Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
[workspace]
members = [
"lib/llm",
"lib/rl",
"lib/runtime",
"lib/tokenizers",
"lib/renderer",
Expand Down Expand Up @@ -45,6 +46,7 @@ keywords = ["llm", "genai", "inference", "nvidia", "distributed"]
# Local crates
dynamo-runtime = { path = "lib/runtime", version = "1.3.0" }
dynamo-llm = { path = "lib/llm", version = "1.3.0" }
dynamo-rl = { path = "lib/rl", version = "1.3.0" }
dynamo-tokenizers = { path = "lib/tokenizers", version = "1.3.0" }
dynamo-renderer = { path = "lib/renderer", version = "1.3.0" }
dynamo-tokens = { path = "lib/tokens", version = "1.3.0" }
Expand Down
15 changes: 15 additions & 0 deletions lib/bindings/kvbm/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

15 changes: 15 additions & 0 deletions lib/bindings/python/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions lib/llm/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ required-features = ["agent-trace-bench"]
dynamo-kv-router = { workspace = true, features = ["metrics", "runtime-protocols"] }
dynamo-memory = { workspace = true }
dynamo-mocker = { workspace = true }
dynamo-rl = { workspace = true }
dynamo-runtime = { workspace = true }
dynamo-tokenizers = { workspace = true }
dynamo-renderer = { workspace = true }
Expand Down
2 changes: 2 additions & 0 deletions lib/llm/src/entrypoint/input/http.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@ pub async fn run(
// with the instance_id as the router_id label.
http_service_builder =
http_service_builder.drt_discovery(Some(distributed_runtime.discovery()));
http_service_builder =
http_service_builder.runtime(Some(Arc::new(distributed_runtime.clone())));

let http_service = match engine_config {
EngineConfig::Dynamic {
Expand Down
8 changes: 8 additions & 0 deletions lib/llm/src/http/service/openai.rs
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,14 @@ const VALIDATION_PREFIX: &str = "Validation: ";

use super::error::SanitizedError;

pub(super) fn rl_router(
drt: Arc<dynamo_runtime::DistributedRuntime>,
) -> anyhow::Result<axum::Router> {
let config = dynamo_rl::RlDiscoveryConfig::from_env(drt);
let state = dynamo_rl::RlDiscoveryState::new(config);
Ok(dynamo_rl::rl_router(state))
}

// Default axum max body limit without configuring is 2MB: https://docs.rs/axum/latest/axum/extract/struct.DefaultBodyLimit.html
/// Default body limit in bytes (45MB) to support 500k+ token payloads.
/// Can be configured at runtime using the DYN_HTTP_BODY_LIMIT_MB environment variable.
Expand Down
90 changes: 90 additions & 0 deletions lib/llm/src/http/service/service_v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ use crate::request_template::RequestTemplate;
use anyhow::Result;
use axum_server::tls_rustls::RustlsConfig;
use derive_builder::Builder;
use dynamo_runtime::DistributedRuntime;
use dynamo_runtime::config::env_is_truthy;
use dynamo_runtime::config::environment_names::llm as env_llm;
use dynamo_runtime::discovery::Discovery;
Expand Down Expand Up @@ -215,6 +216,9 @@ pub struct HttpService {
tls_cert_path: Option<PathBuf>,
tls_key_path: Option<PathBuf>,
route_docs: Vec<RouteDoc>,
/// RL worker discovery router, served on a dedicated port when enabled.
rl_router: Option<axum::Router>,
rl_port: u16,
}

#[derive(Clone, Builder)]
Expand Down Expand Up @@ -270,6 +274,25 @@ pub struct HttpServiceConfig {
/// are registered using discovery.instance_id() and exposed on /metrics.
#[builder(default = "None")]
drt_discovery: Option<Arc<dyn Discovery>>,

/// When true, serve the RL worker discovery API on `rl_port`.
#[builder(default = "false")]
enable_rl: bool,

/// Port for the RL worker discovery listener. Defaults to `DYN_RL_PORT` or 8001.
#[builder(default = "default_rl_port()")]
rl_port: u16,

/// Distributed runtime used by the RL worker discovery API.
#[builder(default = "None")]
runtime: Option<Arc<DistributedRuntime>>,
}

fn default_rl_port() -> u16 {
std::env::var("DYN_RL_PORT")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(8001)
}

impl HttpService {
Expand Down Expand Up @@ -378,6 +401,8 @@ impl HttpService {
.handle(handle.clone())
.serve(router.into_make_service());

self.spawn_rl_listener_if_configured(&cancel_token).await?;

// Spawn canary after all fallible startup so it won't leak on early errors
tokio::spawn(tokio_metrics_and_canary_loop(cancel_token.clone()));

Expand Down Expand Up @@ -426,6 +451,8 @@ impl HttpService {
}
};

self.spawn_rl_listener_if_configured(&cancel_token).await?;

// Spawn canary after all fallible startup so it won't leak on early errors
tokio::spawn(tokio_metrics_and_canary_loop(cancel_token.clone()));

Expand All @@ -447,6 +474,43 @@ impl HttpService {
Ok(())
}

async fn spawn_rl_listener_if_configured(
&self,
cancel_token: &CancellationToken,
) -> Result<()> {
let Some(rl_router) = self.rl_router.clone() else {
return Ok(());
};
let rl_addr = format!("{}:{}", self.host, self.rl_port);
// Bind eagerly and fail fast: when RL discovery is enabled, a bind failure
// should abort service startup rather than silently leave RL discovery
// unavailable while the main HTTP service keeps running.
let listener = tokio::net::TcpListener::bind(&rl_addr).await.map_err(|e| {
tracing::error!(
address = %rl_addr,
error = %e,
"Failed to bind RL worker discovery listener"
);
anyhow::anyhow!("Failed to bind RL worker discovery listener on {rl_addr}: {e}")
})?;
tracing::info!(
address = %rl_addr,
"RL worker discovery listener started"
);
let rl_cancel = cancel_token.child_token();
tokio::spawn(async move {
if let Err(e) = axum::serve(listener, rl_router)
.with_graceful_shutdown(async move {
rl_cancel.cancelled_owned().await;
})
.await
{
tracing::error!("RL worker discovery listener error: {e}");
}
});
Ok(())
}

/// Documentation of exposed HTTP endpoints
pub fn route_docs(&self) -> &[RouteDoc] {
&self.route_docs
Expand Down Expand Up @@ -631,6 +695,30 @@ impl HttpServiceConfigBuilder {
// Echo x-request-id from request to response headers for client correlation
let router = router.layer(axum::middleware::from_fn(echo_request_id_header));

let enable_rl_router = config.enable_rl || env_is_truthy("DYN_ENABLE_RL");
let rl_router = if enable_rl_router {
let Some(drt) = config.runtime.as_ref() else {
return Err(anyhow::anyhow!(
"RL worker discovery was requested (DYN_ENABLE_RL=true \
or enable_rl) but HttpServiceConfig.runtime is not set."
));
};
let router = super::openai::rl_router(drt.clone())?;
tracing::info!(
rl_port = config.rl_port,
"RL worker discovery enabled at /v1/rl/workers"
);
Some(
router.layer(
TraceLayer::new_for_http()
.make_span_with(make_system_request_span)
.on_response(on_response),
),
)
} else {
None
};

Ok(HttpService {
state,
router,
Expand All @@ -640,6 +728,8 @@ impl HttpServiceConfigBuilder {
tls_cert_path: config.tls_cert_path,
tls_key_path: config.tls_key_path,
route_docs: all_docs,
rl_router,
rl_port: config.rl_port,
})
}

Expand Down
22 changes: 22 additions & 0 deletions lib/rl/Cargo.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

[package]
name = "dynamo-rl"
version.workspace = true
edition.workspace = true
authors.workspace = true
license.workspace = true
homepage.workspace = true
repository.workspace = true
description = "Dynamo RL worker discovery API"

[dependencies]
anyhow = { workspace = true }
axum = { workspace = true }
dynamo-runtime = { workspace = true }
futures = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
Loading
Loading