diff --git a/.github/workflows/_st-sim-a2a3.yml b/.github/workflows/_st-sim-a2a3.yml index ae1e27b7d7..79e434a266 100644 --- a/.github/workflows/_st-sim-a2a3.yml +++ b/.github/workflows/_st-sim-a2a3.yml @@ -120,6 +120,14 @@ jobs: --require-pto-isa --manual "${{ inputs.manual_mode == 'only' && 'only' || 'include' }}" \ --enable-chip-swimlane --enable-dep-gen + - name: chip_swimlane smoke (a2a3 host_build_graph) + if: inputs.include_dfx_smokes + run: | + .venv/bin/python -m pytest tests/st/a2a3/host_build_graph/dfx/chip_swimlane/ \ + --platform a2a3sim --device 0-15 -p no:xdist --pto-session-timeout 600 \ + --require-pto-isa --manual "${{ inputs.manual_mode == 'only' && 'only' || 'include' }}" \ + --enable-chip-swimlane 3 + - name: PMU smoke (a2a3) if: inputs.include_dfx_smokes run: | diff --git a/.github/workflows/_st-sim-a5.yml b/.github/workflows/_st-sim-a5.yml index 492884e775..01699ee60b 100644 --- a/.github/workflows/_st-sim-a5.yml +++ b/.github/workflows/_st-sim-a5.yml @@ -113,6 +113,14 @@ jobs: --require-pto-isa --manual "${{ inputs.manual_mode == 'only' && 'only' || 'include' }}" \ --enable-chip-swimlane --enable-dep-gen + - name: chip_swimlane smoke (a5 host_build_graph) + if: inputs.include_dfx_smokes + run: | + .venv/bin/python -m pytest tests/st/a5/host_build_graph/dfx/chip_swimlane/ \ + --platform a5sim --device 0-15 -p no:xdist --pto-session-timeout 600 \ + --require-pto-isa --manual "${{ inputs.manual_mode == 'only' && 'only' || 'include' }}" \ + --enable-chip-swimlane 3 + - name: PMU smoke (a5) if: inputs.include_dfx_smokes run: | diff --git a/docs/dfx/chip-swimlane-profiling.md b/docs/dfx/chip-swimlane-profiling.md index 7768430c1d..3835358a60 100644 --- a/docs/dfx/chip-swimlane-profiling.md +++ b/docs/dfx/chip-swimlane-profiling.md @@ -45,12 +45,16 @@ available. `chip_swimlane_records.json` with `deps.json` from [`dep_gen`](dep-gen.md) at post-process time; see [§3.5](#35-dependency-arrows-from-dep_gen). -- **AICPU scheduler phases** — per-iteration breakdown into seven - mutually time-exclusive **outer** phases (`complete` / `async_poll` - / `dispatch` / `release` / `dummy` / `early_dispatch` / `drain`), plus - `resolve`, `drain_prepare`, and `drain_publish` **inner** phases. `resolve` - is rendered on a sibling scheduler sub-lane with the same `Sched_N` label, - while the drain sub-phases are nested within their `drain` bar, +- **AICPU scheduler phases** — per-iteration breakdown into mutually + time-exclusive **outer** phases (`complete` / `async_poll` / `dispatch` / + `release` / `dummy` / `early_dispatch` / `drain` / `graph_prepare`), plus + nested phases. + In `tensormap_and_ringbuffer`, `resolve` is nested within `complete` or + `dummy`; in `host_build_graph`, `resolve`, `async_poll`, and `dummy` are + standalone, mutually exclusive phases on the dedicated P thread. HBG + `resolve` uses that P thread's main scheduler lane; TMR's nested `resolve` + uses a sibling scheduler sub-lane. The drain sub-phases are nested within + their `drain` bar, and two **separate-lane** phases (`dummy_task` and `predicated_skip`, sampled immediately before `on_task_complete()` begins dependency resolution and rendered as synthetic @@ -257,13 +261,14 @@ field but render differently in Perfetto: | Phase | Role | Lane | `tasks_processed` semantic | | ----- | ---- | ---- | -------------------------- | | `complete` | outer | sched (pid=2) | FIN'd subtasks + sub-block retires this iter | -| `async_poll` | outer | sched | async-wait (SDMA/RoCE/URMA/CCU) subtasks completed this iter; split from `complete` | +| `async_poll` | outer | sched | async-wait completions resolved; zero means polling consumed CPU without completing work | | `dispatch` | outer | sched | subtasks published this iter | | `release` | outer | sched | deferred-release slots drained this iter | | `dummy` | outer | sched | `dummy_ready_queue` entries handled this iter (explicit dummies and false-predicate tasks) | | `early_dispatch` | outer | sched | blocks staged by speculative early-dispatch this pass | | `drain` | outer | sched | blocks staged by this thread's global sync-start drain pass | -| `resolve` | inner | sched sub-lane, same `Sched_N` label as its outer lane | consumers visited in `on_task_complete` | +| `graph_prepare` | outer | sched | Graph Definition nodes expanded this pass | +| `resolve` | inner (TMR); P-thread outer (HBG) | TMR sched sub-lane; HBG P sched lane | consumers visited in `on_task_complete` (TMR); completed SPSC slots (HBG) | | `drain_prepare` | inner | sched, nested in `drain` | subtasks prepared for global sync-start publication | | `drain_publish` | inner | sched, nested in `drain` | subtasks published during global sync-start staging | | `dummy_task` | separate-lane | Worker View AICPU_N (pid=4) | one dummy entering `on_task_complete()`; full identity is in `task_id` | @@ -274,14 +279,32 @@ orchestrator submit path, so it has no swimlane lane. Read its cost from `g_orch_fanin_cycle` in the device-log orch breakdown (the `fanin` line) instead. -Outer phases are mutually time-exclusive within an iter. The converter renders -`resolve` on a sibling `Sched_N` tid so flow arrows attach to the outer -`complete`/`dummy` lane; `drain_prepare` and `drain_publish` remain on the -scheduler lane and are time-contained by `drain`. Separate-lane phases are -routed to a different lane by the converter +Outer phases are mutually time-exclusive within an iter. In +`tensormap_and_ringbuffer`, the converter renders `resolve` on a sibling +`Sched_N` tid because it is time-contained by the outer `complete`/`dummy` +lane. In `host_build_graph`, standalone `resolve` stays beside `async_poll` and +`dummy` on the P thread's main scheduler lane. `drain_prepare` and +`drain_publish` remain on the scheduler lane and are time-contained by `drain`. +Separate-lane phases are routed to a different lane by the converter (Worker View AICPU_N), so they never overlap visually with the sched lane bars even when their timestamps fall inside an outer span. +On the HBG P thread, consecutive empty async-wait polls are compacted into one +`async_poll(0)` record. Its duration is the exact sum of time spent inside the +poll calls, anchored at the point where the aggregate is flushed; it is not a +wall-clock envelope over the intervening loop bookkeeping. The aggregate is +flushed before `resolve` or `dummy`, when a poll resolves work or reports an +error, and when P exits. This keeps polling cost visible without exporting one +record per spin. A non-zero `tasks_processed` counts every resolved async-wait +entry, including internal Graph nodes, rather than only host-submitted stream +tasks. The compacted record's `shared_at_start` snapshot comes from the first +poll in the aggregate, while `loop_iter` names the iteration that flushes the +aggregate. Because the displayed start timestamp is synthesized from summed +poll CPU time, neither field identifies one wall-clock iteration boundary. +The converter still emits the record's real `shared_at_end` snapshot on the +global ready-queue counter track; only the aggregate's start-side metadata has +the synthesized-timestamp caveat. + Legacy phases (`scan` / `poll` / `idle` / `fanout` / `prestage`) are still parsed for old captures but current a2a3/a5 builds no longer emit them. Renames: `fanout` → `resolve`, `prestage` → @@ -323,9 +346,10 @@ in. The trace contains: blocks (level >= 4). - **AICPU Scheduler** (pid=2) — per-iteration scheduler phase blocks coloured by `phase` (level >= 3). Outer phases appear as sibling bars - on each scheduler thread's first `Sched_N` lane. `resolve` appears on an - adjacent `Sched_N` sub-lane, while `drain_prepare` and `drain_publish` nest - within `drain`. + on each scheduler thread's first `Sched_N` lane. TMR's nested `resolve` + appears on an adjacent `Sched_N` sub-lane; HBG's standalone `resolve` stays + on the P thread's first lane. `drain_prepare` and `drain_publish` nest within + `drain`. - **Scheduler View** (pid=3) — task-execution overlay using AICPU dispatch/finish timestamps (level >= 2), with the same labels as Worker View. @@ -377,6 +401,11 @@ report is printed to stdout; it correlates AICPU phase records with the device log to attribute each scheduler iteration to a specific overhead source. +The scheduler-budget parser counts every mutually exclusive outer phase and +standalone HBG P-thread `resolve` bars. It excludes only `resolve` records whose +timestamps are contained by a TMR `complete` or `dummy` parent, preventing the +nested TMR work from being counted twice. + ### 3.4 Adding human-readable names Lane labels degrade in two steps: @@ -677,10 +706,11 @@ Both architectures use split phase streams: - `ChipSwimlaneAicpuSchedPhaseRecord` (64 B) — one record per **emitted phase**, not per scheduler iteration: a single iteration routinely emits - several (e.g. Complete, AsyncPoll, Dispatch, Release, plus the Resolve - inner phase). `ChipSwimlaneSchedPhaseKind` spans the outer phases + several (e.g. Complete, AsyncPoll, Dispatch, Release, plus Resolve). + `ChipSwimlaneSchedPhaseKind` spans the outer phases (Complete, Dispatch, Release, Dummy, EarlyDispatch, AsyncPoll, Drain, - GraphPrepare), the inner ones (Resolve, DrainPrepare, DrainPublish) and + GraphPrepare), runtime-specific Resolve, the inner drain phases + (DrainPrepare, DrainPublish), and the separate-lane markers (DummyTask, PredicatedSkip) — see §3.2 for how each is rendered. Carries loop_iter + tasks_processed + pop_hit / pop_miss deltas and queue-depth snapshots. diff --git a/docs/dfx/sched-overhead-model.md b/docs/dfx/sched-overhead-model.md index a5bf086174..d5e4cfe71f 100644 --- a/docs/dfx/sched-overhead-model.md +++ b/docs/dfx/sched-overhead-model.md @@ -103,10 +103,23 @@ python -m simpler_setup.tools.swimlane_converter .json \ | 1 | Overhead verdict — per-engine + system `all`/`has` overhead (% of makespan) | | 2 | aicore switch — per-core min/mean/max, overhead-vs-independent split, makespan bound | | 3 / 4 | Head OH / Tail OH distributions | -| 5 | AICPU scheduler-loop budget — ns/loop, phase split, pop hit-rate, fanout/fanin | +| 5 | AICPU scheduler-loop budget — separate S/P ns/loop, all mutually exclusive outer phases, standalone HBG P-thread Resolve, pop hit-rate, fanout/fanin | | 6 | Critical-path attribution — compute vs scheduler-injected µs on the makespan path | ```bash python -m simpler_setup.tools.sched_overhead_analysis \ --chip-swimlane-records-json .json --deps-json .json ``` + +For TMR captures, Resolve is nested in Complete or Dummy and is excluded from +the phase total to avoid double counting. For HBG captures, Resolve is +standalone work on the P thread and is included. Empty HBG async polling is +reported as compact `AsyncPoll(0)` bars, so its measured CPU cost contributes +to the scheduler budget instead of being reconstructed as idle. HBG's S +threads detect AICore FIN and dispatch work, while its P thread resolves +completion state and dependencies. Part 5 reports their loop rates separately; +the Tail-OH-to-loop comparison uses only S-thread loops because Tail OH ends at +FIN detection, before P-thread resolution begins. Phase totals still sum all +threads and therefore include P-thread CPU cost. The phase table lists only +phases represented by the capture, so runtime-specific absent phases do not +appear as synthetic zero rows. diff --git a/simpler_setup/scene_test.py b/simpler_setup/scene_test.py index 6ac9373560..79c0b245ad 100644 --- a/simpler_setup/scene_test.py +++ b/simpler_setup/scene_test.py @@ -1905,6 +1905,20 @@ def task_orch(orch, _args, _cfg, _ns=ns, _test_args=test_args, _config=config): # pytest auto test method # ------------------------------------------------------------------ + @staticmethod + def _effective_enable_chip_swimlane(request) -> int: + """Return the multi-round-safe chip-swimlane level for extension hooks.""" + return effective_diagnostic_options( + request.config.getoption("--rounds", default=1), + chip_swimlane=request.config.getoption("--enable-chip-swimlane", default=0), + dump_args=0, + pmu=0, + dep_gen=False, + scope_stats=False, + swimlane_overhead=False, + warn=False, + ).chip_swimlane + @staticmethod def _effective_enable_dep_gen(request) -> bool: """Return the multi-round-safe dep-gen setting for extension hooks. diff --git a/simpler_setup/tools/sched_overhead_analysis.py b/simpler_setup/tools/sched_overhead_analysis.py index 285e38e213..393afcfbf2 100644 --- a/simpler_setup/tools/sched_overhead_analysis.py +++ b/simpler_setup/tools/sched_overhead_analysis.py @@ -35,6 +35,17 @@ from collections import defaultdict from pathlib import Path +_SCHED_OUTER_PHASES = ( + "complete", + "async_poll", + "dispatch", + "release", + "dummy", + "early_dispatch", + "drain", + "graph_prepare", +) + def _to_uint64(v): """Coerce a JSON-encoded uint64 (int, or string — deps.json quotes uint64s @@ -184,6 +195,9 @@ def parse_scheduler_from_json_phases(data): # noqa: PLR0912 # thread; otherwise a task whose subtasks finish on different threads is # counted once by every thread that observed one of its rows. core_to_thread = data.get("core_to_thread") or [] + assigned_thread_indices = { + thread_idx for thread_idx in core_to_thread if isinstance(thread_idx, int) and thread_idx >= 0 + } final_finish_thread_by_task = {} for task in data.get("tasks", []): task_id = _to_uint64(task.get("task_id")) @@ -209,22 +223,37 @@ def parse_scheduler_from_json_phases(data): # noqa: PLR0912 if not records: continue - # Only "complete" and "dispatch" emit records on a2a3 post-#869. - # Legacy a2a3 captures (or current a5 captures, which still emit - # SCHED_IDLE_WAIT) may carry "idle" / "scan" records; both are - # dropped because idle is reconstructed from gaps between work - # records on the same thread — for the a5/legacy case the gap - # exactly equals the dropped idle records' total span, so the - # numeric idle_us is preserved (only the per-iter granularity is - # lost, which Part 2 doesn't surface). + # Scheduler outer phases are mutually exclusive. Resolve needs a + # record-by-record classification: TMR emits it nested inside Complete + # or Dummy, while HBG emits it as standalone work on the dedicated P + # thread. Count only the latter so the P thread is not dropped without + # double-counting TMR's nested bars. + outer_recs = [r for r in records if r.get("phase") in _SCHED_OUTER_PHASES] + resolve_parents = sorted( + ( + (r.get("start_time_us", 0), r.get("end_time_us", 0)) + for r in outer_recs + if r.get("phase") in ("complete", "dummy") + ), + key=lambda interval: interval[0], + ) + resolve_parent_starts = [interval[0] for interval in resolve_parents] + + def is_nested_resolve(rec): + start = rec.get("start_time_us", 0) + end = rec.get("end_time_us", 0) + parent_idx = bisect.bisect_right(resolve_parent_starts, start) - 1 + return parent_idx >= 0 and end <= resolve_parents[parent_idx][1] + + standalone_resolve = [r for r in records if r.get("phase") == "resolve" and not is_nested_resolve(r)] work_recs = sorted( - (r for r in records if r.get("phase") in ("complete", "async_poll", "dispatch")), + outer_recs + standalone_resolve, key=lambda r: r.get("start_time_us", 0), ) if not work_recs: continue - phase_us = {"complete": 0.0, "async_poll": 0.0, "dispatch": 0.0, "idle": 0.0} + phase_us = {phase: 0.0 for phase in (*_SCHED_OUTER_PHASES, "resolve", "idle")} total_finishes = 0 max_loop_iter = 0 pop_hit = 0 @@ -268,6 +297,16 @@ def parse_scheduler_from_json_phases(data): # noqa: PLR0912 finishes_per_loop = total_finishes / loops if loops > 0 else 0.0 pop_total = pop_hit + pop_miss pop_hit_rate = pop_hit / pop_total * 100 if pop_total > 0 else 0.0 + phases_seen = {rec["phase"] for rec in work_recs} + if phase_us["idle"] > 0: + phases_seen.add("idle") + scheduler_only_phases = {"complete", "dispatch", "release", "early_dispatch", "drain", "graph_prepare"} + has_scheduler_work = bool(phases_seen & scheduler_only_phases) + has_resolution_work = bool(phases_seen & {"resolve", "async_poll", "dummy"}) + is_unassigned_thread = bool(assigned_thread_indices) and tid not in assigned_thread_indices + is_resolution_thread = ( + has_resolution_work and not has_scheduler_work and (bool(standalone_resolve) or is_unassigned_thread) + ) t = { # `completed` remains the legacy logical-task field used by the @@ -283,6 +322,8 @@ def parse_scheduler_from_json_phases(data): # noqa: PLR0912 "pop_miss": pop_miss, "pop_hit_rate": pop_hit_rate, "format": "json_phase", + "role": "resolution" if is_resolution_thread else "scheduler", + "phases_seen": phases_seen, } for p, us in phase_us.items(): t[f"{p}_us"] = us @@ -293,6 +334,29 @@ def parse_scheduler_from_json_phases(data): # noqa: PLR0912 return threads +def _summarize_scheduler_loops(threads): + """Aggregate loop budgets without mixing scheduler and resolution loops.""" + summary = {} + for role in ("scheduler", "resolution"): + role_threads = [thread for thread in threads.values() if thread.get("role", "scheduler") == role] + total_us = sum(thread["total_us"] for thread in role_threads) + loops = sum(thread["loops"] for thread in role_threads) + completed = sum(thread["completed"] for thread in role_threads) + summary[role] = { + "total_us": total_us, + "loops": loops, + "completed": completed, + "avg_loop_us": total_us / loops if loops > 0 else 0.0, + } + return summary + + +def _scheduler_phases_for_report(threads): + """Return phase rows that are represented by the current capture.""" + phases_seen = set().union(*(thread.get("phases_seen", set()) for thread in threads.values())) + return [phase for phase in (*_SCHED_OUTER_PHASES, "resolve", "idle") if phase in phases_seen] + + def validate_perf_tasks_for_overhead_analysis(tasks): """Validate required per-task fields for overhead deep-dive analysis. @@ -865,18 +929,35 @@ def _pct(x, denom): print("=" * 90) print() - fmt2 = " {:<10} {:>7} {:>10} {:>12} {:>11}" - print(fmt2.format("Thread", "Loops", "Tasks", "ns/loop", "Total (us)")) - print(" " + "-" * 54) + fmt2 = " {:<10} {:<10} {:>7} {:>10} {:>12} {:>11}" + print(fmt2.format("Thread", "Role", "Loops", "Tasks", "ns/loop", "Total (us)")) + print(" " + "-" * 65) for tid in sorted(threads.keys()): t = threads[tid] ns_per_loop = t["total_us"] * 1000 / t["loops"] if t["loops"] else 0 - print(fmt2.format("T" + str(tid), t["loops"], t["completed"], f"{ns_per_loop:.0f}", f"{t['total_us']:.1f}")) + role_label = "S scheduler" if t["role"] == "scheduler" else "P resolve" + print( + fmt2.format( + "T" + str(tid), role_label, t["loops"], t["completed"], f"{ns_per_loop:.0f}", f"{t['total_us']:.1f}" + ) + ) + loop_summary = _summarize_scheduler_loops(threads) + for role, row_label in (("scheduler", "S SUM"), ("resolution", "P SUM")): + role_summary = loop_summary[role] + if role_summary["loops"] == 0: + continue + print( + fmt2.format( + row_label, + "", + role_summary["loops"], + role_summary["completed"], + f"{role_summary['avg_loop_us'] * 1000:.0f}", + f"{role_summary['total_us']:.1f}", + ) + ) total_us = sum(t["total_us"] for t in threads.values()) total_completed = sum(t["completed"] for t in threads.values()) - total_loops = sum(t["loops"] for t in threads.values()) - avg_ns_per_loop = total_us * 1000 / total_loops if total_loops > 0 else 0 - print(fmt2.format("SUM", total_loops, total_completed, f"{avg_ns_per_loop:.0f}", f"{total_us:.1f}")) total_finishes = sum(t.get("finishes", 0) for t in threads.values()) print(f" FINs observed (Complete phase): {total_finishes}") print() @@ -884,13 +965,19 @@ def _pct(x, denom): # Phase breakdown. Idle is reconstructed from gaps between work # records on the same thread (no explicit idle record is emitted by # the device anymore). - phases = ["complete", "async_poll", "dispatch", "idle"] phase_labels = { - "complete": "Complete (poll handshake, resolve deps)", + "complete": "Complete (poll handshake, completion handling)", "async_poll": "AsyncPoll (async-wait completion: SDMA/RoCE/URMA/CCU)", "dispatch": "Dispatch (pop queue, build payload, flush)", + "release": "Release (deferred producer release)", + "dummy": "Dummy (dependency-only task resolution)", + "early_dispatch": "EarlyDispatch (speculative staging)", + "drain": "Drain (sync-start staging)", + "graph_prepare": "GraphPrepare (Definition expansion)", + "resolve": "Resolve (completion/dependency resolution)", "idle": "Idle (spinning, no progress — reconstructed from gaps)", } + reported_phases = _scheduler_phases_for_report(threads) # Total (us) is summed across all scheduler threads, so it can exceed the # wall-clock window (e.g. idle ~= n_threads x per-thread idle); "% of total" @@ -900,7 +987,7 @@ def _pct(x, denom): print(header) print(" " + "-" * (len(header) - 2)) phase_totals = {} - for p in phases: + for p in reported_phases: key = p + "_us" tot = sum(t.get(key, 0) for t in threads.values()) phase_totals[p] = tot @@ -949,18 +1036,18 @@ def _pct(x, denom): # Tail-vs-loop cause analysis (closes Part 5). # Scheduler loop time, reported in ns — a loop iteration is sub-us, so us # rounds to a misleading 0.0; ns keeps it readable. - avg_loop_us = total_us / total_loops if total_loops > 0 else 0 + avg_loop_us = loop_summary["scheduler"]["avg_loop_us"] avg_loop_ns = avg_loop_us * 1000 avg_tail_oh = sum(tails) / n loop_ratio = avg_tail_oh / avg_loop_us if avg_loop_us > 0 else 0 - print(f" Avg scheduler loop iteration: {avg_loop_ns:.0f} ns (approx avg polling interval per loop)") + print(f" Avg scheduler loop iteration: {avg_loop_ns:.0f} ns (S threads; approx FIN polling interval)") print() print(f" Avg Tail OH = {avg_tail_oh:.1f} us ~= {loop_ratio:.1f} x avg loop iteration ({avg_loop_ns:.0f} ns)") - print(f" -> On average, a completed task waits ~{loop_ratio:.1f} loop iterations before being detected") + print(f" -> On average, a completed task waits ~{loop_ratio:.1f} S-thread loop iterations before FIN detection") print() # Data-driven insight: find the dominant phase (excluding idle which is not useful work) - work_phases = {p: phase_totals.get(p, 0) for p in ["complete", "async_poll", "dispatch"]} + work_phases = {p: phase_totals[p] for p in reported_phases if p != "idle"} dominant_phase = max(work_phases, key=lambda p: work_phases[p]) dominant_pct = work_phases[dominant_phase] / total_us * 100 if total_us > 0 else 0 key_phase_label = phase_labels[dominant_phase].split(" (")[0] diff --git a/simpler_setup/tools/swimlane_converter.py b/simpler_setup/tools/swimlane_converter.py index 8212a15ab1..eb28b5b5d5 100644 --- a/simpler_setup/tools/swimlane_converter.py +++ b/simpler_setup/tools/swimlane_converter.py @@ -139,6 +139,28 @@ def _decode_in_graph_task_id(task_id): return local >> 10, local & 0x3FF +def _nested_resolve_record_ids(records): + """Return Resolve records contained by a Complete or Dummy parent.""" + parents = sorted( + ( + (record.get("start_time_us", 0), record.get("end_time_us", 0)) + for record in records + if record.get("phase") in ("complete", "dummy") + ), + key=lambda interval: interval[0], + ) + parent_starts = [interval[0] for interval in parents] + nested = set() + for record in records: + if record.get("phase") != "resolve": + continue + start_us = record.get("start_time_us", 0) + parent_idx = bisect.bisect_right(parent_starts, start_us) - 1 + if parent_idx >= 0 and record.get("end_time_us", 0) <= parents[parent_idx][1]: + nested.add(id(record)) + return nested + + def _collect_graph_execution_instances(tasks, scheduler_phases): # noqa: PLR0912 """Join in-graph task rows to their outer GraphPrepare records.""" prepare_by_outer = defaultdict(list) @@ -1729,7 +1751,7 @@ def sched_lane_tid(thread_idx, lane=0): "dispatch": "terrible", # red "async_poll": "yellow", # async-wait completion polling (split from complete) "release": "olive", # deferred-release drain (on_task_release work) - "dummy": "grey", # dummy_drain pass (Resolve nests inside) + "dummy": "grey", # dummy_drain pass (TMR Resolve nests inside; HBG P bar is standalone) "early_dispatch": "rail_animation", # speculative early-dispatch staging # sync_start stop-the-world drain: outer bar time-contains the two # inner staging passes, so Perfetto nests them by depth on the track. @@ -1737,7 +1759,7 @@ def sched_lane_tid(thread_idx, lane=0): "drain_prepare": "cq_build_attempt_runnable", # inner: cluster scan + build_payload "drain_publish": "cq_build_attempt_passed", # inner: MMIO write_reg per subtask (the cohort launch) "graph_prepare": "rail_animation", # bounded Scheduler-side Definition expansion - # Inner phase — nests inside Complete or Dummy via time containment + # Inner in TMR; standalone on HBG's dedicated P thread. "resolve": "vsync_highlight_color", # on_task_complete: walk consumer list # Separate-lane (Worker View AICPU_N) — fallback color if it ever lands on Sched "dummy_task": "grey", @@ -1797,9 +1819,24 @@ def _find_containing_complete(thread_idx: int, finish_us: float): # as a 0.02 us sliver so Perfetto does not collapse it to a hairline. AICPU_WORKER_MARKER_MIN_DUR_US = 0.02 # noqa: N806 + assigned_thread_indices = { + assigned for assigned in (core_to_thread or []) if isinstance(assigned, int) and assigned >= 0 + } for thread_idx, thread_records in enumerate(scheduler_phases): tid = sched_lane_tid(thread_idx, 0) resolve_tid = sched_lane_tid(thread_idx, 1) + nested_resolve_ids = _nested_resolve_record_ids(thread_records) + thread_phase_names = {record.get("phase") for record in thread_records} + scheduler_only_phases = {"complete", "dispatch", "release", "early_dispatch", "drain", "graph_prepare"} + has_scheduler_work = bool(thread_phase_names & scheduler_only_phases) + has_resolution_work = bool(thread_phase_names & {"resolve", "async_poll", "dummy"}) + is_unassigned_thread = bool(assigned_thread_indices) and thread_idx not in assigned_thread_indices + has_standalone_resolve = any( + record.get("phase") == "resolve" and id(record) not in nested_resolve_ids for record in thread_records + ) + is_resolution_thread = ( + has_resolution_work and not has_scheduler_work and (has_standalone_resolve or is_unassigned_thread) + ) # Thread name metadata events.append( @@ -1812,7 +1849,7 @@ def _find_containing_complete(thread_idx: int, finish_us: float): "tid": tid, } ) - if any(record.get("phase") == "resolve" for record in thread_records): + if nested_resolve_ids: events.append( { "args": {"name": f"Sched_{thread_idx}"}, @@ -1949,7 +1986,7 @@ def _find_containing_complete(thread_idx: int, finish_us: float): tasks_processed = matched_finish_rows phase_args["tasks_processed"] = tasks_processed display_name = f"{phase}({tasks_processed})" - event_tid = resolve_tid if phase == "resolve" else tid + event_tid = resolve_tid if phase == "resolve" and id(record) in nested_resolve_ids else tid events.append( { "args": phase_args, @@ -1969,10 +2006,12 @@ def _find_containing_complete(thread_idx: int, finish_us: float): # start, so emitting both is redundant. Two samples at the # SAME ts (e.g. final-drain emit where start_time==end_time) # also breaks Perfetto's rate calc (divide-by-zero → NULL). - # Only complete/dispatch carry real queue depths; release/ - # resolve/early_dispatch zero-fill them, so skip their counter - # samples to avoid spurious 0 dips. - if phase not in ("complete", "dispatch"): + # Complete/Dispatch and every HBG P-thread phase carry live + # shared-queue snapshots. Other runtime phases zero-fill them. + phase_has_live_depths = phase in ("complete", "dispatch") or ( + is_resolution_thread and phase in ("resolve", "async_poll", "dummy") + ) + if not phase_has_live_depths: continue if not depths_valid: continue diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp index 830dbed3dc..787d284961 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp @@ -22,6 +22,7 @@ #include "aicpu/platform_regs.h" #include "callable.h" #include "common/chip_swimlane_profiling.h" +#include "host_build_graph/async_poll_phase_accumulator.h" #include "common/memory_barrier.h" #include "common/platform_config.h" #include "host_build_graph/runtime_core.h" @@ -901,6 +902,38 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread auto &chip_swimlane = sched_chip_swimlane_[thread_idx]; chip_swimlane.reset(); chip_swimlane.chip_swimlane_enabled = (chip_swimlane_level_ != ChipSwimlaneLevel::DISABLED); + + const bool record_sched_phases = chip_swimlane_level_ >= ChipSwimlaneLevel::SCHED_PHASES; + auto capture_shared_depth = [&](int16_t shared_depth[CHIP_SWIMLANE_NUM_QUEUE_SHAPES]) { + constexpr size_t kMax = static_cast(std::numeric_limits::max()); + for (int shape = 0; shape < CHIP_SWIMLANE_NUM_QUEUE_SHAPES; shape++) { + const size_t depth = sched_->ready_queues[shape].size() + sched_->ready_sync_queues[shape].size(); + shared_depth[shape] = static_cast(std::min(depth, kMax)); + } + }; + auto record_p_phase = [&](ChipSwimlaneSchedPhaseKind kind, uint64_t start_time, uint64_t end_time, + uint32_t tasks_processed, const int16_t shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES]) { + int16_t shared_at_end[CHIP_SWIMLANE_NUM_QUEUE_SHAPES]; + capture_shared_depth(shared_at_end); + chip_swimlane_aicpu_record_sched_phase( + thread_idx, kind, start_time, end_time, chip_swimlane.sched_loop_count, tasks_processed, + /*pop_hit=*/0, /*pop_miss=*/0, shared_at_start, shared_at_end + ); + }; + simpler::hbg::AsyncPollPhaseAccumulator async_poll_phase; + int16_t async_poll_shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES] = {0}; + auto flush_async_poll = [&](uint64_t end_time) { + if (!async_poll_phase.active()) return; + // Consecutive empty polls are compacted into one bar whose duration is + // their exact summed CPU time. Anchoring the compact bar at the flush + // point preserves phase accounting without exporting one record per + // spin iteration. + const simpler::hbg::AsyncPollPhaseSummary summary = async_poll_phase.flush(end_time); + record_p_phase( + ChipSwimlaneSchedPhaseKind::AsyncPoll, summary.start_time, summary.end_time, summary.resolved, + async_poll_shared_at_start + ); + }; #endif uint64_t last_progress_ts = get_sys_cnt_aicpu(); @@ -914,15 +947,31 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread while (true) { if (completed_.load(std::memory_order_acquire)) break; +#if SIMPLER_DFX + chip_swimlane.sched_loop_count++; +#endif + int32_t published_task_count = 0; if (handle_orchestrator_exit(thread_idx, header, runtime, published_task_count) == LoopAction::BREAK_LOOP) break; int32_t resolved_this_pass = 0; bool resolved_any = false; +#if SIMPLER_DFX + uint64_t resolve_t0 = 0; + int16_t resolve_shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES] = {0}; + uint32_t resolve_count = 0; +#endif for (int32_t s = 0; s < active_sched_threads_ && !completed_.load(std::memory_order_acquire); s++) { ChipTaskSlotState *slot; while ((slot = sp_queues_[s].pop()) != nullptr) { +#if SIMPLER_DFX + if (record_sched_phases && resolve_t0 == 0) { + resolve_t0 = get_sys_cnt_aicpu(); + flush_async_poll(resolve_t0); + capture_shared_depth(resolve_shared_at_start); + } +#endif #if SIMPLER_SCHED_PROFILING SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*slot, thread_idx); #else @@ -934,8 +983,19 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; +#if SIMPLER_DFX + resolve_count++; +#endif } } +#if SIMPLER_DFX + if (resolve_t0 != 0) { + record_p_phase( + ChipSwimlaneSchedPhaseKind::Resolve, resolve_t0, get_sys_cnt_aicpu(), resolve_count, + resolve_shared_at_start + ); + } +#endif if (completed_.load(std::memory_order_acquire)) break; // Async deferred completions, moved off the scheduler threads. Every @@ -944,6 +1004,16 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread // through P alone. if (rt_ != nullptr && rt_->aicore_mailbox != nullptr && (sched_->async_wait_list.count > 0 || rt_->aicore_mailbox->has_pending())) { +#if SIMPLER_DFX + uint64_t async_poll_t0 = 0; + if (record_sched_phases) { + if (!async_poll_phase.active()) { + capture_shared_depth(async_poll_shared_at_start); + async_poll_phase.begin(); + } + async_poll_t0 = get_sys_cnt_aicpu(); + } +#endif AsyncPollResult poll_result = sched_->async_wait_list.poll_and_complete( rt_->aicore_mailbox, sched_ #if SIMPLER_SCHED_PROFILING @@ -951,6 +1021,18 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread thread_idx #endif ); +#if SIMPLER_DFX + if (async_poll_t0 != 0) { + const uint64_t async_poll_t1 = get_sys_cnt_aicpu(); + const bool flush_poll_phase = async_poll_phase.add_poll( + async_poll_t0, async_poll_t1, static_cast(poll_result.resolved), + poll_result.error_code != SIMPLER_ERROR_NONE + ); + if (flush_poll_phase) { + flush_async_poll(async_poll_t1); + } + } +#endif if (poll_result.error_code != SIMPLER_ERROR_NONE) { fail_scheduler(runtime, thread_idx, poll_result.error_code); break; @@ -967,7 +1049,19 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread constexpr int DUMMY_DRAIN_BATCH = 8; ChipTaskSlotState *dummy_batch[DUMMY_DRAIN_BATCH]; int dummy_got; +#if SIMPLER_DFX + uint64_t dummy_t0 = 0; + int16_t dummy_shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES] = {0}; + uint32_t dummy_count = 0; +#endif while ((dummy_got = sched_->dummy_ready_queue.pop_batch(dummy_batch, DUMMY_DRAIN_BATCH)) > 0) { +#if SIMPLER_DFX + if (record_sched_phases && dummy_t0 == 0) { + dummy_t0 = get_sys_cnt_aicpu(); + flush_async_poll(dummy_t0); + capture_shared_depth(dummy_shared_at_start); + } +#endif for (int di = 0; di < dummy_got; di++) { #if SIMPLER_SCHED_PROFILING SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*dummy_batch[di], thread_idx); @@ -980,9 +1074,19 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; +#if SIMPLER_DFX + dummy_count++; +#endif } if (completed_.load(std::memory_order_acquire)) break; } +#if SIMPLER_DFX + if (dummy_t0 != 0) { + record_p_phase( + ChipSwimlaneSchedPhaseKind::Dummy, dummy_t0, get_sys_cnt_aicpu(), dummy_count, dummy_shared_at_start + ); + } +#endif } if (completed_.load(std::memory_order_acquire)) break; @@ -1030,6 +1134,7 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread } #if SIMPLER_DFX + flush_async_poll(get_sys_cnt_aicpu()); // P owns no cores, so the AICore-keyed flushes below iterate an empty core // list; the sched-phase-buffer flush is the one that matters — it drains any // per-thread records P wrote (e.g. under SCHED_PROFILING) so they are not lost. diff --git a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp index 4e62b029cd..a5305285d2 100644 --- a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp @@ -22,6 +22,7 @@ #include "aicpu/platform_regs.h" #include "callable.h" #include "common/chip_swimlane_profiling.h" +#include "host_build_graph/async_poll_phase_accumulator.h" #include "common/memory_barrier.h" #include "common/platform_config.h" #include "host_build_graph/runtime_core.h" @@ -902,6 +903,38 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread auto &chip_swimlane = sched_chip_swimlane_[thread_idx]; chip_swimlane.reset(); chip_swimlane.chip_swimlane_enabled = (chip_swimlane_level_ != ChipSwimlaneLevel::DISABLED); + + const bool record_sched_phases = chip_swimlane_level_ >= ChipSwimlaneLevel::SCHED_PHASES; + auto capture_shared_depth = [&](int16_t shared_depth[CHIP_SWIMLANE_NUM_QUEUE_SHAPES]) { + constexpr size_t kMax = static_cast(std::numeric_limits::max()); + for (int shape = 0; shape < CHIP_SWIMLANE_NUM_QUEUE_SHAPES; shape++) { + const size_t depth = sched_->ready_queues[shape].size() + sched_->ready_sync_queues[shape].size(); + shared_depth[shape] = static_cast(std::min(depth, kMax)); + } + }; + auto record_p_phase = [&](ChipSwimlaneSchedPhaseKind kind, uint64_t start_time, uint64_t end_time, + uint32_t tasks_processed, const int16_t shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES]) { + int16_t shared_at_end[CHIP_SWIMLANE_NUM_QUEUE_SHAPES]; + capture_shared_depth(shared_at_end); + chip_swimlane_aicpu_record_sched_phase( + thread_idx, kind, start_time, end_time, chip_swimlane.sched_loop_count, tasks_processed, + /*pop_hit=*/0, /*pop_miss=*/0, shared_at_start, shared_at_end + ); + }; + simpler::hbg::AsyncPollPhaseAccumulator async_poll_phase; + int16_t async_poll_shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES] = {0}; + auto flush_async_poll = [&](uint64_t end_time) { + if (!async_poll_phase.active()) return; + // Consecutive empty polls are compacted into one bar whose duration is + // their exact summed CPU time. Anchoring the compact bar at the flush + // point preserves phase accounting without exporting one record per + // spin iteration. + const simpler::hbg::AsyncPollPhaseSummary summary = async_poll_phase.flush(end_time); + record_p_phase( + ChipSwimlaneSchedPhaseKind::AsyncPoll, summary.start_time, summary.end_time, summary.resolved, + async_poll_shared_at_start + ); + }; #endif uint64_t last_progress_ts = get_sys_cnt_aicpu(); @@ -915,15 +948,31 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread while (true) { if (completed_.load(std::memory_order_acquire)) break; +#if SIMPLER_DFX + chip_swimlane.sched_loop_count++; +#endif + int32_t published_task_count = 0; if (handle_orchestrator_exit(thread_idx, header, runtime, published_task_count) == LoopAction::BREAK_LOOP) break; int32_t resolved_this_pass = 0; bool resolved_any = false; +#if SIMPLER_DFX + uint64_t resolve_t0 = 0; + int16_t resolve_shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES] = {0}; + uint32_t resolve_count = 0; +#endif for (int32_t s = 0; s < active_sched_threads_ && !completed_.load(std::memory_order_acquire); s++) { ChipTaskSlotState *slot; while ((slot = sp_queues_[s].pop()) != nullptr) { +#if SIMPLER_DFX + if (record_sched_phases && resolve_t0 == 0) { + resolve_t0 = get_sys_cnt_aicpu(); + flush_async_poll(resolve_t0); + capture_shared_depth(resolve_shared_at_start); + } +#endif #if SIMPLER_SCHED_PROFILING SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*slot, thread_idx); #else @@ -935,8 +984,19 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; +#if SIMPLER_DFX + resolve_count++; +#endif } } +#if SIMPLER_DFX + if (resolve_t0 != 0) { + record_p_phase( + ChipSwimlaneSchedPhaseKind::Resolve, resolve_t0, get_sys_cnt_aicpu(), resolve_count, + resolve_shared_at_start + ); + } +#endif if (completed_.load(std::memory_order_acquire)) break; // Async deferred completions, moved off the scheduler threads. Every @@ -945,6 +1005,16 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread // through P alone. if (rt_ != nullptr && rt_->aicore_mailbox != nullptr && (sched_->async_wait_list.count > 0 || rt_->aicore_mailbox->has_pending())) { +#if SIMPLER_DFX + uint64_t async_poll_t0 = 0; + if (record_sched_phases) { + if (!async_poll_phase.active()) { + capture_shared_depth(async_poll_shared_at_start); + async_poll_phase.begin(); + } + async_poll_t0 = get_sys_cnt_aicpu(); + } +#endif AsyncPollResult poll_result = sched_->async_wait_list.poll_and_complete( rt_->aicore_mailbox, sched_ #if SIMPLER_SCHED_PROFILING @@ -952,6 +1022,18 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread thread_idx #endif ); +#if SIMPLER_DFX + if (async_poll_t0 != 0) { + const uint64_t async_poll_t1 = get_sys_cnt_aicpu(); + const bool flush_poll_phase = async_poll_phase.add_poll( + async_poll_t0, async_poll_t1, static_cast(poll_result.resolved), + poll_result.error_code != SIMPLER_ERROR_NONE + ); + if (flush_poll_phase) { + flush_async_poll(async_poll_t1); + } + } +#endif if (poll_result.error_code != SIMPLER_ERROR_NONE) { fail_scheduler(runtime, thread_idx, poll_result.error_code); break; @@ -968,7 +1050,19 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread constexpr int DUMMY_DRAIN_BATCH = 8; ChipTaskSlotState *dummy_batch[DUMMY_DRAIN_BATCH]; int dummy_got; +#if SIMPLER_DFX + uint64_t dummy_t0 = 0; + int16_t dummy_shared_at_start[CHIP_SWIMLANE_NUM_QUEUE_SHAPES] = {0}; + uint32_t dummy_count = 0; +#endif while ((dummy_got = sched_->dummy_ready_queue.pop_batch(dummy_batch, DUMMY_DRAIN_BATCH)) > 0) { +#if SIMPLER_DFX + if (record_sched_phases && dummy_t0 == 0) { + dummy_t0 = get_sys_cnt_aicpu(); + flush_async_poll(dummy_t0); + capture_shared_depth(dummy_shared_at_start); + } +#endif for (int di = 0; di < dummy_got; di++) { #if SIMPLER_SCHED_PROFILING SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*dummy_batch[di], thread_idx); @@ -981,9 +1075,19 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; +#if SIMPLER_DFX + dummy_count++; +#endif } if (completed_.load(std::memory_order_acquire)) break; } +#if SIMPLER_DFX + if (dummy_t0 != 0) { + record_p_phase( + ChipSwimlaneSchedPhaseKind::Dummy, dummy_t0, get_sys_cnt_aicpu(), dummy_count, dummy_shared_at_start + ); + } +#endif } if (completed_.load(std::memory_order_acquire)) break; @@ -1031,6 +1135,7 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread } #if SIMPLER_DFX + flush_async_poll(get_sys_cnt_aicpu()); // P owns no cores, so the AICore-keyed flushes below iterate an empty core // list; the sched-phase-buffer flush is the one that matters — it drains any // per-thread records P wrote (e.g. under SCHED_PROFILING) so they are not lost. diff --git a/src/common/host_build_graph/async_poll_phase_accumulator.h b/src/common/host_build_graph/async_poll_phase_accumulator.h new file mode 100644 index 0000000000..8e3b4b744f --- /dev/null +++ b/src/common/host_build_graph/async_poll_phase_accumulator.h @@ -0,0 +1,57 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#pragma once + +#include + +namespace simpler::hbg { + +struct AsyncPollPhaseSummary { + uint64_t start_time{0}; + uint64_t end_time{0}; + uint32_t resolved{0}; +}; + +// Accumulates only the CPU time spent polling, excluding the gaps between +// scheduler-loop iterations. The caller chooses the flush point, so a run of +// empty polls and its terminating resolved/error poll becomes one compact bar. +class AsyncPollPhaseAccumulator { +public: + bool active() const { return active_; } + + void begin() { active_ = true; } + + bool add_poll(uint64_t start_time, uint64_t end_time, uint32_t resolved, bool failed = false) { + if (!active_) begin(); + accumulated_cycles_ += end_time - start_time; + resolved_ += resolved; + return resolved > 0 || failed; + } + + AsyncPollPhaseSummary flush(uint64_t end_time) { + AsyncPollPhaseSummary summary; + summary.start_time = end_time >= accumulated_cycles_ ? end_time - accumulated_cycles_ : 0; + summary.end_time = end_time; + summary.resolved = resolved_; + active_ = false; + accumulated_cycles_ = 0; + resolved_ = 0; + return summary; + } + +private: + bool active_{false}; + uint64_t accumulated_cycles_{0}; + uint32_t resolved_{0}; +}; + +} // namespace simpler::hbg diff --git a/src/common/platform/include/common/chip_swimlane_profiling.h b/src/common/platform/include/common/chip_swimlane_profiling.h index a9dcc63bfb..bb57f6f560 100644 --- a/src/common/platform/include/common/chip_swimlane_profiling.h +++ b/src/common/platform/include/common/chip_swimlane_profiling.h @@ -488,12 +488,18 @@ static_assert(sizeof(ChipSwimlaneDataHeader) % 64 == 0, "ChipSwimlaneDataHeader * a single discriminator byte: * * OUTER (mutually time-exclusive within an iter; emit advances _t0_phase): - * Complete, Dispatch, Release, Dummy, EarlyDispatch. + * Complete, Dispatch, Release, Dummy, EarlyDispatch, AsyncPoll, Drain, + * GraphPrepare. * Every iter is a sequence of zero-or-more outer bars + optional gap. * * INNER (no anchor advance; Perfetto auto-nests by time containment): - * Resolve. Only parents are Complete and Dummy — those are the two - * FIN-observation sites that call on_task_complete. + * Resolve in the tensormap_and_ringbuffer runtime, plus DrainPrepare and + * DrainPublish. Resolve is contained by Complete or Dummy there. + * + * HBG RESOLUTION-THREAD OUTER: + * Resolve, AsyncPoll, and Dummy. The host_build_graph runtime hands FIN'd + * slots from S threads to a dedicated P thread, so these are standalone, + * mutually exclusive P-thread bars rather than nested S-thread work. * * SEPARATE-LANE (converter routes to Worker View pid=4, not the sched lane): * DummyTask and PredicatedSkip. One zero-width marker per dependency-only @@ -516,10 +522,10 @@ enum class ChipSwimlaneSchedPhaseKind : uint32_t { EarlyDispatch = 5, // try_early_dispatch: early-dispatch pre-staging // of a flagged producer's consumer's gated blocks. // tasks_processed = blocks staged this pass. - // Inner (parent: Complete | Dummy) - Resolve = 6, // on_task_complete: walk consumer list, decrement fanin, - // push newly-ready successors, ring doorbells for - // early-dispatch hits. tasks_processed = # consumers visited. + // Inner in tensormap_and_ringbuffer (parent: Complete | Dummy); standalone + // P-thread outer phase in host_build_graph. + Resolve = 6, // Complete ready work after FIN observation. tasks_processed + // is consumers visited (TMR) or completed SPSC slots (HBG). // Separate-lane (Worker View pid=4 AICPU_N) DummyTask = 7, // Per-dummy identity marker (zero-width). phase_data.dummy_task // carries the local/ring components of the full task identity. diff --git a/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/kernels/aiv/kernel_noop.cpp b/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/kernels/aiv/kernel_noop.cpp new file mode 100644 index 0000000000..6a73f725c4 --- /dev/null +++ b/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/kernels/aiv/kernel_noop.cpp @@ -0,0 +1,23 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include +#include + +#ifndef __gm__ +#define __gm__ +#endif + +#ifndef __aicore__ +#define __aicore__ [aicore] +#endif + +extern "C" __aicore__ void kernel_entry(__gm__ int64_t *args) { (void)args; } diff --git a/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/kernels/orchestration/scheduler_phases_orch.cpp b/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/kernels/orchestration/scheduler_phases_orch.cpp new file mode 100644 index 0000000000..ec912b9f57 --- /dev/null +++ b/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/kernels/orchestration/scheduler_phases_orch.cpp @@ -0,0 +1,42 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include + +#include "orchestration_api.h" + +namespace { + +constexpr int kNoopKernel = 0; + +} // namespace + +extern "C" { + +__attribute__((visibility("default"))) OrchestrationConfig aicpu_orchestration_config(const ChipTaskArgs &args) { + (void)args; + return OrchestrationConfig{.expected_arg_count = 1}; +} + +__attribute__((visibility("default"))) void aicpu_orchestration_entry(const ChipTaskArgs &args) { + const simpler::hbg::Tensor &input = args.tensor(0).ref(); + + CoreTaskArgs normal_args; + normal_args.add_input(input); + normal_args.launch_spec.set_block_num(1); + const TaskId normal_task = rt_submit_aiv_task(kNoopKernel, normal_args).task_id(); + + CoreTaskArgs dummy_args; + dummy_args.set_dependencies(&normal_task, 1); + rt_submit_dummy_task(dummy_args); +} + +} // extern "C" diff --git a/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/test_scheduler_phases.py b/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/test_scheduler_phases.py new file mode 100644 index 0000000000..0191c07e1f --- /dev/null +++ b/tests/st/a2a3/host_build_graph/dfx/chip_swimlane/test_scheduler_phases.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- + +from __future__ import annotations + +import time + +import torch +from simpler.task_interface import ArgDirection as D + +from simpler_setup import SceneTestCase, TaskArgsBuilder, TensorArg, scene_test +from simpler_setup.scene_test import _outputs_dir, _sanitize_for_filename +from simpler_setup.tools.swimlane_converter import read_perf_data + + +@scene_test(level=2, runtime="host_build_graph") +class TestSchedulerPhases(SceneTestCase): + CALLABLE = { + "orchestration": { + "source": "kernels/orchestration/scheduler_phases_orch.cpp", + "function_name": "aicpu_orchestration_entry", + "signature": [D.IN], + }, + "incores": [ + { + "func_id": 0, + "source": "kernels/aiv/kernel_noop.cpp", + "core_type": "aiv", + "signature": [D.IN], + }, + ], + } + + CASES = [ + { + "name": "resolve_dummy", + "platforms": ["a2a3sim", "a2a3"], + "manual": ["a2a3sim"], + "params": {}, + }, + ] + + def generate_args(self, params): + return TaskArgsBuilder( + TensorArg("input", torch.zeros(1, dtype=torch.int32)), + ) + + def compute_golden(self, args, params): + pass + + def test_run(self, st_platform, st_worker, request): + run_marker = int(time.time()) + super().test_run(st_platform, st_worker, request) + if self._effective_enable_chip_swimlane(request) < 3: + return + + for case in self._matching_cases(st_platform, request): + case_label = _sanitize_for_filename(f"TestSchedulerPhases_{case['name']}") + matches = [p for p in _outputs_dir().glob(f"{case_label}_*") if p.stat().st_mtime >= run_marker] + assert matches, f"no output directory created for {case_label}" + perf_path = max(matches, key=lambda p: p.stat().st_mtime) / "chip_swimlane_records.json" + assert perf_path.exists(), f"missing chip swimlane artifact: {perf_path}" + + data = read_perf_data(perf_path) + phase_threads = data.get("aicpu_scheduler_phases") + assert phase_threads, "scheduler phase records are missing" + assigned_threads = {thread_idx for thread_idx in data.get("core_to_thread", []) if thread_idx >= 0} + resolution_threads = [ + records + for thread_idx, records in enumerate(phase_threads) + if records and thread_idx not in assigned_threads + ] + assert len(resolution_threads) == 1, f"expected one core-less P thread, found {len(resolution_threads)}" + resolution_thread = resolution_threads[0] + required = {"resolve", "dummy"} + emitted = {record.get("phase") for record in resolution_thread} + assert required <= emitted, f"missing P-thread phases: {sorted(required - emitted)}" + + records = [record for record in resolution_thread if record.get("phase") in required] + assert all(record["loop_iter"] > 0 for record in records) + assert all(record["end_time_us"] >= record["start_time_us"] for record in records) + assert sum(record["tasks_processed"] for record in records if record["phase"] == "resolve") >= 1 + assert sum(record["tasks_processed"] for record in records if record["phase"] == "dummy") == 1 + assert len(resolution_thread) < 64, "P-thread phase aggregation produced excessive records" + + ordered = sorted(records, key=lambda record: (record["start_time_us"], record["end_time_us"])) + assert all(left["end_time_us"] <= right["start_time_us"] for left, right in zip(ordered, ordered[1:])) + + +if __name__ == "__main__": + SceneTestCase.run_module(__name__) diff --git a/tests/st/a5/host_build_graph/dfx/chip_swimlane/kernels/aiv/kernel_noop.cpp b/tests/st/a5/host_build_graph/dfx/chip_swimlane/kernels/aiv/kernel_noop.cpp new file mode 100644 index 0000000000..6a73f725c4 --- /dev/null +++ b/tests/st/a5/host_build_graph/dfx/chip_swimlane/kernels/aiv/kernel_noop.cpp @@ -0,0 +1,23 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include +#include + +#ifndef __gm__ +#define __gm__ +#endif + +#ifndef __aicore__ +#define __aicore__ [aicore] +#endif + +extern "C" __aicore__ void kernel_entry(__gm__ int64_t *args) { (void)args; } diff --git a/tests/st/a5/host_build_graph/dfx/chip_swimlane/kernels/orchestration/scheduler_phases_orch.cpp b/tests/st/a5/host_build_graph/dfx/chip_swimlane/kernels/orchestration/scheduler_phases_orch.cpp new file mode 100644 index 0000000000..ec912b9f57 --- /dev/null +++ b/tests/st/a5/host_build_graph/dfx/chip_swimlane/kernels/orchestration/scheduler_phases_orch.cpp @@ -0,0 +1,42 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include + +#include "orchestration_api.h" + +namespace { + +constexpr int kNoopKernel = 0; + +} // namespace + +extern "C" { + +__attribute__((visibility("default"))) OrchestrationConfig aicpu_orchestration_config(const ChipTaskArgs &args) { + (void)args; + return OrchestrationConfig{.expected_arg_count = 1}; +} + +__attribute__((visibility("default"))) void aicpu_orchestration_entry(const ChipTaskArgs &args) { + const simpler::hbg::Tensor &input = args.tensor(0).ref(); + + CoreTaskArgs normal_args; + normal_args.add_input(input); + normal_args.launch_spec.set_block_num(1); + const TaskId normal_task = rt_submit_aiv_task(kNoopKernel, normal_args).task_id(); + + CoreTaskArgs dummy_args; + dummy_args.set_dependencies(&normal_task, 1); + rt_submit_dummy_task(dummy_args); +} + +} // extern "C" diff --git a/tests/st/a5/host_build_graph/dfx/chip_swimlane/test_scheduler_phases.py b/tests/st/a5/host_build_graph/dfx/chip_swimlane/test_scheduler_phases.py new file mode 100644 index 0000000000..14ccf818a7 --- /dev/null +++ b/tests/st/a5/host_build_graph/dfx/chip_swimlane/test_scheduler_phases.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- + +from __future__ import annotations + +import time + +import torch +from simpler.task_interface import ArgDirection as D + +from simpler_setup import SceneTestCase, TaskArgsBuilder, TensorArg, scene_test +from simpler_setup.scene_test import _outputs_dir, _sanitize_for_filename +from simpler_setup.tools.swimlane_converter import read_perf_data + + +@scene_test(level=2, runtime="host_build_graph") +class TestSchedulerPhases(SceneTestCase): + CALLABLE = { + "orchestration": { + "source": "kernels/orchestration/scheduler_phases_orch.cpp", + "function_name": "aicpu_orchestration_entry", + "signature": [D.IN], + }, + "incores": [ + { + "func_id": 0, + "source": "kernels/aiv/kernel_noop.cpp", + "core_type": "aiv", + "signature": [D.IN], + }, + ], + } + + CASES = [ + { + "name": "resolve_dummy", + "platforms": ["a5sim", "a5"], + "manual": ["a5sim"], + "params": {}, + }, + ] + + def generate_args(self, params): + return TaskArgsBuilder( + TensorArg("input", torch.zeros(1, dtype=torch.int32)), + ) + + def compute_golden(self, args, params): + pass + + def test_run(self, st_platform, st_worker, request): + run_marker = int(time.time()) + super().test_run(st_platform, st_worker, request) + if self._effective_enable_chip_swimlane(request) < 3: + return + + for case in self._matching_cases(st_platform, request): + case_label = _sanitize_for_filename(f"TestSchedulerPhases_{case['name']}") + matches = [p for p in _outputs_dir().glob(f"{case_label}_*") if p.stat().st_mtime >= run_marker] + assert matches, f"no output directory created for {case_label}" + perf_path = max(matches, key=lambda p: p.stat().st_mtime) / "chip_swimlane_records.json" + assert perf_path.exists(), f"missing chip swimlane artifact: {perf_path}" + + data = read_perf_data(perf_path) + phase_threads = data.get("aicpu_scheduler_phases") + assert phase_threads, "scheduler phase records are missing" + assigned_threads = {thread_idx for thread_idx in data.get("core_to_thread", []) if thread_idx >= 0} + resolution_threads = [ + records + for thread_idx, records in enumerate(phase_threads) + if records and thread_idx not in assigned_threads + ] + assert len(resolution_threads) == 1, f"expected one core-less P thread, found {len(resolution_threads)}" + resolution_thread = resolution_threads[0] + required = {"resolve", "dummy"} + emitted = {record.get("phase") for record in resolution_thread} + assert required <= emitted, f"missing P-thread phases: {sorted(required - emitted)}" + + records = [record for record in resolution_thread if record.get("phase") in required] + assert all(record["loop_iter"] > 0 for record in records) + assert all(record["end_time_us"] >= record["start_time_us"] for record in records) + assert sum(record["tasks_processed"] for record in records if record["phase"] == "resolve") >= 1 + assert sum(record["tasks_processed"] for record in records if record["phase"] == "dummy") == 1 + assert len(resolution_thread) < 64, "P-thread phase aggregation produced excessive records" + + ordered = sorted(records, key=lambda record: (record["start_time_us"], record["end_time_us"])) + assert all(left["end_time_us"] <= right["start_time_us"] for left, right in zip(ordered, ordered[1:])) + + +if __name__ == "__main__": + SceneTestCase.run_module(__name__) diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index 65ac705256..f1c9bf3419 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -750,6 +750,21 @@ target_link_libraries(test_chip_swimlane_aicore PRIVATE add_test(NAME test_chip_swimlane_aicore COMMAND test_chip_swimlane_aicore) set_tests_properties(test_chip_swimlane_aicore PROPERTIES LABELS "no_hardware") +add_executable(test_async_poll_phase_accumulator + common/test_async_poll_phase_accumulator.cpp +) +target_include_directories(test_async_poll_phase_accumulator PRIVATE + ${GTEST_INCLUDE_DIRS} + ${CMAKE_SOURCE_DIR}/../../../src/common +) +target_link_libraries(test_async_poll_phase_accumulator PRIVATE + ${GTEST_MAIN_LIB} + ${GTEST_LIB} + pthread +) +add_test(NAME test_async_poll_phase_accumulator COMMAND test_async_poll_phase_accumulator) +set_tests_properties(test_async_poll_phase_accumulator PROPERTIES LABELS "no_hardware") + # AICPU-side dep_gen writer. The shared-memory layout is byte-identical across # platforms, so compiling against a2a3's platform_config covers a5 too. add_executable(test_dep_gen_collector_aicpu diff --git a/tests/ut/cpp/common/test_async_poll_phase_accumulator.cpp b/tests/ut/cpp/common/test_async_poll_phase_accumulator.cpp new file mode 100644 index 0000000000..6fbb7d796a --- /dev/null +++ b/tests/ut/cpp/common/test_async_poll_phase_accumulator.cpp @@ -0,0 +1,60 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include + +#include "host_build_graph/async_poll_phase_accumulator.h" + +using simpler::hbg::AsyncPollPhaseAccumulator; + +TEST(AsyncPollPhaseAccumulatorTest, PendingPollsAndResolvedPollFormOneCompactSummary) { + AsyncPollPhaseAccumulator accumulator; + + accumulator.begin(); + EXPECT_FALSE(accumulator.add_poll(100, 110, 0)); + EXPECT_FALSE(accumulator.add_poll(200, 220, 0)); + EXPECT_TRUE(accumulator.add_poll(300, 340, 2)); + + ASSERT_TRUE(accumulator.active()); + const auto summary = accumulator.flush(400); + + EXPECT_EQ(summary.start_time, 330u); + EXPECT_EQ(summary.end_time, 400u); + EXPECT_EQ(summary.resolved, 2u); + EXPECT_FALSE(accumulator.active()); +} + +TEST(AsyncPollPhaseAccumulatorTest, FlushWithoutResolutionResetsTheNextBatch) { + AsyncPollPhaseAccumulator accumulator; + + accumulator.add_poll(10, 15, 0); + const auto pending = accumulator.flush(20); + EXPECT_EQ(pending.start_time, 15u); + EXPECT_EQ(pending.end_time, 20u); + EXPECT_EQ(pending.resolved, 0u); + + accumulator.add_poll(30, 37, 1); + const auto resolved = accumulator.flush(40); + EXPECT_EQ(resolved.start_time, 33u); + EXPECT_EQ(resolved.end_time, 40u); + EXPECT_EQ(resolved.resolved, 1u); +} + +TEST(AsyncPollPhaseAccumulatorTest, FailedPollRequestsAFlushWithoutResolution) { + AsyncPollPhaseAccumulator accumulator; + + EXPECT_TRUE(accumulator.add_poll(50, 60, 0, true)); + const auto failed = accumulator.flush(60); + + EXPECT_EQ(failed.start_time, 50u); + EXPECT_EQ(failed.end_time, 60u); + EXPECT_EQ(failed.resolved, 0u); +} diff --git a/tests/ut/py/test_scene_test_cli_contract.py b/tests/ut/py/test_scene_test_cli_contract.py index cc637f950c..c107fb035d 100644 --- a/tests/ut/py/test_scene_test_cli_contract.py +++ b/tests/ut/py/test_scene_test_cli_contract.py @@ -95,6 +95,14 @@ def test_thin_pytest_wrapper_forwards_the_shared_cli_contract() -> None: } +@pytest.mark.parametrize(("rounds", "expected"), [(1, 3), (2, 0)]) +def test_chip_swimlane_extension_hook_uses_the_shared_multi_round_gate(rounds, expected) -> None: + options = {"--rounds": rounds, "--enable-chip-swimlane": 3} + request = SimpleNamespace(config=SimpleNamespace(getoption=lambda name, default=None: options.get(name, default))) + + assert SceneTestCase._effective_enable_chip_swimlane(request) == expected + + def test_swimlane_overhead_allocates_a_diagnostic_output_prefix(monkeypatch) -> None: scene_test_module = importlib.import_module("simpler_setup.scene_test") output_prefix = scene_test_module.Path("diagnostic-output") diff --git a/tests/ut/py/test_sched_overhead_analysis.py b/tests/ut/py/test_sched_overhead_analysis.py index e2724cf260..3a84210257 100644 --- a/tests/ut/py/test_sched_overhead_analysis.py +++ b/tests/ut/py/test_sched_overhead_analysis.py @@ -9,6 +9,8 @@ """Tests for sched_overhead_analysis: overhead model, aicore switch, Head/Tail OH.""" from simpler_setup.tools.sched_overhead_analysis import ( + _scheduler_phases_for_report, + _summarize_scheduler_loops, aicore_switch_stats, build_task_graph, compute_critical_path, @@ -219,3 +221,87 @@ def test_parse_scheduler_attributes_spmd_task_to_final_finish_thread(): assert threads[1]["finishes"] == 1 assert threads[1]["tasks_per_loop"] == 1 assert threads[1]["finishes_per_loop"] == 1 + + +def test_parse_scheduler_counts_hbg_p_thread_standalone_phases(): + data = { + "aicpu_scheduler_phases": [ + [ + {"phase": "resolve", "start_time_us": 1.0, "end_time_us": 2.0, "loop_iter": 7}, + {"phase": "async_poll", "start_time_us": 3.0, "end_time_us": 5.0, "loop_iter": 9}, + {"phase": "dummy", "start_time_us": 6.0, "end_time_us": 7.0, "loop_iter": 9}, + ] + ] + } + + threads = parse_scheduler_from_json_phases(data) + + assert threads[0]["resolve_us"] == 1.0 + assert threads[0]["async_poll_us"] == 2.0 + assert threads[0]["dummy_us"] == 1.0 + assert threads[0]["idle_us"] == 2.0 + assert threads[0]["total_us"] == 6.0 + assert threads[0]["loops"] == 9 + assert threads[0]["role"] == "resolution" + assert threads[0]["phases_seen"] == {"resolve", "async_poll", "dummy", "idle"} + + +def test_parse_scheduler_does_not_double_count_tmr_nested_resolve(): + data = { + "aicpu_scheduler_phases": [ + [ + { + "phase": "complete", + "start_time_us": 1.0, + "end_time_us": 5.0, + "loop_iter": 3, + "tasks_processed": 1, + }, + {"phase": "resolve", "start_time_us": 2.0, "end_time_us": 4.0, "loop_iter": 3}, + {"phase": "dummy", "start_time_us": 6.0, "end_time_us": 9.0, "loop_iter": 4}, + {"phase": "resolve", "start_time_us": 7.0, "end_time_us": 8.0, "loop_iter": 4}, + ] + ] + } + + threads = parse_scheduler_from_json_phases(data) + + assert threads[0]["complete_us"] == 4.0 + assert threads[0]["dummy_us"] == 3.0 + assert threads[0]["resolve_us"] == 0.0 + assert threads[0]["idle_us"] == 1.0 + assert threads[0]["total_us"] == 8.0 + assert threads[0]["role"] == "scheduler" + assert threads[0]["phases_seen"] == {"complete", "dummy", "idle"} + + +def test_scheduler_loop_summary_keeps_scheduler_and_resolution_rates_separate(): + threads = { + 0: {"role": "scheduler", "total_us": 100.0, "loops": 10, "completed": 2}, + 1: {"role": "scheduler", "total_us": 300.0, "loops": 30, "completed": 6}, + 2: {"role": "resolution", "total_us": 20.0, "loops": 200, "completed": 0}, + } + + summary = _summarize_scheduler_loops(threads) + + assert summary["scheduler"] == { + "total_us": 400.0, + "loops": 40, + "completed": 8, + "avg_loop_us": 10.0, + } + assert summary["resolution"] == { + "total_us": 20.0, + "loops": 200, + "completed": 0, + "avg_loop_us": 0.1, + } + + +def test_scheduler_phase_report_suppresses_absent_runtime_phases(): + threads = { + 0: {"phases_seen": {"complete", "dispatch", "idle"}}, + 1: {"phases_seen": {"resolve", "async_poll"}}, + } + + assert _scheduler_phases_for_report(threads) == ["complete", "async_poll", "dispatch", "resolve", "idle"] diff --git a/tests/ut/py/test_swimlane_converter.py b/tests/ut/py/test_swimlane_converter.py index ebd6aa3278..4576de0e77 100644 --- a/tests/ut/py/test_swimlane_converter.py +++ b/tests/ut/py/test_swimlane_converter.py @@ -841,6 +841,71 @@ def test_complete_phase_preserves_runtime_fin_count(tmp_path): assert complete["args"]["finish_rows_attributed"] == 2 +def test_hbg_resolution_thread_uses_one_lane_and_exports_queue_depths(tmp_path): + out = tmp_path / "trace.json" + scheduler_phases = [ + [], + [ + { + "phase": "resolve", + "start_time_us": 1.0, + "end_time_us": 2.0, + "tasks_processed": 1, + "shared_at_start": [1, 2, 3], + "shared_at_end": [4, 5, 6], + }, + { + "phase": "async_poll", + "start_time_us": 2.0, + "end_time_us": 3.0, + "shared_at_start": [4, 5, 6], + "shared_at_end": [7, 8, 9], + }, + { + "phase": "dummy", + "start_time_us": 3.0, + "end_time_us": 4.0, + "shared_at_start": [7, 8, 9], + "shared_at_end": [10, 11, 12], + }, + ], + ] + + sc.generate_chrome_trace_json([], str(out), scheduler_phases=scheduler_phases, core_to_thread=[0]) + + events = json.loads(out.read_text())["traceEvents"] + p_phases = [event for event in events if event.get("cat") == "scheduler" and event.get("tid") // 10 == 3001] + assert [(event["name"], event["tid"]) for event in p_phases] == [ + ("resolve(1)", 30010), + ("async_poll(0)", 30010), + ("dummy(0)", 30010), + ] + queue_samples = [event for event in events if event.get("name") == "shared_ready_queue"] + assert [(event["ts"], event["args"]) for event in queue_samples] == [ + (2.0, {"AIC": 4, "AIV": 5, "MIX": 6}), + (3.0, {"AIC": 7, "AIV": 8, "MIX": 9}), + (4.0, {"AIC": 10, "AIV": 11, "MIX": 12}), + ] + + +def test_tmr_nested_resolve_stays_on_scheduler_sublane(tmp_path): + out = tmp_path / "trace.json" + scheduler_phases = [ + [ + {"phase": "complete", "start_time_us": 1.0, "end_time_us": 4.0}, + {"phase": "resolve", "start_time_us": 2.0, "end_time_us": 3.0}, + ] + ] + + sc.generate_chrome_trace_json([], str(out), scheduler_phases=scheduler_phases, core_to_thread=[0]) + + events = json.loads(out.read_text())["traceEvents"] + complete = next(event for event in events if event.get("name") == "complete(0)") + resolve = next(event for event in events if event.get("name") == "resolve(0)") + assert complete["tid"] == 30000 + assert resolve["tid"] == 30001 + + def test_complete_flow_worker_view_only_without_scheduler_phases(tmp_path): # Without scheduler_phases the complete-flow block is skipped entirely: # neither view gets a complete arrow (regression guard on the gate).