|
|
@@ -12,16 +12,17 @@ pub mod push;
|
|
|
pub use logs::show_logs;
|
|
|
pub mod build_dir;
|
|
|
pub mod stale;
|
|
|
+mod start;
|
|
|
pub mod stop;
|
|
|
pub mod sync;
|
|
|
pub mod tail;
|
|
|
use crate::{
|
|
|
- config::{Cluster, Config, NcclConfig, Node, Recipe, load_recipe},
|
|
|
+ config::{Cluster, Config, NcclConfig, Recipe, load_recipe},
|
|
|
doctor::try_repaired_image,
|
|
|
ssh::run_ssh,
|
|
|
};
|
|
|
use anyhow::{Context, Result};
|
|
|
-use tokio::time::{Duration, sleep};
|
|
|
+
|
|
|
use tracing::{info, warn};
|
|
|
/// Options for the serve operation.
|
|
|
#[derive(Clone, Debug)]
|
|
|
@@ -284,7 +285,7 @@ pub async fn serve(
|
|
|
opts.dry_run,
|
|
|
)
|
|
|
.await?;
|
|
|
- start_cluster(
|
|
|
+ start::start_cluster(
|
|
|
config,
|
|
|
&cluster,
|
|
|
nccl,
|
|
|
@@ -322,6 +323,7 @@ pub async fn serve(
|
|
|
}
|
|
|
Ok(cluster)
|
|
|
}
|
|
|
+
|
|
|
pub async fn resolve_cluster_by_container_label<'c>(
|
|
|
config: &'c Config,
|
|
|
recipe_name: &str,
|
|
|
@@ -351,7 +353,7 @@ pub async fn resolve_cluster_by_container_label<'c>(
|
|
|
anyhow::bail!("No running containers found for recipe '{}'", recipe_name);
|
|
|
}
|
|
|
/// Options for [start_cluster].
|
|
|
-struct StartClusterOpts {
|
|
|
+pub(crate) struct StartClusterOpts {
|
|
|
cache_dir: std::path::PathBuf,
|
|
|
/// When set, the per-node mirror dir mounted at /root/.cache/huggingface
|
|
|
/// (repos are copied into it from the shared cache before launch).
|
|
|
@@ -362,138 +364,3 @@ struct StartClusterOpts {
|
|
|
dry_run: bool,
|
|
|
clear: bool,
|
|
|
}
|
|
|
-async fn start_cluster(
|
|
|
- config: &Config,
|
|
|
- cluster: &Cluster,
|
|
|
- nccl: Option<NcclConfig>,
|
|
|
- recipe_name: &str,
|
|
|
- recipe: &Recipe,
|
|
|
- opts: StartClusterOpts,
|
|
|
-) -> Result<()> {
|
|
|
- let nodes: Vec<&Node> = if let Some(ref nn) = opts.node_name {
|
|
|
- let v: Vec<&Node> = cluster.nodes.iter().filter(|n| n.name == *nn).collect();
|
|
|
- if v.is_empty() {
|
|
|
- anyhow::bail!("Node '{}' not found in cluster '{}'", nn, cluster.name);
|
|
|
- }
|
|
|
- v
|
|
|
- } else {
|
|
|
- // Trim to `required_nodes` when set: a recipe that needs N nodes
|
|
|
- // must launch on exactly N nodes, not on every free node in the
|
|
|
- // cluster. Previously a `required_nodes = 1` recipe on a 4-node
|
|
|
- // cluster launched 4 containers, each told `--nnodes 4 --tp 4` —
|
|
|
- // a broken rendezvous that never coordinated. `prepare_cluster`
|
|
|
- // guarantees >= required_nodes are free, so taking the first N is
|
|
|
- // safe and deterministic.
|
|
|
- let all = cluster.nodes.iter().collect::<Vec<&Node>>();
|
|
|
- match recipe.required_nodes {
|
|
|
- Some(n) if (n as usize) < all.len() => all.into_iter().take(n as usize).collect(),
|
|
|
- _ => all,
|
|
|
- }
|
|
|
- };
|
|
|
- let nnodes = nodes.len() as u32;
|
|
|
- // `--mirror`: ensure every repo the recipe needs exists in the per-node
|
|
|
- // mirror before launch (copies from the shared HF cache when missing).
|
|
|
- if let Some(ref mirror) = opts.mirror {
|
|
|
- mirror::ensure_mirrors(
|
|
|
- config,
|
|
|
- &nodes,
|
|
|
- recipe,
|
|
|
- &config.resolved_hf_cache_dir(),
|
|
|
- mirror,
|
|
|
- opts.dry_run,
|
|
|
- )
|
|
|
- .await?;
|
|
|
- }
|
|
|
- let pp_size = if recipe.pipeline_parallel_size.unwrap_or(0) == 0
|
|
|
- || recipe.required_nodes == recipe.pipeline_parallel_size
|
|
|
- {
|
|
|
- nnodes
|
|
|
- } else {
|
|
|
- recipe.pipeline_parallel_size.unwrap()
|
|
|
- };
|
|
|
- let tp_size = if recipe.required_nodes == Some(recipe.tensor_parallel_size) {
|
|
|
- nnodes
|
|
|
- } else {
|
|
|
- recipe.tensor_parallel_size
|
|
|
- };
|
|
|
- // Resolve static mounts once (alias/host + tilde). Model path references
|
|
|
- // (`repo`+`subfolder`) are kept as specs and resolved per node inside the
|
|
|
- // spawn loop, since the HF snapshot hash can differ across nodes.
|
|
|
- let resolved = command::resolve_extra_mounts(config, recipe)?;
|
|
|
- command::clear_extra_mounts(config, &nodes, &resolved, opts.clear, opts.dry_run).await?;
|
|
|
- let mut handles = Vec::new();
|
|
|
- for node in &nodes {
|
|
|
- let node = (*node).clone();
|
|
|
- let config = config.clone();
|
|
|
- let rn = recipe_name.to_string();
|
|
|
- let dry = opts.dry_run;
|
|
|
- let engine = recipe.engine.clone();
|
|
|
- let is_head = node.rank == 0;
|
|
|
- let cleanup = format!(
|
|
|
- "docker rm -f {} 2>/dev/null; true",
|
|
|
- shq(&format!("sm121_{}_{}", recipe_name, node.rank))
|
|
|
- );
|
|
|
- let dm_ctx = command::DockerCommandCtx {
|
|
|
- cluster: cluster.clone(),
|
|
|
- nccl: nccl.clone(),
|
|
|
- recipe: recipe.clone(),
|
|
|
- recipe_name: rn.clone(),
|
|
|
- pp_size,
|
|
|
- tp_size,
|
|
|
- nnodes,
|
|
|
- cache_dir: opts.cache_dir.to_string_lossy().into_owned(),
|
|
|
- keep: opts.keep,
|
|
|
- init_port: config.init_port,
|
|
|
- mods_remote_path: opts.mods_remote_path.clone(),
|
|
|
- extra_mounts: resolved.static_mounts.clone(),
|
|
|
- };
|
|
|
- let hf_cache = opts.cache_dir.clone();
|
|
|
- let path_refs = resolved.path_refs.clone();
|
|
|
- handles.push(crate::locale::spawn_with_locale(async move {
|
|
|
- launch_node::launch_node(
|
|
|
- &config,
|
|
|
- launch_node::LaunchNodeCtx {
|
|
|
- node: &node,
|
|
|
- rn: &rn,
|
|
|
- dry,
|
|
|
- engine: &engine,
|
|
|
- is_head,
|
|
|
- cleanup,
|
|
|
- dm_ctx: &dm_ctx,
|
|
|
- hf_cache: &hf_cache,
|
|
|
- path_refs: &path_refs,
|
|
|
- },
|
|
|
- )
|
|
|
- .await
|
|
|
- }));
|
|
|
- }
|
|
|
- // NOTE: all node launches are awaited (join_all) before reporting — a
|
|
|
- // failure on one node does not cancel the others, so a partial cluster
|
|
|
- // may be left running. Intentional: reporting every node's result is
|
|
|
- // more useful than aborting mid-launch and leaving the operator to
|
|
|
- // guess which nodes actually started.
|
|
|
- for result in futures::future::join_all(handles).await {
|
|
|
- match result {
|
|
|
- Ok((name, rank, engine, Ok(_))) => info!(
|
|
|
- "{}",
|
|
|
- fluent_i18n::t!("engine-started", { "engine" => engine.to_uppercase(), "name" => &name, "rank" => rank })
|
|
|
- ),
|
|
|
- Ok((name, rank, engine, Err(e))) => anyhow::bail!(
|
|
|
- "{}",
|
|
|
- fluent_i18n::t!("engine-start-failed", { "engine" => engine.to_uppercase(), "name" => &name, "rank" => rank, "error" => e.to_string() })
|
|
|
- ),
|
|
|
- Err(e) => anyhow::bail!("Task join error: {}", e),
|
|
|
- }
|
|
|
- }
|
|
|
- if !opts.dry_run {
|
|
|
- sleep(Duration::from_secs(10)).await;
|
|
|
- info!(
|
|
|
- "{}",
|
|
|
- fluent_i18n::t!("cluster-started", { "ip" => cluster.effective_master_ip(), "port" => recipe.port })
|
|
|
- );
|
|
|
- }
|
|
|
- Ok(())
|
|
|
-}
|
|
|
-
|
|
|
-#[cfg(test)]
|
|
|
-mod tests;
|