Canvas refactor, parts of root_id and parent_id and group no longer evaluates generators

celery/app/amqp.py

@@ -269,7 +269,8 @@ class AMQP(object):
                    expires=None, retries=0, chord=None,
                    callbacks=None, errbacks=None, reply_to=None,
                    time_limit=None, soft_time_limit=None,
-                   create_sent_event=False, now=None, timezone=None):
+                   create_sent_event=False, now=None, timezone=None,
+                   root_id=None, parent_id=None):
         args = args or ()
         kwargs = kwargs or {}
         utc = self.utc
@@ -305,6 +306,8 @@ class AMQP(object):
                 'chord': chord,
                 'retries': retries,
                 'timelimit': (time_limit, soft_time_limit),
+                'root_id': root_id,
+                'parent_id': parent_id,
             },
             properties={
                 'correlation_id': task_id,
@@ -313,6 +316,8 @@ class AMQP(object):
             body=(args, kwargs),
             sent_event={
                 'uuid': task_id,
+                'root': root_id,
+                'parent': parent_id,
                 'name': name,
                 'args': safe_repr(args),
                 'kwargs': safe_repr(kwargs),
@@ -327,7 +332,8 @@ class AMQP(object):
                    expires=None, retries=0,
                    chord=None, callbacks=None, errbacks=None, reply_to=None,
                    time_limit=None, soft_time_limit=None,
-                   create_sent_event=False, now=None, timezone=None):
+                   create_sent_event=False, now=None, timezone=None,
+                   root_id=None, parent_id=None):
         args = args or ()
         kwargs = kwargs or {}
         utc = self.utc

celery/app/base.py

@@ -351,7 +351,7 @@ class Celery(object):
                   publisher=None, link=None, link_error=None,
                   add_to_parent=True, group_id=None, retries=0, chord=None,
                   reply_to=None, time_limit=None, soft_time_limit=None,
-                  **options):
+                  root_id=None, parent_id=None, **options):
         amqp = self.amqp
         task_id = task_id or uuid()
         producer = producer or publisher  # XXX compat
@@ -369,6 +369,7 @@ class Celery(object):
             maybe_list(link), maybe_list(link_error),
             reply_to or self.oid, time_limit, soft_time_limit,
             self.conf.CELERY_SEND_TASK_SENT_EVENT,
+            root_id, parent_id,
         )
 
         if connection:

celery/app/builtins.py

@@ -9,10 +9,7 @@
 """
 from __future__ import absolute_import
 
-from collections import deque
-
 from celery._state import get_current_worker_task, connect_on_app_finalize
-from celery.utils import uuid
 from celery.utils.log import get_logger
 
 __all__ = []
@@ -44,7 +41,7 @@ def add_unlock_chord_task(app):
     It joins chords by creating a task chain polling the header for completion.
 
     """
-    from celery.canvas import signature
+    from celery.canvas import maybe_signature
     from celery.exceptions import ChordError
     from celery.result import allow_join_result, result_from_tuple
 
@@ -66,6 +63,8 @@ def add_unlock_chord_task(app):
             interval = unlock_chord.default_retry_delay
 
         # check if the task group is ready, and if so apply the callback.
+        callback = maybe_signature(callback, app)
+        root_id = callback.options.get('root_id')
         deps = GroupResult(
             group_id,
             [result_from_tuple(r, app=app) for r in result],
@@ -73,7 +72,7 @@ def add_unlock_chord_task(app):
         j = deps.join_native if deps.supports_native_join else deps.join
 
         if deps.ready():
-            callback = signature(callback, app=app)
+            callback = maybe_signature(callback, app=app)
             try:
                 with allow_join_result():
                     ret = j(timeout=3.0, propagate=propagate)
@@ -139,7 +138,7 @@ def add_chunk_task(app):
 @connect_on_app_finalize
 def add_group_task(app):
     _app = app
-    from celery.canvas import maybe_signature, signature
+    from celery.canvas import maybe_signature
     from celery.result import result_from_tuple
 
     class Group(app.Task):
@@ -153,13 +152,8 @@ def add_group_task(app):
             app = self.app
             result = result_from_tuple(result, app)
             # any partial args are added to all tasks in the group
-            taskit = (signature(task, app=app).clone(partial_args)
+            taskit = (maybe_signature(task, app=app).clone(partial_args)
                       for i, task in enumerate(tasks))
-            if self.request.is_eager or app.conf.CELERY_ALWAYS_EAGER:
-                return app.GroupResult(
-                    result.id,
-                    [stask.apply(group_id=group_id) for stask in taskit],
-                )
             with app.producer_or_acquire() as pub:
                 [stask.apply_async(group_id=group_id, producer=pub,
                                    add_to_parent=False) for stask in taskit]
@@ -167,48 +161,11 @@ def add_group_task(app):
             if add_to_parent and parent:
                 parent.add_trail(result)
             return result
-
-        def prepare(self, options, tasks, args, **kwargs):
-            options['group_id'] = group_id = (
-                options.setdefault('task_id', uuid()))
-
-            def prepare_member(task):
-                task = maybe_signature(task, app=self.app)
-                task.options['group_id'] = group_id
-                return task, task.freeze()
-
-            try:
-                tasks, res = list(zip(
-                    *[prepare_member(task) for task in tasks]
-                ))
-            except ValueError:  # tasks empty
-                tasks, res = [], []
-            return (tasks, self.app.GroupResult(group_id, res), group_id, args)
-
-        def apply_async(self, partial_args=(), kwargs={}, **options):
-            if self.app.conf.CELERY_ALWAYS_EAGER:
-                return self.apply(partial_args, kwargs, **options)
-            tasks, result, gid, args = self.prepare(
-                options, args=partial_args, **kwargs
-            )
-            super(Group, self).apply_async((
-                list(tasks), result.as_tuple(), gid, args), **options
-            )
-            return result
-
-        def apply(self, args=(), kwargs={}, **options):
-            return super(Group, self).apply(
-                self.prepare(options, args=args, **kwargs),
-                **options).get()
     return Group
 
 
 @connect_on_app_finalize
 def add_chain_task(app):
-    from celery.canvas import (
-        Signature, chain, chord, group, maybe_signature, maybe_unroll_group,
-    )
-
     _app = app
 
     class Chain(app.Task):
@@ -217,85 +174,6 @@ def add_chain_task(app):
         accept_magic_kwargs = False
         _decorated = True
 
-        def prepare_steps(self, args, tasks):
-            app = self.app
-            steps = deque(tasks)
-            next_step = prev_task = prev_res = None
-            tasks, results = [], []
-            i = 0
-            while steps:
-                # First task get partial args from chain.
-                task = maybe_signature(steps.popleft(), app=app)
-                task = task.clone() if i else task.clone(args)
-                res = task.freeze()
-                i += 1
-
-                if isinstance(task, group):
-                    task = maybe_unroll_group(task)
-                if isinstance(task, chain):
-                    # splice the chain
-                    steps.extendleft(reversed(task.tasks))
-                    continue
-
-                elif isinstance(task, group) and steps and \
-                        not isinstance(steps[0], group):
-                    # automatically upgrade group(..) | s to chord(group, s)
-                    try:
-                        next_step = steps.popleft()
-                        # for chords we freeze by pretending it's a normal
-                        # task instead of a group.
-                        res = Signature.freeze(next_step)
-                        task = chord(task, body=next_step, task_id=res.task_id)
-                    except IndexError:
-                        pass  # no callback, so keep as group
-                if prev_task:
-                    # link previous task to this task.
-                    prev_task.link(task)
-                    # set the results parent attribute.
-                    if not res.parent:
-                        res.parent = prev_res
-
-                if not isinstance(prev_task, chord):
-                    results.append(res)
-                    tasks.append(task)
-                prev_task, prev_res = task, res
-
-            return tasks, results
-
-        def apply_async(self, args=(), kwargs={}, group_id=None, chord=None,
-                        task_id=None, link=None, link_error=None, **options):
-            if self.app.conf.CELERY_ALWAYS_EAGER:
-                return self.apply(args, kwargs, **options)
-            options.pop('publisher', None)
-            tasks, results = self.prepare_steps(args, kwargs['tasks'])
-            result = results[-1]
-            if group_id:
-                tasks[-1].set(group_id=group_id)
-            if chord:
-                tasks[-1].set(chord=chord)
-            if task_id:
-                tasks[-1].set(task_id=task_id)
-                result = tasks[-1].type.AsyncResult(task_id)
-            # make sure we can do a link() and link_error() on a chain object.
-            if link:
-                tasks[-1].set(link=link)
-            # and if any task in the chain fails, call the errbacks
-            if link_error:
-                for task in tasks:
-                    task.set(link_error=link_error)
-            tasks[0].apply_async(**options)
-            return result
-
-        def apply(self, args=(), kwargs={}, signature=maybe_signature,
-                  **options):
-            app = self.app
-            last, fargs = None, args  # fargs passed to first task only
-            for task in kwargs['tasks']:
-                res = signature(task, app=app).clone(fargs).apply(
-                    last and (last.get(), ),
-                )
-                res.parent, last, fargs = last, res, None
-            return last
     return Chain
 
 
@@ -304,10 +182,9 @@ def add_chord_task(app):
     """Every chord is executed in a dedicated task, so that the chord
     can be used as a signature, and this generates the task
     responsible for that."""
-    from celery import group
+    from celery import group, chord as _chord
     from celery.canvas import maybe_signature
     _app = app
-    default_propagate = app.conf.CELERY_CHORD_PROPAGATES
 
     class Chord(app.Task):
         app = _app
@@ -320,53 +197,13 @@ def add_chord_task(app):
                 countdown=1, max_retries=None, propagate=None,
                 eager=False, **kwargs):
             app = self.app
-            propagate = default_propagate if propagate is None else propagate
-            group_id = uuid()
-
             # - convert back to group if serialized
             tasks = header.tasks if isinstance(header, group) else header
             header = group([
-                maybe_signature(s, app=app).clone() for s in tasks
+                maybe_signature(s, app=app) for s in tasks
             ], app=self.app)
-            # - eager applies the group inline
-            if eager:
-                return header.apply(args=partial_args, task_id=group_id)
-
-            body.setdefault('chord_size', len(header.tasks))
-            results = header.freeze(group_id=group_id, chord=body).results
-
-            return self.backend.apply_chord(
-                header, partial_args, group_id,
-                body, interval=interval, countdown=countdown,
-                max_retries=max_retries, propagate=propagate, result=results,
-            )
-
-        def apply_async(self, args=(), kwargs={}, task_id=None,
-                        group_id=None, chord=None, **options):
-            app = self.app
-            if app.conf.CELERY_ALWAYS_EAGER:
-                return self.apply(args, kwargs, **options)
-            header = kwargs.pop('header')
-            body = kwargs.pop('body')
-            header, body = (maybe_signature(header, app=app),
-                            maybe_signature(body, app=app))
-            # forward certain options to body
-            if chord is not None:
-                body.options['chord'] = chord
-            if group_id is not None:
-                body.options['group_id'] = group_id
-            [body.link(s) for s in options.pop('link', [])]
-            [body.link_error(s) for s in options.pop('link_error', [])]
-            body_result = body.freeze(task_id)
-            parent = super(Chord, self).apply_async((header, body, args),
-                                                    kwargs, **options)
-            body_result.parent = parent
-            return body_result
-
-        def apply(self, args=(), kwargs={}, propagate=True, **options):
-            body = kwargs['body']
-            res = super(Chord, self).apply(args, dict(kwargs, eager=True),
-                                           **options)
-            return maybe_signature(body, app=self.app).apply(
-                args=(res.get(propagate=propagate).get(), ))
+            body = maybe_signature(body, app=app)
+            ch = _chord(header, body)
+            return ch.run(header, body, partial_args, app, interval,
+                          countdown, max_retries, propagate, **kwargs)
     return Chord

celery/app/task.py

@@ -93,6 +93,8 @@ class Context(object):
     headers = None
     delivery_info = None
     reply_to = None
+    root_id = None
+    parent_id = None
     correlation_id = None
     taskset = None   # compat alias to group
     group = None

celery/canvas.py

@@ -12,6 +12,7 @@
 """
 from __future__ import absolute_import
 
+from collections import deque
 from copy import deepcopy
 from functools import partial as _partial, reduce
 from operator import itemgetter
@@ -19,7 +20,7 @@ from itertools import chain as _chain
 
 from kombu.utils import cached_property, fxrange, kwdict, reprcall, uuid
 
-from celery._state import current_app
+from celery._state import current_app, get_current_worker_task
 from celery.utils.functional import (
     maybe_list, is_list, regen,
     chunks as _chunks,
@@ -194,12 +195,13 @@ class Signature(dict):
         return s
     partial = clone
 
-    def freeze(self, _id=None, group_id=None, chord=None):
+    def freeze(self, _id=None, group_id=None, chord=None, root_id=None):
         opts = self.options
         try:
             tid = opts['task_id']
         except KeyError:
             tid = opts['task_id'] = _id or uuid()
+        root_id = opts.setdefault('root_id', root_id)
         if 'reply_to' not in opts:
             opts['reply_to'] = self.app.oid
         if group_id:
@@ -348,6 +350,99 @@ class chain(Signature):
         if self.tasks:
             return self.apply_async(args, kwargs)
 
+    def apply_async(self, args=(), kwargs={}, group_id=None, chord=None,
+                    task_id=None, link=None, link_error=None,
+                    publisher=None, root_id=None, **options):
+        app = self.app
+        if app.conf.CELERY_ALWAYS_EAGER:
+            return self.apply(args, kwargs, **options)
+        tasks, results = self.prepare_steps(
+            args, self.tasks, root_id, link_error,
+        )
+        if not results:
+            return
+        result = results[-1]
+        last_task = tasks[-1]
+        if group_id:
+            last_task.set(group_id=group_id)
+        if chord:
+            last_task.set(chord=chord)
+        if task_id:
+            last_task.set(task_id=task_id)
+            result = last_task.type.AsyncResult(task_id)
+        # make sure we can do a link() and link_error() on a chain object.
+        if link:
+            tasks[-1].set(link=link)
+        tasks[0].apply_async(**options)
+        return result
+
+    def prepare_steps(self, args, tasks,
+                      root_id=None, link_error=None, app=None):
+        app = app or self.app
+        steps = deque(tasks)
+        next_step = prev_task = prev_res = None
+        tasks, results = [], []
+        i = 0
+        while steps:
+            task = steps.popleft()
+            if not i:  # first task
+                # first task gets partial args from chain
+                task = task.clone(args)
+                res = task.freeze(root_id=root_id)
+                root_id = res.id if root_id is None else root_id
+            else:
+                task = task.clone()
+                res = task.freeze(root_id=root_id)
+            i += 1
+
+            if isinstance(task, group):
+                task = maybe_unroll_group(task)
+
+            if isinstance(task, chain):
+                # splice the chain
+                steps.extendleft(reversed(task.tasks))
+                continue
+            elif isinstance(task, group) and steps and \
+                    not isinstance(steps[0], group):
+                # automatically upgrade group(...) | s to chord(group, s)
+                try:
+                    next_step = steps.popleft()
+                    # for chords we freeze by pretending it's a normal
+                    # signature instead of a group.
+                    res = Signature.freeze(next_step)
+                    task = chord(
+                        task, body=next_step,
+                        task_id=res.task_id, root_id=root_id,
+                    )
+                except IndexError:
+                    pass  # no callback, so keep as group.
+
+            if prev_task:
+                # link previous task to this task.
+                prev_task.link(task)
+                # set AsyncResult.parent
+                if not res.parent:
+                    res.parent = prev_res
+
+            if link_error:
+                task.set(link_error=link_error)
+
+            if not isinstance(prev_task, chord):
+                results.append(res)
+                tasks.append(task)
+            prev_task, prev_res = task, res
+
+        return tasks, results
+
+    def apply(self, args=(), kwargs={}, **options):
+        last, fargs = None, args
+        for task in self.tasks:
+            res = task.clone(fargs).apply(
+                last and (last.get(), ), **options
+            )
+            res.parent, last, fargs = last, res, None
+        return last
+
     @classmethod
     def from_dict(self, d, app=None):
         tasks = d['kwargs']['tasks']
@@ -357,11 +452,14 @@ class chain(Signature):
         return chain(*d['kwargs']['tasks'], app=app, **kwdict(d['options']))
 
     @property
-    def type(self):
-        try:
-            return self._type or self.tasks[0].type.app.tasks['celery.chain']
-        except KeyError:
-            return self.app.tasks['celery.chain']
+    def app(self):
+        app = self._app
+        if app is None:
+            try:
+                app = self.tasks[0]._app
+            except (KeyError, IndexError):
+                pass
+        return app or current_app
 
     def __repr__(self):
         return ' | '.join(repr(t) for t in self.tasks)
@@ -452,11 +550,6 @@ def _maybe_group(tasks):
     return tasks
 
 
-def _maybe_clone(tasks, app):
-    return [s.clone() if isinstance(s, Signature) else signature(s, app=app)
-            for s in tasks]
-
-
 @Signature.register_type
 class group(Signature):
 
@@ -477,14 +570,58 @@ class group(Signature):
                 task['args'] = task._merge(d['args'])[0]
         return group(tasks, app=app, **kwdict(d['options']))
 
-    def apply_async(self, args=(), kwargs=None, add_to_parent=True, **options):
-        tasks = _maybe_clone(self.tasks, app=self._app)
-        if not tasks:
+    def _prepared(self, tasks, partial_args, group_id, root_id):
+        for task in tasks:
+            task = task.clone(partial_args)
+            yield task, task.freeze(group_id=group_id, root_id=root_id)
+
+    def _apply_tasks(self, tasks, producer=None, app=None, **options):
+        app = app or self.app
+        with app.producer_or_acquire(producer) as producer:
+            for sig, res in tasks:
+                sig.apply_async(producer=producer, add_to_parent=False,
+                                **options)
+                yield res
+
+    def _freeze_gid(self, options):
+        # remove task_id and use that as the group_id,
+        # if we don't remove it then every task will have the same id...
+        options = dict(self.options, **options)
+        options['group_id'] = group_id = (
+            options.pop('task_id', uuid()))
+        return options, group_id, options.get('root_id')
+
+    def apply_async(self, args=(), kwargs=None, add_to_parent=True,
+                    producer=None, **options):
+        app = self.app
+        if app.conf.CELERY_ALWAYS_EAGER:
+            return self.apply(args, kwargs, **options)
+        if not self.tasks:
             return self.freeze()
-        type = self.type
+
+        options, group_id, root_id = self._freeze_gid(options)
+        tasks = self._prepared(self.tasks, args, group_id, root_id)
+        result = self.app.GroupResult(
+            group_id, list(self._apply_tasks(tasks, producer, app, **options)),
+        )
+        parent_task = get_current_worker_task()
+        if add_to_parent and parent_task:
+            parent_task.add_trail(result)
+        return result
+
         return type(*type.prepare(dict(self.options, **options), tasks, args),
                     add_to_parent=add_to_parent)
 
+    def apply(self, args=(), kwargs={}, **options):
+        app = self.app
+        if not self.tasks:
+            return self.freeze()  # empty group returns GroupResult
+        options, group_id, root_id = self._freeze_gid(options)
+        tasks = self._prepared(self.tasks, args, group_id, root_id)
+        return app.GroupResult(group_id, [
+            sig.apply(**options) for sig, _ in tasks
+        ])
+
     def set_immutable(self, immutable):
         for task in self.tasks:
             task.set_immutable(immutable)
@@ -498,15 +635,10 @@ class group(Signature):
         sig = sig.clone().set(immutable=True)
         return self.tasks[0].link_error(sig)
 
-    def apply(self, *args, **kwargs):
-        if not self.tasks:
-            return self.freeze()  # empty group returns GroupResult
-        return Signature.apply(self, *args, **kwargs)
-
     def __call__(self, *partial_args, **options):
         return self.apply_async(partial_args, **options)
 
-    def freeze(self, _id=None, group_id=None, chord=None):
+    def freeze(self, _id=None, group_id=None, chord=None, root_id=None):
         opts = self.options
         try:
             gid = opts['task_id']
@@ -516,10 +648,13 @@ class group(Signature):
             opts['group_id'] = group_id
         if chord:
             opts['chord'] = group_id
+        root_id = opts.setdefault('root_id', root_id)
         new_tasks, results = [], []
         for task in self.tasks:
             task = maybe_signature(task, app=self._app).clone()
-            results.append(task.freeze(group_id=group_id, chord=chord))
+            results.append(task.freeze(
+                group_id=group_id, chord=chord, root_id=root_id,
+            ))
             new_tasks.append(task)
         self.tasks = self.kwargs['tasks'] = new_tasks
         return self.app.GroupResult(gid, results)
@@ -538,14 +673,14 @@ class group(Signature):
         return repr(self.tasks)
 
     @property
-    def type(self):
-        if self._type:
-            return self._type
-        # taking the app from the first task in the list, there may be a
-        # better solution for this, e.g. to consolidate tasks with the same
-        # app and apply them in batches.
-        app = self._app if self._app else self.tasks[0].type.app
-        return app.tasks[self['task']]
+    def app(self):
+        app = self._app
+        if app is None:
+            try:
+                app = self.tasks[0]._app
+            except (KeyError, IndexError):
+                pass
+        return app if app is not None else current_app
 
 
 @Signature.register_type
@@ -560,8 +695,8 @@ class chord(Signature):
         )
         self.subtask_type = 'chord'
 
-    def freeze(self, _id=None, group_id=None, chord=None):
-        return self.body.freeze(_id, group_id=group_id, chord=chord)
+    def freeze(self, *args, **kwargs):
+        return self.body.freeze(*args, **kwargs)
 
     @classmethod
     def from_dict(self, d, app=None):
@@ -574,20 +709,14 @@ class chord(Signature):
         # than manually popping things off.
         return (header, body), kwargs
 
-    @property
-    def type(self):
-        if self._type:
-            return self._type
-        # we will be able to fix this mess in 3.2 when we no longer
-        # require an actual task implementation for chord/group
-        if self._app:
-            app = self._app
-        else:
-            try:
-                app = self.tasks[0].type.app
-            except IndexError:
-                app = self.body.type.app
-        return app.tasks['celery.chord']
+    @cached_property
+    def app(self):
+        app = self._app
+        if app is None:
+            app = self.tasks[0]._app
+            if app is None:
+                app = self.body._app
+        return app if app is not None else current_app
 
     def apply_async(self, args=(), kwargs={}, task_id=None,
                     producer=None, publisher=None, connection=None,
@@ -595,14 +724,41 @@ class chord(Signature):
         body = kwargs.get('body') or self.kwargs['body']
         kwargs = dict(self.kwargs, **kwargs)
         body = body.clone(**options)
+        app = self.app
+        tasks = (self.tasks.clone() if isinstance(self.tasks, group)
+                 else group(self.tasks))
+        if app.conf.CELERY_ALWAYS_EAGER:
+            return self.apply((), kwargs,
+                              body=body, task_id=task_id, **options)
+        return self.run(tasks, body, args, task_id=task_id, **options)
+
+    def apply(self, args=(), kwargs={}, propagate=True, body=None, **options):
+        body = self.body if body is None else body
+        tasks = (self.tasks.clone() if isinstance(self.tasks, group)
+                 else group(self.tasks))
+        return body.apply(
+            args=(tasks.apply().get(propagate=propagate), ),
+        )
 
-        _chord = self.type
-        if _chord.app.conf.CELERY_ALWAYS_EAGER:
-            return self.apply((), kwargs, task_id=task_id, **options)
-        res = body.freeze(task_id)
-        parent = _chord(self.tasks, body, args, **options)
-        res.parent = parent
-        return res
+    def run(self, header, body, partial_args, app=None, interval=None,
+            countdown=1, max_retries=None, propagate=None, eager=False,
+            task_id=None, **options):
+        app = app or self.app
+        propagate = (app.conf.CELERY_CHORD_PROPAGATES
+                     if propagate is None else propagate)
+        group_id = uuid()
+        root_id = body.options.get('root_id')
+        body.setdefault('chord_size', len(header.tasks))
+        results = header.freeze(
+            group_id=group_id, chord=body, root_id=root_id).results
+        bodyres = body.freeze(task_id, root_id=root_id)
+
+        parent = app.backend.apply_chord(
+            header, partial_args, group_id, body,
+            interval=interval, countdown=countdown,
+            max_retries=max_retries, propagate=propagate, result=results)
+        bodyres.parent = parent
+        return bodyres
 
     def __call__(self, body=None, **options):
         return self.apply_async((), {'body': body} if body else {}, **options)

celery/tests/app/test_builtins.py

@@ -136,18 +136,18 @@ class test_chain(BuiltinsCase):
 
     def test_group_to_chord(self):
         c = (
-            group(self.add.s(i, i) for i in range(5)) |
+            group([self.add.s(i, i) for i in range(5)], app=self.app) |
             self.add.s(10) |
             self.add.s(20) |
             self.add.s(30)
         )
-        tasks, _ = c.type.prepare_steps((), c.tasks)
+        tasks, _ = c.prepare_steps((), c.tasks)
         self.assertIsInstance(tasks[0], chord)
         self.assertTrue(tasks[0].body.options['link'])
         self.assertTrue(tasks[0].body.options['link'][0].options['link'])
 
         c2 = self.add.s(2, 2) | group(self.add.s(i, i) for i in range(10))
-        tasks2, _ = c2.type.prepare_steps((), c2.tasks)
+        tasks2, _ = c2.prepare_steps((), c2.tasks)
         self.assertIsInstance(tasks2[1], group)
 
     def test_apply_options(self):
@@ -158,7 +158,7 @@ class test_chain(BuiltinsCase):
                 return self
 
         def s(*args, **kwargs):
-            return static(self.add, args, kwargs, type=self.add)
+            return static(self.add, args, kwargs, type=self.add, app=self.app)
 
         c = s(2, 2) | s(4, 4) | s(8, 8)
         r1 = c.apply_async(task_id='some_id')
@@ -196,18 +196,16 @@ class test_chord(BuiltinsCase):
     def test_forward_options(self):
         body = self.xsum.s()
         x = chord([self.add.s(i, i) for i in range(10)], body=body)
-        x._type = Mock()
-        x._type.app.conf.CELERY_ALWAYS_EAGER = False
+        x.run = Mock(name='chord.run(x)')
         x.apply_async(group_id='some_group_id')
-        self.assertTrue(x._type.called)
-        resbody = x._type.call_args[0][1]
+        self.assertTrue(x.run.called)
+        resbody = x.run.call_args[0][1]
         self.assertEqual(resbody.options['group_id'], 'some_group_id')
         x2 = chord([self.add.s(i, i) for i in range(10)], body=body)
-        x2._type = Mock()
-        x2._type.app.conf.CELERY_ALWAYS_EAGER = False
+        x2.run = Mock(name='chord.run(x2)')
         x2.apply_async(chord='some_chord_id')
-        self.assertTrue(x2._type.called)
-        resbody = x2._type.call_args[0][1]
+        self.assertTrue(x2.run.called)
+        resbody = x2.run.call_args[0][1]
         self.assertEqual(resbody.options['chord'], 'some_chord_id')
 
     def test_apply_eager(self):

celery/tests/tasks/test_chord.py

@@ -205,7 +205,7 @@ class test_chord(ChordCase):
         m = Mock()
         m.app.conf.CELERY_ALWAYS_EAGER = False
         m.AsyncResult = AsyncResult
-        prev, chord._type = chord._type, m
+        prev, chord.run = chord.run, m
         try:
             x = chord(self.add.s(i, i) for i in range(10))
             body = self.add.s(2)
@@ -214,9 +214,9 @@ class test_chord(ChordCase):
             # does not modify original signature
             with self.assertRaises(KeyError):
                 body.options['task_id']
-            self.assertTrue(chord._type.called)
+            self.assertTrue(chord.run.called)
         finally:
-            chord._type = prev
+            chord.run = prev
 
 
 class test_Chord_task(ChordCase):
@@ -227,7 +227,7 @@ class test_Chord_task(ChordCase):
         self.app.backend.cleanup.__name__ = 'cleanup'
         Chord = self.app.tasks['celery.chord']
 
-        body = dict()
+        body = self.add.signature()
         Chord(group(self.add.signature((i, i)) for i in range(5)), body)
         Chord([self.add.signature((j, j)) for j in range(5)], body)
         self.assertEqual(self.app.backend.apply_chord.call_count, 2)

docs/internals/protov2.rst

@@ -50,6 +50,7 @@ Notes
 
 - ``correlation_id`` replaces ``task_id`` field.
 
+- ``root_id`` and ``parent_id`` fields helps keep track of workflows.
 
 - ``c_shadow`` lets you specify a different name for logs, monitors
   can be used for e.g. meta tasks that calls any function::
@@ -115,6 +116,8 @@ Definition
         'chord': (uuid)chord_id,
         'retries': (int)retries,
         'timelimit': (tuple)(soft, hard),
+        'root_id': (uuid)root_id,
+        'parent_id': (uuid)parent_id,
     }
 
     body = (args, kwargs)