From 64833f8158236a7bddeeb89efc6a3bde5d16f468 Mon Sep 17 00:00:00 2001 From: Sahil Singh Date: Mon, 15 Jun 2026 12:21:24 +0530 Subject: [PATCH] =?UTF-8?q?[Rust=20Frontend]=20Add=20external=E2=86=92inte?= =?UTF-8?q?rnal=20request-id=20map=20for=20abort()=20(#45137)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Sahil Singh --- rust/Cargo.lock | 1 + rust/src/chat/src/lib.rs | 6 + rust/src/engine-core-client/src/client.rs | 4 + rust/src/engine-core-client/src/client/imp.rs | 15 ++ .../engine-core-client/src/client/state.rs | 30 ++- .../engine-core-client/src/tests/client.rs | 4 +- rust/src/llm/Cargo.toml | 1 + rust/src/llm/src/inflight.rs | 179 ++++++++++++++++++ rust/src/llm/src/lib.rs | 36 +++- rust/src/llm/src/output.rs | 12 +- rust/src/llm/src/request_metrics.rs | 44 +++-- rust/src/llm/tests/generate.rs | 136 +++++++++++++ rust/src/text/src/lib.rs | 6 + 13 files changed, 447 insertions(+), 27 deletions(-) create mode 100644 rust/src/llm/src/inflight.rs diff --git a/rust/Cargo.lock b/rust/Cargo.lock index c1477092b9..0369dc8d94 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -5804,6 +5804,7 @@ dependencies = [ "enum-as-inner", "expect-test", "futures", + "parking_lot", "rmp-serde", "serde", "serde_json", diff --git a/rust/src/chat/src/lib.rs b/rust/src/chat/src/lib.rs index 130d4c9f46..1148560787 100644 --- a/rust/src/chat/src/lib.rs +++ b/rust/src/chat/src/lib.rs @@ -235,6 +235,12 @@ impl ChatLlm { Ok(token_ids) } + /// Abort in-flight requests by their external (user-supplied) request ids. + pub async fn abort(&self, external_ids: &[String]) -> Result<()> { + self.text.abort(external_ids).await?; + Ok(()) + } + /// Shut down the underlying LLM client and its background tasks. pub async fn shutdown(self) -> Result<()> { self.text.shutdown().await?; diff --git a/rust/src/engine-core-client/src/client.rs b/rust/src/engine-core-client/src/client.rs index 73ebe9ef40..c646de567d 100644 --- a/rust/src/engine-core-client/src/client.rs +++ b/rust/src/engine-core-client/src/client.rs @@ -490,6 +490,10 @@ impl EngineCoreClient { return Ok(()); } + // Finalize the consumer streams first, before the engine round-trip. + let all_request_ids: Vec = abortable.values().flatten().cloned().collect(); + self.inner.abort_requests_locally(&all_request_ids); + for (engine_id, request_ids) in abortable { self.inner.do_abort_requests(&engine_id, &request_ids).await?; } diff --git a/rust/src/engine-core-client/src/client/imp.rs b/rust/src/engine-core-client/src/client/imp.rs index 6f218717ed..1107e415d3 100644 --- a/rust/src/engine-core-client/src/client/imp.rs +++ b/rust/src/engine-core-client/src/client/imp.rs @@ -1,5 +1,6 @@ use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; use arc_swap::ArcSwapOption; use parking_lot::Mutex; @@ -126,6 +127,20 @@ impl ClientInner { self.request_reg.lock().finish_many(request_ids) } + /// Finalize client-initiated aborts by pushing a terminal `Abort` output + /// down each request's stream and removing it from the registry. Returns + /// the request ids that were still active. See [`RequestRegistry::abort_many`]. + pub fn abort_requests_locally<'a>( + &self, + request_ids: impl IntoIterator, + ) -> Vec { + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs_f64()) + .unwrap_or(0.0); + self.request_reg.lock().abort_many(request_ids, timestamp) + } + /// Apply one scheduler stats update for the given engine to the local /// routing state. Returns `false` if the engine is unknown to the /// client. diff --git a/rust/src/engine-core-client/src/client/state.rs b/rust/src/engine-core-client/src/client/state.rs index 062f284d90..51da1c10f6 100644 --- a/rust/src/engine-core-client/src/client/state.rs +++ b/rust/src/engine-core-client/src/client/state.rs @@ -9,7 +9,7 @@ use crate::client::stream::EngineCoreStreamOutput; use crate::error::{Error, Result}; use crate::protocol::stats::SchedulerStats; use crate::protocol::utility::UtilityOutput; -use crate::protocol::{EngineCoreEventType, EngineCoreOutput}; +use crate::protocol::{EngineCoreEventType, EngineCoreFinishReason, EngineCoreOutput}; use crate::transport::ConnectedEngine; pub type OutputSender = mpsc::UnboundedSender>; @@ -289,6 +289,34 @@ impl RequestRegistry { .collect() } + /// Finalize client-initiated aborts: remove each request and push a + /// terminal output with `finish_reason = Abort` down its stream before the + /// sender drops. Returns the request ids that were still active. + pub fn abort_many<'a>( + &mut self, + request_ids: impl IntoIterator, + timestamp: f64, + ) -> Vec { + let mut aborted = Vec::new(); + for request_id in request_ids { + let Some((sender, engine_id)) = self.remove(request_id) else { + continue; + }; + let output = EngineCoreStreamOutput { + engine_index: engine_id.engine_index().unwrap_or(0), + timestamp, + output: EngineCoreOutput { + request_id: request_id.clone(), + finish_reason: Some(EngineCoreFinishReason::Abort), + ..EngineCoreOutput::default() + }, + }; + let _ = sender.send(Ok(output)); + aborted.push(request_id.clone()); + } + aborted + } + /// Remove one request from the local registry. Returns the tracked entry if /// it exists. #[must_use] diff --git a/rust/src/engine-core-client/src/tests/client.rs b/rust/src/engine-core-client/src/tests/client.rs index 322eebfd83..83e6cb7ec2 100644 --- a/rust/src/engine-core-client/src/tests/client.rs +++ b/rust/src/engine-core-client/src/tests/client.rs @@ -1939,7 +1939,7 @@ async fn multi_engine_abort_is_grouped_and_utility_fans_out_to_all_engines() { let (shutdown_tx_0, engine_task_0) = spawn_mock_engine_task( handshake_address.clone(), - b"engine-0".to_vec(), + EngineId::from_engine_index(0).into_frame().to_vec(), |dealer, push| { Box::pin(async move { let utility = recv_engine_message(dealer).await; @@ -1993,7 +1993,7 @@ async fn multi_engine_abort_is_grouped_and_utility_fans_out_to_all_engines() { tokio::time::sleep(Duration::from_millis(50)).await; let (shutdown_tx_1, engine_task_1) = spawn_mock_engine_task( handshake_address.clone(), - b"engine-1".to_vec(), + EngineId::from_engine_index(1).into_frame().to_vec(), |dealer, push| { Box::pin(async move { let utility = recv_engine_message(dealer).await; diff --git a/rust/src/llm/Cargo.toml b/rust/src/llm/Cargo.toml index c7924b85db..982fd32dfd 100644 --- a/rust/src/llm/Cargo.toml +++ b/rust/src/llm/Cargo.toml @@ -11,6 +11,7 @@ test-util = [] easy-ext.workspace = true enum-as-inner.workspace = true futures.workspace = true +parking_lot.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true diff --git a/rust/src/llm/src/inflight.rs b/rust/src/llm/src/inflight.rs new file mode 100644 index 0000000000..37df144117 --- /dev/null +++ b/rust/src/llm/src/inflight.rs @@ -0,0 +1,179 @@ +//! Tracking of the external→internal request-id mapping for in-flight requests. +//! +//! When request-id randomization is enabled (the default), [`crate::Llm`] +//! rewrites the external (user-supplied) request id into a unique internal +//! engine id before reaching engine-core. Engine-core only ever knows the +//! internal id, so aborting a request by its external id requires resolving it +//! back to the internal id(s) first. + +use std::collections::HashMap; +use std::sync::{Arc, Weak}; + +use parking_lot::Mutex; + +/// external id → internal id → number of live guards holding that edge. +type InflightMap = HashMap>; + +/// Maps external (user-supplied) request ids to the set of live internal engine +/// request ids they currently expand into. +/// +/// One external id may map to multiple internal ids: duplicate external ids +/// submitted concurrently each get their own randomized internal id, and an +/// abort by the shared external id must reach all of them. Edges are +/// refcounted: with randomization disabled the same (external, internal) pair +/// can be tracked by several guards in sequence (e.g. a finished request whose +/// stream is still held alongside a fresh submission reusing the id), and the +/// edge must survive until the last guard drops. +#[derive(Default)] +pub(crate) struct InflightRequests { + map: Arc>, +} + +impl InflightRequests { + pub(crate) fn new() -> Self { + Self::default() + } + + /// Record that `internal` is now an in-flight engine request for the + /// `external` request id, returning a guard that removes the edge when the + /// request's output stream is dropped (on clean finish or cancellation). + pub(crate) fn track(&self, external: String, internal: String) -> RequestGuard { + *self + .map + .lock() + .entry(external.clone()) + .or_default() + .entry(internal.clone()) + .or_insert(0) += 1; + RequestGuard { + map: Arc::downgrade(&self.map), + external, + internal, + } + } + + /// Resolve external request ids to the internal engine ids currently + /// in-flight for them. Unknown or already-finished ids contribute nothing. + pub(crate) fn resolve(&self, external_ids: &[String]) -> Vec { + let map = self.map.lock(); + external_ids + .iter() + .filter_map(|external| map.get(external)) + .flat_map(|internal_ids| internal_ids.keys()) + .cloned() + .collect() + } + + #[cfg(test)] + fn is_empty(&self) -> bool { + self.map.lock().is_empty() + } +} + +/// RAII guard that releases one refcount on a single external→internal edge +/// when dropped, removing the edge once no live guard holds it. +/// +/// Held by the per-request output stream, so cleanup runs whether the stream +/// terminates cleanly or is cancelled. A [`Weak`] handle is used so a stream +/// outliving its owning [`InflightRequests`] does not keep the map alive. +pub(crate) struct RequestGuard { + map: Weak>, + external: String, + internal: String, +} + +impl Drop for RequestGuard { + fn drop(&mut self) { + let Some(map) = self.map.upgrade() else { + return; + }; + let mut map = map.lock(); + if let Some(internal_ids) = map.get_mut(&self.external) { + if let Some(count) = internal_ids.get_mut(&self.internal) { + *count -= 1; + if *count == 0 { + internal_ids.remove(&self.internal); + } + } + if internal_ids.is_empty() { + map.remove(&self.external); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resolves_external_to_internal() { + let inflight = InflightRequests::new(); + let _guard = inflight.track("ext".to_string(), "ext-abc".to_string()); + + assert_eq!( + inflight.resolve(&["ext".to_string()]), + vec!["ext-abc".to_string()] + ); + assert!(inflight.resolve(&["unknown".to_string()]).is_empty()); + } + + #[test] + fn one_external_maps_to_many_internal() { + let inflight = InflightRequests::new(); + let _g1 = inflight.track("dup".to_string(), "dup-1".to_string()); + let _g2 = inflight.track("dup".to_string(), "dup-2".to_string()); + + let mut resolved = inflight.resolve(&["dup".to_string()]); + resolved.sort(); + assert_eq!(resolved, vec!["dup-1".to_string(), "dup-2".to_string()]); + } + + #[test] + fn dropping_guard_removes_only_its_own_edge_then_cleans_empty_key() { + let inflight = InflightRequests::new(); + let g1 = inflight.track("dup".to_string(), "dup-1".to_string()); + let g2 = inflight.track("dup".to_string(), "dup-2".to_string()); + + drop(g1); + assert_eq!( + inflight.resolve(&["dup".to_string()]), + vec!["dup-2".to_string()] + ); + + drop(g2); + assert!(inflight.resolve(&["dup".to_string()]).is_empty()); + assert!( + inflight.is_empty(), + "empty external key must be removed, not left dangling" + ); + } + + #[test] + fn identical_edges_are_refcounted_across_guards() { + // With request-id randomization disabled, internal == external, so two + // tracked requests can share the exact same edge. Dropping one guard + // (e.g. a stale stream, or the error path of a rejected duplicate + // submission) must not untrack the other still-live request. + let inflight = InflightRequests::new(); + let g1 = inflight.track("x".to_string(), "x".to_string()); + let g2 = inflight.track("x".to_string(), "x".to_string()); + + drop(g1); + assert_eq!(inflight.resolve(&["x".to_string()]), vec!["x".to_string()]); + + drop(g2); + assert!(inflight.resolve(&["x".to_string()]).is_empty()); + assert!(inflight.is_empty()); + } + + #[test] + fn guard_drop_is_a_noop_after_inflight_is_gone() { + let guard = { + let inflight = InflightRequests::new(); + inflight.track("ext".to_string(), "ext-abc".to_string()) + }; + // Dropping the guard after the owning map is gone must not panic. + drop(guard); + } +} diff --git a/rust/src/llm/src/lib.rs b/rust/src/llm/src/lib.rs index 43d46b02f8..9adfc737b6 100644 --- a/rust/src/llm/src/lib.rs +++ b/rust/src/llm/src/lib.rs @@ -2,6 +2,7 @@ use tracing::Span; use vllm_engine_core_client::EngineCoreClient; mod error; +mod inflight; mod log_stats; mod output; mod request; @@ -15,18 +16,22 @@ pub use output::{ pub use request::GenerateRequest; pub use vllm_engine_core_client::protocol::logprobs::{Logprobs, PositionLogprobs, TokenLogprob}; +use crate::inflight::InflightRequests; use crate::log_stats::StatsLogger; use crate::request_metrics::RequestMetricsTracker; -/// Thin generate-only facade over [`EngineCoreClient`]. +/// Thin generate-and-abort facade over [`EngineCoreClient`]. /// /// This mirrors the narrow public shape of Python `AsyncLLM.generate()` and /// `abort()`, but keeps the boundary close to raw engine-core requests and -/// outputs. +/// outputs. It tracks an in-flight external→internal request-id index (see +/// [`InflightRequests`]) so that aborts issued against external (user-supplied) +/// ids can be resolved to the internal engine ids that engine-core understands. pub struct Llm { client: EngineCoreClient, randomize_request_id: bool, stats_logger: Option, + inflight: InflightRequests, } impl Llm { @@ -37,6 +42,7 @@ impl Llm { client, randomize_request_id: true, stats_logger: None, + inflight: InflightRequests::new(), } } @@ -72,9 +78,15 @@ impl Llm { pub async fn generate(&self, req: GenerateRequest) -> Result { let prepared = req.prepare(self.randomize_request_id)?; let prompt_token_ids = prepared.prompt_token_ids().into(); + let external_request_id = prepared + .engine_request + .external_req_id + .clone() + .expect("prepare always sets external_req_id"); + let internal_request_id = prepared.engine_request.request_id.clone(); // Record internal engine-core request ID in the current tracing span. - Span::current().record("engine_request_id", &prepared.engine_request.request_id); + Span::current().record("engine_request_id", &internal_request_id); let request_metrics = RequestMetricsTracker::new( self.client.model_name().to_string(), @@ -84,14 +96,32 @@ impl Llm { 1, ); let stream = self.client.call(prepared.engine_request).await?; + let guard = self.inflight.track(external_request_id, internal_request_id); Ok(GenerateOutputStream::new( prompt_token_ids, stream, request_metrics, + guard, )) } + /// Abort in-flight requests by their external (user-supplied) request ids. + /// + /// External ids are resolved to the internal engine ids actually known to + /// engine-core (one external id may map to several internal ids). Unknown + /// or already-finished ids resolve to nothing and are a safe no-op. The + /// tracking entries themselves are removed when the corresponding output + /// streams are dropped, not here. + pub async fn abort(&self, external_ids: &[String]) -> Result<()> { + let internal_ids = self.inflight.resolve(external_ids); + if internal_ids.is_empty() { + return Ok(()); + } + self.client.abort(&internal_ids).await?; + Ok(()) + } + /// Shut down the underlying engine-core client and its background tasks. pub async fn shutdown(self) -> Result<()> { self.client.shutdown().await?; diff --git a/rust/src/llm/src/output.rs b/rust/src/llm/src/output.rs index cca7cdca33..8cfc38d0bc 100644 --- a/rust/src/llm/src/output.rs +++ b/rust/src/llm/src/output.rs @@ -12,6 +12,7 @@ use vllm_engine_core_client::protocol::{EngineCoreFinishReason, StopReason}; use vllm_engine_core_client::{AbortCause, EngineCoreOutputStream}; use crate::error::Result; +use crate::inflight::RequestGuard; use crate::request_metrics::{RequestMetricsTracker, current_unix_timestamp_secs}; /// Token usage metadata for one request. @@ -195,12 +196,17 @@ impl GenerateOutput { /// Stream of per-request generate outputs for one request. /// -/// - A normal termination of the stream represents a clean completion of the request. -/// - For errors, unexpected closes, or explicit aborts, the stream terminates with an error. +/// - A normal termination of the stream represents a clean completion of the +/// request, including a client-initiated abort, which yields a final output +/// with `finish_reason = Abort` before the stream ends. +/// - For errors or unexpected engine-side closes, the stream terminates with an error. pub struct GenerateOutputStream { pending_prompt_info: Option, raw_stream: EngineCoreOutputStream, request_metrics: RequestMetricsTracker, + /// Removes this request's external→internal tracking edge on drop. Held for + /// its `Drop` side effect only; never read directly. + _request_guard: RequestGuard, } impl GenerateOutputStream { @@ -210,6 +216,7 @@ impl GenerateOutputStream { prompt_token_ids: Arc<[u32]>, raw_stream: EngineCoreOutputStream, request_metrics: RequestMetricsTracker, + request_guard: RequestGuard, ) -> Self { Self { pending_prompt_info: Some(GeneratePromptInfo { @@ -218,6 +225,7 @@ impl GenerateOutputStream { }), raw_stream, request_metrics, + _request_guard: request_guard, } } diff --git a/rust/src/llm/src/request_metrics.rs b/rust/src/llm/src/request_metrics.rs index d28b83be81..6612fa3cc4 100644 --- a/rust/src/llm/src/request_metrics.rs +++ b/rust/src/llm/src/request_metrics.rs @@ -98,27 +98,33 @@ impl RequestMetricsTracker { self.observe_events(engine_index, events); } - if self.is_prefilling { - if let Some(prefill_stats) = &output.prefill_stats { - record_prompt_tokens(&self.model_name, engine_index, prefill_stats); + // Only outputs that actually carry tokens drive token-timing metrics. + // A terminal output with no new tokens (e.g. the synthesized abort + // output) must not log a stray time-to-first-token or inter-token + // sample. + if !output.new_token_ids.is_empty() { + if self.is_prefilling { + if let Some(prefill_stats) = &output.prefill_stats { + record_prompt_tokens(&self.model_name, engine_index, prefill_stats); + } + self.first_token_latency = received_at - self.arrival_time; + observe_time_to_first_token_seconds( + &self.model_name, + engine_index, + self.first_token_latency, + ); + self.first_token_ts = batch_timestamp; + self.is_prefilling = false; + } else if self.last_token_ts > 0.0 { + observe_inter_token_latency_seconds( + &self.model_name, + engine_index, + batch_timestamp - self.last_token_ts, + ); } - self.first_token_latency = received_at - self.arrival_time; - observe_time_to_first_token_seconds( - &self.model_name, - engine_index, - self.first_token_latency, - ); - self.first_token_ts = batch_timestamp; - self.is_prefilling = false; - } else if self.last_token_ts > 0.0 { - observe_inter_token_latency_seconds( - &self.model_name, - engine_index, - batch_timestamp - self.last_token_ts, - ); - } - self.last_token_ts = batch_timestamp; + self.last_token_ts = batch_timestamp; + } } /// Emit the terminal request metrics once a finished output has been diff --git a/rust/src/llm/tests/generate.rs b/rust/src/llm/tests/generate.rs index 18e05063d9..cc7e7f820f 100644 --- a/rust/src/llm/tests/generate.rs +++ b/rust/src/llm/tests/generate.rs @@ -554,6 +554,142 @@ async fn duplicate_external_request_ids_are_randomized_before_reaching_engine_co llm.shutdown().await.unwrap(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn abort_resolves_external_request_id_to_internal_before_reaching_engine() { + let ipc = IpcNamespace::new().unwrap(); + let handshake_address = ipc.handshake_endpoint(); + let engine_id = b"engine-abort".to_vec(); + + let (shutdown_tx, engine_task) = spawn_mock_engine_task( + handshake_address.clone(), + engine_id.clone(), + |dealer, push| { + Box::pin(async move { + let add = recv_engine_message(dealer).await; + assert_eq!(add[0].as_ref(), &[0x00]); + let request: EngineCoreRequest = rmp_serde::from_slice(&add[1]).unwrap(); + assert_eq!(request.external_req_id.as_deref(), Some("req-abort")); + assert!(request.request_id.starts_with("req-abort-")); + assert_ne!(request.request_id, "req-abort"); + + send_outputs( + push, + EngineCoreOutputs { + outputs: vec![request_output(&request.request_id, vec![7], None)], + ..Default::default() + }, + ) + .await; + + // The abort frame must carry the internal engine id, not the + // external "req-abort" id the caller aborted by. + let abort = + timeout(Duration::from_secs(1), recv_engine_message(dealer)).await.unwrap(); + assert_eq!(abort[0].as_ref(), &[0x01]); + let aborted_ids: Vec = rmp_serde::from_slice(&abort[1]).unwrap(); + assert_eq!(aborted_ids, vec![request.request_id]); + }) + }, + ); + + let llm = connect_async_llm_with_ipc(handshake_address, 0, "test-model", &ipc).await; + let mut stream = llm.generate(sample_generate_request("req-abort", 4)).await.unwrap(); + let internal_id = stream.request_id().to_string(); + assert_ne!(internal_id, "req-abort"); + + assert_eq!(stream.next().await.unwrap().unwrap().token_ids, vec![7]); + + // Abort by the external id; engine-core only knows the internal id. + llm.abort(&["req-abort".to_string()]).await.unwrap(); + + // The consumer stream is finalized locally with a clean abort terminal + // rather than hanging or surfacing as RequestStreamClosed. The engine sends + // no final output for a client abort, so this output is synthesized. + let terminal = stream.next().await.unwrap().unwrap(); + assert_eq!(terminal.finish_reason, Some(FinishReason::Abort)); + assert!(terminal.token_ids.is_empty()); + assert!(stream.next().await.is_none()); + + let _ = shutdown_tx.send(()); + engine_task.await.unwrap(); + drop(stream); + llm.shutdown().await.unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn abort_by_external_id_aborts_all_internal_requests() { + let ipc = IpcNamespace::new().unwrap(); + let handshake_address = ipc.handshake_endpoint(); + let engine_id = b"engine-abort-many".to_vec(); + + let (shutdown_tx, engine_task) = spawn_mock_engine_task( + handshake_address.clone(), + engine_id.clone(), + |dealer, push| { + Box::pin(async move { + let add_1 = recv_engine_message(dealer).await; + assert_eq!(add_1[0].as_ref(), &[0x00]); + let request_1: EngineCoreRequest = rmp_serde::from_slice(&add_1[1]).unwrap(); + + let add_2 = recv_engine_message(dealer).await; + assert_eq!(add_2[0].as_ref(), &[0x00]); + let request_2: EngineCoreRequest = rmp_serde::from_slice(&add_2[1]).unwrap(); + + assert_eq!(request_1.external_req_id.as_deref(), Some("req-dup-abort")); + assert_eq!(request_2.external_req_id.as_deref(), Some("req-dup-abort")); + assert_ne!(request_1.request_id, request_2.request_id); + + send_outputs( + push, + EngineCoreOutputs { + outputs: vec![ + request_output(&request_1.request_id, vec![7], None), + request_output(&request_2.request_id, vec![8], None), + ], + ..Default::default() + }, + ) + .await; + + // A single abort by the shared external id must abort both + // internal engine ids it expanded into. + let abort = + timeout(Duration::from_secs(1), recv_engine_message(dealer)).await.unwrap(); + assert_eq!(abort[0].as_ref(), &[0x01]); + let mut aborted_ids: Vec = rmp_serde::from_slice(&abort[1]).unwrap(); + aborted_ids.sort(); + let mut expected = vec![request_1.request_id, request_2.request_id]; + expected.sort(); + assert_eq!(aborted_ids, expected); + }) + }, + ); + + let llm = connect_async_llm_with_ipc(handshake_address, 0, "test-model", &ipc).await; + let mut stream_1 = llm.generate(sample_generate_request("req-dup-abort", 4)).await.unwrap(); + let mut stream_2 = llm.generate(sample_generate_request("req-dup-abort", 4)).await.unwrap(); + assert_ne!(stream_1.request_id(), stream_2.request_id()); + + assert_eq!(stream_1.next().await.unwrap().unwrap().token_ids, vec![7]); + assert_eq!(stream_2.next().await.unwrap().unwrap().token_ids, vec![8]); + + llm.abort(&["req-dup-abort".to_string()]).await.unwrap(); + + // Both internal requests the external id expanded into are finalized with a + // clean abort terminal. + for stream in [&mut stream_1, &mut stream_2] { + let terminal = stream.next().await.unwrap().unwrap(); + assert_eq!(terminal.finish_reason, Some(FinishReason::Abort)); + assert!(stream.next().await.is_none()); + } + + let _ = shutdown_tx.send(()); + engine_task.await.unwrap(); + drop(stream_1); + drop(stream_2); + llm.shutdown().await.unwrap(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn generate_records_request_metrics_in_prometheus_output() { let ipc = IpcNamespace::new().unwrap(); diff --git a/rust/src/text/src/lib.rs b/rust/src/text/src/lib.rs index a550a8afc5..a8ab4191ef 100644 --- a/rust/src/text/src/lib.rs +++ b/rust/src/text/src/lib.rs @@ -151,6 +151,12 @@ impl TextLlm { Ok((text_request, raw_stream)) } + /// Abort in-flight requests by their external (user-supplied) request ids. + pub async fn abort(&self, external_ids: &[String]) -> Result<()> { + self.llm.abort(external_ids).await?; + Ok(()) + } + /// Shut down the underlying LLM client and its background tasks. pub async fn shutdown(self) -> Result<()> { self.llm.shutdown().await?;