This defines the sequence of actions that are taken when the preemptive concurrency state is executed

        :return:
546257
def run(self):
        
        logger.debug("Starting execution of {0}{1}".format(self, " (backwards)" if self.backward_execution else ""))
        self.setup_run()

        try:
            concurrency_history_item = self.setup_forward_or_backward_execution()
            concurrency_queue = self.start_child_states(concurrency_history_item)

            #######################################################
            # wait for the first threads to finish
            #######################################################
            finished_thread_id = concurrency_queue.get()
            finisher_state = self.states[finished_thread_id]
            finisher_state.join()

            # preempt all child states
            if not self.backward_execution:
                for state_id, state in self.states.items():
                    state.recursively_preempt_states()
            # join all states
            for history_index, state in enumerate(self.states.values()):
                self.join_state(state, history_index, concurrency_history_item)
                self.add_state_execution_output_to_scoped_data(state.output_data, state)
                self.update_scoped_variables_with_output_dictionary(state.output_data, state)

            # add the data of the first state now to overwrite data of the preempted states
            self.add_state_execution_output_to_scoped_data(finisher_state.output_data, finisher_state)
            self.update_scoped_variables_with_output_dictionary(finisher_state.output_data, finisher_state)
            #######################################################
            # handle backward execution case
            #######################################################
            if self.states[finished_thread_id].backward_execution:
                return self.finalize_backward_execution()

            else:
                self.backward_execution = False

            #######################################################
            # handle no transition
            #######################################################
            transition = self.get_transition_for_outcome(self.states[finished_thread_id],
                                                         self.states[finished_thread_id].final_outcome)
            if transition is None:
                # final outcome is set here
                transition = self.handle_no_transition(self.states[finished_thread_id])
            # it the transition is still None, then the state was preempted or aborted, in this case return
            if transition is None:
                self.output_data["error"] = RuntimeError("state aborted")
            else:
                if 'error' in self.states[finished_thread_id].output_data:
                    self.output_data["error"] = self.states[finished_thread_id].output_data['error']
                self.final_outcome = self.outcomes[transition.to_outcome]

            return self.finalize_concurrency_state(self.final_outcome)

        except Exception as e:
            logger.error("{0} had an internal error: {1}\n{2}".format(self, str(e), str(traceback.format_exc())))
            self.output_data["error"] = e
            self.state_execution_status = StateExecutionStatus.WAIT_FOR_NEXT_STATE
            return self.finalize(Outcome(-1, "aborted"))
91903
function Queue(concurrency) {
  this.paused = false;
  this.concurrency = concurrency;
  this.waitTimeout = 0;
  this.processTimeout = 0;
  this.throttleCount = 0;
  this.throttleInterval = 1000;
  this.count = 0;
  this.tasks = [];
  this.waiting = [];
  this.factors = {};
  this.fifoMode = true;
  this.roundRobinMode = false;
  this.priorityMode = false;
  this.onProcess = null;
  this.onDone = null;
  this.onSuccess = null;
  this.onTimeout = null;
  this.onFailure = null;
  this.onDrain = null;
}
508115
def _serviceActions(self) -> int:
        
        if self.aqStash:
            tm = time.perf_counter()
            if tm > self.aqNextCheck:
                earliest = float('inf')
                for d in list(self.aqStash):
                    nxt, action = d
                    if tm > nxt:
                        self.actionQueue.appendleft(action)
                        self.aqStash.remove(d)
                    if nxt < earliest:
                        earliest = nxt
                self.aqNextCheck = earliest
        count = len(self.actionQueue)
        while self.actionQueue:
            action, aid = self.actionQueue.popleft()
            assert action in self.scheduled
            if aid in self.scheduled[action]:
                self.scheduled[action].remove(aid)
                logger.trace("{} running action {} with id {}".
                             format(self, get_func_name(action), aid))
                action()
            else:
                logger.trace("{} not running cancelled action {} with id {}".
                             format(self, get_func_name(action), aid))
        return count
948817
void afterExecution() {
        for (;;) {
            State currentState = state.get();
            switch (currentState) {
                case REQUESTED:
                    state.set(State.RUNNING);
                    scheduleExecution();
                    return;
                case RUNNING:
                    if (state.compareAndSet(State.RUNNING, State.INACTIVE)) {
                        return;
                    }
                    break;
                default:
                    throw new IllegalStateException("Inactive state is illegal here.");
            }
        }
    }
531032
public void prependActionsToActionStateExecutionList(final Flow flow, final ActionState actionStateId, final String... actions) {
        prependActionsToActionStateExecutionList(flow, actionStateId.getId(), actions);
    }
638059
def finalize_backward_execution(self):
        
        # backward_execution needs to be True to signal the parent container state the backward execution
        self.backward_execution = True
        # pop the ConcurrencyItem as we are leaving the barrier concurrency state
        last_history_item = self.execution_history.pop_last_item()
        assert isinstance(last_history_item, ConcurrencyItem)

        last_history_item = self.execution_history.pop_last_item()
        assert isinstance(last_history_item, CallItem)
        # this copy is convenience and not required here
        self.scoped_data = last_history_item.scoped_data
        self.state_execution_status = StateExecutionStatus.WAIT_FOR_NEXT_STATE
        return self.finalize()
958926
def _handle_forward_execution_after_child_execution(self):
        
        self.add_state_execution_output_to_scoped_data(self.child_state.output_data, self.child_state)
        self.update_scoped_variables_with_output_dictionary(self.child_state.output_data, self.child_state)
        self.execution_history.push_return_history_item(
            self.child_state, CallType.EXECUTE, self, self.child_state.output_data)
        # not explicitly connected preempted outcomes are implicit connected to parent preempted outcome
        transition = self.get_transition_for_outcome(self.child_state, self.child_state.final_outcome)

        if transition is None:
            transition = self.handle_no_transition(self.child_state)
        # if the transition is still None, then the child_state was preempted or aborted, in this case
        # return
        if transition is None:
            return True

        self.last_transition = transition
        self.child_state = self.get_state_for_transition(transition)
        if transition is not None and self.child_state is self:
            self.final_outcome = self.outcomes[transition.to_outcome]

        if self.child_state is self:
            singleton.state_machine_execution_engine._modify_run_to_states(self)
        return False
793131
protected function define_execution() {
        // A block in the restore list may be skipped because a duplicate is detected.
        // In this case, there is no new blockid (or context) to get.
        if (!empty($this->task->get_blockid())) {
            $context = \context_block::instance($this->task->get_blockid());
            \core_search\manager::request_index($context);
        }
    }
36804
def execution_context(self):
        
        if self._execution_context is None:
            self._execution_context = ExecutionContextList(
                self._version,
                flow_sid=self._solution['flow_sid'],
                execution_sid=self._solution['sid'],
            )
        return self._execution_context
26242
def _determine_scheduled_actions(scheduled_actions, scheduled_actions_from_pillar):
    '''
    
    '''
    tmp = copy.deepcopy(
        __salt__['config.option'](scheduled_actions_from_pillar, {})
    )
    # merge with data from state
    if scheduled_actions:
        tmp = dictupdate.update(tmp, scheduled_actions)
    return tmp
995908
@Trivial
    private void performAction(Runnable action, boolean addToQueue) {
        ExecutorService exec = executorService.getService();

        if (exec == null) {
            // If we can't find the executor service, we have to run it in place.
            action.run();
        } else {
            // If we can find the executor service, we'll add the action to the queue.
            // If the actionFuture is null (no pending actions) and the configFuture is null (no
            // pending configuration updates), we'll submit the actionsRunner to the executor
            // service to drain the queue
            //
            // configFuture is used to avoid bouncing the endpoint multiple times because of a
            // single configuration update.
            //
            // actionFuture is only set to a non-null value by kicking off the executor service here.
            // actionsRunner syncs on actionQueue, so we can't add any new actions while we are
            // draining the queue. When the queue is empty, actionFuture is explicitly set to null.
            //
            // Long story short, it prevents us from kicking off multiple executors which could run in
            // random order.
            if (addToQueue) {
                synchronized (actionQueue) {
                    actionQueue.add(action);
                    if ((actionFuture == null) && (configFuture == null)) {
                        actionFuture = exec.submit(actionsRunner);
                    }
                }
            } else {
                // Schedule immediately
                exec.submit(action);
            }
        }
    }
1103589
def exec_before_request_actions(actions, **kwargs):
    
    groups = ("before", "before_" + flask.request.method.lower())
    return execute_actions(actions, limit_groups=groups, **kwargs)
847997
def pre_dispatch(self):
        
        middleware = sort_by_priority(self)
        return tuple(m.pre_dispatch for m in middleware if hasattr(m, 'pre_dispatch'))
1308313
private ChunkStatus getNextChunkStatusBasedOnPrevious() {

        // If this is the first chunk
        if (currentChunkStatus == null) {
            return new ChunkStatus();
        }

        ChunkStatus nextChunkStatus = null;

        // At this point the 'current' status is the previous chunk's status.
        if (currentChunkStatus.wasMarkedForRollbackWithRetry()) {

            // Re-position reader & writer
            getTransactionManager().begin();
            positionReaderAtCheckpoint();
            positionWriterAtCheckpoint();
            getTransactionManager().commit();

            nextChunkStatus = new ChunkStatus(ChunkStatusType.RETRY_AFTER_ROLLBACK);

            // What happens if we get a retry-with-rollback on a single item that we were processing
            // after a prior retry with rollback?   We don't want to revert to normal processing
            // after completing only the single item of the "single item chunk".  We want to complete
            // the full portion of the original chunk.  So be careful to propagate this number if
            // it already exists.
            int numToProcessOneByOne = currentChunkStatus.getItemsToProcessOneByOneAfterRollback();
            if (numToProcessOneByOne > 0) {
                // Retry after rollback AFTER a previous retry after rollback
                nextChunkStatus.setItemsToProcessOneByOneAfterRollback(numToProcessOneByOne);
            } else {
                // "Normal" (i.e. the first) retry after rollback.
                nextChunkStatus.setItemsToProcessOneByOneAfterRollback(currentChunkStatus.getItemsTouchedInCurrentChunk());
            }
        } else if (currentChunkStatus.isRetryingAfterRollback()) {
            // In this case the 'current' (actually the last) chunk was a single-item retry after rollback chunk,
            // so we have to see if it's time to revert to normal processing.
            int numToProcessOneByOne = currentChunkStatus.getItemsToProcessOneByOneAfterRollback();
            if (numToProcessOneByOne == 1) {
                // we're done, revert to normal
                nextChunkStatus = new ChunkStatus();
            } else {
                nextChunkStatus = new ChunkStatus(ChunkStatusType.RETRY_AFTER_ROLLBACK);
                nextChunkStatus.setItemsToProcessOneByOneAfterRollback(numToProcessOneByOne - 1);
            }
        } else {
            nextChunkStatus = new ChunkStatus();
        }

        return nextChunkStatus;
    }
828523
def run_to_states(self):
        
        self.execution_engine_lock.acquire()
        return_value = self._run_to_states
        self.execution_engine_lock.release()
        return return_value
1129866
def invoke_prerequisites_concurrently(task_args, invocation_chain)# :nodoc:
      futures = prerequisite_tasks.map do |p|
        prereq_args = task_args.new_scope(p.arg_names)
        application.thread_pool.future(p) do |r|
          r.invoke_with_call_chain(prereq_args, invocation_chain)
        end
      end
      futures.each { |f| f.value }
    end
1007644
def auto_execute
      coordinator.acquire(Coordinator::AutoExecuteLock.new(self)) do
        planned_execution_plans =
            self.persistence.find_execution_plans filters: { 'state' => %w(planned paused), 'result' => (ExecutionPlan.results - [:error]).map(&:to_s) }
        planned_execution_plans.map do |ep|
          if coordinator.find_locks(Dynflow::Coordinator::ExecutionLock.unique_filter(ep.id)).empty?
            execute(ep.id)
          end
        end.compact
      end
    rescue Coordinator::LockError => e
      logger.info "auto-executor lock already aquired: #{e.message}"
      []
    end