Multiprocessing - Pipe vs Queue

Quelles sont les différences fondamentales entre les files d'attente et les tubes dans le package multiprocesseur de Python ?

Dans quels scénarios faut-il choisir l'un plutôt que l'autre? Quand est-il avantageux de l'utiliser Pipe()? Quand est-il avantageux de l'utiliser Queue()?

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

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

Quand les utiliser

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

Si vous avez besoin de performances absolues, a Pipe()est beaucoup plus rapide car il 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. Voici les résultats de chronométrage d'une course de dragsters entre des tests similaires utilisant Pipe()et Queue()... Ceci est sur un ThinkpadT61 exécutant Ubuntu 11.10 et Python 2.7.2.

Pour info, j'ai ajouté des résultats JoinableQueue()en prime; JoinableQueue()tient compte des tâches lorsqu'il queue.task_done()est appelé (il ne connaît même pas la tâche spécifique, il compte simplement les tâches inachevées dans la file d'attente), de sorte que queue.join()sache que le travail est terminé.

Le code pour chacun au 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()c'est environ trois fois plus rapide qu'un Queue(). Ne pensez même pas à JoinableQueue()moins que vous en ayez vraiment les avantages.

MATÉRIEL BONUS 2

Le multitraitement introduit des changements subtils dans le flux 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 via un dictionnaire dans de nombreuses conditions, mais qui échoue rarement avec certaines entrées.

Normalement, nous obtenons des indices sur l'échec lorsque tout le processus python plante; cependant, vous n'obtenez pas de traces de crash non sollicitées imprimées sur la console si la fonction de multitraitement tombe en panne. Traquer les plantages de multitraitement inconnus est difficile sans une idée de ce qui a bloqué le processus.

Le moyen le plus simple que j'ai trouvé pour traquer les informations de plantage du multitraitement est d'envelopper toute la fonction de multitraitement dans un try/ exceptet d'utiliser 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()

Maintenant, lorsque 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(args)
  File "foo.py", line 46, in run
    KeyError: 'that'

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

Une caractéristique supplémentaire Queue()à noter est le fil d'alimentation. Cette section indique "Lorsqu'un processus met pour la première fois un élément dans la file d'attente, un thread d'alimentation est démarré qui transfère les objets d'un tampon dans le tube." Un nombre infini d'éléments (ou maxsize) peut être inséré dans Queue()sans aucun appel au queue.put()blocage. Cela vous permet de stocker plusieurs éléments dans un Queue(), jusqu'à ce que votre programme soit prêt à les traiter.

Pipe(), d'autre part, a une quantité limitée de stockage pour les éléments qui ont été envoyés à une connexion, mais qui n'ont pas été reçus de l'autre connexion. Une fois ce stockage épuisé, les appels à connection.send()se bloqueront jusqu'à ce qu'il y ait de l'espace pour écrire l'intégralité de l'élément. Cela bloquera le thread en train d'écrire jusqu'à ce qu'un autre thread lise à partir du tube. Connectionles objets vous donnent accès au descripteur de fichier sous-jacent. Sur les systèmes * nix, vous pouvez empêcher le connection.send()blocage des appels à l'aide de la os.set_blocking()fonction. Cependant, cela posera des problèmes si vous essayez d'envoyer un seul élément qui ne rentre pas dans le fichier du canal. Les versions récentes de Linux vous permettent d'augmenter la taille d'un fichier, mais la taille maximale autorisée varie en fonction des configurations système. Vous ne devez donc jamais vous fier à la mise Pipe()en mémoire tampon des données. Appels àconnection.send pourrait bloquer jusqu'à ce que les données soient lues à partir du tube quelque part.

En conclusion, Queue est un meilleur choix que pipe lorsque vous avez besoin de tamponner des données. Même lorsque vous n'avez besoin de communiquer qu'entre deux points.