Przetwarzanie wieloprocesowe - rura kontra kolejka

Odpowiedzi:

281
  • A Pipe()może mieć tylko dwa punkty końcowe.

  • Queue()Może mieć wielu producentów i konsumentów.

Kiedy ich używać

Jeśli potrzebujesz więcej niż dwóch punktów do komunikacji, użyj pliku Queue().

Jeśli potrzebujesz absolutnej wydajności, a Pipe()jest znacznie szybszy, ponieważ Queue()jest zbudowany na wierzchu Pipe().

Benchmarking wydajności

Załóżmy, że chcesz uruchomić dwa procesy i jak najszybciej wysyłać między nimi wiadomości. Oto wyniki wyścigu równoległego między podobnymi testami przy użyciu Pipe()i Queue()... To jest na ThinkpadT61 z systemem Ubuntu 11.10 i Python 2.7.2.

FYI, wrzuciłem wyniki JoinableQueue()jako bonus; JoinableQueue()rozlicza zadania, gdy queue.task_done()jest wywoływane (nawet nie wie o konkretnym zadaniu, po prostu liczy zadania niedokończone w kolejce), dzięki czemu queue.join()wie, że praca jest zakończona.

Kod dla każdego na dole tej odpowiedzi ...

mpenning@mpenning-T61:~$ python multi_pipe.py 
Sending 10000 numbers to Pipe() took 0.0369849205017 seconds
Sending 100000 numbers to Pipe() took 0.328398942947 seconds
Sending 1000000 numbers to Pipe() took 3.17266988754 seconds
mpenning@mpenning-T61:~$ python multi_queue.py 
Sending 10000 numbers to Queue() took 0.105256080627 seconds
Sending 100000 numbers to Queue() took 0.980564117432 seconds
Sending 1000000 numbers to Queue() took 10.1611330509 seconds
mpnening@mpenning-T61:~$ python multi_joinablequeue.py 
Sending 10000 numbers to JoinableQueue() took 0.172781944275 seconds
Sending 100000 numbers to JoinableQueue() took 1.5714070797 seconds
Sending 1000000 numbers to JoinableQueue() took 15.8527247906 seconds
mpenning@mpenning-T61:~$

Podsumowując, Pipe()jest około trzy razy szybszy niż plik Queue(). Nawet nie myśl o tym, JoinableQueue()chyba że naprawdę musisz mieć korzyści.

MATERIAŁ BONUSOWY 2

Wieloprocesorowość wprowadza subtelne zmiany w przepływie informacji, które utrudniają debugowanie, chyba że znasz kilka skrótów. Na przykład możesz mieć skrypt, który działa dobrze podczas indeksowania przez słownik w wielu warunkach, ale rzadko kończy się niepowodzeniem z określonymi danymi wejściowymi.

Zwykle wskazówki dotyczące niepowodzenia otrzymujemy, gdy cały proces Pythona ulega awarii; jednakże, nie otrzymujesz niezamawianych informacji o błędach wypisywanych na konsoli, jeśli funkcja wieloprocesorowa ulegnie awarii. Wyśledzenie nieznanych awarii przetwarzania wieloprocesowego jest trudne bez pojęcia, co spowodowało awarię procesu.

Najprostszym sposobem, w jaki znalazłem, aby wyśledzić informacje o awariach wieloprocesorowych, jest zawinięcie całej funkcji wieloprocesowej w a try/ excepti użycie traceback.print_exc():

import traceback
def run(self, args):
    try:
        # Insert stuff to be multiprocessed here
        return args[0]['that']
    except:
        print "FATAL: reader({0}) exited while multiprocessing".format(args) 
        traceback.print_exc()

Teraz, gdy znajdziesz awarię, zobaczysz coś takiego:

FATAL: reader([{'crash': 'this'}]) exited while multiprocessing
Traceback (most recent call last):
  File "foo.py", line 19, in __init__
    self.run(args)
  File "foo.py", line 46, in run
    KeyError: 'that'

Kod źródłowy:


"""
multi_pipe.py
"""
from multiprocessing import Process, Pipe
import time

def reader_proc(pipe):
    ## Read from the pipe; this will be spawned as a separate Process
    p_output, p_input = pipe
    p_input.close()    # We are only reading
    while True:
        msg = p_output.recv()    # Read from the output pipe and do nothing
        if msg=='DONE':
            break

def writer(count, p_input):
    for ii in xrange(0, count):
        p_input.send(ii)             # Write 'count' numbers into the input pipe
    p_input.send('DONE')

if __name__=='__main__':
    for count in [10**4, 10**5, 10**6]:
        # Pipes are unidirectional with two endpoints:  p_input ------> p_output
        p_output, p_input = Pipe()  # writer() writes to p_input from _this_ process
        reader_p = Process(target=reader_proc, args=((p_output, p_input),))
        reader_p.daemon = True
        reader_p.start()     # Launch the reader process

        p_output.close()       # We no longer need this part of the Pipe()
        _start = time.time()
        writer(count, p_input) # Send a lot of stuff to reader_proc()
        p_input.close()
        reader_p.join()
        print("Sending {0} numbers to Pipe() took {1} seconds".format(count,
            (time.time() - _start)))

"""
multi_queue.py
"""

from multiprocessing import Process, Queue
import time
import sys

def reader_proc(queue):
    ## Read from the queue; this will be spawned as a separate Process
    while True:
        msg = queue.get()         # Read from the queue and do nothing
        if (msg == 'DONE'):
            break

def writer(count, queue):
    ## Write to the queue
    for ii in range(0, count):
        queue.put(ii)             # Write 'count' numbers into the queue
    queue.put('DONE')

if __name__=='__main__':
    pqueue = Queue() # writer() writes to pqueue from _this_ process
    for count in [10**4, 10**5, 10**6]:             
        ### reader_proc() reads from pqueue as a separate process
        reader_p = Process(target=reader_proc, args=((pqueue),))
        reader_p.daemon = True
        reader_p.start()        # Launch reader_proc() as a separate python process

        _start = time.time()
        writer(count, pqueue)    # Send a lot of stuff to reader()
        reader_p.join()         # Wait for the reader to finish
        print("Sending {0} numbers to Queue() took {1} seconds".format(count, 
            (time.time() - _start)))

"""
multi_joinablequeue.py
"""
from multiprocessing import Process, JoinableQueue
import time

def reader_proc(queue):
    ## Read from the queue; this will be spawned as a separate Process
    while True:
        msg = queue.get()         # Read from the queue and do nothing
        queue.task_done()

def writer(count, queue):
    for ii in xrange(0, count):
        queue.put(ii)             # Write 'count' numbers into the queue

if __name__=='__main__':
    for count in [10**4, 10**5, 10**6]:
        jqueue = JoinableQueue() # writer() writes to jqueue from _this_ process
        # reader_proc() reads from jqueue as a different process...
        reader_p = Process(target=reader_proc, args=((jqueue),))
        reader_p.daemon = True
        reader_p.start()     # Launch the reader process
        _start = time.time()
        writer(count, jqueue) # Send a lot of stuff to reader_proc() (in different process)
        jqueue.join()         # Wait for the reader to finish
        print("Sending {0} numbers to JoinableQueue() took {1} seconds".format(count, 
            (time.time() - _start)))
Mike Pennington
źródło
2
@Jonathan „Podsumowując, Pipe () jest około trzy razy szybsze niż Queue ()”
James Brady,
13
Doskonały! Dobra odpowiedź i miło, że podałeś testy! Mam tylko dwa drobne wątpliwości: (1) „rzędy wielkości szybciej” to trochę przesada. Różnica wynosi x3, co stanowi około jednej trzeciej jednego rzędu wielkości. Tylko mówię. ;-); i (2) bardziej sprawiedliwe porównanie to uruchomienie N procesów roboczych, z których każdy komunikuje się z głównym wątkiem za pośrednictwem potoku punkt-punkt, w porównaniu z wydajnością uruchomionych N procesów roboczych pobierających z pojedynczej kolejki punkt-wielopunkt.
JJC
3
Do twojego "materiału bonusowego" ... Tak. Jeśli tworzysz podklasę Process, umieść większość metody „run” w bloku try. Jest to również przydatny sposób rejestrowania wyjątków. Aby zreplikować normalne wyjście wyjątku: sys.stderr.write (''. Join (traceback.format_exception (* (sys.exc_info ()))))
travc
2
@ alexpinho98 - ale będziesz potrzebować pewnych danych spoza pasma i związanego z nimi trybu sygnalizacji, aby wskazać, że wysyłane są nie zwykłe dane, ale dane o błędach. biorąc pod uwagę, że proces początkowy jest już w nieprzewidywalnym stanie, może to być zbyt trudne pytanie.
scytale
10
@JJC Aby spierać się ze swoim sporem, 3x to około połowy rzędu wielkości, a nie jedna trzecia - sqrt (10) = ~ 3.
dźgnij
1

Dodatkową cechą, na Queue()którą warto zwrócić uwagę, jest nić podajnika. W tej sekcji zauważono: „Kiedy proces po raz pierwszy umieszcza element w kolejce, uruchamiany jest wątek podajnika, który przesyła obiekty z bufora do potoku”. Nieskończoną liczbę (lub maksymalny rozmiar) elementów można wstawić Queue()bez wywołań queue.put()blokowania. Pozwala to na przechowywanie wielu elementów w a Queue(), dopóki program nie będzie gotowy do ich przetworzenia.

Pipe()z drugiej strony ma skończoną ilość miejsca do przechowywania elementów, które zostały wysłane do jednego połączenia, ale nie zostały odebrane z drugiego połączenia. Po wykorzystaniu tej pamięci wywołania connection.send()będą blokowane do momentu, gdy będzie miejsce na zapisanie całego elementu. Spowoduje to wstrzymanie pisania przez wątek, dopóki inny wątek nie odczyta z potoku. Connectionobiekty dają dostęp do bazowego deskryptora pliku. W systemach * nix możesz zapobiec connection.send()blokowaniu połączeń przy użyciu tej os.set_blocking()funkcji. Jednak spowoduje to problemy, jeśli spróbujesz wysłać pojedynczy element, który nie mieści się w pliku potoku. Najnowsze wersje systemu Linux pozwalają na zwiększenie rozmiaru pliku, ale maksymalny dozwolony rozmiar zależy od konfiguracji systemu. Dlatego nigdy nie należy polegać na Pipe()buforowaniu danych. Wzywa doconnection.send może blokować, dopóki dane nie zostaną odczytane z potoku w innym miejscu.

Podsumowując, kolejka jest lepszym wyborem niż potok, gdy trzeba buforować dane. Nawet jeśli potrzebujesz tylko komunikacji między dwoma punktami.

Roger Iyengar
źródło