Multiprocessing Python – Pipe vs Queue

Quelles sont les différences fondamentales entre les files d’attente et les canaux dans le package multi-traitement de Python ?

Dans quels scénarios choisir l’un sur l’autre? Quand est-il avantageux d’utiliser Pipe() ? Quand est-il avantageux d’utiliser Queue() ?

  • Un Pipe() ne peut avoir que deux points de terminaison.

  • Une Queue() peut avoir plusieurs producteurs et consommateurs.

Quand les utiliser

Si vous avez besoin de plus de deux points pour communiquer, utilisez une Queue() .

Si vous avez besoin de performances absolues, un Pipe() est beaucoup plus rapide car Queue() est construit sur Pipe() .

Analyse comparative des performances

Supposons que vous souhaitiez générer deux processus et envoyer des messages entre eux aussi rapidement que possible. Ce sont les résultats de timing d’une course de glisser entre des tests similaires utilisant Pipe() et Queue() … Ceci est sur un ThinkpadT61 sous Ubuntu 11.10, et Python 2.7.2.

Pour info, j’ai JoinableQueue() résultats pour JoinableQueue() en bonus; JoinableQueue() comptabilise les tâches lorsque queue.task_done() est appelé (il ne connaît même pas la tâche spécifique, il ne compte que les tâches inachevées dans la queue), afin que queue.join() sache que le travail est terminé.

Le code pour chacun en bas de cette réponse …

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

En résumé, Pipe() est environ trois fois plus rapide qu’une Queue() . Ne pensez même pas à JoinableQueue() moins que vous en ayez vraiment besoin.

BONUS MATERIAL 2

Le multitraitement introduit des modifications subtiles du stream d’informations qui rendent le débogage difficile à moins que vous ne connaissiez certains raccourcis. Par exemple, vous pouvez avoir un script qui fonctionne correctement lors de l’indexation dans un dictionnaire dans de nombreuses conditions, mais échoue rarement avec certaines entrées.

Normalement, nous obtenons des indices de défaillance lorsque le processus python entier se bloque; cependant, vous n’obtenez pas les retraits de pannes non sollicités imprimés sur la console si la fonction de multitraitement se bloque. Le suivi des pannes de multitraitement inconnues est difficile sans la moindre idée de ce qui a provoqué le crash du processus.

La manière la plus simple de trouver des informations sur le plantage du multitraitement consiste à envelopper la fonction de multitraitement dans un try / except et à utiliser 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() 

Maintenant, quand vous trouvez un crash, vous voyez quelque chose comme:

 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 

Code source:


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