Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions server/src/qwen4exp/qwen4exp_backend.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ static size_t qwen4exp_free_memory(ggml_backend_t backend) {

int qwen4exp_select_chunk(ggml_backend_t backend, const Qwen4ExpWeights & w,
Qwen4ExpCache & cache, int slots, int resident_slots, size_t * snapshot_budget, Qwen4ExpChunkPlan * plan,
size_t snapshot_keep, size_t reserved) {
size_t snapshot_keep, size_t reserved, int restore_spacing) {
if (slots < 1 || resident_slots < 1 || resident_slots > slots) return 0;
const size_t free_now = qwen4exp_free_memory(backend);
const size_t available = free_now > reserved ? free_now - reserved : 0;
Expand Down Expand Up @@ -284,11 +284,17 @@ int qwen4exp_select_chunk(ggml_backend_t backend, const Qwen4ExpWeights & w,
plan->layer_bytes = (size_t) slots * layer;
if (plan->required > available) return 0; // the context fit moves on without searching smaller chunks
}
const auto restore_point_spacing = [restore_spacing] { return restore_spacing; };
const size_t min_snapshot = snapshot_budget

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: This minimum is applied to single-sequence snapshots too, so a short valid prefix capture is disabled whenever a 4096-token checkpoint does not fit. Preserve the single-sequence path by applying this minimum only to the concurrent budget, or derive its minimum from the actual single-sequence capture boundary.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At server/src/qwen4exp/qwen4exp_backend.cpp, line 280:

<comment>This minimum is applied to single-sequence snapshots too, so a short valid prefix capture is disabled whenever a 4096-token checkpoint does not fit. Preserve the single-sequence path by applying this minimum only to the concurrent budget, or derive its minimum from the actual single-sequence capture boundary.</comment>

<file context>
@@ -277,8 +277,10 @@ int qwen4exp_select_chunk(ggml_backend_t backend, const Qwen4ExpWeights & w,
         plan->layer_bytes = (size_t) slots * layer;
         if (plan->required > available) return 0;   // the context fit moves on without searching smaller chunks
     }
+    const size_t min_snapshot = snapshot_budget
+        ? qwen4exp_snapshot_bytes(backend, cache, std::min(cache.max_ctx, 4096)) : 0;
     const int chunk = qwen4exp_fit_chunk(cache.max_ctx, available, fixed, chunk_workspace, snapshot_budget,
</file context>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: min_snapshot undercounts the checkpoint size that snapshot_fits() actually enforces. When the spare memory falls between the raw cache-copy size and the full estimate, this leaves a nonzero allowance but every checkpoint capture fails; calculate the minimum with the same complete checkpoint estimate used by snapshot_fits().

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At server/src/qwen4exp/qwen4exp_backend.cpp, line 280:

<comment>`min_snapshot` undercounts the checkpoint size that `snapshot_fits()` actually enforces. When the spare memory falls between the raw cache-copy size and the full estimate, this leaves a nonzero allowance but every checkpoint capture fails; calculate the minimum with the same complete checkpoint estimate used by `snapshot_fits()`.</comment>

<file context>
@@ -277,8 +277,10 @@ int qwen4exp_select_chunk(ggml_backend_t backend, const Qwen4ExpWeights & w,
         plan->layer_bytes = (size_t) slots * layer;
         if (plan->required > available) return 0;   // the context fit moves on without searching smaller chunks
     }
+    const size_t min_snapshot = snapshot_budget
+        ? qwen4exp_snapshot_bytes(backend, cache, std::min(cache.max_ctx, 4096)) : 0;
     const int chunk = qwen4exp_fit_chunk(cache.max_ctx, available, fixed, chunk_workspace, snapshot_budget,
</file context>

? qwen4exp_snapshot_bytes(
backend, cache,
std::min(cache.max_ctx, restore_point_spacing()))
: 0;
const int reserved_rows = !snapshot_budget ? floor_rows
: qwen4exp_reserved_rows(cache.max_ctx, available, fixed, chunk_workspace, floor_rows,
std::min(kQwen4ExpPreferredChunk, max_rows), std::min(*snapshot_budget, snapshot_keep));
const int chunk = qwen4exp_fit_chunk(cache.max_ctx, available, fixed, chunk_workspace, snapshot_budget,
reserved_rows, max_rows);
reserved_rows, max_rows, min_snapshot);
if (plan && chunk > 0) {
const size_t workspace = chunk_workspace(plan->rows > 0 ? plan->rows : chunk);
plan->chunk_bytes = workspace == SIZE_MAX ? SIZE_MAX
Expand Down Expand Up @@ -427,7 +433,9 @@ bool Qwen4ExpBackend::load_target() {
}
}
if (!load_hot_experts()) return false;
if (snapshot_budget_ != SIZE_MAX)
if (snapshot_budget_ == 0)
std::fprintf(stderr, "[qwen4exp] prefix snapshots disabled: memory allowance insufficient for one checkpoint\n");
else if (snapshot_budget_ != SIZE_MAX)
std::fprintf(stderr, "[qwen4exp] prefix snapshot allowance=%zu bytes\n", snapshot_budget_);
else if (snap_backend_ != backend_)
std::fprintf(stderr, "[qwen4exp] prefix snapshots in system memory\n");
Expand Down Expand Up @@ -501,7 +509,9 @@ int Qwen4ExpBackend::plan_chunk(Qwen4ExpChunkPlan * plan) {
const int slots = cfg_.max_concurrency;
const int fit = qwen4exp_select_chunk(backend_, weights_, cache_, slots, 1,
allowance(slots > 1 ? 2 * (size_t) slots + 1 : 3), plan,
snapshot_bytes_estimate(cache_.max_ctx), pflash_reserve(backend_));
snapshot_bytes_estimate(cache_.max_ctx),
pflash_reserve(backend_),
restore_point_spacing());
return cfg_.chunk > 0 && fit > 0 ? explicit_chunk : fit;
}

Expand Down
14 changes: 10 additions & 4 deletions server/src/qwen4exp/qwen4exp_chunk.h
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ constexpr int kQwen4ExpConcurrentGranule = 2048;
template<class Measure>
int qwen4exp_fit_chunk(int max_ctx, size_t available, size_t fixed, Measure measure,
size_t * snapshot_budget = nullptr, int floor_rows = 4096,
int max_rows = kQwen4ExpMaxChunk) {
int max_rows = kQwen4ExpMaxChunk, size_t min_snapshot = 0) {
const size_t budget = available - available / 10;
floor_rows = std::min(floor_rows, max_rows);
if (snapshot_budget) {
Expand All @@ -46,8 +46,13 @@ int qwen4exp_fit_chunk(int max_ctx, size_t available, size_t fixed, Measure meas
// Snapshots get only the remaining safe budget.
const int floor = qwen4exp_fit_chunk(std::min(max_ctx, floor_rows), available, fixed, measure);
if (!floor) return 0;
*snapshot_budget = std::min(requested, budget - fixed - measure(floor));
fixed += *snapshot_budget;
const size_t needed = fixed + measure(floor);
const size_t spare = budget > needed ? budget - needed : 0;
const size_t allowance = std::min(requested, spare);
if (allowance >= min_snapshot) {
*snapshot_budget = allowance;
fixed += *snapshot_budget;
}
}
if (max_ctx <= 0 || fixed >= budget) return 0;
int best = 0, lo = 2, hi = std::min(max_ctx, max_rows) / 256;
Expand Down Expand Up @@ -122,6 +127,7 @@ struct Qwen4ExpCache;
// `reserved` is device memory kept free for something else (a PFlash drafter).
int qwen4exp_select_chunk(ggml_backend_t backend, const Qwen4ExpWeights & w,
Qwen4ExpCache & cache, int slots = 1, int resident_slots = 1, size_t * snapshot_budget = nullptr,
Qwen4ExpChunkPlan * plan = nullptr, size_t snapshot_keep = 0, size_t reserved = 0);
Qwen4ExpChunkPlan * plan = nullptr, size_t snapshot_keep = 0, size_t reserved = 0,
int restore_point_spacing = 0);

} // namespace luce::common
6 changes: 4 additions & 2 deletions server/src/server/http_server.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1819,7 +1819,7 @@ static json model_list(const ServerConfig & config, bool codex_schema) {
// Memory the auto budget takes a quarter of: what is available once the
// model has loaded (a container's cgroup v2 limit included; Windows too),
// else total physical memory where the probe cannot read it.
static size_t budgetable_memory_bytes() {
size_t budgetable_memory_bytes() {
if (const auto available = available_kv_offload_memory()) return *available;
#if !defined(_WIN32)
const long pages = sysconf(_SC_PHYS_PAGES);
Expand All @@ -1841,7 +1841,8 @@ size_t auto_concurrent_prefix_budget(size_t per_checkpoint, int slots, size_t me
}

size_t HttpServer::concurrent_prefix_budget(size_t per_checkpoint, int slots) const {
return auto_concurrent_prefix_budget(per_checkpoint, slots, budgetable_memory_bytes());
const size_t auto_budget = auto_concurrent_prefix_budget(per_checkpoint, slots, budgetable_memory_bytes());
return std::min(auto_budget, backend_.snapshot_allowance_bytes());
}

PrefixCacheBudget resolve_prefix_cache_budget(const ServerConfig & config,
Expand All @@ -1857,6 +1858,7 @@ PrefixCacheBudget resolve_prefix_cache_budget(const ServerConfig & config,
// (concurrent_prefix_budget below).
out.bytes = out.automatic ? ServerConfig::kConcurrentPrefixBudgetFloor
: config.concurrent_prefix_cache_max_bytes;
out.bytes = std::min(out.bytes, backend.snapshot_allowance_bytes());
return out;
}
out.automatic =
Expand Down
2 changes: 2 additions & 0 deletions server/src/server/http_server.h
Original file line number Diff line number Diff line change
Expand Up @@ -673,6 +673,8 @@ struct PrefixCacheBudget {
// and, above that, at most memory / 4 when memory is known (non-zero).
size_t auto_concurrent_prefix_budget(size_t per_checkpoint, int slots, size_t memory);

size_t budgetable_memory_bytes();

PrefixCacheBudget resolve_prefix_cache_budget(const ServerConfig & config,
const ModelBackend & backend);

Expand Down
11 changes: 6 additions & 5 deletions server/src/server/kv/prefix_cache.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -131,10 +131,10 @@ PrefixCache::PrefixCache(int cap, const Tokenizer & tokenizer,
disabled_ = false;
if (max_resident_bytes_ > 0) {
std::fprintf(stderr, "[pc] enabled: cap=%d family=%s resident_budget=%zu MiB\n",
cap_, markers_.family.c_str(), max_resident_bytes_ / (1024 * 1024));
cap_.load(std::memory_order_relaxed), markers_.family.c_str(), max_resident_bytes_ / (1024 * 1024));
} else {
std::fprintf(stderr, "[pc] enabled: cap=%d family=%s\n",
cap_, markers_.family.c_str());
cap_.load(std::memory_order_relaxed), markers_.family.c_str());
}
}

Expand Down Expand Up @@ -527,7 +527,8 @@ PrefixCache::InlineReservation PrefixCache::reserve_inline_snap(
std::fprintf(stderr,
"[pc] inline snap skipped, %s: entries=%zu protected=%d "
"in_flight=%zu cap=%d restore_slot=%d cut=%d\n",
reason, entries_.size(), pins, active_reservations_.size(), cap_,
reason, entries_.size(), pins, active_reservations_.size(),
cap_.load(std::memory_order_relaxed),
restore_source_slot, target_cut);
};
if ((int)entries_.size() + reserved_free_slots >= cap_) {
Expand Down Expand Up @@ -1226,8 +1227,8 @@ void PrefixCache::abort_full_snap(int slot) {

PrefixCache::InlineStats PrefixCache::stats() const {
InlineStats out{};
if (disabled_) return out;
out.capacity = cap_;
if (disabled_.load(std::memory_order_acquire)) return out;
out.capacity = cap_.load(std::memory_order_relaxed);
out.in_use =
(int)entries_size_count_.load(std::memory_order_relaxed);
out.lifetime_hits = lifetime_hits_.load(std::memory_order_relaxed);
Expand Down
10 changes: 7 additions & 3 deletions server/src/server/kv/prefix_cache.h
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,11 @@ class PrefixCache {
PrefixCache(int cap, const Tokenizer & tokenizer,
size_t max_resident_bytes = 0);

bool disabled() const { return disabled_; }
bool disabled() const { return disabled_.load(std::memory_order_acquire); }
void disable() {
cap_.store(0, std::memory_order_relaxed);
disabled_.store(true, std::memory_order_release);
}

// Expose chat markers for cold prefix boundary detection.
const ChatMarkers & chat_markers() const { return markers_; }
Expand Down Expand Up @@ -329,8 +333,8 @@ class PrefixCache {
FullStats full_stats() const;

private:
bool disabled_ = true;
int cap_ = 0;
std::atomic<bool> disabled_{true};
std::atomic<int> cap_{0};
ChatMarkers markers_;

// LRU for inline prefix cache: ordered map of hash → slot.
Expand Down
55 changes: 46 additions & 9 deletions server/src/server/scheduler.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -102,16 +102,53 @@ void HttpServer::scheduler_loop(SeqEngine & engine) {
// snapshot slots go through checkpoints_.
EngineCheckpoints engine_checkpoints(engine, checkpoints_);
const int n_slots = engine.slot_count();
if (config_.concurrent_prefix_cache &&
config_.concurrent_prefix_cache_max_bytes == ServerConfig::kPrefixCacheBudgetAuto) {
const size_t per = engine_checkpoints.estimate_bytes(engine.max_context());
if (per > 0) {
const size_t bytes = concurrent_prefix_budget(per, n_slots);
prefix_cache_.set_max_resident_bytes(bytes);
if (config_.concurrent_prefix_cache) {
if (!engine_checkpoints.supported()) {
prefix_cache_.disable();
std::fprintf(stderr,
"[pc] concurrent prefix budget (auto): %zu MiB = %d slots, %zu MiB per "
"checkpoint at %d tokens\n",
bytes / (1024 * 1024), n_slots, per / (1024 * 1024), engine.max_context());
"[pc] concurrent prefix cache disabled: engine does not support prefix store\n");
} else if (backend_.snapshot_allowance_bytes() == 0) {
prefix_cache_.disable();
std::fprintf(stderr,
"[pc] concurrent prefix cache disabled: backend snapshot allowance is 0 bytes\n");
} else if (config_.concurrent_prefix_cache_max_bytes == ServerConfig::kPrefixCacheBudgetAuto) {
const size_t per = engine_checkpoints.estimate_bytes(engine.max_context());
if (per > 0) {
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Resolved
const size_t bytes = concurrent_prefix_budget(per, n_slots);
if (bytes == 0) {
prefix_cache_.disable();
std::fprintf(stderr,
"[pc] concurrent prefix cache disabled: concurrent prefix budget is 0 bytes\n");
} else {
prefix_cache_.set_max_resident_bytes(bytes);
std::fprintf(stderr,
"[pc] concurrent prefix budget (auto): %zu MiB = %d slots, %zu MiB per "
"checkpoint at %d tokens\n",
bytes / (1024 * 1024), n_slots, per / (1024 * 1024), engine.max_context());
}
} else {
prefix_cache_.disable();
std::fprintf(stderr,
"[pc] concurrent prefix cache disabled: estimated checkpoint size is 0 bytes\n");
}
} else {
const size_t configured = config_.concurrent_prefix_cache_max_bytes;
const size_t allowance = backend_.snapshot_allowance_bytes();
const size_t bytes = std::min(configured, allowance);
if (bytes == 0) {
prefix_cache_.disable();
std::fprintf(stderr,
"[pc] concurrent prefix cache disabled: %s\n",
allowance == 0 ? "backend snapshot allowance is 0 bytes"
: "configured budget is 0 bytes");
} else {
if (bytes < configured) {
std::fprintf(stderr,
"[pc] concurrent prefix budget clamped to backend allowance: %zu MiB (configured %zu MiB)\n",
bytes / (1024 * 1024), configured / (1024 * 1024));
}
prefix_cache_.set_max_resident_bytes(bytes);
}
}
}
std::vector<SchedSlot> slots((size_t)n_slots);
Expand Down
32 changes: 32 additions & 0 deletions server/test/unit/test_qwen4exp_chunk.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -69,4 +69,36 @@ int main() {
const int rows = qwen4exp_reserved_rows(10000, 100000, 20000, workspace, 4096, 8192, 10000);
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, rows) == 5888);
assert(snapshots == 90000 - 20000 - 58880 && snapshots >= 10000);

// An allowance smaller than min_snapshot is rejected so prefill chunks can use the spare memory.
snapshots = 100000;
// Insufficient positive allowance: spare (29040) < min_snapshot (30000) -> rejected to 0, chunk widens to 6912.
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 30000) == 6912);
assert(snapshots == 0);
// Rejected snapshot reservation matches no-snapshot chunk sizing exactly.
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 30000) ==
qwen4exp_fit_chunk(10000, 100000, 20000, workspace));
// Exactly the minimum viable allowance: spare == min_snapshot -> accepted.
snapshots = 100000;
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 29040) == 4096);
assert(snapshots == 29040);
// Sufficient allowance: spare (29040) > min_snapshot (10000) -> accepted.
snapshots = 100000;
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 10000) == 4096);
assert(snapshots == 29040);

// Requested allowance smaller than min_snapshot: spare (29040) > min_snapshot (10000), but requested (5000) < min_snapshot (10000).
// Resulting allowance must be zero and rejected reservation must not reduce prefill chunk capacity (fits 6912, matches no-snapshot fit).
snapshots = 5000;
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 10000) == 6912);
assert(snapshots == 0);
assert(qwen4exp_fit_chunk(10000, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 10000) ==
qwen4exp_fit_chunk(10000, 100000, 20000, workspace));
// Short-context boundary case (max_ctx < 4096):
snapshots = 100000;
assert(qwen4exp_fit_chunk(2048, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 50000) == 2048);
assert(snapshots == 0);
snapshots = 100000;
assert(qwen4exp_fit_chunk(2048, 100000, 20000, workspace, &snapshots, 4096, luce::common::kQwen4ExpMaxChunk, 49520) == 2048);
assert(snapshots == 49520);
}
Loading
Loading