Python Pool with worker Processes

Question

Próbuję użyć puli Worker w python za pomocą obiektów Process. Każdy pracownik (proces) wykonuje pewną inicjalizację (zajmuje nieprzydatny czas), przechodzi serię zadań (najlepiej używając map()) i zwraca coś. Komunikacja nie jest konieczna. Nie mogę jednak wymyślić, jak używać funkcji map() do korzystania z funkcji mojego pracownika compute().

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) 

jest praca w kolejce na drogę zamiast, czy mogę użyć map()?

Answer 1

Proponuję użyć kolejki do tego.

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 

Teraz można uruchomić stos z nich, wszystkie z dostaniem pracy z jednej kolejce

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) 

Tego rodzaju rzeczy powinny pozwalają amortyzować drogie koszty startowego w wielu robotników.

Answer 2

initializer spodziewa się arbitralnego wywołania, które przeprowadza inicjalizację, np. Może ustawić niektóre globale, a nie podklasę Process; map akceptuje dowolne iterowalne:

#!/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()))