我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用signal.CTRL_BREAK_EVENT。
def kill(self, sig): '''Sends a Unix signal to the subprocess. Use constants from the :mod:`signal` module to specify which signal. ''' if sys.platform == 'win32': if sig in [signal.SIGINT, signal.CTRL_C_EVENT]: sig = signal.CTRL_C_EVENT elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]: sig = signal.CTRL_BREAK_EVENT else: sig = signal.SIGTERM os.kill(self.proc.pid, sig)
def _default_handler(signum, *args): ''' The default signal handler. Don't register with built-in signal.signal! This needs to be used on the subprocess await death workaround. ''' # All valid cpython windows signals sigs = { signal.SIGABRT: SIGABRT, # signal.SIGFPE: 'fpe', # Don't catch this # signal.SIGSEGV: 'segv', # Don't catch this # signal.SIGILL: 'illegal', # Don't catch this signal.SIGINT: SIGINT, signal.SIGTERM: SIGTERM, # Note that signal.CTRL_C_EVENT and signal.CTRL_BREAK_EVENT are # converted to SIGINT in _await_signal } try: exc = sigs[signum] except KeyError: exc = DaemonikerSignal _sketch_raise_in_main(exc)
def send_signal(self, sig): """Send a signal to the process """ if sig == signal.SIGTERM: self.terminate() elif sig == signal.CTRL_C_EVENT: os.kill(self.pid, signal.CTRL_C_EVENT) elif sig == signal.CTRL_BREAK_EVENT: os.kill(self.pid, signal.CTRL_BREAK_EVENT) else: raise ValueError("Unsupported signal: {}".format(sig))
def submit_code(timeout=5): code = request.form.get("code", "") inp = request.form.get("input", "") print(code, inp) warnings = int(request.form.get("warnings", "0"), 10) use_hex = int(request.form.get("hex", "0"), 10) args = [sys.executable, 'main.py', '--safe', '--', code] stderr = PIPE if warnings: args.insert(2, "--warnings") stderr = STDOUT if use_hex: args.insert(2, "--hex") with Popen(args, stdin=PIPE, stdout=PIPE, stderr=stderr, creationflags=is_windows and subprocess.CREATE_NEW_PROCESS_GROUP) as process: process.stdin.write(bytearray(inp, "utf-8")) process.stdin.close() response = "" try: process.wait(timeout) except TimeoutExpired: response = "Timeout running code.\n" if is_windows: os.kill(process.pid, signal.CTRL_BREAK_EVENT) else: process.send_signal(signal.SIGTERM) try: process.wait(2) except TimeoutExpired: response += "Really timed out code\n" process.kill() response += process.stdout.read().decode("cp1252", errors="replace") return response
def _ctrl_handler(sig): """Handle a sig event and return 0 to terminate the process""" if sig == signal.CTRL_C_EVENT: pass elif sig == signal.CTRL_BREAK_EVENT: pass else: print("UNKNOWN EVENT") return 0
def test_CTRL_BREAK_EVENT(self): self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
def test_ctrl_signals(self): p = psutil.Process(get_test_subprocess().pid) p.send_signal(signal.CTRL_C_EVENT) p.send_signal(signal.CTRL_BREAK_EVENT) p.kill() p.wait() self.assertRaises(psutil.NoSuchProcess, p.send_signal, signal.CTRL_C_EVENT) self.assertRaises(psutil.NoSuchProcess, p.send_signal, signal.CTRL_BREAK_EVENT)
def kill(self): if sys.platform.startswith("win"): # proc.terminate on Windows does not initiate a graceful shutdown # through the processes signal handlers it just kills it hard. So # this sends a SIGBREAK. You cannot sends a SIGINT (CTRL_C_EVENT) # to a process group in Windows, otherwise Ctrl+C would be # sent. self.proc.send_signal(signal.CTRL_BREAK_EVENT) else: self.proc.terminate()
def test_signal_waiting(self): ''' Fixture thine self. ''' proc1 = ProcFixture(signal.SIGINT) proc2 = ProcFixture(signal.SIGTERM) proc3 = ProcFixture(signal.SIGABRT) proc4 = ProcFixture(signal.CTRL_C_EVENT) proc5 = ProcFixture(signal.CTRL_BREAK_EVENT) self.assertEqual(_await_signal(proc1), signal.SIGINT) self.assertEqual(_await_signal(proc2), signal.SIGTERM) self.assertEqual(_await_signal(proc3), signal.SIGABRT) self.assertEqual(_await_signal(proc4), signal.SIGINT) self.assertEqual(_await_signal(proc5), signal.SIGINT)
def _await_signal(process): ''' Waits for the process to die, and then returns the exit code for the process, converting CTRL_C_EVENT and CTRL_BREAK_EVENT into SIGINT. ''' # Note that this is implemented with a busy wait process.wait() code = process.returncode if code == signal.CTRL_C_EVENT: code = signal.SIGINT elif code == signal.CTRL_BREAK_EVENT: code = signal.SIGINT return code
def send_signal(self, sig): """Send a signal to the process.""" # Don't signal a process that we know has already died. if self.returncode is not None: return if sig == signal.SIGTERM: self.terminate() elif sig == signal.CTRL_C_EVENT: os.kill(self.pid, signal.CTRL_C_EVENT) elif sig == signal.CTRL_BREAK_EVENT: os.kill(self.pid, signal.CTRL_BREAK_EVENT) else: raise ValueError("Unsupported signal: {}".format(sig))
def interrupt_process(process): print("Due to limitations of the operating system, we can only kill the process.", file=sys.stderr) process.send_signal(signal.CTRL_BREAK_EVENT)
