From 6b6fd97758db72546923e704cb9f3b399adbc602 Mon Sep 17 00:00:00 2001 From: Ivan Zatevakhin Date: Wed, 19 Aug 2026 00:26:51 +0100 Subject: [PATCH 1/2] feat: add custom logits processors and native diagnostics add panic-contained custom logits processing and expose the linked native version while documenting priority scheduler behavior. replace the legacy test-model downloader with the pinned hugging face resolver and keep ordinary tests network-free. --- CHANGELOG.md | 4 +- Justfile | 86 ++----- README.md | 12 +- vllm-cpp/README.md | 11 +- vllm-cpp/examples/README.md | 3 +- vllm-cpp/examples/setup_test_model.rs | 23 ++ vllm-cpp/src/engine.rs | 47 +++- vllm-cpp/src/error.rs | 5 +- vllm-cpp/src/lib.rs | 33 ++- vllm-cpp/src/params.rs | 321 +++++++++++++++++++++++++- vllm-cpp/src/request.rs | 107 ++++++--- vllm-cpp/tests/qwen3.rs | 123 ++++++++-- vllm-cpp/tests/safe_api.rs | 2 + 13 files changed, 641 insertions(+), 136 deletions(-) create mode 100644 vllm-cpp/examples/setup_test_model.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 31be1a7..cb29658 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ All notable changes to this project will be documented in this file. ### Added - Checked-in raw Rust declarations for the 19-symbol stable vllm.cpp C API at ABI version 10, with header, symbol, layout, and runtime conformance checks. -- A safe API for model loading, blocking completion and streaming, raw-JSON and optional serde chat, structured output, owned sampling parameters, and concurrent request submission, cancellation, waiting, and diagnostics. +- A safe API for model loading, blocking completion and streaming, raw-JSON and optional serde chat, structured output, owned sampling parameters, panic-contained custom logits processors, native version diagnostics, and concurrent request submission, cancellation, waiting, and diagnostics. - An always-available synchronous `hf-hub` resolver for standalone GGUF files and runtime-complete sparse Safetensors snapshots, defaulting to the Hub's mutable `main` revision with an explicit branch/tag/commit override, cache/token/progress/offline controls, and no async runtime. - Consistent local, Hugging Face GGUF, and Hugging Face Safetensors model-source arguments across every runnable example, with cache reuse and optional revisions; plus a weather extraction example and model-backed test using JSON-Schema structured output. - A Clap-based interactive `chat` example with prompt/file startup input, retained system/user/assistant history, supported sampling controls, default streaming or blocking output, and shared local/Hugging Face resolution. @@ -24,6 +24,8 @@ All notable changes to this project will be documented in this file. ### Known limitations +- The priority scheduler is selectable, but the ABI v10 submission surface assigns every request priority zero; safe requests therefore tie by arrival until a future ABI carries per-request priority. + - The supported runtime tier is native Linux x86_64 CPU. Accelerator features are experimental build/configuration surfaces, not runtime-support claims. - Known native blockers include a CUDA teardown SIGSEGV after otherwise successful tests, a CUDA bf16 numerical tolerance failure, CUTLASS concurrent-output differences, incomplete Vulkan attention/model runtime, and external MLX deployment plus unvalidated release-lane model/runtime behavior. - The hosted Metal lane checks compile/link only, the software Vulkan lane checks backend/ops only, and accelerator builds do not establish runtime correctness. diff --git a/Justfile b/Justfile index 08af319..2d96acc 100644 --- a/Justfile +++ b/Justfile @@ -2,7 +2,6 @@ set shell := ["bash", "-euo", "pipefail", "-c"] root := justfile_directory() bindings_file := root + "/vllm-cpp-sys/src/bindings.rs" -model_revision := "c1899de289a04d12100db370d81485cdf75e47ca" # Maintainer workflows require Just 1.40 or newer. @@ -491,6 +490,7 @@ package-test: examples/common/mod.rs \ examples/complete.rs \ examples/concurrent.rs \ + examples/setup_test_model.rs \ examples/stream.rs \ examples/structured.rs \ src/callback.rs \ @@ -757,13 +757,16 @@ package-test: EOF cat > "$safe_consumer/src/main.rs" <<'EOF' use vllm_cpp::{ - abi_version, expected_abi_version, Engine, Error, HuggingFaceModel, SamplingParams, + abi_version, expected_abi_version, version, Engine, Error, HuggingFaceModel, SamplingParams, }; fn main() { assert_eq!(expected_abi_version(), 10); assert_eq!(abi_version(), 10); - let _params = SamplingParams::greedy().max_tokens(1); + assert!(!version().expect("native version").is_empty()); + let _params = SamplingParams::greedy() + .max_tokens(1) + .logits_processor(|_, logits| logits.fill(0.0)); let resolver = HuggingFaceModel::gguf("owner/model", "model.gguf") .revision("revision") .cache_dir("/nonexistent/vllm-cpp-rs-safe-package-hf-cache") @@ -804,45 +807,12 @@ publish-dry-run: # dry-run preserves Cargo's sys-first order without requiring sys on crates.io. cargo publish --workspace --locked --dry-run --allow-dirty --no-verify -# Download and verify the pinned Qwen3-0.6B model fixture. -setup-test-model destination=env_var_or_default("VLLM_CPP_TEST_MODEL", env_var_or_default("XDG_CACHE_HOME", env_var("HOME") + "/.cache") + "/vllm-cpp-rs/Qwen3-0.6B-" + model_revision): +# Resolve the pinned Qwen3-0.6B test fixture into the standard Hugging Face cache. +setup-test-model: #!/usr/bin/env bash set -euo pipefail - revision={{ quote(model_revision) }} - base="https://huggingface.co/Qwen/Qwen3-0.6B/resolve/$revision" - destination={{ quote(destination) }} - mkdir -p "$destination" - - files=( - LICENSE - config.json - generation_config.json - merges.txt - model.safetensors - tokenizer.json - tokenizer_config.json - vocab.json - ) - for file in "${files[@]}"; do - if [[ ! -f $destination/$file ]]; then - echo "downloading $file" >&2 - curl --fail --location --retry 3 --continue-at - \ - "$base/$file" --output "$destination/$file" - fi - done - - cat > "$destination/SHA256SUMS.expected" <<'EOF' - 832dd9e00a68dd83b3c3fb9f5588dad7dcf337a0db50f7d9483f310cd292e92e LICENSE - 660db3b73d788119c04535e48cf9be5f55bc3100841a718637ae695b442f27dd config.json - 2325da0f15bb848e018c5ae071b7943332e9f871d6b60e2ed22ca97d4cb993d2 generation_config.json - 8831e4f1a044471340f7c0a83d7bd71306a5b867e95fd870f74d0c5308a904d5 merges.txt - f47f71177f32bcd101b7573ec9171e6a57f4f4d31148d38e382306f42996874b model.safetensors - aeb13307a71acd8fe81861d94ad54ab689df773318809eed3cbe794b4492dae4 tokenizer.json - d5d09f07b48c3086c508b30d1c9114bd1189145b74e982a265350c923acd8101 tokenizer_config.json - ca10d7e9fb3ed18575dd1e277a2579c16d108e32f27439684afa0e10b1440910 vocab.json - EOF - (cd "$destination" && sha256sum --check SHA256SUMS.expected) >&2 - printf '%s\n' "$destination" + cd {{ quote(root) }} + cargo run --quiet --locked -p vllm-cpp --example setup_test_model # Run the full safe/request/model suites with ASan, UBSan, and leak detection. sanitizers model=env_var_or_default("VLLM_CPP_TEST_MODEL", ""): @@ -850,23 +820,11 @@ sanitizers model=env_var_or_default("VLLM_CPP_TEST_MODEL", ""): set -euo pipefail model={{ quote(model) }} if [[ -z $model ]]; then - echo 'set VLLM_CPP_TEST_MODEL or pass model=' >&2 + 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 + if [[ ! -d $model ]]; then + echo "model fixture is not a directory: $model" >&2 exit 1 fi cd {{ quote(root) }} @@ -908,23 +866,11 @@ tsan model=env_var_or_default("VLLM_CPP_TEST_MODEL", ""): fi model={{ quote(model) }} if [[ -z $model ]]; then - echo 'set VLLM_CPP_TEST_MODEL or pass model=' >&2 + 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 + if [[ ! -d $model ]]; then + echo "model fixture is not a directory: $model" >&2 exit 1 fi cd {{ quote(root) }} diff --git a/README.md b/README.md index 995e1d9..de85cc5 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ Native builds require: - Ninja or another CMake build tool. - A C11 and C++20 compiler. - A system linker and C++ standard library. -- Just 1.40 or newer for maintainer workflows, plus Git, `jq`, GNU tar, and `curl` for the model fixture recipe. +- Just 1.40 or newer for maintainer workflows, plus Git, `jq`, and GNU tar. This repository provides a Nix development shell with the pinned development tools. Linux also has minimal CUDA and Vulkan shells: @@ -43,13 +43,13 @@ git submodule update --init --recursive The packaged [`vllm-cpp` guide](vllm-cpp/README.md) covers local and Hugging Face model resolution, safe ownership, callbacks, concurrency, features, link modes, and deployment. The [`vllm-cpp-sys` guide](vllm-cpp-sys/README.md) documents the raw ABI and native build boundary. -`Engine::load` accepts a native-compatible model directory or standalone GGUF. `HuggingFaceModel` synchronously resolves into the normal Hugging Face cache before engine construction, defaulting to the Hub's mutable `main` revision; `.revision(...)` can pin a branch, tag, or commit. GGUF mode selects one safe root file. Safetensors mode pins downloads to repository metadata's commit SHA and retrieves only native runtime requirements: root configuration/tokenizer files and either unsharded weights or an index plus all root shards. Every runnable example accepts a bare or explicit local path and both Hub artifact forms with optional `--revision`. Cached downloads are reused. Retrieval does not prove model/backend compatibility. +`Engine::load` accepts a native-compatible model directory or standalone GGUF. `HuggingFaceModel` synchronously resolves into the normal Hugging Face cache before engine construction, defaulting to the Hub's mutable `main` revision; `.revision(...)` can pin a branch, tag, or commit. GGUF mode selects one safe root file. Safetensors mode pins downloads to repository metadata's commit SHA and retrieves only native runtime requirements: root configuration/tokenizer files and either unsharded weights or an index plus all root shards. Every inference example accepts a bare or explicit local path and both Hub artifact forms with optional `--revision`. Cached downloads are reused. Retrieval does not prove model/backend compatibility. -`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. +`EngineBuilder` owns model settings and converts them to temporary C strings only for the load call. `SamplingParams` owns stop strings, structured constraints, and optional `Send + Sync` custom logits processors. Processor panics are contained before the C boundary and reported through Rust errors; processor-backed generation must be bounded because ABI v10 has no callback abort channel. Each processor invocation retains its state until the engine is dropped because ABI v10 has no sampler-quiescence primitive. `version()` copies the linked native diagnostic version string. Completion and chat strings are copied into Rust values before the matching native free function runs. `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. +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. `SchedulerPolicy::Priority` selects the native queue, but every ABI v10 safe submission currently has priority zero and therefore ties by arrival. See [the examples guide](vllm-cpp/examples/README.md) for ordinary Linux and optional Nix setup, commands for every example, and the interactive chat CLI's local/Hub model forms and generation options. Release-facing changes are recorded in the [changelog](CHANGELOG.md), and maintainers use the manual [release process](RELEASING.md). @@ -105,7 +105,7 @@ Compilation does not establish runtime correctness. Known native evidence blocke ## 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 15 blocking and request-lifecycle model tests serially, including choice and JSON-Schema structured-output enforcement: +Model-backed tests use Apache-2.0 `Qwen/Qwen3-0.6B` at pinned revision `c1899de289a04d12100db370d81485cdf75e47ca`. Explicitly resolve its complete Safetensors snapshot into the standard Hugging Face cache, then run exactly 18 blocking and request-lifecycle model tests serially, including choice and JSON-Schema structured-output enforcement: ```console model=$(just setup-test-model) @@ -113,7 +113,7 @@ VLLM_CPP_TEST_MODEL="$model" \ cargo test --locked -p vllm-cpp --release --test qwen3 -- --test-threads=1 ``` -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. +`just setup-test-model` is the only explicit test-fixture acquisition step. It uses `HuggingFaceModel` with the immutable revision above, honors normal `HF_HOME` and Hugging Face authentication, reuses the standard cache, and prints the resolved directory. The approximately 1.5 GB model is not included in repository or crate packages. Ordinary tests, sanitizers, and TSan never resolve or download models: `VLLM_CPP_TEST_MODEL` must name an externally prepared model directory. Model-backed tests skip with an explanatory message when it is unset; when set, tests and instrumentation recipes require it to be a directory. 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. diff --git a/vllm-cpp/README.md b/vllm-cpp/README.md index 4c0f5ae..e15b6d3 100644 --- a/vllm-cpp/README.md +++ b/vllm-cpp/README.md @@ -14,7 +14,7 @@ println!("{}", completion.text); # Ok::<(), vllm_cpp::Error>(()) ``` -`Engine::load` accepts either a model directory or a standalone GGUF file understood by the pinned native engine. The known-good Safetensors test layout contains `model.safetensors`, `config.json`, `tokenizer.json`, and `tokenizer_config.json`; model-family compatibility remains a native vllm.cpp concern. See the packaged [examples guide](examples/README.md) for local and Hugging Face loading, blocking completion, streaming, JSON-Schema output, concurrent-request commands, and the Clap-based interactive chat CLI with retained history and supported sampling controls. +`Engine::load` accepts either a model directory or a standalone GGUF file understood by the pinned native engine. `HuggingFaceModel::safetensors` resolves and validates a complete loader snapshot, including unsharded or indexed weights; model-family compatibility remains a native vllm.cpp concern. See the packaged [examples guide](examples/README.md) for local and Hugging Face loading, blocking completion, streaming, JSON-Schema output, concurrent-request commands, and the Clap-based interactive chat CLI with retained history and supported sampling controls. ## Hugging Face models @@ -37,15 +37,20 @@ GGUF mode retrieves one root-level lowercase `.gguf` file and rejects split sets Retrieval validates cache and snapshot completeness; it does not prove that the pinned native engine supports the repository's model architecture, tokenizer, quantization, or backend. +In the repository checkout, `just setup-test-model` explicitly resolves `Qwen/Qwen3-0.6B` at immutable revision `c1899de289a04d12100db370d81485cdf75e47ca`, reusing the standard Hugging Face cache and honoring normal `HF_HOME` and authentication. Ordinary tests and instrumentation commands never resolve or download this fixture. `VLLM_CPP_TEST_MODEL` remains an external contract: when supplied, it must point to a prepared model directory; model tests skip when it is unset. + ## API and ownership - `EngineBuilder` configures and loads a model. `Engine` is `Clone + Send + Sync`; clones share one reference-counted native engine. -- `SamplingParams` owns stop strings and structured constraints. Completion, chat, error, and stream text is copied into Rust-owned values before native storage is released or reused. +- `SamplingParams` owns stop strings, structured constraints, and an optional `Send + Sync` custom logits processor. The processor receives generated token IDs and a mutable logits row each decode step; panics are contained and returned as `Error::LogitsProcessorPanicked`. Processor-backed generation must have a finite `max_tokens` bound because ABI v10 cannot abort from that callback. Each processor invocation retains its state until the engine is dropped because ABI v10 has no sampler-quiescence primitive. +- Completion, chat, error, and stream text is copied into Rust-owned values before native storage is released or reused. - Blocking `complete`, `complete_stream`, `chat_json`, and `chat_stream_json` calls keep borrowed callbacks alive only for the call. Callback panics are caught before crossing C and resumed after the native call returns. - `Engine::submit` returns a `Request` before generation finishes. A request retains its engine and callback until native free/join completes, is `Send`, and is deliberately not `Sync`. - Asynchronous callbacks run on a native delivery thread and must be `Send + 'static`. `wait` reports callback panics as `Error::CallbackPanicked`; waiting or freeing from that same callback thread is prohibited by ABI v10, so callback-thread drop transfers cleanup to a prestarted reaper. - Dropping a live request cancels and joins it. `cancel` is idempotent, `wait` reports the request outcome, and `native_error` copies the request-owned diagnostic after completion into an owned Rust `String`; the native storage remains valid until the request is dropped or freed. +`SchedulerPolicy::Priority` selects the native priority queue, but ABI v10 assigns priority zero to every safe blocking, streaming, chat, and asynchronous submission. Until a future ABI carries per-request priority, safe requests therefore remain ordered by arrival. + ## Features and linking | Feature | Purpose | @@ -67,7 +72,7 @@ Hugging Face resolution is not a Cargo feature: synchronous `hf-hub` support is ## ABI and deployment -This crate is tied to the exact same `vllm-cpp-sys` crate version and the pinned vllm.cpp commit `34aedfbe8ed9779697905541a62e2160ccfd9c05`. Model loading requires exact C ABI version 10 before any versioned struct crosses FFI. A system library must implement the same ABI; the consumer build checks for its header, while maintainer conformance tests check layout and symbols. +This crate is tied to the exact same `vllm-cpp-sys` crate version and the pinned vllm.cpp commit `34aedfbe8ed9779697905541a62e2160ccfd9c05`. Model loading requires exact C ABI version 10 before any versioned struct crosses FFI. `version()` copies the linked library's diagnostic native version string, while `abi_version()` remains the compatibility authority. A system library must implement the same ABI; the consumer build checks for its header, while maintainer conformance tests check layout and symbols. Static bundled builds include the native archive in the application link. Dynamic bundled or system builds do not deploy `libvllm.so`/`libvllm.dylib`: install it and its backend/toolkit dependencies in a loader-visible location using `LD_LIBRARY_PATH`, `DYLD_LIBRARY_PATH`, rpath supplied by the application, or the system loader configuration. System mode uses `VLLM_CPP_ROOT`; `VLLM_CPP_LIB_DIR` can choose a nonstandard library directory. System static linking also requires the matching `libblake3_vendored.a` through `VLLM_CPP_BLAKE3_LIB_DIR` or the selected vllm library directory. diff --git a/vllm-cpp/examples/README.md b/vllm-cpp/examples/README.md index 3eac25b..cca4186 100644 --- a/vllm-cpp/examples/README.md +++ b/vllm-cpp/examples/README.md @@ -1,6 +1,6 @@ # examples -these examples exercise the safe `vllm-cpp` api. four use fixed prompts and settings; `chat` is an interactive command-line application: +five examples exercise the safe `vllm-cpp` api: four use fixed prompts and settings, and `chat` is an interactive command-line application. one additional maintainer utility prepares the pinned test fixture: | example | behavior | |---|---| @@ -9,6 +9,7 @@ these examples exercise the safe `vllm-cpp` api. four use fixed prompts and sett | [`concurrent`](concurrent.rs) | submits two asynchronous streaming requests and waits for both | | [`chat`](chat.rs) | runs a Clap-based interactive chat with conversation history and streaming output | | [`structured`](structured.rs) | extracts a fixed weather report under a JSON Schema | +| [`setup_test_model`](setup_test_model.rs) | resolves the pinned Qwen test fixture for `just setup-test-model`; not a general inference CLI | The four fixed examples (`complete`, `stream`, `concurrent`, and `structured`) accept the same manual model-source forms: diff --git a/vllm-cpp/examples/setup_test_model.rs b/vllm-cpp/examples/setup_test_model.rs new file mode 100644 index 0000000..55e21f7 --- /dev/null +++ b/vllm-cpp/examples/setup_test_model.rs @@ -0,0 +1,23 @@ +use std::process::ExitCode; + +use vllm_cpp::HuggingFaceModel; + +const REPOSITORY: &str = "Qwen/Qwen3-0.6B"; +const REVISION: &str = "c1899de289a04d12100db370d81485cdf75e47ca"; + +fn main() -> ExitCode { + match HuggingFaceModel::safetensors(REPOSITORY) + .revision(REVISION) + .progress(true) + .resolve() + { + Ok(path) => { + println!("{}", path.display()); + ExitCode::SUCCESS + } + Err(error) => { + eprintln!("could not set up test model: {error}"); + ExitCode::FAILURE + } + } +} diff --git a/vllm-cpp/src/engine.rs b/vllm-cpp/src/engine.rs index bef72fd..acf3a18 100644 --- a/vllm-cpp/src/engine.rs +++ b/vllm-cpp/src/engine.rs @@ -3,7 +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 std::sync::{Arc, Mutex}; use vllm_cpp_sys as ffi; @@ -11,7 +11,7 @@ use crate::callback::{ callback_trampoline, CallbackState, StreamControl, StreamEvent, StreamOutcome, }; use crate::error::{invalid_configuration, status_result, Error}; -use crate::params::{SamplingParams, SchedulerPolicy, Toggle}; +use crate::params::{LogitsProcessorState, SamplingParams, SchedulerPolicy, Toggle}; /// A cloneable vllm.cpp serving engine. #[derive(Clone)] @@ -21,6 +21,7 @@ pub struct Engine { pub(crate) struct EngineInner { pub(crate) raw: NonNull, + logits_processors: Mutex>>, } impl std::fmt::Debug for Engine { @@ -74,6 +75,14 @@ pub struct Completion { } impl Engine { + pub(crate) fn retain_logits_processor(&self, state: Arc) { + self.inner + .logits_processors + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(state); + } + /// Starts configuring an engine for a model directory or GGUF file. pub fn builder(model_path: impl Into) -> EngineBuilder { EngineBuilder::new(model_path) @@ -88,6 +97,9 @@ impl Engine { pub fn complete(&self, prompt: &str, params: &SamplingParams) -> Result { let prompt = to_cstring(prompt, "prompt")?; let params = params.marshal()?; + if let Some(logits_processor) = params.logits_processor() { + self.retain_logits_processor(logits_processor); + } let mut raw = MaybeUninit::::uninit(); // SAFETY: the engine is owned and live, all pointers remain valid for the // call, and out storage is initialized by native code on success. @@ -99,10 +111,19 @@ impl Engine { raw.as_mut_ptr(), ) }; - status_result(status)?; + if status != ffi::vllm_status_VLLM_OK { + if let Some(error) = params.logits_processor_error() { + return Err(error); + } + status_result(status)?; + unreachable!("non-OK native status unexpectedly succeeded"); + } // SAFETY: VLLM_OK initializes every completion field. let raw = unsafe { raw.assume_init() }; let guard = CompletionGuard(raw); + if let Some(error) = params.logits_processor_error() { + return Err(error); + } completion_from_raw(&guard.0) } @@ -121,6 +142,9 @@ impl Engine { { let prompt = to_cstring(prompt, "prompt")?; let params = params.marshal()?; + if let Some(logits_processor) = params.logits_processor() { + self.retain_logits_processor(logits_processor); + } let mut state = CallbackState::new(&mut callback); // SAFETY: state has a stable stack address for this blocking call; the C // API does not retain user_data after returning. @@ -139,6 +163,9 @@ impl Engine { if let Some(error) = state.take_error() { return Err(error); } + if let Some(error) = params.logits_processor_error() { + return Err(error); + } status_result(status)?; Ok(StreamOutcome { stopped_by_callback: state.stopped(), @@ -210,8 +237,9 @@ impl Engine { impl Drop for EngineInner { fn drop(&mut self) { - // SAFETY: EngineInner exclusively owns this live handle and drops it once, - // after every Request-owned Arc has been released. + // SAFETY: EngineInner exclusively owns this live handle. Native teardown + // joins engine workers before retained logits states drop with the other + // fields, so no callback can outlive its user_data. unsafe { ffi::vllm_engine_free(self.raw.as_ptr()) }; } } @@ -304,6 +332,10 @@ impl EngineBuilder { self } + /// Selects the native admission queue policy. + /// + /// The stable C submission API currently assigns priority zero to every + /// request, so [`SchedulerPolicy::Priority`] orders safe requests by arrival. #[must_use] pub fn scheduler(mut self, value: SchedulerPolicy) -> Self { self.scheduler = value; @@ -370,7 +402,10 @@ impl EngineBuilder { message: "vllm_engine_load succeeded without a handle".to_owned(), })?; Ok(Engine { - inner: Arc::new(EngineInner { raw }), + inner: Arc::new(EngineInner { + raw, + logits_processors: Mutex::new(Vec::new()), + }), }) } } diff --git a/vllm-cpp/src/error.rs b/vllm-cpp/src/error.rs index d9a15ae..70501bd 100644 --- a/vllm-cpp/src/error.rs +++ b/vllm-cpp/src/error.rs @@ -60,8 +60,10 @@ pub enum Error { PathEncoding, /// Native code returned bytes that are not valid UTF-8. InvalidUtf8 { field: &'static str }, - /// An asynchronous callback panicked. + /// An asynchronous output callback panicked. CallbackPanicked, + /// A custom logits processor panicked. + LogitsProcessorPanicked, /// 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. @@ -93,6 +95,7 @@ impl fmt::Display for Error { 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::LogitsProcessorPanicked => write!(f, "custom logits processor panicked"), Self::RequestCallbackThread { operation } => { write!( f, diff --git a/vllm-cpp/src/lib.rs b/vllm-cpp/src/lib.rs index e4a422a..abf514c 100644 --- a/vllm-cpp/src/lib.rs +++ b/vllm-cpp/src/lib.rs @@ -4,8 +4,9 @@ //! //! Resolve a Hub model with [`HuggingFaceModel`] (default `main`, or an explicit //! revision), then create an [`Engine`] with [`Engine::load`] or configure native -//! model settings through [`EngineBuilder`]. [`SamplingParams`] owns sampling, stop-string, and -//! [`StructuredOutput`] settings for completion calls. The engine provides +//! model settings through [`EngineBuilder`]. [`SamplingParams`] owns sampling, +//! stop-string, [`StructuredOutput`] settings, and an optional host-side logits +//! processor for completion calls. The engine provides //! blocking completion, streaming, raw-JSON chat, and [`Engine::submit`] for a //! concurrent [`Request`]. Enable `serde` for `serde_json::Value` chat helpers. //! @@ -15,7 +16,9 @@ //! native engine. Rust copies completion, stream, chat, and error text before //! native storage is freed or reused. Blocking callbacks may borrow caller data. //! Their panics are caught before the C boundary and resumed after the native -//! call returns. +//! call returns. Custom logits processors are `Send + Sync`, may run concurrently +//! on native worker threads, and report contained panic through +//! [`Error::LogitsProcessorPanicked`]. //! //! A [`Request`] retains its engine and asynchronous callback until native //! free/join completes. Requests are `Send` but intentionally not `Sync`, while @@ -28,7 +31,8 @@ //! # ABI, linking, and deployment //! //! Engine loading requires the linked native library's ABI to equal -//! [`expected_abi_version`] before versioned structs cross FFI. The default +//! [`expected_abi_version`] before versioned structs cross FFI. [`version`] copies +//! the linked library's diagnostic version string. The default //! `bundled` feature builds the pinned native source. `system` selects a //! caller-provided installation, `dynamic-link` selects shared linking, and //! `serde` adds typed JSON helpers. The non-optional `hf-hub` dependency provides @@ -72,3 +76,24 @@ pub fn abi_version() -> i32 { // SAFETY: this base ABI function takes no pointers and returns a plain i32. unsafe { vllm_cpp_sys::vllm_abi_version() } } + +/// Copies the version string reported by the linked vllm.cpp library. +/// +/// This diagnostic does not replace [`abi_version`]: callers must still use the +/// numeric ABI for compatibility decisions. +pub fn version() -> Result { + // SAFETY: the base ABI returns a borrowed, process-lifetime C string. + let pointer = unsafe { vllm_cpp_sys::vllm_version() }; + if pointer.is_null() { + return Err(Error::Runtime { + message: "vllm_version returned a null pointer".to_owned(), + }); + } + // SAFETY: the native contract returns a live NUL-terminated string. + unsafe { std::ffi::CStr::from_ptr(pointer) } + .to_str() + .map(str::to_owned) + .map_err(|_| Error::InvalidUtf8 { + field: "native version", + }) +} diff --git a/vllm-cpp/src/params.rs b/vllm-cpp/src/params.rs index 6154f1d..31842df 100644 --- a/vllm-cpp/src/params.rs +++ b/vllm-cpp/src/params.rs @@ -1,6 +1,13 @@ use std::ffi::CString; -use std::os::raw::c_char; +use std::fmt; +use std::mem::{align_of, size_of}; +use std::os::raw::{c_char, c_void}; +use std::panic::{catch_unwind, AssertUnwindSafe}; use std::ptr; +use std::slice; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::{self, ThreadId}; use vllm_cpp_sys as ffi; @@ -9,13 +16,23 @@ use crate::error::{invalid_configuration, Error}; const NATIVE_DEFAULT_MAX_TOKENS: u32 = 16; /// Native scheduler admission order. +/// +/// The stable C submission API does not expose per-request priorities, so every +/// safe completion, stream, chat, and [`crate::Request`] currently has priority +/// zero. [`SchedulerPolicy::Priority`] therefore preserves arrival order until a +/// future ABI adds an explicit request-priority field. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub enum SchedulerPolicy { /// Process requests in arrival order. #[default] Fcfs, /// Order requests by priority and then arrival time. + /// + /// Current safe submissions all use priority zero, so ties remain ordered by + /// arrival time. This variant selects the native queue; it does not assign a + /// nonzero priority to any request. Priority, + /// Prefer requests sharing the longest cached prefix. LongestPrefixMatch, } @@ -52,6 +69,25 @@ impl Toggle { } } +type DynLogitsProcessor = dyn Fn(&[i32], &mut [f32]) + Send + Sync + 'static; + +#[derive(Clone)] +struct LogitsProcessor { + callback: Arc, +} + +impl fmt::Debug for LogitsProcessor { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("LogitsProcessor { .. }") + } +} + +impl PartialEq for LogitsProcessor { + fn eq(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.callback, &other.callback) + } +} + /// One engine-side structured decoding constraint. #[derive(Clone, Debug, Eq, PartialEq)] pub enum StructuredOutput { @@ -78,6 +114,7 @@ pub struct SamplingParams { ignore_eos: bool, stop: Vec, structured_output: Option, + logits_processor: Option, } impl Default for SamplingParams { @@ -96,6 +133,7 @@ impl Default for SamplingParams { ignore_eos: false, stop: Vec::new(), structured_output: None, + logits_processor: None, } } } @@ -207,6 +245,33 @@ impl SamplingParams { self } + /// Installs a host-side processor that can inspect generated token IDs and + /// mutate one request's logits before each sampling step. + /// + /// The processor may run concurrently for different requests, so it must be + /// `Send + Sync`. Cloned parameters share the processor. A panic is contained + /// before the C boundary and reported as [`Error::LogitsProcessorPanicked`] + /// after the bounded generation call or from [`crate::Request::wait`]. Each + /// invocation retains its processor state until the parent engine is dropped + /// because ABI v10 has no sampler-quiescence primitive. + #[must_use] + pub fn logits_processor(mut self, processor: F) -> Self + where + F: Fn(&[i32], &mut [f32]) + Send + Sync + 'static, + { + self.logits_processor = Some(LogitsProcessor { + callback: Arc::new(processor), + }); + self + } + + /// Removes a previously configured custom logits processor. + #[must_use] + pub fn clear_logits_processor(mut self) -> Self { + self.logits_processor = None; + self + } + pub(crate) fn marshal(&self) -> Result { MarshaledSamplingParams::new(self) } @@ -219,6 +284,7 @@ pub(crate) struct MarshaledSamplingParams { _structured_string: Option, _choices: Vec, _choice_pointers: Vec<*const c_char>, + logits_processor: Option>, } impl MarshaledSamplingParams { @@ -279,6 +345,19 @@ impl MarshaledSamplingParams { } } + let mut logits_processor = None; + if let Some(processor) = ¶ms.logits_processor { + if params.max_tokens.is_none() || params.max_tokens == Some(0) { + return Err(invalid_configuration( + "custom logits processors require bounded max_tokens because the native callback cannot abort generation", + )); + } + let state = Arc::new(LogitsProcessorState::new(Arc::clone(&processor.callback))); + raw.logits_processor = Some(logits_processor_trampoline); + raw.logits_processor_user_data = Arc::as_ptr(&state).cast_mut().cast(); + logits_processor = Some(state); + } + Ok(Self { raw, _stop: stop, @@ -286,12 +365,160 @@ impl MarshaledSamplingParams { _structured_string: structured_string, _choices: choices, _choice_pointers: choice_pointers, + logits_processor, }) } pub(crate) const fn raw(&self) -> &ffi::vllm_sampling_params { &self.raw } + + pub(crate) fn logits_processor_error(&self) -> Option { + self.logits_processor + .as_deref() + .and_then(LogitsProcessorState::error) + } + + pub(crate) fn logits_processor(&self) -> Option> { + self.logits_processor.clone() + } + + pub(crate) fn take_logits_processor(&mut self) -> Option> { + self.logits_processor.take() + } +} + +const PROCESSOR_OK: u8 = 0; +const PROCESSOR_PANICKED: u8 = 1; +const PROCESSOR_INVALID_INPUT: u8 = 2; + +pub(crate) struct LogitsProcessorState { + callback: Arc, + failure: AtomicU8, + active_threads: Mutex>, +} + +impl LogitsProcessorState { + fn new(callback: Arc) -> Self { + Self { + callback, + failure: AtomicU8::new(PROCESSOR_OK), + active_threads: Mutex::new(Vec::new()), + } + } + + pub(crate) fn error(&self) -> Option { + match self.failure.load(Ordering::Acquire) { + PROCESSOR_OK => None, + PROCESSOR_PANICKED => Some(Error::LogitsProcessorPanicked), + PROCESSOR_INVALID_INPUT => Some(Error::Runtime { + message: "native logits processor callback received invalid pointers or lengths" + .to_owned(), + }), + _ => Some(Error::Runtime { + message: "native logits processor callback entered an unknown failure state" + .to_owned(), + }), + } + } + + pub(crate) fn is_active_on_current_thread(&self) -> bool { + let current = thread::current().id(); + lock_unpoisoned(&self.active_threads).contains(¤t) + } + + fn record_failure(&self, failure: u8) { + let _ = self.failure.compare_exchange( + PROCESSOR_OK, + failure, + Ordering::AcqRel, + Ordering::Acquire, + ); + } +} + +struct ActiveProcessorGuard<'state> { + state: &'state LogitsProcessorState, + thread_id: ThreadId, +} + +impl<'state> ActiveProcessorGuard<'state> { + fn enter(state: &'state LogitsProcessorState) -> Self { + let thread_id = thread::current().id(); + lock_unpoisoned(&state.active_threads).push(thread_id); + Self { state, thread_id } + } +} + +impl Drop for ActiveProcessorGuard<'_> { + fn drop(&mut self) { + let mut active = lock_unpoisoned(&self.state.active_threads); + if let Some(index) = active.iter().rposition(|id| *id == self.thread_id) { + active.swap_remove(index); + } + } +} + +unsafe extern "C" fn logits_processor_trampoline( + token_ids: *const i32, + n_token_ids: i32, + logits: *mut f32, + vocab_size: i32, + user_data: *mut c_void, +) { + if user_data.is_null() { + return; + } + // SAFETY: marshaling passes Arc-backed state. Engine entry points retain an + // Arc until native engine teardown joins the sampler worker. + let state = unsafe { &*user_data.cast::() }; + if state.failure.load(Ordering::Acquire) != PROCESSOR_OK { + return; + } + if n_token_ids < 0 + || vocab_size <= 0 + || (n_token_ids > 0 && token_ids.is_null()) + || logits.is_null() + || (n_token_ids > 0 && !valid_slice_layout(token_ids, n_token_ids as usize)) + || !valid_slice_layout(logits, vocab_size as usize) + { + state.record_failure(PROCESSOR_INVALID_INPUT); + return; + } + + let _active = ActiveProcessorGuard::enter(state); + let tokens = if n_token_ids == 0 { + &[] + } else { + // SAFETY: the native callback contract lends this aligned token slice for + // this invocation, and the validated length fits Rust slice bounds. + unsafe { slice::from_raw_parts(token_ids, n_token_ids as usize) } + }; + // SAFETY: the native callback contract lends this aligned, uniquely mutable + // logits row for this invocation, and the validated length fits slice bounds. + let logits = unsafe { slice::from_raw_parts_mut(logits, vocab_size as usize) }; + if let Err(payload) = catch_unwind(AssertUnwindSafe(|| (state.callback)(tokens, logits))) { + state.record_failure(PROCESSOR_PANICKED); + discard_panic_payload(payload); + } +} + +fn valid_slice_layout(pointer: *const T, length: usize) -> bool { + !pointer.is_null() + && (pointer as usize) % align_of::() == 0 + && length <= (isize::MAX as usize) / size_of::() +} + +fn discard_panic_payload(payload: Box) { + if let Err(payload) = catch_unwind(AssertUnwindSafe(|| drop(payload))) { + std::mem::forget(payload); + } +} + +fn lock_unpoisoned(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) } pub(crate) fn to_cstring(value: &str, field: &'static str) -> Result { @@ -328,3 +555,95 @@ fn u32_to_i32(value: u32, field: &'static str) -> Result { fn length_to_i32(value: usize, field: &'static str) -> Result { i32::try_from(value).map_err(|_| invalid_configuration(format!("too many {field}"))) } + +#[cfg(test)] +mod tests { + use super::{logits_processor_trampoline, SamplingParams}; + use crate::Error; + + #[test] + fn marshals_and_invokes_custom_logits_processor() { + let params = SamplingParams::default() + .max_tokens(2) + .logits_processor(|tokens, logits| { + assert_eq!(tokens, &[3, 5]); + logits[1] = 9.0; + }); + let marshaled = params.marshal().expect("marshal processor"); + let callback = marshaled + .raw() + .logits_processor + .expect("processor callback"); + let mut logits = [1.0, 2.0, 3.0]; + let tokens = [3, 5]; + unsafe { + callback( + tokens.as_ptr(), + tokens.len() as i32, + logits.as_mut_ptr(), + logits.len() as i32, + marshaled.raw().logits_processor_user_data, + ); + } + assert_eq!(logits, [1.0, 9.0, 3.0]); + assert_eq!(marshaled.logits_processor_error(), None); + } + + #[test] + fn contains_processor_panic_and_skips_later_calls() { + let params = SamplingParams::default() + .max_tokens(2) + .logits_processor(|_, _| panic!("processor panic")); + let marshaled = params.marshal().expect("marshal processor"); + let mut logits = [1.0, 2.0]; + unsafe { + logits_processor_trampoline( + std::ptr::null(), + 0, + logits.as_mut_ptr(), + logits.len() as i32, + marshaled.raw().logits_processor_user_data, + ); + } + assert_eq!( + marshaled.logits_processor_error(), + Some(Error::LogitsProcessorPanicked) + ); + unsafe { + logits_processor_trampoline( + std::ptr::null(), + 0, + logits.as_mut_ptr(), + logits.len() as i32, + marshaled.raw().logits_processor_user_data, + ); + } + } + + #[test] + fn rejects_unbounded_processor_and_invalid_native_shape() { + let error = SamplingParams::default() + .unbounded() + .logits_processor(|_, _| {}) + .marshal() + .err() + .expect("unbounded processor rejection"); + assert!(matches!(error, Error::InvalidConfiguration { .. })); + + let params = SamplingParams::default().logits_processor(|_, _| {}); + let marshaled = params.marshal().expect("marshal processor"); + unsafe { + logits_processor_trampoline( + std::ptr::null(), + -1, + std::ptr::null_mut(), + 0, + marshaled.raw().logits_processor_user_data, + ); + } + assert!(matches!( + marshaled.logits_processor_error(), + Some(Error::Runtime { .. }) + )); + } +} diff --git a/vllm-cpp/src/request.rs b/vllm-cpp/src/request.rs index e754984..b99b763 100644 --- a/vllm-cpp/src/request.rs +++ b/vllm-cpp/src/request.rs @@ -13,7 +13,7 @@ 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}; +use crate::params::{to_cstring, LogitsProcessorState, SamplingParams}; /// How a successfully waited non-blocking request ended. /// @@ -40,6 +40,7 @@ pub enum RequestOutcome { pub struct Request { raw: Option>, callback: Option>, + logits_processor: Option>, engine: Option>, cancellation_requested: bool, _not_sync: PhantomData>, @@ -72,7 +73,10 @@ impl Engine { { cleanup_sender()?; let prompt = to_cstring(prompt, "prompt")?; - let params = params.marshal()?; + let mut params = params.marshal()?; + if let Some(logits_processor) = params.logits_processor() { + self.retain_logits_processor(logits_processor); + } 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 @@ -88,13 +92,22 @@ impl Engine { &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(), - })?; + if status != ffi::vllm_status_VLLM_OK { + status_result(status)?; + unreachable!("non-OK native status unexpectedly succeeded"); + } + let raw = match NonNull::new(output) { + Some(raw) => raw, + None => { + return Err(Error::Runtime { + message: "vllm_request_submit succeeded without a request handle".to_owned(), + }); + } + }; Ok(Request { raw: Some(raw), callback: Some(callback), + logits_processor: params.take_logits_processor(), engine: Some(Arc::clone(&self.inner)), cancellation_requested: false, _not_sync: PhantomData, @@ -106,9 +119,7 @@ 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()) } + self.native_done() } /// Requests cancellation. @@ -132,13 +143,16 @@ impl Request { /// 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() { + if self.is_native_callback_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); + if let Some(error) = self.logits_processor_error() { + return Err(error); + } let callback_result = self.callback().result(self.cancellation_requested); match callback_result { Err(error) => Err(error), @@ -185,19 +199,44 @@ impl Request { self.raw.expect("live Request always has a native handle") } + fn native_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()) } + } + fn callback(&self) -> &AsyncCallbackState { self.callback .as_deref() .expect("live Request always has callback state") } + + fn logits_processor_error(&self) -> Option { + self.logits_processor + .as_deref() + .and_then(LogitsProcessorState::error) + } + + fn is_native_callback_thread(&self) -> bool { + self.callback().is_delivery_thread() + || self + .logits_processor + .as_deref() + .is_some_and(LogitsProcessorState::is_active_on_current_thread) + } } impl Drop for Request { fn drop(&mut self) { - let parts = (self.raw.take(), self.callback.take(), self.engine.take()); + let parts = ( + self.raw.take(), + self.callback.take(), + self.logits_processor.take(), + self.engine.take(), + ); match parts { - (Some(raw), Some(callback), Some(engine)) => { - CleanupJob::new(raw, callback, engine).run(); + (Some(raw), Some(callback), logits_processor, Some(engine)) => { + CleanupJob::new(raw, callback, logits_processor, engine).run(); } parts => { // A partial Request would make either freeing or dropping its @@ -361,6 +400,7 @@ enum CleanupState { Armed { raw: NonNull, callback: Box, + logits_processor: Option>, engine: Arc, }, Disarmed, @@ -370,12 +410,14 @@ impl CleanupJob { fn new( raw: NonNull, callback: Box, + logits_processor: Option>, engine: Arc, ) -> Self { Self { state: CleanupState::Armed { raw, callback, + logits_processor, engine, }, context: CleanupContext::Caller, @@ -392,7 +434,16 @@ impl CleanupJob { } let needs_deferral = match self.context { CleanupContext::Caller => match &self.state { - CleanupState::Armed { callback, .. } => callback.is_delivery_thread(), + CleanupState::Armed { + callback, + logits_processor, + .. + } => { + callback.is_delivery_thread() + || logits_processor + .as_deref() + .is_some_and(LogitsProcessorState::is_active_on_current_thread) + } CleanupState::Disarmed => return, }, // A successfully sent job is owned only by the prestarted Rust reaper, @@ -414,6 +465,7 @@ impl CleanupJob { let CleanupState::Armed { raw, callback, + logits_processor, engine, } = state else { @@ -422,26 +474,27 @@ impl CleanupJob { // 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. + let mut owners = std::mem::ManuallyDrop::new((callback, logits_processor, engine)); + // ABI coupling: output delivery records its permanent thread ID, while + // logits calls record their active thread. This path cannot free from either + // callback context. New native user_data entrypoints must join this tracking. // - // 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. + // SAFETY: this job owns the request once and retains its callback and engine. + // Native free joins output delivery. EngineInner separately retains every + // logits state until engine teardown joins the sampler worker. 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) }; + // SAFETY: output delivery is joined. Dropping this request-owned Arc is safe + // because EngineInner retains another Arc until native engine teardown. + let (callback, logits_processor, 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); } + if let Err(payload) = catch_unwind(AssertUnwindSafe(|| drop(logits_processor))) { + std::mem::forget(payload); + } drop(engine); } diff --git a/vllm-cpp/tests/qwen3.rs b/vllm-cpp/tests/qwen3.rs index 3ae1aaf..cfe0fcc 100644 --- a/vllm-cpp/tests/qwen3.rs +++ b/vllm-cpp/tests/qwen3.rs @@ -8,25 +8,12 @@ use vllm_cpp::{ StructuredOutput, }; -const REQUIRED_MODEL_FILES: [&str; 4] = [ - "model.safetensors", - "config.json", - "tokenizer.json", - "tokenizer_config.json", -]; - fn model_path() -> Option { let path = std::env::var_os("VLLM_CPP_TEST_MODEL").map(PathBuf::from)?; - let missing = REQUIRED_MODEL_FILES - .iter() - .filter(|file| !path.join(file).is_file()) - .copied() - .collect::>(); assert!( - missing.is_empty(), - "VLLM_CPP_TEST_MODEL fixture is incomplete at {}: missing {}", - path.display(), - missing.join(", ") + path.is_dir(), + "VLLM_CPP_TEST_MODEL is not a directory: {}", + path.display() ); Some(path) } @@ -529,6 +516,110 @@ fn concurrent_request_lifecycle_stress() { }); } +#[test] +fn custom_logits_processor_controls_tokens_and_receives_history() { + with_engine(|engine, _| { + let histories = Arc::new(Mutex::new(Vec::>::new())); + let params = SamplingParams::greedy().max_tokens(3).logits_processor({ + let histories = Arc::clone(&histories); + move |tokens, logits| { + histories + .lock() + .expect("processor histories") + .push(tokens.to_vec()); + let forced = if tokens.is_empty() { 10 } else { 11 }; + logits.fill(f32::NEG_INFINITY); + logits[forced] = f32::INFINITY; + } + }); + let completion = engine + .complete("Say anything", ¶ms) + .expect("custom processor completion"); + assert_eq!(completion.completion_tokens, 3); + assert_eq!( + *histories.lock().expect("processor histories"), + vec![vec![], vec![10], vec![10, 11]] + ); + }); +} + +#[test] +fn logits_processor_panic_is_contained_for_blocking_and_async_requests() { + with_engine(|engine, _| { + let params = SamplingParams::greedy() + .max_tokens(2) + .logits_processor(|_, _| panic!("intentional logits processor panic")); + let error = engine + .complete("Say hello", ¶ms) + .expect_err("blocking processor panic"); + assert_eq!(error, Error::LogitsProcessorPanicked); + + let mut request = engine + .submit("Say hello", ¶ms, |_| StreamControl::Continue) + .expect("submit processor panic request"); + assert_eq!( + request.wait().expect_err("async processor panic"), + Error::LogitsProcessorPanicked + ); + + let completion = engine + .complete("Say hello", &SamplingParams::greedy().max_tokens(1)) + .expect("engine remains reusable"); + assert!(!completion.text.is_empty()); + }); +} + +#[test] +fn logits_processor_self_wait_is_rejected_and_state_is_retained() { + with_engine(|engine, _| { + let slot = Arc::new(Mutex::new(None::)); + let processor_ready = Arc::new(Barrier::new(2)); + let (result_sender, result_receiver) = mpsc::channel(); + let (drop_sender, drop_receiver) = mpsc::channel(); + + struct ProcessorDropProbe(mpsc::Sender); + impl Drop for ProcessorDropProbe { + fn drop(&mut self) { + let _ = self.0.send(thread::current().id()); + } + } + + let params = SamplingParams::greedy().max_tokens(8).logits_processor({ + let slot = Arc::clone(&slot); + let processor_ready = Arc::clone(&processor_ready); + let result_sender = result_sender.clone(); + let drop_probe = ProcessorDropProbe(drop_sender); + move |_, _| { + let _ = &drop_probe; + processor_ready.wait(); + if let Some(mut request) = slot.lock().expect("request slot").take() { + let thread_id = thread::current().id(); + let error = request.wait().expect_err("self wait rejection"); + result_sender + .send((thread_id, error)) + .expect("processor result"); + drop(request); + } + } + }); + let request = engine + .submit("Count from one:", ¶ms, |_| StreamControl::Continue) + .expect("submit processor self-lifecycle request"); + drop(params); + *slot.lock().expect("request slot") = Some(request); + processor_ready.wait(); + + let (_processor_thread, error) = result_receiver + .recv_timeout(Duration::from_secs(30)) + .expect("processor self-wait result"); + assert_eq!(error, Error::RequestCallbackThread { operation: "wait" }); + assert!(matches!( + drop_receiver.try_recv(), + Err(mpsc::TryRecvError::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 f5f69f5..8b9d1c9 100644 --- a/vllm-cpp/tests/safe_api.rs +++ b/vllm-cpp/tests/safe_api.rs @@ -7,6 +7,7 @@ assert_impl_all!(Engine: Send, Sync, Clone); assert_impl_all!(HuggingFaceError: Clone, std::fmt::Debug, Eq, PartialEq); assert_impl_all!(HuggingFaceModel: Clone, std::fmt::Debug); assert_impl_all!(Request: Send); +assert_impl_all!(vllm_cpp::SamplingParams: Clone, Send, Sync); assert_not_impl_any!(Request: Sync); fn missing_model() -> &'static str { @@ -25,6 +26,7 @@ fn hugging_face_constructors_accept_default_and_explicit_revisions() { fn reports_expected_abi() { assert_eq!(vllm_cpp::expected_abi_version(), 10); assert_eq!(vllm_cpp::abi_version(), 10); + assert!(!vllm_cpp::version().expect("native version").is_empty()); } #[test] From 4c6fbe30d9f21075287fbd616e468f5a95a9ddd3 Mon Sep 17 00:00:00 2001 From: Ivan Zatevakhin Date: Wed, 19 Aug 2026 10:54:00 +0100 Subject: [PATCH 2/2] docs: clarify priority scheduler behavior --- CHANGELOG.md | 2 +- README.md | 2 +- vllm-cpp/README.md | 2 +- vllm-cpp/src/engine.rs | 7 +++++-- vllm-cpp/src/params.rs | 14 +++++++------- 5 files changed, 15 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cb29658..77c8401 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,7 +24,7 @@ All notable changes to this project will be documented in this file. ### Known limitations -- The priority scheduler is selectable, but the ABI v10 submission surface assigns every request priority zero; safe requests therefore tie by arrival until a future ABI carries per-request priority. +- The priority scheduler is selectable, and raw and serde chat request JSON can carry a `priority` field that the native OpenAI-compatible path parses and submits. Direct completion, completion streaming, and `Request` submissions currently default to priority zero and tie by arrival; caller-selected priorities for those direct APIs remain deferred until a future C ABI/API change. - The supported runtime tier is native Linux x86_64 CPU. Accelerator features are experimental build/configuration surfaces, not runtime-support claims. - Known native blockers include a CUDA teardown SIGSEGV after otherwise successful tests, a CUDA bf16 numerical tolerance failure, CUTLASS concurrent-output differences, incomplete Vulkan attention/model runtime, and external MLX deployment plus unvalidated release-lane model/runtime behavior. diff --git a/README.md b/README.md index de85cc5..0254ab8 100644 --- a/README.md +++ b/README.md @@ -49,7 +49,7 @@ The packaged [`vllm-cpp` guide](vllm-cpp/README.md) covers local and Hugging Fac `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. `SchedulerPolicy::Priority` selects the native queue, but every ABI v10 safe submission currently has priority zero and therefore ties by arrival. +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. `SchedulerPolicy::Priority` selects the native queue. Raw and serde chat request JSON can carry a `priority` field that the native OpenAI-compatible path parses and submits. Direct completion, completion streaming, and `Request` submissions currently default to priority zero and tie by arrival; caller-selected priorities for those direct APIs require a future C ABI/API change. See [the examples guide](vllm-cpp/examples/README.md) for ordinary Linux and optional Nix setup, commands for every example, and the interactive chat CLI's local/Hub model forms and generation options. Release-facing changes are recorded in the [changelog](CHANGELOG.md), and maintainers use the manual [release process](RELEASING.md). diff --git a/vllm-cpp/README.md b/vllm-cpp/README.md index e15b6d3..30527ca 100644 --- a/vllm-cpp/README.md +++ b/vllm-cpp/README.md @@ -49,7 +49,7 @@ In the repository checkout, `just setup-test-model` explicitly resolves `Qwen/Qw - Asynchronous callbacks run on a native delivery thread and must be `Send + 'static`. `wait` reports callback panics as `Error::CallbackPanicked`; waiting or freeing from that same callback thread is prohibited by ABI v10, so callback-thread drop transfers cleanup to a prestarted reaper. - Dropping a live request cancels and joins it. `cancel` is idempotent, `wait` reports the request outcome, and `native_error` copies the request-owned diagnostic after completion into an owned Rust `String`; the native storage remains valid until the request is dropped or freed. -`SchedulerPolicy::Priority` selects the native priority queue, but ABI v10 assigns priority zero to every safe blocking, streaming, chat, and asynchronous submission. Until a future ABI carries per-request priority, safe requests therefore remain ordered by arrival. +`SchedulerPolicy::Priority` selects the native priority queue. Raw and serde chat request JSON can carry a `priority` field that the native OpenAI-compatible path parses and submits. Direct completion, completion streaming, and `Request` submissions currently default to priority zero and tie by arrival; caller-selected priorities for those direct APIs require a future C ABI/API change. ## Features and linking diff --git a/vllm-cpp/src/engine.rs b/vllm-cpp/src/engine.rs index acf3a18..caf39b8 100644 --- a/vllm-cpp/src/engine.rs +++ b/vllm-cpp/src/engine.rs @@ -334,8 +334,11 @@ impl EngineBuilder { /// Selects the native admission queue policy. /// - /// The stable C submission API currently assigns priority zero to every - /// request, so [`SchedulerPolicy::Priority`] orders safe requests by arrival. + /// Raw and serde chat request JSON can carry a `priority` field that the native + /// OpenAI-compatible path parses and submits. Direct completion, completion + /// streaming, and `Request` submissions currently default to priority zero and + /// tie by arrival; caller-selected priorities for those direct APIs require a + /// future C ABI/API change. #[must_use] pub fn scheduler(mut self, value: SchedulerPolicy) -> Self { self.scheduler = value; diff --git a/vllm-cpp/src/params.rs b/vllm-cpp/src/params.rs index 31842df..130b5af 100644 --- a/vllm-cpp/src/params.rs +++ b/vllm-cpp/src/params.rs @@ -17,10 +17,11 @@ const NATIVE_DEFAULT_MAX_TOKENS: u32 = 16; /// Native scheduler admission order. /// -/// The stable C submission API does not expose per-request priorities, so every -/// safe completion, stream, chat, and [`crate::Request`] currently has priority -/// zero. [`SchedulerPolicy::Priority`] therefore preserves arrival order until a -/// future ABI adds an explicit request-priority field. +/// Raw and serde chat request JSON can carry a `priority` field that the native +/// OpenAI-compatible path parses and submits. Direct completion, completion +/// streaming, and [`crate::Request`] submissions currently default to priority zero +/// and tie by arrival; caller-selected priorities for those direct APIs require a +/// future C ABI/API change. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub enum SchedulerPolicy { /// Process requests in arrival order. @@ -28,9 +29,8 @@ pub enum SchedulerPolicy { Fcfs, /// Order requests by priority and then arrival time. /// - /// Current safe submissions all use priority zero, so ties remain ordered by - /// arrival time. This variant selects the native queue; it does not assign a - /// nonzero priority to any request. + /// This variant selects the native priority queue; it does not itself assign a + /// priority to a request. Priority, /// Prefer requests sharing the longest cached prefix.