Как передать * аргументы пафосу

Question

У меня следующая проблема: я хотел бы использовать amap от pathos.multiprocessing.

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 аргумента, но один был дан.

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

Я знаю, это выглядит странно, однако мне нравится идея передавать данные как итеративные в пафосе.

Чтобы достичь того, что вы ищете, вы должны передать элементы как отдельный повторяемый элемент. Вот пример:

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

Надеюсь это поможет.