Engines¶

Why they exist¶

An engine being the core component which actually makes your flows progress is likely a new concept for many programmers so let’s describe how it operates in more depth and some of the reasoning behind why it exists. This will hopefully make it more clear on their value add to the TaskFlow library user.

First though let us discuss something most are familiar already with; the difference between declarative and imperative programming models. The imperative model involves establishing statements that accomplish a programs action (likely using conditionals and such other language features to do this). This kind of program embeds the how to accomplish a goal while also defining what the goal actually is (and the state of this is maintained in memory or on the stack while these statements execute). In contrast there is the declarative model which instead of combining the how to accomplish a goal along side the what is to be accomplished splits these two into only declaring what the intended goal is and not the how. In TaskFlow terminology the what is the structure of your flows and the tasks and other atoms you have inside those flows, but the how is not defined (the line becomes blurred since tasks themselves contain imperative code, but for now consider a task as more of a pure function that executes, reverts and may require inputs and provide outputs). This is where engines get involved; they do the execution of the what defined via atoms, tasks, flows and the relationships defined there-in and execute these in a well-defined manner (and the engine is responsible for any state manipulation instead).

This mix of imperative and declarative (with a stronger emphasis on the declarative model) allows for the following functionality to become possible:

• Enhancing reliability: Decoupling of state alterations from what should be accomplished allows for a natural way of resuming by allowing the engine to track the current state and know at which point a workflow is in and how to get back into that state when resumption occurs.
• Enhancing scalability: When an engine is responsible for executing your desired work it becomes possible to alter the how in the future by creating new types of execution backends (for example the worker model which does not execute locally). Without the decoupling of the what and the how it is not possible to provide such a feature (since by the very nature of that coupling this kind of functionality is inherently very hard to provide).
• Enhancing consistency: Since the engine is responsible for executing atoms and the associated workflow, it can be one (if not the only) of the primary entities that is working to keep the execution model in a consistent state. Coupled with atoms which should be immutable and have have limited (if any) internal state the ability to reason about and obtain consistency can be vastly improved.
  • With future features around locking (using tooz to help) engines can also help ensure that resources being accessed by tasks are reliably obtained and mutated on. This will help ensure that other processes, threads, or other types of entities are also not executing tasks that manipulate those same resources (further increasing consistency).

Of course these kind of features can come with some drawbacks:

• The downside of decoupling the how and the what is that the imperative model where functions control & manipulate state must start to be shifted away from (and this is likely a mindset change for programmers used to the imperative model). We have worked to make this less of a concern by creating and encouraging the usage of persistence, to help make it possible to have state and transfer that state via a argument input and output mechanism.
• Depending on how much imperative code exists (and state inside that code) there may be significant rework of that code and converting or refactoring it to these new concepts. We have tried to help here by allowing you to have tasks that internally use regular python code (and internally can be written in an imperative style) as well as by providing examples that show how to use these concepts.
• Another one of the downsides of decoupling the what from the how is that it may become harder to use traditional techniques to debug failures (especially if remote workers are involved). We try to help here by making it easy to track, monitor and introspect the actions & state changes that are occurring inside an engine (see notifications for how to use some of these capabilities).

Serial¶

Engine type: 'serial'

Runs all tasks on a single thread – the same thread run() is called from.

Parallel¶

Engine type: 'parallel'

A parallel engine schedules tasks onto different threads/processes to allow for running non-dependent tasks simultaneously. See the documentation of ParallelActionEngine for supported arguments that can be used to construct a parallel engine that runs using your desired execution model.

Workers¶

Engine type: 'worker-based' or 'workers'

Creation¶

The first thing that occurs is that the user creates an engine for a given flow, providing a flow detail (where results will be saved into a provided persistence backend). This is typically accomplished via the methods described above in creating engines. The engine at this point now will have references to your flow and backends and other internal variables are setup.

Compiling¶

During this stage (see compile()) the flow will be converted into an internal graph representation using a compiler (the default implementation for patterns is the PatternCompiler). This class compiles/converts the flow objects and contained atoms into a networkx directed graph (and tree structure) that contains the equivalent atoms defined in the flow and any nested flows & atoms as well as the constraints that are created by the application of the different flow patterns. This graph (and tree) are what will be analyzed & traversed during the engines execution. At this point a few helper object are also created and saved to internal engine variables (these object help in execution of atoms, analyzing the graph and performing other internal engine activities). At the finishing of this stage a Runtime object is created which contains references to all needed runtime components and its compile() is called to compile a cache of frequently used execution helper objects.

Preparation¶

This stage (see prepare()) starts by setting up the storage needed for all atoms in the compiled graph, ensuring that corresponding AtomDetail (or subclass of) objects are created for each node in the graph.

Validation¶

This stage (see validate()) performs any final validation of the compiled (and now storage prepared) engine. It compares the requirements that are needed to start execution and what is currently provided or will be produced in the future. If there are any atom requirements that are not satisfied (no known current provider or future producer is found) then execution will not be allowed to continue.

Execution¶

The graph (and helper objects) previously created are now used for guiding further execution (see run()). The flow is put into the RUNNING state and a MachineBuilder state machine object and runner object are built (using the automaton library). That machine and associated runner then starts to take over and begins going through the stages listed below (for a more visual diagram/representation see the engine state diagram).

Finishing¶

At this point the machine (and runner) that was built using the MachineBuilder class has now finished successfully, failed, or the execution was suspended. Depending on which one of these occurs will cause the flow to enter a new state (typically one of FAILURE, SUSPENDED, SUCCESS or REVERTED). Notifications will be sent out about this final state change (other state changes also send out notifications) and any failures that occurred will be reraised (the failure objects are wrapped exceptions). If no failures have occurred then the engine will have finished and if so desired the persistence can be used to cleanup any details that were saved for this execution.

Suspension¶

Each engine implements a suspend() method that can be used to externally (or in the future internally) request that the engine stop scheduling new work. By default what this performs is a transition of the flow state from RUNNING into a SUSPENDING state (which will later transition into a SUSPENDED state). Since an engine may be remotely executing atoms (or locally executing them) and there is currently no preemption what occurs is that the engines MachineBuilder state machine will detect this transition into SUSPENDING has occurred and the state machine will avoid scheduling new work (it will though let active work continue). After the current work has finished the engine will transition from SUSPENDING into SUSPENDED and return from its run() method.

Default strategy¶

When an engine is executing it internally interacts with the Storage class and that class interacts with the a ScopeWalker instance and the Storage class uses the following lookup order to find (or fail) a atoms requirement lookup/request:

1. Transient injected atom specific arguments.
2. Non-transient injected atom specific arguments.
3. Transient injected arguments (flow specific).
4. Non-transient injected arguments (flow specific).
5. First scope visited provider that produces the named result; note that if multiple providers are found in the same scope the first (the scope walkers yielded ordering defines what first means) that produced that result and can be extracted without raising an error is selected as the provider of the requested requirement.
6. Fails with NotFound if unresolved at this point (the cause attribute of this exception may have more details on why the lookup failed).

Components¶

class taskflow.engines.action_engine.builder.MachineMemory[source]¶

Bases: object

State machine memory.

cancel_futures()[source]¶

Attempts to cancel any not done futures.

class taskflow.engines.action_engine.builder.MachineBuilder(runtime, waiter)[source]¶

Bases: object

State machine builder that powers the engine components.

NOTE(harlowja): the machine (states and events that will trigger transitions) that this builds is represented by the following table:

+--------------+------------------+------------+----------+---------+
|    Start     |      Event       |    End     | On Enter | On Exit |
+--------------+------------------+------------+----------+---------+
|  ANALYZING   |    completed     | GAME_OVER  |    .     |    .    |
|  ANALYZING   |  schedule_next   | SCHEDULING |    .     |    .    |
|  ANALYZING   |  wait_finished   |  WAITING   |    .     |    .    |
|  FAILURE[$]  |        .         |     .      |    .     |    .    |
|  GAME_OVER   |      failed      |  FAILURE   |    .     |    .    |
|  GAME_OVER   |     reverted     |  REVERTED  |    .     |    .    |
|  GAME_OVER   |     success      |  SUCCESS   |    .     |    .    |
|  GAME_OVER   |    suspended     | SUSPENDED  |    .     |    .    |
|   RESUMING   |  schedule_next   | SCHEDULING |    .     |    .    |
| REVERTED[$]  |        .         |     .      |    .     |    .    |
|  SCHEDULING  |  wait_finished   |  WAITING   |    .     |    .    |
|  SUCCESS[$]  |        .         |     .      |    .     |    .    |
| SUSPENDED[$] |        .         |     .      |    .     |    .    |
| UNDEFINED[^] |      start       |  RESUMING  |    .     |    .    |
|   WAITING    | examine_finished | ANALYZING  |    .     |    .    |
+--------------+------------------+------------+----------+---------+

Between any of these yielded states (minus GAME_OVER and UNDEFINED) if the engine has been suspended or the engine has failed (due to a non-resolveable task failure or scheduling failure) the machine will stop executing new tasks (currently running tasks will be allowed to complete) and this machines run loop will be broken.

NOTE(harlowja): If the runtimes scheduler component is able to schedule tasks in parallel, this enables parallel running and/or reversion.

build(statistics, timeout=None, gather_statistics=True)[source]¶

Builds a state-machine (that is used during running).

class taskflow.engines.action_engine.compiler.Terminator(flow)[source]¶

Bases: object

Flow terminator class.

flow¶

The flow which this terminator signifies/marks the end of.

name¶

Useful name this end terminator has (derived from flow name).

class taskflow.engines.action_engine.compiler.Compilation(execution_graph, hierarchy)[source]¶

Bases: object

The result of a compilers compile() is this immutable object.

TASK = 'task'¶

Task nodes will have a kind metadata key with this value.

RETRY = 'retry'¶

Retry nodes will have a kind metadata key with this value.

FLOW = 'flow'¶

Flow entry nodes will have a kind metadata key with this value.

FLOW_END = 'flow_end'¶

Flow exit nodes will have a kind metadata key with this value (only applicable for compilation execution graph, not currently used in tree hierarchy).

execution_graph¶

The execution ordering of atoms (as a graph structure).

hierarchy¶

The hierarchy of patterns (as a tree structure).

class taskflow.engines.action_engine.compiler.TaskCompiler[source]¶

Bases: object

Non-recursive compiler of tasks.

class taskflow.engines.action_engine.compiler.FlowCompiler(deep_compiler_func)[source]¶

Bases: object

Recursive compiler of flows.

compile(flow, parent=None)[source]¶

Decomposes a flow into a graph and scope tree hierarchy.

class taskflow.engines.action_engine.compiler.PatternCompiler(root, freeze=True)[source]¶

Bases: object

Compiles a flow pattern (or task) into a compilation unit.

Let’s dive into the basic idea for how this works:

The compiler here is provided a ‘root’ object via its __init__ method, this object could be a task, or a flow (one of the supported patterns), the end-goal is to produce a Compilation object as the result with the needed components. If this is not possible a CompilationFailure will be raised. In the case where a unknown type is being requested to compile a TypeError will be raised and when a duplicate object (one that has already been compiled) is encountered a ValueError is raised.

The complexity of this comes into play when the ‘root’ is a flow that contains itself other nested flows (and so-on); to compile this object and its contained objects into a graph that preserves the constraints the pattern mandates we have to go through a recursive algorithm that creates subgraphs for each nesting level, and then on the way back up through the recursion (now with a decomposed mapping from contained patterns or atoms to there corresponding subgraph) we have to then connect the subgraphs (and the atom(s) there-in) that were decomposed for a pattern correctly into a new graph and then ensure the pattern mandated constraints are retained. Finally we then return to the caller (and they will do the same thing up until the root node, which by that point one graph is created with all contained atoms in the pattern/nested patterns mandated ordering).

Also maintained in the Compilation object is a hierarchy of the nesting of items (which is also built up during the above mentioned recusion, via a much simpler algorithm); this is typically used later to determine the prior atoms of a given atom when looking up values that can be provided to that atom for execution (see the scopes.py file for how this works). Note that although you could think that the graph itself could be used for this, which in some ways it can (for limited usage) the hierarchy retains the nested structure (which is useful for scoping analysis/lookup) to be able to provide back a iterator that gives back the scopes visible at each level (the graph does not have this information once flattened).

Let’s take an example:

Given the pattern f(a(b, c), d) where f is a Flow with items a(b, c) where a is a Flow composed of tasks (b, c) and task d.

The algorithm that will be performed (mirroring the above described logic) will go through the following steps (the tree hierarchy building is left out as that is more obvious):

Compiling f
  - Decomposing flow f with no parent (must be the root)
  - Compiling a
      - Decomposing flow a with parent f
      - Compiling b
          - Decomposing task b with parent a
          - Decomposed b into:
            Name: b
            Nodes: 1
              - b
            Edges: 0
      - Compiling c
          - Decomposing task c with parent a
          - Decomposed c into:
            Name: c
            Nodes: 1
              - c
            Edges: 0
      - Relinking decomposed b -> decomposed c
      - Decomposed a into:
        Name: a
        Nodes: 2
          - b
          - c
        Edges: 1
          b -> c ({'invariant': True})
  - Compiling d
      - Decomposing task d with parent f
      - Decomposed d into:
        Name: d
        Nodes: 1
          - d
        Edges: 0
  - Relinking decomposed a -> decomposed d
  - Decomposed f into:
    Name: f
    Nodes: 3
      - c
      - b
      - d
    Edges: 2
      c -> d ({'invariant': True})
      b -> c ({'invariant': True})

compile()[source]¶

Compiles the contained item into a compiled equivalent.

class taskflow.engines.action_engine.completer.Strategy(runtime)[source]¶

Bases: object

Failure resolution strategy base class.

apply()[source]¶

Applies some algorithm to resolve some detected failure.

class taskflow.engines.action_engine.completer.RevertAndRetry(runtime, retry)[source]¶

Bases: taskflow.engines.action_engine.completer.Strategy

Sets the associated subflow for revert to be later retried.

apply()[source]¶

Applies some algorithm to resolve some detected failure.

class taskflow.engines.action_engine.completer.RevertAll(runtime)[source]¶

Bases: taskflow.engines.action_engine.completer.Strategy

Sets all nodes/atoms to the REVERT intention.

apply()[source]¶

Applies some algorithm to resolve some detected failure.

class taskflow.engines.action_engine.completer.Revert(runtime, atom)[source]¶

Bases: taskflow.engines.action_engine.completer.Strategy

Sets atom and associated nodes to the REVERT intention.

apply()[source]¶

Applies some algorithm to resolve some detected failure.

class taskflow.engines.action_engine.completer.Completer(runtime)[source]¶

Bases: object

Completes atoms using actions to complete them.

resume()[source]¶

Resumes atoms in the contained graph.

This is done to allow any previously completed or failed atoms to be analyzed, there results processed and any potential atoms affected to be adjusted as needed.

This should return a set of atoms which should be the initial set of atoms that were previously not finished (due to a RUNNING or REVERTING attempt not previously finishing).

complete_failure(node, outcome, failure)[source]¶

Performs post-execution completion of a nodes failure.

Returns whether the result should be saved into an accumulator of failures or whether this should not be done.

complete(node, outcome, result)[source]¶

Performs post-execution completion of a node result.

class taskflow.engines.action_engine.deciders.Decider[source]¶

Bases: object

Base class for deciders.

Provides interface to be implemented by sub-classes.

Deciders check whether next atom in flow should be executed or not.

tally(runtime)[source]¶

Tally edge deciders on whether this decider should allow running.

The returned value is a list of edge deciders that voted ‘nay’ (do not allow running).

affect(runtime, nay_voters)[source]¶

Affects associated atoms due to at least one ‘nay’ edge decider.

This will alter the associated atom + some set of successor atoms by setting there state and intention to IGNORE so that they are ignored in future runtime activities.

check_and_affect(runtime)[source]¶

Handles tally() + affect() in right order.

NOTE(harlowja): If there are zero ‘nay’ edge deciders then it is assumed this decider should allow running.

Returns boolean of whether this decider allows for running (or not).

class taskflow.engines.action_engine.deciders.IgnoreDecider(atom, edge_deciders)[source]¶

Bases: taskflow.engines.action_engine.deciders.Decider

Checks any provided edge-deciders and determines if ok to run.

tally(runtime)[source]¶

Tally edge deciders on whether this decider should allow running.

The returned value is a list of edge deciders that voted ‘nay’ (do not allow running).

affect(runtime, nay_voters)[source]¶

Affects associated atoms due to at least one ‘nay’ edge decider.

This will alter the associated atom + some set of successor atoms by setting there state and intention to IGNORE so that they are ignored in future runtime activities.

class taskflow.engines.action_engine.deciders.NoOpDecider[source]¶

Bases: taskflow.engines.action_engine.deciders.Decider

No-op decider that says it is always ok to run & has no effect(s).

tally(runtime)[source]¶

Always good to go.

affect(runtime, nay_voters)[source]¶

Does nothing.

class taskflow.engines.action_engine.executor.SerialRetryExecutor[source]¶

Bases: object

Executes and reverts retries.

start()[source]¶

Prepare to execute retries.

stop()[source]¶

Finalize retry executor.

execute_retry(retry, arguments)[source]¶

Schedules retry execution.

revert_retry(retry, arguments)[source]¶

Schedules retry reversion.

class taskflow.engines.action_engine.executor.TaskExecutor[source]¶

Bases: object

Executes and reverts tasks.

This class takes task and its arguments and executes or reverts it. It encapsulates knowledge on how task should be executed or reverted: right now, on separate thread, on another machine, etc.

execute_task(task, task_uuid, arguments, progress_callback=None)[source]¶

Schedules task execution.

revert_task(task, task_uuid, arguments, result, failures, progress_callback=None)[source]¶

Schedules task reversion.

start()[source]¶

Prepare to execute tasks.

stop()[source]¶

Finalize task executor.

class taskflow.engines.action_engine.executor.SerialTaskExecutor[source]¶

Bases: taskflow.engines.action_engine.executor.TaskExecutor

Executes tasks one after another.

start()[source]¶

Prepare to execute tasks.

stop()[source]¶

Finalize task executor.

execute_task(task, task_uuid, arguments, progress_callback=None)[source]¶

Schedules task execution.

revert_task(task, task_uuid, arguments, result, failures, progress_callback=None)[source]¶

Schedules task reversion.

class taskflow.engines.action_engine.executor.ParallelTaskExecutor(executor=None, max_workers=None)[source]¶

Bases: taskflow.engines.action_engine.executor.TaskExecutor

Executes tasks in parallel.

Submits tasks to an executor which should provide an interface similar to concurrent.Futures.Executor.

constructor_options = [('max_workers', <function ParallelTaskExecutor.<lambda>>)]¶

Optional constructor keyword arguments this executor supports. These will typically be passed via engine options (by a engine user) and converted into the correct type before being sent into this classes __init__ method.

execute_task(task, task_uuid, arguments, progress_callback=None)[source]¶

Schedules task execution.

revert_task(task, task_uuid, arguments, result, failures, progress_callback=None)[source]¶

Schedules task reversion.

start()[source]¶

Prepare to execute tasks.

stop()[source]¶

Finalize task executor.

class taskflow.engines.action_engine.executor.ParallelThreadTaskExecutor(executor=None, max_workers=None)[source]¶

Bases: taskflow.engines.action_engine.executor.ParallelTaskExecutor

Executes tasks in parallel using a thread pool executor.

class taskflow.engines.action_engine.executor.ParallelGreenThreadTaskExecutor(executor=None, max_workers=None)[source]¶

Bases: taskflow.engines.action_engine.executor.ParallelThreadTaskExecutor

Executes tasks in parallel using a greenthread pool executor.

DEFAULT_WORKERS = 1000¶

Default number of workers when None is passed; being that greenthreads don’t map to native threads or processors very well this is more of a guess/somewhat arbitrary, but it does match what the eventlet greenpool default size is (so at least it’s consistent with what eventlet does).

exception taskflow.engines.action_engine.process_executor.UnknownSender[source]¶

Bases: Exception

Exception raised when message from unknown sender is recvd.

exception taskflow.engines.action_engine.process_executor.ChallengeIgnored[source]¶

Bases: Exception

Exception raised when challenge has not been responded to.

class taskflow.engines.action_engine.process_executor.Reader(auth_key, dispatch_func, msg_limit=-1)[source]¶

Bases: object

Reader machine that streams & parses messages that it then dispatches.

TODO(harlowja): Use python-suitcase in the future when the following are addressed/resolved and released:

• https://github.com/digidotcom/python-suitcase/issues/28
• https://github.com/digidotcom/python-suitcase/issues/29

Binary format format is the following (no newlines in actual format):

<magic-header> (4 bytes)
<mac-header-length> (4 bytes)
<mac> (1 or more variable bytes)
<identity-header-length> (4 bytes)
<identity> (1 or more variable bytes)
<msg-header-length> (4 bytes)
<msg> (1 or more variable bytes)

exception taskflow.engines.action_engine.process_executor.BadHmacValueError[source]¶

Bases: ValueError

Value error raised when an invalid hmac is discovered.

class taskflow.engines.action_engine.process_executor.Channel(port, identity, auth_key)[source]¶

Bases: object

Object that workers use to communicate back to their creator.

class taskflow.engines.action_engine.process_executor.EventSender(channel)[source]¶

Bases: object

Sends event information from a child worker process to its creator.

class taskflow.engines.action_engine.process_executor.DispatcherHandler(sock, addr, dispatcher)[source]¶

Bases: asyncore.dispatcher

Dispatches from a single connection into a target.

CHUNK_SIZE = 8192¶

Read/write chunk size.

class taskflow.engines.action_engine.process_executor.Dispatcher(map, auth_key, identity)[source]¶

Bases: asyncore.dispatcher

Accepts messages received from child worker processes.

MAX_BACKLOG = 5¶

See https://docs.python.org/2/library/socket.html#socket.socket.listen

class taskflow.engines.action_engine.process_executor.ParallelProcessTaskExecutor(executor=None, max_workers=None, wait_timeout=None)[source]¶

Bases: taskflow.engines.action_engine.executor.ParallelTaskExecutor

Executes tasks in parallel using a process pool executor.

NOTE(harlowja): this executor executes tasks in external processes, so that implies that tasks that are sent to that external process are pickleable since this is how the multiprocessing works (sending pickled objects back and forth) and that the bound handlers (for progress updating in particular) are proxied correctly from that external process to the one that is alive in the parent process to ensure that callbacks registered in the parent are executed on events in the child.

WAIT_TIMEOUT = 0.01¶

Default timeout used by asyncore io loop (and eventually select/poll).

constructor_options = [('max_workers', <function ParallelProcessTaskExecutor.<lambda>>), ('wait_timeout', <function ParallelProcessTaskExecutor.<lambda>>)]¶

Optional constructor keyword arguments this executor supports. These will typically be passed via engine options (by a engine user) and converted into the correct type before being sent into this classes __init__ method.

start()[source]¶

Prepare to execute tasks.

stop()[source]¶

Finalize task executor.

class taskflow.engines.action_engine.runtime.Runtime(compilation, storage, atom_notifier, task_executor, retry_executor, options=None)[source]¶

Bases: object

A aggregate of runtime objects, properties, … used during execution.

This object contains various utility methods and properties that represent the collection of runtime components and functionality needed for an action engine to run to completion.

compile()[source]¶

Compiles & caches frequently used execution helper objects.

Build out a cache of commonly used item that are associated with the contained atoms (by name), and are useful to have for quick lookup on (for example, the change state handler function for each atom, the scope walker object for each atom, the task or retry specific scheduler and so-on).

check_atom_transition(atom, current_state, target_state)[source]¶

Checks if the atom can transition to the provided target state.

fetch_edge_deciders(atom)[source]¶

Fetches the edge deciders for the given atom.

fetch_scheduler(atom)[source]¶

Fetches the cached specific scheduler for the given atom.

fetch_action(atom)[source]¶

Fetches the cached action handler for the given atom.

fetch_scopes_for(atom_name)[source]¶

Fetches a walker of the visible scopes for the given atom.

iterate_retries(state=None)[source]¶

Iterates retry atoms that match the provided state.

If no state is provided it will yield back all retry atoms.

iterate_nodes(allowed_kinds)[source]¶

Yields back all nodes of specified kinds in the execution graph.

is_success()[source]¶

Checks if all atoms in the execution graph are in ‘happy’ state.

find_retry(node)[source]¶

Returns the retry atom associated to the given node (or none).

reset_atoms(atoms, state='PENDING', intention='EXECUTE')[source]¶

Resets all the provided atoms to the given state and intention.

reset_all(state='PENDING', intention='EXECUTE')[source]¶

Resets all atoms to the given state and intention.

reset_subgraph(atom, state='PENDING', intention='EXECUTE')[source]¶

Resets a atoms subgraph to the given state and intention.

The subgraph is contained of all of the atoms successors.

retry_subflow(retry)[source]¶

Prepares a retrys + its subgraph for execution.

This sets the retrys intention to EXECUTE and resets all of its subgraph (its successors) to the PENDING state with an EXECUTE intention.

class taskflow.engines.action_engine.scheduler.RetryScheduler(runtime)[source]¶

Bases: object

Schedules retry atoms.

schedule(retry)[source]¶

Schedules the given retry atom for future completion.

Depending on the atoms stored intention this may schedule the retry atom for reversion or execution.

class taskflow.engines.action_engine.scheduler.TaskScheduler(runtime)[source]¶

Bases: object

Schedules task atoms.

schedule(task)[source]¶

Schedules the given task atom for future completion.

Depending on the atoms stored intention this may schedule the task atom for reversion or execution.

class taskflow.engines.action_engine.scheduler.Scheduler(runtime)[source]¶

Bases: object

Safely schedules atoms using a runtime fetch_scheduler routine.

schedule(atoms)[source]¶

Schedules the provided atoms for future completion.

This method should schedule a future for each atom provided and return a set of those futures to be waited on (or used for other similar purposes). It should also return any failure objects that represented scheduling failures that may have occurred during this scheduling process.

class taskflow.engines.action_engine.selector.Selector(runtime)[source]¶

Bases: object

Selector that uses a compilation and aids in execution processes.

Its primary purpose is to get the next atoms for execution or reversion by utilizing the compilations underlying structures (graphs, nodes and edge relations…) and using this information along with the atom state/states stored in storage to provide other useful functionality to the rest of the runtime system.

iter_next_atoms(atom=None)[source]¶

Iterate next atoms to run (originating from atom or all atoms).

class taskflow.engines.action_engine.scopes.ScopeWalker(compilation, atom, names_only=False)[source]¶

Bases: object

Walks through the scopes of a atom using a engines compilation.

NOTE(harlowja): for internal usage only.

This will walk the visible scopes that are accessible for the given atom, which can be used by some external entity in some meaningful way, for example to find dependent values…

__iter__()[source]¶

Iterates over the visible scopes.

How this works is the following:

We first grab all the predecessors of the given atom (lets call it Y) by using the Compilation execution graph (and doing a reverse breadth-first expansion to gather its predecessors), this is useful since we know they always will exist (and execute) before this atom but it does not tell us the corresponding scope level (flow, nested flow…) that each predecessor was created in, so we need to find this information.

For that information we consult the location of the atom Y in the Compilation hierarchy/tree. We lookup in a reverse order the parent X of Y and traverse backwards from the index in the parent where Y exists to all siblings (and children of those siblings) in X that we encounter in this backwards search (if a sibling is a flow itself, its atom(s) will be recursively expanded and included). This collection will then be assumed to be at the same scope. This is what is called a potential single scope, to make an actual scope we remove the items from the potential scope that are not predecessors of Y to form the actual scope which we then yield back.

Then for additional scopes we continue up the tree, by finding the parent of X (lets call it Z) and perform the same operation, going through the children in a reverse manner from the index in parent Z where X was located. This forms another potential scope which we provide back as an actual scope after reducing the potential set to only include predecessors previously gathered. We then repeat this process until we no longer have any parent nodes (aka we have reached the top of the tree) or we run out of predecessors.

class taskflow.engines.action_engine.traversal.Direction[source]¶

Bases: enum.Enum

Traversal direction enum.

FORWARD = 1¶

Go through successors.

BACKWARD = 2¶

Go through predecessors.

taskflow.engines.action_engine.traversal.breadth_first_iterate(execution_graph, starting_node, direction, through_flows=True, through_retries=True, through_tasks=True)[source]¶

Iterates connected nodes in execution graph (from starting node).

Does so in a breadth first manner.

Jumps over nodes with noop attribute (does not yield them back).

taskflow.engines.action_engine.traversal.depth_first_iterate(execution_graph, starting_node, direction, through_flows=True, through_retries=True, through_tasks=True)[source]¶

Iterates connected nodes in execution graph (from starting node).

Does so in a depth first manner.

Jumps over nodes with noop attribute (does not yield them back).

taskflow.engines.action_engine.traversal.depth_first_reverse_iterate(node, start_from_idx=-1)[source]¶

Iterates connected (in reverse) tree nodes (from starting node).

Jumps through nodes with noop attribute (does not yield them back).