From bfe6f4d9ff2ba2cf1374abe2e7015a229ed4b21f Mon Sep 17 00:00:00 2001 From: mrciffa Date: Wed, 23 Sep 2026 16:32:07 +0200 Subject: [PATCH 1/2] feat(qwen35): serve image requests in the concurrent batch A vision projector used to switch concurrent sequence scheduling off for image input. The batched engine now accepts image requests: - SeqEngine gains supports_images() and admit_images(); the default refuses. The scheduler routes image requests there and never gives them a prefix-cache plan (tokens alone do not identify an image). - Qwen35SeqEngine encodes the images at admission, keeps the payload, rows and rope offset with the slot until it retires, overwrites image rows and writes the image's M-RoPE positions in every prefill chunk that covers an image (so an eviction re-prefill sees them again), and shifts decode and chain-verify rotary positions by the slot's offset. - The server enables image input when the engine supports it; the feature gate allows --mmproj with --max-concurrency for Qwen3.5. DeepSeek4 still requires one request at a time. lucebox6 R9700, Qwen3.8-27B-IQ4_XS-pure + Q8_0 projector + DFlash2, --paged-attention --max-concurrency 4: sanity 6/6, one to four images 9/9 (same as single-request); 1/2/4 concurrent 256-token image answers 81/104/149 tok/s total (single-request server 78/73/77), 4 answers in 6.9 s instead of 13.3 s, each answer about its own chart. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/image-input.md | 11 ++- server/src/common/concurrency/seq_engine.h | 17 +++++ server/src/common/feature_gate.cpp | 6 +- .../qwen35/concurrency/qwen35_seq_engine.cpp | 69 +++++++++++++++++-- .../qwen35/concurrency/qwen35_seq_engine.h | 28 ++++++++ server/src/server/http_server.cpp | 3 +- server/src/server/scheduler.cpp | 9 ++- 7 files changed, 128 insertions(+), 15 deletions(-) diff --git a/docs/image-input.md b/docs/image-input.md index d4a264b9d..d676568f8 100644 --- a/docs/image-input.md +++ b/docs/image-input.md @@ -111,9 +111,14 @@ plain autoregressive decoding and bypass the token-keyed prefix, disk and agent-turn caches and prompt compression: tokens alone do not identify an image. Text requests on the same server keep speculative decoding and caching. -Layer or tensor splitting across GPUs, remote target shards, concurrent -sequence scheduling (`--max-concurrency`) and upstream forwarding do not -support images. `/props` reports the effective capability in +Layer or tensor splitting across GPUs, remote target shards and upstream +forwarding do not support images. Qwen3.5 / Qwen3.8 serve images with +concurrent sequence scheduling (`--paged-attention --max-concurrency N`): each +image request is encoded when it is admitted and then prefills and decodes in +the shared batch like text, with the drafter. On one R9700, four concurrent +256-token image answers finish in 6.9 s (149 tok/s in total) against 13.3 s +(77 tok/s) one at a time. DeepSeek V4 image requests still need one request +at a time. `/props` reports the effective capability in `capabilities.image_input_supported` after backend initialization. ## Qwen3.5 / Qwen3.8 diff --git a/server/src/common/concurrency/seq_engine.h b/server/src/common/concurrency/seq_engine.h index 98444f955..2292f259a 100644 --- a/server/src/common/concurrency/seq_engine.h +++ b/server/src/common/concurrency/seq_engine.h @@ -54,6 +54,7 @@ #include #include +#include "common/image_prompt.h" #include "common/sampler.h" #include "prefix_store.h" @@ -181,6 +182,22 @@ class SeqEngine { const std::vector & prompt, const SamplerCfg & sampler) = 0; + // Image requests. The engine keeps the payload with the slot for as long + // as the slot lives, so a re-prefill after eviction sees the images again. + // Engines without image support refuse; the server only routes images to + // an engine that reports supports_images(). + virtual bool supports_images() const { return false; } + virtual AdmitResult admit_images(uint64_t request_id, + const std::vector & prompt, + const SamplerCfg & sampler, + const ImagePromptHandle & images) { + (void) request_id; (void) prompt; (void) sampler; (void) images; + AdmitResult result; + result.status = AdmitResult::Status::failed; + result.error = "this engine does not serve image requests"; + return result; + } + // Optional prefix-checkpoint admission. Unsupported engines remain on // cold admission and never receive a plan from the scheduler. virtual bool supports_prefix_store() const { return false; } diff --git a/server/src/common/feature_gate.cpp b/server/src/common/feature_gate.cpp index fadbb6d40..39dbb617e 100644 --- a/server/src/common/feature_gate.cpp +++ b/server/src/common/feature_gate.cpp @@ -51,9 +51,9 @@ std::string check_feature_compatibility( if (args.mmproj_path.has_value()) { if ((arch != "deepseek4" && arch != "qwen35") || args.device.is_layer_split() || args.device.is_tensor_parallel() || args.remote_target_shard.enabled() || - args.max_concurrency != 1) { - return "--mmproj requires a local single-request DeepSeek4 or Qwen3.5 backend " - "that is not split across GPUs by layer or tensor"; + (arch == "deepseek4" && args.max_concurrency != 1)) { + return "--mmproj requires a local DeepSeek4 (one request at a time) or Qwen3.5 " + "backend that is not split across GPUs by layer or tensor"; } if (arch == "deepseek4" && target_backend != PlacementBackend::Hip) { return "--mmproj with DeepSeek4 requires a HIP backend"; diff --git a/server/src/qwen35/concurrency/qwen35_seq_engine.cpp b/server/src/qwen35/concurrency/qwen35_seq_engine.cpp index e38dc751f..e9852634c 100644 --- a/server/src/qwen35/concurrency/qwen35_seq_engine.cpp +++ b/server/src/qwen35/concurrency/qwen35_seq_engine.cpp @@ -379,6 +379,7 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit( const SamplerCfg & sampler) { AdmitResult result = slots_.admit(request_id, prompt, sampler); if (result.status == AdmitResult::Status::admitted) { + clear_slot_images(result.slot); reset_recurrent_slot(b_.cache_, result.slot); if (result.slot >= 0 && result.slot < static_cast(slot_draft_kv_.size()) && @@ -389,6 +390,43 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit( return result; } +bool Qwen35SeqEngine::supports_images() const { + return b_.supports_images(); +} + +SeqEngine::AdmitResult Qwen35SeqEngine::admit_images( + uint64_t request_id, + const std::vector & prompt, + const SamplerCfg & sampler, + const ImagePromptHandle & images) { + AdmitResult refused; + refused.status = AdmitResult::Status::failed; + const auto * payload = dynamic_cast(images.get()); + if (!payload || payload->owner != &b_ || !payload->matches(prompt)) { + refused.error = "image binding does not match this prompt"; + return refused; + } + // Encode before claiming a slot: a failed encode then leaves no slot to + // unwind. The tower runs on this (the scheduler) thread, between steps. + Qwen35ImageRows rows; + std::string error = "image encoding failed"; + if (!b_.encode_images(*payload, rows, error)) { + refused.error = error; + return refused; + } + AdmitResult result = admit(request_id, prompt, sampler); + if (result.status != AdmitResult::Status::admitted) return result; + if (slot_images_.size() < static_cast(slots_.slot_count())) { + slot_images_.resize(static_cast(slots_.slot_count())); + } + SlotImages & state = slot_images_[static_cast(result.slot)]; + state.payload = images; + state.rows = std::move(rows); + state.rows.prompt = payload; + state.rope_delta = payload->positions.next - static_cast(prompt.size()); + return result; +} + size_t Qwen35SeqEngine::estimate_prefix_store_bytes(int tokens) const { return estimate_paged_target_cache_snapshot_bytes(b_.cache_, tokens); } @@ -425,6 +463,7 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit_with_prefix( const PrefixStorePlan & plan) { AdmitResult result = slots_.admit(request_id, prompt, sampler); if (result.status != AdmitResult::Status::admitted) return result; + clear_slot_images(result.slot); const int slot = result.slot; slots_.slot(slot).pending_capture = {}; @@ -652,6 +691,13 @@ Qwen35SeqEngine::PrefillStage Qwen35SeqEngine::stage_prefill_chunk( "prefill embedding failed"); return PrefillStage{}; } + if (slot < static_cast(slot_images_.size()) && + slot_images_[static_cast(slot)].payload) { + const SlotImages & images = slot_images_[static_cast(slot)]; + images.rows.overwrite(stage.embeddings.data(), stage.kv_pos, stage.chunk, b_.w_.n_embd); + stage.positions.assign(static_cast(4) * stage.chunk, 0); + images.rows.prompt->positions.fill(stage.positions.data(), stage.kv_pos, stage.chunk); + } stage.ready = true; return stage; } @@ -856,7 +902,7 @@ SeqEngine::StepResult Qwen35SeqEngine::step_chain_spec( seq_lens_[static_cast(lane.slot)] = lane.position + 1; for (int axis = 0; axis < 3; ++axis) { positions[static_cast(axis) * total_rows + lane_index] = - lane.position; + lane.position + rope_delta(lane.slot); } for (int head = 0; head < n_head_kv; ++head) { write_rows[static_cast(head) * total_rows + lane_index] = @@ -884,7 +930,8 @@ SeqEngine::StepResult Qwen35SeqEngine::step_chain_spec( ? -1 : node - 1; query_slots[static_cast(row)] = proposal.slot; - const int position = slots_.slot(proposal.slot).cur_pos + node; + const int position = slots_.slot(proposal.slot).cur_pos + node + + rope_delta(proposal.slot); for (int axis = 0; axis < 3; ++axis) { positions[static_cast(axis) * total_rows + row] = position; @@ -1427,14 +1474,23 @@ SeqEngine::StepResult Qwen35SeqEngine::step(const StepPlan & plan) { pos_buf_.assign((size_t)4 * n_total, 0); token_offset = 0; for (const PrefillStage & prefill : prefills) { - fill_qwen35_mrope_positions( - pos_buf_.data(), n_total, token_offset, - prefill.kv_pos, prefill.chunk); + if (prefill.positions.empty()) { + fill_qwen35_mrope_positions( + pos_buf_.data(), n_total, token_offset, + prefill.kv_pos, prefill.chunk); + } else { + for (int axis = 0; axis < 4; ++axis) { + std::copy_n(prefill.positions.data() + (size_t)axis * prefill.chunk, prefill.chunk, + pos_buf_.data() + (size_t)axis * n_total + token_offset); + } + } token_offset += prefill.chunk; } if (with_decode) { for (int row = 0; row < live_count; ++row) { - const int pos = live_positions_[(size_t)row]; + // Rotary positions run ahead of KV positions after an image. + const int pos = live_positions_[(size_t)row] + + rope_delta(live_slot_ids_[(size_t)row]); const int packed_row = n_prefill + row; pos_buf_[(size_t)0 * n_total + packed_row] = pos; pos_buf_[(size_t)1 * n_total + packed_row] = pos; @@ -1650,6 +1706,7 @@ bool Qwen35SeqEngine::evict_kv(int slot, int32_t pending_token, void Qwen35SeqEngine::retire(int slot) { offload_.discard(slot); + clear_slot_images(slot); if (!slots_.is_active(slot)) return; if (slot >= 0 && slot < static_cast(slot_draft_kv_.size()) && slot_draft_kv_[static_cast(slot)]) { diff --git a/server/src/qwen35/concurrency/qwen35_seq_engine.h b/server/src/qwen35/concurrency/qwen35_seq_engine.h index 938bb9356..e33e36892 100644 --- a/server/src/qwen35/concurrency/qwen35_seq_engine.h +++ b/server/src/qwen35/concurrency/qwen35_seq_engine.h @@ -26,6 +26,7 @@ #include "common/dflash_draft_kv.h" #include "common/dflash_feature_ring.h" #include "qwen35_slot_manager.h" +#include "../qwen35_image_request.h" #include #include @@ -114,6 +115,11 @@ class Qwen35SeqEngine final : public SeqEngine { return slots_.kv_restore_feasible(slot); } void retire(int slot) override; + bool supports_images() const override; + AdmitResult admit_images(uint64_t request_id, + const std::vector & prompt, + const SamplerCfg & sampler, + const ImagePromptHandle & images) override; bool token_is_eos(int32_t token) const override; @@ -126,8 +132,30 @@ class Qwen35SeqEngine final : public SeqEngine { bool commit = false; std::vector rows; std::vector embeddings; + // Axis-major [4 x chunk] rotary positions when the slot holds images; + // empty means the plain kv_pos + i positions. + std::vector positions; }; + // Per-slot image state: the payload (kept alive for re-prefill after + // eviction), its encoded rows, and how far rotary positions run ahead of + // KV positions after the images. + struct SlotImages { + ImagePromptHandle payload; + Qwen35ImageRows rows; + int rope_delta = 0; + }; + std::vector slot_images_; + int rope_delta(int slot) const { + return slot >= 0 && slot < static_cast(slot_images_.size()) + ? slot_images_[static_cast(slot)].rope_delta : 0; + } + void clear_slot_images(int slot) { + if (slot >= 0 && slot < static_cast(slot_images_.size())) { + slot_images_[static_cast(slot)] = SlotImages{}; + } + } + struct PreparedChainDraft { std::vector tokens; }; diff --git a/server/src/server/http_server.cpp b/server/src/server/http_server.cpp index 18d8c7973..be2c28f58 100644 --- a/server/src/server/http_server.cpp +++ b/server/src/server/http_server.cpp @@ -1248,7 +1248,8 @@ HttpServer::HttpServer(luce::engine::LuceEngine & engine, config.disk_cache_cold_max_tokens}, backend_) { config_.image_input_enabled = backend_.supports_images() && - config_.pflash_upstream_base.empty() && !backend_.seq_engine(); + config_.pflash_upstream_base.empty() && + (!backend_.seq_engine() || backend_.seq_engine()->supports_images()); if (backend_.supports_images() && !config_.image_input_enabled) { std::fprintf(stderr, "[server] WARNING: a vision projector is loaded but image input is off: it is " diff --git a/server/src/server/scheduler.cpp b/server/src/server/scheduler.cpp index f1b184ce1..91705cdfe 100644 --- a/server/src/server/scheduler.cpp +++ b/server/src/server/scheduler.cpp @@ -451,7 +451,9 @@ void HttpServer::scheduler_loop(SeqEngine & engine) { PrefixCaptureTxn prepared_capture; PrefixCache::InlineReservation capture_reservation; int restore_policy_slot = -1; - const bool prefix_supported = + // Tokens alone do not identify an image: image requests never touch + // the prefix cache. + const bool prefix_supported = !req.images && engine.supports_prefix_store() && !prefix_cache_.disabled(); if (prefix_supported) { const auto hit = prefix_cache_.lookup_candidate( @@ -502,7 +504,10 @@ void HttpServer::scheduler_loop(SeqEngine & engine) { // Admission only claims the slot and queues the prompt. Prefill // advances one chunk per engine step alongside live decode. const PrefixStorePlan requested_plan = prefix_plan; - auto ar = prefix_supported + auto ar = req.images + ? engine.admit_images( + next_request_id, req.prompt_tokens, req.sampler, req.images) + : prefix_supported ? engine.admit_with_prefix( next_request_id, req.prompt_tokens, req.sampler, requested_plan) From 87bce9e99ae069de45f0eea5ddae6a503b86c8d5 Mon Sep 17 00:00:00 2001 From: mrciffa Date: Thu, 24 Sep 2026 13:34:08 +0200 Subject: [PATCH 2/2] fix(qwen35): claim the slot before encoding a batched image request A busy pool defers the request and retries it; encoding first reran the vision tower on every retry. A failed encode now retires the slot. Co-Authored-By: Claude Opus 5.5 (1M context) --- server/src/qwen35/concurrency/qwen35_seq_engine.cpp | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/server/src/qwen35/concurrency/qwen35_seq_engine.cpp b/server/src/qwen35/concurrency/qwen35_seq_engine.cpp index e9852634c..d6da7541e 100644 --- a/server/src/qwen35/concurrency/qwen35_seq_engine.cpp +++ b/server/src/qwen35/concurrency/qwen35_seq_engine.cpp @@ -406,16 +406,19 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit_images( refused.error = "image binding does not match this prompt"; return refused; } - // Encode before claiming a slot: a failed encode then leaves no slot to - // unwind. The tower runs on this (the scheduler) thread, between steps. + // Claim the slot first: a busy pool defers the request and retries it, + // and encoding before that would rerun the tower on every retry. The + // tower runs on this (the scheduler) thread, so no step sees the slot + // before its images are in place. + AdmitResult result = admit(request_id, prompt, sampler); + if (result.status != AdmitResult::Status::admitted) return result; Qwen35ImageRows rows; std::string error = "image encoding failed"; if (!b_.encode_images(*payload, rows, error)) { + retire(result.slot); refused.error = error; return refused; } - AdmitResult result = admit(request_id, prompt, sampler); - if (result.status != AdmitResult::Status::admitted) return result; if (slot_images_.size() < static_cast(slots_.slot_count())) { slot_images_.resize(static_cast(slots_.slot_count())); }