Parameters

Variables

priority = 0

A numeric priority that instances of this class will have when running, used when there are multiple parallel candidates to execute and/or revert. During this situation the candidate list will be stably sorted based on this priority attribute which will result in atoms with higher priorities executing (or reverting) before atoms with lower priorities (higher being defined as a number bigger, or greater tha an atom with a lower priority number). By default all atoms have the same priority (zero).

For example when the following is combined into a graph (where each node in the denoted graph is some task):

a -> b
b -> c
b -> e
b -> f

When b finishes there will then be three candidates that can run (c, e, f) and they may run in any order. What this priority does is sort those three by their priority before submitting them to be worked on (so that instead of say a random run order they will now be ran by there sorted order). This is also true when reverting (in that the sort order of the potential nodes will be used to determine the submission order).

requires

A OrderedSet of inputs this atom requires to function.

pre_execute()

Code to be run prior to executing the atom.

A common pattern for initializing the state of the system prior to running atoms is to define some code in a base class that all your atoms inherit from. In that class, you can define a pre_execute method and it will always be invoked just prior to your atoms running.

abstract execute(*args, **kwargs)

Activate a given atom which will perform some operation and return.

This method can be used to perform an action on a given set of input requirements (passed in via *args and **kwargs) to accomplish some type of operation. This operation may provide some named outputs/results as a result of it executing for later reverting (or for other atoms to depend on).

NOTE(harlowja): the result (if any) that is returned should be persistable so that it can be passed back into this atom if reverting is triggered (especially in the case where reverting happens in a different python process or on a remote machine) and so that the result can be transmitted to other atoms (which may be local or remote).

post_execute()

Code to be run after executing the atom.

A common pattern for cleaning up global state of the system after the execution of atoms is to define some code in a base class that all your atoms inherit from. In that class, you can define a post_execute method and it will always be invoked just after your atoms execute, regardless of whether they succeeded or not.

This pattern is useful if you have global shared database sessions that need to be cleaned up, for example.

pre_revert()

Code to be run prior to reverting the atom.

This works the same as pre_execute(), but for the revert phase.

revert(*args, **kwargs)

Revert this atom.

This method should undo any side-effects caused by previous execution of the atom using the result of the execute() method and information on the failure which triggered reversion of the flow the atom is contained in (if applicable).

post_revert()

Code to be run after reverting the atom.

This works the same as post_execute(), but for the revert phase.

REVERT = 'REVERT'

Reverts only the surrounding/associated subflow.

This strategy first consults the parent atom before reverting the associated subflow to determine if the parent retry object provides a different reconciliation strategy. This allows for safe nesting of flows with different retry strategies.

If the parent flow has no retry strategy, the default behavior is to just revert the atoms in the associated subflow. This is generally not the desired behavior, but is left as the default in order to keep backwards-compatibility. The defer_reverts engine option will let you change this behavior. If that is set to True, a REVERT will always defer to the parent, meaning that if the parent has no retry strategy, it will be reverted as well.

REVERT_ALL = 'REVERT_ALL'

Reverts the entire flow, regardless of parent strategy.

This strategy will revert every atom that has executed thus far, regardless of whether the parent flow has a separate retry strategy associated with it.

RETRY = 'RETRY'

Retries the surrounding/associated subflow again.

>>> class EchoTask(task.Task):
...     def execute(self, *args, **kwargs):
...         print(self.name)
...         print(args)
...         print(kwargs)
...
>>> flow = linear_flow.Flow('f1').add(
...     EchoTask('t1'),
...     linear_flow.Flow('f2', retry=retry.ForEach(values=['a', 'b', 'c'], name='r1', provides='value')).add(
...         EchoTask('t2'),
...         EchoTask('t3', requires='value')),
...     EchoTask('t4'))

>>> class SendMessage(task.Task):
...     def execute(self, message):
...         print("Sending message: %s" % message)
...
>>> flow = linear_flow.Flow('send_message', retry=retry.Times(5)).add(
...      SendMessage('sender'))

>>> class ConnectToServer(task.Task):
...     def execute(self, ip):
...         print("Connecting to %s" % ip)
...
>>> server_ips = ['192.168.1.1', '192.168.1.2', '192.168.1.3' ]
>>> flow = linear_flow.Flow('send_message',
...                         retry=retry.ParameterizedForEach(rebind={'values': 'server_ips'},
...                                                          provides='ip')).add(
...     ConnectToServer(requires=['ip']))

property notifier

Internal notification dispatcher/registry.

A notification object that will dispatch events that occur related to internal notifications that the task internally emits to listeners (for example for progress status updates, telling others that a task has reached 50% completion...).

copy(retain_listeners=True)

Clone/copy this task.

update_progress(progress)

Update task progress and notify all registered listeners.

execute(*args, **kwargs)

Activate a given atom which will perform some operation and return.

This method can be used to perform an action on a given set of input requirements (passed in via *args and **kwargs) to accomplish some type of operation. This operation may provide some named outputs/results as a result of it executing for later reverting (or for other atoms to depend on).

NOTE(harlowja): the result (if any) that is returned should be persistable so that it can be passed back into this atom if reverting is triggered (especially in the case where reverting happens in a different python process or on a remote machine) and so that the result can be transmitted to other atoms (which may be local or remote).

revert(*args, **kwargs)

Revert this atom.

This method should undo any side-effects caused by previous execution of the atom using the result of the execute() method and information on the failure which triggered reversion of the flow the atom is contained in (if applicable).

execute(*args, **kwargs)

Activate a given atom which will perform some operation and return.

This method can be used to perform an action on a given set of input requirements (passed in via *args and **kwargs) to accomplish some type of operation. This operation may provide some named outputs/results as a result of it executing for later reverting (or for other atoms to depend on).

NOTE(harlowja): the result (if any) that is returned should be persistable so that it can be passed back into this atom if reverting is triggered (especially in the case where reverting happens in a different python process or on a remote machine) and so that the result can be transmitted to other atoms (which may be local or remote).

execute(*args, **kwargs)

Activate a given atom which will perform some operation and return.

This method can be used to perform an action on a given set of input requirements (passed in via *args and **kwargs) to accomplish some type of operation. This operation may provide some named outputs/results as a result of it executing for later reverting (or for other atoms to depend on).

NOTE(harlowja): the result (if any) that is returned should be persistable so that it can be passed back into this atom if reverting is triggered (especially in the case where reverting happens in a different python process or on a remote machine) and so that the result can be transmitted to other atoms (which may be local or remote).

abstract execute(history, *args, **kwargs)

Executes the given retry.

This execution activates a given retry which will typically produce data required to start or restart a connected component using previously provided values and a history of prior failures from previous runs. The historical data can be analyzed to alter the resolution strategy that this retry controller will use.

For example, a retry can provide the same values multiple times (after each run), the latest value or some other variation. Old values will be saved to the history of the retry atom automatically, that is a list of tuples (result, failures) are persisted where failures is a dictionary of failures indexed by task names and the result is the execution result returned by this retry during that failure resolution attempt.

revert(history, *args, **kwargs)

Reverts this retry.

On revert call all results that had been provided by previous tries and all errors caused during reversion are provided. This method will be called only if a subflow must be reverted without the retry (that is to say that the controller has ran out of resolution options and has either given up resolution or has failed to handle a execution failure).

abstract on_failure(history, *args, **kwargs)

Makes a decision about the future.

This method will typically use information about prior failures (if this historical failure information is not available or was not persisted the provided history will be empty).

Returns a retry constant (one of):

property failure

Returns the retries own failure or none if not existent.

outcomes_iter(index=None)

Iterates over the contained failure outcomes.

If the index is not provided, then all outcomes are iterated over.

NOTE(harlowja): if the retry itself failed, this will not include those types of failures. Use the failure attribute to access that instead (if it exists, aka, non-none).

provided_iter()

Iterates over all the values the retry has attempted (in order).

caused_by(exception_cls, index=None, include_retry=False)

Checks if the exception class provided caused the failures.

If the index is not provided, then all outcomes are iterated over.

on_failure(*args, **kwargs)

Makes a decision about the future.

This method will typically use information about prior failures (if this historical failure information is not available or was not persisted the provided history will be empty).

Returns a retry constant (one of):

execute(*args, **kwargs)

Executes the given retry.

This execution activates a given retry which will typically produce data required to start or restart a connected component using previously provided values and a history of prior failures from previous runs. The historical data can be analyzed to alter the resolution strategy that this retry controller will use.

For example, a retry can provide the same values multiple times (after each run), the latest value or some other variation. Old values will be saved to the history of the retry atom automatically, that is a list of tuples (result, failures) are persisted where failures is a dictionary of failures indexed by task names and the result is the execution result returned by this retry during that failure resolution attempt.

on_failure(**kwargs)

Makes a decision about the future.

This method will typically use information about prior failures (if this historical failure information is not available or was not persisted the provided history will be empty).

Returns a retry constant (one of):

execute(**kwargs)

Executes the given retry.

This execution activates a given retry which will typically produce data required to start or restart a connected component using previously provided values and a history of prior failures from previous runs. The historical data can be analyzed to alter the resolution strategy that this retry controller will use.

For example, a retry can provide the same values multiple times (after each run), the latest value or some other variation. Old values will be saved to the history of the retry atom automatically, that is a list of tuples (result, failures) are persisted where failures is a dictionary of failures indexed by task names and the result is the execution result returned by this retry during that failure resolution attempt.

Parameters

on_failure(history, *args, **kwargs)

Makes a decision about the future.

This method will typically use information about prior failures (if this historical failure information is not available or was not persisted the provided history will be empty).

Returns a retry constant (one of):

execute(history, *args, **kwargs)

Executes the given retry.

This execution activates a given retry which will typically produce data required to start or restart a connected component using previously provided values and a history of prior failures from previous runs. The historical data can be analyzed to alter the resolution strategy that this retry controller will use.

For example, a retry can provide the same values multiple times (after each run), the latest value or some other variation. Old values will be saved to the history of the retry atom automatically, that is a list of tuples (result, failures) are persisted where failures is a dictionary of failures indexed by task names and the result is the execution result returned by this retry during that failure resolution attempt.

Parameters

on_failure(history, *args, **kwargs)

Makes a decision about the future.

This method will typically use information about prior failures (if this historical failure information is not available or was not persisted the provided history will be empty).

Returns a retry constant (one of):

execute(history, *args, **kwargs)

Executes the given retry.

This execution activates a given retry which will typically produce data required to start or restart a connected component using previously provided values and a history of prior failures from previous runs. The historical data can be analyzed to alter the resolution strategy that this retry controller will use.

For example, a retry can provide the same values multiple times (after each run), the latest value or some other variation. Old values will be saved to the history of the retry atom automatically, that is a list of tuples (result, failures) are persisted where failures is a dictionary of failures indexed by task names and the result is the execution result returned by this retry during that failure resolution attempt.

Parameters

revert_all (bool) -- when provided this will cause the full flow to revert when the number of attempts that have been tried has been reached (when false, it will only locally revert the associated subflow)

on_failure(values, history, *args, **kwargs)

Makes a decision about the future.

This method will typically use information about prior failures (if this historical failure information is not available or was not persisted the provided history will be empty).

Returns a retry constant (one of):

execute(values, history, *args, **kwargs)

Executes the given retry.

This execution activates a given retry which will typically produce data required to start or restart a connected component using previously provided values and a history of prior failures from previous runs. The historical data can be analyzed to alter the resolution strategy that this retry controller will use.

For example, a retry can provide the same values multiple times (after each run), the latest value or some other variation. Old values will be saved to the history of the retry atom automatically, that is a list of tuples (result, failures) are persisted where failures is a dictionary of failures indexed by task names and the result is the execution result returned by this retry during that failure resolution attempt.

>>> class MyTask(task.Task):
...     def execute(self, spam, eggs, bacon=None):
...         return spam + eggs
...
>>> sorted(MyTask().requires)
['eggs', 'spam']
>>> sorted(MyTask().optional)
['bacon']

>>> class UniTask(task.Task):
...     def execute(self, *args, **kwargs):
...         pass
...
>>> sorted(UniTask().requires)
[]

class SpawnVMTask(task.Task):


    def execute(self, vm_name, vm_image_id, **kwargs):


        pass  # TODO(imelnikov): use parameters to spawn vm

SpawnVMTask(rebind={'vm_name': 'name'})

SpawnVMTask(rebind_args=('name', 'vm_image_id'))

SpawnVMTask(rebind=dict(vm_name='name',


                        vm_image_id='vm_image_id'))

SpawnVMTask(rebind=('name', 'vm_image_id', 'admin_key_name'))

>>> class Cat(task.Task):
...     def __init__(self, **kwargs):
...         if 'requires' not in kwargs:
...             kwargs['requires'] = ("food", "milk")
...         super(Cat, self).__init__(**kwargs)
...     def execute(self, food, **kwargs):
...         pass
...
>>> cat = Cat()
>>> sorted(cat.requires)
['food', 'milk']

>>> class Dog(task.Task):
...     def execute(self, food, **kwargs):
...         pass
>>> dog = Dog(requires=("water", "grass"))
>>> sorted(dog.requires)
['food', 'grass', 'water']

>>> class Bird(task.Task):
...     def execute(self, food, **kwargs):
...         pass
>>> bird = Bird(requires=("food", "water", "grass"), auto_extract=False)
>>> sorted(bird.requires)
['food', 'grass', 'water']

>>> class TheAnswerReturningTask(task.Task):
...    def execute(self):
...        return 42
...
>>> sorted(TheAnswerReturningTask(provides='the_answer').provides)
['the_answer']

class BitsAndPiecesTask(task.Task):


    def execute(self):


        return 'BITs', 'PIECEs'

BitsAndPiecesTask(provides=('bits', 'pieces'))

>>> storage.fetch('bits')
'BITs'
>>> storage.fetch('pieces')
'PIECEs'

class BitsAndPiecesTask(task.Task):


    def execute(self):


        return {


            'bits': 'BITs',


            'pieces': 'PIECEs'


        }

BitsAndPiecesTask(provides=set(['bits', 'pieces']))

>>> storage.fetch('bits')
'BITs'
>>> storage.fetch('pieces')
'PIECEs'

class BitsAndPiecesTask(task.Task):


    default_provides = ('bits', 'pieces')


    def execute(self):


        return 'BITs', 'PIECEs'

BitsAndPiecesTask(provides=('b', 'p'))

BitsAndPiecesTask(provides='bnp')

BitsAndPiecesTask(provides=())

from taskflow.types import failure
class RevertingTask(task.Task):


    def execute(self, spam, eggs):


        return do_something(spam, eggs)


    def revert(self, result, spam, eggs):


        if isinstance(result, failure.Failure):


            print("This task failed, exception: %s"


                  % result.exception_str)


        else:


            print("do_something returned %r" % result)

class MyRetry(retry.Retry):


    default_provides = 'value'


    def on_failure(self, history, *args, **kwargs):


        print(list(history))


        return RETRY


    def execute(self, history, *args, **kwargs):


        print(list(history))


        return 5


    def revert(self, history, *args, **kwargs):


        print(list(history))

[('5', {'A': failure.Failure()})]

>>> class MyTask(task.Task):
...     def execute(self, **kwargs):
...         return 1, 2
...
>>> flow = linear_flow.Flow('test').add(
...     MyTask(requires='a', provides=('b', 'c')),
...     MyTask(requires='b', provides='d')
... )
>>> flow.requires
frozenset(['a'])
>>> sorted(flow.provides)
['b', 'c', 'd']

>>> class CatTalk(task.Task):
...   def execute(self, meow):
...     print meow
...     return "cat"
...
>>> class DogTalk(task.Task):
...   def execute(self, woof):
...     print woof
...     return "dog"
...
>>> flo = linear_flow.Flow("cat-dog")
>>> flo.add(CatTalk(), DogTalk(provides="dog"))
<taskflow.patterns.linear_flow.Flow object at 0x...>
>>> engines.run(flo)
Traceback (most recent call last):


   ...
taskflow.exceptions.MissingDependencies: 'linear_flow.Flow: cat-dog(len=2)' requires ['meow', 'woof'] but no other entity produces said requirements


 MissingDependencies: 'execute' method on '__main__.DogTalk==1.0' requires ['woof'] but no other entity produces said requirements


 MissingDependencies: 'execute' method on '__main__.CatTalk==1.0' requires ['meow'] but no other entity produces said requirements

>>> class CatTalk(task.Task):
...   def execute(self, meow):
...     print meow
...     return "cat"
...
>>> class DogTalk(task.Task):
...   def execute(self, woof):
...     print woof
...     return "dog"
...
>>> flo = linear_flow.Flow("cat-dog")
>>> flo.add(CatTalk(), DogTalk(provides="dog"))
<taskflow.patterns.linear_flow.Flow object at 0x...>
>>> result = engines.run(flo, store={'meow': 'meow', 'woof': 'woof'})
meow
woof
>>> pprint(result)
{'dog': 'dog', 'meow': 'meow', 'woof': 'woof'}

>>> flo = linear_flow.Flow("cat-dog")
>>> flo.add(CatTalk(), DogTalk(provides="dog"))
<taskflow.patterns.linear_flow.Flow object at 0x...>
>>> eng = engines.load(flo, store={'meow': 'meow'})
>>> eng.storage.inject({"woof": "bark"})
>>> eng.run()
meow
bark

>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> eng.run()
meow
woof
>>> pprint(eng.storage.fetch_all())
{'dog': 'dog', 'meow': 'meow', 'woof': 'woof'}
>>> print(eng.storage.fetch("dog"))
dog

property name

A non-unique name for this flow (human readable).

property retry

The associated flow retry controller.

This retry controller object will affect & control how (and if) this flow and its contained components retry when execution is underway and a failure occurs.

abstract add(*items)

Adds a given item/items to this flow.

abstract iter_links()

Iterates over dependency links between children of the flow.

abstract iter_nodes()

Iterate over nodes of the flow.

property provides

Set of symbol names provided by the flow.

abstract property requires

Set of unsatisfied symbol names required by the flow.

add(*items)

Adds a given task/tasks/flow/flows to this flow.

property requires

Set of unsatisfied symbol names required by the flow.

iter_nodes()

Iterate over nodes of the flow.

iter_links()

Iterates over dependency links between children of the flow.

add(*items)

Adds a given task/tasks/flow/flows to this flow.

iter_links()

Iterates over dependency links between children of the flow.

iter_nodes()

Iterate over nodes of the flow.

property requires

Set of unsatisfied symbol names required by the flow.

link(u, v, decider=None, decider_depth=None)

Link existing node u as a runtime dependency of existing node v.

Note that if the addition of these edges creates a cyclic graph then a DependencyFailure will be raised and the provided changes will be discarded. If the nodes that are being requested to link do not exist in this graph than a ValueError will be raised.

add(*nodes, **kwargs)

Adds a given task/tasks/flow/flows to this flow.

Note that if the addition of these nodes (and any edges) creates a cyclic graph then a DependencyFailure will be raised and the applied changes will be discarded.

iter_links()

Iterates over dependency links between children of the flow.

iter_nodes()

Iterate over nodes of the flow.

property requires

Set of unsatisfied symbol names required by the flow.

set_target(target_node)

Set target for the flow.

Any node(s) (tasks or subflows) not needed for the target node will not be executed.

reset_target()

Reset target for the flow.

All node(s) of the flow will be executed.

add(*nodes, **kwargs)

Adds a given task/tasks/flow/flows to this flow.

Note that if the addition of these nodes (and any edges) creates a cyclic graph then a DependencyFailure will be raised and the applied changes will be discarded.

link(u, v, decider=None, decider_depth=None)

Link existing node u as a runtime dependency of existing node v.

Note that if the addition of these edges creates a cyclic graph then a DependencyFailure will be raised and the provided changes will be discarded. If the nodes that are being requested to link do not exist in this graph than a ValueError will be raised.

ALL = 'ALL'

Default decider depth that affects all successor atoms (including ones that are in successor nested flows).

FLOW = 'FLOW'

Decider depth that affects all successor tasks in the same flow (it will not affect tasks/retries that are in successor nested flows).

WARNING:

NEIGHBORS = 'NEIGHBORS'

Decider depth that affects only next successor tasks (and does not traverse past one level of successor tasks).

WARNING:

ATOM = 'ATOM'

Decider depth that affects only targeted atom (and does not traverse into any level of successor atoms).

WARNING:

classmethod translate(desired_depth)

Translates a string into a depth enumeration.

•

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).

from taskflow import engines
...
flow = make_flow()
eng = engines.load(flow, engine='serial', backend=my_persistence_conf)
eng.run()
...

Parameters

Returns

engine

Returns

dictionary of all named results (see fetch_all())

Parameters

Parameters

Returns

engine

Parameters

flow_detail -- FlowDetail that holds state of the flow to load

Parameters

flow_detail -- FlowDetail that holds state of the flow to load

Returns

engine

Variables

property notifier

The flow notifier.

property atom_notifier

The atom notifier.

property options

The options that were passed to this engine on construction.

abstract property storage

The storage unit for this engine.

abstract property statistics

A dictionary of runtime statistics this engine has gathered.

This dictionary will be empty when the engine has never been ran. When it is running or has ran previously it should have (but may not) have useful and/or informational keys and values when running is underway and/or completed.

WARNING:

abstract compile()

Compiles the contained flow into a internal representation.

This internal representation is what the engine will actually use to run. If this compilation can not be accomplished then an exception is expected to be thrown with a message indicating why the compilation could not be achieved.

abstract reset()

Reset back to the PENDING state.

If a flow had previously ended up (from a prior engine run()) in the FAILURE, SUCCESS or REVERTED states (or for some reason it ended up in an intermediary state) it can be desirable to make it possible to run it again. Calling this method enables that to occur (without causing a state transition failure, which would typically occur if run() is called directly without doing a reset).

abstract prepare()

Performs any pre-run, but post-compilation actions.

NOTE(harlowja): During preparation it is currently assumed that the underlying storage will be initialized, the atoms will be reset and the engine will enter the PENDING state.

abstract validate()

Performs any pre-run, post-prepare validation actions.

NOTE(harlowja): During validation all final dependencies will be verified and ensured. This will by default check that all atoms have satisfiable requirements (satisfied by some other provider).

abstract run()

Runs the flow in the engine to completion (or die trying).

abstract suspend()

Attempts to suspend the engine.

If the engine is currently running atoms then this will attempt to suspend future work from being started (currently active atoms can not currently be preempted) and move the engine into a suspend state which can then later be resumed from.

NO_RERAISING_STATES = frozenset({'SUCCESS', 'SUSPENDED'})

States that if the engine stops in will not cause any potential failures to be reraised. States not in this list will cause any failure/s that were captured (if any) to get reraised.

IGNORABLE_STATES = frozenset({'ANALYZING', 'GAME_OVER', 'RESUMING', 'SCHEDULING', 'UNDEFINED', 'WAITING'})

Informational states this engines internal machine yields back while running, not useful to have the engine record but useful to provide to end-users when doing execution iterations via run_iter().

MAX_MACHINE_STATES_RETAINED = 10

During run_iter() the last X state machine transitions will be recorded (typically only useful on failure).

suspend()

Attempts to suspend the engine.

If the engine is currently running atoms then this will attempt to suspend future work from being started (currently active atoms can not currently be preempted) and move the engine into a suspend state which can then later be resumed from.

property statistics

A dictionary of runtime statistics this engine has gathered.

This dictionary will be empty when the engine has never been ran. When it is running or has ran previously it should have (but may not) have useful and/or informational keys and values when running is underway and/or completed.

WARNING:

property compilation

The compilation result.

NOTE(harlowja): Only accessible after compilation has completed (None will be returned when this property is accessed before compilation has completed successfully).

storage

The storage unit for this engine.

NOTE(harlowja): the atom argument lookup strategy will change for this storage unit after compile() has completed (since only after compilation is the actual structure known). Before compile() has completed the atom argument lookup strategy lookup will be restricted to injected arguments only (this will not reflect the actual runtime lookup strategy, which typically will be, but is not always different).

run(timeout=None)

Runs the engine (or die trying).

run_iter(timeout=None)

Runs the engine using iteration (or die trying).

Instead of running to completion in a blocking manner, this will return a generator which will yield back the various states that the engine is going through (and can be used to run multiple engines at once using a generator per engine). The iterator returned also responds to the send() method from PEP 0342 and will attempt to suspend itself if a truthy value is sent in (the suspend may be delayed until all active atoms have finished).

NOTE(harlowja): using the run_iter method will not retain the engine lock while executing so the user should ensure that there is only one entity using a returned engine iterator (one per engine) at a given time.

validate()

Performs any pre-run, post-prepare validation actions.

NOTE(harlowja): During validation all final dependencies will be verified and ensured. This will by default check that all atoms have satisfiable requirements (satisfied by some other provider).

prepare()

Performs any pre-run, but post-compilation actions.

NOTE(harlowja): During preparation it is currently assumed that the underlying storage will be initialized, the atoms will be reset and the engine will enter the PENDING state.

reset()

Reset back to the PENDING state.

If a flow had previously ended up (from a prior engine run()) in the FAILURE, SUCCESS or REVERTED states (or for some reason it ended up in an intermediary state) it can be desirable to make it possible to run it again. Calling this method enables that to occur (without causing a state transition failure, which would typically occur if run() is called directly without doing a reset).

compile()

Compiles the contained flow into a internal representation.

This internal representation is what the engine will actually use to run. If this compilation can not be accomplished then an exception is expected to be thrown with a message indicating why the compilation could not be achieved.

cancel_futures()

Attempts to cancel any not done futures.

build(statistics, timeout=None, gather_statistics=True)

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

property flow

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

property name

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

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).

property execution_graph

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

property hierarchy

The hierarchy of patterns (as a tree structure).

compile(flow, parent=None)

Decomposes a flow into a graph and scope tree hierarchy.

compile()

Compiles the contained item into a compiled equivalent.

abstract apply()

Applies some algorithm to resolve some detected failure.

apply()

Applies some algorithm to resolve some detected failure.

apply()

Applies some algorithm to resolve some detected failure.

apply()

Applies some algorithm to resolve some detected failure.

resume()

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)

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)

Performs post-execution completion of a node result.

abstract tally(runtime)

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).

abstract affect(runtime, nay_voters)

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)

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).

tally(runtime)

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)

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.

tally(runtime)

Always good to go.

affect(runtime, nay_voters)

Does nothing.

start()

Prepare to execute retries.

stop()

Finalize retry executor.

execute_retry(retry, arguments)

Schedules retry execution.

revert_retry(retry, arguments)

Schedules retry reversion.

abstract execute_task(task, task_uuid, arguments, progress_callback=None)

Schedules task execution.

abstract revert_task(task, task_uuid, arguments, result, failures, progress_callback=None)

Schedules task reversion.

start()

Prepare to execute tasks.

stop()

Finalize task executor.

start()

Prepare to execute tasks.

stop()

Finalize task executor.

execute_task(task, task_uuid, arguments, progress_callback=None)

Schedules task execution.

revert_task(task, task_uuid, arguments, result, failures, progress_callback=None)

Schedules task reversion.

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)

Schedules task execution.

revert_task(task, task_uuid, arguments, result, failures, progress_callback=None)

Schedules task reversion.

start()

Prepare to execute tasks.

stop()

Finalize task 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).

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

CHUNK_SIZE = 8192

Read/write chunk size.

MAX_BACKLOG = 5

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

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()

Prepare to execute tasks.

stop()

Finalize task executor.

compile()

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)

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

fetch_edge_deciders(atom)

Fetches the edge deciders for the given atom.

fetch_scheduler(atom)

Fetches the cached specific scheduler for the given atom.

fetch_action(atom)

Fetches the cached action handler for the given atom.

fetch_scopes_for(atom_name)

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

iterate_retries(state=None)

Iterates retry atoms that match the provided state.

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

iterate_nodes(allowed_kinds)

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

is_success()

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

find_retry(node)

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

reset_atoms(atoms, state='PENDING', intention='EXECUTE')

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

reset_all(state='PENDING', intention='EXECUTE')

Resets all atoms to the given state and intention.

reset_subgraph(atom, state='PENDING', intention='EXECUTE')

Resets a atoms subgraph to the given state and intention.

The subgraph is contained of all of the atoms successors.

retry_subflow(retry)

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.

schedule(retry)

Schedules the given retry atom for future completion.

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

schedule(task)

Schedules the given task atom for future completion.

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

schedule(atoms)

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.

iter_next_atoms(atom=None)

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

__iter__()

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.

FORWARD = 1

Go through successors.

BACKWARD = 2

Go through predecessors.

1.

The worker dispatches the request (gets desired endpoint that actually executes the task).

2.

If dispatched succeeded then the worker sends a confirmation response to the executor otherwise the worker sends a failed response along with a serialized failure object that contains what has failed (and why).

3.

The worker executes the task and once it is finished sends the result back to the originating executor (every time a task progress event is triggered it sends progress notification to the executor where it is handled by the engine, dispatching to listeners and so-on).

abstract to_dict()

Return json-serializable message representation.

TYPE = 'NOTIFY'

String constant representing this message type.

RESPONSE_SCHEMA = {'additionalProperties': False, 'properties': {'tasks': {'items': {'type': 'string'}, 'type': 'array'}, 'topic': {'type': 'string'}}, 'required': ['topic', 'tasks'], 'type': 'object'}

Expected notify response message schema (in json schema format).

SENDER_SCHEMA = {'additionalProperties': False, 'type': 'object'}

Expected sender request message schema (in json schema format).

to_dict()

Return json-serializable message representation.

TYPE = 'REQUEST'

String constant representing this message type.

SCHEMA = {'properties': {'action': {'enum': ['execute', 'revert'], 'type': 'string'}, 'arguments': {'type': 'object'}, 'failures': {'type': 'object'}, 'result': {}, 'task_cls': {'type': 'string'}, 'task_name': {'type': 'string'}, 'task_version': {'oneOf': [{'type': 'string'}, {'type': 'array'}]}}, 'required': ['task_cls', 'task_name', 'task_version', 'action'], 'type': 'object'}

Expected message schema (in json schema format).

property current_state

Current state the request is in.

set_result(result)

Sets the responses futures result.

property expired

Check if request has expired.

When new request is created its state is set to the WAITING, creation time is stored and timeout is given via constructor arguments.

Request is considered to be expired when it is in the WAITING/PENDING state for more then the given timeout (it is not considered to be expired in any other state).

to_dict()

Return json-serializable request.

To convert requests that have failed due to some exception this will convert all failure.Failure objects into dictionaries (which will then be reconstituted by the receiver).

transition_and_log_error(new_state, logger=None)

Transitions and logs an error if that transitioning raises.

This overlays the transition function and performs nearly the same functionality but instead of raising if the transition was not valid it logs a warning to the provided logger and returns False to indicate that the transition was not performed (note that this is different from the transition function where False means ignored).

transition(new_state)

Transitions the request to a new state.

If transition was performed, it returns True. If transition was ignored, it returns False. If transition was not valid (and will not be performed), it raises an InvalidState exception.

static from_dict(data, task_uuid=None)

Parses validated data into a work unit.

All Failure objects that have been converted to dict(s) on the remote side will now converted back to py:class:~taskflow.types.failure.Failure objects.

TYPE = 'RESPONSE'

String constant representing this message type.

SCHEMA = {'additionalProperties': False, 'definitions': {'completion': {'additionalProperties': False, 'properties': {'result': {}}, 'required': ['result'], 'type': 'object'}, 'empty': {'additionalProperties': False, 'type': 'object'}, 'event': {'additionalProperties': False, 'properties': {'details': {'type': 'object'}, 'event_type': {'type': 'string'}}, 'required': ['event_type', 'details'], 'type': 'object'}}, 'properties': {'data': {'anyOf': [{'$ref': '#/definitions/event'}, {'$ref': '#/definitions/completion'}, {'$ref': '#/definitions/empty'}]}, 'state': {'enum': ['WAITING', 'PENDING', 'RUNNING', 'SUCCESS', 'FAILURE', 'EVENT'], 'type': 'string'}}, 'required': ['state', 'data'], 'type': 'object'}

Expected message schema (in json schema format).

to_dict()

Return json-serializable message representation.

{


    "action": "execute",


    "arguments": {


        "x": 111


    },


    "task_cls": "taskflow.tests.utils.TaskOneArgOneReturn",


    "task_name": "taskflow.tests.utils.TaskOneArgOneReturn",


    "task_version": [


        1,


        0


    ]
}

{


    "action": "revert",


    "arguments": {},


    "failures": {


        "taskflow.tests.utils.TaskWithFailure": {


            "exc_type_names": [


                "RuntimeError",


                "StandardError",


                "Exception"


            ],


            "exception_str": "Woot!",


            "traceback_str": "  File \"/homes/harlowja/dev/os/taskflow/taskflow/engines/action_engine/executor.py\", line 56, in _execute_task\n    result = task.execute(**arguments)\n  File \"/homes/harlowja/dev/os/taskflow/taskflow/tests/utils.py\", line 165, in execute\n    raise RuntimeError('Woot!')\n",


            "version": 1


        }


    },


    "result": [


        "failure",


        {


            "exc_type_names": [


                "RuntimeError",


                "StandardError",


                "Exception"


            ],


            "exception_str": "Woot!",


            "traceback_str": "  File \"/homes/harlowja/dev/os/taskflow/taskflow/engines/action_engine/executor.py\", line 56, in _execute_task\n    result = task.execute(**arguments)\n  File \"/homes/harlowja/dev/os/taskflow/taskflow/tests/utils.py\", line 165, in execute\n    raise RuntimeError('Woot!')\n",


            "version": 1


        }


    ],


    "task_cls": "taskflow.tests.utils.TaskWithFailure",


    "task_name": "taskflow.tests.utils.TaskWithFailure",


    "task_version": [


        1,


        0


    ]
}

{


    "data": {},


    "state": "RUNNING"
}

{


    "details": {


        "progress": 0.5


    },


    "event_type": "update_progress",


    "state": "EVENT"
}

{


    "data": {


        "result": 666


    },


    "state": "SUCCESS"
}

{


    "data": {


        "result": {


            "exc_type_names": [


                "RuntimeError",


                "StandardError",


                "Exception"


            ],


            "exception_str": "Woot!",


            "traceback_str": "  File \"/homes/harlowja/dev/os/taskflow/taskflow/engines/action_engine/executor.py\", line 56, in _execute_task\n    result = task.execute(**arguments)\n  File \"/homes/harlowja/dev/os/taskflow/taskflow/tests/utils.py\", line 165, in execute\n    raise RuntimeError('Woot!')\n",


            "version": 1


        }


    },


    "state": "FAILURE"
}

from taskflow.engines.worker_based import worker as w
config = {


    'url': 'amqp://guest:guest@localhost:5672//',


    'exchange': 'test-exchange',


    'topic': 'test-tasks',


    'tasks': ['tasks:TestTask1', 'tasks:TestTask2'],
}
worker = w.Worker(**config)
worker.run()

flow = lf.Flow('simple-linear').add(...)
eng = taskflow.engines.load(flow, engine='worker-based',


                            url='amqp://guest:guest@localhost:5672//',


                            exchange='test-exchange',


                            topics=['topic1', 'topic2'])
eng.run()

flow = lf.Flow('simple-linear').add(...)
eng = taskflow.engines.load(flow, engine='worker-based',


                            exchange='test-exchange',


                            topics=['topic1', 'topic2'],


                            transport='filesystem',


                            transport_options={


                                'data_folder_in': '/tmp/in',


                                'data_folder_out': '/tmp/out',


                            })
eng.run()

property total_workers

Number of workers currently known.

wait_for_workers(workers=1, timeout=None)

Waits for geq workers to notify they are ready to do work.

NOTE(harlowja): if a timeout is provided this function will wait until that timeout expires, if the amount of workers does not reach the desired amount of workers before the timeout expires then this will return how many workers are still needed, otherwise it will return zero.

property messages_processed

How many notify response messages have been processed.

maybe_publish()

Periodically called to publish notify message to each topic.

These messages (especially the responses) are how this find learns about workers and what tasks they can perform (so that we can then match workers to tasks to run).

process_response(data, message)

Process notify message sent from remote side.

clean()

Cleans out any dead/expired/not responding workers.

Returns how many workers were removed.

reset()

Resets finders internal state.

get_worker_for_task(task)

Gets a worker that can perform a given task.

>>> class CatTalk(task.Task):
...   def execute(self, meow):
...     print(meow)
...     return "cat"
...
>>> class DogTalk(task.Task):
...   def execute(self, woof):
...     print(woof)
...     return 'dog'
...
>>> def flow_transition(state, details):
...     print("Flow '%s' transition to state %s" % (details['flow_name'], state))
...
>>>
>>> flo = linear_flow.Flow("cat-dog").add(
...   CatTalk(), DogTalk(provides="dog"))
>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> eng.notifier.register(ANY, flow_transition)
>>> eng.run()
Flow 'cat-dog' transition to state RUNNING
meow
woof
Flow 'cat-dog' transition to state SUCCESS

>>> class CatTalk(task.Task):
...   def execute(self, meow):
...     print(meow)
...     return "cat"
...
>>> class DogTalk(task.Task):
...   def execute(self, woof):
...     print(woof)
...     return 'dog'
...
>>> def task_transition(state, details):
...     print("Task '%s' transition to state %s" % (details['task_name'], state))
...
>>>
>>> flo = linear_flow.Flow("cat-dog")
>>> flo.add(CatTalk(), DogTalk(provides="dog"))
<taskflow.patterns.linear_flow.Flow object at 0x...>
>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> eng.atom_notifier.register(ANY, task_transition)
>>> eng.run()
Task 'CatTalk' transition to state RUNNING
meow
Task 'CatTalk' transition to state SUCCESS
Task 'DogTalk' transition to state RUNNING
woof
Task 'DogTalk' transition to state SUCCESS

>>> from taskflow.listeners import printing
>>> class CatTalk(task.Task):
...   def execute(self, meow):
...     print(meow)
...     return "cat"
...
>>> class DogTalk(task.Task):
...   def execute(self, woof):
...     print(woof)
...     return 'dog'
...
>>>
>>> flo = linear_flow.Flow("cat-dog").add(
...   CatTalk(), DogTalk(provides="dog"))
>>> eng = engines.load(flo, store={'meow': 'meow', 'woof': 'woof'})
>>> with printing.PrintingListener(eng):
...   eng.run()
...
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved flow 'cat-dog' (...) into state 'RUNNING' from state 'PENDING'
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'CatTalk' (...) into state 'RUNNING' from state 'PENDING'
meow
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'CatTalk' (...) into state 'SUCCESS' from state 'RUNNING' with result 'cat' (failure=False)
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'DogTalk' (...) into state 'RUNNING' from state 'PENDING'
woof
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved task 'DogTalk' (...) into state 'SUCCESS' from state 'RUNNING' with result 'dog' (failure=False)
<taskflow.engines.action_engine.engine.SerialActionEngine object at ...> has moved flow 'cat-dog' (...) into state 'SUCCESS' from state 'RUNNING'

• states.FAILURE
• states.REVERTED

• states.FAILURE
• states.RETRYING
• states.REVERTING
• states.REVERT_FAILURE

Variables

values -- Captured transitions + details (the result of the _format_capture() method) are stored into this list (a previous list to append to may be provided using the constructor keyword argument of the same name); by default this stores tuples of the format (kind, state, details).

FLOW = 'flow'

Kind that denotes a 'flow' capture.

TASK = 'task'

Kind that denotes a 'task' capture.

RETRY = 'retry'

Kind that denotes a 'retry' capture.

format(fail, atom_matcher)

Returns a (exc_info, details) tuple about the failure.

The exc_info tuple should be a standard three element (exctype, value, traceback) tuple that will be used for further logging. A non-empty string is typically returned for details; it should contain any string info about the failure (with any specific details the exc_info may not have/contain).

• Timing information (how long a task took to run).
• User information (who the task ran as).
• When a atom/workflow was ran (and why).

from taskflow.persistence import backends
...
persistence = backends.fetch(conf={


    "connection": "mysql",


    "user": ...,


    "password": ...,
})
book = make_and_save_logbook(persistence)
...

abstract get_connection()

Return a Connection instance based on the configuration settings.

abstract close()

Closes any resources this backend has open.

abstract property backend

Returns the backend this connection is associated with.

abstract close()

Closes any resources this connection has open.

abstract upgrade()

Migrate the persistence backend to the most recent version.

abstract clear_all()

Clear all entries from this backend.

abstract validate()

Validates that a backend is still ok to be used.

The semantics of this may vary depending on the backend. On failure a backend specific exception should be raised that will indicate why the failure occurred.

abstract update_atom_details(atom_detail)

Updates a given atom details and returns the updated version.

NOTE(harlowja): the details that is to be updated must already have been created by saving a flow details with the given atom detail inside of it.

abstract update_flow_details(flow_detail)

Updates a given flow details and returns the updated version.

NOTE(harlowja): the details that is to be updated must already have been created by saving a logbook with the given flow detail inside of it.

abstract save_logbook(book)

Saves a logbook, and all its contained information.

abstract destroy_logbook(book_uuid)

Deletes/destroys a logbook matching the given uuid.

abstract get_logbook(book_uuid, lazy=False)

Fetches a logbook object matching the given uuid.

abstract get_logbooks(lazy=False)

Return an iterable of logbook objects.

abstract get_flows_for_book(book_uuid)

Return an iterable of flowdetails for a given logbook uuid.

abstract get_flow_details(fd_uuid, lazy=False)

Fetches a flowdetails object matching the given uuid.

abstract get_atom_details(ad_uuid)

Fetches a atomdetails object matching the given uuid.

abstract get_atoms_for_flow(fd_uuid)

Return an iterable of atomdetails for a given flowdetails uuid.

DEFAULT_PATH = None

Default path used when none is provided.

property backend

Returns the backend this connection is associated with.

get_logbooks(lazy=False)

Return an iterable of logbook objects.

get_logbook(book_uuid, lazy=False)

Fetches a logbook object matching the given uuid.

save_logbook(book)

Saves a logbook, and all its contained information.

get_flows_for_book(book_uuid, lazy=False)

Return an iterable of flowdetails for a given logbook uuid.

get_flow_details(flow_uuid, lazy=False)

Fetches a flowdetails object matching the given uuid.

update_flow_details(flow_detail, ignore_missing=False)

Updates a given flow details and returns the updated version.

NOTE(harlowja): the details that is to be updated must already have been created by saving a logbook with the given flow detail inside of it.

get_atoms_for_flow(flow_uuid)

Return an iterable of atomdetails for a given flowdetails uuid.

get_atom_details(atom_uuid)

Fetches a atomdetails object matching the given uuid.

update_atom_details(atom_detail, ignore_missing=False)

Updates a given atom details and returns the updated version.

NOTE(harlowja): the details that is to be updated must already have been created by saving a flow details with the given atom detail inside of it.

destroy_logbook(book_uuid)

Deletes/destroys a logbook matching the given uuid.

clear_all()

Clear all entries from this backend.

upgrade()

Migrate the persistence backend to the most recent version.

close()

Closes any resources this connection has open.

Variables

pformat(indent=0, linesep='\n')

Pretty formats this logbook into a string.

>>> from taskflow.persistence import models
>>> tmp = models.LogBook("example")
>>> print(tmp.pformat())
LogBook: 'example'


 - uuid = ...


 - created_at = ...
    

add(fd)

Adds a new flow detail into this logbook.

NOTE(harlowja): if an existing flow detail exists with the same uuid the existing one will be overwritten with the newly provided one.

Does not guarantee that the details will be immediately saved.

find(flow_uuid)

Locate the flow detail corresponding to the given uuid.

merge(lb, deep_copy=False)

Merges the current object state with the given ones state.

If deep_copy is provided as truthy then the local object will use copy.deepcopy to replace this objects local attributes with the provided objects attributes (only if there is a difference between this objects attributes and the provided attributes). If deep_copy is falsey (the default) then a reference copy will occur instead when a difference is detected.

NOTE(harlowja): If the provided object is this object itself then no merging is done. Also note that this does not merge the flow details contained in either.

to_dict(marshal_time=False)

Translates the internal state of this object to a dict.

NOTE(harlowja): The returned dict does not include any contained flow details.

classmethod from_dict(data, unmarshal_time=False)

Translates the given dict into an instance of this class.

NOTE(harlowja): the dict provided should come from a prior call to to_dict().

property uuid

The unique identifer of this logbook.

property name

The name of this logbook.

copy(retain_contents=True)

Copies this logbook.

Creates a shallow copy of this logbook. If this logbook contains flow details and retain_contents is truthy (the default) then the flow details container will be shallow copied (the flow details contained there-in will not be copied). If retain_contents is falsey then the copied logbook will have no contained flow details (but it will have the rest of the local objects attributes copied).

Variables

meta -- A dictionary of meta-data associated with this flow detail.

state

The state of the flow associated with this flow detail.

update(fd)

Updates the objects state to be the same as the given one.

This will assign the private and public attributes of the given flow detail directly to this object (replacing any existing attributes in this object; even if they are the same).

NOTE(harlowja): If the provided object is this object itself then no update is done.

pformat(indent=0, linesep='\n')

Pretty formats this flow detail into a string.

>>> from oslo_utils import uuidutils
>>> from taskflow.persistence import models
>>> flow_detail = models.FlowDetail("example",
...                                 uuid=uuidutils.generate_uuid())
>>> print(flow_detail.pformat())
FlowDetail: 'example'


 - uuid = ...


 - state = ...
    

merge(fd, deep_copy=False)

Merges the current object state with the given one's state.

If deep_copy is provided as truthy then the local object will use copy.deepcopy to replace this objects local attributes with the provided objects attributes (only if there is a difference between this objects attributes and the provided attributes). If deep_copy is falsey (the default) then a reference copy will occur instead when a difference is detected.

NOTE(harlowja): If the provided object is this object itself then no merging is done. Also this does not merge the atom details contained in either.

copy(retain_contents=True)

Copies this flow detail.

Creates a shallow copy of this flow detail. If this detail contains flow details and retain_contents is truthy (the default) then the atom details container will be shallow copied (the atom details contained there-in will not be copied). If retain_contents is falsey then the copied flow detail will have no contained atom details (but it will have the rest of the local objects attributes copied).

to_dict()

Translates the internal state of this object to a dict.

NOTE(harlowja): The returned dict does not include any contained atom details.

classmethod from_dict(data)

Translates the given dict into an instance of this class.

NOTE(harlowja): the dict provided should come from a prior call to to_dict().

add(ad)

Adds a new atom detail into this flow detail.

NOTE(harlowja): if an existing atom detail exists with the same uuid the existing one will be overwritten with the newly provided one.

Does not guarantee that the details will be immediately saved.

find(ad_uuid)

Locate the atom detail corresponding to the given uuid.

property uuid

The unique identifer of this flow detail.

property name

The name of this flow detail.

Variables

state

The state of the atom associated with this atom detail.

property last_results

Gets the atoms last result.

If the atom has produced many results (for example if it has been retried, reverted, executed and ...) this returns the last one of many results.

update(ad)

Updates the object's state to be the same as the given one.

This will assign the private and public attributes of the given atom detail directly to this object (replacing any existing attributes in this object; even if they are the same).

NOTE(harlowja): If the provided object is this object itself then no update is done.

abstract merge(other, deep_copy=False)

Merges the current object state with the given ones state.

If deep_copy is provided as truthy then the local object will use copy.deepcopy to replace this objects local attributes with the provided objects attributes (only if there is a difference between this objects attributes and the provided attributes). If deep_copy is falsey (the default) then a reference copy will occur instead when a difference is detected.

NOTE(harlowja): If the provided object is this object itself then no merging is done. Do note that no results are merged in this method. That operation must to be the responsibilty of subclasses to implement and override this abstract method and provide that merging themselves as they see fit.

abstract put(state, result)

Puts a result (acquired in the given state) into this detail.

to_dict()

Translates the internal state of this object to a dict.

classmethod from_dict(data)

Translates the given dict into an instance of this class.

NOTE(harlowja): the dict provided should come from a prior call to to_dict().

property uuid

The unique identifer of this atom detail.

property name

The name of this atom detail.

abstract reset(state)

Resets this atom detail and sets state attribute value.

abstract copy()

Copies this atom detail.

pformat(indent=0, linesep='\n')

Pretty formats this atom detail into a string.

reset(state)

Resets this task detail and sets state attribute value.

This sets any previously set results, failure, and revert_results attributes back to None and sets the state to the provided one, as well as setting this task details intention attribute to EXECUTE.

put(state, result)

Puts a result (acquired in the given state) into this detail.

Returns whether this object was modified (or whether it was not).

merge(other, deep_copy=False)

Merges the current task detail with the given one.

NOTE(harlowja): This merge does not copy and replace the results or revert_results if it differs. Instead the current objects results and revert_results attributes directly becomes (via assignment) the other objects attributes. Also note that if the provided object is this object itself then no merging is done.

See: https://bugs.launchpad.net/taskflow/+bug/1452978 for what happens if this is copied at a deeper level (for example by using copy.deepcopy or by using copy.copy).

copy()

Copies this task detail.

Creates a shallow copy of this task detail (any meta-data and version information that this object maintains is shallow copied via copy.copy).

NOTE(harlowja): This copy does not copy and replace the results or revert_results attribute if it differs. Instead the current objects results and revert_results attributes directly becomes (via assignment) the cloned objects attributes.

See: https://bugs.launchpad.net/taskflow/+bug/1452978 for what happens if this is copied at a deeper level (for example by using copy.deepcopy or by using copy.copy).

reset(state)

Resets this retry detail and sets state attribute value.

This sets any previously added results back to an empty list and resets the failure and revert_failure and revert_results attributes back to None and sets the state to the provided one, as well as setting this retry details intention attribute to EXECUTE.

copy()

Copies this retry detail.

Creates a shallow copy of this retry detail (any meta-data and version information that this object maintains is shallow copied via copy.copy).

NOTE(harlowja): This copy does not copy the incoming objects results or revert_results attributes. Instead this objects results attribute list is iterated over and a new list is constructed with each (data, failures) element in that list having its failures (a dictionary of each named Failure object that occured) copied but its data is left untouched. After this is done that new list becomes (via assignment) the cloned objects results attribute. The revert_results is directly assigned to the cloned objects revert_results attribute.

See: https://bugs.launchpad.net/taskflow/+bug/1452978 for what happens if the data in results is copied at a deeper level (for example by using copy.deepcopy or by using copy.copy).

property last_results

The last result that was produced.

property last_failures

The last failure dictionary that was produced.

NOTE(harlowja): This is not the same as the local failure attribute as the obtained failure dictionary in the results attribute (which is what this returns) is from associated atom failures (which is different from the directly related failure of the retry unit associated with this atom detail).

put(state, result)

Puts a result (acquired in the given state) into this detail.

Returns whether this object was modified (or whether it was not).

classmethod from_dict(data)

Translates the given dict into an instance of this class.

to_dict()

Translates the internal state of this object to a dict.

merge(other, deep_copy=False)

Merges the current retry detail with the given one.

NOTE(harlowja): This merge does not deep copy the incoming objects results attribute (if it differs). Instead the incoming objects results attribute list is always iterated over and a new list is constructed with each (data, failures) element in that list having its failures (a dictionary of each named Failure objects that occurred) copied but its data is left untouched. After this is done that new list becomes (via assignment) this objects results attribute. Also note that if the provided object is this object itself then no merging is done.

See: https://bugs.launchpad.net/taskflow/+bug/1452978 for what happens if the data in results is copied at a deeper level (for example by using copy.deepcopy or by using copy.copy).

root_path = '/'

Root path of the in-memory filesystem.

classmethod normpath(path)

Return a normalized absolutized version of the pathname path.

static split(p)

Split a pathname into a tuple of (head, tail).

static join(*pieces)

Join many path segments together.

ensure_path(path)

Ensure the path (and parents) exists.

get(path, default=None)

Fetch the value of given path (and return default if not found).

ls_r(path, absolute=False)

Return list of all children of the given path (recursively).

ls(path, absolute=False)

Return list of all children of the given path (not recursive).

clear()

Remove all nodes (except the root) from this filesystem.

delete(path, recursive=False)

Deletes a node (optionally its children) from this filesystem.

pformat()

Pretty format this in-memory filesystem.

symlink(src_path, dest_path)

Link the destionation path to the source path.

DEFAULT_PATH = '/'

Default path used when none is provided.

get_connection()

Return a Connection instance based on the configuration settings.

close()

Closes any resources this backend has open.

validate()

Validates that a backend is still ok to be used.

The semantics of this may vary depending on the backend. On failure a backend specific exception should be raised that will indicate why the failure occurred.

DEFAULT_FILE_ENCODING = 'utf-8'

Default encoding used when decoding or encoding files into or from text/unicode into binary or binary into text/unicode.

get_connection()

Return a Connection instance based on the configuration settings.

close()

Closes any resources this backend has open.

validate()

Validates that a backend is still ok to be used.

The semantics of this may vary depending on the backend. On failure a backend specific exception should be raised that will indicate why the failure occurred.

injector_name = '_TaskFlow_INJECTOR'

Injector task detail name.

This task detail is a special detail that will be automatically created and saved to store persistent injected values (name conflicts with it must be avoided) that are global to the flow being executed.

ensure_atoms(atoms)

Ensure there is an atomdetail for each of the given atoms.

Returns list of atomdetail uuids for each atom processed.

property lock

Reader/writer lock used to ensure multi-thread safety.

This does not protect against the same storage objects being used by multiple engines/users across multiple processes (or different machines); certain backends handle that situation better than others (for example by using sequence identifiers) and it's a ongoing work in progress to make that better).

ensure_atom(atom)

Ensure there is an atomdetail for the given atom.

Returns the uuid for the atomdetail that corresponds to the given atom.

property flow_name

The flow detail name this storage unit is associated with.

property flow_uuid

The flow detail uuid this storage unit is associated with.

property flow_meta

The flow detail metadata this storage unit is associated with.

property backend

The backend this storage unit is associated with.

get_atom_uuid(atom_name)

Gets an atoms uuid given a atoms name.

set_atom_state(atom_name, state)

Sets an atoms state.

get_atom_state(atom_name)

Gets the state of an atom given an atoms name.

set_atom_intention(atom_name, intention)

Sets the intention of an atom given an atoms name.

get_atom_intention(atom_name)

Gets the intention of an atom given an atoms name.

get_atoms_states(atom_names)

Gets a dict of atom name => (state, intention) given atom names.

update_atom_metadata(atom_name, update_with)

Updates a atoms associated metadata.

This update will take a provided dictionary or a list of (key, value) pairs to include in the updated metadata (newer keys will overwrite older keys) and after merging saves the updated data into the underlying persistence layer.

set_task_progress(task_name, progress, details=None)

Set a tasks progress.

get_task_progress(task_name)

Get the progress of a task given a tasks name.

get_task_progress_details(task_name)

Get the progress details of a task given a tasks name.

save(atom_name, result, state='SUCCESS')

Put result for atom with provided name to storage.

save_retry_failure(retry_name, failed_atom_name, failure)

Save subflow failure to retry controller history.

cleanup_retry_history(retry_name, state)

Cleanup history of retry atom with given name.

get_execute_result(atom_name)

Gets the execute results for an atom from storage.

get_execute_failures()

Get all execute failures that happened with this flow.

get(atom_name)

Gets the execute results for an atom from storage.

get_failures()

Get all execute failures that happened with this flow.

get_revert_result(atom_name)

Gets the revert results for an atom from storage.

get_revert_failures()

Get all revert failures that happened with this flow.

has_failures()

Returns true if there are any failures in storage.

reset(atom_name, state='PENDING')

Reset atom with given name (if the atom is not in a given state).

inject_atom_args(atom_name, pairs, transient=True)

Add values into storage for a specific atom only.

This method injects a dictionary/pairs of arguments for an atom so that when that atom is scheduled for execution it will have immediate access to these arguments.

NOTE:

WARNING:

inject(pairs, transient=False)

Add values into storage.

This method should be used to put flow parameters (requirements that are not satisfied by any atom in the flow) into storage.

WARNING:

fetch(name, many_handler=None)

Fetch a named execute result.

fetch_unsatisfied_args(atom_name, args_mapping, scope_walker=None, optional_args=None)

Fetch unsatisfied execute arguments using an atoms args mapping.

NOTE(harlowja): this takes into account the provided scope walker atoms who should produce the required value at runtime, as well as the transient/persistent flow and atom specific injected arguments. It does not check if the providers actually have produced the needed values; it just checks that they are registered to produce it in the future.

fetch_all(many_handler=None)

Fetch all named execute results known so far.

fetch_mapped_args(args_mapping, atom_name=None, scope_walker=None, optional_args=None)

Fetch execute arguments for an atom using its args mapping.

set_flow_state(state)

Set flow details state and save it.

update_flow_metadata(update_with)

Update flowdetails metadata and save it.

change_flow_state(state)

Transition flow from old state to new state.

Returns (True, old_state) if transition was performed, or (False, old_state) if it was ignored, or raises a InvalidState exception if transition is invalid.

get_flow_state()

Get state from flow details.

get_retry_history(retry_name)

Fetch a single retrys history.

get_retry_histories()

Fetch all retrys histories.

•

Guarantees workflow forward progress by transferring partially complete work or work that has not been started to entities which can either resume the previously partially completed work or begin initial work to ensure that the workflow as a whole progresses (where progressing implies transitioning through the workflow patterns and atoms and completing their associated states transitions).

• Ensures that only one workflow is managed (aka owned) by a single owner at a time in an atomic manner (including when the workflow is transferred to a owner that is resuming some other failed owners work). This avoids contention and ensures a workflow is managed by one and only one entity at a time.
• Note: this does not mean that the owner needs to run the workflow itself but instead said owner could use an engine that runs the work in a distributed manner to ensure that the workflow progresses.

•

Jobs can be created with logbooks that contain a specification of the work to be done by a entity (such as an API server). The job then can be completed by a entity that is watching that jobboard (not necessarily the API server itself). This creates a disconnection between work formation and work completion that is useful for scaling out horizontally.

•

When for example a API server posts a job for completion to a jobboard that API server can return a tracking identifier to the user calling the API service. This tracking identifier can be used by the user to poll for status (similar in concept to a shipping tracking identifier created by fedex or UPS).

from taskflow.persistence import backends as persistence_backends
from taskflow.jobs import backends as job_backends
...
persistence = persistence_backends.fetch({


    "connection': "mysql",


    "user": ...,


    "password": ...,
})
book = make_and_save_logbook(persistence)
board = job_backends.fetch('my-board', {


    "board": "zookeeper",
}, persistence=persistence)
job = board.post("my-first-job", book)
...

import time
from taskflow import exceptions as exc
from taskflow.persistence import backends as persistence_backends
from taskflow.jobs import backends as job_backends
...
my_name = 'worker-1'
coffee_break_time = 60
persistence = persistence_backends.fetch({


    "connection': "mysql",


    "user": ...,


    "password": ...,
})
board = job_backends.fetch('my-board', {


    "board": "zookeeper",
}, persistence=persistence)
while True:


    my_job = None


    for job in board.iterjobs(only_unclaimed=True):


        try:


            board.claim(job, my_name)


        except exc.UnclaimableJob:


            pass


        else:


            my_job = job


            break


    if my_job is not None:


        try:


            perform_job(my_job)


        except Exception:


            LOG.exception("I failed performing job: %s", my_job)


            board.abandon(my_job, my_name)


        else:


            # I finished it, now cleanup.


            board.consume(my_job)


            persistence.get_connection().destroy_logbook(my_job.book.uuid)


    time.sleep(coffee_break_time)
...

from oslo_utils import uuidutils
from taskflow import engines
from taskflow.persistence import backends as persistence_backends
from taskflow.persistence import models
from taskflow.jobs import backends as job_backends
...
persistence = persistence_backends.fetch({


    "connection': "mysql",


    "user": ...,


    "password": ...,
})
board = job_backends.fetch('my-board', {


    "board": "zookeeper",
}, persistence=persistence)
book = models.LogBook('my-book', uuidutils.generate_uuid())
flow_detail = models.FlowDetail('my-job', uuidutils.generate_uuid())
book.add(flow_detail)
connection = persistence.get_connection()
connection.save_logbook(book)
flow_detail.meta['store'] = {'a': 1, 'c': 3}
job_details = {


    "flow_uuid": flow_detail.uuid,


    "store": {'a': 2, 'b': 1}
}
engines.save_factory_details(flow_detail, flow_factory,


                             factory_args=[],


                             factory_kwargs={},


                             backend=persistence)
jobboard = get_jobboard(zk_client)
jobboard.connect()
job = jobboard.post('my-job', book=book, details=job_details)
# the flow global parameters are now the combined store values
# {'a': 2, 'b': 1', 'c': 3}
...

• Rolling back the last atom/set of atoms that finished.
• Rolling back the last state change that occurred.

VERY_HIGH = 'VERY_HIGH'

Extremely urgent job priority.

HIGH = 'HIGH'

Mildly urgent job priority.

NORMAL = 'NORMAL'

Default job priority.

LOW = 'LOW'

Not needed anytime soon job priority.

VERY_LOW = 'VERY_LOW'

Very much not needed anytime soon job priority.

classmethod reorder(*values)

Reorders (priority, value) tuples -> priority ordered values.

abstract property last_modified

The datetime the job was last modified.

abstract property created_on

The datetime the job was created on.

property board

The board this job was posted on or was created from.

abstract property state

Access the current state of this job.

abstract property priority

The JobPriority of this job.

wait(timeout=None, delay=0.01, delay_multiplier=2.0, max_delay=60.0, sleep_func=<built-in function sleep>)

Wait for job to enter completion state.

If the job has not completed in the given timeout, then return false, otherwise return true (a job failure exception may also be raised if the job information can not be read, for whatever reason). Periodic state checks will happen every delay seconds where delay will be multiplied by the given multipler after a state is found that is not complete.

Note that if no timeout is given this is equivalent to blocking until the job has completed. Also note that if a jobboard backend can optimize this method then its implementation may not use delays (and backoffs) at all. In general though no matter what optimizations are applied implementations must always respect the given timeout value.

property book

Logbook associated with this job.

If no logbook is associated with this job, this property is None.

property book_uuid

UUID of logbook associated with this job.

If no logbook is associated with this job, this property is None.

property book_name

Name of logbook associated with this job.

If no logbook is associated with this job, this property is None.

property uuid

The uuid of this job.

property details

A dictionary of any details associated with this job.

property name

The non-uniquely identifying name of this job.

• only_unclaimed: boolean that indicates whether to only iterate over unclaimed jobs
• ensure_fresh: boolean that requests that during every fetch of a new set of jobs this will cause the iterator to force the backend to refresh (ensuring that the jobboard has the most recent job listings)
• board: the board this iterator was created from

property board

The board this iterator was created from.

abstract iterjobs(only_unclaimed=False, ensure_fresh=False)

Returns an iterator of jobs that are currently on this board.

NOTE(harlowja): the ordering of this iteration should be by posting order (oldest to newest) with higher priority jobs being provided before lower priority jobs, but it is left up to the backing implementation to provide the order that best suits it..

NOTE(harlowja): the iterator that is returned may support other attributes which can be used to further customize how iteration can be accomplished; check with the backends iterator object to determine what other attributes are supported.

abstract wait(timeout=None)

Waits a given amount of time for any jobs to be posted.

When jobs are found then an iterator will be returned that can be used to iterate over those jobs.

NOTE(harlowja): since a jobboard can be mutated on by multiple external entities at the same time the iterator that can be returned may still be empty due to other entities removing those jobs after the iterator has been created (be aware of this when using it).

abstract property job_count

Returns how many jobs are on this jobboard.

NOTE(harlowja): this count may change as jobs appear or are removed so the accuracy of this count should not be used in a way that requires it to be exact & absolute.

abstract find_owner(job)

Gets the owner of the job if one exists.

property name

The non-uniquely identifying name of this jobboard.

abstract consume(job, who)

Permanently (and atomically) removes a job from the jobboard.

Consumption signals to the board (and any others examining the board) that this job has been completed by the entity that previously claimed that job.

Only the entity that has claimed that job is able to consume the job.

A job that has been consumed can not be reclaimed or reposted by another entity (job postings are immutable). Any entity consuming a unclaimed job (or a job they do not have a claim on) will cause an exception.

abstract post(name, book=None, details=None, priority=JobPriority.NORMAL)

Atomically creates and posts a job to the jobboard.

This posting allowing others to attempt to claim that job (and subsequently work on that job). The contents of the provided logbook, details dictionary, or name (or a mix of these) must provide enough information for consumers to reference to construct and perform that jobs contained work (whatever it may be).

Once a job has been posted it can only be removed by consuming that job (after that job is claimed). Any entity can post/propose jobs to the jobboard (in the future this may be restricted).

Returns a job object representing the information that was posted.

abstract claim(job, who)

Atomically attempts to claim the provided job.

If a job is claimed it is expected that the entity that claims that job will at sometime in the future work on that jobs contents and either fail at completing them (resulting in a reposting) or consume that job from the jobboard (signaling its completion). If claiming fails then a corresponding exception will be raised to signal this to the claim attempter.

abstract abandon(job, who)

Atomically attempts to abandon the provided job.

This abandonment signals to others that the job may now be reclaimed. This would typically occur if the entity that has claimed the job has failed or is unable to complete the job or jobs it had previously claimed.

Only the entity that has claimed that job can abandon a job. Any entity abandoning a unclaimed job (or a job they do not own) will cause an exception.

abstract trash(job, who)

Trash the provided job.

Trashing a job signals to others that the job is broken and should not be reclaimed. This is provided as an option for users to be able to remove jobs from the board externally. The trashed job details should be kept around in an alternate location to be reviewed, if desired.

Only the entity that has claimed that job can trash a job. Any entity trashing a unclaimed job (or a job they do not own) will cause an exception.

abstract register_entity(entity)

Register an entity to the jobboard('s backend), e.g: a conductor.

abstract property connected

Returns if this jobboard is connected.

abstract connect()

Opens the connection to any backend system.

abstract close()

Close the connection to any backend system.

Once closed the jobboard can no longer be used (unless reconnection occurs).

property key

Key (in board listings/trash hash) the job data is stored under.

property priority

The JobPriority of this job.

property last_modified_key

Key the job last modified data is stored under.

property owner_key

Key the job claim + data of the owner is stored under.

property sequence

Sequence number of the current job.

expires_in()

How many seconds until the claim expires.

Returns the number of seconds until the ownership entry expires or DOES_NOT_EXPIRE or KEY_NOT_FOUND if it does not expire or if the expiry can not be determined (perhaps the owner_key expired at/before time of inquiry?).

extend_expiry(expiry)

Extends the owner key (aka the claim) expiry for this job.

NOTE(harlowja): if the claim for this job did not previously have an expiry associated with it, calling this method will create one (and after that time elapses the claim on this job will cease to exist).

Returns True if the expiry request was performed otherwise False.

property created_on

The datetime the job was created on.

property last_modified

The datetime the job was last modified.

property state

Access the current state of this job.

CLIENT_CONF_TRANSFERS = (('host', <class 'str'>), ('port', <class 'int'>), ('password', <class 'str'>), ('encoding', <class 'str'>), ('encoding_errors', <class 'str'>), ('socket_timeout', <class 'float'>), ('socket_connect_timeout', <class 'float'>), ('unix_socket_path', <class 'str'>), ('ssl', <function bool_from_string>), ('ssl_keyfile', <class 'str'>), ('ssl_certfile', <class 'str'>), ('ssl_cert_reqs', <class 'str'>), ('ssl_ca_certs', <class 'str'>), ('db', <class 'int'>))

Keys (and value type converters) that we allow to proxy from the jobboard configuration into the redis client (used to configure the redis client internals if no explicit client is provided via the client keyword argument).

See: http://redis-py.readthedocs.org/en/latest/#redis.Redis

See: https://github.com/andymccurdy/redis-py/blob/2.10.3/redis/client.py

OWNED_POSTFIX = b'.owned'

Postfix (combined with job key) used to make a jobs owner key.

LAST_MODIFIED_POSTFIX = b'.last_modified'

Postfix (combined with job key) used to make a jobs last modified key.

DEFAULT_NAMESPACE = b'taskflow'

Default namespace for keys when none is provided.

MIN_REDIS_VERSION = (2, 6)

Minimum redis version this backend requires.

This version is required since we need the built-in server-side lua scripting support that is included in 2.6 and newer.

NAMESPACE_SEP = b':'

Separator that is used to combine a key with the namespace (to get the actual key that will be used).

KEY_PIECE_SEP = b'.'

Separator that is used to combine a bunch of key pieces together (to get the actual key that will be used).

SCRIPT_STATUS_OK = 'ok'

Expected lua response status field when call is ok.

SCRIPT_STATUS_ERROR = 'error'

Expected lua response status field when call is not ok.

SCRIPT_NOT_EXPECTED_OWNER = 'Not expected owner!'

Expected lua script error response when the owner is not as expected.

SCRIPT_UNKNOWN_OWNER = 'Unknown owner!'

Expected lua script error response when the owner is not findable.

SCRIPT_UNKNOWN_JOB = 'Unknown job!'

Expected lua script error response when the job is not findable.

SCRIPT_ALREADY_CLAIMED = 'Job already claimed!'

Expected lua script error response when the job is already claimed.

SCRIPT_TEMPLATES = {'abandon': '\n-- Extract *all* the variables (so we can easily know what they are)...\nlocal owner_key = KEYS[1]\nlocal listings_key = KEYS[2]\nlocal last_modified_key = KEYS[3]\n\nlocal expected_owner = ARGV[1]\nlocal job_key = ARGV[2]\nlocal last_modified_blob = ARGV[3]\nlocal result = {}\nif redis.call("hexists", listings_key, job_key) == 1 then\n if redis.call("exists", owner_key) == 1 then\n local owner = redis.call("get", owner_key)\n if owner ~= expected_owner then\n result["status"] = "${error}"\n result["reason"] = "${not_expected_owner}"\n result["owner"] = owner\n else\n redis.call("del", owner_key)\n redis.call("set", last_modified_key, last_modified_blob)\n result["status"] = "${ok}"\n end\n else\n result["status"] = "${error}"\n result["reason"] = "${unknown_owner}"\n end\nelse\n result["status"] = "${error}"\n result["reason"] = "${unknown_job}"\nend\nreturn cmsgpack.pack(result)\n', 'claim': '\nlocal function apply_ttl(key, ms_expiry)\n if ms_expiry ~= nil then\n redis.call("pexpire", key, ms_expiry)\n end\nend\n\n-- Extract *all* the variables (so we can easily know what they are)...\nlocal owner_key = KEYS[1]\nlocal listings_key = KEYS[2]\nlocal last_modified_key = KEYS[3]\n\nlocal expected_owner = ARGV[1]\nlocal job_key = ARGV[2]\nlocal last_modified_blob = ARGV[3]\n\n-- If this is non-numeric (which it may be) this becomes nil\nlocal ms_expiry = nil\nif ARGV[4] ~= "none" then\n ms_expiry = tonumber(ARGV[4])\nend\nlocal result = {}\nif redis.call("hexists", listings_key, job_key) == 1 then\n if redis.call("exists", owner_key) == 1 then\n local owner = redis.call("get", owner_key)\n if owner == expected_owner then\n -- Owner is the same, leave it alone...\n redis.call("set", last_modified_key, last_modified_blob)\n apply_ttl(owner_key, ms_expiry)\n end\n result["status"] = "${error}"\n result["reason"] = "${already_claimed}"\n result["owner"] = owner\n else\n redis.call("set", owner_key, expected_owner)\n redis.call("set", last_modified_key, last_modified_blob)\n apply_ttl(owner_key, ms_expiry)\n result["status"] = "${ok}"\n end\nelse\n result["status"] = "${error}"\n result["reason"] = "${unknown_job}"\nend\nreturn cmsgpack.pack(result)\n', 'consume': '\n-- Extract *all* the variables (so we can easily know what they are)...\nlocal owner_key = KEYS[1]\nlocal listings_key = KEYS[2]\nlocal last_modified_key = KEYS[3]\n\nlocal expected_owner = ARGV[1]\nlocal job_key = ARGV[2]\nlocal result = {}\nif redis.call("hexists", listings_key, job_key) == 1 then\n if redis.call("exists", owner_key) == 1 then\n local owner = redis.call("get", owner_key)\n if owner ~= expected_owner then\n result["status"] = "${error}"\n result["reason"] = "${not_expected_owner}"\n result["owner"] = owner\n else\n -- The order is important here, delete the owner first (and if\n -- that blows up, the job data will still exist so it can be\n -- worked on again, instead of the reverse)...\n redis.call("del", owner_key, last_modified_key)\n redis.call("hdel", listings_key, job_key)\n result["status"] = "${ok}"\n end\n else\n result["status"] = "${error}"\n result["reason"] = "${unknown_owner}"\n end\nelse\n result["status"] = "${error}"\n result["reason"] = "${unknown_job}"\nend\nreturn cmsgpack.pack(result)\n', 'trash': '\n-- Extract *all* the variables (so we can easily know what they are)...\nlocal owner_key = KEYS[1]\nlocal listings_key = KEYS[2]\nlocal last_modified_key = KEYS[3]\nlocal trash_listings_key = KEYS[4]\n\nlocal expected_owner = ARGV[1]\nlocal job_key = ARGV[2]\nlocal last_modified_blob = ARGV[3]\nlocal result = {}\nif redis.call("hexists", listings_key, job_key) == 1 then\n local raw_posting = redis.call("hget", listings_key, job_key)\n if redis.call("exists", owner_key) == 1 then\n local owner = redis.call("get", owner_key)\n if owner ~= expected_owner then\n result["status"] = "${error}"\n result["reason"] = "${not_expected_owner}"\n result["owner"] = owner\n else\n -- This ordering is important (try to first move the value\n -- and only if that works do we try to do any deletions)...\n redis.call("hset", trash_listings_key, job_key, raw_posting)\n redis.call("set", last_modified_key, last_modified_blob)\n redis.call("del", owner_key)\n redis.call("hdel", listings_key, job_key)\n result["status"] = "${ok}"\n end\n else\n result["status"] = "${error}"\n result["reason"] = "${unknown_owner}"\n end\nelse\n result["status"] = "${error}"\n result["reason"] = "${unknown_job}"\nend\nreturn cmsgpack.pack(result)\n'}

Lua template scripts that will be used by various methods (they are turned into real scripts and loaded on call into the connect() method).

Some things to note:

So overall after thinking, it seemed like having little lua scripts was not that bad (even if it is somewhat convoluted) due to the above and public mentioned issues with transactions. In general using lua scripts for this purpose seems to be somewhat common practice and it solves the issues that came up when transactions were considered & implemented.

Some links about redis (and redis + lua) that may be useful to look over:

join(key_piece, *more_key_pieces)

Create and return a namespaced key from many segments.

NOTE(harlowja): all pieces that are text/unicode are converted into their binary equivalent (if they are already binary no conversion takes place) before being joined (as redis expects binary keys and not unicode/text ones).

property namespace

The namespace all keys will be prefixed with (or none).

trash_key

Key where a hash will be stored with trashed jobs in it.

sequence_key

Key where a integer will be stored (used to sequence jobs).

listings_key

Key where a hash will be stored with active jobs in it.

property job_count

Returns how many jobs are on this jobboard.

NOTE(harlowja): this count may change as jobs appear or are removed so the accuracy of this count should not be used in a way that requires it to be exact & absolute.

property connected

Returns if this jobboard is connected.

connect()

Opens the connection to any backend system.

close()

Close the connection to any backend system.

Once closed the jobboard can no longer be used (unless reconnection occurs).

find_owner(job)

Gets the owner of the job if one exists.

post(name, book=None, details=None, priority=JobPriority.NORMAL)

Atomically creates and posts a job to the jobboard.

This posting allowing others to attempt to claim that job (and subsequently work on that job). The contents of the provided logbook, details dictionary, or name (or a mix of these) must provide enough information for consumers to reference to construct and perform that jobs contained work (whatever it may be).

Once a job has been posted it can only be removed by consuming that job (after that job is claimed). Any entity can post/propose jobs to the jobboard (in the future this may be restricted).

Returns a job object representing the information that was posted.

wait(timeout=None, initial_delay=0.005, max_delay=1.0, sleep_func=<built-in function sleep>)

Waits a given amount of time for any jobs to be posted.

When jobs are found then an iterator will be returned that can be used to iterate over those jobs.

NOTE(harlowja): since a jobboard can be mutated on by multiple external entities at the same time the iterator that can be returned may still be empty due to other entities removing those jobs after the iterator has been created (be aware of this when using it).

iterjobs(only_unclaimed=False, ensure_fresh=False)

Returns an iterator of jobs that are currently on this board.

NOTE(harlowja): the ordering of this iteration should be by posting order (oldest to newest) with higher priority jobs being provided before lower priority jobs, but it is left up to the backing implementation to provide the order that best suits it..

NOTE(harlowja): the iterator that is returned may support other attributes which can be used to further customize how iteration can be accomplished; check with the backends iterator object to determine what other attributes are supported.

register_entity(entity)

Register an entity to the jobboard('s backend), e.g: a conductor.

consume(job, who)

Permanently (and atomically) removes a job from the jobboard.

Consumption signals to the board (and any others examining the board) that this job has been completed by the entity that previously claimed that job.

Only the entity that has claimed that job is able to consume the job.

A job that has been consumed can not be reclaimed or reposted by another entity (job postings are immutable). Any entity consuming a unclaimed job (or a job they do not have a claim on) will cause an exception.

claim(job, who, expiry=None)

Atomically attempts to claim the provided job.

If a job is claimed it is expected that the entity that claims that job will at sometime in the future work on that jobs contents and either fail at completing them (resulting in a reposting) or consume that job from the jobboard (signaling its completion). If claiming fails then a corresponding exception will be raised to signal this to the claim attempter.

abandon(job, who)

Atomically attempts to abandon the provided job.

This abandonment signals to others that the job may now be reclaimed. This would typically occur if the entity that has claimed the job has failed or is unable to complete the job or jobs it had previously claimed.

Only the entity that has claimed that job can abandon a job. Any entity abandoning a unclaimed job (or a job they do not own) will cause an exception.

trash(job, who)

Trash the provided job.

Trashing a job signals to others that the job is broken and should not be reclaimed. This is provided as an option for users to be able to remove jobs from the board externally. The trashed job details should be kept around in an alternate location to be reviewed, if desired.

Only the entity that has claimed that job can trash a job. Any entity trashing a unclaimed job (or a job they do not own) will cause an exception.

ENTITY_KIND = 'conductor'

Entity kind used when creating new entity objects

conductor

Entity object that represents this conductor.

property notifier

The conductor actions (or other state changes) notifier.

NOTE(harlowja): different conductor implementations may emit different events + event details at different times, so refer to your conductor documentation to know exactly what can and what can not be subscribed to.

connect()

Ensures the jobboard is connected (noop if it is already).

close()

Closes the contained jobboard, disallowing further use.

abstract run(max_dispatches=None)

Continuously claims, runs, and consumes jobs (and repeat).

LOG = None

Logger that will be used for listening to events (if none then the module level logger will be used instead).

REFRESH_PERIODICITY = 30

Every 30 seconds the jobboard will be resynced (if for some reason a watch or set of watches was not received) using the ensure_fresh option to ensure this (for supporting jobboard backends only).

WAIT_TIMEOUT = 0.5

Default timeout used to idle/wait when no jobs have been found.

MAX_SIMULTANEOUS_JOBS = -1

Default maximum number of jobs that can be in progress at the same time.

Negative or zero values imply no limit (do note that if a executor is used that is built on a queue, as most are, that this will imply that the queue will contain a potentially large & unfinished backlog of submitted jobs). This may get better someday if https://bugs.python.org/issue22737 is ever implemented and released.

NO_CONSUME_EXCEPTIONS = (<class 'taskflow.exceptions.ExecutionFailure'>, <class 'taskflow.exceptions.StorageFailure'>)

Exceptions that will not cause consumption to occur.

EVENTS_EMITTED = ('compilation_start', 'compilation_end', 'preparation_start', 'preparation_end', 'validation_start', 'validation_end', 'running_start', 'running_end', 'job_consumed', 'job_abandoned')

Events will be emitted for each of the events above. The event is emitted to listeners registered with the conductor.

stop()

Requests the conductor to stop dispatching.

This method can be used to request that a conductor stop its consumption & dispatching loop.

The method returns immediately regardless of whether the conductor has been stopped.

property dispatching

Whether or not the dispatching loop is still dispatching.

run(max_dispatches=None)

Continuously claims, runs, and consumes jobs (and repeat).

wait(timeout=None)

Waits for the conductor to gracefully exit.

This method waits for the conductor to gracefully exit. An optional timeout can be provided, which will cause the method to return within the specified timeout. If the timeout is reached, the returned value will be False, otherwise it will be True.

MAX_SIMULTANEOUS_JOBS = 1

Default maximum number of jobs that can be in progress at the same time.

NOTE(harlowja): A custom executor factory can be provided via keyword

argument executor_factory, if provided it will be invoked at run() time with one positional argument (this conductor) and it must return a compatible executor which can be used to submit jobs to. If None is a provided a thread pool backed executor is selected by default (it will have an equivalent number of workers as this conductors simultaneous job count).

MAX_SIMULTANEOUS_JOBS = 33

Default maximum number of jobs that can be in progress at the same time.

import logging
import os
import sys
logging.basicConfig(level=logging.ERROR)
top_dir = os.path.abspath(os.path.join(os.path.dirname(__file__),


                                       os.pardir,


                                       os.pardir))
sys.path.insert(0, top_dir)
from taskflow import engines
from taskflow.patterns import linear_flow as lf
from taskflow.patterns import unordered_flow as uf
from taskflow import task
# INTRO: This is the defacto hello world equivalent for taskflow; it shows how
# an overly simplistic workflow can be created that runs using different
# engines using different styles of execution (all can be used to run in
# parallel if a workflow is provided that is parallelizable).
class PrinterTask(task.Task):


    def __init__(self, name, show_name=True, inject=None):


        super(PrinterTask, self).__init__(name, inject=inject)


        self._show_name = show_name


    def execute(self, output):


        if self._show_name:


            print("%s: %s" % (self.name, output))


        else:


            print(output)
# This will be the work that we want done, which for this example is just to
# print 'hello world' (like a song) using different tasks and different
# execution models.
song = lf.Flow("beats")
# Unordered flows when ran can be ran in parallel; and a chorus is everyone
# singing at once of course!
hi_chorus = uf.Flow('hello')
world_chorus = uf.Flow('world')
for (name, hello, world) in [('bob', 'hello', 'world'),


                             ('joe', 'hellooo', 'worllllld'),


                             ('sue', "helloooooo!", 'wooorllld!')]:


    hi_chorus.add(PrinterTask("%s@hello" % name,


                              # This will show up to the execute() method of


                              # the task as the argument named 'output' (which


                              # will allow us to print the character we want).


                              inject={'output': hello}))


    world_chorus.add(PrinterTask("%s@world" % name,


                                 inject={'output': world}))
# The composition starts with the conductor and then runs in sequence with
# the chorus running in parallel, but no matter what the 'hello' chorus must
# always run before the 'world' chorus (otherwise the world will fall apart).
song.add(PrinterTask("conductor@begin",


                     show_name=False, inject={'output': "*ding*"}),


         hi_chorus,


         world_chorus,


         PrinterTask("conductor@end",


                     show_name=False, inject={'output': "*dong*"}))
# Run in parallel using eventlet green threads...
try:


    import eventlet as _eventlet  # noqa
except ImportError:


    # No eventlet currently active, skip running with it...


    pass
else:


    print("-- Running in parallel using eventlet --")


    e = engines.load(song, executor='greenthreaded', engine='parallel',


                     max_workers=1)


    e.run()
# Run in parallel using real threads...
print("-- Running in parallel using threads --")
e = engines.load(song, executor='threaded', engine='parallel',


                 max_workers=1)
e.run()
# Run in parallel using external processes...
print("-- Running in parallel using processes --")
e = engines.load(song, executor='processes', engine='parallel',


                 max_workers=1)
e.run()
# Run serially (aka, if the workflow could have been ran in parallel, it will
# not be when ran in this mode)...
print("-- Running serially --")
e = engines.load(song, engine='serial')
e.run()
print("-- Statistics gathered --")
print(e.statistics)