Come si passa un riferimento alla coda a una funzione gestita da pool.map_async()?

Question

Voglio un processo a esecuzione prolungata per restituire il suo avanzamento su una coda (o qualcosa di simile) che alimenterò a una finestra di dialogo della barra di avanzamento. Ho anche bisogno del risultato quando il processo è completato. Un esempio di test qui non riesce con 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() 

Sono stato in grado di ottenere questo al lavoro utilizzando oggetti singoli Process (dove io sto alowed a passare un riferimento coda), ma poi non ho un pool per gestire i molti processi che voglio lanciare. Qualche consiglio su un modello migliore per questo?

Answer 1

Il seguente codice sembra funzionare:

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() 

Si noti che la coda è ottenuto da una manager.Queue() piuttosto che multiprocessing.Queue(). Grazie Alex per avermi indicato in questa direzione.

Answer 2

Fare qglobali opere ...:

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() 

Se avete bisogno di più code, per esempio per evitare di confondere i progressi dei vari processi del pool, dovrebbe funzionare un elenco globale di code (ovviamente, ogni processo dovrà quindi sapere quale indice nella lista da utilizzare, ma va bene passare come argomento;).