Wieloprocesorowe przetwarzanie w Pythonie dla procesów równoległych

Question

Przykro mi, jeśli dla niektórych osób jest to zbyt proste, ale nadal nie dostaję sztuczki z wieloprocesorowania Pythona. Przeczytałem
http://pymotw.com/2/multiprocessing/basics.html i wiele innych tutoriali i przykładów, które Google daje mi ... wielu z nich również tutaj.

Cóż, moja sytuacja polega na tym, że muszę obliczyć wiele zbędnych macierzy, a następnie zapisać je w jednej macierzy. Powiedzmy, że chcę użyć 20 rdzeni (lub, że mogę użyć 20 rdzeni), ale nie udało mi się z powodzeniem korzystać z zasobu puli, ponieważ utrzymuje on procesy przy życiu aż do momentu, gdy pula "umrze". Więc pomyślałem robić coś takiego:

from multiprocessing import Process, Queue 
import numpy as np 

def f(q,i): 
    q.put(np.zeros((4,4))) 

if __name__ == '__main__': 
    q = Queue() 
    for i in range(30): 
      p = Process(target=f, args=(q,)) 
      p.start() 
      p.join() 
    result = q.get() 
    while q.empty() == False: 
      result += q.get() 
    print result 

ale to wygląda procesy nie działają równolegle, ale prowadzony kolejno (proszę mnie poprawić, jeśli się mylę), a ja nie wiem, czy umierają po wykonaniu obliczeń (tak więc w przypadku ponad 20 procesów te, które wykonały swoją część, pozostawiają rdzeń wolną dla innego procesu). Plus, dla bardzo dużej liczby (powiedzmy 100 000), przechowywanie wszystkich tych macierzy (które mogą być naprawdę duże) w kolejce będzie zużywać dużo pamięci, czyniąc kod bezużytecznym, ponieważ celem jest umieszczenie każdego wyniku w każdej iteracji. w ostatecznym wyniku, jak przy użyciu metody blokowania (i jego nabycia() i wydania()), ale jeśli ten kod nie jest przeznaczony do przetwarzania równoległego, blokada jest również bezużyteczna ...

Mam nadzieję, że ktoś może pomóc mnie.

Z góry dziękuję!

Answer 1

Masz rację, wykonują sekwencyjnie w twoim przykładzie.

p.join() powoduje, że bieżący wątek blokuje się, dopóki nie zakończy się wykonywanie. Będziesz chciał albo dołączyć do procesów indywidualnie poza pętlą for (np. Przechowując je na liście, a następnie iterując nad nim) lub używając czegoś w rodzaju numpy.Pool i apply_async z funkcją zwrotną. Umożliwi to również bezpośrednie dodanie wyników do wyników, a nie utrzymywanie obiektów w pobliżu.

Na przykład:

def f(i): 
    return i*np.identity(4) 

if __name__ == '__main__': 
    p=Pool(5) 
    result = np.zeros((4,4)) 
    def adder(value): 
     global result 
     result += value 

    for i in range(30): 
     p.apply_async(f, args=(i,), callback=adder) 
    p.close() 
    p.join() 
    print result 

Zamknięcie a następnie przystąpienie do basenu w końcu sprawia, że ​​procesy basenie za zakończone i obiekt result jest zakończona jest obliczana. Możesz również zbadać, używając Pool.imap jako rozwiązania Twojego problemu. Że dane rozwiązanie będzie wyglądać mniej więcej tak:

if __name__ == '__main__': 
    p=Pool(5) 
    result = np.zeros((4,4)) 

    im = p.imap_unordered(f, range(30), chunksize=5) 

    for x in im: 
     result += x 

    print result 

To jest środek czyszczący do konkretnej sytuacji, ale nie może być za cokolwiek ostatecznie zrobić.

Jeśli chodzi o przechowywanie wszystkich zróżnicowanych wyników, jeśli rozumiem Twoje pytanie, możesz po prostu dodać je do wyniku w metodzie oddzwaniania (jak wyżej) lub element-o-czasie używając imap/imap_unordered (która nadal przechowuje wyniki, ale wyczyścisz je podczas jego tworzenia). Następnie nie trzeba go przechowywać dłużej, niż wymaga to dodania do wyniku.