Add: sustained four-chip step jitter reproducer

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:

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
 

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.

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()

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 <cstdint>
+#include <pto/pto-inst.hpp>
+
+#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<uint64_t>(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<float, Shape<1, 1, 1, kRows, kCols>, pto::Stride<1, 1, 1, kCols, 1>>;
+    using TileData = Tile<TileType::Vec, float, kRows, kCols, BLayout::RowMajor, -1, -1>;
+
+    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();
+    }
+}

examples/workers/l3/step_jitter_repro/kernels/orchestration/repeated_vector_add.cpp

@@ -0,0 +1,38 @@
+/*
+ * 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 <stddef.h>
+#include <stdint.h>
+
+#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"