[router] bugfix: cache_aware in grpc inbalance forward (#15473)
Signed-off-by: Kun(llfl) <i@imux.top>
This commit is contained in:
@@ -8,7 +8,7 @@ use super::PipelineStage;
|
|||||||
use crate::routers::{
|
use crate::routers::{
|
||||||
error,
|
error,
|
||||||
grpc::{
|
grpc::{
|
||||||
context::{ClientSelection, ExecutionResult, RequestContext},
|
context::{ClientSelection, ExecutionResult, LoadGuards, RequestContext, WorkerSelection},
|
||||||
proto_wrapper::{ProtoGenerateRequest, ProtoStream},
|
proto_wrapper::{ProtoGenerateRequest, ProtoStream},
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
@@ -56,6 +56,30 @@ impl PipelineStage for RequestExecutionStage {
|
|||||||
)
|
)
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
|
// Create load guards for worker load tracking (increment load when created)
|
||||||
|
// They will be automatically dropped (and decrement load) when RequestContext is dropped
|
||||||
|
let workers = ctx.state.workers.as_ref().ok_or_else(|| {
|
||||||
|
error!(
|
||||||
|
function = "RequestExecutionStage::execute",
|
||||||
|
"Worker selection not completed"
|
||||||
|
);
|
||||||
|
error::internal_error(
|
||||||
|
"worker_selection_not_completed",
|
||||||
|
"Worker selection not completed",
|
||||||
|
)
|
||||||
|
})?;
|
||||||
|
|
||||||
|
let load_guards = match workers {
|
||||||
|
WorkerSelection::Single { worker } => {
|
||||||
|
LoadGuards::Single(crate::core::WorkerLoadGuardV2::new(worker.clone()))
|
||||||
|
}
|
||||||
|
WorkerSelection::Dual { prefill, decode } => LoadGuards::Dual {
|
||||||
|
prefill: crate::core::WorkerLoadGuardV2::new(prefill.clone()),
|
||||||
|
decode: crate::core::WorkerLoadGuardV2::new(decode.clone()),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
ctx.state.load_guards = Some(load_guards);
|
||||||
|
|
||||||
// Extract dispatch metadata for tracing span
|
// Extract dispatch metadata for tracing span
|
||||||
let request_id = ctx
|
let request_id = ctx
|
||||||
.state
|
.state
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ use super::{
|
|||||||
proto_wrapper::{ProtoGenerateComplete, ProtoGenerateRequest, ProtoStream},
|
proto_wrapper::{ProtoGenerateComplete, ProtoGenerateRequest, ProtoStream},
|
||||||
};
|
};
|
||||||
use crate::{
|
use crate::{
|
||||||
core::Worker,
|
core::{Worker, WorkerLoadGuardV2},
|
||||||
protocols::{
|
protocols::{
|
||||||
chat::{ChatCompletionRequest, ChatCompletionResponse},
|
chat::{ChatCompletionRequest, ChatCompletionResponse},
|
||||||
generate::{GenerateRequest, GenerateResponse},
|
generate::{GenerateRequest, GenerateResponse},
|
||||||
@@ -76,6 +76,9 @@ pub struct ProcessingState {
|
|||||||
// Stage 5: Dispatch metadata
|
// Stage 5: Dispatch metadata
|
||||||
pub dispatch: Option<DispatchMetadata>,
|
pub dispatch: Option<DispatchMetadata>,
|
||||||
|
|
||||||
|
// Load guard for worker load tracking (created at execution stage)
|
||||||
|
pub load_guards: Option<LoadGuards>,
|
||||||
|
|
||||||
// Stage 6: Response processing state
|
// Stage 6: Response processing state
|
||||||
pub response: ResponseState,
|
pub response: ResponseState,
|
||||||
}
|
}
|
||||||
@@ -143,6 +146,16 @@ pub struct DispatchMetadata {
|
|||||||
pub is_streaming: bool,
|
pub is_streaming: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Load guards for worker load tracking
|
||||||
|
/// Automatically decrements load when dropped
|
||||||
|
pub enum LoadGuards {
|
||||||
|
Single(WorkerLoadGuardV2),
|
||||||
|
Dual {
|
||||||
|
prefill: WorkerLoadGuardV2,
|
||||||
|
decode: WorkerLoadGuardV2,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
/// Response processing state (Step 6)
|
/// Response processing state (Step 6)
|
||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
pub struct ResponseState {
|
pub struct ResponseState {
|
||||||
|
|||||||
Reference in New Issue
Block a user