信号signal

信号，signal

信号是一种软件中断机制，基于该机制能实现进程间通信，故信号是实现进程通信的一种方式。

• 信号由内核线程或者用户线程发起生成，例如，SIGINT 信号通常由用户通过按下 Ctrl+C 产生。

• 内核中会保存每个进程的信息，信号生成后会根据目标进程号找到内核中的进程信息，写入信号（此处可以设置响应策略暂时阻塞信号写入）。

• 目标进程进行系统调用时，在完成调用退出内核态之前会检查是否有信号，有信号则会根据信号按照对应方式处理

• 操作系统为每个信号设置好了默认操作，进程可以通过自定义设置覆盖操作系统默认处理方式，或者直接忽略该信号。因此程序在接收到信号后，可能会按系统既定的默认操作执行，也可能会忽略该信号，或者使用我们编写的替代方案。但有些信号是不可被替换的

• 不同操作系统的信号及信号的编号有所不同，使用信号时应当注意不同操作系统的差异

接下来将在Ubuntu操作系统上基于Python的signal模块演示信号的用处。

信号种类

信号

信号处理的应用场景

1. 定时任务：使用信号实现定时任务，例如定时发送通知、定时检查资源等。
2. 异步事件处理：使用信号处理异步事件，例如用户输入、网络事件等。
3. 程序控制：使用信号控制程序的行为，例如优雅地关闭程序、处理程序错误等。

API

signal.signal(signalnum, handler)

用于设置信号处理函数

• singnalnum：信号编号，代表具体的信号
• handler：接收到信号后，调用handler进行处理该信号，handler允许三种情况：
  1. singal.SIG_DFL：进程采用操作系统的默认操作
  2. signal.SIG_IGN：进程忽略该信号
  3. 其他函数：按照函数进行调用，系统会自动给该函数传两个参数，一个是signalnum，一个是frame

signal.getsignal(signalnum)

获取当前进程对应信号的处理函数

signal.pause()

使程序进入睡眠，直到程序接收到任意一个信号

signal.alarm(time)

每隔time秒发出一个ALRM信号，若注册了ALRM信号的处理函数，则相关处理器会被调用。当time为0时，取消注册ALRM信号处理函数。

如何向指定进程发送信号

1. 程序在前台运行：可以使用例如ctrl+c等快捷键发送信号

2. 程序在后台运行：使用命令工具发送信号，例如kill命令或者Python中的os.kill

   • kill命令：
     • kill命令用于向进程发送信号，杀死进程仅仅是其一个功能
     • kill -1 pid
     • kill -9 pid
   • Python os.kill
     • os.kill(pid, signal)
     • os.getpid()

Python的signal模块注意事项

• Python signal模块要求所有信号handler必须在进程的主线程中注册
• 所以进程接收到信号之后，所有回调也是在主线程进行处理
• 主线程和子线程中任意线程均可以发送信号，包括使用signal.alarm()发送信号，信号会被进程接收
• 主线程或子线程任意线程使用signal.pause()，均可以由任意信号唤醒，该信号到达进程时唤醒所有正在等待信号的线程，信号到达之后进入等待信号状态的线程需要等待下一个信号

import signal
import threading


# 在一个子线程中设置信号handler
def usr1_handler_set():
    signal.signal(signal.SIGUSR1, lambda signal_num, frame: ...)
    # Python的signal模块不支持在子线程中为进程设置信号handler
    # ValueError: signal only works in main thread
    print("sub thread do other things")


set_signal_handler_in_sub_thread = threading.Thread(target=usr1_handler_set)
set_signal_handler_in_sub_thread.start()
set_signal_handler_in_sub_thread.join()  # 等待子线程执行完毕

示例

自定义信号处理

import os
import signal
import sys
import time


def handle_int(sig, frame):
    print("get signal: %s, I will quit" % sig)
    sys.exit(0)


def handle_hup(sig, frame):
    print("get signal: %s" % sig)


if __name__ == "__main__":
    signal.signal(1, handle_hup)
    signal.signal(2, handle_int)
    print("PID %s" % os.getpid())
    while True:
        time.sleep(3)

# kill -1 pid
# kill -9 pid

alarm信号处理

import signal


def signal_alarm_handler(signum, frame):
    """
    处理alarm信号
    :param signum: 信号数字
    :param frame: 栈帧
    :return: 
    """
    print(signum, type(frame))
    print("Now, it's the time to exit")
    exit()


signal.signal(signal.SIGALRM, signal_alarm_handler)
signal.alarm(1)
while True:
    print('not yet')

查看所有信号

import signal

signals_to_names = {
    getattr(signal, n): n
    for n in dir(signal)
    if n.startswith('SIG') and '_' not in n
}

for s, name in sorted(signals_to_names.items()):
    handler = signal.getsignal(s)
    if handler is signal.SIG_DFL:
        handler = 'SIG_DFL'
    elif handler is signal.SIG_IGN:
        handler = 'SIG_IGN'
    print('{:<10} ({:2d}):'.format(name, s), handler)

# SIGHUP     ( 1): SIG_DFL
# SIGINT     ( 2): <built-in function default_int_handler>
# SIGQUIT    ( 3): SIG_DFL
# SIGILL     ( 4): SIG_DFL
# SIGTRAP    ( 5): SIG_DFL
# SIGIOT     ( 6): SIG_DFL
# SIGBUS     ( 7): SIG_DFL
# SIGFPE     ( 8): SIG_DFL
# SIGKILL    ( 9): SIG_DFL
# SIGUSR1    (10): SIG_DFL
# SIGSEGV    (11): SIG_DFL
# SIGUSR2    (12): SIG_DFL
# SIGPIPE    (13): SIG_IGN
# SIGALRM    (14): SIG_DFL
# SIGTERM    (15): SIG_DFL
# SIGCLD     (17): SIG_DFL
# SIGCONT    (18): SIG_DFL
# SIGSTOP    (19): SIG_DFL
# SIGTSTP    (20): SIG_DFL
# SIGTTIN    (21): SIG_DFL
# SIGTTOU    (22): SIG_DFL
# SIGURG     (23): SIG_DFL
# SIGXCPU    (24): SIG_DFL
# SIGXFSZ    (25): SIG_IGN
# SIGVTALRM  (26): SIG_DFL
# SIGPROF    (27): SIG_DFL
# SIGWINCH   (28): SIG_DFL
# SIGPOLL    (29): SIG_DFL
# SIGPWR     (30): SIG_DFL
# SIGSYS     (31): SIG_DFL
# SIGRTMIN   (34): SIG_DFL
# SIGRTMAX   (64): SIG_DFL

信号的软中断特性，中断sleep执行

这是一段有助于理解sleep与信号间关系的代码

import signal
import time


def receive_alarm(signum, stack):
    print('Alarm :', time.ctime())
    time.sleep(6)
    print('Alarm :', time.ctime())


signal.signal(signal.SIGALRM, receive_alarm)
signal.alarm(2)

print('Before:', time.ctime())
time.sleep(4)
print('After :', time.ctime())

# Before: Sat Nov 23 10:15:41 2024
# Alarm : Sat Nov 23 10:15:43 2024
# Alarm : Sat Nov 23 10:15:49 2024
# After : Sat Nov 23 10:15:49 2024

信号阻塞

不适用信号阻塞，执行input_10_num_and_print函数时，若kill -2 pid会导致该程序直接报错KeyboardInterrupt

使用信号阻塞，对SIGINT=2进行阻塞，则即使在执行输入部分时进行kill -2 pid，也不会直接报错KeyboardInterrupt，而是等阻塞解除后再处理信号然后报错

import os
import signal


def input_10_num_and_print(block_signal=False):
    if block_signal:
        signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGINT])
    # start input
    nums = []
    for _ in range(10):
        nums.append(int(input()))
    print(nums)
    # end input and print nums
    if block_signal:
        signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGINT])


if __name__ == '__main__':
    print(f"PID = {os.getpid()}")
    input_10_num_and_print(True) # 置True表示使用信号阻塞

子线程中的signal.pause()无法取消

import os
import signal
import threading
import time


# 设置一个信号handler
def usr_handler(signal_num, frame):
    print("received signal %s %s" % (signal_num, threading.currentThread()))
signal.signal(signal.SIGUSR2, usr_handler)

def wait_a_signal():
    print("wait a signal ...")
    signal.pause()
    print("received signal %s" % threading.currentThread())
wait_a_signal_thread = threading.Thread(target=wait_a_signal)
wait_a_signal_thread.start()
time.sleep(3)  # 保证子线程处于一个等待信号状态


def send_signal():
    print("sending signal", threading.currentThread())
    os.kill(os.getpid(), signal.SIGUSR2)
sender = threading.Thread(target=send_signal, name="sender")
sender.start()
sender.join()


signal.alarm(2)  # 先前的信号并不会触发子线程中的pause，这里需要主动退出
wait_a_signal_thread.join()

ref

https://blog.csdn.net/abc123lzf/article/details/101245167

https://docs.python.org/3/library/signal.html