Count of the number of times a :class:`Breakpoint` has been hit.
-.. class:: Bdb(skip=None)
+.. class:: Bdb(skip=None, backend='settrace')
The :class:`Bdb` class acts as a generic Python debugger base class.
frame is considered to originate in a certain module is determined
by the ``__name__`` in the frame globals.
+ The *backend* argument specifies the backend to use for :class:`Bdb`. It
+ can be either ``'settrace'`` or ``'monitoring'``. ``'settrace'`` uses
+ :func:`sys.settrace` which has the best backward compatibility. The
+ ``'monitoring'`` backend uses the new :mod:`sys.monitoring` that was
+ introduced in Python 3.12, which can be much more efficient because it
+ can disable unused events. We are trying to keep the exact interfaces
+ for both backends, but there are some differences. The debugger developers
+ are encouraged to use the ``'monitoring'`` backend to achieve better
+ performance.
+
.. versionchanged:: 3.1
Added the *skip* parameter.
+ .. versionchanged:: 3.14
+ Added the *backend* parameter.
+
The following methods of :class:`Bdb` normally don't need to be overridden.
.. method:: canonic(filename)
<os.path.abspath>`. A *filename* with angle brackets, such as ``"<stdin>"``
generated in interactive mode, is returned unchanged.
+ .. method:: start_trace(self)
+
+ Start tracing. For ``'settrace'`` backend, this method is equivalent to
+ ``sys.settrace(self.trace_dispatch)``
+
+ .. versionadded:: 3.14
+
+ .. method:: stop_trace(self)
+
+ Stop tracing. For ``'settrace'`` backend, this method is equivalent to
+ ``sys.settrace(None)``
+
+ .. versionadded:: 3.14
+
.. method:: reset()
Set the :attr:`!botframe`, :attr:`!stopframe`, :attr:`!returnframe` and
Return all breakpoints that are set.
+ Derived classes and clients can call the following methods to disable and
+ restart events to achieve better performance. These methods only work
+ when using the ``'monitoring'`` backend.
+
+ .. method:: disable_current_event()
+
+ Disable the current event until the next time :func:`restart_events` is
+ called. This is helpful when the debugger is not interested in the current
+ line.
+
+ .. versionadded:: 3.14
+
+ .. method:: restart_events()
+
+ Restart all the disabled events. This function is automatically called in
+ ``dispatch_*`` methods after ``user_*`` methods are called. If the
+ ``dispatch_*`` methods are not overridden, the disabled events will be
+ restarted after each user interaction.
+
+ .. versionadded:: 3.14
+
+
Derived classes and clients can call the following methods to get a data
structure representing a stack trace.
import fnmatch
import sys
+import threading
import os
import weakref
from contextlib import contextmanager
"""Exception to give up completely."""
+E = sys.monitoring.events
+
+class _MonitoringTracer:
+ EVENT_CALLBACK_MAP = {
+ E.PY_START: 'call',
+ E.PY_RESUME: 'call',
+ E.PY_THROW: 'call',
+ E.LINE: 'line',
+ E.JUMP: 'jump',
+ E.PY_RETURN: 'return',
+ E.PY_YIELD: 'return',
+ E.PY_UNWIND: 'unwind',
+ E.RAISE: 'exception',
+ E.STOP_ITERATION: 'exception',
+ E.INSTRUCTION: 'opcode',
+ }
+
+ GLOBAL_EVENTS = E.PY_START | E.PY_RESUME | E.PY_THROW | E.PY_UNWIND | E.RAISE
+ LOCAL_EVENTS = E.LINE | E.JUMP | E.PY_RETURN | E.PY_YIELD | E.STOP_ITERATION
+
+ def __init__(self):
+ self._tool_id = sys.monitoring.DEBUGGER_ID
+ self._name = 'bdbtracer'
+ self._tracefunc = None
+ self._disable_current_event = False
+ self._tracing_thread = None
+ self._enabled = False
+
+ def start_trace(self, tracefunc):
+ self._tracefunc = tracefunc
+ self._tracing_thread = threading.current_thread()
+ curr_tool = sys.monitoring.get_tool(self._tool_id)
+ if curr_tool is None:
+ sys.monitoring.use_tool_id(self._tool_id, self._name)
+ elif curr_tool == self._name:
+ sys.monitoring.clear_tool_id(self._tool_id)
+ else:
+ raise ValueError('Another debugger is using the monitoring tool')
+ E = sys.monitoring.events
+ all_events = 0
+ for event, cb_name in self.EVENT_CALLBACK_MAP.items():
+ callback = getattr(self, f'{cb_name}_callback')
+ sys.monitoring.register_callback(self._tool_id, event, callback)
+ if event != E.INSTRUCTION:
+ all_events |= event
+ self.check_trace_func()
+ self.check_trace_opcodes()
+ sys.monitoring.set_events(self._tool_id, self.GLOBAL_EVENTS)
+ self._enabled = True
+
+ def stop_trace(self):
+ self._enabled = False
+ self._tracing_thread = None
+ curr_tool = sys.monitoring.get_tool(self._tool_id)
+ if curr_tool != self._name:
+ return
+ sys.monitoring.clear_tool_id(self._tool_id)
+ self.check_trace_opcodes()
+ sys.monitoring.free_tool_id(self._tool_id)
+
+ def disable_current_event(self):
+ self._disable_current_event = True
+
+ def restart_events(self):
+ if sys.monitoring.get_tool(self._tool_id) == self._name:
+ sys.monitoring.restart_events()
+
+ def callback_wrapper(func):
+ import functools
+
+ @functools.wraps(func)
+ def wrapper(self, *args):
+ if self._tracing_thread != threading.current_thread():
+ return
+ try:
+ frame = sys._getframe().f_back
+ ret = func(self, frame, *args)
+ if self._enabled and frame.f_trace:
+ self.check_trace_func()
+ if self._disable_current_event:
+ return sys.monitoring.DISABLE
+ else:
+ return ret
+ except BaseException:
+ self.stop_trace()
+ sys._getframe().f_back.f_trace = None
+ raise
+ finally:
+ self._disable_current_event = False
+
+ return wrapper
+
+ @callback_wrapper
+ def call_callback(self, frame, code, *args):
+ local_tracefunc = self._tracefunc(frame, 'call', None)
+ if local_tracefunc is not None:
+ frame.f_trace = local_tracefunc
+ if self._enabled:
+ sys.monitoring.set_local_events(self._tool_id, code, self.LOCAL_EVENTS)
+
+ @callback_wrapper
+ def return_callback(self, frame, code, offset, retval):
+ if frame.f_trace:
+ frame.f_trace(frame, 'return', retval)
+
+ @callback_wrapper
+ def unwind_callback(self, frame, code, *args):
+ if frame.f_trace:
+ frame.f_trace(frame, 'return', None)
+
+ @callback_wrapper
+ def line_callback(self, frame, code, *args):
+ if frame.f_trace and frame.f_trace_lines:
+ frame.f_trace(frame, 'line', None)
+
+ @callback_wrapper
+ def jump_callback(self, frame, code, inst_offset, dest_offset):
+ if dest_offset > inst_offset:
+ return sys.monitoring.DISABLE
+ inst_lineno = self._get_lineno(code, inst_offset)
+ dest_lineno = self._get_lineno(code, dest_offset)
+ if inst_lineno != dest_lineno:
+ return sys.monitoring.DISABLE
+ if frame.f_trace and frame.f_trace_lines:
+ frame.f_trace(frame, 'line', None)
+
+ @callback_wrapper
+ def exception_callback(self, frame, code, offset, exc):
+ if frame.f_trace:
+ if exc.__traceback__ and hasattr(exc.__traceback__, 'tb_frame'):
+ tb = exc.__traceback__
+ while tb:
+ if tb.tb_frame.f_locals.get('self') is self:
+ return
+ tb = tb.tb_next
+ frame.f_trace(frame, 'exception', (type(exc), exc, exc.__traceback__))
+
+ @callback_wrapper
+ def opcode_callback(self, frame, code, offset):
+ if frame.f_trace and frame.f_trace_opcodes:
+ frame.f_trace(frame, 'opcode', None)
+
+ def check_trace_opcodes(self, frame=None):
+ if frame is None:
+ frame = sys._getframe().f_back
+ while frame is not None:
+ self.set_trace_opcodes(frame, frame.f_trace_opcodes)
+ frame = frame.f_back
+
+ def set_trace_opcodes(self, frame, trace_opcodes):
+ if sys.monitoring.get_tool(self._tool_id) != self._name:
+ return
+ if trace_opcodes:
+ sys.monitoring.set_local_events(self._tool_id, frame.f_code, E.INSTRUCTION)
+ else:
+ sys.monitoring.set_local_events(self._tool_id, frame.f_code, 0)
+
+ def check_trace_func(self, frame=None):
+ if frame is None:
+ frame = sys._getframe().f_back
+ while frame is not None:
+ if frame.f_trace is not None:
+ sys.monitoring.set_local_events(self._tool_id, frame.f_code, self.LOCAL_EVENTS)
+ frame = frame.f_back
+
+ def _get_lineno(self, code, offset):
+ import dis
+ last_lineno = None
+ for start, lineno in dis.findlinestarts(code):
+ if offset < start:
+ return last_lineno
+ last_lineno = lineno
+ return last_lineno
+
+
class Bdb:
"""Generic Python debugger base class.
is determined by the __name__ in the frame globals.
"""
- def __init__(self, skip=None):
+ def __init__(self, skip=None, backend='settrace'):
self.skip = set(skip) if skip else None
self.breaks = {}
self.fncache = {}
self.trace_opcodes = False
self.enterframe = None
self.code_linenos = weakref.WeakKeyDictionary()
+ self.backend = backend
+ if backend == 'monitoring':
+ self.monitoring_tracer = _MonitoringTracer()
+ elif backend == 'settrace':
+ self.monitoring_tracer = None
+ else:
+ raise ValueError(f"Invalid backend '{backend}'")
self._load_breaks()
self.fncache[filename] = canonic
return canonic
+ def start_trace(self):
+ if self.monitoring_tracer:
+ self.monitoring_tracer.start_trace(self.trace_dispatch)
+ else:
+ sys.settrace(self.trace_dispatch)
+
+ def stop_trace(self):
+ if self.monitoring_tracer:
+ self.monitoring_tracer.stop_trace()
+ else:
+ sys.settrace(None)
+
def reset(self):
"""Set values of attributes as ready to start debugging."""
import linecache
"""
if self.stop_here(frame) or self.break_here(frame):
self.user_line(frame)
+ self.restart_events()
if self.quitting: raise BdbQuit
+ elif not self.get_break(frame.f_code.co_filename, frame.f_lineno):
+ self.disable_current_event()
return self.trace_dispatch
def dispatch_call(self, frame, arg):
if self.stopframe and frame.f_code.co_flags & GENERATOR_AND_COROUTINE_FLAGS:
return self.trace_dispatch
self.user_call(frame, arg)
+ self.restart_events()
if self.quitting: raise BdbQuit
return self.trace_dispatch
try:
self.frame_returning = frame
self.user_return(frame, arg)
+ self.restart_events()
finally:
self.frame_returning = None
if self.quitting: raise BdbQuit
if not (frame.f_code.co_flags & GENERATOR_AND_COROUTINE_FLAGS
and arg[0] is StopIteration and arg[2] is None):
self.user_exception(frame, arg)
+ self.restart_events()
if self.quitting: raise BdbQuit
# Stop at the StopIteration or GeneratorExit exception when the user
# has set stopframe in a generator by issuing a return command, or a
and self.stopframe.f_code.co_flags & GENERATOR_AND_COROUTINE_FLAGS
and arg[0] in (StopIteration, GeneratorExit)):
self.user_exception(frame, arg)
+ self.restart_events()
if self.quitting: raise BdbQuit
return self.trace_dispatch
unconditionally.
"""
self.user_opcode(frame)
+ self.restart_events()
if self.quitting: raise BdbQuit
return self.trace_dispatch
frame = self.enterframe
while frame is not None:
frame.f_trace_opcodes = trace_opcodes
+ if self.monitoring_tracer:
+ self.monitoring_tracer.set_trace_opcodes(frame, trace_opcodes)
if frame is self.botframe:
break
frame = frame.f_back
If frame is not specified, debugging starts from caller's frame.
"""
- sys.settrace(None)
+ self.stop_trace()
if frame is None:
frame = sys._getframe().f_back
self.reset()
frame.f_trace_lines = True
frame = frame.f_back
self.set_stepinstr()
- sys.settrace(self.trace_dispatch)
+ self.enterframe = None
+ self.start_trace()
def set_continue(self):
"""Stop only at breakpoints or when finished.
self._set_stopinfo(self.botframe, None, -1)
if not self.breaks:
# no breakpoints; run without debugger overhead
- sys.settrace(None)
+ self.stop_trace()
frame = sys._getframe().f_back
while frame and frame is not self.botframe:
del frame.f_trace
frame = frame.f_back
for frame, (trace_lines, trace_opcodes) in self.frame_trace_lines_opcodes.items():
frame.f_trace_lines, frame.f_trace_opcodes = trace_lines, trace_opcodes
+ if self.backend == 'monitoring':
+ self.monitoring_tracer.set_trace_opcodes(frame, trace_opcodes)
self.frame_trace_lines_opcodes = {}
def set_quit(self):
self.stopframe = self.botframe
self.returnframe = None
self.quitting = True
- sys.settrace(None)
+ self.stop_trace()
# Derived classes and clients can call the following methods
# to manipulate breakpoints. These methods return an
s += f'{lprefix}Warning: lineno is None'
return s
+ def disable_current_event(self):
+ """Disable the current event."""
+ if self.backend == 'monitoring':
+ self.monitoring_tracer.disable_current_event()
+
+ def restart_events(self):
+ """Restart all events."""
+ if self.backend == 'monitoring':
+ self.monitoring_tracer.restart_events()
+
# The following methods can be called by clients to use
# a debugger to debug a statement or an expression.
# Both can be given as a string, or a code object.
self.reset()
if isinstance(cmd, str):
cmd = compile(cmd, "<string>", "exec")
- sys.settrace(self.trace_dispatch)
+ self.start_trace()
try:
exec(cmd, globals, locals)
except BdbQuit:
pass
finally:
self.quitting = True
- sys.settrace(None)
+ self.stop_trace()
def runeval(self, expr, globals=None, locals=None):
"""Debug an expression executed via the eval() function.
if locals is None:
locals = globals
self.reset()
- sys.settrace(self.trace_dispatch)
+ self.start_trace()
try:
return eval(expr, globals, locals)
except BdbQuit:
pass
finally:
self.quitting = True
- sys.settrace(None)
+ self.stop_trace()
def runctx(self, cmd, globals, locals):
"""For backwards-compatibility. Defers to run()."""
Return the result of the function call.
"""
self.reset()
- sys.settrace(self.trace_dispatch)
+ self.start_trace()
res = None
try:
res = func(*args, **kwds)
pass
finally:
self.quitting = True
- sys.settrace(None)
+ self.stop_trace()
return res
pass
__all__ = ["run", "pm", "Pdb", "runeval", "runctx", "runcall", "set_trace",
- "post_mortem", "help"]
+ "post_mortem", "set_default_backend", "get_default_backend", "help"]
def find_first_executable_line(code):
line_prefix = '\n-> ' # Probably a better default
+# The default backend to use for Pdb instances if not specified
+# Should be either 'settrace' or 'monitoring'
+_default_backend = 'settrace'
+
+
+def set_default_backend(backend):
+ """Set the default backend to use for Pdb instances."""
+ global _default_backend
+ if backend not in ('settrace', 'monitoring'):
+ raise ValueError("Invalid backend: %s" % backend)
+ _default_backend = backend
+
+
+def get_default_backend():
+ """Get the default backend to use for Pdb instances."""
+ return _default_backend
+
class Pdb(bdb.Bdb, cmd.Cmd):
_previous_sigint_handler = None
_last_pdb_instance = None
def __init__(self, completekey='tab', stdin=None, stdout=None, skip=None,
- nosigint=False, readrc=True, mode=None):
- bdb.Bdb.__init__(self, skip=skip)
+ nosigint=False, readrc=True, mode=None, backend=None):
+ bdb.Bdb.__init__(self, skip=skip, backend=backend if backend else get_default_backend())
cmd.Cmd.__init__(self, completekey, stdin, stdout)
sys.audit("pdb.Pdb")
if stdout:
if not arg:
self._print_invalid_arg(arg)
return
- sys.settrace(None)
+ self.stop_trace()
globals = self.curframe.f_globals
locals = self.curframe.f_locals
p = Pdb(self.completekey, self.stdin, self.stdout)
except Exception:
self._error_exc()
self.message("LEAVING RECURSIVE DEBUGGER")
- sys.settrace(self.trace_dispatch)
+ self.start_trace()
self.lastcmd = p.lastcmd
complete_debug = _complete_expression
if Pdb._last_pdb_instance is not None:
pdb = Pdb._last_pdb_instance
else:
- pdb = Pdb(mode='inline')
+ pdb = Pdb(mode='inline', backend='monitoring')
if header is not None:
pdb.message(header)
pdb.set_trace(sys._getframe().f_back, commands=commands)
# modified by the script being debugged. It's a bad idea when it was
# changed by the user from the command line. There is a "restart" command
# which allows explicit specification of command line arguments.
- pdb = Pdb(mode='cli')
+ pdb = Pdb(mode='cli', backend='monitoring')
pdb.rcLines.extend(opts.commands)
while True:
try:
4
"""
+def test_pdb_breakpoint_ignore_and_condition():
+ """
+ >>> reset_Breakpoint()
+
+ >>> def test_function():
+ ... import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
+ ... for i in range(5):
+ ... print(i)
+
+ >>> with PdbTestInput([ # doctest: +NORMALIZE_WHITESPACE
+ ... 'break 4',
+ ... 'ignore 1 2', # ignore once
+ ... 'continue',
+ ... 'condition 1 i == 4',
+ ... 'continue',
+ ... 'clear 1',
+ ... 'continue',
+ ... ]):
+ ... test_function()
+ > <doctest test.test_pdb.test_pdb_breakpoint_ignore_and_condition[1]>(2)test_function()
+ -> import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
+ (Pdb) break 4
+ Breakpoint 1 at <doctest test.test_pdb.test_pdb_breakpoint_ignore_and_condition[1]>:4
+ (Pdb) ignore 1 2
+ Will ignore next 2 crossings of breakpoint 1.
+ (Pdb) continue
+ 0
+ 1
+ > <doctest test.test_pdb.test_pdb_breakpoint_ignore_and_condition[1]>(4)test_function()
+ -> print(i)
+ (Pdb) condition 1 i == 4
+ New condition set for breakpoint 1.
+ (Pdb) continue
+ 2
+ 3
+ > <doctest test.test_pdb.test_pdb_breakpoint_ignore_and_condition[1]>(4)test_function()
+ -> print(i)
+ (Pdb) clear 1
+ Deleted breakpoint 1 at <doctest test.test_pdb.test_pdb_breakpoint_ignore_and_condition[1]>:4
+ (Pdb) continue
+ 4
+ """
+
def test_pdb_breakpoint_on_annotated_function_def():
"""Test breakpoints on function definitions with annotation.
(Pdb) continue
"""
+def test_pdb_breakpoint_on_disabled_line():
+ """New breakpoint on once disabled line should work
+
+ >>> reset_Breakpoint()
+ >>> def test_function():
+ ... import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
+ ... for i in range(3):
+ ... j = i * 2
+ ... print(j)
+
+ >>> with PdbTestInput([ # doctest: +NORMALIZE_WHITESPACE
+ ... 'break 5',
+ ... 'c',
+ ... 'clear 1',
+ ... 'break 4',
+ ... 'c',
+ ... 'clear 2',
+ ... 'c'
+ ... ]):
+ ... test_function()
+ > <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>(2)test_function()
+ -> import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
+ (Pdb) break 5
+ Breakpoint 1 at <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>:5
+ (Pdb) c
+ > <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>(5)test_function()
+ -> print(j)
+ (Pdb) clear 1
+ Deleted breakpoint 1 at <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>:5
+ (Pdb) break 4
+ Breakpoint 2 at <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>:4
+ (Pdb) c
+ 0
+ > <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>(4)test_function()
+ -> j = i * 2
+ (Pdb) clear 2
+ Deleted breakpoint 2 at <doctest test.test_pdb.test_pdb_breakpoint_on_disabled_line[1]>:4
+ (Pdb) c
+ 2
+ 4
+ """
+
def test_pdb_breakpoints_preserved_across_interactive_sessions():
"""Breakpoints are remembered between interactive sessions
def load_tests(loader, tests, pattern):
from test import test_pdb
- tests.addTest(doctest.DocTestSuite(test_pdb))
+ def setUpPdbBackend(backend):
+ def setUp(test):
+ import pdb
+ pdb.set_default_backend(backend)
+ return setUp
+ tests.addTest(doctest.DocTestSuite(test_pdb, setUp=setUpPdbBackend('monitoring')))
+ tests.addTest(doctest.DocTestSuite(test_pdb, setUp=setUpPdbBackend('settrace')))
return tests