Teilen Sie große, schreibgeschützte Numpy-Arrays zwischen Multiprocessing-Prozessen

Ich habe ein 60-GB-SciPy-Array (Matrix), das ich zwischen 5+ teilen muss multiprocessing Process Objekte. Ich habe numpy-sharedmem gesehen und gelesen diese Diskussion auf der SciPy-Liste. Es scheint zwei Ansätze zu geben:numpy-sharedmem und mit a multiprocessing.RawArray() und Zuordnung von NumPy dtypes zu ctypeS. Jetzt, numpy-sharedmem scheint der richtige Weg zu sein, aber ich habe noch kein gutes Referenzbeispiel gesehen. Ich brauche keine Sperren, da das Array (eigentlich eine Matrix) schreibgeschützt ist. Nun möchte ich aufgrund der Größe auf eine Kopie verzichten. Es hört sich an wie Die richtige Methode besteht darin, die zu erstellen nur Kopie des Arrays als sharedmem Array, und übergeben Sie es dann an die Process Objekte? Ein paar konkrete Fragen:

1. Was ist der beste Weg, um die Sharedmem-Handles tatsächlich an sub-Process()es? Benötige ich eine Warteschlange, nur um ein Array herumzureichen? Wäre ein Rohr besser? Kann ich es einfach als Argument an die übergeben Process() init der Unterklasse (wobei ich annehme, dass es eingelegt ist)?

2. In der Diskussion, die ich oben verlinkt habe, wird erwähnt numpy-sharedmem nicht 64bit-sicher? Ich verwende definitiv einige Strukturen, die nicht 32-Bit-adressierbar sind.

3. Gibt es Kompromisse zu den RawArray() sich nähern? Langsamer, fehlerhafter?

4. Benötige ich eine ctype-to-dtype-Zuordnung für die numpy-sharedmem-Methode?

5. Hat jemand ein Beispiel für einen OpenSource-Code, der dies tut? Ich bin ein sehr praktischer Gelehrter und es ist schwer, dies zum Laufen zu bringen, ohne ein gutes Beispiel zu sehen.

Wenn es zusätzliche Informationen gibt, die ich bereitstellen kann, um dies für andere zu verdeutlichen, kommentieren Sie dies bitte und ich werde es hinzufügen. Danke!

Dies muss auf Ubuntu Linux und laufen Könnte sein Mac OS, aber Portabilität ist kein großes Problem.

Unter Linux (oder einem anderen POSIX-kompatiblen System) können Sie dieses Array als globale Variable definieren. multiprocessing benutzt fork() unter Linux, wenn ein neuer untergeordneter Prozess gestartet wird. Ein neu erzeugter untergeordneter Prozess teilt den Speicher automatisch mit seinem übergeordneten Prozess, solange er ihn nicht ändert (Copy-on-Write Mechanismus).

Da Sie sagen “Ich brauche keinerlei Sperren, da das Array (eigentlich eine Matrix) schreibgeschützt ist”, wäre die Nutzung dieses Verhaltens ein sehr einfacher und dennoch äußerst effizienter Ansatz: Alle untergeordneten Prozesse werden zugreifen dieselben Daten im physischen Speicher beim Lesen dieses großen numpy-Arrays.

Übergeben Sie Ihr Array nicht an die Process() Konstrukteur, dies wird anweisen multiprocessing zu pickle die Daten an das Kind, was in Ihrem Fall äußerst ineffizient oder unmöglich wäre. Unter Linux gleich danach fork() das untergeordnete Element ist eine exakte Kopie des übergeordneten Elements, das denselben physischen Speicher verwendet, sodass Sie lediglich sicherstellen müssen, dass auf die Python-Variable, die die Matrix „enthält“, von innerhalb der zugegriffen werden kann target Funktion, an die Sie übergeben Process(). Dies können Sie typischerweise mit einer ‘globalen’ Variablen erreichen.

Beispielcode:

from multiprocessing import Process
from numpy import random


global_array = random.random(10**4)


def child():
    print sum(global_array)


def main():
    processes = [Process(target=child) for _ in xrange(10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()


if __name__ == "__main__":
    main()

Unter Windows – was nicht unterstützt wird fork() — multiprocessing verwendet den Win32-API-Aufruf CreateProcess. Es erstellt einen völlig neuen Prozess aus einer beliebigen ausführbaren Datei. Das ist, warum auf Windows man ist erforderlich um Daten an das Kind zu picken, wenn man Daten benötigt, die während der Laufzeit des Elternteils erstellt wurden.

@Velimir Mlaker hat eine tolle Antwort gegeben. Ich dachte, ich könnte einige Kommentare und ein kleines Beispiel hinzufügen.

(Ich konnte nicht viel Dokumentation zu Sharedmem finden – dies sind die Ergebnisse meiner eigenen Experimente.)

1. Müssen Sie die Handles übergeben, wenn der Unterprozess gestartet wird oder nachdem er gestartet wurde? Wenn es nur das erstere ist, können Sie einfach das verwenden target und args Argumente für Process. Dies ist möglicherweise besser als die Verwendung einer globalen Variablen.
2. Aus der von Ihnen verlinkten Diskussionsseite geht hervor, dass Sharedmem vor einiger Zeit Unterstützung für 64-Bit-Linux hinzugefügt wurde, sodass dies möglicherweise kein Problem darstellt.
3. Ich weiß nichts über diesen.
4. Nr. Siehe Beispiel unten.

Beispiel

#!/usr/bin/env python
from multiprocessing import Process
import sharedmem
import numpy

def do_work(data, start):
    data[start] = 0;

def split_work(num):
    n = 20
    width  = n/num
    shared = sharedmem.empty(n)
    shared[:] = numpy.random.rand(1, n)[0]
    print "values are %s" % shared

    processes = [Process(target=do_work, args=(shared, i*width)) for i in xrange(num)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print "values are %s" % shared
    print "type is %s" % type(shared[0])

if __name__ == '__main__':
    split_work(4)

Ausgabe

values are [ 0.81397784  0.59667692  0.10761908  0.6736734   0.46349645  0.98340718
  0.44056863  0.10701816  0.67167752  0.29158274  0.22242552  0.14273156
  0.34912309  0.43812636  0.58484507  0.81697513  0.57758441  0.4284959
  0.7292129   0.06063283]
values are [ 0.          0.59667692  0.10761908  0.6736734   0.46349645  0.
  0.44056863  0.10701816  0.67167752  0.29158274  0.          0.14273156
  0.34912309  0.43812636  0.58484507  0.          0.57758441  0.4284959
  0.7292129   0.06063283]
type is <type 'numpy.float64'>

Diese verwandte Frage könnte nützlich sein.

Vielleicht interessiert Sie ein kleines Stück Code, das ich geschrieben habe: github.com/vmlaker/benchmark-sharedmem

Die einzige Datei von Interesse ist main.py. Es ist ein Maßstab für numpy-sharedmem — der Code übergibt einfach Arrays (entweder numpy oder sharedmem) zu gespawnten Prozessen über Pipe. Die Arbeiter rufen einfach an sum() auf den Daten. Ich war nur daran interessiert, die Datenkommunikationszeiten zwischen den beiden Implementierungen zu vergleichen.

Ich habe auch einen anderen, komplexeren Code geschrieben: github.com/vmlaker/sherlock.

Hier verwende ich die numpy-sharedmem Modul für die Echtzeit-Bildverarbeitung mit OpenCV – die Bilder sind NumPy-Arrays, gemäß den neueren von OpenCV cv2 API. Die Bilder, eigentlich Verweise darauf, werden zwischen Prozessen über das erstellte Wörterbuchobjekt geteilt multiprocessing.Manager (im Gegensatz zur Verwendung von Queue oder Pipe.) Ich erhalte große Leistungsverbesserungen im Vergleich zur Verwendung von einfachen NumPy-Arrays.

Pipe vs. Warteschlange:

Meiner Erfahrung nach ist IPC mit Pipe schneller als Queue. Und das macht Sinn, da Queue Sperren hinzufügt, um es für mehrere Produzenten/Konsumenten sicher zu machen. Rohr nicht. Aber wenn Sie nur zwei Prozesse haben, die hin und her sprechen, ist es sicher, Pipe zu verwenden, oder, wie es in der Dokumentation heißt:

… es besteht keine Verfälschungsgefahr durch Prozesse, die gleichzeitig verschiedene Rohrenden verwenden.

sharedmem Sicherheit:

Das Hauptproblem mit sharedmem Modul ist die Möglichkeit eines Speicherlecks bei unspektakulärem Beenden des Programms. Dies wird in einer längeren Diskussion beschrieben Hier. Obwohl Sturla am 10. April 2011 einen Fix für ein Speicherleck erwähnt, habe ich seitdem immer noch Lecks erlebt, indem ich beide Repos verwende, Sturla Moldens eigenen auf GitHub (github.com/sturlamolden/sharedmem-numpy) und Chris Lee-Messer auf Bitbucket (bitbucket.org/cleemesser/numpy-sharedmem).

Wenn Ihr Array so groß ist, können Sie verwenden numpy.memmap. Wenn Sie beispielsweise ein Array auf der Festplatte gespeichert haben, sagen Sie 'test.array'können Sie gleichzeitige Prozesse verwenden, um auf die darin enthaltenen Daten auch im “Schreibmodus” zuzugreifen, aber Ihr Fall ist einfacher, da Sie nur den “Lesemodus” benötigen.

Erstellen des Arrays:

a = np.memmap('test.array', dtype="float32", mode="w+", shape=(100000,1000))

Sie können dieses Array dann genauso füllen wie ein gewöhnliches Array. Beispielsweise:

a[:10,:100]=1.
a[10:,100:]=2.

Die Daten werden auf der Festplatte gespeichert, wenn Sie die Variable löschen a.

Später können Sie mehrere Prozesse verwenden, die auf die Daten zugreifen test.array:

# read-only mode
b = np.memmap('test.array', dtype="float32", mode="r", shape=(100000,1000))

# read and writing mode
c = np.memmap('test.array', dtype="float32", mode="r+", shape=(100000,1000))

Verwandte Antworten:

• Arbeiten mit Big Data in Python und Numpy, nicht genug RAM, wie kann man Teilergebnisse auf Disc speichern?

• Ist es möglich, diskontinuierliche Daten auf der Festplatte einem Array mit Python zuzuordnen?

Es kann auch hilfreich sein, einen Blick in die Dokumentation für zu werfen Pyro Als ob Sie Ihre Aufgabe entsprechend partitionieren könnten, könnten Sie damit verschiedene Abschnitte auf verschiedenen Maschinen sowie auf verschiedenen Kernen derselben Maschine ausführen.

Warum nicht Multithreading verwenden? Ressourcen des Hauptprozesses können von seinen Threads nativ gemeinsam genutzt werden, daher ist Multithreading offensichtlich eine bessere Möglichkeit, Objekte gemeinsam zu nutzen, die dem Hauptprozess gehören.

Wenn Sie sich Sorgen um den GIL-Mechanismus von Python machen, können Sie vielleicht auf den zurückgreifen nogil von numba.