feat(server): PFlash prefill compression under concurrent serving (stacked on #746)

server/deps/llama.cpp/ggml/src/ggml-cuda/ggml-cuda.cu

@@ -6829,9 +6829,11 @@ static enum ggml_status ggml_backend_cuda_graph_compute(ggml_backend_t backend,
 
 #if defined(GGML_USE_HIP)
     // Drop last graph's mmb activation cache: keyed by raw data pointers the pool recycles.
+    // Empty containers are left alone: graphs without the qwen4exp profile never fill them,
+    // and another backend may be computing on another thread.
     ggml_cuda_mmb_begin_graph();
-    g_hc_marked_xn.clear();
-    g_hc_q8_xn.clear();
+    if (!g_hc_marked_xn.empty()) g_hc_marked_xn.clear();
+    if (!g_hc_q8_xn.empty()) g_hc_q8_xn.clear();
 
     // xn is a view of HC_COMBINE_NORM, so reads() keys on (view_src, view_offs); a plain pointer test misses consumers.
     if (ggml_cuda_qwen4exp_enabled() && GGML_CUDA_CC_IS_RDNA3_5(ggml_cuda_info().devices[cuda_ctx->device].cc)) {

server/deps/llama.cpp/ggml/src/ggml-cuda/mmb.cu

@@ -786,8 +786,23 @@ void ggml_cuda_mmb_mark_f16_only(const ggml_tensor * t) { if (mmb_arch()) g_mmb_
 bool ggml_cuda_mmb_is_f16_only(const ggml_tensor * t) { return g_mmb_f16_only.count(t) > 0; }
 void ggml_cuda_mmb_mark_bf16_only(const ggml_tensor * t) { if (mmb_arch()) g_mmb_bf16_only.insert(t); }
 bool ggml_cuda_mmb_is_bf16_only(const ggml_tensor * t) { return g_mmb_bf16_only.count(t) > 0; }
-void ggml_cuda_mmb_begin_graph() { for (auto & e : g_mmb_cache) delete e.buf; g_mmb_cache.clear(); for (auto & e : g_mmb_f16_cache) delete e.buf; g_mmb_f16_cache.clear(); for (auto & e : g_mmb_slots) { e.root = nullptr; e.data = nullptr; e.n = 0; } g_mmb_bf16_only.clear(); g_mmb_f16_only.clear(); g_mmb_q8_root = nullptr; g_mmb_q8_data = nullptr; }
+// True when no graph left MMB state behind. Only graphs computed under the qwen4exp profile create any, so a graph
+// without it (another model's backend, e.g. the PFlash drafter scoring on its own thread beside a serving target)
+// then only reads these process-wide containers instead of clearing them under the other thread.
+static bool mmb_graph_state_empty() {
+    if (!g_mmb_cache.empty() || !g_mmb_f16_cache.empty() || !g_mmb_bf16_only.empty() || !g_mmb_f16_only.empty()) return false;
+    if (g_mmb_q8_root || g_mmb_q8_data) return false;
+    for (const auto & e : g_mmb_slots) if (e.root || e.data || e.n) return false;
+    return true;
+}
+static bool mmb_buffers_empty() {
+    if (g_mmb_q8_buf || !g_mmb_shadow.empty()) return false;
+    for (const auto & e : g_mmb_slots) if (e.buf) return false;
+    return true;
+}
+void ggml_cuda_mmb_begin_graph() { if (!mmb_enabled() && mmb_graph_state_empty()) return; for (auto & e : g_mmb_cache) delete e.buf; g_mmb_cache.clear(); for (auto & e : g_mmb_f16_cache) delete e.buf; g_mmb_f16_cache.clear(); for (auto & e : g_mmb_slots) { e.root = nullptr; e.data = nullptr; e.n = 0; } g_mmb_bf16_only.clear(); g_mmb_f16_only.clear(); g_mmb_q8_root = nullptr; g_mmb_q8_data = nullptr; }
 void ggml_cuda_mmb_release_all() {
+    if (mmb_graph_state_empty() && mmb_buffers_empty()) return;
     ggml_cuda_mmb_begin_graph();
     delete g_mmb_q8_buf; g_mmb_q8_buf = nullptr; g_mmb_q8_cap = 0;
     for (int i = 0; i < 4; ++i) { if (g_mmb_slots[i].buf) delete g_mmb_slots[i].buf; g_mmb_slots[i].buf = nullptr; g_mmb_slot_cap[i] = 0; }

server/src/common/backend_args.h

@@ -105,6 +105,9 @@ struct BackendArgs {
     // Total paged K/V pool in tokens shared by all slots (--kv-pool-tokens;
     // model-page-rounded). 0 = derive capacity from available device memory.
     long long       kv_pool_tokens   = 0;
+    // Device memory an automatically sized pool leaves free for a model that
+    // runs beside concurrent decode (the PFlash drafter). Bytes; 0 = none.
+    long long       kv_pool_reserve_bytes = 0;
     int             kq_stride_pad    = 32;
     int             draft_block_size = 0;  // 0 = drafter metadata
     int             draft_swa_window = 0;

server/src/common/backend_factory.cpp

@@ -209,6 +209,7 @@ std::unique_ptr<ModelBackend> construct_backend(
         cfg.paged_attention    = cache.paged_attention;
         cfg.max_concurrency    = execution.max_concurrency;
         cfg.kv_pool_tokens     = cache.kv_pool_tokens;
+        cfg.kv_pool_reserve_bytes = cache.kv_pool_reserve_bytes;
         cfg.kq_stride_pad      = cache.kq_stride_pad;
         cfg.draft_block_size   = speculation.draft_block_size;
         cfg.draft_swa_window   = cache.draft_swa_window;

server/src/common/backend_factory.h

@@ -53,6 +53,7 @@ class BackendPlan final {
         int fa_window = 0;
         bool paged_attention = false;
         long long kv_pool_tokens = 0;
+        long long kv_pool_reserve_bytes = 0;
         ggml_type cache_type_k = GGML_TYPE_COUNT;
         ggml_type cache_type_v = GGML_TYPE_COUNT;
         int kq_stride_pad = 32;

server/src/common/backend_plan.cpp

@@ -144,6 +144,7 @@ BackendPreparation BackendPlanBuilder::resolve(
     plan.cache_.fa_window = args.fa_window;
     plan.cache_.paged_attention = args.paged_attention;
     plan.cache_.kv_pool_tokens = args.kv_pool_tokens;
+    plan.cache_.kv_pool_reserve_bytes = args.kv_pool_reserve_bytes;
     plan.cache_.cache_type_k = args.cache_type_k;
     plan.cache_.cache_type_v = args.cache_type_v;
     plan.cache_.kq_stride_pad = args.kq_stride_pad;

server/src/common/feature_gate.cpp

@@ -254,9 +254,14 @@ std::string check_feature_compatibility(
         if (args.fa_window != 0) {
             return "--paged-attention requires full attention (--fa-window 0)";
         }
-        if (admission.pflash_enabled) {
-            return "--paged-attention cannot be combined with PFlash prefill "
-                   "compression";
+        // Only concurrent dense Qwen3.5 serving compresses beside paged
+        // decode: its drafter scores on a thread of its own while the slots
+        // keep decoding, in memory the KV pool leaves free for it.
+        if (admission.pflash_enabled &&
+            (args.max_concurrency <= 1 || arch != "qwen35")) {
+            return "--paged-attention combines with PFlash prefill compression "
+                   "only for concurrent dense Qwen3.5 serving "
+                   "(--max-concurrency > 1)";
         }
         if (admission.fixed_kvflash_requested()) {
             return "--paged-attention cannot be combined with KVFlash";

server/src/errors.cpp

@@ -1,27 +1,27 @@
-// Thread-safe last-error string used by loaders and graph builders.
-// Consumed by tests and the test_dflash driver via luce_last_error().
+// Last-error string used by loaders and graph builders, one per thread: a
+// caller reads the error its own failing call set, and the pointer stays
+// valid until that thread sets another. A concurrent server scores PFlash
+// prompts on one thread while another serves, and neither may read, or
+// free, the other's. Consumed by tests and the test_dflash driver via
+// luce_last_error().
 
 #include "luce.h"
 #include "internal.h"
 
-#include <mutex>
 #include <string>
 
 namespace luce::common {
 
 namespace {
-std::mutex g_err_mu;
-std::string g_last_error;
+thread_local std::string g_last_error;
 }
 
 void set_last_error(std::string msg) {
-    std::lock_guard<std::mutex> lk(g_err_mu);
     g_last_error = std::move(msg);
 }
 
 } // namespace luce::common
 
 extern "C" const char * luce_last_error(void) {
-    std::lock_guard<std::mutex> lk(luce::common::g_err_mu);
     return luce::common::g_last_error.c_str();
 }

server/src/kv_quant.cpp

@@ -171,7 +171,8 @@ void validate_kv_pair_or_abort(ggml_type k, ggml_type v, const char * who) {
 }
 
 void resolve_kv_types(ggml_type & k_out, ggml_type & v_out,
-                      ggml_type k_override, ggml_type v_override) {
+                      ggml_type k_override, ggml_type v_override,
+                      bool tq3_shorthand) {
     ggml_type k = GGML_TYPE_Q4_0;
     ggml_type v = GGML_TYPE_Q4_0;
 
@@ -182,7 +183,7 @@ void resolve_kv_types(ggml_type & k_out, ggml_type & v_out,
     if (const char * s = std::getenv("LUCE_KV_Q4")) {
         if (std::atoi(s) != 0) { k = GGML_TYPE_Q4_0;  v = GGML_TYPE_Q4_0;  }
     }
-    if (const char * s = std::getenv("LUCE_KV_TQ3")) {
+    if (const char * s = tq3_shorthand ? std::getenv("LUCE_KV_TQ3") : nullptr) {
         if (std::atoi(s) != 0) { k = GGML_TYPE_TQ3_0; v = GGML_TYPE_TQ3_0; }
     }
 

server/src/kv_quant.h

@@ -27,13 +27,15 @@ void validate_kv_pair_or_abort(ggml_type k, ggml_type v, const char * who);
 // Precedence (high -> low):
 //   0. Explicit k_override / v_override (COUNT means unspecified)
 //   1. LUCE_KV_K=<type> / LUCE_KV_V=<type>  (independent override)
-//   2. LUCE_KV_F16 / _KV_Q4 / _KV_TQ3            (legacy shorthand, K==V)
+//   2. LUCE_KV_F16 / _KV_Q4 / _KV_TQ3            (legacy shorthand, K==V;
+//      tq3_shorthand=false ignores LUCE_KV_TQ3)
 //   3. Default: GGML_TYPE_Q4_0 for both (with FWHT K-rotation)
 // On invalid input or unsupported (K,V) pair, prints an explanatory message
 // and calls std::abort(). Returns the resolved pair via out params.
 void resolve_kv_types(ggml_type & k_out, ggml_type & v_out,
                       ggml_type k_override = GGML_TYPE_COUNT,
-                      ggml_type v_override = GGML_TYPE_COUNT);
+                      ggml_type v_override = GGML_TYPE_COUNT,
+                      bool tq3_shorthand = true);
 
 // KV reservation bytes per token for a hybrid attention/SSM model. Single source
 // of truth for both qwen35 (dense) and qwen35moe expert-placement budgeting.

server/src/pflash/qwen35_drafter.cpp

@@ -19,6 +19,7 @@
 #include "common/gguf_inspect.h"
 #include "anchor_params.h"
 #include "internal.h"
+#include "kv_quant.h"
 
 #include "ggml.h"
 #include "ggml-alloc.h"
@@ -52,36 +53,17 @@ static size_t causal_mask_bytes(int kv_start, int n_tokens) {
     return (size_t)align_up_i(kv_start + n_tokens, 256) * align_up_i(n_tokens, 32) * sizeof(uint16_t);
 }
 
-// create_target_cache honours LUCE_KV_TQ3; the drafter cache never wants
-// the TurboQuant rotation, so force it off while the cache is created.
-struct ScopedKvTq3Off {
-    ScopedKvTq3Off() {
-#if defined(_WIN32)
-        char * raw = nullptr;
-        size_t len = 0;
-        _dupenv_s(&raw, &len, "LUCE_KV_TQ3");
-        had_ = raw != nullptr;
-        old_ = had_ ? raw : "";
-        free(raw);
-        _putenv_s("LUCE_KV_TQ3", "0");
-#else
-        const char * raw = std::getenv("LUCE_KV_TQ3");
-        had_ = raw != nullptr;
-        old_ = had_ ? raw : "";
-        setenv("LUCE_KV_TQ3", "0", 1);
-#endif
-    }
-    ~ScopedKvTq3Off() {
-#if defined(_WIN32)
-        // _putenv_s with empty value removes the variable on MSVCRT.
-        _putenv_s("LUCE_KV_TQ3", had_ ? old_.c_str() : "");
-#else
-        if (had_) setenv("LUCE_KV_TQ3", old_.c_str(), 1);
-        else unsetenv("LUCE_KV_TQ3");
-#endif
-    }
-    bool had_ = false;
-    std::string old_;
+// The drafter cache's KV types: the target's resolution without the
+// LUCE_KV_TQ3 shorthand, since the drafter never wants the TurboQuant
+// rotation. Resolved without touching the environment, which other threads
+// read while a concurrent server scores a prompt.
+struct DrafterKvTypes {
+    ggml_type k = GGML_TYPE_COUNT;
+    ggml_type v = GGML_TYPE_COUNT;
+    DrafterKvTypes() {
+        luce::resolve_kv_types(k, v, GGML_TYPE_COUNT, GGML_TYPE_COUNT,
+                               /*tq3_shorthand=*/false);
+    }
 };
 
 } // namespace
@@ -144,11 +126,11 @@ std::vector<int32_t> qwen35_score_and_compress(
     std::vector<float> running_max((size_t)n_lookahead * S, -INFINITY);
 
     TargetCache cache;
-    {
-        ScopedKvTq3Off tq3_off;
-        if (!create_target_cache(w, S, 0, w.backend, cache, true)) {
-            return {};
-        }
+    const DrafterKvTypes kv;
+    if (!create_target_cache(w, S, 0, w.backend, cache, true, 0,
+                             /*paged_attention=*/false, 1,
+                             /*concurrent_tree=*/false, kv.k, kv.v)) {
+        return {};
     }
 
     ggml_init_params act_ip{};
@@ -621,13 +603,14 @@ int scoring_session_limit() {
 
 bool allocate_scoring_session(TargetWeights & w, int capacity,
                               Qwen35ScoringSession & session) {
-    {
-        ScopedKvTq3Off tq3_off;
-        if (!create_target_cache_partial(w, capacity, 0, w.backend, session.cache,
-                                         /*prefill_only=*/true, 0, kQwen35HeadBlock,
-                                         /*allocate_target_feat=*/false)) {
-            return false;
-        }
+    const DrafterKvTypes kv;
+    if (!create_target_cache_partial(w, capacity, 0, w.backend, session.cache,
+                                     /*prefill_only=*/true, 0, kQwen35HeadBlock,
+                                     /*allocate_target_feat=*/false, 0,
+                                     /*f32_ssm_intermediates=*/false,
+                                     /*paged_attention=*/false, 1,
+                                     /*concurrent_tree=*/false, kv.k, kv.v)) {
+        return false;
     }
     if (!ensure_ssm_snapshot(session.cache, w.backend)) {
         free_qwen35_scoring_session(session);

server/src/qwen35/qwen35_backend.cpp

@@ -509,23 +509,42 @@ bool Qwen35Backend::init() {
     // pool's index space) as the write target of dead decode-batch rows.
     int64_t pool_tokens = 0;
     if (n_slots > 1) {
+        PagedKvAutoBudget budget;
+        // TODO: Size tensor-parallel pools from each device's free memory
+        // and account for fixed cache tensors mirrored by the meta backend.
+        budget.free_bytes = (int64_t)gpu_free;
+        budget.bytes_per_token = kvf_budget.bytes_per_token;
+        budget.reserve_bytes =
+            kvf_budget.reserve_bytes + cfg_.kv_pool_reserve_bytes;
+        budget.fixed_cache_bytes = concurrent_fixed_cache_bytes(
+            w_, cfg_.device.max_ctx, n_slots, budget.bytes_per_token,
+            scratch_tokens, fixed_chain.enabled);
         if (cfg_.kv_pool_tokens > 0) {
             const int64_t max_pool_tokens =
                 ((int64_t)INT32_MAX - scratch_tokens) /
                 PAGED_BLOCK_SIZE * PAGED_BLOCK_SIZE;
             pool_tokens = (int64_t)paged_token_capacity(
                 (int)std::min<int64_t>(
                     cfg_.kv_pool_tokens, max_pool_tokens));
+            // Memory set aside for a model beside the slots (the PFlash
+            // drafter) bounds an explicit pool too.
+            if (cfg_.kv_pool_reserve_bytes > 0 && budget.bytes_per_token > 0) {
+                const int64_t fit = (std::max<int64_t>)(0,
+                    budget.free_bytes - budget.fixed_cache_bytes -
+                    budget.reserve_bytes) / budget.bytes_per_token;
+                if (pool_tokens > fit) {
+                    set_last_error(
+                        "--kv-pool-tokens " + std::to_string(pool_tokens) +
+                        " leaves less than the " +
+                        std::to_string(cfg_.kv_pool_reserve_bytes >> 20) +
+                        " MiB reserved for the PFlash drafter; at most " +
+                        std::to_string(fit / PAGED_BLOCK_SIZE * PAGED_BLOCK_SIZE) +
+                        " tokens fit");
+                    std::fprintf(stderr, "[parallel] %s\n", luce_last_error());
+                    return false;
+                }
+            }
         } else {
-            PagedKvAutoBudget budget;
-            // TODO: Size tensor-parallel pools from each device's free memory
-            // and account for fixed cache tensors mirrored by the meta backend.
-            budget.free_bytes = (int64_t)gpu_free;
-            budget.bytes_per_token = kvf_budget.bytes_per_token;
-            budget.reserve_bytes = kvf_budget.reserve_bytes;
-            budget.fixed_cache_bytes = concurrent_fixed_cache_bytes(
-                w_, cfg_.device.max_ctx, n_slots, budget.bytes_per_token,
-                scratch_tokens, fixed_chain.enabled);
             pool_tokens = paged_kv_auto_pool_tokens(
                 cfg_.device.max_ctx, n_slots, budget);
             const int64_t one_context =
@@ -1439,9 +1458,14 @@ std::vector<ModelBackend::CompressResult> Qwen35Backend::compress_batch(
 
     // Synchronize all backends to flush any outstanding async CUDA work
     // before loading the drafter. Without this, pending operations on
-    // target/draft streams can corrupt the drafter's allocations.
-    ggml_backend_synchronize(target_backend_);
-    if (draft_backend_) ggml_backend_synchronize(draft_backend_);
+    // target/draft streams can corrupt the drafter's allocations. Only
+    // parking frees their memory: a drafter beside resident models
+    // allocates its own, and concurrent serving scores on a thread of its
+    // own while the scheduler keeps the target busy.
+    if (should_park) {
+        ggml_backend_synchronize(target_backend_);
+        if (draft_backend_) ggml_backend_synchronize(draft_backend_);
+    }
 
     // Load drafter with its OWN backend (not target_backend_).
     // Matches test_dflash.cpp: separate backend supports multi-GPU

server/src/qwen35/qwen35_backend.h

@@ -75,6 +75,9 @@ struct Qwen35Config {
     // (--kv-pool-tokens; rounded up to whole 16-token blocks).
     // 0 derives the shared physical capacity from available device memory.
     int64_t      kv_pool_tokens  = 0;
+    // Device memory the automatically sized pool leaves free for the PFlash
+    // drafter, which scores beside concurrent decode (bytes).
+    int64_t      kv_pool_reserve_bytes = 0;
 
     // Draft
     int          draft_block_size = 0;  // 0 = use drafter metadata

server/src/server/admission.h

@@ -6,6 +6,21 @@
 
 #include <algorithm>
 
+// Device memory concurrent serving leaves free beside the paged KV pool for the
+// PFlash drafter, which scores on its own thread while the slots decode: its
+// weights plus a workspace that grows linearly with the prompt it scores
+// (activations, its own cache, the largest ubatch's attention). The
+// per-token figure is the Qwen3.5-0.8B drafter's measured peak, 53 KiB per
+// token from 33K to 206K tokens, rounded up; its fixed part (0.6 GiB) is
+// under the weights.
+inline constexpr long long kPFlashScoringBytesPerToken = 56LL * 1024;
+
+inline long long pflash_concurrent_reserve_bytes(long long drafter_weight_bytes, int max_prompt_tokens)
+{
+    return (std::max)(0LL, drafter_weight_bytes) +
+           (long long)(std::max)(0, max_prompt_tokens) * kPFlashScoringBytesPerToken;
+}
+
 // The reply a request can get: what it asked for, trimmed to the context its
 // prompt leaves (0 when the prompt fills it). Clients ask for their usual cap
 // whatever the prompt, Claude Code 32000 tokens, so a long session would