From 21cd5a7488109c7e32409db48eb30b145acf28d0 Mon Sep 17 00:00:00 2001 From: HighCloud Date: Mon, 24 Aug 2026 18:56:34 -0700 Subject: [PATCH 1/2] Add: sustained four-chip step jitter reproducer Add a Simpler-only L3 workload that keeps two four-chip groups in flight and checks deterministic vector-add output over long runs. Include STRACE analysis for runner start/end skew, host-versus-device runner excess, runner-to-validate gaps, validate latency, and complete step latency, plus a wrapper that emits raw logs and a merged host swimlane. Related: #1995 --- examples/workers/README.md | 2 + examples/workers/l3/README.md | 1 + .../workers/l3/step_jitter_repro/README.md | 31 +++ .../l3/step_jitter_repro/analyze_strace.py | 209 ++++++++++++++++++ .../kernels/aiv/repeated_vector_add.cpp | 64 ++++++ .../orchestration/repeated_vector_add.cpp | 39 ++++ examples/workers/l3/step_jitter_repro/main.py | 170 ++++++++++++++ .../workers/l3/step_jitter_repro/run_4card.sh | 31 +++ 8 files changed, 547 insertions(+) create mode 100644 examples/workers/l3/step_jitter_repro/README.md create mode 100644 examples/workers/l3/step_jitter_repro/analyze_strace.py create mode 100644 examples/workers/l3/step_jitter_repro/kernels/aiv/repeated_vector_add.cpp create mode 100644 examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp create mode 100644 examples/workers/l3/step_jitter_repro/main.py create mode 100644 examples/workers/l3/step_jitter_repro/run_4card.sh diff --git a/examples/workers/README.md b/examples/workers/README.md index 6516dff1df..f96be2f21e 100644 --- a/examples/workers/README.md +++ b/examples/workers/README.md @@ -33,6 +33,7 @@ workers/ l3/ # Multi-chip examples (host-level DAG) multi_chip_dispatch/ # Worker(level=3) + orchestration + SubWorker child_memory/ # orch.malloc + child_memory=True, weight reuse across tasks + step_jitter_repro/ # Four-chip sustained dispatch + host STRACE analysis l4/ # Multi-machine examples (one L3 here, one over TCP or mpirun) vector_add_mixed_l3/ # Worker(level=4) + add_remote_worker, golden checked on both sides global_tload_mixed_l3/ # Global CommDomain build + cross-machine peer TLOAD on both ranks @@ -144,6 +145,7 @@ python examples/workers/l2/worker_malloc/main.py -p a2a3sim -d 0 python examples/workers/l2/vector_add/main.py -p a2a3sim -d 0 python examples/workers/l3/multi_chip_dispatch/main.py -p a2a3sim -d 0-1 python examples/workers/l3/child_memory/main.py -p a2a3sim -d 0 +python examples/workers/l3/step_jitter_repro/main.py -p a2a3sim -d 0-3 --rounds 10 ``` Flags: diff --git a/examples/workers/l3/README.md b/examples/workers/l3/README.md index 64dcf4c045..82e3836965 100644 --- a/examples/workers/l3/README.md +++ b/examples/workers/l3/README.md @@ -68,6 +68,7 @@ has its own README with the run commands and what the golden check proves. | [`multi_chip_dispatch/`](multi_chip_dispatch/) | Two chips + one SubWorker. An orchestration fn dispatches a `ChipCallable` to each chip, then submits a Python callable to collect/verify results. The smallest correct L3 program. | | [`child_memory/`](child_memory/) | `orch.malloc` + `ChipTensor(child_memory=True)` to load a weight once and reuse it across multiple kernel invocations on the same chip. | | [`per_task_runtime_env/`](per_task_runtime_env/) | One L3 launch where each L2 task binds its own ring sizes through `CallConfig.runtime_env`. | +| [`step_jitter_repro/`](step_jitter_repro/) | Sustained four-chip depth-two dispatch with STRACE analysis for host scheduling and validation latency outliers. | ### Communication domains and collectives diff --git a/examples/workers/l3/step_jitter_repro/README.md b/examples/workers/l3/step_jitter_repro/README.md new file mode 100644 index 0000000000..42eeac5eb2 --- /dev/null +++ b/examples/workers/l3/step_jitter_repro/README.md @@ -0,0 +1,31 @@ +# Simpler-only multi-chip step-jitter reproducer + +This example depends only on the current Simpler checkout and its normal runtime dependencies. It does not import +PyPTO, pypto-lib, or any serving package. + +One logical step submits a four-member chip group. Two top-level `Worker.submit` handles remain in flight, and the +oldest handle is retired only after its successor has been submitted. Each chip run submits repeated vector work so +the device remains busy long enough for host scheduling jitter to be observable. + +```bash +export SIMPLER_LOG_LEVEL=TIMING +python examples/workers/l3/step_jitter_repro/main.py \ + -p a2a3 -d "$TASK_DEVICE" --warmup 5 --rounds 1000 --depth 2 \ + --kernel-repeats 4096 2>&1 | tee /tmp/simpler-step-jitter.log + +python examples/workers/l3/step_jitter_repro/analyze_strace.py \ + /tmp/simpler-step-jitter.log --warmup 5 --rounds 1000 \ + --json-out /tmp/simpler-step-jitter-analysis.json \ + --trace-out /tmp/simpler-step-jitter-swimlane.json +``` + +The wrapper below runs the workload and writes the raw STRACE, analysis, and merged four-device swimlane to one +output directory. `TASK_DEVICE` must contain exactly four comma-separated device IDs: + +```bash +TASK_DEVICE=0,1,2,3 PYTHON=.venv/bin/python \ + bash examples/workers/l3/step_jitter_repro/run_4card.sh /tmp/simpler-jitter +``` + +On shared hardware, run this wrapper inside a four-device allocation so all devices remain reserved for the complete +workload. diff --git a/examples/workers/l3/step_jitter_repro/analyze_strace.py b/examples/workers/l3/step_jitter_repro/analyze_strace.py new file mode 100644 index 0000000000..05d87d8477 --- /dev/null +++ b/examples/workers/l3/step_jitter_repro/analyze_strace.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- +"""Analyze Simpler chip.run STRACE spans and emit a Perfetto swimlane.""" + +from __future__ import annotations + +import argparse +import json +import re +import statistics +from collections import defaultdict +from pathlib import Path + +READY_RE = re.compile(r"\[chip_process pid=(\d+) dev=(\d+)\] ready") +SPAN_RE = re.compile(r"\[STRACE\].*\bpid=(\d+).*\binv=(\d+).*\bname=(\S+).*\bts=(\d+)\s+dur=(\d+)") +FIELDS_RE = re.compile(r"\b([a-zA-Z_][a-zA-Z0-9_]*)=([^ ]+)") +TRACKED = { + "chip.run", + "chip.run.pre_bind", + "chip.run.bind", + "chip.run.post_bind", + "chip.run.runner_run", + "chip.run.runner_run.device_wall", + "chip.run.validate", + "chip.run.claim_release", +} +REQUIRED = {"chip.run", "chip.run.runner_run", "chip.run.runner_run.device_wall", "chip.run.validate"} + + +def _percentile(values: list[float], quantile: float) -> float: + ordered = sorted(values) + return ordered[round((len(ordered) - 1) * quantile)] + + +def _summary(values: list[float]) -> dict[str, float]: + return { + "p50": _percentile(values, 0.50), + "p95": _percentile(values, 0.95), + "p99": _percentile(values, 0.99), + "max": max(values), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("log", type=Path) + parser.add_argument("--warmup", type=int, default=5) + parser.add_argument("--rounds", type=int, default=1000) + parser.add_argument("--threshold-ms", type=float, default=5.0) + parser.add_argument("--json-out", type=Path, required=True) + parser.add_argument("--trace-out", type=Path, required=True) + args = parser.parse_args() + + text = args.log.read_text(errors="replace") + pid_to_device = {int(pid): int(device) for pid, device in READY_RE.findall(text)} + spans = defaultdict(dict) + trace_events = [] + for line in text.splitlines(): + match = SPAN_RE.search(line) + if not match: + continue + pid, invocation, name, ts, dur = match.groups() + pid = int(pid) + if pid not in pid_to_device or name not in TRACKED: + continue + device = pid_to_device[pid] + invocation = int(invocation) + ts_ns = int(ts) + dur_ns = int(dur) + fields = {key: value for key, value in FIELDS_RE.findall(line)} + spans[(invocation, device)][name] = (ts_ns, dur_ns) + if name == "chip.run.runner_run.device_wall": + continue + trace_events.append( + { + "name": name.removeprefix("chip.run."), + "cat": "simpler.host_strace", + "ph": "X", + "pid": 1000 + device, + "tid": device, + "ts": ts_ns / 1000.0, + "dur": dur_ns / 1000.0, + "args": {"round": invocation, **fields}, + } + ) + + devices = sorted(pid_to_device.values()) + first = args.warmup + 1 + last = args.warmup + args.rounds + rounds = [] + outliers = [] + for invocation in range(first, last + 1): + rows = {} + for device in devices: + row = spans.get((invocation, device), {}) + missing = REQUIRED - row.keys() + if missing: + raise RuntimeError(f"round {invocation} device {device} missing spans: {sorted(missing)}") + root_ts, root_dur = row["chip.run"] + runner_ts, runner_dur = row["chip.run.runner_run"] + _device_ts, device_wall_dur = row["chip.run.runner_run.device_wall"] + validate_ts, validate_dur = row["chip.run.validate"] + rows[device] = { + "root_start_ms": root_ts / 1e6, + "step_ms": root_dur / 1e6, + "runner_start_ms": runner_ts / 1e6, + "runner_ms": runner_dur / 1e6, + "device_wall_ms": device_wall_dur / 1e6, + "runner_host_excess_ms": (runner_dur - device_wall_dur) / 1e6, + "runner_end_ms": (runner_ts + runner_dur) / 1e6, + "runner_to_validate_gap_ms": (validate_ts - runner_ts - runner_dur) / 1e6, + "validate_ms": validate_dur / 1e6, + "step_end_ms": (root_ts + root_dur) / 1e6, + } + starts = [row["runner_start_ms"] for row in rows.values()] + ends = [row["runner_end_ms"] for row in rows.values()] + step_ends = [row["step_end_ms"] for row in rows.values()] + record = { + "round": invocation - args.warmup, + "invocation": invocation, + "runner_start_skew_ms": max(starts) - min(starts), + "runner_end_skew_ms": max(ends) - min(ends), + "step_end_skew_ms": max(step_ends) - min(step_ends), + "rows": rows, + } + rounds.append(record) + + median_start = statistics.median(starts) + median_end = statistics.median(ends) + for device, row in rows.items(): + if row["runner_start_ms"] - median_start > args.threshold_ms: + outliers.append( + { + "round": record["round"], + "device": device, + "category": "runner_late_start", + "above_median_ms": row["runner_start_ms"] - median_start, + } + ) + if row["runner_end_ms"] - median_end > args.threshold_ms: + outliers.append( + { + "round": record["round"], + "device": device, + "category": "runner_long_tail", + "above_median_ms": row["runner_end_ms"] - median_end, + } + ) + + metrics = {} + for field in ("runner_start_skew_ms", "runner_end_skew_ms", "step_end_skew_ms"): + metrics[field] = _summary([record[field] for record in rounds]) + for field in ( + "runner_ms", + "device_wall_ms", + "runner_host_excess_ms", + "runner_to_validate_gap_ms", + "validate_ms", + "step_ms", + ): + metrics[field] = _summary([row[field] for record in rounds for row in record["rows"].values()]) + + result = { + "source": str(args.log), + "devices": devices, + "warmup": args.warmup, + "rounds": args.rounds, + "threshold_ms": args.threshold_ms, + "metrics_ms": metrics, + "outlier_counts": { + category: sum(item["category"] == category for item in outliers) + for category in ("runner_late_start", "runner_long_tail") + }, + "outliers": outliers, + "round_data": rounds, + } + args.json_out.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n") + + for device in devices: + trace_events.extend( + [ + { + "name": "process_name", + "ph": "M", + "pid": 1000 + device, + "tid": 0, + "args": {"name": f"Device {device}"}, + }, + {"name": "thread_name", "ph": "M", "pid": 1000 + device, "tid": device, "args": {"name": "chip.run"}}, + ] + ) + trace = { + "displayTimeUnit": "ms", + "metadata": {"source": str(args.log), "devices": devices, "simpler_only": True}, + "traceEvents": trace_events, + } + args.trace_out.write_text(json.dumps(trace, separators=(",", ":"))) + print(json.dumps({"devices": devices, "metrics_ms": metrics, "outlier_counts": result["outlier_counts"]})) + + +if __name__ == "__main__": + main() diff --git a/examples/workers/l3/step_jitter_repro/kernels/aiv/repeated_vector_add.cpp b/examples/workers/l3/step_jitter_repro/kernels/aiv/repeated_vector_add.cpp new file mode 100644 index 0000000000..e9812d6590 --- /dev/null +++ b/examples/workers/l3/step_jitter_repro/kernels/aiv/repeated_vector_add.cpp @@ -0,0 +1,64 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include +#include + +#include "tensor.h" + +using namespace pto; + +#include "pipe_sync.h" + +#ifndef __gm__ +#define __gm__ +#endif + +#ifndef __aicore__ +#define __aicore__ [aicore] +#endif + +extern "C" __aicore__ __attribute__((always_inline)) void kernel_entry(__gm__ int64_t *args) { + __gm__ ChipTensor *src0_tensor = reinterpret_cast<__gm__ ChipTensor *>(args[0]); + __gm__ ChipTensor *src1_tensor = reinterpret_cast<__gm__ ChipTensor *>(args[1]); + __gm__ ChipTensor *out_tensor = reinterpret_cast<__gm__ ChipTensor *>(args[2]); + const uint64_t repeats = static_cast(args[3]); + __gm__ float *src0 = reinterpret_cast<__gm__ float *>(src0_tensor->buffer.addr) + src0_tensor->start_offset; + __gm__ float *src1 = reinterpret_cast<__gm__ float *>(src1_tensor->buffer.addr) + src1_tensor->start_offset; + __gm__ float *out = reinterpret_cast<__gm__ float *>(out_tensor->buffer.addr) + out_tensor->start_offset; + + constexpr int kRows = 128; + constexpr int kCols = 128; + using GlobalData = GlobalTensor, pto::Stride<1, 1, 1, kCols, 1>>; + using TileData = Tile; + + TileData src0_tile(kRows, kCols); + TileData src1_tile(kRows, kCols); + TileData dst_tile(kRows, kCols); + TASSIGN(src0_tile, 0x0); + TASSIGN(src1_tile, 0x10000); + TASSIGN(dst_tile, 0x20000); + GlobalData src0_global(src0); + GlobalData src1_global(src1); + GlobalData dst_global(out); + + for (uint64_t index = 0; index < repeats; ++index) { + TLOAD(src0_tile, src0_global); + TLOAD(src1_tile, src1_global); + set_flag(PIPE_MTE2, PIPE_V, EVENT_ID0); + wait_flag(PIPE_MTE2, PIPE_V, EVENT_ID0); + TADD(dst_tile, src0_tile, src1_tile); + set_flag(PIPE_V, PIPE_MTE3, EVENT_ID0); + wait_flag(PIPE_V, PIPE_MTE3, EVENT_ID0); + TSTORE(dst_global, dst_tile); + pipe_sync(); + } +} diff --git a/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp b/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp new file mode 100644 index 0000000000..3554370ab4 --- /dev/null +++ b/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp @@ -0,0 +1,39 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include +#include + +#include "orchestration_api.h" // NOLINT(build/include_subdir) + +extern "C" { + +__attribute__((visibility("default"))) OrchestrationConfig +aicpu_orchestration_config(const ChipTaskArgs &orch_args) { + (void)orch_args; // NOLINT(readability/casting) + return OrchestrationConfig{ + .expected_arg_count = 4, + }; +} + +__attribute__((visibility("default"))) void repeated_vector_add(const ChipTaskArgs &orch_args) { + const ChipTensor &a = orch_args.tensor(0).ref(); + const ChipTensor &b = orch_args.tensor(1).ref(); + const ChipTensor &out = orch_args.tensor(2).ref(); + CoreTaskArgs params; + params.add_input(a); + params.add_input(b); + params.add_output(out); + params.add_scalar(orch_args.scalar(0)); + rt_submit_aiv_task(0, params); +} + +} // extern "C" diff --git a/examples/workers/l3/step_jitter_repro/main.py b/examples/workers/l3/step_jitter_repro/main.py new file mode 100644 index 0000000000..f98327e442 --- /dev/null +++ b/examples/workers/l3/step_jitter_repro/main.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- +"""Reproduce multi-chip step jitter with Simpler APIs only. + +Each logical step is one four-member ``submit_next_level_group``. The parent +keeps two ``Worker.submit`` handles in flight and retires the oldest handle +only after the successor has been submitted, matching a serving-style +continuous depth-two pipeline. +""" + +from __future__ import annotations + +import argparse +import os +import time +from pathlib import Path + +os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE") + +import torch # noqa: E402 +from simpler.task_interface import ArgDirection, CallConfig, ChipCallable, CoreCallable, TaskArgs, TensorArgType +from simpler.worker import Worker + +from simpler_setup.elf_parser import extract_text_section +from simpler_setup.kernel_compiler import KernelCompiler +from simpler_setup.pto_isa import ensure_pto_isa_root +from simpler_setup.torch_interop import make_tensor_arg + +HERE = Path(__file__).resolve().parent +VECTOR_KERNEL = HERE / "kernels" / "aiv" / "repeated_vector_add.cpp" +ORCHESTRATION = HERE / "kernels" / "orchestration" / "repeated_vector_add.cpp" +ROWS = 128 +COLS = 128 + + +def _parse_devices(value: str) -> list[int]: + devices = [] + for item in value.split(","): + if "-" in item: + first, last = (int(part) for part in item.split("-", maxsplit=1)) + devices.extend(range(first, last + 1)) + elif item: + devices.append(int(item)) + if len(devices) != 4 or len(set(devices)) != 4: + raise argparse.ArgumentTypeError(f"exactly four distinct devices are required, got {devices}") + return devices + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("-p", "--platform", default="a2a3", choices=["a2a3", "a2a3sim"]) + parser.add_argument("-d", "--device", required=True, type=_parse_devices) + parser.add_argument("--warmup", type=int, default=5) + parser.add_argument("--rounds", type=int, default=1000) + parser.add_argument("--depth", type=int, default=2, choices=[1, 2]) + parser.add_argument("--kernel-repeats", type=int, default=4096) + return parser.parse_args() + + +def _build_callable(platform: str) -> ChipCallable: + compiler = KernelCompiler(platform=platform) + runtime = "tensormap_and_ringbuffer" + include_dirs = compiler.get_orchestration_include_dirs(runtime) + kernel = compiler.compile_incore( + source_path=str(VECTOR_KERNEL), + core_type="aiv", + pto_isa_root=ensure_pto_isa_root(), + extra_include_dirs=include_dirs, + ) + if not platform.endswith("sim"): + kernel = extract_text_section(kernel) + orchestration = compiler.compile_orchestration(runtime_name=runtime, source_path=str(ORCHESTRATION)) + core = CoreCallable.build( + signature=[ArgDirection.IN, ArgDirection.IN, ArgDirection.OUT, ArgDirection.IN], + binary=kernel, + ) + return ChipCallable.build( + signature=[ArgDirection.IN, ArgDirection.IN, ArgDirection.OUT, ArgDirection.IN], + func_name="repeated_vector_add", + binary=orchestration, + children=[(0, core)], + ) + + +def _task_args(worker: Worker, tensors: tuple[torch.Tensor, torch.Tensor, torch.Tensor], repeats: int) -> TaskArgs: + a, b, out = tensors + args = TaskArgs() + args.add_tensor(make_tensor_arg(worker, a), TensorArgType.INPUT) + args.add_tensor(make_tensor_arg(worker, b), TensorArgType.INPUT) + args.add_tensor(make_tensor_arg(worker, out), TensorArgType.OUTPUT_EXISTING) + args.add_scalar(repeats) + return args + + +def run(config: argparse.Namespace) -> None: + if config.warmup < 0 or config.rounds <= 0 or config.kernel_repeats <= 0: + raise ValueError("warmup must be non-negative; rounds and kernel-repeats must be positive") + + devices = config.device + workers = list(range(len(devices))) + total = config.warmup + config.rounds + print( + f"[repro] devices={devices} warmup={config.warmup} rounds={config.rounds} " + f"depth={config.depth} kernel_repeats={config.kernel_repeats}", + flush=True, + ) + + torch.manual_seed(20260824) + slots = [] + for _slot in range(config.depth): + per_device = [] + for _device in devices: + per_device.append( + ( + torch.full((ROWS, COLS), 1.0, dtype=torch.float32).share_memory_(), + torch.full((ROWS, COLS), 2.0, dtype=torch.float32).share_memory_(), + torch.zeros((ROWS, COLS), dtype=torch.float32).share_memory_(), + ) + ) + slots.append(per_device) + + worker = Worker( + level=3, + platform=config.platform, + runtime="tensormap_and_ringbuffer", + device_ids=devices, + num_sub_workers=0, + ) + chip_handle = worker.register(_build_callable(config.platform)) + worker.init() + started = time.monotonic() + handles = [] + try: + slot_args = [ + [_task_args(worker, tensors, config.kernel_repeats) for tensors in per_device] for per_device in slots + ] + call_config = CallConfig() + + for step in range(total): + while len(handles) >= config.depth: + handles.pop(0).wait() + args_for_step = slot_args[step % config.depth] + + def graph(orch, _args, cfg, group_args=args_for_step): + orch.submit_next_level_group(chip_handle, group_args, cfg, workers=workers) + + handles.append(worker.submit(graph, config=call_config)) + + for handle in handles: + handle.wait() + + elapsed = time.monotonic() - started + for slot in slots: + for _a, _b, out in slot: + if not torch.allclose(out, torch.full_like(out, 3.0), rtol=0.0, atol=0.0): + raise AssertionError("vector-add output mismatch") + print(f"[repro] PASS total_steps={total} elapsed_s={elapsed:.3f}", flush=True) + finally: + worker.close() + + +if __name__ == "__main__": + run(_parse_args()) diff --git a/examples/workers/l3/step_jitter_repro/run_4card.sh b/examples/workers/l3/step_jitter_repro/run_4card.sh new file mode 100644 index 0000000000..3935fc914a --- /dev/null +++ b/examples/workers/l3/step_jitter_repro/run_4card.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- +set -euo pipefail + +script_dir=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +repo_root=$(cd -- "$script_dir/../../../.." && pwd) +output_dir=${1:?usage: run_4card.sh OUTPUT_DIR [ROUNDS] [KERNEL_REPEATS]} +rounds=${2:-1000} +kernel_repeats=${3:-4096} +python_bin=${PYTHON:-python} +devices=${TASK_DEVICE:?TASK_DEVICE must contain four allocated device IDs} + +cd "$repo_root" +mkdir -p "$output_dir" + +SIMPLER_LOG_LEVEL=TIMING "$python_bin" \ + examples/workers/l3/step_jitter_repro/main.py \ + -p a2a3 -d "$devices" --warmup 5 --rounds "$rounds" --depth 2 \ + --kernel-repeats "$kernel_repeats" >"$output_dir/run.log" 2>&1 + +"$python_bin" examples/workers/l3/step_jitter_repro/analyze_strace.py \ + "$output_dir/run.log" --warmup 5 --rounds "$rounds" \ + --json-out "$output_dir/analysis.json" \ + --trace-out "$output_dir/swimlane.json" From 42cc1ae1cb70808023e4cb91fd2d9d96a0960d32 Mon Sep 17 00:00:00 2001 From: HighCloud Date: Mon, 24 Aug 2026 19:31:59 -0700 Subject: [PATCH 2/2] Fix: format step jitter orchestration entry point Apply the repository clang-format layout required by the pre-commit check. Related: #1995 --- .../kernels/orchestration/repeated_vector_add.cpp | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp b/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp index 3554370ab4..b19e7061ec 100644 --- a/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp +++ b/examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp @@ -16,8 +16,7 @@ extern "C" { -__attribute__((visibility("default"))) OrchestrationConfig -aicpu_orchestration_config(const ChipTaskArgs &orch_args) { +__attribute__((visibility("default"))) OrchestrationConfig aicpu_orchestration_config(const ChipTaskArgs &orch_args) { (void)orch_args; // NOLINT(readability/casting) return OrchestrationConfig{ .expected_arg_count = 4,