ThreadPoolExecutor KeyboardInterrupt

Question

我得到了以下代码，它使用concurrent.futures.ThreadPoolExecutor以一种有节制的方式启动另一个程序的进程(一次不超过30个)。我还希望能够停止所有工作，如果我按ctrl-C的python进程。这段代码需要注意的是:我必须按ctrl-C键两次。我第一次发送SIGINT时，什么也没有发生；第二次，我看到“将SIGKILL发送到进程”，进程死了，它就工作了。我的第一个SIGINT发生了什么？

execution_list = [['prog', 'arg1'], ['prog', 'arg2']] ... etc
processes = []

def launch_instance(args):
    process = subprocess.Popen(args)
    processes.append(process)
    process.wait()

try:
    with concurrent.futures.ThreadPoolExecutor(max_workers=30) as executor:
        results = list(executor.map(launch_instance, execution_list))
except KeyboardInterrupt:
    print('sending SIGKILL to processes')
    for p in processes:
        if p.poll() is None: #If process is still alive
            p.send_signal(signal.SIGKILL)

Answer 1

我在尝试解决类似的问题时偶然发现了你的问题。不是百分之百确定它会解决你的用例(我没有使用子进程)，但我认为它会的。

只要作业仍在运行，您的代码就会留在executor的上下文管理器中。我有根据的猜测是，第一个KeyboardInterrupt将被ThreadPoolExecutor捕获，它的默认行为是不启动任何新作业，等到当前作业完成，然后清理(可能还会重新启动KeyboardInterrupt)。但是这些进程可能是长时间运行的，所以您不会注意到。然后，第二个KeyboardInterrupt中断这种错误处理。

我是如何用下面的代码解决我的问题(独立线程中的无限后台进程)的：

from concurrent.futures import ThreadPoolExecutor
import signal
import threading
from time import sleep


def loop_worker(exiting):
    while not exiting.is_set():
        try:
            print("started work")
            sleep(10)
            print("finished work")
        except KeyboardInterrupt:
            print("caught keyboardinterrupt")  # never caught here. just for demonstration purposes


def loop_in_worker():
    exiting = threading.Event()
    def signal_handler(signum, frame):
        print("Setting exiting event")
        exiting.set()

    signal.signal(signal.SIGTERM, signal_handler)
    with ThreadPoolExecutor(max_workers=1) as executor:
        executor.submit(loop_worker, exiting)

        try:
            while not exiting.is_set():
                sleep(1)
                print('waiting')
        except KeyboardInterrupt:
            print('Caught keyboardinterrupt')
            exiting.set()
    print("Main thread finished (and thus all others)")


if __name__ == '__main__':
    loop_in_worker()

它使用Event向线程发出信号，告诉它们应该停止正在做的事情。在主循环中，有一个循环只是为了保持忙碌和检查任何异常。请注意，此循环位于ThreadPoolExecutor的上下文中。

作为额外的好处，它还使用相同的exiting事件来处理信号。

如果您在processes.append(process)和process.wait()之间添加一个检查信号的循环，那么它可能也会解决您的用例。这取决于你想要对正在运行的进程做什么，你应该在那里采取什么行动。

如果您从命令行运行我的脚本并按ctrl-C，您应该会看到类似以下内容：

started work
waiting
waiting
^CCaught keyboardinterrupt

   # some time passes here

finished work
Main thread finished (and thus all others)

我的解决方案的灵感来自this blog post