¿Cómo obtener el valor de retorno de una función pasada a multiprocessing.Process?

Question 1

En el código de ejemplo a continuación, me gustaría obtener el valor de retorno de la función worker. ¿Cómo puedo hacer esto? ¿Dónde se almacena este valor?

Código de ejemplo:

import multiprocessing

def worker(procnum):
    '''worker function'''
    print str(procnum) + ' represent!'
    return procnum


if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        jobs.append(p)
        p.start()

    for proc in jobs:
        proc.join()
    print jobs

Producción:

0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
[<Process(Process-1, stopped)>, <Process(Process-2, stopped)>, <Process(Process-3, stopped)>, <Process(Process-4, stopped)>, <Process(Process-5, stopped)>]

Parece que no puedo encontrar el atributo relevante en los objetos almacenados en jobs.

Question 2

Usar variable compartida comunicar. Por ejemplo como este:

import multiprocessing


def worker(procnum, return_dict):
    """worker function"""
    print(str(procnum) + " represent!")
    return_dict[procnum] = procnum


if __name__ == "__main__":
    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i, return_dict))
        jobs.append(p)
        p.start()

    for proc in jobs:
        proc.join()
    print(return_dict.values())

Question 3

Creo que el enfoque sugerido por @sega_sai es mejor. Pero realmente necesita un ejemplo de código, así que aquí va:

import multiprocessing
from os import getpid

def worker(procnum):
    print('I am number %d in process %d' % (procnum, getpid()))
    return getpid()

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes = 3)
    print(pool.map(worker, range(5)))

Que imprimirá los valores de retorno:

I am number 0 in process 19139
I am number 1 in process 19138
I am number 2 in process 19140
I am number 3 in process 19139
I am number 4 in process 19140
[19139, 19138, 19140, 19139, 19140]

Si está familiarizado con map (el Python 2 incorporado) esto no debería ser demasiado desafiante. De lo contrario, echa un vistazo a enlace de sega_Sai.

Tenga en cuenta lo poco que se necesita código. (También tenga en cuenta cómo se reutilizan los procesos).

Question 4

Para cualquier otra persona que esté buscando cómo obtener un valor de un Process usando Queue:

import multiprocessing

ret = {'foo': False}

def worker(queue):
    ret = queue.get()
    ret['foo'] = True
    queue.put(ret)

if __name__ == '__main__':
    queue = multiprocessing.Queue()
    queue.put(ret)
    p = multiprocessing.Process(target=worker, args=(queue,))
    p.start()
    p.join()
    print(queue.get())  # Prints {"foo": True}

Tenga en cuenta que en Windows o Jupyter Notebook, con multithreading tienes que guardar esto como un archivo y ejecutar el archivo. Si lo hace en un símbolo del sistema, verá un error como este:

 AttributeError: Can't get attribute 'worker' on <module '__main__' (built-in)>

Question 5

Por alguna razón, no pude encontrar un ejemplo general de cómo hacer esto con Queue en cualquier lugar (incluso los ejemplos de documentos de Python no generan múltiples procesos), así que esto es lo que obtuve después de 10 intentos:

def add_helper(queue, arg1, arg2): # the func called in child processes
    ret = arg1 + arg2
    queue.put(ret)

def multi_add(): # spawns child processes
    q = Queue()
    processes = []
    rets = []
    for _ in range(0, 100):
        p = Process(target=add_helper, args=(q, 1, 2))
        processes.append(p)
        p.start()
    for p in processes:
        ret = q.get() # will block
        rets.append(ret)
    for p in processes:
        p.join()
    return rets

Queue es una cola de bloqueo segura para subprocesos que puede usar para almacenar los valores de retorno de los procesos secundarios. Entonces hay que pasar la cola a cada proceso. Algo menos obvio aquí es que tienes que get() de la cola antes que tú join la Processes o de lo contrario la cola se llena y bloquea todo.

Actualizar para aquellos que están orientados a objetos (probado en Python 3.4):

from multiprocessing import Process, Queue

class Multiprocessor():

    def __init__(self):
        self.processes = []
        self.queue = Queue()

    @staticmethod
    def _wrapper(func, queue, args, kwargs):
        ret = func(*args, **kwargs)
        queue.put(ret)

    def run(self, func, *args, **kwargs):
        args2 = [func, self.queue, args, kwargs]
        p = Process(target=self._wrapper, args=args2)
        self.processes.append(p)
        p.start()

    def wait(self):
        rets = []
        for p in self.processes:
            ret = self.queue.get()
            rets.append(ret)
        for p in self.processes:
            p.join()
        return rets

# tester
if __name__ == "__main__":
    mp = Multiprocessor()
    num_proc = 64
    for _ in range(num_proc): # queue up multiple tasks running `sum`
        mp.run(sum, [1, 2, 3, 4, 5])
    ret = mp.wait() # get all results
    print(ret)
    assert len(ret) == num_proc and all(r == 15 for r in ret)

Question 6

Este ejemplo muestra cómo utilizar una lista de multiprocesamiento.Tubo instancias para devolver cadenas de un número arbitrario de procesos:

import multiprocessing

def worker(procnum, send_end):
    '''worker function'''
    result = str(procnum) + ' represent!'
    print result
    send_end.send(result)

def main():
    jobs = []
    pipe_list = []
    for i in range(5):
        recv_end, send_end = multiprocessing.Pipe(False)
        p = multiprocessing.Process(target=worker, args=(i, send_end))
        jobs.append(p)
        pipe_list.append(recv_end)
        p.start()

    for proc in jobs:
        proc.join()
    result_list = [x.recv() for x in pipe_list]
    print result_list

if __name__ == '__main__':
    main()

Producción:

0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
['0 represent!', '1 represent!', '2 represent!', '3 represent!', '4 represent!']

Esta solución utiliza menos recursos que una multiprocesamiento.Cola que utiliza

un tubo
al menos un bloqueo
un amortiguador
un hilo

o un multiprocesamiento.SimpleQueue que utiliza

un tubo
al menos un bloqueo

Es muy instructivo mirar la fuente de cada uno de estos tipos.

Question 7

Parece que deberías usar el multiprocesamiento.Pool class en su lugar y use los métodos .apply() .apply_async(), map()

http://docs.python.org/library/multiprocessing.html?highlight=pool#multiprocessing.pool.AsyncResult

Question 8

Puedes usar el exit incorporado para establecer el código de salida de un proceso. Se puede obtener de la exitcode atributo del proceso:

import multiprocessing

def worker(procnum):
    print str(procnum) + ' represent!'
    exit(procnum)

if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        jobs.append(p)
        p.start()

    result = []
    for proc in jobs:
        proc.join()
        result.append(proc.exitcode)
    print result

Producción:

0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
[0, 1, 2, 3, 4]

¿Ha sido útil esta solución?