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

188

Im folgenden Beispielcode möchte ich den Rückgabewert der Funktion wiederherstellen worker. Wie kann ich das machen? Wo ist 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 in den darin gespeicherten Objekten nicht finden jobs.

blz
quelle

Antworten:

187

Verwenden Sie eine gemeinsam genutzte Variable für die Kommunikation. 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()
vartec
quelle
46
Ich würde multiprocessing.Queueeher die Verwendung von a als von Managerhier empfehlen . Die Verwendung von Managererfordert das Laichen eines völlig neuen Prozesses, der übertrieben ist, wenn a Queuedies tun würde.
Dano
1
@dano: Ich frage mich, ob wir das Queue () - Objekt verwenden, wir können die Reihenfolge nicht sicher bestimmen, wenn jeder Prozess den Wert zurückgibt. Ich meine, wenn wir die Reihenfolge im Ergebnis brauchen, um die nächste Arbeit zu erledigen. Wie können wir sicher sein, wo genau welche Ausgabe von welchem ​​Prozess ist
Catbuilts
4
@Catbuilts Sie können von jedem Prozess ein Tupel zurückgeben, wobei ein Wert der tatsächliche Rückgabewert ist, den Sie interessieren, und der andere eine eindeutige Kennung aus dem Prozess ist. Ich frage mich aber auch, warum Sie wissen müssen, welcher Prozess welchen Wert zurückgibt. Wenn das das ist, was Sie tatsächlich über den Prozess wissen müssen, oder müssen Sie zwischen Ihrer Liste der Eingaben und der Liste der Ausgaben korrelieren? In diesem Fall würde ich empfehlen multiprocessing.Pool.map, Ihre Liste der Arbeitselemente zu verarbeiten.
Dano
4
Vorsichtsmaßnahmen für Funktionen mit nur einem Argument : sollte verwenden args=(my_function_argument, ). Beachten Sie das ,Komma hier! Andernfalls beschwert sich Python über "fehlende Positionsargumente". Ich habe 10 Minuten gebraucht, um das herauszufinden. Überprüfen Sie auch die manuelle Verwendung (im Abschnitt "Prozessklasse").
Yuqli
2
@vartec Ein Nachteil bei der Verwendung eines Multipriocessing.Manager () -Wörterbuchs besteht darin, dass das zurückgegebene Objekt ausgewählt (serialisiert) wird. Daher weist die Pickle-Bibliothek einen Engpass mit einer maximalen Größe von 2 GB für das zurückzugebende Objekt auf. Gibt es eine andere Möglichkeit, um die Serialisierung des zurückkehrenden Objekts zu vermeiden?
Hirschme
67

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

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

Welches gibt die Rückgabewerte aus:

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 mit map(dem integrierten Python 2) vertraut sind, sollte dies nicht zu schwierig sein. Ansonsten schau dir den Link von sega_Sai an .

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

Kennzeichen
quelle
1
Irgendwelche Ideen, warum meine getpid()Rückgabe alle den gleichen Wert hat? Ich verwende Python3
Zelusp
Ich bin nicht sicher, wie Pool Aufgaben auf Arbeiter verteilt. Vielleicht können sie alle beim selben Arbeiter landen, wenn sie wirklich schnell sind? Passiert es konsequent? Auch wenn Sie eine Verzögerung hinzufügen?
Mark
Ich dachte auch, dass es eine geschwindigkeitsbezogene Sache ist, aber wenn ich pool.mapeinen Bereich von 1.000.000 mit mehr als 10 Prozessen füttere, sehe ich höchstens zwei verschiedene Pids.
Zelusp
1
Dann bin ich mir nicht sicher. Ich denke, es wäre interessant, dafür eine separate Frage zu stellen.
Mark
Wenn die Dinge, die Sie eine andere Funktion an jeden Prozess senden möchten, verwenden Sie pool.apply_async: docs.python.org/3/library/…
Kyle
24

Dieses Beispiel zeigt, wie eine Liste von Multiprocessing.Pipe- Instanzen verwendet wird, um Zeichenfolgen aus 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 eine Multiprocessing.Queue, die verwendet wird

  • ein Rohr
  • mindestens ein Schloss
  • ein Puffer
  • ein Faden

oder eine Multiprocessing.SimpleQueue, die verwendet

  • ein Rohr
  • mindestens ein Schloss

Es ist sehr lehrreich, die Quelle für jeden dieser Typen zu betrachten.

David Cullen
quelle
Was wäre der beste Weg, dies zu tun, ohne die Pipes zu einer globalen Variablen zu machen?
Nickpick
Ich habe alle globalen Daten und Codes in eine Hauptfunktion eingefügt und es funktioniert genauso. Beantwortet das deine Frage?
David Cullen
Muss die Pipe immer gelesen werden, bevor ein neuer Wert hinzugefügt (gesendet) werden kann?
Nickpick
+1, gute Antwort. Da die Lösung jedoch effizienter ist, besteht der Kompromiss darin, dass Sie eine Pipepro Prozess gegenüber einer Queuefür alle Prozesse erstellen. Ich weiß nicht, ob das in allen Fällen effizienter ist.
Sudo
2
Diese Antwort verursacht einen Deadlock, wenn das zurückkehrende Objekt groß ist. Anstatt zuerst proc.join () auszuführen, würde ich zuerst versuchen, den Rückgabewert zu recv () und dann den Join auszuführen.
L. Pes
21

Aus irgendeinem Grund konnte ich Queuenirgendwo ein allgemeines Beispiel dafür finden (selbst Pythons Dokumentbeispiele erzeugen nicht mehrere Prozesse). Nach 10 Versuchen habe ich Folgendes getan:

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

Queueist eine blockierende, threadsichere Warteschlange, in der Sie die Rückgabewerte der untergeordneten Prozesse speichern können. Sie müssen also die Warteschlange an jeden Prozess übergeben. Etwas weniger offensichtliches ist hier, dass Sie get()aus der Warteschlange müssen, bevor Sie joindie Processes oder die Warteschlange füllt und blockiert alles.

Update für objektorientierte Benutzer (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)
sudo
quelle
18

Für alle anderen, die suchen, wie sie einen Wert aus einer ProcessVerwendung ziehen können 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()
    print queue.get()  # Prints {"foo": True}
    p.join()
Matthew Moisen
quelle
1
Wenn ich in meinem Arbeitsprozess etwas in eine Warteschlange stelle, wird mein Join nie erreicht. Irgendeine Idee, wie das kommen könnte?
Laurens Koppenol
@LaurensKoppenol meinst du, dass dein Hauptcode permanent bei p.join () hängt und nie weitergeht? Hat Ihr Prozess eine Endlosschleife?
Matthew Moisen
4
Ja, es hängt dort unendlich. Meine Arbeiter sind alle fertig (Schleife innerhalb der Arbeiterfunktion endet, Druckanweisung wird anschließend für alle Arbeiter gedruckt). Der Join macht nichts. Wenn ich das Queueaus meiner Funktion entferne , kann ich dasjoin()
Laurens Koppenol
@LaurensKoppenol Rufen Sie vielleicht nicht queue.put(ret)vor dem Anruf an p.start()? In diesem Fall bleibt der Worker-Thread für immer hängen queue.get(). Sie können dies replizieren, indem Sie meinen obigen Ausschnitt kopieren, während Sie ihn auskommentieren queue.put(ret).
Matthew Moisen
Ich habe diese Antwort bearbeitet, das queue.get()muss vor dem passieren p.join(). Es funktioniert jetzt für mich.
jfunk
10

Mit der integrierten Funktion können Sie exitden Exit-Code eines Prozesses festlegen. Es kann aus dem exitcodeAttribut des Prozesses erhalten werden:

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]
David Cullen
quelle
4
Seien Sie gewarnt, dass dieser Ansatz verwirrend werden könnte. Prozesse sollten im Allgemeinen mit dem Exit-Code 0 beendet werden, wenn sie fehlerfrei abgeschlossen wurden. Wenn Ihre Systemprozess-Exit-Codes von irgendetwas überwacht werden, werden diese möglicherweise als Fehler gemeldet.
Eisenrad
1
Perfekt, wenn Sie bei einem Fehler nur eine Ausnahme im übergeordneten Prozess auslösen möchten.
CrizCraig
5

Das Kieselpaket hat eine nette Abstraktion, multiprocessing.Pipedie dies recht einfach macht:

from pebble import concurrent

@concurrent.process
def function(arg, kwarg=0):
    return arg + kwarg

future = function(1, kwarg=1)

print(future.result())

Beispiel von: https://pythonhosted.org/Pebble/#concurrent-decorators

erikreed
quelle
3

Ich dachte, ich würde die einfachsten Beispiele, die von oben kopiert wurden, vereinfachen und für mich an Py3.6 arbeiten. Am einfachsten ist multiprocessing.Pool:

import multiprocessing
import time

def worker(x):
    time.sleep(1)
    return x

pool = multiprocessing.Pool()
print(pool.map(worker, range(10)))

Sie können die Anzahl der Prozesse im Pool festlegen, z Pool(processes=5). Standardmäßig wird jedoch die CPU-Anzahl verwendet. Lassen Sie diese Option daher für CPU-gebundene Aufgaben leer. (E / A-gebundene Aufgaben passen ohnehin oft zu Threads, da die Threads meistens warten und einen CPU-Kern gemeinsam nutzen können.) WendetPool auch die Chunking-Optimierung an .

(Beachten Sie, dass die Worker-Methode nicht in einer Methode verschachtelt werden kann. Ich habe meine Worker-Methode zunächst in der Methode definiert, die aufgerufen pool.mapwird, um alles in sich geschlossen zu halten, aber dann konnten die Prozesse sie nicht importieren und haben "AttributeError" ausgelöst : Lokales Objekt Outer_Method..inner_method kann nicht ausgewählt werden. Mehr hier . Es kann sich innerhalb einer Klasse befinden.)

(Schätzen Sie den Druck der ursprünglichen Frage 'represent!' und nicht time.sleep(), aber ohne sie dachte ich, dass ein Code gleichzeitig ausgeführt wird, wenn dies nicht der Fall ist.)


Py3 ProcessPoolExecutorist auch zwei Zeilen ( .mapgibt einen Generator zurück, so dass Sie die benötigenlist() ):

from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor() as executor:
    print(list(executor.map(worker, range(10))))

Mit schlicht Process es:

import multiprocessing
import time

def worker(x, queue):
    time.sleep(1)
    queue.put(x)

queue = multiprocessing.SimpleQueue()
tasks = range(10)

for task in tasks:
    multiprocessing.Process(target=worker, args=(task, queue,)).start()

for _ in tasks:
    print(queue.get())

Verwenden Sie, SimpleQueuewenn Sie nur putund benötigen get. Die erste Schleife startet alle Prozesse, bevor die zweite die blockierenden queue.getAufrufe ausführt . Ich glaube nicht, dass es einen Grund gibt, auch anzurufen p.join().

Chris
quelle
2

Eine einfache Lösung:

import multiprocessing

output=[]
data = range(0,10)

def f(x):
    return x**2

def handler():
    p = multiprocessing.Pool(64)
    r=p.map(f, data)
    return r

if __name__ == '__main__':
    output.append(handler())

print(output[0])

Ausgabe:

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
Rubens_Zimbres
quelle
2

Wenn Sie Python 3 verwenden, können Sie Folgendes verwenden concurrent.futures.ProcessPoolExecutor:

from concurrent.futures import ProcessPoolExecutor

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


if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(worker, range(5))))

Ausgabe:

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

Ich habe die Antwort von vartec ein wenig geändert, da ich die Fehlercodes von der Funktion erhalten musste. (Danke vertec !!! es ist ein toller Trick)

Dies kann auch mit einem gemacht werden, manager.listaber ich denke, es ist besser, es in einem Diktat zu haben und eine Liste darin zu speichern. Auf diese Weise behalten wir die Funktion und die Ergebnisse bei, da wir nicht sicher sind, in welcher Reihenfolge die Liste gefüllt wird.

from multiprocessing import Process
import time
import datetime
import multiprocessing


def func1(fn, m_list):
    print 'func1: starting'
    time.sleep(1)
    m_list[fn] = "this is the first function"
    print 'func1: finishing'
    # return "func1"  # no need for return since Multiprocess doesnt return it =(

def func2(fn, m_list):
    print 'func2: starting'
    time.sleep(3)
    m_list[fn] = "this is function 2"
    print 'func2: finishing'
    # return "func2"

def func3(fn, m_list):
    print 'func3: starting'
    time.sleep(9)
    # if fail wont join the rest because it never populate the dict
    # or do a try/except to get something in return.
    raise ValueError("failed here")
    # if we want to get the error in the manager dict we can catch the error
    try:
        raise ValueError("failed here")
        m_list[fn] = "this is third"
    except:
        m_list[fn] = "this is third and it fail horrible"
        # print 'func3: finishing'
        # return "func3"


def runInParallel(*fns):  # * is to accept any input in list
    start_time = datetime.datetime.now()
    proc = []
    manager = multiprocessing.Manager()
    m_list = manager.dict()
    for fn in fns:
        # print fn
        # print dir(fn)
        p = Process(target=fn, name=fn.func_name, args=(fn, m_list))
        p.start()
        proc.append(p)
    for p in proc:
        p.join()  # 5 is the time out

    print datetime.datetime.now() - start_time
    return m_list, proc

if __name__ == '__main__':
    manager, proc = runInParallel(func1, func2, func3)
    # print dir(proc[0])
    # print proc[0]._name
    # print proc[0].name
    # print proc[0].exitcode

    # here you can check what did fail
    for i in proc:
        print i.name, i.exitcode  # name was set up in the Process line 53

    # here will only show the function that worked and where able to populate the 
    # manager dict
    for i, j in manager.items():
        print dir(i)  # things you can do to the function
        print i, j
Pelos
quelle