proceso - python 2 multiprocessing

Multiproceso de Python-Pipe vs Queue

Question

¿Cuáles son las diferencias fundamentales entre las colas y las tuberías en el paquete de multiprocesamiento de Python ?

¿En qué escenarios debería uno elegir uno sobre el otro? ¿Cuándo es ventajoso usar Pipe() ? ¿Cuándo es ventajoso usar Queue() ?

Answer 1

• Un Pipe() solo puede tener dos puntos finales.

• A Queue() puede tener múltiples productores y consumidores.

Cuándo usarlos

Si necesita más de dos puntos para comunicarse, use una Queue() .

Si necesita un rendimiento absoluto, un Pipe() es mucho más rápido porque Queue() se basa en Pipe() .

Benchmarking de rendimiento

Supongamos que desea generar dos procesos y enviar mensajes entre ellos lo más rápido posible. Estos son los resultados de tiempo de una carrera de arrastre entre pruebas similares usando Pipe() y Queue() ... Esto está en un ThinkpadT61 ejecutando Ubuntu 11.10, y Python 2.7.2.

FYI, arrojé los resultados de JoinableQueue() como una bonificación; JoinableQueue() cuenta de las tareas cuando se llama a queue.task_done() (ni siquiera conoce la tarea específica, solo cuenta las tareas sin terminar en la cola), de modo que queue.join() sabe que el trabajo ha finalizado.

El código para cada uno al final de esta respuesta ...

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 resumen, Pipe() es aproximadamente tres veces más rápido que una Queue() . Ni siquiera pienses en JoinableQueue() menos que realmente tengas los beneficios.

MATERIAL BONIFICADO 2

El multiprocesamiento introduce cambios sutiles en el flujo de información que dificultan la depuración a menos que conozca algunos accesos directos. Por ejemplo, es posible que tenga una secuencia de comandos que funcione bien al indexar a través de un diccionario en muchas condiciones, pero con poca frecuencia falla con ciertas entradas.

Normalmente obtenemos pistas sobre el fallo cuando todo el proceso de Python falla; sin embargo, no se registran los registros de fallos no solicitados en la consola si la función de multiprocesamiento falla. El seguimiento de bloqueos de multiprocesamiento desconocidos es difícil sin una pista de lo que colapsó el proceso.

La forma más simple que he encontrado para rastrear la información de falla de multiprocesamiento es ajustar toda la función de multiproceso en try / except y usar 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()

Ahora, cuando encuentras un bloqueo, ves algo como:

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

Código fuente:

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