set_start_method - python pool process

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

Question

Quiero un proceso de larga ejecución para devolver su progreso a través de una cola (o algo similar) que alimentaré a un cuadro de diálogo de barras 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 podido hacer que esto funcione usando objetos de proceso individuales (donde me permite pasar una referencia de cola) pero luego no tengo un grupo para administrar los muchos procesos que deseo iniciar. ¿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 obtiene de un manager.Queue () en lugar de multiprocesamiento.Queue (). Gracias Alex por señalarme en esta dirección.

Answer 2

Hacer 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 de grupo, debería funcionar una lista global de colas (por supuesto, cada proceso necesitará saber qué índice de la lista usar, pero está bien pasarlo). como argumento ;-).