Posso usare una coda di multiprocessing in una funzione chiamata da Pool.imap?

Question

Sto usando python 2.7 e sto provando a eseguire alcuni compiti pesanti della CPU nei loro processi. Vorrei poter inviare i messaggi al processo principale per tenerlo informato sullo stato corrente del processo. La coda multiprocessing sembra perfetta per questo, ma non riesco a capire come farlo funzionare.

Quindi, questo è il mio esempio di lavoro di base meno l'uso di una coda.

import multiprocessing as mp 
import time 

def f(x): 
    return x*x 

def main(): 
    pool = mp.Pool() 
    results = pool.imap_unordered(f, range(1, 6)) 
    time.sleep(1) 

    print str(results.next()) 

    pool.close() 
    pool.join() 

if __name__ == '__main__': 
    main() 

Ho provato passando la coda in diversi modi, e il messaggio di errore "RuntimeError: oggetti coda dovrebbero essere condivisi solo tra i processi attraverso l'ereditarietà". Ecco uno dei modi in cui ho provato in base a una risposta precedente che ho trovato. (Ho lo stesso problema cercando di utilizzare Pool.map_async e Pool.imap)

import multiprocessing as mp 
import time 

def f(args): 
    x = args[0] 
    q = args[1] 
    q.put(str(x)) 
    time.sleep(0.1) 
    return x*x 

def main(): 
    q = mp.Queue() 
    pool = mp.Pool() 
    results = pool.imap_unordered(f, ([i, q] for i in range(1, 6))) 

    print str(q.get()) 

    pool.close() 
    pool.join() 

if __name__ == '__main__': 
    main() 

Infine, l'approccio di fitness 0 (renderlo globale) non genera alcun messaggio, tutto si blocca.

import multiprocessing as mp 
import time 

q = mp.Queue() 

def f(x): 
    q.put(str(x)) 
    return x*x 

def main(): 
    pool = mp.Pool() 
    results = pool.imap_unordered(f, range(1, 6)) 
    time.sleep(1) 

    print q.get() 

    pool.close() 
    pool.join() 

if __name__ == '__main__': 
    main() 

mi rendo conto che probabilmente funzionerà con multiprocessing.Process direttamente e che ci sono altre librerie per ottenere questo risultato, ma io odio a indietreggiare dalle funzioni della libreria standard che sono una grande misura fino a quando ho' sono sicuro che non è solo la mia mancanza di conoscenza che mi impedisce di essere in grado di sfruttarli.

Grazie.

Answer 1

Il trucco è passare la coda come argomento all'inizializzatore. Sembra funzionare con tutti i metodi di invio della piscina.

import multiprocessing as mp 

def f(x): 
    f.q.put('Doing: ' + str(x)) 
    return x*x 

def f_init(q): 
    f.q = q 

def main(): 
    jobs = range(1,6) 

    q = mp.Queue() 
    p = mp.Pool(None, f_init, [q]) 
    results = p.imap(f, jobs) 
    p.close() 

    for i in range(len(jobs)): 
     print q.get() 
     print results.next() 

if __name__ == '__main__': 
    main()