¿Cómo se pasa una referencia de cola a una función administrada por pool.map_async()?

Question

Quiero un proceso de larga duración para devolver su progreso a través de una cola (o algo similar) que alimentaré a un cuadro de diálogo de barra de progreso. También necesito el resultado cuando el proceso se completa. Aquí falla un ejemplo de prueba con 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() 

he sido capaz de conseguir que esto funcione utilizando objetos individuales del proceso (en el que estoy alowed a pasar una referencia de cola), pero entonces no tener una piscina para manejar los muchos procesos que quiero almorzar. ¿Algún consejo sobre un mejor patrón para esto?

Answer 1

El siguiente código parece funcionar:

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

Tenga en cuenta que la cola se obtuvo a partir de un manager.Queue() en lugar de multiprocessing.Queue(). Gracias Alex por señalarme en esta dirección.

Answer 2

Haciendo q trabajos globales ...:

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 necesita varias colas, por ejemplo para evitar mezclar el progreso de los diversos procesos del grupo, debería funcionar una lista global de colas (por supuesto, cada proceso necesitará saber qué índice en la lista a usar, pero está bien pasarlo como argumento;)