Wer führt die callback-bei Verwendung apply_async Methode einer multiprocessing-pool?

Es ist in der Tat ein Hinweis in der Dokumentation:

Rückruf sollte sofort vollständig da sonst der thread
verarbeitet die Ergebnisse wird blockiert.

Rückrufe behandelt werden die wichtigsten Verfahren, aber Sie laufen in einer eigenen thread. Beim erstellen einer Pool es tatsächlich schafft ein paar Thread Objekte intern:

class Pool(object):
    Process = Process

    def __init__(self, processes=None, initializer=None, initargs=(),
                 maxtasksperchild=None):
        self._setup_queues()
        self._taskqueue = Queue.Queue()
        self._cache = {}
        ... # stuff we don't care about
        self._worker_handler = threading.Thread(
            target=Pool._handle_workers,
            args=(self, )
            )
        self._worker_handler.daemon = True
        self._worker_handler._state = RUN 
        self._worker_handler.start()

        self._task_handler = threading.Thread(
            target=Pool._handle_tasks,
            args=(self._taskqueue, self._quick_put, self._outqueue,
                  self._pool, self._cache)
            )
        self._task_handler.daemon = True
        self._task_handler._state = RUN 
        self._task_handler.start()

        self._result_handler = threading.Thread(
            target=Pool._handle_results,
            args=(self._outqueue, self._quick_get, self._cache)
            )
        self._result_handler.daemon = True
        self._result_handler._state = RUN
        self._result_handler.start()

Den interessanten thread für uns ist _result_handler; wir kommen zu kurz, warum.

Umschaltgetriebe für einen zweiten, wenn Sie Sie ausführen apply_async wird, wird ein ApplyResult Objekt intern verwalten immer das Ergebnis vom Kind:

def apply_async(self, func, args=(), kwds={}, callback=None):
    assert self._state == RUN
    result = ApplyResult(self._cache, callback)
    self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
    return result

class ApplyResult(object):

    def __init__(self, cache, callback):
        self._cond = threading.Condition(threading.Lock())
        self._job = job_counter.next()
        self._cache = cache
        self._ready = False
        self._callback = callback
        cache[self._job] = self


    def _set(self, i, obj):
        self._success, self._value = obj
        if self._callback and self._success:
            self._callback(self._value)
        self._cond.acquire()
        try:
            self._ready = True
            self._cond.notify()
        finally:
            self._cond.release()
        del self._cache[self._job]

Wie Sie sehen können, die _set Methode ist die eine, die endet eigentlich die Ausführung der callback übergeben, vorausgesetzt, die Aufgabe erfolgreich war. Beachten Sie auch, dass es fügt sich zu einem globalen cache dict am Ende __init__.

Nun, zurück zu den _result_handler thread-Objekt. Das Objekt ruft die _handle_results Funktion, die wie folgt aussieht:

    while 1:
        try:
            task = get()
        except (IOError, EOFError):
            debug('result handler got EOFError/IOError -- exiting')
            return

        if thread._state:
            assert thread._state == TERMINATE
            debug('result handler found thread._state=TERMINATE')
            break

        if task is None:
            debug('result handler got sentinel')
            break

        job, i, obj = task
        try:
            cache[job]._set(i, obj)  # Here is _set (and therefore our callback) being called!
        except KeyError:
            pass

        # More stuff

Es ist eine Schleife, die zieht nur die Ergebnisse von Kindern aus der Warteschlange, findet den Eintrag in cache, und fordert _set führt unser callback. Es ist in der Lage zu laufen, obwohl Sie in einer Schleife, weil es nicht läuft im Haupt-thread.