多处理池是否为每个进程提供相同数量的任务，或者它们是否被分配为可用？

Question

当您 map 可迭代到 multiprocessing.Pool 时，迭代是否在开始时为池中的每个进程划分为一个队列，或者是否存在一个公共队列，当一个进程空闲时从该队列中获取任务？

    def generate_stuff():
        for foo in range(100):
             yield foo

    def process(moo):
        print moo

    pool = multiprocessing.Pool()
    pool.map(func=process, iterable=generate_stuff())
    pool.close()

因此，鉴于此未经测试的建议代码；如果池中有 4 个进程，每个进程是否被分配了 25 个任务，或者 100 个任务被进程一个接一个地挑选出来，以便每个进程执行不同数量的任务，例如 30 , 26, 24, 20.

Answer 1

要估计 Python 实现使用的 chunksize 而无需查看其 multiprocessing 模块源代码，请运行：

#!/usr/bin/env python
import multiprocessing as mp
from itertools import groupby

def work(index):
    mp.get_logger().info(index)
    return index, mp.current_process().name

if __name__ == "__main__":
    import logging
    import sys
    logger = mp.log_to_stderr()

    # process cmdline args
    try:
        sys.argv.remove('--verbose')
    except ValueError:
        pass  # not verbose
    else:
        logger.setLevel(logging.INFO)  # verbose
    nprocesses, nitems = int(sys.argv.pop(1)), int(sys.argv.pop(1))
    # choices: 'map', 'imap', 'imap_unordered'
    map_name = sys.argv[1] if len(sys.argv) > 1 else 'map'
    kwargs = dict(chunksize=int(sys.argv[2])) if len(sys.argv) > 2 else {}

    # estimate chunksize used
    max_chunksize = 0
    map_func = getattr(mp.Pool(nprocesses), map_name)
    for _, group in groupby(sorted(map_func(work, range(nitems), **kwargs),
                                   key=lambda x: x[0]),  # sort by index
                            key=lambda x: x[1]):  # group by process name
        max_chunksize = max(max_chunksize, len(list(group)))
    print("%s: max_chunksize %d" % (map_name, max_chunksize))

这表明imap，imap_unordered默认使用chunksize=1，max_chunksize用于map取决于nprocesses，nitem（每个进程的块数不固定）和max_chunksize取决于python版本。如果指定了chunksize 参数，所有*map* 函数都会考虑它。

用法

$ ./estimate_chunksize.py nprocesses nitems [map_name [chunksize]] [--verbose]

查看各个职位的分配情况；指定--verbose参数。

Answer 2

因此，鉴于此未经测试的建议代码；如果池中有 4 个进程，每个进程是否被分配了 25 个任务，或者 100 个任务被进程一个接一个地挑选出来，以便每个进程执行不同数量的任务，例如 30 , 26, 24, 20.

嗯，显而易见的答案是测试它。

按原样，测试可能不会告诉您太多，因为作业将尽快完成，而且即使池化进程在准备就绪时抢占作业，事情最终也可能会均匀分布。但是有一个简单的方法可以解决这个问题：

import collections
import multiprocessing
import os
import random
import time

def generate_stuff():
    for foo in range(100):
        yield foo

def process(moo):
    #print moo
    time.sleep(random.randint(0, 50) / 10.)
    return os.getpid()

pool = multiprocessing.Pool()
pids = pool.map(func=process, iterable=generate_stuff(), chunksize=1)
pool.close()
print collections.Counter(pids)

如果数字是“参差不齐的”，则您知道池中的进程必须在准备就绪时抓取新作业。 （我明确地将chunksize 设置为 1 以确保块不会太大以至于每个块首先只得到一个块。）

当我在 8 核机器上运行它时：

Counter({98935: 16, 98936: 16, 98939: 13, 98937: 12, 98942: 12, 98938: 11, 98940: 11, 98941: 9})

因此，看起来流程正在快速获得新的工作。

因为你专门问了4个工人，我把Pool()改成Pool(4)，得到了这个：

Counter({98965: 31, 98962: 24, 98964: 23, 98963: 22})

但是，有一种比测试更好的方法来找出答案：阅读the source。

如您所见，map 只是调用map_async，这会创建一堆批次并将它们放在self._taskqueue 对象（Queue.Queue 实例）上。如果您进一步阅读，此队列不会直接与其他进程共享，但是有一个池管理器线程，每当一个进程完成并返回结果时，它就会从队列中弹出下一个作业并将其提交回进程。

这也是您可以找出map 的默认块大小的方法。上面链接的 2.7 实现表明它只是对 len(iterable) / (len(self._pool) * 4) 进行了四舍五入（比避免小数算术稍微详细一点）——或者，换一种说法，对于每个进程大约 4 个块来说足够大。但是你真的不应该依赖这个；该文档含糊地和间接地暗示它将使用某种启发式方法，但并没有为您提供任何关于那将是什么的保证。因此，如果您真的需要“每个进程大约 4 个块”，请明确计算。更现实地说，如果您需要除默认值之外的任何内容，您可能需要一个特定于域的值，您将要计算出（通过计算、猜测或分析）。

Answer 3

http://docs.python.org/2/library/multiprocessing.html#multiprocessing.pool.multiprocessing.Pool.map

map(func, iterable[, chunksize])

此方法将可迭代对象切成若干块 作为单独的任务提交到进程池。 （近似）大小 可以通过将 chunksize 设置为正数来指定这些块中的哪一个 整数。

我假设一个进程在处理完前一个块后会从队列中提取下一个块。

默认的chunksize 取决于iterable 的长度，因此选择的块数大约是进程数的四倍。 (source)