Method flush

celery/concurrency/asynpool.py:999–1083

(self)

                active_writes.discard(fd)

    def flush(self):
        if self._state == TERMINATE:
            return
        # cancel all tasks that haven't been accepted so that NACK is sent
        # if synack is enabled, otherwise discard them from the cache
        # since they will be redelivered by the broker.
        for job in tuple(self._cache.values()):
            if not job._accepted:
                if self.synack:
                    job._cancel()
                else:
                    job.discard()

        # clear the outgoing buffer as the tasks will be redelivered by
        # the broker anyway.
        if self.outbound_buffer:
            self.outbound_buffer.clear()

        self.maintain_pool()

        try:
            # ...but we must continue writing the payloads we already started
            # to keep message boundaries.
            # The messages may be NACK'ed later if synack is enabled.
            if self._state == RUN:
                # flush outgoing buffers
                intervals = fxrange(0.01, 0.1, 0.01, repeatlast=True)

                # TODO: Rewrite this as a dictionary comprehension once we drop support for Python 3.7
                #       This dict comprehension requires the walrus operator which is only available in 3.8.
                owned_by = {}
                for job in self._cache.values():
                    writer = _get_job_writer(job)
                    if writer is not None:
                        owned_by[writer] = job

                while self._active_writers:
                    writers = list(self._active_writers)
                    for gen in writers:
                        if (gen.__name__ == '_write_job' and
                                gen_not_started(gen)):
                            # hasn't started writing the job so can
                            # discard the task, but we must also remove
                            # it from the Pool._cache.
                            try:
                                job = owned_by[gen]
                            except KeyError:
                                pass
                            else:
                                # removes from Pool._cache
                                job.discard()
                            self._active_writers.discard(gen)
                        else:
                            try:
                                job = owned_by[gen]
                            except KeyError:
                                # Generator not in owned_by — not a _write_job
                                # (e.g. a _write_ack coroutine added by send_ack()).