python - set_start_method - ¿Puedo usar una Cola de multiprocesamiento en una función llamada por Pool.imap?

python pool process

Question

Estoy usando python 2.7 e intento ejecutar algunas tareas pesadas de la CPU en sus propios procesos. Me gustaría poder enviar mensajes al proceso principal para mantenerlo informado del estado actual del proceso. La cola de multiprocesamiento parece perfecta para esto, pero no puedo encontrar la manera de hacerlo funcionar.

Entonces, este es mi ejemplo de trabajo básico menos el uso de una cola.

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

Intenté pasar la cola de varias maneras, y reciben el mensaje de error "RuntimeError: los objetos de la cola solo se deben compartir entre los procesos a través de la herencia". Esta es una de las formas en que lo intenté en base a una respuesta anterior que encontré. (Tengo el mismo problema al tratar de usar Pool.map_async y 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()

Finalmente, el enfoque de 0 aptitud (hacerlo global) no genera ningún mensaje, simplemente se bloquea.

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

Soy consciente de que probablemente funcione con multiprocesamiento. Procesa directamente y que hay otras bibliotecas para lograr esto, pero odio alejarme de las funciones estándar de la biblioteca que son una excelente opción hasta que estoy seguro de que no es solo mi falta. de conocimiento que me impide ser capaz de explotarlos.

Gracias.

Answer 1

El truco es pasar la cola como un argumento para el inicializador. Parece que funciona con todos los métodos de despacho de Pool.

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