gh-71052: Use `raise_signal` in `ThreadSignals.test_signals` (#116423)

Use `raise_signal` rather than `kill` in `ThreadSignals.test_signals`
This commit is contained in:
Malcolm Smith 2024-03-11 20:39:17 +00:00 committed by GitHub
parent 1d0d49a7e8
commit 34920f3691
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 7 additions and 18 deletions

View File

@ -32,39 +32,28 @@ def handle_signals(sig,frame):
# a function that will be spawned as a separate thread. # a function that will be spawned as a separate thread.
def send_signals(): def send_signals():
os.kill(process_pid, signal.SIGUSR1) # We use `raise_signal` rather than `kill` because:
os.kill(process_pid, signal.SIGUSR2) # * It verifies that a signal delivered to a background thread still has
# its Python-level handler called on the main thread.
# * It ensures the signal is handled before the thread exits.
signal.raise_signal(signal.SIGUSR1)
signal.raise_signal(signal.SIGUSR2)
signalled_all.release() signalled_all.release()
@threading_helper.requires_working_threading() @threading_helper.requires_working_threading()
@unittest.skipUnless(hasattr(signal, "alarm"), "test requires signal.alarm")
class ThreadSignals(unittest.TestCase): class ThreadSignals(unittest.TestCase):
def test_signals(self): def test_signals(self):
with threading_helper.wait_threads_exit(): with threading_helper.wait_threads_exit():
# Test signal handling semantics of threads. # Test signal handling semantics of threads.
# We spawn a thread, have the thread send two signals, and # We spawn a thread, have the thread send itself two signals, and
# wait for it to finish. Check that we got both signals # wait for it to finish. Check that we got both signals
# and that they were run by the main thread. # and that they were run by the main thread.
signalled_all.acquire() signalled_all.acquire()
self.spawnSignallingThread() self.spawnSignallingThread()
signalled_all.acquire() signalled_all.acquire()
# the signals that we asked the kernel to send
# will come back, but we don't know when.
# (it might even be after the thread exits
# and might be out of order.) If we haven't seen
# the signals yet, send yet another signal and
# wait for it return.
if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
try:
signal.alarm(1)
signal.pause()
finally:
signal.alarm(0)
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1) self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'], self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
thread.get_ident()) thread.get_ident())