From 064bc2adc30dece69b0d17eb23a95ce63fa14341 Mon Sep 17 00:00:00 2001 From: majin0824 Date: Tue, 1 Sep 2026 19:08:19 -0700 Subject: [PATCH] Feature: add dispatch-correlated same-host multi-rank swimlane timeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capture and merge same-host multi-rank Chip Swimlane data on one timeline. Carry the parent run, task-slot, and group identity through the L3 mailbox so rank-local captures can be paired by stable dispatch identity instead of coincidental local dN indexes. Request the Host/Device clock anchors through CallConfig.capture_clock_anchors so the runtime enables them from an explicit field rather than by parsing the output directory. The field names what the runtime does, never who consumes it: Rank, group and merge stay concepts of the layer above, so the platform runner that reads it has no notion of a Rank and no runtime or platform code parses the rankN/dN path. The two runtimes open that interval at different points — before Host orchestration for host_build_graph, before kernel launch for tensormap_and_ringbuffer — because each takes it at the earliest point preceding its own device timestamps; the serialized position name predates the second case and is documented rather than renamed, being an on-wire value in every existing capture. Separate one ChipWorker child's artifacts from its siblings' for every diagnostic that writes below output_prefix, not the swimlane alone. Those filenames are fixed, so N children sharing one prefix overwrite each other. The clock anchors stay swimlane-only, because only that reader places its records against a Host timeline. A failure to write that per-dispatch directory is reported as the task's own error, so a full or read-only output_prefix cannot leave the mailbox loop without publishing TASK_DONE and strand the parent on a run that never completes. Offline discovery of chip_swimlane_records.json searches recursively, since an L3 capture now sits two levels deeper than an L2 one. Forward the parent's case, diagnostic and round selection to the resource child that runs an L3 case. That child's argv is built from scratch so it can be narrowed to one nodeid, so an option absent from it never reaches the case at all: an L3 chip-swimlane run passed while writing no artifact. The gap predates this change and was unreachable while L3 rejected the flag outright, and it is invisible to a green pipeline because a dropped diagnostic changes no assertion. A nodeid names a whole SceneTestCase class and --case filters inside its single test_run item at run time, so a child without the selector runs every case of the class the parent narrowed to one. Validate group membership, Rank uniqueness, callable identity, and sidecar/path agreement during conversion. Retain explicit local-dN fallback for legacy and individual submissions, add dispatch-id selection, and reject a view pid too wide for the per-Rank namespace stride. Discovery keeps each capture's identity in the same record as its path rather than in a parallel list, which also holds the tool to the Python 3.9 floor the project declares. A remainder that cannot be paired by dN downgrades to a warning instead of discarding the parent-identity targets, which pair by run and task slot and are unaffected by it. Convert each rankN/dN capture on its own relative timeline when a capture is known to be below level 4, since cross-Rank merging needs the Host/Device clock anchors that only level 4 collects. Update SceneTest postprocessing, and cover the mailbox, identity, converter, child-argv, and worker paths. --- conftest.py | 73 ++- docs/dfx/chip-swimlane-profiling.md | 106 +++- docs/dfx/hbg-bind-phases.md | 5 +- docs/task-flow.md | 1 + docs/user/reference/python-api.md | 1 + python/bindings/task_interface.cpp | 12 +- python/simpler/worker.py | 170 +++++- simpler_setup/scene_test.py | 174 +++++- simpler_setup/tools/README.md | 45 +- simpler_setup/tools/clock_correlation.py | 27 + .../tools/sched_overhead_analysis.py | 13 +- simpler_setup/tools/swimlane_converter.py | 526 +++++++++++++++++- src/a2a3/platform/sim/host/device_runner.cpp | 5 +- src/a5/platform/sim/host/device_runner.cpp | 5 +- src/common/hierarchical/remote_wire.cpp | 3 + src/common/hierarchical/worker_manager.cpp | 15 + src/common/hierarchical/worker_manager.h | 11 +- .../platform/include/host/clock_correlation.h | 13 + .../onboard/host/device_runner_base.cpp | 13 +- .../onboard/host/device_runner_base.h | 3 + .../shared/host/chip_swimlane_collector.cpp | 30 + .../platform/sim/host/device_runner_base.cpp | 12 +- .../platform/sim/host/device_runner_base.h | 3 + src/common/task_interface/call_config.h | 9 +- tests/ut/cpp/hierarchical/test_scheduler.cpp | 33 +- tests/ut/cpp/types/test_call_config.cpp | 2 +- tests/ut/py/test_chip_worker.py | 81 +++ tests/ut/py/test_clock_correlation.py | 4 + tests/ut/py/test_scene_level_selection.py | 76 ++- tests/ut/py/test_scene_test_cli_contract.py | 123 ++++ tests/ut/py/test_sched_overhead_analysis.py | 23 + tests/ut/py/test_swimlane_converter.py | 329 +++++++++++ tests/ut/py/test_worker/test_host_worker.py | 100 +++- 33 files changed, 1928 insertions(+), 118 deletions(-) diff --git a/conftest.py b/conftest.py index f7d38660eb..478161220d 100644 --- a/conftest.py +++ b/conftest.py @@ -722,24 +722,22 @@ def sort_key(item): items.sort(key=sort_key) - # L3 perf collection is not supported yet: a single L3 case forks N chip-processes - # that all write chip_swimlane_records_.json to the same directory with - # second-precision timestamps, so they trample each other. Block the - # combination up front; waiting for a proper device-id-in-filename fix. + # The automatic rankN/dN layout is scoped to one same-host L3 Worker. Every + # level above NODE owns several L3 Workers, each of which numbers its local + # chips from zero, so accepting one would reintroduce directory collisions. if config.getoption("--enable-chip-swimlane", default=0) and config.getoption("--rounds", default=1) <= 1: - l3_items = [ - i - for i in items - if _item_scene_level(i) == SceneTestLevel.NODE and not any(m.name == "skip" for m in i.iter_markers()) + multi_node_items = [ + item + for item in items + if (_item_scene_level(item) or SceneTestLevel.CHIP) > SceneTestLevel.NODE + and not any(marker.name == "skip" for marker in item.iter_markers()) ] - if l3_items: - sample = ", ".join(sorted({i.nodeid for i in l3_items})[:3]) - more = "" if len(l3_items) <= 3 else f" (+{len(l3_items) - 3} more)" + if multi_node_items: + sample = ", ".join(sorted({item.nodeid for item in multi_node_items})[:3]) + more = "" if len(multi_node_items) <= 3 else f" (+{len(multi_node_items) - 3} more)" raise pytest.UsageError( - f"--enable-chip-swimlane is not supported for L3 tests yet — " - f"multi-chip-process filename collision unresolved. " - f"L3 items in this session: {sample}{more}. " - f"Either drop --enable-chip-swimlane or scope to L2 with --level 2." + "--enable-chip-swimlane supports automatic multi-Rank merging only for same-host L3 tests; " + f"NETWORK1/L4 needs a node namespace before it is safe. Items: {sample}{more}." ) @@ -872,7 +870,45 @@ def _strip_value_options(args, options): return stripped -def _resource_child_command(spec, device_ids, platform, manual_mode): +# Options that change what a case does rather than which case runs. The +# resource child's argv is built from scratch so it can be narrowed to one +# nodeid (see _build in the dispatcher), which means nothing reaches it that is +# not listed here — a diagnostic the parent asked for is silently dropped +# otherwise, and the run passes while producing no artifact at all. +# +# The options that select which case runs — --manual and --case — are forwarded +# by _resource_child_command alongside the nodeid they refine, because a nodeid +# names a whole SceneTestCase class: `--case` filters inside its single +# `test_run` item at run time, so a child that does not receive it runs every +# case of the class the parent narrowed to one. +_RESOURCE_CHILD_VALUE_OPTIONS = ( + ("--rounds", 1), + ("--enable-chip-swimlane", 0), + ("--dump-args", 0), + ("--enable-pmu", 0), +) +_RESOURCE_CHILD_FLAG_OPTIONS = ( + "--skip-golden", + "--enable-dep-gen", + "--enable-scope-stats", + "--enable-swimlane-overhead", +) + + +def _resource_child_diagnostic_argv(cfg): + """Forward the parent's diagnostic and round selection to a resource child.""" + argv = [] + for option, unset in _RESOURCE_CHILD_VALUE_OPTIONS: + value = cfg.getoption(option, default=unset) + if value != unset: + argv.extend([option, str(value)]) + for option in _RESOURCE_CHILD_FLAG_OPTIONS: + if cfg.getoption(option, default=False): + argv.append(option) + return argv + + +def _resource_child_command(spec, device_ids, platform, manual_mode, cfg): command = [ sys.executable, "-m", @@ -888,6 +924,9 @@ def _resource_child_command(spec, device_ids, platform, manual_mode): if platform: command.extend(["--platform", platform]) command.extend(["--manual", manual_mode]) + for selector in cfg.getoption("--case", default=None) or []: + command.extend(["--case", str(selector)]) + command.extend(_resource_child_diagnostic_argv(cfg)) return command @@ -1050,7 +1089,7 @@ def _build(ids, _spec=spec): # this job in the same subprocess, which has only this job's # allocated devices — e.g. TestL3Group (needs 2) would fail # inside TestL3ChildMemory's 1-device subprocess. - return _resource_child_command(_spec, ids, platform, manual_mode) + return _resource_child_command(_spec, ids, platform, manual_mode, session.config) jobs.append( _ps.Job( diff --git a/docs/dfx/chip-swimlane-profiling.md b/docs/dfx/chip-swimlane-profiling.md index 1521480851..414fa6bf98 100644 --- a/docs/dfx/chip-swimlane-profiling.md +++ b/docs/dfx/chip-swimlane-profiling.md @@ -165,6 +165,43 @@ runs): Filenames are fixed (no per-file timestamp) — the directory is the per-task uniqueness boundary. +For L3 runs, each forked ChipWorker writes below its own `rankN/dN` directory. +The filenames above are fixed, so N children sharing one `output_prefix` would +overwrite each other — the separation therefore covers **every** diagnostic that +writes below `output_prefix`, not just the swimlane: + +```text +/ +├── rank0/d0/ +│ ├── chip_swimlane_records.json # --enable-chip-swimlane +│ ├── dispatch_identity.json # always, whenever any diagnostic is on +│ ├── deps.json # --enable-dep-gen +│ └── scope_stats/ # --enable-scope-stats +├── rank1/d0/ +│ └── ... +└── l3_swimlane.json # cross-Rank trace (added by converter) +``` + +Here `rankN` is the logical ChipWorker index and `dN` is that worker's local +capture index. It is a storage-order suffix, not a globally comparable dispatch +ID. `dispatch_identity.json` records the parent scheduler identity: `run_id`, +`task_slot`, `group_index`, and `group_size`, plus the endpoint-local dispatch +and pipeline diagnostics. All members submitted through one +`submit_next_level_group` share `(run_id, task_slot)` and have distinct +`group_index` values. Individually submitted tasks do not share that identity; +the current postprocessor therefore retains local-capture-index pairing for +them and requires symmetric `dN` sets. + +Automatic merging is limited to one same-host L3 Worker. NETWORK1/L4 is +rejected until the layout also carries a node namespace. Every Rank must expose +the same complete set of local capture indexes; the postprocessor refuses +asymmetric sets instead of guessing pairings. + +Cross-Rank merging needs `--enable-chip-swimlane 4` on every Rank, because the +Host/Device clock anchors that level 4 collects are what put the Ranks on a +common timeline. A lower level still captures per Rank; the postprocessor then +converts each `rankN/dN` capture on its own relative timeline and says so. + `chip_swimlane_records.json` carries the raw records. **There are two layers to be aware of:** @@ -338,11 +375,74 @@ python -m simpler_setup.tools.swimlane_converter \ # Custom output path python -m simpler_setup.tools.swimlane_converter \ outputs/_/chip_swimlane_records.json -o my_trace.json + +# Same-host L3: merge rankN/d0 captures onto one CLOCK_MONOTONIC timeline +python -m simpler_setup.tools.swimlane_converter \ + build_output//dfx_outputs --dispatch d0 + +# Prefer the parent group identity when Rank-local dN suffixes differ +python -m simpler_setup.tools.swimlane_converter \ + build_output//dfx_outputs --dispatch-id 17:5 ``` -The output is `outputs/_/merged_swimlane.json` (or your -`-o` override). Open and drag the file -in. The trace contains: +For directory input, the default output is `dfx_outputs/l3_swimlane.json`. +Every Rank must be a level-4 capture under +`rankN//`, with successful clock anchors and the same +`metadata.host_clock_domain_id`. The converter preserves real Rank start skew, +adds Rank-specific PID/name/flow namespaces, and reports clock uncertainty and +anchor-group observer overhead in trace metadata. + +For new group captures, `--dispatch-id RUN_ID:TASK_SLOT` selects the common +parent DAG node and resolves each Rank's actual `dN` path through +`dispatch_identity.json`. SceneTest does this automatically. `--dispatch dN` +remains the compatibility selector for old captures and for independently +submitted per-Rank tasks; it fails if available sidecars show that the selected +paths belong to different parent groups. + +Host-orchestrated level-4 runs retain their existing clock anchors. For +Device/AICPU orchestration, anchors are additionally enabled only when the +ChipWorker marks the capture with `CallConfig.capture_clock_anchors`, which it +does for an L3 chip-swimlane capture, at the common launch boundary before +collectors and kernels start. Both modes sample again after AICPU/AICore +execution completes. Existing single-card Device/AICPU level-4 captures +therefore keep their prior relative timeline and do not pay the new anchor cost. + +`capture_clock_anchors` says only *what the runtime does* — sample the two +clocks — never why. Rank, group and merge are concepts of the layer above: the +platform runner that reads this flag has no notion of a Rank, and no runtime or +platform code parses the `rankN/dN` path. The two are deliberately separate +switches, because the directory is artifact separation that every diagnostic +needs while the anchors are consumed only by the swimlane reader. An L3 run with +`--enable-dep-gen` alone therefore gets its own `rankN/dN` directory and pays no +anchor cost. + +**The opening anchor sits at a different point in each runtime**, because each +takes it at the earliest point preceding every device timestamp it records: + +| Runtime | Opening anchor | Calibrated interval covers | +| ------- | -------------- | -------------------------- | +| `host_build_graph` | before Host orchestration (`host_phase_pool_arm`) | bind, H2D, and execution | +| `tensormap_and_ringbuffer` | before kernel launch (`start_shared_collectors_for_run`) | execution only | + +Both close on `post_device_execution`. So the two runtimes' calibrated intervals +are not comparable in length, and a `host_build_graph` interpolation spans work +a `tensormap_and_ringbuffer` one does not. This does not affect +`max_uncertainty_ns`, which depends only on each anchor group's own sampling +RTT. The serialized position name `pre_host_orchestration` predates the +Device/AICPU case — read it as "start of the calibrated interval", not as a +claim about Host orchestration. + +The default output depends on which input form was used, and `-o` overrides +either: + +| Input | Default output | +| ----- | -------------- | +| a records file | `outputs/_/merged_swimlane.json` | +| a `dfx_outputs` directory | `/l3_swimlane.json` | + +Open and drag the file in. Both forms produce the +same lane structure — the directory form repeats it once per Rank under the +`rankN / ` process names. The trace contains: - **Orchestrator** (pid=1) — per-submit `orch_submit` envelope blocks (level >= 4). diff --git a/docs/dfx/hbg-bind-phases.md b/docs/dfx/hbg-bind-phases.md index 2152c8dd75..dec438bd98 100644 --- a/docs/dfx/hbg-bind-phases.md +++ b/docs/dfx/hbg-bind-phases.md @@ -347,8 +347,9 @@ silently: `bind phase=` lines alone, so Recipe A's environment collects no records. 2. **A diagnostic flag must be on**, because that is what makes `CallConfig.output_prefix` non-empty. `--enable-scope-stats` is the cheapest - for an L3 case; `--enable-chip-swimlane` raises `NotImplementedError` for - `level=3` (per-chip-process filename collision). + choice when only Host phase records are needed. `--enable-chip-swimlane` is + also supported for same-host L3 and writes each ChipWorker capture below a + separate `rankN/dN` directory, but it collects substantially more data. 3. **`--rounds` must be 1.** `rounds > 1` force-disables every diagnostic flag — this one does warn, ` disabled: --rounds > 1` per flag ([`simpler_setup/scene_test.py`](../../simpler_setup/scene_test.py)), but the diff --git a/docs/task-flow.md b/docs/task-flow.md index abfb7a0068..cfb29f325c 100644 --- a/docs/task-flow.md +++ b/docs/task-flow.md @@ -220,6 +220,7 @@ struct CallConfig { int32_t enable_pmu = 0; // 0 = disabled; >0 selects PMU event type int32_t enable_dep_gen = 0; int32_t enable_scope_stats = 0; + int32_t capture_clock_anchors = 0; // set by the ChipWorker child, not by callers char output_prefix[1024] = {}; // future fields here - same POD used at all levels }; diff --git a/docs/user/reference/python-api.md b/docs/user/reference/python-api.md index b162eda7ad..c0f3e0d240 100644 --- a/docs/user/reference/python-api.md +++ b/docs/user/reference/python-api.md @@ -140,6 +140,7 @@ takes a device pointer; `DataType` carries the element types. | `enable_pmu` | `0` | `0` off; `>0` selects the event type | | `enable_dep_gen` | `0` | Emit the dependency graph | | `enable_scope_stats` | `0` | Writes `/scope_stats/scope_stats.jsonl` | +| `capture_clock_anchors` | `False` | Anchors the Host and Device clocks so device timestamps land on an absolute Host timeline. Set by the ChipWorker child for an L3 chip-swimlane capture; not a caller knob | | `output_prefix` | `""` | **Required whenever any diagnostic is enabled** | | `runtime_env` | — | `ring_task_window`, `ring_heap`, `ring_dep_pool`; `tensormap_and_ringbuffer` only | diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index abd737aadb..a65dc8e4d8 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -2787,6 +2787,15 @@ NB_MODULE(_task_interface, m) { c.enable_scope_stats = v ? 1 : 0; } ) + .def_prop_rw( + "capture_clock_anchors", + [](const CallConfig &c) { + return static_cast(c.capture_clock_anchors); + }, + [](CallConfig &c, bool v) { + c.capture_clock_anchors = v ? 1 : 0; + } + ) .def_prop_rw( "output_prefix", [](const CallConfig &c) -> std::string { @@ -2809,7 +2818,8 @@ NB_MODULE(_task_interface, m) { << ", enable_chip_swimlane=" << self.enable_chip_swimlane << ", enable_dump_args=" << self.enable_dump_args << ", enable_pmu=" << self.enable_pmu << ", enable_dep_gen=" << (self.enable_dep_gen ? "True" : "False") - << ", enable_scope_stats=" << (self.enable_scope_stats ? "True" : "False"); + << ", enable_scope_stats=" << (self.enable_scope_stats ? "True" : "False") + << ", capture_clock_anchors=" << (self.capture_clock_anchors ? "True" : "False"); if (self.runtime_env.any()) { append_ring_values(os, "runtime_env.ring_task_window", true, self.runtime_env.ring_task_window); append_ring_values(os, "runtime_env.ring_heap", true, self.runtime_env.ring_heap); diff --git a/python/simpler/worker.py b/python/simpler/worker.py index 785bb74cc5..774b801c80 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -319,13 +319,14 @@ def _host_spans_active() -> bool: _OFF_CALLABLE = 8 _OFF_CONFIG = 16 # Packed CallConfig wire layout — must match call_config.h byte for byte: -# 6 int32 (aicpu_thread_num, enable_chip_swimlane, enable_dump_args, -# enable_pmu, enable_dep_gen, enable_scope_stats) + uint64 ring sizing -# overrides (3 per-ring arrays of RUNTIME_ENV_RING_COUNT: ring_task_window, -# ring_heap, ring_dep_pool) + 1024-byte NUL-terminated output_prefix. Log config -# travels separately via ChipWorker.init(log_level) — not on per-task wire. +# 7 int32 (aicpu_thread_num, enable_chip_swimlane, enable_dump_args, +# enable_pmu, enable_dep_gen, enable_scope_stats, capture_clock_anchors) + uint64 +# ring sizing overrides (3 per-ring arrays of RUNTIME_ENV_RING_COUNT: +# ring_task_window, ring_heap, ring_dep_pool) + 1024-byte NUL-terminated +# output_prefix. Log config travels separately via ChipWorker.init(log_level) — +# not on per-task wire. _RUNTIME_ENV_UINT64_FIELD_COUNT = 3 * RUNTIME_ENV_RING_COUNT -_CFG_FMT = struct.Struct("=iiiiii" + ("Q" * _RUNTIME_ENV_UINT64_FIELD_COUNT) + "1024s") +_CFG_FMT = struct.Struct("=iiiiiii" + ("Q" * _RUNTIME_ENV_UINT64_FIELD_COUNT) + "1024s") # The generation-safe pipeline lease follows CONFIG. Args start after the # lease, rounded up to 8 bytes so the first # Tensor.data (uint64_t at OFF_ARGS+8) is 8-byte aligned, avoiding @@ -357,14 +358,17 @@ def _host_spans_active() -> bool: _OFF_FRAME_SLOT_ID = _OFF_ACCEPTED - 24 _OFF_FRAME_GENERATION = _OFF_ACCEPTED - 16 _OFF_FRAME_DISPATCH_ID = _OFF_ACCEPTED - 8 -_TASK_PROTOCOL_VERSION = 3 +_OFF_FRAME_TASK_SLOT = _OFF_ACCEPTED - 48 +_OFF_FRAME_GROUP_INDEX = _OFF_ACCEPTED - 56 +_OFF_FRAME_GROUP_SIZE = _OFF_ACCEPTED - 64 +_TASK_PROTOCOL_VERSION = 4 # Mirrors MAILBOX_OFF_SHUTDOWN / MAILBOX_SHUTDOWN_REQUESTED: termination is a # sticky one-way word on the control frame, not a MailboxState. _OFF_STATE has # three writers (parent CONTROL_REQUEST, child CONTROL_DONE, C++ # return-to-IDLE), any of which overwrites a _SHUTDOWN store; only a # terminating parent writes this word, 0 -> 1, and nothing clears it. The word # is reserved on every frame so a task-args blob can never reach it. -_OFF_SHUTDOWN = _OFF_FRAME_PROTOCOL - 8 +_OFF_SHUTDOWN = _OFF_ACCEPTED - 72 _SHUTDOWN_REQUESTED = 1 _MAILBOX_ARGS_CAPACITY = _OFF_SHUTDOWN - _OFF_TASK_ARGS_BLOB _OFF_CONTROL_CALLABLE_HASH = _OFF_ARGS + 32 @@ -1980,6 +1984,74 @@ def _read_task_digest(buf) -> bytes: return bytes(buf[_OFF_TASK_CALLABLE_HASH : _OFF_TASK_CALLABLE_HASH + CALLABLE_HASH_DIGEST_BYTES]) +def _read_task_frame_identity(buf: memoryview) -> tuple[int, int, int, int, int, int, int, int]: + """Read the versioned parent-run identity from one task-frame trailer.""" + return ( + struct.unpack_from("=Q", buf, _OFF_FRAME_PROTOCOL)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_RUN_ID)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_SLOT_ID)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_GENERATION)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_DISPATCH_ID)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_TASK_SLOT)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_GROUP_INDEX)[0], + struct.unpack_from("=Q", buf, _OFF_FRAME_GROUP_SIZE)[0], + ) + + +def _config_diagnostics_any(cfg: CallConfig) -> bool: + """Mirror of `CallConfig::diagnostics_any()`, which nanobind does not bind.""" + return bool( + cfg.enable_chip_swimlane + or cfg.enable_dump_args + or cfg.enable_pmu + or cfg.enable_dep_gen + or cfg.enable_scope_stats + ) + + +def _write_dispatch_identity_sidecar( + output_prefix: str, + *, + frame_identity: tuple[int, int, int, int, int, int, int, int], + chip_rank: int, + capture_index: int, + callable_digest: bytes, +) -> None: + """Write parent-DAG identity beside one Rank-local diagnostic artifact.""" + protocol, run_id, pipeline_slot, generation, endpoint_dispatch_id, task_slot, group_index, group_size = ( + frame_identity + ) + if ( + protocol != _TASK_PROTOCOL_VERSION + or run_id == 0 + or generation == 0 + or endpoint_dispatch_id == 0 + or group_size == 0 + or group_index >= group_size + ): + raise RuntimeError(f"invalid diagnostic task frame identity {frame_identity}") + os.makedirs(output_prefix, exist_ok=True) + path = os.path.join(output_prefix, "dispatch_identity.json") + temporary_path = f"{path}.{os.getpid()}.tmp" + payload = { + "schema_version": 1, + "run_id": run_id, + "task_slot": task_slot, + "group_index": group_index, + "group_size": group_size, + "chip_rank": chip_rank, + "local_capture_index": capture_index, + "endpoint_dispatch_id": endpoint_dispatch_id, + "pipeline_slot": pipeline_slot, + "pipeline_generation": generation, + "callable_digest": callable_digest.hex(), + } + with open(temporary_path, "w") as file: + json.dump(payload, file, indent=2) + file.write("\n") + os.replace(temporary_path, path) + + def _format_digest(digest: bytes) -> str: return "sha256:" + digest.hex() @@ -2801,6 +2873,7 @@ def _run_chip_main_loop( # noqa: PLR0913, PLR0915 -- fork-child entry: every de on_task_done_success=None, prepared: set[int] | None = None, task_frame_count: int = 1, + chip_rank: int | None = None, ) -> None: """Chip-process handlers for `_run_mailbox_loop`. @@ -2836,16 +2909,48 @@ def _run_chip_main_loop( # noqa: PLR0913, PLR0915 -- fork-child entry: every de ) ) global_domain_store = _L2GlobalDomainStore() + diagnostic_capture_index = 0 + + def read_task_config( + task_buf: memoryview, + frame_identity: tuple[int, int, int, int, int, int, int, int], + callable_digest: bytes, + ) -> CallConfig: + nonlocal diagnostic_capture_index + capture_index = diagnostic_capture_index + cfg = _read_config_from_mailbox( + task_buf, + chip_rank=chip_rank, + capture_index=capture_index, + ) + # Same gate as the rankN/dN redirect in _read_config_from_mailbox: a loop + # with no Rank identity writes at the case root, and a sidecar naming a + # Rank it does not have would be a lie. + if chip_rank is None or not (_config_diagnostics_any(cfg) and cfg.output_prefix): + return cfg + _write_dispatch_identity_sidecar( + cfg.output_prefix, + frame_identity=frame_identity, + chip_rank=chip_rank, + capture_index=capture_index, + callable_digest=callable_digest, + ) + diagnostic_capture_index += 1 + return cfg def handle_task(task_buf) -> tuple[int, str]: task_addr = ctypes.addressof(ctypes.c_char.from_buffer(task_buf)) digest = _read_task_digest(task_buf) + frame_identity = _read_task_frame_identity(task_buf) cid = identity_table.get(digest) - cfg = _read_config_from_mailbox(task_buf) code = 0 msg = "" try: + # Inside the try because it writes the diagnostic sidecar: a full or + # read-only output_prefix must surface as this task's error, not as + # an exception that leaves the loop before TASK_DONE is published. + cfg = read_task_config(task_buf, frame_identity, digest) if cid is None: raise RuntimeError(f"callable hash {_format_digest(digest)} not registered") # Run only consumes a prepared slot — it never lazily @@ -3049,7 +3154,7 @@ class _StagedFrame: index: int frame_buf: memoryview frame_addr: int - identity: tuple[int, int, int, int, int] + identity: tuple[int, int, int, int, int, int, int, int] cid: int config: CallConfig activated: bool @@ -3058,15 +3163,6 @@ class _StagedFrame: staged_frames: dict[int, _StagedFrame] = {} - def read_identity(frame_buf: memoryview) -> tuple[int, int, int, int, int]: - return ( - struct.unpack_from("=Q", frame_buf, _OFF_FRAME_PROTOCOL)[0], - struct.unpack_from("=Q", frame_buf, _OFF_FRAME_RUN_ID)[0], - struct.unpack_from("=Q", frame_buf, _OFF_FRAME_SLOT_ID)[0], - struct.unpack_from("=Q", frame_buf, _OFF_FRAME_GENERATION)[0], - struct.unpack_from("=Q", frame_buf, _OFF_FRAME_DISPATCH_ID)[0], - ) - def task_frame_references_digest(digest: bytes) -> bool: live_states = (_TASK_READY, _PREPARE_READY, _ACTIVATE, _FRAME_STAGED, _TASK_LAUNCHED) for index, frame_buf in enumerate(frame_bufs): @@ -3084,8 +3180,8 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: frame_buf = frame_bufs[index] frame_addr = frame_addrs[index] try: - identity = read_identity(frame_buf) - protocol, run_id, slot_id, generation, dispatch_id = identity + identity = _read_task_frame_identity(frame_buf) + protocol, run_id, slot_id, generation, dispatch_id, _task_slot, _group_index, _group_size = identity pipeline_slot, pipeline_reserved, pipeline_generation = _PIPELINE_LEASE_FMT.unpack_from( frame_buf, _OFF_PIPELINE_LEASE ) @@ -3118,7 +3214,7 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: frame_addr=frame_addr, identity=identity, cid=int(cid), - config=_read_config_from_mailbox(frame_buf), + config=read_task_config(frame_buf, identity, digest), activated=initial_state in (_TASK_READY, _ACTIVATE), ) except Exception as e: # noqa: BLE001 @@ -3127,7 +3223,7 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: return None def submit_frame(frame: _StagedFrame) -> None: - _protocol, run_id, slot_id, generation, dispatch_id = frame.identity + _protocol, run_id, slot_id, generation, dispatch_id, _task_slot, _group_index, _group_size = frame.identity # The frame carries the wire blob; the runtime reads the chip POD. The bytes decode # once into the wire TaskArgs, whose tensors resolve to local bases (map-once, cached # by canonical identity) and rebuild at those bases, as the non-pipelined task path @@ -3190,7 +3286,7 @@ def submit_frame(frame: _StagedFrame) -> None: new_frames.append(staged) continue if frame_state == _ACTIVATE and not staged.activated: - if read_identity(staged.frame_buf) != staged.identity: + if _read_task_frame_identity(staged.frame_buf) != staged.identity: stale_message = f"chip_process dev={device_id}: stale activation identity" try: staged.chip_run.abandon() @@ -3309,6 +3405,7 @@ def _chip_process_loop( # noqa: PLR0913 -- fork-child entry: all context (bins, runtime: str = "", prewarm_config=None, enable_sdma: bool = False, + chip_rank: int | None = None, ) -> None: """Runs in forked child process. Loads host_runtime.so in own address space. @@ -3386,12 +3483,18 @@ def _chip_process_loop( # noqa: PLR0913 -- fork-child entry: all context (bins, chip_runtime=runtime, prepared=prepared, task_frame_count=_local_task_frame_count(platform, runtime, int(cw.pipeline_depth)), + chip_rank=chip_rank, ) finally: cw.finalize() -def _read_config_from_mailbox(buf: memoryview) -> CallConfig: +def _read_config_from_mailbox( + buf: memoryview, + *, + chip_rank: int | None = None, + capture_index: int | None = None, +) -> CallConfig: """Reconstruct a CallConfig from the unified mailbox layout.""" ( aicpu_tn, @@ -3400,6 +3503,7 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig: pmu, dep_gen, scope_stats, + _capture_clock_anchors, *ring_values, prefix_bytes, ) = _CFG_FMT.unpack_from(buf, _OFF_CONFIG) @@ -3418,9 +3522,22 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig: cfg.runtime_env.ring_dep_pool = ring_dep_pool # NUL-terminated C string in a 1024-byte field. cfg.output_prefix = prefix_bytes.split(b"\x00", 1)[0].decode("utf-8") - # A forked chip child owns its own log file under the same directory. + # Keep per-process host logs at the case root. Profiling artifacts are + # routed below, after the log directory has been configured, so changing + # capture directories does not add log-directory churn to every dispatch. if cfg.output_prefix: _native_set_host_log_directory(cfg.output_prefix) + if cfg.output_prefix and chip_rank is not None and capture_index is not None and _config_diagnostics_any(cfg): + # Every diagnostic below output_prefix uses a fixed filename, so N + # ChipWorker children sharing one prefix overwrite each other's + # artifacts. rankN/dN is the storage convention that separates them, and + # it is read only by the offline tools: no runtime or platform code + # parses this path, or knows that a Rank is what produced it. + cfg.output_prefix = os.path.join(cfg.output_prefix, f"rank{chip_rank}", f"d{capture_index}") + # Only the swimlane reader places its records against a Host timeline, + # so it alone needs both clocks anchored; the other diagnostics get the + # directory separation without paying for the anchors. + cfg.capture_clock_anchors = bool(cfg.enable_chip_swimlane) return cfg @@ -7966,6 +8083,7 @@ def _setup(): runtime=str(self._config["runtime"]), prewarm_config=self._prewarm_config, enable_sdma=bool(self._config.get("enable_sdma", False)), + chip_rank=idx, ) except BaseException as e: # noqa: BLE001 import traceback as _tb # noqa: PLC0415 diff --git a/simpler_setup/scene_test.py b/simpler_setup/scene_test.py index b09ad3a2aa..5d074d5f93 100644 --- a/simpler_setup/scene_test.py +++ b/simpler_setup/scene_test.py @@ -1042,6 +1042,10 @@ def _run_swimlane_converter( input_path: Path | None = None, func_names_path: Path | None = None, enable_overhead: bool = False, + *, + dispatch: str | None = None, + dispatch_id: str | None = None, + output_path: Path | None = None, ) -> None: """Invoke the bundled swimlane converter as a subprocess. @@ -1063,6 +1067,12 @@ def _run_swimlane_converter( cmd.append(str(input_path)) if func_names_path is not None: cmd += ["--func-names", str(func_names_path)] + if dispatch is not None: + cmd += ["--dispatch", dispatch] + if dispatch_id is not None: + cmd += ["--dispatch-id", dispatch_id] + if output_path is not None: + cmd += ["--output", str(output_path)] if enable_overhead: cmd.append("--overhead") try: @@ -1082,6 +1092,111 @@ def _sanitize_for_filename(s: str) -> str: return "".join(c if c.isalnum() or c in "._-" else "_" for c in s) +# Host/Device clock anchors — and therefore a common cross-Rank timeline — exist +# only at this chip-swimlane level. +_MULTI_RANK_SWIMLANE_LEVEL = 4 + + +def _rank_dirs(output_prefix: Path) -> list[Path]: + """Return the ``rankN`` roots below a case prefix, ordered by Rank.""" + return sorted( + (path for path in output_prefix.glob("rank*") if path.is_dir() and path.name.removeprefix("rank").isdigit()), + key=lambda path: int(path.name.removeprefix("rank")), + ) + + +def _rank_capture_dirs(output_prefix: Path) -> list[Path]: + """Return deterministic ``rankN/dN`` capture roots below a case prefix.""" + captures = [] + for rank_dir in _rank_dirs(output_prefix): + captures.extend( + sorted( + (path for path in rank_dir.glob("d*") if path.is_dir() and path.name.removeprefix("d").isdigit()), + key=lambda path: int(path.name.removeprefix("d")), + ) + ) + return captures + + +def _capture_swimlane_level(records_path: Path) -> int | None: + """Return one capture's ``chip_swimlane_level``, or None if it is unreadable.""" + try: + with records_path.open() as file: + return int(json.load(file)["chip_swimlane_level"]) + except (OSError, KeyError, TypeError, ValueError): + return None + + +def _convert_rank_swimlanes( + case_label: str, + output_prefix: Path, + *, + callable_spec: dict | None, + enable_overhead: bool, + logger: logging.Logger, +) -> None: + """Convert the ``rankN/dN`` captures below one L3 case prefix. + + Cross-Rank merging is what puts every Rank on a common Host timeline, and + only level 4 carries the Host/Device clock anchors that make that possible. + A capture known to be below level 4 is therefore converted on its own Rank's + relative timeline instead. An unreadable level is left to the converter to + reject, so a malformed capture still fails loudly rather than downgrading. + """ + from simpler_setup.tools.swimlane_converter import discover_l3_conversion_targets # noqa: PLC0415 + + def dump_name_map(capture_dir: Path) -> Path | None: + if not callable_spec: + return None + safe_label = _sanitize_for_filename(case_label) + return _dump_name_map(_extract_name_map(callable_spec), capture_dir / f"name_map_{safe_label}.json") + + captures = [path for path in _rank_capture_dirs(output_prefix) if (path / "chip_swimlane_records.json").is_file()] + if not captures: + logger.warning(f"[{case_label}] no Rank capture is present under {output_prefix}") + return + + known_levels = { + level + for level in (_capture_swimlane_level(path / "chip_swimlane_records.json") for path in captures) + if level is not None + } + if known_levels - {_MULTI_RANK_SWIMLANE_LEVEL}: + logger.warning( + f"[{case_label}] cross-Rank merging needs --enable-chip-swimlane {_MULTI_RANK_SWIMLANE_LEVEL} on every " + f"Rank (found {sorted(known_levels)}); converting each Rank capture on its own timeline instead" + ) + for capture_dir in captures: + _run_swimlane_converter( + input_path=capture_dir / "chip_swimlane_records.json", + func_names_path=dump_name_map(capture_dir), + enable_overhead=enable_overhead, + ) + return + + try: + targets = discover_l3_conversion_targets(output_prefix) + except ValueError as error: + logger.warning(f"[{case_label}] {error}") + return + if not targets: + logger.warning(f"[{case_label}] no complete Rank capture is present under {output_prefix}") + return + + for target in targets: + # Directory mode auto-loads each Rank's own sibling name map, so these + # are dumped in place and never passed as a global override. + for capture_dir in target["capture_dirs"]: + dump_name_map(capture_dir) + _run_swimlane_converter( + input_path=output_prefix, + enable_overhead=enable_overhead, + dispatch=target["dispatch"], + dispatch_id=target["dispatch_id"], + output_path=output_prefix / f"{target['output_stem']}.json" if len(targets) > 1 else None, + ) + + def _convert_case_swimlane( case_label: str, output_prefix: Path, @@ -1091,10 +1206,23 @@ def _convert_case_swimlane( """Post-case: invoke the swimlane converter on the perf file the runtime just wrote into ``/chip_swimlane_records.json``. No diff/rename dance — the path is known a priori from CallConfig.output_prefix. + + A run whose chips are forked ChipWorker children writes below ``rankN/dN`` + instead, and its presence is what selects the multi-Rank postprocessor. """ import logging # noqa: PLC0415 logger = logging.getLogger(__name__) + if _rank_dirs(output_prefix): + _convert_rank_swimlanes( + case_label, + output_prefix, + callable_spec=callable_spec, + enable_overhead=enable_overhead, + logger=logger, + ) + return + perf_file = output_prefix / "chip_swimlane_records.json" if not perf_file.exists(): logger.warning(f"[{case_label}] {perf_file} not produced; skipping conversion") @@ -1205,12 +1333,23 @@ def finalize_diagnostic_outputs( ) -> None: """Run the postprocessors shared by SceneTest and standalone drivers.""" prefix = Path(output_prefix) + rank_capture_dirs = _rank_capture_dirs(prefix) if chip_swimlane: _convert_case_swimlane(case_label, prefix, callable_spec=callable_spec, enable_overhead=swimlane_overhead) if dep_gen: - _graph_case_dep_gen(case_label, prefix, callable_spec=callable_spec) + dep_targets = [path for path in rank_capture_dirs if (path / "deps.json").is_file()] + if dep_targets: + for target in dep_targets: + _graph_case_dep_gen(case_label, target, callable_spec=callable_spec) + else: + _graph_case_dep_gen(case_label, prefix, callable_spec=callable_spec) if scope_stats: - _plot_case_scope_stats(case_label, prefix) + scope_targets = [path for path in rank_capture_dirs if (path / "scope_stats" / "scope_stats.jsonl").is_file()] + if scope_targets: + for target in scope_targets: + _plot_case_scope_stats(case_label, target) + else: + _plot_case_scope_stats(case_label, prefix) def _name_failing_case(exc: BaseException, cls_name: str, case_name: str) -> None: @@ -1819,18 +1958,6 @@ def _run_and_validate_l3( # noqa: PLR0913 -- threads CLI diagnostic flags + L3 enable_scope_stats=False, output_prefix="", ): - # Defensive belt-and-braces: the pytest dispatcher and run_module both - # block --enable-chip-swimlane for L3 at the CLI boundary. Catch any code - # path that reaches here with the flag on anyway (direct API use, - # future refactors) so we fail loud rather than produce garbage perf - # files. Lift once the runtime embeds device_id in the perf filename. - if enable_chip_swimlane: - raise NotImplementedError( - "L3 profiling is not supported yet (multi-chip-process perf " - "filename collision). Gate at the CLI level in " - "conftest.pytest_collection_modifyitems / scene_test.run_module." - ) - params = case.get("params", {}) config_dict = case.get("config", {}) skip_golden = skip_golden or bool(case.get("skip_golden", self.SKIP_GOLDEN)) @@ -2125,6 +2252,11 @@ def run_module(module_name): # noqa: PLR0912, PLR0915 -- CLI parsing + dispatch parser.add_argument( "--level", type=int, + # SceneTestCase reaches level 2 and 3 only: build_callable rejects + # anything else, and the NETWORK1 scene tests are plain pytest + # functions rather than classes. Should a class ever reach L4, + # mirror the pytest-side multi-Rank swimlane guard here first — + # rankN numbering is per L3 Worker and collides above it. choices=[2, 3], default=None, help="Only run classes with this _st_level (child-mode marker when combined with --runtime)", @@ -2254,20 +2386,6 @@ def run_module(module_name): # noqa: PLR0912, PLR0915 -- CLI parsing + dispatch for cls, case in selected: selected_by_cls.setdefault(cls, []).append(case) - # L3 profiling not supported yet (multi-chip-process filename collision). - # Mirror the pytest-side guard so standalone users get the same early-fail. - if args.enable_chip_swimlane: - l3_classes = sorted(cls.__name__ for cls in selected_by_cls if cls._st_level == 3) - if l3_classes: - print( - f"ERROR: --enable-chip-swimlane is not supported for L3 tests yet — " - f"multi-chip-process filename collision unresolved. " - f"L3 classes selected: {', '.join(l3_classes)}. " - f"Either drop --enable-chip-swimlane or scope to L2 with --level 2.", - file=sys.stderr, - ) - sys.exit(2) - # Child mode: both --runtime and --level set. Run inline without # spawning further subprocesses; this is the path dispatcher # children take after we re-enter run_module. diff --git a/simpler_setup/tools/README.md b/simpler_setup/tools/README.md index 2b7722f8d5..07f0fc7f4e 100644 --- a/simpler_setup/tools/README.md +++ b/simpler_setup/tools/README.md @@ -156,8 +156,34 @@ python -m simpler_setup.tools.swimlane_converter outputs/_/chip_swimla # Reuse a deps.json captured in an earlier dep_gen run (different output dir) python -m simpler_setup.tools.swimlane_converter outputs/_/chip_swimlane_records.json \ --deps-json outputs/_/deps.json + +# Merge one same-host L3 dispatch laid out as rankN/d0/chip_swimlane_records.json +python -m simpler_setup.tools.swimlane_converter build_output//dfx_outputs \ + --dispatch d0 -o build_output//dfx_outputs/l3_swimlane.json + +# Merge one parent group dispatch even when its members use different dN paths +python -m simpler_setup.tools.swimlane_converter build_output//dfx_outputs \ + --dispatch-id 17:5 -o build_output//dfx_outputs/l3_swimlane.json ``` +Directory mode requires level-4 captures with successful Host/Device clock +anchors and the same non-empty `metadata.host_clock_domain_id`. New captures +derive that ID from the Linux boot ID; older captures remain supported in +single-file mode. Every Rank loads its own sibling `deps.json` and unique +`name_map*.json`, so the single-file override options are intentionally rejected +in directory mode. + +L3 SceneTest runs create `rank/d/` automatically and +invoke this directory mode after the case. Each new capture also contains +`dispatch_identity.json`. Members of one `submit_next_level_group` are paired by +their common `(run_id, task_slot)` even if their local `dN` suffixes differ; the +trace metadata records `dispatch_pairing: parent_dispatch_identity`. Old +captures and individually submitted per-Rank tasks fall back to symmetric `dN` +pairing and record `dispatch_pairing: local_capture_index`. A `dN` selector whose +sidecars identify different parent groups is rejected instead of producing a +plausible but incorrectly paired trace. This layout is scoped to one same-host +L3 Worker; NETWORK1/L4 needs an additional node namespace. + > Dependency arrows in the Perfetto trace come from `deps.json` (dep_gen > replay). The device hot path no longer records fanout, so the typical > workflow is **two runs**: a one-time `--enable-dep-gen` capture per @@ -196,14 +222,20 @@ SPMD tasks are present. | Option | Short | Description | | ------ | ----- | ----------- | -| `input` | | Input JSON file (chip_swimlane_records_*.json). If omitted, the latest file in outputs/ is used | -| `--output` | `-o` | Output JSON file (default: outputs/merged_swimlane_``.json) | -| `--kernel-config` | `-k` | Path to kernel_config.py, used for function name mapping | -| `--func-names` | | Path to name_map*.json (SceneTest format) for function name mapping | -| `--deps-json` | | Path to a dep_gen `deps.json` (defaults to sibling of input). Without one, no dependency arrows are drawn. | +| `input` | | Input JSON file (chip_swimlane_records_*.json), **or** a `dfx_outputs` directory containing `rank*/dN/` for directory mode. If omitted, the latest file in outputs/ is used | +| `--output` | `-o` | Output JSON file (default: `merged_swimlane.json` beside a file input, `l3_swimlane.json` inside a directory input) | +| `--dispatch` | | Directory mode only: local capture directory to merge across Ranks, e.g. `d0`. Mutually exclusive with `--dispatch-id` | +| `--dispatch-id` | | Directory mode only: parent dispatch identity to merge, formatted `RUN_ID:TASK_SLOT`. Resolves each Rank's own `dN` through `dispatch_identity.json`. Mutually exclusive with `--dispatch` | +| `--kernel-config` | `-k` | Path to kernel_config.py, used for function name mapping. Rejected in directory mode | +| `--func-names` | | Path to name_map*.json (SceneTest format) for function name mapping. Rejected in directory mode | +| `--deps-json` | | Path to a dep_gen `deps.json` (defaults to sibling of input). Without one, no dependency arrows are drawn. Rejected in directory mode | | `--overhead` | | Add the 8-line Overhead Analysis counter group (needs `deps.json`). See [sched-overhead-model](../../docs/dfx/sched-overhead-model.md). | | `--verbose` | `-v` | Enable verbose output | +Directory mode auto-loads each Rank's own sibling `name_map*.json` and +`deps.json`, which is why the three global override options above are rejected +there rather than silently applied to every Rank. + ### Outputs The tool produces three kinds of output: @@ -212,7 +244,8 @@ The tool produces three kinds of output: A Chrome Trace Event format JSON file that can be visualized in Perfetto: -- File location: `outputs/merged_swimlane_.json` +- File location: `merged_swimlane.json` beside the input records file, or + `l3_swimlane.json` inside the input `dfx_outputs` directory - Open and drag-and-drop the file to visualize #### 2. Task Statistics diff --git a/simpler_setup/tools/clock_correlation.py b/simpler_setup/tools/clock_correlation.py index 14f0fb5c14..061d4bfbd2 100644 --- a/simpler_setup/tools/clock_correlation.py +++ b/simpler_setup/tools/clock_correlation.py @@ -34,6 +34,8 @@ class ClockAlignment: host_timestamp_quantization_ns: int = 0 start: Optional[ClockAnchor] = None end: Optional[ClockAnchor] = None + pre_anchor_group_duration_ns: Optional[int] = None + post_anchor_group_duration_ns: Optional[int] = None @property def anchor_uncertainty_ns(self): @@ -62,6 +64,12 @@ def metadata(self): self.start.position: self.start.sample_idx, self.end.position: self.end.sample_idx, } + group_durations = { + _A0_POSITION: self.pre_anchor_group_duration_ns, + _A2_POSITION: self.post_anchor_group_duration_ns, + } + if any(duration is not None for duration in group_durations.values()): + out["anchor_group_duration_ns"] = group_durations return out def contains(self, device_cycles): @@ -129,6 +137,23 @@ def _parse_anchor(sample, position, device_quantization_ns): ) +def _anchor_group_duration_ns(samples, position): + bounds = [] + for sample in samples: + if not isinstance(sample, dict) or sample.get("position") != position: + continue + try: + before = int(sample["host_before_ns"]) + after = int(sample["host_after_ns"]) + except (KeyError, TypeError, ValueError): + continue + if before > 0 and after >= before: + bounds.append((before, after)) + if not bounds: + return None + return max(after for _, after in bounds) - min(before for before, _ in bounds) + + def build_clock_alignment( # noqa: PLR0912 clock_anchors, frequency_hz, device_timestamps=(), host_timestamp_quantization_ns=0 ): @@ -184,6 +209,8 @@ def build_clock_alignment( # noqa: PLR0912 host_timestamp_quantization_ns=host_timestamp_quantization_ns, start=start, end=end, + pre_anchor_group_duration_ns=_anchor_group_duration_ns(samples, _A0_POSITION), + post_anchor_group_duration_ns=_anchor_group_duration_ns(samples, _A2_POSITION), ) for timestamp in device_timestamps: timestamp = int(timestamp) diff --git a/simpler_setup/tools/sched_overhead_analysis.py b/simpler_setup/tools/sched_overhead_analysis.py index 76066edca1..62dbb94a13 100644 --- a/simpler_setup/tools/sched_overhead_analysis.py +++ b/simpler_setup/tools/sched_overhead_analysis.py @@ -162,11 +162,18 @@ def task_thread(task): def auto_select_chip_swimlane_records_json(): - """Find the latest outputs//chip_swimlane_records.json (sorted by mtime).""" + """Find the newest ``chip_swimlane_records.json`` under ``outputs/`` by mtime. + + Recursive because the depth varies with the level that produced the capture: + an L2 case writes it at ``outputs//``, while each chip of an L3 case + writes its own below ``outputs//rank/d/``. A fixed one-level + glob finds only the former and reports "no records" for a run that produced + several. + """ outputs_dir = Path.cwd() / "outputs" - files = sorted(outputs_dir.glob("*/chip_swimlane_records.json"), key=lambda p: p.stat().st_mtime, reverse=True) + files = sorted(outputs_dir.rglob("chip_swimlane_records.json"), key=lambda p: p.stat().st_mtime, reverse=True) if not files: - raise FileNotFoundError(f"No outputs/*/chip_swimlane_records.json found under {outputs_dir}") + raise FileNotFoundError(f"No chip_swimlane_records.json found anywhere under {outputs_dir}") return files[0] diff --git a/simpler_setup/tools/swimlane_converter.py b/simpler_setup/tools/swimlane_converter.py index 21898d7dd4..4fc9a8fa08 100644 --- a/simpler_setup/tools/swimlane_converter.py +++ b/simpler_setup/tools/swimlane_converter.py @@ -29,6 +29,7 @@ import bisect import importlib.util import json +import re import sys import traceback from collections import defaultdict @@ -224,8 +225,15 @@ def _collect_graph_execution_instances(tasks, scheduler_phases): # noqa: PLR091 return instances -def read_perf_data(filepath): # noqa: PLR0912, PLR0915 - """Read performance data from a swimlane JSON file. +def read_perf_data(filepath, *, timeline_origin_ns=None): + """Read and decode performance data from a swimlane JSON file.""" + with open(filepath) as file: + data = json.load(file) + return _decode_perf_data(data, timeline_origin_ns=timeline_origin_ns) + + +def _decode_perf_data(data, *, timeline_origin_ns=None): # noqa: PLR0912, PLR0915 + """Decode performance data from an already-loaded swimlane document. Host dumps raw cycle-domain per-stream records plus metadata; this function does the AICore↔AICPU join. Schema: @@ -256,6 +264,10 @@ def read_perf_data(filepath): # noqa: PLR0912, PLR0915 software-tunable). Archived v2 JSON without this column still parses; the field is exposed as 0 for those. + ``timeline_origin_ns`` optionally supplies a Host CLOCK_MONOTONIC origin + shared by several same-host Rank files. The default preserves the existing + single-file origin. + Returns a dict shaped for `generate_chrome_trace_json`, `print_task_statistics`, and `sched_overhead_analysis`: `tasks`, `aicpu_scheduler_phases`, `aicpu_orchestrator_phases`, @@ -279,9 +291,6 @@ def read_perf_data(filepath): # noqa: PLR0912, PLR0915 Raises: ValueError: If the JSON is malformed. """ - with open(filepath) as f: - data = json.load(f) - level = int(data.get("chip_swimlane_level")) if level not in [1, 2, 3, 4]: raise ValueError(f"Unsupported chip_swimlane_level: {level} (expected 1, 2, 3, or 4)") @@ -304,6 +313,7 @@ def read_perf_data(filepath): # noqa: PLR0912, PLR0915 or bool(host_orch_phases_raw) or isinstance(raw_host_capture, dict) ) + clock_anchor_mode = isinstance(metadata.get("clock_anchors"), dict) if orch_phases_raw and host_mode: raise ValueError("both AICPU and host orchestrator phases are present; clock-domain source is ambiguous") @@ -352,9 +362,12 @@ def read_perf_data(filepath): # noqa: PLR0912, PLR0915 for pr in thread_records for field in ("start_host_ns", "end_host_ns") ] - host_origin_ns = int(metadata.get("host_orchestration_origin_ns") or 0) - if host_timestamps and host_origin_ns == 0: - host_origin_ns = min(host_timestamps) + source_host_origin_ns = int( + metadata.get("host_orchestration_origin_ns") or metadata.get("host_timeline_origin_ns") or 0 + ) + if host_timestamps and source_host_origin_ns == 0: + source_host_origin_ns = min(host_timestamps) + host_origin_ns = source_host_origin_ns host_composite_end_us = (max(host_timestamps) - host_origin_ns) / 1000.0 if host_timestamps else 0.0 # AICore lookup keyed by (core_id, reg_task_id). Two dispatches of the @@ -424,7 +437,7 @@ def _track(v): device_timestamps.extend((int(phase.get("start_cycles", 0)), int(phase.get("end_cycles", 0)))) clock_alignment = None - if host_mode: + if host_mode or clock_anchor_mode: clock_alignment = build_clock_alignment( metadata.get("clock_anchors"), clock_freq_hz, @@ -434,6 +447,14 @@ def _track(v): if host_origin_ns == 0 and clock_alignment.start is not None: host_origin_ns = clock_alignment.start.host_mid_ns + source_host_origin_ns = host_origin_ns + if timeline_origin_ns is not None: + if clock_alignment is None or clock_alignment.status != "calibrated": + raise ValueError("a shared timeline origin requires calibrated Host/Device clock anchors") + host_origin_ns = int(timeline_origin_ns) + if host_origin_ns <= 0: + raise ValueError(f"invalid shared timeline origin: {host_origin_ns}") + cycles_to_us_factor = 1_000_000.0 / float(clock_freq_hz) def _to_us(cycles): @@ -622,7 +643,12 @@ def _phase_us(pr): "host_capture": host_capture, "host_records_complete": host_capture_complete, "cross_domain_latency_available": calibrated and host_capture_complete, + "source_timeline_origin_ns": source_host_origin_ns, + "timeline_origin_ns": host_origin_ns, } + host_clock_domain_id = metadata.get("host_clock_domain_id") + if host_clock_domain_id: + out["timeline_metadata"]["host_clock_domain_id"] = str(host_clock_domain_id) if not calibrated: out["timeline_metadata"].update( { @@ -634,6 +660,22 @@ def _phase_us(pr): out["aicpu_orchestrator_phases"] = host_orchestrator_phases else: out["timeline_metadata"]["host_records_missing"] = True + elif clock_anchor_mode: + if clock_alignment is None: + raise RuntimeError("clock anchors are missing their alignment result") + calibrated = clock_alignment.status == "calibrated" + out["timeline_metadata"] = { + "layout": "clock_aligned" if calibrated else "device_relative", + "trace_status": "complete" if calibrated else "partial", + "clock_alignment": clock_alignment.metadata(), + "host_records_complete": False, + "cross_domain_latency_available": False, + "source_timeline_origin_ns": source_host_origin_ns, + "timeline_origin_ns": host_origin_ns, + } + host_clock_domain_id = metadata.get("host_clock_domain_id") + if host_clock_domain_id: + out["timeline_metadata"]["host_clock_domain_id"] = str(host_clock_domain_id) if core_to_thread: out["core_to_thread"] = core_to_thread return out @@ -2804,11 +2846,13 @@ def _find_containing_complete(thread_idx: int, finish_us: float): trace = {"traceEvents": events} if timeline_metadata: trace["metadata"] = timeline_metadata - with open(output_path, "w") as f: - json.dump(trace, f, indent=2) + if output_path is not None: + with open(output_path, "w") as f: + json.dump(trace, f, indent=2) - if verbose: + if verbose and output_path is not None: print(f"JSON written to: {output_path}") + return trace def _build_parser(): @@ -2823,14 +2867,22 @@ def _build_parser(): %(prog)s outputs/_/chip_swimlane_records.json \ -k examples/host_build_graph/paged_attention/kernels/kernel_config.py %(prog)s outputs/_/chip_swimlane_records.json -v + %(prog)s build_output//dfx_outputs --dispatch d0 """, ) parser.add_argument( "input", nargs="?", - help="Input JSON file (.json). If not specified, uses the latest chip_swimlane_records_*.json in outputs/", + help=( + "Input JSON file, or a dfx_outputs directory containing rank*/dN/. " + "If omitted, uses the latest chip_swimlane_records_*.json in outputs/." + ), + ) + parser.add_argument( + "-o", + "--output", + help="Output JSON file (default: merged_swimlane.json for a file, l3_swimlane.json for a directory)", ) - parser.add_argument("-o", "--output", help="Output JSON file (default: /merged_swimlane.json)") parser.add_argument( "-k", "--kernel-config", @@ -2851,6 +2903,14 @@ def _build_parser(): ), ) parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output") + parser.add_argument( + "--dispatch", + help="Local capture directory to merge for directory input, for example d0", + ) + parser.add_argument( + "--dispatch-id", + help="Parent dispatch identity to merge for directory input, formatted as RUN_ID:TASK_SLOT", + ) parser.add_argument( "--overhead", action="store_true", @@ -2887,6 +2947,9 @@ def _resolve_output_path(args, input_path): if args.output: return Path(args.output) + if input_path.is_dir(): + return input_path / "l3_swimlane.json" + # Default: write merged_swimlane.json next to the input. The parent # directory name (e.g. outputs/_/) already disambiguates runs. return input_path.parent / "merged_swimlane.json" @@ -2980,6 +3043,431 @@ def _load_func_names(args, input_path): return {}, None +_RANK_DIR_PATTERN = re.compile(r"rank([0-9]+)") +_DISPATCH_DIR_PATTERN = re.compile(r"d[0-9]+") +_DISPATCH_ID_PATTERN = re.compile(r"([0-9]+):([0-9]+)") +_RANK_PID_STRIDE = 100 + + +def _l3_rank_dirs(root): + root = Path(root) + rank_dirs = sorted(path for path in root.glob("rank*") if path.is_dir()) + if not rank_dirs: + raise ValueError(f"no rankN directories found under {root}") + + discovered = [] + seen_ranks = set() + for rank_dir in rank_dirs: + match = _RANK_DIR_PATTERN.fullmatch(rank_dir.name) + if match is None: + raise ValueError(f"invalid Rank directory name: {rank_dir.name} (expected rankN)") + rank = int(match.group(1)) + if rank in seen_ranks: + raise ValueError(f"duplicate Rank number {rank} under {root}") + seen_ranks.add(rank) + discovered.append((rank, rank_dir)) + return sorted(discovered) + + +def _load_dispatch_identity(capture_dir): + path = Path(capture_dir) / "dispatch_identity.json" + if not path.is_file(): + return None + with path.open() as file: + identity = json.load(file) + if ( + not isinstance(identity, dict) + or isinstance(identity.get("schema_version"), bool) + or identity.get("schema_version") != 1 + ): + raise ValueError(f"unsupported dispatch identity schema: {path}") + + integer_fields = ( + "run_id", + "task_slot", + "group_index", + "group_size", + "chip_rank", + "local_capture_index", + "endpoint_dispatch_id", + "pipeline_slot", + "pipeline_generation", + ) + for field in integer_fields: + value = identity.get(field) + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"dispatch identity {field} must be an integer: {path}") + if identity["run_id"] <= 0 or identity["task_slot"] < 0: + raise ValueError(f"dispatch identity has an invalid parent key: {path}") + if identity["endpoint_dispatch_id"] <= 0 or identity["pipeline_slot"] < 0 or identity["pipeline_generation"] <= 0: + raise ValueError(f"dispatch identity has invalid endpoint diagnostics: {path}") + if identity["group_size"] <= 0 or not 0 <= identity["group_index"] < identity["group_size"]: + raise ValueError(f"dispatch identity has invalid group membership: {path}") + if not re.fullmatch(r"[0-9a-f]{64}", identity.get("callable_digest", "")): + raise ValueError(f"dispatch identity has an invalid callable digest: {path}") + + rank_match = _RANK_DIR_PATTERN.fullmatch(Path(capture_dir).parent.name) + dispatch_match = _DISPATCH_DIR_PATTERN.fullmatch(Path(capture_dir).name) + if rank_match is None or dispatch_match is None: + raise ValueError(f"dispatch identity is not below rankN/dN: {path}") + path_rank = int(rank_match.group(1)) + path_capture_index = int(Path(capture_dir).name.removeprefix("d")) + if identity["chip_rank"] != path_rank or identity["local_capture_index"] != path_capture_index: + raise ValueError(f"dispatch identity disagrees with its rankN/dN path: {path}") + return identity + + +def _validate_parent_dispatch_group(discovered, expected_key): + if not discovered: + raise ValueError(f"no Rank captures found for parent dispatch {expected_key[0]}:{expected_key[1]}") + identities = [identity for _, _, identity in discovered] + group_sizes = {identity["group_size"] for identity in identities} + callable_digests = {identity.get("callable_digest") for identity in identities} + if len(group_sizes) != 1 or len(callable_digests) != 1: + raise ValueError(f"inconsistent metadata for parent dispatch {expected_key[0]}:{expected_key[1]}") + group_size = next(iter(group_sizes)) + if group_size <= 1: + raise ValueError( + f"parent dispatch {expected_key[0]}:{expected_key[1]} is an individual submission; pair it by dN" + ) + ranks = [rank for rank, _, _ in discovered] + if len(set(ranks)) != len(ranks): + duplicates = sorted({rank for rank in ranks if ranks.count(rank) > 1}) + raise ValueError( + f"parent dispatch {expected_key[0]}:{expected_key[1]} has several captures on the same " + f"Rank {duplicates}; one Rank contributes at most one capture to a group" + ) + group_indexes = [identity["group_index"] for identity in identities] + if len(discovered) != group_size or sorted(group_indexes) != list(range(group_size)): + raise ValueError( + f"incomplete parent dispatch {expected_key[0]}:{expected_key[1]}: " + f"expected group indexes 0..{group_size - 1}, found {sorted(group_indexes)}" + ) + return { + "dispatch_pairing": "parent_dispatch_identity", + "dispatch_identity": { + "run_id": expected_key[0], + "task_slot": expected_key[1], + "group_size": group_size, + "callable_digest": next(iter(callable_digests)), + }, + } + + +def _discover_l3_parent_dispatch_inputs(rank_dirs, dispatch_identity): + match = _DISPATCH_ID_PATTERN.fullmatch(dispatch_identity) + if match is None: + raise ValueError("--dispatch-id must use RUN_ID:TASK_SLOT (for example, 17:5)") + expected_key = (int(match.group(1)), int(match.group(2))) + discovered = [] + for rank, rank_dir in rank_dirs: + for capture_dir in sorted( + (path for path in rank_dir.glob("d*") if path.is_dir()), + key=lambda path: int(path.name.removeprefix("d")) + if _DISPATCH_DIR_PATTERN.fullmatch(path.name) + else sys.maxsize, + ): + if not (capture_dir / "chip_swimlane_records.json").is_file(): + continue + identity = _load_dispatch_identity(capture_dir) + if identity is None: + continue + if (identity["run_id"], identity["task_slot"]) == expected_key: + discovered.append((rank, capture_dir / "chip_swimlane_records.json", identity)) + pairing = _validate_parent_dispatch_group(discovered, expected_key) + return [(rank, records_path) for rank, records_path, _ in sorted(discovered)], pairing + + +def _discover_l3_local_capture_inputs(rank_dirs, dispatch): + if not _DISPATCH_DIR_PATTERN.fullmatch(dispatch or ""): + raise ValueError("--dispatch must name dN (for example, --dispatch d0)") + + # One list of (rank, records_path, identity) triples rather than two parallel + # lists: the identity belongs to the capture it was read from, and a pairing + # that depends on two lists staying index-aligned is the failure this + # function exists to prevent. + captures = [] + for rank, rank_dir in rank_dirs: + records_path = rank_dir / dispatch / "chip_swimlane_records.json" + if not records_path.is_file(): + raise ValueError(f"rank{rank} is missing {dispatch}/chip_swimlane_records.json") + captures.append((rank, records_path, _load_dispatch_identity(records_path.parent))) + + present_count = sum(identity is not None for _, _, identity in captures) + if present_count not in (0, len(captures)): + raise ValueError(f"{dispatch} has dispatch identity metadata for only {present_count}/{len(captures)} Ranks") + pairing = {"dispatch_pairing": "local_capture_index"} + if present_count: + group_keys = {(identity["run_id"], identity["task_slot"]) for _, _, identity in captures} + has_group = any(identity["group_size"] > 1 for _, _, identity in captures) + if has_group: + if len(group_keys) != 1: + raise ValueError( + f"{dispatch} refers to different parent dispatches across Ranks; use --dispatch-id RUN_ID:TASK_SLOT" + ) + pairing = _validate_parent_dispatch_group(captures, next(iter(group_keys))) + else: + pairing["dispatch_identity_status"] = "individual_submissions" + return sorted((rank, records_path) for rank, records_path, _ in captures), pairing + + +def _discover_l3_rank_inputs(root, dispatch, dispatch_identity=None): + if bool(dispatch) == bool(dispatch_identity): + raise ValueError("directory input requires exactly one of --dispatch dN or --dispatch-id RUN_ID:TASK_SLOT") + rank_dirs = _l3_rank_dirs(root) + if dispatch_identity: + return _discover_l3_parent_dispatch_inputs(rank_dirs, dispatch_identity) + return _discover_l3_local_capture_inputs(rank_dirs, dispatch) + + +def discover_l3_conversion_targets(root): + """Discover safe automatic conversion units below one L3 output root. + + Public because the SceneTest postprocessor needs the same units this + module's directory mode would pick, one CLI invocation per unit. + + Raises ValueError only when nothing below ``root`` can be paired safely. A + remainder that cannot be paired by dN downgrades to a stderr warning while + the parent-identity targets are still returned. + """ + captures_by_rank = {} + parent_groups = defaultdict(list) + for rank, rank_dir in _l3_rank_dirs(root): + captures = [] + for capture_dir in sorted( + (path for path in rank_dir.glob("d*") if path.is_dir()), + key=lambda path: int(path.name.removeprefix("d")) + if _DISPATCH_DIR_PATTERN.fullmatch(path.name) + else sys.maxsize, + ): + if not _DISPATCH_DIR_PATTERN.fullmatch(capture_dir.name): + raise ValueError(f"invalid capture directory name: {capture_dir} (expected dN)") + records_path = capture_dir / "chip_swimlane_records.json" + if not records_path.is_file(): + continue + identity = _load_dispatch_identity(capture_dir) + capture = (rank, records_path, identity) + captures.append(capture) + if identity is not None and identity["group_size"] > 1: + parent_groups[(identity["run_id"], identity["task_slot"])].append(capture) + captures_by_rank[rank] = captures + + targets = [] + semantically_paired_paths = set() + for parent_key, captures in sorted(parent_groups.items()): + pairing = _validate_parent_dispatch_group(captures, parent_key) + paths = [records_path for _, records_path, _ in sorted(captures)] + semantically_paired_paths.update(paths) + targets.append( + { + "dispatch": None, + "dispatch_id": f"{parent_key[0]}:{parent_key[1]}", + "output_stem": f"l3_swimlane_run{parent_key[0]}_task{parent_key[1]}", + "capture_dirs": [path.parent for path in paths], + "pairing": pairing, + } + ) + + fallback_sets = [] + for rank, captures in sorted(captures_by_rank.items()): + fallback_sets.append( + ( + rank, + { + records_path.parent.name: records_path.parent + for _, records_path, _ in captures + if records_path not in semantically_paired_paths + }, + ) + ) + if fallback_sets: + expected = set(fallback_sets[0][1]) + if any(set(captures) != expected for _, captures in fallback_sets[1:]): + # Only the dN-paired remainder is unsafe here. A parent-identity + # target is paired by (run_id, task_slot) and is unaffected by what + # the leftover dN sets look like, so dropping those too would + # discard exactly the pairings this identity exists to make. + detail = ", ".join(f"rank{rank}={sorted(captures)}" for rank, captures in fallback_sets) + message = f"refusing to pair asymmetric local capture indexes under {root}: {detail}" + if not targets: + raise ValueError(message) + print(f"Warning: {message}", file=sys.stderr) + return targets + for dispatch in sorted(expected, key=lambda name: int(name.removeprefix("d"))): + targets.append( + { + "dispatch": dispatch, + "dispatch_id": None, + "output_stem": f"l3_swimlane_{dispatch}", + "capture_dirs": [captures[dispatch] for _, captures in fallback_sets], + "pairing": {"dispatch_pairing": "local_capture_index"}, + } + ) + return targets + + +def _validate_l3_rank_data(rank, records_path, data): + if data.get("chip_swimlane_level") != 4: + raise ValueError(f"rank{rank} must use chip_swimlane_level 4: {records_path}") + timeline = data.get("timeline_metadata") or {} + alignment = timeline.get("clock_alignment") or {} + if alignment.get("status") != "calibrated": + reason = alignment.get("reason", "unknown") + raise ValueError(f"rank{rank} clock calibration failed ({reason}): {records_path}") + clock_domain = timeline.get("host_clock_domain_id") + if not clock_domain: + raise ValueError( + f"rank{rank} is missing metadata.host_clock_domain_id; old captures remain usable only in single-file mode" + ) + origin_ns = int(timeline.get("source_timeline_origin_ns") or 0) + if origin_ns <= 0: + raise ValueError(f"rank{rank} has no valid Host timeline origin: {records_path}") + return str(clock_domain), origin_ns + + +def _load_rank_local_artifacts(records_path): + name_map_path = _find_sibling_name_map(records_path) + if name_map_path is None: + func_names, orchestrator_name = {}, None + else: + func_names, orchestrator_name = load_func_names_json(name_map_path) + + deps_path = records_path.parent / "deps.json" + return { + "dispatch_identity": _load_dispatch_identity(records_path.parent), + "func_names": func_names, + "orchestrator_name": orchestrator_name, + "deps_path": deps_path, + "deps_edges": load_deps_json(deps_path), + "deps_kernel_map": load_deps_kernel_map(deps_path), + "deps_block_map": load_deps_block_map(deps_path), + } + + +def _namespace_rank_trace(trace, rank): + pid_base = rank * _RANK_PID_STRIDE + for event in trace.get("traceEvents", []): + if "pid" in event: + # Every single-Rank view pid must fit inside one stride, or two Ranks + # land on the same namespaced pid and their lanes silently merge. + base_pid = int(event["pid"]) + if not 0 <= base_pid < _RANK_PID_STRIDE: + raise ValueError(f"single-Rank view pid {base_pid} does not fit the per-Rank stride {_RANK_PID_STRIDE}") + event["pid"] = pid_base + base_pid + if event.get("ph") == "M" and event.get("name") == "process_name": + name = event.get("args", {}).get("name") + if name: + event["args"]["name"] = f"rank{rank} / {name}" + elif event.get("ph") == "M" and event.get("name") == "process_sort_index": + sort_index = int(event.get("args", {}).get("sort_index", 0)) + event["args"]["sort_index"] = pid_base + sort_index + for id_field in ("id", "bind_id"): + if id_field in event: + event[id_field] = f"r{rank}:{event[id_field]}" + # Perfetto treats every counter arg as a separate numeric series. Rank + # identity is already encoded in the PID, so adding it to ``ph: C`` + # would create a bogus constant counter alongside the real values. + if event.get("ph") not in ("M", "C"): + event.setdefault("args", {})["rank"] = rank + return trace + + +def _generate_l3_trace(args, root): # noqa: PLR0912 + if args.func_names or args.kernel_config or args.deps_json: + raise ValueError("directory input auto-loads per-Rank name/dependency files; global overrides are not allowed") + + rank_inputs, pairing_metadata = _discover_l3_rank_inputs(root, args.dispatch, args.dispatch_id) + raw_inputs = {} + clock_domains = set() + origins = [] + for rank, records_path in rank_inputs: + with records_path.open() as f: + raw_inputs[rank] = json.load(f) + data = _decode_perf_data(raw_inputs[rank]) + clock_domain, origin_ns = _validate_l3_rank_data(rank, records_path, data) + clock_domains.add(clock_domain) + origins.append(origin_ns) + if len(clock_domains) != 1: + raise ValueError(f"Rank inputs use different Host clock domains: {sorted(clock_domains)}") + + global_origin_ns = min(origins) + all_events = [] + rank_metadata = [] + pre_group_durations = [] + for rank, records_path in rank_inputs: + data = _decode_perf_data(raw_inputs[rank], timeline_origin_ns=global_origin_ns) + artifacts = _load_rank_local_artifacts(records_path) + dispatch_identity = artifacts["dispatch_identity"] + trace = generate_chrome_trace_json( + data["tasks"], + None, + artifacts["func_names"], + args.verbose, + orchestrator_name=artifacts["orchestrator_name"], + scheduler_phases=data.get("aicpu_scheduler_phases"), + orchestrator_phases=data.get("aicpu_orchestrator_phases"), + orchestrator_source=data.get("orchestrator_source"), + timeline_metadata=data.get("timeline_metadata"), + core_to_thread=data.get("core_to_thread"), + host_device_uploads=data.get("host_device_uploads"), + deps_edges=artifacts["deps_edges"], + deps_kernel_map=artifacts["deps_kernel_map"], + deps_block_map=artifacts["deps_block_map"], + emit_overhead=args.overhead, + ) + _namespace_rank_trace(trace, rank) + all_events.extend(trace["traceEvents"]) + + timeline = data["timeline_metadata"] + alignment = timeline["clock_alignment"] + durations = alignment.get("anchor_group_duration_ns") or {} + pre_duration = durations.get("pre_host_orchestration") + if pre_duration is not None: + pre_group_durations.append(int(pre_duration)) + rank_metadata.append( + { + "rank": rank, + "input": str(records_path), + "trace_status": timeline["trace_status"], + "source_timeline_origin_ns": timeline["source_timeline_origin_ns"], + "clock_alignment": alignment, + "host_capture": timeline.get("host_capture"), + "dispatch_identity": dispatch_identity, + } + ) + + # Worst case for an interval read between two Ranks: each end carries its + # own Rank's alignment error, so the two largest bound any pair. null means + # the bound is unknown — fewer than two Ranks reported one — never that the + # comparison is exact. + uncertainties = sorted( + int(rank["clock_alignment"]["max_uncertainty_ns"]) + for rank in rank_metadata + if rank["clock_alignment"].get("max_uncertainty_ns") is not None + ) + metadata = { + "layout": "same_host_multi_rank", + "dispatch": args.dispatch, + "dispatch_id": args.dispatch_id, + "host_clock_domain_id": next(iter(clock_domains)), + "global_origin_ns": global_origin_ns, + "rank_count": len(rank_metadata), + **pairing_metadata, + "trace_status": "partial" if any(rank["trace_status"] != "complete" for rank in rank_metadata) else "complete", + "ranks": rank_metadata, + "cross_rank_uncertainty_ns": sum(uncertainties[-2:]) if len(uncertainties) >= 2 else None, + "pre_anchor_group_duration_spread_ns": ( + max(pre_group_durations) - min(pre_group_durations) if len(pre_group_durations) >= 2 else 0 + ), + "pre_anchor_group_duration_max_ns": max(pre_group_durations) if pre_group_durations else None, + } + output_path = _resolve_output_path(args, Path(root)) + output_path.parent.mkdir(parents=True, exist_ok=True) + with output_path.open("w") as f: + json.dump({"traceEvents": all_events, "metadata": metadata}, f, indent=2) + return output_path, rank_metadata + + def main(): args = _build_parser().parse_args() @@ -2988,6 +3476,16 @@ def main(): return 1 try: + if input_path.is_dir(): + output_path, rank_metadata = _generate_l3_trace(args, input_path) + print("\n✓ Multi-Rank conversion complete") + print(f" Input: {input_path}") + print(f" Ranks: {', '.join('rank' + str(item['rank']) for item in rank_metadata)}") + print(f" Output: {output_path}") + print(f"\nTo visualize: Open https://ui.perfetto.dev/ and drag in {output_path}") + return 0 + if args.dispatch or args.dispatch_id: + raise ValueError("--dispatch and --dispatch-id are only valid when input is a dfx_outputs directory") if args.verbose: print(f"Reading performance data from: {input_path}") data = read_perf_data(input_path) diff --git a/src/a2a3/platform/sim/host/device_runner.cpp b/src/a2a3/platform/sim/host/device_runner.cpp index d676835883..aa18e54bef 100644 --- a/src/a2a3/platform/sim/host/device_runner.cpp +++ b/src/a2a3/platform/sim/host/device_runner.cpp @@ -554,7 +554,10 @@ DeviceRunner::launch_execution(std::unique_ptr prepared, Laun auto thread_factory = [this](std::function fn) { return create_thread(std::move(fn)); }; - if (enable_chip_swimlane_) chip_swimlane_collector_.start(thread_factory); + if (enable_chip_swimlane_) { + if (capture_clock_anchors_) begin_clock_correlation_session_if_needed(); + chip_swimlane_collector_.start(thread_factory); + } if (enable_dump_args_) dump_collector_.start(thread_factory); if (enable_pmu_) pmu_collector_.start(thread_factory); if (enable_dep_gen_ && !dep_gen_host_graph_active()) dep_gen_collector_.start(thread_factory); diff --git a/src/a5/platform/sim/host/device_runner.cpp b/src/a5/platform/sim/host/device_runner.cpp index 85e10bf18e..b56f9c5960 100644 --- a/src/a5/platform/sim/host/device_runner.cpp +++ b/src/a5/platform/sim/host/device_runner.cpp @@ -526,7 +526,10 @@ DeviceRunner::launch_execution(std::unique_ptr prepared, Laun auto thread_factory = [this](std::function fn) { return create_thread(std::move(fn)); }; - if (enable_chip_swimlane_) chip_swimlane_collector_.start(thread_factory); + if (enable_chip_swimlane_) { + if (capture_clock_anchors_) begin_clock_correlation_session_if_needed(); + chip_swimlane_collector_.start(thread_factory); + } if (enable_dump_args_) dump_collector_.start(thread_factory); if (enable_pmu_) pmu_collector_.start(thread_factory); if (enable_dep_gen_ && !dep_gen_host_graph_active()) dep_gen_collector_.start(thread_factory); diff --git a/src/common/hierarchical/remote_wire.cpp b/src/common/hierarchical/remote_wire.cpp index c63bb1424b..eaf136a9bc 100644 --- a/src/common/hierarchical/remote_wire.cpp +++ b/src/common/hierarchical/remote_wire.cpp @@ -315,6 +315,9 @@ std::vector encode_call_config(const CallConfig &config) { put_i32(out, config.enable_pmu); put_i32(out, config.enable_dep_gen); put_i32(out, config.enable_scope_stats); + // CallConfig::capture_clock_anchors is absent on purpose: every ChipWorker + // child decides it locally when it reads the config out of its mailbox, so a + // transported value would be overwritten before any runtime reads it. put_string(out, call_config_prefix(config), MAX_STRING_BYTES, "CallConfig.output_prefix"); return out; } diff --git a/src/common/hierarchical/worker_manager.cpp b/src/common/hierarchical/worker_manager.cpp index ca72ec1b04..8af1a9151a 100644 --- a/src/common/hierarchical/worker_manager.cpp +++ b/src/common/hierarchical/worker_manager.cpp @@ -799,17 +799,24 @@ void LocalMailboxEndpoint::submit_progress(Ring *ring, const WorkerDispatch &dis const uint64_t protocol = MAILBOX_TASK_PROTOCOL_VERSION; const uint64_t slot_id = state.pipeline_lease.slot_id; + const uint64_t task_slot = static_cast(dispatch.task_slot); + const uint64_t group_index = static_cast(dispatch.group_index); + const uint64_t group_size = static_cast(state.group_size()); std::memcpy(frame + MAILBOX_OFF_FRAME_PROTOCOL, &protocol, sizeof(protocol)); std::memcpy(frame + MAILBOX_OFF_FRAME_RUN_ID, &state.run_id, sizeof(state.run_id)); std::memcpy(frame + MAILBOX_OFF_FRAME_SLOT_ID, &slot_id, sizeof(slot_id)); std::memcpy(frame + MAILBOX_OFF_FRAME_GENERATION, &state.pipeline_lease.generation, sizeof(uint64_t)); std::memcpy(frame + MAILBOX_OFF_FRAME_DISPATCH_ID, &dispatch.dispatch_id, sizeof(dispatch.dispatch_id)); + std::memcpy(frame + MAILBOX_OFF_FRAME_TASK_SLOT, &task_slot, sizeof(task_slot)); + std::memcpy(frame + MAILBOX_OFF_FRAME_GROUP_INDEX, &group_index, sizeof(group_index)); + std::memcpy(frame + MAILBOX_OFF_FRAME_GROUP_SIZE, &group_size, sizeof(group_size)); record.occupied = true; record.dispatch = dispatch; record.run_id = state.run_id; record.slot_id = slot_id; record.generation = state.pipeline_lease.generation; + record.group_size = group_size; write_mailbox_state(dispatch.prepare_only ? MailboxState::PREPARE_READY : MailboxState::TASK_READY, frame); } @@ -819,15 +826,23 @@ bool LocalMailboxEndpoint::frame_identity_matches(const FrameRecord &record, con uint64_t slot_id = 0; uint64_t generation = 0; uint64_t dispatch_id = 0; + uint64_t task_slot = 0; + uint64_t group_index = 0; + uint64_t group_size = 0; PipelineSlotLease lease{}; std::memcpy(&protocol, frame + MAILBOX_OFF_FRAME_PROTOCOL, sizeof(protocol)); std::memcpy(&run_id, frame + MAILBOX_OFF_FRAME_RUN_ID, sizeof(run_id)); std::memcpy(&slot_id, frame + MAILBOX_OFF_FRAME_SLOT_ID, sizeof(slot_id)); std::memcpy(&generation, frame + MAILBOX_OFF_FRAME_GENERATION, sizeof(generation)); std::memcpy(&dispatch_id, frame + MAILBOX_OFF_FRAME_DISPATCH_ID, sizeof(dispatch_id)); + std::memcpy(&task_slot, frame + MAILBOX_OFF_FRAME_TASK_SLOT, sizeof(task_slot)); + std::memcpy(&group_index, frame + MAILBOX_OFF_FRAME_GROUP_INDEX, sizeof(group_index)); + std::memcpy(&group_size, frame + MAILBOX_OFF_FRAME_GROUP_SIZE, sizeof(group_size)); std::memcpy(&lease, frame + MAILBOX_OFF_PIPELINE_LEASE, sizeof(lease)); return protocol == MAILBOX_TASK_PROTOCOL_VERSION && run_id == record.run_id && slot_id == record.slot_id && generation == record.generation && dispatch_id == record.dispatch.dispatch_id && + task_slot == static_cast(record.dispatch.task_slot) && + group_index == static_cast(record.dispatch.group_index) && group_size == record.group_size && lease.slot_id == record.slot_id && lease.reserved == 0 && lease.generation == record.generation; } diff --git a/src/common/hierarchical/worker_manager.h b/src/common/hierarchical/worker_manager.h index 2257e1c7c2..9206b13b66 100644 --- a/src/common/hierarchical/worker_manager.h +++ b/src/common/hierarchical/worker_manager.h @@ -99,7 +99,7 @@ static constexpr size_t MAILBOX_TASK_FRAME_COUNT = 2; static constexpr size_t MAILBOX_CONTROL_FRAME = 0; static constexpr size_t MAILBOX_FIRST_TASK_FRAME = 1; static constexpr size_t MAILBOX_SIZE = MAILBOX_FRAME_SIZE * (1 + MAILBOX_TASK_FRAME_COUNT); -static constexpr uint32_t MAILBOX_TASK_PROTOCOL_VERSION = 3; +static constexpr uint32_t MAILBOX_TASK_PROTOCOL_VERSION = 4; // Error message region lives at the mailbox tail. 256 B of headroom is // enough for `: ` produced by the child-side @@ -149,13 +149,19 @@ static constexpr ptrdiff_t MAILBOX_OFF_FRAME_RUN_ID = MAILBOX_OFF_ACCEPTED - 32; static constexpr ptrdiff_t MAILBOX_OFF_FRAME_SLOT_ID = MAILBOX_OFF_ACCEPTED - 24; static constexpr ptrdiff_t MAILBOX_OFF_FRAME_GENERATION = MAILBOX_OFF_ACCEPTED - 16; static constexpr ptrdiff_t MAILBOX_OFF_FRAME_DISPATCH_ID = MAILBOX_OFF_ACCEPTED - 8; +// Parent-DAG identity shared by every member of one NEXT_LEVEL group. Unlike +// dispatch_id (which is local to one WorkerThread), task_slot is comparable +// across the group's Rank endpoints within run_id. +static constexpr ptrdiff_t MAILBOX_OFF_FRAME_TASK_SLOT = MAILBOX_OFF_ACCEPTED - 48; +static constexpr ptrdiff_t MAILBOX_OFF_FRAME_GROUP_INDEX = MAILBOX_OFF_ACCEPTED - 56; +static constexpr ptrdiff_t MAILBOX_OFF_FRAME_GROUP_SIZE = MAILBOX_OFF_ACCEPTED - 64; // Termination is a sticky one-way word on the control frame, not a MailboxState: // MAILBOX_OFF_STATE has three writers (the parent's CONTROL_REQUEST, the child's // CONTROL_DONE, and this endpoint's return-to-IDLE), any of which overwrites a // SHUTDOWN store. Only a terminating parent writes this word, 0 -> 1, and // nothing ever clears it, so a child that observes it exits its serve loop no // matter what state word a concurrent control command leaves behind. -static constexpr ptrdiff_t MAILBOX_OFF_SHUTDOWN = MAILBOX_OFF_FRAME_PROTOCOL - 8; +static constexpr ptrdiff_t MAILBOX_OFF_SHUTDOWN = MAILBOX_OFF_ACCEPTED - 72; static constexpr int32_t MAILBOX_SHUTDOWN_REQUESTED = 1; static constexpr ptrdiff_t MAILBOX_OFF_TASK_CALLABLE_HASH = MAILBOX_OFF_ARGS; static constexpr ptrdiff_t MAILBOX_OFF_TASK_ARGS_BLOB = @@ -522,6 +528,7 @@ class LocalMailboxEndpoint : public WorkerEndpoint { RunId run_id{INVALID_RUN_ID}; uint64_t slot_id{0}; uint64_t generation{0}; + uint64_t group_size{0}; }; std::array frames_{}; diff --git a/src/common/platform/include/host/clock_correlation.h b/src/common/platform/include/host/clock_correlation.h index 38c3e56ab3..b367fb428b 100644 --- a/src/common/platform/include/host/clock_correlation.h +++ b/src/common/platform/include/host/clock_correlation.h @@ -21,6 +21,19 @@ namespace simpler::dfx { constexpr std::size_t kClockAnchorSamplesPerPosition = 3; +// The two ends of the calibrated interval. Device timestamps outside +// [HostOrchestrationBegin, DeviceExecutionComplete] cannot be mapped to the +// Host clock, so the opening anchor is taken at whatever point in a runtime's +// sequence precedes every device timestamp it will record: +// +// host_build_graph before Host orchestration (host_phase_pool_arm), +// so the interval also spans bind and H2D +// tensormap_and_ringbuffer before kernel launch (start_shared_collectors_ +// for_run), the earliest point it has +// +// The serialized name "pre_host_orchestration" predates the second case and is +// kept because it is an on-wire value in every existing capture; read it as +// "start of the calibrated interval", not as a claim about Host orchestration. enum class ClockAnchorPosition : uint32_t { HostOrchestrationBegin = 0, DeviceExecutionComplete = 1, diff --git a/src/common/platform/onboard/host/device_runner_base.cpp b/src/common/platform/onboard/host/device_runner_base.cpp index 016ec60a9b..50ffb816ca 100644 --- a/src/common/platform/onboard/host/device_runner_base.cpp +++ b/src/common/platform/onboard/host/device_runner_base.cpp @@ -1277,6 +1277,7 @@ void DeviceRunnerBase::apply_call_config(const CallConfig &config) { // without dep_gen falls through to the base no-op. set_dep_gen_enabled(config.enable_dep_gen != 0); set_scope_stats_enabled(config.enable_scope_stats != 0); + capture_clock_anchors_ = config.capture_clock_anchors != 0; set_output_prefix(config.output_prefix); } @@ -1299,8 +1300,14 @@ HostPhaseRecordPool *DeviceRunnerBase::host_phase_pool_arm(bool producer_wants_r } if (!swimlane_wants_records) return pool; - // Only the chip-swimlane reader places these records against device - // timestamps, so only it needs the two clocks anchored. + begin_clock_correlation_session_if_needed(); + return pool; +} + +void DeviceRunnerBase::begin_clock_correlation_session_if_needed() noexcept { + if (chip_swimlane_level_ != ChipSwimlaneLevel::ORCH_PHASES || chip_swimlane_collector_.clock_correlation_active()) { + return; + } try { clock_correlation_provider_ = simpler::dfx::make_clock_correlation_provider(); chip_swimlane_collector_.begin_clock_correlation_session( @@ -1321,7 +1328,6 @@ HostPhaseRecordPool *DeviceRunnerBase::host_phase_pool_arm(bool producer_wants_r chip_swimlane_collector_.finish_clock_correlation_session(); } } - return pool; } void DeviceRunnerBase::publish_host_phase_records_to_swimlane() { @@ -1883,6 +1889,7 @@ void DeviceRunnerBase::start_shared_collectors_for_run() { return create_thread(std::move(fn)); }; if (enable_chip_swimlane_) { + if (capture_clock_anchors_) begin_clock_correlation_session_if_needed(); chip_swimlane_collector_.start(thread_factory); } if (enable_dump_args_) { diff --git a/src/common/platform/onboard/host/device_runner_base.h b/src/common/platform/onboard/host/device_runner_base.h index 169e00a048..71ebf5d6d0 100644 --- a/src/common/platform/onboard/host/device_runner_base.h +++ b/src/common/platform/onboard/host/device_runner_base.h @@ -740,6 +740,8 @@ class DeviceRunnerBase { const simpler::dfx::HostPhaseRecordStore &host_phase_records() const { return host_phase_records_; } /** Hand this pass's records to the swimlane reader, just before its export. */ void publish_host_phase_records_to_swimlane(); + /** Start the level-4 Host/Device clock correlation once per run. */ + void begin_clock_correlation_session_if_needed() noexcept; /** * Write this pass's per-event host phase records to `output_prefix_`. * @@ -1299,5 +1301,6 @@ class DeviceRunnerBase { bool enable_scope_stats_{false}; ChipSwimlaneLevel chip_swimlane_level_{ChipSwimlaneLevel::DISABLED}; // resolved from set_chip_swimlane_enabled() PmuEventType pmu_event_type_{PmuEventType::PIPE_UTILIZATION}; // resolved from set_pmu_enabled() + bool capture_clock_anchors_{false}; // from CallConfig::capture_clock_anchors std::string output_prefix_{}; // diagnostic artifact root directory }; diff --git a/src/common/platform/shared/host/chip_swimlane_collector.cpp b/src/common/platform/shared/host/chip_swimlane_collector.cpp index 8c35783408..3d3aea80bf 100644 --- a/src/common/platform/shared/host/chip_swimlane_collector.cpp +++ b/src/common/platform/shared/host/chip_swimlane_collector.cpp @@ -22,6 +22,7 @@ #include #include +#include #include #include #include @@ -47,6 +48,16 @@ namespace { +std::string linux_boot_clock_domain_id() { + std::ifstream boot_id_file("/proc/sys/kernel/random/boot_id"); + std::string boot_id; + if (!(boot_id_file >> boot_id) || boot_id.empty()) return {}; + for (unsigned char ch : boot_id) { + if (!std::isalnum(ch) && ch != '-') return {}; + } + return "linux-boot-id:" + boot_id; +} + int owner_recycled_shard_for_core(int core_index, int thread_count) { int cluster_index = core_index / PLATFORM_CORES_PER_BLOCKDIM; return cluster_index % thread_count; @@ -1044,7 +1055,26 @@ int ChipSwimlaneCollector::export_swimlane_json() { outfile << "\"record_count_mismatch\"}"; } } + if (host_phase_records_present_ || clock_correlation_session_.started()) { + const std::string host_clock_domain_id = linux_boot_clock_domain_id(); + if (!host_clock_domain_id.empty()) { + outfile << ",\n \"host_clock_domain_id\": \"" << host_clock_domain_id << "\""; + } + } if (clock_correlation_session_.started()) { + uint64_t host_timeline_origin_ns = 0; + for (const auto &sample : clock_correlation_session_.samples()) { + if (sample.position != simpler::dfx::ClockAnchorPosition::HostOrchestrationBegin || !sample.valid()) { + continue; + } + const uint64_t midpoint = sample.host_before_ns + (sample.host_after_ns - sample.host_before_ns) / 2; + if (host_timeline_origin_ns == 0 || midpoint < host_timeline_origin_ns) { + host_timeline_origin_ns = midpoint; + } + } + if (host_timeline_origin_ns != 0) { + outfile << ",\n \"host_timeline_origin_ns\": " << host_timeline_origin_ns; + } outfile << ",\n \"clock_anchors\": {"; outfile << "\n \"provider\": \"" << clock_correlation_session_.provider_name() << "\","; outfile << "\n \"device_timestamp_unit\": \"syscnt_cycles\","; diff --git a/src/common/platform/sim/host/device_runner_base.cpp b/src/common/platform/sim/host/device_runner_base.cpp index f8f48df4b8..669de166d6 100644 --- a/src/common/platform/sim/host/device_runner_base.cpp +++ b/src/common/platform/sim/host/device_runner_base.cpp @@ -721,6 +721,7 @@ void SimDeviceRunnerBase::apply_call_config(const CallConfig &config) { // a2a3 and a5 override set_dep_gen_enabled; an arch without dep_gen no-ops. set_dep_gen_enabled(config.enable_dep_gen != 0); set_scope_stats_enabled(config.enable_scope_stats != 0); + capture_clock_anchors_ = config.capture_clock_anchors != 0; set_output_prefix(config.output_prefix); } @@ -743,8 +744,14 @@ HostPhaseRecordPool *SimDeviceRunnerBase::host_phase_pool_arm(bool producer_want } if (!swimlane_wants_records) return pool; - // Only the chip-swimlane reader places these records against device - // timestamps, so only it needs the two clocks anchored. + begin_clock_correlation_session_if_needed(); + return pool; +} + +void SimDeviceRunnerBase::begin_clock_correlation_session_if_needed() noexcept { + if (chip_swimlane_level_ != ChipSwimlaneLevel::ORCH_PHASES || chip_swimlane_collector_.clock_correlation_active()) { + return; + } try { clock_correlation_provider_ = simpler::dfx::make_clock_correlation_provider(); chip_swimlane_collector_.begin_clock_correlation_session( @@ -765,7 +772,6 @@ HostPhaseRecordPool *SimDeviceRunnerBase::host_phase_pool_arm(bool producer_want chip_swimlane_collector_.finish_clock_correlation_session(); } } - return pool; } void SimDeviceRunnerBase::publish_host_phase_records_to_swimlane() { diff --git a/src/common/platform/sim/host/device_runner_base.h b/src/common/platform/sim/host/device_runner_base.h index 8c17ca89f7..7d677f719b 100644 --- a/src/common/platform/sim/host/device_runner_base.h +++ b/src/common/platform/sim/host/device_runner_base.h @@ -280,6 +280,8 @@ class SimDeviceRunnerBase { const simpler::dfx::HostPhaseRecordStore &host_phase_records() const { return host_phase_records_; } /** Hand this pass's records to the swimlane reader, just before its export. */ void publish_host_phase_records_to_swimlane(); + /** Start the level-4 Host/Device clock correlation once per run. */ + void begin_clock_correlation_session_if_needed() noexcept; void finish_clock_correlation_session(bool capture_device_complete) noexcept; void set_dump_args_enabled(int level) { dump_args_level_ = static_cast(level); @@ -512,6 +514,7 @@ class SimDeviceRunnerBase { bool enable_scope_stats_{false}; ChipSwimlaneLevel chip_swimlane_level_{ChipSwimlaneLevel::DISABLED}; // resolved from set_chip_swimlane_enabled() PmuEventType pmu_event_type_{PmuEventType::PIPE_UTILIZATION}; // resolved from set_pmu_enabled() + bool capture_clock_anchors_{false}; // from CallConfig::capture_clock_anchors std::string output_prefix_{}; // diagnostic artifact root directory }; diff --git a/src/common/task_interface/call_config.h b/src/common/task_interface/call_config.h index 7905addd5f..df185ac5b2 100644 --- a/src/common/task_interface/call_config.h +++ b/src/common/task_interface/call_config.h @@ -117,7 +117,12 @@ struct CallConfig { int32_t enable_pmu = 0; // 0 = disabled; >0 = enabled, value selects event type int32_t enable_dep_gen = 0; int32_t enable_scope_stats = 0; // writes /scope_stats/scope_stats.jsonl - RuntimeEnv runtime_env; // per-task ring sizing + // Anchor the Host and Device clocks for this capture even when no Host + // orchestration records exist, which is what places its device timestamps on + // an absolute Host timeline. Independent of why a caller wants that timeline: + // the runtime samples the anchors and never learns who consumes them. + int32_t capture_clock_anchors = 0; + RuntimeEnv runtime_env; // per-task ring sizing char output_prefix[1024] = {}; bool diagnostics_any() const noexcept { @@ -145,6 +150,6 @@ struct CallConfig { #pragma pack(pop) static_assert(sizeof(RuntimeEnv) == RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t), "RuntimeEnv wire layout drift"); static_assert( - sizeof(CallConfig) == 6 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024, + sizeof(CallConfig) == 7 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024, "CallConfig wire layout drift" ); diff --git a/tests/ut/cpp/hierarchical/test_scheduler.cpp b/tests/ut/cpp/hierarchical/test_scheduler.cpp index d59d24e64f..d7e3889c7f 100644 --- a/tests/ut/cpp/hierarchical/test_scheduler.cpp +++ b/tests/ut/cpp/hierarchical/test_scheduler.cpp @@ -673,6 +673,24 @@ static uint64_t test_frame_dispatch_id(const char *frame) { return dispatch_id; } +static uint64_t test_frame_task_slot(const char *frame) { + uint64_t task_slot = 0; + std::memcpy(&task_slot, frame + MAILBOX_OFF_FRAME_TASK_SLOT, sizeof(task_slot)); + return task_slot; +} + +static uint64_t test_frame_group_index(const char *frame) { + uint64_t group_index = 0; + std::memcpy(&group_index, frame + MAILBOX_OFF_FRAME_GROUP_INDEX, sizeof(group_index)); + return group_index; +} + +static uint64_t test_frame_group_size(const char *frame) { + uint64_t group_size = 0; + std::memcpy(&group_size, frame + MAILBOX_OFF_FRAME_GROUP_SIZE, sizeof(group_size)); + return group_size; +} + class ScopedChildProcess { public: explicit ScopedChildProcess(pid_t pid) : @@ -1510,6 +1528,12 @@ TEST(WorkerManagerTest, TwoFrameLeaseSlotsDoNotDefineFifoOrAcceptance) { EXPECT_EQ(test_frame_state(upper_frame), MailboxState::TASK_READY); EXPECT_EQ(test_frame_dispatch_id(lower_frame), 42u); EXPECT_EQ(test_frame_dispatch_id(upper_frame), 41u); + EXPECT_EQ(test_frame_task_slot(lower_frame), static_cast(staged_slot)); + EXPECT_EQ(test_frame_task_slot(upper_frame), static_cast(active_slot)); + EXPECT_EQ(test_frame_group_index(lower_frame), 0u); + EXPECT_EQ(test_frame_group_index(upper_frame), 0u); + EXPECT_EQ(test_frame_group_size(lower_frame), 1u); + EXPECT_EQ(test_frame_group_size(upper_frame), 1u); EXPECT_TRUE(endpoint.activate_progress(/*run_id=*/22)); EXPECT_EQ(test_frame_state(lower_frame), MailboxState::PREPARE_READY); @@ -1542,16 +1566,23 @@ TEST(WorkerManagerTest, CapacityOneMailboxUsesTheProgressTaskFrame) { allocator.init(/*heap_bytes=*/0); TaskSlot task_slot = make_progress_slot(allocator, /*run_id=*/21, /*pipeline_slot=*/1, /*generation=*/7); ASSERT_NE(task_slot, INVALID_SLOT); + TaskSlotState *state = allocator.slot_state(task_slot); + ASSERT_NE(state, nullptr); + state->is_group_ = true; + state->task_args_list.resize(3); LocalMailboxEndpoint endpoint(/*worker_id=*/0, mailbox.data(), /*child_pid=*/-1, /*task_frame_count=*/1); EXPECT_FALSE(endpoint.caps().supports_frame_staging); - WorkerDispatch dispatch{task_slot, 0, /*dispatch_id=*/41, /*prepare_only=*/false}; + WorkerDispatch dispatch{task_slot, 1, /*dispatch_id=*/41, /*prepare_only=*/false}; endpoint.submit_progress(&allocator, dispatch); char *frame = test_task_frame(mailbox, 0); EXPECT_EQ(test_frame_state(frame), MailboxState::TASK_READY); EXPECT_EQ(test_frame_dispatch_id(frame), 41u); + EXPECT_EQ(test_frame_task_slot(frame), static_cast(task_slot)); + EXPECT_EQ(test_frame_group_index(frame), 1u); + EXPECT_EQ(test_frame_group_size(frame), 3u); set_test_frame_accepted(frame); WorkerEndpointProgress progress; diff --git a/tests/ut/cpp/types/test_call_config.cpp b/tests/ut/cpp/types/test_call_config.cpp index 6778d13621..4a9d5f63e0 100644 --- a/tests/ut/cpp/types/test_call_config.cpp +++ b/tests/ut/cpp/types/test_call_config.cpp @@ -19,7 +19,7 @@ // Wire contract: parent and forked child move CallConfig with one memcpy. TEST(CallConfig, WireLayoutMatchesConstant) { EXPECT_EQ(sizeof(RuntimeEnv), RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t)); - EXPECT_EQ(sizeof(CallConfig), 6 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024); + EXPECT_EQ(sizeof(CallConfig), 7 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024); } TEST(CallConfig, RuntimeEnvDefaultsAreUnset) { diff --git a/tests/ut/py/test_chip_worker.py b/tests/ut/py/test_chip_worker.py index 6cc70a498c..f77fedf7d0 100644 --- a/tests/ut/py/test_chip_worker.py +++ b/tests/ut/py/test_chip_worker.py @@ -8,6 +8,7 @@ # ----------------------------------------------------------------------------------------------------------- """Tests for CallConfig and ChipWorker state machine.""" +import json import threading import pytest @@ -383,6 +384,7 @@ def test_config_roundtrip(self): cfg.enable_pmu, int(cfg.enable_dep_gen), int(cfg.enable_scope_stats), + int(cfg.capture_clock_anchors), *cfg.runtime_env.ring_task_window, *cfg.runtime_env.ring_heap, *cfg.runtime_env.ring_dep_pool, @@ -400,3 +402,82 @@ def test_config_roundtrip(self): assert decoded.runtime_env.ring_heap == [1024, 2048, 4096, 8192] assert decoded.runtime_env.ring_dep_pool == [64, 128, 256, 512] assert decoded.output_prefix == "/tmp/out" + assert decoded.capture_clock_anchors is False + + ranked = _read_config_from_mailbox(memoryview(buf), chip_rank=2, capture_index=7) + assert ranked.output_prefix == "/tmp/out/rank2/d7" + assert ranked.capture_clock_anchors is True + + def test_rank_directory_covers_every_diagnostic_but_anchors_stay_swimlane_only(self): + # rankN/dN separates one ChipWorker child's artifacts from its siblings', + # which every diagnostic needs; capture_clock_anchors only turns on the + # Host/Device clock anchors, which only the swimlane reader consumes. + from simpler.worker import ( # noqa: PLC0415 # pyright: ignore[reportAttributeAccessIssue] + _CFG_FMT, + _OFF_CONFIG, + _read_config_from_mailbox, + ) + + def decode(**flags): + cfg = CallConfig() + cfg.output_prefix = "/tmp/out" + for name, value in flags.items(): + setattr(cfg, name, value) + buf = bytearray(_OFF_CONFIG + _CFG_FMT.size) + _CFG_FMT.pack_into( + buf, + _OFF_CONFIG, + cfg.aicpu_thread_num, + cfg.enable_chip_swimlane, + int(cfg.enable_dump_args), + cfg.enable_pmu, + int(cfg.enable_dep_gen), + int(cfg.enable_scope_stats), + int(cfg.capture_clock_anchors), + *cfg.runtime_env.ring_task_window, + *cfg.runtime_env.ring_heap, + *cfg.runtime_env.ring_dep_pool, + cfg.output_prefix.encode(), + ) + return _read_config_from_mailbox(memoryview(buf), chip_rank=1, capture_index=0) + + dep_gen_only = decode(enable_dep_gen=True) + assert dep_gen_only.output_prefix == "/tmp/out/rank1/d0" + assert dep_gen_only.capture_clock_anchors is False + + swimlane = decode(enable_chip_swimlane=4) + assert swimlane.output_prefix == "/tmp/out/rank1/d0" + assert swimlane.capture_clock_anchors is True + + # No diagnostic at all: nothing is written below output_prefix, so the + # child leaves the case root alone. + assert decode().output_prefix == "/tmp/out" + + def test_dispatch_identity_sidecar_uses_parent_dag_slot(self, tmp_path): + from simpler.worker import ( # noqa: PLC0415 # pyright: ignore[reportAttributeAccessIssue] + _TASK_PROTOCOL_VERSION, + _write_dispatch_identity_sidecar, + ) + + _write_dispatch_identity_sidecar( + str(tmp_path), + frame_identity=(_TASK_PROTOCOL_VERSION, 17, 1, 23, 41, 5, 2, 4), + chip_rank=2, + capture_index=7, + callable_digest=b"\xab" * 32, + ) + + identity = json.loads((tmp_path / "dispatch_identity.json").read_text()) + assert identity == { + "schema_version": 1, + "run_id": 17, + "task_slot": 5, + "group_index": 2, + "group_size": 4, + "chip_rank": 2, + "local_capture_index": 7, + "endpoint_dispatch_id": 41, + "pipeline_slot": 1, + "pipeline_generation": 23, + "callable_digest": "ab" * 32, + } diff --git a/tests/ut/py/test_clock_correlation.py b/tests/ut/py/test_clock_correlation.py index b2ef7d33ae..f5f94d1c11 100644 --- a/tests/ut/py/test_clock_correlation.py +++ b/tests/ut/py/test_clock_correlation.py @@ -50,6 +50,10 @@ def test_alignment_selects_minimum_rtt_and_interpolates_offset_with_integer_math "post_device_execution": 1, } assert alignment.max_uncertainty_ns == 20 + assert alignment.metadata()["anchor_group_duration_ns"] == { + "pre_host_orchestration": 200, + "post_device_execution": 400, + } assert alignment.map_cycles_to_host_ns(100) == 1_000 assert alignment.map_cycles_to_host_ns(2_100) == 3_050 assert alignment.map_cycles_to_host_ns(4_100) == 5_100 diff --git a/tests/ut/py/test_scene_level_selection.py b/tests/ut/py/test_scene_level_selection.py index 0adafc702c..896c8fffe5 100644 --- a/tests/ut/py/test_scene_level_selection.py +++ b/tests/ut/py/test_scene_level_selection.py @@ -204,7 +204,7 @@ def host_fn(): assert [item.nodeid for item in items] == ["tests::host"] -def test_single_round_chip_swimlane_rejects_l3_items(): +def test_single_round_chip_swimlane_allows_l3_items(): @scene_level(SceneTestLevel.NODE) def host_fn(): return None @@ -221,7 +221,29 @@ def host_fn(): }, ) - with pytest.raises(pytest.UsageError, match="not supported for L3 tests"): + root_conftest.pytest_collection_modifyitems(None, config, items) + + assert [item.nodeid for item in items] == ["tests::host"] + + +def test_single_round_chip_swimlane_rejects_network1_items(): + @scene_level(SceneTestLevel.NETWORK1) + def network1_fn(): + return None + + items = [_FakeItem("tests::network1", function=network1_fn)] + config = _FakeConfig( + platform="a2a3", + level=None, + **{ + "exclude-level": None, + "runtime": None, + "rounds": 1, + "enable-chip-swimlane": 4, + }, + ) + + with pytest.raises(pytest.UsageError, match="NETWORK1/L4 needs a node namespace"): root_conftest.pytest_collection_modifyitems(None, config, items) @@ -241,3 +263,53 @@ def chip_fn(): with pytest.raises(Failed, match="SceneTestLevel\\.NETWORK1"): root_conftest.st_network1_logs.__wrapped__(request, monkeypatch) + + +def test_resource_child_inherits_the_parents_diagnostic_selection(): + # The resource child's argv is built from scratch rather than inherited, so + # a diagnostic the parent asked for reaches it only if it is forwarded here. + # Without this, an L3 chip-swimlane run passes and writes nothing at all. + spec = SimpleNamespace(nodeid="tests/st/x.py::TestL3", runtime="tensormap_and_ringbuffer", kind="l3") + config = _FakeConfig( + **{ + "--enable-chip-swimlane": 4, + "--enable-dep-gen": True, + "--enable-scope-stats": True, + "--rounds": 3, + } + ) + + command = root_conftest._resource_child_command(spec, [0, 1], "a2a3", "exclude", config) + + assert command[command.index("--enable-chip-swimlane") + 1] == "4" + assert command[command.index("--rounds") + 1] == "3" + assert "--enable-dep-gen" in command + assert "--enable-scope-stats" in command + # Options the parent did not ask for stay off the child's command line. + assert "--enable-pmu" not in command + assert "--dump-args" not in command + assert "--skip-golden" not in command + assert "--enable-swimlane-overhead" not in command + + +def test_resource_child_stays_bare_when_no_diagnostic_is_requested(): + spec = SimpleNamespace(nodeid="tests/st/x.py::TestL3", runtime="tensormap_and_ringbuffer", kind="l3") + + command = root_conftest._resource_child_command(spec, [0], "a2a3", "exclude", _FakeConfig()) + + assert not any(arg.startswith("--enable-") or arg in ("--rounds", "--dump-args") for arg in command) + assert "--case" not in command + + +def test_resource_child_inherits_the_parents_case_selection(): + # A nodeid names a whole SceneTestCase class, and --case filters inside its + # single test_run item at run time. A child that does not receive the + # selector therefore runs every case of the class, not the one asked for. + spec = SimpleNamespace(nodeid="tests/st/x.py::TestL3", runtime="tensormap_and_ringbuffer", kind="l3") + config = _FakeConfig(**{"--case": ["TestL3::Case1", "Case2"]}) + + command = root_conftest._resource_child_command(spec, [0, 1], "a2a3", "only", config) + + assert command[command.index("--manual") + 1] == "only" + selectors = [command[index + 1] for index, arg in enumerate(command) if arg == "--case"] + assert selectors == ["TestL3::Case1", "Case2"] diff --git a/tests/ut/py/test_scene_test_cli_contract.py b/tests/ut/py/test_scene_test_cli_contract.py index c107fb035d..e2f412f245 100644 --- a/tests/ut/py/test_scene_test_cli_contract.py +++ b/tests/ut/py/test_scene_test_cli_contract.py @@ -12,7 +12,9 @@ from __future__ import annotations import importlib +import json import sys +from pathlib import Path from types import SimpleNamespace import pytest @@ -27,6 +29,127 @@ ) +def test_l3_swimlane_postprocess_merges_dispatches_present_on_every_rank(tmp_path, monkeypatch) -> None: + scene_test_module = importlib.import_module("simpler_setup.scene_test") + for rank in (0, 1): + for dispatch in ("d0", "d1"): + records = tmp_path / f"rank{rank}" / dispatch / "chip_swimlane_records.json" + records.parent.mkdir(parents=True) + records.write_text("{}") + + calls = [] + monkeypatch.setattr(scene_test_module, "_run_swimlane_converter", lambda **kwargs: calls.append(kwargs)) + + scene_test_module._convert_case_swimlane("case", tmp_path) + + assert [call["dispatch"] for call in calls] == ["d0", "d1"] + assert [call["output_path"] for call in calls] == [ + Path(tmp_path) / "l3_swimlane_d0.json", + Path(tmp_path) / "l3_swimlane_d1.json", + ] + + +def test_l3_swimlane_postprocess_falls_back_per_rank_below_level_four(tmp_path, monkeypatch, caplog) -> None: + scene_test_module = importlib.import_module("simpler_setup.scene_test") + for rank in (0, 1): + records = tmp_path / f"rank{rank}" / "d0" / "chip_swimlane_records.json" + records.parent.mkdir(parents=True) + records.write_text(json.dumps({"chip_swimlane_level": 3})) + + calls = [] + monkeypatch.setattr(scene_test_module, "_run_swimlane_converter", lambda **kwargs: calls.append(kwargs)) + + scene_test_module._convert_case_swimlane("case", tmp_path) + + # No cross-Rank merge without clock anchors — one single-file conversion per Rank. + assert [call["input_path"] for call in calls] == [ + tmp_path / "rank0" / "d0" / "chip_swimlane_records.json", + tmp_path / "rank1" / "d0" / "chip_swimlane_records.json", + ] + assert all("dispatch" not in call for call in calls) + assert "cross-Rank merging needs --enable-chip-swimlane 4" in caplog.text + + +def test_l3_swimlane_postprocess_refuses_asymmetric_local_capture_indexes(tmp_path, monkeypatch, caplog) -> None: + scene_test_module = importlib.import_module("simpler_setup.scene_test") + for rank, dispatches in ((0, ("d0", "d1")), (1, ("d0",))): + for dispatch in dispatches: + records = tmp_path / f"rank{rank}" / dispatch / "chip_swimlane_records.json" + records.parent.mkdir(parents=True) + records.write_text("{}") + + calls = [] + monkeypatch.setattr(scene_test_module, "_run_swimlane_converter", lambda **kwargs: calls.append(kwargs)) + + scene_test_module._convert_case_swimlane("case", tmp_path) + + assert calls == [] + assert "refusing to pair asymmetric local capture indexes" in caplog.text + + +def test_l3_swimlane_postprocess_uses_parent_identity_when_rank_d_paths_are_reordered(tmp_path, monkeypatch) -> None: + scene_test_module = importlib.import_module("simpler_setup.scene_test") + captures = { + (0, "d0"): (5, 0), + (0, "d1"): (6, 0), + (1, "d0"): (6, 1), + (1, "d1"): (5, 1), + } + for (rank, dispatch), (task_slot, group_index) in captures.items(): + capture = tmp_path / f"rank{rank}" / dispatch + capture.mkdir(parents=True) + (capture / "chip_swimlane_records.json").write_text("{}") + (capture / "dispatch_identity.json").write_text( + json.dumps( + { + "schema_version": 1, + "run_id": 17, + "task_slot": task_slot, + "group_index": group_index, + "group_size": 2, + "chip_rank": rank, + "local_capture_index": int(dispatch.removeprefix("d")), + "endpoint_dispatch_id": int(dispatch.removeprefix("d")) + 1, + "pipeline_slot": 0, + "pipeline_generation": 1, + "callable_digest": "ab" * 32, + } + ) + ) + + calls = [] + monkeypatch.setattr(scene_test_module, "_run_swimlane_converter", lambda **kwargs: calls.append(kwargs)) + + scene_test_module._convert_case_swimlane("case", tmp_path) + + assert [(call["dispatch"], call["dispatch_id"]) for call in calls] == [(None, "17:5"), (None, "17:6")] + assert [call["output_path"].name for call in calls] == [ + "l3_swimlane_run17_task5.json", + "l3_swimlane_run17_task6.json", + ] + + +def test_rank_local_dep_and_scope_postprocessors_follow_swimlane_output_prefix(tmp_path, monkeypatch) -> None: + scene_test_module = importlib.import_module("simpler_setup.scene_test") + captures = [tmp_path / f"rank{rank}" / "d0" for rank in (0, 1)] + for capture in captures: + (capture / "scope_stats").mkdir(parents=True) + (capture / "deps.json").write_text("{}") + (capture / "scope_stats" / "scope_stats.jsonl").write_text("") + + dep_calls = [] + scope_calls = [] + monkeypatch.setattr( + scene_test_module, "_graph_case_dep_gen", lambda _label, path, **_kwargs: dep_calls.append(path) + ) + monkeypatch.setattr(scene_test_module, "_plot_case_scope_stats", lambda _label, path: scope_calls.append(path)) + + scene_test_module.finalize_diagnostic_outputs("case", tmp_path, dep_gen=True, scope_stats=True) + + assert dep_calls == captures + assert scope_calls == captures + + def test_multi_rounds_disable_every_diagnostic() -> None: options = effective_diagnostic_options( 2, diff --git a/tests/ut/py/test_sched_overhead_analysis.py b/tests/ut/py/test_sched_overhead_analysis.py index a4764716e8..a4944e3c19 100644 --- a/tests/ut/py/test_sched_overhead_analysis.py +++ b/tests/ut/py/test_sched_overhead_analysis.py @@ -8,10 +8,13 @@ # ----------------------------------------------------------------------------------------------------------- """Tests for sched_overhead_analysis: overhead model, aicore switch, Head/Tail OH.""" +import os + from simpler_setup.tools.sched_overhead_analysis import ( _scheduler_phases_for_report, _summarize_scheduler_loops, aicore_switch_stats, + auto_select_chip_swimlane_records_json, build_task_graph, compute_critical_path, compute_head_tail, @@ -375,3 +378,23 @@ def test_scheduler_phase_report_suppresses_absent_runtime_phases(): } assert _scheduler_phases_for_report(threads) == ["complete", "async_poll", "dispatch", "resolve", "idle"] + + +def test_auto_select_reaches_both_the_l2_and_the_l3_capture_depths(tmp_path, monkeypatch): + # An L2 case writes its records at outputs//, an L3 case writes one + # per chip below outputs//rankN/dN/ — two levels deeper. A depth-fixed + # glob resolves only one of them and calls the other run "no records". + outputs = tmp_path / "outputs" + l2_records = outputs / "case_l2" / "chip_swimlane_records.json" + l3_records = outputs / "case_l3" / "rank1" / "d0" / "chip_swimlane_records.json" + for path in (l2_records, l3_records): + path.parent.mkdir(parents=True) + path.write_text("{}") + monkeypatch.chdir(tmp_path) + + os.utime(l2_records, (1_000, 1_000)) + os.utime(l3_records, (2_000, 2_000)) + assert auto_select_chip_swimlane_records_json() == l3_records + + os.utime(l2_records, (3_000, 3_000)) + assert auto_select_chip_swimlane_records_json() == l2_records diff --git a/tests/ut/py/test_swimlane_converter.py b/tests/ut/py/test_swimlane_converter.py index bd26e2f20f..2d7b1d5bfa 100644 --- a/tests/ut/py/test_swimlane_converter.py +++ b/tests/ut/py/test_swimlane_converter.py @@ -9,6 +9,9 @@ # ----------------------------------------------------------------------------------------------------------- import json +from pathlib import Path + +import pytest from simpler_setup.tools import swimlane_converter as sc @@ -108,6 +111,267 @@ def _generate_trace(tasks, deps_edges, deps_block_map, tmp_path): return out +def _write_l3_rank(root, rank, *, host_shift_ns, task_id, clock_domain="same-boot", dispatch="d0"): + rank_dir = root / f"rank{rank}" / dispatch + rank_dir.mkdir(parents=True) + device_base = 100 + rank * 100_000 + records = { + "chip_swimlane_level": 4, + "metadata": { + "clock_freq_hz": 1_000_000_000, + "num_cores": 1, + "core_types": ["aiv"], + "core_to_thread": [0], + "orchestrator_source": "host", + "orchestrator_clock_domain": "host_monotonic_ns", + "host_clock_domain_id": clock_domain, + "host_orchestration_origin_ns": host_shift_ns + 1_500, + "host_capture": { + "status": "complete", + "expected_records": 1, + "recorded_records": 1, + "dropped_records": 0, + "error": None, + }, + "clock_anchors": { + "device_timestamp_unit": "syscnt_cycles", + "samples": [ + { + "position": "pre_host_orchestration", + "sample_idx": 0, + "host_before_ns": host_shift_ns + 990, + "device_cycles": device_base, + "host_after_ns": host_shift_ns + 1_010, + "error": None, + }, + { + "position": "post_device_execution", + "sample_idx": 0, + "host_before_ns": host_shift_ns + 8_980, + "device_cycles": device_base + 8_000, + "host_after_ns": host_shift_ns + 9_020, + "error": None, + }, + ], + }, + }, + "aicore_tasks": [[0, task_id, 1, device_base + 1_000, device_base + 1_100, 0]], + "aicpu_tasks": [[0, 1, device_base + 900, device_base + 1_200]], + "aicpu_scheduler_phases": [ + [{"kind": "dispatch", "start_cycles": device_base + 800, "end_cycles": device_base + 850}] + ], + "host_orchestrator_phases": [ + [ + { + "submit_idx": 0, + "task_id": task_id, + "start_host_ns": host_shift_ns + 1_500, + "end_host_ns": host_shift_ns + 1_800, + } + ] + ], + } + (rank_dir / "chip_swimlane_records.json").write_text(json.dumps(records)) + (rank_dir / "name_map.json").write_text(json.dumps({"callable_id_to_name": {"0": f"kernel_r{rank}"}})) + return rank_dir + + +def _write_dispatch_identity(capture_dir, *, run_id, task_slot, group_index, group_size): + rank = int(capture_dir.parent.name.removeprefix("rank")) + capture_index = int(capture_dir.name.removeprefix("d")) + (capture_dir / "dispatch_identity.json").write_text( + json.dumps( + { + "schema_version": 1, + "run_id": run_id, + "task_slot": task_slot, + "group_index": group_index, + "group_size": group_size, + "chip_rank": rank, + "local_capture_index": capture_index, + "endpoint_dispatch_id": capture_index + 1, + "pipeline_slot": 0, + "pipeline_generation": 1, + "callable_digest": "ab" * 32, + } + ) + ) + + +def test_l3_directory_merge_uses_common_host_origin_and_rank_namespaces(tmp_path): + root = tmp_path / "dfx_outputs" + _write_l3_rank(root, 0, host_shift_ns=0, task_id=7) + _write_l3_rank(root, 1, host_shift_ns=10_000, task_id=8) + output = tmp_path / "l3.json" + args = sc._build_parser().parse_args([str(root), "--dispatch", "d0", "-o", str(output)]) + + output_path, rank_metadata = sc._generate_l3_trace(args, root) + + assert output_path == output + assert [item["rank"] for item in rank_metadata] == [0, 1] + trace = json.loads(output.read_text()) + assert trace["metadata"]["global_origin_ns"] == 1_500 + assert trace["metadata"]["host_clock_domain_id"] == "same-boot" + assert trace["metadata"]["cross_rank_uncertainty_ns"] == 40 + assert trace["metadata"]["pre_anchor_group_duration_spread_ns"] == 0 + assert trace["metadata"]["pre_anchor_group_duration_max_ns"] == 20 + assert trace["metadata"]["dispatch_pairing"] == "local_capture_index" + + process_names = { + event["args"]["name"] + for event in trace["traceEvents"] + if event.get("ph") == "M" and event.get("name") == "process_name" + } + assert "rank0 / Worker View" in process_names + assert "rank1 / Worker View" in process_names + worker_events = { + event["args"]["taskId"]: event + for event in trace["traceEvents"] + if event.get("ph") == "X" and event.get("cat") == "event" and event.get("pid") % 100 == 4 + } + assert worker_events[7]["pid"] == 4 + assert worker_events[7]["ts"] == 0.5 + assert worker_events[8]["pid"] == 104 + assert worker_events[8]["ts"] == 10.5 + + +def test_l3_parent_dispatch_identity_pairs_different_local_capture_indexes(tmp_path): + root = tmp_path / "dfx_outputs" + rank0 = _write_l3_rank(root, 0, host_shift_ns=0, task_id=7, dispatch="d0") + rank1 = _write_l3_rank(root, 1, host_shift_ns=10_000, task_id=8, dispatch="d1") + _write_dispatch_identity(rank0, run_id=17, task_slot=5, group_index=0, group_size=2) + _write_dispatch_identity(rank1, run_id=17, task_slot=5, group_index=1, group_size=2) + output = tmp_path / "semantic.json" + args = sc._build_parser().parse_args([str(root), "--dispatch-id", "17:5", "-o", str(output)]) + + _, rank_metadata = sc._generate_l3_trace(args, root) + + trace = json.loads(output.read_text()) + assert [Path(item["input"]).parent.name for item in rank_metadata] == ["d0", "d1"] + assert trace["metadata"]["dispatch_pairing"] == "parent_dispatch_identity" + assert trace["metadata"]["dispatch_identity"] == { + "run_id": 17, + "task_slot": 5, + "group_size": 2, + "callable_digest": "ab" * 32, + } + assert [item["dispatch_identity"]["group_index"] for item in rank_metadata] == [0, 1] + + +def test_l3_local_capture_selector_rejects_different_parent_dispatches(tmp_path): + root = tmp_path / "dfx_outputs" + rank0 = _write_l3_rank(root, 0, host_shift_ns=0, task_id=7) + rank1 = _write_l3_rank(root, 1, host_shift_ns=10_000, task_id=8) + _write_dispatch_identity(rank0, run_id=17, task_slot=5, group_index=0, group_size=2) + _write_dispatch_identity(rank1, run_id=17, task_slot=6, group_index=1, group_size=2) + args = sc._build_parser().parse_args([str(root), "--dispatch", "d0"]) + + with pytest.raises(ValueError, match="different parent dispatches"): + sc._generate_l3_trace(args, root) + + +def test_l3_auto_discovery_rejects_incomplete_parent_group(tmp_path): + root = tmp_path / "dfx_outputs" + rank0 = _write_l3_rank(root, 0, host_shift_ns=0, task_id=7) + _write_l3_rank(root, 1, host_shift_ns=10_000, task_id=8) + _write_dispatch_identity(rank0, run_id=17, task_slot=5, group_index=0, group_size=2) + + with pytest.raises(ValueError, match="incomplete parent dispatch 17:5"): + sc.discover_l3_conversion_targets(root) + + +def test_l3_auto_discovery_pairs_two_groups_despite_reordered_d_paths(tmp_path): + root = tmp_path / "dfx_outputs" + captures = { + (0, "d0"): (17, 5, 0), + (0, "d1"): (17, 6, 0), + (1, "d0"): (17, 6, 1), + (1, "d1"): (17, 5, 1), + } + for (rank, dispatch), (run_id, task_slot, group_index) in captures.items(): + capture_dir = _write_l3_rank( + root, + rank, + host_shift_ns=rank * 10_000, + task_id=rank * 10 + int(dispatch.removeprefix("d")), + dispatch=dispatch, + ) + _write_dispatch_identity( + capture_dir, + run_id=run_id, + task_slot=task_slot, + group_index=group_index, + group_size=2, + ) + + targets = sc.discover_l3_conversion_targets(root) + + assert [(target["dispatch"], target["dispatch_id"]) for target in targets] == [ + (None, "17:5"), + (None, "17:6"), + ] + assert [[path.name for path in target["capture_dirs"]] for target in targets] == [["d0", "d1"], ["d1", "d0"]] + + +def test_l3_auto_discovery_keeps_paired_groups_when_the_remainder_is_asymmetric(tmp_path, capsys): + # A group is paired by (run_id, task_slot), so it is unaffected by what the + # leftover dN sets look like. Refusing the whole root would discard exactly + # the pairing the parent identity exists to make. + root = tmp_path / "dfx_outputs" + for rank, group_index in ((0, 0), (1, 1)): + capture_dir = _write_l3_rank(root, rank, host_shift_ns=rank * 10_000, task_id=rank, dispatch="d0") + _write_dispatch_identity(capture_dir, run_id=17, task_slot=5, group_index=group_index, group_size=2) + # An extra individually submitted capture on rank0 only: no sibling to pair + # it with, and no identity that would let it pair by anything but its name. + _write_l3_rank(root, 0, host_shift_ns=0, task_id=99, dispatch="d1") + + targets = sc.discover_l3_conversion_targets(root) + + assert [(target["dispatch"], target["dispatch_id"]) for target in targets] == [(None, "17:5")] + assert "refusing to pair asymmetric local capture indexes" in capsys.readouterr().err + + +def test_rank_namespace_does_not_turn_rank_into_a_counter_series(): + trace = { + "traceEvents": [ + {"ph": "C", "pid": 2, "tid": 1, "args": {"AIC": 3, "AIV": 4}}, + {"ph": "X", "pid": 4, "tid": 2, "args": {"taskId": 9}}, + ] + } + + sc._namespace_rank_trace(trace, 2) + + counter, task = trace["traceEvents"] + assert counter["pid"] == 202 + assert counter["args"] == {"AIC": 3, "AIV": 4} + assert task["pid"] == 204 + assert task["args"]["rank"] == 2 + + +def test_rank_namespace_rejects_a_view_pid_wider_than_the_stride(): + trace = {"traceEvents": [{"ph": "X", "pid": sc._RANK_PID_STRIDE, "tid": 1, "args": {}}]} + + with pytest.raises(ValueError, match="does not fit the per-Rank stride"): + sc._namespace_rank_trace(trace, 1) + + +def test_l3_directory_merge_rejects_different_or_missing_host_clock_domains(tmp_path): + root = tmp_path / "dfx_outputs" + _write_l3_rank(root, 0, host_shift_ns=0, task_id=7, clock_domain="boot-a") + rank1_dir = _write_l3_rank(root, 1, host_shift_ns=10_000, task_id=8, clock_domain="boot-b") + args = sc._build_parser().parse_args([str(root), "--dispatch", "d0"]) + + with pytest.raises(ValueError, match="different Host clock domains"): + sc._generate_l3_trace(args, root) + + rank1_path = rank1_dir / "chip_swimlane_records.json" + rank1 = json.loads(rank1_path.read_text()) + rank1["metadata"].pop("host_clock_domain_id") + rank1_path.write_text(json.dumps(rank1)) + with pytest.raises(ValueError, match="missing metadata.host_clock_domain_id"): + sc._generate_l3_trace(args, root) + + def test_task_statistics_level_one_hides_aicpu_metrics(capsys): tasks = [ { @@ -228,6 +492,8 @@ def test_host_orchestrator_phases_without_anchors_are_marked_unaligned(tmp_path) "cross_domain_gap_unknown": True, "cross_domain_latency_available": False, "logical_seam_us": 2.0, + "source_timeline_origin_ns": 1_000, + "timeline_origin_ns": 1_000, } trace_path = tmp_path / "merged_swimlane.json" @@ -253,6 +519,63 @@ def test_host_orchestrator_phases_without_anchors_are_marked_unaligned(tmp_path) ) +def test_aicpu_orchestrator_uses_host_timeline_when_clock_anchors_exist(tmp_path): + raw = tmp_path / "chip_swimlane_records.json" + raw.write_text( + json.dumps( + { + "chip_swimlane_level": 4, + "metadata": { + "clock_freq_hz": 1_000_000_000, + "num_cores": 1, + "core_types": ["aiv"], + "core_to_thread": [0], + "host_clock_domain_id": "same-boot", + "host_timeline_origin_ns": 1_000, + "clock_anchors": { + "device_timestamp_unit": "syscnt_cycles", + "samples": [ + { + "position": "pre_host_orchestration", + "sample_idx": 0, + "host_before_ns": 990, + "device_cycles": 100, + "host_after_ns": 1_010, + "error": None, + }, + { + "position": "post_device_execution", + "sample_idx": 0, + "host_before_ns": 5_080, + "device_cycles": 4_100, + "host_after_ns": 5_120, + "error": None, + }, + ], + }, + }, + "aicore_tasks": [[0, 7, 1, 2_100, 2_200, 0]], + "aicpu_tasks": [[0, 1, 2_000, 2_300]], + "aicpu_scheduler_phases": [ + [{"kind": "dispatch", "start_cycles": 1_900, "end_cycles": 1_950, "tasks_processed": 1}] + ], + "aicpu_orchestrator_phases": [ + [{"submit_idx": 0, "task_id": 7, "start_cycles": 1_800, "end_cycles": 1_850}] + ], + } + ) + ) + + data = sc.read_perf_data(raw) + + assert data["orchestrator_source"] == "aicpu" + assert data["tasks"][0]["start_time_us"] == 2.05 + assert data["timeline_metadata"]["layout"] == "clock_aligned" + assert data["timeline_metadata"]["clock_alignment"]["status"] == "calibrated" + assert data["timeline_metadata"]["host_clock_domain_id"] == "same-boot" + assert data["timeline_metadata"]["source_timeline_origin_ns"] == 1_000 + + def test_host_capture_is_complete_when_the_pool_holds_more_than_the_submit_projection(tmp_path): """A pool record count above the projected one is normal, not incomplete. @@ -380,6 +703,10 @@ def test_host_and_device_timestamps_use_calibrated_clock_alignment(tmp_path): "pre_host_orchestration": 0, "post_device_execution": 0, }, + "anchor_group_duration_ns": { + "pre_host_orchestration": 20, + "post_device_execution": 40, + }, }, "host_capture": { "status": "complete", @@ -390,6 +717,8 @@ def test_host_and_device_timestamps_use_calibrated_clock_alignment(tmp_path): }, "host_records_complete": True, "cross_domain_latency_available": True, + "source_timeline_origin_ns": 1_500, + "timeline_origin_ns": 1_500, } diff --git a/tests/ut/py/test_worker/test_host_worker.py b/tests/ut/py/test_worker/test_host_worker.py index 4e3d3aa19c..95754e5cae 100644 --- a/tests/ut/py/test_worker/test_host_worker.py +++ b/tests/ut/py/test_worker/test_host_worker.py @@ -225,7 +225,10 @@ def init(self, device_id, bins, *, log_level, prewarm_config=None, enable_sdma=F def finalize(self) -> None: events.append(("finalize",)) - def fake_run_chip_main_loop(cw, *_args, chip_platform, chip_runtime, prepared=None, task_frame_count=1): + def fake_run_chip_main_loop( + cw, *_args, chip_platform, chip_runtime, prepared=None, task_frame_count=1, chip_rank=None + ): + assert chip_rank is None published_depths.append(worker_mod._PIPELINE_LEASE_FMT.unpack_from(_args[0], worker_mod._OFF_PIPELINE_LEASE)[0]) published_frame_counts.append(task_frame_count) events.append(("main_loop", cw, chip_platform, chip_runtime)) @@ -1082,13 +1085,16 @@ def publish( state: int = worker_mod._TASK_READY, generation: int = 11, diagnostics: bool = False, + task_slot: Optional[int] = None, + group_index: int = 0, + group_size: int = 1, ) -> None: offset = self._frame_offset(index) frame = self.buf[offset : offset + worker_mod.MAILBOX_FRAME_SIZE] try: frame[worker_mod._OFF_TASK_CALLABLE_HASH : worker_mod._OFF_TASK_ARGS_BLOB] = self.digest struct.pack_into("=ii", frame, worker_mod._OFF_TASK_ARGS_BLOB, 0, 0) - cfg_values = [0] * (6 + 3 * worker_mod.RUNTIME_ENV_RING_COUNT) + cfg_values = [0] * (7 + 3 * worker_mod.RUNTIME_ENV_RING_COUNT) cfg_values[3] = int(diagnostics) output_prefix = b"/tmp/simpler-test" if diagnostics else b"" worker_mod._CFG_FMT.pack_into(frame, worker_mod._OFF_CONFIG, *cfg_values, output_prefix) @@ -1098,6 +1104,11 @@ def publish( struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_SLOT_ID, index) struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_GENERATION, generation) struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_DISPATCH_ID, dispatch_id) + struct.pack_into( + "=Q", frame, worker_mod._OFF_FRAME_TASK_SLOT, dispatch_id if task_slot is None else task_slot + ) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_GROUP_INDEX, group_index) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_GROUP_SIZE, group_size) finally: frame.release() _mailbox_store_i32(self.accepted_addr(index), 0) @@ -8169,6 +8180,91 @@ def test_unregister_removes_only_after_last_digest_ref(self): payload_shm.unlink() +def test_a_failed_diagnostic_sidecar_write_fails_the_task_not_the_loop(tmp_path): + """A rankN/dN write failure is this task's error, not the loop's. + + The sidecar is written while the task config is read, before anything the + task itself does. An exception escaping there leaves ``_run_mailbox_loop`` + without publishing TASK_DONE or an error message, so the parent waits on a + mailbox that never completes — a far worse outcome than the diagnostic + artifact this path exists to produce. + """ + from unittest.mock import MagicMock # noqa: PLC0415 + + # rank0 as a regular file makes os.makedirs("/rank0/d0") raise, + # while output_prefix itself stays a directory the host log setter accepts. + (tmp_path / "rank0").write_text("") + + shm = SharedMemory(create=True, size=MAILBOX_SIZE) + buf = shm.buf + assert buf is not None + mailbox_addr = _mailbox_addr(shm) + state_addr = mailbox_addr + _OFF_STATE + frame_state_addr = mailbox_addr + worker_mod.MAILBOX_FRAME_SIZE + _OFF_STATE + _mailbox_store_i32(state_addr, worker_mod._IDLE) + + digest = bytes([0x42]) * worker_mod.CALLABLE_HASH_DIGEST_BYTES + frame = buf[worker_mod.MAILBOX_FRAME_SIZE : 2 * worker_mod.MAILBOX_FRAME_SIZE] + try: + frame[worker_mod._OFF_TASK_CALLABLE_HASH : worker_mod._OFF_TASK_ARGS_BLOB] = digest + struct.pack_into("=ii", frame, worker_mod._OFF_TASK_ARGS_BLOB, 0, 0) + cfg_values = [0] * (7 + 3 * worker_mod.RUNTIME_ENV_RING_COUNT) + cfg_values[1] = 4 # enable_chip_swimlane + worker_mod._CFG_FMT.pack_into(frame, worker_mod._OFF_CONFIG, *cfg_values, str(tmp_path).encode()) + worker_mod._PIPELINE_LEASE_FMT.pack_into(frame, worker_mod._OFF_PIPELINE_LEASE, 0, 0, 11) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_PROTOCOL, worker_mod._TASK_PROTOCOL_VERSION) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_RUN_ID, 5) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_SLOT_ID, 0) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_GENERATION, 11) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_DISPATCH_ID, 1) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_TASK_SLOT, 3) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_GROUP_INDEX, 0) + struct.pack_into("=Q", frame, worker_mod._OFF_FRAME_GROUP_SIZE, 1) + finally: + frame.release() + + thread = threading.Thread( + target=worker_mod._run_chip_main_loop, + args=( + MagicMock(), + buf, + mailbox_addr, + state_addr, + 0, + {}, + {digest: 7}, + {digest: 1}, + worker_mod.mint_owner_instance_id(), + ), + kwargs={"chip_platform": "a2a3", "chip_runtime": "", "prepared": {7}, "chip_rank": 0}, + daemon=True, + ) + try: + _mailbox_store_i32(frame_state_addr, worker_mod._TASK_READY) + thread.start() + deadline = time.monotonic() + 5.0 + while _mailbox_load_i32(frame_state_addr) != worker_mod._TASK_DONE: + assert time.monotonic() < deadline, "loop exited without publishing TASK_DONE" + time.sleep(0.001) + + task_frame = buf[worker_mod.MAILBOX_FRAME_SIZE : 2 * worker_mod.MAILBOX_FRAME_SIZE] + try: + error_code = struct.unpack_from("i", task_frame, worker_mod._OFF_ERROR)[0] + raw = bytes(task_frame[MAILBOX_OFF_ERROR_MSG : MAILBOX_OFF_ERROR_MSG + MAILBOX_ERROR_MSG_SIZE]) + finally: + task_frame.release() + assert error_code == 1 + assert raw.split(b"\x00", 1)[0].decode("utf-8", "replace").startswith("chip_process dev=0") + assert not (tmp_path / "rank0" / "d0").exists() + assert thread.is_alive() + finally: + _mailbox_store_i32(state_addr, worker_mod._SHUTDOWN) + thread.join(5.0) + assert not thread.is_alive() + shm.close() + shm.unlink() + + def test_the_cpp_pre_bind_level_word_is_the_ladder_word_for_l3(): """`host_span_names.h` hand-writes the L3 word a third time, as its pre-bind default.