Method _apply_tasks

celery/canvas.py:1742–1804

Run all the tasks in the group. This is used by :meth:`apply_async` to run all the tasks in the group and return a generator of their results. Arguments: tasks (list): List of tasks in the group. producer (Producer): The producer to use to publish th

(self, tasks, producer=None, app=None, p=None,
                     add_to_parent=None, chord=None,
                     args=None, kwargs=None, group_index=None, **options)

                yield task, task.freeze(group_id=group_id, root_id=root_id, group_index=index), group_id

    def _apply_tasks(self, tasks, producer=None, app=None, p=None,
                     add_to_parent=None, chord=None,
                     args=None, kwargs=None, group_index=None, **options):
        """Run all the tasks in the group.

        This is used by :meth:`apply_async` to run all the tasks in the group
        and return a generator of their results.

        Arguments:
            tasks (list): List of tasks in the group.
            producer (Producer): The producer to use to publish the tasks.
            app (Celery): The Celery app instance.
            p (barrier): Barrier object to synchronize the tasks results.
            args (list): List of arguments to be prepended to
                the arguments of each task.
            kwargs (dict): Dict of keyword arguments to be merged with
                the keyword arguments of each task.
            **options (dict): Options to be merged with the options of each task.

        Returns:
            generator: A generator for the AsyncResult of the tasks in the group.
        """
        # pylint: disable=redefined-outer-name
        #   XXX chord is also a class in outer scope.
        app = app or self.app
        with app.producer_or_acquire(producer) as producer:
            # Iterate through tasks two at a time. If tasks is a generator,
            # we are able to tell when we are at the end by checking if
            # next_task is None.  This enables us to set the chord size
            # without burning through the entire generator.  See #3021.
            chord_size = 0
            tasks_shifted, tasks = itertools.tee(tasks)
            next(tasks_shifted, None)
            next_task = next(tasks_shifted, None)

            for task_index, current_task in enumerate(tasks):
                # We expect that each task must be part of the same group which
                # seems sensible enough. If that's somehow not the case we'll
                # end up messing up chord counts and there are all sorts of
                # awful race conditions to think about. We'll hope it's not!
                sig, res, group_id = current_task
                chord_obj = chord if chord is not None else sig.options.get("chord")
                # We need to check the chord size of each contributing task so
                # that when we get to the final one, we can correctly set the
                # size in the backend and the chord can be sensible completed.
                chord_size += _chord._descend(sig)
                if chord_obj is not None and next_task is None:
                    # Per above, sanity check that we only saw one group
                    app.backend.set_chord_size(group_id, chord_size)
                sig.apply_async(producer=producer, add_to_parent=False,
                                chord=chord_obj, args=args, kwargs=kwargs,
                                **options)
                # adding callback to result, such that it will gradually
                # fulfill the barrier.
                #
                # Using barrier.add would use result.then, but we need
                # to add the weak argument here to only create a weak
                # reference to the object.