python - set_start_method - Piscina de pitón con procesos de trabajo.

python multithreading

Question

Estoy tratando de usar un grupo de trabajadores en python usando objetos de proceso. Cada trabajador (un Proceso) realiza alguna inicialización (toma una cantidad de tiempo no trivial), obtiene una serie de trabajos (idealmente utilizando map() ) y devuelve algo. Ninguna comunicación es necesaria más allá de eso. Sin embargo, parece que no puedo entender cómo usar map () para usar la función compute() mi trabajador.

from multiprocessing import Pool, Process
class Worker(Process):
def __init__(self):
print ''Worker started''
# do some initialization here
super(Worker, self).__init__()
def compute(self, data):
print ''Computing things!''
return data * data
if __name__ == ''__main__'':
# This works fine
worker = Worker()
print worker.compute(3)
# workers get initialized fine
pool = Pool(processes = 4,
initializer = Worker)
data = range(10)
# How to use my worker pool?
result = pool.map(compute, data)

¿Es una cola de trabajos el camino a seguir, o puedo usar map() ?

Answer 1

Le sugiero que use una cola para esto.

class Worker(Process):
def __init__(self, queue):
super(Worker, self).__init__()
self.queue = queue
def run(self):
print(''Worker started'')
# do some initialization here
print(''Computing things!'')
for data in iter(self.queue.get, None):
# Use data

Ahora puede comenzar un montón de estos, todos trabajando desde una sola cola

request_queue = Queue()
for i in range(4):
Worker(request_queue).start()
for data in the_real_source:
request_queue.put(data)
# Sentinel objects to allow clean shutdown: 1 per worker.
for i in range(4):
request_queue.put(None)

Ese tipo de cosas debería permitirle amortizar el costoso costo de inicio entre varios trabajadores.

Answer 2

initializer espera un llamamiento arbitrario que realice la iniciación, por ejemplo, puede establecer algunas variables globales, no una subclase de Process ; map acepta un iterable arbitrario:

#!/usr/bin/env python
import multiprocessing as mp
def init(val):
print(''do some initialization here'')
def compute(data):
print(''Computing things!'')
return data * data
def produce_data():
yield -100
for i in range(10):
yield i
yield 100
if __name__=="__main__":
p = mp.Pool(initializer=init, initargs=(''arg'',))
print(p.map(compute, produce_data()))