我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用signal.SIGALRM。
def __init__(self, child, timelimit, greedy=False, use_alarm=True): """ :param child: the collector to wrap. :param timelimit: the maximum amount of time (in seconds) to allow for searching. If the search takes longer than this, it will raise a ``TimeLimit`` exception. :param greedy: if ``True``, the collector will finish adding the most recent hit before raising the ``TimeLimit`` exception. :param use_alarm: if ``True`` (the default), the collector will try to use signal.SIGALRM (on UNIX). """ self.child = child self.timelimit = timelimit self.greedy = greedy if use_alarm: import signal self.use_alarm = use_alarm and hasattr(signal, "SIGALRM") else: self.use_alarm = False self.timer = None self.timedout = False
def test_select_interrupt(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = self.make_socketpair() orig_alrm_handler = signal.signal(signal.SIGALRM, lambda *args: None) self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler) self.addCleanup(signal.alarm, 0) signal.alarm(1) s.register(rd, selectors.EVENT_READ) t = time() self.assertFalse(s.select(2)) self.assertLess(time() - t, 2.5)
def __init__(self, master=False): # Setup signal fd, this allows signal to behave correctly if os.name == 'posix': self.signal_pipe_r, self.signal_pipe_w = os.pipe() self._set_nonblock(self.signal_pipe_r) self._set_nonblock(self.signal_pipe_w) signal.set_wakeup_fd(self.signal_pipe_w) self._signals_received = collections.deque() signal.signal(signal.SIGINT, signal.SIG_DFL) if os.name == 'posix': signal.signal(signal.SIGCHLD, signal.SIG_DFL) signal.signal(signal.SIGTERM, self._signal_catcher) signal.signal(signal.SIGALRM, self._signal_catcher) signal.signal(signal.SIGHUP, self._signal_catcher) else: # currently a noop on window... signal.signal(signal.SIGTERM, self._signal_catcher) # FIXME(sileht): should allow to catch signal CTRL_BREAK_EVENT, # but we to create the child process with CREATE_NEW_PROCESS_GROUP # to make this work, so current this is a noop for later fix signal.signal(signal.SIGBREAK, self._signal_catcher)
def test_communicate_eintr(self): # Issue #12493: communicate() should handle EINTR def handler(signum, frame): pass old_handler = signal.signal(signal.SIGALRM, handler) self.addCleanup(signal.signal, signal.SIGALRM, old_handler) # the process is running for 2 seconds args = [sys.executable, "-c", 'import time; time.sleep(2)'] for stream in ('stdout', 'stderr'): kw = {stream: subprocess.PIPE} with subprocess.Popen(args, **kw) as process: signal.alarm(1) # communicate() will be interrupted by SIGALRM process.communicate() # context manager
def test_lock_acquire_interruption(self): # 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. oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt) try: lock = thread.allocate_lock() lock.acquire() signal.alarm(1) t1 = time.time() self.assertRaises(KeyboardInterrupt, lock.acquire, timeout=5) dt = time.time() - 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.assertLess(dt, 3.0) finally: signal.signal(signal.SIGALRM, oldalrm)
def test_rlock_acquire_interruption(self): # 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. oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt) try: rlock = thread.RLock() # For reentrant locks, the initial acquisition must be in another # thread. def other_thread(): rlock.acquire() thread.start_new_thread(other_thread, ()) # Wait until we can't acquire it without blocking... while rlock.acquire(blocking=False): rlock.release() time.sleep(0.01) signal.alarm(1) t1 = time.time() self.assertRaises(KeyboardInterrupt, rlock.acquire, timeout=5) dt = time.time() - t1 # See rationale above in test_lock_acquire_interruption self.assertLess(dt, 3.0) finally: signal.signal(signal.SIGALRM, oldalrm)
def test_wakeup_fd_early(self): self.check_wakeup("""def test(): import select import time TIMEOUT_FULL = 10 TIMEOUT_HALF = 5 signal.alarm(1) before_time = time.time() # We attempt to get a signal during the sleep, # before select is called time.sleep(TIMEOUT_FULL) mid_time = time.time() dt = mid_time - before_time if dt >= TIMEOUT_HALF: raise Exception("%s >= %s" % (dt, TIMEOUT_HALF)) select.select([read], [], [], TIMEOUT_FULL) after_time = time.time() dt = after_time - mid_time if dt >= TIMEOUT_HALF: raise Exception("%s >= %s" % (dt, TIMEOUT_HALF)) """, signal.SIGALRM)
def test_wakeup_fd_during(self): self.check_wakeup("""def test(): import select import time TIMEOUT_FULL = 10 TIMEOUT_HALF = 5 signal.alarm(1) before_time = time.time() # We attempt to get a signal during the select call try: select.select([read], [], [], TIMEOUT_FULL) except select.error: pass else: raise Exception("select.error not raised") after_time = time.time() dt = after_time - before_time if dt >= TIMEOUT_HALF: raise Exception("%s >= %s" % (dt, TIMEOUT_HALF)) """, signal.SIGALRM)
def test_sigwaitinfo_interrupted(self): self.wait_helper(signal.SIGUSR1, ''' def test(signum): import errno hndl_called = True def alarm_handler(signum, frame): hndl_called = False signal.signal(signal.SIGALRM, alarm_handler) signal.alarm(1) try: signal.sigwaitinfo([signal.SIGUSR1]) except OSError as e: if e.errno == errno.EINTR: if not hndl_called: raise Exception("SIGALRM handler not called") else: raise Exception("Expected EINTR to be raised by sigwaitinfo") else: raise Exception("Expected EINTR to be raised by sigwaitinfo") ''')
def check_reentrant_write(self, data, **fdopen_kwargs): def on_alarm(*args): # Will be called reentrantly from the same thread wio.write(data) 1/0 signal.signal(signal.SIGALRM, on_alarm) r, w = os.pipe() wio = self.io.open(w, **fdopen_kwargs) try: signal.alarm(1) # Either the reentrant call to wio.write() fails with RuntimeError, # or the signal handler raises ZeroDivisionError. with self.assertRaises((ZeroDivisionError, RuntimeError)) as cm: while 1: for i in range(100): wio.write(data) wio.flush() # Make sure the buffer doesn't fill up and block further writes os.read(r, len(data) * 100) exc = cm.exception if isinstance(exc, RuntimeError): self.assertTrue(str(exc).startswith("reentrant call"), str(exc)) finally: wio.close() os.close(r)
def check_interrupted_read_retry(self, decode, **fdopen_kwargs): """Check that a buffered read, when it gets interrupted (either returning a partial result or EINTR), properly invokes the signal handler and retries if the latter returned successfully.""" r, w = os.pipe() fdopen_kwargs["closefd"] = False def alarm_handler(sig, frame): os.write(w, b"bar") signal.signal(signal.SIGALRM, alarm_handler) try: rio = self.io.open(r, **fdopen_kwargs) os.write(w, b"foo") signal.alarm(1) # Expected behaviour: # - first raw read() returns partial b"foo" # - second raw read() returns EINTR # - third raw read() returns b"bar" self.assertEqual(decode(rio.read(6)), "foobar") finally: rio.close() os.close(w) os.close(r)
def connect_socket(url, timeout=TIMEOUT, **kwargs): """set up a websocket and return the socket connection object""" signal.signal( signal.SIGALRM, lambda s, f: timeout_handler(s, f, f'connecting ({timeout}s)') ) signal.alarm(timeout) try: sock = create_connection(url, **kwargs) signal.alarm(0) return sock except Exception: signal.alarm(0) print(f'[X] Failed to connect, is runserver running on {url}?') raise except Exception: signal.alarm(0) raise
def send_json(socket, data: dict, timeout=TIMEOUT): """ send a json-ified dictionary, throws an exception if it takes more than [timeout] seconds """ signal.signal( signal.SIGALRM, lambda s, f: timeout_handler(s, f, f'sending ({timeout}s)') ) signal.alarm(timeout) try: result = socket.send(json.dumps(data)) signal.alarm(0) return result except Exception: signal.alarm(0) raise
def recv_json(socket, timeout=TIMEOUT): """ block until a message is received [timeout] seconds, returns None if nothing is received """ signal.alarm(0) signal.signal( signal.SIGALRM, lambda s, f: timeout_handler(s, f, f'receiving ({timeout}s)') ) signal.alarm(timeout) try: result = json.loads(socket.recv()) signal.alarm(0) return result except TimeOutException: signal.alarm(0) return None except Exception: signal.alarm(0) raise
def recv_all_json(socket, timeout=TIMEOUT): """ block for [timeout] seconds, and return a list of all received messages in that period """ results = [] try: last_result = True while last_result: signal.alarm(0) signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: last_result = json.loads(socket.recv()) except TimeOutException: last_result = None signal.alarm(0) if last_result: results.append(last_result) return results except TimeOutException: return results
def test_wakeup_fd_during(self): self.check_wakeup("""def test(): import select import time TIMEOUT_FULL = 10 TIMEOUT_HALF = 5 signal.alarm(1) before_time = time.time() # We attempt to get a signal during the select call try: select.select([read], [], [], TIMEOUT_FULL) except OSError: pass else: raise Exception("OSError not raised") after_time = time.time() dt = after_time - before_time if dt >= TIMEOUT_HALF: raise Exception("%s >= %s" % (dt, TIMEOUT_HALF)) """, signal.SIGALRM)
def __init__(self, wait_interval=0.01): """Creates the ServiceManager object :param wait_interval: time between each new process spawn :type wait_interval: float """ if self._process_runner_already_created: raise RuntimeError("Only one instance of ProcessRunner per " "application is allowed") ServiceManager._process_runner_already_created = True self._wait_interval = wait_interval self._shutdown = threading.Event() self._running_services = collections.defaultdict(dict) self._services = [] self._forktimes = [] self._current_process = None # Try to create a session id if possible try: os.setsid() except OSError: pass self.readpipe, self.writepipe = os.pipe() signal.signal(signal.SIGTERM, self._clean_exit) signal.signal(signal.SIGINT, self._fast_exit) signal.signal(signal.SIGALRM, self._alarm_exit) signal.signal(signal.SIGHUP, self._reload_services)
def _fast_exit(self, signo, frame, reason='Caught SIGINT signal, instantaneous exiting'): signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGALRM, signal.SIG_IGN) LOG.info(reason) os.killpg(0, signal.SIGINT) os._exit(1)
def prepare(self, top_searcher, q, context): self.child.prepare(top_searcher, q, context) self.timedout = False if self.use_alarm: import signal signal.signal(signal.SIGALRM, self._was_signaled) # Start a timer thread. If the timer fires, it will call this object's # _timestop() method self.timer = threading.Timer(self.timelimit, self._timestop) self.timer.start()
def _timestop(self): # Called when the timer expires self.timer = None # Set an attribute that will be noticed in the collect_matches() loop self.timedout = True if self.use_alarm: import signal os.kill(os.getpid(), signal.SIGALRM)
def test_signal_handling_while_selecting(self): # Test with a signal actually arriving during a select() call. caught = 0 def my_handler(): nonlocal caught caught += 1 self.loop.stop() self.loop.add_signal_handler(signal.SIGALRM, my_handler) signal.setitimer(signal.ITIMER_REAL, 0.01, 0) # Send SIGALRM once. self.loop.run_forever() self.assertEqual(caught, 1)
def test_signal_handling_args(self): some_args = (42,) caught = 0 def my_handler(*args): nonlocal caught caught += 1 self.assertEqual(args, some_args) self.loop.add_signal_handler(signal.SIGALRM, my_handler, *some_args) signal.setitimer(signal.ITIMER_REAL, 0.1, 0) # Send SIGALRM once. self.loop.call_later(0.5, self.loop.stop) self.loop.run_forever() self.assertEqual(caught, 1)
def run(self): time.sleep(self.timeout) if not self.canceled: os.kill(os.getpid(), signal.SIGALRM)
def _begin_alarm_thread(self, timeout): if not hasattr(signal, "SIGALRM"): self.skipTest("Platform doesn't have signal.SIGALRM") self.addCleanup(self._cancel_alarm_thread) self.alarm_thread = AlarmThread(timeout) self.alarm_thread.start()
def set_alarm(self, duration, handler): sigalrm_handler = signal.signal(signal.SIGALRM, handler) self.addCleanup(signal.signal, signal.SIGALRM, sigalrm_handler) self._begin_alarm_thread(duration)
def handler(signum, frame): # logs message with level debug on this logger if signum == signal.SIGXCPU: # when process reaches soft limit --> a SIGXCPU signal is sent # (it normally terminats the process) raise CpuTimeoutException elif signum == signal.SIGALRM: # SIGALRM is sent to process when the specified time limit to an alarm function elapses # (when real or clock time elapses) raise TimeoutException raise AnythingException
def _signal_catcher(self, sig, frame): # NOTE(sileht): This is useful only for python < 3.5 # in python >= 3.5 we could read the signal number # from the wakeup_fd pipe if sig in (SIGALRM, signal.SIGTERM): self._signals_received.appendleft(sig) else: self._signals_received.append(sig)
def test_exit_code_signal(self): self.mktmp("import signal, time\n" "signal.setitimer(signal.ITIMER_REAL, 0.1)\n" "time.sleep(1)\n") self.system("%s %s" % (sys.executable, self.fname)) self.assertEqual(ip.user_ns['_exit_code'], -signal.SIGALRM)
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 # The signal handler. Just note that the signal occurred and # from who.
