sigwait未按预期工作：SIGALRM触发后为何不返回？

仅适用于Unix系统 #

from threading import Thread
from signal import signal, alarm, sigwait, SIGALRM


class Check(Thread):
    def __init__(self):
        super().__init__()
        signal(SIGALRM, Check.handler)

    @staticmethod
    def handler(*_):
        print("Hello")

    def run(self):
        for _ in range(5):
            alarm(1)
            print("Waiting...")
            sigwait((SIGALRM,))
            print("done")


if __name__ == "__main__":
    (check := Check()).start()
    check.join()

预期行为 #

期望以下输出重复5次：

Waiting...
Hello
done

实际问题 #

实际运行中done从未打印，程序一直阻塞在*sigwait()上。
"Hello"正常打印，说明信号处理函数已被调用。
既然SIGALRM已通过alarm()触发并处理，为何sigwait()*不返回？

运行平台 #

macOS 14.2.1
Python 3.12.1

补充尝试 #

添加pthread_sigmask调用后的代码如下，但行为与原代码完全一致：

from threading import Thread
from signal import signal, alarm, sigwait, pthread_sigmask, SIGALRM, SIG_BLOCK


class Check(Thread):
    def __init__(self):
        super().__init__()
        signal(SIGALRM, self.handler)

    def handler(self, *_):
        print("Hello")

    def run(self):
        mask = SIGALRM,
        pthread_sigmask(SIG_BLOCK, mask)
        
        for _ in range(5):
            alarm(1)
            print("Waiting...")
            sigwait(mask)
            print("done")


if __name__ == "__main__":
    (check := Check()).start()
    check.join()

问题核心在于信号处理机制的互斥性：

1. 当你同时用signal()注册自定义信号处理函数，又调用sigwait()等待同一信号时，二者是互斥关系：SIGALRM触发后会优先被注册的handler捕获处理，信号被消耗后不会再传递给sigwait()，导致sigwait()一直阻塞等待不存在的信号。

2. 补充的pthread_sigmask调用无法解决问题，是因为Python的信号处理默认逻辑：所有信号会被发送到主线程处理，即使子线程阻塞了信号，SIGALRM还是会被主线程的handler捕获，子线程的sigwait()拿不到信号。

解决方法 #

要让sigwait()捕获到信号，需要取消注册信号处理函数，同时确保信号被当前线程阻塞（避免被其他线程截获）：

from threading import Thread
from signal import alarm, sigwait, pthread_sigmask, SIGALRM, SIG_BLOCK, SIG_DFL


class Check(Thread):
    def __init__(self):
        super().__init__()
        # 将SIGALRM处理逻辑重置为默认（不主动处理）
        signal(SIGALRM, SIG_DFL)

    def run(self):
        mask = (SIGALRM,)
        # 阻塞SIGALRM，确保信号只会被当前线程的sigwait捕获
        pthread_sigmask(SIG_BLOCK, mask)
        
        for _ in range(5):
            alarm(1)
            print("Waiting...")
            sigwait(mask)
            print("done")


if __name__ == "__main__":
    (check := Check()).start()
    check.join()

核心原则：同一信号不能同时被自定义handler处理和sigwait捕获，二者只能选其一。

内容的提问来源于stack exchange，提问作者jackal