bobfrank / pydebug Goto Github PK

# …
import platform
import warn
# …
_EMPTY_OBJ = ctypes.py_object()
# …

def install_thread_tracer(trace_function, arg=None, skip_thread_ids=()):
    trace_f = Py_tracefunc(trace_function)
    trace_f_arg = arg
    _set_thread_tracer(trace_f, trace_f_arg, skip_thread_ids)

def uninstall_thread_tracer(skip_thread_ids=()):
    trace_f = ctypes.cast(None, Py_tracefunc)
    trace_f_arg = _EMPTY_OBJ
    _set_thread_tracer(trace_f, trace_f_arg, skip_thread_ids)

def _set_thread_tracer_cpython(trace_f, trace_f_arg, skip_thread_ids):
    interp = ctypes.pythonapi.PyInterpreterState_Head()
    t = ctypes.pythonapi.PyInterpreterState_ThreadHead(interp)
    while t != 0:
        t_p = ctypes.cast(t, ctypes.POINTER(PyThreadState))
        if t_p[0].thread_id != _thread.get_ident() and t_p[0].thread_id not \
                in skip_thread_ids:
            try:
                temp = t_p[0].c_traceobj
            except ValueError:
                temp = None
            if trace_f_arg != _EMPTY_OBJ:  # Py_XINCREF
                # ctypes.pythonapi._Total
                refcount = ctypes.c_long.from_address(id(trace_f_arg))
                refcount.value += 1
            t_p[0].c_tracefunc = ctypes.cast(None, Py_tracefunc)
            t_p[0].c_traceobj = _EMPTY_OBJ
            t_p[0].use_tracing = int(t_p[0].c_profilefunc is not None)
            if temp is not None:  # Py_XDECREF
                refcount = ctypes.c_long.from_address(id(temp))
                # Don't need to dealloc since we have a ref in here and it'll
                # always be >0
                refcount.value -= 1
            t_p[0].c_tracefunc = trace_f
            t_p[0].c_traceobj = trace_f_arg
            t_p[0].use_tracing = int(trace_f is not None or
                t_p[0].c_profilefunc is not None)
        t = ctypes.pythonapi.PyThreadState_Next(t)

_THREAD_TRACER_SETTERS = {
    'CPython': _set_thread_tracer_cpython,
    # Future implementations, maybe?
}

def _set_thread_tracer_noop(*arg, **kw):  # pylint: disable=unused-arg
    pass

def _set_thread_tracer(trace_f, trace_f_arg, skip_thread_ids):
    _thread_tracer_setter = _THREAD_TRACER_SETTERS.get(
        platform.python_implementation(), _set_thread_tracer_noop)
    if _thread_tracer_setter is _set_thread_tracer_noop:
        warnings.warn('no thread tracer found for {}'.format(
            platform.python_implementation()), RuntimeWarning)
    _thread_tracer_setter(trace_f, trace_f_arg, skip_thread_ids)
    # Skip lookup next time
    global _set_thread_tracer
    _set_thread_tracer = _thread_tracer_setter

class ServerHandler(object):
    # …
    def __setup_thread_tracer(self, setting):
        global debug__has_loaded_thread
        if setting:
            install_thread_tracer(breakpoint_tracer, self,
                skip_threads=(debug__has_loaded_thread,))
        else:
            uninstall_thread_tracer(skip_threads=(debug__has_loaded_thread,))
    # …