Wieloprocesorowość: jak używać Pool.map na funkcji zdefiniowanej w klasie?

Kiedy biegnę coś takiego:

from multiprocessing import Pool

p = Pool(5)
def f(x):
     return x*x

p.map(f, [1,2,3])

to działa dobrze. Jednak umieszczając to jako funkcję klasy:

class calculate(object):
    def run(self):
        def f(x):
            return x*x

        p = Pool()
        return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

Daje mi następujący błąd:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/sw/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/sw/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/sw/lib/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

Widziałem post Alexa Martelliego dotyczący tego samego problemu, ale nie był wystarczająco jasny.

Denerwowały mnie również ograniczenia dotyczące tego, jakie funkcje może akceptować pool.map. Napisałem następujące informacje, aby to obejść. Wydaje się, że działa, nawet przy rekurencyjnym użyciu parmap.

from multiprocessing import Process, Pipe
from itertools import izip

def spawn(f):
    def fun(pipe, x):
        pipe.send(f(x))
        pipe.close()
    return fun

def parmap(f, X):
    pipe = [Pipe() for x in X]
    proc = [Process(target=spawn(f), args=(c, x)) for x, (p, c) in izip(X, pipe)]
    [p.start() for p in proc]
    [p.join() for p in proc]
    return [p.recv() for (p, c) in pipe]

if __name__ == '__main__':
    print parmap(lambda x: x**x, range(1, 5))

Nie mogłem użyć dotychczas opublikowanych kodów, ponieważ kody używające "multiprocessing.Pool" nie działają z wyrażeniami lambda, a kody nie używające "multiprocessing.Pool" generują tyle procesów, ile jest elementów pracy.

Dostosowałem kod tak, aby tworzył predefiniowaną liczbę pracowników i iterował listę wejściową tylko wtedy, gdy istnieje bezczynny pracownik. Włączyłem również tryb "demona" dla pracowników. Ctrl-c działa zgodnie z oczekiwaniami.

import multiprocessing


def fun(f, q_in, q_out):
    while True:
        i, x = q_in.get()
        if i is None:
            break
        q_out.put((i, f(x)))


def parmap(f, X, nprocs=multiprocessing.cpu_count()):
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()

    proc = [multiprocessing.Process(target=fun, args=(f, q_in, q_out))
            for _ in range(nprocs)]
    for p in proc:
        p.daemon = True
        p.start()

    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [q_in.put((None, None)) for _ in range(nprocs)]
    res = [q_out.get() for _ in range(len(sent))]

    [p.join() for p in proc]

    return [x for i, x in sorted(res)]


if __name__ == '__main__':
    print(parmap(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8]))

Wieloprocesowe przetwarzanie i wytrawianie jest zepsute i ograniczone, chyba że wyskoczysz poza standardową bibliotekę.

Jeśli używasz rozwidlenia multiprocessingwywołanego pathos.multiprocesssing, możesz bezpośrednio używać klas i metod klas w mapfunkcjach wieloprocesorowych . Dzieje się tak, ponieważ dilljest używany zamiast picklelub cPicklei dillmoże serializować prawie wszystko w Pythonie.

pathos.multiprocessingzapewnia również asynchroniczną funkcję mapy… i może mapdziałać z wieloma argumentami (np. map(math.pow, [1,2,3], [4,5,6]))

Zobacz dyskusje: Co może zrobić wieloprocesorowość i koperek?

i: http://matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization

Obsługuje nawet kod, który napisałeś na początku, bez modyfikacji i od interpretera. Po co cokolwiek, co jest bardziej delikatne i specyficzne dla pojedynczego przypadku?

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> class calculate(object):
...  def run(self):
...   def f(x):
...    return x*x
...   p = Pool()
...   return p.map(f, [1,2,3])
... 
>>> cl = calculate()
>>> print cl.run()
[1, 4, 9]

Pobierz kod tutaj: https://github.com/uqfoundation/pathos

I żeby pokazać, co potrafi:

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> 
>>> p = Pool(4)
>>> 
>>> def add(x,y):
...   return x+y
... 
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>> 
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>> 
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> 
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>> 
>>> res = p.amap(t.plus, x, y)
>>> res.get()
[4, 6, 8, 10]

O ile wiem, obecnie nie ma rozwiązania twojego problemu: funkcja, którą dajesz, map()musi być dostępna poprzez import twojego modułu. Dlatego działa kod Roberta: funkcję f()można uzyskać, importując następujący kod:

def f(x):
    return x*x

class Calculate(object):
    def run(self):
        p = Pool()
        return p.map(f, [1,2,3])

if __name__ == '__main__':
    cl = Calculate()
    print cl.run()

Właściwie dodałem sekcję „main”, ponieważ jest ona zgodna z zaleceniami dla platformy Windows („Upewnij się, że moduł główny można bezpiecznie zaimportować przez nowy interpreter Pythona bez powodowania niezamierzonych efektów ubocznych”).

Dodałem również wielką literę przed Calculate, aby zastosować się do PEP 8 . :)

Rozwiązanie mrule jest poprawne, ale zawiera błąd: jeśli dziecko odeśle dużą ilość danych, może wypełnić bufor potoku, blokując bufor dziecka pipe.send(), podczas gdy rodzic czeka, aż dziecko wyjdzie pipe.join(). Rozwiązaniem jest odczytanie danych dziecka przed join()nim. Ponadto dziecko powinno zamknąć koniec rury rodzica, aby zapobiec zakleszczeniu. Poniższy kod rozwiązuje ten problem. Należy również pamiętać, że parmaptworzy to jeden proces na element w X. Bardziej zaawansowanym rozwiązaniem jest multiprocessing.cpu_count()podzielenie Xna kilka fragmentów, a następnie scalenie wyników przed zwróceniem. Zostawiam to czytelnikowi jako ćwiczenie, aby nie zepsuć zwięzłości ładnej odpowiedzi muła. ;)

from multiprocessing import Process, Pipe
from itertools import izip

def spawn(f):
    def fun(ppipe, cpipe,x):
        ppipe.close()
        cpipe.send(f(x))
        cpipe.close()
    return fun

def parmap(f,X):
    pipe=[Pipe() for x in X]
    proc=[Process(target=spawn(f),args=(p,c,x)) for x,(p,c) in izip(X,pipe)]
    [p.start() for p in proc]
    ret = [p.recv() for (p,c) in pipe]
    [p.join() for p in proc]
    return ret

if __name__ == '__main__':
    print parmap(lambda x:x**x,range(1,5))

Ja też się z tym zmagałem. Miałem funkcje jako członkowie danych klasy, jako uproszczony przykład:

from multiprocessing import Pool
import itertools
pool = Pool()
class Example(object):
    def __init__(self, my_add): 
        self.f = my_add  
    def add_lists(self, list1, list2):
        # Needed to do something like this (the following line won't work)
        return pool.map(self.f,list1,list2)  

Musiałem użyć funkcji self.f w wywołaniu Pool.map () z tej samej klasy, a self.f nie przyjęło krotki jako argumentu. Ponieważ ta funkcja została osadzona w klasie, nie było dla mnie jasne, jak napisać typ opakowania, które sugerowały inne odpowiedzi.

Rozwiązałem ten problem, używając innego opakowania, które przyjmuje krotkę / listę, gdzie pierwszym elementem jest funkcja, a pozostałe elementy to argumenty tej funkcji, zwanej eval_func_tuple (f_args). Używając tego, problematyczna linia może zostać zastąpiona przez return pool.map (eval_func_tuple, itertools.izip (itertools.repeat (self.f), list1, list2)). Oto pełny kod:

Plik: util.py

def add(a, b): return a+b

def eval_func_tuple(f_args):
    """Takes a tuple of a function and args, evaluates and returns result"""
    return f_args[0](*f_args[1:])  

Plik: main.py

from multiprocessing import Pool
import itertools
import util  

pool = Pool()
class Example(object):
    def __init__(self, my_add): 
        self.f = my_add  
    def add_lists(self, list1, list2):
        # The following line will now work
        return pool.map(util.eval_func_tuple, 
            itertools.izip(itertools.repeat(self.f), list1, list2)) 

if __name__ == '__main__':
    myExample = Example(util.add)
    list1 = [1, 2, 3]
    list2 = [10, 20, 30]
    print myExample.add_lists(list1, list2)  

Uruchomienie main.py da [11, 22, 33]. Możesz to poprawić, na przykład eval_func_tuple można również zmodyfikować tak, aby przyjmował argumenty słów kluczowych.

Z drugiej strony, w innych odpowiedziach, funkcja „parmap” może być bardziej wydajna w przypadku większej liczby Procesów niż liczba dostępnych procesorów. Kopiuję edytowaną wersję poniżej. To jest mój pierwszy post i nie byłem pewien, czy powinienem bezpośrednio edytować oryginalną odpowiedź. Zmieniłem też nazwy niektórych zmiennych.

from multiprocessing import Process, Pipe  
from itertools import izip  

def spawn(f):  
    def fun(pipe,x):  
        pipe.send(f(x))  
        pipe.close()  
    return fun  

def parmap(f,X):  
    pipe=[Pipe() for x in X]  
    processes=[Process(target=spawn(f),args=(c,x)) for x,(p,c) in izip(X,pipe)]  
    numProcesses = len(processes)  
    processNum = 0  
    outputList = []  
    while processNum < numProcesses:  
        endProcessNum = min(processNum+multiprocessing.cpu_count(), numProcesses)  
        for proc in processes[processNum:endProcessNum]:  
            proc.start()  
        for proc in processes[processNum:endProcessNum]:  
            proc.join()  
        for proc,c in pipe[processNum:endProcessNum]:  
            outputList.append(proc.recv())  
        processNum = endProcessNum  
    return outputList    

if __name__ == '__main__':  
    print parmap(lambda x:x**x,range(1,5))         

Wziąłem odpowiedź klausa se i aganders3 i stworzyłem udokumentowany moduł, który jest bardziej czytelny i przechowuje w jednym pliku. Możesz go po prostu dodać do swojego projektu. Ma nawet opcjonalny pasek postępu!

"""
The ``processes`` module provides some convenience functions
for using parallel processes in python.

Adapted from http://stackoverflow.com/a/16071616/287297

Example usage:

    print prll_map(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8], 32, verbose=True)

Comments:

"It spawns a predefined amount of workers and only iterates through the input list
 if there exists an idle worker. I also enabled the "daemon" mode for the workers so
 that KeyboardInterupt works as expected."

Pitfalls: all the stdouts are sent back to the parent stdout, intertwined.

Alternatively, use this fork of multiprocessing: 
https://github.com/uqfoundation/multiprocess
"""

# Modules #
import multiprocessing
from tqdm import tqdm

################################################################################
def apply_function(func_to_apply, queue_in, queue_out):
    while not queue_in.empty():
        num, obj = queue_in.get()
        queue_out.put((num, func_to_apply(obj)))

################################################################################
def prll_map(func_to_apply, items, cpus=None, verbose=False):
    # Number of processes to use #
    if cpus is None: cpus = min(multiprocessing.cpu_count(), 32)
    # Create queues #
    q_in  = multiprocessing.Queue()
    q_out = multiprocessing.Queue()
    # Process list #
    new_proc  = lambda t,a: multiprocessing.Process(target=t, args=a)
    processes = [new_proc(apply_function, (func_to_apply, q_in, q_out)) for x in range(cpus)]
    # Put all the items (objects) in the queue #
    sent = [q_in.put((i, x)) for i, x in enumerate(items)]
    # Start them all #
    for proc in processes:
        proc.daemon = True
        proc.start()
    # Display progress bar or not #
    if verbose:
        results = [q_out.get() for x in tqdm(range(len(sent)))]
    else:
        results = [q_out.get() for x in range(len(sent))]
    # Wait for them to finish #
    for proc in processes: proc.join()
    # Return results #
    return [x for i, x in sorted(results)]

################################################################################
def test():
    def slow_square(x):
        import time
        time.sleep(2)
        return x**2
    objs    = range(20)
    squares = prll_map(slow_square, objs, 4, verbose=True)
    print "Result: %s" % squares

EDYCJA : Dodano sugestię @ alexander-mcfarlane i funkcję testową

Wiem, że zadawano to ponad 6 lat temu, ale chciałem tylko dodać moje rozwiązanie, ponieważ niektóre z powyższych sugestii wydają się strasznie skomplikowane, ale moje rozwiązanie było w rzeczywistości bardzo proste.

Wszystko, co musiałem zrobić, to zawinąć wywołanie pool.map () do funkcji pomocniczej. Przekazanie obiektu klasy wraz z argumentami dla metody jako krotki, która wyglądała trochę tak.

def run_in_parallel(args):
    return args[0].method(args[1])

myclass = MyClass()
method_args = [1,2,3,4,5,6]
args_map = [ (myclass, arg) for arg in method_args ]
pool = Pool()
pool.map(run_in_parallel, args_map)

Funkcje zdefiniowane w klasach (nawet w funkcjach wewnątrz klas) tak naprawdę nie trawią. Jednak to działa:

def f(x):
    return x*x

class calculate(object):
    def run(self):
        p = Pool()
    return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

Wiem, że to pytanie padło 8 lat i 10 miesięcy temu, ale chcę Wam przedstawić moje rozwiązanie:

from multiprocessing import Pool

class Test:

    def __init__(self):
        self.main()

    @staticmethod
    def methodForMultiprocessing(x):
        print(x*x)

    def main(self):
        if __name__ == "__main__":
            p = Pool()
            p.map(Test.methodForMultiprocessing, list(range(1, 11)))
            p.close()

TestObject = Test()

Musisz tylko przekształcić swoją klasę w metodę statyczną. Ale jest to również możliwe z metodą klasową:

from multiprocessing import Pool

class Test:

    def __init__(self):
        self.main()

    @classmethod
    def methodForMultiprocessing(cls, x):
        print(x*x)

    def main(self):
        if __name__ == "__main__":
            p = Pool()
            p.map(Test.methodForMultiprocessing, list(range(1, 11)))
            p.close()

TestObject = Test()

Przetestowano w Pythonie 3.7.3

Zmodyfikowałem metodę klaus se, ponieważ podczas pracy z małymi listami zawieszała się, gdy liczba pozycji wynosiła ~ 1000 lub więcej. Zamiast przesuwać zadania pojedynczo z Nonewarunkiem zatrzymania, ładuję kolejkę wejściową naraz i po prostu pozwalam procesom miażdżyć ją, aż będzie pusta.

from multiprocessing import cpu_count, Queue, Process

def apply_func(f, q_in, q_out):
    while not q_in.empty():
        i, x = q_in.get()
        q_out.put((i, f(x)))

# map a function using a pool of processes
def parmap(f, X, nprocs = cpu_count()):
    q_in, q_out   = Queue(), Queue()
    proc = [Process(target=apply_func, args=(f, q_in, q_out)) for _ in range(nprocs)]
    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [p.start() for p in proc]
    res = [q_out.get() for _ in sent]
    [p.join() for p in proc]

    return [x for i,x in sorted(res)]

Edycja: niestety teraz napotykam ten błąd w moim systemie: Limit maksymalnego rozmiaru kolejki wieloprocesowej wynosi 32767 , mam nadzieję, że obejścia tam pomogą.

Możesz uruchomić kod bez żadnych problemów, jeśli w jakiś sposób ręcznie zignorujesz Poolobiekt z listy obiektów w klasie, ponieważ nie jest on w picklestanie, jak mówi błąd. Możesz to zrobić za pomocą __getstate__funkcji (patrz również tutaj ) w następujący sposób. PoolObiekt będzie starał się znaleźć __getstate__i __setstate__funkcje i wykonać je, jeśli uzna to po uruchomieniu map, map_asyncitp:

class calculate(object):
    def __init__(self):
        self.p = Pool()
    def __getstate__(self):
        self_dict = self.__dict__.copy()
        del self_dict['p']
        return self_dict
    def __setstate__(self, state):
        self.__dict__.update(state)

    def f(self, x):
        return x*x
    def run(self):
        return self.p.map(self.f, [1,2,3])

Następnie wykonaj:

cl = calculate()
cl.run()

da ci wynik:

[1, 4, 9]

Przetestowałem powyższy kod w Pythonie 3.xi działa.

Nie jestem pewien, czy przyjęto takie podejście, ale obejście, którego używam, jest:

from multiprocessing import Pool

t = None

def run(n):
    return t.f(n)

class Test(object):
    def __init__(self, number):
        self.number = number

    def f(self, x):
        print x * self.number

    def pool(self):
        pool = Pool(2)
        pool.map(run, range(10))

if __name__ == '__main__':
    t = Test(9)
    t.pool()
    pool = Pool(2)
    pool.map(run, range(10))

Wynik powinien być:

0
9
18
27
36
45
54
63
72
81
0
9
18
27
36
45
54
63
72
81

class Calculate(object):
  # Your instance method to be executed
  def f(self, x, y):
    return x*y

if __name__ == '__main__':
  inp_list = [1,2,3]
  y = 2
  cal_obj = Calculate()
  pool = Pool(2)
  results = pool.map(lambda x: cal_obj.f(x, y), inp_list)

Istnieje możliwość, że zechcesz zastosować tę funkcję dla każdej innej instancji klasy. Oto rozwiązanie również na to

class Calculate(object):
  # Your instance method to be executed
  def __init__(self, x):
    self.x = x

  def f(self, y):
    return self.x*y

if __name__ == '__main__':
  inp_list = [Calculate(i) for i in range(3)]
  y = 2
  pool = Pool(2)
  results = pool.map(lambda x: x.f(y), inp_list)

Oto moje rozwiązanie, które moim zdaniem jest nieco mniej hakerskie niż większość innych tutaj. Jest podobny do odpowiedzi Nightowl.

someclasses = [MyClass(), MyClass(), MyClass()]

def method_caller(some_object, some_method='the method'):
    return getattr(some_object, some_method)()

othermethod = partial(method_caller, some_method='othermethod')

with Pool(6) as pool:
    result = pool.map(othermethod, someclasses)

Od http://www.rueckstiess.net/research/snippets/show/ca1d7d90 i http://qingkaikong.blogspot.com/2016/12/python-parallel-method-in-class.html

Możemy utworzyć funkcję zewnętrzną i zapełnić ją obiektem klasy self:

from joblib import Parallel, delayed
def unwrap_self(arg, **kwarg):
    return square_class.square_int(*arg, **kwarg)

class square_class:
    def square_int(self, i):
        return i * i

    def run(self, num):
        results = []
        results = Parallel(n_jobs= -1, backend="threading")\
            (delayed(unwrap_self)(i) for i in zip([self]*len(num), num))
        print(results)

LUB bez Joblib:

from multiprocessing import Pool
import time

def unwrap_self_f(arg, **kwarg):
    return C.f(*arg, **kwarg)

class C:
    def f(self, name):
        print 'hello %s,'%name
        time.sleep(5)
        print 'nice to meet you.'

    def run(self):
        pool = Pool(processes=2)
        names = ('frank', 'justin', 'osi', 'thomas')
        pool.map(unwrap_self_f, zip([self]*len(names), names))

if __name__ == '__main__':
    c = C()
    c.run()

Może to nie jest zbyt dobre rozwiązanie, ale w moim przypadku rozwiązuję to w ten sposób.

from multiprocessing import Pool

def foo1(data):
    self = data.get('slf')
    lst = data.get('lst')
    return sum(lst) + self.foo2()

class Foo(object):
    def __init__(self, a, b):
        self.a = a
        self.b = b

    def foo2(self):
        return self.a**self.b   

    def foo(self):
        p = Pool(5)
        lst = [1, 2, 3]
        result = p.map(foo1, (dict(slf=self, lst=lst),))
        return result

if __name__ == '__main__':
    print(Foo(2, 4).foo())

Musiałem przejść selfdo mojej funkcji, ponieważ muszę uzyskać dostęp do atrybutów i funkcji mojej klasy za pośrednictwem tej funkcji. To działa dla mnie. Zawsze mile widziane są poprawki i sugestie.

Oto szablon, który napisałem dla korzystania z wieloprocesorowej puli w python3, w szczególności python3.7.7 został użyty do uruchomienia testów. Moje najszybsze biegi uzyskałem przy użyciu imap_unordered. Po prostu podłącz swój scenariusz i wypróbuj go. Możesz użyć timeitlub po prostu time.time()dowiedzieć się, co jest dla Ciebie najlepsze.

import multiprocessing
import time

NUMBER_OF_PROCESSES = multiprocessing.cpu_count()
MP_FUNCTION = 'starmap'  # 'imap_unordered' or 'starmap' or 'apply_async'

def process_chunk(a_chunk):
    print(f"processig mp chunk {a_chunk}")
    return a_chunk


map_jobs = [1, 2, 3, 4]

result_sum = 0

s = time.time()
if MP_FUNCTION == 'imap_unordered':
    pool = multiprocessing.Pool(processes=NUMBER_OF_PROCESSES)
    for i in pool.imap_unordered(process_chunk, map_jobs):
        result_sum += i
elif MP_FUNCTION == 'starmap':
    pool = multiprocessing.Pool(processes=NUMBER_OF_PROCESSES)
    try:
        map_jobs = [(i, ) for i in map_jobs]
        result_sum = pool.starmap(process_chunk, map_jobs)
        result_sum = sum(result_sum)
    finally:
        pool.close()
        pool.join()
elif MP_FUNCTION == 'apply_async':
    with multiprocessing.Pool(processes=NUMBER_OF_PROCESSES) as pool:
        result_sum = [pool.apply_async(process_chunk, [i, ]).get() for i in map_jobs]
    result_sum = sum(result_sum)
print(f"result_sum is {result_sum}, took {time.time() - s}s")

W powyższym scenariuszu imap_unorderedfaktycznie wydaje się działać najgorzej dla mnie. Wypróbuj swoją obudowę i porównaj ją na maszynie, na której planujesz ją uruchomić. Przeczytaj także o pulach procesów . Twoje zdrowie!