使用pathos的Python多处理

Question

我正在尝试使用Python的pathos将计算指定到单独的进程中，以便使用多核处理器进行加速。我的代码是这样组织的：

class:
   def foo(self,name):
    ...
    setattr(self,name,something)
    ...
   def boo(self):
      for name in list:
         self.foo(name)

由于我在使用multiprocessing.Pool时遇到了一些问题，我决定尝试一下pathos。我试过了，正如前面的主题所建议的那样：

import pathos.multiprocessing

但它导致了错误:没有模块多处理-这是我在最新的pathos版本中找不到的。

然后我尝试修改boo方法：

def boo(self):
 import pathos
 pathos.pp_map.pp_map(self.foo,list)

现在没有抛出错误，但是foo不起作用--我的类的实例没有新的属性。请帮帮我，因为我不知道下一步要搬到哪里去，花了一天的时间。

Answer 1

我是pathos的作者。我不确定你想从上面的代码中做什么。不过，也许我能给你一些启发。下面是一些类似的代码：

>>> from pathos.multiprocessing import ProcessingPool
>>> class Bar:
...   def foo(self, name):
...     return len(str(name))
...   def boo(self, things):
...     for thing in things:
...       self.sum += self.foo(thing)
...     return self.sum
...   sum = 0
... 
>>> b = Bar()
>>> results = ProcessingPool().map(b.boo, [[12,3,456],[8,9,10],['a','b','cde']])
>>> results
[6, 4, 5]
>>> b.sum
0

因此，上面发生的事情是调用Bar实例b的boo方法，其中b.boo被传递给一个新的python进程，然后针对每个嵌套列表进行计算。您可以看到结果是正确的…len("12")+len("3")+len("456")是6，依此类推。

然而，您也可以看到，当您查看b.sum时，它神秘地仍然是0。为什么b.sum仍然是零？那么，multiprocessing (以及pathos.multiprocessing)所做的就是为您通过映射传递到其他python进程…的任何内容创建一个副本然后，复制的实例被调用(并行地)，并返回被调用的方法调用的任何结果。注意:您必须返回结果，或者打印它们，或者记录它们，或者将它们发送到文件中，或者以其他方式。它们不能像您预期的那样返回到原始实例，因为它不是发送到其他处理器的原始实例。实例的副本被创建，然后被销毁--每个副本的sum属性都增加了，但原始的`b.sum属性保持不变。

但是，在pathos中有计划让上面的东西像您期望的那样工作--其中原始对象是更新的，但它还不是这样工作的。

EDIT:如果您使用pip进行安装，请注意最新发布的pathos版本已有数年之久，可能无法正确安装，也可能无法安装所有子模块。新的pathos版本正在等待发布，但在此之前，最好从github获取最新版本的代码，并从那里安装。主干在开发中大部分都是稳定的。我认为您的问题可能是由于安装过程中的“新”pip --“旧”pathos不兼容，所以并没有安装所有的包。如果pathos.multiprocessing丢失，这是最有可能的罪魁祸首。

从github这里获取pathos：https://github.com/uqfoundation/pathos

Answer 2

我是这样做的--我将函数放在类的外部并行运行，并在调用pool.map时将对象作为arg传递。然后，我返回要重新分配的对象。

from pathos.multiprocessing import ProcessingPool


def boo(args):
    b, things = args
    for thing in things:
        b.sum += b.foo(thing)
    return [b, b.sum]

class Bar:
    def __init__(self):
       self.sum = 0
    def foo(self, name):
       return len(str(name))

pool = ProcessingPool(2)
b1 = Bar()
b2 = Bar()
print(b1, b2)

results = pool.map(boo, [[b1, [12,3,456]],[b2, ['a','b','cde']]])

b1, b1s = results[0]
b2, b2s = results[1]
print(b1,b1s,b1.sum)
print(b2, b2s, b2.sum)

输出：

(<__main__.Bar instance at 0x10b341518>, <__main__.Bar instance at 0x10b341560>)
(<__main__.Bar instance at 0x10b3504d0>, 6, 6)
(<__main__.Bar instance at 0x10b350560>, 5, 5)

请注意，b1和b2不再与调用map之前相同，因为它们的副本是用来传递的，正如@Mike McKerns所描述的那样。但是，它们所有属性的值都是完整的，因为它们被传递、返回和重新分配。