python - 使用多处理时无法 pickle 静态方法

Question

我正在对我的代码应用一些并行化，我在其中使用了类。我知道如果没有任何其他不同于 Python 提供的方法，就不可能 pickle 一个类方法。我找到了解决方案 here .

在我的代码中，我有两个部分应该并行化，都使用类。在这里，我发布了一个非常简单的代码，只代表我的结构(是一样的，但我删除了方法内容，这是很多数学演算，对我得到的输出来说无关紧要)。

问题是当我可以 pickle 一种方法 (shepard_interpolation) 时，我遇到了 pickle 错误。我不知道为什么会这样，或者为什么它部分起作用。

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


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)

class ImageData(object):

    def __init__(self, width=60, height=60):
        self.width = width
        self.height = height
        self.data = []
        for i in range(width):
            self.data.append([0] * height)
            
    def shepard_interpolation(self, seeds=20):
        print "ImD - Sucess"       

import copy_reg
import types
from itertools import product
from multiprocessing import Pool

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

class VariabilityOfGradients(object):
    def __init__(self):
        pass
        
    @staticmethod
    def aux():
        return "VoG - Sucess" 
            
    @staticmethod
    def calculate_orientation_uncertainty():
        results = []
        pool = Pool()
        for x, y in product(range(1, 5), range(1, 5)):
            result = pool.apply_async(VariabilityOfGradients.aux) 
        results.append(result.get())
        pool.close()
        pool.join()        

        
if __name__ == '__main__':  
    results = []
    pool = Pool()
    for _ in range(3):
        result = pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()])
        results.append(result.get())
    pool.close()
    pool.join()
    
    VariabilityOfGradients.calculate_orientation_uncertainty()   

 

运行时，我得到了

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

这几乎是相同的发现here .我看到的唯一区别是我的方法是静态的。

我注意到在我的 calculate_orientation_uncertainty 中，当我将函数调用为 result = pool.apply_async(VariabilityOfGradients.aux()) 时，即带有括号(在我从未见过的文档示例)，它似乎有效。但是，当我尝试获得结果时，我收到了

TypeError: 'int' object is not callable

我怎样才能正确地做到这一点？

Answer 1

您可以在模块级别和定义一个普通函数，也可以定义一个静态方法。这保留了静态方法的调用语法、自省(introspection)和可继承性特性，同时避免了 pickling 问题:

def aux():
    return "VoG - Sucess" 

class VariabilityOfGradients(object):
    aux = staticmethod(aux)

---

例如，

import copy_reg
import types
from itertools import product
import multiprocessing as mp

def _pickle_method(method):
    """
    Author: Steven Bethard (author of argparse)
    http://bytes.com/topic/python/answers/552476-why-cant-you-pickle-instancemethods
    """
    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):
    """
    Author: Steven Bethard
    http://bytes.com/topic/python/answers/552476-why-cant-you-pickle-instancemethods
    """
    for cls in cls.mro():
        try:
            func = cls.__dict__[func_name]
        except KeyError:
            pass
        else:
            break
    return func.__get__(obj, cls)

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

class ImageData(object):

    def __init__(self, width=60, height=60):
        self.width = width
        self.height = height
        self.data = []
        for i in range(width):
            self.data.append([0] * height)

    def shepard_interpolation(self, seeds=20):
        print "ImD - Success"       

def aux():
    return "VoG - Sucess" 

class VariabilityOfGradients(object):
    aux = staticmethod(aux)

    @staticmethod
    def calculate_orientation_uncertainty():
        pool = mp.Pool()
        results = []
        for x, y in product(range(1, 5), range(1, 5)):
            # result = pool.apply_async(aux) # this works too
            result = pool.apply_async(VariabilityOfGradients.aux, callback=results.append)
        pool.close()
        pool.join()
        print(results)


if __name__ == '__main__':  
    results = []
    pool = mp.Pool()
    for _ in range(3):
        result = pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()])
        results.append(result.get())
    pool.close()
    pool.join()

    VariabilityOfGradients.calculate_orientation_uncertainty()   

产量

ImD - Success
ImD - Success
ImD - Success
['VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess', 'VoG - Sucess']

---

顺便说一下，result.get()阻塞调用进程，直到 pool.apply_async(例如 ImageData.shepard_interpolation)调用的函数完成。所以

for _ in range(3):
    result = pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()])
    results.append(result.get())

实际上是按顺序调用 ImageData.shepard_interpolation，违背了池的目的。

相反你可以使用

for _ in range(3):
    pool.apply_async(ImageData.shepard_interpolation, args=[ImageData()],
                     callback=results.append)

回调函数(例如results.append)在函数完成时在调用进程的线程中被调用。它被发送一个参数——函数的返回值。因此，没有什么能阻止三个 pool.apply_async 调用被快速调用，并且由对 ImageData.shepard_interpolation 的三个调用完成的工作将同时执行。

或者，在这里使用 pool.map 可能更简单。

results = pool.map(ImageData.shepard_interpolation, [ImageData()]*3)