Python 多处理 PicklingError：无法腌制 <type 'function'>

Question

很抱歉，我无法用更简单的示例重现错误，而且我的代码太复杂而无法发布。如果我在 IPython shell 而不是常规 Python 中运行该程序，一切都会顺利进行。

我查了一些以前关于这个问题的笔记。它们都是由使用池调用类函数中定义的函数引起的。但对我来说情况并非如此。

 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

我将不胜感激任何帮助。

更新：我 pickle 的函数定义在模块的顶层。尽管它调用了一个包含嵌套函数的函数。 ie, f() calls g() calls h() which has a nested function i() , and I am calling pool.apply_async(f) . f() , g() , h() 都是在顶层定义的。我用这个模式尝试了更简单的例子，但它确实有效。

原文由 CuriousMind 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

以下是 可以 pickle 的列表。特别是，只有在模块的顶层定义的函数才可以选择。

这段代码：

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

产生一个与您发布的错误几乎相同的错误：

 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

问题是 pool 方法都使用 mp.SimpleQueue 将任务传递给工作进程。通过 mp.SimpleQueue 的所有内容都必须是可挑选的，并且 foo.work 不可挑选，因为它未在模块的顶层定义。

它可以通过在顶层定义一个函数来修复，该函数调用 foo.work() ：

 def work(foo):
    foo.work()

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

请注意 foo 是可挑选的，因为 Foo 是在顶层定义的，而 foo.__dict__ 是可挑选的。

原文由 unutbu 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

我会使用 pathos.multiprocesssing ，而不是 multiprocessing 。 pathos.multiprocessing 是 multiprocessing 的一个分支，它使用 dill 。 dill 几乎可以在 python 中序列化任何内容，因此您可以并行发送更多内容。 pathos fork 还可以直接使用多个参数函数，因为您需要类方法。

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

在此处获取 pathos （如果您愿意，也可以获取 --- dill ）： https ://github.com/uqfoundation

原文由 Mike McKerns 发布，翻译遵循 CC BY-SA 3.0 许可协议