python - 为什么 mp.Pool fast 会失败，而 ProcessPoolExecutor 不会？

Question

1.

    with ProcessPoolExecutor() as executor:
        futures = [executor.submit(foo, user_number, user_id)
                   for user_number, user_id in enumerate(user_ids, start=1)]

        for future in as_completed(futures):
            future.result()

2.

    pool = Pool()
    results = [pool.apply_async(foo, args=(user_number, user_id))
               for user_number, user_id in enumerate(user_ids, start=1)]
    for result in results:
        result.get()
    pool.close()
    pool.join()

3.

    pool = Pool()
    results = [pool.apply_async(foo, args=(user_number, user_id))
               for user_number, user_id in enumerate(user_ids, start=1)]
    try:
        for result in results:
            result.get()
    finally:
        pool.close()
        pool.join()

foo 完成一些工作然后引发值错误。

使用第一个示例时，我仅在所有 future 完成后才收到异常。
在第二个示例中，一旦第一个作业失败，我就会收到异常。
对于第三个示例，其行为与第一个示例类似

如何快速失败并在退出前清理资源？

为什么会发生这种情况？根据文档，as_completed 在完成后立即返回 future，并且调用 future.result() 应该引发异常。

Python版本为3.6.9

Answer 1

问题是 Python 无法安全地取消已经开始的作业。区别只在于你告诉 Python 做什么:

• 情况 1: 异常是由 future.result() 引发的。然后，控制流跳出 with 语句，并触发 ProcessPoolExecutor.__exit__ 。默认情况下，会等待所有挂起的作业完成，因此执行会挂起，直到出现这种情况为止。

• 情况2:Python解释器遇到异常立即退出。但这并不意味着您的作业已停止运行!你永远不会等待他们完成。

• 情况 3:引发异常后，您调用 pool.join()，这与情况 1 中发生的情况大致相同。执行等待作业完成，然后退出。

您可以使用此脚本检查案例 2 到底发生了什么:

import signal
from multiprocessing import Pool
import time


def throw():
    raise ValueError()


def foo():
    def sigterm_handler(*args):
        print('received sigterm')
        raise SystemExit()

    signal.signal(signal.SIGTERM, sigterm_handler)

    while True:
        print('still alive')
        time.sleep(0.1)

pool = Pool()
results = [pool.apply_async(throw), pool.apply_async(foo)]
time.sleep(1)

for result in results:
    result.get()

pool.close()
pool.join()

在 OSX 上，输出:

$ python mp_test.py
still alive
still alive
still alive
still alive
still alive
still alive
still alive
still alive
still alive
still alive
multiprocessing.pool.RemoteTraceback:
"""
Traceback (most recent call last):
  File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 121, in worker
    result = (True, func(*args, **kwds))
  File "mp_test.py", line 8, in throw
    raise ValueError()
ValueError
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "mp_test.py", line 27, in <module>
    result.get()
  File "/usr/local/Cellar/python/3.7.3/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 657, in get
    raise self._value
ValueError
still alive
received sigterm

因此，当解释器退出时，工作人员会收到 SIGTERM 信号(不过，行为可能取决于操作系统)。请注意，SIGTERM可以被忽略(例如，通过您在工作程序中使用的第三方库)，因此不能保证您的工作程序在这种情况下实际退出。

<小时/>

现在，如果您有长时间运行的作业并且确定可以安全地取消它们(例如，因为它们不执行任何 I/O)，您可以使用类似的方法来模拟案例2的行为:

with concurrent.futures.ProcessPoolExecutor() as executor:
    try:
        futures = [executor.submit(foo, user_number, user_id)
                   for user_number, user_id in enumerate(user_ids, start=1)]

        for future in concurrent.futures.as_completed(futures):
            future.result()

    except Exception:
        # abort workers immediately if anything goes wrong
        for process in executor._processes.values():
            process.terminate()
        raise

这会将 SIGTERM 发送到遇到异常时仍在运行的所有作业，然后引发异常(并等待所有进程完成，以便您可以确定它们已停止)。再次强调，这不是一个优雅的退出 - 当您在 I/O 期间中断时，它可能并且将会导致数据丢失或悬空资源。

Python docs关于 terminate 方法，请这样说:

Terminate the process. On Unix this is done using the SIGTERM signal; on Windows TerminateProcess() is used. Note that exit handlers and finally clauses, etc., will not be executed.

Note that descendant processes of the process will not be terminated – they will simply become orphaned.

Warning

If this method is used when the associated process is using a pipe or queue then the pipe or queue is liable to become corrupted and may become unusable by other process. Similarly, if the process has acquired a lock or semaphore etc. then terminating it is liable to cause other processes to deadlock.