From 4ca6d196985b43fcac1b703f349be876eeceb1a6 Mon Sep 17 00:00:00 2001 From: Teo Mahnic Date: Mon, 28 Sep 2026 14:33:56 +0200 Subject: [PATCH 1/6] coresight: target: apply ctrace-run configuration after TraceCapture --- pyocd/coresight/coresight_target.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/pyocd/coresight/coresight_target.py b/pyocd/coresight/coresight_target.py index 238d6f879..8b4bd7305 100644 --- a/pyocd/coresight/coresight_target.py +++ b/pyocd/coresight/coresight_target.py @@ -417,15 +417,16 @@ def _run_trace_sequence(self, name: str) -> None: delegate.run_sequence(name, pname=seq.pname) def trace_capture(self) -> None: - changed = False - if self.session.ctrace_run is not None: - changed = self.session.ctrace_run.apply(self) - result = self.call_delegate('trace_capture', target=self, mode=0) if not result and self.has_debug_sequence('TraceCapture'): assert self.debug_sequence_delegate if self.debug_sequence_delegate.trace_setup == TraceSetup.FULL: self.debug_sequence_delegate.run_sequence('TraceCapture') + + changed = False + if self.session.ctrace_run is not None: + changed = self.session.ctrace_run.apply(self) + self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, changed) def trace_flush(self) -> None: @@ -434,6 +435,7 @@ def trace_flush(self) -> None: assert self.debug_sequence_delegate if self.debug_sequence_delegate.trace_setup == TraceSetup.FULL: self.debug_sequence_delegate.run_sequence('TraceFlush') + self.session.notify(self.session.Event.TRACE_DATA_FLUSH, self.session) def write_trace_buffer(self, name: str, data: bytes) -> int: From cefc2b4f01a8a7bf3a4759f157aeab5179e785e6 Mon Sep 17 00:00:00 2001 From: Teo Mahnic Date: Mon, 28 Sep 2026 19:38:05 +0200 Subject: [PATCH 2/6] coresight: target: add optional core parameter to trace capture/flush --- pyocd/core/soc_target.py | 4 ++-- pyocd/coresight/coresight_target.py | 24 +++++++++++++++--------- pyocd/gdbserver/gdbserver.py | 4 ++-- 3 files changed, 19 insertions(+), 13 deletions(-) diff --git a/pyocd/core/soc_target.py b/pyocd/core/soc_target.py index 9a6120473..276fe6800 100644 --- a/pyocd/core/soc_target.py +++ b/pyocd/core/soc_target.py @@ -326,10 +326,10 @@ def trace_start(self): def trace_stop(self): self.call_delegate('trace_stop', target=self, mode=0) - def trace_capture(self): + def trace_capture(self, core: Optional[CoreTarget] = None): self.call_delegate('trace_capture', target=self, mode=0) - def trace_flush(self): + def trace_flush(self, core: Optional[CoreTarget] = None): self.call_delegate('trace_flush', target=self, mode=0) def add_target_command_groups(self, command_set: CommandSet): diff --git a/pyocd/coresight/coresight_target.py b/pyocd/coresight/coresight_target.py index 8b4bd7305..7065bcf5a 100644 --- a/pyocd/coresight/coresight_target.py +++ b/pyocd/coresight/coresight_target.py @@ -32,6 +32,7 @@ from ..target.pack.flm_region_builder import FlmFlashRegionBuilder if TYPE_CHECKING: + from ..core.core_target import CoreTarget from ..core.session import Session from .ap import (APAddressBase, AccessPort) from ..debug.svd.model import SVDDevice @@ -416,25 +417,30 @@ def _run_trace_sequence(self, name: str) -> None: if seq.name == name: delegate.run_sequence(name, pname=seq.pname) - def trace_capture(self) -> None: + def trace_capture(self, core: Optional["CoreTarget"] = None) -> None: + pname = core.node_name if core else None + ctrace_run = self.session.ctrace_run + changed = False + result = self.call_delegate('trace_capture', target=self, mode=0) - if not result and self.has_debug_sequence('TraceCapture'): + if not result and self.has_debug_sequence('TraceCapture', pname=pname): assert self.debug_sequence_delegate if self.debug_sequence_delegate.trace_setup == TraceSetup.FULL: - self.debug_sequence_delegate.run_sequence('TraceCapture') + self.debug_sequence_delegate.run_sequence('TraceCapture', pname=pname) - changed = False - if self.session.ctrace_run is not None: - changed = self.session.ctrace_run.apply(self) + if ctrace_run is not None: + changed = ctrace_run.apply(self) self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, changed) - def trace_flush(self) -> None: + def trace_flush(self, core: Optional["CoreTarget"] = None) -> None: + pname = core.node_name if core else None + result = self.call_delegate('trace_flush', target=self, mode=0) - if not result and self.has_debug_sequence('TraceFlush'): + if not result and self.has_debug_sequence('TraceFlush', pname=pname): assert self.debug_sequence_delegate if self.debug_sequence_delegate.trace_setup == TraceSetup.FULL: - self.debug_sequence_delegate.run_sequence('TraceFlush') + self.debug_sequence_delegate.run_sequence('TraceFlush', pname=pname) self.session.notify(self.session.Event.TRACE_DATA_FLUSH, self.session) diff --git a/pyocd/gdbserver/gdbserver.py b/pyocd/gdbserver/gdbserver.py index e47af762a..937a6c2f7 100644 --- a/pyocd/gdbserver/gdbserver.py +++ b/pyocd/gdbserver/gdbserver.py @@ -426,11 +426,11 @@ def __init__(self, session, core=None): def trace_flush(self) -> None: if self.board.target.trace_enabled: - self.board.target.trace_flush() + self.board.target.trace_flush(self.target) def trace_capture(self) -> None: if self.board.target.trace_enabled: - self.board.target.trace_capture() + self.board.target.trace_capture(self.target) def _init_remote_commands(self): """@brief Initialize the remote command processor infrastructure.""" From 1b4bc2073da07c8caee0eab73141b6cd25122cd0 Mon Sep 17 00:00:00 2001 From: Teo Mahnic Date: Fri, 2 Oct 2026 14:11:30 +0200 Subject: [PATCH 3/6] trace: sink: add lock to file sink --- pyocd/trace/sink.py | 31 ++++++++++++++++++------------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/pyocd/trace/sink.py b/pyocd/trace/sink.py index 3d5058c0f..c0c6900e1 100644 --- a/pyocd/trace/sink.py +++ b/pyocd/trace/sink.py @@ -17,6 +17,7 @@ import collections.abc import logging +import threading from pathlib import Path from typing import (Any, BinaryIO, Dict, Mapping, TYPE_CHECKING, Iterable, List, Optional, Sequence, Union) @@ -65,25 +66,29 @@ def __init__(self, path: Path) -> None: self._path = path self._file: Optional[BinaryIO] = None self._started = False + self._lock = threading.RLock() def start(self, changed: bool) -> None: - self.flush() - if self._path.parent.name == '.trace': - self._path.parent.mkdir(exist_ok=True) - self._file = self._path.open('wb' if changed or not self._started else 'ab') - self._started = True + with self._lock: + self.flush() + if self._path.parent.name == '.trace': + self._path.parent.mkdir(exist_ok=True) + self._file = self._path.open('wb' if changed or not self._started else 'ab') + self._started = True def write(self, data: bytes) -> int: - if self._file is None: - return 0 - self._file.write(data) - return len(data) + with self._lock: + if self._file is None: + return 0 + self._file.write(data) + return len(data) def flush(self) -> None: - if self._file is not None: - self._file.flush() - self._file.close() - self._file = None + with self._lock: + if self._file is not None: + self._file.flush() + self._file.close() + self._file = None def shutdown(self) -> None: self.flush() From ba443b6094eda51df724139e1249b4e33586ceb7 Mon Sep 17 00:00:00 2001 From: Teo Mahnic Date: Fri, 2 Oct 2026 13:27:17 +0200 Subject: [PATCH 4/6] ctrace-run: accept DWT VMASK --- pyocd/target/csolution/ctrace_run.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pyocd/target/csolution/ctrace_run.py b/pyocd/target/csolution/ctrace_run.py index 4398f26b0..ac8906bc4 100644 --- a/pyocd/target/csolution/ctrace_run.py +++ b/pyocd/target/csolution/ctrace_run.py @@ -118,6 +118,7 @@ class _CTraceRunParser: (re.compile(r'^DWT_COMP(\d+)$'), 0x020, 0x10), (re.compile(r'^DWT_MASK(\d+)$'), 0x024, 0x10), (re.compile(r'^DWT_FUNCTION(\d+)$'), 0x028, 0x10), + (re.compile(r'^DWT_VMASK(\d+)$'), 0x02C, 0x10), ), 'PMU': ( (re.compile(r'^PMU_EVCNTR(\d+)$'), 0x000, 4), From f53941a8c8d25c8b861576d3958e74391f6b15b3 Mon Sep 17 00:00:00 2001 From: Teo Mahnic Date: Fri, 2 Oct 2026 13:25:28 +0200 Subject: [PATCH 5/6] ctrace-run: add support for ctrace-disable --- pyocd/coresight/coresight_target.py | 16 +- pyocd/target/csolution/ctrace_run.py | 217 +++++++++++++++++---------- 2 files changed, 146 insertions(+), 87 deletions(-) diff --git a/pyocd/coresight/coresight_target.py b/pyocd/coresight/coresight_target.py index 7065bcf5a..fee8ff2af 100644 --- a/pyocd/coresight/coresight_target.py +++ b/pyocd/coresight/coresight_target.py @@ -419,8 +419,6 @@ def _run_trace_sequence(self, name: str) -> None: def trace_capture(self, core: Optional["CoreTarget"] = None) -> None: pname = core.node_name if core else None - ctrace_run = self.session.ctrace_run - changed = False result = self.call_delegate('trace_capture', target=self, mode=0) if not result and self.has_debug_sequence('TraceCapture', pname=pname): @@ -428,10 +426,16 @@ def trace_capture(self, core: Optional["CoreTarget"] = None) -> None: if self.debug_sequence_delegate.trace_setup == TraceSetup.FULL: self.debug_sequence_delegate.run_sequence('TraceCapture', pname=pname) - if ctrace_run is not None: - changed = ctrace_run.apply(self) - - self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, changed) + if (ctrace_run := self.session.ctrace_run) is not None: + with ctrace_run.update(): + changed = ctrace_run.load() + if changed: + ctrace_run.apply_disable(self) + self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, ctrace_run.reloaded) + if changed: + ctrace_run.apply_refs(self) + else: + self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, False) def trace_flush(self, core: Optional["CoreTarget"] = None) -> None: pname = core.node_name if core else None diff --git a/pyocd/target/csolution/ctrace_run.py b/pyocd/target/csolution/ctrace_run.py index ac8906bc4..d4234666c 100644 --- a/pyocd/target/csolution/ctrace_run.py +++ b/pyocd/target/csolution/ctrace_run.py @@ -16,13 +16,14 @@ from __future__ import annotations +from contextlib import contextmanager from dataclasses import dataclass import hashlib import logging from pathlib import Path import re import threading -from typing import (Any, Dict, Iterable, List, Mapping, Optional, Set, Tuple, TYPE_CHECKING, Union) +from typing import (Any, Dict, Iterable, Iterator, List, Mapping, Optional, Set, Tuple, TYPE_CHECKING, Union) import yaml @@ -69,11 +70,12 @@ class _CTraceRunData: """Immutable parsed ctrace-run configuration.""" references: Tuple[Tuple[str, Optional[str]], ...] + disable_register_access: Tuple[_CTraceRunRegister, ...] register_access: Tuple[_CTraceRunRegister, ...] class _CTraceRunParser: - """Cached parser for a .ctrace-run.yml file.""" + """Parser for a .ctrace-run.yml file.""" _DEFAULT_BASE_ADDRESSES: Dict[str, int] = { 'ITM': 0xE0000000, @@ -129,29 +131,22 @@ class _CTraceRunParser: def __init__(self, yml_path: Union[str, Path]) -> None: """Create a ctrace-run processor for a YAML file.""" self._path = Path(yml_path).expanduser().resolve() - self._digest: Optional[bytes] = None - self._data: Optional[_CTraceRunData] = None + self.digest: Optional[bytes] = None - def load(self, force: bool = False) -> Optional[Tuple[bytes, _CTraceRunData]]: - """Read and parse the file if its content has changed.""" + def load(self, last_digest: Optional[bytes] = None) -> Optional[Tuple[bool, Optional[_CTraceRunData]]]: + """Read the file and return whether it changed, together with any parsed configuration.""" + self.digest = None try: yml_content = self._path.read_bytes() except FileNotFoundError: LOG.debug("No ctrace-run file found at '%s'", self._path) - self._digest = None - self._data = None return None except OSError as err: raise CTraceRunError(f"cannot access file: {err.strerror}") from err - digest = hashlib.sha256(yml_content).digest() - if not force and digest == self._digest and self._data is not None: - return digest, self._data - - data = self._parse(yml_content) - self._digest = digest - self._data = data - return digest, data + self.digest = hashlib.sha256(yml_content).digest() + changed = self.digest != last_digest + return changed, self._parse(yml_content) if changed else None def _parse(self, yml_content: bytes) -> _CTraceRunData: try: @@ -166,15 +161,40 @@ def _parse(self, yml_content: bytes) -> _CTraceRunData: if not isinstance(data, dict): raise CTraceRunError("'ctrace-run' must be a dictionary") + disable = data.get('ctrace-disable') + if disable is None: + disable = [] + if not isinstance(disable, list): + raise CTraceRunError("'ctrace-disable' must be a list") + refs = data.get('ctrace-refs') + if refs is None: + refs = [] if not isinstance(refs, list): raise CTraceRunError("'ctrace-refs' must be a list") references: List[Tuple[str, Optional[str]]] = [] + disable_register_access: List[_CTraceRunRegister] = [] register_access: List[_CTraceRunRegister] = [] - for ref_index, ref in enumerate(refs, start=1): + + for idx, ref in enumerate(disable, start=1): + reference, reg_access = self._parse_register_entry(ref, f"ctrace-disable#{idx}") + if reg_access: + references.append(reference) + disable_register_access.extend(reg_access) + + for idx, ref in enumerate(refs, start=1): + if isinstance(ref, dict) and ref.get('regs') is None: + continue try: - reference, ref_access = self._parse_reference(ref, ref_index) + if not isinstance(ref, dict): + raise CTraceRunError(f"reference entry {idx} must be a dictionary") + ref_name = ref.get('ref') + if not isinstance(ref_name, str) or not ref_name: + raise CTraceRunError(f"reference entry {idx} requires 'ref'") + if not isinstance(ref.get('type'), str) or not ref['type']: + raise CTraceRunError(f"reference '{ref_name}' requires 'type'") + reference, ref_access = self._parse_register_entry(ref, ref_name) except CTraceRunError as err: LOG.warning("Ignoring invalid ctrace-run reference in '%s': %s", self._path, err) continue @@ -183,26 +203,17 @@ def _parse(self, yml_content: bytes) -> _CTraceRunData: register_access.extend(ref_access) LOG.debug("Parsed ctrace-run configuration from '%s'", self._path) - return _CTraceRunData(tuple(references), tuple(register_access)) - - def _parse_reference(self, ref: Any, ref_index: int) -> Tuple[Tuple[str, Optional[str]], List[_CTraceRunRegister]]: - if not isinstance(ref, dict): - raise CTraceRunError(f"ctrace-run reference entry {ref_index} must be a dictionary") - - ref_name = ref.get('ref') - if not isinstance(ref_name, str) or not ref_name: - raise CTraceRunError(f"ctrace-run reference entry {ref_index} requires a non-empty 'ref'") - - regs = ref.get('regs') - if regs is None: - regs = [] + return _CTraceRunData(tuple(references), tuple(disable_register_access), tuple(register_access)) + + def _parse_register_entry(self, entry: Any, ref_name: str) -> Tuple[Tuple[str, Optional[str]], List[_CTraceRunRegister]]: + if not isinstance(entry, dict): + raise CTraceRunError(f"ctrace-run entry '{ref_name}' must be a dictionary") + pname = entry.get('pname') + if pname is not None and (not isinstance(pname, str) or not pname): + raise CTraceRunError(f"'pname' in ctrace-run entry '{ref_name}' must be a non-empty string") + regs = entry.get('regs') if not isinstance(regs, list): - raise CTraceRunError(f"'regs' in ctrace-run reference '{ref_name}' must be a list") - - pname = ref.get('pname') - if regs and pname is not None and (not isinstance(pname, str) or not pname): - raise CTraceRunError(f"'pname' in ctrace-run reference '{ref_name}' must be a non-empty string") - + raise CTraceRunError(f"'regs' in ctrace-run entry '{ref_name}' must be a list") return ((ref_name, pname), self._parse_regs(regs, pname, ref_name)) def _parse_regs(self, regs: Iterable[Mapping[str, Any]], pname: Optional[str], ref_name: str) -> List[_CTraceRunRegister]: @@ -279,9 +290,10 @@ class CTraceRun: def __init__(self, session: "Session") -> None: self._lock = threading.RLock() - self._last_applied_digest: Optional[bytes] = None - self._last_capture_digest: Optional[bytes] = None + self._last_digest: Optional[bytes] = None + self._pending_config: Optional[Tuple[bool, Optional[_CTraceRunData]]] = None self._last_error: Optional[str] = None + self._reloaded: bool = False cbuild_run = session.cbuild_run if cbuild_run is None or not cbuild_run.trace.enabled: @@ -303,40 +315,76 @@ def __init__(self, session: "Session") -> None: self._parser = _CTraceRunParser(self._path) session.subscribe(self._trace_restart_handler, session.Event.TRACE_RESTART, session) - def apply(self, target: "SoCTarget", force: bool = False) -> bool: - """Apply configuration and report success or whether it changed for this capture.""" + @property + def reloaded(self) -> bool: + """Whether the configuration was reloaded during the last update context.""" + return self._reloaded + + @contextmanager + def update(self, reload: bool = False) -> Iterator[None]: + """Serialize the three configuration stages and discard any abandoned pending data.""" with self._lock: try: - loaded = self._parser.load(force=force) - if loaded is None: - self._last_applied_digest = None + yield + self._last_digest = self._parser.digest + self._last_error = None + finally: + self._reloaded = reload + self._pending_config = None + + def load(self, force: bool = False) -> bool: + """Prepare pending configuration and report whether it differs from the last call.""" + with self._lock: + try: + self._pending_config = self._parser.load(last_digest=None if force else self._last_digest) + if self._pending_config is None: self._last_error = None return False - digest, data = loaded - if force or digest != self._last_applied_digest: - self._apply_to_target(target, data) - self._last_applied_digest = digest - self._last_error = None - - if force: - return True + changed, data = self._pending_config - changed = digest != self._last_capture_digest - self._last_capture_digest = digest + self._reloaded = self._reloaded or changed return changed except exceptions.Error as err: + self.invalidate() self._report_error(err) return False + def apply_disable(self, target: "SoCTarget") -> bool: + """Apply pending disable registers, if reapplication is needed.""" + with self._lock: + if self._pending_config is None: + return False + _, data = self._pending_config + if data is None: + return True + return self._apply_to_target(target, data, data.disable_register_access) + + def apply_refs(self, target: "SoCTarget") -> bool: + """Apply pending active registers, if reapplication is needed.""" + with self._lock: + if self._pending_config is None: + return False + _, data = self._pending_config + if data is None: + return False + return self._apply_to_target(target, data, data.register_access) + def reload(self, target: "SoCTarget") -> bool: """Reload and apply the file even if it has not changed.""" - return self.apply(target, force=True) + with self.update(reload=True): + self.load(force=True) + if not self.apply_disable(target): + return False + if not self.apply_refs(target): + return False + return True def invalidate(self) -> None: - """Invalidate the last applied configuration, so that it will be reapplied on the next call to apply().""" + """Invalidate the applied configuration.""" with self._lock: - self._last_applied_digest = None + self._last_digest = None + self._pending_config = None def _trace_restart_handler(self, notification: "Notification") -> None: """Invalidate the applied configuration after target trace support restarts.""" @@ -353,30 +401,37 @@ def _report_error(self, error: exceptions.Error) -> None: LOG.warning("Ignoring invalid ctrace-run configuration '%s': %s", self._path, error) self._last_error = error_message - def _apply_to_target(self, target: "SoCTarget", data: _CTraceRunData) -> None: - if not data.register_access: - LOG.debug("No ctrace-run register access to apply") - return - access_targets = self._resolve_access_targets(target, data.references) - - enabled_targets: Set[int] = set() - unlocked_components: Set[Tuple[int, int]] = set() - - for reg in data.register_access: - access_target = access_targets[reg.pname] - target_key = id(access_target) - - if target_key not in enabled_targets: - self._enable_trace_access(access_target) - enabled_targets.add(target_key) - - if reg.component in ('DWT', 'ITM'): - component_key = (target_key, reg.base_address) - if component_key not in unlocked_components: - self._unlock_component(access_target, reg.component, reg.base_address) - unlocked_components.add(component_key) - - self._modify_register(access_target, reg) + def _apply_to_target(self, target: "SoCTarget", data: _CTraceRunData, register_access: Tuple[_CTraceRunRegister, ...]) -> bool: + """Apply register writes and invalidate the update on failure.""" + try: + if not register_access: + LOG.debug("No ctrace-run register access to apply") + return True + access_targets = self._resolve_access_targets(target, data.references) + + enabled_targets: Set[int] = set() + unlocked_components: Set[Tuple[int, int]] = set() + + for reg in register_access: + access_target = access_targets[reg.pname] + target_key = id(access_target) + + if target_key not in enabled_targets: + self._enable_trace_access(access_target) + enabled_targets.add(target_key) + + if reg.component in ('DWT', 'ITM'): + component_key = (target_key, reg.base_address) + if component_key not in unlocked_components: + self._unlock_component(access_target, reg.component, reg.base_address) + unlocked_components.add(component_key) + + self._modify_register(access_target, reg) + return True + except exceptions.Error as err: + self.invalidate() + self._report_error(err) + return False def _resolve_access_targets(self, target: "SoCTarget", references: Iterable[Tuple[str, Optional[str]]]) -> Dict[Optional[str], "CoreTarget"]: cores_by_pname = {core.node_name: core for core in target.cores.values()} From 2b3b9d8082c27ae4c2528d8fc8f34ac9a4717e4b Mon Sep 17 00:00:00 2001 From: Teo Mahnic Date: Fri, 2 Oct 2026 13:26:34 +0200 Subject: [PATCH 6/6] ctrace-run: add support for sync-on-run node --- pyocd/coresight/coresight_target.py | 1 + pyocd/coresight/dwt.py | 53 +++++++++++++---------- pyocd/target/csolution/ctrace_run.py | 63 +++++++++++++++++++++++++++- 3 files changed, 92 insertions(+), 25 deletions(-) diff --git a/pyocd/coresight/coresight_target.py b/pyocd/coresight/coresight_target.py index fee8ff2af..6c8995757 100644 --- a/pyocd/coresight/coresight_target.py +++ b/pyocd/coresight/coresight_target.py @@ -434,6 +434,7 @@ def trace_capture(self, core: Optional["CoreTarget"] = None) -> None: self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, ctrace_run.reloaded) if changed: ctrace_run.apply_refs(self) + ctrace_run.itm_force_sync(core if core is not None else self.primary_core) else: self.session.notify(self.session.Event.TRACE_DATA_CAPTURE, self.session, False) diff --git a/pyocd/coresight/dwt.py b/pyocd/coresight/dwt.py index fcd347005..8b8e042a1 100644 --- a/pyocd/coresight/dwt.py +++ b/pyocd/coresight/dwt.py @@ -1,5 +1,5 @@ # pyOCD debugger -# Copyright (c) 2015-2019 Arm Limited +# Copyright (c) 2015-2019,2026 Arm Limited # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -55,6 +55,7 @@ class DWT(CoreSightComponent): DWT_CTRL_NUM_COMP_MASK = (0xF << 28) DWT_CTRL_NUM_COMP_SHIFT = 28 + DWT_CTRL_NOCYCCNT_MASK = (1 << 25) DWT_CTRL_CYCEVTENA_MASK = (1 << 22) DWT_CTRL_FOLDEVTENA_MASK = (1 << 21) DWT_CTRL_LSUEVTENA_MASK = (1 << 20) @@ -73,13 +74,13 @@ class DWT(CoreSightComponent): DWT_CTRL_CYCCNTENA_MASK = (1 << 0) WATCH_TYPE_TO_FUNCT = { - Target.WatchpointType.READ: 5, - Target.WatchpointType.WRITE: 6, - Target.WatchpointType.READ_WRITE: 7, - 5: Target.WatchpointType.READ, - 6: Target.WatchpointType.WRITE, - 7: Target.WatchpointType.READ_WRITE, - } + Target.WatchpointType.READ: 5, + Target.WatchpointType.WRITE: 6, + Target.WatchpointType.READ_WRITE: 7, + 5: Target.WatchpointType.READ, + 6: Target.WatchpointType.WRITE, + 7: Target.WatchpointType.READ_WRITE, + } # Only sizes that are powers of 2 are supported # Breakpoint size = MASK**2 @@ -90,6 +91,7 @@ def __init__(self, ap, cmpid=None, addr=None): self.watchpoints = [] self.watchpoint_used = 0 self.dwt_configured = False + self._has_cycle_counter = False @property def watchpoint_count(self): @@ -108,6 +110,7 @@ def init(self): self.ap.write_memory(DEMCR, demcr) dwt_ctrl = self.ap.read_memory(self.address + self.DWT_CTRL) + self._has_cycle_counter = not bool(dwt_ctrl & self.DWT_CTRL_NOCYCCNT_MASK) watchpoint_count = (dwt_ctrl & self.DWT_CTRL_NUM_COMP_MASK) >> self.DWT_CTRL_NUM_COMP_SHIFT LOG.info("%d hardware watchpoints", watchpoint_count) for i in range(watchpoint_count): @@ -116,7 +119,10 @@ def init(self): self.ap.write_memory(comparatorAddress + self.DWT_FUNCTION_OFFSET, 0) # Enable cycle counter. - self.ap.write32(self.address + self.DWT_CTRL, self.DWT_CTRL_CYCCNTENA_MASK) + if self._has_cycle_counter: + self.ap.write32(self.address + self.DWT_CTRL, self.DWT_CTRL_CYCCNTENA_MASK) + else: + self.ap.write32(self.address + self.DWT_CTRL, 0x0) self.dwt_configured = True def find_watchpoint(self, addr, size, type): @@ -184,6 +190,10 @@ def remove_all_watchpoints(self): def get_watchpoints(self): return [watch for watch in self.watchpoints if watch.func != 0] + @property + def has_cycle_counter(self) -> bool: + return self._has_cycle_counter + @property def cycle_count(self): return self.ap.read32(self.address + self.DWT_CYCCNT) @@ -205,20 +215,20 @@ class DWTv2(DWT): ## Map from watchpoint type to FUNCTIONn.MATCH field value. WATCH_TYPE_TO_FUNCT = { - Target.WatchpointType.READ: 0b0110, - Target.WatchpointType.WRITE: 0b0101, - Target.WatchpointType.READ_WRITE: 0b0100, - 0b0110: Target.WatchpointType.READ, - 0b0101: Target.WatchpointType.WRITE, - 0b0100: Target.WatchpointType.READ_WRITE, - } + Target.WatchpointType.READ: 0b0110, + Target.WatchpointType.WRITE: 0b0101, + Target.WatchpointType.READ_WRITE: 0b0100, + 0b0110: Target.WatchpointType.READ, + 0b0101: Target.WatchpointType.WRITE, + 0b0100: Target.WatchpointType.READ_WRITE, + } ## Map from data access size to pre-shifted DATAVSIZE field value. DATAVSIZE_MAP = { - 1: (0 << 10), - 2: (1 << 10), - 4: (2 << 10), - } + 1: (0 << 10), + 2: (1 << 10), + 4: (2 << 10), + } def set_watchpoint(self, addr, size, type): """@brief Set a hardware watchpoint.""" @@ -254,6 +264,3 @@ def set_watchpoint(self, addr, size, type): LOG.error('No more watchpoints are available, dropped watchpoint at 0x%08x', addr) return False - - - diff --git a/pyocd/target/csolution/ctrace_run.py b/pyocd/target/csolution/ctrace_run.py index d4234666c..ecbba4571 100644 --- a/pyocd/target/csolution/ctrace_run.py +++ b/pyocd/target/csolution/ctrace_run.py @@ -40,6 +40,8 @@ DEMCR = 0xE000EDFC DEMCR_TRCENA = (1 << 24) +CYCCNT_SYNC_TOGGLE = (1 << 24) | (1 << 26) | (1 << 28) + CORESIGHT_LAR_OFFSET = 0xFB0 CORESIGHT_LAR_KEY = 0xC5ACCE55 @@ -72,6 +74,7 @@ class _CTraceRunData: references: Tuple[Tuple[str, Optional[str]], ...] disable_register_access: Tuple[_CTraceRunRegister, ...] register_access: Tuple[_CTraceRunRegister, ...] + sync_on_run: Dict[Optional[str], bool] class _CTraceRunParser: @@ -202,8 +205,37 @@ def _parse(self, yml_content: bytes) -> _CTraceRunData: references.append(reference) register_access.extend(ref_access) + sync_on_run = self._parse_sync_on_run(data.get('ctrace-setup')) LOG.debug("Parsed ctrace-run configuration from '%s'", self._path) - return _CTraceRunData(tuple(references), tuple(disable_register_access), tuple(register_access)) + return _CTraceRunData(tuple(references), tuple(disable_register_access), tuple(register_access), sync_on_run) + + @staticmethod + def _parse_sync_on_run(setup: Any) -> Dict[Optional[str], bool]: + if setup is None: + return {} + if not isinstance(setup, list): + raise CTraceRunError("'ctrace-setup' must be a list") + + settings: Dict[Optional[str], bool] = {} + for entry in setup: + if not isinstance(entry, dict): + raise CTraceRunError("entries in 'ctrace-setup' must be dictionaries") + pname = entry.get('pname') + if pname is not None and (not isinstance(pname, str) or not pname): + raise CTraceRunError("'pname' in 'ctrace-setup' must be a non-empty string") + if 'disable' in entry and entry['disable'] is not False: + settings[pname] = False + continue + synchronization = entry.get('synchronization') + if synchronization is None: + synchronization = {} + if not isinstance(synchronization, dict): + raise CTraceRunError("'synchronization' in 'ctrace-setup' must be a dictionary") + sync_enabled = synchronization.get('sync-on-run', True) + if not isinstance(sync_enabled, bool): + raise CTraceRunError("'sync-on-run' in 'ctrace-setup' must be a boolean") + settings[pname] = sync_enabled + return settings def _parse_register_entry(self, entry: Any, ref_name: str) -> Tuple[Tuple[str, Optional[str]], List[_CTraceRunRegister]]: if not isinstance(entry, dict): @@ -292,6 +324,7 @@ def __init__(self, session: "Session") -> None: self._lock = threading.RLock() self._last_digest: Optional[bytes] = None self._pending_config: Optional[Tuple[bool, Optional[_CTraceRunData]]] = None + self._sync_on_run: Dict[Optional[str], bool] = {} self._last_error: Optional[str] = None self._reloaded: bool = False @@ -338,10 +371,13 @@ def load(self, force: bool = False) -> bool: try: self._pending_config = self._parser.load(last_digest=None if force else self._last_digest) if self._pending_config is None: + self._sync_on_run = {} self._last_error = None return False changed, data = self._pending_config + if data is not None: + self._sync_on_run = data.sync_on_run self._reloaded = self._reloaded or changed return changed @@ -381,10 +417,33 @@ def reload(self, target: "SoCTarget") -> bool: return True def invalidate(self) -> None: - """Invalidate the applied configuration.""" + """Invalidate the applied configuration and sync settings.""" with self._lock: self._last_digest = None self._pending_config = None + self._sync_on_run = {} + + def itm_force_sync(self, core: "CoreTarget") -> None: + """Request ITM synchronization on the selected core before execution.""" + with self._lock: + LOG.debug("Requesting ITM synchronization for core '%s'", core.node_name) + sync_on_run = self._sync_on_run.get(core.node_name) + if sync_on_run is None: + sync_on_run = self._sync_on_run.get(None, False) + if not sync_on_run: + return + dwt = getattr(core, 'dwt', None) + if dwt is None or not dwt.has_cycle_counter: + return + + try: + cyccnt = dwt.cycle_count + try: + dwt.cycle_count = cyccnt ^ CYCCNT_SYNC_TOGGLE + finally: + dwt.cycle_count = cyccnt + except exceptions.Error as err: + LOG.warning("Failed to force ITM synchronization for processor '%s': %s", core.node_name, err) def _trace_restart_handler(self, notification: "Notification") -> None: """Invalidate the applied configuration after target trace support restarts."""