Terminology¶

Client

Code or program or service (or user) that uses this library to define flows and run them via engines.

Transport + protocol

Mechanism (and protocol on top of that mechanism) used to pass information between the client and worker (for example amqp as a transport and a json encoded message format as the protocol).

Executor

Part of the worker-based engine and is used to publish task requests, so these requests can be accepted and processed by remote workers.

Worker

Workers are started on remote hosts and each has a list of tasks it can perform (on request). Workers accept and process task requests that are published by an executor. Several requests can be processed simultaneously in separate threads (or processes…). For example, an executor can be passed to the worker and configured to run in as many threads (green or not) as desired.

Proxy

Executors interact with workers via a proxy. The proxy maintains the underlying transport and publishes messages (and invokes callbacks on message reception).

Requirements¶

• Transparent: it should work as ad-hoc replacement for existing (local) engines with minimal, if any refactoring (e.g. it should be possible to run the same flows on it without changing client code if everything is set up and configured properly).
• Transport-agnostic: the means of transport should be abstracted so that we can use oslo.messaging, gearmand, amqp, zookeeper, marconi, websockets or anything else that allows for passing information between a client and a worker.
• Simple: it should be simple to write and deploy.
• Non-uniformity: it should support non-uniform workers which allows different workers to execute different sets of atoms depending on the workers published capabilities.

High level architecture¶

Executor and worker communication¶

Let’s consider how communication between an executor and a worker happens. First of all an engine resolves all atoms dependencies and schedules atoms that can be performed at the moment. This uses the same scheduling and dependency resolution logic that is used for every other engine type. Then the atoms which can be executed immediately (ones that are dependent on outputs of other tasks will be executed when that output is ready) are executed by the worker-based engine executor in the following manner:

1. The executor initiates task execution/reversion using a proxy object.
2. Proxy publishes task request (format is described below) into a named exchange using a routing key that is used to deliver request to particular workers topic. The executor then waits for the task requests to be accepted and confirmed by workers. If the executor doesn’t get a task confirmation from workers within the given timeout the task is considered as timed-out and a timeout exception is raised.
3. A worker receives a request message and starts a new thread for processing it.
   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).
4. The executor gets the task request confirmation from the worker and the task request state changes from the PENDING to the RUNNING state. Once a task request is in the RUNNING state it can’t be timed-out (considering that the task execution process may take an unpredictable amount of time).
5. The executor gets the task execution result from the worker and passes it back to the executor and worker-based engine to finish task processing (this repeats for subsequent tasks).

{
    "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"
}

Request state transitions¶

WAITING - Request placed on queue (or other kombu message bus/transport) but not yet consumed.

PENDING - Worker accepted request and is pending to run using its executor (threads, processes, or other).

FAILURE - Worker failed after running request (due to task exception) or no worker moved/started executing (by placing the request into RUNNING state) with-in specified time span (this defaults to 60 seconds unless overridden).

RUNNING - Workers executor (using threads, processes…) has started to run requested task (once this state is transitioned to any request timeout no longer becomes applicable; since at this point it is unknown how long a task will run since it can not be determined if a task is just taking a long time or has failed).

SUCCESS - Worker finished running task without exception.

Workers¶

To use the worker based engine a set of workers must first be established on remote machines. These workers must be provided a list of task objects, task names, modules names (or entrypoints that can be examined for valid tasks) they can respond to (this is done so that arbitrary code execution is not possible).

For complete parameters and object usage please visit Worker.

Example:

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

Engines¶

To use the worker based engine a flow must be constructed (which contains tasks that are visible on remote machines) and the specific worker based engine entrypoint must be selected. Certain configuration options must also be provided so that the transport backend can be configured and initialized correctly. Otherwise the usage should be mostly transparent (and is nearly identical to using any other engine type).

For complete parameters and object usage please see WorkerBasedActionEngine.

Example with amqp transport:

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

Example with filesystem transport:

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

Additional supported keyword arguments:

• executor: a class that provides a WorkerTaskExecutor interface; it will be used for executing, reverting and waiting for remote tasks.

Components¶

class taskflow.engines.worker_based.dispatcher.Handler(process_message, validator=None)[source]¶

Bases: object

Component(s) that will be called on reception of messages.

process_message¶

Main callback that is called to process a received message.

This is only called after the format has been validated (using the validator callback if applicable) and only after the message has been acknowledged.

validator¶

Optional callback that will be activated before processing.

This callback if present is expected to validate the message and raise InvalidFormat if the message is not valid.

class taskflow.engines.worker_based.dispatcher.TypeDispatcher(type_handlers=None, requeue_filters=None)[source]¶

Bases: object

Receives messages and dispatches to type specific handlers.

type_handlers¶

Dictionary of message type -> callback to handle that message.

The callback(s) will be activated by looking for a message property ‘type’ and locating a callback in this dictionary that maps to that type; if one is found it is expected to be a callback that accepts two positional parameters; the first being the message data and the second being the message object. If a callback is not found then the message is rejected and it will be up to the underlying message transport to determine what this means/implies…

requeue_filters¶

List of filters (callbacks) to request a message to be requeued.

The callback(s) will be activated before the message has been acked and it can be used to instruct the dispatcher to requeue the message instead of processing it. The callback, when called, will be provided two positional parameters; the first being the message data and the second being the message object. Using these provided parameters the filter should return a truthy object if the message should be requeued and a falsey object if it should not.

on_message(data, message)[source]¶

This method is called on incoming messages.

class taskflow.engines.worker_based.endpoint.Endpoint(task_cls)[source]¶

Bases: object

Represents a single task with execute/revert methods.

class taskflow.engines.worker_based.executor.WorkerTaskExecutor(uuid, exchange, topics, transition_timeout=60, url=None, transport=None, transport_options=None, retry_options=None, worker_expiry=60)[source]¶

Bases: taskflow.engines.action_engine.executor.TaskExecutor

Executes tasks on remote workers.

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

Schedules task execution.

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

Schedules task reversion.

wait_for_workers(workers=1, timeout=None)[source]¶

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.

start()[source]¶

Starts message processing thread.

stop()[source]¶

Stops message processing thread.

class taskflow.engines.worker_based.proxy.Proxy(topic, exchange, type_handlers=None, on_wait=None, url=None, transport=None, transport_options=None, retry_options=None)[source]¶

Bases: object

A proxy processes messages from/to the named exchange.

For internal usage only (not for public consumption).

DEFAULT_RETRY_OPTIONS = {'interval_max': 1, 'interval_start': 1, 'interval_step': 1, 'max_retries': 3}¶

Settings used (by default) to reconnect under transient failures.

See: http://kombu.readthedocs.org/ (and connection ensure_options) for what these values imply/mean…

dispatcher¶

Dispatcher internally used to dispatch message(s) that match.

connection_details¶

Details about the connection (read-only).

is_running¶

Return whether the proxy is running.

publish(msg, routing_key, reply_to=None, correlation_id=None)[source]¶

Publish message to the named exchange with given routing key.

start()[source]¶

Start proxy.

wait()[source]¶

Wait until proxy is started.

stop()[source]¶

Stop proxy.

class taskflow.engines.worker_based.worker.Worker(exchange, topic, tasks, executor=None, threads_count=None, url=None, transport=None, transport_options=None, retry_options=None)[source]¶

Bases: object

Worker that can be started on a remote host for handling tasks requests.

Parameters:

url – broker url exchange – broker exchange name topic – topic name under which worker is stated tasks – task list that worker is capable of performing, items in the list can be one of the following types; 1, a string naming the python module name to search for tasks in or the task class name; 2, a python module to search for tasks in; 3, a task class object that will be used to create tasks from. executor – custom executor object that can used for processing requests in separate threads (if not provided one will be created) threads_count – threads count to be passed to the default executor (used only if an executor is not passed in) transport – transport to be used (e.g. amqp, memory, etc.) transport_options – transport specific options (see: http://kombu.readthedocs.org/ for what these options imply and are expected to be) retry_options – retry specific options (see: DEFAULT_RETRY_OPTIONS)

banner¶

A banner that can be useful to display before running.

run(display_banner=True, banner_writer=None)[source]¶

Runs the worker.

wait()[source]¶

Wait until worker is started.

stop()[source]¶

Stop worker.

class taskflow.engines.worker_based.types.TopicWorker(topic, tasks, identity=<object object>)[source]¶

Bases: object

A (read-only) worker and its relevant information + useful methods.

class taskflow.engines.worker_based.types.ProxyWorkerFinder(uuid, proxy, topics, beat_periodicity=5, worker_expiry=60)[source]¶

Bases: object

Requests and receives responses about workers topic+task details.

total_workers¶

Number of workers currently known.

wait_for_workers(workers=1, timeout=None)[source]¶

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.

messages_processed¶

How many notify response messages have been processed.

maybe_publish()[source]¶

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)[source]¶

Process notify message sent from remote side.

clean()[source]¶

Cleans out any dead/expired/not responding workers.

Returns how many workers were removed.

reset()[source]¶

Resets finders internal state.

get_worker_for_task(task)[source]¶

Gets a worker that can perform a given task.