przetwarzanie wieloprocesowe: współdzielenie dużego obiektu tylko do odczytu między procesami?

Czy procesy potomne powstały w wyniku współdzielonych obiektów wieloprocesowych utworzonych wcześniej w programie?

Mam następującą konfigurację:

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'))

Ładuję duży obiekt do pamięci, a następnie tworzę pulę pracowników, którzy muszą z niego korzystać. Duży obiekt jest dostępny tylko do odczytu, nie muszę przekazywać jego modyfikacji między procesami.

Moje pytanie brzmi: czy duży obiekt jest ładowany do pamięci współdzielonej, tak jak by to było, gdybym uruchomił proces w unix / c, czy też każdy proces ładuje własną kopię dużego obiektu?

Aktualizacja: aby wyjaśnić dalej - big_lookup_object jest współdzielonym obiektem wyszukiwania. Nie muszę tego dzielić i przetwarzać osobno. Muszę zachować jedną kopię. Praca, której potrzebuję, aby go podzielić, polega na czytaniu wielu innych dużych plików i sprawdzaniu elementów w tych dużych plikach względem obiektu wyszukiwania.

Dalsza aktualizacja: baza danych jest dobrym rozwiązaniem, memcached może być lepszym rozwiązaniem, a plik na dysku (półka lub dbm) może być jeszcze lepszy. W tym pytaniu byłem szczególnie zainteresowany rozwiązaniem w pamięci. Jako ostateczne rozwiązanie użyję hadoopa, ale chciałem sprawdzić, czy mogę również mieć lokalną wersję w pamięci.

„Czy procesy potomne powstały w wyniku współdzielonych obiektów wieloprocesowych utworzonych wcześniej w programie?”

Nie (python przed 3.8) i tak w 3.8 ( https://docs.python.org/3/library/multiprocessing.shared_memory.html#module-multiprocessing.shared_memory )

Procesy mają niezależną przestrzeń pamięci.

Rozwiązanie 1

Aby jak najlepiej wykorzystać dużą strukturę z dużą liczbą pracowników, zrób to.

1. Zapisz każdego pracownika jako „filtr” - odczytuje wyniki pośrednie ze standardowego wejścia, działa, zapisuje wyniki pośrednie na standardowe wyjście.

2. Połącz wszystkich pracowników jako rurociąg:

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

Każdy proces czyta, pracuje i pisze.

Jest to niezwykle wydajne, ponieważ wszystkie procesy działają jednocześnie. Zapisy i odczyty przechodzą bezpośrednio przez bufory współdzielone między procesami.

Rozwiązanie 2

W niektórych przypadkach masz bardziej złożoną strukturę - często strukturę „rozłożoną”. W takim przypadku masz rodzica z kilkoma dziećmi.

1. Rodzic otwiera dane źródłowe. Rodzic rozwidla wiele dzieci.

2. Rodzic czyta źródło, wypuszcza części źródła do każdego współbieżnie działającego dziecka.

3. Kiedy rodzic dojdzie do końca, zamknij rurkę. Dziecko otrzymuje koniec pliku i kończy normalnie.

Części dziecięce są przyjemne w pisaniu, ponieważ każde dziecko po prostu czyta sys.stdin.

Rodzic ma trochę wymyślnej pracy nóg przy odradzaniu wszystkich dzieci i prawidłowym utrzymywaniu rur, ale nie jest tak źle.

Fan-in to odwrotna konstrukcja. Wiele niezależnie działających procesów musi przeplatać swoje dane wejściowe do wspólnego procesu. Kolektor nie jest tak łatwy do napisania, ponieważ musi czytać z wielu źródeł.

Czytanie z wielu nazwanych potoków jest często wykonywane za pomocą selectmodułu, aby zobaczyć, które potoki mają oczekujące dane wejściowe.

Rozwiązanie 3

Wspólne wyszukiwanie to definicja bazy danych.

Rozwiązanie 3A - załaduj bazę danych. Pozwól pracownikom przetwarzać dane w bazie danych.

Rozwiązanie 3B - utwórz bardzo prosty serwer przy użyciu werkzeug (lub podobnego), aby zapewnić aplikacje WSGI, które odpowiadają na HTTP GET, aby pracownicy mogli wysyłać zapytania do serwera.

Rozwiązanie 4

Współdzielony obiekt systemu plików. Unix OS oferuje współdzielone obiekty pamięci. Są to tylko pliki, które są mapowane do pamięci, dzięki czemu zamiana we / wy jest wykonywana zamiast bardziej konwencjonalnych odczytów buforowanych.

Możesz to zrobić z kontekstu Pythona na kilka sposobów

1. Napisz program startowy, który (1) rozbija twój oryginalny gigantyczny obiekt na mniejsze obiekty i (2) uruchamia pracowników, każdy z mniejszym obiektem. Mniejsze obiekty mogą być piklowanymi obiektami Pythona, aby zaoszczędzić trochę czasu na odczyt plików.

2. Napisz program startowy, który (1) czyta twój oryginalny gigantyczny obiekt i zapisuje plik ze strukturą strony, zakodowany bajtowo, używając seekoperacji, aby zapewnić, że poszczególne sekcje są łatwe do znalezienia za pomocą prostych wyszukiwań. To właśnie robi silnik bazy danych - dzieli dane na strony, ułatwiając zlokalizowanie każdej strony za pomocą pliku seek.

   Przywołaj pracowników mających dostęp do tego dużego pliku o strukturze strony. Każdy pracownik może szukać odpowiednich części i tam wykonywać swoją pracę.

Czy procesy potomne powstały w wyniku współdzielonych obiektów wieloprocesowych utworzonych wcześniej w programie?

To zależy. W przypadku zmiennych globalnych tylko do odczytu można to często uznać za tak (oprócz zajętej pamięci), w przeciwnym razie nie powinno.

Dokumentacja multiprocessingu mówi:

Better to inherit than pickle/unpickle

W systemie Windows wiele typów z przetwarzania wieloprocesowego musi być możliwych do pobrania, aby procesy potomne mogły ich używać. Jednak generalnie należy unikać wysyłania obiektów współdzielonych do innych procesów przy użyciu potoków lub kolejek. Zamiast tego należy tak ustawić program, aby proces, który potrzebuje dostępu do współdzielonego zasobu utworzonego w innym miejscu, mógł odziedziczyć go po procesie nadrzędnym.

Explicitly pass resources to child processes

W systemie Unix proces potomny może korzystać ze współużytkowanego zasobu utworzonego w procesie nadrzędnym przy użyciu zasobu globalnego. Jednak lepiej jest przekazać obiekt jako argument do konstruktora procesu potomnego.

Oprócz uczynienia kodu (potencjalnie) kompatybilnym z Windows zapewnia to również, że dopóki proces potomny nadal żyje, obiekt nie zostanie usunięty jako śmieci w procesie nadrzędnym. Może to być ważne, jeśli jakiś zasób zostanie zwolniony, gdy obiekt zostanie usunięty w procesie nadrzędnym.

Global variables

Należy pamiętać, że jeśli kod uruchamiany w procesie potomnym próbuje uzyskać dostęp do zmiennej globalnej, wówczas wartość, którą widzi (jeśli istnieje) może nie być taka sama, jak wartość w procesie nadrzędnym w momencie wywołania metody Process.start () .

Przykład

W systemie Windows (pojedynczy procesor):

#!/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))

Z 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

Bez 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 ma rację. Skróty wieloprocesorowe Pythona skutecznie zapewniają oddzielny, zduplikowany fragment pamięci.

W większości systemów * nix użycie wywołania niższego poziomu os.fork()w rzeczywistości da ci pamięć typu „kopiuj przy zapisie”, o której myślisz. AFAIK, w teorii, w najbardziej uproszczonych programach można było czytać te dane bez ich powielania.

Jednak w przypadku interpretera Pythona sprawy nie są takie proste. Dane obiektu i metadane są przechowywane w tym samym segmencie pamięci, więc nawet jeśli obiekt nigdy się nie zmienia, coś w rodzaju licznika odniesienia dla zwiększanego obiektu spowoduje zapis do pamięci, a tym samym kopię. Prawie każdy program w języku Python, który robi więcej niż „print 'hello'”, spowoduje zwiększenie liczby odwołań, więc prawdopodobnie nigdy nie zdasz sobie sprawy z korzyści płynących z kopiowania przy zapisie.

Nawet gdyby komuś udało się zhakować rozwiązanie pamięci współdzielonej w Pythonie, próba skoordynowania czyszczenia pamięci między procesami byłaby prawdopodobnie dość bolesna.

Jeśli pracujesz pod Uniksem, mogą dzielić ten sam obiekt, ze względu na to, jak działa fork (tj. Procesy potomne mają oddzielną pamięć, ale jest to kopiowanie przy zapisie, więc może być współdzielone, o ile nikt go nie modyfikuje). Wypróbowałem następujące:

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))

i otrzymałem następujący wynik:

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

Oczywiście nie dowodzi to, że kopia nie została utworzona, ale powinieneś być w stanie to sprawdzić w swojej sytuacji, patrząc na wynik, psaby zobaczyć, ile rzeczywistej pamięci używa każdy podproces.

Różne procesy mają różną przestrzeń adresową. Jak uruchamianie różnych instancji tłumacza. Do tego służy IPC (komunikacja międzyprocesowa).

W tym celu można użyć kolejek lub potoków. Możesz także użyć rpc over tcp, jeśli chcesz później rozprowadzić procesy w sieci.

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

Nie jest to bezpośrednio związane z przetwarzaniem wieloprocesowym jako takim, ale z twojego przykładu wydaje się, że możesz po prostu użyć modułu półki lub czegoś podobnego. Czy „big_lookup_object” naprawdę musi być całkowicie w pamięci?

Nie, ale możesz załadować swoje dane jako proces podrzędny i pozwolić mu na udostępnianie swoich danych innym dzieciom. patrz poniżej.

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)    

W przypadku platformy Linux / Unix / MacOS forkmap jest szybkim rozwiązaniem.