使用multiprocessing Pool.map()时无法对<type 'instancemethod'>进行pickle操作

Question

我正在尝试使用 multiprocessing 的 Pool.map() 函数来同时分配工作。当我使用以下代码时，一切正常：

import multiprocessing

def f(x):
    return x*x

def go():
    pool = multiprocessing.Pool(processes=4)        
    print pool.map(f, range(10))


if __name__== '__main__' :
    go()

但是，当我用更面向对象的方法来做时，它就不工作了。它给出的错误信息是：

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

这个问题出现在我的主程序是这样的：

import someClass

if __name__== '__main__' :
    sc = someClass.someClass()
    sc.go()

而我的 someClass 类是这样的：

import multiprocessing

class someClass(object):
    def __init__(self):
        pass

    def f(self, x):
        return x*x

    def go(self):
        pool = multiprocessing.Pool(processes=4)       
        print pool.map(self.f, range(10))

有没有人知道可能是什么问题，或者有什么简单的解决办法？

Answer 1

你可以在你的 someClass() 里面定义一个 __call__() 方法，这个方法会调用 someClass.go()，然后把 someClass() 的一个实例传给池子。这个对象是可以被序列化的，效果很好（对我来说）...

Answer 2

这些解决方案都不太好，因为多进程和序列化（也就是把数据变成可以存储或传输的格式）在标准库中有很多限制，除非你使用一些额外的工具。

如果你使用一个叫做 pathos.multiprocessing 的库，它是 multiprocessing 的一个分支，你就可以在多进程的 map 函数中直接使用类和类的方法。这是因为它用的是 dill，而不是 pickle 或 cPickle，而 dill 能够序列化几乎所有的 Python 对象。

pathos.multiprocessing 还提供了一个异步的 map 函数……而且它可以处理多个参数的函数，比如 map(math.pow, [1,2,3], [4,5,6])。

你可以查看以下链接了解更多信息： 多进程和 dill 可以一起做什么？

还有： http://matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization/

>>> import pathos.pools as pp
>>> p = pp.ProcessPool(4)
>>> 
>>> def add(x,y):
...   return x+y
... 
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>> 
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>> 
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> 
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>> 
>>> p.map(t.plus, x, y)
[4, 6, 8, 10]

为了明确一点，你可以完全按照你最初想做的那样去做，而且如果你愿意的话，可以直接在解释器中执行。

>>> import pathos.pools as pp
>>> class someClass(object):
...   def __init__(self):
...     pass
...   def f(self, x):
...     return x*x
...   def go(self):
...     pool = pp.ProcessPool(4)
...     print pool.map(self.f, range(10))
... 
>>> sc = someClass()
>>> sc.go()
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
>>> 

获取代码请访问： https://github.com/uqfoundation/pathos

Answer 3

问题在于，使用多进程时，程序需要把数据打包（也就是“pickle”）才能在不同的进程之间传递，而绑定的方法是无法被打包的。解决这个问题的方法（不管你觉得这个方法“简单”与否；-)）是给你的程序添加一些功能，让这些方法可以被打包，并通过copy_reg这个标准库来注册。

比如，Steven Bethard在这个讨论串（在讨论的最后部分）中分享了一种非常有效的方法，展示了如何通过copy_reg来实现方法的打包和解包。