Multiprocessing Python PicklingError: Can't pickle <type 'function'>

Je suis désolé de ne pas pouvoir reproduire l'erreur avec un exemple plus simple et mon code est trop compliqué à publier. Si j'exécute le programme dans le shell IPython au lieu du Python normal, les choses fonctionnent bien.

J'ai recherché quelques notes précédentes sur ce problème. Ils ont tous été causés par l'utilisation de pool pour appeler la fonction définie dans une fonction de classe. Mais ce n'est pas le cas pour moi.

Exception in thread Thread-3:
Traceback (most recent call last):
  File "/usr/lib64/python2.7/threading.py", line 552, in __bootstrap_inner
    self.run()
  File "/usr/lib64/python2.7/threading.py", line 505, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 313, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

J'apprécierais toute aide.

Mise à jour : La fonction I pickle est définie au niveau supérieur du module. Bien qu'il appelle une fonction qui contient une fonction imbriquée. c'est-à-dire, f()appelle des g()appels h()qui ont une fonction imbriquée i(), et j'appelle pool.apply_async(f). f(), g(), h()Sont tous définis au niveau supérieur. J'ai essayé un exemple plus simple avec ce modèle et cela fonctionne cependant.

Voici une liste de ce qui peut être mariné . En particulier, les fonctions ne sont picklables que si elles sont définies au niveau supérieur d'un module.

Ce morceau de code:

import multiprocessing as mp

class Foo():
    @staticmethod
    def work(self):
        pass

if __name__ == '__main__':   
    pool = mp.Pool()
    foo = Foo()
    pool.apply_async(foo.work)
    pool.close()
    pool.join()

renvoie une erreur presque identique à celle que vous avez publiée:

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

Le problème est que pooltoutes les méthodes utilisent a mp.SimpleQueuepour transmettre des tâches aux processus de travail. Tout ce qui passe par le mp.SimpleQueuedoit être pickable, et foo.workn'est pas picklable car il n'est pas défini au niveau supérieur du module.

Il peut être corrigé en définissant une fonction au niveau supérieur, qui appelle foo.work():

def work(foo):
    foo.work()

pool.apply_async(work,args=(foo,))

Notez que fooc'est pickable, car Fooest défini au niveau supérieur et foo.__dict__est picklable.

J'utiliserais pathos.multiprocesssing, au lieu de multiprocessing. pathos.multiprocessingest une fourchette de multiprocessingces utilisations dill. dillpeut sérialiser presque n'importe quoi en python, vous pouvez donc en envoyer beaucoup plus en parallèle. Le pathosfork a également la possibilité de travailler directement avec plusieurs fonctions d'argument, comme vous en avez besoin pour les méthodes de classe.

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> p = Pool(4)
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> p.map(t.plus, x, y)
[4, 6, 8, 10]
>>> 
>>> class Foo(object):
...   @staticmethod
...   def work(self, x):
...     return x+1
... 
>>> f = Foo()
>>> p.apipe(f.work, f, 100)
<processing.pool.ApplyResult object at 0x10504f8d0>
>>> res = _
>>> res.get()
101

Obtenez pathos(et si vous le souhaitez dill) ici: https://github.com/uqfoundation

Comme d'autres l'ont dit, multiprocessingil ne peut transférer que des objets Python vers des processus de travail qui peuvent être décapés. Si vous ne pouvez pas réorganiser votre code comme décrit par unutbu, vous pouvez utiliserdill les capacités étendues de pickling / unpickling pour transférer des données (en particulier des données de code) comme je le montre ci-dessous.

Cette solution nécessite uniquement l'installation de dillet aucune autre bibliothèque comme pathos:

import os
from multiprocessing import Pool

import dill


def run_dill_encoded(payload):
    fun, args = dill.loads(payload)
    return fun(*args)


def apply_async(pool, fun, args):
    payload = dill.dumps((fun, args))
    return pool.apply_async(run_dill_encoded, (payload,))


if __name__ == "__main__":

    pool = Pool(processes=5)

    # asyn execution of lambda
    jobs = []
    for i in range(10):
        job = apply_async(pool, lambda a, b: (a, b, a * b), (i, i + 1))
        jobs.append(job)

    for job in jobs:
        print job.get()
    print

    # async execution of static method

    class O(object):

        @staticmethod
        def calc():
            return os.getpid()

    jobs = []
    for i in range(10):
        job = apply_async(pool, O.calc, ())
        jobs.append(job)

    for job in jobs:
        print job.get()

J'ai trouvé que je peux également générer exactement cette sortie d'erreur sur un morceau de code fonctionnant parfaitement en essayant d'utiliser le profileur dessus.

Notez que c'était sur Windows (où la fourche est un peu moins élégante).

Je courais:

python -m profile -o output.pstats <script> 

Et constaté que la suppression du profilage supprimait l'erreur et que le placement du profilage la restaurait. Me rendait fou aussi parce que je savais que le code fonctionnait. Je vérifiais si quelque chose avait mis à jour pool.py ... puis avait un sentiment de naufrage et éliminé le profilage et c'était tout.

Publier ici pour les archives au cas où quelqu'un d'autre les rencontrerait.

Lorsque ce problème survient, multiprocessingune solution simple consiste à passer de Poolà ThreadPool. Cela peut être fait sans changement de code autre que l’importation

from multiprocessing.pool import ThreadPool as Pool

Cela fonctionne parce que ThreadPool partage la mémoire avec le thread principal, plutôt que de créer un nouveau processus - cela signifie que le décapage n'est pas requis.

L'inconvénient de cette méthode est que python n'est pas le plus grand langage de gestion des threads - il utilise quelque chose appelé Global Interpreter Lock pour rester sûr, ce qui peut ralentir certains cas d'utilisation ici. Cependant, si vous interagissez principalement avec d'autres systèmes (exécution de commandes HTTP, conversation avec une base de données, écriture sur des systèmes de fichiers), votre code n'est probablement pas lié par le processeur et ne prendra pas beaucoup de succès. En fait, j'ai trouvé lors de l'écriture de benchmarks HTTP / HTTPS que le modèle fileté utilisé ici a moins de temps et de délais, car le temps de création de nouveaux processus est beaucoup plus élevé que le temps de création de nouveaux threads.

Donc, si vous traitez une tonne de choses dans l'espace utilisateur python, ce n'est peut-être pas la meilleure méthode.

Cette solution nécessite uniquement l'installation d'aneth et pas d'autres bibliothèques comme pathos

def apply_packed_function_for_map((dumped_function, item, args, kwargs),):
    """
    Unpack dumped function as target function and call it with arguments.

    :param (dumped_function, item, args, kwargs):
        a tuple of dumped function and its arguments
    :return:
        result of target function
    """
    target_function = dill.loads(dumped_function)
    res = target_function(item, *args, **kwargs)
    return res


def pack_function_for_map(target_function, items, *args, **kwargs):
    """
    Pack function and arguments to object that can be sent from one
    multiprocessing.Process to another. The main problem is:
        «multiprocessing.Pool.map*» or «apply*»
        cannot use class methods or closures.
    It solves this problem with «dill».
    It works with target function as argument, dumps it («with dill»)
    and returns dumped function with arguments of target function.
    For more performance we dump only target function itself
    and don't dump its arguments.
    How to use (pseudo-code):

        ~>>> import multiprocessing
        ~>>> images = [...]
        ~>>> pool = multiprocessing.Pool(100500)
        ~>>> features = pool.map(
        ~...     *pack_function_for_map(
        ~...         super(Extractor, self).extract_features,
        ~...         images,
        ~...         type='png'
        ~...         **options,
        ~...     )
        ~... )
        ~>>>

    :param target_function:
        function, that you want to execute like  target_function(item, *args, **kwargs).
    :param items:
        list of items for map
    :param args:
        positional arguments for target_function(item, *args, **kwargs)
    :param kwargs:
        named arguments for target_function(item, *args, **kwargs)
    :return: tuple(function_wrapper, dumped_items)
        It returs a tuple with
            * function wrapper, that unpack and call target function;
            * list of packed target function and its' arguments.
    """
    dumped_function = dill.dumps(target_function)
    dumped_items = [(dumped_function, item, args, kwargs) for item in items]
    return apply_packed_function_for_map, dumped_items

Il fonctionne également pour les tableaux numpy.

Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

Cette erreur se produira également si vous avez une fonction intégrée dans l'objet modèle qui a été transmise au travail asynchrone.

Assurez-vous donc de vérifier que les objets de modèle transmis n'ont pas de fonctions intégrées. (Dans notre cas, nous utilisions laFieldTracker() fonction de django-model-utils à l'intérieur du modèle pour suivre un certain champ). Voici le lien vers le problème GitHub pertinent.

S'appuyant sur la solution @rocksportrocker, il serait judicieux de l'analyser lors de l'envoi et de la récupération des résultats.

import dill
import itertools
def run_dill_encoded(payload):
    fun, args = dill.loads(payload)
    res = fun(*args)
    res = dill.dumps(res)
    return res

def dill_map_async(pool, fun, args_list,
                   as_tuple=True,
                   **kw):
    if as_tuple:
        args_list = ((x,) for x in args_list)

    it = itertools.izip(
        itertools.cycle([fun]),
        args_list)
    it = itertools.imap(dill.dumps, it)
    return pool.map_async(run_dill_encoded, it, **kw)

if __name__ == '__main__':
    import multiprocessing as mp
    import sys,os
    p = mp.Pool(4)
    res = dill_map_async(p, lambda x:[sys.stdout.write('%s\n'%os.getpid()),x][-1],
                  [lambda x:x+1]*10,)
    res = res.get(timeout=100)
    res = map(dill.loads,res)
    print(res)