Comment passez-vous une référence de file d'attente à une fonction gérée par pool.map_async()?

Question

Je veux un processus de longue durée pour retourner sa progression sur une file d'attente (ou quelque chose de similaire) que je vais alimenter dans une boîte de dialogue de progression. J'ai aussi besoin du résultat lorsque le processus est terminé. Un exemple de test échoue ici avec un RuntimeError: Queue objects should only be shared between processes through inheritance.

import multiprocessing, time 

def task(args): 
    count = args[0] 
    queue = args[1] 
    for i in xrange(count): 
     queue.put("%d mississippi" % i) 
    return "Done" 

def main(): 
    q = multiprocessing.Queue() 
    pool = multiprocessing.Pool() 
    result = pool.map_async(task, [(x, q) for x in range(10)]) 
    time.sleep(1) 
    while not q.empty(): 
     print q.get() 
    print result.get() 

if __name__ == "__main__": 
    main() 

Je suis en mesure d'obtenir ce travail en utilisant des objets individuels de processus (où je suis alowed pour passer une référence file d'attente), mais je n'ai pas une piscine pour gérer les nombreux processus que je veux lancer. Des conseils sur un meilleur modèle pour cela?

Answer 1

Le code suivant semble fonctionner:

import multiprocessing, time 

def task(args): 
    count = args[0] 
    queue = args[1] 
    for i in xrange(count): 
     queue.put("%d mississippi" % i) 
    return "Done" 


def main(): 
    manager = multiprocessing.Manager() 
    q = manager.Queue() 
    pool = multiprocessing.Pool() 
    result = pool.map_async(task, [(x, q) for x in range(10)]) 
    time.sleep(1) 
    while not q.empty(): 
     print q.get() 
    print result.get() 

if __name__ == "__main__": 
    main() 

Notez que la file d'attente est obtenu à partir d'un manager.Queue() plutôt que multiprocessing.Queue(). Merci Alex de m'avoir indiqué dans cette direction.

Answer 2

Faire q mondiaux travaux ...:

import multiprocessing, time 

q = multiprocessing.Queue() 

def task(count): 
    for i in xrange(count): 
     q.put("%d mississippi" % i) 
    return "Done" 

def main(): 
    pool = multiprocessing.Pool() 
    result = pool.map_async(task, range(10)) 
    time.sleep(1) 
    while not q.empty(): 
     print q.get() 
    print result.get() 

if __name__ == "__main__": 
    main() 

Si vous avez besoin de plusieurs files d'attente, par exemple pour éviter de mélanger la progression des différents processus de pool, une liste globale de files d'attente devrait fonctionner (bien sûr, chaque processus devra alors savoir quel index dans la liste à utiliser, mais c'est OK pour passer en argument; -).