特定ord中的多处理池.map调用函数

Question

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

---

更多信息：
我有一个程序可以处理几千个数据文件，每一个都有一个绘图。我正在使用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

---

回答：

From@Hayden:使用'chunksize'参数，def map(self, func, iterable, chunksize=None)。

更多信息：
chunksize确定一次分配给每个处理器的迭代次数。例如，我上面的例子使用了一个2的chunksize——这意味着每个处理器都会关闭，并在函数的两次迭代中执行它的操作，然后返回更多（“check-in”）。chunksize背后的折衷是，当处理器必须与其他处理器同步时，“签入”会有开销——这表明您需要一个大的chunksize。另一方面，如果你有大的块，那么一个处理器可能会完成它的块，而另一个处理器还有很长的时间——所以你应该使用一个小的块大小。我想附加的有用信息是每个函数调用可以花费多长时间，有多少范围。如果它们真的需要同样的时间，那么使用大数据块的效率会更高。另一方面，如果某些函数调用的时间可能是其他函数调用的两倍，则需要较小的块大小，以便处理器不会被捕获等待。

对于我的问题，每个函数调用所需的时间应该非常接近（我认为），所以如果我希望按顺序调用进程，我将牺牲效率，因为签入开销。

Answer 1

之所以会发生这种情况，是因为每个进程在调用map开始时都有一个预定义的工作量，这取决于chunksize。我们可以通过查看pool.map的源代码来计算默认的chunksize

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

因此，对于20的范围，以及4个进程，我们将得到2的chunksize。

如果我们修改您的代码以反映这一点，我们将得到与您现在得到的结果相似的结果：

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为1会产生输出：

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

Answer 2

把map改成imap怎么样：

import os
from multiprocessing import Pool
import time

num_proc = 4
num_calls = 20
sleeper = 0.1

def SomeFunc(arg):
    time.sleep(sleeper)
    print "%s %5d" % (os.getpid(), arg)
    return arg

proc_pool = Pool(num_proc)
list(proc_pool.imap(SomeFunc, range(num_calls)))

原因可能是imap的默认chunksize是1，所以它可能不会运行到map。