From 9e551daf95f16e3bcb7b94b61528b0d4f5cf1493 Mon Sep 17 00:00:00 2001 From: Lars Schneidenbach Date: Tue, 28 Jul 2026 16:10:11 -0400 Subject: [PATCH] fix drain-chain hierarchy for barrier context; convert overlap counter to tracewarning Signed-off-by: Lars Schneidenbach --- src/aiu_trace_analyzer/pipeline/barrier.py | 14 ++++--- src/aiu_trace_analyzer/pipeline/hashqueue.py | 3 +- src/aiu_trace_analyzer/pipeline/overlap.py | 39 ++++++++++--------- .../pipeline/test_deactivated_stage_output.py | 11 +++--- 4 files changed, 38 insertions(+), 29 deletions(-) diff --git a/src/aiu_trace_analyzer/pipeline/barrier.py b/src/aiu_trace_analyzer/pipeline/barrier.py index 6bbfbec..455cb1b 100644 --- a/src/aiu_trace_analyzer/pipeline/barrier.py +++ b/src/aiu_trace_analyzer/pipeline/barrier.py @@ -42,10 +42,14 @@ def collection_phase(self) -> bool: def drain(self) -> list[TraceEvent]: if self.phase == self._COLLECTION_PHASE: + # first drain call: switch to the application phase. Defer to the application-phase + # drain so parent drain is not called twice and cross-phase state in self.queues + # survives the transition. self.phase = self._APPLICATION_PHASE + revents = [] else: - # do nothing if this is the application phase - pass - # the queues for these contexts don't contain events (events are held in barrier context), - # so nothing to drain here - return [] + # application phase (final drain call): the cross-phase state has been consumed, so + # it is safe to chain the parent drain, which flushes the (event-less) queues and + # emits any additional events if needed. + revents = super().drain() + return revents diff --git a/src/aiu_trace_analyzer/pipeline/hashqueue.py b/src/aiu_trace_analyzer/pipeline/hashqueue.py index 6982370..180359f 100644 --- a/src/aiu_trace_analyzer/pipeline/hashqueue.py +++ b/src/aiu_trace_analyzer/pipeline/hashqueue.py @@ -26,7 +26,8 @@ def drain(self) -> list[TraceEvent]: item = self.queues.popitem() if isinstance(item, TraceEvent): revents += item - return revents + # chain to the base drain anything left there is emitted + return revents + super().drain() def insert(self, event: TraceEvent, queue_id=None) -> int: ''' diff --git a/src/aiu_trace_analyzer/pipeline/overlap.py b/src/aiu_trace_analyzer/pipeline/overlap.py index 408f380..d92ff91 100644 --- a/src/aiu_trace_analyzer/pipeline/overlap.py +++ b/src/aiu_trace_analyzer/pipeline/overlap.py @@ -4,7 +4,7 @@ import aiu_trace_analyzer.logger as aiulog from aiu_trace_analyzer.pipeline import AbstractContext, AbstractHashQueueContext, TwoPhaseWithBarrierContext -from aiu_trace_analyzer.types import TraceEvent, GlobalIngestData +from aiu_trace_analyzer.types import TraceEvent, GlobalIngestData, TraceWarning from aiu_trace_analyzer.pipeline.tools import PipelineContextTool @@ -44,21 +44,20 @@ def __init__(self, ts_shift_threshold=0.0, max_tid_streams=5, ) -> None: - super().__init__() + super().__init__(warnings=[ + TraceWarning( + name="overlaps", + text="Partial-overlap slices resolved: {d[count]}", + data={"count": 0}, + ) + ]) self.overlap_resolve = overlap_resolve - self.resolved = 0 self.async_id = 0 self.async_queues = {} self.ts_shift_threshold = ts_shift_threshold self.tid_space = {} self.max_tid_streams = max_tid_streams - def __del__(self) -> None: - if not self.is_enabled(): - return - level = aiulog.WARN if self.resolved else aiulog.INFO - aiulog.log(level, "Partial-overlap slices resolved:", self.resolved) - # search for events within the same pid/tid # accumulate a queue of events for each pid/tid # once the queue is full, run detection and emit events that are fine @@ -145,11 +144,11 @@ def handle_overlap(self, queue_id: int) -> list[TraceEvent]: if self.overlap_resolve == self.OVERLAP_RESOLVE_DROP: aiulog.log(aiulog.WARN, "Solving overlap conflict by dropping:", oevent) - self.resolved += 1 + self.issue_warning("overlaps") return [] elif self.overlap_resolve == self.OVERLAP_RESOLVE_WARN: aiulog.log(aiulog.WARN, "Detected overlap conflict: ", oevent["name"]) - self.resolved += 1 + self.issue_warning("overlaps") return [oevent] elif self.overlap_resolve == self.OVERLAP_RESOLVE_SHIFT: ts_shift = self.get_overlap_time(oevent["ts"], oevent["ts"]+oevent["dur"], self.queues[queue_id]) @@ -174,21 +173,21 @@ def handle_overlap(self, "us: increase threshold or use different overlap res option.") rlist = [oevent] - self.resolved += 1 + self.issue_warning("overlaps") return rlist elif self.overlap_resolve == self.OVERLAP_RESOLVE_TID: oevent["tid"] = self.find_next_tid(oevent) # feed offending event back into the detector with the new TID to make sure # there are no collisions there either rlist = self.overlap_detection(oevent) - self.resolved += 1 + self.issue_warning("overlaps") return rlist elif self.overlap_resolve == self.OVERLAP_RESOLVE_ASYNC: oevent["id"] = self.async_id end_ts = oevent["ts"] + oevent["dur"] oevent.pop("dur") self.async_id += 1 - self.resolved += 1 + self.issue_warning("overlaps") e_event = copy.deepcopy(oevent) oevent["ph"] = "b" @@ -264,9 +263,9 @@ def drain(self): if self.overlap_resolve == self.OVERLAP_RESOLVE_TID: if self.phase == self._COLLECTION_PHASE: self._collect_and_build_tid_space() - return super().drain() - else: - return [] + # chain to the parent drain in both phases: it advances the two-phase state and + # propagates accumulated warnings as trace_issue events + return super().drain() else: revents = [] # make sure to drain the queue of async 'e' events that might have been hold @@ -276,7 +275,11 @@ def drain(self): # make sure to keep everything sorted aq.sort(key=lambda e: e['ts']) revents += aq - return revents + # these modes don't use the two-phase mechanism and are drained only once, so switch + # to the application phase before chaining up to make the parent emit the accumulated + # warnings as trace_issue events (rather than deferring to a second drain that never comes) + self.phase = self._APPLICATION_PHASE + return revents + super().drain() def detect_partial_overlap_tids(event: TraceEvent, context: AbstractContext) -> list[TraceEvent]: diff --git a/tests/aiu_trace_analyzer/pipeline/test_deactivated_stage_output.py b/tests/aiu_trace_analyzer/pipeline/test_deactivated_stage_output.py index 7d0b199..df59b7b 100644 --- a/tests/aiu_trace_analyzer/pipeline/test_deactivated_stage_output.py +++ b/tests/aiu_trace_analyzer/pipeline/test_deactivated_stage_output.py @@ -12,9 +12,6 @@ def test_deactivated_stage_contexts_do_not_emit_output(monkeypatch, capsys): - log_calls = [] - monkeypatch.setattr(aiulog, "log", lambda *args: log_calls.append(args)) - contexts = [] dma_context = DataTransferExtractionContext() @@ -29,9 +26,9 @@ def test_deactivated_stage_contexts_do_not_emit_output(monkeypatch, capsys): contexts.append(inverse_context) overlap_context = OverlapDetectionContext() - overlap_context.resolved = 5 + overlap_context.issue_warning("overlaps") overlap_context.disable() - contexts.append(overlap_context) + assert overlap_context.warnings["overlaps"].has_warning() is False power_context = PowerExtractionContext() power_context.bad_events = 1 @@ -52,6 +49,10 @@ def test_deactivated_stage_contexts_do_not_emit_output(monkeypatch, capsys): coll_context.disable() contexts.append(coll_context) + # capture output only around teardown: context construction may legitimately emit debug logs + log_calls = [] + monkeypatch.setattr(aiulog, "log", lambda *args: log_calls.append(args)) + for context in contexts: type(context).__del__(context)