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

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.

experimental/CollectiveX/bench/ep_mori.py

@@ -86,29 +86,30 @@ def __init__(self, args, rank, world_size, local_rank, device):
             else ("IntraNode", "intranode", (80, 0, 16, 16))
         )
         if self.mode == "low-latency":
-            # LOW-LATENCY (decode) mode: IntraNodeLL, the scale-up low-latency kernel. It is
-            # single-phase (plain dispatch()/combine(), no dispatch_recv/combine_recv split),
-            # pure-intranode (shares the no-RDMA ShmemBufsIntraNode staging with IntraNode, so
-            # no symmetric-heap registration), and returns the same compact [max_recv, hidden]
-            # layout. Its combine keeps the plain rank-deduplicated additive sum (combine is
-            # called with weights=None -> weight_ptr 0 in mori.ops, so the gate is NOT applied
-            # in-kernel), identical in semantics to IntraNode/normal mode ("unweighted-rank-sum",
-            # the base default the harness admits for low-latency) over the same compact
-            # rank-deduplicated receive (the base "token-rank" receive_layout, so the
-            # artifact's wire basis stays rank-deduplicated). So LL differs from the normal
-            # IntraNode path ONLY by kernel_type (set here vs omitted) and timing; every transport
-            # method (dispatch/stage/combine/inspect_dispatch/combine_transformed) is reused as-is.
-            # AsyncLL (enum 4) is deliberately NOT used: it is split-phase (dispatch_recv/
-            # combine_recv) and RDMA-staged, which does not fit the single-call dispatch/stage/
-            # combine contract. Scale-up EP8 decode only; scale-out EP16 LL is out of scope (kept
-            # out of ll_backends). Reuse the IntraNode launch tuning under MANUAL launch mode.
+            # LOW-LATENCY (decode) mode: AsyncLL, because that is what production selects —
+            # SGLang's moriep dispatcher maps EpMode.LOW_LATENCY to AsyncLL (split-phase
+            # dispatch_send/dispatch_recv + combine_send/combine_recv) with block_num 64,
+            # rdma_block_num 32, warp_num_per_block 8, and its low-latency impl asserts the
+            # kernel IS AsyncLL. The previous adapter measured IntraNodeLL, a kernel no engine
+            # deploys for LL, so those rows described an off-production path (their
+            # kernel_generation "intranode-ll" is the discriminator). AsyncLL must be driven
+            # split-phase: mori.ops' dispatch()/combine() under this kernel_type launch only
+            # the SEND side and return without the payload landing (the earlier "fails
+            # silently single-call" finding), so the timed dispatch here is send+recv
+            # back-to-back — the full transport the serving step pays, merely without the
+            # expert GEMM interleaved between the phases.
+            # Combine keeps weights=None (gate not applied in-kernel), and the receive is the
+            # same compact rank-deduplicated [max_recv, hidden] layout, so semantics stay
+            # "unweighted-rank-sum" over "token-rank" and the wire basis stays
+            # rank-deduplicated. Scale-up EP8 decode only; scale-out EP16 LL stays out of
+            # ll_backends.
             if scale_out:
                 raise RuntimeError(
                     "MoRI low-latency is scale-up EP8 only (scale-out EP16 low-latency "
                     "is out of scope; see platform_config ll_backends)"
                 )
             kernel_name, self.kernel_generation, blocks = (
-                "IntraNodeLL", "intranode-ll", (80, 0, 16, 16)
+                "AsyncLL", "async-ll", (64, 32, 8, 8)
             )
         self._kernel_type = None
         if kernel_name != "IntraNode":
@@ -118,6 +119,9 @@ def __init__(self, args, rank, world_size, local_rank, device):
                     f"this MoRI image lacks EpDispatchCombineKernelType.{kernel_name}"
                 )
             self._kernel_type = getattr(kernel_enum, kernel_name)
+        # AsyncLL's dispatch()/combine() launch only the SEND kernels; the payload lands in
+        # the recv phase, so both timed components must run send+recv back-to-back.
+        self._split_phase = kernel_name == "AsyncLL"
         self._inter_node = kernel_name == "InterNodeV1"
         self.num_qps = 1
         self.block_num, self.rdma_block_num, self.dispatch_warps, self.combine_warps = blocks
@@ -227,6 +231,16 @@ def create_buffer(self, spec):
         self.op = mori.ops.EpDispatchCombineOp(self.config)
         if getattr(self.op, "launch_config_mode", None) != "MANUAL":
             raise RuntimeError("MoRI explicit launch configuration was not applied")
+        # Fail closed rather than measure a silent send-only AsyncLL: without the recv
+        # methods the single-call dispatch returns before any payload lands.
+        if self._split_phase and not (
+            callable(getattr(self.op, "dispatch_recv", None))
+            and callable(getattr(self.op, "combine_recv", None))
+        ):
+            raise RuntimeError(
+                "this MoRI image's AsyncLL lacks the split-phase "
+                "dispatch_recv/combine_recv API the deployed low-latency path drives"
+            )
 
     def semantic_payload(self, x):
         if not self._fp8:
@@ -265,6 +279,10 @@ def dispatch(self, p):
                 warp_per_block=self.dispatch_warps,
             )
         )
+        if self._split_phase:
+            # AsyncLL: the call above launched only the send kernels; production pays the
+            # recv phase in the same step (SGLang's dispatch_b), so the timed window does too.
+            self.op.dispatch_recv()
         return types.SimpleNamespace(
             dispatch_output=dispatch_output,
             dispatch_weights=dispatch_weights,
@@ -302,6 +320,9 @@ def combine(self, p, h):
             rdma_block_num=self.rdma_block_num,
             warp_per_block=self.combine_warps,
         )
+        if self._split_phase:
+            # AsyncLL combine is send-only until the recv phase lands the reduction.
+            self.op.combine_recv()
         return combined[:p.T]
 
     def inspect_dispatch(self, p, h):

experimental/CollectiveX/bench/ep_nccl.py

@@ -95,7 +95,11 @@ class NCCLEPBackend(EPBackend):
     # semantics are switched to their LL values in __init__ (mirrors ep_deepep_v2).
     #   normal      -> HT / FLAT layout / unweighted-rank-sum combine.
     #   low-latency -> LL / EXPERT_MAJOR layout / source-side weighted-kernel-sum combine.
-    kernel_generation = "nccl-ep-ht"
+    # "-routed" marks the generation whose timed HT dispatch charges the per-step
+    # ncclEpUpdateHandle (see dispatch()); earlier "nccl-ep-ht" rows excluded it and the
+    # docs record that those cannot be separated by any other field — this suffix is the
+    # per-row discriminator that change lacked.
+    kernel_generation = "nccl-ep-ht-routed"
     SUPPORTED_MODES = ("normal", "low-latency")
     SUPPORTED_PRECISIONS = ("bf16",)
     stage_device_work = False
@@ -385,6 +389,17 @@ def _rebind(self, h):
     def dispatch(self, p):
         h = self._ensure_handle(p)
         stream = self._stream()
+        if not self._ll:
+            # Charge the per-step routing collective to the window. `ncclEpUpdateHandle` is
+            # documented as a "per-step collective: prepare the handle for the given top-k
+            # routing decisions", and production routing changes every MoE layer — a serving
+            # step pays this update before every HT dispatch, at the handle's full token
+            # capacity, exactly as here. Excluding it (as NVIDIA's ep_bench does) made HT
+            # dispatch the one window that omitted its routing/layout work while deepep-v2,
+            # uccl-ep, MoRI and FlashInfer all carry theirs per call. No sync or counter
+            # read here: the bound problem's counters are deterministic and already read
+            # (_bind_ht_recv_count) in the untimed rebind.
+            h.handle.update(h.topk_idx_t, layout_info=h.layout_info, stream=stream)
         if self._ll:
             # LL EXPERT_MAJOR: tokens in, 3D per-expert padded tokens out, per-expert recv
             # counts written into expert_counters. No weights on the dispatch (the gate is

experimental/CollectiveX/configs/platform_config.json

@@ -11,7 +11,7 @@
       "scale_up_transport": "nvlink",
       "launcher": "single-slurm",
       "backends": {"deepep-v2": [8, 16], "uccl-ep": [8], "nccl-ep": [8]},
-      "ll_backends": {"deepep-v2": [8], "uccl-ep": [8], "nccl-ep": [8]},
+      "ll_backends": {"deepep-v2": [8], "uccl-ep": [8]},
       "fabric": {"nic": "ConnectX-7 2x200GbE", "switch": "Arista 7060DX5-64S (Tomahawk4, 25.6T)"},
       "operator": {
         "partition": "hpc-gpu-1",
@@ -35,7 +35,7 @@
       "scale_up_transport": "nvlink",
       "launcher": "single-slurm",
       "backends": {"deepep-v2": [8, 16], "uccl-ep": [8], "nccl-ep": [8]},
-      "ll_backends": {"deepep-v2": [8], "uccl-ep": [8], "nccl-ep": [8]},
+      "ll_backends": {"deepep-v2": [8], "uccl-ep": [8]},
       "fabric": {"nic": "ConnectX-7 400G", "switch": "NVIDIA Quantum-2 QM9790 (25.6T, InfiniBand)"},
       "operator": {
         "partition": "main",
@@ -55,7 +55,7 @@
       "scale_up_transport": "nvlink",
       "launcher": "single-slurm",
       "backends": {"deepep-v2": [8, 16], "uccl-ep": [8, 16], "nccl-ep": [8]},
-      "ll_backends": {"deepep-v2": [8, 16], "uccl-ep": [8], "nccl-ep": [8]},
+      "ll_backends": {"deepep-v2": [8, 16], "uccl-ep": [8]},
       "fabric": {"nic": "8x ConnectX-7-class 400G InfiniBand rails (bare metal, gdrdrv)", "switch": "InfiniBand (model not surveyed)"},
       "operator": {
         "partition": "batch_1",
@@ -77,7 +77,7 @@
       "scale_up_transport": "nvlink",
       "launcher": "single-slurm",
       "backends": {"deepep-v2": [8, 16], "nccl-ep": [8]},
-      "ll_backends": {"deepep-v2": [8], "nccl-ep": [8]},
+      "ll_backends": {"deepep-v2": [8]},
       "fabric": {"nic": "ConnectX-8 2x400GbE", "switch": "NVIDIA Spectrum-X SN5600 (51.2T)"},
       "operator": {
         "partition": "batch_1",
@@ -104,7 +104,7 @@
       "scale_up_transport": "mnnvl",
       "launcher": "gb-nv",
       "backends": {"deepep-v2": [8, 16], "nccl-ep": [8, 16], "flashinfer-ep": [8, 16]},
-      "ll_backends": {"deepep-v2": [8, 16], "nccl-ep": [8]},
+      "ll_backends": {"deepep-v2": [8, 16]},
       "fabric": {"nic": "MNNVL (scale-out not used)", "switch": "NVLink NVL72"},
       "operator": {
         "partition": "batch",
@@ -124,7 +124,7 @@
       "scale_up_transport": "mnnvl",
       "launcher": "gb-nv",
       "backends": {"deepep-v2": [8, 16], "nccl-ep": [8, 16], "flashinfer-ep": [8, 16]},
-      "ll_backends": {"deepep-v2": [8, 16], "nccl-ep": [8]},
+      "ll_backends": {"deepep-v2": [8, 16]},
       "fabric": {"nic": "MNNVL (scale-out not used)", "switch": "NVLink NVL72"},
       "operator": {
         "partition": "batch_1",

experimental/CollectiveX/configs/sweep.json

@@ -17,10 +17,10 @@
     "chain_drop": 16
   },
   "workload": {
-    "name": "deepseek-v3",
+    "name": "deepseek-v4-pro",
     "hidden": 7168,
-    "topk": 8,
-    "routed_experts": 256,
+    "topk": 6,
+    "routed_experts": 384,
     "seed": 67,
     "token_ladders": {
       "decode": [1, 2, 4, 8, 16, 32, 64, 128, 256, 512],