fuel_dispatch/ranker/route_picker.rs
1// SPDX-License-Identifier: MIT OR Apache-2.0
2//! The **runtime route picker** ("Picker 2") — selects an arm at each
3//! `Op::Branch` decision point by **live telemetry**.
4//!
5//! Phase C PR-C1 of the "plan IS the graph" rebuild
6//! ([`../../docs/session-prompts/plan-is-graph-rebuild.md`](
7//! ../../docs/session-prompts/plan-is-graph-rebuild.md) Phase C,
8//! capability 7; [`../../docs/architecture/06-runtime.md`](
9//! ../../docs/architecture/06-runtime.md) §"Route picker (the runtime
10//! selector / Picker 2)").
11//!
12//! ## What this is
13//!
14//! Phases A+B emit the optimized form as an in-graph multi-path
15//! structure: `optimize_graph` records `Op::Branch { reconverge_at }`
16//! decision points whose arms (`inputs`) are competing `(device, backend)`
17//! placements — arm-0 the cost-vector winner, arm-1+ runner-up
18//! placements bounded by the per-device Pareto frontier. Phase B realize
19//! followed **arm-0 statically** (`lower_runs_arm0`). This module is the
20//! runtime selector that **chooses an arm per branch by live telemetry**,
21//! replacing the static arm-0 lowering with [`fuel_graph::lower_picked_route`].
22//!
23//! ## How it picks
24//!
25//! [`pick_route`] walks the reachable branches in **topological order**
26//! ([`fuel_graph::branches_in_topo_order`], keyed by each branch's
27//! `reconverge_at` merge position) so coupled upstream decisions are
28//! committed before a downstream branch is reached. At each branch it
29//! builds a per-branch [`AlternativeSet`] over the arms — one
30//! [`Candidate`] per arm carrying the arm's real `(backend, device)` and
31//! its plan cost — and consults the supplied [`RuntimeSelector`] (the
32//! production [`crate::ranker::ChainedSelector`]: VRAM-pressure guard
33//! over the [`BackendRuntimeLookup`]'s per-tier free memory, then
34//! Judge-measured rank, then the static winner). The picked candidate's
35//! `(backend, device)` maps back to an arm index, stored in the
36//! [`PickedRoute`].
37//!
38//! Under **no pressure / no telemetry** the chained selector degrades to
39//! `set.winner()` = arm-0 (the cost-ranked first candidate), so the route
40//! is arm-0 everywhere and [`fuel_graph::lower_picked_route`] reproduces
41//! Phase B's `lower_runs_arm0` exactly. Under **VRAM pressure** the guard
42//! demotes/skips the GPU arm and the picker takes a host-RAM arm
43//! instead — exactly the "multiple paths per device" payoff
44//! (06-runtime §"Memory pressure as the parallelism limit").
45//!
46//! ## Caching + re-resolution
47//!
48//! [`RouteCache`] holds the last resolved [`PickedRoute`] plus a
49//! [`TelemetryFingerprint`] (per-tier free-memory buckets across the
50//! candidate backends). [`RouteCache::resolve`] re-resolves only when the
51//! fingerprint changes meaningfully (a bucket shift); otherwise it reuses
52//! the cached route. In steady state the picker is a table lookup;
53//! re-resolution fires on transitions (pressure rising, a device
54//! appearing). A CPU-only build has zero branches ⇒ the route is empty ⇒
55//! the picker is a no-op and realize is unchanged.
56
57use std::sync::Arc;
58
59use fuel_graph::{Graph, NodeId, PickedRoute, branches_in_topo_order};
60use fuel_ir::dispatch::{OpKind, SizeClass};
61use fuel_ir::probe::BackendId;
62use fuel_ir::{DType, DeviceLocation, Shape};
63
64use crate::fused::{CostEstimate, PrecisionGuarantee};
65use crate::kernel::{KernelBindingTable, KernelCaps, OpParams};
66use crate::pipelined::{build_lookup_dtypes, op_to_op_kind};
67use crate::plan::default_device_for;
68use crate::ranker::{
69 AlternativeSet, BackendRuntimeLookup, Candidate, DecisionContext, RuntimeSelector,
70 enumerate_candidates,
71};
72
73/// Bounded lookahead for adversarially-coupled branches (06-runtime
74/// §"Resolution order matters when decisions are coupled"). The default
75/// resolution is locally-greedy in topological order — by the time the
76/// picker reaches a branch its upstream decisions are already committed.
77/// Rare adversarial coupling is bounded by considering at most this many
78/// branches jointly. PR-C1 commits upstream-first in topo order (the
79/// greedy default); `K` is the documented ceiling the joint-resolution
80/// window must never exceed.
81pub const LOOKAHEAD_K: usize = 3;
82
83/// A coarse fingerprint of the live per-tier free memory across the
84/// backends a route's branches care about — the cache key for
85/// [`RouteCache`]. Re-resolution fires only on a *meaningful* delta: we
86/// bucket each backend's free bytes into coarse log-scale tiers so noise
87/// (a few MB jitter) does not invalidate the cached route, but a genuine
88/// pressure transition (free memory dropping an order of magnitude) does.
89#[derive(Clone, Debug, Default, PartialEq, Eq)]
90pub struct TelemetryFingerprint {
91 /// `(backend, free-memory bucket)` pairs, sorted by backend for a
92 /// stable equality check. An absent backend (no live handle) is
93 /// omitted — its bucket is "unknown", which never forces a
94 /// re-resolve on its own.
95 buckets: Vec<(BackendId, u8)>,
96}
97
98impl TelemetryFingerprint {
99 /// Fingerprint the backends in `backends` against the live lookup.
100 /// Each backend's free bytes are bucketed via [`free_bytes_bucket`].
101 fn sample(backends: &[BackendId], lookup: Option<&BackendRuntimeLookup>) -> Self {
102 let mut buckets: Vec<(BackendId, u8)> = Vec::new();
103 if let Some(lookup) = lookup {
104 let mut seen: Vec<BackendId> = Vec::new();
105 for &b in backends {
106 if seen.contains(&b) {
107 continue;
108 }
109 seen.push(b);
110 if let Some(handle) = lookup(b, default_device_for(b))
111 && let Some(free) = handle.available_bytes()
112 {
113 buckets.push((b, free_bytes_bucket(free)));
114 }
115 }
116 }
117 buckets.sort_by_key(|&(b, _)| b as u8);
118 Self { buckets }
119 }
120}
121
122/// Bucket free-memory bytes into coarse log-scale tiers (powers of ~16×)
123/// so the [`TelemetryFingerprint`] is stable under small jitter but
124/// flips on an order-of-magnitude pressure transition. `0` for zero free;
125/// otherwise `floor(log2(free)/4) + 1` clamped to a byte.
126fn free_bytes_bucket(free: u64) -> u8 {
127 if free == 0 {
128 return 0;
129 }
130 let log2 = 63 - free.leading_zeros(); // floor(log2(free))
131 ((log2 / 4) + 1).min(u8::MAX as u32) as u8
132}
133
134/// Per-realize cache of the resolved route + the telemetry it was
135/// resolved under (06-runtime §"Telemetry caching for picker speed").
136///
137/// Live across realizes by being held by the caller (the bridge). The
138/// first [`resolve`](RouteCache::resolve) populates it; subsequent calls
139/// reuse the cached route while the [`TelemetryFingerprint`] is stable,
140/// and re-resolve on a meaningful delta.
141#[derive(Clone, Debug, Default)]
142pub struct RouteCache {
143 cached: Option<(TelemetryFingerprint, PickedRoute)>,
144 /// Diagnostics: how many times [`resolve`](RouteCache::resolve) hit
145 /// the cache vs. re-resolved. Tests assert the re-resolve discipline.
146 pub hits: u64,
147 pub resolves: u64,
148}
149
150impl RouteCache {
151 /// A fresh, empty cache.
152 pub fn new() -> Self {
153 Self::default()
154 }
155
156 /// Resolve the route for `roots`, reusing the cached route while the
157 /// live telemetry fingerprint is unchanged. Re-resolves (and updates
158 /// the cache) on the first call or a meaningful telemetry delta.
159 ///
160 /// `bindings` is the runtime kernel registry (the per-arm candidates are
161 /// re-enumerated from it + the graph); `selector` is the production runtime
162 /// selector; `lookup` is the live per-tier free-memory closure (`None` ⇒ no
163 /// pressure signal ⇒ arm-0).
164 pub fn resolve(
165 &mut self,
166 graph: &Graph,
167 roots: &[NodeId],
168 bindings: &KernelBindingTable,
169 selector: &dyn RuntimeSelector,
170 lookup: Option<&BackendRuntimeLookup>,
171 ) -> PickedRoute {
172 let backends = route_backends(graph, roots);
173 let fingerprint = TelemetryFingerprint::sample(&backends, lookup);
174
175 if let Some((cached_fp, cached_route)) = &self.cached
176 && *cached_fp == fingerprint
177 {
178 self.hits += 1;
179 return cached_route.clone();
180 }
181
182 let route = pick_route(graph, roots, bindings, selector, lookup);
183 self.resolves += 1;
184 self.cached = Some((fingerprint, route.clone()));
185 route
186 }
187}
188
189/// The set of `(backend)`s the route's branches' arms target — the
190/// backends whose live free memory the picker (and the cache
191/// fingerprint) consults.
192fn route_backends(graph: &Graph, roots: &[NodeId]) -> Vec<BackendId> {
193 let mut backends: Vec<BackendId> = Vec::new();
194 for branch in branches_in_topo_order(graph, roots) {
195 for &arm in &graph.node(branch).inputs {
196 if let Some(b) = graph.target_backend(arm)
197 && !backends.contains(&b)
198 {
199 backends.push(b);
200 }
201 }
202 }
203 backends
204}
205
206/// Walk the multi-path graph's `Op::Branch` decision points in
207/// **topological order** and choose one arm at each via the
208/// `RuntimeSelector` over the arms' `(device, backend)` + plan cost +
209/// live per-tier free memory. Returns the [`PickedRoute`]
210/// ([`fuel_graph::lower_picked_route`] consumes it).
211///
212/// Topological order means a downstream branch is resolved only after its
213/// upstream branches are committed (the picks recorded in `picked`), so
214/// coupled decisions resolve consistently (06-runtime §"Resolution order
215/// matters when decisions are coupled"). The greedy upstream-first pass
216/// is the default; [`LOOKAHEAD_K`] bounds any joint-resolution window for
217/// adversarial coupling.
218///
219/// With no pressure / no telemetry the selector degrades to `set.winner()`
220/// = arm-0 for every branch, so the returned route is empty-equivalent
221/// (every pick is arm 0) and realize is identical to Phase B.
222pub fn pick_route(
223 graph: &Graph,
224 roots: &[NodeId],
225 bindings: &KernelBindingTable,
226 selector: &dyn RuntimeSelector,
227 lookup: Option<&BackendRuntimeLookup>,
228) -> PickedRoute {
229 // The live per-tier free-memory lookup is consulted *through* the
230 // selector — the production `ChainedSelector` holds its own
231 // `BackendRuntimeLookup` and reads each arm's `(backend, device)` fit
232 // status in `select`. `lookup` is part of the picker's documented
233 // contract (06-runtime §"Route picker") and is the same handle the
234 // `RouteCache` fingerprints telemetry through; the per-arm guard
235 // itself lives in the selector, so `pick_route` does not re-query it.
236 let _ = lookup;
237 let mut picked = PickedRoute::new();
238 let branches = branches_in_topo_order(graph, roots);
239
240 for branch in branches {
241 // Only record a non-arm-0 pick; arm 0 is the default in
242 // `lower_picked_route`, so keeping the route minimal makes the
243 // "no pressure ⇒ empty route ⇒ Phase B behavior" contract direct.
244 let chosen = resolve_branch(graph, branch, bindings, selector);
245 if chosen != 0 {
246 picked.insert(branch, chosen);
247 }
248 }
249
250 picked
251}
252
253/// Resolve **one** `Op::Branch` to its chosen arm index — the body of
254/// [`pick_route`]'s per-branch loop, lifted out so the **streaming
255/// run-walk** (Step E Phase C, PR C1) can resolve branches **lazily, as
256/// the frontier reaches each one**, rather than resolving the whole route
257/// up front (06-runtime §"selecting arms by live device load … per-
258/// decision-point during the dispatch walk").
259///
260/// This is the SINGLE arm-decision primitive: [`pick_route`] (the eager
261/// lane — [`RouteCache`] + the route-picker tests) calls it in
262/// topological order over every branch; the streaming lowering
263/// ([`fuel_graph::lower_picked_route_streaming`]) calls it one branch at a
264/// time at the branch's arm-entry boundary. Because it consults the **same
265/// [`RuntimeSelector`]** either way — and the production VRAM-pressure
266/// chain reads only free-memory state (not walk progress) — the lazy and
267/// eager resolutions are byte-identical on every input. C1 changes *when*
268/// a branch resolves, never *which* arm it picks; the load signal that
269/// will make the moment matter is C2.
270///
271/// A single-arm (or zero-arm) branch has no decision — returns arm 0
272/// (`lower_picked_route`'s default), so it never enters the route map.
273pub fn resolve_branch(
274 graph: &Graph,
275 branch: NodeId,
276 bindings: &KernelBindingTable,
277 selector: &dyn RuntimeSelector,
278) -> usize {
279 let arms = graph.node(branch).inputs.clone();
280 if arms.len() < 2 {
281 return 0;
282 }
283 pick_arm(graph, &arms, bindings, selector)
284}
285
286/// Choose one arm index for a single branch. Builds a per-arm
287/// [`AlternativeSet`] (one [`Candidate`] per arm carrying the arm's real
288/// `(backend, device)` + plan cost + the fork's decision context), runs
289/// the selector, and maps the picked candidate's `(backend, device)`
290/// back to its arm index.
291fn pick_arm(
292 graph: &Graph,
293 arms: &[NodeId],
294 bindings: &KernelBindingTable,
295 selector: &dyn RuntimeSelector,
296) -> usize {
297 // Step D: re-derive the fork's candidate set from the graph + the runtime
298 // binding registry (was the cached `plan.alternatives(fork)`). The fork
299 // node = arm-0's exit; enumerating its `(op, dtypes)` across the arms'
300 // placements reproduces the per-placement Candidates + the DecisionContext
301 // the selector's VRAM-pressure guard + Judge-aware rank read. We mirror
302 // them into a per-arm set in ARM ORDER so the picked index maps straight
303 // back to an arm.
304 let fork = arms[0];
305 let fork_set = enumerate_fork_set(graph, fork, arms, bindings);
306
307 let mut arm_set = AlternativeSet::empty();
308 for &arm in arms {
309 let backend = graph.target_backend(arm).unwrap_or(BackendId::Cpu);
310 let device = default_device_for(backend);
311 let candidate = candidate_for_placement(&fork_set, backend, device)
312 .cloned()
313 .unwrap_or_else(|| synthetic_candidate(backend, device));
314 arm_set.push(candidate);
315 }
316 // Carry the fork's decision context so the Judge-aware rank leg can
317 // re-query measurements per arm.
318 if let Some(ctx) = fork_set.context() {
319 arm_set.set_context(*ctx);
320 }
321
322 // The selector picks one candidate; arm index = its position in the
323 // arm-ordered set. `select` returns `None` only on an empty set
324 // (impossible here — arms.len() >= 2), in which case arm 0.
325 match selector.select(&arm_set) {
326 Some(picked) => arm_index_of(&arm_set, picked),
327 None => 0,
328 }
329}
330
331/// Re-derive the fork node's candidate set from the graph + the runtime
332/// binding registry — the dispatch-time replacement for the cached
333/// `ExecutionPlan::alternatives(fork)` (Step D). Enumerates the fork's
334/// `(op, dtypes)` across the arms' `(backend, device)` placements and stamps
335/// the fork's [`DecisionContext`] so the selector's Judge leg keys on the same
336/// `(op, dtype, size_class)` triple the planner used. A non-kernel fork (no
337/// `OpKind` mapping) yields an empty set ⇒ the per-arm loop falls back to
338/// [`synthetic_candidate`], preserving the prior behavior when the plan
339/// carried no entry for the fork. Layer-1 cost only (the candidates' default
340/// `static_cost`); the selector's own Judge leg supplies the Layer-2
341/// measurement, so no cost composition is needed here.
342fn enumerate_fork_set(
343 graph: &Graph,
344 fork: NodeId,
345 arms: &[NodeId],
346 bindings: &KernelBindingTable,
347) -> AlternativeSet {
348 let node = graph.node(fork);
349 let Some(op_kind) = op_to_op_kind(&node.op) else {
350 return AlternativeSet::empty();
351 };
352 let dtypes = build_lookup_dtypes(graph, node);
353 let placements: Vec<(BackendId, DeviceLocation)> = arms
354 .iter()
355 .map(|&arm| {
356 let backend = graph.target_backend(arm).unwrap_or(BackendId::Cpu);
357 (backend, default_device_for(backend))
358 })
359 .collect();
360 // `op_params` is unread by the selector (it ranks on backend/device/cost/
361 // kernel_source), so `None` suffices — the chosen arm is lowered to real
362 // runs and dispatched through the normal binding-table path regardless.
363 let mut set = enumerate_candidates(op_kind, &dtypes, &placements, &OpParams::None, bindings);
364 if let Some(ctx) = fork_decision_context(graph, node, op_kind, &dtypes) {
365 set.set_context(ctx);
366 }
367 set
368}
369
370/// The fork's [`DecisionContext`] — `(op, principal dtype, size class)`,
371/// the triple the Judge keys measurements on. Mirrors `compile_plan`'s
372/// stamping (plan.rs) exactly — `SizeClass::for_op` over the input
373/// operand shapes — so dispatch-time re-derivation hits the same Judge
374/// entries (a matmul keys on its `(m,n,k)` aspect, aligned with the
375/// planner and producer).
376fn fork_decision_context(
377 graph: &Graph,
378 node: &fuel_graph::Node,
379 op_kind: OpKind,
380 dtypes: &[DType],
381) -> Option<DecisionContext> {
382 let &principal_dtype = dtypes.first()?;
383 let input_shapes: Vec<Shape> = node
384 .inputs
385 .iter()
386 .map(|&input_id| graph.node(input_id).shape.clone())
387 .collect();
388 let size_class = SizeClass::for_op(op_kind, &input_shapes);
389 Some(DecisionContext {
390 op: op_kind,
391 principal_dtype,
392 size_class,
393 })
394}
395
396/// Find a plan candidate matching `(backend, device)` (the arm's real
397/// placement). Returns the first match — sibling kernels at the same
398/// `(backend, device)` are interchangeable for the arm-selection purpose
399/// (they differ only by `kernel_source`, which the selector still ranks).
400fn candidate_for_placement(
401 set: &AlternativeSet,
402 backend: BackendId,
403 device: DeviceLocation,
404) -> Option<&Candidate> {
405 set.alternatives()
406 .iter()
407 .find(|c| c.backend == backend && c.device == device)
408 .or_else(|| {
409 // Fall back to a backend-only match (the plan may carry a
410 // different GPU ordinal than the arm's default device).
411 set.alternatives().iter().find(|c| c.backend == backend)
412 })
413}
414
415/// The arm index of `picked` within the arm-ordered `set` — by
416/// `(backend, device)` identity. The set was built in arm order, so the
417/// first matching position is the chosen arm.
418fn arm_index_of(set: &AlternativeSet, picked: &Candidate) -> usize {
419 set.alternatives()
420 .iter()
421 .position(|c| c.backend == picked.backend && c.device == picked.device)
422 .unwrap_or(0)
423}
424
425/// A minimal stand-in [`Candidate`] for an arm with no matching plan
426/// entry: real `(backend, device)`, zero cost, no-op kernel. The
427/// selector reads only `backend` / `device` / `static_cost` /
428/// `inbound_transfer_ns` / `kernel_source` for its pick, so the kernel
429/// pointer is never invoked — this candidate exists purely to carry the
430/// arm's placement to the VRAM-pressure guard.
431fn synthetic_candidate(backend: BackendId, device: DeviceLocation) -> Candidate {
432 Candidate {
433 kernel: noop_kernel,
434 caps: KernelCaps::empty(),
435 backend,
436 device,
437 precision: PrecisionGuarantee::PRIMITIVE_DETERMINISTIC_CPU,
438 static_cost: CostEstimate {
439 flops: 0,
440 bytes_moved: 0,
441 kernel_overhead_ns: 0,
442 },
443 inbound_transfer_ns: 0,
444 op_params: OpParams::None,
445 coupling: Vec::new(),
446 kernel_source: "route-picker-synthetic",
447 kernel_revision_hash: 0,
448 }
449}
450
451/// Never-invoked kernel for [`synthetic_candidate`] (see its docs). The
452/// picker reads placement metadata only; the chosen arm is lowered to
453/// runs and dispatched through the normal binding-table path, not via
454/// this pointer.
455fn noop_kernel(
456 _i: &[Arc<std::sync::RwLock<fuel_memory::Storage>>],
457 _o: &mut [Arc<std::sync::RwLock<fuel_memory::Storage>>],
458 _l: &[fuel_ir::Layout],
459 _p: &OpParams,
460) -> fuel_ir::Result<()> {
461 Ok(())
462}
463
464#[cfg(test)]
465mod tests {
466 use super::*;
467 use crate::ranker::{BackendRuntimeHandle, ChainedSelector, WinnerSelector};
468 use fuel_backend_contract::backend::BackendRuntime;
469 use fuel_graph::{Node, Op};
470 use fuel_ir::{DType, Shape};
471
472 fn f32_node(g: &mut Graph, op: Op, inputs: Vec<NodeId>) -> NodeId {
473 g.push(Node {
474 op,
475 inputs,
476 shape: Shape::from_dims(&[2]),
477 dtype: DType::F32,
478 })
479 }
480
481 /// Build a 2-arm diamond with arm-0 on `arm0_backend` and arm-1 on
482 /// `arm1_backend`. Returns `(graph, branch, arm0, arm1, post)`.
483 ///
484 /// Topology mirrors the run.rs diamond: `diverge -> {arm0, arm1}`;
485 /// `reconverge` reads arm0 (the runnability invariant); the Branch
486 /// merges {arm0, arm1} at `reconverge`. The arm exits carry their
487 /// `target_backend` so the picker can read each arm's placement.
488 fn diamond(
489 arm0_backend: BackendId,
490 arm1_backend: BackendId,
491 ) -> (Graph, NodeId, NodeId, NodeId, NodeId) {
492 let mut g = Graph::new();
493 let pre = f32_node(&mut g, Op::Const, vec![]);
494 let diverge = f32_node(&mut g, Op::Relu, vec![pre]);
495 let arm0 = f32_node(&mut g, Op::Silu, vec![diverge]);
496 let arm1 = f32_node(&mut g, Op::GeluTanh, vec![diverge]);
497 g.set_target_backend(arm0, arm0_backend);
498 g.set_target_backend(arm1, arm1_backend);
499 let reconverge = f32_node(&mut g, Op::Relu, vec![arm0]);
500 let mut b = g.open_branch(diverge);
501 b.add_arm(arm0);
502 b.add_arm(arm1);
503 let branch = b
504 .finalize_branches(&mut g, reconverge)
505 .expect("well-formed 2-arm branch")
506 .expect("2 arms survive");
507 let post = f32_node(&mut g, Op::Tanh, vec![reconverge]);
508 (g, branch, arm0, arm1, post)
509 }
510
511 /// A binding table with the diamond's fork op (`Op::Silu` ⇒
512 /// `SiluElementwise`) registered at both arm backends, so `pick_arm`
513 /// re-enumerates a real candidate per placement (Step D — was a hand-built
514 /// `ExecutionPlan`). No-op kernels: the picker reads only placement
515 /// metadata + `kernel_source`, never invoking the pointer.
516 fn diamond_bindings() -> KernelBindingTable {
517 let mut t = KernelBindingTable::new();
518 for backend in [BackendId::Cuda, BackendId::Cpu] {
519 t.register(
520 OpKind::SiluElementwise,
521 &[DType::F32, DType::F32],
522 backend,
523 noop_kernel,
524 );
525 }
526 t
527 }
528
529 struct MockRuntime {
530 available: Option<u64>,
531 total: Option<u64>,
532 }
533 impl BackendRuntime for MockRuntime {
534 fn available_bytes(&self) -> Option<u64> {
535 self.available
536 }
537 fn total_bytes(&self) -> Option<u64> {
538 self.total
539 }
540 }
541
542 /// Per-backend `(available, total)` lookup; backends not listed
543 /// resolve to `None` (= Unknown / no signal).
544 fn lookup_for(entries: Vec<(BackendId, Option<u64>, Option<u64>)>) -> BackendRuntimeLookup {
545 Arc::new(move |b, _d| {
546 entries
547 .iter()
548 .find(|(eb, _, _)| *eb == b)
549 .map(|&(_, a, t)| {
550 Box::new(MockRuntime {
551 available: a,
552 total: t,
553 }) as BackendRuntimeHandle
554 })
555 })
556 }
557
558 /// (a) NO PRESSURE / NO TELEMETRY ⇒ the picker chooses arm-0 (the
559 /// winner) at every branch ⇒ an empty route ⇒ realize is identical to
560 /// Phase B. Tested with both the bare `WinnerSelector` and the
561 /// production `ChainedSelector` with no signals.
562 #[test]
563 fn no_telemetry_picks_arm0_empty_route() {
564 let (g, _branch, _fork, _arm1, post) = diamond(BackendId::Cuda, BackendId::Cpu);
565 let bindings = diamond_bindings();
566
567 // Bare winner selector: always arm 0.
568 let winner = WinnerSelector;
569 let route = pick_route(&g, &[post], &bindings, &winner, None);
570 assert!(
571 route.is_empty(),
572 "no telemetry ⇒ arm-0 everywhere ⇒ empty route; got {route:?}",
573 );
574
575 // Production chained selector with NO signals (no judge, no
576 // lookup) degrades to the static winner = arm 0.
577 let chained = ChainedSelector::with_default_estimator(None, None);
578 let route = pick_route(&g, &[post], &bindings, &chained, None);
579 assert!(
580 route.is_empty(),
581 "ChainedSelector with no signals ⇒ arm-0 everywhere; got {route:?}",
582 );
583 }
584
585 /// (b) SIMULATED VRAM PRESSURE: the GPU arm's backend reports tiny
586 /// free VRAM (WontFit), so the chained selector's guard skips it and
587 /// the picker takes the host-RAM (CPU) arm instead — arm 1.
588 #[test]
589 fn vram_pressure_picks_host_ram_arm() {
590 let (g, branch, _fork, _arm1, post) = diamond(BackendId::Cuda, BackendId::Cpu);
591 // Both placements resolve to the same no-op kernel; the only signal is
592 // memory pressure, so the guard alone must flip the pick.
593 let bindings = diamond_bindings();
594
595 // CUDA has 1 byte free of 10_000 → WontFit for the default
596 // bytes_moved estimate; CPU has ample room → Comfortable.
597 let lookup = lookup_for(vec![
598 (BackendId::Cuda, Some(1), Some(10_000)),
599 (BackendId::Cpu, Some(8_000), Some(10_000)),
600 ]);
601 let chained = ChainedSelector::with_default_estimator(None, Some(lookup.clone()));
602
603 let route = pick_route(&g, &[post], &bindings, &chained, Some(&lookup));
604 assert_eq!(
605 route.get(&branch).copied(),
606 Some(1),
607 "under VRAM pressure the picker takes the host-RAM (CPU) arm; \
608 route={route:?}",
609 );
610 }
611
612 /// (c) TWO COUPLED BRANCHES resolve consistently in topological
613 /// order: both GPU arms are under pressure, so the picker takes the
614 /// CPU arm at BOTH branches — the downstream branch sees the same
615 /// pressure the upstream one did and resolves the same way.
616 #[test]
617 fn two_coupled_branches_resolve_consistently() {
618 // Two diamonds chained: post0 feeds the second diverge.
619 let mut g = Graph::new();
620 let pre = f32_node(&mut g, Op::Const, vec![]);
621 // --- branch 1 ---
622 let div1 = f32_node(&mut g, Op::Relu, vec![pre]);
623 let a0_1 = f32_node(&mut g, Op::Silu, vec![div1]);
624 let a1_1 = f32_node(&mut g, Op::GeluTanh, vec![div1]);
625 g.set_target_backend(a0_1, BackendId::Cuda);
626 g.set_target_backend(a1_1, BackendId::Cpu);
627 let recon1 = f32_node(&mut g, Op::Relu, vec![a0_1]);
628 let mut b1 = g.open_branch(div1);
629 b1.add_arm(a0_1);
630 b1.add_arm(a1_1);
631 let branch1 = b1
632 .finalize_branches(&mut g, recon1)
633 .expect("branch1 valid")
634 .expect("2 arms");
635 // --- branch 2 (downstream of branch 1's merge) ---
636 let div2 = f32_node(&mut g, Op::Tanh, vec![recon1]);
637 let a0_2 = f32_node(&mut g, Op::Silu, vec![div2]);
638 let a1_2 = f32_node(&mut g, Op::GeluTanh, vec![div2]);
639 g.set_target_backend(a0_2, BackendId::Cuda);
640 g.set_target_backend(a1_2, BackendId::Cpu);
641 let recon2 = f32_node(&mut g, Op::Relu, vec![a0_2]);
642 let mut b2 = g.open_branch(div2);
643 b2.add_arm(a0_2);
644 b2.add_arm(a1_2);
645 let branch2 = b2
646 .finalize_branches(&mut g, recon2)
647 .expect("branch2 valid")
648 .expect("2 arms");
649 let post = f32_node(&mut g, Op::Tanh, vec![recon2]);
650
651 // Topo order must place branch1 before branch2.
652 let order = branches_in_topo_order(&g, &[post]);
653 assert_eq!(
654 order,
655 vec![branch1, branch2],
656 "coupled branches resolve upstream-first in topo order",
657 );
658
659 // Both forks (`Op::Silu`) re-enumerate from the same registered
660 // bindings (Silu at both arm backends).
661 let bindings = diamond_bindings();
662
663 // GPU under pressure ⇒ both branches pick the CPU arm (arm 1).
664 let lookup = lookup_for(vec![
665 (BackendId::Cuda, Some(1), Some(10_000)),
666 (BackendId::Cpu, Some(8_000), Some(10_000)),
667 ]);
668 let chained = ChainedSelector::with_default_estimator(None, Some(lookup.clone()));
669 let route = pick_route(&g, &[post], &bindings, &chained, Some(&lookup));
670
671 assert_eq!(route.get(&branch1).copied(), Some(1), "branch1 → CPU arm");
672 assert_eq!(route.get(&branch2).copied(), Some(1), "branch2 → CPU arm");
673 }
674
675 /// (d) The route is CACHED and re-resolved only on a meaningful
676 /// telemetry delta. Same fingerprint ⇒ cache hit (no re-resolve); a
677 /// genuine pressure transition ⇒ re-resolve + a possibly different
678 /// route.
679 #[test]
680 fn route_cached_and_reresolved_on_telemetry_delta() {
681 let (g, branch, _fork, _arm1, post) = diamond(BackendId::Cuda, BackendId::Cpu);
682 let bindings = diamond_bindings();
683
684 // Start comfortable: both backends have ample free memory.
685 let comfy = lookup_for(vec![
686 (BackendId::Cuda, Some(8_000), Some(10_000)),
687 (BackendId::Cpu, Some(8_000), Some(10_000)),
688 ]);
689 let chained_comfy = ChainedSelector::with_default_estimator(None, Some(comfy.clone()));
690
691 let mut cache = RouteCache::new();
692 let r1 = cache.resolve(&g, &[post], &bindings, &chained_comfy, Some(&comfy));
693 assert!(r1.is_empty(), "comfortable ⇒ arm-0 (winner) route");
694 assert_eq!(cache.resolves, 1, "first resolve populates the cache");
695 assert_eq!(cache.hits, 0);
696
697 // Same telemetry ⇒ cache hit, no re-resolve.
698 let r2 = cache.resolve(&g, &[post], &bindings, &chained_comfy, Some(&comfy));
699 assert_eq!(r2, r1, "stable telemetry reuses the cached route");
700 assert_eq!(cache.resolves, 1, "no re-resolve on a stable fingerprint");
701 assert_eq!(cache.hits, 1, "stable telemetry is a cache hit");
702
703 // Meaningful delta: CUDA free memory collapses to 1 byte
704 // (WontFit). Re-resolve fires and the route flips to the CPU arm.
705 let pressured = lookup_for(vec![
706 (BackendId::Cuda, Some(1), Some(10_000)),
707 (BackendId::Cpu, Some(8_000), Some(10_000)),
708 ]);
709 let chained_pressured =
710 ChainedSelector::with_default_estimator(None, Some(pressured.clone()));
711 let r3 = cache.resolve(&g, &[post], &bindings, &chained_pressured, Some(&pressured));
712 assert_eq!(
713 cache.resolves, 2,
714 "a meaningful telemetry delta re-resolves"
715 );
716 assert_eq!(cache.hits, 1, "the delta is not counted as a hit");
717 assert_eq!(
718 r3.get(&branch).copied(),
719 Some(1),
720 "after the pressure transition the route picks the CPU arm",
721 );
722 }
723
724 /// A graph with ZERO branches (a CPU-only build) ⇒ the picker is a
725 /// no-op ⇒ the route is empty ⇒ realize is unchanged. The cache
726 /// resolves it once and caches the empty route.
727 #[test]
728 fn branchless_graph_is_a_noop() {
729 let mut g = Graph::new();
730 let a = f32_node(&mut g, Op::Const, vec![]);
731 let b = f32_node(&mut g, Op::Relu, vec![a]);
732 let c = f32_node(&mut g, Op::Tanh, vec![b]);
733 let bindings = KernelBindingTable::new();
734 let winner = WinnerSelector;
735 let route = pick_route(&g, &[c], &bindings, &winner, None);
736 assert!(route.is_empty(), "no branches ⇒ empty route (picker no-op)");
737 }
738
739 /// Step D: the fork's candidate set is re-enumerated from the runtime
740 /// registry (not a cached plan). With the fork op registered at both arm
741 /// backends, the set carries a real candidate per placement + the fork's
742 /// `DecisionContext` (so the Judge leg keys on the same triple the planner
743 /// used).
744 #[test]
745 fn enumerate_fork_set_pulls_registered_candidates() {
746 let (g, _branch, fork, arm1, _post) = diamond(BackendId::Cuda, BackendId::Cpu);
747 let bindings = diamond_bindings();
748 let set = enumerate_fork_set(&g, fork, &[fork, arm1], &bindings);
749
750 assert_eq!(
751 set.alternatives().len(),
752 2,
753 "Silu registered at both arm backends ⇒ one candidate per placement",
754 );
755 assert!(
756 set.alternatives()
757 .iter()
758 .any(|c| c.backend == BackendId::Cuda)
759 );
760 assert!(
761 set.alternatives()
762 .iter()
763 .any(|c| c.backend == BackendId::Cpu)
764 );
765 let ctx = set.context().expect("fork DecisionContext stamped");
766 assert_eq!(ctx.op, OpKind::SiluElementwise);
767 assert_eq!(ctx.principal_dtype, DType::F32);
768 }
769
770 /// A fork op with no registered binding falls back to placement-only
771 /// synthetic candidates — they still carry each arm's `(backend, device)`,
772 /// so the VRAM-pressure guard still flips to the CPU arm. Preserves the
773 /// prior no-plan-entry behavior.
774 #[test]
775 fn synthetic_fallback_when_fork_op_unregistered() {
776 let (g, branch, _fork, _arm1, post) = diamond(BackendId::Cuda, BackendId::Cpu);
777 let empty = KernelBindingTable::new();
778 let lookup = lookup_for(vec![
779 (BackendId::Cuda, Some(1), Some(10_000)),
780 (BackendId::Cpu, Some(8_000), Some(10_000)),
781 ]);
782 let chained = ChainedSelector::with_default_estimator(None, Some(lookup.clone()));
783 let route = pick_route(&g, &[post], &empty, &chained, Some(&lookup));
784 assert_eq!(
785 route.get(&branch).copied(),
786 Some(1),
787 "no binding ⇒ synthetic placement candidates ⇒ guard still picks CPU",
788 );
789 }
790
791 // ===== Step E Phase C / C2: live-load arm re-pick (the headline gate) =====
792
793 /// Silu bindings at CUDA + Vulkan (the two streaming arms of the
794 /// load-gate diamond). No-op kernels — the picker reads only placement
795 /// metadata for the arm decision.
796 fn diamond_bindings_cuda_vulkan() -> KernelBindingTable {
797 let mut t = KernelBindingTable::new();
798 for backend in [BackendId::Cuda, BackendId::Vulkan] {
799 t.register(
800 OpKind::SiluElementwise,
801 &[DType::F32, DType::F32],
802 backend,
803 noop_kernel,
804 );
805 }
806 t
807 }
808
809 /// THE C2 HEADLINE GATE (selection logic, CPU — no GPU needed). A
810 /// 2-device branched graph (arm-0 on CUDA, arm-1 on Vulkan, BOTH
811 /// VRAM-fit) with a FAKE `BackendStreams` lookup reporting HIGH
812 /// pending_work_count on arm-0's device (CUDA, saturated) and LOW on
813 /// arm-1's (Vulkan, idle): the load-aware `ChainedSelector` re-picks
814 /// the UNLOADED arm-1, and the chosen arm's `target_backend` is the
815 /// unloaded device (Vulkan). Mirrors `vram_pressure_picks_host_ram_arm`
816 /// but with a LOAD signal instead of a memory signal — the live-load
817 /// re-pick that is the whole point of Step E, proven deterministically
818 /// with a fake signal (the real signal is B1, already verified).
819 #[test]
820 fn load_pressure_picks_unloaded_arm() {
821 let (g, branch, _arm0, arm1, post) = diamond(BackendId::Cuda, BackendId::Vulkan);
822 let bindings = diamond_bindings_cuda_vulkan();
823
824 // Both arms VRAM-fit (ample free / total). CUDA saturated (4 in
825 // flight / 1 slot), Vulkan idle (0). The ONLY differentiator is
826 // load — VRAM ties, cost ties (no-op kernels, same static cost),
827 // so the load leg alone must flip the pick to arm-1.
828 let lookup = crate::ranker::mock_combined_lookup(vec![
829 (BackendId::Cuda, Some(10_000), Some(10_000), Some(4), 1),
830 (BackendId::Vulkan, Some(10_000), Some(10_000), Some(0), 1),
831 ]);
832 let chained = ChainedSelector::with_default_estimator(None, Some(lookup.clone()));
833
834 let route = pick_route(&g, &[post], &bindings, &chained, Some(&lookup));
835 assert_eq!(
836 route.get(&branch).copied(),
837 Some(1),
838 "under simulated CUDA saturation the picker takes the UNLOADED \
839 Vulkan arm (arm-1); route={route:?}",
840 );
841 // The chosen arm's node lands on the unloaded device.
842 assert_eq!(
843 g.target_backend(arm1),
844 Some(BackendId::Vulkan),
845 "the picked arm's target_backend is the unloaded device",
846 );
847 }
848
849 /// No-load mirror of the headline gate (the degenerate-fallback): the
850 /// SAME 2-device diamond with BOTH devices reporting 0 in flight ⇒
851 /// every load_tier is 0 ⇒ the route is arm-0 everywhere ⇒ empty route,
852 /// byte-identical to `no_telemetry_picks_arm0_empty_route`. Load
853 /// changes nothing unless devices genuinely contend.
854 #[test]
855 fn no_load_picks_arm0_empty_route() {
856 let (g, _branch, _arm0, _arm1, post) = diamond(BackendId::Cuda, BackendId::Vulkan);
857 let bindings = diamond_bindings_cuda_vulkan();
858
859 // Both VRAM-fit, both idle (0 in flight) → no load signal anywhere.
860 let lookup = crate::ranker::mock_combined_lookup(vec![
861 (BackendId::Cuda, Some(10_000), Some(10_000), Some(0), 1),
862 (BackendId::Vulkan, Some(10_000), Some(10_000), Some(0), 1),
863 ]);
864 let chained = ChainedSelector::with_default_estimator(None, Some(lookup.clone()));
865
866 let route = pick_route(&g, &[post], &bindings, &chained, Some(&lookup));
867 assert!(
868 route.is_empty(),
869 "no load anywhere ⇒ arm-0 everywhere ⇒ empty route; got {route:?}",
870 );
871 }
872}