use super::caps::NodeCaps;
use super::config::{NodeConfig, Objective, Optimize, PlacementPolicy, UnknownKv};
use super::cost::ModelCost;
use anyhow::{Result, bail};
use std::ops::Range;
#[derive(Debug, Clone)]
pub struct Assignment {
pub addr: String,
pub ssh: Option<String>,
pub layers: Range<usize>,
pub first: bool,
pub last: bool,
pub est_bytes: u64,
pub budget_bytes: u64,
pub est_disk_bytes: u64,
pub experts_paged: bool,
pub est_stage_secs: f64,
pub est_kv_bytes: u64,
pub device: String,
pub role: super::config::NodeRole,
pub backs: Option<String>,
}
const GPU_ALLOC_SAFETY: f64 = 0.85;
fn budget_bytes(caps: &NodeCaps, cfg: &NodeConfig, reserve: u64) -> u64 {
let mut b = match cfg.max_ram_gb {
Some(gb) => ((gb * 1e9) as u64).min(caps.ram_total),
None => caps.ram_avail,
};
let dev = cfg.primary_device();
let paged = matches!(dev, rlx_runtime::Device::Cuda);
if dev != rlx_runtime::Device::Cpu && !paged {
let vram = caps.accel_mem();
if vram > 0 && vram < caps.ram_total {
b = b.min((vram as f64 * GPU_ALLOC_SAFETY) as u64);
}
}
b.saturating_sub(reserve)
}
fn layer_cap(budget: u64, per_layer: u64, overhead: u64) -> usize {
if per_layer == 0 {
return usize::MAX;
}
(budget.saturating_sub(overhead) / per_layer) as usize
}
const DISK_SAFETY: f64 = 0.90;
fn experts_fit_in_ram(budget: u64, model: &ModelCost, layers: u64, overhead: u64) -> bool {
let need = layers * (model.per_layer_bytes + model.per_layer_expert_bytes) + overhead;
need <= budget
}
fn disk_cap(caps: &NodeCaps, model: &ModelCost) -> usize {
if model.per_layer_expert_bytes == 0 {
return usize::MAX;
}
let usable = (caps.disk_free as f64 * DISK_SAFETY) as u64;
(usable / model.per_layer_expert_bytes) as usize
}
pub fn resolve_kv(model: &ModelCost, objective: &Objective) -> Result<super::cost::KvProfile> {
use super::cost::KvProfile;
if let Some(b) = objective.kv_bytes_per_layer_token {
return Ok(KvProfile::declared(b));
}
if model.kv.is_known() {
return Ok(model.kv.clone());
}
match objective.on_unknown_kv {
UnknownKv::Ignore => Ok(KvProfile::declared(0)),
UnknownKv::Assume => {
let hidden = model.hidden_size as u64;
if hidden == 0 {
bail!(
"`on_unknown_kv = \"assume\"` needs the model width to bound the \
cache, and the checkpoint did not yield one (no rank-1 norm \
inside a block). Estimating it from byte counts under-reserves \
for quantized weights, so set \
`[placement.objective] kv_bytes_per_layer_token = <bytes>` \
instead."
);
}
let elem = super::cost::kv_elem_bytes(objective.kv_dtype.as_deref().unwrap_or("f32"));
Ok(KvProfile::declared(2 * hidden * elem))
}
UnknownKv::Fail => bail!(
"KV cache size is unknown, so the planner cannot reserve for it at \
context {}. A plan that ignores the cache fits on paper and dies \
when it fills. Fix by one of:\n \
• set `[placement.objective] kv_bytes_per_layer_token = <bytes>`\n \
• set `ModelCost::kv` from the model crate (it knows its head dims)\n \
• `on_unknown_kv = \"assume\"` for a pessimistic MHA-width bound\n \
• `on_unknown_kv = \"ignore\"` if there is genuinely no cache",
objective.context
),
}
}
fn node_capacity(caps: &NodeCaps, model: &ModelCost, budget: u64, overhead: u64) -> (usize, bool) {
let resident = layer_cap(
budget,
model.per_layer_bytes + model.per_layer_expert_bytes,
overhead,
);
if model.per_layer_expert_bytes == 0 {
return (resident, false);
}
let paged = layer_cap(budget, model.per_layer_bytes, overhead).min(disk_cap(caps, model));
if paged > resident {
(paged, true)
} else {
(resident, false)
}
}
fn stage_secs(caps: &NodeCaps, gflops: f64, model: &ModelCost, layers: u64, paged: bool) -> f64 {
let gflops = gflops.max(1.0);
let compute = layers as f64 * model.per_layer_flops / (gflops * 1e9);
if !paged || model.per_layer_expert_active_bytes == 0 {
return compute;
}
let bps = (caps.io_mbps.max(1.0)) * 1e6;
compute + (layers as f64 * model.per_layer_expert_active_bytes as f64) / bps
}
fn balance_critical_path(
cost: &Costing<'_>,
nodes: &[(NodeCaps, NodeConfig)],
caps_layers: &[usize],
counts: &mut [usize],
optimize: Optimize,
) {
let k = counts.len();
if k < 2 {
return;
}
let secs = |i: usize, n: usize| {
if n == 0 {
return 0.0;
}
let (caps, cfg) = &nodes[i];
cost.secs_for(caps, cfg, n, i == 0, i == k - 1)
};
let key = |c: &[usize]| {
let mut v: Vec<f64> = (0..k).map(|i| secs(i, c[i])).collect();
v.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal));
v
};
let better = |a: &[f64], b: &[f64]| -> bool {
for (x, y) in a.iter().zip(b) {
if (x - y).abs() > 1e-9 * y.abs().max(1e-6) {
return x < y;
}
}
false
};
let mut cur = key(counts);
for _ in 0..(counts.iter().sum::<usize>() + 1) * k {
let Some(from) = (0..k).filter(|&i| counts[i] > 0).max_by(|&a, &b| {
secs(a, counts[a])
.partial_cmp(&secs(b, counts[b]))
.unwrap_or(std::cmp::Ordering::Equal)
}) else {
return;
};
let mut best: Option<(usize, Vec<f64>)> = None;
for to in 0..k {
if to == from || counts[to] >= caps_layers[to] {
continue;
}
if optimize == Optimize::Latency && counts[to] == 0 {
continue;
}
counts[from] -= 1;
counts[to] += 1;
let cand = key(counts);
counts[from] += 1;
counts[to] -= 1;
if better(&cand, &cur) && best.as_ref().is_none_or(|(_, b)| better(&cand, b)) {
best = Some((to, cand));
}
}
match best {
Some((to, k2)) => {
counts[from] -= 1;
counts[to] += 1;
cur = k2;
}
None => return,
}
}
}
pub fn plan_placement(
model: &ModelCost,
nodes: &[(NodeCaps, NodeConfig)],
policy: PlacementPolicy,
reserve_bytes: u64,
) -> Result<Vec<Assignment>> {
plan_placement_with(model, nodes, policy, reserve_bytes, &Objective::default())
}
pub fn plan_placement_with(
model: &ModelCost,
nodes: &[(NodeCaps, NodeConfig)],
policy: PlacementPolicy,
reserve_bytes: u64,
objective: &Objective,
) -> Result<Vec<Assignment>> {
let k = nodes.len();
if k == 0 {
bail!("no nodes");
}
if policy == PlacementPolicy::Manual {
let cost = Costing {
model,
kv: &resolve_kv(model, objective)?,
context: objective.context,
reserve: reserve_bytes,
};
let mut out = Vec::new();
for (i, (caps, cfg)) in nodes.iter().enumerate() {
let r = cfg.manual_range().ok_or_else(|| {
anyhow::anyhow!("node {} has no `layers` for manual policy", cfg.addr)
})?;
out.push(cost.assign(caps, cfg, r, i == 0, i == k - 1));
}
return Ok(out);
}
let kv = resolve_kv(model, objective)?;
let fair = (model.n_layers).div_ceil(k).max(1);
let kv_reserve = kv.max_window_bytes(fair, objective.context);
let budgets: Vec<u64> = nodes
.iter()
.map(|(c, cfg)| budget_bytes(c, cfg, reserve_bytes).saturating_sub(kv_reserve))
.collect();
let modes: Vec<(usize, bool)> = (0..k)
.map(|i| {
let overhead = if i == 0 { model.embed_bytes } else { 0 }
+ if i == k - 1 { model.head_bytes } else { 0 };
node_capacity(&nodes[i].0, model, budgets[i], overhead)
})
.collect();
let caps_layers: Vec<usize> = modes.iter().map(|&(c, _)| c).collect();
let total_cap: usize = caps_layers
.iter()
.copied()
.map(|c| c.min(model.n_layers))
.sum();
if total_cap < model.n_layers {
let disk_bound = model.per_layer_expert_bytes > 0
&& (0..k).any(|i| modes[i].1 && disk_cap(&nodes[i].0, model) <= caps_layers[i]);
bail!(
"model ({} layers, {:.1} GB total / {:.1} GB resident) does not fit the \
cluster (capacity {} layers).{} Lower precision, add a node, raise \
max_ram, or reduce reserve.",
model.n_layers,
model.total_bytes() as f64 / 1e9,
model.total_resident_bytes() as f64 / 1e9,
total_cap,
if disk_bound {
" At least one node is bounded by FREE DISK for the routed expert \
banks, not by RAM — free space or move the checkpoint."
} else {
""
}
);
}
let weight: Vec<f64> = match policy {
PlacementPolicy::Throughput => nodes
.iter()
.map(|(c, cfg)| {
c.gflops_for(cfg.primary_device()).max(1.0) * model.per_layer_flops.max(0.001)
})
.collect(),
PlacementPolicy::Auto => (0..k)
.map(|i| {
let (caps, cfg) = &nodes[i];
let t = stage_secs(
caps,
caps.gflops_for(cfg.primary_device()),
model,
1,
modes[i].1,
);
if t > 0.0 { 1.0 / t } else { 1.0 }
})
.collect(),
_ => budgets.iter().map(|&b| b as f64).collect(),
};
let weight: Vec<f64> = if objective.optimize == Optimize::Latency {
let mut left = model.n_layers;
(0..k)
.map(|i| {
if left == 0 {
return 0.0;
}
let take = caps_layers[i].min(left);
left -= take;
take as f64
})
.collect()
} else {
weight
};
let wsum: f64 = weight.iter().sum::<f64>().max(f64::MIN_POSITIVE);
let cost = Costing {
model,
kv: &kv,
context: objective.context,
reserve: reserve_bytes,
};
let mut counts: Vec<usize> = (0..k)
.map(|i| ((model.n_layers as f64 * weight[i] / wsum).round() as usize).min(caps_layers[i]))
.collect();
let mut assigned: usize = counts.iter().sum();
while assigned < model.n_layers {
let i = (0..k)
.filter(|&i| counts[i] < caps_layers[i])
.max_by_key(|&i| caps_layers[i] - counts[i]);
match i {
Some(i) => {
counts[i] += 1;
assigned += 1;
}
None => bail!("placement: could not distribute remaining layers"),
}
}
while assigned > model.n_layers {
let i = (0..k).filter(|&i| counts[i] > 0).max_by_key(|&i| counts[i]);
match i {
Some(i) => {
counts[i] -= 1;
assigned -= 1;
}
None => break,
}
}
if policy == PlacementPolicy::Auto {
balance_critical_path(&cost, nodes, &caps_layers, &mut counts, objective.optimize);
}
let mut out = Vec::with_capacity(k);
let mut start = 0usize;
for (i, ((caps, cfg), &cnt)) in nodes.iter().zip(&counts).enumerate() {
let r = start..start + cnt;
start += cnt;
out.push(cost.assign(caps, cfg, r, i == 0, i == k - 1));
}
Ok(out)
}
struct Costing<'a> {
model: &'a ModelCost,
kv: &'a super::cost::KvProfile,
context: usize,
reserve: u64,
}
impl Costing<'_> {
fn paged_for(
&self,
caps: &NodeCaps,
cfg: &NodeConfig,
n: u64,
first: bool,
last: bool,
) -> bool {
let overhead = if first { self.model.embed_bytes } else { 0 }
+ if last { self.model.head_bytes } else { 0 };
let budget = budget_bytes(caps, cfg, self.reserve);
self.model.per_layer_expert_bytes > 0
&& !experts_fit_in_ram(budget, self.model, n, overhead)
}
fn secs_for(
&self,
caps: &NodeCaps,
cfg: &NodeConfig,
n: usize,
first: bool,
last: bool,
) -> f64 {
let paged = self.paged_for(caps, cfg, n as u64, first, last);
stage_secs(
caps,
caps.gflops_for(cfg.primary_device()),
self.model,
n as u64,
paged,
)
}
fn assign(
&self,
caps: &NodeCaps,
cfg: &NodeConfig,
layers: Range<usize>,
first: bool,
last: bool,
) -> Assignment {
let (model, reserve) = (self.model, self.reserve);
let layers_for_kv = layers.clone();
let n = (layers.end - layers.start) as u64;
let overhead =
if first { model.embed_bytes } else { 0 } + if last { model.head_bytes } else { 0 };
let budget = budget_bytes(caps, cfg, reserve);
let paged = self.paged_for(caps, cfg, n, first, last);
let resident_experts = !paged && model.per_layer_expert_bytes > 0;
let est = n * model.per_layer_bytes
+ overhead
+ if resident_experts {
n * model.per_layer_expert_bytes
} else {
0
};
Assignment {
addr: cfg.addr.clone(),
ssh: cfg.ssh.clone(),
layers,
first,
last,
est_bytes: est,
budget_bytes: budget,
est_disk_bytes: if paged {
n * model.per_layer_expert_bytes
} else {
0
},
experts_paged: paged,
est_stage_secs: stage_secs(
caps,
caps.gflops_for(cfg.primary_device()),
model,
n,
paged,
),
est_kv_bytes: self.kv.bytes_for(layers_for_kv, self.context),
device: rlx_runtime::device_label(cfg.primary_device()).to_string(),
role: cfg.role,
backs: cfg.standby_for.clone(),
}
}
}
pub fn assign_standbys(
model: &ModelCost,
nodes: &[(NodeCaps, NodeConfig)],
active: &[Assignment],
reserve_bytes: u64,
kv: &super::cost::KvProfile,
context: usize,
) -> Vec<Assignment> {
use super::config::NodeRole;
let cost = Costing {
model,
kv,
context,
reserve: reserve_bytes,
};
let mut covered: Vec<String> = Vec::new();
let mut out = Vec::new();
for (caps, cfg) in nodes {
if cfg.role != NodeRole::Standby {
continue;
}
let principal = match &cfg.standby_for {
Some(addr) => active.iter().find(|a| &a.addr == addr),
None => active
.iter()
.filter(|a| !covered.contains(&a.addr))
.max_by_key(|a| a.layers.len())
.or_else(|| active.iter().max_by_key(|a| a.layers.len())),
};
let Some(p) = principal else { continue };
covered.push(p.addr.clone());
let mut a = cost.assign(caps, cfg, p.layers.clone(), p.first, p.last);
a.backs = Some(p.addr.clone());
out.push(a);
}
out
}
#[cfg(test)]
mod tests {
use super::super::cost::KvProfile;
use super::*;
fn caps(addr: &str, ram_gb: f64, gflops: f64) -> NodeCaps {
caps_full(addr, ram_gb, gflops, 500.0, 0.0)
}
fn caps_full(addr: &str, ram_gb: f64, gflops: f64, disk_gb: f64, io_mbps: f64) -> NodeCaps {
NodeCaps {
addr: addr.into(),
os: "linux".into(),
cores: 8,
ram_total: (ram_gb * 1e9) as u64,
ram_avail: (ram_gb * 1e9) as u64,
disk_free: (disk_gb * 1e9) as u64,
devices: vec![],
gflops,
io_mbps,
}
}
fn moe_model(n_layers: usize) -> ModelCost {
ModelCost {
n_layers,
per_layer_bytes: 200_000_000,
per_layer_expert_bytes: 2_000_000_000,
per_layer_expert_active_bytes: 56_000_000,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::declared(0),
params: 0,
hidden_size: 0,
}
}
fn node(addr: &str) -> NodeConfig {
NodeConfig {
addr: addr.into(),
ssh: None,
ckpt_dir: "/x".into(),
device: "cpu".parse().unwrap(),
precision: "bf16".parse().unwrap(),
kv_cache: Default::default(),
rng_seed: None,
max_ram_gb: None,
layers: None,
role: Default::default(),
standby_for: None,
}
}
#[test]
fn ram_balanced_fits_and_covers_all_layers() {
let model = ModelCost {
n_layers: 43,
per_layer_bytes: 3_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 200_000_000,
head_bytes: 200_000_000,
per_layer_flops: 1.0,
kv: KvProfile::declared(0),
params: 0,
hidden_size: 0,
};
let nodes = vec![
(caps("a", 60.0, 100.0), node("a")),
(caps("b", 55.0, 90.0), node("b")),
(caps("c", 44.0, 60.0), node("c")),
];
let plan =
plan_placement(&model, &nodes, PlacementPolicy::RamBalanced, 5_000_000_000).unwrap();
assert_eq!(plan[0].layers.start, 0);
assert_eq!(plan.last().unwrap().layers.end, 43);
for w in plan.windows(2) {
assert_eq!(w[0].layers.end, w[1].layers.start);
}
for a in &plan {
assert!(
a.est_bytes <= a.budget_bytes,
"{} over budget: {} > {}",
a.addr,
a.est_bytes,
a.budget_bytes
);
}
}
#[test]
fn experts_are_budgeted_to_disk_not_ram() {
let model = moe_model(20);
let nodes = vec![(caps_full("a", 16.0, 100.0, 500.0, 2000.0), node("a"))];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
assert_eq!(plan[0].layers, 0..20);
assert!(plan[0].experts_paged, "experts should page from disk");
assert_eq!(plan[0].est_disk_bytes, 20 * 2_000_000_000);
assert_eq!(plan[0].est_bytes, 20 * 200_000_000);
}
#[test]
fn experts_stay_resident_when_ram_allows() {
let model = moe_model(4);
let nodes = vec![(caps_full("a", 64.0, 100.0, 500.0, 2000.0), node("a"))];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
assert!(!plan[0].experts_paged);
assert_eq!(plan[0].est_disk_bytes, 0);
assert_eq!(plan[0].est_bytes, 4 * 2_200_000_000);
}
#[test]
fn auto_gives_more_layers_to_the_faster_disk() {
let model = moe_model(24);
let nodes = vec![
(caps_full("nvme", 16.0, 100.0, 500.0, 6000.0), node("nvme")),
(caps_full("sata", 16.0, 100.0, 500.0, 500.0), node("sata")),
];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
let (fast, slow) = (plan[0].layers.len(), plan[1].layers.len());
assert!(
fast > slow,
"fast-NVMe node got {fast} layers, slow-disk node {slow}"
);
assert_eq!(fast + slow, 24);
let rb =
plan_placement(&model, &nodes, PlacementPolicy::RamBalanced, 2_000_000_000).unwrap();
assert_eq!(rb[0].layers.len(), rb[1].layers.len());
}
#[test]
fn auto_returns_a_plan_no_single_move_improves() {
let model = moe_model(46);
for nodes in [
vec![
(caps_full("a", 34.0, 400.0, 500.0, 700.0), node("a")),
(caps_full("b", 34.0, 400.0, 500.0, 2200.0), node("b")),
],
vec![
(caps_full("a", 24.0, 400.0, 500.0, 1500.0), node("a")),
(caps_full("b", 48.0, 200.0, 500.0, 1500.0), node("b")),
(caps_full("c", 34.0, 900.0, 500.0, 900.0), node("c")),
],
] {
let plan =
plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
let counts: Vec<usize> = plan.iter().map(|a| a.layers.len()).collect();
assert_eq!(counts.iter().sum::<usize>(), 46, "every layer placed");
let k = nodes.len();
let secs = |c: &[usize]| -> Vec<f64> {
let mut v: Vec<f64> = (0..k)
.map(|i| {
let n = c[i];
if n == 0 {
return 0.0;
}
let budget = budget_bytes(&nodes[i].0, &nodes[i].1, 2_000_000_000);
let paged = !experts_fit_in_ram(budget, &model, n as u64, 0);
let g = nodes[i].0.gflops_for(nodes[i].1.primary_device());
stage_secs(&nodes[i].0, g, &model, n as u64, paged)
})
.collect();
v.sort_by(|a, b| b.partial_cmp(a).unwrap());
v
};
let base = secs(&counts);
for from in 0..k {
if counts[from] == 0 {
continue;
}
for to in 0..k {
if to == from {
continue;
}
let mut c = counts.clone();
c[from] -= 1;
c[to] += 1;
let cand = secs(&c);
assert!(
cand >= base,
"moving a layer {} -> {} improves the critical path \
{base:?} -> {cand:?}; the planner left it on the table \
(counts {counts:?})",
nodes[from].1.addr,
nodes[to].1.addr,
);
}
}
}
}
fn benched(label: &str, gflops: f64) -> super::super::caps::DeviceInfo {
super::super::caps::DeviceInfo {
device: label.into(),
name: label.into(),
mem_bytes: 0,
unified: true,
gflops,
available: true,
}
}
#[test]
fn a_node_is_costed_by_the_device_its_stage_uses() {
let model = moe_model(20);
let mut fast_cpu_slow_gpu = caps_full("a", 64.0, 1100.0, 500.0, 1500.0);
fast_cpu_slow_gpu.devices = vec![benched("cpu", 1100.0), benched("metal", 540.0)];
let plain = caps_full("b", 64.0, 1100.0, 500.0, 1500.0);
let mut on_metal = node("a");
on_metal.device = "metal".parse().unwrap();
let nodes = vec![(fast_cpu_slow_gpu, on_metal), (plain, node("b"))];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
assert!(
plan[0].layers.len() < plan[1].layers.len(),
"the Metal node runs at half the CPU node's rate but got {} layers \
against {} — it was priced on its fastest device, not its stage's",
plan[0].layers.len(),
plan[1].layers.len(),
);
}
#[test]
fn auto_splits_identical_nodes_evenly() {
let model = moe_model(46);
let nodes = vec![
(caps_full("a", 34.0, 400.0, 500.0, 1500.0), node("a")),
(caps_full("b", 34.0, 400.0, 500.0, 1500.0), node("b")),
];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
assert_eq!(plan[0].layers.len(), 23);
assert_eq!(plan[1].layers.len(), 23);
}
#[test]
fn reported_mode_matches_the_costed_mode() {
let model = moe_model(46);
let nodes = vec![
(caps_full("a", 34.0, 400.0, 500.0, 700.0), node("a")),
(caps_full("b", 34.0, 400.0, 500.0, 2200.0), node("b")),
];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
for a in &plan {
if a.experts_paged {
assert!(a.est_disk_bytes > 0, "paged stage with no disk cost: {a:?}");
} else {
assert_eq!(a.est_disk_bytes, 0, "resident stage charged disk: {a:?}");
assert!(
a.est_bytes >= a.layers.len() as u64 * model.per_layer_expert_bytes,
"resident stage does not account for its experts: {a:?}"
);
}
}
}
#[test]
fn disk_bound_cluster_reports_disk() {
let model = moe_model(40);
let nodes = vec![(caps_full("a", 16.0, 100.0, 20.0, 6000.0), node("a"))];
let err = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000)
.unwrap_err()
.to_string();
assert!(err.contains("FREE DISK"), "{err}");
}
#[test]
fn dense_models_are_unaffected_by_the_expert_path() {
let model = ModelCost {
n_layers: 8,
per_layer_bytes: 1_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::declared(0),
params: 0,
hidden_size: 0,
};
let nodes = vec![(caps("a", 32.0, 100.0), node("a"))];
let plan = plan_placement(&model, &nodes, PlacementPolicy::Auto, 2_000_000_000).unwrap();
assert_eq!(plan[0].layers, 0..8);
assert!(!plan[0].experts_paged);
assert_eq!(plan[0].est_disk_bytes, 0);
}
fn obj(o: Optimize) -> Objective {
Objective {
optimize: o,
..Default::default()
}
}
#[test]
fn latency_uses_as_few_nodes_as_will_fit() {
let model = ModelCost {
n_layers: 12,
per_layer_bytes: 1_000_000_000,
..moe_model(12)
};
let model = ModelCost {
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
..model
};
let nodes = vec![
(caps("a", 32.0, 100.0), node("a")),
(caps("b", 32.0, 100.0), node("b")),
(caps("c", 32.0, 100.0), node("c")),
];
let lat = plan_placement_with(
&model,
&nodes,
PlacementPolicy::Auto,
2_000_000_000,
&obj(Optimize::Latency),
)
.unwrap();
assert_eq!(lat[0].layers.len(), 12, "first node should take everything");
assert!(
lat[1..].iter().all(|a| a.layers.is_empty()),
"latency should leave the other nodes idle: {:?}",
lat.iter().map(|a| a.layers.len()).collect::<Vec<_>>()
);
let tp = plan_placement_with(
&model,
&nodes,
PlacementPolicy::Auto,
2_000_000_000,
&obj(Optimize::Throughput),
)
.unwrap();
assert!(tp.iter().all(|a| !a.layers.is_empty()));
}
#[test]
fn kv_cache_is_reserved_before_weights() {
let model = ModelCost {
n_layers: 8,
per_layer_bytes: 1_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::declared(1_000_000),
params: 0,
hidden_size: 0,
};
let nodes = vec![(caps("a", 16.0, 100.0), node("a"))];
let short = Objective {
context: 512,
..Default::default()
};
let long = Objective {
context: 32_768,
..Default::default()
};
assert!(
plan_placement_with(&model, &nodes, PlacementPolicy::Auto, 1_000_000_000, &short)
.is_ok(),
"8 GB weights + 4 GB KV fits 16 GB"
);
assert!(
plan_placement_with(&model, &nodes, PlacementPolicy::Auto, 1_000_000_000, &long)
.is_err(),
"at 32k the KV cache alone exceeds the node"
);
}
#[test]
fn standby_mirrors_the_stage_it_backs() {
let model = ModelCost {
n_layers: 8,
per_layer_bytes: 1_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::declared(0),
params: 0,
hidden_size: 0,
};
let mut spare = node("spare");
spare.role = super::super::config::NodeRole::Standby;
spare.standby_for = Some("b".into());
let active = vec![
(caps("a", 32.0, 100.0), node("a")),
(caps("b", 32.0, 100.0), node("b")),
];
let plan = plan_placement(&model, &active, PlacementPolicy::Auto, 1_000_000_000).unwrap();
let mut all = active.clone();
all.push((caps("spare", 32.0, 100.0), spare));
let sb = assign_standbys(
&model,
&all,
&plan,
1_000_000_000,
&KvProfile::declared(0),
4096,
);
assert_eq!(sb.len(), 1);
assert_eq!(sb[0].backs.as_deref(), Some("b"));
let b = plan.iter().find(|a| a.addr == "b").unwrap();
assert_eq!(sb[0].layers, b.layers, "standby must hold the same layers");
assert_eq!(sb[0].est_bytes, b.est_bytes);
}
#[test]
fn unassigned_standby_backs_the_largest_stage() {
let model = ModelCost {
n_layers: 12,
per_layer_bytes: 1_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::declared(0),
params: 0,
hidden_size: 0,
};
let active = vec![
(caps("big", 64.0, 100.0), node("big")),
(caps("small", 16.0, 100.0), node("small")),
];
let plan = plan_placement(&model, &active, PlacementPolicy::Auto, 1_000_000_000).unwrap();
let biggest = plan
.iter()
.max_by_key(|a| a.layers.len())
.unwrap()
.addr
.clone();
let mut spare = node("spare");
spare.role = super::super::config::NodeRole::Standby;
let mut all = active.clone();
all.push((caps("spare", 64.0, 100.0), spare));
let sb = assign_standbys(
&model,
&all,
&plan,
1_000_000_000,
&KvProfile::declared(0),
4096,
);
assert_eq!(sb[0].backs.as_deref(), Some(biggest.as_str()));
}
fn unknown_kv_model() -> ModelCost {
ModelCost {
n_layers: 8,
per_layer_bytes: 1_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::default(), params: 0,
hidden_size: 0,
}
}
#[test]
fn unknown_kv_fails_with_the_ways_out() {
let nodes = vec![(caps("a", 32.0, 100.0), node("a"))];
let err = plan_placement_with(
&unknown_kv_model(),
&nodes,
PlacementPolicy::Auto,
1_000_000_000,
&Objective::default(),
)
.unwrap_err()
.to_string();
assert!(err.contains("KV cache size is unknown"), "{err}");
assert!(err.contains("kv_bytes_per_layer_token"), "{err}");
assert!(err.contains("assume"), "{err}");
assert!(err.contains("ignore"), "{err}");
}
#[test]
fn assume_uses_the_real_width_and_dtype() {
let m = ModelCost {
hidden_size: 4096,
..unknown_kv_model()
};
let f32_obj = Objective {
on_unknown_kv: UnknownKv::Assume,
..Default::default()
};
let kv = resolve_kv(&m, &f32_obj).unwrap();
assert_eq!(kv.bytes_per_layer_token(), 2 * 4096 * 4);
let f16_obj = Objective {
kv_dtype: Some("f16".into()),
..f32_obj.clone()
};
assert_eq!(
resolve_kv(&m, &f16_obj).unwrap().bytes_per_layer_token(),
2 * 4096 * 2
);
}
#[test]
fn assume_refuses_rather_than_inventing_a_width() {
let hidden = 4096u64;
let m = ModelCost {
hidden_size: 0,
per_layer_bytes: 3 * hidden * hidden,
..unknown_kv_model()
};
let old_estimate = ((m.per_layer_bytes as f64 / 12.0).sqrt()) as u64;
assert_eq!(
old_estimate,
hidden / 2,
"the old fallback halved the width"
);
let err = resolve_kv(
&m,
&Objective {
on_unknown_kv: UnknownKv::Assume,
..Default::default()
},
)
.unwrap_err()
.to_string();
assert!(err.contains("needs the model width"), "{err}");
assert!(err.contains("kv_bytes_per_layer_token"), "{err}");
}
#[test]
fn assume_reserves_more_than_ignore() {
let nodes = vec![(caps("a", 32.0, 100.0), node("a"))];
let mk = |u| Objective {
on_unknown_kv: u,
context: 8192,
..Default::default()
};
let ignored = plan_placement_with(
&ModelCost {
hidden_size: 4096,
..unknown_kv_model()
},
&nodes,
PlacementPolicy::Auto,
1_000_000_000,
&mk(UnknownKv::Ignore),
)
.unwrap();
assert_eq!(ignored[0].layers, 0..8);
let assumed = plan_placement_with(
&ModelCost {
hidden_size: 4096,
..unknown_kv_model()
},
&nodes,
PlacementPolicy::Auto,
1_000_000_000,
&mk(UnknownKv::Assume),
);
match assumed {
Ok(p) => assert!(
p[0].layers.len() <= 8,
"assume must not reserve less than ignore"
),
Err(e) => assert!(e.to_string().contains("does not fit"), "{e}"),
}
}
#[test]
fn declared_kv_overrides_unknown() {
let nodes = vec![(caps("a", 32.0, 100.0), node("a"))];
let obj = Objective {
kv_bytes_per_layer_token: Some(1024),
context: 1024,
..Default::default()
};
let p = plan_placement_with(
&unknown_kv_model(),
&nodes,
PlacementPolicy::Auto,
1_000_000_000,
&obj,
)
.unwrap();
assert_eq!(p[0].layers, 0..8);
}
#[test]
fn rejects_when_too_big() {
let model = ModelCost {
n_layers: 100,
per_layer_bytes: 5_000_000_000,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: KvProfile::declared(0),
params: 0,
hidden_size: 0,
};
let nodes = vec![(caps("a", 40.0, 100.0), node("a"))];
assert!(
plan_placement(&model, &nodes, PlacementPolicy::RamBalanced, 5_000_000_000).is_err()
);
}
}