pub mod caps;
pub mod config;
pub mod cost;
pub mod discover;
pub mod placement;
pub mod stats;
pub use caps::{DeviceInfo, NodeCaps, probe_local, probe_remote};
pub use config::{
ClusterConfig, DeviceList, KvPolicy, NodeConfig, NodeRole, Objective, Optimize,
PlacementPolicy, PlacementSection, Precision, UnknownKv,
};
pub use cost::{
BlockKv, CostOptions, KvDetail, KvMarkers, KvProfile, KvSource, Markers, ModelCost, TensorEntry,
};
pub use discover::{DiscoverSection, Discovered, discover_ssh, discover_subnet};
pub use placement::{Assignment, assign_standbys, plan_placement, plan_placement_with, resolve_kv};
pub use rlx_driver::{ProcessGroup, ReduceKind};
pub use stats::{ClusterRun, NodeReport, StageTiming};
use crate::graph::transport::run_pipeline_tcp_timed;
use crate::graph::{NamedTensor, Stage};
use anyhow::{Context, Result};
use std::path::Path;
use std::process::Child;
pub fn all_reduce(pg: &ProcessGroup, buf: &mut [f32], kind: ReduceKind) -> Result<()> {
pg.all_reduce(buf, kind)
.map_err(|e| anyhow::anyhow!("all_reduce: {e:?}"))
}
pub struct Cluster {
pub cfg: ClusterConfig,
pub caps: Vec<NodeCaps>,
pub plan: Vec<Assignment>,
pub chosen_precision: Option<String>,
last_kv: Option<KvProfile>,
}
impl Cluster {
pub fn from_config(cfg: ClusterConfig) -> Self {
Self {
cfg,
caps: Vec::new(),
plan: Vec::new(),
chosen_precision: None,
last_kv: None,
}
}
pub fn from_path(p: impl AsRef<Path>) -> Result<Self> {
Ok(Self::from_config(ClusterConfig::from_path(p)?))
}
pub fn probe(&mut self, remote_bin: &str) -> Result<&[NodeCaps]> {
self.caps.clear();
for n in &self.cfg.nodes {
let caps = match &n.ssh {
Some(host) => probe_remote(host, remote_bin, &n.addr, &n.ckpt_dir)
.with_context(|| format!("probe {host}"))?,
None => probe_local(&n.addr, &n.ckpt_dir, true),
};
self.caps.push(caps);
}
Ok(&self.caps)
}
pub fn discover(&mut self, remote_bin: &str) -> Result<(usize, Vec<(String, String)>)> {
let d = self.cfg.discover.clone();
if d.is_empty() {
return Ok((0, Vec::new()));
}
let mut added = 0;
let mut failed = Vec::new();
if !d.ssh_hosts.is_empty() {
let (found, f) = discover::discover_ssh(&d, remote_bin)?;
added += discover::merge_nodes(&mut self.cfg.nodes, found);
failed.extend(f);
}
if d.subnet.is_some() {
let found = discover::discover_subnet(&d)?;
added += discover::merge_nodes(&mut self.cfg.nodes, found);
}
if self.cfg.nodes.is_empty() {
anyhow::bail!(
"discovery found no usable nodes{}",
if failed.is_empty() {
String::new()
} else {
format!(
" ({} host(s) failed to probe: {})",
failed.len(),
failed
.iter()
.map(|(h, e)| format!("{h}: {e}"))
.collect::<Vec<_>>()
.join("; ")
)
}
);
}
Ok((added, failed))
}
pub fn autoplan(
&mut self,
index: &[TensorEntry],
opts: &CostOptions,
remote_bin: &str,
) -> Result<&[Assignment]> {
if !self.cfg.discover.is_empty() {
let (added, failed) = self.discover(remote_bin)?;
for (h, e) in &failed {
eprintln!("discover: skipping {h}: {e}");
}
if added > 0 {
eprintln!("discover: added {added} node(s)");
}
}
self.probe(remote_bin)?;
let cost = ModelCost::from_tensor_index(index, opts)?;
eprintln!("model: {}", cost.summary());
self.plan(cost)
}
pub fn retire(&mut self, addr: &str, model: ModelCost) -> Result<&[Assignment]> {
use config::NodeRole;
let i = self
.cfg
.nodes
.iter()
.position(|n| n.addr == addr)
.with_context(|| format!("no node {addr} in this cluster"))?;
anyhow::ensure!(
self.cfg.nodes[i].role != NodeRole::Draining,
"node {addr} is already draining"
);
self.cfg.nodes[i].role = NodeRole::Draining;
let promoted = self
.cfg
.nodes
.iter_mut()
.find(|n| n.role == NodeRole::Standby && n.standby_for.as_deref() == Some(addr))
.map(|n| {
n.role = NodeRole::Active;
n.standby_for = None;
n.addr.clone()
});
if let Some(p) = &promoted {
eprintln!("retire {addr}: promoted standby {p} (weights already loaded)");
} else {
eprintln!("retire {addr}: no standby — re-planning across the remaining nodes");
}
self.plan(model)
}
pub fn active_nodes(&self) -> Vec<&NodeConfig> {
self.cfg
.nodes
.iter()
.filter(|n| n.role == config::NodeRole::Active)
.collect()
}
pub fn kv_profile(&self) -> KvProfile {
self.last_kv.clone().unwrap_or_default()
}
pub fn plan_table(&self) -> String {
let mut out =
String::from("node device layers resident disk stage\n");
for a in &self.plan {
out.push_str(&format!(
"{:<20} {:<9} {:>3}..{:<3} {:>7.1}/{:<5.1}G {:>7.1}G {:>7.2}G {:>6.3}s{}\n",
a.addr,
a.device,
a.layers.start,
a.layers.end,
a.est_bytes as f64 / 1e9,
a.budget_bytes as f64 / 1e9,
a.est_disk_bytes as f64 / 1e9,
a.est_kv_bytes as f64 / 1e9,
a.est_stage_secs,
match (a.role, a.backs.as_deref()) {
(config::NodeRole::Standby, Some(b)) => format!(" standby<-{b}"),
(config::NodeRole::Standby, None) => " standby".into(),
(config::NodeRole::Draining, _) => " draining".into(),
_ if a.experts_paged => " paged".into(),
_ => String::new(),
},
));
}
let active_layers: u64 = self
.plan
.iter()
.filter(|a| a.role == config::NodeRole::Active)
.map(|a| a.layers.len() as u64)
.sum();
if active_layers > 0 {
out.push_str(&format!(
"{}\n",
self.kv_profile()
.summary(active_layers, self.cfg.placement.objective.context)
));
}
if let Some(slowest) = self
.plan
.iter()
.filter(|a| a.role == config::NodeRole::Active)
.max_by(|x, y| x.est_stage_secs.total_cmp(&y.est_stage_secs))
{
out.push_str(&format!(
"critical path: {} at {:.3}s/token\n",
slowest.addr, slowest.est_stage_secs
));
}
out
}
pub fn plan(&mut self, model: ModelCost) -> Result<&[Assignment]> {
anyhow::ensure!(!self.caps.is_empty(), "probe() before plan()");
anyhow::ensure!(
self.caps.len() == self.cfg.nodes.len(),
"probed {} node(s) but the config has {} — re-probe after discovery",
self.caps.len(),
self.cfg.nodes.len()
);
let all: Vec<(NodeCaps, NodeConfig)> = self
.caps
.iter()
.cloned()
.zip(self.cfg.nodes.iter().cloned())
.collect();
let active: Vec<(NodeCaps, NodeConfig)> = all
.iter()
.filter(|(_, c)| c.role == config::NodeRole::Active)
.cloned()
.collect();
anyhow::ensure!(
!active.is_empty(),
"no active nodes: {} standby, {} draining",
all.iter()
.filter(|(_, c)| c.role == config::NodeRole::Standby)
.count(),
all.iter()
.filter(|(_, c)| c.role == config::NodeRole::Draining)
.count()
);
let reserve = (self.cfg.reserve_ram_gb * 1e9) as u64;
let obj = self.cfg.placement.objective.clone();
let policy = self.cfg.placement.policy;
let mut attempts: Vec<(Option<String>, ModelCost)> = vec![(None, model.clone())];
for name in &obj.precision_ladder {
match name.bits().and_then(|b| model.at_bits(b)) {
Some(m) => attempts.push((Some(name.to_string()), m)),
None => eprintln!("plan: ignoring precision `{name}` (no fixed width)"),
}
}
let mut last_err = None;
for (name, m) in attempts {
match placement::plan_placement_with(&m, &active, policy, reserve, &obj) {
Ok(mut p) => {
if let Some(n) = &name {
eprintln!(
"plan: does not fit as stored; using precision `{n}` \
({:.1} GB)",
m.total_bytes() as f64 / 1e9
);
self.chosen_precision = Some(n.clone());
}
let kv = placement::resolve_kv(&m, &obj)?;
p.extend(placement::assign_standbys(
&m,
&all,
&p,
reserve,
&kv,
obj.context,
));
self.last_kv = Some(kv);
self.plan = p;
return Ok(&self.plan);
}
Err(e) => last_err = Some(e),
}
}
Err(last_err.unwrap_or_else(|| anyhow::anyhow!("placement failed")))
}
pub fn worker_argv(&self, i: usize) -> Vec<String> {
let a = &self.plan[i];
let n = &self.cfg.nodes[i];
let mut v = vec![
"--role".into(),
"worker".into(),
"--index".into(),
i.to_string(),
"--layers".into(),
format!("{}:{}", a.layers.start, a.layers.end),
"--ckpt".into(),
n.ckpt_dir.clone(),
"--addr".into(),
n.addr.clone(),
"--seq".into(),
self.cfg.seq.to_string(),
"--device".into(),
n.device.to_string(),
"--precision".into(),
n.precision.to_string(),
"--kv".into(),
format!("{:?}", n.kv_cache).to_lowercase(),
"--rng".into(),
self.cfg.seed_for(i).to_string(),
];
if a.first {
v.push("--first".into());
}
if a.last {
v.push("--last".into());
}
v
}
pub fn launch(&self, remote_bin: &str) -> Result<Vec<Child>> {
anyhow::ensure!(!self.plan.is_empty(), "plan() before launch()");
use std::process::Stdio;
let mut kids = Vec::new();
for (i, n) in self.cfg.nodes.iter().enumerate() {
let argv = self.worker_argv(i);
let child = match &n.ssh {
Some(host) => std::process::Command::new("ssh")
.arg(host)
.arg(format!("{remote_bin} {}", argv.join(" ")))
.stdout(Stdio::piped())
.spawn()?,
None => {
let mut c = std::process::Command::new(std::env::current_exe()?);
c.args(&argv).stdout(Stdio::piped());
if let Some(dir) = rlx_ir::env::var("RLX_WORKER_ERR_DIR")
&& let Ok(f) = std::fs::File::create(format!("{dir}/local_worker_{i}.err"))
{
c.stderr(Stdio::from(f));
}
c.spawn()?
}
};
kids.push(child);
}
Ok(kids)
}
pub fn drive(&self, inputs: Vec<NamedTensor>) -> Result<ClusterRun> {
let stages: Vec<Stage> = (0..self.plan.len()).map(|i| self.meta_stage(i)).collect();
let addrs: Vec<String> = self.cfg.nodes.iter().map(|n| n.addr.clone()).collect();
let (out, per_stage_ms) = run_pipeline_tcp_timed(&stages, &addrs, inputs)?;
let timings = self
.plan
.iter()
.zip(per_stage_ms)
.map(|(a, ms)| StageTiming {
addr: a.addr.clone(),
layers: a.layers.clone(),
device: a.device.clone(),
build_ms: 0,
forward_ms: ms,
resident_bytes: a.est_bytes,
})
.collect::<Vec<_>>();
let total: u64 = timings.iter().map(|t| t.forward_ms).sum();
Ok(ClusterRun {
timings,
total_forward_ms: total,
output: out.into_iter().next().map(|t| t.data).unwrap_or_default(),
})
}
fn meta_stage(&self, i: usize) -> Stage {
let (first, last) = (self.plan[i].first, self.plan[i].last);
Stage {
index: i,
graph: rlx_ir::Graph::new("meta"),
inputs: vec![if first { "input_ids" } else { "hidden_in" }.into()],
outputs: vec![if last { "logits" } else { "hidden_in" }.into()],
output_shapes: vec![],
params: vec![],
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use config::NodeRole;
fn caps_for(addr: &str, ram_gb: 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: 1_000_000_000_000,
devices: vec![],
gflops: 100.0,
io_mbps: 2000.0,
}
}
fn node_for(addr: &str, role: NodeRole, backs: Option<&str>) -> NodeConfig {
NodeConfig {
addr: addr.into(),
ssh: None,
ckpt_dir: "/m".into(),
device: "cpu".parse().unwrap(),
precision: "bf16".parse().unwrap(),
kv_cache: Default::default(),
rng_seed: None,
max_ram_gb: None,
layers: None,
role,
standby_for: backs.map(String::from),
}
}
fn dense(n_layers: usize, per_layer: u64) -> ModelCost {
ModelCost {
n_layers,
per_layer_bytes: per_layer,
per_layer_expert_bytes: 0,
per_layer_expert_active_bytes: 0,
embed_bytes: 0,
head_bytes: 0,
per_layer_flops: 1.0,
kv: cost::KvProfile::declared(0),
params: 0,
hidden_size: 0,
}
}
fn cluster(nodes: Vec<NodeConfig>) -> Cluster {
let caps = nodes.iter().map(|n| caps_for(&n.addr, 32.0)).collect();
let cfg = ClusterConfig {
model: "m".into(),
seq: 8,
rng_seed: None,
reserve_ram_gb: 1.0,
placement: Default::default(),
discover: Default::default(),
nodes,
};
let mut c = Cluster::from_config(cfg);
c.caps = caps;
c
}
#[test]
fn plan_covers_actives_and_mirrors_standbys() {
let mut cx = cluster(vec![
node_for("a", NodeRole::Active, None),
node_for("b", NodeRole::Active, None),
node_for("spare", NodeRole::Standby, Some("b")),
]);
let plan = cx.plan(dense(8, 1_000_000_000)).unwrap().to_vec();
let active: Vec<_> = plan.iter().filter(|a| a.role == NodeRole::Active).collect();
let standby: Vec<_> = plan
.iter()
.filter(|a| a.role == NodeRole::Standby)
.collect();
assert_eq!(active.len(), 2);
assert_eq!(standby.len(), 1);
assert_eq!(active[0].layers.start, 0);
assert_eq!(active.last().unwrap().layers.end, 8);
let b = active.iter().find(|a| a.addr == "b").unwrap();
assert_eq!(standby[0].layers, b.layers);
}
#[test]
fn retire_promotes_the_standby_without_moving_layers() {
let mut cx = cluster(vec![
node_for("a", NodeRole::Active, None),
node_for("b", NodeRole::Active, None),
node_for("spare", NodeRole::Standby, Some("b")),
]);
let before = cx.plan(dense(8, 1_000_000_000)).unwrap().to_vec();
let b_layers = before
.iter()
.find(|a| a.addr == "b")
.unwrap()
.layers
.clone();
let after = cx.retire("b", dense(8, 1_000_000_000)).unwrap().to_vec();
assert!(
!after
.iter()
.any(|a| a.addr == "b" && a.role == NodeRole::Active)
);
let promoted = after.iter().find(|a| a.addr == "spare").unwrap();
assert_eq!(promoted.role, NodeRole::Active);
assert_eq!(promoted.layers, b_layers, "promotion must not move layers");
let mut act: Vec<_> = after
.iter()
.filter(|a| a.role == NodeRole::Active)
.collect();
act.sort_by_key(|a| a.layers.start);
assert_eq!(act[0].layers.start, 0);
assert_eq!(act.last().unwrap().layers.end, 8);
}
#[test]
fn retire_without_a_standby_replans_the_rest() {
let mut cx = cluster(vec![
node_for("a", NodeRole::Active, None),
node_for("b", NodeRole::Active, None),
]);
cx.plan(dense(8, 1_000_000_000)).unwrap();
let after = cx.retire("a", dense(8, 1_000_000_000)).unwrap().to_vec();
assert_eq!(after.len(), 1);
assert_eq!(after[0].addr, "b");
assert_eq!(after[0].layers, 0..8);
}
#[test]
fn retiring_the_last_active_node_is_refused() {
let mut cx = cluster(vec![node_for("a", NodeRole::Active, None)]);
cx.plan(dense(4, 1_000_000_000)).unwrap();
let err = cx
.retire("a", dense(4, 1_000_000_000))
.unwrap_err()
.to_string();
assert!(err.contains("no active nodes"), "{err}");
}
#[test]
fn retiring_an_unknown_node_is_an_error() {
let mut cx = cluster(vec![node_for("a", NodeRole::Active, None)]);
cx.caps = vec![caps_for("a", 32.0)];
let err = cx.retire("nope", dense(4, 1)).unwrap_err().to_string();
assert!(err.contains("no node nope"), "{err}");
}
#[test]
fn precision_ladder_steps_down_until_it_fits() {
let mut cx = cluster(vec![node_for("a", NodeRole::Active, None)]);
cx.cfg.placement.objective.precision_ladder = vec!["mxfp4".parse().unwrap()];
let mut model = dense(8, 4_000_000_000);
model.params = model.total_bytes() / 2; let plan = cx.plan(model).unwrap().to_vec();
assert_eq!(plan[0].layers, 0..8);
assert_eq!(cx.chosen_precision.as_deref(), Some("mxfp4"));
}
}