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