class ActionScheduler implements Runnable {

  public void doWork() throws AmbariException {
    try {
      unitOfWork.begin();

      // grab a reference to this UnitOfWork's EM
      threadEntityManager = entityManagerProvider.get();

      // The first thing to do is to abort requests that are cancelled
      processCancelledRequestsList();

      // !!! getting the stages in progress could be a very expensive call due
      // to the join being used; there's no need to make it if there are
      // no commands in progress
      if (db.getCommandsInProgressCount() == 0) {
        // Nothing to do
        if (LOG.isDebugEnabled()) {
          LOG.debug("There are no stages currently in progress.");
        }

        return;
      }

      Set<Long> runningRequestIds = new HashSet<>();
      List<Stage> firstStageInProgressPerRequest = db.getFirstStageInProgressPerRequest();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Scheduler wakes up");
        LOG.debug("Processing {} in progress stages", firstStageInProgressPerRequest.size());
      }

      publishInProgressTasks(firstStageInProgressPerRequest);

      if (firstStageInProgressPerRequest.isEmpty()) {
        // Nothing to do
        if (LOG.isDebugEnabled()) {
          LOG.debug("There are no stages currently in progress.");
        }

        return;
      }

      int i_stage = 0;

      // get the range of requests in progress
      long iLowestRequestIdInProgress = firstStageInProgressPerRequest.get(0).getRequestId();
      long iHighestRequestIdInProgress = firstStageInProgressPerRequest.get(
          firstStageInProgressPerRequest.size() - 1).getRequestId();

      List<String> hostsWithPendingTasks = hostRoleCommandDAO.getHostsWithPendingTasks(
          iLowestRequestIdInProgress, iHighestRequestIdInProgress);

      // filter the stages in progress down to those which can be scheduled in
      // parallel
      List<Stage> stages = filterParallelPerHostStages(firstStageInProgressPerRequest);

      boolean exclusiveRequestIsGoing = false;
      // This loop greatly depends on the fact that order of stages in
      // a list does not change between invocations
      for (Stage stage : stages) {
        // Check if we can process this stage in parallel with another stages
        i_stage++;
        long requestId = stage.getRequestId();
        LOG.debug("==> STAGE_i = {}(requestId={},StageId={})", i_stage, requestId, stage.getStageId());

        RequestEntity request = db.getRequestEntity(requestId);

        if (request.isExclusive()) {
          if (runningRequestIds.size() > 0) {
            // As a result, we will wait until any previous stages are finished
            LOG.debug("Stage requires exclusive execution, but other requests are already executing. Stopping for now");
            break;
          }
          exclusiveRequestIsGoing = true;
        }

        if (runningRequestIds.contains(requestId)) {
          // We don't want to process different stages from the same request in parallel
          LOG.debug("==> We don't want to process different stages from the same request in parallel");
          continue;
        } else {
          runningRequestIds.add(requestId);
          if (!requestsInProgress.contains(requestId)) {
            requestsInProgress.add(requestId);
            db.startRequest(requestId);
          }
        }

        // Commands that will be scheduled in current scheduler wakeup
        List<ExecutionCommand> commandsToSchedule = new ArrayList<>();
        Multimap<Long, AgentCommand> commandsToEnqueue = ArrayListMultimap.create();

        Map<String, RoleStats> roleStats =
          processInProgressStage(stage, commandsToSchedule, commandsToEnqueue);

        // Check if stage is failed
        boolean failed = false;
        for (Map.Entry<String, RoleStats> entry : roleStats.entrySet()) {

          String role = entry.getKey();
          RoleStats stats = entry.getValue();

          if (LOG.isDebugEnabled()) {
            LOG.debug("Stats for role: {}, stats={}", role, stats);
          }

          // only fail the request if the role failed and the stage is not
          // skippable
          if (stats.isRoleFailed() && !stage.isSkippable()) {
            LOG.warn("{} failed, request {} will be aborted", role, request.getRequestId());

            failed = true;
            break;
          }
        }

        if (!failed) {
          // Prior stage may have failed and it may need to fail the whole request
          failed = hasPreviousStageFailed(stage);
        }

        if (failed) {
          LOG.error("Operation completely failed, aborting request id: {}", stage.getRequestId());
          cancelHostRoleCommands(stage.getOrderedHostRoleCommands(), FAILED_TASK_ABORT_REASONING);
          abortOperationsForStage(stage);
          return;
        }

        List<ExecutionCommand> commandsToStart = new ArrayList<>();
        List<ExecutionCommand> commandsToUpdate = new ArrayList<>();

        //Schedule what we have so far


        for (ExecutionCommand cmd : commandsToSchedule) {
          ConfigHelper.processHiddenAttribute(cmd.getConfigurations(), cmd.getConfigurationAttributes(), cmd.getRole(), false);
          processHostRole(request, stage, cmd, commandsToStart, commandsToUpdate);
        }

        LOG.debug("==> Commands to start: {}", commandsToStart.size());
        LOG.debug("==> Commands to update: {}", commandsToUpdate.size());

        //Multimap is analog of Map<Object, List<Object>> but allows to avoid nested loop
        ListMultimap<String, ServiceComponentHostEvent> eventMap = formEventMap(stage, commandsToStart);
        Map<ExecutionCommand, String> commandsToAbort = new HashMap<>();
        if (!eventMap.isEmpty()) {
          LOG.debug("==> processing {} serviceComponentHostEvents...", eventMap.size());
          Cluster cluster = clusters.getCluster(stage.getClusterName());
          if (cluster != null) {
            Map<ServiceComponentHostEvent, String> failedEvents = cluster.processServiceComponentHostEvents(eventMap);

            if (failedEvents.size() > 0) {
              LOG.error("==> {} events failed.", failedEvents.size());
            }

            for (Iterator<ExecutionCommand> iterator = commandsToUpdate.iterator(); iterator.hasNext(); ) {
              ExecutionCommand cmd = iterator.next();
              for (ServiceComponentHostEvent event : failedEvents.keySet()) {
                if (StringUtils.equals(event.getHostName(), cmd.getHostname()) &&
                  StringUtils.equals(event.getServiceComponentName(), cmd.getRole())) {
                  iterator.remove();
                  commandsToAbort.put(cmd, failedEvents.get(event));
                  break;
                }
              }
            }
          } else {
            LOG.warn("There was events to process but cluster {} not found", stage.getClusterName());
          }
        }

        LOG.debug("==> Scheduling {} tasks...", commandsToUpdate.size());
        db.bulkHostRoleScheduled(stage, commandsToUpdate);

        if (commandsToAbort.size() > 0) { // Code branch may be a bit slow, but is extremely rarely used
          LOG.debug("==> Aborting {} tasks...", commandsToAbort.size());
          // Build a list of HostRoleCommands
          List<Long> taskIds = new ArrayList<>();
          for (ExecutionCommand command : commandsToAbort.keySet()) {
            taskIds.add(command.getTaskId());
          }
          Collection<HostRoleCommand> hostRoleCommands = db.getTasks(taskIds);

          cancelHostRoleCommands(hostRoleCommands, FAILED_TASK_ABORT_REASONING);
          db.bulkAbortHostRole(stage, commandsToAbort);
        }

        LOG.debug("==> Adding {} tasks to queue...", commandsToUpdate.size());
        for (ExecutionCommand cmd : commandsToUpdate) {
          // Do not queue up server actions; however if we encounter one, wake up the ServerActionExecutor
          if (Role.AMBARI_SERVER_ACTION.name().equals(cmd.getRole())) {
            serverActionExecutor.awake();
          } else {
            commandsToEnqueue.put(clusters.getHost(cmd.getHostname()).getHostId(), cmd);
          }
        }
        if (!commandsToEnqueue.isEmpty()) {
          agentCommandsPublisher.sendAgentCommand(commandsToEnqueue);
        }
        LOG.debug("==> Finished.");

        if (!configuration.getParallelStageExecution()) { // If disabled
          return;
        }

        if (exclusiveRequestIsGoing) {
          // As a result, we will prevent any further stages from being executed
          LOG.debug("Stage requires exclusive execution, skipping all executing any further stages");
          break;
        }
      }

      requestsInProgress.retainAll(runningRequestIds);

    } finally {
      LOG.debug("Scheduler finished work.");
      unitOfWork.end();
    }
  }

  @Inject public  ActionScheduler(@Named("schedulerSleeptime") long sleepTime,
                         @Named("actionTimeout") long actionTimeout, ActionDBAccessor db,
                         JPAEventPublisher jpaPublisher);
  protected  ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory,
                            RoleCommandOrderProvider roleCommandOrderProvider, AgentCommandsPublisher agentCommandsPublisher);
  protected  ActionScheduler(long sleepTimeMilliSec, long actionTimeoutMilliSec, ActionDBAccessor db,
                            Clusters fsmObject, int maxAttempts, HostsMap hostsMap,
                            UnitOfWork unitOfWork, AmbariEventPublisher ambariEventPublisher,
                            Configuration configuration, Provider<EntityManager> entityManagerProvider,
                            HostRoleCommandDAO hostRoleCommandDAO, HostRoleCommandFactory hostRoleCommandFactory,
                            AgentCommandsPublisher agentCommandsPublisher);

}