|
|
@@ -14,7 +14,6 @@ use std::borrow::Cow;
|
|
|
use std::collections::HashSet;
|
|
|
use tokio::time::{Duration, sleep};
|
|
|
use tracing::info;
|
|
|
-
|
|
|
/// Options for the serve operation.
|
|
|
#[derive(Clone, Debug)]
|
|
|
pub struct ServeOpts {
|
|
|
@@ -23,7 +22,6 @@ pub struct ServeOpts {
|
|
|
pub dry_run: bool,
|
|
|
pub auto_power_on: bool,
|
|
|
}
|
|
|
-
|
|
|
pub(crate) fn shq(s: &str) -> String {
|
|
|
let mut out = String::with_capacity(s.len() + 2);
|
|
|
out.push('\'');
|
|
|
@@ -37,7 +35,6 @@ pub(crate) fn shq(s: &str) -> String {
|
|
|
out.push('\'');
|
|
|
out
|
|
|
}
|
|
|
-
|
|
|
pub async fn serve(
|
|
|
config: &Config,
|
|
|
recipe_name: &str,
|
|
|
@@ -52,7 +49,6 @@ pub async fn serve(
|
|
|
"{}",
|
|
|
fluent_i18n::t!("recipe-starting", { "name" => rn, "engine" => re, "backend" => recipe.distributed_backend.map_or("none".to_string(), |b| b.to_string()) })
|
|
|
);
|
|
|
-
|
|
|
// Check if cluster switch is off and power on if requested
|
|
|
if opts.auto_power_on {
|
|
|
crate::cluster_power::power_on_cluster_if_off(
|
|
|
@@ -64,7 +60,6 @@ pub async fn serve(
|
|
|
)
|
|
|
.await?;
|
|
|
}
|
|
|
-
|
|
|
let (cluster, nccl): (Cluster, NcclConfig) = if let Some(nn) = node_name {
|
|
|
let (solo_c, solo_n) = config
|
|
|
.clusters
|
|
|
@@ -86,6 +81,11 @@ pub async fn serve(
|
|
|
socket_ifname: "lo".to_string(),
|
|
|
ib_gid_index: 3,
|
|
|
ib_disable: 1,
|
|
|
+ gloo_socket_ifname: None,
|
|
|
+ ucx_net_devices: None,
|
|
|
+ nccl_cross_nic: None,
|
|
|
+ nccl_proto: None,
|
|
|
+ nccl_p2p_level: None,
|
|
|
});
|
|
|
(solo, nc)
|
|
|
})
|
|
|
@@ -124,7 +124,6 @@ pub async fn serve(
|
|
|
opts.dry_run,
|
|
|
)
|
|
|
.await?;
|
|
|
-
|
|
|
let mods_remote_path = mods::upload_mods_to_cluster(
|
|
|
config,
|
|
|
&cluster,
|
|
|
@@ -134,7 +133,6 @@ pub async fn serve(
|
|
|
opts.dry_run,
|
|
|
)
|
|
|
.await?;
|
|
|
-
|
|
|
let image = try_repaired_image(config, &cluster, &recipe).await?;
|
|
|
let mut recipe = recipe;
|
|
|
if image != recipe.docker_image {
|
|
|
@@ -204,7 +202,6 @@ pub async fn stop_cluster(
|
|
|
info!("{}", fluent_i18n::t!("cluster-all-stopped"));
|
|
|
Ok(())
|
|
|
}
|
|
|
-
|
|
|
/// Find the cluster that has running containers for a recipe by reading
|
|
|
/// the `sm121.cluster` label from live containers. Used by stop/logs
|
|
|
/// when `--cluster auto`.
|
|
|
@@ -236,7 +233,6 @@ pub async fn resolve_cluster_by_container_label<'c>(
|
|
|
}
|
|
|
anyhow::bail!("No running containers found for recipe '{}'", recipe_name);
|
|
|
}
|
|
|
-
|
|
|
/// Query running containers across all unique nodes and return the set of
|
|
|
/// occupied node names. A node is occupied if any sm121 container is
|
|
|
/// running on it.
|
|
|
@@ -269,7 +265,6 @@ pub async fn occupied_nodes(config: &Config) -> Result<HashSet<String>> {
|
|
|
}
|
|
|
Ok(occupied)
|
|
|
}
|
|
|
-
|
|
|
/// Find the cluster that has running containers for a recipe by reading
|
|
|
/// the `sm121.cluster` label from live containers. Used by stop/logs
|
|
|
/// when `--cluster auto`.
|
|
|
@@ -280,7 +275,6 @@ struct StartClusterOpts {
|
|
|
mods_remote_path: String,
|
|
|
dry_run: bool,
|
|
|
}
|
|
|
-
|
|
|
/// Shared context for [`build_docker_command`] — constructed once, cloned per node.
|
|
|
#[derive(Clone)]
|
|
|
struct DockerCommandCtx {
|
|
|
@@ -296,7 +290,6 @@ struct DockerCommandCtx {
|
|
|
init_port: u16,
|
|
|
mods_remote_path: String,
|
|
|
}
|
|
|
-
|
|
|
async fn start_cluster(
|
|
|
config: &Config,
|
|
|
cluster: &Cluster,
|
|
|
@@ -406,7 +399,6 @@ fn build_docker_command(ctx: &DockerCommandCtx, node: &Node, is_head: bool) -> V
|
|
|
if !ctx.mods_remote_path.is_empty() && !ctx.recipe.mods.commands.is_empty() {
|
|
|
cmd.push(format!("-v {}:/mods:ro", ctx.mods_remote_path));
|
|
|
}
|
|
|
-
|
|
|
let mut env = vec![("HF_HOME", Cow::Borrowed("/root/.cache/huggingface"))];
|
|
|
if ctx.nnodes > 1 {
|
|
|
let worker_ip = Cow::Borrowed(node.ib_ip.as_str());
|
|
|
@@ -415,25 +407,37 @@ fn build_docker_command(ctx: &DockerCommandCtx, node: &Node, is_head: bool) -> V
|
|
|
} else {
|
|
|
env.push(("VLLM_HOST_IP", worker_ip));
|
|
|
}
|
|
|
- let nccl_socket_ifname = ctx.nccl.socket_ifname.as_str();
|
|
|
- env.extend([
|
|
|
+ let sock = ctx.nccl.socket_ifname.as_str();
|
|
|
+ let gloo_if = ctx.nccl.gloo_socket_ifname.as_deref().unwrap_or(sock);
|
|
|
+ let ucx_dev = ctx.nccl.ucx_net_devices.as_deref().unwrap_or(sock);
|
|
|
+ let mut ncclext: Vec<(&str, Cow<str>)> = vec![
|
|
|
("NCCL_IB_DISABLE", format!("{}", ctx.nccl.ib_disable).into()),
|
|
|
("NCCL_IB_HCA", ctx.nccl.ib_hca.as_str().into()),
|
|
|
- ("NCCL_SOCKET_IFNAME", nccl_socket_ifname.into()),
|
|
|
- ("GLOO_SOCKET_IFNAME", nccl_socket_ifname.into()),
|
|
|
- ("TP_SOCKET_IFNAME", nccl_socket_ifname.into()),
|
|
|
+ ("NCCL_SOCKET_IFNAME", sock.into()),
|
|
|
+ ("GLOO_SOCKET_IFNAME", gloo_if.into()),
|
|
|
+ ("TP_SOCKET_IFNAME", gloo_if.into()),
|
|
|
(
|
|
|
"NCCL_IB_GID_INDEX",
|
|
|
ctx.nccl.ib_gid_index.to_string().into(),
|
|
|
),
|
|
|
- ("MN_IF_NAME", nccl_socket_ifname.into()),
|
|
|
- ("UCX_NET_DEVICES", nccl_socket_ifname.into()),
|
|
|
- ("OMPI_MCA_btl_tcp_if_include", nccl_socket_ifname.into()),
|
|
|
+ ("MN_IF_NAME", gloo_if.into()),
|
|
|
+ ("UCX_NET_DEVICES", ucx_dev.into()),
|
|
|
+ ("OMPI_MCA_btl_tcp_if_include", gloo_if.into()),
|
|
|
("OMP_NUM_THREADS", "4".into()),
|
|
|
("NCCL_IGNORE_CPU_AFFINITY", "1".into()),
|
|
|
("NCCL_NET", "IB".into()),
|
|
|
("NCCL_DEBUG", "WARN".into()),
|
|
|
- ]);
|
|
|
+ ];
|
|
|
+ if let Some(v) = ctx.nccl.nccl_cross_nic {
|
|
|
+ ncclext.push(("NCCL_CROSS_NIC", v.to_string().into()));
|
|
|
+ }
|
|
|
+ if let Some(ref v) = ctx.nccl.nccl_proto {
|
|
|
+ ncclext.push(("NCCL_PROTO", v.as_str().into()));
|
|
|
+ }
|
|
|
+ if let Some(ref v) = ctx.nccl.nccl_p2p_level {
|
|
|
+ ncclext.push(("NCCL_P2P_LEVEL", v.as_str().into()));
|
|
|
+ }
|
|
|
+ env.extend(ncclext);
|
|
|
}
|
|
|
if is_head {
|
|
|
env.push(("VLLM_LOGGING_COLOR", "1".into()));
|
|
|
@@ -462,7 +466,6 @@ fn build_docker_command(ctx: &DockerCommandCtx, node: &Node, is_head: bool) -> V
|
|
|
for (k, v) in env {
|
|
|
cmd.push(format!("-e {k}={v}"));
|
|
|
}
|
|
|
-
|
|
|
let tctx = tail::TailCtx {
|
|
|
recipe: &ctx.recipe,
|
|
|
tp_size: ctx.tp_size,
|
|
|
@@ -486,6 +489,5 @@ fn build_docker_command(ctx: &DockerCommandCtx, node: &Node, is_head: bool) -> V
|
|
|
}
|
|
|
cmd
|
|
|
}
|
|
|
-
|
|
|
#[cfg(test)]
|
|
|
mod tests;
|