我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用_thread.interrupt_main()。
def run(self): exists = os.path.exists mtime = lambda p: os.stat(p).st_mtime files = dict() for module in list(sys.modules.values()): path = getattr(module, '__file__', '') if path[-4:] in ('.pyo', '.pyc'): path = path[:-1] if path and exists(path): files[path] = mtime(path) while not self.status: if not exists(self.lockfile)\ or mtime(self.lockfile) < time.time() - self.interval - 5: self.status = 'error' thread.interrupt_main() for path, lmtime in list(files.items()): if not exists(path) or mtime(path) > lmtime: self.status = 'reload' thread.interrupt_main() break time.sleep(self.interval)
def shutdown_system(): worker_thread.stop_all_threads() _thread.interrupt_main() current_pid = funcs.process__get_current_pid() matched = False for hwnd in funcs.window__find_handles(): pid = funcs.window__get_process_id(hwnd) if pid == current_pid: print("Sending PostMessage to hwnd owned by {0}".format(pid)) funcs.window__post_message(hwnd, WM_QUIT, 0, 0) matched = True # Continue in case there are more windows we own if not matched: time.sleep(0.1) # print("DEBUG could not find a window to post a quit to. Forcing quit.") # for t in threading.enumerate(): # print("Running thread: {0}".format(t)) sys.exit()
def run(self): files = dict() for module in list(sys.modules.values()): path = getattr(module, '__file__', '') if path[-4:] in ('.pyo', '.pyc'): path = path[:-1] if path and os.path.exists(path): files[path] = os.stat(path).st_mtime while not self.status: if not os.path.exists(self.lockfile) or \ os.stat(self.lockfile).st_mtime < time.time() - self.interval - 5: self.status = 'error' _thread.interrupt_main() for path, last_mtime in files.items(): if not os.path.exists(path) or os.stat(path).st_mtime > last_mtime: self.status = 'reload' _thread.interrupt_main() break time.sleep(self.interval)
def execution_loop(self): """loop on the main thread which is responsible for executing code""" if sys.platform == 'cli' and sys.version_info[:3] < (2, 7, 1): # IronPython doesn't support thread.interrupt_main until 2.7.1 import System self.main_thread = System.Threading.Thread.CurrentThread # save ourselves so global lookups continue to work (required pre-2.6)... cur_modules = set() try: cur_ps1 = sys.ps1 cur_ps2 = sys.ps2 except: # CPython/IronPython don't set sys.ps1 for non-interactive sessions, Jython and PyPy do sys.ps1 = cur_ps1 = '>>> ' sys.ps2 = cur_ps2 = '... ' self.send_prompt(cur_ps1, cur_ps2, allow_multiple_statements=False) while True: exit, cur_modules, cur_ps1, cur_ps2 = self.run_one_command(cur_modules, cur_ps1, cur_ps2) if exit: return
def run(self): exists = os.path.exists mtime = lambda path: os.stat(path).st_mtime files = dict() for module in list(sys.modules.values()): path = getattr(module, '__file__', '') if path[-4:] in ('.pyo', '.pyc'): path = path[:-1] if path and exists(path): files[path] = mtime(path) while not self.status: if not exists(self.lockfile)\ or mtime(self.lockfile) < time.time() - self.interval - 5: self.status = 'error' thread.interrupt_main() for path, lmtime in list(files.items()): if not exists(path) or mtime(path) > lmtime: self.status = 'reload' thread.interrupt_main() break time.sleep(self.interval)
def _terminate_execution(self): # called from receiverthread self._trace("shutting down execution pool") self._execpool.trigger_shutdown() if not self._execpool.waitall(5.0): self._trace( "execution ongoing after 5 secs,"" trying interrupt_main") # We try hard to terminate execution based on the assumption # that there is only one gateway object running per-process. if sys.platform != "win32": self._trace("sending ourselves a SIGINT") os.kill(os.getpid(), 2) # send ourselves a SIGINT elif interrupt_main is not None: self._trace("calling interrupt_main()") interrupt_main() if not self._execpool.waitall(10.0): self._trace("execution did not finish in another 10 secs, " "calling os._exit()") os._exit(1)
def recv_bytes(self, connection): """ Given a socket connection, receive up to 2048 bytes from its buffer and return that data as a bytes object. Returns an empty bytes object if there is no data to receive. """ try: bytes_message = connection.recv(2048) except BlockingIOError: # Non-blocking socket couldn't immediately receive. return None except ConnectionResetError: self._print_message('Connection was reset.') self.connection.close() _thread.interrupt_main() return bytes_message
def run(self): # We don't want too frequent retries try_interval = 1 while True: try: try_interval *= 2 with self.capp.connection() as conn: recv = EventReceiver(conn, handlers={'*': self.on_event}, app=self.capp) try_interval = 1 recv.capture(limit=None, timeout=None, wakeup=True) except (KeyboardInterrupt, SystemExit): import _thread _thread.interrupt_main() except Exception as e: logger.error('Failed to capture events: "%s", ' 'trying again in %s seconds.', e, try_interval) logger.debug(e, exc_info=True) time.sleep(try_interval)
def run(self): try: with self.sess.as_default(): self._run() except: traceback.print_exc() _thread.interrupt_main()
def _cmd_abrt(self): """aborts the current running command""" # abort command, interrupts execution of the main thread. self.interrupt_main()
def interrupt_main(self): """aborts the current running command""" raise NotImplementedError
def interrupt_main(self): # acquire the send lock so we dont interrupt while we're communicting w/ the debugger with self.send_lock: if sys.platform == 'cli' and sys.version_info[:3] < (2, 7, 1): # IronPython doesn't get thread.interrupt_main until 2.7.1 self.main_thread.Abort(ReplAbortException()) else: thread.interrupt_main()
