python - 如何将关键字列表传递给 pathos.multiprocessing？

Question

我正在使用 pathos.multiprocessing 来并行化需要使用实例方法的程序。这是一个最小的工作示例:

import time
import numpy as np
from pathos.multiprocessing import Pool, ProcessingPool, ThreadingPool

class dummy(object):
    def __init__(self, arg, key1=None, key2=-11):

        np.random.seed(arg)

        randnum = np.random.randint(0, 5)

        print 'Sleeping {} seconds'.format(randnum)
        time.sleep(randnum)

        self.value = arg
        self.more1 = key1
        self.more2 = key2

args = [0, 10, 20, 33, 82] 
keys = ['key1', 'key2']
k1val = ['car', 'borg', 'syria', 'aurora', 'libera']
k2val = ['a', 'b', 'c', 'd', 'e']
allks = [dict(zip(keys, [k1val[i], k2val[i]])) for i in range(5)]

pool = ThreadingPool(4)
result = pool.map(dummy, args, k1val, k2val)

print [[r.value, r.more1, r.more2] for r in result]

打印的结果是(如预期的那样):

Sleeping 4 seconds
Sleeping 1 seconds
Sleeping 3 seconds
Sleeping 4 seconds
Sleeping 3 seconds
[[0, 'car', 'a'], [10, 'borg', 'b'], [20, 'syria', 'c'], [33, 'aurora', 'd'], [82, 'libera', 'e']]

但是，在调用 map 时，最后两个参数的顺序很重要，如果我这样做:

result2 = pool.map(dummy, args, k2val, k1val)

我得到:

[[0, 'a', 'car'], [10, 'b', 'borg'], [20, 'c', 'syria'], [33, 'd', 'aurora'], [82, 'e', 'libera']]

而我想获得与第一个结果相同的结果。该行为与 apply_async kwds 在标准模块 multiprocessing 中可以执行的操作相同，即传递字典列表，其中在每个字典中键是关键字名称，项目是关键字参数(参见 allks)。请注意，标准模块 multiprocessing 不能使用实例方法，因此甚至不满足最低要求。

暂定为: result = pool.map(dummy, args, kwds=allks) # 这不起作用

Answer 1

我是 pathos 的作者。是的，你找到了我知道需要一些工作的东西。目前，来自 ProcessPool、ThreadPool 的 map 和 pipe(即 apply)方法，并且 ParallelPool 不能接受 kwds - 您必须将它们作为 args 传递。但是，如果您使用 _ProcessPool 或 _ThreadPool，那么您可以将 kwds 传递给它们的 map 并 apply 方法。pathos.pools 中以下划线开头的池实际上直接来自 multiprocess，因此它们具有与 multiprocessing 中相同的 API(但具有更好的序列化，因此可以传递类方法等)。

>>> from pathos.pools import _ProcessPool
>>> from multiprocess.pool import Pool
>>> Pool is _ProcessPool
True

因此，对原始代码的编辑看起来像这样(来自 OP 的建议编辑):

>>> from pathos.pools import _ThreadPool
>>> pool = _ThreadPool(4)
>>> 
[…]
>>> result = []
>>> def callback(x):
>>>    result.append(x)
>>>
>>> for a, k in zip(args, allks):
>>>     pool.apply_async(dummy, args=(a,), kwds=k, callback=callback)
>>> 
>>> pool.close()
>>> pool.join()