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