20

I would like to use a decorator on a function that I will subsequently pass to a multiprocessing pool. However, the code fails with "PicklingError: Can't pickle : attribute lookup __builtin__.function failed". I don't quite see why it fails here. I feel certain that it's something simple, but I can't find it. Below is a minimal "working" example. I thought that using the functools function would be enough to let this work.

If I comment out the function decoration, it works without an issue. What is it about multiprocessing that I'm misunderstanding here? Is there any way to make this work?

Edit: After adding both a callable class decorator and a function decorator, it turns out that the function decorator works as expected. The callable class decorator continues to fail. What is it about the callable class version that keeps it from being pickled?

import random
import multiprocessing
import functools

class my_decorator_class(object):
    def __init__(self, target):
        self.target = target
        try:
            functools.update_wrapper(self, target)
        except:
            pass

    def __call__(self, elements):
        f = []
        for element in elements:
            f.append(self.target([element])[0])
        return f

def my_decorator_function(target):
    @functools.wraps(target)
    def inner(elements):
        f = []
        for element in elements:
            f.append(target([element])[0])
        return f
    return inner

@my_decorator_function
def my_func(elements):
    f = []
    for element in elements:
        f.append(sum(element))
    return f

if __name__ == '__main__':
    elements = [[random.randint(0, 9) for _ in range(5)] for _ in range(10)]
    pool = multiprocessing.Pool(processes=4)
    results = [pool.apply_async(my_func, ([e],)) for e in elements]
    pool.close()
    f = [r.get()[0] for r in results]
    print(f)
agarrett
  • 423
  • 1
  • 5
  • 15
  • This post seems to indicate that pickling decorated objects is tricky: http://gael-varoquaux.info/blog/?p=120 – Daenyth Feb 17 '12 at 23:03
  • Yes, I found that page, as well. That's why I added the `functools` wrapper. But it doesn't seem to make any difference. I confess that I don't really understand what is happening underneath to see why it fails. – agarrett Feb 17 '12 at 23:07
  • 1
    If anyone is still looking for the post, its here now: http://gael-varoquaux.info/programming/decoration-in-python-done-right-decorating-and-pickling.html – wusatosi May 13 '21 at 01:46

4 Answers4

18

The problem is that pickle needs to have some way to reassemble everything that you pickle. See here for a list of what can be pickled:

http://docs.python.org/library/pickle.html#what-can-be-pickled-and-unpickled

When pickling my_func, the following components need to be pickled:

  • An instance of my_decorator_class, called my_func.

    This is fine. Pickle will store the name of the class and pickle its __dict__ contents. When unpickling, it uses the name to find the class, then creates an instance and fills in the __dict__ contents. However, the __dict__ contents present a problem...

  • The instance of the original my_func that's stored in my_func.target.

    This isn't so good. It's a function at the top-level, and normally these can be pickled. Pickle will store the name of the function. The problem, however, is that the name "my_func" is no longer bound to the undecorated function, it's bound to the decorated function. This means that pickle won't be able to look up the undecorated function to recreate the object. Sadly, pickle doesn't have any way to know that object it's trying to pickle can always be found under the name __main__.my_func.

You can change it like this and it will work:

import random
import multiprocessing
import functools

class my_decorator(object):
    def __init__(self, target):
        self.target = target
        try:
            functools.update_wrapper(self, target)
        except:
            pass

    def __call__(self, candidates, args):
        f = []
        for candidate in candidates:
            f.append(self.target([candidate], args)[0])
        return f

def old_my_func(candidates, args):
    f = []
    for c in candidates:
        f.append(sum(c))
    return f

my_func = my_decorator(old_my_func)

if __name__ == '__main__':
    candidates = [[random.randint(0, 9) for _ in range(5)] for _ in range(10)]
    pool = multiprocessing.Pool(processes=4)
    results = [pool.apply_async(my_func, ([c], {})) for c in candidates]
    pool.close()
    f = [r.get()[0] for r in results]
    print(f)

You have observed that the decorator function works when the class does not. I believe this is because functools.wraps modifies the decorated function so that it has the name and other properties of the function it wraps. As far as the pickle module can tell, it is indistinguishable from a normal top-level function, so it pickles it by storing its name. Upon unpickling, the name is bound to the decorated function so everything works out.

Community
  • 1
  • 1
Weeble
  • 17,058
  • 3
  • 60
  • 75
  • OK. So if I want these things to pickle, and if I want to use a callable class as my decorator, then I won't be able to use the `@` decoration approach. I'll have to use it as if I were instantiating the class. Is that correct? – agarrett Feb 17 '12 at 23:30
  • I believe that is correct. Alternatively, you could avoid pickling it at all by creating a trivial non-decorated top-level function that just forwards to the decorated function. – Weeble Feb 17 '12 at 23:39
4

I also had some problem using decorators in multiprocessing. I'm not sure if it's the same problem as yours:

My code looked like this:

from multiprocessing import Pool

def decorate_func(f):
    def _decorate_func(*args, **kwargs):
        print "I'm decorating"
        return f(*args, **kwargs)
    return _decorate_func

@decorate_func
def actual_func(x):
    return x ** 2

my_swimming_pool = Pool()
result = my_swimming_pool.apply_async(actual_func,(2,))
print result.get()

and when I run the code I get this:

Traceback (most recent call last):
  File "test.py", line 15, in <module>
    print result.get()
  File "somedirectory_too_lengthy_to_put_here/lib/python2.7/multiprocessing/pool.py", line 572, in get
    raise self._value
cPickle.PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

I fixed it by defining a new function to wrap the function in the decorator function, instead of using the decorator syntax

from multiprocessing import Pool

def decorate_func(f):
    def _decorate_func(*args, **kwargs):
        print "I'm decorating"
        return f(*args, **kwargs)
    return _decorate_func

def actual_func(x):
    return x ** 2

def wrapped_func(*args, **kwargs):
    return decorate_func(actual_func)(*args, **kwargs)

my_swimming_pool = Pool()
result = my_swimming_pool.apply_async(wrapped_func,(2,))
print result.get()

The code ran perfectly and I got:

I'm decorating
4

I'm not very experienced at Python, but this solution solved my problem for me

blueishpoop
  • 226
  • 1
  • 10
1

If you want the decorators too bad (like me), you can also use the exec() command on the function string, to circumvent the mentioned pickling.

I wanted to be able to pass all the arguments to an original function and then use them successively. The following is my code for it.

At first, I made a make_functext() function to convert the target function object to a string. For that, I used the getsource() function from the inspect module (see doctumentation here and note that it can't retrieve source code from compiled code etc.). Here it is:

from inspect import getsource

def make_functext(func):
    ft = '\n'.join(getsource(func).split('\n')[1:]) # Removing the decorator, of course
    ft = ft.replace(func.__name__, 'func')          # Making function callable with 'func'
    ft = ft.replace('#§ ', '').replace('#§', '')    # For using commented code starting with '#§'
    ft = ft.strip()                                 # In case the function code was indented
    return ft

It is used in the following _worker() function that will be the target of the processes:

def _worker(functext, args):
    scope = {}               # This is needed to keep executed definitions
    exec(functext, scope)
    scope['func'](args)      # Using func from scope

And finally, here's my decorator:

from multiprocessing import Process 

def parallel(num_processes, **kwargs):
    def parallel_decorator(func, num_processes=num_processes):
        functext = make_functext(func)
        print('This is the parallelized function:\n', functext)
        def function_wrapper(funcargs, num_processes=num_processes):
            workers = []
            print('Launching processes...')
            for k in range(num_processes):
                p = Process(target=_worker, args=(functext, funcargs[k])) # use args here
                p.start()
                workers.append(p)
        return function_wrapper
    return parallel_decorator

The code can finally be used by defining a function like this:

@parallel(4)
def hello(args):
    #§ from time import sleep     # use '#§' to avoid unnecessary (re)imports in main program
    name, seconds = tuple(args)   # unpack args-list here
    sleep(seconds)
    print('Hi', name)

... which can now be called like this:

hello([['Marty', 0.5],
       ['Catherine', 0.9],
       ['Tyler', 0.7],
       ['Pavel', 0.3]])

... which outputs:

This is the parallelized function:
 def func(args):
        from time import sleep
        name, seconds = tuple(args)
        sleep(seconds)
        print('Hi', name)
Launching processes...
Hi Pavel
Hi Marty
Hi Tyler
Hi Catherine

Thanks for reading, this is my very first post. If you find any mistakes or bad practices, feel free to leave a comment. I know that these string conversions are quite dirty, though...

N. Jonas Figge
  • 92
  • 7
0

If you use this code for your decorator:

import multiprocessing
from types import MethodType


DEFAULT_POOL = []


def run_parallel(_func=None, *, name: str = None, context_pool: list = DEFAULT_POOL):

    class RunParallel:
        def __init__(self, func):
            self.func = func

        def __call__(self, *args, **kwargs):
            process = multiprocessing.Process(target=self.func, name=name, args=args, kwargs=kwargs)
            context_pool.append(process)
            process.start()

        def __get__(self, instance, owner):
            return self if instance is None else MethodType(self, instance)

    if _func is None:
        return RunParallel
    else:
        return RunParallel(_func)


def wait_context(context_pool: list = DEFAULT_POOL, kill_others_if_one_fails: bool = False):
    finished = []
    for process in context_pool:
        process.join()
        finished.append(process)

        if kill_others_if_one_fails and process.exitcode != 0:
            break

    if kill_others_if_one_fails:
        # kill unfinished processes
        for process in context_pool:
            if process not in finished:
                process.kill()

        # wait for every process to be dead
        for process in context_pool:
            process.join()

Then you can use it like this, in these 4 examples:

@run_parallel
def m1(a, b="b"):
    print(f"m1 -- {a=} {b=}")


@run_parallel(name="mym2", context_pool=DEFAULT_POOL)
def m2(d, cc="cc"):
    print(f"m2 -- {d} {cc=}")
    a = 1/0


class M:

    @run_parallel
    def c3(self, k, n="n"):
        print(f"c3 -- {k=} {n=}")

    @run_parallel(name="Mc4", context_pool=DEFAULT_POOL)
    def c4(self, x, y="y"):
        print(f"c4 -- {x=} {y=}")


if __name__ == "__main__":
    m1(11)
    m2(22)
    M().c3(33)
    M().c4(44)
    wait_context(kill_others_if_one_fails=True)

The output will be:

m1 -- a=11 b='b'

m2 -- 22 cc='cc'

c3 -- k=33 n='n'

(followed by the exception raised in method m2)

Pedro Nunes
  • 367
  • 2
  • 5