code: def test_interrupted_timed_acquire(self):
      # 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.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():
              # Send the signals from the non-main thread, since the main thread
              # is the only one that can process signals.
              thread.start_new_thread(send_signals, ())
              timed_acquire()
              # Wait for thread to finish
              done.acquire()
              # This allows for some timing and scheduling imprecision
              self.assertLess(self.end - self.start, 2.0)
              self.assertGreater(self.end - self.start, 0.3)
              # 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.
              self.assertGreater(self.sigs_recvd, 0)
      finally:
          signal.signal(signal.SIGUSR1, old_handler)
  
imports: import unittest
  import signal
  import os
  import sys
  from test.support import threading_helper
  import _thread
  import time
label_desc: Use image node_id 9 for calling the ThreadSignals obj's underlying member method code with example usage: obj.test_interrupted_timed_acquire() without return types