如何使用 ProcessPoolExecutor 优雅地终止 loop.run_in_executor？

Question

如何终止loop.run_in_executor与ProcessPoolExecutor优雅？启动程序后不久，发送 SIGINT (ctrl + c)。

def blocking_task():
    sleep(3)

async def main():
    exe = concurrent.futures.ProcessPoolExecutor(max_workers=4)
    loop = asyncio.get_event_loop()
    tasks = [loop.run_in_executor(exe, blocking_task) for i in range(3)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print('ctrl + c')

随着max_workers比的任务，一切正常的数量等于或更少。但是如果max_workers更大，上面代码的输出如下：

Process ForkProcess-4:
Traceback (most recent call last):
  File "/usr/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/usr/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.8/concurrent/futures/process.py", line 233, in _process_worker
    call_item = call_queue.get(block=True)
  File "/usr/lib/python3.8/multiprocessing/queues.py", line 97, in get
    res = self._recv_bytes()
  File "/usr/lib/python3.8/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/usr/lib/python3.8/multiprocessing/connection.py", line 414, in _recv_bytes
    buf = self._recv(4)
  File "/usr/lib/python3.8/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt
ctrl + c

我想只捕获一次异常（KeyboardInterrupt）并忽略或静音进程池中的其他异常，但是如何？

---

更新额外信用：

• 你能解释（原因）多重异常吗？
• 在 Windows 上添加信号处理程序是否有效？
• 如果没有，是否有没有信号处理程序的解决方案？

Answer 1

您可以使用initializer 参数的ProcessPoolExecutor安装一个处理程序SIGINT中的每个过程。

更新：在 Unix 上，当进程被创建时，它成为其父进程组的成员。如果您正在生成SIGINTwith Ctrl+C，则信号将被发送到整个进程组。

import asyncio
import concurrent.futures
import os
import signal
import sys
from time import sleep


def handler(signum, frame):
    print('SIGINT for PID=', os.getpid())
    sys.exit(0)


def init():    
    signal.signal(signal.SIGINT, handler)


def blocking_task():
    sleep(15)


async def main():
    exe = concurrent.futures.ProcessPoolExecutor(max_workers=5, initializer=init)
    loop = asyncio.get_event_loop()
    tasks = [loop.run_in_executor(exe, blocking_task) for i in range(2)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print('ctrl + c')

Ctrl-C 开始后不久：

^CSIGINT for PID= 59942
SIGINT for PID= 59943
SIGINT for PID= 59941
SIGINT for PID= 59945
SIGINT for PID= 59944
ctrl + c