Python greenlet 模块，settrace() 实例源码

def test_greenlet_tracing(self):
            main = greenlet.getcurrent()
            actions = []
            def trace(*args):
                actions.append(args)
            def dummy():
                pass
            def dummyexc():
                raise SomeError()
            oldtrace = greenlet.settrace(trace)
            try:
                g1 = greenlet.greenlet(dummy)
                g1.switch()
                g2 = greenlet.greenlet(dummyexc)
                self.assertRaises(SomeError, g2.switch)
            finally:
                greenlet.settrace(oldtrace)
            self.assertEqual(actions, [
                ('switch', (main, g1)),
                ('switch', (g1, main)),
                ('switch', (main, g2)),
                ('throw', (g2, main)),
            ])

def test_exception_disables_tracing(self):
            main = greenlet.getcurrent()
            actions = []
            def trace(*args):
                actions.append(args)
                raise SomeError()
            def dummy():
                main.switch()
            g = greenlet.greenlet(dummy)
            g.switch()
            oldtrace = greenlet.settrace(trace)
            try:
                self.assertRaises(SomeError, g.switch)
                self.assertEqual(greenlet.gettrace(), None)
            finally:
                greenlet.settrace(oldtrace)
            self.assertEqual(actions, [
                ('switch', (main, g)),
            ])

def enable_debug():
    if IS_PYPY:
        sys.stderr.write("settrace api unsupported on pypy")
        sys.stderr.flush()
        return

    import inspect
    def trace_green(event, args):
        src, target = args
        if event == "switch":
            print("from %s switch to %s" % (src, target))
        elif event == "throw":
            print("from %s throw exception to %s" % (src, target))

        if src.gr_frame:
            tracebacks = inspect.getouterframes(src.gr_frame)
            buff = []
            for traceback in tracebacks:
                srcfile, lineno, func_name, codesample = traceback[1:-1]
                trace_line = '''File "%s", line %d, in %s\n%s '''
                buff.append(trace_line % (srcfile, lineno, func_name, "".join(codesample)))

            print("".join(buff))

    greenlet.settrace(trace_green)

def start_profiler():
    global _state

    _state = GlobalState()

    frame = sys._getframe(0)
    current_greenlet = greenlet.getcurrent()  # pylint: disable=no-member

    thread_state = ensure_thread_state(current_greenlet, frame)
    _state.last = thread_state

    # this needs to be instantiate before the handler is installed
    greenlet.settrace(greenlet_profiler)  # pylint: disable=no-member
    sys.setprofile(thread_profiler)
    threading.setprofile(thread_profiler)

def stop_profiler():
    # we keep the _state around for the user until the next session

    # Unregister the profiler in this order, otherwise we will have extra
    # measurements in the end
    sys.setprofile(None)
    threading.setprofile(None)
    greenlet.settrace(None)  # pylint: disable=no-member

def __init__(self, timeout=1000):
        self.timeout = timeout
        self._active_greenlet = None
        self._greenlet_switch_counter = 0
        self._greenlet_last_switch_time = None
        greenlet.settrace(self._greenlet_switch_tracer)

def enable_debug():
    if IS_PYPY:
        sys.stderr.write("settrace api unsupported on pypy")
        sys.stderr.flush()
        return

    import inspect

    def trace_green(event, args):
        src, target = args
        if event == "switch":
            print("from %s switch to %s" % (src, target))
        elif event == "throw":
            print("from %s throw exception to %s" % (src, target))

        if src.gr_frame:
            tracebacks = inspect.getouterframes(src.gr_frame)
            buff = []
            for traceback in tracebacks:
                srcfile, lineno, func_name, codesample = traceback[1:-1]
                trace_line = '''File "%s", line %d, in %s\n%s '''
                buff.append(trace_line %
                            (srcfile, lineno, func_name, "".join(codesample)))

            print("".join(buff))

    greenlet.settrace(trace_green)

def apply_patch(hogging_detection=False, real_threads=1):
    _logger.info('applying gevent patch (%s real threads)', real_threads)

    # real_threads is 1 by default so it will be possible to run watch_threads concurrently
    if hogging_detection:
        real_threads += 1

    if real_threads:
        _RealThreadsPool(real_threads)

    _patch_module_locks()

    import gevent
    import gevent.monkey

    for m in ["easypy.threadtree", "easypy.concurrency"]:
        assert m not in sys.modules, "Must apply the gevent patch before importing %s" % m

    gevent.monkey.patch_all(Event=True, sys=True)

    _unpatch_logging_handlers_lock()

    global HUB
    HUB = gevent.get_hub()

    global threading
    import threading
    for thread in threading.enumerate():
        _set_thread_uuid(thread.ident)
    _set_main_uuid()  # the patched threading has a new ident for the main thread

    # this will declutter the thread dumps from gevent/greenlet frames
    from .threadtree import _BOOTSTRAPPERS
    import gevent, gevent.threading, gevent.greenlet
    _BOOTSTRAPPERS.update([gevent, gevent.threading, gevent.greenlet])

    if hogging_detection:
        import greenlet
        greenlet.settrace(lambda *args: _greenlet_trace_func(*args))
        defer_to_thread(detect_hogging, 'detect-hogging')