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

Lamento no poder reproducir el error con un ejemplo más simple, y mi código es demasiado complicado para publicar. Si ejecuto el programa en el shell de IPython en lugar del Python normal, las cosas funcionan bien.

Busqué algunas notas anteriores sobre este problema. Todos fueron causados ​​por el uso de grupo para llamar a la función definida dentro de una función de clase. Pero este no es el caso para mí.

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

Apreciaría cualquier ayuda.

Actualización : la función I pickle se define en el nivel superior del módulo. Aunque llama a una función que contiene una función anidada. es decir, f()llama a g()llamadas h()que tienen una función anidada i(), y estoy llamando pool.apply_async(f). f(), g(), h()Están todos definidos en el nivel superior. Intenté un ejemplo más simple con este patrón y funciona.

Aquí hay una lista de lo que se puede encurtir . En particular, las funciones solo son seleccionables si se definen en el nivel superior de un módulo.

Esta pieza de código:

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

produce un error casi idéntico al publicado:

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

El problema es que pooltodos los métodos usan a mp.SimpleQueuepara pasar tareas a los procesos de trabajo. Todo lo que pasa por el mp.SimpleQueuedebe ser seleccionable y foo.workno es seleccionable ya que no está definido en el nivel superior del módulo.

Se puede solucionar definiendo una función en el nivel superior, que llama foo.work():

def work(foo):
    foo.work()

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

Tenga en cuenta que fooes seleccionable, ya que Foose define en el nivel superior y foo.__dict__es seleccionable.

Yo usaría pathos.multiprocesssing, en lugar de multiprocessing. pathos.multiprocessingEs una bifurcación de multiprocessingque utiliza dill. dillpuede serializar casi cualquier cosa en python, por lo que puede enviar mucho más en paralelo. La pathosbifurcación también tiene la capacidad de trabajar directamente con múltiples funciones de argumento, como lo necesita para los métodos de clase.

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

Obtenga pathos(y si lo desea dill) aquí: https://github.com/uqfoundation

Como han dicho otros, multiprocessingsolo se pueden transferir objetos de Python a procesos de trabajo que se pueden encurtir. Si no puede reorganizar su código como lo describe unutbu, puede usar dilllas capacidades extendidas de decapado / desempaquetado para transferir datos (especialmente datos de código) como se muestra a continuación.

Esta solución requiere solo la instalación de dilly no otras bibliotecas como 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()

He descubierto que también puedo generar exactamente esa salida de error en un fragmento de código que funciona perfectamente al intentar usar el generador de perfiles en él.

Tenga en cuenta que esto fue en Windows (donde la bifurcación es un poco menos elegante).

Yo estaba corriendo:

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

Y descubrió que al eliminar el perfil se eliminó el error y al colocar el perfil se restableció. También me estaba volviendo loco porque sabía que el código solía funcionar. Estaba comprobando si algo había actualizado pool.py ... luego tuve una sensación de hundimiento y eliminé el perfil y eso fue todo.

Publicar aquí para los archivos en caso de que alguien más se encuentre con él.

Cuando surge este problema, multiprocessinguna solución simple es cambiar de Poola ThreadPool. Esto se puede hacer sin cambiar el código que no sea la importación

from multiprocessing.pool import ThreadPool as Pool

Esto funciona porque ThreadPool comparte memoria con el hilo principal, en lugar de crear un nuevo proceso, esto significa que no es necesario el decapado.

La desventaja de este método es que Python no es el mejor lenguaje para el manejo de subprocesos: utiliza algo llamado Bloqueo global del intérprete para mantenerse seguro, lo que puede ralentizar algunos casos de uso aquí. Sin embargo, si está interactuando principalmente con otros sistemas (ejecutando comandos HTTP, hablando con una base de datos, escribiendo en sistemas de archivos), es probable que su código no esté vinculado por la CPU y no reciba mucho impacto. De hecho, cuando escribí puntos de referencia HTTP / HTTPS, descubrí que el modelo de subprocesos utilizado aquí tiene menos sobrecarga y demoras, ya que la sobrecarga de crear nuevos procesos es mucho mayor que la sobrecarga para crear nuevos subprocesos.

Entonces, si está procesando un montón de cosas en el espacio de usuario de Python, este podría no ser el mejor método.

Esta solución requiere solo la instalación de eneldo y no otras bibliotecas como 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

También funciona para matrices numpy.

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

Este error también se producirá si tiene alguna función incorporada dentro del objeto modelo que se pasó al trabajo asíncrono.

Así que asegúrese de verificar que los objetos del modelo que se pasan no tienen funciones incorporadas. (En nuestro caso, estábamos usando la FieldTracker()función de django-model-utils dentro del modelo para rastrear cierto campo). Aquí está el enlace a la cuestión relevante de GitHub.

Basándose en la solución @rocksportrocker, tendría sentido hacer eneldo al enviar y RECUPERAR los resultados.

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)