From a9bbbfc2d51e151d404baf340109eb1f974b454d Mon Sep 17 00:00:00 2001 From: Ivan Zatevakhin Date: Tue, 11 Aug 2026 04:18:53 +0100 Subject: [PATCH] feat: add concurrent request API --- Cargo.lock | 7 + Justfile | 70 ++++- README.md | 30 +- vllm-cpp-sys/build.rs | 8 +- vllm-cpp/Cargo.toml | 1 + vllm-cpp/examples/concurrent.rs | 26 ++ vllm-cpp/src/engine.rs | 39 ++- vllm-cpp/src/error.rs | 11 + vllm-cpp/src/lib.rs | 9 +- vllm-cpp/src/request.rs | 508 ++++++++++++++++++++++++++++++++ vllm-cpp/tests/qwen3.rs | 445 +++++++++++++++++++++++++++- vllm-cpp/tests/safe_api.rs | 7 +- 12 files changed, 1116 insertions(+), 45 deletions(-) create mode 100644 vllm-cpp/examples/concurrent.rs create mode 100644 vllm-cpp/src/request.rs diff --git a/Cargo.lock b/Cargo.lock index 2ea39b2..d7e4e88 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -105,6 +105,12 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + [[package]] name = "syn" version = "3.0.3" @@ -127,6 +133,7 @@ name = "vllm-cpp" version = "0.1.0" dependencies = [ "serde_json", + "static_assertions", "vllm-cpp-sys", ] diff --git a/Justfile b/Justfile index 487f7a1..e5a4da7 100644 --- a/Justfile +++ b/Justfile @@ -577,7 +577,7 @@ setup-test-model destination=env_var_or_default("VLLM_CPP_TEST_MODEL", env_var_o (cd "$destination" && sha256sum --check SHA256SUMS.expected) >&2 printf '%s\n' "$destination" -# Run the blocking safe API and Qwen model suites with ASan, UBSan, and leak detection. +# Run the full safe/request/model suites with ASan, UBSan, and leak detection. sanitizers model=env_var_or_default("VLLM_CPP_TEST_MODEL", ""): #!/usr/bin/env bash set -euo pipefail @@ -631,6 +631,74 @@ sanitizers model=env_var_or_default("VLLM_CPP_TEST_MODEL", ""): "$binary" --test-threads=1 done +# Run selected request lifecycle tests under native-only GCC TSan on Linux x86_64. +tsan model=env_var_or_default("VLLM_CPP_TEST_MODEL", ""): + #!/usr/bin/env bash + set -euo pipefail + if [[ $(uname -s) != Linux || $(uname -m) != x86_64 ]]; then + echo 'GCC TSan validation is supported only on Linux x86_64' >&2 + exit 1 + fi + model={{ quote(model) }} + if [[ -z $model ]]; then + echo 'set VLLM_CPP_TEST_MODEL or pass model=' >&2 + exit 1 + fi + required_model_files=( + model.safetensors + config.json + tokenizer.json + tokenizer_config.json + ) + missing=() + for file in "${required_model_files[@]}"; do + [[ -f $model/$file ]] || missing+=("$file") + done + if ((${#missing[@]})); then + printf 'model fixture is incomplete at %s; missing:' "$model" >&2 + printf ' %s' "${missing[@]}" >&2 + printf '\n' >&2 + exit 1 + fi + cd {{ quote(root) }} + export VLLM_CPP_TEST_MODEL="$model" + export VLLM_CPP_TEST_ISOLATED_ENGINE=1 + export VLLM_CPP_SANITIZE=thread + export CARGO_TARGET_DIR=${CARGO_TARGET_DIR:-{{ quote(root + "/target/tsan") }}} + + cargo test --locked -p vllm-cpp --test qwen3 --no-run + + tsan=$(gcc -print-file-name=libtsan.so) + if [[ ! -f $tsan ]]; then + echo 'GCC ThreadSanitizer runtime is unavailable' >&2 + exit 1 + fi + export LD_PRELOAD="$tsan${LD_PRELOAD:+:$LD_PRELOAD}" + # Rust and its standard library are not instrumented in this GCC lane. Keep + # native vllm.cpp races visible while ignoring uninstrumented Rust modules. + export TSAN_OPTIONS=${TSAN_OPTIONS:-halt_on_error=1:ignore_noninstrumented_modules=1} + export VT_POOL_BYPASS=1 + + binary=$(find "$CARGO_TARGET_DIR/debug/deps" -maxdepth 1 -type f \ + -name 'qwen3-*' -executable -printf '%T@ %p\n' \ + | sort -n | tail -1 | cut -d' ' -f2-) || true + if [[ -z $binary ]]; then + echo 'could not find qwen3 test binary' >&2 + exit 1 + fi + # Self-drop uses uninstrumented Rust synchronization, so normal and ASan/LSan + # cover it. Every selected native lifecycle case runs in its own process. + for test in \ + concurrent_requests_batch_with_correct_output \ + engine_clones_submit_and_wait_from_multiple_rust_threads \ + live_request_moves_to_rust_thread_for_cancel_wait_and_drop \ + request_outcomes_and_probes_are_precise \ + callback_panic_is_reported_and_engine_is_reusable \ + request_retains_engine_and_live_drop_is_safe \ + concurrent_request_lifecycle_stress; do + "$binary" "$test" --exact --test-threads=1 + done + # Check Just and Rust formatting. fmt-check: just --unstable --justfile {{ quote(root + "/Justfile") }} --fmt --check diff --git a/README.md b/README.md index a6d5d4a..153d7aa 100644 --- a/README.md +++ b/README.md @@ -3,11 +3,11 @@ Rust bindings for [vllm.cpp](https://github.com/mudler/vllm.cpp), organized as: - `vllm-cpp-sys`: raw C API bindings and the pinned native source build. -- `vllm-cpp`: the application-facing safe blocking API. +- `vllm-cpp`: the application-facing safe inference API. ## Status -The safe crate provides an owned blocking engine API for model loading, completion, streaming, structured output, and raw-JSON chat. An optional `serde` feature adds `serde_json::Value` chat helpers. The sys crate provides checked-in generated FFI declarations with C/Rust layout checks and coverage for all 19 exported C symbols. +The safe crate provides a cloneable engine API for model loading, blocking completion and streaming, non-blocking concurrent requests, structured output, and raw-JSON chat. An optional `serde` feature adds `serde_json::Value` chat helpers. The sys crate provides checked-in generated FFI declarations with C/Rust layout checks and coverage for all 19 exported C symbols. Linux x86_64 CPU builds support bundled static, bundled dynamic, system static, and system dynamic linking. vllm.cpp is pinned at `34aedfbe8ed9779697905541a62e2160ccfd9c05`, which exposes C ABI version 10. @@ -39,26 +39,33 @@ git submodule update --init --recursive ## Safe API ```rust -use vllm_cpp::{Engine, SamplingParams}; +use vllm_cpp::{Engine, SamplingParams, StreamControl}; let engine = Engine::load("/models/Qwen3-0.6B")?; -let completion = engine.complete( - "The capital of France is", - &SamplingParams::greedy().max_tokens(16), -)?; +let params = SamplingParams::greedy().max_tokens(16); +let completion = engine.complete("The capital of France is", ¶ms)?; println!("{}", completion.text); + +let mut request = engine.submit("The capital of Germany is", ¶ms, |event| { + print!("{}", event.delta); + StreamControl::Continue +})?; +println!("{:?}", request.wait()?); # Ok::<(), vllm_cpp::Error>(()) ``` `EngineBuilder` owns model settings and converts them to temporary C strings only for the load call. `SamplingParams` owns stop strings and structured constraints. Completion and chat strings are copied into Rust values before the matching native free function runs. -Blocking streaming callbacks receive copied UTF-8 deltas. Callback panics are caught before the C boundary and resumed only after the native call has stopped and returned. Chat methods accept raw OpenAI-compatible request JSON; enable `serde` for `serde_json::Value` request and response helpers. +`Engine` is `Clone + Send + Sync`; each `Request` retains the shared engine until native callback delivery has joined. A request is `Send` but deliberately not `Sync`. `submit` returns before generation finishes, and `Request` provides `is_done`, idempotent `cancel`, `wait`, and copied `native_error` diagnostics. `wait` classifies completion as `Completed`, `StoppedByCallback`, or `Cancelled`; an explicit asynchronous `Stop` is classified as `StoppedByCallback` even when returned for the terminal event. + +All streaming callbacks receive copied UTF-8 deltas. Blocking callbacks may borrow stack data; their panics are caught before the C boundary and resumed only after the native call returns. Asynchronous callbacks must be `Send + 'static`, run on a native delivery thread, and report panic as `Error::CallbackPanicked` from `wait`. Waiting for or freeing a request from its own callback thread is prohibited by ABI v10: `wait` returns `Error::RequestCallbackThread`, while drop transfers cleanup to a prestarted reaper that owns the request, callback, and engine until native free/cancel/join completes. Chat methods accept raw OpenAI-compatible request JSON; enable `serde` for `serde_json::Value` request and response helpers. Run the practical examples with a model directory: ```console cargo run -p vllm-cpp --example complete -- cargo run -p vllm-cpp --example stream -- +cargo run -p vllm-cpp --example concurrent -- cargo run -p vllm-cpp --example chat -- cargo run -p vllm-cpp --example structured -- ``` @@ -80,7 +87,7 @@ Set `CMAKE_BUILD_PARALLEL_LEVEL` to control native parallelism. The bundled buil ## Test Model and Sanitizers -Model-backed tests use Apache-2.0 `Qwen/Qwen3-0.6B` at pinned revision `c1899de289a04d12100db370d81485cdf75e47ca`. Download or reuse the cache and verify every file, then run exactly the six blocking model tests serially: +Model-backed tests use Apache-2.0 `Qwen/Qwen3-0.6B` at pinned revision `c1899de289a04d12100db370d81485cdf75e47ca`. Download or reuse the cache and verify every file, then run exactly 14 blocking and request-lifecycle model tests serially: ```console model=$(just setup-test-model) @@ -90,10 +97,11 @@ VLLM_CPP_TEST_MODEL="$model" \ The approximately 1.5 GB model stays in the user cache and is not included in repository or crate packages. Model-backed tests skip with an explanatory message when `VLLM_CPP_TEST_MODEL` is unset. When it is set, the test helper and sanitizer gate require `model.safetensors`, `config.json`, `tokenizer.json`, and `tokenizer_config.json` and report every missing file. -Address, undefined-behavior, and leak checks run the blocking safe API and six model tests with native instrumentation: +AddressSanitizer, UndefinedBehaviorSanitizer, and leak detection run the full safe/request/model suites with native instrumentation. The Linux x86_64 GCC ThreadSanitizer lane runs selected request lifecycle tests individually and instruments native C++ only; it does not claim race coverage for Rust or the Rust standard library. Callback-thread self-drop remains in the normal and ASan/leak suites because its handoff uses uninstrumented Rust synchronization. ```console just sanitizers "$model" +just tsan "$model" ``` `VLLM_CPP_SANITIZE` is a bundled-build test input. System mode rejects it because Cargo cannot infer whether an externally built native library carries matching instrumentation. @@ -121,7 +129,7 @@ The package gate preserves the sys crate inventory, tests the extracted sys crat ## Support -The supported target is native Linux x86_64 CPU. Maintainer tests cover the four bundled/system static/dynamic link modes and bundled blocking inference with the pinned Qwen fixture. Other operating systems, architectures, and accelerator builds are not supported by this Rust build. +The supported target is native Linux x86_64 CPU. Maintainer tests cover the four bundled/system static/dynamic link modes plus bundled blocking and concurrent request inference with the pinned Qwen fixture. Sanitizer evidence covers native ASan/UBSan/leak detection and selected native-only GCC TSan lifecycle paths as described above. Other operating systems, architectures, and accelerator builds are not supported by this Rust build. ## Licensing and Affiliation diff --git a/vllm-cpp-sys/build.rs b/vllm-cpp-sys/build.rs index 3dc6ff0..44c5465 100644 --- a/vllm-cpp-sys/build.rs +++ b/vllm-cpp-sys/build.rs @@ -233,10 +233,9 @@ fn sanitizer_config(bundled: bool) -> Option { panic!("VLLM_CPP_SANITIZE is supported only for bundled builds"); } match value.as_str() { - "address" | "undefined" | "address,undefined" => Some(value), - "thread" => panic!("thread sanitization is not part of the blocking API test lane"), + "address" | "undefined" | "address,undefined" | "thread" => Some(value), _ => panic!( - "unsupported VLLM_CPP_SANITIZE value `{value}`; expected OFF, address, undefined, or address,undefined" + "unsupported VLLM_CPP_SANITIZE value `{value}`; expected OFF, address, undefined, address,undefined, or thread" ), } } @@ -251,6 +250,9 @@ fn link_sanitizer_runtimes(sanitizer: Option<&str>) { if sanitizer.split(',').any(|name| name == "undefined") { println!("cargo:rustc-link-lib=dylib=ubsan"); } + if sanitizer.split(',').any(|name| name == "thread") { + println!("cargo:rustc-link-lib=dylib=tsan"); + } } fn link_platform_dependencies() { diff --git a/vllm-cpp/Cargo.toml b/vllm-cpp/Cargo.toml index 6ebcf79..b411250 100644 --- a/vllm-cpp/Cargo.toml +++ b/vllm-cpp/Cargo.toml @@ -20,3 +20,4 @@ vllm-cpp-sys = { workspace = true, default-features = false } [dev-dependencies] serde_json = "1.0.149" +static_assertions = "1.1.0" diff --git a/vllm-cpp/examples/concurrent.rs b/vllm-cpp/examples/concurrent.rs new file mode 100644 index 0000000..51432f7 --- /dev/null +++ b/vllm-cpp/examples/concurrent.rs @@ -0,0 +1,26 @@ +use std::io::{self, Write}; + +use vllm_cpp::{Engine, SamplingParams, StreamControl}; + +fn main() -> Result<(), Box> { + let model = std::env::args_os() + .nth(1) + .ok_or("usage: concurrent ")?; + let engine = Engine::load(model)?; + let params = SamplingParams::greedy().max_tokens(16); + + let mut france = engine.submit("The capital of France is", ¶ms, |event| { + print!("[france] {}", event.delta); + io::stdout().flush().expect("flush stdout"); + StreamControl::Continue + })?; + let mut germany = engine.submit("The capital of Germany is", ¶ms, |event| { + print!("[germany] {}", event.delta); + io::stdout().flush().expect("flush stdout"); + StreamControl::Continue + })?; + + println!("\nfrance: {:?}", france.wait()?); + println!("germany: {:?}", germany.wait()?); + Ok(()) +} diff --git a/vllm-cpp/src/engine.rs b/vllm-cpp/src/engine.rs index cc5f084..bef72fd 100644 --- a/vllm-cpp/src/engine.rs +++ b/vllm-cpp/src/engine.rs @@ -3,6 +3,7 @@ use std::mem::MaybeUninit; use std::os::raw::c_char; use std::path::{Path, PathBuf}; use std::ptr::{self, NonNull}; +use std::sync::Arc; use vllm_cpp_sys as ffi; @@ -12,16 +13,21 @@ use crate::callback::{ use crate::error::{invalid_configuration, status_result, Error}; use crate::params::{SamplingParams, SchedulerPolicy, Toggle}; -/// An owned vllm.cpp serving engine. +/// A cloneable vllm.cpp serving engine. +#[derive(Clone)] pub struct Engine { - raw: NonNull, + pub(crate) inner: Arc, +} + +pub(crate) struct EngineInner { + pub(crate) raw: NonNull, } impl std::fmt::Debug for Engine { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter .debug_struct("Engine") - .field("raw", &self.raw) + .field("raw", &self.inner.raw) .finish_non_exhaustive() } } @@ -87,7 +93,7 @@ impl Engine { // call, and out storage is initialized by native code on success. let status = unsafe { ffi::vllm_complete( - self.raw.as_ptr(), + self.inner.raw.as_ptr(), prompt.as_ptr(), params.raw(), raw.as_mut_ptr(), @@ -120,7 +126,7 @@ impl Engine { // API does not retain user_data after returning. let status = unsafe { ffi::vllm_complete_stream( - self.raw.as_ptr(), + self.inner.raw.as_ptr(), prompt.as_ptr(), params.raw(), Some(callback_trampoline::), @@ -145,7 +151,8 @@ impl Engine { let mut output: *mut c_char = ptr::null_mut(); // SAFETY: the engine and request pointers are valid for the call and the // returned string is released by NativeStringGuard. - let status = unsafe { ffi::vllm_chat(self.raw.as_ptr(), request.as_ptr(), &mut output) }; + let status = + unsafe { ffi::vllm_chat(self.inner.raw.as_ptr(), request.as_ptr(), &mut output) }; status_result(status)?; let output = NonNull::new(output).ok_or_else(|| Error::Runtime { message: "vllm_chat succeeded without a response".to_owned(), @@ -169,7 +176,7 @@ impl Engine { // not retain it after returning. let status = unsafe { ffi::vllm_chat_stream( - self.raw.as_ptr(), + self.inner.raw.as_ptr(), request.as_ptr(), Some(callback_trampoline::), ptr::from_mut(&mut state).cast(), @@ -201,16 +208,20 @@ impl Engine { } } -impl Drop for Engine { +impl Drop for EngineInner { fn drop(&mut self) { - // SAFETY: Engine exclusively owns this live handle and drops it once. + // SAFETY: EngineInner exclusively owns this live handle and drops it once, + // after every Request-owned Arc has been released. unsafe { ffi::vllm_engine_free(self.raw.as_ptr()) }; } } -// Moving the sole owner is safe because the native handle has no thread affinity; -// ownership still keeps the handle live until all Rust access has ended. -unsafe impl Send for Engine {} +// SAFETY: vllm.cpp documents concurrent completion submissions as thread-safe, +// and EngineInner keeps the engine alive until the last shared owner is dropped. +unsafe impl Send for EngineInner {} +// SAFETY: shared references may submit concurrently through native AsyncLLM; +// destruction cannot race because Arc retains the handle for each active owner. +unsafe impl Sync for EngineInner {} impl EngineBuilder { #[must_use] @@ -358,7 +369,9 @@ impl EngineBuilder { let raw = NonNull::new(output).ok_or_else(|| Error::ModelLoad { message: "vllm_engine_load succeeded without a handle".to_owned(), })?; - Ok(Engine { raw }) + Ok(Engine { + inner: Arc::new(EngineInner { raw }), + }) } } diff --git a/vllm-cpp/src/error.rs b/vllm-cpp/src/error.rs index c2694cf..9090f93 100644 --- a/vllm-cpp/src/error.rs +++ b/vllm-cpp/src/error.rs @@ -25,6 +25,10 @@ pub enum Error { PathEncoding, /// Native code returned bytes that are not valid UTF-8. InvalidUtf8 { field: &'static str }, + /// An asynchronous callback panicked. + CallbackPanicked, + /// A request operation was attempted from that request's callback thread. + RequestCallbackThread { operation: &'static str }, /// A Rust-side parameter cannot be represented by the native API. InvalidConfiguration { message: String }, /// JSON serialization or parsing failed. @@ -53,6 +57,13 @@ impl fmt::Display for Error { Self::InteriorNul { field } => write!(f, "{field} contains an interior NUL byte"), Self::PathEncoding => write!(f, "path cannot be represented by the native API"), Self::InvalidUtf8 { field } => write!(f, "native {field} is not valid UTF-8"), + Self::CallbackPanicked => write!(f, "asynchronous request callback panicked"), + Self::RequestCallbackThread { operation } => { + write!( + f, + "cannot {operation} a request from its own callback thread" + ) + } Self::InvalidConfiguration { message } => { write!(f, "invalid configuration: {message}") } diff --git a/vllm-cpp/src/lib.rs b/vllm-cpp/src/lib.rs index 755c20a..1d21c30 100644 --- a/vllm-cpp/src/lib.rs +++ b/vllm-cpp/src/lib.rs @@ -1,18 +1,21 @@ //! Safe Rust bindings for the stable vllm.cpp C API. //! -//! The central [`Engine`] owns a complete native serving stack. Construct -//! request parameters in Rust, then use blocking completion or chat methods -//! without handling native pointers or free functions. +//! [`Engine`] is a cloneable, shared owner of a complete native serving stack. +//! It provides blocking completion, streaming, and chat methods plus +//! [`Engine::submit`] for non-blocking requests. Each [`Request`] retains the +//! engine until native request free/join completes. mod callback; mod engine; mod error; mod params; +mod request; pub use callback::{StreamControl, StreamEvent, StreamOutcome}; pub use engine::{Completion, Engine, EngineBuilder, FinishReason}; pub use error::Error; pub use params::{SamplingParams, SchedulerPolicy, StructuredOutput, Toggle}; +pub use request::{Request, RequestOutcome}; /// Returns the compile-time C ABI expected by this crate. #[must_use] diff --git a/vllm-cpp/src/request.rs b/vllm-cpp/src/request.rs new file mode 100644 index 0000000..e754984 --- /dev/null +++ b/vllm-cpp/src/request.rs @@ -0,0 +1,508 @@ +use std::any::Any; +use std::cell::Cell; +use std::ffi::CStr; +use std::marker::PhantomData; +use std::os::raw::{c_char, c_void}; +use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::ptr::{self, NonNull}; +use std::sync::{mpsc, Arc, Mutex, OnceLock}; +use std::thread::{self, ThreadId}; + +use vllm_cpp_sys as ffi; + +use crate::callback::{StreamControl, StreamEvent}; +use crate::engine::{Engine, EngineInner}; +use crate::error::{status_result, Error}; +use crate::params::{to_cstring, SamplingParams}; + +/// How a successfully waited non-blocking request ended. +/// +/// This is a Rust-side classification because the native ABI does not expose its +/// cancellation flag. Callback panic/error takes precedence, followed by an +/// explicit callback stop, an observed terminal callback, and cancellation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[non_exhaustive] +pub enum RequestOutcome { + /// Native generation delivered its terminal callback. + Completed, + /// The Rust callback returned [`StreamControl::Stop`]. + /// + /// ABI v10 treats this as an explicit stop even for the terminal event. + StoppedByCallback, + /// Rust requested cancellation before completion was observable. + Cancelled, +} + +/// An owned non-blocking streaming request. +/// +/// A request keeps its parent [`Engine`] alive. Lifecycle methods require mutable +/// access, and the request is intentionally `Send` but not `Sync`. +pub struct Request { + raw: Option>, + callback: Option>, + engine: Option>, + cancellation_requested: bool, + _not_sync: PhantomData>, +} + +impl std::fmt::Debug for Request { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("Request") + .field("raw", &self.raw) + .field("cancellation_requested", &self.cancellation_requested) + .finish_non_exhaustive() + } +} + +impl Engine { + /// Submits a non-blocking streaming completion to the shared engine. + /// + /// The callback runs on a native delivery thread and receives an owned UTF-8 + /// copy of each delta. A callback panic is contained and later reported by + /// [`Request::wait`] as [`Error::CallbackPanicked`]. + pub fn submit( + &self, + prompt: &str, + params: &SamplingParams, + callback: F, + ) -> Result + where + F: FnMut(StreamEvent) -> StreamControl + Send + 'static, + { + cleanup_sender()?; + let prompt = to_cstring(prompt, "prompt")?; + let params = params.marshal()?; + let mut callback = Box::new(AsyncCallbackState::new(callback)); + let mut output = ptr::null_mut(); + // SAFETY: the engine is retained by the returned Request, native code + // copies prompt/parameters before returning, callback has a stable boxed + // address, and callback state remains live until request_free joins. + let status = unsafe { + ffi::vllm_request_submit( + self.inner.raw.as_ptr(), + prompt.as_ptr(), + params.raw(), + Some(async_callback_trampoline), + ptr::from_mut(&mut *callback).cast(), + &mut output, + ) + }; + status_result(status)?; + let raw = NonNull::new(output).ok_or_else(|| Error::Runtime { + message: "vllm_request_submit succeeded without a request handle".to_owned(), + })?; + Ok(Request { + raw: Some(raw), + callback: Some(callback), + engine: Some(Arc::clone(&self.inner)), + cancellation_requested: false, + _not_sync: PhantomData, + }) + } +} + +impl Request { + /// Returns whether native callback delivery has finished. + #[must_use] + pub fn is_done(&self) -> bool { + // SAFETY: raw remains a live request handle until Drop, and the native + // completion probe is atomic and accepts concurrent callback delivery. + unsafe { ffi::vllm_request_done(self.raw().as_ptr()) } + } + + /// Requests cancellation. + /// + /// Cancellation is idempotent. The ABI does not report whether this call + /// changed native state, so [`wait`](Self::wait) returns + /// [`RequestOutcome::Cancelled`] when cancellation succeeded after a false + /// completion probe and no terminal callback was subsequently observed. A + /// terminal callback wins that race unless it stops or panics. + pub fn cancel(&mut self) -> Result<(), Error> { + let was_done = self.is_done(); + // SAFETY: mutable access serializes safe lifecycle calls and raw is live. + let status = unsafe { ffi::vllm_request_cancel(self.raw().as_ptr()) }; + status_result(status)?; + self.cancellation_requested |= !was_done; + Ok(()) + } + + /// Waits for callback delivery to finish and returns its terminal outcome. + /// + /// Calling this from this request's own callback returns + /// [`Error::RequestCallbackThread`] without entering native code. + pub fn wait(&mut self) -> Result { + if self.callback().is_delivery_thread() { + return Err(Error::RequestCallbackThread { operation: "wait" }); + } + // SAFETY: mutable access serializes safe lifecycle calls, raw is live, + // and the delivery-thread case was rejected before the FFI call. + let status = unsafe { ffi::vllm_request_wait(self.raw().as_ptr()) }; + let native_result = status_result(status); + let callback_result = self.callback().result(self.cancellation_requested); + match callback_result { + Err(error) => Err(error), + Ok(Some(outcome)) => native_result.map(|()| outcome), + Ok(None) => { + native_result?; + Err(Error::Runtime { + message: "request completed without a terminal callback or locally observable stop/cancellation" + .to_owned(), + }) + } + } + } + + /// Copies the native request diagnostic after completion. + /// + /// Returns `None` while the request is running or when it completed without + /// a native diagnostic. Callback panics are reported by [`wait`](Self::wait) + /// rather than through this native string. + pub fn native_error(&self) -> Result, Error> { + if !self.is_done() { + return Ok(None); + } + // SAFETY: done has acquired native publication of the request-owned error + // string, and raw remains live for this copy. + let pointer = unsafe { ffi::vllm_request_error(self.raw().as_ptr()) }; + if pointer.is_null() { + return Err(Error::Runtime { + message: "vllm_request_error returned a null pointer".to_owned(), + }); + } + // SAFETY: the C contract promises a NUL-terminated string valid until + // request_free; this method copies it before returning. + let error = unsafe { CStr::from_ptr(pointer) } + .to_str() + .map_err(|_| Error::InvalidUtf8 { + field: "request error", + })? + .to_owned(); + Ok((!error.is_empty()).then_some(error)) + } + + fn raw(&self) -> NonNull { + self.raw.expect("live Request always has a native handle") + } + + fn callback(&self) -> &AsyncCallbackState { + self.callback + .as_deref() + .expect("live Request always has callback state") + } +} + +impl Drop for Request { + fn drop(&mut self) { + let parts = (self.raw.take(), self.callback.take(), self.engine.take()); + match parts { + (Some(raw), Some(callback), Some(engine)) => { + CleanupJob::new(raw, callback, engine).run(); + } + parts => { + // A partial Request would make either freeing or dropping its + // Rust owners unsafe. This private invariant cannot fail without + // an implementation bug, so preserve everything before aborting. + std::mem::forget(parts); + std::process::abort(); + } + } + } +} + +// Moving exclusive request ownership between threads is valid under the native +// request contract. Callback state is Send, EngineInner is Send + Sync, and +// wait/free explicitly reject or defer the one prohibited delivery-thread case. +unsafe impl Send for Request {} + +struct CallbackOutcome { + stopped: bool, + saw_finished: bool, + error: Option, + panic: Option>, + delivery_thread: Option, +} + +struct AsyncCallbackState { + callback: Mutex StreamControl + Send + 'static>>, + outcome: Mutex, +} + +impl AsyncCallbackState { + fn new(callback: F) -> Self + where + F: FnMut(StreamEvent) -> StreamControl + Send + 'static, + { + Self { + callback: Mutex::new(Box::new(callback)), + outcome: Mutex::new(CallbackOutcome { + stopped: false, + saw_finished: false, + error: None, + panic: None, + delivery_thread: None, + }), + } + } + + fn record_delivery_thread(&self) { + // ABI v10 invokes user_data only from this request's single library-owned + // delivery thread. Retain its ID through cleanup instead of marking only + // an active trampoline, so every possible Rust re-entry from that thread + // remains ineligible for synchronous wait/free. + lock_unpoisoned(&self.outcome).delivery_thread = Some(thread::current().id()); + } + + fn is_delivery_thread(&self) -> bool { + lock_unpoisoned(&self.outcome) + .delivery_thread + .as_ref() + .is_some_and(|id| *id == thread::current().id()) + } + + fn record_error(&self, error: Error) { + let mut outcome = lock_unpoisoned(&self.outcome); + outcome.error = Some(error); + outcome.stopped = true; + } + + fn record_result( + &self, + result: Result>, + finished: bool, + ) -> bool { + let mut outcome = lock_unpoisoned(&self.outcome); + outcome.saw_finished |= finished; + match result { + Ok(StreamControl::Continue) => true, + Ok(StreamControl::Stop) => { + outcome.stopped = true; + false + } + Err(payload) => { + outcome.panic = Some(payload); + outcome.stopped = true; + false + } + } + } + + fn result(&self, cancellation_requested: bool) -> Result, Error> { + let outcome = lock_unpoisoned(&self.outcome); + if outcome.panic.is_some() { + return Err(Error::CallbackPanicked); + } + if let Some(error) = &outcome.error { + return Err(error.clone()); + } + if outcome.stopped { + return Ok(Some(RequestOutcome::StoppedByCallback)); + } + if outcome.saw_finished { + return Ok(Some(RequestOutcome::Completed)); + } + if cancellation_requested { + return Ok(Some(RequestOutcome::Cancelled)); + } + Ok(None) + } +} + +unsafe extern "C" fn async_callback_trampoline( + delta_text: *const c_char, + finished: bool, + user_data: *mut c_void, +) -> bool { + // SAFETY: submit passes a stable boxed AsyncCallbackState, and request_free + // joins this delivery before the box can be destroyed. + let state = unsafe { &*user_data.cast::() }; + state.record_delivery_thread(); + if delta_text.is_null() { + state.record_error(Error::InvalidUtf8 { + field: "stream delta", + }); + return false; + } + // SAFETY: native code lends a NUL-terminated string for this invocation. + let delta = match unsafe { CStr::from_ptr(delta_text) }.to_str() { + Ok(delta) => delta.to_owned(), + Err(_) => { + state.record_error(Error::InvalidUtf8 { + field: "stream delta", + }); + return false; + } + }; + let event = StreamEvent { delta, finished }; + let result = catch_unwind(AssertUnwindSafe(|| { + let mut callback = lock_unpoisoned(&state.callback); + callback(event) + })); + state.record_result(result, finished) +} + +fn lock_unpoisoned(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +struct CleanupJob { + state: CleanupState, + context: CleanupContext, +} + +enum CleanupContext { + Caller, + Reaper, +} + +enum CleanupState { + Armed { + raw: NonNull, + callback: Box, + engine: Arc, + }, + Disarmed, +} + +impl CleanupJob { + fn new( + raw: NonNull, + callback: Box, + engine: Arc, + ) -> Self { + Self { + state: CleanupState::Armed { + raw, + callback, + engine, + }, + context: CleanupContext::Caller, + } + } + + fn run(mut self) { + self.finish(); + } + + fn finish(&mut self) { + if matches!(self.state, CleanupState::Disarmed) { + return; + } + let needs_deferral = match self.context { + CleanupContext::Caller => match &self.state { + CleanupState::Armed { callback, .. } => callback.is_delivery_thread(), + CleanupState::Disarmed => return, + }, + // A successfully sent job is owned only by the prestarted Rust reaper, + // so it cannot be running in the native request callback. + CleanupContext::Reaper => false, + }; + if needs_deferral { + self.defer_to_reaper(); + } else if let Err(payload) = catch_unwind(AssertUnwindSafe(|| self.cleanup_now())) { + // Retrying free after a Rust unwind could double-free an opaque void + // native operation. Disarm and leak the unknown remainder instead. + self.leak_armed(); + std::mem::forget(payload); + } + } + + fn cleanup_now(&mut self) { + let state = std::mem::replace(&mut self.state, CleanupState::Disarmed); + let CleanupState::Armed { + raw, + callback, + engine, + } = state + else { + return; + }; + // If this function unwinds, forget every owner before CleanupJob::Drop can + // run. Repeating an opaque void free could double-free, while releasing + // callback/engine without a known join would be unsafe. + let mut owners = std::mem::ManuallyDrop::new((callback, engine)); + // ABI coupling: the native delivery thread can enter Rust only through + // async_callback_trampoline, which records its permanent ThreadId before + // user code runs. Therefore this path cannot call free on that thread. + // Any new native user_data entrypoint or delivery model must update that + // tracking before this wrapper can safely adopt it. + // + // SAFETY: this armed job owns the request exactly once, runs off the + // tracked delivery thread or on the dedicated reaper after ownership + // transfer, and retains both callback state and parent engine. Native + // free cancels if needed and joins before returning. + unsafe { ffi::vllm_request_free(raw.as_ptr()) }; + // SAFETY: native free returned, so delivery is joined and Rust owners can + // be reclaimed. ManuallyDrop prevents premature release on unwind above. + let (callback, engine) = unsafe { std::mem::ManuallyDrop::take(&mut owners) }; + // User callback captures and a stored panic payload can have arbitrary + // destructors. Never let them unwind out of cleanup. + if let Err(payload) = catch_unwind(AssertUnwindSafe(|| drop(callback))) { + std::mem::forget(payload); + } + drop(engine); + } + + fn leak_armed(&mut self) { + let state = std::mem::replace(&mut self.state, CleanupState::Disarmed); + std::mem::forget(state); + } + + fn defer_to_reaper(&mut self) { + let sender = match CLEANUP_REAPER.get() { + Some(Ok(sender)) => sender, + // Submission starts the process-lifetime reaper before native code + // can create a request. Without it, self-thread cleanup is impossible. + _ => std::process::abort(), + }; + let job = Self { + state: std::mem::replace(&mut self.state, CleanupState::Disarmed), + context: CleanupContext::Reaper, + }; + if let Err(error) = sender.send(job) { + // SendError owns the still-live native handle. Its Drop would recurse + // here on the callback thread and eventually release callback/engine + // before native join, so leak it and terminate instead. + std::mem::forget(error); + std::process::abort(); + } + } +} + +impl Drop for CleanupJob { + fn drop(&mut self) { + // This is the ownership backstop: every caller-side armed drop either + // frees and joins or transfers all owners to the reaper; a reaper-owned + // drop always completes cleanup locally, so send failure cannot recurse. + self.finish(); + } +} + +// The job transfers unique native-handle ownership to the reaper. Its callback +// is Send and its retained engine is Send + Sync; no aliases perform lifecycle +// operations while the job owns the handle. +unsafe impl Send for CleanupJob {} + +static CLEANUP_REAPER: OnceLock, String>> = OnceLock::new(); + +fn cleanup_sender() -> Result<&'static mpsc::Sender, Error> { + match CLEANUP_REAPER.get_or_init(|| { + let (sender, receiver) = mpsc::channel::(); + thread::Builder::new() + .name("vllm-request-reaper".to_owned()) + .spawn(move || { + while let Ok(job) = receiver.recv() { + job.run(); + } + }) + .map(|_| sender) + .map_err(|error| error.to_string()) + }) { + Ok(sender) => Ok(sender), + Err(message) => Err(Error::Runtime { + message: format!("failed to start request cleanup reaper: {message}"), + }), + } +} diff --git a/vllm-cpp/tests/qwen3.rs b/vllm-cpp/tests/qwen3.rs index 5e6bee0..967f49e 100644 --- a/vllm-cpp/tests/qwen3.rs +++ b/vllm-cpp/tests/qwen3.rs @@ -1,7 +1,12 @@ use std::path::{Path, PathBuf}; -use std::sync::{Mutex, OnceLock}; +use std::sync::{mpsc, Arc, Barrier, Mutex, OnceLock}; +use std::thread; +use std::time::{Duration, Instant}; -use vllm_cpp::{Engine, FinishReason, SamplingParams, StreamControl, StructuredOutput}; +use vllm_cpp::{ + Engine, Error, FinishReason, Request, RequestOutcome, SamplingParams, StreamControl, + StructuredOutput, +}; const REQUIRED_MODEL_FILES: [&str; 4] = [ "model.safetensors", @@ -31,22 +36,38 @@ fn with_engine(test: impl FnOnce(&Engine, &Path)) { eprintln!("skipping model test; set VLLM_CPP_TEST_MODEL with `just setup-test-model`"); return; }; + if std::env::var_os("VLLM_CPP_TEST_ISOLATED_ENGINE").is_some() { + let engine = load_engine(&path); + test(&engine, &path); + return; + } static ENGINE: OnceLock> = OnceLock::new(); - let engine = ENGINE.get_or_init(|| { - Mutex::new( - Engine::builder(&path) - .num_blocks(64) - .max_model_len(256) - .max_num_seqs(2) - .max_num_batched_tokens(256) - .load() - .expect("load Qwen3-0.6B"), - ) - }); + let engine = ENGINE.get_or_init(|| Mutex::new(load_engine(&path))); let engine = engine.lock().expect("model test engine lock"); test(&engine, &path); } +fn load_engine(path: &Path) -> Engine { + Engine::builder(path) + .num_blocks(64) + .max_model_len(256) + .max_num_seqs(2) + .max_num_batched_tokens(256) + .load() + .expect("load Qwen3-0.6B") +} + +fn wait_until_done(request: &Request) { + let deadline = Instant::now() + Duration::from_secs(180); + while !request.is_done() { + assert!( + Instant::now() < deadline, + "request did not finish before timeout" + ); + thread::sleep(Duration::from_millis(1)); + } +} + #[test] fn greedy_completion_and_streaming_match() { with_engine(|engine, _| { @@ -110,6 +131,404 @@ fn early_stop_and_callback_panic_leave_engine_reusable() { }); } +#[test] +fn concurrent_requests_batch_with_correct_output() { + with_engine(|engine, _| { + let params = SamplingParams::greedy().max_tokens(12); + let expected_first = engine + .complete("The capital of France is", ¶ms) + .expect("first reference completion") + .text; + let expected_second = engine + .complete("The capital of Germany is", ¶ms) + .expect("second reference completion") + .text; + + let first_text = Arc::new(Mutex::new(String::new())); + let second_text = Arc::new(Mutex::new(String::new())); + let mut first = engine + .submit("The capital of France is", ¶ms, { + let text = Arc::clone(&first_text); + move |event| { + text.lock() + .expect("first callback text") + .push_str(&event.delta); + StreamControl::Continue + } + }) + .expect("submit first request"); + let mut second = engine + .submit("The capital of Germany is", ¶ms, { + let text = Arc::clone(&second_text); + move |event| { + text.lock() + .expect("second callback text") + .push_str(&event.delta); + StreamControl::Continue + } + }) + .expect("submit second request"); + + assert_eq!(first.wait().expect("wait first"), RequestOutcome::Completed); + assert_eq!( + second.wait().expect("wait second"), + RequestOutcome::Completed + ); + assert_eq!(*first_text.lock().expect("first result"), expected_first); + assert_eq!(*second_text.lock().expect("second result"), expected_second); + assert_eq!(first.native_error().expect("first native error"), None); + assert_eq!(second.native_error().expect("second native error"), None); + }); +} + +#[test] +fn engine_clones_submit_and_wait_from_multiple_rust_threads() { + with_engine(|engine, _| { + let start = Arc::new(Barrier::new(3)); + let workers = ["The capital of France is", "The capital of Germany is"] + .into_iter() + .map(|prompt| { + let engine = engine.clone(); + let start = Arc::clone(&start); + thread::spawn(move || { + let output = Arc::new(Mutex::new(String::new())); + start.wait(); + let mut request = engine + .submit(prompt, &SamplingParams::greedy().max_tokens(8), { + let output = Arc::clone(&output); + move |event| { + output + .lock() + .expect("cross-thread callback output") + .push_str(&event.delta); + StreamControl::Continue + } + }) + .expect("submit from Rust worker thread"); + let outcome = request.wait().expect("wait on Rust worker thread"); + let native_error = request + .native_error() + .expect("native error on Rust worker thread"); + let output = output.lock().expect("cross-thread output").clone(); + (outcome, native_error, output) + }) + }) + .collect::>(); + + start.wait(); + for worker in workers { + let (outcome, native_error, output) = worker.join().expect("Rust request worker"); + assert_eq!(outcome, RequestOutcome::Completed); + assert_eq!(native_error, None); + assert!(!output.is_empty()); + } + }); +} + +#[test] +fn live_request_moves_to_rust_thread_for_cancel_wait_and_drop() { + with_engine(|engine, _| { + let callback_release = Arc::new(Barrier::new(2)); + let (callback_started_sender, callback_started_receiver) = mpsc::channel(); + let (callback_drop_sender, callback_drop_receiver) = mpsc::channel(); + struct CallbackDropProbe(mpsc::Sender); + impl Drop for CallbackDropProbe { + fn drop(&mut self) { + let _ = self.0.send(thread::current().id()); + } + } + + let mut callback_started_sender = Some(callback_started_sender); + let request = engine + .submit( + "Write a long numbered list:", + &SamplingParams::greedy().max_tokens(64), + { + let callback_release = Arc::clone(&callback_release); + let drop_probe = CallbackDropProbe(callback_drop_sender); + move |_| { + let _ = &drop_probe; + if let Some(sender) = callback_started_sender.take() { + sender.send(()).expect("report live callback"); + callback_release.wait(); + } + StreamControl::Continue + } + }, + ) + .expect("submit request before moving it"); + + let worker = thread::spawn(move || { + let mut request = request; + callback_started_receiver + .recv_timeout(Duration::from_secs(180)) + .expect("callback starts while request is live"); + let cancel_result = request.cancel(); + callback_release.wait(); + cancel_result.expect("cancel moved request"); + let outcome = request.wait().expect("wait for moved request"); + let native_error = request.native_error().expect("moved request error"); + let worker_thread = thread::current().id(); + drop(request); + (worker_thread, outcome, native_error) + }); + + let (worker_thread, outcome, native_error) = worker.join().expect("moved request worker"); + assert!(matches!( + outcome, + RequestOutcome::Cancelled | RequestOutcome::Completed + )); + assert_eq!(native_error, None); + assert_eq!( + callback_drop_receiver + .recv_timeout(Duration::from_secs(30)) + .expect("callback state drops with moved request"), + worker_thread + ); + }); +} + +#[test] +fn request_outcomes_and_probes_are_precise() { + with_engine(|engine, _| { + let params = SamplingParams::greedy().max_tokens(64); + let mut stopped = engine + .submit("Count upward forever:", ¶ms, |_| StreamControl::Stop) + .expect("submit callback-stop request"); + assert_eq!( + stopped.wait().expect("wait callback-stop request"), + RequestOutcome::StoppedByCallback + ); + assert!(stopped.is_done()); + assert!(stopped.is_done()); + assert_eq!( + stopped.wait().expect("repeat callback-stop wait"), + RequestOutcome::StoppedByCallback + ); + + let mut terminal_stopped = engine + .submit( + "Say hello", + &SamplingParams::greedy().max_tokens(1), + |event| { + if event.finished { + StreamControl::Stop + } else { + StreamControl::Continue + } + }, + ) + .expect("submit terminal callback-stop request"); + assert_eq!( + terminal_stopped + .wait() + .expect("wait terminal callback-stop request"), + RequestOutcome::StoppedByCallback + ); + + let barrier = Arc::new(Barrier::new(2)); + let mut cancelled = engine + .submit("Write a long numbered list:", ¶ms, { + let barrier = Arc::clone(&barrier); + let mut first = true; + move |_| { + if first { + first = false; + barrier.wait(); + thread::sleep(Duration::from_millis(20)); + } + StreamControl::Continue + } + }) + .expect("submit cancellable request"); + barrier.wait(); + cancelled.cancel().expect("first cancel"); + cancelled.cancel().expect("idempotent cancel"); + assert_eq!( + cancelled.wait().expect("wait cancelled request"), + RequestOutcome::Cancelled + ); + assert!(cancelled.is_done()); + assert!(cancelled.is_done()); + assert_eq!(cancelled.native_error().expect("cancel native error"), None); + }); +} + +#[test] +fn callback_panic_is_reported_and_engine_is_reusable() { + with_engine(|engine, _| { + let params = SamplingParams::greedy().max_tokens(16); + let mut request = engine + .submit("Say hello", ¶ms, |_| { + panic!("intentional async callback panic") + }) + .expect("submit panic request"); + assert_eq!(request.wait().unwrap_err(), Error::CallbackPanicked); + assert_eq!(request.wait().unwrap_err(), Error::CallbackPanicked); + + let completion = engine + .complete("Say hello", &SamplingParams::greedy().max_tokens(2)) + .expect("engine remains reusable after async panic"); + assert!(!completion.text.is_empty()); + }); +} + +#[test] +fn request_retains_engine_and_live_drop_is_safe() { + let Some(path) = model_path() else { + eprintln!("skipping model test; set VLLM_CPP_TEST_MODEL using `just setup-test-model`"); + return; + }; + let engine = Engine::builder(path) + .num_blocks(64) + .max_model_len(256) + .max_num_seqs(2) + .max_num_batched_tokens(256) + .load() + .expect("load drop-order engine"); + let params = SamplingParams::greedy().max_tokens(64); + let final_clone = engine.clone(); + let live = engine + .submit("Write a long numbered list:", ¶ms, |_| { + thread::sleep(Duration::from_millis(1)); + StreamControl::Continue + }) + .expect("submit live request"); + drop(engine); + drop(live); + let completion = final_clone + .complete("Say hello", &SamplingParams::greedy().max_tokens(2)) + .expect("final public engine clone remains usable"); + assert!(!completion.text.is_empty()); + + let retained = final_clone + .submit( + "Count from one:", + &SamplingParams::greedy().max_tokens(4), + |_| StreamControl::Continue, + ) + .expect("submit engine-retaining request"); + drop(final_clone); + let mut retained = retained; + assert_eq!( + retained.wait().expect("request outlives public engines"), + RequestOutcome::Completed + ); +} + +#[test] +fn callback_thread_self_wait_is_rejected_and_self_drop_is_deferred() { + with_engine(|engine, _| { + let params = SamplingParams::greedy().max_tokens(16); + let slot = Arc::new(Mutex::new(None::)); + let wait_error = Arc::new(Mutex::new(None)); + let callback_finished = Arc::new(Barrier::new(2)); + let callback_started = Arc::new(Barrier::new(2)); + let callback_thread = Arc::new(Mutex::new(None)); + let (drop_sender, drop_receiver) = mpsc::channel(); + struct CallbackDropProbe(mpsc::Sender); + impl Drop for CallbackDropProbe { + fn drop(&mut self) { + let _ = self.0.send(thread::current().id()); + } + } + let request = engine + .submit("Count from one:", ¶ms, { + let slot = Arc::clone(&slot); + let wait_error = Arc::clone(&wait_error); + let callback_started = Arc::clone(&callback_started); + let callback_finished = Arc::clone(&callback_finished); + let callback_thread = Arc::clone(&callback_thread); + let drop_probe = CallbackDropProbe(drop_sender); + move |_| { + let _ = &drop_probe; + *callback_thread.lock().expect("callback thread") = + Some(thread::current().id()); + callback_started.wait(); + if let Some(mut request) = slot.lock().expect("request slot").take() { + *wait_error.lock().expect("wait result") = + Some(request.wait().unwrap_err()); + drop(request); + callback_finished.wait(); + } + StreamControl::Stop + } + }) + .expect("submit self-lifecycle request"); + *slot.lock().expect("request slot") = Some(request); + callback_started.wait(); + callback_finished.wait(); + assert_eq!( + wait_error.lock().expect("wait result").take(), + Some(Error::RequestCallbackThread { operation: "wait" }) + ); + let callback_thread = callback_thread + .lock() + .expect("callback thread") + .expect("callback thread recorded"); + let cleanup_thread = drop_receiver + .recv_timeout(Duration::from_secs(30)) + .expect("deferred callback state drop"); + assert_ne!(cleanup_thread, callback_thread); + }); +} + +#[test] +fn concurrent_request_lifecycle_stress() { + with_engine(|engine, _| { + let params = SamplingParams::greedy().max_tokens(16); + for round in 0..16 { + let mut requests = Vec::new(); + for index in 0..4 { + let request = engine + .submit( + &format!("Round {round}, item {index}:"), + ¶ms, + move |event| { + if (round + index) % 4 != 0 + && (round + index) % 5 == 0 + && !event.finished + { + StreamControl::Stop + } else { + StreamControl::Continue + } + }, + ) + .expect("submit stress request"); + requests.push(request); + } + + for (index, mut request) in requests.drain(..).enumerate() { + match (round + index) % 4 { + 0 => { + request.cancel().expect("stress cancel"); + request.cancel().expect("stress repeated cancel"); + assert!(matches!( + request.wait().expect("wait stress cancel"), + RequestOutcome::Cancelled | RequestOutcome::Completed + )); + } + 1 => { + wait_until_done(&request); + assert!(request.is_done()); + request.wait().expect("wait probed stress request"); + } + 2 => { + request.wait().expect("wait stress request"); + } + _ => drop(request), + } + } + } + + let completion = engine + .complete("Say hello", &SamplingParams::greedy().max_tokens(2)) + .expect("engine remains reusable after lifecycle stress"); + assert!(!completion.text.is_empty()); + }); +} + #[test] fn structured_choice_is_enforced() { with_engine(|engine, _| { diff --git a/vllm-cpp/tests/safe_api.rs b/vllm-cpp/tests/safe_api.rs index 8ce090f..8e25025 100644 --- a/vllm-cpp/tests/safe_api.rs +++ b/vllm-cpp/tests/safe_api.rs @@ -1,4 +1,9 @@ -use vllm_cpp::{Engine, Error, SchedulerPolicy, Toggle}; +use static_assertions::{assert_impl_all, assert_not_impl_any}; +use vllm_cpp::{Engine, Error, Request, SchedulerPolicy, Toggle}; + +assert_impl_all!(Engine: Send, Sync, Clone); +assert_impl_all!(Request: Send); +assert_not_impl_any!(Request: Sync); fn missing_model() -> &'static str { "/nonexistent/vllm-cpp-rs-safe-api-model"