Maintenance 25 week 50

src/aiu_trace_analyzer/pipeline/dma.py

@@ -171,15 +171,13 @@ def extract_data_transfer_event(event: TraceEvent, context: AbstractContext) ->
             # If event name contains " DmaO" use TS4
             elif " DmaO" in event['name']:
                 ts = int(get_cycle_ts_as_clock(4, event["args"]["ts_all"]))
+            else:
+                return [event]
 
             counter = {
                 "ph": "C",
                 "pid": event["pid"],
                 "ts": ts,
-                "DmaI_start": get_cycle_ts_as_clock(1, event["args"]['ts_all']),
-                "DmaI_end": get_cycle_ts_as_clock(2, event["args"]['ts_all']),
-                "DmaO_start": get_cycle_ts_as_clock(4, event["args"]['ts_all']),
-                "DmaO_end": get_cycle_ts_as_clock(5, event["args"]['ts_all']),
                 "cat": event['name'],
                 "name": "BW",
                 "args": {

src/aiu_trace_analyzer/pipeline/normalize.py

@@ -196,60 +196,61 @@ def get_overflow_count(self, qid, job: str, ts: float, cycle: int) -> tuple[floa
         return elapsed_epochs, drift, actual_freq
 
     def tsx_32bit_local_correction(self, event: TraceEvent) -> dict:
-        if "TS1" in event["args"]:
-            args = event["args"]
-            prev = -(1 << 48)  # set something very small to cover for some negative overflow epochs to happen
-            for ts in ["TS1", "TS2", "TS3", "TS4", "TS5"]:
-                curr = int(args[ts], 0)
-                if curr < prev:
-                    if "TSxOF" not in event["args"]:
-                        event["args"]["TSxOF"] = ts
-                    aiulog.log(aiulog.TRACE, "OVC: intra-event TSx overflow:", event["args"])
-                    # currently hard-coded 1 epoch
-                    # event-duration-based epochs require analysis of circular dependency
-                    # between cycle->time and time->cycle conversions
-                    curr += 1 << 32
-                args[ts] = str(curr)
-                prev = curr
-
-            if event["dur"] > self.OVERFLOW_TIME_SPAN_US:
-                self.warnings["long_dur"].update()
-
-            if "Cmpt Exec" not in event["name"]:
-                return args
-
-            # compute anticipated frequency based on duration
-            qid = self.queue_hash(event)
-            if qid not in self.prev_event_data:
-                self.prev_event_data[qid] = {
-                    self._DURATION_KEY: EventStats(),
-                    self._INTERVAL_KEY: EventStats()}
-
-            ts_a, ts_b = self.flex_name_ts_map[event["name"]]
-            dur_cycles = int(event["args"][ts_b]) - int(event["args"][ts_a])
-            dur_freq = float(dur_cycles) / event["dur"]
-            aiulog.log(aiulog.TRACE,
-                       f"{event['args'][ts_a]:10} {event['args'][ts_b]:10} {dur_cycles:10}"
-                       f" {event['dur']:15} {dur_freq:12.3f} |{event['name']}")
-            self.prev_event_data[qid][self._DURATION_KEY].update(
-                (int(event["args"][ts_a]), int(event["args"][ts_b])),
-                (event["ts"], event["dur"]),
-                dur_freq)
-
-            # compute anticipated frequency based on event interval to previous event
-            if self.prev_event_data[qid][self._INTERVAL_KEY].count > 0:
-                gap_cycles = int(event["args"][ts_a]) - self.prev_event_data[qid][self._INTERVAL_KEY].get_start_cycle()
-                gap_time = event["ts"] - self.prev_event_data[qid][self._INTERVAL_KEY].get_start_ts()
-                gap_freq = float(gap_cycles) / gap_time
-            else:
-                gap_freq = dur_freq
-            self.prev_event_data[qid][self._INTERVAL_KEY].update(
-                (int(event["args"][ts_a]), int(event["args"][ts_b])),
-                (event["ts"], event["dur"]),
-                gap_freq)
-
-            return args
-        return event["args"]
+        if "TS1" not in event["args"]:
+            return event["args"]
+
+        args = event["args"]
+        prev = -(1 << 48)  # set something very small to cover for some negative overflow epochs to happen
+        for ts in ["TS1", "TS2", "TS3", "TS4", "TS5"]:
+            curr = int(args[ts], 0)
+            if curr < prev:
+                if "TSxOF" not in event["args"]:
+                    event["args"]["TSxOF"] = ts
+                aiulog.log(aiulog.TRACE, "OVC: intra-event TSx overflow:", event["args"])
+                # currently hard-coded 1 epoch
+                # event-duration-based epochs require analysis of circular dependency
+                # between cycle->time and time->cycle conversions
+                curr += 1 << 32
+            args[ts] = str(curr)
+            prev = curr
+
+        if event["dur"] > self.OVERFLOW_TIME_SPAN_US:
+            self.warnings["long_dur"].update()
+
+        if "Cmpt Exec" in event["name"]:
+            self.frequency_stats(event)
+        return args
+
+    def frequency_stats(self, event: TraceEvent) -> None:
+        # compute anticipated frequency based on duration
+        qid = self.queue_hash(event)
+        if qid not in self.prev_event_data:
+            self.prev_event_data[qid] = {
+                self._DURATION_KEY: EventStats(),
+                self._INTERVAL_KEY: EventStats()}
+
+        ts_a, ts_b = self.flex_name_ts_map[event["name"]]
+        dur_cycles = int(event["args"][ts_b]) - int(event["args"][ts_a])
+        dur_freq = float(dur_cycles) / event["dur"]
+        aiulog.log(aiulog.TRACE,
+                   f"{event['args'][ts_a]:10} {event['args'][ts_b]:10} {dur_cycles:10}"
+                   f" {event['dur']:15} {dur_freq:12.3f} |{event['name']}")
+        self.prev_event_data[qid][self._DURATION_KEY].update(
+            (int(event["args"][ts_a]), int(event["args"][ts_b])),
+            (event["ts"], event["dur"]),
+            dur_freq)
+
+        # compute anticipated frequency based on event interval to previous event
+        if self.prev_event_data[qid][self._INTERVAL_KEY].count > 0:
+            gap_cycles = int(event["args"][ts_a]) - self.prev_event_data[qid][self._INTERVAL_KEY].get_start_cycle()
+            gap_time = event["ts"] - self.prev_event_data[qid][self._INTERVAL_KEY].get_start_ts()
+            gap_freq = float(gap_cycles) / gap_time
+        else:
+            gap_freq = dur_freq
+        self.prev_event_data[qid][self._INTERVAL_KEY].update(
+            (int(event["args"][ts_a]), int(event["args"][ts_b])),
+            (event["ts"], event["dur"]),
+            gap_freq)
 
     def tsx_32bit_global_correction(self, qid, event: TraceEvent) -> dict:
         if "TS1" in event["args"]:

src/aiu_trace_analyzer/pipeline/overlap.py

@@ -126,10 +126,14 @@ def get_overlap_time(self, ts: float, end: float, qstate: OverlapTracking) -> fl
         return overlap_time
 
     def find_next_tid(self, event: TraceEvent) -> int:
+        if self.max_tid_streams == -1:
+            return event["tid"] + 1
+
         if event["tid"] not in self.tid_space[event["pid"]]:
             aiulog.log(
                 aiulog.ERROR,
-                "POD: insufficient dynamic range for tid-based overlap resolution. Increase max_tid_space.")
+                f"POD: insufficient dynamic range for tid-based overlap resolution ({self.max_tid_streams})",
+                f"of job: {event['args']['jobname']}. Increase max_tid_space.")
         new_tid = self.tid_space[event["pid"]][event["tid"]]
         return new_tid
 
@@ -228,33 +232,39 @@ def collect_tid_space(self, event: TraceEvent) -> None:
     def _create_tid_space(self, tid: int, exclude: list[int]) -> list[int]:
         tlist = []
         next_tid = tid
-        while len(tlist) < self.max_tid_streams:
+        while len(tlist) < max(self.max_tid_streams, 1):
             next_tid += 1
             if next_tid not in exclude:
                 tlist.append(next_tid)
         return tlist
 
-    def drain(self):
-        if self.overlap_resolve == self.OVERLAP_RESOLVE_TID and self.phase == self._COLLECTION_PHASE:
+    def _collect_and_build_tid_space(self) -> None:
+        new_tspace = {}
+        for pid, tspace in self.tid_space.items():
             new_tspace = {}
-            for pid, tspace in self.tid_space.items():
-                new_tspace = {}
-                # collect candidate lists for each known tid from input
-                exclude: set = tspace[-1]
-                for tid in tspace.keys():
-                    if tid == -1:
-                        continue
-                    tcandidates = self._create_tid_space(tid, exclude)
-                    self.tid_space[pid][tid] = tcandidates
-                    exclude.update(tcandidates)
-                    new_tspace[tid] = tcandidates[0]
-                    for src_tid, next_tid in zip(tcandidates[:-1], tcandidates[1:]):
-                        new_tspace[src_tid] = next_tid
-
-                aiulog.log(aiulog.TRACE, "POD: total tid_space:", self.tid_space[pid])
-                self.tid_space[pid] = copy.deepcopy(new_tspace)
-                aiulog.log(aiulog.TRACE, "POD: tid neighbors:", new_tspace)
-            return super().drain()
+            # collect candidate lists for each known tid from input
+            exclude: set = tspace[-1]
+            for tid in tspace.keys():
+                if tid == -1:
+                    continue
+                tcandidates = self._create_tid_space(tid, exclude)
+                self.tid_space[pid][tid] = tcandidates
+                exclude.update(tcandidates)
+                new_tspace[tid] = tcandidates[0]
+                for src_tid, next_tid in zip(tcandidates[:-1], tcandidates[1:]):
+                    new_tspace[src_tid] = next_tid
+
+            aiulog.log(aiulog.TRACE, "POD: total tid_space:", self.tid_space[pid])
+            self.tid_space[pid] = copy.deepcopy(new_tspace)
+            aiulog.log(aiulog.TRACE, "POD: tid neighbors:", new_tspace)
+
+    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 []
         else:
             revents = []
             # make sure to drain the queue of async 'e' events that might have been hold