python - ProcessPoolExecutor 锁定了超出必要数量的 future

Question

我正在使用 ProcessPoolExecutor 来生成子进程。目前，我正在尝试通过键盘中断/Ctrl+C 优雅地退出脚本。

我正在创建包含 2 个工作人员的池并提交 5 个 future。在中断时，我试图取消所有尚未执行的 future 。如果我在前两个 future 执行期间中断，池只能取消两个 future，这意味着当前正在运行三个 future。但我只有两个 worker ，每个进程运行 5 秒。我的 future 正在执行什么或为什么？

import subprocess
from concurrent.futures import ProcessPoolExecutor
import signal
import sys


def executecommands(commands):
    # Replace signal handler of parent process, so child processes will ignore terminate signals
    original_sigint_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
    processpool = ProcessPoolExecutor(1)
    # Restore original handler, so the parent process can handle terminate signals
    signal.signal(signal.SIGINT, original_sigint_handler)
    futures = []
    try:
        for command in commands:
            futures.append(processpool.submit(executecommand, command))

        processpool.shutdown()
        print("Executed commands without interruption")
    except KeyboardInterrupt:
        print("\nAttempting to cancel pending commands..")
        for future in futures:
            if future.cancel():
                print("Cancelled one command")
            else:
                print("One command was already running, couldn't cancel")
        print("Waiting for running processes to finish..")
        processpool.shutdown()
        print("Shutdown complete")
        sys.exit(0)


def executecommand(command):
    # create a subprocess and run it
    print("running command")
    process = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    print("finished command")
    return process

if __name__ == '__main__':
    print("lets go")
    commandlist = [['/bin/sleep', '5'], ['/bin/sleep', '5'], ['/bin/sleep', '5'], ['/bin/sleep', '5'], ['/bin/sleep', '5']]
    executecommands(commandlist)

Answer 1

这是 CPython 实现细节，但您唯一可以取消的 future 是不在“调用队列”中的 future。调用队列包含接下来要执行的所有 future。其大小为 max_workers + EXTRA_QUEUED_CALLS。 ( EXTRA_QUEUED_CALLS is currently set to 1 .)

在您的情况下，当前两个 future 开始执行时，调用队列将填充接下来的 3 个 future(max_workers 为 2，EXTRA_QUEUED_CALLS 为 1)。由于您总共只有 5 个 future ，因此您无法取消其中任何一个。

如果您在命令列表中填写了 2 个 worker 的 10 个 future，您将能够取消最后 5 个 future:

lets go
running command
running command
^C
Attempting to cancel pending commands..
One command was already running, couldn't cancel
One command was already running, couldn't cancel
One command was already running, couldn't cancel
One command was already running, couldn't cancel
One command was already running, couldn't cancel
Cancelled one command
Cancelled one command
Cancelled one command
Cancelled one command
Cancelled one command
Waiting for running processes to finish..
running command
running command
finished command
finished command
running command
finished command
Shutdown complete