Partage d'une file d'attente de résultats entre plusieurs processus

La documentation du multiprocessingmodule montre comment transmettre une file d'attente à un processus démarré avec multiprocessing.Process. Mais comment puis-je partager une file d'attente avec des processus de travail asynchrones démarrés avec apply_async? Je n'ai pas besoin de jointure dynamique ou de quoi que ce soit d'autre, juste un moyen pour les travailleurs de rapporter (à plusieurs reprises) leurs résultats à la base.

import multiprocessing
def worker(name, que):
    que.put("%d is done" % name)

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=3)
    q = multiprocessing.Queue()
    workers = pool.apply_async(worker, (33, q))

Cela échoue avec: RuntimeError: Queue objects should only be shared between processes through inheritance. Je comprends ce que cela signifie, et je comprends les conseils d'hériter plutôt que d'exiger le décapage / décapage (et toutes les restrictions Windows spéciales). Mais comment puis - je passer la file d'attente d'une manière qui fonctionne? Je ne trouve pas d'exemple et j'ai essayé plusieurs alternatives qui ont échoué de différentes manières. Aidez-moi, s'il vous plaît?

Essayez d'utiliser le multiprocessing.Manager pour gérer votre file d'attente et la rendre également accessible à différents travailleurs.

import multiprocessing
def worker(name, que):
    que.put("%d is done" % name)

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=3)
    m = multiprocessing.Manager()
    q = m.Queue()
    workers = pool.apply_async(worker, (33, q))

multiprocessing.Poola déjà une file d'attente de résultats partagée, il n'est pas nécessaire d'impliquer en plus un fichier Manager.Queue. Manager.Queueest une queue.Queue(file d'attente multithreading) sous le capot, située sur un processus serveur séparé et exposée via des proxys. Cela ajoute une surcharge supplémentaire par rapport à la file d'attente interne du pool. Contrairement au fait de s'appuyer sur la gestion native des résultats de Pool, les résultats dans le Manager.Queuene sont pas non plus garantis d'être commandés.

Les processus de travail ne sont pas démarrés avec .apply_async(), cela se produit déjà lorsque vous instanciez Pool. Ce qui est commencé lorsque vous appelez pool.apply_async()est un nouveau « job ». Les processus de travail de Pool exécutent la multiprocessing.pool.workerfonction sous le capot. Cette fonction prend en charge le traitement des nouvelles «tâches» transférées via le pool interne Pool._inqueueet le renvoi des résultats au parent via le Pool._outqueue. Votre spécifié funcsera exécuté dans multiprocessing.pool.worker. funcn'a qu'à returnquelque chose et le résultat sera automatiquement renvoyé au parent.

.apply_async() immédiatement (de manière asynchrone) renvoie un AsyncResultobjet (alias pour ApplyResult). Vous devez appeler .get()(bloque) sur cet objet pour recevoir le résultat réel. Une autre option serait d'enregistrer une fonction de rappel , qui est déclenchée dès que le résultat est prêt.

from multiprocessing import Pool

def busy_foo(i):
    """Dummy function simulating cpu-bound work."""
    for _ in range(int(10e6)):  # do stuff
        pass
    return i

if __name__ == '__main__':

    with Pool(4) as pool:
        print(pool._outqueue)  # DEMO
        results = [pool.apply_async(busy_foo, (i,)) for i in range(10)]
        # `.apply_async()` immediately returns AsyncResult (ApplyResult) object
        print(results[0])  # DEMO
        results = [res.get() for res in results]
        print(f'result: {results}')       

Exemple de sortie:

<multiprocessing.queues.SimpleQueue object at 0x7fa124fd67f0>
<multiprocessing.pool.ApplyResult object at 0x7fa12586da20>
result: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Remarque: la spécification du timeoutparamètre-pour .get()n'arrêtera pas le traitement réel de la tâche dans le worker, elle débloque uniquement le parent en attente en levant un multiprocessing.TimeoutError.