228

我正在尝试使用multiprocessing'sPool.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))

任何人都知道问题可能是什么,或者解决它的简单方法?

4

13 回答 13

128

问题是多处理必须腌制事物以将它们吊在进程之间,并且绑定的方法是不可腌制的。解决方法(无论您是否认为它“容易”;-)是将基础结构添加到您的程序中以允许对此类方法进行腌制,并将其注册到copy_reg标准库方法。

例如,Steven Bethard 对这个线程的贡献(接近线程的末尾)展示了一种完全可行的方法来允许通过copy_reg.

于 2009-11-29T22:16:14.613 回答
78

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

multiprocessing如果你使用被调用的分支pathos.multiprocesssing,你可以直接在多处理的map函数中使用类和类方法。这是因为dillis 用来代替pickleor cPickle,并且dill可以在 python 中序列化几乎任何东西。

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

请参阅: 多处理和莳萝可以一起做什么?

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

于 2014-01-25T01:11:46.427 回答
36

您还可以在您的 中定义一个__call__()方法someClass(),该方法调用someClass.go()然后将 的实例传递someClass()给池。这个对象是可腌制的,它工作正常(对我来说)......

于 2011-08-07T20:40:35.593 回答
23

Steven Bethard 的解决方案有一些限制:

当你将你的类方法注册为一个函数时,你的类的析构函数会在每次你的方法处理完成时被惊人地调用。因此,如果您有 1 个类的实例调用其方法的 n 次,则成员可能会在 2 次运行之间消失,并且您可能会收到一条消息malloc: *** error for object 0x...: pointer being freed was not allocated(例如打开成员文件)或pure virtual method called, terminate called without an active exception(这意味着我使用的成员对象的生命周期比我的想法)。我在处理大于池大小的 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 !

所以这两种方法都不令人满意......

于 2011-09-05T14:50:52.940 回答
16

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

正如每个人所说的那样,问题在于multiprocessing代码必须腌制它发送到它已经启动的子流程的东西,而腌制器不做实例方法。

但是,您可以不发送实例方法,而是将实际的类实例以及要调用的函数的名称发送到一个普通函数,然后该函数用于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 次(每次复制一次 = 每个 pool-worker-process 需要 2 或 3 次,加上原始的一次过程)。这通常是可以的,就像在这种情况下一样,因为默认的pickler会复制整个实例并(半)秘密地重新填充它——在这种情况下,这样做:

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

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

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

例如在这种情况下。

于 2012-04-18T20:01:54.997 回答
15

您还可以在您的 中定义一个__call__()方法someClass(),该方法调用someClass.go()然后将 的实例传递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()
于 2017-01-31T14:18:07.453 回答
3

上面parisjohn的解决方案对我来说很好。此外,代码看起来干净且易于理解。在我的例子中,有几个函数可以使用 Pool 调用,所以我在下面修改了 parisjohn 的代码。我打电话是为了能够调用几个函数,函数名是在参数dict中传递的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()
于 2018-06-27T14:10:21.210 回答
1

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

from pyVmomi.VmomiSupport import VmomiJSONEncoder

使用它来创建您的列表:

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

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

pfVmomiObj = json.loads(jsonSerialized)
于 2019-07-28T19:52:41.723 回答
1

为什么不使用单独的函数?

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

print pool.map(func, arr)
于 2018-05-17T14:33:11.387 回答
1

在这个简单的情况下, wheresomeClass.f没有从类继承任何数据并且没有将任何东西附加到类,一个可能的解决方案是分离 out 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))
于 2018-03-23T18:14:11.463 回答
1

一个潜在的微不足道的解决方案是切换到使用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/

于 2016-12-20T05:08:35.787 回答
0

pathos.multiprocessing为我工作。

它有一个pool方法并序列化所有不同的东西multiprocessing

import pathos.multiprocessing as mp
pool = mp.Pool(processes=2) 
于 2021-11-29T11:28:43.640 回答
0

更新:截至撰写本文时,namedTuples 是可选的(从 python 2.7 开始)

这里的问题是子进程无法导入对象的类 - 在这种情况下是类 P-,在多模型项目的情况下,类 P 应该可以在任何使用子进程的地方导入

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

globals()["P"] = P
于 2018-06-11T13:27:12.007 回答