Python信号处理与异步事件响应

signal模块处理Unix信号。信号是操作系统发送给进程的异步通知，Python通过信号处理器响应。

注册信号处理器：

import signal
import time

def handler(signum, frame):
print(f"Received signal {signum}")
print(f"Frame: {frame.f_code.co_name} at line {frame.f_lineno}")

signal.signal(signal.SIGINT, handler)
print("Press Ctrl+C...")
time.sleep(10)

signal.signal注册处理函数。signum是信号编号，frame是中断时的调用栈帧。

SIGCHLD处理子进程退出：

def sigchld_handler(signum, frame):
while True:
try:
pid, status = os.waitpid(-1, os.WNOHANG)
if pid == 0:
break
print(f"Child {pid} exited with status {status}")
except ChildProcessError:
break

signal.signal(signal.SIGCHLD, sigchld_handler)

SIGCHLD在子进程状态改变时触发。WNOHANG非阻塞等待子进程。

信号与多线程：

signal.signal(signal.SIGINT, handler)
# 在主线程中注册

在子线程中不能设置信号处理器。signal.signal只能在主线程中调用。

signal.getsignal获取当前处理器：

old_handler = signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT, new_handler)
# 保存旧处理器以便恢复

SIGALRM定时器（Unix）：

def alarm_handler(signum, frame):
print("Alarm!")

signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(5) # 5秒后触发SIGALRM

signal.alarm设置一次性定时器。再次调用取消前一定时器。返回前一定时器的剩余秒数。

SIGALRM实现超时：

def timeout_handler(signum, frame):
raise TimeoutError("Operation timed out")

class Timeout:
def __init__(self, seconds):
self.seconds = seconds

def __enter__(self):
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(self.seconds)
return self

def __exit__(self, *args):
signal.alarm(0)

with Timeout(5):
# 超过5秒会抛出TimeoutError

信号驱动IO（SIGIO）：

signal.signal(signal.SIGIO, io_handler)
fcntl.fcntl(fd, fcntl.F_SETOWN, os.getpid())
fcntl.fcntl(fd, fcntl.F_SETFL, fcntl.fcntl(fd, fcntl.F_GETFL) | os.O_ASYNC)

SIGIO在文件描述符就绪时触发。需要设置文件描述符的所有权和异步模式。

signal.set_wakeup_fd与事件循环集成：

import socket

r, w = socket.socketpair()
signal.set_wakeup_fd(w.fileno())

def signal_handler(signum, frame):
pass

signal.signal(signal.SIGINT, signal_handler)

# 在事件循环中监控r
# select.select([r], [], [])

set_wakeup_fd在信号处理时向fd写入一个字节，用于打破select/poll/epoll等待。

信号与asyncio：

import asyncio

async def main():
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGINT, lambda: print("Got SIGINT"))
loop.add_signal_handler(signal.SIGTERM, lambda: print("Got SIGTERM"))

while True:
await asyncio.sleep(1)

asyncio事件循环通过信号处理器安全集成信号。add_signal_handler在事件循环上下文中调用回调。

signal.pthread_sigmask控制信号屏蔽（Python 3.11+）：

signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
# SIGINT被屏蔽
signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGINT})
# SIGINT解除屏蔽

SIG_BLOCK添加信号到屏蔽集，SIG_UNBLOCK移除，SIG_SETMASK设置整个屏蔽集。

signal.valid_signals获取所有有效信号（Python 3.8+）：

print(signal.valid_signals())

valid_signals返回当前平台上所有可用信号的集合。

使用信号中断长时间运行的Python代码：

class InterruptibleLoop:
def __init__(self):
self._interrupted = False
signal.signal(signal.SIGINT, self._on_signal)
signal.signal(signal.SIGTERM, self._on_signal)

def _on_signal(self, signum, frame):
self._interrupted = True
print(f"Received signal {signum}, will stop at next checkpoint")

def run(self):
for i in range(1000000):
if self._interrupted:
break
# 正常工作