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

Tiny clean router load report logic (#14889)

Browse Source
This commit is contained in:
fzyzcjy
2025-12-11 06:50:32 -08:00
committed by GitHub
parent 60560c07a2
commit ba88f1cace
1 changed files with 17 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sgl-model-gateway/src/routers/http/router.rs
+17 -20
View File
@@ -205,8 +205,7 @@ impl Router {
}; };
let load_incremented = if policy.name() == "cache_aware" { let load_incremented = if policy.name() == "cache_aware" {
worker.increment_load(); increment_load(&worker);
RouterMetrics::set_running_requests(worker.url(), worker.load());
true true
} else { } else {
false false
@@ -246,11 +245,7 @@ impl Router {
// won't have done it (it only decrements on success or non-retryable failures) // won't have done it (it only decrements on success or non-retryable failures)
if is_retryable_status(response.status()) && load_incremented { if is_retryable_status(response.status()) && load_incremented {
if let Some(cleanup_worker) = worker_for_cleanup { if let Some(cleanup_worker) = worker_for_cleanup {
cleanup_worker.decrement_load(); decrement_load(&cleanup_worker);
RouterMetrics::set_running_requests(
cleanup_worker.url(),
cleanup_worker.load(),
);
} }
} }
@@ -518,8 +513,7 @@ impl Router {
// Decrement load on error if it was incremented // Decrement load on error if it was incremented
if load_incremented { if load_incremented {
if let Some(ref w) = worker { if let Some(ref w) = worker {
w.decrement_load(); decrement_load(w);
RouterMetrics::set_running_requests(worker_url, w.load());
} }
} }
@@ -551,8 +545,7 @@ impl Router {
// IMPORTANT: Decrement load on error before returning // IMPORTANT: Decrement load on error before returning
if load_incremented { if load_incremented {
if let Some(ref w) = worker { if let Some(ref w) = worker {
w.decrement_load(); decrement_load(w);
RouterMetrics::set_running_requests(worker_url, w.load());
} }
} }
@@ -564,8 +557,7 @@ impl Router {
// Decrement load counter for non-streaming requests if it was incremented // Decrement load counter for non-streaming requests if it was incremented
if load_incremented { if load_incremented {
if let Some(ref w) = worker { if let Some(ref w) = worker {
w.decrement_load(); decrement_load(w);
RouterMetrics::set_running_requests(worker_url, w.load());
} }
} }
@@ -594,11 +586,7 @@ impl Router {
// Check for stream end marker using memmem for efficiency // Check for stream end marker using memmem for efficiency
if memmem::find(&bytes, b"data: [DONE]").is_some() { if memmem::find(&bytes, b"data: [DONE]").is_some() {
if let Some(ref w) = stream_worker { if let Some(ref w) = stream_worker {
w.decrement_load(); decrement_load(w);
RouterMetrics::set_running_requests(
&worker_url_owned,
w.load(),
);
decremented = true; decremented = true;
} }
} }
@@ -614,8 +602,7 @@ impl Router {
} }
if !decremented { if !decremented {
if let Some(ref w) = stream_worker { if let Some(ref w) = stream_worker {
w.decrement_load(); decrement_load(w);
RouterMetrics::set_running_requests(&worker_url_owned, w.load());
} }
} }
}); });
@@ -685,6 +672,16 @@ impl Router {
} }
} }
fn increment_load(w: &Arc<dyn Worker>) {
w.increment_load();
RouterMetrics::set_running_requests(w.url(), w.load());
}
fn decrement_load(w: &Arc<dyn Worker>) {
w.decrement_load();
RouterMetrics::set_running_requests(w.url(), w.load());
}
use async_trait::async_trait; use async_trait::async_trait;
#[async_trait] #[async_trait]