skip launch-flow event creation if inconsistent timestamps are deteted

src/aiu_trace_analyzer/pipeline/flow_launch.py

@@ -3,14 +3,29 @@
 import re
 
 import aiu_trace_analyzer.logger as aiulog
-from aiu_trace_analyzer.types import TraceEvent
+from aiu_trace_analyzer.types import TraceEvent, TraceWarning
 from aiu_trace_analyzer.pipeline import AbstractContext, TwoPhaseWithBarrierContext
 
 
 class LaunchFLowContext(TwoPhaseWithBarrierContext):
     launch_pattern = re.compile(r'Launch.*ControlBlock')
 
     def __init__(self, warnings=None):
+        if warnings is None:
+            warnings = []
+        warnings.extend([
+            TraceWarning(
+                name="ts_inconsistency",
+                text="FLOWS: Detected {d[count]} timestamp inconsistencies,"
+                " skipped flow creation for affected iterations",
+                data={"count": 0}
+            ),
+            TraceWarning(
+                name="ts_after_schedwait",
+                text="FLOWS: Ignored {d[count]} events with timestamp after schedule wait",
+                data={"count": 0}
+            )
+        ])
         super().__init__(warnings)
         self.flow_id_seq = 0
 
@@ -67,8 +82,14 @@ def update_last_ts(self, qid: int, event: TraceEvent) -> None:
             self.queues[qid]["last_pid_tid"] = (event["pid"], event["tid"])
             self.queues[qid]["last_event"] = event
         else:
-            aiulog.log(aiulog.WARN, "FLOWS: Ignoring event with ts after schedule wait", event)
-        assert self.queues[qid]["last_ts"] <= sched_wait_end, f"{qid}: {event}, {self.queues[qid]}"
+            aiulog.log(aiulog.TRACE, "FLOWS: Ignoring event with ts after schedule wait", event)
+            self.warnings["ts_after_schedwait"].update({"count": 1})
+
+        # Check for timestamp inconsistency and mark queue as invalid if detected
+        if self.queues[qid]["last_ts"] > sched_wait_end:
+            self.warnings["ts_inconsistency"].update({"count": 1})
+            # Mark this queue as invalid to prevent flow event creation
+            self.queues[qid]["invalid"] = True
 
     def max_flow_id_detection(self, observed_id: int) -> None:
         self.flow_id_seq = max(self.flow_id_seq, observed_id)
@@ -102,6 +123,10 @@ def create_missing(self, event: TraceEvent) -> list[TraceEvent]:
         if qid not in self.queues or "src" not in self.queues[qid]:
             return []
 
+        # Skip flow creation if queue is marked as invalid due to timestamp issues
+        if self.queues[qid].get("invalid", False):
+            return []
+
         launcher = self.queues[qid]["src"]
         new_flow_id = self.get_new_flow_id()
         flow_events = [
@@ -135,6 +160,10 @@ def drain(self) -> list[TraceEvent]:
         for id, qdata in self.queues.items():
             if "src" not in qdata or "schedwait" not in qdata:
                 continue
+
+            # Skip flow creation if queue is marked as invalid due to timestamp issues
+            if qdata.get("invalid", False):
+                continue
             launcher = qdata["src"]
             waiter = qdata["schedwait"]
             new_flow_id = self.get_new_flow_id()