Python Itertools mit Multiprocessing - riesige Liste vs ineffiziente CPUs Nutzung mit Iterator

Question

Ich arbeite an n Elemente (namens "Paar" unten) Variationen mit Wiederholung als Argument meiner Funktion verwendet. Natürlich funktioniert alles gut, solange die "r" -Liste nicht groß genug ist, um den gesamten Speicher zu verbrauchen. Das Problem ist, dass ich mehr als 16 Wiederholungen für 6 Elemente machen muss. Ich benutze dafür 40-Core-System in der Cloud.

Der Code sieht sieht wie folgt aus:

if __name__ == '__main__': 
    pool = Pool(39) 
    r = itertools.product(pairs,repeat=16) 
    pool.map(f, r) 

Ich glaube, ich Iterator statt der Schaffung der riesigen Liste im Voraus und hier beginnt das Problem verwenden sollten ..

Ich habe versucht, das Problem mit lösen der folgende Code:

if __name__ == '__main__': 
    pool = Pool(39) 
    for r in itertools.product(pairs,repeat=14): 
    pool.map(f, r) 

Das Speicherproblem geht weg, aber die CPU-Nutzung ist wie 5% pro Kern. Jetzt ist die Single-Core-Version des Codes schneller als diese.

Ich würde wirklich schätzen, wenn Sie mich ein bisschen führen könnte ..

Dank.

Answer 1

Ihr Original-Code schafft kein list im Voraus in Ihrem eigenen Code (itertools.product gibt einen Generator), aber pool.map wird den gesamten Generator zu realisieren (weil es annimmt, wenn Sie alle Ausgänge speichern können, können Sie alle Eingaben speichern) .

Verwenden Sie nicht pool.map hier. Wenn Sie geordnete Ergebnisse benötigen, verwenden Sie pool.imap, oder ist die Reihenfolge der Ergebnisse unwichtig, verwenden Sie pool.imap_unordered.Iterate das Ergebnis entweder telefonisch unter (Wickeln nicht in list), und die Ergebnisse verarbeiten, wie sie kommen, und Speicher sollte kein Problem sein:

if __name__ == '__main__': 
    pool = Pool(39) 
    for result in pool.imap(f, itertools.product(pairs, repeat=16)): 
     print(result) 

Wenn Sie mit pool.map für Nebenwirkungen, so Sie müssen es nur zur Fertigstellung ausführen, aber die Ergebnisse und die Reihenfolge spielen keine Rolle, Sie können die Leistung drastisch verbessern, indem Sie imap_unordered verwenden und collections.deque verwenden, um die "Ergebnisse" effizient zu leeren, ohne etwas zu speichern (deque von ist das schnellste , niedrigster Speicher, um zu erzwingen, dass ein Iterator vollständig ausgeführt wird, ohne die Ergebnisse zu speichern):

from collections import deque 

if __name__ == '__main__': 
    pool = Pool(39) 
    deque(pool.imap_unordered(f, itertools.product(pairs, repeat=16)), 0) 

Schließlich bin ich ein wenig verdächtig, 39 Arbeiter zu spezifizieren; multiprocessing ist für CPU-gebundene Aufgaben von großem Vorteil; Wenn Sie mit mehr Arbeitern arbeiten, als Sie CPU-Kerne haben und einen Vorteil erhalten, ist es möglich, dass multiprocessing Sie mehr in IPC kostet als es gewinnt, und die Verwendung von mehr Arbeitern maskiert nur das Problem, indem mehr Daten gepuffert werden.

Wenn Ihre Arbeit weitgehend I/O-gebunden ist, können Sie versuchen, einen threadbasierten Pool zu verwenden, der den Aufwand für das Beizen und Entpacken sowie die Kosten für IPC zwischen übergeordneten und untergeordneten Prozessen vermeidet. Im Gegensatz zu prozessbasierten Pools unterliegt das Python-Threading GIL-Problemen, sodass Ihre CPU-Bindung in Python funktioniert (ausgenommen GIL-Aufrufe für E/A, ctypes Aufrufe in .dll/.so-Dateien und bestimmte Erweiterungen von Drittanbietern wie numpy) GIL für schwere CPU-Arbeit) ist auf einen einzelnen Kern beschränkt (und in Python 2.x für CPU-gebundene Arbeit verschwenden Sie oft einen ansehnlichen Teil dieses auflösenden GIL-Konflikts und führen Kontextwechsel aus; Python 3 entfernt den größten Teil des Abfalls). Aber wenn Ihre Arbeit weitgehend I/O-gebunden ist, wird durch Sperren auf I/O die GIL freigegeben, damit andere Threads ausgeführt werden können, sodass Sie viele Threads haben können, solange die meisten auf I/O verzögern. Es ist auch leicht zu wechseln (solange Sie Ihr Programm nicht so ausgelegt haben, dass es sich auf separate Adressräume für jeden Worker verlässt, indem Sie annehmen, dass Sie in den Status "shared" schreiben und andere Worker oder den übergeordneten Prozess nicht beeinflussen können), ändern Sie einfach:

from multiprocessing import Pool 

zu:

from multiprocessing.dummy import Pool 

und Sie erhalten die multiprocessing.dummy Version des Pools, basierend auf Threads anstelle von Prozessen.

Answer 2

Das zweite Codebeispiel ist langsamer, weil Sie ein einzelnes Paar an einen Pool mit 39 Arbeiten senden. Nur ein Mitarbeiter bearbeitet Ihre Anfrage und die anderen 38 werden nichts tun! Wird langsamer sein, da Sie beim Verteilen von Daten vom Hauptthread an die Worker-Prozesse Overhead haben.

Sie können einige Paare "puffern" und dann den Satz von Paaren ausführen, um die Speichernutzung auszugleichen, aber dennoch die Vorteile der Multiprozessumgebung zu nutzen.

import itertools 
from multiprocessing import Pool 

def foo(x): 
    return sum(x) 

cpus = 3 
pool = Pool(cpus) 
# 10 is buffer size multiplier - the number of pair that each process will get 
buff_size = 10*cpus 
buff = [] 
for i, r in enumerate(itertools.product(range(20), range(10))): 
    if (i % buff_size) == (buff_size-1): 
     print pool.map(foo, buff) 
     buff = [] 
    else: 
     buff.append(r) 

if len(buff) > 0: 
    print pool.map(foo, buff) 
    buff = [] 

Der Ausgang der oben wie diese

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 2, 3, 4, 5, 6, 7, 8, 9, 10] 
[3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 5, 6, 7, 8, 9, 10, 11, 12, 13] 
[6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 8, 9, 10, 11, 12, 13, 14, 15, 16] 
[9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 11, 12, 13, 14, 15, 16, 17, 18, 19] 
[12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 14, 15, 16, 17, 18, 19, 20, 21, 22] 
[15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 17, 18, 19, 20, 21, 22, 23, 24, 25] 
[18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28] 

Wiedergabe mit dem buff_size Multiplikator aussehen wird, die richtige Balance für Ihr System zu erhalten!