Search code examples
pythonsignals

sigwait not behaving as expected


Unix only

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()

Expected behaviour:

I expect the following output to be repeated 5 times:

Waiting...
Hello
done

However, "done" is never printed because the runtime is "blocking" on sigwait()

"Hello" is printed. Therefore I know that the signal handler has been invoked.

If SIGALARM has been signalled (via alarm()) and handled, why does sigwait() not return?

Platform:

macOS 14.2.1
Python 3.12.1

EDIT to include call to 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()

Behaviour is identical to original code


Solution

  • This answer helped clear things up some.

    Since signal() can, for sanity reasons, only be called from the main thread in Python ("The effects of signal() in a multithreaded process are unspecified", says the Linux manpage for signal()), in order to receive a signal in another thread you'll have to use synchronous signals (unless you arrange for your main-thread-configured signal to e.g. set a threading.Event).

    Since the default action for SIGALRM is to terminate the process, you'll have to arrange for it not to be received on the main thread at all in order to receive it on another thread, by way of pthread_sigmask(SIG_IGN, ...) there, and then set it to blocking mode on the signal receiver thread. (Other threads spawned from that main thread will inherit the sigmask otherwise.)

    This uses a threading.Event() to let the main thread know the receiver received a thing.

    import signal
    import threading
    
    mask = (signal.SIGALRM,)
    ev = threading.Event()
    
    
    class SignalReceiver(threading.Thread):
        def run(self):
            signal.pthread_sigmask(signal.SIG_BLOCK, mask)
            while True:
                signal.sigwait(mask)
                print("Got signal")
                ev.set()
    
    
    if __name__ == "__main__":
        SignalReceiver(daemon=True).start()
        signal.pthread_sigmask(signal.SIG_IGN, mask)
    
        for _ in range(3):
            signal.alarm(1)
            print("Waiting...")
            ev.wait()
            ev.clear()