celery/concurrency/asynpool.py

@@ -596,7 +596,8 @@ def _create_write_handlers(self, hub,
        fileno_to_synq = self._fileno_to_synq
        outbound = self.outbound_buffer
        pop_message = outbound.popleft
        put_message = outbound.append
        append_message = outbound.append
        put_back_message = outbound.appendleft
        all_inqueues = self._all_inqueues
        active_writes = self._active_writes
        active_writers = self._active_writers
@@ -693,12 +694,15 @@ def schedule_writes(ready_fds, curindex=[0]):
                ready_fd = ready_fds[curindex[0] % total]
                if ready_fd in active_writes:
                    # already writing to this fd
                    curindex[0] += 1
                    continue
                if is_fair_strategy and ready_fd in busy_workers:
                    # worker is already busy with another task
                    curindex[0] += 1
                    continue
                if ready_fd not in all_inqueues:
                    hub_remove(ready_fd)
                    curindex[0] += 1
                    continue
                try:
                    job = pop_message()
@@ -711,7 +715,6 @@ def schedule_writes(ready_fds, curindex=[0]):
                    for inqfd in diff(active_writes):
                        hub_remove(inqfd)
                    break

                else:
                    if not job._accepted:  # job not accepted by another worker
                        try:
@@ -722,9 +725,9 @@ def schedule_writes(ready_fds, curindex=[0]):
                            # write was scheduled for this fd but the process
                            # has since exited and the message must be sent to
                            # another process.
                            put_message(job)
                            put_back_message(job)
                            curindex[0] += 1
                            continue
                        curindex[0] += 1
                        cor = _write_job(proc, ready_fd, job)
                        job._writer = ref(cor)
                        mark_write_gen_as_active(cor)
@@ -741,6 +744,7 @@ def schedule_writes(ready_fds, curindex=[0]):
                                raise
                        else:
                            add_writer(ready_fd, cor)
                curindex[0] += 1
        hub.consolidate_callback = schedule_writes

        def send_job(tup):
@@ -752,7 +756,7 @@ def send_job(tup):
            # index 1,0 is the job ID.
            job = get_job(tup[1][0])
            job._payload = buf_t(header), buf_t(body), body_size
            put_message(job)
            append_message(job)
        self._quick_put = send_job

        def on_not_recovering(proc, fd, job):