Python多处理PicklingError:无法pickle <type'function'>

Question

很抱歉,我无法用更简单的示例重现错误,而且我的代码太复杂而无法发布.如果我在IPython shell而不是常规Python中运行程序,那么事情就会很顺利.

我查看了之前关于这个问题的一些注意事项.它们都是由在类函数中定义的pool to call函数引起的.但对我来说情况并非如此.

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

我将不胜感激任何帮助.

更新:我挑选的功能是在模块的顶层定义的.虽然它调用包含嵌套函数的函数.即f()要求g()调用h()具有嵌套函数i(),和我打电话pool.apply_async(f).f(),g(),h()都在顶层定义.我用这个模式尝试了更简单的例子,但它确实有效.

Answer 1

这是一个可以腌制的清单.特别是,如果函数在模块的顶层定义,则它们只能被选中.

这段代码:

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方法都使用a mp.SimpleQueue将任务传递给工作进程.经过的所有东西mp.SimpleQueue必须是可挑选的,并且foo.work不可挑选,因为它没有在模块的顶层定义.

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

def work(foo):
    foo.work()

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

请注意,这foo是可选择的,因为它Foo是在顶层定义的并且 foo.__dict__是可选择的.

Answer 2

我会用pathos.multiprocesssing,而不是multiprocessing. pathos.multiprocessing是一个multiprocessing使用的分支dill.dill可以在python中序列化几乎所有内容,因此您可以并行发送更多内容.该pathos叉也有直接与多个参数的函数的工作,因为你需要为类方法的能力.

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

Answer 3

当这个问题出现时，multiprocessing一个简单的解决方案是从 切换Pool到ThreadPool。除了导入之外，无需更改代码即可完成此操作

from multiprocessing.pool import ThreadPool as Pool

这是有效的，因为 ThreadPool 与主线程共享内存，而不是创建一个新进程——这意味着不需要酸洗。

这种方法的缺点是 python 不是处理线程的最佳语言——它使用称为全局解释器锁的东西来保持线程安全，这会减慢这里的某些用例。但是，如果您主要与其他系统交互（运行 HTTP 命令、与数据库通信、写入文件系统），那么您的代码可能不受 CPU 的约束，不会受到太大影响。事实上，我在编写 HTTP/HTTPS 基准测试时发现这里使用的线程模型具有较少的开销和延迟，因为创建新进程的开销远高于创建新线程的开销，否则程序只是在等待 HTTP回应。

因此，如果您在 python 用户空间中处理大量内容，这可能不是最好的方法。

Answer 4

正如其他人所说multiprocessing,只能将Python对象转移到可以被腌制的工作进程.如果您无法按照unutbu的描述重新组织代码,则可以使用dills扩展的pickle/unpickling功能来传输数据(尤其是代码数据),如下所示.

此解决方案仅需要安装dill而不需要其他库pathos:

import os
from multiprocessing import Pool

import dill


def run_dill_encoded(payload):
    fun, args = dill.loads(payload)
    return fun(*args)


def apply_async(pool, fun, args):
    payload = dill.dumps((fun, args))
    return pool.apply_async(run_dill_encoded, (payload,))


if __name__ == "__main__":

    pool = Pool(processes=5)

    # asyn execution of lambda
    jobs = []
    for i in range(10):
        job = apply_async(pool, lambda a, b: (a, b, a * b), (i, i + 1))
        jobs.append(job)

    for job in jobs:
        print job.get()
    print

    # async execution of static method

    class O(object):

        @staticmethod
        def calc():
            return os.getpid()

    jobs = []
    for i in range(10):
        job = apply_async(pool, O.calc, ())
        jobs.append(job)

    for job in jobs:
        print job.get()

Answer 5

我发现通过尝试在其上使用分析器,我还可以在完美的代码片段上生成完全错误输出.

请注意,这是在Windows上(分叉不太优雅).

我之前在跑步:

python -m profile -o output.pstats <script> 

并发现删除分析删除了错误并放置分析恢复它.因为我知道过去常用的代码,所以也让我感到沮丧.我正在检查是否有什么东西更新了pool.py ...然后有一种下沉的感觉并消除了分析,就是这样.

在此处发布档案以防其他任何人遇到它.

Answer 6

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

如果模型对象中有任何内置函数传递给异步作业，也会出现此错误。

因此，请确保检查传递的模型对象没有内置函数。（在我们的例子中，我们使用模型内部的django-model-utilsFieldTracker()函数来跟踪某个字段）。这是相关 GitHub 问题的链接。