如何切换* args到pathos'amap

Question

我有以下问题：我想使用来自pathos.multiprocessing的amap。

import pathos as pt 

class Foo: 

    def __init__(self): 
     pass 

    def f(self, a, b, c): 
     return a + b + c 

    def g(self): 
     pool = pt.multiprocessing.ProcessPool() 
     def _h(a, b, c): 
      k = self.f(a, b, c) 
      return k/2 
     result = [pool.amap(_h, (i, j, k)) for i in range(3) for j in range(5) for k in range(7)] 
     return result 

a = Foo() 
b = a.g() 
b[0].get() 

尽管我可以在f中完成所有工作，但有必要使用这两个函数f和g。

如果我运行这个代码，我得到的g期望3个参数，但一个给出。

TypeError: _h() takes exactly 3 arguments (1 given) 

我该如何解决这个问题？

Answer 1

因为amap定义为：

149  def amap(self, f, *args, **kwds): # register a callback ? 
... 
152   return _pool.map_async(star(f), zip(*args)) # chunksize 

存在源代码的使用示例：

pool.amap(pow, [1,2,3,4], [5,6,7,8]) 

给出l作为输入：

l = [(i, j, k) for i in range(3) for j in range(5) for k in range(7)] 

你可以调换你的输入：

results = pool.amap(_h, *map(list, zip(*l))) 

或使用发电机，它应该是更快：

def getter(n): 
    for e in l: 
     yield e[n] 
result = pool.amap(_h, *[getter(n) for n in range(3)]) 

，或者使用apipe API来代替：

results = [pool.apipe(_h, l)] 

当然

，你可以让输入更适合其界面一旦你明白了。但为什么不只是使用multiprocessing.pool.async_apply来代替它，它的界面与您最初的预期完全相同。

Answer 2

我知道它看起来很奇怪，但是我喜欢将数据作为迭代传递给pathos的想法。

要实现您正在寻找的内容，您必须将元素作为单独的可迭代项传递。这里有一个例子：

def g(self): 
    pool = pt.multiprocessing.ProcessPool() 
    def _h(a, b, c): 
     k = self.f(a, b, c) 
     return k/2 
    result = [pool.amap(_h, (i,),(j,),(k,))) for i in range(3) for j in range(5) for k in range(7)] 
    return result 

请注意这里传递参数的奇怪的方式：pool.amap(_h, (i,),(j,),(k,))

你可以得到一个想法，为什么会这样，如果你跟踪_pool.map_async(star(f), zip(*args))通话位完成。

一般来说，您可能想要使用不同的参数多次调用您的函数。以下是我所做的演示：

def g(self): 
    pool = pt.multiprocessing.ProcessPool() 
    def _h(a, b, c): 
     print('a: {0} b: {1}. c: {2}'.format(a, b, c)) 
     k = self.f(a, b, c) 
     return k/2 
    return [pool.amap(_h, (1,4), (2,5), (3,6))] 

即使当我明确调用函数时，它会执行两次。 输出：

a: 1 b: 2. c: 3 
a: 4 b: 5. c: 6 

希望这会有所帮助。