python - multiprocessing中map与Pool结合使用时如何划分数据？

Question

我有一个函数 f，我想在某个大数据上并行计算。数据可以以多种方式划分，我正在尝试就如何划分它做出决定。我试图了解 multiprocessing.Pool 中的“映射”如何准确地分发/划分数据，以便我做出正确的数据拆分决定以及选择处理器数量。我的输入数据不仅仅是一个列表，如下例所示，而是字典列表和列表列表，因此了解 Pool.map 如何划分数据似乎很重要。

话虽这么说，但我认为理解简单的例子会说明更复杂的例子。

以下 scipt 显示我们正在选择一个包含 5 个进程的池和 [1,2,3] 中的数据。这里为划分数据所做的隐式选择是什么？

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    p = Pool(5)
    print(p.map(f, [1, 2, 3]))

Answer 1

它没有记录，所以你不应该依赖任何特定的行为。您可以通过传递可选的 chunksize= 来强制执行它争论。如果您不这样做，则会使用启发式方法为您构成 chunksize 的值。这可以在私有(private)函数中找到 _map_async() ，在您的源代码树的 Lib/multiprocessing/Pool.py 中:

def _map_async(self, func, iterable, mapper, chunksize=None, ...
    '''
    Helper function to implement map, starmap and their async counterparts.
    '''
    ...
    if chunksize is None:
        chunksize, extra = divmod(len(iterable), len(self._pool) * 4)
        if extra:
            chunksize += 1
    if len(iterable) == 0:
        chunksize = 0
    ...

len(self._pool)是工作进程的数量。因此，默认情况下，如果工作项少于进程数的 4 倍，则一次传递一个。您的具体示例 ( 3 <= 4*5 ) 就是这种情况。如果工作项比进程多得多，则选择 block 大小，以便每个进程在 map() 的生命周期内将工作 block 交给大约 4​​ 次。 .例如，如果您的列表中有 500 个项目，500 / (5*4) == 25 ，因此一次将 25 个项目传递给工作进程。

为什么不是一次 100 个，这样 5 个工作人员中的每一个都只被调用一次？因为它是一种启发式 ;-) 传递少于该值是一种权衡，平衡需要完成进程间通信的次数与负载平衡(不同的工作项可能需要不同的时间才能完成)。但是关于负载平衡的任何事情都无法提前知道，因此启发式方法给予更多(但不是绝对!)权重以保持进程间调用的数量较低。

这就是它没有被记录的原因。很可能有一天会使用更智能的启发式方法。