Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes
test.test_threadsignals.ThreadSignals Class Reference

List of all members.

Public Member Functions

def test_signals
def spawnSignallingThread
def alarm_interrupt
def test_lock_acquire_interruption
def test_rlock_acquire_interruption
def acquire_retries_on_intr
def test_lock_acquire_retries_on_intr
def test_rlock_acquire_retries_on_intr
def test_interrupted_timed_acquire

Public Attributes

 sig_recvd
 start
 end
 sigs_recvd

Detailed Description

Definition at line 41 of file test_threadsignals.py.


Member Function Documentation

Definition at line 128 of file test_threadsignals.py.

00128 
00129     def acquire_retries_on_intr(self, lock):
00130         self.sig_recvd = False
00131         def my_handler(signal, frame):
00132             self.sig_recvd = True
00133         old_handler = signal.signal(signal.SIGUSR1, my_handler)
00134         try:
00135             def other_thread():
00136                 # Acquire the lock in a non-main thread, so this test works for
00137                 # RLocks.
00138                 lock.acquire()
00139                 # Wait until the main thread is blocked in the lock acquire, and
00140                 # then wake it up with this.
00141                 time.sleep(0.5)
00142                 os.kill(process_pid, signal.SIGUSR1)
00143                 # Let the main thread take the interrupt, handle it, and retry
00144                 # the lock acquisition.  Then we'll let it run.
00145                 time.sleep(0.5)
00146                 lock.release()
00147             thread.start_new_thread(other_thread, ())
00148             # Wait until we can't acquire it without blocking...
00149             while lock.acquire(blocking=False):
00150                 lock.release()
00151                 time.sleep(0.01)
00152             result = lock.acquire()  # Block while we receive a signal.
00153             self.assertTrue(self.sig_recvd)
00154             self.assertTrue(result)
00155         finally:
00156             signal.signal(signal.SIGUSR1, old_handler)

Here is the caller graph for this function:

def test.test_threadsignals.ThreadSignals.alarm_interrupt (   self,
  sig,
  frame 
)

Definition at line 74 of file test_threadsignals.py.

00074 
00075     def alarm_interrupt(self, sig, frame):
00076         raise KeyboardInterrupt

Here is the caller graph for this function:

Definition at line 71 of file test_threadsignals.py.

00071 
00072     def spawnSignallingThread(self):
00073         thread.start_new_thread(send_signals, ())

Here is the caller graph for this function:

Definition at line 163 of file test_threadsignals.py.

00163 
00164     def test_interrupted_timed_acquire(self):
00165         # Test to make sure we recompute lock acquisition timeouts when we
00166         # receive a signal.  Check this by repeatedly interrupting a lock
00167         # acquire in the main thread, and make sure that the lock acquire times
00168         # out after the right amount of time.
00169         # NOTE: this test only behaves as expected if C signals get delivered
00170         # to the main thread.  Otherwise lock.acquire() itself doesn't get
00171         # interrupted and the test trivially succeeds.
00172         self.start = None
00173         self.end = None
00174         self.sigs_recvd = 0
00175         done = thread.allocate_lock()
00176         done.acquire()
00177         lock = thread.allocate_lock()
00178         lock.acquire()
00179         def my_handler(signum, frame):
00180             self.sigs_recvd += 1
00181         old_handler = signal.signal(signal.SIGUSR1, my_handler)
00182         try:
00183             def timed_acquire():
00184                 self.start = time.time()
00185                 lock.acquire(timeout=0.5)
00186                 self.end = time.time()
00187             def send_signals():
00188                 for _ in range(40):
00189                     time.sleep(0.02)
00190                     os.kill(process_pid, signal.SIGUSR1)
00191                 done.release()
00192 
00193             # Send the signals from the non-main thread, since the main thread
00194             # is the only one that can process signals.
00195             thread.start_new_thread(send_signals, ())
00196             timed_acquire()
00197             # Wait for thread to finish
00198             done.acquire()
00199             # This allows for some timing and scheduling imprecision
00200             self.assertLess(self.end - self.start, 2.0)
00201             self.assertGreater(self.end - self.start, 0.3)
00202             # If the signal is received several times before PyErr_CheckSignals()
00203             # is called, the handler will get called less than 40 times. Just
00204             # check it's been called at least once.
00205             self.assertGreater(self.sigs_recvd, 0)
00206         finally:
00207             signal.signal(signal.SIGUSR1, old_handler)
00208 

Definition at line 79 of file test_threadsignals.py.

00079 
00080     def test_lock_acquire_interruption(self):
00081         # Mimic receiving a SIGINT (KeyboardInterrupt) with SIGALRM while stuck
00082         # in a deadlock.
00083         # XXX this test can fail when the legacy (non-semaphore) implementation
00084         # of locks is used in thread_pthread.h, see issue #11223.
00085         oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
00086         try:
00087             lock = thread.allocate_lock()
00088             lock.acquire()
00089             signal.alarm(1)
00090             t1 = time.time()
00091             self.assertRaises(KeyboardInterrupt, lock.acquire, timeout=5)
00092             dt = time.time() - t1
00093             # Checking that KeyboardInterrupt was raised is not sufficient.
00094             # We want to assert that lock.acquire() was interrupted because
00095             # of the signal, not that the signal handler was called immediately
00096             # after timeout return of lock.acquire() (which can fool assertRaises).
00097             self.assertLess(dt, 3.0)
00098         finally:
00099             signal.signal(signal.SIGALRM, oldalrm)

Here is the call graph for this function:

Definition at line 157 of file test_threadsignals.py.

00157 
00158     def test_lock_acquire_retries_on_intr(self):
00159         self.acquire_retries_on_intr(thread.allocate_lock())

Here is the call graph for this function:

Definition at line 102 of file test_threadsignals.py.

00102 
00103     def test_rlock_acquire_interruption(self):
00104         # Mimic receiving a SIGINT (KeyboardInterrupt) with SIGALRM while stuck
00105         # in a deadlock.
00106         # XXX this test can fail when the legacy (non-semaphore) implementation
00107         # of locks is used in thread_pthread.h, see issue #11223.
00108         oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
00109         try:
00110             rlock = thread.RLock()
00111             # For reentrant locks, the initial acquisition must be in another
00112             # thread.
00113             def other_thread():
00114                 rlock.acquire()
00115             thread.start_new_thread(other_thread, ())
00116             # Wait until we can't acquire it without blocking...
00117             while rlock.acquire(blocking=False):
00118                 rlock.release()
00119                 time.sleep(0.01)
00120             signal.alarm(1)
00121             t1 = time.time()
00122             self.assertRaises(KeyboardInterrupt, rlock.acquire, timeout=5)
00123             dt = time.time() - t1
00124             # See rationale above in test_lock_acquire_interruption
00125             self.assertLess(dt, 3.0)
00126         finally:
00127             signal.signal(signal.SIGALRM, oldalrm)

Here is the call graph for this function:

Definition at line 160 of file test_threadsignals.py.

00160 
00161     def test_rlock_acquire_retries_on_intr(self):
00162         self.acquire_retries_on_intr(thread.RLock())

Here is the call graph for this function:

Definition at line 43 of file test_threadsignals.py.

00043 
00044     def test_signals(self):
00045         # Test signal handling semantics of threads.
00046         # We spawn a thread, have the thread send two signals, and
00047         # wait for it to finish. Check that we got both signals
00048         # and that they were run by the main thread.
00049         signalled_all.acquire()
00050         self.spawnSignallingThread()
00051         signalled_all.acquire()
00052         # the signals that we asked the kernel to send
00053         # will come back, but we don't know when.
00054         # (it might even be after the thread exits
00055         # and might be out of order.)  If we haven't seen
00056         # the signals yet, send yet another signal and
00057         # wait for it return.
00058         if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
00059            or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
00060             signal.alarm(1)
00061             signal.pause()
00062             signal.alarm(0)
00063 
00064         self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
00065         self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
00066                            thread.get_ident())
00067         self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)
00068         self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],
00069                            thread.get_ident())
00070         signalled_all.release()

Here is the call graph for this function:


Member Data Documentation

Definition at line 172 of file test_threadsignals.py.

Definition at line 129 of file test_threadsignals.py.

Definition at line 173 of file test_threadsignals.py.

Definition at line 171 of file test_threadsignals.py.


The documentation for this class was generated from the following file: