多处理pool.map按特定顺序调用函数

Question

如何使multiprocessing.pool.map按数字顺序分配进程？

---

更多信息:
我有一个程序可以处理几千个数据文件,并绘制每个文件的图.我正在使用a multiprocessing.pool.map将每个文件分发到处理器,它运行良好.有时这需要很长时间,在程序运行时查看输出图像会很好.如果地图进程按顺序分发快照,这将会容易得多; 相反,对于我刚刚执行的特定运行,分析的前8个快照是: 0, 78, 156, 234, 312, 390, 468, 546.有没有办法让它按数字顺序更紧密地分配它们？

---

示例:
这是一个包含相同键元素的示例代码,并显示相同的基本结果:

import sys
from multiprocessing import Pool
import time

num_proc  = 4; num_calls = 20; sleeper   = 0.1

def SomeFunc(arg):
    time.sleep(sleeper)
    print "%5d" % (arg),
    sys.stdout.flush()     # otherwise doesn't print properly on single line

proc_pool = Pool(num_proc)
proc_pool.map( SomeFunc, range(num_calls) )

产量:

   0  4  2  6   1   5   3   7   8  10  12  14  13  11   9  15  16  18  17  19

---

回答:

来自@Hayden:使用'chunksize'参数,def map(self, func, iterable, chunksize=None).

更多信息:
该chunksize决定多少次迭代同时被分配到每个处理器.例如,我上面的例子使用了2的chunksize - 这意味着每个处理器都会关闭并执行函数的2次迭代,然后返回更多('check-in').chunksize背后的权衡是,当处理器必须与其他处理器同步时,"签入"会产生开销 - 这表明你想要一个大的块.另一方面,如果你有大块,那么一个处理器可能会完成它的块,而另一个处理器还有很长的时间可以完成---所以你应该使用一个小的块.我想其他有用的信息是每个函数调用可以花多长时间.如果他们真的应该花费相同的时间 - 使用大块大小的方式更有效.另一方面,如果某些函数调用可能需要两倍的时间,那么您需要一个小的chunksize,以便处理器不会被等待.

对于我的问题,每个函数调用应该花费非常接近相同的时间(我认为),所以如果我想按顺序调用进程,我会因为签入开销而牺牲效率.

Answer 1

发生这种情况的原因是因为每个进程都有一个预定义的工作量要在映射调用开始时执行,这取决于chunksize.我们可以chunksize通过查看pool.map的源代码来计算出默认值

chunksize, extra = divmod(len(iterable), len(self._pool) * 4)
if extra:
  chunksize += 1

因此,对于20个范围和4个进程,我们将获得chunksize2个.

如果我们修改您的代码以反映这一点,我们应该得到与您现在获得的结果类似的结果:

proc_pool.map(SomeFunc, range(num_calls), chunksize=2)

这产生了输出:

0 2 6 4 1 7 5 3 8 10 12 14 9 13 15 11 16 18 17 19

现在,设置chunksize=1将确保池中的每个进程一次只能被赋予一个任务.

proc_pool.map(SomeFunc, range(num_calls), chunksize=1)

与未指定chunksize时相比,这应该确保相当好的数字排序.例如,chunksize为1会产生输出:

0 1 2 3 4 5 6 7 9 10 8 11 13 12 15 14 16 17 19 18