minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

config: the readback and the resolving view say what they are (#35027)

Browse Source
This commit is contained in:
Cheng Wan
2026-08-17 16:18:19 -07:00
committed by GitHub
parent cba3c5d5ac
commit c70c7d72a8
12 changed files with 775 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
experimental/sgl-router/src/workers/introspect.rs
+173 -50
View File
@@ -1,23 +1,28 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
// SPDX-License-Identifier: Apache-2.0
//! Single-shot `/server_info` introspection for newly-discovered workers.
//! Introspection for newly-discovered workers.
//!
//! Combines what used to be two separate round-trips (the worker
//! manager's `served_model_name` fetch and `KvEventIndex::add_worker`'s
//! `fetch_event_config`) into one HTTP request. The result is dispatched
//! by the manager: registry consumes `served_model_name`, the optional
//! `KvEventIndex` consumes the resolved `EventConfig`.
//! Two concurrent requests, because the worker answers two different
//! questions on two different endpoints: `/model_info` reports the identity
//! the worker currently serves under (a weight update moves it), while
//! `/server_info` reports its launch configuration — kv-event publisher and
//! disaggregation role. The result is dispatched by the manager: registry
//! consumes `served_model_name`, the optional `KvEventIndex` consumes the
//! resolved `EventConfig`.
//!
//! `served_model_name` is taken from `/model_info`, falling back to
//! `/server_info` for workers that predate the field there.
//!
//! # Failure semantics
//!
//! `fetch` is **infallible** — any error (network, non-2xx, JSON parse,
//! invalid worker URL) is logged at `warn!` and returns an empty
//! `ServerInfo` so the caller can register the worker with empty
//! `model_ids` and no kv-events attachment. Workers that need accuracy
//! around publisher availability use `kv_events::discovery::fetch_event_config`
//! directly (it returns `Result<Option<EventConfig>>`); the manager
//! intentionally doesn't.
//! invalid worker URL) is logged and yields `None` for whatever that request
//! was carrying, so the caller can register the worker with empty `model_ids`
//! and no kv-events attachment. The two requests fail independently. Workers
//! that need accuracy around publisher availability use
//! `kv_events::discovery::fetch_event_config` directly (it returns
//! `Result<Option<EventConfig>>`); the manager intentionally doesn't.
use std::time::Duration;
@@ -68,9 +73,9 @@ pub enum DisaggregationRole {
Decode,
}
/// Performs the single `/server_info` round-trip and projects the
/// response into both halves of `ServerInfo`. Cheap to clone — wraps a
/// `reqwest::Client` (which is internally `Arc`-backed).
/// Performs the two round-trips concurrently and projects the responses into
/// `ServerInfo`. Cheap to clone — wraps a `reqwest::Client` (which is
/// internally `Arc`-backed).
#[derive(Clone)]
pub struct WorkerIntrospector {
client: reqwest::Client,
@@ -95,9 +100,9 @@ impl WorkerIntrospector {
Self { client }
}
/// Fetch `/server_info` for the worker. Never returns an error:
/// any failure is logged at `warn!` and yields a default
/// `ServerInfo` with both halves `None`. Callers register the
/// Fetch `/model_info` and `/server_info` for the worker, concurrently.
/// Never returns an error: any failure is logged at `warn!` and yields
/// `None` for the fields that request carried. Callers register the
/// worker with empty model IDs and no event subscription on the
/// failure path; future re-discovery will retry.
///
@@ -106,24 +111,31 @@ impl WorkerIntrospector {
/// responses and JSON-parse errors short-circuit immediately —
/// the worker answered authoritatively, retrying won't help.
pub async fn fetch(&self, worker_url: &str) -> ServerInfo {
let server_info_url = format!("{}/server_info", worker_url.trim_end_matches('/'));
let parsed = match Self::fetch_with_retry(&self.client, &server_info_url, worker_url).await
{
Some(p) => p,
None => return ServerInfo::default(),
};
let base = worker_url.trim_end_matches('/');
let server_info_url = format!("{base}/server_info");
let model_info_url = format!("{base}/model_info");
let (parsed, model_info) = tokio::join!(
Self::fetch_with_retry::<ServerInfoBody>(&self.client, &server_info_url, worker_url),
Self::fetch_with_retry::<ModelInfoBody>(&self.client, &model_info_url, worker_url),
);
// A worker that answers one endpoint and not the other still gets
// registered with whatever did answer.
let parsed = parsed.unwrap_or_default();
let served_model_name = match parsed.served_model_name {
Some(name) if !name.is_empty() => Some(name),
Some(_) => {
warn!(
worker_url = %worker_url,
"introspect: /server_info has empty `served_model_name`; registering worker with empty model_ids"
);
None
}
None => None,
};
// `/model_info` is the effective identity; `/server_info` is the launch
// record, kept as the fallback for workers that predate the field
// there. An empty string is the same as absent on either.
let non_empty = |name: String| Some(name).filter(|name| !name.is_empty());
let served_model_name = model_info
.and_then(|body| body.served_model_name)
.and_then(non_empty)
.or_else(|| parsed.served_model_name.and_then(non_empty));
if served_model_name.is_none() {
warn!(
worker_url = %worker_url,
"introspect: no `served_model_name` on /model_info or /server_info; registering worker with empty model_ids"
);
}
// EAGLE-family speculative decoding ⇒ the worker hashes KV blocks over
// token bigrams; the router must mirror that on the selection side.
@@ -148,48 +160,53 @@ impl WorkerIntrospector {
}
}
/// Issue the `/server_info` GET with bounded retry on transient
/// errors. Returns `Some(body)` on success, `None` after exhausting
/// retries (the caller falls back to default `ServerInfo`).
async fn fetch_with_retry(
/// Issue one introspection GET with bounded retry on transient errors.
/// Returns `Some(body)` on success, `None` after exhausting retries or on
/// an authoritative answer (4xx, unparsable JSON). The caller decides what
/// a missing body costs — each endpoint carries different fields.
async fn fetch_with_retry<T: serde::de::DeserializeOwned>(
client: &reqwest::Client,
server_info_url: &str,
url: &str,
worker_url: &str,
) -> Option<ServerInfoBody> {
) -> Option<T> {
let mut delay = FETCH_BACKOFF_BASE;
for attempt in 1..=FETCH_MAX_ATTEMPTS {
match client.get(server_info_url).send().await {
match client.get(url).send().await {
Err(e) => {
warn!(
worker_url = %worker_url,
url = %url,
attempt,
error = %e,
"introspect: /server_info request failed; will retry"
"introspect: request failed; will retry"
);
}
Ok(resp) if resp.status().is_server_error() => {
warn!(
worker_url = %worker_url,
url = %url,
attempt,
status = %resp.status(),
"introspect: /server_info returned 5xx; will retry"
"introspect: returned 5xx; will retry"
);
}
Ok(resp) if !resp.status().is_success() => {
warn!(
worker_url = %worker_url,
url = %url,
status = %resp.status(),
"introspect: /server_info returned non-2xx; registering worker with empty model_ids"
"introspect: returned non-2xx; treating the endpoint as absent"
);
return None;
}
Ok(resp) => match resp.json::<ServerInfoBody>().await {
Ok(resp) => match resp.json::<T>().await {
Ok(body) => return Some(body),
Err(e) => {
warn!(
worker_url = %worker_url,
url = %url,
error = %e,
"introspect: /server_info JSON parse failed; registering worker with empty model_ids"
"introspect: JSON parse failed; treating the endpoint as absent"
);
return None;
}
@@ -202,8 +219,9 @@ impl WorkerIntrospector {
}
warn!(
worker_url = %worker_url,
url = %url,
attempts = FETCH_MAX_ATTEMPTS,
"introspect: /server_info failed after retries; registering worker with empty model_ids"
"introspect: failed after retries; treating the endpoint as absent"
);
None
}
@@ -306,11 +324,21 @@ pub(crate) fn resolve_event_config(
}
}
/// Projection of `/model_info` used by the introspector: the identity the
/// worker currently serves under, which a weight update moves.
#[derive(Debug, Default, Deserialize)]
struct ModelInfoBody {
#[serde(default)]
served_model_name: Option<String>,
}
/// Projection of `/server_info` used by the introspector. Every field is
/// `#[serde(default)]` so a worker that exposes only some of them still
/// deserialises; downstream callers handle `None` as "absent".
#[derive(Debug, Default, Deserialize)]
struct ServerInfoBody {
/// The launch record's value, and the fallback for a worker whose
/// `/model_info` predates the field.
#[serde(default)]
served_model_name: Option<String>,
#[serde(default)]
@@ -359,17 +387,36 @@ mod tests {
use tokio::net::TcpListener;
use tokio::sync::oneshot;
/// A worker that serves `/server_info` only: an SGLang that predates
/// `served_model_name` on `/model_info`.
async fn spawn_fake_worker(body: Value) -> (String, oneshot::Sender<()>) {
let body = Arc::new(body);
spawn_fake_worker_with_model_info(body, None).await
}
async fn spawn_fake_worker_with_model_info(
server_info: Value,
model_info: Option<Value>,
) -> (String, oneshot::Sender<()>) {
let body = Arc::new(server_info);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let app = Router::new().route(
let mut app = Router::new().route(
"/server_info",
get(move || {
let body = body.clone();
async move { Json((*body).clone()) }
}),
);
if let Some(model_info) = model_info {
let model_info = Arc::new(model_info);
app = app.route(
"/model_info",
get(move || {
let model_info = model_info.clone();
async move { Json((*model_info).clone()) }
}),
);
}
let (tx, rx) = oneshot::channel::<()>();
tokio::spawn(async move {
let _ = axum::serve(listener, app)
@@ -500,6 +547,82 @@ mod tests {
);
}
#[tokio::test]
async fn fetch_prefers_model_info_over_server_info_for_the_name() {
let (url, _shutdown) = spawn_fake_worker_with_model_info(
json!({"served_model_name": "at-launch"}),
Some(json!({"served_model_name": "after-weight-update"})),
)
.await;
let got = fast_introspector().fetch(&url).await;
assert_eq!(
got.served_model_name.as_deref(),
Some("after-weight-update")
);
}
#[tokio::test]
async fn fetch_falls_back_to_server_info_when_model_info_lacks_the_name() {
let (url, _shutdown) = spawn_fake_worker_with_model_info(
json!({"served_model_name": "at-launch"}),
Some(json!({"model_path": "/models/m"})),
)
.await;
let got = fast_introspector().fetch(&url).await;
assert_eq!(got.served_model_name.as_deref(), Some("at-launch"));
}
/// An empty name on either surface is the same as absent.
#[tokio::test]
async fn fetch_treats_an_empty_name_as_absent_on_both() {
let (url, _shutdown) = spawn_fake_worker_with_model_info(
json!({"served_model_name": "at-launch"}),
Some(json!({"served_model_name": ""})),
)
.await;
assert_eq!(
fast_introspector()
.fetch(&url)
.await
.served_model_name
.as_deref(),
Some("at-launch"),
);
let (url, _shutdown) =
spawn_fake_worker_with_model_info(json!({"served_model_name": ""}), None).await;
assert!(fast_introspector()
.fetch(&url)
.await
.served_model_name
.is_none());
}
#[tokio::test]
async fn fetch_keeps_the_name_when_server_info_is_absent() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let app = Router::new().route(
"/model_info",
get(|| async { Json(json!({"served_model_name": "m"})) }),
);
let (tx, rx) = oneshot::channel::<()>();
tokio::spawn(async move {
let _ = axum::serve(listener, app)
.with_graceful_shutdown(async move {
let _ = rx.await;
})
.await;
});
let _shutdown = tx;
let got = fast_introspector()
.fetch(&format!("http://127.0.0.1:{port}"))
.await;
assert_eq!(got.served_model_name.as_deref(), Some("m"));
assert!(got.event_config.is_none());
assert!(got.disaggregation_role.is_none());
}
#[tokio::test]
async fn fetch_only_served_model_name_when_kv_events_absent() {
let (url, _shutdown) = spawn_fake_worker(json!({"served_model_name": "m"})).await;