Die Speicherauslastung wächst mit Pythons multiprocessing.pool

Hier ist das Programm:

#!/usr/bin/python

import multiprocessing

def dummy_func(r):
    pass

def worker():
    pass

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=16)
    for index in range(0,100000):
        pool.apply_async(worker, callback=dummy_func)

    # clean up
    pool.close()
    pool.join()

Fand ich die Speicherauslastung (beide und VIRT RES) wuchs bis close()/join(), gibt es eine Lösung, um loszuwerden dieses? Ich habe versucht, maxtasksperchild mit 2,7 aber es hat auch nicht geholfen.

Ich habe ein eher kompliziertes Programm, dass calles apply_async() ~6M Zeiten, und bei ~1,5 M Punkt ich hab schon 6G+ RES, um zu vermeiden, alle anderen Faktoren, die ich vereinfacht das Programm um die oben genannten version.

EDIT:

Stellte sich heraus diese version besser funktioniert, vielen Dank für jedermanns Eingang:

#!/usr/bin/python

import multiprocessing

ready_list = []
def dummy_func(index):
    global ready_list
    ready_list.append(index)

def worker(index):
    return index

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=16)
    result = {}
    for index in range(0,1000000):
        result[index] = (pool.apply_async(worker, (index,), callback=dummy_func))
        for ready in ready_list:
            result[ready].wait()
            del result[ready]
        ready_list = []

    # clean up
    pool.close()
    pool.join()

Ich nicht alle sperren gibt es, wie ich glaube, wichtigsten Prozess single-threaded (callback ist mehr oder weniger wie eine event-driven Sache pro docs gelesen habe).

Ich geändert v1 index-Bereich bis 1.000.000, gleichen als v2 und hat einige tests - es ist seltsam zu mir, v2 ist noch ~10% schneller als die v1 (33s vs 37s), vielleicht die v1 war dabei zu viele interne Liste Wartungsarbeiten. v2 ist definitiv ein Gewinner auf die Speichernutzung, es ging nie über 300M (VIRT) und 50M (RES), während v1 verwendet werden 370M/120M, das beste war 330M/85M. Alle zahlen waren nur 3~4 mal testen, nur als Referenz.

InformationsquelleAutor der Frage C.B. | 2013-08-24

17

Hatte ich Speicherprobleme, vor kurzem, da war ich mit mehrmals die multiprocessing-Funktion, so dass es halten Laichzeit Prozesse, und verlassen Sie in Erinnerung.

Hier ist die Lösung, die ich jetzt verwende:
```
def myParallelProcess(ahugearray)
 from multiprocessing import Pool
 from contextlib import closing
 with closing( Pool(15) ) as p:
    res = p.imap_unordered(simple_matching, ahugearray, 100)
 return res
```
Ich ❤ mit

InformationsquelleAutor der Antwort deddu
6

Verwenden map_async statt apply_async zur Vermeidung einer übermäßigen Speicherauslastung.

Für Ihr erstes Beispiel, ändern Sie die folgenden zwei Zeilen:
```
for index in range(0,100000):
    pool.apply_async(worker, callback=dummy_func)
```
zu
```
pool.map_async(worker, range(100000), callback=dummy_func)
```
Es wird fertig in einem Blinzeln, bevor Sie sehen können die Speichernutzung im top. Ändern Sie die Liste, um eine größere, um den Unterschied zu sehen. Aber beachten Sie map_async zuerst konvertieren die iterierbar Sie sich an ihn übergeben, um eine Liste für die Berechnung der Länge, wenn es nicht __len__ Methode. Wenn Sie einen iterator, der eine große Anzahl von Elementen, die Sie verwenden können itertools.islice zu verarbeiten, Sie in kleinere Stücke.

Hatte ich ein problem mit dem Arbeitsspeicher in eine real-life-Programm mit viel mehr Daten und schließlich den Schuldigen gefunden wurde apply_async.

P. S. in Bezug auf die Speichernutzung, deine beiden Beispiele haben keine offensichtliche Unterschied.

InformationsquelleAutor der Antwort fossilet
4

Ich habe eine sehr große 3d-point-cloud-Daten, die ich bin-Verarbeitung. Ich habe versucht, mit dem multiprocessing-Modul, um die Geschwindigkeit der Verarbeitung, aber ich habe immer out of memory-Fehler. Nach einigen Recherchen und Tests habe ich festgestellt, dass ich füllte die Warteschlange von Aufgaben, die verarbeitet werden, viel schneller, als die Teilprozesse, könnte es leer. Ich bin mir sicher, dass durch chunking, oder mit map_async oder etwas, was ich eingestellt haben könnte die Last, aber ich wollte nicht zu größeren änderungen der umgebenden Logik.

Dumme Lösung, die ich Treffer auf die überprüfung des pool._cache Länge intermittierend, und wenn der cache zu groß ist, dann warten, bis die Warteschlange leer ist.

In meiner mainloop ich hatte auch schon ein Zähler-und ein status-ticker:
```
# Update status
count += 1
if count%10000 == 0:
    sys.stdout.write('.')
    if len(pool._cache) > 1e6:
        print "waiting for cache to clear..."
        last.wait() # Where last is assigned the latest ApplyResult
```
So alle 10k einfügen in den pool ich prüfen, wenn es mehr als 1 million Operationen in der Warteschlange (über 1G Speicher verwendet, in der main-Prozess). Wenn die Warteschlange voll ist, ich warte einfach auf das zuletzt eingefügte Aufgabe zu beenden.

Nun mein Programm kann stundenlang laufen, ohne running out of memory. Der wichtigste Prozess nur gelegentlich Pausen, während die Arbeiter weiterhin die Verarbeitung der Daten.

BTW die _cache-Mitglied dokumentiert ist, die das multiprocessing-Modul-pool Beispiel:
```
#
# Check there are no outstanding tasks
#

assert not pool._cache, 'cache = %r' % pool._cache
```
InformationsquelleAutor der Antwort kitsu.eb
2

Erstellen Sie einfach am pool in der Schleife und schließen Sie es an das Ende der Schleife mit
pool.close().

InformationsquelleAutor der Antwort Ullullu
1

Ich denke, das ist ähnlich wie die Frage, die ich gepostetaber ich bin mir nicht sicher, Sie haben die gleiche Verzögerung. Mein problem war, dass ich zu Ergebnissen aus dem multiprocessing-pool schneller als ich war Sie verbraucht, also Bauten Sie sich in Erinnerung. Um das zu vermeiden, habe ich eine semaphore Gas die Eingänge in den pool, so dass Sie nicht zu weit vor den Ausgängen war ich konsumieren.

InformationsquelleAutor der Antwort Don Kirkby

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.