Multitraitement Python - Pipe vs file d’attente

Question

Quelles sont les différences fondamentales entre les tuyaux et les files d’attente dans le paquet Python multitraitement?

Dans quels scénarios doit on choisir un sur l’autre ? Quand est-ce avantageux d’utiliser ? Quand est-ce avantageux d’utiliser ?

Answer 1

• Un Pipe() ne peut avoir que deux points d'extrémité.

• Un Queue() peut avoir plusieurs producteurs et les consommateurs.

Quand les utiliser

Si vous avez besoin de plus de deux points de communiquer, d'utiliser un Queue().

Si vous avez besoin de performance absolue, Pipe() est beaucoup plus rapide, car Queue() est construit au sommet d' Pipe().

Analyse Comparative De La Performance

Imaginons que vous voulez lancer deux processus et envoyer des messages entre eux le plus rapidement possible. Ce sont les résultats du réglage d'une course de drag entre des tests à l'aide d' Pipe() et Queue()... C'est sur un ThinkpadT61 sous Ubuntu 11.10, et Python 2.7.2.

Pour info, je l'ai jeté dans les résultats pour JoinableQueue() comme un bonus, JoinableQueue() comptes pour des tâches lors de l' queue.task_done() est appelé (il ne sait même pas à propos de la tâche, il a juste compte les tâches en cours dans la file d'attente), de sorte qu' queue.join() sait que le travail est terminé.

Le code pour chaque 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() est environ trois fois plus rapide qu'un Queue(). Ne pensez même pas à l' JoinableQueue() , sauf si vous devez vraiment avoir les avantages.

BONUS 2

Multitraitement introduit des changements subtils dans le flux d'informations qui font de débogage difficile, sauf si vous savez quelques raccourcis. Par exemple, vous pourriez avoir un script qui fonctionne très bien lors de l'indexation par le biais d'un dictionnaire en sous de nombreuses conditions, mais rarement échoue avec certaines entrées.

Normalement, nous obtenir des indices à l'échec lorsque l'ensemble de python le processus se bloque; cependant, vous ne recevez crash retraçage imprimé sur la console si le multitraitement fonction des plantages. La traque de l'inconnu multitraitement accidents est dur sans la moindre idée de ce qui s'est écrasé le processus.

La façon la plus simple que j'ai trouvé la trace de multitraitement crash de détails est de renvoyer l'ensemble de multitraitement fonction en try / except et l'utilisation sys.exc_info():

import sys
def reader(args):
    try:
        # Insert stuff to be multiprocessed here
        return args[0]['that']
    except:
        print "reader(%s) exited with '%s' while multiprocessing" % (args, 
            sys.exc_info())

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

reader([{'crash', 'this'}]) exited with '(<type 'exceptions.KeyError'>, 
KeyError(0,), <traceback object at 0x287bdd0>)' while multiprocessing

Code Source:

---

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