Fix: group bind phases correctly, and give the dsv4 driver the observability calls it omitted

docs/dfx/hbg-bind-phases.md

@@ -47,9 +47,25 @@ predates the change still totals correctly, and names them under its `total` row
 as absent from every bind. The table above is what a current run emits: ten
 segments, three of them control plane.
 
-**The control plane is a sum of costs, not an interval.** `arena_h2d` runs
-*after* `host_view_close`, so the segments do not form one contiguous window.
-Sum the ones the bind has; do not subtract two timestamps.
+**The control plane is a sum of costs, not an interval.** Its three segments are
+not adjacent: `static_arena`, `shared_mem` and `gm_heap` run between
+`graph_upload` and `arena_h2d`, so the segments do not form one contiguous
+window. Sum the ones the bind has; do not subtract two timestamps.
+
+**A bind runs its segments in one order and prints them in another**, and the
+two are easy to confuse because only the second is visible in a log.
+
+| order | segments |
+| ----- | -------- |
+| **execution** — the sequence `runtime_maker.cpp` calls them in, and what `start_ns` shows | `args`, `arena_build`, `runtime_init`, `host_orch`, `graph_upload`, `static_arena`, `shared_mem`, `gm_heap`, `arena_h2d`, `host_view_close` |
+| **emission** — `HostPhaseKind` order, printed as one burst when the bind ends | `args`, `arena_build`, `static_arena`, `gm_heap`, `shared_mem`, `runtime_init`, `host_orch`, `graph_upload`, `arena_h2d`, `host_view_close` |
+
+The execution order is what puts `static_arena`, `shared_mem` and `gm_heap`
+between `graph_upload` and `arena_h2d`, which is why the control plane is a sum
+and not an interval. The emission order is what a line-by-line reader of the log
+sees; `host_view_close` is last in both, and `arena_h2d` second to last, so
+reading `arena_h2d` as the segment that closes a bind shifts every
+`host_view_close` into the following bind.
 
 ## Prerequisites
 
@@ -71,8 +87,8 @@ device lock for the whole job (see
 | Path | `examples/a2a3/host_build_graph/qwen3_14b_decode/` | `examples/a2a3/host_build_graph/deepseek_v4_flash_decode/` |
 | Entry point | standalone `main.py`, which owns its L2 `Worker` | standalone `main.py`, which owns its L3 `Worker` |
 | Devices | 1 | 2 (EP2/TP2) |
-| Host tasks | 47 | 1131 |
-| Graph replays | 40, of a 277-task Definition | 20, of a 743-task Definition |
+| Host tasks (`host_orch tasks=`) | 47 | 129 |
+| Graph submissions (`graph_upload submissions=`/`defs=`) | 40, of 1 Definition | 86, of 8 Definitions |
 | Graph boundary | 26 tensors | 118 tensors, 31 scalars |
 | First-run compile | seconds | **minutes** (369 kernel sources + an 11.6k-line orchestration) |
 | Parameters | device memory; valid fixture streamed once before all rounds | child memory, and `--skip-golden` leaves it uninitialized |
@@ -81,6 +97,13 @@ device lock for the whole job (see
 The entry point decides how a case's output is captured, which is what the recipe
 below has to work around.
 
+**The two count rows name the markers they come from, because they are properties
+of the cases and the cases get edited.** They read 47 / 129 tasks and 40 / 86
+submissions on `4d31f482`; they previously read 1131 tasks and 20 replays for
+DeepSeek-V4, from before its orchestration moved most task submission onto the
+recording threads. Re-read them from a current log rather than trusting this
+table — a bind's `host_orch` and `graph_upload` lines carry both.
+
 ## Recipe A — stable numbers, many rounds
 
 The ready-made invocation for either case lives in the
@@ -169,18 +192,18 @@ grep -oE 'bind phase=[a-z0-9_]+ start_ns=[0-9]+ dur_ns=[0-9]+[^[]*' outputs/hbg_
 ```
 
 The character class has to admit digits. `[a-z_]+` matches no segment whose name
-carries one, so it silently drops every `arena_h2d` line — the bind-closing
-segment, the only H2D left, and the one that itemizes the whole upload. On a
-two-bind log that is 18 lines where 20 exist, with nothing to say a segment went
-missing.
+carries one, so it silently drops every `arena_h2d` line — the only H2D
+left, and the one that itemizes the whole upload. On a two-bind log that is 18
+lines where 20 exist, with nothing to say a segment went missing.
 
 Each line carries `start_ns` (a `CLOCK_MONOTONIC` timestamp) plus the segment's
 own attributes — `tasks=` and `heap_used=` on `host_orch`, `defs=`, `bytes=`,
 `submissions=` and `spilled=` on `graph_upload`, and `arena_h2d`'s itemized upload.
 Group the
-lines into binds — `arena_h2d` is the last segment of a bind, so it closes one —
-then sum the control-plane segments **within each bind** and take the minimum of
-those sums. Never sum
+lines into binds — a bind prints each segment it has once, so a segment name you
+have already seen is the first line of the next bind; do **not** close on
+`arena_h2d`, which is second to last — then sum the control-plane segments
+**within each bind** and take the minimum of those sums. Never sum
 minima taken across binds; that total belongs to no bind and can point the wrong
 way (see below).
 

examples/a2a3/tensormap_and_ringbuffer/deepseek_v4_flash_decode/main.py

@@ -88,7 +88,9 @@
     build_output_prefix,
     compile_chip_callable_spec,
     effective_diagnostic_options,
+    finalize_diagnostic_outputs,
     l3_compile_cache_key,
+    log_torch_backend_autoload_once,
 )
 
 HERE = Path(__file__).resolve().parent
@@ -861,6 +863,7 @@ def run(  # noqa: PLR0913 -- one knob per CLI flag
             enable_scope_stats=diagnostics.scope_stats,
             output_prefix=output_prefix,
         )
+        log_torch_backend_autoload_once()
         for round_idx in range(rounds):
             print(f"[dsv4] round {round_idx + 1}/{rounds}", flush=True)
             keepalive: list = []
@@ -871,6 +874,14 @@ def task_orch(orch, _args, _cfg, _keep=keepalive):
             worker.run(task_orch)
     finally:
         worker.close()
+        if output_prefix:
+            finalize_diagnostic_outputs(
+                f"{CASE_LABEL}_{runtime}",
+                output_prefix,
+                callable_spec=spec,
+                dep_gen=diagnostics.dep_gen,
+                scope_stats=diagnostics.scope_stats,
+            )
     print("[dsv4] PASSED", flush=True)
     return 0
 

simpler_setup/tools/README.md

@@ -404,10 +404,19 @@ warm-up pass **per rank** rather than one in total; `--ranks` sets that directly
 and `--keep-first` keeps the cold passes. A run whose every pass is a rank's
 warm-up — `--rounds 1` — is refused rather than reported, since the one number it
 could print is the cold one. Three grouping rules are encoded rather than left to
-the caller, because each silently produces a wrong number: `arena_h2d` closes a
-pass (the segments are not contiguous in time, so timestamp order does not group
-them), the control-plane total is summed **within** a pass before any minimum is
-taken, and the first pass of each rank is warm-up.
+the caller, because each silently produces a wrong number: a repeated segment
+name opens the next pass (the segments are not contiguous in time, so timestamp
+order does not group them, and no single segment reliably closes a pass — reading
+`arena_h2d` as the closing one shifted every `host_view_close` by a pass), the
+control-plane total is summed **within** a pass before any minimum is taken, and
+the first pass of each rank is warm-up.
+
+Grouping on the repeat assumes each pass prints its segments as an uninterrupted
+burst. That holds on every log measured so far, but nothing enforces it: ranks
+share one stream and the line prefix carries no pid, and the thread id it does
+carry is identical across ranks. A burst split by another rank's is therefore
+reported as a pass whose segment set differs from its neighbours', which the
+non-uniform-segment warning names rather than passing off as a number.
 
 A run whose control plane is missing a phase entirely — a change can retire one —
 is still totalled, over the phases it has, with the absent ones named. A phase

simpler_setup/tools/hbg_bind_phases.py

@@ -54,13 +54,32 @@
     "host_view_close",
 )
 
-# The last segment of a bind, and so what closes one: the segments are not
-# contiguous in time, so timestamp order does not group them.
-BIND_CLOSING_PHASE = "arena_h2d"
-
 
 def parse_binds(path: str) -> list[dict[str, float]]:
-    """Group `bind phase=` lines into binds, in milliseconds."""
+    """Group `bind phase=` lines into binds, in milliseconds.
+
+    A bind emits every segment it has once, so a repeated segment name is the
+    first line of the next bind. Grouping on the repeat rather than on a named
+    closing segment survives a bind that omits a segment, a change to the
+    emission order, and a new segment being added; the segments are not
+    contiguous in time, so timestamp order does not group them.
+
+    This assumes each bind's burst reaches the log uninterrupted, and **nothing
+    enforces that**. Ranks write one stream through no lock, the line prefix
+    carries no pid, and its thread id is identical across ranks — measured as one
+    value over all 400 bind lines of a two-rank run — so there is no field to
+    group by instead.
+
+    It also assumes no bind omits a segment its successor emits *before* any they
+    share, which would put that segment in the earlier bind. `args` is emitted
+    unconditionally and first, so today nothing can precede a shared segment.
+
+    Both assumptions fail the same visible way — a bind whose segment set differs
+    from its neighbours' — which `warn_on_ragged_binds` names. That is why the
+    boundary stays free of any knowledge about segment order: an order constant
+    gone stale would split every bind at the same point, leaving the sets uniform
+    and the mis-grouping undetectable.
+    """
     binds: list[dict[str, float]] = []
     current: dict[str, float] = {}
     with open(path, encoding="utf-8", errors="replace") as handle:
@@ -69,10 +88,10 @@ def parse_binds(path: str) -> list[dict[str, float]]:
             if match is None:
                 continue
             phase, _start_ns, dur_ns = match.group(1), int(match.group(2)), int(match.group(3))
-            current[phase] = dur_ns / 1e6
-            if phase == BIND_CLOSING_PHASE:
+            if phase in current:
                 binds.append(current)
                 current = {}
+            current[phase] = dur_ns / 1e6
     if current:
         binds.append(current)
     # A group without `host_orch` is not a bind.
@@ -110,6 +129,26 @@ def spread(values: list[float]) -> tuple[float, float, float]:
     return min(values), statistics.median(values), max(values)
 
 
+def warn_on_ragged_binds(binds: list[dict[str, float]]) -> None:
+    """Report binds whose segment set differs from their neighbours'.
+
+    `parse_binds` assumes each bind's segments reach the log as an uninterrupted
+    burst, and nothing enforces that. A rank whose burst was split by another
+    rank's, and a truncated log, both leave a bind short of segments its
+    neighbours have; either way the rows below are not all describing the same
+    thing, so name it rather than print a clean table over it.
+    """
+    everywhere = {k for k in binds[0] if all(k in b for b in binds)}
+    ragged = sorted({k for b in binds for k in b} - everywhere)
+    if not ragged:
+        return
+    verb = "is" if len(ragged) == 1 else "are"
+    print(f"\n  WARNING: not every bind has the same segments; {', '.join(ragged)}")
+    print(f"  {verb} missing from some. A bind is grouped by its first repeated segment")
+    print("  name, so a rank whose burst was split by another rank's, and a truncated")
+    print("  log, both land here -- read the counts column before quoting a duration.")
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
     parser.add_argument("log", help="run log carrying the `bind phase=` lines")
@@ -180,6 +219,8 @@ def main() -> int:
         print(f"\n  phases this tool does not know about: {', '.join(unknown)}")
         print("  (add them to PHASE_ORDER, and to CONTROL_PLANE if a dispatch change can move them)")
 
+    warn_on_ragged_binds(warm)
+
     # The control-plane set is not fixed: a change can retire a phase outright, so
     # the total covers the phases this run has and names the ones absent from every
     # bind. Absent from only some binds is a truncated log rather than a retired

tests/ut/py/test_hbg_bind_phases_grouping.py

@@ -0,0 +1,133 @@
+# 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.
+# -----------------------------------------------------------------------------------------------------------
+"""Grouping of `bind phase=` lines into binds."""
+
+from __future__ import annotations
+
+import sys
+
+from simpler_setup.tools import hbg_bind_phases
+
+# A bind emits its segments in one contiguous burst, in this order. `arena_h2d`
+# is second to last, so it does not close a bind; `host_view_close` does.
+EMISSION_ORDER = (
+    "args",
+    "arena_build",
+    "static_arena",
+    "gm_heap",
+    "shared_mem",
+    "runtime_init",
+    "host_orch",
+    "graph_upload",
+    "arena_h2d",
+    "host_view_close",
+)
+
+
+def _write_binds(path, count: int, *, order=EMISSION_ORDER, omit: tuple[str, ...] = ()) -> None:
+    """One line per segment per bind, each duration encoding its own bind index."""
+    lines = ["[stamp] command commit=abc"]
+    clock = 0
+    for bind_index in range(count):
+        for phase in order:
+            if phase in omit:
+                continue
+            clock += 1
+            # dur_ns = (bind_index + 1) ms, so a misattributed segment is visible.
+            lines.append(f"bind phase={phase} start_ns={clock} dur_ns={(bind_index + 1) * 1000000}")
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def test_each_segment_lands_in_its_own_bind(tmp_path):
+    log = tmp_path / "run.log"
+    _write_binds(log, 3)
+
+    binds = hbg_bind_phases.parse_binds(str(log))
+
+    assert len(binds) == 3
+    for index, bind in enumerate(binds):
+        assert sorted(bind) == sorted(EMISSION_ORDER), f"bind {index} is not whole"
+        for phase, duration in bind.items():
+            assert duration == index + 1, f"bind {index} carries another bind's {phase}"
+
+
+def test_a_bind_missing_its_closing_segment_does_not_swallow_the_next(tmp_path):
+    """An interrupted bind stays one bind rather than merging with its successor."""
+    log = tmp_path / "partial.log"
+    _write_binds(log, 1)
+    with log.open("a", encoding="utf-8") as handle:
+        for phase in ("args", "host_orch", "arena_h2d"):
+            handle.write(f"bind phase={phase} start_ns=999 dur_ns=2000000\n")
+
+    binds = hbg_bind_phases.parse_binds(str(log))
+
+    assert len(binds) == 2
+    assert binds[0]["host_view_close"] == 1
+    assert binds[1]["host_orch"] == 2
+    assert "host_view_close" not in binds[1]
+
+
+def test_a_bind_omitting_a_leading_segment_is_reported(tmp_path, monkeypatch, capsys):
+    """A bind short of a leading segment absorbs the next bind's copy of it.
+
+    Grouping closes on a repeated name, so a segment the previous bind lacks and
+    the next one emits before any they share lands in the previous bind. `args`
+    is emitted unconditionally and first, so no current bind can omit it — this
+    pins what happens if that ever changes, and that the reader is told.
+    """
+    log = tmp_path / "no_args.log"
+    _write_binds(log, 1, order=EMISSION_ORDER[1:])
+    with log.open("a", encoding="utf-8") as handle:
+        for phase in EMISSION_ORDER:
+            handle.write(f"bind phase={phase} start_ns=999 dur_ns=2000000\n")
+
+    binds = hbg_bind_phases.parse_binds(str(log))
+
+    assert [len(bind) for bind in binds] == [10, 9]
+    assert binds[0]["args"] == 2, "the second bind's args landed in the first"
+    assert "args" not in binds[1]
+
+    monkeypatch.setattr(sys, "argv", ["hbg_bind_phases", str(log), "--keep-first"])
+    assert hbg_bind_phases.main() == 0
+    out = capsys.readouterr().out
+    assert "not every bind has the same segments; args" in out
+
+
+def test_a_split_burst_is_reported_rather_than_passed_off_as_a_bind(tmp_path, monkeypatch, capsys):
+    """Grouping assumes uninterrupted bursts, and nothing enforces that.
+
+    Ranks share one stream through no lock and the line prefix carries no pid, so
+    one rank's burst can land inside another's. The result is binds whose segment
+    sets differ, which has to reach the reader.
+    """
+    log = tmp_path / "split_burst.log"
+    lines = ["[stamp] command commit=abc"]
+
+    def emit(phase: str, rank: int) -> None:
+        lines.append(f"bind phase={phase} start_ns={len(lines)} dur_ns={rank * 1000000}")
+
+    for phase in EMISSION_ORDER[:4]:  # rank 1 starts
+        emit(phase, 1)
+    for phase in EMISSION_ORDER:  # rank 2 lands whole, inside it
+        emit(phase, 2)
+    for phase in EMISSION_ORDER[4:]:  # rank 1 finishes
+        emit(phase, 1)
+    log.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    binds = hbg_bind_phases.parse_binds(str(log))
+    assert [len(bind) for bind in binds] == [10, 6], "the split burst is not silently made whole"
+
+    # --keep-first so both binds are reported; dropping the cold one would leave a
+    # single bind, whose segment set is trivially uniform with itself.
+    monkeypatch.setattr(sys, "argv", ["hbg_bind_phases", str(log), "--keep-first"])
+    assert hbg_bind_phases.main() == 0
+    out = capsys.readouterr().out
+    assert "not every bind has the same segments" in out
+    for phase in ("args", "arena_build", "static_arena", "gm_heap"):
+        assert phase in out