python - 当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

现在我用 imap_unordered 替换 map:

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？
• 对于imap_unordered:为什么这次它一直到9？有什么不同？

Answer 1

map、imap 和 imap_unordered 以 block 的形式处理数据。从本质上讲，他们准备将这些 block 并行提交给多个进程。

for imap_unordered: why does it go all the way to 9 this time? What's different?

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

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

for map: why is it jumping from 0 to 3, 6, and 9?

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

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