在多处理中与Pool结合使用时,map如何划分数据？

Question

我有一个函数f,我想要并行计算某些大数据.数据可以分为多种方式,我试图决定如何划分它.我试图理解多处理中的"映射".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的值.这可以在私有函数中找到_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)中就是这种情况.如果有多个工作项而不是进程,则选择块大小,这样每个进程在整个生命周期中将被传递大约4次map().例如,如果列表中有500个项目500 / (5*4) == 25,那么一次将有25个项目传递给工作进程.

为什么不一次100个,这样5个工人中的每一个都会被调用一次？因为它是一种启发式;-)传递少于这一点是权衡,平衡进程间通信需要完成的次数与负载平衡(不同工作项需要不同时间完成的可能性).但是事先不知道负载均衡是什么,所以启发式提供了更多(但不是绝对的)权重来保持进程间调用的数量很少.

这就是为什么它没有记录.有一天,很有可能会使用更智能的启发式算法.