无法腌制<类型 “实例方法”>当使用多处理 Pool.map() 时

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

问题在于，多处理必须腌制一些东西才能在进程之间吊起它们，而绑定方法是不可腌制的。解决方法（无论您是否认为它“简单”；-）是将基础结构添加到您的程序中，以允许对此类方法进行腌制，并将其注册到 copy_reg 标准库方法。

例如，Steven Bethard 对此主题的贡献 （接近线程的末尾）展示了一种完全可行的方法，允许通过 copy_reg 进行方法 pickling/unpickling。

Answer 2

所有这些解决方案都很丑陋，因为除非您跳出标准库，否则多重处理和酸洗会被破坏和限制。

如果您使用名为 pathos.multiprocesssing 的 multiprocessing 分支，则可以直接在 multiprocessing 的 map 函数中使用类和类方法。这是因为使用 dill 代替 pickle 或 cPickle，并且 dill 几乎可以序列化 python 中的任何内容。

pathos.multiprocessing 还提供了一个异步映射函数...并且它可以map 具有多个参数的函数（例如 map(math.pow, [1,2,3] , [4,5,6]))

请参阅：
multiprocessing 和 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

您还可以在 someClass() 中定义一个 __call__() 方法，该方法调用 someClass.go() ，然后传递 someClass.go() 的实例code>someClass() 到池中。这个对象是可腌制的并且它工作得很好（对我来说）......

Answer 4

Steven Bethard 的解决方案存在一些限制：

当您将类方法注册为函数时，每次方法处理完成时都会令人惊讶地调用类的析构函数。因此，如果您的类有 1 个实例调用了 n 次其方法，则成员可能会在两次运行之间消失，并且您可能会收到一条消息 malloc: *** error for object 0x...:pointer being freed was not allocate（例如打开成员文件）或调用纯虚拟方法，
在没有活动异常的情况下终止调用（这意味着我使用的成员对象的生命周期比我想象的要短）。我在处理 n 大于池大小时得到了这个。这是一个简短的示例：

from multiprocessing import Pool, cpu_count
from multiprocessing.pool import ApplyResult

# --------- see Stenven's solution above -------------
from copy_reg import pickle
from types import MethodType

def _pickle_method(method):
    func_name = method.im_func.__name__
    obj = method.im_self
    cls = method.im_class
    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 Myclass(object):

    def __init__(self, nobj, workers=cpu_count()):

        print "Constructor ..."
        # multi-processing
        pool = Pool(processes=workers)
        async_results = [ pool.apply_async(self.process_obj, (i,)) for i in range(nobj) ]
        pool.close()
        # waiting for all results
        map(ApplyResult.wait, async_results)
        lst_results=[r.get() for r in async_results]
        print lst_results

    def __del__(self):
        print "... Destructor"

    def process_obj(self, index):
        print "object %d" % index
        return "results"

pickle(MethodType, _pickle_method, _unpickle_method)
Myclass(nobj=8, workers=3)
# problem !!! the destructor is called nobj times (instead of once)

输出：

Constructor ...
object 0
object 1
object 2
... Destructor
object 3
... Destructor
object 4
... Destructor
object 5
... Destructor
object 6
... Destructor
object 7
... Destructor
... Destructor
... Destructor
['results', 'results', 'results', 'results', 'results', 'results', 'results', 'results']
... Destructor

__call__ 方法并不那么等价，因为 [None,...] 是从结果中读取的：

from multiprocessing import Pool, cpu_count
from multiprocessing.pool import ApplyResult

class Myclass(object):

    def __init__(self, nobj, workers=cpu_count()):

        print "Constructor ..."
        # multiprocessing
        pool = Pool(processes=workers)
        async_results = [ pool.apply_async(self, (i,)) for i in range(nobj) ]
        pool.close()
        # waiting for all results
        map(ApplyResult.wait, async_results)
        lst_results=[r.get() for r in async_results]
        print lst_results

    def __call__(self, i):
        self.process_obj(i)

    def __del__(self):
        print "... Destructor"

    def process_obj(self, i):
        print "obj %d" % i
        return "result"

Myclass(nobj=8, workers=3)
# problem !!! the destructor is called nobj times (instead of once), 
# **and** results are empty !

因此这两种方法都不令人满意...

Answer 5

您可以使用另一个快捷方式，尽管它可能效率低下，具体取决于类实例中的内容。

正如大家所说，问题在于多处理代码必须对发送到已启动的子流程的内容进行pickle，而pickler 不执行实例方法。

但是，您可以将实际的类实例加上要调用的函数名称发送到普通函数，然后使用 getattr 来调用实例方法，而不是发送实例方法，从而在 Pool 子流程中创建绑定方法。这与定义 __call__ 方法类似，只不过您可以调用多个成员函数。

从他的答案中窃取@EricH.的代码并对其进行一些注释（我重新输入了它，因此所有名称都发生了更改等，出于某种原因，这似乎比剪切和粘贴更容易:-)）以说明所有的魔力：

import multiprocessing
import os

def call_it(instance, name, args=(), kwargs=None):
    "indirect caller for instance methods and multiprocessing"
    if kwargs is None:
        kwargs = {}
    return getattr(instance, name)(*args, **kwargs)

class Klass(object):
    def __init__(self, nobj, workers=multiprocessing.cpu_count()):
        print "Constructor (in pid=%d)..." % os.getpid()
        self.count = 1
        pool = multiprocessing.Pool(processes = workers)
        async_results = [pool.apply_async(call_it,
            args = (self, 'process_obj', (i,))) for i in range(nobj)]
        pool.close()
        map(multiprocessing.pool.ApplyResult.wait, async_results)
        lst_results = [r.get() for r in async_results]
        print lst_results

    def __del__(self):
        self.count -= 1
        print "... Destructor (in pid=%d) count=%d" % (os.getpid(), self.count)

    def process_obj(self, index):
        print "object %d" % index
        return "results"

Klass(nobj=8, workers=3)

输出显示，事实上，构造函数被调用一次（在原始 pid 中），析构函数被调用 9 次（每个副本调用一次 = 每个池工作进程根据需要调用 2 或 3 次，加上原始 pid 中的一次）过程）。这通常是可以的，就像在本例中一样，因为默认的pickler会复制整个实例并（半）秘密地重新填充它——在本例中，执行以下操作：

obj = object.__new__(Klass)
obj.__dict__.update({'count':1})

——这就是为什么即使析构函数在三个工作进程，每次都会从 1 倒数到 0——当然，这样仍然会遇到麻烦。如有必要，您可以提供自己的 __setstate__：

    def __setstate__(self, adict):
        self.count = adict['count']

例如在本例中。

Answer 6

您还可以在 someClass() 中定义一个 __call__() 方法，该方法调用 someClass.go() ，然后传递 someClass.go() 的实例code>someClass() 到池中。这个对象是可腌制的并且它工作得很好（对我来说）......

class someClass(object):
   def __init__(self):
       pass
   def f(self, x):
       return x*x

   def go(self):
      p = Pool(4)
      sc = p.map(self, range(4))
      print sc

   def __call__(self, x):   
     return self.f(x)

sc = someClass()
sc.go()

Answer 7

上面的 parisjohn 的解决方案对我来说效果很好。另外，代码看起来干净且易于理解。就我而言，有一些函数需要使用 Pool 来调用，所以我在下面修改了 parisjohn 的代码。我使 __call__ 能够调用多个函数，并且函数名称在来自 go() 的参数字典中传递：

from multiprocessing import Pool
class someClass(object):
    def __init__(self):
        pass
    
    def f(self, x):
        return x*x
    
    def g(self, x):
        return x*x+1    

    def go(self):
        p = Pool(4)
        sc = p.map(self, [{"func": "f", "v": 1}, {"func": "g", "v": 2}])
        print sc

    def __call__(self, x):
        if x["func"]=="f":
            return self.f(x["v"])
        if x["func"]=="g":
            return self.g(x["v"])        

sc = someClass()
sc.go()

Answer 8

在这个简单的情况下，someClass.f 没有从类继承任何数据，也没有向类附加任何内容，一个可能的解决方案是分离出 f，因此它可以被腌制：

import multiprocessing


def f(x):
    return x*x


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

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

Answer 9

一个可能简单的解决方案是改用 multiprocessing.dummy。这是多处理接口的基于线程的实现，在 Python 2.7 中似乎没有这个问题。我在这里没有太多经验，但这种快速导入更改允许我在类方法上调用 apply_async 。

multiprocessing.dummy 上的一些好资源：

https://docs.python.org/2/library/multiprocessing.html#module-multiprocessing.dummy

http://chriskiehl.com/article/parallelism-in-one-line/

Answer 10

为什么不使用单独的功能？

def func(*args, **kwargs):
    return inst.method(args, kwargs)

print pool.map(func, arr)

Answer 11

我遇到了同样的问题，但发现有一个 JSON 编码器可用于在进程之间移动这些对象。

from pyVmomi.VmomiSupport import VmomiJSONEncoder

使用它来创建列表：

jsonSerialized = json.dumps(pfVmomiObj, cls=VmomiJSONEncoder)

然后在映射函数中，使用它来恢复对象：

pfVmomiObj = json.loads(jsonSerialized)

Answer 12

更新：截至撰写本文之日，namedTuples 是可以选择的（从 python 2.7 开始）

这里的问题是子进程无法导入对象的类 - 在本例中为类 P-，在本例中在多模型项目中，P 类应该在子进程使用的任何地方都是可导入的，

一个快速的解决方法是通过影响 globals() 来使其可导入

globals()["P"] = P

Answer 13

pathos.multiprocessing 对我有用。

它有一个pool方法并序列化所有内容，这与multiprocessing不同

import pathos.multiprocessing as mp
pool = mp.Pool(processes=2) 

Answer 14

甚至不需要安装完整的 pathos 包。

实际上，唯一需要的包是 dill (pip install dill)，然后用 dill 覆盖多处理 Pickler：

dill.Pickler.dumps, dill.Pickler.loads = dill.dumps, dill.loads
multiprocessing.reduction.ForkingPickler = dill.Pickler
multiprocessing.reduction.dump = dill.dump

这个答案借自 https://stackoverflow.com/a/69253561/10686785