Task Message Protocol v2 (Draft Spec.)

Notes

  • Support for multiple languages via the lang header.

Worker may redirect the message to a worker that supports the language. - Metadata moved to headers.

This means that workers/intermediates can inspect the message and make decisions based on the headers without decoding the payload (which may be language specific, e.g. serialized by the Python specific pickle serializer). - Body is only for language specific data.

  • Python stores args/kwargs in body.
  • If a message uses raw encoding then the raw data will be passed as a single argument to the function.
  • Java/C, etc. can use a thrift/protobuf document as the body
  • Dispatches to actor based on c_type, c_meth headers

c_meth is unused by python, but may be used in the future to specify class+method pairs. - Chain gains a dedicated field.

Reducing the chain into a recursive callbacks argument causes problems when the recursion limit is exceeded.

This is fixed in the new message protocol by specifying a list of signatures, each task will then pop a task off the list when sending the next message:

```

  1. execute_task(message)
  2. chain = message.headers['chain']
  3. if chain:
  4. sig = maybe_signature(chain.pop())
  5. sig.apply_async(chain=chain) ``` - correlation_id replaces task_id field. - c_shadow lets you specify a different name for logs, monitors can be used for e.g. meta tasks that calls any function:

```

  1. from celery.utils.imports import qualname

  2. class PickleTask(Task):

  3. abstract = True

  4. def unpack_args(self, fun, args=()):

  5. return fun, args

  6. def apply_async(self, args, kwargs, **options):

  7. fun, real_args = self.unpack_args(*args)
  8. return super(PickleTask, self).apply_async(
  9. (fun, real_args, kwargs), shadow=qualname(fun), **options

  10. )

  11. @app.task(base=PickleTask)

  12. def call(fun, args, kwargs):
  13. return fun(args, *kwargs) ```

Undecided

  • May consider moving callbacks/errbacks/chain into body.

Will huge lists in headers cause overhead? The downside of keeping them in the body is that intermediates won’t be able to introspect these values.

Definition


1. # protocol v2 implies UTC=True
2. # 'class' header existing means protocol is v2

4. properties = {
5. 'correlation_id': (uuid)task_id,
6. 'content_type': (string)mime,
7. 'content_encoding': (string)encoding,

9. # optional
10. 'reply_to': (string)queue_or_url,
11. }
12. headers = {
13. 'lang': (string)'py'
14. 'c_type': (string)task,

16. # optional
17. 'c_meth': (string)unused,
18. 'c_shadow': (string)replace_name,
19. 'eta': (iso8601)eta,
20. 'expires'; (iso8601)expires,
21. 'callbacks': (list)Signature,
22. 'errbacks': (list)Signature,
23. 'chain': (list)Signature,  # non-recursive, reversed list of signatures
24. 'group': (uuid)group_id,
25. 'chord': (uuid)chord_id,
26. 'retries': (int)retries,
27. 'timelimit': (tuple)(soft, hard),
28. }

30. body = (args, kwargs)

Example


1. # chain: add(add(add(2, 2), 4), 8) == 2 + 2 + 4 + 8

3. task_id = uuid()
4. basic_publish(
5. message=json.dumps([[2, 2], {}]),
6. application_headers={
7. 'lang': 'py',
8. 'c_type': 'proj.tasks.add',
9. 'chain': [
10. # reversed chain list
11. {'task': 'proj.tasks.add', 'args': (8, )},
12. {'task': 'proj.tasks.add', 'args': (4, )},
13. ]
14. }
15. properties={
16. 'correlation_id': task_id,
17. 'content_type': 'application/json',
18. 'content_encoding': 'utf-8',
19. }
20. )