我们从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 exec_timed(code, context, timeout_secs): """ Dynamically execute 'code' using 'context' as the global enviroment. SafeEvalTimeoutException is raised if execution does not finish within the given timelimit. """ assert(timeout_secs > 0) signal_finished = False def alarm(secs): def wait(secs): for n in xrange(timeout_secs): time.sleep(1) if signal_finished: break else: thread.interrupt_main() thread.start_new_thread(wait, (secs,)) try: alarm(timeout_secs) exec code in context signal_finished = True except KeyboardInterrupt: raise SafeEvalTimeoutException(timeout_secs)
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 get_elapsed_time(): # Create new time object based on time subtract start time time_elapsed = int(time.time() - gSTARTTIME) # Perform math to get elapsed time. hour = (time_elapsed / 3600) mins = (time_elapsed % 3600) / 60 secs = (time_elapsed % 60) if gKILLTIME and time_elapsed > gKILLTIME: gALIVE = False thread.interrupt_main() if hour > 0: return "%d h %d m %d s" % (hour, mins, secs) elif mins > 0: return "%d m %d s" % (mins, secs) return "%d s" % secs
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 run(self): exists = os.path.exists mtime = lambda path: os.stat(path).st_mtime files = dict() for module in sys.modules.values(): try: path = inspect.getsourcefile(module) if path and exists(path): files[path] = mtime(path) except TypeError: pass while not self.status: for path, lmtime in files.iteritems(): if not exists(path) or mtime(path) > lmtime: self.status = 3 if not exists(self.lockfile): self.status = 2 elif mtime(self.lockfile) < time.time() - self.interval - 5: self.status = 1 if not self.status: time.sleep(self.interval) if self.status != 5: thread.interrupt_main()
def run(self): exists = os.path.exists mtime = lambda path: os.stat(path).st_mtime files = dict() for module in 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 files.iteritems(): if not exists(path) or mtime(path) > lmtime: self.status = 'reload' thread.interrupt_main() break time.sleep(self.interval)
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 timer(t, flag): global FLAG_COMPLETE global TIME_OVER_FLAG if (not flag): thread.exit() start = time.time() while(True): if(FLAG_COMPLETE): thread.exit() time.sleep(5) # print "#######\ntime.time() - start %f : t is %d" %(time.time() - # start,(int)(t)) if(time.time() - start > float(t)): print "****timer:time out!" # if(FLAG_COMPLETE): # thread.exit() TIME_OVER_FLAG = True thread.interrupt_main() #raise NameError("Time Out!") thread.exit()
def run(self): exists = os.path.exists mtime = lambda path: os.stat(path).st_mtime files = dict() for module in 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: for path, lmtime in files.iteritems(): if not exists(path) or mtime(path) > lmtime: self.status = 3 if not exists(self.lockfile): self.status = 2 elif mtime(self.lockfile) < time.time() - self.interval - 5: self.status = 1 if not self.status: time.sleep(self.interval) if self.status != 5: 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()
