Worker 抛出异常时 pool.map 和 pool.imap_unordered 的区别

Question

假设我这样做：

import multiprocessing as mp

def f(x):
    print x
    raise OverflowError 

if __name__ == '__main__':
    pool = mp.Pool(processes=1)
    pool.map(f, range(10))
    pool.close()
    pool.join()

出来：

0
3
6
9
Traceback (most recent call last):
  File "test1.py", line 9, in <module>
    pool.map(f, range(10))
  File "/Users/usualme/anaconda/lib/python2.7/multiprocessing/pool.py", line 251, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/Users/usualme/anaconda/lib/python2.7/multiprocessing/pool.py", line 558, in get
    raise self._value
OverflowError

现在我将map 替换为imap_unordered：

import multiprocessing as mp

def f(x):
    print x
    raise OverflowError 

if __name__ == '__main__':
    pool = mp.Pool(processes=1)
    for _ in pool.imap_unordered(f, range(10)):
        pass
    pool.close()
    pool.join()

出来：

0
1
2
3
4
5
Traceback (most recent call last):
  File "test0.py", line 9, in <module>
6
7
    for _ in pool.imap_unordered(f, range(10)):
8
  File "/Users/usualme/anaconda/lib/python2.7/multiprocessing/pool.py", line 659, in next
9
    raise value
OverflowError

我的问题：

• map：为什么会从 0 跳到 3、6 和 9？
• forimap_unordered：这次怎么一直到9？有什么不同？

Answer 1

map、imap 和 imap_unordered 处理块中的数据。就其性质而言，它们准备将这些块并行提交给多个进程。

for imap_unordered：这次怎么一直到9？什么是 不一样？

对于imap（大概是imap_unordered），the default chunksize is 1。因此，f 实际上将开始对所有值执行。

您可以通过传递 chunksize 参数来检查此行为。如果您将chunksize=1 提供给map，您将获得与其他示例类似的行为。

对于地图：为什么会从 0 跳到 3、6 和 9？

虽然文档中没有提到，但似乎map 的默认chunksize 是“更智能”的。在这里，块大小看起来是 3，所以块将是[[0,1,2],[3,4,5],[6,7,8],[9]]。

我不知道为什么当你只有一个进程时这一切仍然会发生，但我的猜测是实现会在检查它们之前聚合所有结果。异常，IIRC，在子进程中被捕获并通过 IPC 序列化 - 它们是任何其他的结果。