diff --git a/pyocd/debug/breakpoints/manager.py b/pyocd/debug/breakpoints/manager.py index 13c816022..df012fa0a 100644 --- a/pyocd/debug/breakpoints/manager.py +++ b/pyocd/debug/breakpoints/manager.py @@ -1,6 +1,7 @@ # pyOCD debugger # Copyright (c) 2015-2019,2025 Arm Limited # Copyright (c) 2021 Chris Reed +# Copyright (c) 2026 j4rvisstant # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -20,6 +21,7 @@ from typing import (Dict, List, TYPE_CHECKING, Iterable, MutableSequence, Optional, Sequence, Tuple) from .provider import Breakpoint +from ...core import exceptions from ...core.target import Target if TYPE_CHECKING: @@ -46,6 +48,9 @@ class BreakpointManager: add/remove request is recorded for later. Then, before the target is stepped or resumed, the manager flushes breakpoint changes to the target. It is at this point when it decides which provider to use for each new breakpoint. + + A caller that must guarantee a removal has reached the target before it returns control, for + instance because the session may end without a further step or resume, calls flush_removals(). """ ## Number of hardware breakpoints to try to keep available. @@ -275,6 +280,50 @@ def flush(self, is_step: bool = False) -> None: finally: self._ignore_notifications = False + def flush_removals(self) -> None: + """@brief Physically remove breakpoints whose removal is pending. + + Only removals are committed; pending additions stay deferred until flush(). Nothing is + done unless the core is halted, since restoring memory or clearing a comparator under a + running core is not safe; such a removal is committed by the next flush() instead. + + After this returns, the owning provider has executed the removal and the probe's queued + transfers have been drained, so the change cannot still be sitting in the host queue. + + @exception Error A removal could not be committed. If a provider fails to remove a + breakpoint, that breakpoint stays pending and the next flush() asks the provider + again. If draining the queued transfers fails, the removals already handed to the + providers are not retried. + """ + _, removed = self._get_updated_breakpoints() + if not removed: + return + if not self._core.is_halted(): + LOG.debug("core not halted; deferring removal of %s", removed) + return + + try: + # Ignore any notifications while we modify breakpoints. + self._ignore_notifications = True + + LOG.debug("flushing removals=%s", removed) + for bp in removed: + assert bp.provider is not None + bp.provider.remove_breakpoint(bp) + + # A provider may report failure by keeping the breakpoint rather than raising. + if bp.provider.find_breakpoint(bp.addr) is not None: + raise exceptions.DebugError("failed to remove %s breakpoint at 0x%08x" + % (bp.type.name, bp.addr)) + + del self._breakpoints[bp.addr] + + # Flush all providers, then make sure the removals are really on the wire. + self._flush_all() + self._core.flush() + finally: + self._ignore_notifications = False + def get_breakpoint_type(self, addr: int) -> Optional[Target.BreakpointType]: bp = self.find_breakpoint(addr) return bp.type if (bp is not None) else None diff --git a/pyocd/gdbserver/gdbserver.py b/pyocd/gdbserver/gdbserver.py index 7d3aa85cf..b8e47d8d6 100644 --- a/pyocd/gdbserver/gdbserver.py +++ b/pyocd/gdbserver/gdbserver.py @@ -2,6 +2,7 @@ # Copyright (c) 2006-2020,2025-2026 Arm Limited # Copyright (c) 2021-2022 Chris Reed # Copyright (c) 2022 Clay McClure +# Copyright (c) 2026 j4rvisstant # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -963,6 +964,8 @@ def breakpoint(self, client, data): else: LOG.debug("Command: Clear software breakpoint (addr=0x%08x)", addr) self.target.remove_breakpoint(addr) + if not self._commit_breakpoint_removal(addr): + return self.create_rsp_packet(b'E01') return self.create_rsp_packet(b"OK") # handle hardware breakpoint Z1/z1 @@ -975,6 +978,8 @@ def breakpoint(self, client, data): else: LOG.debug("Command: Clear hardware breakpoint (addr=0x%08x)", addr) self.target.remove_breakpoint(addr) + if not self._commit_breakpoint_removal(addr): + return self.create_rsp_packet(b'E01') return self.create_rsp_packet(b"OK") # handle hardware watchpoint Z2/z2/Z3/z3/Z4/z4 @@ -1008,6 +1013,27 @@ def breakpoint(self, client, data): self.target.remove_watchpoint(addr, size, watchpoint_type) return self.create_rsp_packet(b"OK") + def _commit_breakpoint_removal(self, addr: int) -> bool: + """@brief Commit pending breakpoint removals to the target before acknowledging gdb. + + A removal that is acknowledged with OK while the core is halted must already have reached + the target, so a session that ends without a further resume or step cannot leave a deleted + breakpoint armed. Pending additions stay deferred until the next resume or step. A removal + that arrives while the core is running (non-stop mode) stays pending, as before. + + @return True if OK may be sent, False if the removal failed and E01 must be sent. + """ + bp_manager = getattr(self.target, 'bp_manager', None) + if bp_manager is None: + return True + try: + bp_manager.flush_removals() + return True + except exceptions.Error as e: + LOG.error("Command: Clear breakpoint (addr=0x%08x): failed to commit removal to target: %s", + addr, e, exc_info=self.session.log_tracebacks) + return False + def set_thread(self, client, data): op = data[0:1] thread_id = int(data[1:-3], 16) diff --git a/test/unit/mock_breakpoints.py b/test/unit/mock_breakpoints.py new file mode 100644 index 000000000..9bac1b7a5 --- /dev/null +++ b/test/unit/mock_breakpoints.py @@ -0,0 +1,294 @@ +# pyOCD debugger +# Copyright (c) 2026 j4rvisstant +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""@brief Hardware-free doubles for the breakpoint removal tests. + +The doubles are deliberately thin: the real BreakpointManager, the real SoftwareBreakpointProvider +and the real FPB are exercised over a fake core and a dict-backed fake access port, so the tests +can assert on actual comparator values and on the actual halfword in ram. + +The fake access port and the fake core append to one shared trace, so a test can assert the order +in which the target was touched, not just the end state. + +This module is intentionally not named test_* so pytest does not collect it. +""" + +import threading +import types +from typing import (Any, Dict, List, Optional, Tuple) + +from pyocd.core import exceptions +from pyocd.core.memory_interface import MemoryInterface +from pyocd.core.target import Target +from pyocd.coresight.fpb import FPB +from pyocd.debug.breakpoints.manager import BreakpointManager +from pyocd.debug.breakpoints.software import SoftwareBreakpointProvider +from pyocd.gdbserver.gdbserver import GDBServer +from pyocd.utility.notification import (Notification, Notifier) + +from .mockcore import MockCore + +## Base address used for the FPB in the fake access port. +FPB_BASE = 0xE0002000 +FP_CTRL_ADDR = FPB_BASE + FPB.FP_CTRL +FP_COMP0_ADDR = FPB_BASE + FPB.FP_COMP0 +FP_COMP1_ADDR = FP_COMP0_ADDR + 4 + +## FP_CTRL value presented by the fake access port. +# +# REV field 1 selects FPBv2, so ram addresses are hardware-breakpoint capable; NUM_CODE is 2 and +# NUM_LIT is 0. +FP_CTRL_REV2_TWO_COMPARATORS = 0x10000020 + +## Addresses in the MockCore flash region (0x0-0x3ff, not writable, so hardware breakpoints only). +FLASH_A = 0x100 +FLASH_B = 0x200 +FLASH_C = 0x300 + +## Address in the MockCore ram region (writable, so software breakpoints are possible). +RAM_R = 0x20000010 + +## Comparator values the FPBv2 encoding produces for the addresses above. +COMP_FLASH_A = (FLASH_A & 0xfffffffe) | 1 +COMP_FLASH_B = (FLASH_B & 0xfffffffe) | 1 +COMP_RAM_R = (RAM_R & 0xfffffffe) | 1 + +## Instruction placed in ram underneath a software breakpoint ('bx lr'). +ORIGINAL_INSTR = 0x4770 + +## The instruction a software breakpoint writes over it. +BKPT_INSTR = SoftwareBreakpointProvider.BKPT_INSTR + +## Expected gdb remote protocol replies. +OK_REPLY = b'$OK#9a' +E01_REPLY = b'$E01#a6' + +## Operation names used in the shared trace. An entry is a tuple of the name and its arguments. +OP_AP_WRITE = 'ap_write' +OP_CORE_WRITE = 'core_write' +OP_INVALIDATE = 'invalidate' +OP_FLUSH = 'flush' + +## One entry of the shared trace: the operation name followed by its arguments. +TraceEntry = Tuple[Any, ...] + + +class FakeAP(MemoryInterface): + """@brief Dict-backed access port, so the real FPB can be driven without hardware. + + Only single-location accesses are implemented; that is all the FPB needs. The read32/write32 + shorthands of MemoryInterface work unchanged on top of them. + """ + + def __init__(self, trace: Optional[List[TraceEntry]] = None) -> None: + self.mem: Dict[int, int] = {FP_CTRL_ADDR: FP_CTRL_REV2_TWO_COMPARATORS} + self.writes: List[Tuple[int, int]] = [] + self.fail_writes: bool = False + self.trace: List[TraceEntry] = trace if (trace is not None) else [] + + def read_memory(self, addr: int, transfer_size: int = 32, now: bool = True) -> int: + return self.mem.get(addr, 0) + + def write_memory(self, addr: int, value: int, transfer_size: int = 32) -> None: + if self.fail_writes: + raise exceptions.TransferError("fake AP write failure at 0x%08x" % addr) + self.mem[addr] = value + self.writes.append((addr, value)) + self.trace.append((OP_AP_WRITE, addr, value)) + + def comp_writes(self) -> List[Tuple[int, int]]: + """@brief Recorded writes that target an FPB comparator register.""" + return [w for w in self.writes if w[0] in (FP_COMP0_ADDR, FP_COMP1_ADDR)] + + +class BreakpointTestCore(MockCore): + """@brief MockCore extended with what BreakpointManager and the sw provider require. + + MockCore has the memory map and byte-backed memory, but no session to subscribe to, no + settable run state, no instruction cache maintenance and no flush(). + """ + + def __init__(self, trace: Optional[List[TraceEntry]] = None) -> None: + super().__init__() + self.session = Notifier() + self.state: Target.State = Target.State.HALTED + self.has_cache: bool = False + self.invalidated: List[Optional[int]] = [] + self.flush_calls: int = 0 + self.fail_writes: bool = False + self.fail_flush: bool = False + self.pre_run_events: List[Target.RunType] = [] + self.trace: List[TraceEntry] = trace if (trace is not None) else [] + self.session.subscribe(self._record_pre_run, Target.Event.PRE_RUN) + + def _record_pre_run(self, notification: Notification) -> None: + self.pre_run_events.append(notification.data) + + def get_state(self) -> Target.State: + return self.state + + def is_halted(self) -> bool: + return self.state == Target.State.HALTED + + def is_running(self) -> bool: + return self.state == Target.State.RUNNING + + def invalidate_instruction_cache(self, address: Optional[int] = None) -> None: + self.invalidated.append(address) + self.trace.append((OP_INVALIDATE, address)) + + def flush(self) -> None: + """@brief Drain the probe's queued transfers, as CortexM.flush() does. + + With fail_flush set the drain raises, which is how a transport error that only surfaces + once deferred transfers are actually sent is simulated. The call is traced before it + fails, so a test can tell an attempted drain from a missing one. + """ + self.flush_calls += 1 + self.trace.append((OP_FLUSH,)) + if self.fail_flush: + raise exceptions.TransferError("fake core flush failure") + + def write_memory(self, addr: int, value: int, transfer_size: int = 32) -> None: + if self.fail_writes: + raise exceptions.TransferError("fake core write failure at 0x%08x" % addr) + self.trace.append((OP_CORE_WRITE, addr, value)) + super().write_memory(addr, value, transfer_size) + + def halt(self) -> None: + self.state = Target.State.HALTED + + def resume(self) -> None: + """@brief Send PRE_RUN and start running. + + Mirrors CortexM.resume(), which returns without sending PRE_RUN when the core is not + halted. The fake deliberately does not call flush(), so flush_calls counts only the + flushes requested by the code under test. + """ + if self.state != Target.State.HALTED: + return + self.session.notify(Target.Event.PRE_RUN, self, Target.RunType.RESUME) + self.state = Target.State.RUNNING + + def step(self) -> None: + """@brief Send PRE_RUN for a single step; the core stays halted afterwards.""" + self.session.notify(Target.Event.PRE_RUN, self, Target.RunType.STEP) + + +class BreakpointHarness: + """@brief The fake core plus the real manager and providers wired together.""" + + def __init__( + self, + core: BreakpointTestCore, + manager: BreakpointManager, + sw_provider: SoftwareBreakpointProvider, + fpb: FPB, + ap: FakeAP, + trace: List[TraceEntry], + ) -> None: + self.core = core + self.manager = manager + self.sw_provider = sw_provider + self.fpb = fpb + self.ap = ap + ## Access port writes, core writes, cache invalidations and flushes in the order they + ## happened. Shared by the fake access port and the fake core. + self.trace = trace + + +def make_harness() -> BreakpointHarness: + """@brief Build a harness wired like CortexM does it.""" + trace: List[TraceEntry] = [] + core = BreakpointTestCore(trace) + manager = BreakpointManager(core) + sw_provider = SoftwareBreakpointProvider(core) + sw_provider.init() + manager.add_provider(sw_provider) + ap = FakeAP(trace) + fpb = FPB(ap, addr=FPB_BASE) + fpb.init() + manager.add_provider(fpb) + # Discard the writes made by FPB.init(): the FP_CTRL disable plus both comparators. + ap.writes.clear() + trace.clear() + return BreakpointHarness(core, manager, sw_provider, fpb, ap, trace) + + +class FakeGdbTarget: + """@brief Stand-in for the CortexM that the gdbserver handler talks to. + + The breakpoint methods are pure delegations to the breakpoint manager, exactly as in + CortexM. + """ + + def __init__(self, harness: BreakpointHarness) -> None: + self.core = harness.core + self.bp_manager = harness.manager + + def set_breakpoint(self, addr: int, type: Target.BreakpointType = Target.BreakpointType.AUTO) -> bool: + return self.bp_manager.set_breakpoint(addr, type) + + def remove_breakpoint(self, addr: int) -> None: + self.bp_manager.remove_breakpoint(addr) + + def get_breakpoint_type(self, addr: int) -> Optional[Target.BreakpointType]: + return self.bp_manager.get_breakpoint_type(addr) + + def get_state(self) -> Target.State: + return self.core.get_state() + + def is_halted(self) -> bool: + return self.core.is_halted() + + def resume(self) -> None: + self.core.resume() + + def halt(self) -> None: + self.core.halt() + + def flush(self) -> None: + self.core.flush() + + +class ServerUnderTest(GDBServer): + """@brief GDBServer carrying just enough state to run the breakpoint packet handler. + + GDBServer.__init__() needs a full session and binds a listening socket, so it is not called. + Only the Thread base class is initialised, which keeps this object a usable Thread (pytest + reads its repr when an assertion fails) without starting anything. + """ + + def __init__(self, target: FakeGdbTarget, soft_bkpt_as_hard: bool = False) -> None: + threading.Thread.__init__(self, daemon=True) + self.target = target + self.soft_bkpt_as_hard = soft_bkpt_as_hard + self.session = types.SimpleNamespace(log_tracebacks=False) + self.lock = threading.RLock() + + +def make_server(harness: BreakpointHarness, soft_bkpt_as_hard: bool = False) -> ServerUnderTest: + """@brief Build a socket-less gdbserver over the given harness.""" + return ServerUnderTest(FakeGdbTarget(harness), soft_bkpt_as_hard) + + +def bp_packet(command: bytes, addr: int, kind: int = 2) -> bytes: + """@brief Build the payload that GDBServer.breakpoint() receives for a z/Z packet. + + The handler is passed msg[1:] and parses it with data.split(b'#')[0].split(b','), so the + checksum is not validated and a placeholder is good enough. + """ + return command + b',' + (b'%x' % addr) + b',' + (b'%d' % kind) + b'#00' diff --git a/test/unit/test_breakpoint_manager.py b/test/unit/test_breakpoint_manager.py new file mode 100644 index 000000000..95d414098 --- /dev/null +++ b/test/unit/test_breakpoint_manager.py @@ -0,0 +1,370 @@ +# pyOCD debugger +# Copyright (c) 2026 j4rvisstant +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""@brief Tests for committing pending breakpoint removals to the target. + +A removal that has been acknowledged must be physically committed before control returns, so +that a session which ends without a further resume or step leaves no breakpoint armed in the +target. Pending additions must stay deferred until the next flush(). +""" + +import pytest + +from pyocd.core import exceptions +from pyocd.core.target import Target +from pyocd.debug.breakpoints.manager import UnrealizedBreakpoint + +from .mock_breakpoints import ( + BKPT_INSTR, + BreakpointHarness, + COMP_FLASH_A, + COMP_FLASH_B, + COMP_RAM_R, + FLASH_A, + FLASH_B, + FLASH_C, + FP_COMP0_ADDR, + FP_COMP1_ADDR, + OP_AP_WRITE, + OP_CORE_WRITE, + OP_FLUSH, + OP_INVALIDATE, + ORIGINAL_INSTR, + RAM_R, + make_harness, +) + +HW = Target.BreakpointType.HW +SW = Target.BreakpointType.SW + + +@pytest.fixture +def harness() -> BreakpointHarness: + return make_harness() + + +class TestBreakpointManagerRemovals: + def test_hw_removal_committed_before_next_run(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + core.resume() + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert fpb.num_hw_breakpoint_used == 1 + + core.halt() + manager.remove_breakpoint(FLASH_A) + events_before = list(core.pre_run_events) + harness.trace.clear() + + manager.flush_removals() + + assert ap.mem[FP_COMP0_ADDR] == 0 + assert fpb.available_breakpoints == 2 + assert fpb.find_breakpoint(FLASH_A) is None + assert FLASH_A not in manager.get_breakpoints() + assert manager.find_breakpoint(FLASH_A) is None + assert core.flush_calls == 1 + # The comparator is cleared first and the probe queue is drained afterwards, so the + # write cannot still be sitting in the host queue when the commit returns. + assert harness.trace == [ + (OP_AP_WRITE, FP_COMP0_ADDR, 0), + (OP_FLUSH,), + ] + # The commit must not resume or step the core. + assert core.pre_run_events == events_before + + def test_sw_removal_committed_before_next_run(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + sw_provider = harness.sw_provider + + core.write16(RAM_R, ORIGINAL_INSTR) + assert manager.set_breakpoint(RAM_R, SW) is True + core.resume() + assert core.read16(RAM_R) == BKPT_INSTR + assert sw_provider.find_breakpoint(RAM_R) is not None + + core.halt() + manager.remove_breakpoint(RAM_R) + core.invalidated.clear() + harness.trace.clear() + + manager.flush_removals() + + assert core.read16(RAM_R) == ORIGINAL_INSTR + assert core.invalidated == [RAM_R] + assert sw_provider.find_breakpoint(RAM_R) is None + assert RAM_R not in manager.get_breakpoints() + assert core.flush_calls == 1 + # The halfword is restored and the cache invalidated first, the probe queue drained last. + assert harness.trace == [ + (OP_CORE_WRITE, RAM_R, ORIGINAL_INSTR), + (OP_INVALIDATE, RAM_R), + (OP_FLUSH,), + ] + # A software breakpoint must not touch a comparator. + assert ap.comp_writes() == [] + + def test_soft_bkpt_as_hard_removal_frees_comparator(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + + core.write16(RAM_R, ORIGINAL_INSTR) + # An FPBv2 accepts ram addresses, which is what soft_bkpt_as_hard relies on. + assert manager.set_breakpoint(RAM_R, HW) is True + core.resume() + assert ap.mem[FP_COMP0_ADDR] == COMP_RAM_R + assert core.read16(RAM_R) == ORIGINAL_INSTR + assert harness.sw_provider.find_breakpoint(RAM_R) is None + + core.halt() + manager.remove_breakpoint(RAM_R) + + manager.flush_removals() + + assert ap.mem[FP_COMP0_ADDR] == 0 + assert fpb.available_breakpoints == 2 + assert core.read16(RAM_R) == ORIGINAL_INSTR + assert RAM_R not in manager.get_breakpoints() + assert core.flush_calls == 1 + + def test_stop_resume_churn_installs_once(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + core.resume() + core.halt() + manager.remove_breakpoint(FLASH_A) + + manager.flush_removals() + + assert ap.mem[FP_COMP0_ADDR] == 0 + assert fpb.num_hw_breakpoint_used == 0 + + # gdb reinserts the breakpoint before the next resume; that addition stays deferred. + assert manager.set_breakpoint(FLASH_A, HW) is True + assert len(ap.comp_writes()) == 2 + core.resume() + + enabled = [bp for bp in fpb.hw_breakpoints if bp.enabled] + assert len(enabled) == 1 + assert enabled[0].addr == FLASH_A + assert fpb.num_hw_breakpoint_used == 1 + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert ap.mem[FP_COMP1_ADDR] == 0 + assert ap.comp_writes() == [ + (FP_COMP0_ADDR, COMP_FLASH_A), + (FP_COMP0_ADDR, 0), + (FP_COMP0_ADDR, COMP_FLASH_A), + ] + + def test_pending_addition_stays_deferred(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + + manager.flush_removals() + + assert ap.writes == [] + assert ap.mem[FP_COMP0_ADDR] == 0 + assert ap.mem[FP_COMP1_ADDR] == 0 + assert core.flush_calls == 0 + assert FLASH_A not in manager.get_breakpoints() + assert manager.find_breakpoint(FLASH_A) is not None + + core.resume() + + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert FLASH_A in manager.get_breakpoints() + + def test_mixed_removal_and_addition_commits_only_removal(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + core.resume() + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + core.halt() + + # gdb deletes one breakpoint and creates another one while the core is stopped, so a + # removal and an addition are pending at the same time. + manager.remove_breakpoint(FLASH_A) + assert manager.set_breakpoint(FLASH_B, HW) is True + harness.trace.clear() + + manager.flush_removals() + + # The removal is committed, ... + assert ap.mem[FP_COMP0_ADDR] == 0 + assert fpb.find_breakpoint(FLASH_A) is None + assert fpb.num_hw_breakpoint_used == 0 + assert FLASH_A not in manager.get_breakpoints() + assert manager.find_breakpoint(FLASH_A) is None + # ... and the addition is not: clearing the comparator and draining the queue are the + # only things that touched the target. + assert harness.trace == [ + (OP_AP_WRITE, FP_COMP0_ADDR, 0), + (OP_FLUSH,), + ] + assert ap.mem[FP_COMP1_ADDR] == 0 + assert FLASH_B not in manager.get_breakpoints() + assert fpb.find_breakpoint(FLASH_B) is None + assert isinstance(manager.find_breakpoint(FLASH_B), UnrealizedBreakpoint) + + core.resume() + + assert FLASH_B in manager.get_breakpoints() + assert manager.get_breakpoint_type(FLASH_B) == HW + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_B + assert ap.mem[FP_COMP1_ADDR] == 0 + assert fpb.num_hw_breakpoint_used == 1 + + def test_hw_allocation_unchanged_for_representative_sequence(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + assert manager.set_breakpoint(FLASH_B, HW) is True + core.resume() + assert fpb.num_hw_breakpoint_used == 2 + assert fpb.available_breakpoints == 0 + + core.halt() + manager.remove_breakpoint(FLASH_A) + manager.remove_breakpoint(FLASH_B) + + manager.flush_removals() + + assert fpb.available_breakpoints == 2 + assert ap.mem[FP_COMP0_ADDR] == 0 + assert ap.mem[FP_COMP1_ADDR] == 0 + + # Both comparators are free again, so both breakpoints can be re-added and the third + # one is still refused: the same allocation result as before the early commit, where the + # pending removals credited two comparators and the pending additions debited them. + assert manager.set_breakpoint(FLASH_A, HW) is True + assert manager.set_breakpoint(FLASH_B, HW) is True + assert manager.set_breakpoint(FLASH_C, HW) is False + assert manager.find_breakpoint(FLASH_C) is None + + core.resume() + + assert manager.get_breakpoint_type(FLASH_A) == HW + assert manager.get_breakpoint_type(FLASH_B) == HW + assert FLASH_C not in manager.get_breakpoints() + assert fpb.num_hw_breakpoint_used == 2 + assert fpb.find_breakpoint(FLASH_A) is not None + assert fpb.find_breakpoint(FLASH_B) is not None + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert ap.mem[FP_COMP1_ADDR] == COMP_FLASH_B + + def test_remove_never_set_is_noop(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + + manager.remove_breakpoint(FLASH_B) + + manager.flush_removals() + + assert ap.writes == [] + assert core.flush_calls == 0 + assert bytes(core.ram) == bytes(1024) + assert list(manager.get_breakpoints()) == [] + + def test_remove_unflushed_addition_is_bookkeeping_only(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + manager.remove_breakpoint(FLASH_A) + + manager.flush_removals() + + assert ap.writes == [] + assert fpb.num_hw_breakpoint_used == 0 + assert core.flush_calls == 0 + assert manager.find_breakpoint(FLASH_A) is None + assert FLASH_A not in manager.get_breakpoints() + + core.resume() + + assert ap.writes == [] + + def test_removal_deferred_while_running(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + core.resume() + assert core.get_state() == Target.State.RUNNING + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + + manager.remove_breakpoint(FLASH_A) + + manager.flush_removals() + + # Restoring memory or clearing a comparator under a running core is not safe, so the + # removal stays pending exactly as it does today. + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert FLASH_A in manager.get_breakpoints() + assert core.flush_calls == 0 + + core.halt() + core.resume() + + assert ap.mem[FP_COMP0_ADDR] == 0 + assert FLASH_A not in manager.get_breakpoints() + + def test_failed_hw_removal_raises_and_keeps_breakpoint_pending(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + + assert manager.set_breakpoint(FLASH_A, HW) is True + core.resume() + core.halt() + + ap.fail_writes = True + manager.remove_breakpoint(FLASH_A) + + with pytest.raises(exceptions.Error): + manager.flush_removals() + + # The breakpoint stays live and its removal stays pending, so the next flush retries it. + assert FLASH_A in manager.get_breakpoints() + assert manager.find_breakpoint(FLASH_A) is None + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert core.flush_calls == 0 + + def test_failed_sw_removal_is_detected(self, harness: BreakpointHarness) -> None: + core, manager = harness.core, harness.manager + sw_provider = harness.sw_provider + + core.write16(RAM_R, ORIGINAL_INSTR) + assert manager.set_breakpoint(RAM_R, SW) is True + core.resume() + core.halt() + + core.fail_writes = True + manager.remove_breakpoint(RAM_R) + + # The software provider swallows the transfer error and keeps its entry, so the failure + # has to be detected through the provider instead of an exception from it. + with pytest.raises(exceptions.DebugError): + manager.flush_removals() + + assert RAM_R in manager.get_breakpoints() + assert sw_provider.find_breakpoint(RAM_R) is not None + assert core.read16(RAM_R) == BKPT_INSTR + assert core.flush_calls == 0 + + core.fail_writes = False + core.resume() + + assert core.read16(RAM_R) == ORIGINAL_INSTR + assert sw_provider.find_breakpoint(RAM_R) is None diff --git a/test/unit/test_gdbserver_breakpoints.py b/test/unit/test_gdbserver_breakpoints.py new file mode 100644 index 000000000..4828ae534 --- /dev/null +++ b/test/unit/test_gdbserver_breakpoints.py @@ -0,0 +1,285 @@ +# pyOCD debugger +# Copyright (c) 2026 j4rvisstant +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""@brief Tests for the gdb breakpoint removal packets z0 and z1. + +A z0/z1 that is acknowledged with OK while the core is halted must already be committed to the +target, so that a session ending without a further resume, step or disconnect leaves no +breakpoint armed. A failed commit must be reported as E01 rather than OK. +""" + +import logging + +import pytest + +from pyocd.core.target import Target +from pyocd.debug.breakpoints.manager import UnrealizedBreakpoint + +from .mock_breakpoints import ( + BKPT_INSTR, + BreakpointHarness, + COMP_FLASH_A, + COMP_FLASH_B, + COMP_RAM_R, + E01_REPLY, + FLASH_A, + FLASH_B, + FLASH_C, + FP_COMP0_ADDR, + FP_COMP1_ADDR, + OK_REPLY, + OP_AP_WRITE, + OP_FLUSH, + ORIGINAL_INSTR, + RAM_R, + bp_packet, + make_harness, + make_server, +) + +GDBSERVER_LOGGER = 'pyocd.gdbserver.gdbserver' + + +@pytest.fixture +def harness() -> BreakpointHarness: + return make_harness() + + +class TestGdbServerBreakpointRemoval: + def test_z1_commits_removal_before_any_run(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + server = make_server(harness) + + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + core.resume() + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + core.halt() + + reply = server.breakpoint(None, bp_packet(b'z1', FLASH_A)) + + assert reply == server.create_rsp_packet(b'OK') + assert ap.mem[FP_COMP0_ADDR] == 0 + assert fpb.available_breakpoints == 2 + assert FLASH_A not in manager.get_breakpoints() + assert core.flush_calls == 1 + # Nothing resumed, stepped or disconnected after the z1. + assert core.pre_run_events == [Target.RunType.RESUME] + + def test_z0_ram_commits_removal_before_any_run(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + sw_provider = harness.sw_provider + server = make_server(harness, soft_bkpt_as_hard=False) + + core.write16(RAM_R, ORIGINAL_INSTR) + assert server.breakpoint(None, bp_packet(b'Z0', RAM_R)) == OK_REPLY + core.resume() + assert core.read16(RAM_R) == BKPT_INSTR + core.halt() + + reply = server.breakpoint(None, bp_packet(b'z0', RAM_R)) + + assert reply == server.create_rsp_packet(b'OK') + assert core.read16(RAM_R) == ORIGINAL_INSTR + assert sw_provider.find_breakpoint(RAM_R) is None + assert RAM_R not in manager.get_breakpoints() + assert core.flush_calls == 1 + assert ap.comp_writes() == [] + + def test_z0_soft_bkpt_as_hard_frees_comparator(self, harness: BreakpointHarness) -> None: + core, fpb, ap = harness.core, harness.fpb, harness.ap + server = make_server(harness, soft_bkpt_as_hard=True) + + core.write16(RAM_R, ORIGINAL_INSTR) + assert server.breakpoint(None, bp_packet(b'Z0', RAM_R)) == OK_REPLY + core.resume() + assert ap.mem[FP_COMP0_ADDR] == COMP_RAM_R + assert core.read16(RAM_R) == ORIGINAL_INSTR + core.halt() + + reply = server.breakpoint(None, bp_packet(b'z0', RAM_R)) + + assert reply == server.create_rsp_packet(b'OK') + assert ap.mem[FP_COMP0_ADDR] == 0 + assert fpb.available_breakpoints == 2 + assert core.read16(RAM_R) == ORIGINAL_INSTR + assert core.flush_calls == 1 + + def test_gdb_stop_resume_churn(self, harness: BreakpointHarness) -> None: + core, fpb, ap = harness.core, harness.fpb, harness.ap + server = make_server(harness) + + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + core.resume() + core.halt() + assert server.breakpoint(None, bp_packet(b'z1', FLASH_A)) == OK_REPLY + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + # The reinsertion is still deferred until the resume. + assert len(ap.comp_writes()) == 2 + core.resume() + + enabled = [bp for bp in fpb.hw_breakpoints if bp.enabled] + assert len(enabled) == 1 + assert enabled[0].addr == FLASH_A + assert fpb.num_hw_breakpoint_used == 1 + assert ap.mem[FP_COMP1_ADDR] == 0 + assert ap.comp_writes() == [ + (FP_COMP0_ADDR, COMP_FLASH_A), + (FP_COMP0_ADDR, 0), + (FP_COMP0_ADDR, COMP_FLASH_A), + ] + + def test_z1_with_pending_addition_commits_only_removal(self, harness: BreakpointHarness) -> None: + core, manager, fpb, ap = harness.core, harness.manager, harness.fpb, harness.ap + server = make_server(harness) + + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + core.resume() + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + core.halt() + + # gdb inserts a new breakpoint and deletes the old one while the core is stopped, so a + # removal and an addition are pending when the z1 is handled. + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_B)) == OK_REPLY + harness.trace.clear() + + reply = server.breakpoint(None, bp_packet(b'z1', FLASH_A)) + + assert reply == OK_REPLY + assert ap.mem[FP_COMP0_ADDR] == 0 + assert FLASH_A not in manager.get_breakpoints() + # Only the removal reached the target; the addition is still deferred. + assert harness.trace == [ + (OP_AP_WRITE, FP_COMP0_ADDR, 0), + (OP_FLUSH,), + ] + assert ap.mem[FP_COMP1_ADDR] == 0 + assert FLASH_B not in manager.get_breakpoints() + assert isinstance(manager.find_breakpoint(FLASH_B), UnrealizedBreakpoint) + + core.resume() + + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_B + assert FLASH_B in manager.get_breakpoints() + assert fpb.num_hw_breakpoint_used == 1 + + def test_pending_Z1_stays_deferred_until_resume(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + server = make_server(harness) + + reply = server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) + + assert reply == OK_REPLY + assert ap.writes == [] + assert core.flush_calls == 0 + assert FLASH_A not in manager.get_breakpoints() + assert manager.find_breakpoint(FLASH_A) is not None + + core.resume() + + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + + def test_z1_for_unknown_address_is_ok_without_provider_call(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + server = make_server(harness) + + reply = server.breakpoint(None, bp_packet(b'z1', FLASH_C)) + + assert reply == OK_REPLY + assert ap.writes == [] + assert core.flush_calls == 0 + assert bytes(core.ram) == bytes(1024) + assert list(manager.get_breakpoints()) == [] + + def test_commit_failure_replies_E01_and_logs_error( + self, + harness: BreakpointHarness, + caplog: "pytest.LogCaptureFixture", + ) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + server = make_server(harness) + + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + core.resume() + core.halt() + ap.fail_writes = True + + with caplog.at_level(logging.ERROR, logger=GDBSERVER_LOGGER): + reply = server.breakpoint(None, bp_packet(b'z1', FLASH_A)) + + assert reply == server.create_rsp_packet(b'E01') + assert reply == E01_REPLY + errors = [r for r in caplog.records if r.name == GDBSERVER_LOGGER and r.levelno == logging.ERROR] + assert len(errors) >= 1 + assert any('0x%08x' % FLASH_A in r.getMessage() for r in errors) + # The breakpoint stays live with its removal pending, so the next flush retries it. + assert FLASH_A in manager.get_breakpoints() + assert manager.find_breakpoint(FLASH_A) is None + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert core.flush_calls == 0 + + def test_flush_failure_replies_E01_and_logs_error( + self, + harness: BreakpointHarness, + caplog: "pytest.LogCaptureFixture", + ) -> None: + core, ap = harness.core, harness.ap + server = make_server(harness) + + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + core.resume() + core.halt() + # The comparator write itself succeeds and only the drain of the probe's transfer queue + # fails, which is how an error on a deferred transfer surfaces. + core.fail_flush = True + harness.trace.clear() + + with caplog.at_level(logging.ERROR, logger=GDBSERVER_LOGGER): + reply = server.breakpoint(None, bp_packet(b'z1', FLASH_A)) + + assert reply == server.create_rsp_packet(b'E01') + assert reply == E01_REPLY + errors = [r for r in caplog.records if r.name == GDBSERVER_LOGGER and r.levelno == logging.ERROR] + assert len(errors) >= 1 + assert any('0x%08x' % FLASH_A in r.getMessage() for r in errors) + # The drain was attempted, and its failure is reported instead of being hidden behind OK. + assert core.flush_calls == 1 + assert ap.mem[FP_COMP0_ADDR] == 0 + assert harness.trace == [ + (OP_AP_WRITE, FP_COMP0_ADDR, 0), + (OP_FLUSH,), + ] + + def test_removal_while_running_is_deferred(self, harness: BreakpointHarness) -> None: + core, manager, ap = harness.core, harness.manager, harness.ap + server = make_server(harness) + + assert server.breakpoint(None, bp_packet(b'Z1', FLASH_A)) == OK_REPLY + # Non-stop mode: gdb may send z while the core runs. + core.resume() + assert core.get_state() == Target.State.RUNNING + + reply = server.breakpoint(None, bp_packet(b'z1', FLASH_A)) + + assert reply == OK_REPLY + assert ap.mem[FP_COMP0_ADDR] == COMP_FLASH_A + assert FLASH_A in manager.get_breakpoints() + assert core.flush_calls == 0 + + core.halt() + core.resume() + + assert ap.mem[FP_COMP0_ADDR] == 0 + assert FLASH_A not in manager.get_breakpoints()