Python Multiprocessing PicklingError: <Typ 'Funktion'> kann nicht eingelegt werden

Es tut mir leid, dass ich den Fehler nicht mit einem einfacheren Beispiel reproduzieren kann und mein Code zu kompliziert ist, um ihn zu veröffentlichen. Wenn ich das Programm in der IPython-Shell anstelle des regulären Python ausführe, funktionieren die Dinge gut.

Ich habe einige frühere Hinweise zu diesem Problem nachgeschlagen. Sie wurden alle durch die Verwendung von pool zum Aufrufen einer Funktion verursacht, die innerhalb einer Klassenfunktion definiert wurde. Dies ist aber bei mir nicht der Fall.

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

Ich würde mich über jede Hilfe freuen.

Update : Die Funktion, die ich auswähle, wird auf der obersten Ebene des Moduls definiert. Es wird jedoch eine Funktion aufgerufen, die eine verschachtelte Funktion enthält. dh f()ruft g()Anrufe auf, h()die eine verschachtelte Funktion haben i(), und ich rufe an pool.apply_async(f). f(), g(), h()Sind alle auf der obersten Ebene definiert. Ich habe ein einfacheres Beispiel mit diesem Muster versucht und es funktioniert jedoch.

Hier ist eine Liste, was eingelegt werden kann . Insbesondere können Funktionen nur ausgewählt werden, wenn sie auf der obersten Ebene eines Moduls definiert sind.

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

ergibt einen Fehler, der fast identisch mit dem von Ihnen geposteten ist:

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

Das Problem ist, dass poolalle Methoden a verwenden mp.SimpleQueue, um Aufgaben an die Arbeitsprozesse zu übergeben. Alles, was durch das geht, mp.SimpleQueuemuss auswählbar sein und foo.workist nicht auswählbar, da es nicht auf der obersten Ebene des Moduls definiert ist.

Sie kann behoben werden, indem auf der obersten Ebene eine Funktion definiert wird, die Folgendes aufruft foo.work():

def work(foo):
    foo.work()

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

Beachten Sie, dass dies fooauswählbar ist, da Fooes auf der obersten Ebene definiert und foo.__dict__auswählbar ist.

Ich würde pathos.multiprocesssingstattdessen verwenden multiprocessing. pathos.multiprocessingist ein Fork von , multiprocessingdass Anwendungen dill. dillkann fast alles in Python serialisieren, so dass Sie viel mehr parallel senden können. Der pathosFork kann auch direkt mit mehreren Argumentfunktionen arbeiten, wie Sie dies für Klassenmethoden benötigen.

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

Holen Sie sich pathos(und wenn Sie möchten dill) hier: https://github.com/uqfoundation

Wie andere gesagt haben, multiprocessingkönnen Python-Objekte nur an Arbeitsprozesse übertragen werden, die eingelegt werden können. Wenn Sie Ihren Code nicht wie von unutbu beschrieben reorganisieren können, können Sie die dillerweiterten Beiz- / Entpickelfunktionen zum Übertragen von Daten (insbesondere Codedaten) verwenden, wie unten gezeigt.

Diese Lösung erfordert nur die Installation dillund keine anderen Bibliotheken wie folgt 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()

Ich habe festgestellt, dass ich auch genau diese Fehlerausgabe auf einem perfekt funktionierenden Code generieren kann, indem ich versuche, den Profiler darauf zu verwenden.

Beachten Sie, dass dies unter Windows war (wo das Gabeln etwas weniger elegant ist).

Ich lief:

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

Und festgestellt, dass durch Entfernen der Profilerstellung der Fehler behoben und durch Platzieren der Profilerstellung wiederhergestellt wurde. Hat mich auch verrückt gemacht, weil ich wusste, dass der Code funktioniert hat. Ich überprüfte, ob etwas pool.py aktualisiert hatte ... dann hatte ich ein sinkendes Gefühl und beseitigte die Profilerstellung und das wars.

Hier für die Archive posten, falls jemand anderes darauf stößt.

Wenn dieses Problem auftritt, besteht multiprocessingeine einfache Lösung darin, von Poolzu zu wechseln ThreadPool. Dies kann ohne Änderung des Codes außer dem Import erfolgen.

from multiprocessing.pool import ThreadPool as Pool

Dies funktioniert, weil ThreadPool den Speicher mit dem Hauptthread teilt, anstatt einen neuen Prozess zu erstellen. Dies bedeutet, dass kein Beizen erforderlich ist.

Der Nachteil dieser Methode ist, dass Python nicht die beste Sprache für den Umgang mit Threads ist. Es verwendet die sogenannte globale Interpreter-Sperre, um die Thread-Sicherheit zu gewährleisten, was hier einige Anwendungsfälle verlangsamen kann. Wenn Sie jedoch hauptsächlich mit anderen Systemen interagieren (HTTP-Befehle ausführen, mit einer Datenbank sprechen, in Dateisysteme schreiben), ist Ihr Code wahrscheinlich nicht an die CPU gebunden und wird keinen großen Einfluss haben. Tatsächlich habe ich beim Schreiben von HTTP / HTTPS-Benchmarks festgestellt, dass das hier verwendete Thread-Modell weniger Overhead und Verzögerungen aufweist, da der Overhead beim Erstellen neuer Prozesse viel höher ist als der Overhead beim Erstellen neuer Threads.

Wenn Sie also eine Menge Dinge im Python-Benutzerbereich verarbeiten, ist dies möglicherweise nicht die beste Methode.

Diese Lösung erfordert nur die Installation von Dill und keine anderen Bibliotheken als 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

Es funktioniert auch für numpy Arrays.

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

Dieser Fehler tritt auch auf, wenn Sie eine integrierte Funktion im Modellobjekt haben, die an den asynchronen Job übergeben wurde.

So stellen Sie sicher , die überprüfen Modellobjekte , die vergangen sind , keine eingebauten Funktionen. (In unserem Fall haben wir die FieldTracker()Funktion von Django-Modell-Utils innerhalb des Modells verwendet, um ein bestimmtes Feld zu verfolgen). Hier ist der Link zum relevanten GitHub-Problem.

Aufbauend auf der @ rocksportrocker-Lösung wäre es sinnvoll, beim Senden und Empfangen der Ergebnisse Dill zu verwenden.

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)