如何处理 SIGTERM 并且仍然有一个工作 process.terminate()

Question

我正在尝试找到一种方法来很好地处理 SIGTERM，并在主进程收到 SIGTERM 时终止我的子进程。

基本上，我是手动创建进程（但我认为问题与mp.pool 相同）

import multiprocessing as mp

...

workers = [
    mp.Process(
        target=worker,
        args=(...,)
    ) for _ in range(nb_workers)
]

我正在捕捉信号

signal.signal(signal.SIGTERM, term)
signal.signal(signal.SIGINT, term)
signal.signal(signal.SIGQUIT, term)
signal.signal(signal.SIGABRT, term)

当信号被捕获时，我想终止我的所有子进程并退出。我不想等待他们完成运行，因为他们的个人运行时间可能会很长（理解几分钟）。

同样，我不能真正设置threading.Event() 所有进程都会定期查看，因为它们基本上只是在做一个巨大但缓慢的操作（取决于几个库）。

我的想法是在捕获信号时设置一个标志，然后在设置标志时让一个看门狗终止所有子进程。但是使用.terminate() 也会使用SIGTERM，它会被我的信号处理程序再次捕获。

例如，简化代码：

import multiprocessing as mp
import signal
import time

FLAG = False


def f(x):
    time.sleep(5)
    print(x)
    return x * x


def term(signum, frame):
    print(f'Received Signal {signum}')
    global FLAG
    FLAG = True


def terminate(w):
    for process in w:
        print('Terminating worker {}'.format(process.pid))
        process.terminate()
        process.join()
        process.close()


signal.signal(signal.SIGTERM, term)
signal.signal(signal.SIGINT, term)
signal.signal(signal.SIGQUIT, term)
signal.signal(signal.SIGABRT, term)


if __name__ == '__main__':
    workers = [
        mp.Process(
            target=f,
            args=(i,)
        ) for i in range(4)
    ]
    for process in workers:
        process.start()
    while not FLAG:
        time.sleep(0.1)
    print('flag set')
    terminate(workers)
    print('Done')

如果我在进程完成之前中断代码（使用 ctrl-c）：

Received Signal 2
Received Signal 2
Received Signal 2
Received Signal 2
Received Signal 2

flag set
Terminating worker 27742
Received Signal 15
0
Terminating worker 27743
Received Signal 15
1
3
2
Terminating worker 27744
Terminating worker 27745
Done

如您所见，.terminate() 似乎没有终止子进程，因为它们一直运行到最后，而且看起来我们也捕获了生成的 SIGTERM (15)。

到目前为止，我的解决方案是：

• 不知何故设法让进程定期检查threading.Event()。这意味着要彻底重新思考我们当前的流程在做什么。
• 使用.kill() 而不是.terminate()。这适用于 Linux，但它是一个不太干净的出口。不确定 Windows，但我的印象是在 Windows 上 .kill == .terminate。
• 不要再捕获 SIGTERM，假设程序永远不会以这种方式被杀死（不太可能）

有什么干净的方法来处理这个问题吗？

Answer 1

解决方案很大程度上取决于您在哪个平台上运行，这通常是用[multiprocessing] 标记的 Python 问题的情况，因此也应该使用特定平台标记此类问题，例如 @987654322 @， 也。我推断您的平台不是 Windows，因为没有为该平台定义 signal.SIGQUIT。所以我会选择 Linux。

1. 对于 Linux，您根本不希望您的子进程处理信号（例如，它们在 Ctrl-C 中断上调用函数 term 有点荒谬）。但是，对于 Windows，您希望您的子进程忽略这些中断。这意味着您希望您的主进程仅在创建子进程之后调用signal。
2. 与其使用FLAG 来指示主进程应该终止并且必须让主进程循环定期测试该值，不如让主进程等待threading.Event 更简单、更简洁、更高效例如，done_event。虽然。出于某种原因，这似乎不适用于 Windows；主进程wait 调用不会立即得到满足。
3. 如果您的进程正常完成并且触发了 so 信号，您希望某些配置能够正常终止。实现所有目标（包括此目标）的最简单方法是使您的子进程成为守护进程，该进程将在主进程终止时终止。然后创建一个守护线程，它只是等待子进程正常终止并在发生这种情况时设置done_event。因此，主进程将在调用done_event.wait() 时失败或者某种中断或正常完成。它现在要做的就是正常结束；无需针对子进程调用terminate，因为它们将在主进程结束时结束。

import multiprocessing as mp
from threading import Thread, Event
import signal
import time
import sys


IS_WINDOWS = sys.platform == 'win32'

def f(x):
    if IS_WINDOWS:
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        signal.signal(signal.SIGABRT, signal.SIG_IGN)

    time.sleep(5)
    print(x)
    return x * x

def term(signum, frame):
    print(f'Received Signal {signum}')
    if IS_WINDOWS:
        globals()['FLAG'] = True
    else:
        done_event.set()

def process_wait_thread():
    """
    wait for processes to finish normally and set done_event
    """
    for process in workers:
        process.join()

    if IS_WINDOWS:
        globals()['FLAG'] = True
    else:
        done_event.set()

if __name__ == '__main__':

    if IS_WINDOWS:
        globals()['FLAG'] = False
    else:
        done_event = Event()

    workers = [
        mp.Process(
            target=f,
            args=(i,),
            daemon=True
        ) for i in range(4)
    ]
    for process in workers:
        process.start()

    # We don't want subprocesses to inherit these so
    # call signal after we start the processes:
    signal.signal(signal.SIGTERM, term)
    signal.signal(signal.SIGINT, term)
    if not IS_WINDOWS:
        signal.signal(signal.SIGQUIT, term) # Not supported by Windows at all
    signal.signal(signal.SIGABRT, term)

    Thread(target=process_wait_thread, daemon=True).start()

    if IS_WINDOWS:
        while not globals()['FLAG']:
            time.sleep(0.1)
    else:
        done_event.wait()

    print('Done')