Merge branch '3.0'

Conflicts:
	celery/app/builtins.py

Author: ask

Changelog

@@ -45,6 +45,21 @@ If you're looking for versions prior to 3.x you should see :ref:`history`.
     - Pool now sets a ``current_process().index`` attribute that can be used to create
       as many log files as there are processes in the pool.
 
+- Canvas: chord/group/chain no longer modifies the state when called
+
+    Previously calling a chord/group/chain would modify the ids of subtasks
+    so that::
+
+        >>> c = chord([add.s(2, 2), add.s(4, 4)], xsum.s())
+        >>> c()
+        >>> c() <-- call again
+
+    at the second time the ids for the tasks would be the same as in the
+    previous invocation.  This is now fixed, so that calling a subtask
+    won't mutate any options.
+
+- Canvas: Chaining a chord to another task now works.
+
 - Worker: Fixed a bug where the request stack could be corrupted if
   relative imports are used.
 
@@ -53,9 +68,6 @@ If you're looking for versions prior to 3.x you should see :ref:`history`.
 
     Fix contributed by Sam Cooke.
 
-- ``subtask.clone`` now deepcopies options so that original subtask
-  is not modified when used in chains/groups etc.
-
 - Because of many bugs the fast local optimization has been disabled,
   and can only be enabled by setting the :envvar:`USE_FAST_LOCALS` attribute.
 

celery/app/builtins.py

@@ -213,8 +213,8 @@ def prepare_steps(self, args, tasks):
                         next_step = steps.popleft()
                     except IndexError:
                         next_step = None
-                if next_step is not None:
-                    task = chord(task, body=next_step, task_id=tid)
+                    if next_step is not None:
+                        task = chord(task, body=next_step, task_id=tid)
                 if prev_task:
                     # link previous task to this task.
                     prev_task.link(task)
@@ -276,8 +276,8 @@ def run(self, header, body, partial_args=(), interval=1,
             prepare_member = self._prepare_member
 
             # - convert back to group if serialized
-            if not isinstance(header, group):
-                header = group([maybe_subtask(t) for t in  header])
+            tasks = header.tasks if isinstance(header, group) else header
+            header = group([maybe_subtask(s).clone() for s in tasks])
             # - eager applies the group inline
             if eager:
                 return header.apply(args=partial_args, task_id=group_id)

celery/canvas.py

@@ -353,11 +353,11 @@ def _unpack_args(header=None, body=None, **kwargs):
 
     def __call__(self, body=None, **kwargs):
         _chord = self.Chord
-        body = self.kwargs['body'] = body or self.kwargs['body']
+        body = (body or self.kwargs['body']).clone()
         if _chord.app.conf.CELERY_ALWAYS_EAGER:
             return self.apply((), kwargs)
         callback_id = body.options.setdefault('task_id', uuid())
-        _chord(**dict(self.kwargs, **kwargs))
+        _chord(**dict(self.kwargs, body=body, **kwargs))
         return _chord.AsyncResult(callback_id)
 
     def clone(self, *args, **kwargs):