Wie bekomme ich den Rückgabewert einer an multiprocessing.Process übergebenen Funktion?

Question 1

Im folgenden Beispielcode möchte ich den Rückgabewert der Funktion erhalten worker. Wie kann ich das anstellen? Wo wird dieser Wert gespeichert?

Beispielcode:

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

Ausgabe:

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

Ich kann das relevante Attribut anscheinend nicht in den gespeicherten Objekten finden jobs.

Question 2

Verwenden freigegebene Variable kommunizieren. Zum Beispiel so:

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

Ich denke, der von @sega_sai vorgeschlagene Ansatz ist der bessere. Aber es braucht wirklich ein Codebeispiel, also hier gehts:

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

Was die Rückgabewerte druckt:

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]

Wenn Sie sich auskennen map (das in Python 2 eingebaute) sollte dies nicht zu schwierig sein. Ansonsten schau mal sega_Sais Link.

Beachten Sie, wie wenig Code benötigt wird. (Beachten Sie auch, wie Prozesse wiederverwendet werden).

Question 4

Für alle anderen, die suchen, wie man einen Wert von a erhält Process verwenden 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}

Beachten Sie, dass in Windows oder Jupyter Notebook mit multithreading Sie müssen dies als Datei speichern und die Datei ausführen. Wenn Sie dies in einer Eingabeaufforderung tun, wird ein Fehler wie dieser angezeigt:

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

Question 5

Aus irgendeinem Grund konnte ich kein allgemeines Beispiel dafür finden, wie man das mit macht Queue überall (selbst die Doc-Beispiele von Python erzeugen nicht mehrere Prozesse), also habe ich nach etwa 10 Versuchen Folgendes zum Laufen gebracht:

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 ist eine blockierende, Thread-sichere Warteschlange, die Sie verwenden können, um die Rückgabewerte der untergeordneten Prozesse zu speichern. Sie müssen also die Warteschlange an jeden Prozess weitergeben. Etwas weniger Offensichtliches ist hier, dass Sie es tun müssen get() aus der Warteschlange vor Ihnen join das Processsonst füllt sich die Warteschlange und blockiert alles.

Aktualisieren für diejenigen, die objektorientiert sind (getestet in 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

Dieses Beispiel zeigt, wie eine Liste von verwendet wird Multiprocessing.Pipe Instanzen, um Strings von einer beliebigen Anzahl von Prozessen zurückzugeben:

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

Ausgabe:

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

Diese Lösung verbraucht weniger Ressourcen als a multiprocessing.Queue was nutzt

ein Rohr
mindestens ein Schloss
ein Puffer

ein Faden

oder ein multiprocessing.SimpleQueue was nutzt

ein Rohr

mindestens ein Schloss

Es ist sehr aufschlussreich, sich die Quelle für jeden dieser Typen anzusehen.

Question 7

Es scheint, dass Sie die verwenden sollten Multiprocessing.Pool Klasse statt und verwenden Sie die Methoden .apply() .apply_async(), map()

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

Question 8

Du kannst den … benutzen exit eingebaut, um den Beendigungscode eines Prozesses festzulegen. Es ist erhältlich bei der exitcode Attribut des Prozesses:

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

Ausgabe:

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