diff --git a/docs/image-input.md b/docs/image-input.md index 250ed3eb1..f6e88c753 100644 --- a/docs/image-input.md +++ b/docs/image-input.md @@ -136,9 +136,31 @@ 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. +(77 tok/s) one at a time. + +DeepSeek V4 Flash Vision batches too, with the batched launch from the DeepSeek +guide plus `--mmproj` (and `--mmproj-device` for an R9700 encoder): + +``` +luce_server models/DeepSeek-V4-Flash-Vision-Exp-ROCMFPX-MIX-STRIX.gguf \ + --target-device hip:1 --mmproj-device hip:0 \ + --paged-attention --max-concurrency 4 --kv-pool-tokens 24576 --max-ctx 8192 \ + --ds4-prefill exact --prefix-cache-slots 0 --ds4-expert-top-k 6 \ + --mmproj models/DeepSeek-V4-Flash-Vision-Exp-mmproj-BF16.gguf +``` + +Its image blocks need whole-block bidirectional prefill, which the batched +engine's 16-row step cannot run. Image requests admitted since the last step +are therefore prefilled up to their last token together, in shared +layer-major sparse passes into per-request staging caches (each layer's +experts are read once for all of them); that state is copied into each +request's paged slot and the last token prefills in the batch, so the answers +decode alongside everyone else. On the Strix Halo with the encoder on the +R9700, four concurrent image answers of 256 tokens finish in 35 s (29 tok/s in +total), two images plus two text requests at 31 tok/s; four text requests +reach 38 tok/s. Image requests beyond the free slots wait in the queue. `/props` reports the +effective capability in `capabilities.image_input_supported` after backend +initialization. ## Qwen3.5 / Qwen3.8 @@ -238,10 +260,11 @@ per-expert layout is used expert by expert; the community publishes one for this model) or `--absmax-only`. The converter uses every core: about 40 minutes for this checkpoint on 32 cores. -One image request may be outstanding per backend. Its admission lease remains -with the immutable payload through queueing and generation; another image -request is rejected until that payload is released. This bounds simultaneous -preprocessing and prepared-image memory. Text requests retain the normal queue. +Image requests wait in the same queue as text requests. A waiting request +holds only its preprocessed patches, a few MB per image; its encoded rows +exist only while it runs, so the number of slots bounds them. When host +memory is too short to prepare another image request, the server answers +HTTP 503 and the client should retry. The server expands image markers after final rendering and tokenization. Expanded image tokens count toward context and usage. Image blocks remain diff --git a/server/src/common/feature_gate.cpp b/server/src/common/feature_gate.cpp index bc69afd7e..d3a703880 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() || - (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"; + (arch == "deepseek4" && args.max_concurrency != 1 && !args.paged_attention)) { + return "--mmproj requires a local DeepSeek4 or Qwen3.5 backend that is not split " + "across GPUs by layer or tensor (DeepSeek4 batching needs --paged-attention)"; } if (arch == "deepseek4" && target_backend != PlacementBackend::Hip) { return "--mmproj with DeepSeek4 requires a HIP backend"; diff --git a/server/src/common/image_prompt.h b/server/src/common/image_prompt.h index 44e46af11..50cbd2722 100644 --- a/server/src/common/image_prompt.h +++ b/server/src/common/image_prompt.h @@ -12,6 +12,11 @@ namespace luce::common { // transport. Each image still has its own token and byte bounds. inline constexpr size_t MAX_REQUEST_IMAGES = 16; +// Outcome of binding a request's images to its prompt. `busy` means the +// request is valid but the backend already holds as many image requests as it +// serves at once; the server answers 503 so the client retries. +enum class ImagePrepareStatus { ok, invalid, busy }; + struct EncodedImage { std::string mime_type; std::vector bytes; diff --git a/server/src/common/model_backend.h b/server/src/common/model_backend.h index 1f6e6afc4..f25d8ec86 100644 --- a/server/src/common/model_backend.h +++ b/server/src/common/model_backend.h @@ -138,21 +138,21 @@ struct ModelBackend { // turns into the image marker, and binds decoded images to a rendered prompt. virtual bool supports_images() const { return false; } virtual std::string image_placeholder() const { return {}; } - virtual bool prepare_images(std::vector & tokens, - std::vector images, - uint64_t context_capacity, - uint64_t output_reserve, - ImagePromptHandle & payload, - std::string & error) const { + virtual ImagePrepareStatus prepare_images(std::vector & tokens, + std::vector images, + uint64_t context_capacity, + uint64_t output_reserve, + ImagePromptHandle & payload, + std::string & error) const { (void) tokens; (void) context_capacity; (void) output_reserve; if (!images.empty()) { error = "this backend does not support image input"; - return false; + return ImagePrepareStatus::invalid; } payload.reset(); - return true; + return ImagePrepareStatus::ok; } // Print the "[-daemon] ready ..." banner on stdout. diff --git a/server/src/deepseek4/deepseek4_backend.cpp b/server/src/deepseek4/deepseek4_backend.cpp index 134db6cb6..1b40aa756 100644 --- a/server/src/deepseek4/deepseek4_backend.cpp +++ b/server/src/deepseek4/deepseek4_backend.cpp @@ -55,9 +55,8 @@ class DeepSeek4ImagePrompt final : public ImagePromptPayload { friend class DeepSeek4Backend; DeepSeek4ImagePrompt(const DeepSeek4Backend * owner, vision::PreparedImagePrompt prepared, - std::vector encoded, std::shared_ptr lease) - : owner_(owner), prepared_(std::move(prepared)), encoded_(std::move(encoded)), - lease_(std::move(lease)) { + std::vector encoded) + : owner_(owner), prepared_(std::move(prepared)), encoded_(std::move(encoded)) { for (const auto & image : prepared_.images) spans_.push_back(image.layout.span); } @@ -108,7 +107,6 @@ class DeepSeek4ImagePrompt final : public ImagePromptPayload { const DeepSeek4Backend * const owner_; const vision::PreparedImagePrompt prepared_; const std::vector encoded_; - const std::shared_ptr lease_; std::vector spans_; mutable std::vector> materialized_; mutable std::mutex stream_mutex_; @@ -1076,7 +1074,7 @@ DeepSeek4Backend::~DeepSeek4Backend() { shutdown(); } -bool DeepSeek4Backend::prepare_images( +ImagePrepareStatus DeepSeek4Backend::prepare_images( std::vector & tokens, std::vector images, uint64_t context_capacity, uint64_t output_reserve, ImagePromptHandle & payload, std::string & error) const { @@ -1088,29 +1086,27 @@ bool DeepSeek4Backend::prepare_images( for (int32_t token : tokens) { if (token == marker || token < 0 || token >= w_.n_vocab) { error = "unbound image marker or invalid token in rendered prompt"; - return false; + return ImagePrepareStatus::invalid; } } } payload.reset(); - return true; + return ImagePrepareStatus::ok; } if (!image_capable_) { error = "image input requires a validated --mmproj projector and heterogeneous HIP sparse prefill"; - return false; + return ImagePrepareStatus::invalid; } try { if (images.size() > MAX_REQUEST_IMAGES) { error = "too many images in request"; - return false; - } - auto lease = image_request_gate_.try_acquire(); - if (!lease) { - error = "an image request is already in progress; retry after it completes"; - return false; + return ImagePrepareStatus::invalid; } + // Image requests queue like text ones: a waiting request holds only + // its patches (a few MB per image); the encoded rows exist only once + // it runs, so the slots bound them. Short host memory is capacity. if (!vision::check_deepseek4_image_host_preparation(4ULL * 1024 * 1024 * 1024, error)) { - return false; + return ImagePrepareStatus::busy; } std::vector patches; patches.reserve(images.size()); @@ -1119,13 +1115,13 @@ bool DeepSeek4Backend::prepare_images( if (image.bytes.size() > 16ULL * 1024 * 1024 || encoded_bytes > 32ULL * 1024 * 1024 - image.bytes.size()) { error = "images exceed request byte limit"; - return false; + return ImagePrepareStatus::invalid; } encoded_bytes += image.bytes.size(); auto decoded = vision::decode_image({image.bytes.data(), image.bytes.size()}); - if (!decoded) { error = decoded.status.message; return false; } + if (!decoded) { error = decoded.status.message; return ImagePrepareStatus::invalid; } auto processed = vision::preprocess_rgb(decoded.image.view(), 0); - if (!processed) { error = processed.status.message; return false; } + if (!processed) { error = processed.status.message; return ImagePrepareStatus::invalid; } patches.push_back({processed.image.plan, std::move(processed.image.patches_bf16)}); } vision::ImagePromptLimits limits; @@ -1133,21 +1129,110 @@ bool DeepSeek4Backend::prepare_images( limits.output_reserve = output_reserve; limits.max_expanded_tokens = std::min(context_capacity, vision::MAX_PREPARED_PROMPT_TOKENS); auto prepared = vision::prepare_image_prompt(tokens, patches, limits); - if (!prepared) { error = prepared.message; return false; } + if (!prepared) { error = prepared.message; return ImagePrepareStatus::invalid; } auto binding = std::shared_ptr( - new DeepSeek4ImagePrompt(this, std::move(prepared), std::move(images), std::move(lease))); + new DeepSeek4ImagePrompt(this, std::move(prepared), std::move(images))); if (!vision::valid_image_spans(binding->spans(), binding->prepared_.tokens.size())) { error = "invalid prepared image spans"; - return false; + return ImagePrepareStatus::invalid; } std::vector expanded = binding->prepared_.tokens; tokens.swap(expanded); payload = std::move(binding); - return true; + return ImagePrepareStatus::ok; } catch (const std::bad_alloc &) { error = "image preparation allocation failed"; + return ImagePrepareStatus::invalid; + } +} + +bool DeepSeek4Backend::encode_image_request(const std::vector & prompt, + const ImagePromptHandle & handle, std::string & error) { + const auto * images = dynamic_cast(handle.get()); + if (!images || images->owner_ != this || !images->matches(prompt)) { + error = "image binding does not match this prompt"; + return false; + } + DaemonIO io; + const auto t0 = Clock::now(); + const bool ok = materialize_images(*images, io, error); + join_image_stream(); + if (ok && !images->complete()) { + error = "image encoding did not complete"; return false; } + std::fprintf(stderr, "[deepseek4] batched image request encoded in %.0f ms\n", elapsed_s(t0) * 1000.0); + return ok; +} + +void DeepSeek4Backend::prefill_staged(std::vector & batch) { + // Embeddings for every request's prefix (image rows + text rows). + struct Seq { StagedPrefill * item; const DeepSeek4ImagePrompt * images; std::vector embed; int done = 0; }; + std::vector seqs; + for (auto & item : batch) { + const auto * images = dynamic_cast(item.images.get()); + if (!images || !item.prompt || !item.staging || item.prefix < DS4_MIN_LAYER_MAJOR_PREFILL_TOKENS || + item.prefix > int(item.prompt->size()) || item.prefix > item.staging->max_ctx) { + item.error = "invalid staged prefill request"; + continue; + } + std::vector embed(size_t(item.prefix) * size_t(w_.n_embd)); + if (!images->embed_chunk(w_.embedder, 0, item.prefix, embed.data())) { + item.error = "staged prefill embedding failed"; + continue; + } + reset_deepseek4_cache(*item.staging); + item.staging->prefill_mode = PrefillAttentionMode::Sparse; + seqs.push_back({&item, images, std::move(embed), 0}); + } + ggml_backend_synchronize(backend_); + deepseek4_release_image_scratch(cache_, moe_hybrid_.get()); + const int budget_total = std::min(1024, DS4_MAX_LAYER_MAJOR_PREFILL_TOKENS); + constexpr int min_rows = DS4_MIN_LAYER_MAJOR_PREFILL_TOKENS; + const auto t0 = Clock::now(); + int passes = 0, rows = 0; + for (;;) { + // One pass takes the next chunk of every unfinished request that fits: + // whole image blocks only, and never a chunk or tail below the + // layer-major minimum. + std::vector pass; + std::vector members; + int budget = budget_total; + for (auto & s : seqs) { + const int remaining = s.item->prefix - s.done; + if (remaining <= 0 || !s.item->error.empty() || budget < min_rows) continue; + int n = std::min(remaining, budget); + if (remaining - n > 0 && remaining - n < min_rows) n = std::max(0, remaining - min_rows); + n = n >= min_rows ? vision::atomic_image_chunk(s.images->spans(), uint64_t(s.done), n, + uint64_t(remaining), budget) : 0; + if (n < min_rows) continue; + DeepSeek4PrefillSeq seq; + seq.cache = s.item->staging; + seq.embed = s.embed.data() + size_t(s.done) * size_t(w_.n_embd); + seq.token_ids = s.item->prompt->data() + s.done; + seq.n_tokens = n; + seq.kv_start = s.done; + seq.image_spans = s.images->spans(); + pass.push_back(seq); + members.push_back(&s); + budget -= n; + } + if (pass.empty()) break; + std::string error; + if (!deepseek4_prefill_multi(backend_, cfg_.device.gpu, w_, pass, error)) { + for (Seq * m : members) m->item->error = error.empty() ? "staged prefill failed" : error; + continue; + } + for (size_t k = 0; k < members.size(); ++k) members[k]->done += pass[k].n_tokens; + ++passes; + for (const auto & p : pass) rows += p.n_tokens; + } + for (auto & s : seqs) { + if (s.item->error.empty() && s.done != s.item->prefix) s.item->error = "staged prefill could not be chunked"; + s.item->ok = s.item->error.empty(); + } + std::fprintf(stderr, "[deepseek4] staged prefill: %zu requests, %d rows in %d shared passes, %.0f ms\n", + seqs.size(), rows, passes, elapsed_s(t0) * 1000.0); } void DeepSeek4Backend::join_image_stream() { @@ -1432,8 +1517,10 @@ bool DeepSeek4Backend::load_model() { const bool two_gpu_ok = tp.in_process && tp.backend_valid && tp.secondary_backend == PlacementBackend::Hip && tp.secondary_gpu != cfg_.device.gpu && !tp.all_on_secondary && !force_full; + // Batched serving keeps exact prefill for text; its image admissions + // prefill into a sparse staging cache instead. if (target_backend != PlacementBackend::Hip || cfg_.device.is_layer_split() || - cfg_.prefill_mode != PrefillAttentionMode::Sparse || + (cfg_.prefill_mode != PrefillAttentionMode::Sparse && !cfg_.paged_attention) || (tp.requested && !two_gpu_ok) || env_flag_enabled("LUCE_DS4_DENSE_TP_MASK")) { std::fprintf(stderr, "[deepseek4] --mmproj requires a HIP target with --ds4-prefill sparse, " @@ -1816,6 +1903,17 @@ bool DeepSeek4Backend::init() { (unsigned long long)requested); return false; } + if (vision_) { + // Image admissions prefill on the single-request sparse path into + // this staging cache, then copy it into their paged slot. + if (!create_deepseek4_cache(backend_, w_, max_ctx, cache_)) { + std::fprintf(stderr, "[deepseek4] image staging cache allocation failed (ctx=%d)\n", max_ctx); + return false; + } + cache_.prefill_mode = PrefillAttentionMode::Sparse; + std::fprintf(stderr, "[deepseek4] batched image serving: %d slots, staging cache ctx=%d\n", + cfg_.max_concurrency, max_ctx); + } } else { if (!create_deepseek4_cache(backend_, w_, max_ctx, cache_)) { std::fprintf(stderr, "[deepseek4] failed to allocate KV cache (ctx=%d)\n", max_ctx); @@ -1834,7 +1932,7 @@ bool DeepSeek4Backend::init() { return false; } if (cfg_.paged_attention && moe_hybrid_ && - moe_hybrid_->streams_cold_experts()) { + !moe_hybrid_->materialized_cold_experts) { std::fprintf(stderr, "[deepseek4] paged serving requires statically materialized " "expert ownership; enable in-process LUCE_DS4_MOE_TP\n"); @@ -2499,7 +2597,7 @@ bool DeepSeek4Backend::init_hybrid_model() { "[deepseek4] speculative verifier routes all experts " "to the duplicated secondary stack\n"); } - if (hybrid->has_mmap() && hybrid->streams_cold_experts()) { + if (hybrid->has_mmap() && !hybrid->materialized_cold_experts) { size_t max_expert_bytes = 0; for (const auto & layer : hybrid->layers) { const size_t per_expert_bytes = layer.fused_gate_up @@ -2526,9 +2624,7 @@ bool DeepSeek4Backend::init_hybrid_model() { w_.moe_hybrid = true; const int total_cold = w_.n_layer * w_.n_expert - moe_placement_.total_hot; const char * cold_backend = - moe_hybrid_->cold_backend_kind == MoeHybridColdBackend::Gpu ? "gpu" - : moe_hybrid_->cold_backend_kind == MoeHybridColdBackend::None ? "none" - : "cpu"; + moe_hybrid_->cold_backend_kind == MoeHybridColdBackend::Gpu ? "gpu" : "cpu"; std::fprintf(stderr, "[deepseek4] hybrid experts ready: hot=%d cold=%d cold_backend=%s%s\n", moe_placement_.total_hot, total_cold, cold_backend, ""); return true; @@ -2750,12 +2846,13 @@ int DeepSeek4Backend::do_prefill(const std::vector & tokens, int kv_offset, int snap_slot, int snap_pos, - const DeepSeek4ImagePrompt * images) { + const DeepSeek4ImagePrompt * images, + int prefix_tokens) { // Image prompts capture DSpark features from their text chunks only: the // image graph takes no capture hooks (see the chunking below). const bool capture_spec = spec_enabled_ && spec_drafter_; const InferencePhase phase = deepseek4_roctx_prefill_phase( - prefill_attention_mode_name(cfg_.prefill_mode)); + prefill_attention_mode_name(cache_.prefill_mode)); const DeepSeek4RoctxPhaseScope roctx_phase(phase); const DeepSeek4RoctxRange roctx_range( "ds4.prefill", @@ -2776,7 +2873,8 @@ int DeepSeek4Backend::do_prefill(const std::vector & tokens, // Mixed hot/cold hybrid execution still has single-token HC semantics, so // retain the reference path there. --chunk 1 is the explicit fallback. const int requested_chunk = cfg_.chunk > 0 ? cfg_.chunk : w_.n_swa; - const int n_total = (int)tokens.size(); + const int n_total = prefix_tokens > 0 + ? std::min(prefix_tokens, (int)tokens.size()) : (int)tokens.size(); // Bound the layer-major graph to the topology validated by the prefill // kernels. Smaller tail chunks use the same scheduler or its reference // fallback. @@ -2787,17 +2885,17 @@ int DeepSeek4Backend::do_prefill(const std::vector & tokens, // hybrid execution remains tokenwise; batching it would skip per-token HC // post-mixing and corrupt the hidden state. const bool hybrid_batch_supported = - !moe_hybrid_ || cfg_.prefill_mode == PrefillAttentionMode::Sparse; + !moe_hybrid_ || cache_.prefill_mode == PrefillAttentionMode::Sparse; const int base_chunk = !hybrid_batch_supported || - (cfg_.prefill_mode == PrefillAttentionMode::Exact && + (cache_.prefill_mode == PrefillAttentionMode::Exact && spec_drafter_ != nullptr) ? 1 : std::max(1, std::min(requested_chunk, layer_major_cap)); const bool bound_hybrid_scratch = moe_hybrid_ && - cfg_.prefill_mode == PrefillAttentionMode::Sparse; + cache_.prefill_mode == PrefillAttentionMode::Sparse; const int chunk = bound_hybrid_scratch ? deepseek4_hybrid_prefill_chunk_tokens( base_chunk, kv_offset + n_total, @@ -3052,7 +3150,7 @@ int DeepSeek4Backend::do_prefill(const std::vector & tokens, need_logits ? &logits : nullptr, tokens.data() + i, timing ? &step_tel : nullptr, - cfg_.prefill_mode != PrefillAttentionMode::Sparse, hp, + cache_.prefill_mode != PrefillAttentionMode::Sparse, hp, /*moe_hybrid=*/nullptr, /*expert_runtime=*/nullptr, /*routing_stats=*/nullptr, images ? images->spans() : vision::ImageSpanView{}); diff --git a/server/src/deepseek4/deepseek4_backend.h b/server/src/deepseek4/deepseek4_backend.h index 51ea90169..729dd13ee 100644 --- a/server/src/deepseek4/deepseek4_backend.h +++ b/server/src/deepseek4/deepseek4_backend.h @@ -75,12 +75,12 @@ class DeepSeek4Backend : public ModelBackend { void print_ready_banner() const override; bool supports_images() const override { return image_capable_ && vision_ != nullptr; } std::string image_placeholder() const override { return vision::DS4V_IMAGE_PLACEHOLDER; } - bool prepare_images(std::vector & tokens, - std::vector images, - uint64_t context_capacity, - uint64_t output_reserve, - ImagePromptHandle & payload, - std::string & error) const override; + ImagePrepareStatus prepare_images(std::vector & tokens, + std::vector images, + uint64_t context_capacity, + uint64_t output_reserve, + ImagePromptHandle & payload, + std::string & error) const override; bool park(ParkTarget target) override; bool unpark(ParkTarget target) override; @@ -137,7 +137,6 @@ class DeepSeek4Backend : public ModelBackend { ggml_backend_t vision_backend_ = nullptr; // Encodes images on vision_backend_ while prefill consumes them. std::thread image_stream_; - vision::ImageRequestGate image_request_gate_; vision::ImageAdmissionReserves image_reserves_; // Sampler @@ -196,13 +195,31 @@ class DeepSeek4Backend : public ModelBackend { int snapshot_capture_to); // Prefill prompt tokens in chunks, return absolute committed position. + // prefix_tokens > 0 prefills only that many leading tokens (the batched + // image admission leaves the last prompt token to the paged engine). int do_prefill(const std::vector & tokens, const DaemonIO & io, int kv_offset = 0, int snap_slot = -1, int snap_pos = -1, - const DeepSeek4ImagePrompt * images = nullptr); + const DeepSeek4ImagePrompt * images = nullptr, + int prefix_tokens = 0); bool load_vision(); bool init_single_gpu_vision(); // Waits for a streaming image encode started by materialize_images. void join_image_stream(); + // Batched serving. encode_image_request materializes an image request's + // rows; prefill_staged fills each request's first `prefix` tokens into + // its own staging cache in shared layer-major passes (expert weights read + // once per pass for every request in it). + struct StagedPrefill { + ImagePromptHandle images; + const std::vector * prompt = nullptr; + int prefix = 0; + DeepSeek4Cache * staging = nullptr; + bool ok = false; + std::string error; + }; + bool encode_image_request(const std::vector & prompt, const ImagePromptHandle & images, + std::string & error); + void prefill_staged(std::vector & batch); bool materialize_images(const DeepSeek4ImagePrompt & images, const DaemonIO & io, std::string & error); diff --git a/server/src/deepseek4/deepseek4_graph.cpp b/server/src/deepseek4/deepseek4_graph.cpp index cb91d7387..c7a77c918 100644 --- a/server/src/deepseek4/deepseek4_graph.cpp +++ b/server/src/deepseek4/deepseek4_graph.cpp @@ -7739,6 +7739,241 @@ static int ds4_try_layer_major_prefill( return (out_logits && out_logits->empty()) ? -1 : 1; } + +static bool initialize_layer_range_cache( + DeepSeek4LayerRangeCache & runtime, ggml_backend_t backend, int device, + const DeepSeek4Weights & w, int layer_begin, int layer_end, bool owns_output); + +bool deepseek4_prefill_multi(ggml_backend_t backend, int device, + const DeepSeek4Weights & w, + const std::vector & seqs, + std::string & error) { + const auto fail_early = [&](const char * why) { error = why; return false; }; + if (!backend || seqs.empty() || w.moe_hybrid || !ds4_backend_is_gpu(backend)) + return fail_early("shared prefill needs a full model on one GPU"); + int total = 0; + for (const auto & s : seqs) { + if (!s.cache || !s.embed || !s.token_ids || s.n_tokens < DS4_MIN_LAYER_MAJOR_PREFILL_TOKENS || s.kv_start < 0 || + s.cache->prefill_mode == PrefillAttentionMode::Exact || + s.kv_start + s.n_tokens > s.cache->max_ctx) + return fail_early("invalid shared prefill sequence"); + total += s.n_tokens; + } + if (total > DS4_MAX_LAYER_MAJOR_PREFILL_TOKENS) return fail_early("shared prefill exceeds the pass size"); + + // Runtime (HC weights, hash tables, HC mirrors) from the first cache. + DeepSeek4Cache & owner = *seqs.front().cache; + if (!owner.layer_range_cache) owner.layer_range_cache = new DeepSeek4LayerRangeCache(); + DeepSeek4LayerRangeCache & runtime = *owner.layer_range_cache; + if (!runtime.matches(w, backend, device, 0, w.n_layer, true) && + !initialize_layer_range_cache(runtime, backend, device, w, 0, w.n_layer, true)) + return fail_early("layer runtime initialization failed"); + auto & fc = runtime.fused_decode_graph_cache; + const auto & hc_weights = runtime.hc_layer_weights; + const auto & hc_out_weights = runtime.hc_output_weights; + const auto & hash_tables = runtime.hash_routing_tables; + if (fc.owner_ctx != w.ctx || fc.backend != backend) { + fc.destroy(); fc.owner_ctx = w.ctx; fc.backend = backend; + } + if (!ds4_fused_ensure_fn_mirrors(fc, backend, w, hc_weights, hc_out_weights)) + return fail_early("HC weight mirrors failed"); + + const int n_embd = w.n_embd, n_hc = w.n_hc; + const int64_t hc_dim = (int64_t) n_embd * n_hc; + const int64_t mix_dim = 2 * (int64_t) n_hc + (int64_t) n_hc * n_hc; + + // Row offsets, concatenated ids and "is image row" flags for the pass. + std::vector offset(seqs.size()); + std::vector ids((size_t) total); + std::vector image_row((size_t) total, 0); + bool any_image = false; + for (size_t k = 0, off = 0; k < seqs.size(); off += (size_t) seqs[k].n_tokens, ++k) { + offset[k] = (int) off; + std::copy_n(seqs[k].token_ids, seqs[k].n_tokens, ids.begin() + (ptrdiff_t) off); + for (int t = 0; t < seqs[k].n_tokens; ++t) { + if (vision::image_block_at(seqs[k].image_spans, uint64_t(seqs[k].kv_start + t))) { + image_row[off + (size_t) t] = 1; + any_image = true; + } + } + } + + ggml_init_params state_params{}; + state_params.mem_size = 4 * ggml_tensor_overhead() + 4096; + state_params.no_alloc = true; + ggml_context * state_ctx = ggml_init(state_params); + if (!state_ctx) return fail_early("state context failed"); + ggml_tensor * state_a = ggml_new_tensor_2d(state_ctx, GGML_TYPE_F32, hc_dim, total); + ggml_tensor * state_b = ggml_new_tensor_2d(state_ctx, GGML_TYPE_F32, hc_dim, total); + ggml_backend_buffer_t state_buf = ggml_backend_alloc_ctx_tensors(state_ctx, backend); + if (!state_buf) { ggml_free(state_ctx); return fail_early("state allocation failed"); } + { + std::vector initial((size_t) hc_dim * total); + for (size_t k = 0; k < seqs.size(); ++k) { + for (int t = 0; t < seqs[k].n_tokens; ++t) { + float * dst = initial.data() + (size_t) (offset[k] + t) * hc_dim; + for (int h = 0; h < n_hc; ++h) { + std::memcpy(dst + (size_t) h * n_embd, seqs[k].embed + (size_t) t * n_embd, + sizeof(float) * (size_t) n_embd); + } + } + } + ggml_backend_tensor_set(state_a, initial.data(), 0, sizeof(float) * initial.size()); + } + ggml_gallocr_t alloc = ds4_layer_major_get_shared_alloc(w, backend); + const size_t meta_bytes = 160u * 1024 * 1024; + if (ds4_layer_major_meta_owner != w.ctx) { + ds4_layer_major_meta_arena.clear(); + ds4_layer_major_meta_arena.shrink_to_fit(); + ds4_layer_major_meta_owner = w.ctx; + } + if (ds4_layer_major_meta_arena.size() < meta_bytes) ds4_layer_major_meta_arena.resize(meta_bytes); + auto fail = [&](const char * what, int il) { + std::fprintf(stderr, "[deepseek4-prefill-multi] %s at layer %d\n", what, il); + ggml_backend_buffer_free(state_buf); + ggml_free(state_ctx); + error = what; + return false; + }; + if (!alloc) return fail("shared allocator unavailable", -1); + + std::vector hash_scratch; + ggml_tensor * state_in = state_a; + ggml_tensor * state_out = state_b; + for (int il = 0; il < w.n_layer; ++il) { + ggml_init_params params{}; + params.mem_size = ds4_layer_major_meta_arena.size(); + params.mem_buffer = ds4_layer_major_meta_arena.data(); + params.no_alloc = true; + ggml_context * ctx = ggml_init(params); + if (!ctx) return fail("metadata allocation failed", il); + ggml_cgraph * gf = ggml_new_graph_custom(ctx, 65536, false); + const DeepSeek4Layer & L = w.layers[(size_t) il]; + const HcLayerWeightsCpu & hlw = hc_weights[(size_t) il]; + + // HC pre over every row of every sequence. + ggml_tensor * norm_hc = ggml_rms_norm(ctx, state_in, w.hc_eps); + ggml_tensor * mix_attn = ggml_reshape_2d(ctx, ggml_mul_mat(ctx, fc.fn_attn_f16[(size_t) il], norm_hc), + mix_dim, total); + ggml_tensor * pre_attn = ggml_ds4_hc_pre( + ctx, mix_attn, ds4_fused_hc_base_f32(ctx, L.hc_attn_base), state_in, n_hc, + w.n_hc_sinkhorn_iter, hlw.attn.scale_data[0], hlw.attn.scale_data[1], hlw.attn.scale_data[2]); + ggml_tensor * attn_in = ggml_view_2d(ctx, pre_attn, n_embd, total, pre_attn->nb[1], 0); + ggml_tensor * split_attn = ggml_view_2d(ctx, pre_attn, mix_dim, total, pre_attn->nb[1], + (size_t) n_embd * sizeof(float)); + ggml_tensor * attn_normed = ggml_cont(ctx, build_rms_norm(ctx, attn_in, L.attn_norm, w.rms_eps)); + + // Attention per sequence, each against its own cache. + std::vector i32_inputs; + std::vector i32_array_inputs; + std::vector i64_array_inputs; + std::vector f32_array_inputs; + ggml_tensor * attn_out = nullptr; + for (size_t k = 0; k < seqs.size(); ++k) { + const DeepSeek4PrefillSeq & s = seqs[k]; + ggml_tensor * rows = ggml_view_2d(ctx, attn_normed, n_embd, s.n_tokens, attn_normed->nb[1], + (size_t) offset[k] * attn_normed->nb[1]); + ggml_tensor * out = build_mla_attention( + ctx, gf, rows, w, L, s.cache->layers[(size_t) il], il, s.kv_start, s.n_tokens, + nullptr, i32_inputs, i32_array_inputs, i64_array_inputs, &f32_array_inputs, + DeepSeek4AttentionImpl::SparseFlash, /*boundary_checkpoint=*/nullptr, s.image_spans); + if (!out) { ggml_free(ctx); return fail("attention graph build failed", il); } + attn_out = attn_out ? ggml_concat(ctx, attn_out, out, 1) : out; + } + ggml_tensor * hc_after_attn = ggml_ds4_hc_post(ctx, state_in, attn_out, split_attn, n_hc); + + // HC pre -> one MoE FFN over all rows. + norm_hc = ggml_rms_norm(ctx, hc_after_attn, w.hc_eps); + ggml_tensor * mix_ffn = ggml_reshape_2d(ctx, ggml_mul_mat(ctx, fc.fn_ffn_f16[(size_t) il], norm_hc), + mix_dim, total); + ggml_tensor * pre_ffn = ggml_ds4_hc_pre( + ctx, mix_ffn, ds4_fused_hc_base_f32(ctx, L.hc_ffn_base), hc_after_attn, n_hc, + w.n_hc_sinkhorn_iter, hlw.ffn.scale_data[0], hlw.ffn.scale_data[1], hlw.ffn.scale_data[2]); + ggml_tensor * ffn_in = ggml_view_2d(ctx, pre_ffn, n_embd, total, pre_ffn->nb[1], 0); + ggml_tensor * split_ffn = ggml_view_2d(ctx, pre_ffn, mix_dim, total, pre_ffn->nb[1], + (size_t) n_embd * sizeof(float)); + ggml_tensor * ffn_normed = build_rms_norm(ctx, ffn_in, L.ffn_norm, w.rms_eps); + const bool hash_routed = il < w.n_hash_layer && L.ffn_gate_tid2eid && hash_tables[(size_t) il].loaded; + ggml_tensor * selection_bias = nullptr; + ggml_tensor * hash_ids = nullptr; + ggml_tensor * ffn_out = nullptr; + if (any_image) { + if (!L.ffn_gate_bias_vl) { ggml_free(ctx); return fail("image rows without an image router bias", il); } + selection_bias = ggml_new_tensor_2d(ctx, GGML_TYPE_F32, w.n_expert, total); + ggml_set_input(selection_bias); + ffn_out = build_moe_ffn(ctx, ffn_normed, w, L, il, total, selection_bias); + } else if (hash_routed) { + hash_ids = ggml_new_tensor_2d(ctx, GGML_TYPE_I32, w.n_expert_used, total); + ggml_set_input(hash_ids); + ffn_out = ds4_build_hash_routed_ffn(ctx, w, L, ffn_normed, hash_ids, total); + } else { + ffn_out = build_moe_ffn(ctx, ffn_normed, w, L, il, total); + } + if (!ffn_out) { ggml_free(ctx); return fail("FFN graph build failed", il); } + ggml_tensor * hc_next = ggml_ds4_hc_post(ctx, hc_after_attn, ffn_out, split_ffn, n_hc); + ggml_tensor * state_copy = ggml_cpy(ctx, hc_next, state_out); + ggml_set_output(state_copy); + ggml_build_forward_expand(gf, state_copy); + + if (!ggml_gallocr_alloc_graph(alloc, gf)) { ggml_free(ctx); return fail("scratch allocation failed", il); } + for (const auto & b : i32_inputs) ggml_backend_tensor_set(b.tensor, &b.value, 0, sizeof(b.value)); + for (const auto & b : i32_array_inputs) + ggml_backend_tensor_set(b.tensor, b.values.data(), 0, sizeof(int32_t) * b.values.size()); + for (const auto & b : i64_array_inputs) + ggml_backend_tensor_set(b.tensor, b.values.data(), 0, sizeof(int64_t) * b.values.size()); + for (const auto & b : f32_array_inputs) + ggml_backend_tensor_set(b.tensor, b.values.data(), 0, sizeof(float) * b.values.size()); + if (hash_ids) { + const int n_used = w.n_expert_used; + hash_scratch.resize((size_t) n_used * total); + const auto & table = hash_tables[(size_t) il].ids; + for (int t = 0; t < total; ++t) { + std::memcpy(hash_scratch.data() + (size_t) t * n_used, table.data() + (size_t) ids[(size_t) t] * n_used, + sizeof(int32_t) * (size_t) n_used); + } + ggml_backend_tensor_set(hash_ids, hash_scratch.data(), 0, sizeof(int32_t) * hash_scratch.size()); + } + if (selection_bias) { + // Same rule as the single-sequence image path: image rows take the + // image router bias; text rows the layer bias, and on hash-routed + // layers a large bias on exactly the experts the table names. + constexpr float HASH_PICK = 1.0e4f; + const size_t n_expert = (size_t) w.n_expert; + std::vector image_bias(n_expert), text_bias(n_expert, 0.0f); + ggml_backend_tensor_get(L.ffn_gate_bias_vl, image_bias.data(), 0, sizeof(float) * n_expert); + if (!hash_routed && L.ffn_exp_probs_b) + ggml_backend_tensor_get(L.ffn_exp_probs_b, text_bias.data(), 0, sizeof(float) * n_expert); + std::vector bias(n_expert * (size_t) total); + for (int t = 0; t < total; ++t) { + float * row = bias.data() + (size_t) t * n_expert; + if (image_row[(size_t) t]) { std::copy(image_bias.begin(), image_bias.end(), row); continue; } + std::copy(text_bias.begin(), text_bias.end(), row); + if (hash_routed) { + const int32_t * picks = hash_tables[(size_t) il].ids.data() + + (size_t) ids[(size_t) t] * (size_t) w.n_expert_used; + for (int j = 0; j < w.n_expert_used; ++j) row[picks[j]] = HASH_PICK; + } + } + ggml_backend_tensor_set(selection_bias, bias.data(), 0, sizeof(float) * bias.size()); + } + if (ggml_backend_graph_compute(backend, gf) != GGML_STATUS_SUCCESS) { ggml_free(ctx); return fail("compute failed", il); } + ggml_free(ctx); + const int ratio = (int) w.compress_ratios[(size_t) il]; + for (const auto & s : seqs) { + if (ratio <= 0) continue; + DeepSeek4LayerCache & lc = s.cache->layers[(size_t) il]; + const int next_pos = s.kv_start + s.n_tokens; + lc.n_comp = std::max(lc.n_comp, next_pos / ratio); + if (ratio == 4) lc.n_index_comp = std::max(lc.n_index_comp, next_pos / ratio); + } + std::swap(state_in, state_out); + } + for (const auto & s : seqs) s.cache->cur_pos = s.kv_start + s.n_tokens; + ggml_backend_buffer_free(state_buf); + ggml_free(state_ctx); + return true; +} + static bool ds4_hc_layer_weights_ready(const HcWeightsCpu & weights, int n_embd, int n_hc) { diff --git a/server/src/deepseek4/deepseek4_image_assembly.cpp b/server/src/deepseek4/deepseek4_image_assembly.cpp index 204107476..de3135307 100644 --- a/server/src/deepseek4/deepseek4_image_assembly.cpp +++ b/server/src/deepseek4/deepseek4_image_assembly.cpp @@ -57,15 +57,6 @@ void cancelled(const ImageCancelled & callback) { } } // namespace -std::shared_ptr ImageRequestGate::try_acquire() const { - bool expected = false; - if (!active_->compare_exchange_strong(expected, true, std::memory_order_acq_rel)) return {}; - // shared_ptr invokes the deleter if control-block allocation throws too. - return std::shared_ptr(active_.get(), [active = active_](void *) { - active->store(false, std::memory_order_release); - }); -} - bool assemble_image_rows(const ImageLayout & layout, const ImageRaster & raster, const ImageSentinels & sentinels, size_t dimension, std::vector & output, std::string & error) { diff --git a/server/src/deepseek4/deepseek4_image_assembly.h b/server/src/deepseek4/deepseek4_image_assembly.h index f343a6a41..aede51119 100644 --- a/server/src/deepseek4/deepseek4_image_assembly.h +++ b/server/src/deepseek4/deepseek4_image_assembly.h @@ -1,21 +1,11 @@ #pragma once #include "deepseek4_image_prompt.h" -#include #include #include namespace luce::vision { -// Bound decoded/prepared image memory to one outstanding request. The lease -// travels with the immutable payload and may be released by another thread. -class ImageRequestGate { -public: - std::shared_ptr try_acquire() const; -private: - std::shared_ptr> active_ = std::make_shared>(false); -}; - struct ImageRaster { size_t rows = 0, columns = 0; std::vector values; diff --git a/server/src/deepseek4/deepseek4_internal.h b/server/src/deepseek4/deepseek4_internal.h index 48c20aba5..7e7432da7 100644 --- a/server/src/deepseek4/deepseek4_internal.h +++ b/server/src/deepseek4/deepseek4_internal.h @@ -37,6 +37,8 @@ namespace luce::common { // the raw-cache rounding boundary between them. inline constexpr int DS4_NUMERICAL_PREFILL_BAND = 2048; inline constexpr int DS4_MAX_LAYER_MAJOR_PREFILL_TOKENS = 10240; +// Chunks of four rows or fewer take the decode-shaped path, not layer-major. +inline constexpr int DS4_MIN_LAYER_MAJOR_PREFILL_TOKENS = 5; // Normal verification stays within one ratio-4 compressor window. Q5 is an // explicit opt-in whose fused graph models a second boundary. inline constexpr int DS4_CONSERVATIVE_VERIFY_MAX_TOKENS = 4; @@ -451,6 +453,15 @@ bool create_deepseek4_paged_cache(ggml_backend_t backend, uint32_t slots, uint32_t max_ctx, uint32_t physical_blocks, DeepSeek4PagedCache & out); +// Copies the first n_tokens of a single-request cache (prefilled from position +// 0) into one paged slot: the raw SWA ring, the completed compressed and +// indexer rows through the slot's block table, and the compressor states. +// Both caches must come from the same weights. The slot's first n_tokens +// blocks must already be in block_table. +bool import_deepseek4_paged_slot(const DeepSeek4Cache & src, int n_tokens, + DeepSeek4PagedCache & dst, uint32_t slot, + const int32_t * block_table, uint32_t block_table_len, + std::string & error); void reset_deepseek4_paged_slot(DeepSeek4PagedCache & c, uint32_t slot); void free_deepseek4_paged_cache(DeepSeek4PagedCache & c); // Exact gathered-reference decode for up to six independent lanes. Inputs are @@ -565,6 +576,28 @@ bool deepseek4_step_layer_range( MoeHybridRoutingStats * routing_stats = nullptr, vision::ImageSpanView image_spans = {}); +// One sequence of a shared prefill pass: `n_tokens` rows of `embed` starting +// at `kv_start` of `cache`, with the sequence's image spans in its own prompt +// positions. +struct DeepSeek4PrefillSeq { + DeepSeek4Cache * cache = nullptr; + const float * embed = nullptr; // [n_tokens, n_embd] + const int32_t * token_ids = nullptr; // n_tokens ids (image rows use their marker ids) + int n_tokens = 0; + int kv_start = 0; + vision::ImageSpanView image_spans; +}; + +// Prefills several independent sequences in one layer-major pass on a full +// (non-hybrid) GPU model with sparse attention. Attention runs per sequence +// against its own cache; the HC mixing and the MoE FFN run once over all the +// sequences' rows, so every layer's expert weights are read once for all of +// them. Produces no logits and no feature capture. +bool deepseek4_prefill_multi(ggml_backend_t backend, int device, + const DeepSeek4Weights & w, + const std::vector & seqs, + std::string & error); + bool deepseek4_validate_image_batch( const DeepSeek4Weights & w, const DeepSeek4Cache & cache, const MoeHybridStorage * hybrid, const int32_t * tokens, diff --git a/server/src/deepseek4/deepseek4_paged_cache.cpp b/server/src/deepseek4/deepseek4_paged_cache.cpp index da8f2b8e0..1ee2af1ad 100644 --- a/server/src/deepseek4/deepseek4_paged_cache.cpp +++ b/server/src/deepseek4/deepseek4_paged_cache.cpp @@ -6,7 +6,10 @@ #include "deepseek4_internal.h" #endif +#include #include +#include +#include #include #include @@ -222,6 +225,78 @@ void reset_deepseek4_paged_slot(DeepSeek4PagedCache & c, uint32_t slot) { } } +bool import_deepseek4_paged_slot(const DeepSeek4Cache & src, int n_tokens, + DeepSeek4PagedCache & dst, uint32_t slot, + const int32_t * block_table, uint32_t block_table_len, + std::string & error) { + const auto fail = [&](const char * why) { error = why; return false; }; + if (!dst.buf || slot >= dst.plan.slots || n_tokens <= 0 || !block_table || + src.layers.size() != dst.layers.size()) return fail("invalid paged import request"); + const uint64_t blocks_needed = (uint64_t(n_tokens) + DS4_PAGE_TOKENS - 1) / DS4_PAGE_TOKENS; + if (blocks_needed > block_table_len) return fail("paged import exceeds the block table"); + for (uint64_t b = 0; b < blocks_needed; ++b) { + if (block_table[b] < 0 || uint32_t(block_table[b]) >= dst.plan.physical_blocks) + return fail("paged import block is not allocated"); + } + const auto same_rows = [](const ggml_tensor * a, const ggml_tensor * b) { + return a && b && a->type == b->type && a->ne[0] == b->ne[0] && a->nb[1] == b->nb[1]; + }; + // One whole slot plane of a [width, rows, slots] tensor from a [width, rows] one. + const auto copy_plane = [&](const ggml_tensor * from, ggml_tensor * to) { + if (!from && !to) return true; + if (!same_rows(from, to) || from->ne[1] != to->ne[1] || ggml_nbytes(from) != to->nb[2]) return false; + std::vector host(ggml_nbytes(from)); + ggml_backend_tensor_get(from, host.data(), 0, host.size()); + ggml_backend_tensor_set(to, host.data(), size_t(slot) * to->nb[2], host.size()); + return true; + }; + // Completed compression groups, chronological in src, paged in dst. + // Groups inside one logical block land on consecutive rows of its page. + const auto copy_groups = [&](const ggml_tensor * from, ggml_tensor * to, uint32_t ratio, int groups) { + if (!groups) return true; + if (!same_rows(from, to) || groups > from->ne[1]) return false; + std::vector host(size_t(groups) * from->nb[1]); + ggml_backend_tensor_get(from, host.data(), 0, host.size()); + const int per_block = int(DS4_PAGE_TOKENS / ratio); + for (int g = 0; g < groups;) { + const uint64_t end_token = uint64_t(g) * ratio + ratio - 1; + const uint64_t logical = end_token / DS4_PAGE_TOKENS; + uint64_t row = 0; bool emitted = false; + if (!ds4_compressed_page_row(end_token, uint32_t(block_table[logical]), ratio, row, emitted) || + !emitted || row >= uint64_t(to->ne[1])) return false; + const int run = std::min(groups - g, per_block - int((end_token % DS4_PAGE_TOKENS) / ratio)); + ggml_backend_tensor_set(to, host.data() + size_t(g) * from->nb[1], + size_t(row) * to->nb[1], size_t(run) * from->nb[1]); + g += run; + } + return true; + }; + for (size_t il = 0; il < dst.layers.size(); ++il) { + const DeepSeek4LayerCache & s = src.layers[il]; + DeepSeek4PagedLayerCache & d = dst.layers[il]; + // The single-request ring has the paged ring's 128 rows and the same + // position % 128 indexing, so the whole ring moves as one plane. + if (!same_rows(s.raw_kv, d.raw_kv) || s.raw_kv->ne[1] != int64_t(DS4_PAGE_TOKENS) || + ggml_nbytes(s.raw_kv) != d.raw_kv->nb[2]) return fail("raw ring layouts differ"); + if (!copy_plane(s.raw_kv, d.raw_kv)) return fail("raw ring copy failed"); + if (!d.ratio) continue; + const int groups = n_tokens / int(d.ratio); + if (s.n_comp != groups) return fail("compressed row count does not match the prefix"); + if (!copy_groups(s.comp_kv, d.comp_kv, d.ratio, groups)) return fail("compressed row copy failed"); + if (!copy_plane(s.attn_compressor.state_kv, d.attn_compressor.state_kv) || + !copy_plane(s.attn_compressor.state_score, d.attn_compressor.state_score)) + return fail("compressor state copy failed"); + if (d.index_comp_kv) { + if (s.n_index_comp != groups) return fail("indexer row count does not match the prefix"); + if (!copy_groups(s.index_comp_kv, d.index_comp_kv, d.ratio, groups) || + !copy_plane(s.indexer_compressor.state_kv, d.indexer_compressor.state_kv) || + !copy_plane(s.indexer_compressor.state_score, d.indexer_compressor.state_score)) + return fail("indexer copy failed"); + } + } + return true; +} + void free_deepseek4_paged_cache(DeepSeek4PagedCache & c) { deepseek4_release_paged_gathered_runtime(c); if (c.buf) { ggml_backend_buffer_free(c.buf); c.buf = nullptr; } diff --git a/server/src/deepseek4/deepseek4_seq_engine.cpp b/server/src/deepseek4/deepseek4_seq_engine.cpp index a8856f6fe..f1f620248 100644 --- a/server/src/deepseek4/deepseek4_seq_engine.cpp +++ b/server/src/deepseek4/deepseek4_seq_engine.cpp @@ -80,6 +80,99 @@ SeqEngine::AdmitResult DeepSeek4SeqEngine::admit( return result; } +DeepSeek4SeqEngine::~DeepSeek4SeqEngine() { + for (auto & cache : staging_caches_) { + if (cache) free_deepseek4_cache(*cache); + } +} + +bool DeepSeek4SeqEngine::supports_images() const { + return b_.image_capable_ && b_.vision_ && b_.cache_.buf; +} + +SeqEngine::AdmitResult DeepSeek4SeqEngine::admit_images( + uint64_t request_id, const std::vector & prompt, + const SamplerCfg & sampler, const ImagePromptHandle & images) { + AdmitResult refused; + refused.status = AdmitResult::Status::failed; + // DS4V image blocks need whole-block bidirectional prefill, which the + // 16-row gathered graph cannot run. Admission claims a slot, encodes the + // images and seeds the slot with every prompt token but the last; the + // next step() prefills all pending image requests together on the + // layer-major sparse path and copies their state into the slots, and the + // last (text) token then prefills in the batch, yielding the first + // sampled token as usual. + const int prefix = int(prompt.size()) - 1; + if (!supports_images() || prefix < DS4_MIN_LAYER_MAJOR_PREFILL_TOKENS || + prompt.size() > size_t(b_.cache_.max_ctx)) { + refused.error = "image support or prompt length is invalid"; + return refused; + } + // Claim the slot before encoding: a busy pool defers the request and + // retries it, and encoding first would rerun the encoder on every retry. + AdmitResult result = admit(request_id, prompt, sampler); + if (result.status != AdmitResult::Status::admitted) return result; + std::string error; + if (!b_.encode_image_request(prompt, images, error)) { + retire(result.slot); + refused.error = error.empty() ? "image encoding failed" : error; + return refused; + } + SeqSlotManager::PrefillChunk seeded = slots_.seed_restored_prefix(result.slot, prefix); + bool ok = seeded.ok && seeded.rows.size() == size_t(prefix); + for (size_t i = 0; ok && i < seeded.new_blocks.size(); ++i) { + ok = set_block(result.slot, seeded.first_new_block + int(i), seeded.new_blocks[i]); + } + if (!ok) { + retire(result.slot); + refused.error = "image prefix could not be seeded into the paged slot"; + return refused; + } + pending_images_.push_back({result.slot, images, prompt, prefix}); + return result; +} + +void DeepSeek4SeqEngine::run_pending_images(std::vector & failures, + std::vector & failed_slots) { + if (pending_images_.empty()) return; + std::vector batch(pending_images_.size()); + for (size_t k = 0; k < pending_images_.size(); ++k) { + // Staging cache k: the backend's own for the first request, then + // one more per concurrent request (allocated once, reused). + DeepSeek4Cache * staging = &b_.cache_; + if (k > 0) { + while (staging_caches_.size() < k) staging_caches_.emplace_back(); + auto & owned = staging_caches_[k - 1]; + if (!owned) { + owned = std::make_unique(); + if (!create_deepseek4_cache(b_.backend_, b_.w_, b_.cache_.max_ctx, *owned)) owned.reset(); + } + staging = owned.get(); + } + batch[k].images = pending_images_[k].images; + batch[k].prompt = &pending_images_[k].prompt; + batch[k].prefix = pending_images_[k].prefix; + batch[k].staging = staging; + if (!staging) batch[k].error = "staging cache allocation failed"; + } + b_.prefill_staged(batch); + for (size_t k = 0; k < batch.size(); ++k) { + const PendingImage & p = pending_images_[k]; + std::string error = batch[k].error; + bool ok = batch[k].ok; + if (ok) { + ok = import_deepseek4_paged_slot( + *batch[k].staging, p.prefix, b_.paged_cache_, uint32_t(p.slot), + host_tables_.data() + size_t(p.slot) * stride_, stride_, error); + } + if (!ok) { + fail_prefill(p.slot, failures, error.empty() ? "staged image prefill failed" : error); + if (p.slot >= 0 && p.slot < int(failed_slots.size())) failed_slots[size_t(p.slot)] = 1; + } + } + pending_images_.clear(); +} + bool DeepSeek4SeqEngine::set_block(int slot, int logical, int32_t physical) { if (slot < 0 || slot >= slots_.slot_count() || logical < 0 || (uint32_t) logical >= stride_) return false; @@ -150,6 +243,9 @@ SeqEngine::StepResult DeepSeek4SeqEngine::step(const StepPlan & plan) { } if (inputs.empty() && plan.prefills.empty()) return result; + std::vector image_failed((size_t)n_slots, 0); + run_pending_images(result.prefills, image_failed); + std::vector lane_tokens; std::vector lane_positions; std::vector lane_slots; @@ -200,6 +296,7 @@ SeqEngine::StepResult DeepSeek4SeqEngine::step(const StepPlan & plan) { std::vector prefill_lanes; prefill_lanes.reserve(plan.prefills.size()); for (const PrefillSlice & slice : plan.prefills) { + if (image_failed[(size_t)slice.slot]) continue; const SeqSlot & before = slots_.slot(slice.slot); const int remaining = before.prompt_len - before.cur_pos; const int n_rows = std::min(slice.max_tokens, remaining); @@ -369,6 +466,9 @@ bool DeepSeek4SeqEngine::evict_kv(int slot, int32_t pending_token, } void DeepSeek4SeqEngine::retire(int slot) { + pending_images_.erase(std::remove_if(pending_images_.begin(), pending_images_.end(), + [slot](const PendingImage & p) { return p.slot == slot; }), + pending_images_.end()); offload_.discard(slot); if (!slots_.is_active(slot)) return; slots_.retire(slot); diff --git a/server/src/deepseek4/deepseek4_seq_engine.h b/server/src/deepseek4/deepseek4_seq_engine.h index cec1d0d2d..c10126622 100644 --- a/server/src/deepseek4/deepseek4_seq_engine.h +++ b/server/src/deepseek4/deepseek4_seq_engine.h @@ -5,11 +5,14 @@ #include "common/concurrency/seq_slot_manager.h" #include +#include +#include #include namespace luce::common { class DeepSeek4Backend; +struct DeepSeek4Cache; // Exact concurrent serving path for DeepSeek4. Model state remains in // DeepSeek4PagedCache; this class owns only scheduler-facing slot state and @@ -18,6 +21,7 @@ class DeepSeek4SeqEngine final : public SeqEngine { public: DeepSeek4SeqEngine(DeepSeek4Backend & backend, PagedKvPool & pool, int max_ctx, uint32_t table_stride); + ~DeepSeek4SeqEngine() override; int slot_count() const override { return slots_.slot_count(); } int max_context() const override { return slots_.max_context(); } @@ -38,12 +42,30 @@ class DeepSeek4SeqEngine final : public SeqEngine { } void retire(int slot) override; bool token_is_eos(int32_t token) const override; + bool supports_images() const override; + AdmitResult admit_images(uint64_t request_id, + const std::vector & prompt, + const SamplerCfg & sampler, + const ImagePromptHandle & images) override; private: bool set_block(int slot, int logical, int32_t physical); void fail_prefill(int slot, std::vector & outputs, const std::string & error); + // Image requests admitted since the last step: their slots hold the + // prompt minus its last token as seeded blocks, and step() fills those + // blocks from shared staged prefill passes before running the batch. + struct PendingImage { + int slot = -1; + ImagePromptHandle images; + std::vector prompt; + int prefix = 0; + }; + std::vector pending_images_; + std::vector> staging_caches_; + void run_pending_images(std::vector & failures, std::vector & failed_slots); + DeepSeek4Backend & b_; SeqSlotManager slots_; PagedKvOffload offload_; diff --git a/server/src/qwen35/qwen35_backend.h b/server/src/qwen35/qwen35_backend.h index 423bb95d8..275dab3a8 100644 --- a/server/src/qwen35/qwen35_backend.h +++ b/server/src/qwen35/qwen35_backend.h @@ -163,12 +163,12 @@ class Qwen35Backend : public ModelBackend { // decode one token at a time. bool supports_images() const override { return image_input_; } std::string image_placeholder() const override; - bool prepare_images(std::vector & tokens, - std::vector images, - uint64_t context_capacity, - uint64_t output_reserve, - ImagePromptHandle & payload, - std::string & error) const override; + ImagePrepareStatus prepare_images(std::vector & tokens, + std::vector images, + uint64_t context_capacity, + uint64_t output_reserve, + ImagePromptHandle & payload, + std::string & error) const override; bool supports_remote_draft() const override { return true; } // ── Concurrent slot serving (paged AR decode over N sequences) ──── diff --git a/server/src/qwen35/qwen35_backend_images.cpp b/server/src/qwen35/qwen35_backend_images.cpp index abe0616b2..55d2f6bf3 100644 --- a/server/src/qwen35/qwen35_backend_images.cpp +++ b/server/src/qwen35/qwen35_backend_images.cpp @@ -50,7 +50,7 @@ std::string Qwen35Backend::image_placeholder() const { return image_input_ ? QWEN35_IMAGE_PLACEHOLDER : ""; } -bool Qwen35Backend::prepare_images(std::vector & tokens, std::vector images, +ImagePrepareStatus Qwen35Backend::prepare_images(std::vector & tokens, std::vector images, uint64_t context_capacity, uint64_t output_reserve, ImagePromptHandle & payload, std::string & error) const { payload.reset(); @@ -59,20 +59,20 @@ bool Qwen35Backend::prepare_images(std::vector & tokens, std::vector MAX_IMAGES_PER_REQUEST) { error = "too many images in request"; return false; } + if (!image_input_) { error = "this model was started without --mmproj"; return ImagePrepareStatus::invalid; } + if (images.size() > MAX_IMAGES_PER_REQUEST) { error = "too many images in request"; return ImagePrepareStatus::invalid; } try { auto prompt = std::make_shared(); prompt->owner = this; for (const EncodedImage & image : images) { auto decoded = vision::decode_image({image.bytes.data(), image.bytes.size()}); - if (!decoded) { error = decoded.status.message; return false; } + if (!decoded) { error = decoded.status.message; return ImagePrepareStatus::invalid; } vision::Qwen35Pixels pixels; - if (!vision::qwen35_vision_preprocess(vision_config_, decoded.image, pixels, error)) return false; + if (!vision::qwen35_vision_preprocess(vision_config_, decoded.image, pixels, error)) return ImagePrepareStatus::invalid; Qwen35ImageSlot slot; slot.columns = pixels.grid_columns; slot.rows = pixels.grid_rows; @@ -80,14 +80,14 @@ bool Qwen35Backend::prepare_images(std::vector & tokens, std::vectorpixels.push_back(std::move(pixels)); } const uint64_t limit = context_capacity > output_reserve ? context_capacity - output_reserve : 0; - if (!qwen35_expand_image_tokens(tokens, w_.image_pad_id, prompt->slots, limit, error)) return false; + if (!qwen35_expand_image_tokens(tokens, w_.image_pad_id, prompt->slots, limit, error)) return ImagePrepareStatus::invalid; prompt->expanded_tokens = tokens; prompt->positions = qwen35_image_rope_positions((int) tokens.size(), prompt->slots); payload = std::move(prompt); - return true; + return ImagePrepareStatus::ok; } catch (const std::bad_alloc &) { error = "image preparation allocation failed"; - return false; + return ImagePrepareStatus::invalid; } } diff --git a/server/src/server/http_server.cpp b/server/src/server/http_server.cpp index 5925b4299..4cb7ce0c1 100644 --- a/server/src/server/http_server.cpp +++ b/server/src/server/http_server.cpp @@ -2494,10 +2494,13 @@ bool HttpServer::handle_model_request(SocketHandle fd, ParsedRequest & req, if (!render_and_tokenize_request(fd, render_messages, req)) return true; std::string image_error; - if (!backend_.prepare_images(req.prompt_tokens, std::move(encoded_images), + const ImagePrepareStatus image_status = backend_.prepare_images( + req.prompt_tokens, std::move(encoded_images), uint64_t(std::max(0, config_.max_ctx)), uint64_t(std::max(0, req.max_output)), - req.images, image_error)) { - send_error(fd, 400, image_error); + req.images, image_error); + if (image_status != ImagePrepareStatus::ok) { + // A full image gate is capacity, not a bad request: clients retry 503. + send_error(fd, image_status == ImagePrepareStatus::busy ? 503 : 400, image_error); return true; } diff --git a/server/test/test_ds4v_image_assembly.cpp b/server/test/test_ds4v_image_assembly.cpp index 91fbf4830..f73664bf0 100644 --- a/server/test/test_ds4v_image_assembly.cpp +++ b/server/test/test_ds4v_image_assembly.cpp @@ -22,20 +22,6 @@ static PromptImage fixture(uint64_t position) { int main() { try { - ImageRequestGate gate; - auto lease = gate.try_acquire(); - check(bool(lease) && !gate.try_acquire(), "concurrent image request was admitted"); - auto retained_lease = lease; - lease.reset(); - check(!gate.try_acquire(), "request copy released image admission early"); - retained_lease.reset(); - lease = gate.try_acquire(); - check(bool(lease), "completed request did not release image admission"); - { - ImageRequestGate transient; - retained_lease = transient.try_acquire(); - } - retained_lease.reset(); const ImageSentinels sentinels{{10,11}, {20,21}, {30,31}, {40,41}}; const ImageRaster raster{4, 2, {100,101,200,201,300,301,400,401}}; const std::vector expected{20,21,10,11,300,301,100,101,30,31, diff --git a/server/test/test_server_unit.cpp b/server/test/test_server_unit.cpp index 7f7348eb6..370ba8868 100644 --- a/server/test/test_server_unit.cpp +++ b/server/test/test_server_unit.cpp @@ -9445,11 +9445,12 @@ TEST_CASE(ServerUnitFixture, test_default_backend_rejects_encoded_images_without ImagePromptHandle payload; std::string error; TEST_ASSERT(!backend.supports_images()); - TEST_ASSERT(!backend.prepare_images(tokens, {{"image/png", {137, 80, 78, 71}}}, - 8192, 32, payload, error)); + TEST_ASSERT(backend.prepare_images(tokens, {{"image/png", {137, 80, 78, 71}}}, + 8192, 32, payload, error) == ImagePrepareStatus::invalid); TEST_ASSERT(tokens == std::vector({1, 2, 3})); TEST_ASSERT(!payload && !error.empty()); - TEST_ASSERT(backend.prepare_images(tokens, {}, 8192, 32, payload, error)); + TEST_ASSERT(backend.prepare_images(tokens, {}, 8192, 32, payload, error) == + ImagePrepareStatus::ok); TEST_ASSERT(tokens == std::vector({1, 2, 3})); }