如何将*args移交给pathos的amap

Question

我有以下问题：我想使用一个来自病理学.多处理. 在

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和g，尽管我可以在f中做任何事情

如果我运行这段代码，我会得到g需要3个参数，但是给出了一个。在

^{pr2}$

我该怎么解决这个问题？在

Answer 1

我知道这看起来很奇怪，但是我喜欢传递数据的想法。在

要实现所需的内容，必须将元素作为单独的iterable项传递。下面是一个例子：

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))调用一个位，就可以知道为什么要这样做。在

通常，您可能希望使用不同的参数多次调用函数。以下是我所做的演示：

^{pr2}$

即使我显式地调用了一次函数，它也被执行了两次。 输出：

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

希望这有帮助。在

Answer 2

因为amap定义为：

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

源代码中有一个用法示例：

^{pr2}$

以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)])

或者，使用apipeapi代替：

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

当然，一旦你有了这个想法，你就可以用它的接口使输入更适合。但是为什么不直接使用multiprocessing.pool.async_apply，它的接口与您最初预期的完全一样。在