Interleave runtime collective startup measurements

helios/src/execution/tests.rs

@@ -143,14 +143,14 @@ fn collective_overview_uses_graph_geometry_and_individual_bindings() {
     value.graphs[0].tuned_bindings = packed.tuned_bindings.clone();
     value.graphs.extend([packed, mixed]);
     let choices = vec![vec![
-        ("target-choice".into(), 4u32.to_ne_bytes().to_vec()),
-        ("mixed-choice".into(), ((1u32 << 16) | 64).to_ne_bytes().to_vec()),
+        ("target-choice".into(), ((5u32 << 16) | 1).to_ne_bytes().to_vec()),
+        ("mixed-choice".into(), ((6u32 << 16) | 64).to_ne_bytes().to_vec()),
     ]];
     let text = super::transport::collective_choices_overview(&value, &choices);
-    assert!(text.contains("Mixed/unknown: 1 group\n"));
-    assert!(text.contains("arbitrary-decode-name · Decode b=4 t=1 rows=4 drafts=0 prefill=0 · Full (4 CTAs)\n"));
-    assert!(text.contains("arbitrary-target-name · Speculative b=4 t=6 rows=16 drafts=5 prefill=0 · Full (4 CTAs)\n"));
-    assert!(text.contains("arbitrary-mixed-name · Mixed b=4 t=1 rows=131 drafts=0 prefill=128 · Payload RSAG (64 CTAs)\n"));
+    assert!(text.contains("Mode 5: 1 binding · Mode 6: 1 binding\n"));
+    assert!(text.contains("arbitrary-decode-name · Decode b=4 t=1 rows=4 drafts=0 prefill=0 · Mode 5 (1 CTAs)\n"));
+    assert!(text.contains("arbitrary-target-name · Speculative b=4 t=6 rows=16 drafts=5 prefill=0 · Mode 5 (1 CTAs)\n"));
+    assert!(text.contains("arbitrary-mixed-name · Mixed b=4 t=1 rows=131 drafts=0 prefill=128 · Mode 6 (64 CTAs)\n"));
 }
 
 pub(super) fn manifest() -> ExecutionManifest {

helios/src/execution/transport.rs

@@ -388,14 +388,14 @@ pub(super) fn collective_choices_overview(
     let mut totals = BTreeMap::new();
     let mut outputs = BTreeMap::new();
     for group in choices {
-        let modes = group.iter().map(|(_, bytes)| decode(bytes).0).collect::<BTreeSet<_>>();
-        let name = if modes.len() == 1 { modes.into_iter().next().unwrap() } else { "Mixed/unknown".into() };
-        *totals.entry(name).or_insert(0usize) += 1;
+        for (_, bytes) in group {
+            *totals.entry(decode(bytes).0).or_insert(0usize) += 1;
+        }
         outputs.extend(group.iter().map(|(name, bytes)| (name.as_str(), bytes.as_slice())));
     }
     let mut text = format!("  Collective choices · device {} · diagnostic\n", manifest.device);
     let counts = totals.iter().map(|(name, count)| {
-        format!("{name}: {count} group{}", if *count == 1 { "" } else { "s" })
+        format!("{name}: {count} binding{}", if *count == 1 { "" } else { "s" })
     }).collect::<Vec<_>>().join(" · ");
     writeln!(text, "  {counts}").unwrap();
     let mut graphs = manifest.graphs.iter().filter(|graph| !graph.tuned_bindings.is_empty()).collect::<Vec<_>>();

ignition/preparation/collective_ladder.py

@@ -6,8 +6,8 @@
 from ignition.errors import UsageError
 
 MODES = {
-    "joint_latency": ("full", "payload_rsag", "pull"),
-    "joint_bandwidth": ("payload_rsag", "bulk_push", "bulk_pull", "pull"),
+    "joint_latency": ("full", "payload_rsag", "pull", "serial_push", "queued_push"),
+    "joint_bandwidth": ("payload_rsag", "bulk_push", "bulk_pull", "pull", "serial_push", "queued_push"),
 }
 
 
@@ -45,9 +45,12 @@ def require(condition, message):
         require(isinstance(row["profiles"], dict) and bool(row["profiles"]), "profiles must be nonempty")
         for mode, counts in row["profiles"].items():
             require(mode in MODES[row["implementation"]], "mode does not belong to joint kernel")
+            require(data["world_size"] > 2 or mode not in ("serial_push", "queued_push"),
+                    "destination serialization requires more than two ranks")
             require(isinstance(counts, list) and bool(counts)
                     and all(type(v) is int and 1 <= v <= row["blocks"] for v in counts), "invalid active CTA counts")
             require(len(set(counts)) == len(counts), "duplicate active CTA counts")
+            require(mode != "serial_push" or counts == [1], "serial_push requires one active CTA")
     return data
 
 

ignition/preparation/ladders/README.md

@@ -18,6 +18,17 @@ including configuration-file and recipe defaults. Static policy rejects the
 option. File contents travel with the preparation plan and package rebuild
 settings; workers do not reopen the path.
 
+Startup interleaves 21 timing samples per profile across the required reduction
+groups. Each sample measures a complete chain of reductions and uses the slowest
+rank's duration. Selection minimizes their sum, so stalls contribute to the
+decision; exact ties follow the profile order in the file. Every profile must
+pass the arithmetic checks before serving can start.
+
+The groups share one captured startup graph. Reducing serving concurrency or
+draft width at load time retains this graph and its workspace whenever a selected
+serving graph uses a runtime collective. A package containing only one reduction
+group has no other groups to separate its samples in time.
+
 ## Version 1 format
 
 ```json
@@ -46,7 +57,11 @@ Each bucket declares one kernel family, thread count, and physical CTA grid.
 `joint_latency` permits `full`, `payload_rsag`, and `pull`; `joint_bandwidth`
 permits `payload_rsag`, `bulk_push`, `bulk_pull`, and `pull`. Each profile lists
 active CTA counts between one and the physical grid size. Several counts may
-belong to the same submode. `cache_policy` controls the captured profile race's
+belong to the same submode. With more than two ranks, both families also permit
+`serial_push` and `queued_push`, which send payloads to one peer at a time.
+`serial_push` requires exactly one active CTA; `queued_push` distributes tiles
+among active CTAs without requiring the entire grid to be resident.
+`cache_policy` controls the captured profile race's
 cache treatment and accepts `cold` or `warm`.
 
 The ladder applies to eligible plain BF16 PCIe reductions. TP size, compute

ignition/preparation/startup_tuning.py

@@ -109,11 +109,13 @@ def memory_report(self):
         from b12x.preparation import MemoryRequirements
 
         memory = MemoryRequirements.sequential(producer.memory_requirements() for producer in self.groups.values())
-        return {"startup_tuning_bytes": sum(entry.resident_nbytes for entry in memory.persistent),
+        # The catalog adds one aggregate status and one int32 scratch scalar.
+        return {"startup_tuning_bytes": sum(entry.resident_nbytes for entry in memory.persistent) + (8 if self.groups else 0),
                 "startup_tuning_producers": len(self.groups)}
 
     def catalog(self, graphs):
         import torch
+        from b12x.comm.pcie import run_joint_startup_tuners
 
         consumers = {name: self.captures.get(id(graph), set()) for name, graph in graphs.items()}
         needed = {self.owner_groups[owner] for owners in consumers.values() for owner in owners
@@ -122,6 +124,8 @@ def catalog(self, graphs):
         if needed and storage is None:
             raise RuntimeError("collective startup capture requires package storage")
         buffers, tuners, bindings, resources = {}, {}, {}, []
+        statuses, all_outputs = [], []
+        tuning_name = "collective-tuning"
         for index, (key, producer) in enumerate(self.groups.items()):
             if key not in needed:
                 continue
@@ -136,29 +140,58 @@ def catalog(self, graphs):
                     buffers[field] = spec.selector.view(torch.int32)
             status = f"{name}-status"
             buffers[status] = producer.status
-            device = self.session.device.ordinal
-            coordinator = getattr(self.session, "_coordinator", None)
-            if coordinator is not None:
-                coordinator.barrier(f"capture {name}", coordinator.ranks)
-            with torch.cuda.device(device):
-                stream = torch.cuda.Stream(device=device)
-                stream.wait_stream(torch.cuda.current_stream(device))
-                graph = torch.cuda.CUDAGraph(keep_graph=True)
-                with self.session.capture(), torch.cuda.graph(graph, pool=storage.graph_pool.id, stream=stream):
-                    producer.run()
-                graph.instantiate()
-                stream.synchronize()
-            tuners[name] = TuningGraph(graph, tuple(outputs), status)
+            all_outputs.extend(outputs)
+            statuses.append((index, producer.status))
             resources.append(producer)
             for serving, owners in consumers.items():
                 selected = {id(owner): owner for owner in self.owners if id(owner) in owners
                             and self.owner_groups[id(owner)] == key}
                 fields = {addresses[owner.plan.prepared.state.selector.data_ptr()] for owner in selected.values()}
-                bindings.setdefault(serving, []).extend(TunedBinding(field, name) for field in sorted(fields))
-        torch.cuda.synchronize(self.session.device.ordinal)
+                bindings.setdefault(serving, []).extend(TunedBinding(field, tuning_name) for field in sorted(fields))
+        if not resources:
+            return buffers, tuners, bindings, ()
+        device = self.session.device.ordinal
+        status_name = f"{tuning_name}-status"
+        scratch_name = f"{status_name}-scratch"
+        coordinator = getattr(self.session, "_coordinator", None)
+        if coordinator is not None:
+            coordinator.barrier(f"capture {tuning_name}", coordinator.ranks)
+        with torch.cuda.device(device):
+            status = resources[0].status.new_full((1,), -1)
+            scratch = status.new_empty((1,))
+            buffers[status_name], buffers[scratch_name] = status, scratch
+            _publish_group_status(statuses, status, scratch)
+            status.fill_(-1)
+            stream = torch.cuda.Stream(device=device)
+            stream.wait_stream(torch.cuda.current_stream(device))
+            graph = torch.cuda.CUDAGraph(keep_graph=True)
+            with self.session.capture(), torch.cuda.graph(graph, pool=storage.graph_pool.id, stream=stream):
+                status.fill_(-1)
+                run_joint_startup_tuners(resources)
+                _publish_group_status(statuses, status, scratch)
+            graph.instantiate()
+            stream.synchronize()
+        tuners[tuning_name] = TuningGraph(graph, tuple(all_outputs), status_name)
+        torch.cuda.synchronize(device)
         return buffers, tuners, bindings, tuple(resources)
 
 
+def _publish_group_status(groups, status, scratch):
+    """Publish zero or the highest failing prepared-group index plus one.
+
+    Per-group status buffers retain the original failure values. The aggregate
+    identifies a failing group without interpreting its protocol in native code.
+    All operations write caller-owned scalars and are captured with the race.
+    """
+    import torch
+
+    status.zero_()
+    for index, group_status in groups:
+        torch.ne(group_status, 0, out=scratch)
+        scratch.mul_(index + 1)
+        torch.maximum(status, scratch, out=status)
+
+
 def register_collective(owner, session):
     from .tuning import RecordedPreparationSession, register_finalizer
 

pyproject.toml

@@ -22,7 +22,7 @@ prepare = [
     "torch==2.14.1+cu132",
     "torchvision==0.29.1+cu132",
     "transformers[torch,vision,chat_template] @ git+https://github.com/huggingface/transformers.git@1d9654e78a23fb167ef539cbd699dfbd3e4830cf",
-    "flashinfer-python[b12x] @ git+https://github.com/local-inference-lab/flashinfer.git@f91f257e079d8fa632d64cc6db49a1e609ff02ed",
+    "flashinfer-python[b12x] @ git+https://github.com/local-inference-lab/flashinfer.git@7294648399133bae4997d0072590b0f977313145",
     "cuda-python==13.3.1",
     "cuda-tile[tileiras]==1.6.0",
     "safetensors==0.8.0",

tests/components/test_automatic_collective_export_cuda.py

@@ -93,7 +93,10 @@ def _rank(rank, directory, readiness):
                 tuner = prepared.tuning_graphs[binding.tuning_graph]
                 assert tuner.outputs == (binding.buffer,)
                 assert prepared.buffers[binding.buffer].data_ptr() == operation.plan.prepared.state.selector.data_ptr()
-                assert prepared.buffers[tuner.status_buffer] is producer.status
+                assert prepared.buffers["collective-tuning-0-status"] is producer.status
+                aggregate = prepared.buffers[tuner.status_buffer]
+                assert aggregate is not producer.status
+                assert aggregate.dtype == torch.int32 and aggregate.shape == (1,)
                 assert set(prepared.graphs) == {*signatures, binding.tuning_graph}
                 from ignition.preparation.startup_tuning import replay_startup_tuning