我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用threading.settrace()。
def stoptrace(): global connected if connected: pydevd_tracing.restore_sys_set_trace_func() sys.settrace(None) try: #not available in jython! threading.settrace(None) # for all future threads except: pass from _pydev_bundle.pydev_monkey import undo_patch_thread_modules undo_patch_thread_modules() debugger = get_global_debugger() if debugger: debugger.set_trace_for_frame_and_parents( get_frame(), also_add_to_passed_frame=True, overwrite_prev_trace=True, dispatch_func=lambda *args:None) debugger.exiting() kill_all_pydev_threads() connected = False
def enable_tracing(self): """ Enable tracing if it is disabled and debugged program is running, else do nothing. Do this on all threads but the debugger thread. :return: True if tracing has been enabled, False else. """ _logger.x_debug("enable_tracing()") #self.dump_tracing_state("before enable_tracing()") if not self.tracing_enabled and self.execution_started: # Restore or set trace function on all existing frames appart from # debugger threading.settrace(self._tracer) # then enable on all threads to come for thr in threading.enumerate(): if thr.ident != self.debugger_thread_ident: # skip debugger thread a_frame = sys._current_frames()[thr.ident] while a_frame: a_frame.f_trace = self._tracer a_frame = a_frame.f_back iksettrace._set_trace_on(self._tracer, self.debugger_thread_ident) self.tracing_enabled = True #self.dump_tracing_state("after enable_tracing()") return self.tracing_enabled
def test_finalize_with_trace(self): # Issue1733757 # Avoid a deadlock when sys.settrace steps into threading._shutdown assert_python_ok("-c", """if 1: import sys, threading # A deadlock-killer, to prevent the # testsuite to hang forever def killer(): import os, time time.sleep(2) print('program blocked; aborting') os._exit(2) t = threading.Thread(target=killer) t.daemon = True t.start() # This is the trace function def func(frame, event, arg): threading.current_thread() return func sys.settrace(func) """)
def _unsettrace(): sys.settrace(None)
def _settrace(func): threading.settrace(func) sys.settrace(func)
def _unsettrace(): sys.settrace(None) threading.settrace(None)
def run(self, cmd): import __main__ dict = __main__.__dict__ if not self.donothing: threading.settrace(self.globaltrace) sys.settrace(self.globaltrace) try: exec cmd in dict, dict finally: if not self.donothing: sys.settrace(None) threading.settrace(None)
def runfunc(self, func, *args, **kw): result = None if not self.donothing: sys.settrace(self.globaltrace) try: result = func(*args, **kw) finally: if not self.donothing: sys.settrace(None) return result
def opt_spew(self): """Print an insanely verbose log of everything that happens. Useful when debugging freezes or locks in complex code.""" sys.settrace(util.spewer) try: import threading except ImportError: return threading.settrace(util.spewer)
def no_tracing(f): @functools.wraps(f) def wrapper(*args, **kwargs): trace_func = sys.gettrace() try: sys.settrace(None) threading.settrace(None) return f(*args, **kwargs) finally: sys.settrace(trace_func) threading.settrace(trace_func) return wrapper
def start_instrumenting(output_file, to_include=(), to_exclude=()): """Enable tracing of all function calls (from specified modules).""" trace_event.trace_enable(output_file) traceFunc = _generate_trace_function(to_include, to_exclude) sys.settrace(traceFunc) threading.settrace(traceFunc)
def stop_instrumenting(): trace_event.trace_disable() sys.settrace(None) threading.settrace(None)
def start(self): """Start this Tracer. Return a Python function suitable for use with sys.settrace(). """ self.thread = threading.currentThread() sys.settrace(self._trace) return self._trace
def stop(self): """Stop this Tracer.""" self.stopped = True if self.thread != threading.currentThread(): # Called on a different thread than started us: we can't unhook # ourseves, but we've set the flag that we should stop, so we won't # do any more tracing. return if hasattr(sys, "gettrace") and self.warn: if sys.gettrace() != self._trace: msg = "Trace function changed, measurement is likely wrong: %r" self.warn(msg % (sys.gettrace(),)) #print("Stopping tracer on %s" % threading.current_thread().ident) sys.settrace(None)
def _installation_trace(self, frame_unused, event_unused, arg_unused): """Called on new threads, installs the real tracer.""" # Remove ourselves as the trace function sys.settrace(None) # Install the real tracer. fn = self._start_tracer() # Invoke the real trace function with the current event, to be sure # not to lose an event. if fn: fn = fn(frame_unused, event_unused, arg_unused) # Return the new trace function to continue tracing in this scope. return fn
def start(self): """Start collecting trace information.""" if self._collectors: self._collectors[-1].pause() self._collectors.append(self) #print("Started: %r" % self._collectors, file=sys.stderr) # Check to see whether we had a fullcoverage tracer installed. traces0 = [] if hasattr(sys, "gettrace"): fn0 = sys.gettrace() if fn0: tracer0 = getattr(fn0, '__self__', None) if tracer0: traces0 = getattr(tracer0, 'traces', []) # Install the tracer on this thread. fn = self._start_tracer() for args in traces0: (frame, event, arg), lineno = args try: fn(frame, event, arg, lineno=lineno) except TypeError: raise Exception( "fullcoverage must be run with the C trace function." ) # Install our installation tracer in threading, to jump start other # threads. threading.settrace(self._installation_trace)
def pause(self): """Pause tracing, but be prepared to `resume`.""" for tracer in self.tracers: tracer.stop() stats = tracer.get_stats() if stats: print("\nCoverage.py tracer stats:") for k in sorted(stats.keys()): print("%16s: %s" % (k, stats[k])) threading.settrace(None)
def test_finalize_with_trace(self): # Issue1733757 # Avoid a deadlock when sys.settrace steps into threading._shutdown p = subprocess.Popen([sys.executable, "-c", """if 1: import sys, threading # A deadlock-killer, to prevent the # testsuite to hang forever def killer(): import os, time time.sleep(2) print 'program blocked; aborting' os._exit(2) t = threading.Thread(target=killer) t.daemon = True t.start() # This is the trace function def func(frame, event, arg): threading.current_thread() return func sys.settrace(func) """], stdout=subprocess.PIPE, stderr=subprocess.PIPE) self.addCleanup(p.stdout.close) self.addCleanup(p.stderr.close) stdout, stderr = p.communicate() rc = p.returncode self.assertFalse(rc == 2, "interpreted was blocked") self.assertTrue(rc == 0, "Unexpected error: " + repr(stderr))
def test_frame_tstate_tracing(self): # Issue #14432: Crash when a generator is created in a C thread that is # destroyed while the generator is still used. The issue was that a # generator contains a frame, and the frame kept a reference to the # Python state of the destroyed C thread. The crash occurs when a trace # function is setup. def noop_trace(frame, event, arg): # no operation return noop_trace def generator(): while 1: yield "generator" def callback(): if callback.gen is None: callback.gen = generator() return next(callback.gen) callback.gen = None old_trace = sys.gettrace() sys.settrace(noop_trace) try: # Install a trace function threading.settrace(noop_trace) # Create a generator in a C thread which exits after the call _testcapi.call_in_temporary_c_thread(callback) # Call the generator in a different Python thread, check that the # generator didn't keep a reference to the destroyed thread state for test in range(3): # The trace function is still called here callback() finally: sys.settrace(old_trace)
def patch_threads(self): try: # not available in jython! import threading threading.settrace(self.trace_dispatch) # for all future threads except: pass from _pydev_bundle.pydev_monkey import patch_thread_modules patch_thread_modules()
def has_data_to_redirect(): if getattr(sys, 'stdoutBuf', None): if not sys.stdoutBuf.empty(): return True if getattr(sys, 'stderrBuf', None): if not sys.stderrBuf.empty(): return True return False #======================================================================================================================= # settrace #=======================================================================================================================
def settrace_forked(): ''' When creating a fork from a process in the debugger, we need to reset the whole debugger environment! ''' host, port = dispatch() from _pydevd_bundle import pydevd_tracing pydevd_tracing.restore_sys_set_trace_func() if port is not None: global connected connected = False custom_frames_container_init() settrace( host, port=port, suspend=False, trace_only_current_thread=False, overwrite_prev_trace=True, patch_multiprocessing=True, ) #======================================================================================================================= # SetupHolder #=======================================================================================================================
def enable(self): """enable logging on all threads.""" assert(self.fh is not None) sys.settrace(self.__callback) threading.settrace(self.__callback)
def disable(self): """disable logging on all threads.""" sys.settrace(None) threading.settrace(None)
def disable_tracing(self): """ Disable tracing if it is disabled and debugged program is running, else do nothing. Do this on all threads but the debugger thread. :return: False if tracing has disabled, False else. """ _logger.x_debug("disable_tracing()") #self.dump_tracing_state("before disable_tracing()") if self.tracing_enabled and self.execution_started: threading.settrace(None) # don't trace threads to come iksettrace._set_trace_off() self.tracing_enabled = False #self.dump_tracing_state("after disable_tracing()") return self.tracing_enabled
def _runscript(self, filename): """ Launchs debugged program execution using the execfile() builtin. We reset and setup the __main__ dict to allow the script to run in __main__ namespace. This is required for imports from __main__ to run correctly. Note that this has the effect to wipe IKPdb's vars created at this point. """ import __main__ __main__.__dict__.clear() __main__.__dict__.update({"__name__" : "__main__", "__file__" : filename, "__builtins__": __builtins__,}) self.mainpyfile = self.canonic(filename) statement = 'execfile(%r)\n' % filename globals = __main__.__dict__ locals = globals # When IKPdb sets tracing, a number of call and line events happens # BEFORE debugger even reaches user's code (and the exact sequence of # events depends on python version). So we take special measures to # avoid stopping before we reach the main script (see reset(), # _tracer() and _line_tracer() methods for details). self.reset() self.execution_started = True # Turn on limited tracing by setting trace function for # current_thread only. This allow self.frame_beginning to be set at # first tracer "call" invocation. sys.settrace(self._tracer) try: exec(statement, globals, locals) except IKPdbQuit: pass finally: self.disable_tracing()
