1

我是装饰器的新手,也许这比我第一个装饰器项目所能咀嚼的更多,但我想做的是制作一个parallel装饰器,它接受一个看起来谦逊地应用于单个参数的函数,并自动分发它multiprocessing并将其转换为适用于参数列表的函数。

我正在跟进对较早问题的这个非常有用的答案,因此我可以成功地腌制类实例方法,并且我可以获得像那里的答案这样的示例,并且工作得很好。

这是我第一次尝试并行装饰器(在咨询了一些关于线程装饰器的网络点击之后)。

###########
# Imports #
###########
import types, copy_reg, multiprocessing as mp
import pandas, numpy as np
### End Imports

##################
# Module methods #
##################

# Parallel decorator
def parallel(f):

    def executor(*args):
        _pool   = mp.Pool(2)
        _result = _pool.map_async(f, args[1:])
        # I used args[1:] because the input will be a
        # class instance method, so gotta skip over the self object.
        # but it seems like there ought to be a better way...

        _pool.close()
        _pool.join()
        return _result.get()
    return executor
### End parallel

def _pickle_method(method):
    func_name = method.im_func.__name__
    obj = method.im_self
    cls = method.im_class
    cls_name = ''
    if func_name.startswith('__') and not func_name.endswith('__'):
        cls_name = cls.__name__.lstrip('_')
    if cls_name:
        func_name = '_' + cls_name + func_name
    return _unpickle_method, (func_name, obj, cls)
### End _pickle_method

def _unpickle_method(func_name, obj, cls):
    for cls in cls.mro():
        try:
            func = cls.__dict__[func_name]
        except KeyError:
            pass
        else:
            break
    return func.__get__(obj, cls)
### End _unpickle_method

# This call to copy_reg.pickle allows you to pass methods as the first arg to
# mp.Pool methods. If you comment out this line, `pool.map(self.foo, ...)` results in
# PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup
# __builtin__.instancemethod failed
copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)
copy_reg.pickle(types.FunctionType, _pickle_method, _unpickle_method)
### End Module methods


##################
# Module classes #
##################
class Foo(object):


    def __init__(self, args):
        self.my_args = args
    ### End __init__

    def squareArg(self, arg):
        return arg**2
    ### End squareArg

    def par_squareArg(self):
        p = mp.Pool(2) # Replace 2 with the number of processors.
        q = p.map_async(self.squareArg, self.my_args)

        p.close()
        p.join()

        return q.get()
    ### End par_SquarArg

    @parallel
    def parSquare(self, num):
        return self.squareArg(num)
    ### End parSquare
### End Foo
### End Module classes


###########
# Testing #
###########
if __name__ == "__main__":

    myfoo = Foo([1,2,3,4])
    print myfoo.par_squareArg()
    print myfoo.parSquare(myfoo.my_args)

### End Testing

但是当我使用这种方法时(愚蠢地尝试用相同的_pickle_method和强大的臂酸洗功能_unpickle_method)我首先得到一个错误,AttributeError: 'function' object has no attribute 'im_func'但更一般地说,该错误表示函数不能被酸洗。

所以这个问题是双重的。(1)如何修改装饰器,如果f它获取的对象是一个类的实例方法,那么executor它返回的也是该类对象的实例方法(这样这个关于无法腌制的业务就不会发生了,因为我可以腌制那些实例方法)?_pickle_function(2) 创建附加方法和_unpickle_function方法更好吗?我认为 Python 可以腌制模块级函数,所以如果我的代码没有导致executor成为实例方法,那么它似乎应该是模块级函数,但是为什么不能腌制呢?

4

2 回答 2

3

(1)如何修改decorator,如果它取的f对象是一个类的实例方法,那么它返回的executor也是那个类对象的实例方法(这样这个业务就没法pickle了不会发生,因为我可以腌制那些实例方法)?

>>> myfoo.parSquare
<bound method Foo.executor of <__main__.Foo object at 0x101332510>>

如您所见, parSquare 实际上是已成为实例方法的执行程序,这不足为奇,因为装饰器是某种函数包装器...

如何制作功能装饰器链?可能对装饰器有最好的描述。

(2) 创建额外的 _pickle_function 和 _unpickle_function 方法更好吗?

你不需要 python 已经支持它们,事实上这copy_reg.pickle(types.FunctionType, _pickle_method, _unpickle_method)似乎有点奇怪,因为你使用相同的算法来腌制这两种类型。

现在更大的问题是,为什么我们得到PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed的错误本身看起来有些模糊,但看起来它无法查找我们的函数?
我认为发生的事情是装饰器正在用您的情况内部定义的函数覆盖该函数,parSquareexecutorexecutor是一个内部函数,parallel因此它是不可导入的,因此查找似乎失败了,这只是一种预感。

让我们尝试一个更简单的例子。

>>> def parallel(function):                        
...     def apply(values):
...         from multiprocessing import Pool
...         pool = Pool(4)
...         result = pool.map(function, values)
...         pool.close()
...         pool.join()
...         return result    
...     return apply
... 
>>> @parallel
... def square(value):
...     return value**2
... 
>>> 
>>> square([1,2,3,4])
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/System/Library/Frameworks/Python.framework/Versions/2.6/lib/python2.6/threading.py", line 522, in __bootstrap_inner
    self.run()
  File "/System/Library/Frameworks/Python.framework/Versions/2.6/lib/python2.6/threading.py", line 477, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/System/Library/Frameworks/Python.framework/Versions/2.6/lib/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

我们得到的错误几乎相同。
注意上面的代码等价于:

def parallel(function):                        
    def apply(values):
        from multiprocessing import Pool
        pool = Pool(4)
        result = pool.map(function, values)
        pool.close()
        pool.join()
        return result    
    return apply    

def square(value):
    return value**2

square = parallel(square)

这会产生相同的错误,还请注意,如果我们不重命名函数。

>>> def parallel(function):                        
...     def apply(values):
...         from multiprocessing import Pool
...         pool = Pool(4)
...         result = pool.map(function, values)
...         pool.close()
...         pool.join()
...         return result    
...     return apply    
... 
>>> def _square(value):
...     return value**2
... 
>>> square = parallel(_square)
>>> square([1,2,3,4])
[1, 4, 9, 16]
>>>

它工作得很好,我一直在寻找一种方法来控制装饰器使用名称的方式,但无济于事,我仍然想将它们与多处理一起使用,所以我想出了一个有点丑陋的解决方法:

>>> def parallel(function):                
...     def temp(_):    
...         def apply(values):
...             from multiprocessing import Pool
...             pool = Pool(4)
...             result = pool.map(function, values)
...             pool.close()
...             pool.join()
...             return result    
...         return apply
...     return temp
... 
>>> def _square(value):
...     return value*value    
... 
>>> @parallel(_square)
... def square(values):
...     pass 
... 
>>> square([1,2,3,4])
[1, 4, 9, 16]
>>>

所以基本上我将真正的函数传递给了装饰器,然后我使用了第二个函数来处理这些值,你可以看到它工作得很好。

我稍微修改了您的初始代码以更好地处理装饰器,尽管它并不完美。

import types, copy_reg, multiprocessing as mp

def parallel(f):    
    def executor(*args):
        _pool   = mp.Pool(2)
        func = getattr(args[0], f.__name__) # This will get the actual method function so we can use our own pickling procedure
        _result = _pool.map(func, args[1])
        _pool.close()
        _pool.join()
        return _result
    return executor

def _pickle_method(method):
    func_name = method.im_func.__name__
    obj = method.im_self
    cls = method.im_class
    cls_name = ''
    if func_name.startswith('__') and not func_name.endswith('__'):
        cls_name = cls.__name__.lstrip('_')
    if cls_name:
        func_name = '_' + cls_name + func_name
    return _unpickle_method, (func_name, obj, cls)

def _unpickle_method(func_name, obj, cls):
    func = None
    for cls in cls.mro():        
        if func_name in cls.__dict__:
            func = cls.__dict__[func_name] # This will fail with the decorator, since parSquare is being wrapped around as executor             
            break
        else:
            for attr in dir(cls):
                prop = getattr(cls, attr)                
                if hasattr(prop, '__call__') and prop.__name__ == func_name:
                    func = cls.__dict__[attr]
                    break
    if func == None:
        raise KeyError("Couldn't find function %s withing %s" % (str(func_name), str(cls)))        
    return func.__get__(obj, cls)

copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)

class Foo(object):
    def __init__(self, args):
        self.my_args = args
    def squareArg(self, arg):
        return arg**2
    def par_squareArg(self):
        p = mp.Pool(2) # Replace 2 with the number of processors.
        q = p.map(self.squareArg, self.my_args)
        p.close()
        p.join()
        return q    
    @parallel
    def parSquare(self, num):
        return self.squareArg(num)  

if __name__ == "__main__":
    myfoo = Foo([1,2,3,4])
    print myfoo.par_squareArg()
    print myfoo.parSquare(myfoo.my_args)  

基本上这仍然失败,AssertionError: daemonic processes are not allowed to have children因为子进程试图调用函数,所以给我们,请记住,子进程并没有真正复制代码只是名称......

一种解决方法类似于我之前提到的一种:

import types, copy_reg, multiprocessing as mp

def parallel(f):    
    def temp(_):
        def executor(*args):
            _pool   = mp.Pool(2)
            func = getattr(args[0], f.__name__) # This will get the actual method function so we can use our own pickling procedure
            _result = _pool.map(func, args[1])
            _pool.close()
            _pool.join()
            return _result        
        return executor
    return temp

def _pickle_method(method):
    func_name = method.im_func.__name__
    obj = method.im_self
    cls = method.im_class
    cls_name = ''
    if func_name.startswith('__') and not func_name.endswith('__'):
        cls_name = cls.__name__.lstrip('_')
    if cls_name:
        func_name = '_' + cls_name + func_name
    return _unpickle_method, (func_name, obj, cls)

def _unpickle_method(func_name, obj, cls):
    func = None
    for cls in cls.mro():        
        if func_name in cls.__dict__:
            func = cls.__dict__[func_name] # This will fail with the decorator, since parSquare is being wrapped around as executor             
            break
        else:
            for attr in dir(cls):
                prop = getattr(cls, attr)                
                if hasattr(prop, '__call__') and prop.__name__ == func_name:
                    func = cls.__dict__[attr]
                    break
    if func == None:
        raise KeyError("Couldn't find function %s withing %s" % (str(func_name), str(cls)))        
    return func.__get__(obj, cls)

copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)

class Foo(object):
    def __init__(self, args):
        self.my_args = args
    def squareArg(self, arg):
        return arg**2
    def par_squareArg(self):
        p = mp.Pool(2) # Replace 2 with the number of processors.
        q = p.map(self.squareArg, self.my_args)
        p.close()
        p.join()
        return q
    def _parSquare(self, num):    
        return self.squareArg(num)
    @parallel(_parSquare)
    def parSquare(self, num):
        pass    


if __name__ == "__main__":
    myfoo = Foo([1,2,3,4])
    print myfoo.par_squareArg()
    print myfoo.parSquare(myfoo.my_args)

[1, 4, 9, 16]
[1, 4, 9, 16]

最后一件事,多线程时要非常小心,这取决于您如何分割数据,实际上多线程的时间可能比单线程慢,这主要是由于来回复制值以及创建和销毁子进程的开销。

始终对单/多线程进行基准测试,并在可能的情况下正确分割您的数据。

一个例子:

import numpy
import time
from multiprocessing import Pool

def square(value):
    return value*value

if __name__ == '__main__':
    pool = Pool(5)
    values = range(1000000)
    start = time.time()
    _ = pool.map(square, values)
    pool.close()
    pool.join()
    end = time.time()

    print "multithreaded time %f" % (end - start)
    start = time.time()
    _ = map(square, values)
    end = time.time()
    print "single threaded time %f" % (end - start)

    start = time.time()
    _ = numpy.asarray(values)**2
    end = time.time()
    print "numpy time %f" % (end - start)

    v = numpy.asarray(values)
    start = time.time()
    _ = v**2
    end = time.time()
    print "numpy without pre-initialization %f" % (end - start)

给我们:

multithreaded time 0.484441
single threaded time 0.196421
numpy time 0.184163
numpy without pre-initialization 0.004490
于 2012-07-31T09:57:40.730 回答
-1

好吧,这不是您要寻找的答案,但是 Sage 有一个@parallel符合您要寻找的内容的装饰器。您可以在线找到文档源代码

但是,作为一般规则,import pdb;pdb.set_trace()请在您看到失败的行之前添加并检查所有可见的对象。如果您正在使用ipython,您可以只使用魔法命令或按照这些方式%pdb做一些事情。

于 2012-07-31T02:28:37.130 回答