Python 多处理池从第一个 block 的输入中吞下异常

Question

我正在编写一个脚本来读取一堆文件，然后并行处理所有这些文件中的行。

我的问题是，如果脚本无法打开某些文件，它会表现得很奇怪。如果它是列表中较晚的文件之一，则它会处理较早的文件，并在到达坏文件时报告异常。但是，如果它无法打开列表中的第一个文件，那么它什么也不处理，也不会报告异常。

如何让脚本报告所有异常，无论它们在列表中的什么位置？

关键问题似乎是pool.imap() 的 block 大小。如果在提交第一个 block 之前发生异常，它将静默失败。

这里有一个重现问题的小脚本:

from multiprocessing.pool import Pool


def prepare():
    for i in range(5):
        yield i+1

    raise RuntimeError('foo')


def process(x):
    return x


def test(chunk_size):
    pool = Pool(10)
    n = raised = None
    try:
        for n in pool.imap(process, prepare(), chunksize=chunk_size):
            pass
    except RuntimeError as ex:
        raised = ex
    print(chunk_size, n, raised)


def main():
    print('chunksize n raised')
    for chunk_size in range(1, 10):
        test(chunk_size)


if __name__ == '__main__':
    main()

prepare() 函数生成五个整数，然后引发异常。该生成器被传递给 pool.imap()， block 大小从 1 到 10。然后它打印出 block 大小、收到的结果数以及引发的任何异常。

chunksize n raised
1 5 foo
2 4 foo
3 3 foo
4 4 foo
5 5 foo
6 None None
7 None None
8 None None
9 None None

您可以看到异常被正确报告，直到 block 大小增加到足以在提交第一个 block 之前发生异常。然后它默默地失败了，没有返回任何结果。

Answer 1

如果我在我自己的可移植系统上使用 Python 2.7.13 和 3.5.4 运行它(为了 py2k 和 py3k 的交叉兼容性，我稍微修改了它)，我得到:

$ python2 --version
Python 2.7.13
$ python2 mptest.py
chunksize    n raised
        1    5 foo
        2    4 foo
        3    3 foo
        4    4 foo
        5    5 foo
        6 None None
        7 None None
        8 None None
        9 None None
$ python3 --version
Python 3.5.4
$ python3 mptest.py
chunksize    n raised
        1    5 foo
        2    4 foo
        3    3 foo
        4    4 foo
        5    5 foo
        6 None foo
        7 None foo
        8 None foo
        9 None foo

我认为对于大于 5 的 block 大小它失败(并因此打印 None)这一事实并不令人惊讶，因为自生成器以来没有池进程可以获取六个参数调用 mptest 产生的结果只能调用 5 次。

令人惊讶的是，Python2.7.9 对大于 5 的 block 大小的异常表示 None，而 Python 3.5 对异常表示 foo。

这是 Issue #28699 , 固定在 commit 794623bdb2 .该修复程序显然已向后移植到 Python 3.5.4，但没有移植到 Python 2.7.9，显然也没有移植到您自己的 Python 3 版本。