async - python multiprocessing windows

¿Cómo puedo paralelizar un simple bucle de Python?

Question

Esta es probablemente una pregunta trivial, pero ¿cómo puedo paralelizar el siguiente ciclo en Python?

# setup output lists
output1 = list()
output2 = list()
output3 = list()
for j in range(0, 10):
# calc individual parameter value
parameter = j * offset
# call the calculation
out1, out2, out3 = calc_stuff(parameter = parameter)
# put results into correct output list
output1.append(out1)
output2.append(out2)
output3.append(out3)

Sé cómo iniciar subprocesos individuales en Python, pero no sé cómo "recopilar" los resultados.

Múltiples procesos también estarían bien, lo que sea más fácil para este caso. Estoy usando actualmente Linux, pero el código debería ejecutarse en Windows y Mac como ... bueno.

¿Cuál es la forma más fácil de paralelizar este código?

Answer 1

¿Cuál es la forma más fácil de paralelizar este código?

Me gusta concurrent.futures para esto, disponible en Python3 desde la versión 3.2 - y a través de backport a 2.6 y 2.7 en PyPi .

Puede usar hilos o procesos y usar exactamente la misma interfaz.

Multiprocesamiento

Pon esto en un archivo - futuretest.py:

import concurrent.futures
import time, random # add some random sleep time
offset = 2 # you don''t supply these so
def calc_stuff(parameter=None): # these are examples.
sleep_time = random.choice([0, 1, 2, 3, 4, 5])
time.sleep(sleep_time)
return parameter / 2, sleep_time, parameter * parameter
def procedure(j): # just factoring out the
parameter = j * offset # procedure
# call the calculation
return calc_stuff(parameter=parameter)
def main():
output1 = list()
output2 = list()
output3 = list()
start = time.time() # let''s see how long this takes
# we can swap out ProcessPoolExecutor for ThreadPoolExecutor
with concurrent.futures.ProcessPoolExecutor() as executor:
for out1, out2, out3 in executor.map(procedure, range(0, 10)):
# put results into correct output list
output1.append(out1)
output2.append(out2)
output3.append(out3)
finish = time.time()
# these kinds of format strings are only available on Python 3.6:
# time to upgrade!
print(f''original inputs: {repr(output1)}'')
print(f''total time to execute {sum(output2)} = sum({repr(output2)})'')
print(f''time saved by parallelizing: {sum(output2) - (finish-start)}'')
print(f''returned in order given: {repr(output3)}'')
if __name__ == ''__main__'':
main()

Y aquí está la salida:

$ python3 -m futuretest
original inputs: [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
total time to execute 33 = sum([0, 3, 3, 4, 3, 5, 1, 5, 5, 4])
time saved by parallellizing: 27.68999981880188
returned in order given: [0, 4, 16, 36, 64, 100, 144, 196, 256, 324]

Multihilo

Ahora cambie ProcessPoolExecutor a ThreadPoolExecutor y ThreadPoolExecutor ejecutar el módulo:

$ python3 -m futuretest
original inputs: [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
total time to execute 19 = sum([0, 2, 3, 5, 2, 0, 0, 3, 3, 1])
time saved by parallellizing: 13.992000102996826
returned in order given: [0, 4, 16, 36, 64, 100, 144, 196, 256, 324]

¡Ahora has hecho tanto multihilo como multiprocesamiento!

Tenga en cuenta el rendimiento y el uso de ambos juntos.

El muestreo es demasiado pequeño para comparar los resultados.

Sin embargo, sospecho que el multihilo será más rápido que el multiprocesamiento en general, especialmente en Windows, ya que Windows no admite bifurcaciones, por lo que cada nuevo proceso debe tomarse un tiempo para su lanzamiento. En Linux o Mac, probablemente estarán más cerca.

Puede anidar varios subprocesos en varios procesos, pero se recomienda no utilizar varios subprocesos para derivar procesos múltiples.

Answer 2

¿por qué no usas hilos y un mutex para proteger una lista global?

import os
import re
import time
import sys
import thread
from threading import Thread
class thread_it(Thread):
def __init__ (self,param):
Thread.__init__(self)
self.param = param
def run(self):
mutex.acquire()
output.append(calc_stuff(self.param))
mutex.release()
threads = []
output = []
mutex = thread.allocate_lock()
for j in range(0, 10):
current = thread_it(j * offset)
threads.append(current)
current.start()
for t in threads:
t.join()
#here you have output list filled with data

ten en cuenta que serás tan rápido como tu hilo más lento

Answer 3

Echa un vistazo a esto;

http://docs.python.org/library/queue.html

Esta podría no ser la manera correcta de hacerlo, pero haría algo así como;

Código real;

from multiprocessing import Process, JoinableQueue as Queue
class CustomWorker(Process):
def __init__(self,workQueue, out1,out2,out3):
Process.__init__(self)
self.input=workQueue
self.out1=out1
self.out2=out2
self.out3=out3
def run(self):
while True:
try:
value = self.input.get()
#value modifier
temp1,temp2,temp3 = self.calc_stuff(value)
self.out1.put(temp1)
self.out2.put(temp2)
self.out3.put(temp3)
self.input.task_done()
except Queue.Empty:
return
#Catch things better here
def calc_stuff(self,param):
out1 = param * 2
out2 = param * 4
out3 = param * 8
return out1,out2,out3
def Main():
inputQueue = Queue()
for i in range(10):
inputQueue.put(i)
out1 = Queue()
out2 = Queue()
out3 = Queue()
processes = []
for x in range(2):
p = CustomWorker(inputQueue,out1,out2,out3)
p.daemon = True
p.start()
processes.append(p)
inputQueue.join()
while(not out1.empty()):
print out1.get()
print out2.get()
print out3.get()
if __name__ == ''__main__'':
Main()

Espero que ayude.

Answer 4

Esto podría ser útil al implementar multiprocesamiento y computación distribuida / paralela en Python.

Tutorial de YouTube sobre el uso del paquete techila

Techila es un middleware de informática distribuida, que se integra directamente con Python utilizando el paquete techila. La función de melocotón en el paquete puede ser útil para paralelizar estructuras de bucle. (El siguiente fragmento de código es de los foros de la comunidad de Techila )

techila.peach(funcname = ''theheavyalgorithm'', # Function that will be called on the compute nodes/ Workers
files = ''theheavyalgorithm.py'', # Python-file that will be sourced on Workers
jobs = jobcount # Number of Jobs in the Project
)

Answer 5

Para paralelizar un bucle for simple, joblib aporta un gran valor al uso crudo de multiprocesamiento. No solo la breve sintaxis, sino también elementos como el agrupamiento transparente de iteraciones cuando son muy rápidas (para eliminar la sobrecarga) o la captura del rastreo del proceso secundario, para tener un mejor informe de errores.

Descargo de responsabilidad: soy el autor original de joblib.

Answer 6

Usar múltiples hilos en CPython no le dará un mejor rendimiento para el código Python puro debido al bloqueo de intérprete global (GIL). Sugiero usar el módulo de multiprocessing lugar:

pool = multiprocessing.Pool(4)
out1, out2, out3 = zip(*pool.map(calc_stuff, range(0, 10 * offset, offset)))

Tenga en cuenta que esto no funcionará en el intérprete interactivo.

Para evitar el FUD usual alrededor del GIL: de todos modos, no habría ninguna ventaja al usar subprocesos para este ejemplo. Desea utilizar procesos aquí, no hilos, ya que evitan un montón de problemas.

Answer 7

ejemplo muy simple de procesamiento paralelo es

from multiprocessing import Process
output1 = list()
output2 = list()
output3 = list()
def yourfunction():
for j in range(0, 10):
# calc individual parameter value
parameter = j * offset
# call the calculation
out1, out2, out3 = calc_stuff(parameter = parameter)
# put results into correct output list
output1.append(out1)
output2.append(out2)
output3.append(out3)
if __name__ == ''__main__'':
p = Process(target=pa.yourfunction, args=(''bob'',))
p.start()
p.join()