Multitraitement: comment utiliser Pool.map sur une fonction définie dans une classe?

Quand je lance quelque chose comme:

from multiprocessing import Pool

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

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

ça fonctionne bien. Cependant, en mettant cela en fonction d'une classe:

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

Me donne l'erreur suivante:

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

J'ai vu un article d'Alex Martelli traitant du même genre de problème, mais ce n'était pas assez explicite.

J'ai également été ennuyé par les restrictions sur le type de fonctions que pool.map pouvait accepter. J'ai écrit ce qui suit pour contourner cela. Cela semble fonctionner, même pour une utilisation récursive de 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))

Je n'ai pas pu utiliser les codes affichés jusqu'à présent car les codes utilisant "multiprocessing.Pool" ne fonctionnent pas avec les expressions lambda et les codes n'utilisant pas "multiprocessing.Pool" engendrent autant de processus qu'il y a d'éléments de travail.

J'ai adapté le code pour qu'il génère un nombre prédéfini de travailleurs et n'itère dans la liste d'entrée que s'il existe un travailleur inactif. J'ai également activé le mode "démon" pour les travailleurs st ctrl-c fonctionne comme prévu.

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

Le multitraitement et le décapage sont interrompus et limités à moins que vous ne sautiez hors de la bibliothèque standard.

Si vous utilisez un fork de multiprocessingcalled pathos.multiprocesssing, vous pouvez directement utiliser des classes et des méthodes de classe dans les mapfonctions de multiprocessing . C'est parce qu'il dillest utilisé à la place de pickleou cPickle, et dillpeut sérialiser presque tout en python.

pathos.multiprocessingfournit également une fonction de carte asynchrone ... et il peut mapfonctionner avec plusieurs arguments (par exemple map(math.pow, [1,2,3], [4,5,6]))

Voir les discussions: Que peuvent faire le multitraitement et l'aneth ensemble?

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

Il gère même le code que vous avez écrit initialement, sans modification, et à partir de l'interpréteur. Pourquoi faire autre chose de plus fragile et spécifique à un seul cas?

>>> 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]

Obtenez le code ici: https://github.com/uqfoundation/pathos

Et, juste pour montrer un peu plus ce qu'il peut faire:

>>> 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]

Il n'y a actuellement pas de solution à votre problème, à ma connaissance: la fonction que vous donnez map()doit être accessible via un import de votre module. C'est pourquoi le code de robert fonctionne: la fonction f()peut être obtenue en important le code suivant:

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

J'ai en fait ajouté une section "principale", car elle suit les recommandations pour la plate-forme Windows ("Assurez-vous que le module principal peut être importé en toute sécurité par un nouvel interpréteur Python sans provoquer d'effets secondaires involontaires").

J'ai également ajouté une lettre majuscule devant Calculate, afin de suivre PEP 8 . :)

La solution de mrule est correcte mais présente un bug: si l'enfant renvoie une grande quantité de données, il peut remplir le tampon du tube, bloquant sur l'enfant pipe.send(), pendant que le parent attend que l'enfant quitte pipe.join(). La solution est de lire les données de join()l'enfant avant de l'engager. De plus, l'enfant doit fermer l'extrémité parent du tuyau pour éviter un blocage. Le code ci-dessous corrige cela. Sachez également que cela parmapcrée un processus par élément dans X. Une solution plus avancée consiste à multiprocessing.cpu_count()diviser Xen plusieurs blocs, puis à fusionner les résultats avant de revenir. Je laisse cela comme un exercice au lecteur pour ne pas gâcher la concision de la belle réponse par mrule. ;)

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

J'ai également eu du mal avec ça. J'avais des fonctions en tant que membres de données d'une classe, à titre d'exemple simplifié:

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)  

J'avais besoin d'utiliser la fonction self.f dans un appel Pool.map () depuis la même classe et self.f ne prenait pas de tuple comme argument. Puisque cette fonction était intégrée dans une classe, je ne savais pas comment écrire le type de wrapper que d'autres réponses suggéraient.

J'ai résolu ce problème en utilisant un wrapper différent qui prend un tuple / liste, où le premier élément est la fonction, et les éléments restants sont les arguments de cette fonction, appelés eval_func_tuple (f_args). En utilisant cela, la ligne problématique peut être remplacée par return pool.map (eval_func_tuple, itertools.izip (itertools.repeat (self.f), list1, list2)). Voici le code complet:

Fichier: 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:])  

Fichier: 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)  

Lancer main.py donnera [11, 22, 33]. N'hésitez pas à améliorer cela, par exemple eval_func_tuple pourrait également être modifié pour prendre des arguments de mots-clés.

Sur une autre note, dans une autre réponse, la fonction "parmap" peut être rendue plus efficace pour le cas de plus de Processus que de nombre de CPU disponibles. Je copie une version modifiée ci-dessous. C'est mon premier message et je n'étais pas sûr de devoir modifier directement la réponse originale. J'ai également renommé certaines variables.

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

J'ai pris la réponse de klaus se et aganders3 et j'ai créé un module documenté qui est plus lisible et tient dans un seul fichier. Vous pouvez simplement l'ajouter à votre projet. Il a même une barre de progression en option!

"""
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

EDIT : Ajout de la suggestion @ alexander-mcfarlane et d'une fonction de test

Je sais que cela a été demandé il y a plus de 6 ans maintenant, mais je voulais simplement ajouter ma solution, car certaines des suggestions ci-dessus semblent horriblement compliquées, mais ma solution était en fait très simple.

Tout ce que j'avais à faire était d'envelopper l'appel pool.map () à une fonction d'assistance. Passer l'objet de classe avec les arguments de la méthode sous forme de tuple, qui ressemblait un peu à ceci.

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)

Les fonctions définies dans les classes (même dans les fonctions au sein des classes) ne sont pas vraiment pickle. Cependant, cela fonctionne:

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

Je sais que cette question a été posée il y a 8 ans et 10 mois mais je souhaite vous présenter ma solution:

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

Vous avez juste besoin de faire de votre fonction de classe une méthode statique. Mais c'est aussi possible avec une méthode de classe:

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

Testé en Python 3.7.3

J'ai modifié la méthode de klaus se car alors qu'elle fonctionnait pour moi avec de petites listes, elle se bloquait lorsque le nombre d'éléments était d'environ 1000 ou plus. Au lieu de pousser les travaux un par un avec la Nonecondition d'arrêt, je charge la file d'attente d'entrée en une seule fois et je laisse simplement les processus la grignoter jusqu'à ce qu'elle soit vide.

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

Edit: malheureusement maintenant, je rencontre cette erreur sur mon système: la limite de taille maximale de la file d'attente de multitraitement est de 32767 , j'espère que les solutions de contournement aideront.

Vous pouvez exécuter votre code sans aucun problème si vous ignorez manuellement l' Poolobjet de la liste des objets de la classe car il n'est pas picklepossible comme le dit l'erreur. Vous pouvez le faire avec la __getstate__fonction (regardez ici aussi) comme suit. L' Poolobjet va essayer de trouver les __getstate__et les __setstate__fonctions et les exécuter si elle le juge lors de l' exécution map, map_asyncetc:

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

Alors fais:

cl = calculate()
cl.run()

vous donnera la sortie:

[1, 4, 9]

J'ai testé le code ci-dessus en Python 3.x et cela fonctionne.

Je ne sais pas si cette approche a été adoptée, mais une solution que j'utilise est:

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

La sortie doit être:

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)

Il est possible que vous souhaitiez appliquer cette fonction pour chaque instance différente de la classe. Alors voici la solution pour cela aussi

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)

Voici ma solution, qui je pense est un peu moins hackish que la plupart des autres ici. C'est similaire à la réponse de 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)

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

Nous pouvons créer une fonction externe et la semer avec l'objet self de classe:

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)

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

Ce n'est peut-être pas une très bonne solution mais dans mon cas, je le résous comme ça.

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

J'ai dû passer selfà ma fonction car je dois accéder aux attributs et aux fonctions de ma classe via cette fonction. Cela fonctionne pour moi. Les corrections et suggestions sont toujours les bienvenues.

Voici un passe-partout que j'ai écrit pour utiliser le pool multiprocesseur dans python3, en particulier python3.7.7 a été utilisé pour exécuter les tests. J'ai obtenu mes courses les plus rapides en utilisant imap_unordered. Branchez simplement votre scénario et essayez-le. Vous pouvez utiliser timeitou simplement time.time()pour déterminer ce qui vous convient le mieux.

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

Dans le scénario ci-dessus, il imap_unorderedsemble que ce soit le pire pour moi. Essayez votre cas et comparez-le sur la machine sur laquelle vous prévoyez de l'exécuter. Lisez également sur les pools de processus . À votre santé!