pyunit testing that threads honor our signal semantics import unittest import signal import os import sys from test support import threadinghelper import thread as thread import time if sys platform 3 win raise unittest skiptestcan t test signal on s sys platform processpid os getpid signalledallthread allocatelock usingpthreadcond sys threadinfo name pthread and sys threadinfo lock mutexcond def registersignalsforusr1 forusr2 foralrm usr1 signal signalsignal sigusr1 forusr1 usr2 signal signalsignal sigusr2 forusr2 alrm signal signalsignal sigalrm foralrm return usr1 usr2 alrm the signal handler just note that the signal occurred and from who def handlesignalssig frame signalblackboardsig tripped 1 signalblackboardsig trippedby thread getident a function that will be spawned as a separate thread def sendsignals os killprocesspid signal sigusr1 os killprocesspid signal sigusr2 signalledall release threadinghelper requiresworkingthreading unittest skipunlesshasattrsignal alarm test requires signal alarm class threadsignalsunittest testcase def testsignalsself with threadinghelper waitthreadsexit test signal handling semantics of threads we spawn a thread have the thread send two signals and wait for it to finish check that we got both signals and that they were run by the main thread signalledall acquire self spawnsignallingthread signalledall 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 signalblackboardsignal sigusr1 tripped 0 or signalblackboardsignal sigusr2 tripped 0 try signal alarm1 signal pause finally signal alarm0 self assertequal signalblackboardsignal sigusr1 tripped 1 self assertequal signalblackboardsignal sigusr1 trippedby thread getident self assertequal signalblackboardsignal sigusr2 tripped 1 self assertequal signalblackboardsignal sigusr2 trippedby thread getident signalledall release def spawnsignallingthreadself thread startnewthreadsendsignals def alarminterruptself sig frame raise keyboardinterrupt unittest skipifusingpthreadcond posix condition variables cannot be interrupted unittest skipifsys platform startswith linux and not sys threadinfo version issue 34004 musl does not allow interruption of locks by signals issue 20564 semtimedwait cannot be interrupted on openbsd unittest skipifsys platform startswith openbsd lock cannot be interrupted on openbsd def testlockacquireinterruptionself mimic receiving a sigint keyboardinterrupt with sigalrm while stuck in a deadlock xxx this test can fail when the legacy nonsemaphore implementation of locks is used in threadpthread h see issue 11223 oldalrm signal signalsignal sigalrm self alarminterrupt try lock thread allocatelock lock acquire signal alarm1 t1 time monotonic self assertraiseskeyboardinterrupt lock acquire timeout5 dt time monotonic t1 checking that keyboardinterrupt was raised is not sufficient we want to assert that lock acquire was interrupted because of the signal not that the signal handler was called immediately after timeout return of lock acquire which can fool assertraises self assertlessdt 3 0 finally signal alarm0 signal signalsignal sigalrm oldalrm unittest skipifusingpthreadcond posix condition variables cannot be interrupted unittest skipifsys platform startswith linux and not sys threadinfo version issue 34004 musl does not allow interruption of locks by signals issue 20564 semtimedwait cannot be interrupted on openbsd unittest skipifsys platform startswith openbsd lock cannot be interrupted on openbsd def testrlockacquireinterruptionself mimic receiving a sigint keyboardinterrupt with sigalrm while stuck in a deadlock xxx this test can fail when the legacy nonsemaphore implementation of locks is used in threadpthread h see issue 11223 oldalrm signal signalsignal sigalrm self alarminterrupt try rlock thread rlock for reentrant locks the initial acquisition must be in another thread def otherthread rlock acquire with threadinghelper waitthreadsexit thread startnewthreadotherthread wait until we can t acquire it without blocking while rlock acquireblockingfalse rlock release time sleep0 01 signal alarm1 t1 time monotonic self assertraiseskeyboardinterrupt rlock acquire timeout5 dt time monotonic t1 see rationale above in testlockacquireinterruption self assertlessdt 3 0 finally signal alarm0 signal signalsignal sigalrm oldalrm def acquireretriesonintrself lock self sigrecvd false def myhandlersignal frame self sigrecvd true oldhandler signal signalsignal sigusr1 myhandler try def otherthread acquire the lock in a nonmain thread so this test works for rlocks lock acquire wait until the main thread is blocked in the lock acquire and then wake it up with this time sleep0 5 os killprocesspid signal sigusr1 let the main thread take the interrupt handle it and retry the lock acquisition then we ll let it run time sleep0 5 lock release with threadinghelper waitthreadsexit thread startnewthreadotherthread wait until we can t acquire it without blocking while lock acquireblockingfalse lock release time sleep0 01 result lock acquire block while we receive a signal self asserttrueself sigrecvd self asserttrueresult finally signal signalsignal sigusr1 oldhandler def testlockacquireretriesonintrself self acquireretriesonintrthread allocatelock def testrlockacquireretriesonintrself self acquireretriesonintrthread rlock def testinterruptedtimedacquireself test to make sure we recompute lock acquisition timeouts when we receive a signal check this by repeatedly interrupting a lock acquire in the main thread and make sure that the lock acquire times out after the right amount of time note this test only behaves as expected if c signals get delivered to the main thread otherwise lock acquire itself doesn t get interrupted and the test trivially succeeds self start none self end none self sigsrecvd 0 done thread allocatelock done acquire lock thread allocatelock lock acquire def myhandlersignum frame self sigsrecvd 1 oldhandler signal signalsignal sigusr1 myhandler try def timedacquire self start time monotonic lock acquiretimeout0 5 self end time monotonic def sendsignals for in range40 time sleep0 02 os killprocesspid signal sigusr1 done release with threadinghelper waitthreadsexit send the signals from the nonmain thread since the main thread is the only one that can process signals thread startnewthreadsendsignals timedacquire wait for thread to finish done acquire this allows for some timing and scheduling imprecision self assertlessself end self start 2 0 self assertgreaterself end self start 0 3 if the signal is received several times before pyerrchecksignals is called the handler will get called less than 40 times just check it s been called at least once self assertgreaterself sigsrecvd 0 finally signal signalsignal sigusr1 oldhandler def setupmodule global signalblackboard signalblackboard signal sigusr1 tripped 0 trippedby 0 signal sigusr2 tripped 0 trippedby 0 signal sigalrm tripped 0 trippedby 0 oldsigs registersignalshandlesignals handlesignals handlesignals unittest addmodulecleanupregistersignals oldsigs if name main unittest main the signal handler just note that the signal occurred and from who a function that will be spawned as a separate thread test signal handling semantics of threads we spawn a thread have the thread send two signals and wait for it to finish check that we got both signals and that they were run by the main thread 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 issue 20564 sem_timedwait cannot be interrupted on openbsd mimic receiving a sigint keyboardinterrupt with sigalrm while stuck in a deadlock xxx this test can fail when the legacy non semaphore implementation of locks is used in thread_pthread h see issue 11223 checking that keyboardinterrupt was raised is not sufficient we want to assert that lock acquire was interrupted because of the signal not that the signal handler was called immediately after timeout return of lock acquire which can fool assertraises issue 20564 sem_timedwait cannot be interrupted on openbsd mimic receiving a sigint keyboardinterrupt with sigalrm while stuck in a deadlock xxx this test can fail when the legacy non semaphore implementation of locks is used in thread_pthread h see issue 11223 for reentrant locks the initial acquisition must be in another thread wait until we can t acquire it without blocking see rationale above in test_lock_acquire_interruption acquire the lock in a non main thread so this test works for rlocks wait until the main thread is blocked in the lock acquire and then wake it up with this let the main thread take the interrupt handle it and retry the lock acquisition then we ll let it run wait until we can t acquire it without blocking block while we receive a signal test to make sure we recompute lock acquisition timeouts when we receive a signal check this by repeatedly interrupting a lock acquire in the main thread and make sure that the lock acquire times out after the right amount of time note this test only behaves as expected if c signals get delivered to the main thread otherwise lock acquire itself doesn t get interrupted and the test trivially succeeds send the signals from the non main thread since the main thread is the only one that can process signals wait for thread to finish this allows for some timing and scheduling imprecision if the signal is received several times before pyerr_checksignals is called the handler will get called less than 40 times just check it s been called at least once
import unittest
import signal
import os
import sys
from test.support import threading_helper
import _thread as thread
import time

if (sys.platform[:3] == 'win'):
    raise unittest.SkipTest("Can't test signal on %s" % sys.platform)

process_pid = os.getpid()
signalled_all=thread.allocate_lock()

USING_PTHREAD_COND = (sys.thread_info.name == 'pthread'
                      and sys.thread_info.lock == 'mutex+cond')

def registerSignals(for_usr1, for_usr2, for_alrm):
    usr1 = signal.signal(signal.SIGUSR1, for_usr1)
    usr2 = signal.signal(signal.SIGUSR2, for_usr2)
    alrm = signal.signal(signal.SIGALRM, for_alrm)
    return usr1, usr2, alrm




def handle_signals(sig,frame):
    signal_blackboard[sig]['tripped'] += 1
    signal_blackboard[sig]['tripped_by'] = thread.get_ident()


def send_signals():
    os.kill(process_pid, signal.SIGUSR1)
    os.kill(process_pid, signal.SIGUSR2)
    signalled_all.release()


@threading_helper.requires_working_threading()
@unittest.skipUnless(hasattr(signal, "alarm"), "test requires signal.alarm")
class ThreadSignals(unittest.TestCase):

    def test_signals(self):
        with threading_helper.wait_threads_exit():
            
            
            
            
            signalled_all.acquire()
            self.spawnSignallingThread()
            signalled_all.acquire()

        
        
        
        
        
        
        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_by'],
                           thread.get_ident())
        self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)
        self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],
                           thread.get_ident())
        signalled_all.release()

    def spawnSignallingThread(self):
        thread.start_new_thread(send_signals, ())

    def alarm_interrupt(self, sig, frame):
        raise KeyboardInterrupt

    @unittest.skipIf(USING_PTHREAD_COND,
                     'POSIX condition variables cannot be interrupted')
    @unittest.skipIf(sys.platform.startswith('linux') and
                     not sys.thread_info.version,
                     'Issue 34004: musl does not allow interruption of locks '
                     'by signals.')
    
    @unittest.skipIf(sys.platform.startswith('openbsd'),
                     'lock cannot be interrupted on OpenBSD')
    def test_lock_acquire_interruption(self):
        
        
        
        
        oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
        try:
            lock = thread.allocate_lock()
            lock.acquire()
            signal.alarm(1)
            t1 = time.monotonic()
            self.assertRaises(KeyboardInterrupt, lock.acquire, timeout=5)
            dt = time.monotonic() - t1
            
            
            
            
            self.assertLess(dt, 3.0)
        finally:
            signal.alarm(0)
            signal.signal(signal.SIGALRM, oldalrm)

    @unittest.skipIf(USING_PTHREAD_COND,
                     'POSIX condition variables cannot be interrupted')
    @unittest.skipIf(sys.platform.startswith('linux') and
                     not sys.thread_info.version,
                     'Issue 34004: musl does not allow interruption of locks '
                     'by signals.')
    
    @unittest.skipIf(sys.platform.startswith('openbsd'),
                     'lock cannot be interrupted on OpenBSD')
    def test_rlock_acquire_interruption(self):
        
        
        
        
        oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
        try:
            rlock = thread.RLock()
            
            
            def other_thread():
                rlock.acquire()

            with threading_helper.wait_threads_exit():
                thread.start_new_thread(other_thread, ())
                
                while rlock.acquire(blocking=False):
                    rlock.release()
                    time.sleep(0.01)
                signal.alarm(1)
                t1 = time.monotonic()
                self.assertRaises(KeyboardInterrupt, rlock.acquire, timeout=5)
                dt = time.monotonic() - t1
                
                self.assertLess(dt, 3.0)
        finally:
            signal.alarm(0)
            signal.signal(signal.SIGALRM, oldalrm)

    def acquire_retries_on_intr(self, lock):
        self.sig_recvd = False
        def my_handler(signal, frame):
            self.sig_recvd = True

        old_handler = signal.signal(signal.SIGUSR1, my_handler)
        try:
            def other_thread():
                
                
                lock.acquire()
                
                
                time.sleep(0.5)
                os.kill(process_pid, signal.SIGUSR1)
                
                
                time.sleep(0.5)
                lock.release()

            with threading_helper.wait_threads_exit():
                thread.start_new_thread(other_thread, ())
                
                while lock.acquire(blocking=False):
                    lock.release()
                    time.sleep(0.01)
                result = lock.acquire()  
                self.assertTrue(self.sig_recvd)
                self.assertTrue(result)
        finally:
            signal.signal(signal.SIGUSR1, old_handler)

    def test_lock_acquire_retries_on_intr(self):
        self.acquire_retries_on_intr(thread.allocate_lock())

    def test_rlock_acquire_retries_on_intr(self):
        self.acquire_retries_on_intr(thread.RLock())

    def test_interrupted_timed_acquire(self):
        
        
        
        
        
        
        
        self.start = None
        self.end = None
        self.sigs_recvd = 0
        done = thread.allocate_lock()
        done.acquire()
        lock = thread.allocate_lock()
        lock.acquire()
        def my_handler(signum, frame):
            self.sigs_recvd += 1
        old_handler = signal.signal(signal.SIGUSR1, my_handler)
        try:
            def timed_acquire():
                self.start = time.monotonic()
                lock.acquire(timeout=0.5)
                self.end = time.monotonic()
            def send_signals():
                for _ in range(40):
                    time.sleep(0.02)
                    os.kill(process_pid, signal.SIGUSR1)
                done.release()

            with threading_helper.wait_threads_exit():
                
                
                thread.start_new_thread(send_signals, ())
                timed_acquire()
                
                done.acquire()
                
                self.assertLess(self.end - self.start, 2.0)
                self.assertGreater(self.end - self.start, 0.3)
                
                
                
                self.assertGreater(self.sigs_recvd, 0)
        finally:
            signal.signal(signal.SIGUSR1, old_handler)


def setUpModule():
    global signal_blackboard

    signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 },
                          signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 },
                          signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } }

    oldsigs = registerSignals(handle_signals, handle_signals, handle_signals)
    unittest.addModuleCleanup(registerSignals, *oldsigs)


if __name__ == '__main__':
    unittest.main()