use anyhow::{Context, Result}; use std::time::Duration; use tokio::time::sleep; use tracing::{info, warn}; use crate::config::{Config, Recipe}; use crate::tmux::TmuxManager; /// Launches a cluster via tmux sessions. /// For Ray backend: starts head node first, waits for readiness, then workers. pub async fn tmux_serve( config: &Config, recipe_name: &str, cluster_name: &str, skip_before_serve: bool, dry_run: bool, ) -> Result<()> { let recipe_path = format!("recipes/{}/recipe.toml", recipe_name); let recipe: Recipe = toml::from_str(&tokio::fs::read_to_string(&recipe_path).await?)?; info!( "🚀 TMUX launch: {} ({}) [backend: {}] [engine: {}]", recipe.name, recipe_name, recipe .distributed_backend .map_or("none".into(), |b| b.to_string()), recipe.engine ); let choice = if cluster_name != "auto" { cluster_name.to_string() } else if let Some(ref rc) = recipe.cluster { rc.clone() } else { "auto".to_string() }; let (cluster, nccl) = crate::config::select_cluster(config, &choice, recipe.required_nodes)?; if let Some(required) = recipe.required_nodes { let actual = cluster.nodes.len() as u32; if actual != required { anyhow::bail!( "Recipe '{}' requires {} nodes, but cluster '{}' has {} nodes", recipe.name, required, cluster.name, actual ); } } crate::before_serve::run(config, &cluster, &recipe, skip_before_serve, dry_run).await?; let manager = TmuxManager { session_prefix: recipe_name.to_string(), }; let is_ray = recipe.distributed_backend == Some(crate::config::DistributedBackend::Ray); if is_ray { info!("🐍 Ray backend detected — starting head node first..."); let head_node = cluster .nodes .iter() .find(|n| n.rank == 0) .context("No head node (rank 0) in cluster")?; let head_cmds = manager.build_session_commands(head_node, &recipe, nccl, &cluster); manager .create_session(config, head_node, &head_cmds, dry_run) .await?; if !dry_run { info!("⏳ Waiting for head node to be ready..."); let sess = manager.session_name(&recipe, 0); let head_ready = wait_for_head_ready(config, head_node, &sess, &recipe).await?; if head_ready { info!("✅ Head node ready. Starting worker nodes..."); } else { warn!("⚠️ Head node may not be fully ready — starting workers anyway."); } } for node in &cluster.nodes { if node.rank == 0 { continue; } let worker_cmds = manager.build_session_commands(node, &recipe, nccl, &cluster); manager .create_session(config, node, &worker_cmds, dry_run) .await?; } } else { info!("🔗 MP backend detected — starting all nodes in parallel..."); let mut handles = Vec::new(); for node in &cluster.nodes { let config = config.clone(); let node = node.clone(); let recipe = recipe.clone(); let nccl = nccl.clone(); let cluster = cluster.clone(); let manager = manager.clone(); let dry = dry_run; handles.push(tokio::spawn(async move { let cmds = manager.build_session_commands(&node, &recipe, &nccl, &cluster); manager .create_session(&config, &node, &cmds, dry) .await .map(|_| node.name.clone()) })); } let results = futures::future::join_all(handles).await; let mut failures = Vec::new(); for result in results { match result { Ok(Ok(name)) => info!(" ✅ Session started on {}", name), Ok(Err(e)) => failures.push(e), Err(e) => failures.push(anyhow::anyhow!("Task join: {}", e)), } } if !failures.is_empty() { for e in &failures { warn!(" ✗ {}", e); } anyhow::bail!("Failed to start sessions on {} node(s)", failures.len()); } } if !dry_run { sleep(Duration::from_secs(5)).await; info!( "🎉 Cluster launched via tmux! API → http://{}:{}", cluster.master_ib_ip, recipe.port ); info!( "📌 To view sessions: `cargo run tmux status --recipe {}`", recipe_name ); info!( "📌 To attach: `cargo run tmux attach --recipe {}`", recipe_name ); } Ok(()) } /// Polls the head node until the API endpoint responds or timeout. async fn wait_for_head_ready( config: &Config, head_node: &crate::config::Node, session_name: &str, recipe: &Recipe, ) -> Result { let max_attempts = 30; for i in 0..max_attempts { let pid_cmd = format!( "tmux list-panes -t {}:0 -F '#{{pane_pid}}' | head -1", session_name ); match crate::ssh::run_ssh(config, head_node, pid_cmd, false).await { Ok(pid_str) => { let pid = pid_str.trim(); if pid.is_empty() || pid == "0" { sleep(Duration::from_secs(5)).await; continue; } } Err(_) => { sleep(Duration::from_secs(5)).await; continue; } } let api_check = format!( "curl -s -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{}/v1/models 2>/dev/null || echo 'not_ready'", recipe.port ); if let Ok(out) = crate::ssh::run_ssh(config, head_node, api_check, false).await { let code = out.trim(); if code == "200" { info!(" Head node API ready (attempt {})", i + 1); return Ok(true); } } if (i + 1) % 5 == 0 { info!( " Still waiting for head node... ({}/{})", i + 1, max_attempts ); } sleep(Duration::from_secs(5)).await; } warn!(" Head node did not become ready within timeout"); Ok(false) }