Python 如何优雅地处理 SIGTERM 信号？

How to process SIGTERM signal gracefully?

提问by zerkms

Let's assume we have such a trivial daemon written in python:

假设我们有一个用 python 编写的简单守护进程：

def mainloop():
    while True:
        # 1. do
        # 2. some
        # 3. important
        # 4. job
        # 5. sleep

mainloop()

and we daemonize it using start-stop-daemonwhich by default sends SIGTERM(TERM) signal on --stop.

并且我们使用start-stop-daemonwhich 默认情况下发送SIGTERM( TERM) 信号来守护它--stop。

Let's suppose the current step performed is #2. And at this very moment we're sending TERMsignal.

假设当前执行的步骤是#2。而此时此刻，我们正在发送TERM信号。

What happens is that the execution terminates immediately.

发生的情况是执行立即终止。

I've found that I can handle the signal event using signal.signal(signal.SIGTERM, handler)but the thing is that it still interrupts the current execution and passes the control to handler.

我发现我可以使用处理信号事件，signal.signal(signal.SIGTERM, handler)但问题是它仍然中断当前执行并将控制权传递给handler.

So, my question is - is it possible to not interrupt the current execution but handle the TERMsignal in a separated thread (?) so that I was able to set shutdown_flag = Trueso that mainloop()had a chance to stop gracefully?

所以，我的问题是 - 是否有可能不中断当前执行而是TERM在单独的线程（？）中处理信号，以便我能够设置shutdown_flag = True以便mainloop()有机会优雅地停止？

采纳答案by Mayank Jaiswal

A class based clean to use solution:

基于类的清洁使用解决方案：

import signal
import time

class GracefulKiller:
  kill_now = False
  def __init__(self):
    signal.signal(signal.SIGINT, self.exit_gracefully)
    signal.signal(signal.SIGTERM, self.exit_gracefully)

  def exit_gracefully(self,signum, frame):
    self.kill_now = True

if __name__ == '__main__':
  killer = GracefulKiller()
  while not killer.kill_now:
    time.sleep(1)
    print("doing something in a loop ...")

  print("End of the program. I was killed gracefully :)")

回答by moliware

I think you are near to a possible solution.

我认为你接近一个可能的解决方案。

Execute mainloopin a separate thread and extend it with the property shutdown_flag. The signal can be caught with signal.signal(signal.SIGTERM, handler)in the main thread (not in a separate thread). The signal handler should set shutdown_flagto True and wait for the thread to end with thread.join()

mainloop在单独的线程中执行并使用属性扩展它shutdown_flag。信号可以signal.signal(signal.SIGTERM, handler)在主线程中被捕获（而不是在单独的线程中）。信号处理程序应设置shutdown_flag为 True 并等待线程结束thread.join()

回答by Will Manley

First, I'm not certain that you need a second thread to set the shutdown_flag.
Why not set it directly in the SIGTERM handler?

首先，我不确定您是否需要第二个线程来设置shutdown_flag.
为什么不直接在 SIGTERM 处理程序中设置它？

An alternative is to raise an exception from the SIGTERMhandler, which will be propagated up the stack. Assuming you've got proper exception handling (e.g. with with/contextmanagerand try: ... finally:blocks) this should be a fairly graceful shutdown, similar to if you were to Ctrl+Cyour program.

另一种方法是从SIGTERM处理程序引发异常，该异常将沿堆栈向上传播。假设您有适当的异常处理（例如使用with/contextmanager和try: ... finally:块），这应该是一个相当优雅的关闭，类似于Ctrl+C您的程序。

Example program signals-test.py:

示例程序signals-test.py：

#!/usr/bin/python

from time import sleep
import signal
import sys


def sigterm_handler(_signo, _stack_frame):
    # Raises SystemExit(0):
    sys.exit(0)

if sys.argv[1] == "handle_signal":
    signal.signal(signal.SIGTERM, sigterm_handler)

try:
    print "Hello"
    i = 0
    while True:
        i += 1
        print "Iteration #%i" % i
        sleep(1)
finally:
    print "Goodbye"

Now see the Ctrl+Cbehaviour:

现在看Ctrl+C行为：

$ ./signals-test.py default
Hello
Iteration #1
Iteration #2
Iteration #3
Iteration #4
^CGoodbye
Traceback (most recent call last):
  File "./signals-test.py", line 21, in <module>
    sleep(1)
KeyboardInterrupt
$ echo $?
1

This time I send it SIGTERMafter 4 iterations with kill $(ps aux | grep signals-test | awk '/python/ {print $2}'):

这次我SIGTERM在 4 次迭代后发送它kill $(ps aux | grep signals-test | awk '/python/ {print $2}')：

$ ./signals-test.py default
Hello
Iteration #1
Iteration #2
Iteration #3
Iteration #4
Terminated
$ echo $?
143

This time I enable my custom SIGTERMhandler and send it SIGTERM:

这次我启用我的自定义SIGTERM处理程序并发送它SIGTERM：

$ ./signals-test.py handle_signal
Hello
Iteration #1
Iteration #2
Iteration #3
Iteration #4
Goodbye
$ echo $?
0

回答by thoughtarray

Here is a simple example without threads or classes.

这是一个没有线程或类的简单示例。

import signal

run = True

def handler_stop_signals(signum, frame):
    global run
    run = False

signal.signal(signal.SIGINT, handler_stop_signals)
signal.signal(signal.SIGTERM, handler_stop_signals)

while run:
    pass # do stuff including other IO stuff

回答by Okke

Based on the previous answers, I have created a context manager which protects from sigint and sigterm.

基于之前的答案，我创建了一个上下文管理器，可以防止 sigint 和 sigterm。

import logging
import signal
import sys


class TerminateProtected:
    """ Protect a piece of code from being killed by SIGINT or SIGTERM.
    It can still be killed by a force kill.

    Example:
        with TerminateProtected():
            run_func_1()
            run_func_2()

    Both functions will be executed even if a sigterm or sigkill has been received.
    """
    killed = False

    def _handler(self, signum, frame):
        logging.error("Received SIGINT or SIGTERM! Finishing this block, then exiting.")
        self.killed = True

    def __enter__(self):
        self.old_sigint = signal.signal(signal.SIGINT, self._handler)
        self.old_sigterm = signal.signal(signal.SIGTERM, self._handler)

    def __exit__(self, type, value, traceback):
        if self.killed:
            sys.exit(0)
        signal.signal(signal.SIGINT, self.old_sigint)
        signal.signal(signal.SIGTERM, self.old_sigterm)


if __name__ == '__main__':
    print("Try pressing ctrl+c while the sleep is running!")
    from time import sleep
    with TerminateProtected():
        sleep(10)
        print("Finished anyway!")
    print("This only prints if there was no sigint or sigterm")

回答by Kron

Found easiest way for me. Here an example with fork for clarity that this way is useful for flow control.

为我找到了最简单的方法。为了清楚起见，这里有一个带有 fork 的示例，这种方式对于流量控制很有用。

import signal
import time
import sys
import os

def handle_exit(sig, frame):
    raise(SystemExit)

def main():
    time.sleep(120)

signal.signal(signal.SIGTERM, handle_exit)

p = os.fork()
if p == 0:
    main()
    os._exit()

try:
    os.waitpid(p, 0)
except (KeyboardInterrupt, SystemExit):
    print('exit handled')
    os.kill(p, 15)
    os.waitpid(p, 0)

回答by loretoparisi

The simplest solution I have found, taking inspiration by responses above is

我找到的最简单的解决方案，从上面的回答中汲取灵感是

class SignalHandler:

    def __init__(self):

        # register signal handlers
        signal.signal(signal.SIGINT, self.exit_gracefully)
        signal.signal(signal.SIGTERM, self.exit_gracefully)

        self.logger = Logger(level=ERROR)

    def exit_gracefully(self, signum, frame):
        self.logger.info('captured signal %d' % signum)
        traceback.print_stack(frame)

        ###### do your resources clean up here! ####

        raise(SystemExit)