From b4d5b133ee1d4717701a9813e26218861483138c Mon Sep 17 00:00:00 2001 From: Erhan Kurubas Date: Fri, 17 Jul 2026 17:06:58 +0200 Subject: [PATCH] test(sysview_tracing): validate TRACE_STOP per core --- conftest.py | 8 +++ .../pytest_function_tracing.py | 4 +- .../sysview_tracing/pytest_sysview_tracing.py | 72 ++++++++++++++----- 3 files changed, 67 insertions(+), 17 deletions(-) diff --git a/conftest.py b/conftest.py index 0eadd337877..089dad6e3b0 100644 --- a/conftest.py +++ b/conftest.py @@ -271,6 +271,14 @@ class OpenOCD: return '' return to_str(resp) + def consume_output(self, duration: float) -> None: + if self.telnet is None: + return + end = time.time() + duration + while time.time() < end: + self.telnet.read_very_eager() + time.sleep(0.05) + def apptrace_wait_stop(self, timeout: int = 30) -> None: stopped = False end_before = time.time() + timeout diff --git a/examples/system/tracing/function_tracing/pytest_function_tracing.py b/examples/system/tracing/function_tracing/pytest_function_tracing.py index 7e54bdcc38f..f401e47edae 100644 --- a/examples/system/tracing/function_tracing/pytest_function_tracing.py +++ b/examples/system/tracing/function_tracing/pytest_function_tracing.py @@ -129,7 +129,9 @@ def _test_function_tracing_jtag(openocd_dut: 'OpenOCD', idf_path: str, dut: IdfD dut.expect(re.compile(rb'function-tracing: workload iteration \d+'), timeout=30) # Let function-trace samples accumulate while recording. - time.sleep(1) + # Keep reading telnet so OpenOCD's blocking log writes don't stall its + # main loop and leave the 'stop' below unserviced. + openocd.consume_output(1) openocd.write('esp sysview_mcore stop') openocd.apptrace_wait_stop() diff --git a/examples/system/tracing/sysview_tracing/pytest_sysview_tracing.py b/examples/system/tracing/sysview_tracing/pytest_sysview_tracing.py index a60370d909c..068957bc2b0 100644 --- a/examples/system/tracing/sysview_tracing/pytest_sysview_tracing.py +++ b/examples/system/tracing/sysview_tracing/pytest_sysview_tracing.py @@ -16,18 +16,60 @@ if typing.TYPE_CHECKING: from conftest import OpenOCD +STOP_EVENT_ID = 0x0B # SYSVIEW_EVTID_TRACE_STOP + + +def _assert_has_stop_record(segment: bytes, label: str) -> None: + """Assert a SysView data segment ends with a TRACE_STOP record. + + A STOP record is the STOP event ID followed by a variable-length timestamp + delta. Walk back over the trailing continuation bytes (0x80 bit set) to find + the event ID, since its offset is not fixed. + """ + size = len(segment) + assert size >= 2, f'{label}: segment too small to contain STOP record' + i = size - 2 + while i >= 0 and (segment[i] & 0x80): + i -= 1 + assert i >= 0 and segment[i] == STOP_EVENT_ID, f'{label}: does not end with a TRACE_STOP record' + + +def _split_mcore_file(content: bytes) -> list[bytes]: + """Split an ``esp sysview_mcore`` combined file into per-core data segments. + + Layout is an ASCII comment header followed by core0 data then core1 data:: + + ; ... + ; Offset Core0 0 + ; Offset Core1 + ; + + + The header ends at the first line that does not start with ';', (the data + begins with the all-zero sync sequence). + """ + header = re.match(rb'(?:;[^\n]*\n)*', content) + assert header is not None + pos = header.end() + m = re.search(rb';\s*Offset\s+Core1\s+(\d+)', content[:pos]) + assert m is not None, 'mcore file header missing "Offset Core1"' + offset_core1 = int(m.group(1)) + core0 = content[pos : pos + offset_core1] + core1 = content[pos + offset_core1 :] + return [core0, core1] + + def _validate_trace_data(trace_log: str, target: str, dual_core: bool = False, is_uart: bool = False) -> None: """Validate SysView trace data in a single trace log file. Args: trace_log: Path to the trace log file target: Target chip name (e.g., 'esp32', 'esp32s3') - dual_core: If True, expect a per-core description block for both cores - (the ``esp sysview_mcore`` multi-core capture embeds one per core) + dual_core: If True, this is an 'esp sysview_mcore' combined file that + embeds core0 and core1 data segments (see '_split_mcore_file'); + each segment is validated separately. is_uart: If True, also validate STOP record at end of file """ - STOP_EVENT_ID = 0x0B # SYSVIEW_EVTID_TRACE_STOP - with open(trace_log, 'rb') as f: content = f.read() @@ -35,16 +77,13 @@ def _validate_trace_data(trace_log: str, target: str, dual_core: bool = False, i search_str = f'N=FreeRTOS Application,D={target},C=core{idx},O=FreeRTOS'.encode() assert search_str in content, f'SysView core{idx} trace data not found in {trace_log}' - # The file must end with a TRACE_STOP record: the STOP event ID - # followed by a variable-length timestamp delta. Walk back - # over the trailing continuation bytes (0x80 bit set) - # to find the event ID, since its offset is not fixed. - size = len(content) - assert size >= 2, 'Trace file too small to contain STOP record' - i = size - 2 - while i >= 0 and (content[i] & 0x80): - i -= 1 - assert i >= 0 and content[i] == STOP_EVENT_ID, 'STOP record does not start with STOP eventID' + if dual_core: + # Seek to each core's segment via the header offset + # and require each to end with its own TRACE_STOP record. + for idx, segment in enumerate(_split_mcore_file(content)): + _assert_has_stop_record(segment, f'core{idx}') + else: + _assert_has_stop_record(content, 'trace') def _capture_sysview_trace(ser: serial.Serial, trace_log_path: str) -> None: @@ -144,8 +183,9 @@ def _test_sysview_tracing_jtag(openocd_dut: 'OpenOCD', dut: IdfDut) -> None: dut.expect('example: Created task') # dut has been restarted by gdb since the last dut.expect() dut_expect_task_event() - # Do a sleep while sysview samples are captured. - time.sleep(3) + # Capture sysview samples and keep reading telnet so OpenOCD's blocking log writes don't stall its + # main loop and leave the 'stop' below unserviced. + openocd.consume_output(3) openocd.write('esp sysview_mcore stop') openocd.apptrace_wait_stop()