Difference between revisions of "TaskFlow/Worker-based Engine Protocol"

Revision as of 16:53, 24 December 2013

Worker-based engine protocol

Worker-based flow

Worker-based flow use remote workers to perform high load, computational and time-consuming tasks. There are two communication components - Proxy and Worker.

Proxy

Proxy is a part of the worker-based engine. It is used to publish task requests, so these requests can be accepted and processed by remote Workers.

Worker

Worker is started on remote host and has list of tasks it can perform. Worker accepts and processes task requests that are published by Proxy. Several requests can be processed simultaneously in separate threads. Workers that can process same set of tasks are grouped under one topic.

Protocol

Proxy and Workers communicate in one common named exchange. Proxy publishes task requests to the named exchange and Workers take requests from it. When Worker gets task request it parses all parameters, dispatches endpoint and starts task processing. During task processing Worker sends responses back to Proxy.

Proxy and Worker communication

Let's consider how communication between Proxy and Worker happens. First of all engine resolves all tasks dependencies and schedules tasks that can be performed at the moment. Tasks are executed by worker-based engine with WorkerTaskExecutor. WorkerTaskExecutor initiates task execution/reversion using Proxy.

• Proxy publishes task request (format is described below) into named exchange with routing key, that is used to deliver request to particular workers topic and waits for task requests to be accepted and confirmed by Workers. If Proxy doesn't get task confirmation from Workers within the given timeout task is considered as timed-out and the Timeout exception is raised.
• Worker receives a request message it starts a new thread for processing it.
• Worker dispatches request (gets desired endpoint that actually executes task):
  

• If request dispatching succeeded Worker sends confirmation response to the Proxy:
  

{'status': 'RUNNING'}

• If request dispatching failed Worker sends failure response to the Proxy with misc.Failure object:
  

{'status': 'FAILURE', 'data': (None, misc.Failure)}

• Proxy gets task request confirmation from the Worker and task request status changed from the Pending to the Running state. Once task request is in the Running state it can't be timed-out (considering that task execution process may take unpredictable time).
• Worker executes task and once it is finished sends result back to the Proxy:

• If result is of the misc.Failure type response with the failure status is sent back:
  

{'status': 'FAILURE', 'data': ('executed'/'reverted', misc.Failure)}

• If result is not of the misc.Failure type response with the success status is sent back:
  

{'status': 'SUCCESS', 'data': ('executed'/'reverted', result)}

• Proxy gets task execution result from Worker and passes it back to the WorkerTaskExecutor and worker-based engine to finish task processing.
• Worker is subscribed on the task progress and every time task progress event is triggered it sends progress status notification to the Proxy where it is handled by engine:

{'status': 'PROGRESS', 'data': (event, progress)}

Notes

• misc.Failure objects are not json-serializable, so they are converted to dict before sending and converted from dict after receiving on both Proxy/Worker sides.
• When Worker sends response to the Proxy it publishes to the exchange that is taken from the reply_to parameter of the AMQP message to reply to the correct Proxy.

Proxy request format

• task - full task name to be performed
• action - task action to be performed (e.g. execute, revert)
• arguments - arguments the task action to be called with
• result - task execution result (result or misc.Failure) [passed to revert only]
• failures - flow tasks failures map (map of tasks names and misc.Failures) [passed to revert only]

Example

{'task': 'tasks.CallJoe', 'action': 'execute', 'arguments': {'joe_number': 444}}

Additionally, the following parameters are added to the AMQP request message:

• reply_to - Proxy named exchange workers will send responses back to
• correlation_id - Proxy request id (since there can be multiple request being processed simultaneously)

Worker response format

• Task accepted: {'status': 'RUNNING'}
• Task progress: {'status': 'PROGRESS', 'data': (event, progress)}
• Task succeeded: {'status': 'SUCCESS', 'data': (event, result)}
• Task failed: {'status': 'FAILURE', 'data': (None, misc.Failure)}