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.