Multiprocessing: Teilen eines großen schreibgeschützten Objekts zwischen Prozessen?

Werden untergeordnete Prozesse über Multiprozessor- Freigabeobjekte erzeugt, die zuvor im Programm erstellt wurden?

Ich habe folgendes Setup:

do_some_processing(filename):
    for line in file(filename):
        if line.split(',')[0] in big_lookup_object:
            # something here

if __name__ == '__main__':
    big_lookup_object = marshal.load('file.bin')
    pool = Pool(processes=4)
    print pool.map(do_some_processing, glob.glob('*.data'))

Ich lade ein großes Objekt in den Speicher und erstelle dann einen Pool von Arbeitern, die dieses große Objekt verwenden müssen. Auf das große Objekt wird schreibgeschützt zugegriffen. Ich muss keine Änderungen zwischen den Prozessen übergeben.

Meine Frage ist: Wird das große Objekt in den gemeinsam genutzten Speicher geladen, wie es der Fall wäre, wenn ich einen Prozess unter Unix / C erzeugen würde, oder lädt jeder Prozess seine eigene Kopie des großen Objekts?

Update: zur weiteren Verdeutlichung - big_lookup_object ist ein gemeinsam genutztes Lookup-Objekt. Ich muss das nicht aufteilen und separat verarbeiten. Ich muss eine einzige Kopie davon behalten. Die Arbeit, die ich zum Teilen brauche, besteht darin, viele andere große Dateien zu lesen und die Elemente in diesen großen Dateien anhand des Suchobjekts nachzuschlagen.

Weiteres Update: Datenbank ist eine gute Lösung, Memcached ist möglicherweise eine bessere Lösung und Datei auf der Festplatte (Shelve oder DBM) ist möglicherweise noch besser. Bei dieser Frage interessierte mich besonders eine In-Memory-Lösung. Für die endgültige Lösung werde ich hadoop verwenden, aber ich wollte sehen, ob ich auch eine lokale In-Memory-Version haben kann.

"Werden untergeordnete Prozesse über Multiprozessor-Freigabeobjekte erzeugt, die zuvor im Programm erstellt wurden?"

Nein (Python vor 3.8) und Ja in 3.8 ( https://docs.python.org/3/library/multiprocessing.shared_memory.html#module-multiprocessing.shared_memory )

Prozesse haben einen unabhängigen Speicherplatz.

Lösung 1

Tun Sie dies, um eine große Struktur mit vielen Arbeitern optimal zu nutzen.

1. Schreiben Sie jeden Worker als "Filter" - liest Zwischenergebnisse von stdin, funktioniert, schreibt Zwischenergebnisse auf stdout.

2. Verbinden Sie alle Mitarbeiter als Pipeline:

   process1 <source | process2 | process3 | ... | processn >result

Jeder Prozess liest, arbeitet und schreibt.

Dies ist bemerkenswert effizient, da alle Prozesse gleichzeitig ausgeführt werden. Die Schreib- und Lesevorgänge durchlaufen direkt gemeinsam genutzte Puffer zwischen den Prozessen.

Lösung 2

In einigen Fällen haben Sie eine komplexere Struktur - häufig eine "Fan-Out" -Struktur. In diesem Fall haben Sie einen Elternteil mit mehreren Kindern.

1. Übergeordnetes Element öffnet Quelldaten. Eltern gabeln eine Reihe von Kindern.

2. Das übergeordnete Element liest die Quelle und gibt Teile der Quelle an jedes gleichzeitig ausgeführte untergeordnete Element weiter.

3. Wenn der Elternteil das Ende erreicht hat, schließen Sie das Rohr. Das Kind bekommt das Dateiende und endet normal.

Die Kinderteile sind angenehm zu schreiben, weil jedes Kind einfach liest sys.stdin.

Die Eltern haben ein wenig ausgefallene Beinarbeit, um alle Kinder zu laichen und die Pfeifen richtig zu halten, aber es ist nicht so schlimm.

Fan-In ist die entgegengesetzte Struktur. Eine Reihe von unabhängig laufenden Prozessen muss ihre Eingaben zu einem gemeinsamen Prozess verschachteln. Der Sammler ist nicht so einfach zu schreiben, da er aus vielen Quellen lesen muss.

Das Lesen von vielen benannten Pipes erfolgt häufig mithilfe des selectModuls, um festzustellen, welche Pipes ausstehende Eingaben haben.

Lösung 3

Shared Lookup ist die Definition einer Datenbank.

Lösung 3A - Laden Sie eine Datenbank. Lassen Sie die Mitarbeiter die Daten in der Datenbank verarbeiten.

Lösung 3B - Erstellen Sie einen sehr einfachen Server mit werkzeug (oder ähnlichem), um WSGI-Anwendungen bereitzustellen, die auf HTTP GET reagieren, damit die Mitarbeiter den Server abfragen können.

Lösung 4

Freigegebenes Dateisystemobjekt. Unix OS bietet Shared Memory-Objekte. Dies sind nur Dateien, die dem Speicher zugeordnet sind, sodass das Austauschen von E / A anstelle von mehr konventionell gepufferten Lesevorgängen erfolgt.

Sie können dies aus einem Python-Kontext auf verschiedene Arten tun

1. Schreiben Sie ein Startprogramm, das (1) Ihr ursprüngliches gigantisches Objekt in kleinere Objekte zerlegt und (2) Arbeiter mit jeweils einem kleineren Objekt startet. Die kleineren Objekte könnten eingelegte Python-Objekte sein, um ein wenig Zeit beim Lesen von Dateien zu sparen.

2. Schreiben Sie ein Startprogramm, das (1) Ihr ursprüngliches gigantisches Objekt liest und mithilfe von seekOperationen eine seitenstrukturierte, bytecodierte Datei schreibt, um sicherzustellen, dass einzelne Abschnitte mit einfachen Suchvorgängen leicht zu finden sind. Dies ist, was ein Datenbankmodul tut - die Daten in Seiten aufteilen, jede Seite über a leicht zu finden machen seek.

   Spawn-Mitarbeiter mit Zugriff auf diese große Datei mit Seitenstruktur. Jeder Arbeiter kann nach den relevanten Teilen suchen und dort seine Arbeit erledigen.

Werden untergeordnete Prozesse über Multiprozessor-Freigabeobjekte erzeugt, die zuvor im Programm erstellt wurden?

Es hängt davon ab, ob. Bei globalen schreibgeschützten Variablen kann dies häufig in Betracht gezogen werden (abgesehen vom verbrauchten Speicher), andernfalls sollte dies nicht der Fall sein.

In der Dokumentation zu Multiprocessing heißt es:

Better to inherit than pickle/unpickle

Unter Windows müssen viele Typen aus der Mehrfachverarbeitung auswählbar sein, damit untergeordnete Prozesse sie verwenden können. Im Allgemeinen sollte jedoch vermieden werden, gemeinsam genutzte Objekte mithilfe von Pipes oder Warteschlangen an andere Prozesse zu senden. Stattdessen sollten Sie das Programm so anordnen, dass ein Prozess, der Zugriff auf eine an anderer Stelle erstellte freigegebene Ressource benötigt, diese von einem Vorgängerprozess erben kann.

Explicitly pass resources to child processes

Unter Unix kann ein untergeordneter Prozess eine gemeinsam genutzte Ressource verwenden, die in einem übergeordneten Prozess mithilfe einer globalen Ressource erstellt wurde. Es ist jedoch besser, das Objekt als Argument an den Konstruktor für den untergeordneten Prozess zu übergeben.

Abgesehen davon, dass der Code (möglicherweise) mit Windows kompatibel ist, wird dadurch auch sichergestellt, dass das Objekt im übergeordneten Prozess nicht durch Müll gesammelt wird, solange der untergeordnete Prozess noch aktiv ist. Dies kann wichtig sein, wenn eine Ressource freigegeben wird, wenn das Objekt im übergeordneten Prozess als Müll gesammelt wird.

Global variables

Beachten Sie, dass wenn Code, der in einem untergeordneten Prozess ausgeführt wird, versucht, auf eine globale Variable zuzugreifen, der angezeigte Wert (falls vorhanden) möglicherweise nicht mit dem Wert im übergeordneten Prozess zum Zeitpunkt des Aufrufs von Process.start () übereinstimmt .

Beispiel

Unter Windows (einzelne CPU):

#!/usr/bin/env python
import os, sys, time
from multiprocessing import Pool

x = 23000 # replace `23` due to small integers share representation
z = []    # integers are immutable, let's try mutable object

def printx(y):
    global x
    if y == 3:
       x = -x
    z.append(y)
    print os.getpid(), x, id(x), z, id(z) 
    print y
    if len(sys.argv) == 2 and sys.argv[1] == "sleep":
       time.sleep(.1) # should make more apparant the effect

if __name__ == '__main__':
    pool = Pool(processes=4)
    pool.map(printx, (1,2,3,4))

Mit sleep:

$ python26 test_share.py sleep
2504 23000 11639492 [1] 10774408
1
2564 23000 11639492 [2] 10774408
2
2504 -23000 11639384 [1, 3] 10774408
3
4084 23000 11639492 [4] 10774408
4

Ohne sleep:

$ python26 test_share.py
1148 23000 11639492 [1] 10774408
1
1148 23000 11639492 [1, 2] 10774408
2
1148 -23000 11639324 [1, 2, 3] 10774408
3
1148 -23000 11639324 [1, 2, 3, 4] 10774408
4

S.Lott ist richtig. Mit den Multiprozessor-Verknüpfungen von Python erhalten Sie effektiv einen separaten, doppelten Speicherblock.

Auf den meisten * nix-Systemen erhalten Sie durch die Verwendung eines Aufrufs auf niedrigerer Ebene os.fork()tatsächlich Copy-on-Write-Speicher, was möglicherweise Ihre Meinung ist. AFAIK konnte theoretisch in den einfachsten Programmen, die möglich sind, aus diesen Daten lesen, ohne sie duplizieren zu lassen.

Im Python-Interpreter sind die Dinge jedoch nicht ganz so einfach. Objektdaten und Metadaten werden im selben Speichersegment gespeichert. Selbst wenn sich das Objekt nie ändert, führt ein Referenzzähler für das inkrementierte Objekt zu einem Speicherschreibvorgang und damit zu einer Kopie. Fast jedes Python-Programm, das mehr als "Hallo drucken" tut, führt zu Inkrementen der Referenzanzahl, sodass Sie den Vorteil des Copy-on-Write wahrscheinlich nie erkennen werden.

Selbst wenn es jemandem gelungen wäre, eine Shared-Memory-Lösung in Python zu hacken, wäre der Versuch, die Speicherbereinigung prozessübergreifend zu koordinieren, wahrscheinlich ziemlich schmerzhaft.

Wenn Sie unter Unix ausgeführt werden, können sie aufgrund der Funktionsweise von Fork dasselbe Objekt gemeinsam nutzen (dh die untergeordneten Prozesse verfügen über einen separaten Speicher, sie werden jedoch beim Schreiben kopiert, sodass sie möglicherweise gemeinsam genutzt werden, solange niemand sie ändert). Ich habe folgendes versucht:

import multiprocessing

x = 23

def printx(y):
    print x, id(x)
    print y

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=4)
    pool.map(printx, (1,2,3,4))

und bekam die folgende Ausgabe:

$ ./mtest.py
23 22995656
1
23 22995656
2
23 22995656
3
23 22995656
4

Dies beweist natürlich nicht , dass keine Kopie erstellt wurde, aber Sie sollten dies in Ihrer Situation überprüfen können, indem Sie die Ausgabe von überprüfen, um festzustellen, pswie viel realen Speicher jeder Unterprozess verwendet.

Unterschiedliche Prozesse haben unterschiedliche Adressräume. Als würde man verschiedene Instanzen des Interpreters ausführen. Dafür ist IPC (Interprozesskommunikation) gedacht.

Zu diesem Zweck können Sie entweder Warteschlangen oder Pipes verwenden. Sie können rpc auch über tcp verwenden, wenn Sie die Prozesse später über ein Netzwerk verteilen möchten.

http://docs.python.org/dev/library/multiprocessing.html#exchanging-objects-between-processes

Nicht direkt mit Multiprocessing an sich verbunden, aber aus Ihrem Beispiel geht hervor, dass Sie einfach das Regalmodul oder ähnliches verwenden könnten . Muss das "big_lookup_object" wirklich vollständig im Speicher sein?

Nein, aber Sie können Ihre Daten als untergeordneten Prozess laden und ihm erlauben, seine Daten mit anderen untergeordneten Prozessen zu teilen. siehe unten.

import time
import multiprocessing

def load_data( queue_load, n_processes )

    ... load data here into some_variable

    """
    Store multiple copies of the data into
    the data queue. There needs to be enough
    copies available for each process to access. 
    """

    for i in range(n_processes):
        queue_load.put(some_variable)


def work_with_data( queue_data, queue_load ):

    # Wait for load_data() to complete
    while queue_load.empty():
        time.sleep(1)

    some_variable = queue_load.get()

    """
    ! Tuples can also be used here
    if you have multiple data files
    you wish to keep seperate.  
    a,b = queue_load.get()
    """

    ... do some stuff, resulting in new_data

    # store it in the queue
    queue_data.put(new_data)


def start_multiprocess():

    n_processes = 5

    processes = []
    stored_data = []

    # Create two Queues
    queue_load = multiprocessing.Queue()
    queue_data = multiprocessing.Queue()

    for i in range(n_processes):

        if i == 0:
            # Your big data file will be loaded here...
            p = multiprocessing.Process(target = load_data,
            args=(queue_load, n_processes))

            processes.append(p)
            p.start()   

        # ... and then it will be used here with each process
        p = multiprocessing.Process(target = work_with_data,
        args=(queue_data, queue_load))

        processes.append(p)
        p.start()

    for i in range(n_processes)
        new_data = queue_data.get()
        stored_data.append(new_data)    

    for p in processes:
        p.join()
    print(processes)    

Für die Linux / Unix / MacOS-Plattform ist forkmap eine schnelle und schmutzige Lösung.