Multiprocessing di Python - Pipe vs Queue

Question

Quali sono le differenze fondamentali tra code e pipe in Python's multiprocessing package?

In quali scenari si dovrebbe scegliere uno rispetto all'altro? Quando è vantaggioso utilizzare Pipe()? Quando è vantaggioso utilizzare Queue()?

Answer 1

• A Pipe() può avere solo due endpoint.

• A Queue() può avere più produttori e consumatori.

Quando usarli

Se avete bisogno di più di due punti di comunicare, utilizzare un Queue().

Se avete bisogno di prestazioni assolute, un Pipe() è molto più veloce perché è stato costruito su .

Benchmark delle prestazioni

Supponiamo di voler deporre le uova due processi e inviare messaggi tra di loro il più rapidamente possibile. Questi sono i risultati dei tempi di una corsa di resistenza tra test simili usando Pipe() e Queue() ... Questo è su un ThinkpadT61 con Ubuntu 11.10 e Python 2.7.2.

FYI, ho inserito i risultati per JoinableQueue() come bonus; JoinableQueue() account per le attività quando viene chiamato queue.task_done() (non sa nemmeno dell'attività specifica, conta solo le attività non completate nella coda), in modo che queue.join() sappia che il lavoro è finito.

Il codice per ciascuna in fondo a questa risposta ...

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:~$ 

In sintesi Pipe() è circa tre volte più veloce di un Queue(). Non pensare nemmeno allo JoinableQueue() a meno che tu non debba davvero avere i benefici.

BONUS MATERIALE 2

Multiprocessing introduce sottili cambiamenti nel flusso di informazioni che rendono debug difficile se non si conosce alcune scorciatoie. Ad esempio, potresti avere uno script che funziona bene quando indicizzi attraverso un dizionario in molte condizioni, ma raramente fallisce con determinati input.

Normalmente otteniamo indizi sull'errore quando l'intero processo Python si interrompe; tuttavia, non si ottengono traceback degli arresti anomali non richiesti stampati sulla console se la funzione di multiprocessing si interrompe. Rintracciare i crash di multiprocessing sconosciuti è difficile senza avere idea di cosa abbia causato il crash del processo.

Il modo più semplice che ho trovato per rintracciare multiprocessing incidente informaiton è quello di avvolgere l'intera funzione multiprocessing in un try/except e utilizzare traceback.print_exc():

import traceback 
def reader(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() 

Ora, quando si trova un crash si vede qualcosa di simile : Codice

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

Fonte:

---

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

def reader(pipe): 
    output_p, input_p = pipe 
    input_p.close() # We are only reading 
    while True: 
     try: 
      msg = output_p.recv() # Read from the output pipe and do nothing 
     except EOFError: 
      break 

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

if __name__=='__main__': 
    for count in [10**4, 10**5, 10**6]: 
     output_p, input_p = Pipe() 
     reader_p = Process(target=reader, args=((output_p, input_p),)) 
     reader_p.start()  # Launch the reader process 

     output_p.close()  # We no longer need this part of the Pipe() 
     _start = time.time() 
     writer(count, input_p) # Send a lot of stuff to reader() 
     input_p.close()  # Ask the reader to stop when it reads EOF 
     reader_p.join() 
     print "Sending %s numbers to Pipe() took %s seconds" % (count, 
      (time.time() - _start)) 

---

""" 
multi_queue.py 
""" 
from multiprocessing import Process, Queue 
import time 

def reader(queue): 
    while True: 
     msg = queue.get()   # Read from the queue and do nothing 
     if (msg == 'DONE'): 
      break 

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

if __name__=='__main__': 
    for count in [10**4, 10**5, 10**6]: 
     queue = Queue() # reader() reads from queue 
          # writer() writes to queue 
     reader_p = Process(target=reader, args=((queue),)) 
     reader_p.daemon = True 
     reader_p.start()  # Launch the reader process 

     _start = time.time() 
     writer(count, queue) # Send a lot of stuff to reader() 
     reader_p.join()   # Wait for the reader to finish 
     print "Sending %s numbers to Queue() took %s seconds" % (count, 
      (time.time() - _start)) 

---

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

def reader(queue): 
    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]: 
     queue = JoinableQueue() # reader() reads from queue 
            # writer() writes to queue 
     reader_p = Process(target=reader, args=((queue),)) 
     reader_p.daemon = True 
     reader_p.start()  # Launch the reader process 

     _start = time.time() 
     writer(count, queue) # Send a lot of stuff to reader() 
     queue.join()   # Wait for the reader to finish 
     print "Sending %s numbers to JoinableQueue() took %s seconds" % (count, 
      (time.time() - _start))