| # Copyright 2022-2025 Free Software Foundation, Inc. |
| |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation; either version 3 of the License, or |
| # (at your option) any later version. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License |
| # along with this program. If not, see <http://www.gnu.org/licenses/>. |
| |
| import re |
| from contextlib import contextmanager |
| |
| # These are deprecated in 3.9, but required in older versions. |
| from typing import Optional, Sequence |
| |
| import gdb |
| |
| from .server import capability, export_line, import_line, request, send_event |
| from .sources import make_source |
| from .startup import ( |
| DAPException, |
| LogLevel, |
| exec_mi_and_log, |
| in_gdb_thread, |
| log_stack, |
| parse_and_eval, |
| ) |
| from .typecheck import type_check |
| |
| # True when suppressing new breakpoint events. |
| _suppress_bp = False |
| |
| |
| @contextmanager |
| def suppress_new_breakpoint_event(): |
| """Return a new context manager that suppresses new breakpoint events.""" |
| global _suppress_bp |
| saved = _suppress_bp |
| _suppress_bp = True |
| try: |
| yield None |
| finally: |
| _suppress_bp = saved |
| |
| |
| @in_gdb_thread |
| def _bp_modified(event): |
| if not _suppress_bp: |
| send_event( |
| "breakpoint", |
| { |
| "reason": "changed", |
| "breakpoint": _breakpoint_descriptor(event), |
| }, |
| ) |
| |
| |
| @in_gdb_thread |
| def _bp_created(event): |
| if not _suppress_bp: |
| send_event( |
| "breakpoint", |
| { |
| "reason": "new", |
| "breakpoint": _breakpoint_descriptor(event), |
| }, |
| ) |
| |
| |
| @in_gdb_thread |
| def _bp_deleted(event): |
| if not _suppress_bp: |
| send_event( |
| "breakpoint", |
| { |
| "reason": "removed", |
| "breakpoint": _breakpoint_descriptor(event), |
| }, |
| ) |
| |
| |
| gdb.events.breakpoint_created.connect(_bp_created) |
| gdb.events.breakpoint_modified.connect(_bp_modified) |
| gdb.events.breakpoint_deleted.connect(_bp_deleted) |
| |
| |
| # Map from the breakpoint "kind" (like "function") to a second map, of |
| # breakpoints of that type. The second map uses the breakpoint spec |
| # as a key, and the gdb.Breakpoint itself as a value. This is used to |
| # implement the clearing behavior specified by the protocol, while |
| # allowing for reuse when a breakpoint can be kept. |
| breakpoint_map = {} |
| |
| |
| @in_gdb_thread |
| def _breakpoint_descriptor(bp): |
| "Return the Breakpoint object descriptor given a gdb Breakpoint." |
| # If there are no objfiles (that is, before the launch request), |
| # we consider all breakpoints to be pending. This is done to work |
| # around the gdb oddity that setting a breakpoint by address will |
| # always succeed. |
| pending = bp.pending or len(gdb.objfiles()) == 0 |
| result = { |
| "id": bp.number, |
| "verified": not pending, |
| } |
| if pending: |
| result["reason"] = "pending" |
| if bp.locations: |
| # Just choose the first location, because DAP doesn't allow |
| # multiple locations. See |
| # https://github.com/microsoft/debug-adapter-protocol/issues/13 |
| loc = bp.locations[0] |
| if loc.source: |
| (filename, line) = loc.source |
| if loc.fullname is not None: |
| filename = loc.fullname |
| |
| result.update( |
| { |
| "source": make_source(filename), |
| "line": export_line(line), |
| } |
| ) |
| |
| if loc.address: |
| result["instructionReference"] = hex(loc.address) |
| |
| return result |
| |
| |
| # Extract entries from a hash table and return a list of them. Each |
| # entry is a string. If a key of that name appears in the hash table, |
| # it is removed and pushed on the result list; if it does not appear, |
| # None is pushed on the list. |
| def _remove_entries(table, *names): |
| return [table.pop(name, None) for name in names] |
| |
| |
| # Helper function to set some breakpoints according to a list of |
| # specifications and a callback function to do the work of creating |
| # the breakpoint. |
| @in_gdb_thread |
| def _set_breakpoints_callback(kind, specs, creator): |
| # Try to reuse existing breakpoints if possible. |
| if kind in breakpoint_map: |
| saved_map = breakpoint_map[kind] |
| else: |
| saved_map = {} |
| breakpoint_map[kind] = {} |
| result = [] |
| with suppress_new_breakpoint_event(): |
| for spec in specs: |
| # It makes sense to reuse a breakpoint even if the condition |
| # or ignore count differs, so remove these entries from the |
| # spec first. |
| (condition, hit_condition) = _remove_entries( |
| spec, "condition", "hitCondition" |
| ) |
| keyspec = frozenset(spec.items()) |
| |
| # Create or reuse a breakpoint. If asked, set the condition |
| # or the ignore count. Catch errors coming from gdb and |
| # report these as an "unverified" breakpoint. |
| bp = None |
| try: |
| if keyspec in saved_map: |
| bp = saved_map.pop(keyspec) |
| else: |
| bp = creator(**spec) |
| |
| bp.condition = condition |
| if hit_condition is None: |
| bp.ignore_count = 0 |
| else: |
| bp.ignore_count = int( |
| parse_and_eval(hit_condition, global_context=True) |
| ) |
| |
| # Reaching this spot means success. |
| breakpoint_map[kind][keyspec] = bp |
| result.append(_breakpoint_descriptor(bp)) |
| # Exceptions other than gdb.error are possible here. |
| except Exception as e: |
| # Don't normally want to see this, as it interferes with |
| # the test suite. |
| log_stack(LogLevel.FULL) |
| # Maybe the breakpoint was made but setting an attribute |
| # failed. We still want this to fail. |
| if bp is not None: |
| bp.delete() |
| # Breakpoint creation failed. |
| result.append( |
| { |
| "verified": False, |
| "reason": "failed", |
| "message": str(e), |
| } |
| ) |
| |
| # Delete any breakpoints that were not reused. |
| for entry in saved_map.values(): |
| entry.delete() |
| return result |
| |
| |
| class _PrintBreakpoint(gdb.Breakpoint): |
| def __init__(self, logMessage, **args): |
| super().__init__(**args) |
| # Split the message up for easier processing. |
| self._message = re.split("{(.*?)}", logMessage) |
| |
| def stop(self): |
| output = "" |
| for idx, item in enumerate(self._message): |
| if idx % 2 == 0: |
| # Even indices are plain text. |
| output += item |
| else: |
| # Odd indices are expressions to substitute. The {} |
| # have already been stripped by the placement of the |
| # regex capture in the 'split' call. |
| try: |
| # No real need to use the DAP parse_and_eval here. |
| val = gdb.parse_and_eval(item) |
| output += str(val) |
| except Exception as e: |
| output += "<" + str(e) + ">" |
| send_event( |
| "output", |
| { |
| "category": "console", |
| "output": output, |
| }, |
| ) |
| # Do not stop. |
| return False |
| |
| |
| # Set a single breakpoint or a log point. Returns the new breakpoint. |
| # Note that not every spec will pass logMessage, so here we use a |
| # default. |
| @in_gdb_thread |
| def _set_one_breakpoint(*, logMessage=None, **args): |
| if logMessage is not None: |
| return _PrintBreakpoint(logMessage, **args) |
| else: |
| return gdb.Breakpoint(**args) |
| |
| |
| # Helper function to set ordinary breakpoints according to a list of |
| # specifications. |
| @in_gdb_thread |
| def _set_breakpoints(kind, specs): |
| return _set_breakpoints_callback(kind, specs, _set_one_breakpoint) |
| |
| |
| # A helper function that rewrites a SourceBreakpoint into the internal |
| # form passed to the creator. This function also allows for |
| # type-checking of each SourceBreakpoint. |
| @type_check |
| def _rewrite_src_breakpoint( |
| *, |
| # This is a Source but we don't type-check it. |
| source, |
| line: int, |
| condition: Optional[str] = None, |
| hitCondition: Optional[str] = None, |
| logMessage: Optional[str] = None, |
| **args, |
| ): |
| return { |
| "source": source["path"], |
| "line": import_line(line), |
| "condition": condition, |
| "hitCondition": hitCondition, |
| "logMessage": logMessage, |
| } |
| |
| |
| @request("setBreakpoints", expect_stopped=False) |
| @capability("supportsHitConditionalBreakpoints") |
| @capability("supportsConditionalBreakpoints") |
| @capability("supportsLogPoints") |
| def set_breakpoint(*, source, breakpoints: Sequence = (), **args): |
| if "path" not in source: |
| result = [] |
| else: |
| # Setting 'source' in BP avoids any Python error if BP already |
| # has a 'source' parameter. Setting this isn't in the spec, |
| # but it is better to be safe. See PR dap/30820. |
| specs = [] |
| for bp in breakpoints: |
| bp["source"] = source |
| specs.append(_rewrite_src_breakpoint(**bp)) |
| # Be sure to include the path in the key, so that we only |
| # clear out breakpoints coming from this same source. |
| key = "source:" + source["path"] |
| result = _set_breakpoints(key, specs) |
| return { |
| "breakpoints": result, |
| } |
| |
| |
| # A helper function that rewrites a FunctionBreakpoint into the |
| # internal form passed to the creator. This function also allows for |
| # type-checking of each FunctionBreakpoint. |
| @type_check |
| def _rewrite_fn_breakpoint( |
| *, |
| name: str, |
| condition: Optional[str] = None, |
| hitCondition: Optional[str] = None, |
| **args, |
| ): |
| return { |
| "function": name, |
| "condition": condition, |
| "hitCondition": hitCondition, |
| } |
| |
| |
| @request("setFunctionBreakpoints", expect_stopped=False) |
| @capability("supportsFunctionBreakpoints") |
| def set_fn_breakpoint(*, breakpoints: Sequence, **args): |
| specs = [_rewrite_fn_breakpoint(**bp) for bp in breakpoints] |
| return { |
| "breakpoints": _set_breakpoints("function", specs), |
| } |
| |
| |
| # A helper function that rewrites an InstructionBreakpoint into the |
| # internal form passed to the creator. This function also allows for |
| # type-checking of each InstructionBreakpoint. |
| @type_check |
| def _rewrite_insn_breakpoint( |
| *, |
| instructionReference: str, |
| offset: Optional[int] = None, |
| condition: Optional[str] = None, |
| hitCondition: Optional[str] = None, |
| **args, |
| ): |
| # There's no way to set an explicit address breakpoint from |
| # Python, so we rely on "spec" instead. |
| val = "*" + instructionReference |
| if offset is not None: |
| val = val + " + " + str(offset) |
| return { |
| "spec": val, |
| "condition": condition, |
| "hitCondition": hitCondition, |
| } |
| |
| |
| @request("setInstructionBreakpoints", expect_stopped=False) |
| @capability("supportsInstructionBreakpoints") |
| def set_insn_breakpoints( |
| *, breakpoints: Sequence, offset: Optional[int] = None, **args |
| ): |
| specs = [_rewrite_insn_breakpoint(**bp) for bp in breakpoints] |
| return { |
| "breakpoints": _set_breakpoints("instruction", specs), |
| } |
| |
| |
| @in_gdb_thread |
| def _catch_exception(filterId, **args): |
| if filterId in ("assert", "exception", "throw", "rethrow", "catch"): |
| cmd = "-catch-" + filterId |
| else: |
| raise DAPException("Invalid exception filterID: " + str(filterId)) |
| result = exec_mi_and_log(cmd) |
| # While the Ada catchpoints emit a "bkptno" field here, the C++ |
| # ones do not. So, instead we look at the "number" field. |
| num = result["bkpt"]["number"] |
| # A little lame that there's no more direct way. |
| for bp in gdb.breakpoints(): |
| if bp.number == num: |
| return bp |
| # Not a DAPException because this is definitely unexpected. |
| raise Exception("Could not find catchpoint after creating") |
| |
| |
| @in_gdb_thread |
| def _set_exception_catchpoints(filter_options): |
| return _set_breakpoints_callback("exception", filter_options, _catch_exception) |
| |
| |
| # A helper function that rewrites an ExceptionFilterOptions into the |
| # internal form passed to the creator. This function also allows for |
| # type-checking of each ExceptionFilterOptions. |
| @type_check |
| def _rewrite_exception_breakpoint( |
| *, |
| filterId: str, |
| condition: Optional[str] = None, |
| # Note that exception breakpoints do not support a hit count. |
| **args, |
| ): |
| return { |
| "filterId": filterId, |
| "condition": condition, |
| } |
| |
| |
| @request("setExceptionBreakpoints", expect_stopped=False) |
| @capability("supportsExceptionFilterOptions") |
| @capability( |
| "exceptionBreakpointFilters", |
| ( |
| { |
| "filter": "assert", |
| "label": "Ada assertions", |
| "supportsCondition": True, |
| }, |
| { |
| "filter": "exception", |
| "label": "Ada exceptions", |
| "supportsCondition": True, |
| }, |
| { |
| "filter": "throw", |
| "label": "C++ exceptions, when thrown", |
| "supportsCondition": True, |
| }, |
| { |
| "filter": "rethrow", |
| "label": "C++ exceptions, when re-thrown", |
| "supportsCondition": True, |
| }, |
| { |
| "filter": "catch", |
| "label": "C++ exceptions, when caught", |
| "supportsCondition": True, |
| }, |
| ), |
| ) |
| def set_exception_breakpoints( |
| *, filters: Sequence[str], filterOptions: Sequence = (), **args |
| ): |
| # Convert the 'filters' to the filter-options style. |
| options = [{"filterId": filter} for filter in filters] |
| options.extend(filterOptions) |
| options = [_rewrite_exception_breakpoint(**bp) for bp in options] |
| return { |
| "breakpoints": _set_exception_catchpoints(options), |
| } |