Beseitigung von Overhead in Multiprocessing mit Pool

Question

Ich bin derzeit in einer Situation, in der ich Code wiederholt aufgerufen haben und versuchen, den Mehraufwand im Zusammenhang mit der Multiprozessverarbeitung zu reduzieren. Also, betrachten Sie das folgende Beispiel, die absichtlich keine „teure“ Berechnungen enthalten:

import multiprocessing as mp 
def f(x): 
    # toy function 
    return x*x 

if __name__ == '__main__': 
    for x in range(500): 
     pool = mp.Pool(processes=2) 
     print(pool.map(f, range(x, x + 50))) 
     pool.close() 
     pool.join() # necessary? 

Dieser Code dauert 53 Sekunden im Vergleich zu 0,04 Sekunden für den sequentiellen Ansatz.

Erste Frage: muss ich wirklich pool.join() in diesem Fall aufrufen, wenn nur pool.map() jemals verwendet wird? Ich kann keine negativen Effekte finden, wenn ich es weglasse und die Laufzeit würde auf 4,8 Sekunden fallen. (Ich verstehe, dass das Auslassen von pool.close() nicht möglich ist, da wir dann Threads verlieren würden.)

Jetzt, während dies eine nette Verbesserung wäre, würde ich als erste Antwort wahrscheinlich "gut, nicht" t erst den Pool in der Schleife erstellen ". Ok, kein Problem, aber der parallelisiert Code lebt tatsächlich in einer Instanzmethode, so würde ich verwenden:

class MyObject: 
    def __init__(self): 
     self.pool = mp.Pool(processes=2) 
    def function(self, x): 
     print(self.pool.map(f, range(x, x + 50))) 

if __name__ == '__main__': 
    my_object = MyObject() 
    for x in range(500): 
     my_object.function(x) 

Das wäre meine favorisierte Lösung sein, wie es in einem ausgezeichneten 0,4 Sekunden läuft.

Zweite Frage: sollte ich pool.close()/pool.join() irgendwo explizit aufrufen (z. B. im Destruktor von MyObject) oder ist der aktuelle Code ausreichend? (Wenn es darauf ankommt: Es ist in Ordnung, anzunehmen, dass es in meinem Projekt nur wenige langlebige Instanzen von MyObject gibt.)

Answer 1

Natürlich dauert es lange: Sie reservieren weiterhin einen neuen Pool und löschen ihn für jede x .

Es läuft viel schneller, wenn Sie stattdessen tun:

if __name__ == '__main__': 
    pool = mp.Pool(processes=2) # allocate the pool only once 
    for x in range(500): 
     print(pool.map(f, range(x, x + 50))) 

    pool.close() # close it only after all the requests are submitted 
    pool.join() # wait for the last worker to finish 

bereites und Sie werden es jetzt sehen viel schneller funktioniert.

Hier sind die Links zu den docs für join und close:

Sobald close genannt wird Sie nicht mehr Aufgaben an den Pool einreichen können, und join wartet, bis der letzte Arbeiter seine Arbeit beendet. Sie sollten in dieser Reihenfolge aufgerufen werden (zuerst schließen und dann verbinden).

Answer 2

Nun, man könnte tatsächlich bereits zugeteilten Pool als Argument zu Ihrem Objekt übergeben:

class MyObject: 
    def __init__(self, pool): 
     self.pool = pool 

    def function(self, x): 
     print(self.pool.map(f, range(x, x + 50))) 


if __name__ == '__main__': 
    with mp.Pool(2) as pool: 
     my_object = MyObject(pool) 
     my_second_object = MyObject(pool) 

     for x in range(500): 
      my_object.function(x) 
      my_second_object.function(x) 

     pool.close() 

ich keinen Grund finden, warum es notwendig sein könnte, verschiedene Pools in verschiedenen Objekten