signal — 为异步事件设置处理程序¶
源代码： Lib/signal.py
此模块提供了在 Python 中使用信号处理程序的机制。
通用规则¶
signal.signal() 函数允许定义自定义处理程序，以便在接收到信号时执行。安装了少量默认处理程序：SIGPIPE 被忽略（因此管道和套接字上的写入错误可以作为普通的 Python 异常报告），如果父进程没有更改 SIGINT，则它会被转换为 KeyboardInterrupt 异常。
一旦为特定信号设置了处理程序，它就会一直保持安装状态，直到显式重置（Python 模拟 BSD 风格的接口，无论底层实现如何），但 SIGCHLD 的处理程序除外，它遵循底层实现。
在 WebAssembly 平台上，信号是模拟的，因此行为有所不同。这些平台上不提供某些函数和信号。
Python 信号处理程序的执行¶
Python 信号处理程序不在低级 (C) 信号处理程序内部执行。相反，低级信号处理程序会设置一个标志，告诉 虚拟机 在稍后（例如在下一个 字节码 指令）执行相应的 Python 信号处理程序。这会带来以下后果：
捕获由 C 代码中的无效操作引起的同步错误（如
SIGFPE或SIGSEGV）意义不大。Python 将从信号处理程序返回到 C 代码，C 代码很可能会再次引发相同的信号，导致 Python 似乎挂起。从 Python 3.3 开始，你可以使用faulthandler模块报告同步错误。纯粹用 C 实现的长时间运行的计算（例如对大量文本进行正则表达式匹配）可能会不间断地运行任意长时间，无论接收到任何信号。Python 信号处理程序将在计算完成后被调用。
如果处理程序引发异常，它将在主线程中“凭空”引发。有关讨论，请参阅下面的注释。
信号与线程¶
Python 信号处理程序始终在主解释器的主 Python 线程中执行，即使信号是在另一个线程中接收到的。这意味着信号不能用作线程间通信的手段。你可以改用 threading 模块中的同步原语。
此外，只有主解释器的主线程才能设置新的信号处理程序。
模块内容¶
版本 3.5 中已更改: 下面列出的信号 (SIG*)、处理程序 (SIG_DFL, SIG_IGN) 和信号掩码 (SIG_BLOCK, SIG_UNBLOCK, SIG_SETMASK) 相关常量已转换为 enums（分别为 Signals、Handlers 和 Sigmasks）。getsignal()、pthread_sigmask()、sigpending() 和 sigwait() 函数返回人类可读的 enums 作为 Signals 对象。
信号模块定义了三个枚举
- class signal.Signals¶
enum.IntEnum集合，包含 SIG* 常量和 CTRL_* 常量。在 3.5 版本加入。
- class signal.Handlers¶
enum.IntEnum集合，包含常量SIG_DFL和SIG_IGN。在 3.5 版本加入。
- class signal.Sigmasks¶
enum.IntEnum集合，包含常量SIG_BLOCK、SIG_UNBLOCK和SIG_SETMASK。可用性: Unix。
有关更多信息，请参阅手册页 sigprocmask(2) 和 pthread_sigmask(3)。
在 3.5 版本加入。
signal 模块中定义的变量是
- signal.SIG_DFL¶
这是两个标准信号处理选项之一；它将简单地执行信号的默认功能。例如，在大多数系统上，
SIGQUIT的默认操作是转储核心并退出，而SIGCHLD的默认操作是简单地忽略它。
- signal.SIG_IGN¶
这是另一个标准信号处理程序，它将简单地忽略给定的信号。
- signal.SIGFPE¶
浮点异常。例如，除以零。
参见
当除法或模运算的第二个参数为零时，会引发
ZeroDivisionError。
- signal.SIGILL¶
非法指令。
- signal.SIGINT¶
来自键盘的中断 (CTRL + C)。
默认操作是引发
KeyboardInterrupt。
- signal.SIGSEGV¶
段错误：无效内存引用。
- signal.SIGSTOP¶
停止执行（无法捕获或忽略）。
- signal.SIGSTKFLT¶
协处理器上的栈错误。Linux 内核不会引发此信号：它只能在用户空间中引发。
在 3.11 版本中新增。
- signal.SIGTERM¶
终止信号。
- SIG*
所有信号编号都以符号形式定义。例如，挂起信号定义为
signal.SIGHUP；变量名与 C 程序中使用的名称相同，如<signal.h>中所示。Unix 手册页 'signal' 列出了现有信号（在某些系统上是 signal(2)，在其他系统上列表在 signal(7) 中）。请注意，并非所有系统都定义相同的信号名称集；此模块仅定义系统定义的那些名称。
- signal.NSIG¶
比最高信号编号多一。使用
valid_signals()获取有效信号编号。
- signal.ITIMER_VIRTUAL¶
仅在进程执行时递减间隔计时器，并在到期时发送 SIGVTALRM。
- signal.ITIMER_PROF¶
在进程执行时和系统代表进程执行时都递减间隔计时器。结合 ITIMER_VIRTUAL，此计时器通常用于分析应用程序在用户空间和内核空间中花费的时间。到期时发送 SIGPROF。
- signal.SIG_BLOCK¶
pthread_sigmask()的 how 参数的可能值之一，表示信号将被阻塞。在 3.3 版本加入。
- signal.SIG_UNBLOCK¶
pthread_sigmask()的 how 参数的可能值之一，表示信号将被解除阻塞。在 3.3 版本加入。
- signal.SIG_SETMASK¶
pthread_sigmask()的 how 参数的可能值之一，表示信号掩码将被替换。在 3.3 版本加入。
signal 模块定义了一个异常
- exception signal.ItimerError¶
当底层
setitimer()或getitimer()实现出现错误时引发。如果将无效的间隔计时器或负时间传递给setitimer()，则会引发此错误。此错误是OSError的子类型。
signal 模块定义了以下函数
- signal.alarm(time)¶
如果 time 非零，则此函数请求在 time 秒后向进程发送
SIGALRM信号。任何先前计划的警报都将被取消（一次只能计划一个警报）。返回值是任何先前设置的警报被发送之前的秒数。如果 time 为零，则不计划警报，并且任何已计划的警报都将被取消。如果返回值为零，则当前没有计划警报。
- signal.getsignal(signalnum)¶
返回信号 signalnum 的当前信号处理程序。返回值可能是一个可调用的 Python 对象，或者特殊值
signal.SIG_IGN、signal.SIG_DFL或None。这里，signal.SIG_IGN表示信号先前被忽略，signal.SIG_DFL表示先前使用信号的默认处理方式，而None表示以前的信号处理程序不是从 Python 安装的。
- signal.strsignal(signalnum)¶
返回信号 signalnum 的描述，例如
SIGINT的“中断”。如果 signalnum 没有描述，则返回None。如果 signalnum 无效，则引发ValueError。在 3.8 版本加入。
- signal.valid_signals()¶
返回此平台上有效信号编号的集合。如果某些信号被系统保留用于内部使用，此值可能小于
range(1, NSIG)。在 3.8 版本加入。
- signal.pause()¶
使进程休眠直到接收到信号；然后将调用相应的处理程序。不返回任何内容。
- signal.raise_signal(signum)¶
向调用进程发送信号。不返回任何内容。
在 3.8 版本加入。
- signal.pidfd_send_signal(pidfd, sig, siginfo=None, flags=0)¶
向文件描述符 pidfd 指代的进程发送信号 sig。Python 当前不支持 siginfo 参数；它必须为
None。flags 参数用于未来的扩展；目前没有定义任何标志值。有关更多信息，请参阅 pidfd_send_signal(2) 手册页。
在 3.9 版本中新增。
- signal.pthread_kill(thread_id, signalnum)¶
向线程 thread_id 发送信号 signalnum，该线程是调用者进程中的另一个线程。目标线程可以执行任何代码（Python 或非 Python）。但是，如果目标线程正在执行 Python 解释器，则 Python 信号处理程序将由主解释器的主线程执行。因此，向特定 Python 线程发送信号的唯一目的是强制运行的系统调用以
InterruptedError失败。使用
threading.get_ident()或threading.Thread对象的ident属性来获取适合 thread_id 的值。如果 signalnum 为 0，则不发送信号，但仍执行错误检查；这可用于检查目标线程是否仍在运行。
引发 审计事件
signal.pthread_kill，参数为thread_id、signalnum。可用性: Unix。
有关更多信息，请参阅手册页 pthread_kill(3)。
另请参阅
os.kill()。在 3.3 版本加入。
- signal.pthread_sigmask(how, mask)¶
获取和/或更改调用线程的信号掩码。信号掩码是当前为调用者阻塞的信号集。返回旧的信号掩码作为信号集。
此调用的行为取决于 how 的值，如下所示。
SIG_BLOCK：被阻塞信号的集合是当前集合与 mask 参数的并集。SIG_UNBLOCK：mask 中的信号将从当前被阻塞信号集合中移除。允许尝试解除阻塞未被阻塞的信号。SIG_SETMASK：被阻塞信号的集合被设置为 mask 参数。
mask 是一个信号编号集（例如 {
signal.SIGINT,signal.SIGTERM}）。使用valid_signals()获取包含所有信号的完整掩码。例如，
signal.pthread_sigmask(signal.SIG_BLOCK, [])读取调用线程的信号掩码。可用性: Unix。
有关更多信息，请参阅手册页 sigprocmask(2) 和 pthread_sigmask(3)。
另请参阅
pause()、sigpending()和sigwait()。在 3.3 版本加入。
- signal.setitimer(which, seconds, interval=0.0)¶
设置由 which 指定的给定间隔计时器（
signal.ITIMER_REAL、signal.ITIMER_VIRTUAL或signal.ITIMER_PROF之一），使其在 seconds 秒后触发（接受浮点数，与alarm()不同），之后每隔 interval 秒触发一次（如果 interval 非零）。通过将 seconds 设置为零，可以清除由 which 指定的间隔计时器。当间隔计时器触发时，会向进程发送一个信号。发送的信号取决于所使用的计时器；
signal.ITIMER_REAL将发送SIGALRM，signal.ITIMER_VIRTUAL发送SIGVTALRM，而signal.ITIMER_PROF将发送SIGPROF。旧值作为元组返回：(延迟，间隔)。
尝试传递无效的间隔计时器将导致
ItimerError。可用性: Unix。
- signal.set_wakeup_fd(fd, *, warn_on_full_buffer=True)¶
将唤醒文件描述符设置为 fd。当程序注册的信号处理程序收到信号时，信号编号将作为单个字节写入 fd。如果未为关心的信号注册信号处理程序，则不会向唤醒 fd 写入任何内容。这可用于库唤醒 poll 或 select 调用，从而允许完全处理信号。
返回旧的唤醒 fd（如果未启用文件描述符唤醒，则返回 -1）。如果 fd 为 -1，则禁用文件描述符唤醒。如果非 -1，则 fd 必须是非阻塞的。在再次调用 poll 或 select 之前，由库负责从 fd 中移除任何字节。
启用线程时，此函数只能从主解释器的主线程调用；尝试从其他线程调用它将导致引发
ValueError异常。使用此函数有两种常见方法。两种方法中，您都使用 fd 在信号到达时唤醒，但它们在确定 哪些 信号已到达的方式上有所不同。
在第一种方法中，我们从 fd 的缓冲区中读取数据，字节值表示信号编号。这很简单，但在极少数情况下可能会遇到问题：通常 fd 将具有有限的缓冲区空间，如果太多信号太快到达，则缓冲区可能会充满，并且某些信号可能会丢失。如果您使用此方法，则应设置
warn_on_full_buffer=True，这至少会在信号丢失时导致警告打印到 stderr。在第二种方法中，我们 仅 将唤醒 fd 用于唤醒，并忽略实际的字节值。在这种情况下，我们只关心 fd 的缓冲区是空的还是非空的；缓冲区已满根本不表示问题。如果您使用此方法，则应设置
warn_on_full_buffer=False，以免您的用户被虚假的警告消息混淆。版本 3.5 中已更改: 在 Windows 上，此函数现在也支持套接字句柄。
版本 3.7 中已更改: 添加了
warn_on_full_buffer参数。
- signal.siginterrupt(signalnum, flag)¶
更改系统调用重启行为：如果 flag 为
False，则系统调用在被信号 signalnum 中断时将重启，否则系统调用将被中断。不返回任何内容。可用性: Unix。
有关更多信息，请参阅手册页 siginterrupt(3)。
请注意，使用
signal()安装信号处理程序将通过隐式调用带有给定信号的 true flag 值的siginterrupt()将重启行为重置为可中断。
- signal.signal(signalnum, handler)¶
将信号 signalnum 的处理程序设置为函数 handler。handler 可以是接受两个参数的可调用 Python 对象（见下文），或者是特殊值
signal.SIG_IGN或signal.SIG_DFL。将返回先前的信号处理程序（参见上面getsignal()的描述）。（有关更多信息，请参见 Unix 手册页 signal(2)。）启用线程时，此函数只能从主解释器的主线程调用；尝试从其他线程调用它将导致引发
ValueError异常。handler 使用两个参数调用：信号编号和当前堆栈帧（
None或帧对象；有关帧对象的描述，请参阅类型层次结构中的描述或inspect模块中的属性描述）。在 Windows 上，
signal()只能与SIGABRT、SIGFPE、SIGILL、SIGINT、SIGSEGV、SIGTERM或SIGBREAK一起使用。在任何其他情况下，都将引发ValueError。请注意，并非所有系统都定义相同的信号名称集；如果信号名称未定义为SIG*模块级常量，则将引发AttributeError。
- signal.sigpending()¶
检查等待传递到调用线程的信号集（即在阻塞时已引发的信号）。返回待处理信号的集合。
可用性: Unix。
有关更多信息，请参阅手册页 sigpending(2)。
另请参阅
pause()、pthread_sigmask()和sigwait()。在 3.3 版本加入。
- signal.sigwait(sigset)¶
暂停调用线程的执行，直到传递 sigset 中指定的信号之一。该函数接受信号（将其从待处理信号列表中移除），并返回信号编号。
可用性: Unix。
有关更多信息，请参阅手册页 sigwait(3)。
另请参阅
pause()、pthread_sigmask()、sigpending()、sigwaitinfo()和sigtimedwait()。在 3.3 版本加入。
- signal.sigwaitinfo(sigset)¶
暂停调用线程的执行，直到传递 sigset 中指定的信号之一。该函数接受信号并将其从待处理信号列表中移除。如果 sigset 中的某个信号已在调用线程中待处理，则该函数将立即返回有关该信号的信息。不会为已传递的信号调用信号处理程序。如果被 sigset 中未包含的信号中断，则该函数会引发
InterruptedError。返回值是一个对象，表示
siginfo_t结构中包含的数据，即：si_signo、si_code、si_errno、si_pid、si_uid、si_status、si_band。可用性: Unix。
有关更多信息，请参阅手册页 sigwaitinfo(2)。
另请参阅
pause()、sigwait()和sigtimedwait()。在 3.3 版本加入。
版本 3.5 中已更改: 如果函数被不在 sigset 中的信号中断且信号处理程序不引发异常，则现在会重试该函数（参见 PEP 475 中的原理）。
- signal.sigtimedwait(sigset, timeout)¶
与
sigwaitinfo()类似，但接受一个额外的 timeout 参数，指定超时时间。如果 timeout 指定为0，则执行轮询。如果发生超时，则返回None。可用性: Unix。
有关更多信息，请参阅手册页 sigtimedwait(2)。
另请参阅
pause()、sigwait()和sigwaitinfo()。在 3.3 版本加入。
版本 3.5 中已更改: 如果函数被不在 sigset 中的信号中断且信号处理程序不引发异常，则现在会使用重新计算的 timeout 重试该函数（参见 PEP 475 中的原理）。
示例¶
这是一个最小的示例程序。它使用 alarm() 函数限制等待打开文件的时间；这在文件用于串行设备且可能未打开的情况下很有用，否则通常会导致 os.open() 无限期挂起。解决方案是在打开文件之前设置一个 5 秒的警报；如果操作时间过长，将发送警报信号，处理程序将引发异常。
import signal, os
def handler(signum, frame):
signame = signal.Signals(signum).name
print(f'Signal handler called with signal {signame} ({signum})')
raise OSError("Couldn't open device!")
# Set the signal handler and a 5-second alarm
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)
# This open() may hang indefinitely
fd = os.open('/dev/ttyS0', os.O_RDWR)
signal.alarm(0) # Disable the alarm
关于 SIGPIPE 的注意事项¶
将程序输出通过管道传递给 head(1) 等工具时，当其标准输出的接收者提前关闭时，将向程序发送 SIGPIPE 信号。这会导致 BrokenPipeError: [Errno 32] Broken pipe 这样的异常。要处理这种情况，请将入口点包装起来以捕获此异常，如下所示
import os
import sys
def main():
try:
# simulate large output (your code replaces this loop)
for x in range(10000):
print("y")
# flush output here to force SIGPIPE to be triggered
# while inside this try block.
sys.stdout.flush()
except BrokenPipeError:
# Python flushes standard streams on exit; redirect remaining output
# to devnull to avoid another BrokenPipeError at shutdown
devnull = os.open(os.devnull, os.O_WRONLY)
os.dup2(devnull, sys.stdout.fileno())
sys.exit(1) # Python exits with error code 1 on EPIPE
if __name__ == '__main__':
main()
不要将 SIGPIPE 的处理方式设置为 SIG_DFL，以避免 BrokenPipeError。这样做会导致程序在任何套接字连接在程序仍在写入时中断时意外退出。
关于信号处理程序和异常的注意事项¶
如果信号处理程序引发异常，则该异常将传播到主线程，并可能在任何 字节码 指令之后引发。最值得注意的是，KeyboardInterrupt 可能会在执行的任何时候出现。大多数 Python 代码，包括标准库，都无法对此进行健壮处理，因此 KeyboardInterrupt（或任何其他由信号处理程序导致的异常）在极少数情况下可能会使程序处于意外状态。
为了说明这个问题，请考虑以下代码
class SpamContext:
def __init__(self):
self.lock = threading.Lock()
def __enter__(self):
# If KeyboardInterrupt occurs here, everything is fine
self.lock.acquire()
# If KeyboardInterrupt occurs here, __exit__ will not be called
...
# KeyboardInterrupt could occur just before the function returns
def __exit__(self, exc_type, exc_val, exc_tb):
...
self.lock.release()
对于许多程序，尤其是那些只想在 KeyboardInterrupt 时退出的程序，这不是问题，但复杂或需要高可靠性的应用程序应避免从信号处理程序引发异常。它们还应避免捕获 KeyboardInterrupt 作为优雅关闭的手段。相反，它们应安装自己的 SIGINT 处理程序。下面是一个避免 KeyboardInterrupt 的 HTTP 服务器示例
import signal
import socket
from selectors import DefaultSelector, EVENT_READ
from http.server import HTTPServer, SimpleHTTPRequestHandler
interrupt_read, interrupt_write = socket.socketpair()
def handler(signum, frame):
print('Signal handler called with signal', signum)
interrupt_write.send(b'\0')
signal.signal(signal.SIGINT, handler)
def serve_forever(httpd):
sel = DefaultSelector()
sel.register(interrupt_read, EVENT_READ)
sel.register(httpd, EVENT_READ)
while True:
for key, _ in sel.select():
if key.fileobj == interrupt_read:
interrupt_read.recv(1)
return
if key.fileobj == httpd:
httpd.handle_request()
print("Serving on port 8000")
httpd = HTTPServer(('', 8000), SimpleHTTPRequestHandler)
serve_forever(httpd)
print("Shutdown...")