use crate::judge::{Criterion, DispatchOptions, DispatchTable, Pick};
use crate::judge::{Judge, OpKind, ProfileEntry, ProfileReport, SizeClass};
use fuel_graph::{Graph, NodeId, Op};
use fuel_hardware::probe::{HardwareChange, ProbeReport};
use fuel_hardware::transfer_cost::BandwidthMatrix;
use fuel_ir::probe::BackendId;
use fuel_ir::{DType, DeviceLocation, Result};
use std::collections::HashMap;
use std::path::PathBuf;
#[derive(Default)]
pub struct ScheduleOptions {
pub probe_path: Option<PathBuf>,
pub profile_path: Option<PathBuf>,
pub bandwidth_path: Option<PathBuf>,
pub force_rejudge: bool,
pub force_remeasure_bandwidth: bool,
pub judge: Judge,
pub dispatch: DispatchOptions,
}
pub fn prepare_dispatch_table(opts: ScheduleOptions) -> Result<(DispatchTable, ProfileReport)> {
let probe_path = opts
.probe_path
.clone()
.or_else(fuel_hardware::probe::default_report_path);
let profile_path = opts
.profile_path
.clone()
.or_else(crate::judge::default_report_path);
let current_probe = ProbeReport::probe_all();
let mut reuse_profile = false;
if !opts.force_rejudge
&& let Some(pp) = probe_path.as_ref()
&& let Ok(Some(prior)) = ProbeReport::load(pp)
&& matches!(current_probe.diff(&prior), HardwareChange::Unchanged)
{
reuse_profile = true;
}
let profile = if reuse_profile {
if let Some(pp) = profile_path.as_ref() {
let current: Vec<_> = current_probe.equivalence_classes().into_keys().collect();
match ProfileReport::load(pp, ¤t)? {
Some(r) => r,
None => run_and_persist(¤t_probe, &opts, &probe_path, &profile_path)?,
}
} else {
run_and_persist(¤t_probe, &opts, &probe_path, &profile_path)?
}
} else {
run_and_persist(¤t_probe, &opts, &probe_path, &profile_path)?
};
let table = DispatchTable::build_with(&profile, opts.dispatch);
Ok((table, profile))
}
fn op_to_kind(op: &Op) -> Option<OpKind> {
match op {
Op::MatMul => Some(OpKind::MatMul),
Op::Add => Some(OpKind::AddElementwise),
_ => None,
}
}
fn pick_to_location(pick: Pick) -> Option<DeviceLocation> {
match pick.backend {
BackendId::Cpu => Some(DeviceLocation::Cpu),
BackendId::Cuda => Some(DeviceLocation::Cuda {
gpu_id: pick.device_index as usize,
}),
BackendId::Vulkan => Some(DeviceLocation::Vulkan {
gpu_id: pick.device_index as usize,
}),
BackendId::Metal => Some(DeviceLocation::Metal {
gpu_id: pick.device_index as usize,
}),
_ => Some(DeviceLocation::Cpu),
}
}
pub fn recommend_placement(
graph: &Graph,
table: &DispatchTable,
criterion: Criterion,
fallback_device: DeviceLocation,
) -> HashMap<NodeId, DeviceLocation> {
let mut plan = HashMap::with_capacity(graph.len());
for i in 0..graph.len() {
let id = NodeId(i);
let node = graph.node(id);
let device = recommend_for_node(node, table, criterion).unwrap_or(fallback_device);
plan.insert(id, device);
}
plan
}
pub fn recommend_for_node(
node: &fuel_graph::Node,
table: &DispatchTable,
criterion: Criterion,
) -> Option<DeviceLocation> {
let kind = op_to_kind(&node.op)?;
if node.dtype != DType::F32 {
return None;
}
let size_class = SizeClass::from_elem_count(node.shape.elem_count());
let pick = table.pick_nearest(kind, node.dtype, size_class, criterion)?;
pick_to_location(pick)
}
pub fn apply_placement_plan(
graph: &fuel_graph::SharedGraph,
plan: &HashMap<NodeId, DeviceLocation>,
) {
let mut g = graph.write().unwrap();
for (&id, &loc) in plan {
if g.placement(id).is_none() {
g.set_placement(id, loc);
}
}
}
pub fn force_apply_placement_plan(
graph: &fuel_graph::SharedGraph,
plan: &HashMap<NodeId, DeviceLocation>,
) {
let mut g = graph.write().unwrap();
for (&id, &loc) in plan {
g.set_placement(id, loc);
}
}
pub fn auto_place_and_route(
graph: &fuel_graph::SharedGraph,
roots: &[NodeId],
table: &DispatchTable,
criterion: Criterion,
fallback_device: DeviceLocation,
) -> Vec<NodeId> {
let plan = {
let g = graph.read().unwrap();
recommend_placement(&g, table, criterion, fallback_device)
};
apply_placement_plan(graph, &plan);
fuel_graph::opt::insert_copies(graph, roots)
}
fn default_bandwidth_path() -> Option<PathBuf> {
fuel_hardware::probe::default_report_path().and_then(|p| {
p.parent()
.map(|parent| parent.join(fuel_hardware::transfer_cost::BANDWIDTH_REPORT_FILENAME))
})
}
pub fn prepare_dp_inputs(
opts: ScheduleOptions,
) -> Result<(ProbeReport, ProfileReport, BandwidthMatrix)> {
let probe_path = opts
.probe_path
.clone()
.or_else(fuel_hardware::probe::default_report_path);
let profile_path = opts
.profile_path
.clone()
.or_else(crate::judge::default_report_path);
let bandwidth_path = opts.bandwidth_path.clone().or_else(default_bandwidth_path);
let current_probe = ProbeReport::probe_all();
let mut hardware_unchanged = false;
if let Some(pp) = probe_path.as_ref()
&& let Ok(Some(prior)) = ProbeReport::load(pp)
&& matches!(current_probe.diff(&prior), HardwareChange::Unchanged)
{
hardware_unchanged = true;
}
let profile = if hardware_unchanged && !opts.force_rejudge {
if let Some(pp) = profile_path.as_ref() {
let current: Vec<_> = current_probe.equivalence_classes().into_keys().collect();
ProfileReport::load(pp, ¤t)?.unwrap_or_else(|| opts.judge.run(¤t_probe))
} else {
opts.judge.run(¤t_probe)
}
} else {
opts.judge.run(¤t_probe)
};
let bandwidth = if hardware_unchanged && !opts.force_remeasure_bandwidth {
if let Some(bp) = bandwidth_path.as_ref() {
BandwidthMatrix::load(bp)?.unwrap_or_else(|| BandwidthMatrix::measure(¤t_probe))
} else {
BandwidthMatrix::measure(¤t_probe)
}
} else {
BandwidthMatrix::measure(¤t_probe)
};
if let Some(p) = probe_path.as_ref() {
if let Some(parent) = p.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = current_probe.save(p) {
eprintln!("fuel scheduling: failed to persist probe report to {p:?}: {e}");
}
}
if let Some(p) = profile_path.as_ref() {
if let Some(parent) = p.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = profile.save(p) {
eprintln!("fuel scheduling: failed to persist profile report to {p:?}: {e}");
}
}
if let Some(p) = bandwidth_path.as_ref() {
if let Some(parent) = p.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = bandwidth.save(p) {
eprintln!("fuel scheduling: failed to persist bandwidth report to {p:?}: {e}");
}
}
Ok((current_probe, profile, bandwidth))
}
pub fn auto_place_and_route_with_transfer_cost(
graph: &fuel_graph::SharedGraph,
roots: &[NodeId],
opts: ScheduleOptions,
fallback_device: DeviceLocation,
) -> Result<Vec<NodeId>> {
let (probe, profile, bandwidth) = prepare_dp_inputs(opts)?;
let mut backends_seen: std::collections::HashSet<BackendId> = Default::default();
let mut available_backends: Vec<BackendId> = Vec::new();
for d in &probe.devices {
if backends_seen.insert(d.backend) {
available_backends.push(d.backend);
}
}
let plan = {
let g = graph.read().unwrap();
dp_plan(
&g,
roots,
&profile,
&bandwidth,
&available_backends,
fallback_device,
)
};
apply_placement_plan(graph, &plan);
Ok(fuel_graph::opt::insert_copies(graph, roots))
}
pub fn dp_plan(
graph: &fuel_graph::Graph,
roots: &[NodeId],
profile: &ProfileReport,
bandwidth: &fuel_hardware::transfer_cost::BandwidthMatrix,
available_backends: &[BackendId],
fallback_device: DeviceLocation,
) -> HashMap<NodeId, DeviceLocation> {
use fuel_graph::topo_order_multi;
let order = topo_order_multi(graph, roots);
let mut cost: HashMap<(NodeId, BackendId), (f64, Vec<BackendId>)> = HashMap::new();
let backends = if available_backends.is_empty() {
vec![BackendId::Cpu]
} else {
available_backends.to_vec()
};
for &id in &order {
let node = graph.node(id);
let kind = op_to_kind(&node.op);
let size_class = SizeClass::from_elem_count(node.shape.elem_count());
let dtype = node.dtype;
let is_const = matches!(node.op, fuel_graph::Op::Const);
for &b in &backends {
if is_const && b != BackendId::Cpu {
cost.insert((id, b), (f64::INFINITY, vec![]));
continue;
}
let compute = if let Some(k) = kind {
if dtype == DType::F32 {
profile_lookup_latency_ns(profile, k, dtype, size_class, b)
.unwrap_or(UNPROFILED_BACKEND_PENALTY_NS)
} else {
UNPROFILED_BACKEND_PENALTY_NS
}
} else {
UNPROFILED_OP_COMPUTE_NS
};
let mut total_cost = compute;
let mut input_choices = Vec::with_capacity(node.inputs.len());
for &input in &node.inputs {
let input_node = graph.node(input);
let bytes = input_node.shape.elem_count() * dtype_size_bytes(input_node.dtype);
let mut best = f64::INFINITY;
let mut best_b = backends[0];
for &b_i in &backends {
let prior = cost
.get(&(input, b_i))
.map(|(c, _)| *c)
.unwrap_or(f64::INFINITY);
if !prior.is_finite() {
continue;
}
let xfer = transfer_cost_with_cpu_fallback(bandwidth, b_i, b, bytes);
let candidate = prior + xfer;
if candidate < best {
best = candidate;
best_b = b_i;
}
}
if best.is_finite() {
total_cost += best;
input_choices.push(best_b);
} else {
total_cost = f64::INFINITY;
break;
}
}
cost.insert((id, b), (total_cost, input_choices));
}
}
let mut placement: HashMap<NodeId, BackendId> = HashMap::new();
for &root in roots {
let mut best = f64::INFINITY;
let mut best_b = backends[0];
for &b in &backends {
if let Some(&(c, _)) = cost.get(&(root, b)).iter().next().copied()
&& c < best
{
best = c;
best_b = b;
}
}
placement.insert(root, best_b);
}
let mut stack: Vec<NodeId> = roots.to_vec();
while let Some(id) = stack.pop() {
let chosen_b = match placement.get(&id) {
Some(&b) => b,
None => continue,
};
if let Some((_, inputs_chosen)) = cost.get(&(id, chosen_b)).cloned() {
let node = graph.node(id);
for (idx, &input) in node.inputs.iter().enumerate() {
if let Some(&b_i) = inputs_chosen.get(idx)
&& let std::collections::hash_map::Entry::Vacant(e) = placement.entry(input)
{
e.insert(b_i);
stack.push(input);
}
}
}
}
placement
.into_iter()
.map(|(id, b)| {
let pick = crate::judge::Pick {
backend: b,
device_index: 0,
kernel_source: "",
kernel_revision_hash: 0,
};
(id, pick_to_location(pick).unwrap_or(fallback_device))
})
.collect()
}
fn transfer_cost_with_cpu_fallback(
bandwidth: &fuel_hardware::transfer_cost::BandwidthMatrix,
src: BackendId,
dst: BackendId,
bytes: usize,
) -> f64 {
if src == dst {
return 0.0;
}
if let Some(t) = bandwidth.lookup(src, dst) {
return t.ns_per_byte * bytes as f64;
}
let leg1 = bandwidth.lookup(src, BackendId::Cpu).map(|t| t.ns_per_byte);
let leg2 = bandwidth.lookup(BackendId::Cpu, dst).map(|t| t.ns_per_byte);
match (leg1, leg2) {
(Some(a), Some(b)) => (a + b) * bytes as f64,
_ => f64::INFINITY, }
}
fn dtype_size_bytes(d: DType) -> usize {
match d {
DType::F32 | DType::U32 | DType::I32 => 4,
DType::F64 | DType::I64 => 8,
DType::F16 | DType::BF16 | DType::I16 => 2,
DType::U8 | DType::I8 | DType::F8E4M3 | DType::F8E5M2 | DType::F8E8M0 | DType::F8E6M2 => 1,
DType::Bool => 1, DType::F4 => 1, DType::F6E2M3 | DType::F6E3M2 => 1,
}
}
fn profile_lookup_latency_ns(
profile: &ProfileReport,
op: OpKind,
dtype: DType,
size_class: SizeClass,
backend: BackendId,
) -> Option<f64> {
if let Some(e) = profile.entries.iter().find(|e| {
e.op == op && e.dtype == dtype && e.size_class == size_class && e.backend == backend
}) {
return Some(e.latency_ns as f64);
}
let candidates: Vec<&ProfileEntry> = profile
.entries
.iter()
.filter(|e| e.op == op && e.dtype == dtype && e.backend == backend)
.collect();
if candidates.is_empty() {
return None;
}
let target = size_class.0 as i32;
let nearest = candidates
.iter()
.min_by_key(|e| (e.size_class.0 as i32 - target).abs())?;
Some(nearest.latency_ns as f64)
}
const UNPROFILED_BACKEND_PENALTY_NS: f64 = 1_000_000_000.0;
const UNPROFILED_OP_COMPUTE_NS: f64 = 1_000.0;
fn run_and_persist(
probe: &ProbeReport,
opts: &ScheduleOptions,
probe_path: &Option<PathBuf>,
profile_path: &Option<PathBuf>,
) -> Result<ProfileReport> {
let profile = opts.judge.run(probe);
if let Some(p) = probe_path {
if let Some(parent) = p.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = probe.save(p) {
eprintln!("fuel scheduling: failed to persist probe report to {p:?}: {e}");
}
}
if let Some(p) = profile_path {
if let Some(parent) = p.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = profile.save(p) {
eprintln!("fuel scheduling: failed to persist profile report to {p:?}: {e}");
}
}
Ok(profile)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::judge::test_equiv_key;
use crate::judge::{OpKind, OpSize};
#[test]
fn end_to_end_probe_judge_dispatch() {
let scratch =
std::env::temp_dir().join(format!("fuel-schedule-test-{}", std::process::id()));
let _ = std::fs::create_dir_all(&scratch);
let opts = ScheduleOptions {
probe_path: Some(scratch.join("probe.json")),
profile_path: Some(scratch.join("judge.json")),
bandwidth_path: Some(scratch.join("bandwidth.json")),
force_rejudge: true,
force_remeasure_bandwidth: true,
judge: Judge {
iterations: 3,
warmup: 1,
size_plan_override: Some(vec![(
OpKind::MatMul,
OpSize::MatMul {
m: 32,
n: 32,
k: 32,
},
)]),
fixtures: None,
},
dispatch: Default::default(),
};
let (table, profile) = prepare_dispatch_table(opts).expect("prepare");
assert!(
profile.entries.iter().any(|e| e.backend == BackendId::Cpu),
"profile should have at least one cpu entry, got {:?}",
profile.entries
);
assert!(
!table.is_empty(),
"dispatch table should have at least one entry"
);
assert!(scratch.join("probe.json").exists());
assert!(scratch.join("judge.json").exists());
let _ = std::fs::remove_dir_all(&scratch);
}
#[test]
fn reuses_profile_when_hardware_unchanged() {
let scratch =
std::env::temp_dir().join(format!("fuel-schedule-reuse-{}", std::process::id()));
let _ = std::fs::create_dir_all(&scratch);
let tiny_judge = || Judge {
iterations: 3,
warmup: 1,
size_plan_override: Some(vec![(
OpKind::MatMul,
OpSize::MatMul {
m: 16,
n: 16,
k: 16,
},
)]),
fixtures: None,
};
let first = prepare_dispatch_table(ScheduleOptions {
probe_path: Some(scratch.join("probe.json")),
profile_path: Some(scratch.join("judge.json")),
bandwidth_path: Some(scratch.join("bandwidth.json")),
force_rejudge: true,
force_remeasure_bandwidth: true,
judge: tiny_judge(),
dispatch: Default::default(),
})
.expect("first run");
let second = prepare_dispatch_table(ScheduleOptions {
probe_path: Some(scratch.join("probe.json")),
profile_path: Some(scratch.join("judge.json")),
bandwidth_path: Some(scratch.join("bandwidth.json")),
force_rejudge: false, force_remeasure_bandwidth: false,
judge: tiny_judge(),
dispatch: Default::default(),
})
.expect("second run");
assert_eq!(
first.1, second.1,
"hardware unchanged + no force_rejudge should yield identical profile"
);
let _ = std::fs::remove_dir_all(&scratch);
}
use crate::judge::Criterion;
use crate::judge::{PROFILE_REPORT_VERSION, ProfileEntry, ProfileReport};
use fuel_graph::NodeHandle;
use fuel_ir::Shape;
use std::sync::Arc;
#[test]
fn recommend_placement_routes_per_node() {
let mk = |backend: BackendId, size: u32, latency: u64| ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(size),
backend,
device_index: 0,
latency_ns: latency,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(backend),
};
let report = ProfileReport {
version: PROFILE_REPORT_VERSION,
entries: vec![
mk(BackendId::Cpu, 12, 10_000),
mk(BackendId::Cuda, 12, 2_000_000),
mk(BackendId::Cpu, 20, 50_000_000),
mk(BackendId::Cuda, 20, 5_000_000),
],
};
let table = DispatchTable::build(&report);
let small_a = NodeHandle::from_f32(
vec![0.0_f32; 64 * 64],
Shape::from_dims(&[64, 64]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let small_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 64 * 64]),
Shape::from_dims(&[64, 64]),
)
.unwrap();
let small_mm = small_a.matmul(&small_b);
let big_a = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let big_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let big_mm = big_a.matmul(&big_b);
let unprofiled = small_mm.sub(&small_b);
let graph = small_a.graph().read().unwrap();
let plan = recommend_placement(&graph, &table, Criterion::Fastest, DeviceLocation::Cpu);
assert_eq!(plan[&small_mm.id()], DeviceLocation::Cpu);
assert_eq!(plan[&big_mm.id()], DeviceLocation::Cuda { gpu_id: 0 });
assert_eq!(plan[&unprofiled.id()], DeviceLocation::Cpu);
assert!(matches!(plan[&small_a.id()], DeviceLocation::Cpu));
}
#[test]
fn apply_placement_plan_skips_pre_set_hints() {
let entries = vec![
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 50_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 5_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
];
let report = ProfileReport {
version: PROFILE_REPORT_VERSION,
entries,
};
let table = DispatchTable::build(&report);
let a = NodeHandle::from_f32(
vec![0.0_f32; 1024 * 1024],
Shape::from_dims(&[1024, 1024]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let b = a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let mm = a.matmul(&b);
a.graph()
.write()
.unwrap()
.set_placement(mm.id(), DeviceLocation::Cpu);
let plan = recommend_placement(
&a.graph().read().unwrap(),
&table,
Criterion::Fastest,
DeviceLocation::Cpu,
);
assert_eq!(plan[&mm.id()], DeviceLocation::Cuda { gpu_id: 0 });
apply_placement_plan(a.graph(), &plan);
assert_eq!(
a.graph().read().unwrap().placement(mm.id()),
Some(DeviceLocation::Cpu),
"user's explicit hint must not be overwritten by recommend_placement",
);
}
#[test]
fn end_to_end_dp_orchestrator() {
let scratch =
std::env::temp_dir().join(format!("fuel-dp-orch-test-{}", std::process::id()));
let _ = std::fs::create_dir_all(&scratch);
let a = NodeHandle::from_f32(
vec![0.0_f32; 64 * 64],
Shape::from_dims(&[64, 64]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let b = a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 64 * 64]),
Shape::from_dims(&[64, 64]),
)
.unwrap();
let mm = a.matmul(&b);
let opts = ScheduleOptions {
probe_path: Some(scratch.join("probe.json")),
profile_path: Some(scratch.join("judge.json")),
bandwidth_path: Some(scratch.join("bandwidth.json")),
force_rejudge: true,
force_remeasure_bandwidth: true,
judge: Judge {
iterations: 3,
warmup: 1,
size_plan_override: Some(vec![(
OpKind::MatMul,
OpSize::MatMul {
m: 32,
n: 32,
k: 32,
},
)]),
fixtures: None,
},
dispatch: Default::default(),
};
let new_roots = auto_place_and_route_with_transfer_cost(
a.graph(),
&[mm.id()],
opts,
DeviceLocation::Cpu,
)
.expect("orchestrator");
assert!(!new_roots.is_empty());
assert!(scratch.join("probe.json").exists());
assert!(scratch.join("judge.json").exists());
assert!(scratch.join("bandwidth.json").exists());
let _ = std::fs::remove_dir_all(&scratch);
}
#[test]
fn dp_plan_avoids_costly_transfers() {
use fuel_hardware::transfer_cost::{
BANDWIDTH_REPORT_VERSION, BandwidthMatrix, TransferCost,
};
let entries = vec![
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(12),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 10_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(12),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 5_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
];
let report = ProfileReport {
version: PROFILE_REPORT_VERSION,
entries,
};
let _table = DispatchTable::build(&report);
let bandwidth = BandwidthMatrix {
version: BANDWIDTH_REPORT_VERSION,
measurement_bytes: 1 << 24,
entries: vec![
TransferCost {
src: BackendId::Cpu,
dst: BackendId::Cpu,
ns_per_byte: 0.05,
},
TransferCost {
src: BackendId::Cpu,
dst: BackendId::Cuda,
ns_per_byte: 100.0,
},
TransferCost {
src: BackendId::Cuda,
dst: BackendId::Cpu,
ns_per_byte: 100.0,
},
],
};
let small_a = NodeHandle::from_f32(
vec![0.0_f32; 64 * 64],
Shape::from_dims(&[64, 64]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let small_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 64 * 64]),
Shape::from_dims(&[64, 64]),
)
.unwrap();
let mm = small_a.matmul(&small_b);
let plan = dp_plan(
&small_a.graph().read().unwrap(),
&[mm.id()],
&report,
&bandwidth,
&[BackendId::Cpu, BackendId::Cuda],
DeviceLocation::Cpu,
);
assert_eq!(
plan[&mm.id()],
DeviceLocation::Cpu,
"DP planner should pick CPU when transfer cost dominates"
);
}
#[test]
fn dp_plan_picks_dispatch_winner_when_transfer_is_cheap() {
use fuel_hardware::transfer_cost::{
BANDWIDTH_REPORT_VERSION, BandwidthMatrix, TransferCost,
};
let entries = vec![
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 50_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 5_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
];
let report = ProfileReport {
version: PROFILE_REPORT_VERSION,
entries,
};
let _table = DispatchTable::build(&report);
let bandwidth = BandwidthMatrix {
version: BANDWIDTH_REPORT_VERSION,
measurement_bytes: 1 << 24,
entries: vec![
TransferCost {
src: BackendId::Cpu,
dst: BackendId::Cpu,
ns_per_byte: 0.05,
},
TransferCost {
src: BackendId::Cpu,
dst: BackendId::Cuda,
ns_per_byte: 0.15,
},
TransferCost {
src: BackendId::Cuda,
dst: BackendId::Cpu,
ns_per_byte: 0.20,
},
],
};
let big_a = NodeHandle::from_f32(
vec![0.0_f32; 1024 * 1024],
Shape::from_dims(&[1024, 1024]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let big_b = big_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let mm = big_a.matmul(&big_b);
let plan = dp_plan(
&big_a.graph().read().unwrap(),
&[mm.id()],
&report,
&bandwidth,
&[BackendId::Cpu, BackendId::Cuda],
DeviceLocation::Cpu,
);
assert_eq!(plan[&mm.id()], DeviceLocation::Cuda { gpu_id: 0 });
}
#[test]
fn auto_place_and_route_inserts_copies_for_split_graph() {
let entries = vec![
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(12),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 10_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(12),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 2_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 50_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 5_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
];
let report = ProfileReport {
version: PROFILE_REPORT_VERSION,
entries,
};
let table = DispatchTable::build(&report);
let small_a = NodeHandle::from_f32(
vec![0.0_f32; 64 * 64],
Shape::from_dims(&[64, 64]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let small_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 64 * 64]),
Shape::from_dims(&[64, 64]),
)
.unwrap();
let small_mm = small_a.matmul(&small_b);
let big_a = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let big_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let big_mm = big_a.matmul(&big_b);
let n_before = small_a.graph().read().unwrap().len();
let _new_roots = auto_place_and_route(
small_a.graph(),
&[small_mm.id(), big_mm.id()],
&table,
Criterion::Fastest,
DeviceLocation::Cpu,
);
let n_after = small_a.graph().read().unwrap().len();
let g = small_a.graph().read().unwrap();
let mut cuda_copies = 0;
for i in n_before..n_after {
if let fuel_graph::Op::Copy { target } = &g.node(NodeId(i)).op
&& matches!(target, DeviceLocation::Cuda { .. })
{
cuda_copies += 1;
}
}
assert!(
cuda_copies >= 2,
"auto_place_and_route should have inserted ≥2 Copy(_, Cuda) nodes; got {cuda_copies}"
);
}
#[test]
fn apply_then_insert_copies_emits_copies_at_device_boundaries() {
let entries = vec![
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(12),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 10_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(12),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 2_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cpu,
device_index: 0,
latency_ns: 50_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cpu),
},
ProfileEntry {
op: OpKind::MatMul,
dtype: DType::F32,
size_class: SizeClass(20),
backend: BackendId::Cuda,
device_index: 0,
latency_ns: 5_000_000,
iterations: 1,
max_rel_error: 0.0,
kernel_source: String::new(),
kernel_revision_hash: 0,
device: test_equiv_key(BackendId::Cuda),
},
];
let report = ProfileReport {
version: PROFILE_REPORT_VERSION,
entries,
};
let table = DispatchTable::build(&report);
let small_a = NodeHandle::from_f32(
vec![0.0_f32; 64 * 64],
Shape::from_dims(&[64, 64]),
crate::Device::cpu().as_dyn(),
)
.unwrap();
let small_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 64 * 64]),
Shape::from_dims(&[64, 64]),
)
.unwrap();
let small_mm = small_a.matmul(&small_b);
let big_a = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let big_b = small_a
.const_f32_like(
Arc::<[f32]>::from(vec![0.0_f32; 1024 * 1024]),
Shape::from_dims(&[1024, 1024]),
)
.unwrap();
let big_mm = big_a.matmul(&big_b);
let n_before = small_a.graph().read().unwrap().len();
let plan = recommend_placement(
&small_a.graph().read().unwrap(),
&table,
Criterion::Fastest,
DeviceLocation::Cpu,
);
apply_placement_plan(small_a.graph(), &plan);
let roots = vec![small_mm.id(), big_mm.id()];
let _new_roots = fuel_graph::opt::insert_copies(small_a.graph(), &roots);
let g = small_a.graph().read().unwrap();
let n_after = g.len();
let mut cuda_copies = 0;
for i in n_before..n_after {
if let fuel_graph::Op::Copy { target } = &g.node(NodeId(i)).op
&& matches!(target, DeviceLocation::Cuda { .. })
{
cuda_copies += 1;
}
}
assert!(
cuda_copies >= 2,
"expected ≥2 Copy(_, Cuda) nodes for big_a + big_b; got {cuda_copies} \
(n_before={n_before} n_after={n_after})"
);
}
}