我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用signal.SIGUSR1。
def test_register_signal(self): if runtime_info.OS_WIN: return result = {'handler': 0} def _handler(signum, frame): result['handler'] += 1 register_signal(signal.SIGUSR1, _handler) os.kill(os.getpid(), signal.SIGUSR1) os.kill(os.getpid(), signal.SIGUSR1) signal.signal(signal.SIGUSR1, signal.SIG_DFL) self.assertEqual(result['handler'], 2)
def test_log(self): """Logging tests""" # Not much to test here but exercise the code nonetheless # for regression/coverage. log = Log(verbose=False, prefix='test') log.debug("Invisible debug.") try: raise ValueError("Test exception") except ValueError: log.exception("Test exception with message") for num in range(1, 5): log.debug("Debug") log.info("Info") log.warning("Warning") log.error("Error") log.critical("Crtitical") os.kill(os.getpid(), signal.SIGUSR1 if (num % 2) != 0 else signal.SIGUSR2)
def t01(factory): pretty = '%s t1' % __file__ print(pretty) ctrl = factory.make_master('master') pid = ctrl.get_pid() os.kill(pid, signal.SIGUSR1) try: pid = ctrl.get_pid() except ConnectionClosed, e: print('FAIL %s: SIGUSR1 killed the process: %s' % (pretty, e)) return False except Exception, e: print('FAIL %s: unknown error: %s' % (pretty, e)) return False return True # check that trace files are written to <home>/.ave/hickup with predictable # file names
def t04(factory): pretty = '%s t4' % __file__ print(pretty) ctrl = factory.make_master('master') bpid = ctrl.get_pid() spids = ctrl.get_session_pids() if [pid for pid in spids if pid < 2] != []: print('FAIL %s: impossible session PIDs: %d' % (pretty, spids)) return False # signal the broker and wait for the hickup directory to appear. the signal # should be propagated to the equipment lister long before the directory is # created. os.kill(bpid, signal.SIGUSR1) path = os.path.join(factory.HOME.path, '.ave', 'hickup') wait_hickup_dir(path, 3) spids2 = ctrl.get_session_pids() if spids2 != spids: print('FAIL %s: sessions affected: %s != %s' % (pretty, spids2, spids)) return False return True
def t01(pretty, factory): ctrl = factory.make_control(home=factory.HOME.path) pid = ctrl.get_pid() os.kill(pid, signal.SIGUSR1) try: pid = ctrl.get_pid() except ConnectionClosed, e: print('FAIL %s: SIGUSR1 killed the process: %s' % (pretty, e)) return False except Exception, e: print('FAIL %s: unknown error: %s' % (pretty, e)) return False return True # check that trace files are written to <home>/.ave/hickup with predictable # file names
def t03(pretty, factory): ctrl = factory.make_control(home=factory.HOME.path) pid = ctrl.get_pid() ctrl.make_child() os.kill(pid, signal.SIGUSR1) path = os.path.join(factory.HOME.path, '.ave', 'hickup') wait_hickup_dir(path, 3) files = glob.glob(os.path.join(path, '*')) if len(files) != 2: print('FAIL %s: wrong number of files: %s' % (pretty, files)) return False return True # check that the signal is not propagated to non-ave processes such as external # tools. these will otherwise terminate if they do not handle the signal.
def handle_SIGUSR1(self, signum, frame): # propagate the signal to children, but only if they are AVE processes for pid in self.get_children(): name = get_proc_name(pid) if name.startswith('ave-'): os.kill(pid, signal.SIGUSR1) # make the dump directory if it doesn't exist hickup_dir = os.path.join(self.home, '.ave', 'hickup') try: os.makedirs(hickup_dir) except OSError, e: if e.errno != errno.EEXIST: self.log('ERROR: could not create %s: %s' % (hickup_dir, e)) return # create the trace file date = time.strftime('%Y%m%d-%H%M%S') name = '%s-%s-%d' % (date, self.proc_name, os.getpid()) path = os.path.join(hickup_dir, name) with open(path, 'w') as f: f.write('stack:\n%s' % ''.join(traceback.format_stack(frame))) f.write('locals:\n%s\n' % frame.f_locals) f.write('globals:\n%s' % frame.f_globals)
def main(): """ Script main. This script is designed so that a process watcher like runit or monit can watch this process and take corrective action if it ever goes away. """ print ("starting hot-restarter with target: {}".format(sys.argv[1])) signal.signal(signal.SIGTERM, sigterm_handler) signal.signal(signal.SIGHUP, sighup_handler) signal.signal(signal.SIGCHLD, sigchld_handler) signal.signal(signal.SIGUSR1, sigusr1_handler) # Start the first child process and then go into an endless loop since everything else happens via # signals. fork_and_exec() while True: time.sleep(60)
def handle_logging_signal(self, signum, frame): """ Handle a signal sent to this process (SIGUSR1 or SIGUSR2) to enable or disable logging. :param signum: signal number sent to process :type signum: int :param frame: current stack frame when signal was caught """ if signum == signal.SIGUSR1: logger.warning('Logging enabled via signal; send SIGUSR2 to PID ' '%d to disable logging', getpid()) self.log_enabled = True elif signum == signal.SIGUSR2: logger.warning('Logging disabled via signal; send SIGUSR1 to PID ' '%d to enable logging', getpid()) self.log_enabled = False # else don't know how we got here, but ignore it
def test_signals(self): signalled_all.acquire() self.spawnSignallingThread() 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: signal.alarm(1) signal.pause() 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 test_poll_eintr(self): got_signal = [False] def record(*args): got_signal[0] = True pid = os.getpid() oldhandler = signal.signal(signal.SIGUSR1, record) try: killer = self.Process(target=self._killer, args=(pid,)) killer.start() p = self.Process(target=time.sleep, args=(1,)) p.start() p.join() self.assertTrue(got_signal[0]) self.assertEqual(p.exitcode, 0) killer.join() finally: signal.signal(signal.SIGUSR1, oldhandler) # # Test to verify handle verification, see issue 3321 #
def test_ignore(self): conn, child_conn = multiprocessing.Pipe() try: p = multiprocessing.Process(target=self._test_ignore, args=(child_conn,)) p.daemon = True p.start() child_conn.close() self.assertEqual(conn.recv(), 'ready') time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) time.sleep(0.1) conn.send(1234) self.assertEqual(conn.recv(), 1234) time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) self.assertEqual(conn.recv_bytes(), b'x'*(1024*1024)) time.sleep(0.1) p.join() finally: conn.close()
def init_signals(self): # Set up signals through the event loop API. self.loop.add_signal_handler(signal.SIGQUIT, self.handle_quit, signal.SIGQUIT, None) self.loop.add_signal_handler(signal.SIGTERM, self.handle_exit, signal.SIGTERM, None) self.loop.add_signal_handler(signal.SIGINT, self.handle_quit, signal.SIGINT, None) self.loop.add_signal_handler(signal.SIGWINCH, self.handle_winch, signal.SIGWINCH, None) self.loop.add_signal_handler(signal.SIGUSR1, self.handle_usr1, signal.SIGUSR1, None) self.loop.add_signal_handler(signal.SIGABRT, self.handle_abort, signal.SIGABRT, None) # Don't let SIGTERM and SIGUSR1 disturb active requests # by interrupting system calls signal.siginterrupt(signal.SIGTERM, False) signal.siginterrupt(signal.SIGUSR1, False)
def test_sigpending(self): code = """if 1: import os import signal def handler(signum, frame): 1/0 signum = signal.SIGUSR1 signal.signal(signum, handler) signal.pthread_sigmask(signal.SIG_BLOCK, [signum]) os.kill(os.getpid(), signum) pending = signal.sigpending() if pending != {signum}: raise Exception('%s != {%s}' % (pending, signum)) try: signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) except ZeroDivisionError: pass else: raise Exception("ZeroDivisionError not raised") """ assert_python_ok('-c', code)
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 test_pthread_kill_main_thread(self): # Test that a signal can be sent to the main thread with pthread_kill() # before any other thread has been created (see issue #12392). code = """if True: import threading import signal import sys def handler(signum, frame): sys.exit(3) signal.signal(signal.SIGUSR1, handler) signal.pthread_kill(threading.get_ident(), signal.SIGUSR1) sys.exit(2) """ with spawn_python('-c', code) as process: stdout, stderr = process.communicate() exitcode = process.wait() if exitcode != 3: raise Exception("Child error (exit code %s): %s" % (exitcode, stdout))
def test_communicate_eintr(self): # Issue #12493: communicate() should handle EINTR def handler(signum, frame): pass old_handler = signal.signal(signal.SIGUSR1, handler) self.addCleanup(signal.signal, signal.SIGUSR1, old_handler) args = [sys.executable, "-c", 'import os, signal;' 'os.kill(os.getppid(), signal.SIGUSR1)'] for stream in ('stdout', 'stderr'): kw = {stream: subprocess.PIPE} with subprocess.Popen(args, **kw) as process: # communicate() will be interrupted by SIGUSR1 process.communicate() # This test is Linux-ish specific for simplicity to at least have # some coverage. It is not a platform specific bug.
def launch(self): self.check() received = False def set_received(*_): nonlocal received received = True signal.signal(signal.SIGUSR1, set_received) pid = self.fork_zeo() if not received: while not signal.sigtimedwait([signal.SIGUSR1], 1): pid, waitres = os.waitpid(pid, os.WNOHANG) if pid: log.error('Database server failed to start (check log above).') sys.exit(1) settings.DB_URI = urlunsplit(('zeo', '', self.zeo_path, '', '')) log.debug('Launched built-in ZEO') return pid
def command_write(): """ Writes changes to file, and resets the change_count. """ viento_utils.directories_check() with open(viento_utils.f_drafts, 'w') as f: json.dump(drafts, f) print(str(change_count.count) + " change", end='') if change_count.count != 1: print("s", end='') print(" saved to file.") viento_utils.log('file_write', args=[change_count.count, viento_utils.f_drafts]) change_count.reset() try: with open(viento_utils.f_pid, 'r') as f: pid = int(f.read()) os.kill(pid, signal.SIGUSR1) except FileNotFoundError: pass
def test_poll_eintr(self): got_signal = [False] def record(*args): got_signal[0] = True pid = os.getpid() oldhandler = signal.signal(signal.SIGUSR1, record) try: killer = self.Process(target=self._killer, args=(pid,)) killer.start() try: p = self.Process(target=time.sleep, args=(2,)) p.start() p.join() finally: killer.join() self.assertTrue(got_signal[0]) self.assertEqual(p.exitcode, 0) finally: signal.signal(signal.SIGUSR1, oldhandler) # # Test to verify handle verification, see issue 3321 #
def test_ignore_listener(self): conn, child_conn = multiprocessing.Pipe() try: p = multiprocessing.Process(target=self._test_ignore_listener, args=(child_conn,)) p.daemon = True p.start() child_conn.close() address = conn.recv() time.sleep(0.1) os.kill(p.pid, signal.SIGUSR1) time.sleep(0.1) client = multiprocessing.connection.Client(address) self.assertEqual(client.recv(), 'welcome') p.join() finally: conn.close()
def setup_limit(self): """set up the process limit""" assert currentThread().getName() == 'MainThread' os.setpgrp() if self._limit_set <= 0: if self.max_time is not None: self._old_usr2_hdlr = signal(SIGUSR2, self._hangle_sig_timeout) self._timer = Timer(max(1, int(self.max_time) - self._elapse_time), self._time_out) self._start_time = int(time()) self._timer.start() if self.max_cpu_time is not None: self._old_max_cpu_time = getrlimit(RLIMIT_CPU) cpu_limit = (int(self.max_cpu_time), self._old_max_cpu_time[1]) self._old_sigxcpu_hdlr = signal(SIGXCPU, self._handle_sigxcpu) setrlimit(RLIMIT_CPU, cpu_limit) if self.max_memory is not None: self._msentinel = MemorySentinel(1, int(self.max_memory) ) self._old_max_memory = getrlimit(RLIMIT_AS) self._old_usr1_hdlr = signal(SIGUSR1, self._hangle_sig_memory) as_limit = (int(self.max_memory), self._old_max_memory[1]) setrlimit(RLIMIT_AS, as_limit) self._msentinel.start() self._limit_set += 1
def clean_limit(self): """reinstall the old process limit""" if self._limit_set > 0: if self.max_time is not None: self._timer.cancel() self._elapse_time += int(time())-self._start_time self._timer = None signal(SIGUSR2, self._old_usr2_hdlr) if self.max_cpu_time is not None: setrlimit(RLIMIT_CPU, self._old_max_cpu_time) signal(SIGXCPU, self._old_sigxcpu_hdlr) if self.max_memory is not None: self._msentinel.stop() self._msentinel = None setrlimit(RLIMIT_AS, self._old_max_memory) signal(SIGUSR1, self._old_usr1_hdlr) self._limit_set -= 1
def main(): """ Script main. This script is designed so that a process watcher like runit or monit can watch this process and take corrective action if it ever goes away. """ print "starting hot-restarter with target: {}".format(sys.argv[1]) signal.signal(signal.SIGTERM, sigterm_handler) signal.signal(signal.SIGHUP, sighup_handler) signal.signal(signal.SIGCHLD, sigchld_handler) signal.signal(signal.SIGUSR1, sigusr1_handler) # Start the first child process and then go into an endless loop since everything else happens via # signals. fork_and_exec() while True: time.sleep(60)
def run(self): def log_failure(failure): logger.exception(failure.value) if failure.frames: logger.critical(str("").join(format_tb(failure.getTracebackObject()))) def errback_main(failure): log_failure(failure) self.task.start(interval=0).addErrback(errback_main) def errback_flush_states(failure): log_failure(failure) self._flush_states_task.start(interval=300).addErrback(errback_flush_states) def debug(sig, frame): logger.critical("Signal received: printing stack trace") logger.critical(str("").join(format_stack(frame))) self.task.start(interval=0).addErrback(errback_main) self._logging_task.start(interval=30) self._flush_states_task.start(interval=300).addErrback(errback_flush_states) signal(SIGUSR1, debug) reactor.addSystemEventTrigger('before', 'shutdown', self.stop) reactor.run()
def test_childSignalHandling(self): """ The disposition of signals which are ignored in the parent process is reset to the default behavior for the child process. """ # Somewhat arbitrarily select SIGUSR1 here. It satisfies our # requirements that: # - The interpreter not fiddle around with the handler # behind our backs at startup time (this disqualifies # signals like SIGINT and SIGPIPE). # - The default behavior is to exit. # # This lets us send the signal to the child and then verify # that it exits with a status code indicating that it was # indeed the signal which caused it to exit. which = signal.SIGUSR1 # Ignore the signal in the parent (and make sure we clean it # up). handler = signal.signal(which, signal.SIG_IGN) self.addCleanup(signal.signal, signal.SIGUSR1, handler) # Now do the test. return self._testSignal(signal.SIGUSR1)
def initialize_exception_listener(): # must be invoked in main thread in "geventless" runs in order for raise_in_main_thread to work global REGISTERED_SIGNAL if REGISTERED_SIGNAL: # already registered return if threading.current_thread() is not threading.main_thread(): raise NotMainThread() def handle_signal(sig, stack): global LAST_ERROR error = LAST_ERROR LAST_ERROR = None if error: raise error raise LastErrorEmpty(signal=sig) custom_signal = signal.SIGUSR1 if signal.getsignal(custom_signal) in (signal.SIG_DFL, signal.SIG_IGN): # check if signal is already trapped signal.signal(custom_signal, handle_signal) REGISTERED_SIGNAL = custom_signal else: raise SignalAlreadyBound(signal=custom_signal)
def test_server_signals(self): server = spoon.server.TCPSpoon(("0.0.0.0", 30783)) calls = [ call(signal.SIGUSR1, server.reload_handler), call(signal.SIGTERM, server.shutdown_handler) ] self.mock_signal.assert_has_calls(calls)
def test_master_reload(self): server = spoon.server.TCPSpork(("0.0.0.0", 30783)) server.prefork = 2 server.pids = [100, 101] server.load_config() calls = [ call(100, signal.SIGUSR1), call(101, signal.SIGUSR1), ] self.mock_kill.assert_has_calls(calls)
def send_action(action, pidfile, logger=None): """Send a signal to an existing running daemon.""" if logger is None: logger = logging if not os.path.exists(pidfile): logger.critical("No pid file available: %s", pidfile) return with open(pidfile) as pidf: pid = int(pidf.read()) if action == "reload": os.kill(pid, signal.SIGUSR1) elif action == "stop": os.kill(pid, signal.SIGTERM)
