CollectiveX: wire-basis bandwidth, nccl-ep LL hold, routed HT window, AsyncLL, DSv4-Pro workload

experimental/CollectiveX/README.md

@@ -26,8 +26,9 @@ in one of two modes:
 - `low-latency` uses each backend's decode-optimized kernel family: on DeepEP the legacy
   `deep_ep.Buffer` IBGDA `low_latency_dispatch`/`low_latency_combine` (a per-expert padded receive
   and a source-side gate-weighted combine). On UCCL-EP the same legacy `Buffer` low-latency kernels,
-  which at the scoped EP8 run `cudaIpc` over NVLink rather than its CPU-proxy transport. On MoRI the `IntraNodeLL` kernel (single-call,
-  pure-intranode, same compact layout and unweighted rank-sum combine as `IntraNode`). It is a
+  which at the scoped EP8 run `cudaIpc` over NVLink rather than its CPU-proxy transport. On MoRI the `AsyncLL` kernel driven split-phase
+  (send+recv inside each timed component — the kernel SGLang deploys for low-latency; same compact
+  layout and unweighted rank-sum combine as `IntraNode`). It is a
   decode-phase-only, per-SKU-capability-gated addition whose runnable set differs from `normal`'s, so
   it is enabled from each SKU's `ll_backends` registry entry (currently DeepEP V2 at EP8 on
   H100/H200 and at EP8 *and EP16* on B200 (the nscale bare-metal pool, whose gdrdrv-backed IBGDA
@@ -128,7 +129,7 @@ scale-up domain.
 | Backend | Engine availability | Current scope |
 |---|---|---|
 | DeepEP V2 | `production`, with vLLM `--all2all-backend deepep_v2`, SGLang `--moe-a2a-backend deepep` | `normal` mode is PR #605 `ElasticBuffer` plus exact upstream #630 and #640 fixes: LSA for scale-up and GIN for x86 EP16 scale-out. FP8 dispatch via `use_fp8_dispatch` (blockwise e4m3fn) alongside BF16. `low-latency` mode is the legacy `deep_ep.Buffer` IBGDA decode kernels (per-expert padded layout, weighted combine, `use_fp8` e4m3fn), decode only, with EP8 wherever enabled, plus EP16 on GB200/GB300 (inside the MNNVL domain) and on B200's nscale bare-metal pool (IBGDA over native IB rails with `/dev/gdrdrv`, although the prior virtualized b200 pool could never run it). B300 is an unsupported coverage row in `low-latency`: the legacy Buffer self-enables NVSHMEM IBGDA even for a single-node EP8 run, and on B300 address-handle creation fails (`ibgda.cpp:2234 Unable to create ah`), rc255 on all eight ranks. `NVSHMEM_DISABLE_IB=1` does not help. The Buffer re-enables IBGDA regardless, and the run fails identically with it set and unset (measured on b300-002 and b300-011) |
-| MoRI | `production`, with vLLM `--all2all-backend mori_*`, SGLang `--moe-a2a-backend mori` | `normal` mode uses the direct `IntraNode` kernel for scale-up EP8 on every CDNA SKU. EP16 remains an unsupported coverage row on all three CDNA SKUs. Part of the old ROCm/mori#475 corruption was this harness passing dispatch's returned recv-slot indices to `combine()` instead of the rank's own routing (root-caused upstream, guarded by ROCm/mori#546, kernels unchanged) — with the corrected call, single-shot InterNodeV1 is clean through T=512 on mi355x — but a residual stochastic corruption remains from T~128 up under repeated execution and is near-certain at prefill sizes (run 33045314017/33050026476; unaffected by per-pair drains, so not a buffer-reuse race; upstream cannot reproduce on ionic driver 26.03 vs our 25.11). The tw pairs additionally have no cross-node GPU fabric, and mi355x EP16 has no publishable transport today: uccl-ep's CPU-proxy RDMA is functional on Pollara but ~13x under its documented bandwidth (~6 GB/s vs 82; unchanged by registration mode or traffic class — ionic-driver suspect, same 25.11-vs-26.03 delta as the mori residual). `low-latency` mode selects the `IntraNodeLL` decode kernel (single-call, pure-intranode, same compact layout and unweighted combine as `IntraNode`), decode/EP8 only. FP8 dispatch is caller-prequantized (per-SKU e4m3fnuz on gfx942, e4m3fn on gfx950). Combine stays BF16 (`quant_type=none`) alongside BF16 dispatch |
+| MoRI | `production`, with vLLM `--all2all-backend mori_*`, SGLang `--moe-a2a-backend mori` | `normal` mode uses the direct `IntraNode` kernel for scale-up EP8 on every CDNA SKU. EP16 remains an unsupported coverage row on all three CDNA SKUs. Part of the old ROCm/mori#475 corruption was this harness passing dispatch's returned recv-slot indices to `combine()` instead of the rank's own routing (root-caused upstream, guarded by ROCm/mori#546, kernels unchanged) — with the corrected call, single-shot InterNodeV1 is clean through T=512 on mi355x — but a residual stochastic corruption remains from T~128 up under repeated execution and is near-certain at prefill sizes (run 33045314017/33050026476; unaffected by per-pair drains, so not a buffer-reuse race; upstream cannot reproduce on ionic driver 26.03 vs our 25.11). The tw pairs additionally have no cross-node GPU fabric, and mi355x EP16 has no publishable transport today: uccl-ep's CPU-proxy RDMA is functional on Pollara but ~13x under its documented bandwidth (~6 GB/s vs 82; unchanged by registration mode or traffic class — ionic-driver suspect, same 25.11-vs-26.03 delta as the mori residual). `low-latency` mode selects the `AsyncLL` decode kernel driven split-phase (dispatch send+recv, combine send+recv — the kernel SGLang deploys; same compact layout and unweighted combine as `IntraNode`; earlier rows measured `IntraNodeLL`, discriminated by `kernel_generation`), decode/EP8 only. FP8 dispatch is caller-prequantized (per-SKU e4m3fnuz on gfx942, e4m3fn on gfx950). Combine stays BF16 (`quant_type=none`) alongside BF16 dispatch |
 | UCCL-EP | `candidate` (no engine exposes a UCCL-EP selector) | [UCCL](https://github.com/uccl-project/uccl) EP: a drop-in, API-identical DeepEP replacement whose CPU proxies issue GPUDirect RDMA over plain `libibverbs` (no NVSHMEM/IBGDA), with software message ordering, atomics, and flow control. Scale-up is single-node `cudaIpc` over NVLink/XGMI (never MNNVL). `normal` mode is the legacy `Buffer` `dispatch`/`combine` (unweighted rank-sum). `low-latency` reuses the legacy `low_latency_dispatch`/`low_latency_combine` decode kernels (weighted combine), decode/EP8 only. FP8 dispatch is caller-prequantized in `normal` mode (blockwise e4m3fn, per-SKU e4m3fnuz on gfx942). In `low-latency` mode the caller sends BF16 and the decode kernel quantizes to e4m3 internally (`use_fp8`). Combine is BF16. Runs on NVIDIA and AMD (H100/H200/B200 + MI300X/MI325X/MI355X), EP8 scale-up. Cross-node EP16 is functional (the internode RDMA path connects and the light case passes correctness) but its CPU-proxy throughput overruns the standardized per-case wall-clock budget on heavy token counts, so EP16 is an unsupported coverage row for now |
 | NCCL EP | `candidate` (NVIDIA's own library, but no engine exposes an NCCL-EP selector) | [NCCL EP](https://github.com/NVIDIA/nccl/tree/master/contrib/nccl_ep): NVIDIA's native MoE dispatch/combine on the NCCL Device API, using LSA (NVLink load/store) intra-node and GIN (GPU-Initiated Networking) inter-node, driven through the `nccl4py` bindings. `normal` mode selects the `HIGH_THROUGHPUT` algorithm (FLAT `[N, hidden]` receive, unweighted rank-sum combine). The `LOW_LATENCY` algorithm carries an EP8 `ll_backends` row on all six NVIDIA SKUs, restored once the single-handle fix removed the NVIDIA/nccl#2303 signal aliasing. That LL decode ladder is clamped to T<=128, below its 256-slot receive: `nccl_ep`'s combine recv pipeline is a port of DeepEP's pre-#642 kernel and is missing the same shared-memory fence before `mbarrier_arrive`, which corrupted T=256 on GB300 in 1 of 5 executions. It was bimodal, with healthy rows at 0.0039 relative error against 0.4704 on the failure. The fence is absent at NVIDIA/nccl master, so it is unfixed upstream. The clamp lowers exposure and is **not** a safety boundary: the fence is missing on every combine recv and T=256 is merely the rung with the most pipeline iterations, so lower rungs are less likely to hit the race rather than immune. Restore when a fixed wheel ships BF16 only: `contrib/nccl_ep/RELEASE.md` says "No FP8 support", so no FP8 case is emitted. That note is worth re-testing rather than trusting, because the C library at our pinned commit does read `inputs->scales` and switch on e4m3/e5m2, the two documented FP8 exclusions are expert-major layouts we do not use, and `NVIDIA/nccl` has not moved since 2026-06-11 while `NVIDIA/nccl-extensions` has replaced that row outright. NVIDIA-only and CUDA 13 only. EP8 scale-up on H100/H200/B200/B300 plus EP8 and EP16 on GB200/GB300, where EP16 stays inside the MNNVL scale-up domain. x86 EP16 scale-out is an unsupported coverage row: the cross-node GIN path faults inside `nccl_ep.cc` identically on RoCE and IB across four SKUs, a GDAKI limit rather than a fabric-selection one |
 | FlashInfer EP | `production`, with vLLM `--all2all-backend flashinfer_nvlink_one_sided` | [FlashInfer](https://github.com/flashinfer-ai/flashinfer) `MoeAlltoAll`: TensorRT-LLM's one-sided MNNVL all-to-all, where each rank writes tokens straight into its peers' workspace windows and combine reads them back, with no send/recv pairing and no NVSHMEM. `normal` mode only (there is one kernel family and no separate decode path), and GB200/GB300 only, since the transport is MNNVL. FP8 dispatch is caller-prequantized blockwise e4m3fn, carried as a fourth dispatch payload alongside its per-128-block FP32 scales, with the combine plane forced to BF16. The C++ `toNvDataType` accepts only fp16/bf16/fp32 for combine, so an FP8 combine buffer would raise rather than corrupt. EP8 and EP16, both inside the scale-up domain. Unlike every other backend here, its combine accumulates in the PAYLOAD dtype rather than FP32: wheels before 0.6.16 reduce the top-k contributions with a pairwise BF16 tree that rounds at every level, so the oracle models that reduction directly (`combine_reduction = "topk-slot-tree"`) instead of widening the tolerance. 0.6.16 moved the accumulator to FP32, and the adapter switches models on the installed version |

experimental/CollectiveX/bandwidth.py

@@ -8,15 +8,16 @@
   B. alpha/beta fit -- OLS of p50 latency vs bytes, latency ~= alpha + bytes/beta, which
      separates the bandwidth term from the per-call floor that dominates small T.
 
-Bytes are LOGICAL payload (one copy per unique (token, dest-rank) pair), excluding
-protocol/padding. Reading them against a link's peak needs two adjustments: logical bytes
-include the copies a rank routes to itself, which never cross the interconnect, so wire
-traffic is (1 - routing.locality.local_rank_fraction) x these figures; and beta is a
-MARGINAL rate, so it legitimately sits above every measured point.
-
-This basis is goodput; low-latency layouts send one copy per (token, expert), so their wire
-traffic is topk/fanout higher (~1.5x at EP8) and their GB/s is not comparable to a normal row
-or to a vendor table counting every expert copy.
+Bytes are the WIRE basis (`wire_byte_provenance`): one copy per unique (token, dest-rank)
+pair for rank-deduplicating layouts, one copy per (token, expert) for the low-latency
+kernels that do not deduplicate — whichever the row's `logical_copies.wire` declares. That
+makes GB/s comparable across backends and modes. Two adjustments still apply when reading
+against a link's peak: these bytes include the copies a rank routes to itself, which never
+cross the interconnect, so fabric traffic is (1 - routing.locality.local_rank_fraction) x
+these figures; and beta is a MARGINAL rate, so it legitimately sits above every measured
+point. Pre-wire artifacts fall back to the rank-deduplicated `byte_provenance`, which for
+per-assignment LL rows understates the wire rate (a lower bound, ~34% low on nccl-ep LL
+EP8 at T=128), never overstates it.
 
 A line through a ladder is a MODEL and a positive slope alone is not evidence, so every Fit
 carries its own quality and beta is withheld unless it clears both gates below. Deliberately
@@ -108,8 +109,7 @@ def fit_alpha_beta(document: dict, component: str, pct: str = "p50") -> Fit | No
         if not row.get("correctness", {}).get("passed", True):
             excluded += 1
             continue
-        usable.append((row["byte_provenance"][component]["total_logical_bytes"],
-                       percentiles[pct]))
+        usable.append((_wire_bytes(row, component), percentiles[pct]))
     if len(usable) < MIN_FIT_POINTS or len({x for x, _ in usable}) < 2:
         return None  # too few points, or zero-variance x (e.g. a degenerate ladder)
     xs, ys = [x for x, _ in usable], [y for _, y in usable]
@@ -127,6 +127,18 @@ def fit_alpha_beta(document: dict, component: str, pct: str = "p50") -> Fit | No
     )
 
 
+def _wire_bytes(row: dict, component: str) -> int:
+    """Bytes the kernels actually move for this component.
+
+    Prefers `wire_byte_provenance` (per-assignment for token-expert LL receives); artifacts
+    written before that field carry only the rank-deduplicated `byte_provenance`, which for
+    those LL rows is a LOWER BOUND on wire traffic (~topk/fanout below it), so a bandwidth
+    derived from the fallback understates the wire rate rather than overstating it.
+    """
+    provenance = row.get("wire_byte_provenance") or row["byte_provenance"]
+    return provenance[component]["total_logical_bytes"]
+
+
 def _format_fit(component: str, fit: Fit) -> str:
     """One component's fit, withholding what the data does not support."""
     if not fit.beta_is_reliable:
@@ -142,7 +154,7 @@ def _cell(row: dict, component: str, ep: int) -> str:
     percentiles = row["components"][component]["percentiles_us"]
     if not percentiles:
         return f"{component}=n/a"
-    nbytes = row["byte_provenance"][component]["total_logical_bytes"]
+    nbytes = _wire_bytes(row, component)
     p50 = _algbw_per_gpu(nbytes, percentiles["p50"], ep)
     p99 = _algbw_per_gpu(nbytes, percentiles["p99"], ep)
     return f"{component}=n/a" if p50 is None or p99 is None \
@@ -170,7 +182,7 @@ def _provenance(document: dict) -> str:
 
 def render(documents: list[dict]) -> str:
     lines = [
-        "## CollectiveX EP bandwidth (per-GPU, logical payload)",
+        "## CollectiveX EP bandwidth (per-GPU, wire-basis payload)",
         "",
         "GB/s at p50/p99 *latency* -- the p99-latency figure is worst-case bandwidth, not a "
         f"'p99 bandwidth'. `fit`: latency ~= alpha + bytes/beta over the ladder; beta is "

experimental/CollectiveX/bench/ep_harness.py

@@ -1344,7 +1344,7 @@ def run_sweep(args, backend, torch, dist, device, rank: int, world_size: int) ->
         )
         combine_bytes = logical_byte_provenance(rstats["routed_copies"], args.hidden)
         # Second byte basis, for backends whose wire carries one copy per (token, expert). Which
-        # applies is a property of the RECEIVE, not the mode -- MoRI's IntraNodeLL deduplicates
+        # applies is a property of the RECEIVE, not the mode -- MoRI's LL kernels deduplicate
         # where the other low-latency kernels do not -- so key it on the declared
         # receive layout. `routed_copies` stays the canonical comparable basis.
         assignment_copies = int(sum(rstats["expert_assignments_per_rank"]))
@@ -1357,6 +1357,25 @@ def run_sweep(args, backend, torch, dist, device, rank: int, world_size: int) ->
             field: dispatch_bytes[field] + combine_bytes[field] for field in dispatch_bytes
         }
         stage_bytes = dict.fromkeys(dispatch_bytes, 0)
+        # WIRE bytes: what the kernels actually move, on the basis `wire_basis` declares.
+        # For token-expert receives this is the per-assignment count (topk/fanout above the
+        # rank-deduplicated basis, +34% observed on nccl-ep LL EP8 at T=128); for token-rank
+        # receives it equals the canonical figures. A bandwidth divided from `byte_provenance`
+        # on a token-expert backend is a LOWER BOUND, not the wire rate, and is not comparable
+        # across backends -- consumers computing GB/s must divide from THESE bytes.
+        wire_copies = (
+            assignment_copies if wire_basis == "per-assignment"
+            else int(rstats["routed_copies"])
+        )
+        wire_dispatch_bytes = logical_byte_provenance(
+            wire_copies, args.hidden,
+            backend.dispatch_value_bytes, backend.dispatch_scale_bytes_per_copy,
+        )
+        wire_combine_bytes = logical_byte_provenance(wire_copies, args.hidden)
+        wire_roundtrip_bytes = {
+            field: wire_dispatch_bytes[field] + wire_combine_bytes[field]
+            for field in wire_dispatch_bytes
+        }
         spread = samples[T].spread
         chain = samples[T].chain
         chain_spread = samples[T].chain_spread
@@ -1441,6 +1460,15 @@ def run_sweep(args, backend, torch, dist, device, rank: int, world_size: int) ->
                 "roundtrip": roundtrip_bytes,
                 "stage": stage_bytes,
             },
+            # Same fields on the wire basis (`logical_copies.wire`). Identical to
+            # `byte_provenance` for token-rank receives; per-assignment for the LL kernels
+            # that move one copy per (token, expert). Bandwidth = wire bytes / latency.
+            "wire_byte_provenance": {
+                "combine": wire_combine_bytes,
+                "dispatch": wire_dispatch_bytes,
+                "roundtrip": wire_roundtrip_bytes,
+                "stage": stage_bytes,
+            },
             # Copy counts behind the byte figures above, so a reader can rebase them: `routed` is
             # the basis they use, `assignments` the per-(token, expert) count, `wire` which the
             # kernels move. Kept out of `byte_provenance`, whose values are all per-component.