feat: add concurrent request API

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=<verified-model-directory>' >&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

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", &params)?;
 println!("{}", completion.text);
+
+let mut request = engine.submit("The capital of Germany is", &params, |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 -- <model-directory>
 cargo run -p vllm-cpp --example stream -- <model-directory>
+cargo run -p vllm-cpp --example concurrent -- <model-directory>
 cargo run -p vllm-cpp --example chat -- <model-directory>
 cargo run -p vllm-cpp --example structured -- <model-directory>
 ```
@@ -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
 

vllm-cpp-sys/build.rs

@@ -233,10 +233,9 @@ fn sanitizer_config(bundled: bool) -> Option<String> {
         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() {

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"

vllm-cpp/examples/concurrent.rs

@@ -0,0 +1,26 @@
+use std::io::{self, Write};
+
+use vllm_cpp::{Engine, SamplingParams, StreamControl};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let model = std::env::args_os()
+        .nth(1)
+        .ok_or("usage: concurrent <model-directory>")?;
+    let engine = Engine::load(model)?;
+    let params = SamplingParams::greedy().max_tokens(16);
+
+    let mut france = engine.submit("The capital of France is", &params, |event| {
+        print!("[france] {}", event.delta);
+        io::stdout().flush().expect("flush stdout");
+        StreamControl::Continue
+    })?;
+    let mut germany = engine.submit("The capital of Germany is", &params, |event| {
+        print!("[germany] {}", event.delta);
+        io::stdout().flush().expect("flush stdout");
+        StreamControl::Continue
+    })?;
+
+    println!("\nfrance: {:?}", france.wait()?);
+    println!("germany: {:?}", germany.wait()?);
+    Ok(())
+}

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<ffi::vllm_engine>,
+    pub(crate) inner: Arc<EngineInner>,
+}
+
+pub(crate) struct EngineInner {
+    pub(crate) raw: NonNull<ffi::vllm_engine>,
 }
 
 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::<F>),
@@ -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::<F>),
                 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 }),
+        })
     }
 }
 

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}")
             }