python类settrace()的实例源码

def _unsettrace():
        sys.settrace(None)
        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 _unsettrace():
        sys.settrace(None)

def _settrace(func):
        threading.settrace(func)
        sys.settrace(func)

def _unsettrace():
        sys.settrace(None)
        threading.settrace(None)

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 _unsettrace():
        sys.settrace(None)

def _settrace(func):
        threading.settrace(func)
        sys.settrace(func)

def _unsettrace():
        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 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()

def _unsettrace():
        sys.settrace(None)

def _settrace(func):
        threading.settrace(func)
        sys.settrace(func)