Difference between revisions of "TaskFlow/Worker-based Engine Protocol"
(Response format changed) |
(Adjusted header levels) |
||
Line 1: | Line 1: | ||
− | |||
− | + | == Worker-based flow == | |
Worker-based flow uses remote workers to perform high load, computational and time-consuming tasks. There are two communication components - <code>Proxy</code> and <code>Worker</code>. | Worker-based flow uses remote workers to perform high load, computational and time-consuming tasks. There are two communication components - <code>Proxy</code> and <code>Worker</code>. | ||
Line 10: | Line 9: | ||
<code>Worker</code> is started on remote host and has list of tasks it can perform. <code>Worker</code> accepts and processes task requests that are published by <code>Proxy</code>. Several requests can be processed simultaneously in separate threads (executor can be passed to the <code>Worker</code> and configured). Workers that can process same set of tasks can be grouped under common <code>topic</code>. | <code>Worker</code> is started on remote host and has list of tasks it can perform. <code>Worker</code> accepts and processes task requests that are published by <code>Proxy</code>. Several requests can be processed simultaneously in separate threads (executor can be passed to the <code>Worker</code> and configured). Workers that can process same set of tasks can be grouped under common <code>topic</code>. | ||
− | + | == Protocol == | |
<code>Proxy</code> and <code>Workers</code> communicate in one common named exchange. <code>Proxy</code> publishes task requests to the named exchange and <code>Workers</code> take requests from it. When <code>Worker</code> gets task request it parses all parameters, dispatches endpoint and starts task processing. During task processing <code>Worker</code> sends responses back to <code>Proxy</code>. | <code>Proxy</code> and <code>Workers</code> communicate in one common named exchange. <code>Proxy</code> publishes task requests to the named exchange and <code>Workers</code> take requests from it. When <code>Worker</code> gets task request it parses all parameters, dispatches endpoint and starts task processing. During task processing <code>Worker</code> sends responses back to <code>Proxy</code>. | ||
Revision as of 21:05, 26 December 2013
Contents
Worker-based flow
Worker-based flow uses 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 and 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 (executor can be passed to the Worker
and configured). Workers that can process same set of tasks can be grouped under common 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 withrouting key
, that is used to deliver request to particular workerstopic
and waits for task requests to be accepted and confirmed byWorkers
. IfProxy
doesn't get task confirmation fromWorkers
within the given timeout task is considered as timed-out and theTimeout
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:
-
{'state': 'RUNNING'}
- If request dispatching
failed
Worker sends failure response to theProxy
withmisc.Failure
object:
{'state': 'FAILURE', 'result': <misc.Failure>}
- If request dispatching
-
Proxy
gets task request confirmation from theWorker
and task request state changed from thePending
to theRunning
state. Once task request is in theRunning
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 theProxy
:
- If result is of the
misc.Failure
type response with the failure state is sent back:
-
{'state': 'FAILURE', 'event': <event>, 'result': <misc.Failure>)}
- If result is not of the
misc.Failure
type response with the success state is sent back:
-
{'state': 'SUCCESS', 'event': <event>, 'result': <result>)}
- If result is of the
-
Proxy
gets task execution result fromWorker
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 notification to theProxy
where it is handled by engine:
-
{'state': 'PROGRESS', 'event_data': <event_data>, 'progress': <progress>}
-
Notes
-
misc.Failure
objects are not json-serializable, so they are converted to dict before sending and converted from dict after receiving on bothProxy
/Worker
sides. - When
Worker
sends response to theProxy
it publishes to the exchange that is taken from thereply_to
parameter of the AMQP message to reply to the correctProxy
.
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 running:
{'status': 'RUNNING'}
- Task progress:
{'state': 'PROGRESS', 'event_data': <event_data>, 'progress': <progress>}
- Task succeeded:
{'state': 'SUCCESS', 'event': <event>, 'result': <result>)}
- Task failed:
{'state': 'FAILURE', 'event': <event>, 'result': <misc.Failure>)}