ferrox_core/cache.rs
1//! A per-layer KV cache, growable one position at a time during decode.
2//! Two growth strategies exist:
3//!
4//! - `with_pool`: PagedAttention-style block allocation. Many caches
5//! (one per concurrent request, typically) draw fixed-size blocks
6//! from one shared, bounded `KvBlockPool` instead of each
7//! independently pre-committing to a worst-case context length.
8//! Growth happens in fixed block-sized quanta, and a cache's blocks
9//! return to the shared pool when it's dropped, so the pool's free
10//! count is a real, live admission-control signal a caller can check
11//! before accepting a new request. This is the block-*allocation*
12//! half of PagedAttention; it does not (yet) change how attention
13//! reads a cache -- `k`/`v` are still read as one contiguous slice
14//! per sequence (see `Decoder::forward_token`/`forward_batch`), just
15//! backed by capacity that grows in block-sized steps instead of
16//! Rust's default exponential `Vec` growth. Wiring this into
17//! `ferrox-server` as live per-request admission control via
18//! `FERROX_KV_POOL_BLOCKS`/`FERROX_KV_POOL_BLOCK_SIZE`.
19
20use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
21
22use crate::kv_swa::KvWindow;
23
24/// Returned by `KvCache::push` (and `with_pool`) when a pool-backed
25/// cache needs another block but its shared `KvBlockPool` has none
26/// free. Caches built with `new`/`with_capacity` never return this --
27/// their growth is unconditional, matching their pre-paging behavior
28/// exactly.
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub struct KvPoolExhausted;
31
32impl std::fmt::Display for KvPoolExhausted {
33 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34 write!(f, "KV cache block pool exhausted: no free blocks remain")
35 }
36}
37
38impl std::error::Error for KvPoolExhausted {}
39
40/// A bounded pool of fixed-size KV-cache blocks (in positions) shared
41/// across many `KvCache` instances, typically one pool per server
42/// process. Each `KvCache::with_pool` acquires one block up front and
43/// one more each time it grows past its currently held capacity;
44/// `free_blocks` is therefore a live, accurate admission-control
45/// signal -- a caller can check it before accepting a new request
46/// rather than discovering exhaustion only after committing memory.
47pub struct KvBlockPool {
48 block_size: usize,
49 total_blocks: usize,
50 free_blocks: usize,
51}
52
53impl KvBlockPool {
54 /// `block_size` positions per block, `total_blocks` blocks in the
55 /// whole shared budget (so `block_size * total_blocks` positions
56 /// total, across however many caches draw from this pool at once).
57 pub fn new(block_size: usize, total_blocks: usize) -> Self {
58 assert!(block_size > 0, "block_size must be positive");
59 KvBlockPool {
60 block_size,
61 total_blocks,
62 free_blocks: total_blocks,
63 }
64 }
65
66 pub fn block_size(&self) -> usize {
67 self.block_size
68 }
69
70 pub fn total_blocks(&self) -> usize {
71 self.total_blocks
72 }
73
74 pub fn free_blocks(&self) -> usize {
75 self.free_blocks
76 }
77
78 /// Re-budget the pool.
79 ///
80 /// The pool is an *accounting* budget, not an allocator: each
81 /// `KvCache` owns its own buffer and this counts how many blocks
82 /// the deployment has promised. So a resize is arithmetic, with one
83 /// rule that is not.
84 ///
85 /// Shrinking below what is currently held is REFUSED and the pool
86 /// is left exactly as it was. `free_blocks` would have to go
87 /// negative to represent it, and the alternative -- clamping it to
88 /// zero -- silently over-promises: the caches already holding those
89 /// blocks do not give them back, so every later `try_acquire`
90 /// would be deciding against a budget that does not describe the
91 /// memory in use.
92 ///
93 /// Returns the number of blocks currently held when it refuses, so
94 /// the caller can say what the floor actually is rather than making
95 /// the operator find it by being rejected.
96 pub fn resize(&mut self, total_blocks: usize) -> Result<(), usize> {
97 let in_use = self.total_blocks - self.free_blocks;
98 if total_blocks < in_use {
99 return Err(in_use);
100 }
101 self.free_blocks = total_blocks - in_use;
102 self.total_blocks = total_blocks;
103 Ok(())
104 }
105
106 fn try_acquire(&mut self, n: usize) -> bool {
107 if n <= self.free_blocks {
108 self.free_blocks -= n;
109 true
110 } else {
111 false
112 }
113 }
114
115 fn release(&mut self, n: usize) {
116 self.free_blocks = (self.free_blocks + n).min(self.total_blocks);
117 }
118}
119
120struct PooledState {
121 pool: Arc<Mutex<KvBlockPool>>,
122 block_size: usize,
123 blocks_held: usize,
124}
125
126pub struct KvCache {
127 pub n_kv_heads: usize,
128 pub head_dim: usize,
129 pub k: Vec<f32>, // [rows, n_kv_heads, head_dim], flattened
130 pub v: Vec<f32>,
131 /// Positions this sequence has consumed.
132 ///
133 /// **Not the same thing as the number of rows in `k`/`v`**, and the
134 /// distinction is the whole reason this field is private. They are
135 /// equal today because nothing evicts, and they stop being equal
136 /// the moment a windowed layer drops a position behind its window
137 /// (#61): `positions` keeps counting, `rows` does not.
138 ///
139 /// Every reader has to say which one it meant, so there is no
140 /// `seq_len` any more. [`Self::positions`] is what RoPE, a resume
141 /// point and a truncate target mean; [`Self::rows`] is what
142 /// attention iterates and what the bytes cost.
143 ///
144 /// Two bugs have already been caused by the two being one field.
145 /// `PrefillState` read the KV's length as the position to resume at
146 /// (#37), and `DraftModelSpeculator::sync` trusted a counter beside
147 /// a cache that a device-resident backend leaves empty. Both were
148 /// right to read the store rather than keep a copy; both would be
149 /// wrong the day the store evicts.
150 positions: usize,
151 /// The capacity (in positions) this cache was pre-allocated for,
152 /// if any. `None` for caches built with `new` or `with_pool`.
153 planned_capacity: Option<usize>,
154 /// `Some` for caches built with `with_pool`; tracks the shared
155 /// pool and how many blocks this cache currently holds, so its
156 /// blocks can be returned on drop.
157 pool_state: Option<PooledState>,
158 /// `Some` once a windowed layer's cache has been told it may drop
159 /// rows behind its window (#61). `None` -- the default, and what
160 /// every constructor produces -- means this cache keeps every
161 /// position it was ever pushed, which is what every store in this
162 /// engine did before.
163 ///
164 /// Armed by the decoder, never by a constructor, because the fact
165 /// that a layer is windowed lives in `ModelConfig` and the decision
166 /// that eviction is *safe for this run* lives in
167 /// `ferrox_models::decoder::kv_window`. See [`Self::arm_window`].
168 window: Option<KvWindow>,
169}
170
171/// Cloning a pool-backed cache detaches the clone from pool accounting
172/// (its `k`/`v`/`seq_len` data is copied normally, but the clone does
173/// not hold or later release any blocks itself) -- mirroring how
174/// `ferrox-models::prefix_cache` already uses `KvCache::clone` to fork
175/// a cached prefix into a new, independent request's cache. Only the
176/// original cache's blocks are released, exactly once, when it drops.
177impl Clone for KvCache {
178 fn clone(&self) -> Self {
179 KvCache {
180 n_kv_heads: self.n_kv_heads,
181 head_dim: self.head_dim,
182 k: self.k.clone(),
183 v: self.v.clone(),
184 positions: self.positions,
185 planned_capacity: self.planned_capacity,
186 pool_state: None,
187 // Carried, not reset: a clone of a cache that has already
188 // dropped rows is a cache that has already dropped rows,
189 // and pretending otherwise would let the clone's
190 // `positions` be read as a row count again.
191 window: self.window,
192 }
193 }
194}
195
196impl Drop for KvCache {
197 fn drop(&mut self) {
198 if let Some(state) = &self.pool_state {
199 if let Ok(mut pool) = state.pool.lock() {
200 pool.release(state.blocks_held);
201 }
202 }
203 }
204}
205
206impl KvCache {
207 /// Positions this sequence has consumed: what RoPE means, what a
208 /// resume point means, and what a truncate target is measured in.
209 ///
210 /// Monotonic except through [`Self::truncate`] and [`Self::clear`].
211 /// Equal to [`Self::rows`] today, and deliberately a different
212 /// method so it stops being equal safely (#61).
213 #[inline]
214 pub fn positions(&self) -> usize {
215 self.positions
216 }
217
218 /// Sets the position counter directly, for a constructor that
219 /// filled `k`/`v` by hand rather than through `push`.
220 ///
221 /// Deliberately narrow and deliberately not `pub`: the only honest
222 /// caller is one that has just written exactly this many rows, and
223 /// a public setter on a counter the buffer should imply is how the
224 /// two drift apart again.
225 pub(crate) fn set_positions(&mut self, positions: usize) {
226 debug_assert_eq!(
227 positions,
228 self.rows(),
229 "set_positions must agree with the rows just written"
230 );
231 self.positions = positions;
232 }
233
234 /// Test-only: sets the position counter WITHOUT the agreement check
235 /// [`Self::set_positions`] makes.
236 ///
237 /// Exists for one caller: `kv_signature`'s test that a serialized
238 /// payload whose declared count contradicts its buffers is
239 /// rejected. That contradiction is the thing under test, so it has
240 /// to be constructible, and it must not be constructible anywhere
241 /// else.
242 #[cfg(test)]
243 pub(crate) fn force_positions_for_test(&mut self, positions: usize) {
244 self.positions = positions;
245 }
246
247 /// Rows of K/V actually resident: what attention iterates over and
248 /// what the memory costs.
249 ///
250 /// Derived from the buffer rather than counted alongside it, so it
251 /// cannot drift from what is really there. That is the same rule
252 /// the batched prefill learned in #37: read the cursor, do not keep
253 /// a copy of it.
254 #[inline]
255 pub fn rows(&self) -> usize {
256 let elems_per_position = self.n_kv_heads * self.head_dim;
257 if elems_per_position == 0 {
258 return 0;
259 }
260 self.k.len() / elems_per_position
261 }
262
263 pub fn new(n_kv_heads: usize, head_dim: usize) -> Self {
264 KvCache {
265 n_kv_heads,
266 head_dim,
267 k: Vec::new(),
268 v: Vec::new(),
269 positions: 0,
270 planned_capacity: None,
271 pool_state: None,
272 window: None,
273 }
274 }
275
276 /// Pre-allocates storage for up to `max_seq_len` positions, so
277 /// `push` never triggers a reallocation-and-copy during decode.
278 /// Use this when the maximum context length is known ahead of time
279 pub fn with_capacity(n_kv_heads: usize, head_dim: usize, max_seq_len: usize) -> Self {
280 let elems_per_position = n_kv_heads * head_dim;
281 KvCache {
282 n_kv_heads,
283 head_dim,
284 k: Vec::with_capacity(max_seq_len * elems_per_position),
285 v: Vec::with_capacity(max_seq_len * elems_per_position),
286 positions: 0,
287 planned_capacity: Some(max_seq_len),
288 pool_state: None,
289 window: None,
290 }
291 }
292
293 /// Acquires up front however many blocks from `pool` are needed to
294 /// cover `max_seq_len` positions (at least one, even if
295 /// `max_seq_len` is `0`), so a caller that knows its worst-case
296 /// sequence length ahead of time (as `ferrox-server` does: prompt
297 /// length + `max_tokens`) never needs to acquire another block
298 /// mid-decode. This matters beyond performance: `push` growing past
299 /// its currently held capacity can fail if the pool is exhausted by
300 /// *other* requests by then, and callers like
301 /// `ferrox_models::Decoder::forward_token` treat `push` as
302 /// infallible for non-pooled caches -- a pooled cache that
303 /// under-reserves at construction and then fails to grow later
304 /// would violate that assumption and panic mid-decode. Sizing to
305 /// `max_seq_len` up front turns that into an admission-control
306 /// decision made once, honestly, before any generation work starts,
307 /// exactly mirroring `with_capacity`'s worst-case pre-allocation --
308 /// just drawn from a shared pool instead of a private allocation.
309 /// Returns `Err(KvPoolExhausted)` without mutating anything if the
310 /// pool doesn't have that many blocks free.
311 pub fn with_pool(
312 n_kv_heads: usize,
313 head_dim: usize,
314 pool: Arc<Mutex<KvBlockPool>>,
315 max_seq_len: usize,
316 ) -> Result<Self, KvPoolExhausted> {
317 let block_size = pool.lock().unwrap().block_size();
318 let blocks_needed = max_seq_len.div_ceil(block_size).max(1);
319 if !pool.lock().unwrap().try_acquire(blocks_needed) {
320 return Err(KvPoolExhausted);
321 }
322 let elems_per_position = n_kv_heads * head_dim;
323 Ok(KvCache {
324 n_kv_heads,
325 head_dim,
326 k: Vec::with_capacity(blocks_needed * block_size * elems_per_position),
327 v: Vec::with_capacity(blocks_needed * block_size * elems_per_position),
328 positions: 0,
329 planned_capacity: None,
330 pool_state: Some(PooledState {
331 pool,
332 block_size,
333 blocks_held: blocks_needed,
334 }),
335 window: None,
336 })
337 }
338
339 /// Appends one position's key/value vectors (each
340 /// `n_kv_heads * head_dim` long) to the cache. For pool-backed
341 /// caches, this may need to acquire another block first; if the
342 /// shared pool has none free, no data is appended and
343 /// `Err(KvPoolExhausted)` is returned. Caches built with `new` or
344 /// `with_capacity` always return `Ok`.
345 pub fn push(&mut self, k_step: &[f32], v_step: &[f32]) -> Result<(), KvPoolExhausted> {
346 assert_eq!(k_step.len(), self.n_kv_heads * self.head_dim);
347 assert_eq!(v_step.len(), self.n_kv_heads * self.head_dim);
348
349 let elems_per_position = self.n_kv_heads * self.head_dim;
350 if let Some(state) = &mut self.pool_state {
351 let capacity_positions = self.k.capacity() / elems_per_position;
352 // ROWS, not positions: this asks whether the buffer is
353 // full, and an evicting cache's buffer is shorter than its
354 // position count. Equal for a cache that never evicts.
355 let rows = self.k.len() / elems_per_position;
356 if rows == capacity_positions {
357 if !state.pool.lock().unwrap().try_acquire(1) {
358 return Err(KvPoolExhausted);
359 }
360 state.blocks_held += 1;
361 self.k.reserve_exact(state.block_size * elems_per_position);
362 self.v.reserve_exact(state.block_size * elems_per_position);
363 }
364 }
365
366 self.k.extend_from_slice(k_step);
367 self.v.extend_from_slice(v_step);
368 self.positions += 1;
369 Ok(())
370 }
371
372 /// Advance length by `n` positions without storing real K/V values
373 /// (zero-fill). Used when Metal owns the KV plane and the host cache
374 /// only needs matching `seq_len` for sync checks.
375 pub fn advance_len(&mut self, n: usize) -> Result<(), KvPoolExhausted> {
376 if n == 0 {
377 return Ok(());
378 }
379 let elems_per_position = self.n_kv_heads * self.head_dim;
380 let zeros = vec![0f32; elems_per_position];
381 for _ in 0..n {
382 self.push(&zeros, &zeros)?;
383 }
384 Ok(())
385 }
386
387 /// Returns this cache's blocks to its shared pool immediately
388 /// (rather than waiting for `Drop`) and detaches it from pool
389 /// accounting; a no-op for caches that aren't pool-backed, and
390 /// idempotent if called more than once.
391 pub fn release_to_pool(&mut self) {
392 if let Some(state) = self.pool_state.take() {
393 if let Ok(mut pool) = state.pool.lock() {
394 pool.release(state.blocks_held);
395 }
396 }
397 }
398
399 pub fn clear(&mut self) {
400 self.k.clear();
401 self.v.clear();
402 self.positions = 0;
403 }
404
405 /// Rolls the cache back to exactly `new_seq_len` positions,
406 /// discarding everything after. Used to reject speculatively
407 /// decoded draft tokens that turned out wrong: their K/V were
408 /// already pushed during batched verification, and rejection means
409 /// removing them so the next real decode step continues from the
410 /// last *accepted* position, not the last *attempted* one.
411 /// Rolls the cache back to exactly `new_seq_len` POSITIONS.
412 ///
413 /// A windowed cache can only roll back into rows it still holds. A
414 /// target further back than [`Self::rows`] names a position this
415 /// cache dropped behind its window, and there is no honest thing to
416 /// return for it -- so it stops, rather than silently rolling back
417 /// to the oldest row it happens to have and answering the next
418 /// token out of a history with a hole in it.
419 ///
420 /// Unreachable for a cache that has not been armed by
421 /// [`Self::arm_window`], which is every cache unless
422 /// `FERROX_KV_WINDOW` is on: `rows == positions` there, so the
423 /// second precondition is implied by the first.
424 pub fn truncate(&mut self, new_seq_len: usize) {
425 assert!(
426 new_seq_len <= self.positions,
427 "truncate target {new_seq_len} must not exceed current seq_len {}",
428 self.positions
429 );
430 let rows = self.rows();
431 let dropped = self.positions - new_seq_len;
432 assert!(
433 dropped <= rows,
434 "truncate target {new_seq_len} is {dropped} positions back but only {rows} rows \
435 are resident: this cache evicted behind a {:?} window (#61). Turn off \
436 FERROX_KV_WINDOW for a workload that rolls the KV cache back this far.",
437 self.window.map(|w| w.window())
438 );
439 let elems_per_position = self.n_kv_heads * self.head_dim;
440 let keep_rows = rows - dropped;
441 self.k.truncate(keep_rows * elems_per_position);
442 self.v.truncate(keep_rows * elems_per_position);
443 self.positions = new_seq_len;
444 }
445
446 /// Tells this cache it may drop rows that have fallen behind
447 /// `window` (#61 step 2).
448 ///
449 /// Arming alone drops nothing: [`Self::evict_behind_window`] is what
450 /// drops, and the holder calls it at a point where it knows nothing
451 /// is mid-read. That split is deliberate. `push` is the obvious
452 /// place to evict and it is the wrong one: `Decoder::forward_batch`
453 /// writes a whole prefill batch into the cache and only then reads
454 /// it back, against a row offset it captured BEFORE the writes.
455 /// Evicting inside `push` would move every row out from under that
456 /// offset, and the prompt would be attended over shifted keys. So
457 /// eviction is something the holder asks for, and asking for it too
458 /// rarely only costs memory -- over-retention is always correct,
459 /// under-retention is wrong logits.
460 ///
461 /// Idempotent; a second call with a different window replaces the
462 /// first, and rows already dropped stay dropped.
463 pub fn arm_window(&mut self, window: KvWindow) {
464 self.window = Some(window);
465 }
466
467 /// The window this cache evicts behind, or `None` if it keeps
468 /// everything.
469 pub fn window(&self) -> Option<KvWindow> {
470 self.window
471 }
472
473 /// Drops rows that have fallen behind the armed window, returning
474 /// how many rows went. Zero, always, for an unarmed cache.
475 ///
476 /// The rows dropped are the OLDEST ones, so what remains is still a
477 /// contiguous suffix of the sequence: row `i` of `rows` holds
478 /// absolute position `positions - rows + i`. Every windowed
479 /// attention kernel reads the last `window` rows and nothing else,
480 /// and [`KvWindow::rows_after`] guarantees at least that many
481 /// survive, so the set of rows a kernel reads is byte-for-byte the
482 /// set it would have read with no eviction at all.
483 pub fn evict_behind_window(&mut self) -> usize {
484 let Some(window) = self.window else {
485 return 0;
486 };
487 let elems_per_position = self.n_kv_heads * self.head_dim;
488 if elems_per_position == 0 {
489 return 0;
490 }
491 let rows = self.k.len() / elems_per_position;
492 let keep = window.rows_after(self.positions).min(rows);
493 let drop_rows = rows - keep;
494 if drop_rows == 0 {
495 return 0;
496 }
497 // One `drain` per eviction, not one per token: the whole reason
498 // `KvWindow` carries slack. This moves `keep` rows down, and it
499 // happens once every `slack + 1` positions.
500 let drop_elems = drop_rows * elems_per_position;
501 self.k.drain(..drop_elems);
502 self.v.drain(..drop_elems);
503 // `drain` frees no memory, and the saving this exists for is
504 // memory. A cache that just absorbed a 32k-token prefill holds
505 // 32k rows of capacity behind `window + slack` rows of data
506 // until something hands it back. Only when the excess is large
507 // enough to be worth a realloc-and-copy: shrinking on every
508 // eviction would trade the block drain for a full copy.
509 //
510 // **Never for a pool-backed cache**, and the reason is a bug
511 // rather than a preference. `push` asks whether the buffer is
512 // full by comparing rows against `k.capacity()`, and takes
513 // another block from the shared pool when they meet. Shrinking
514 // the buffer to a window's worth would make that condition true
515 // every `slack + 1` tokens forever, so a windowed pooled cache
516 // would draw a fresh block from the pool on a cadence, hold
517 // every one of them until it drops, and exhaust the pool
518 // mid-answer -- where `push` is documented infallible and the
519 // caller panics. The pool has already promised this cache its
520 // blocks; handing the capacity back without handing the blocks
521 // back saves nothing and costs that.
522 if self.pool_state.is_none() {
523 let want = window.max_rows() * elems_per_position;
524 if self.k.capacity() > want.saturating_mul(2) {
525 self.k.shrink_to(want);
526 self.v.shrink_to(want);
527 }
528 }
529 drop_rows
530 }
531
532 /// Bytes currently resident for this cache's K and V buffers
533 /// combined (actual allocated capacity, not just used length) --
534 /// the number that matters for "does this fit in the context
535 /// budget,"
536 pub fn allocated_bytes(&self) -> usize {
537 (self.k.capacity() + self.v.capacity()) * std::mem::size_of::<f32>()
538 }
539
540 /// True if this cache was pre-allocated via `with_capacity` and
541 /// has not yet grown past that planned capacity (i.e. `push` has
542 /// never had to reallocate). Useful for tests/diagnostics
543 /// confirming the pre-allocation path actually avoided reallocs.
544 pub fn is_within_planned_capacity(&self) -> bool {
545 match self.planned_capacity {
546 Some(cap) => {
547 // POSITIONS against the plan (the plan was made in
548 // positions), ROWS against the buffer (the buffer holds
549 // rows). Equal unless this cache evicts.
550 self.positions <= cap
551 && self.k.capacity() >= self.rows() * self.n_kv_heads * self.head_dim
552 }
553 None => false,
554 }
555 }
556}
557
558/// The other half of PagedAttention that `KvBlockPool`/`KvCache::with_pool`
559/// deliberately don't implement (see this module's doc comment): real,
560/// *shared* physical block storage that many sequences' block tables can
561/// address into, instead of each `KvCache` still owning its own private,
562/// contiguous `Vec`. `KvBlockPool` only ever bounds a *count* of blocks
563/// each cache may grow to; `PagedKvStore` is the actual backing memory,
564/// and a sequence's `PagedKvCache` holds a block table (an ordered list
565/// of block IDs into this shared store) instead of owning K/V data
566/// directly. This is what makes non-contiguous-block reads during
567/// attention (`causal_gqa_attention_paged`, in `attention.rs`) possible
568/// at all -- `causal_gqa_attention`'s existing contiguous-slice read
569/// pattern has no way to express "position 37 lives in block 12, cached
570/// out of order relative to block 5."
571pub struct PagedKvStore {
572 block_size: usize,
573 n_kv_heads: usize,
574 head_dim: usize,
575 k: Vec<f32>, // [total_blocks * block_size, n_kv_heads, head_dim], flattened
576 v: Vec<f32>,
577 free_block_ids: Vec<usize>,
578}
579
580impl PagedKvStore {
581 pub fn new(block_size: usize, total_blocks: usize, n_kv_heads: usize, head_dim: usize) -> Self {
582 assert!(block_size > 0, "block_size must be positive");
583 let elems_per_block = block_size * n_kv_heads * head_dim;
584 PagedKvStore {
585 block_size,
586 n_kv_heads,
587 head_dim,
588 k: vec![0.0; total_blocks * elems_per_block],
589 v: vec![0.0; total_blocks * elems_per_block],
590 // Pushed in descending order so `pop()` hands out ascending
591 // block IDs -- not load-bearing for correctness (any free ID
592 // works), just makes manual debugging/inspection saner.
593 free_block_ids: (0..total_blocks).rev().collect(),
594 }
595 }
596
597 pub fn block_size(&self) -> usize {
598 self.block_size
599 }
600
601 pub fn free_block_count(&self) -> usize {
602 self.free_block_ids.len()
603 }
604
605 pub fn n_kv_heads(&self) -> usize {
606 self.n_kv_heads
607 }
608
609 pub fn head_dim(&self) -> usize {
610 self.head_dim
611 }
612
613 fn acquire_block(&mut self) -> Option<usize> {
614 self.free_block_ids.pop()
615 }
616
617 fn release_block(&mut self, id: usize) {
618 self.free_block_ids.push(id);
619 }
620
621 fn elems_per_block(&self) -> usize {
622 self.block_size * self.n_kv_heads * self.head_dim
623 }
624
625 /// One position's K (or V) row within block `id` at `offset` (0-based
626 /// within the block) -- `[n_kv_heads * head_dim]` long. Used by
627 /// `causal_gqa_attention_paged` to read attention inputs directly out
628 /// of shared physical storage via a block table, and by
629 /// `PagedKvCache::push` to write a new position into it.
630 pub fn k_row(&self, id: usize, offset: usize) -> &[f32] {
631 let elems_per_position = self.n_kv_heads * self.head_dim;
632 let start = id * self.elems_per_block() + offset * elems_per_position;
633 &self.k[start..start + elems_per_position]
634 }
635
636 pub fn v_row(&self, id: usize, offset: usize) -> &[f32] {
637 let elems_per_position = self.n_kv_heads * self.head_dim;
638 let start = id * self.elems_per_block() + offset * elems_per_position;
639 &self.v[start..start + elems_per_position]
640 }
641
642 fn k_row_mut(&mut self, id: usize, offset: usize) -> &mut [f32] {
643 let elems_per_position = self.n_kv_heads * self.head_dim;
644 let start = id * self.elems_per_block() + offset * elems_per_position;
645 &mut self.k[start..start + elems_per_position]
646 }
647
648 fn v_row_mut(&mut self, id: usize, offset: usize) -> &mut [f32] {
649 let elems_per_position = self.n_kv_heads * self.head_dim;
650 let start = id * self.elems_per_block() + offset * elems_per_position;
651 &mut self.v[start..start + elems_per_position]
652 }
653}
654
655/// Returned when a `PagedKvCache` needs another block but its
656/// `PagedKvStore` has none free -- the paged-storage analog of
657/// `KvPoolExhausted`.
658#[derive(Debug, Clone, Copy, PartialEq, Eq)]
659pub struct PagedStoreExhausted;
660
661impl std::fmt::Display for PagedStoreExhausted {
662 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
663 write!(f, "paged KV store exhausted: no free blocks remain")
664 }
665}
666
667impl std::error::Error for PagedStoreExhausted {}
668
669/// One sequence's view into a shared `PagedKvStore`: a block table
670/// (which physical blocks this sequence's positions live in, in order)
671/// plus how many positions have been written so far. Unlike `KvCache`,
672/// this holds no K/V data itself -- every read and write goes through
673/// the shared store.
674#[derive(Debug, Clone, Default)]
675pub struct PagedKvCache {
676 block_table: Vec<usize>,
677 seq_len: usize,
678}
679
680impl PagedKvCache {
681 pub fn new() -> Self {
682 PagedKvCache {
683 block_table: Vec::new(),
684 seq_len: 0,
685 }
686 }
687
688 pub fn seq_len(&self) -> usize {
689 self.seq_len
690 }
691
692 pub fn block_table(&self) -> &[usize] {
693 &self.block_table
694 }
695
696 /// Appends one position's key/value vectors, acquiring a new block
697 /// from `store` first if the current tail block is full (or none
698 /// held yet). Mirrors `KvCache::push`'s signature/semantics exactly,
699 /// just against shared storage instead of a private buffer.
700 pub fn push(
701 &mut self,
702 store: &mut PagedKvStore,
703 k_step: &[f32],
704 v_step: &[f32],
705 ) -> Result<(), PagedStoreExhausted> {
706 let block_size = store.block_size();
707 let offset_in_block = self.seq_len % block_size;
708 // Index by position rather than taking the tail block: a
709 // sequence that pre-reserved (see `reserve`) already holds the
710 // block this position belongs in, and appending another would
711 // both leak a block and write the row in the wrong place.
712 let block_index = self.seq_len / block_size;
713 if block_index >= self.block_table.len() {
714 let id = store.acquire_block().ok_or(PagedStoreExhausted)?;
715 self.block_table.push(id);
716 }
717 let block_id = self.block_table[block_index];
718 store
719 .k_row_mut(block_id, offset_in_block)
720 .copy_from_slice(k_step);
721 store
722 .v_row_mut(block_id, offset_in_block)
723 .copy_from_slice(v_step);
724 self.seq_len += 1;
725 Ok(())
726 }
727
728 /// Appends a block the caller already owns, without taking one from
729 /// the store.
730 ///
731 /// This is how a sliding window recycles. A block whose positions
732 /// have fallen behind the window is never read again -- the paged
733 /// attention kernel indexes `block_table[t / block_size]` only for
734 /// `t >= seq_len - window` -- so its storage can back a *later*
735 /// position instead of being handed back and re-acquired. The table
736 /// keeps its absolute-position indexing and simply names the same
737 /// physical block at two indices: the stale one, which nothing
738 /// reads, and the live one.
739 ///
740 /// That aliasing is the reason this is a separate method rather than
741 /// a flag on [`Self::reserve`]. A caller that recycles owns the
742 /// obligation to release each distinct block exactly once, and to
743 /// have established that the donor index really is out of window --
744 /// neither of which this type can check for itself.
745 pub fn append_block(&mut self, block_id: usize) {
746 self.block_table.push(block_id);
747 }
748
749 /// Releases every block this sequence holds back to `store`. Must be
750 /// called explicitly (there's no `Drop` here, since dropping needs a
751 /// `&mut PagedKvStore` this type doesn't own a reference to) --
752 /// mirrors `KvCache::release_to_pool`, just not automatic.
753 ///
754 /// Each *distinct* block once: a table that has recycled through
755 /// [`Self::append_block`] names one block at more than one index, and
756 /// releasing per index would put the same id on the free list twice,
757 /// after which two sequences are handed the same memory.
758 pub fn release(&mut self, store: &mut PagedKvStore) {
759 let mut seen: Vec<usize> = Vec::new();
760 for id in self.block_table.drain(..) {
761 if !seen.contains(&id) {
762 seen.push(id);
763 store.release_block(id);
764 }
765 }
766 self.seq_len = 0;
767 }
768
769 /// How many *additional* blocks appending `n_new` positions would
770 /// take from `store`, given what this sequence already holds.
771 ///
772 /// Counted against held CAPACITY rather than against `seq_len`, so
773 /// it is right in both cases. The tail block is usually part-full,
774 /// so the answer is never simply `n_new / block_size`: positions
775 /// that land in a block already held cost nothing. And a sequence
776 /// that pre-reserved (see [`Self::reserve`]) holds blocks beyond
777 /// its length, which a `seq_len`-only sum would ask for twice.
778 ///
779 /// Callers that must not fail part-way through a write check this
780 /// against [`PagedKvStore::free_block_count`] before touching
781 /// anything.
782 pub fn blocks_needed_for(&self, store: &PagedKvStore, n_new: usize) -> usize {
783 let held_capacity = self.block_table.len() * store.block_size();
784 let unused = held_capacity.saturating_sub(self.seq_len);
785 n_new.saturating_sub(unused).div_ceil(store.block_size())
786 }
787
788 /// Takes the blocks `n_new` more positions will need, without
789 /// advancing `seq_len`.
790 ///
791 /// This is what makes a multi-layer append all-or-nothing. The
792 /// check and the taking happen together, so every later
793 /// [`Self::push`] writes into a block this sequence already owns
794 /// and cannot fail. Reserving and then not filling is harmless: the
795 /// blocks are this sequence's until it releases, and `seq_len`
796 /// still says how far it really got.
797 pub fn reserve(
798 &mut self,
799 store: &mut PagedKvStore,
800 n_new: usize,
801 ) -> Result<(), PagedStoreExhausted> {
802 let need = self.blocks_needed_for(store, n_new);
803 if need > store.free_block_count() {
804 return Err(PagedStoreExhausted);
805 }
806 for _ in 0..need {
807 let id = store
808 .acquire_block()
809 .expect("checked against free_block_count immediately above");
810 self.block_table.push(id);
811 }
812 Ok(())
813 }
814
815 /// Installs a block table the caller allocated, with `seq_len`
816 /// positions already computed in it.
817 ///
818 /// This is how a sequence starts life on top of a cached prefix:
819 /// the blocks are somebody else's, already full, and this sequence
820 /// appends past them.
821 ///
822 /// The `seq_len` installed here is therefore also the POSITION the
823 /// caller's next forward pass must run at, and the caller has no
824 /// second source for that number: [`Self::push`] writes at `seq_len`
825 /// and ignores whatever position its caller believes it is at. A
826 /// prefill that started from zero over an adopted prefix put the
827 /// prompt in the rows *after* the prefix while carrying positions
828 /// `0..n`, which is a wrong answer served with a 200.
829 ///
830 /// `seq_len` MUST be a whole number of blocks,
831 /// because the first append writes at `seq_len` and a shared block
832 /// must never be written -- another sequence is attending over it.
833 /// A ragged length would put that write inside the last shared
834 /// block, corrupting a prefix every other holder is reading.
835 pub fn adopt_blocks(&mut self, block_table: Vec<usize>, seq_len: usize, block_size: usize) {
836 assert_eq!(
837 seq_len % block_size,
838 0,
839 "an adopted prefix must end on a block boundary, or the first \
840 append writes into a block another sequence is reading"
841 );
842 assert!(
843 seq_len / block_size <= block_table.len(),
844 "block table too short for the adopted length"
845 );
846 self.block_table = block_table;
847 self.seq_len = seq_len;
848 }
849
850 /// Copies this sequence's KV out of the shared store into a plain
851 /// contiguous [`KvCache`].
852 ///
853 /// This is what lets the batched prefill path run *unchanged* over
854 /// paged storage. Its fast arm hands `cache.k` / `cache.v` to a
855 /// blocked kernel that reads them as flat slices, and a block table
856 /// cannot be expressed that way. Rather than maintain a second
857 /// prefill kernel that reads through the table -- a copy that could
858 /// drift from the one every other model path uses -- the pages are
859 /// materialised once per layer, the existing kernel runs, and the
860 /// new rows go back with [`Self::append_contiguous`].
861 ///
862 /// The cost is one `seq_len * n_kv_heads * head_dim` copy per layer
863 /// per prefill call, against matmuls that dominate prefill. Decode
864 /// still reads through the block table and copies nothing, which is
865 /// where page sharing actually pays.
866 pub fn to_contiguous(&self, store: &PagedKvStore) -> KvCache {
867 let elems_per_position = store.n_kv_heads * store.head_dim;
868 let mut cache = KvCache::with_capacity(store.n_kv_heads, store.head_dim, self.seq_len);
869 cache.k.reserve_exact(self.seq_len * elems_per_position);
870 cache.v.reserve_exact(self.seq_len * elems_per_position);
871 for pos in 0..self.seq_len {
872 let block_id = self.block_table[pos / store.block_size];
873 let offset = pos % store.block_size;
874 cache.k.extend_from_slice(store.k_row(block_id, offset));
875 cache.v.extend_from_slice(store.v_row(block_id, offset));
876 }
877 // The paged store does not evict either, so its positions and
878 // its rows agree and this one assignment is both.
879 cache.set_positions(self.seq_len);
880 cache
881 }
882
883 /// Appends `count` positions' worth of contiguous K/V rows, the
884 /// inverse of [`Self::to_contiguous`].
885 ///
886 /// Blocks are reserved for the whole append *before* the first row
887 /// is written, so a store that cannot hold the request refuses it
888 /// having changed nothing. Writing rows until the store runs dry
889 /// would leave the sequence with a `seq_len` that disagrees with
890 /// the model's own idea of how far it has got, which is not a
891 /// recoverable state.
892 pub fn append_contiguous(
893 &mut self,
894 store: &mut PagedKvStore,
895 k: &[f32],
896 v: &[f32],
897 count: usize,
898 ) -> Result<(), PagedStoreExhausted> {
899 let elems_per_position = store.n_kv_heads * store.head_dim;
900 assert_eq!(k.len(), count * elems_per_position, "k row count");
901 assert_eq!(v.len(), count * elems_per_position, "v row count");
902 if self.blocks_needed_for(store, count) > store.free_block_count() {
903 return Err(PagedStoreExhausted);
904 }
905 for i in 0..count {
906 let lo = i * elems_per_position;
907 let hi = lo + elems_per_position;
908 self.push(store, &k[lo..hi], &v[lo..hi])
909 .expect("blocks reserved above, so no push here can exhaust the store");
910 }
911 Ok(())
912 }
913}
914
915/// Per-layer [`PagedKvStore`]s that many concurrent requests share.
916///
917/// # Why a lock per layer, and why two phases
918///
919/// `ferrox-server` runs generation on `spawn_blocking` with, in its own
920/// words, "no I/O and no shared lock". `KvBlockPool` survives that
921/// because it only bounds a *count*: each `KvCache` owns a private
922/// `Vec`, and the pool mutex is taken briefly at acquire and release,
923/// never during a forward. A `PagedKvStore` is the opposite -- it IS
924/// the backing memory -- so sharing one across concurrent requests
925/// needs an answer to "who may touch these bytes when".
926///
927/// The answer the API already implies: attention takes
928/// `&PagedKvStore` and only `push` takes `&mut`. So the accesses split
929/// cleanly into many concurrent readers and one short exclusive write
930/// per position, which is exactly an `RwLock` -- and one per LAYER
931/// rather than one for the whole model, so two requests contend only
932/// when both are writing the same layer at the same instant.
933///
934/// A caller must therefore take the write guard for the push alone and
935/// drop it before attending under a read guard. Holding the write
936/// guard across attention would serialise the expensive half and give
937/// back a global lock with extra steps. Nothing breaks in the gap: a
938/// sequence's block table and length are its own, and another
939/// request's push in between only touches blocks it exclusively holds.
940///
941/// # Deadlock
942///
943/// [`Self::write_all`] is the one place several layers are held at
944/// once, and it takes them in ascending layer order. Every caller
945/// getting the same order is what makes that safe; there is no other
946/// multi-layer acquisition in the codebase, and a new one must follow
947/// the same rule.
948///
949/// # Poisoning
950///
951/// A panic while holding a store leaves the KV mid-write, which is not
952/// recoverable state, but it is also not *unsound* -- the bytes are
953/// plain `f32`. Poison is stepped over with `into_inner`, matching how
954/// `ferrox-server` already treats its pool mutex: a poisoned lock
955/// should not turn one request's panic into a permanently dead server.
956pub struct SharedPagedKv {
957 layers: Vec<RwLock<PagedKvStore>>,
958 /// Guarded separately from the layers, and always taken BEFORE
959 /// them, never while a layer guard is held. That one-way order is
960 /// what keeps group allocation and the per-layer push paths from
961 /// deadlocking against each other.
962 groups: Mutex<GroupTable>,
963}
964
965impl SharedPagedKv {
966 /// One store per layer, each with `blocks_per_layer` blocks.
967 pub fn new(
968 n_layers: usize,
969 block_size: usize,
970 blocks_per_layer: usize,
971 n_kv_heads: usize,
972 head_dim: usize,
973 ) -> Self {
974 SharedPagedKv {
975 layers: (0..n_layers)
976 .map(|_| {
977 RwLock::new(PagedKvStore::new(
978 block_size,
979 blocks_per_layer,
980 n_kv_heads,
981 head_dim,
982 ))
983 })
984 .collect(),
985 groups: Mutex::new(GroupTable::default()),
986 }
987 }
988
989 /// Wraps stores the caller built, for tests and for callers that
990 /// size layers differently.
991 pub fn from_stores(stores: Vec<PagedKvStore>) -> Self {
992 SharedPagedKv {
993 layers: stores.into_iter().map(RwLock::new).collect(),
994 groups: Mutex::new(GroupTable::default()),
995 }
996 }
997
998 pub fn layer_count(&self) -> usize {
999 self.layers.len()
1000 }
1001
1002 /// Shared access to one layer, for attention.
1003 pub fn read(&self, layer: usize) -> RwLockReadGuard<'_, PagedKvStore> {
1004 self.layers[layer]
1005 .read()
1006 .unwrap_or_else(|poisoned| poisoned.into_inner())
1007 }
1008
1009 /// Exclusive access to one layer, for a push. Hold it for the push
1010 /// and nothing else -- see the type docs.
1011 pub fn write(&self, layer: usize) -> RwLockWriteGuard<'_, PagedKvStore> {
1012 self.layers[layer]
1013 .write()
1014 .unwrap_or_else(|poisoned| poisoned.into_inner())
1015 }
1016
1017 /// Every layer at once, in ascending order, so a multi-layer append
1018 /// is atomic against other requests.
1019 ///
1020 /// This is what makes "all layers advance or none do" hold under
1021 /// concurrency rather than only single-threaded: checking free
1022 /// space and then appending are separate steps, and without the
1023 /// guards spanning both, another request can take the blocks in
1024 /// between and leave this one half-written.
1025 ///
1026 /// Ascending order is the deadlock rule; see the type docs.
1027 pub fn write_all(&self) -> Vec<RwLockWriteGuard<'_, PagedKvStore>> {
1028 self.layers
1029 .iter()
1030 .map(|l| l.write().unwrap_or_else(|poisoned| poisoned.into_inner()))
1031 .collect()
1032 }
1033
1034 /// Free blocks in one layer, for admission control. A snapshot: by
1035 /// the time a caller acts on it another request may have taken
1036 /// them, which is why the append itself re-checks under the guard.
1037 pub fn free_blocks(&self, layer: usize) -> usize {
1038 self.read(layer).free_block_count()
1039 }
1040
1041 /// Takes one block from EVERY layer as a single group, refcount 1.
1042 ///
1043 /// All layers or none: a group that existed in some layers and not
1044 /// others could not answer "which block holds position p in layer
1045 /// l", which is the only question it exists to answer.
1046 pub fn acquire_group(&self) -> Option<PageGroup> {
1047 let mut guards = self.write_all();
1048 if guards.iter().any(|s| s.free_block_count() == 0) {
1049 return None;
1050 }
1051 let blocks: Vec<usize> = guards
1052 .iter_mut()
1053 .map(|s| {
1054 s.acquire_block()
1055 .expect("checked every layer under these same guards")
1056 })
1057 .collect();
1058 let mut groups = self
1059 .groups
1060 .lock()
1061 .unwrap_or_else(|poisoned| poisoned.into_inner());
1062 Some(PageGroup(groups.insert(blocks)))
1063 }
1064
1065 /// One more holder of `group`.
1066 ///
1067 /// Called when a second sequence adopts a cached prefix. Without
1068 /// it, the first sequence to finish frees pages the second is
1069 /// still attending over -- a use-after-free that shows up as
1070 /// another conversation's tokens rather than as a crash.
1071 pub fn retain_group(&self, group: PageGroup) {
1072 let mut groups = self
1073 .groups
1074 .lock()
1075 .unwrap_or_else(|poisoned| poisoned.into_inner());
1076 groups.retain(group.0);
1077 }
1078
1079 /// One fewer holder. At zero the blocks go back to their layers.
1080 ///
1081 /// Returns whether this was the last holder, so a caller can assert
1082 /// on it rather than guess.
1083 pub fn release_group(&self, group: PageGroup) -> bool {
1084 let blocks = {
1085 let mut groups = self
1086 .groups
1087 .lock()
1088 .unwrap_or_else(|poisoned| poisoned.into_inner());
1089 match groups.release(group.0) {
1090 Some(blocks) => blocks,
1091 None => return false,
1092 }
1093 };
1094 // The groups lock is dropped before the layer guards are taken,
1095 // so the lock order is always groups-then-layers and never the
1096 // reverse. See the type docs on deadlock.
1097 let mut guards = self.write_all();
1098 for (store, block) in guards.iter_mut().zip(blocks) {
1099 store.release_block(block);
1100 }
1101 true
1102 }
1103
1104 /// Which block in each layer this group owns, indexed by layer.
1105 pub fn group_blocks(&self, group: PageGroup) -> Vec<usize> {
1106 let groups = self
1107 .groups
1108 .lock()
1109 .unwrap_or_else(|poisoned| poisoned.into_inner());
1110 groups.blocks(group.0).to_vec()
1111 }
1112
1113 /// How many holders `group` has. Zero means it does not exist.
1114 pub fn group_refs(&self, group: PageGroup) -> u32 {
1115 let groups = self
1116 .groups
1117 .lock()
1118 .unwrap_or_else(|poisoned| poisoned.into_inner());
1119 groups.refs(group.0)
1120 }
1121
1122 /// Groups that could still be allocated, bounded by the layer with
1123 /// the fewest free blocks: a group needs one from each.
1124 pub fn free_groups(&self) -> usize {
1125 (0..self.layers.len())
1126 .map(|l| self.free_blocks(l))
1127 .min()
1128 .unwrap_or(0)
1129 }
1130}
1131
1132/// A handle to one block in every layer.
1133///
1134/// The unit of sharing between sequences, and the only thing small
1135/// enough to be what a radix prefix cache stores: that cache maps a
1136/// token prefix to ONE index per token, while a position's KV lives in
1137/// `n_layers` different blocks. A group is the name for all of them.
1138#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
1139pub struct PageGroup(pub u32);
1140
1141/// Group ids, their per-layer blocks, and how many holders each has.
1142#[derive(Debug, Default)]
1143struct GroupTable {
1144 /// Indexed by group id. `None` for an id currently on the free list.
1145 blocks: Vec<Option<Vec<usize>>>,
1146 refs: Vec<u32>,
1147 free_ids: Vec<u32>,
1148}
1149
1150impl GroupTable {
1151 fn insert(&mut self, blocks: Vec<usize>) -> u32 {
1152 if let Some(id) = self.free_ids.pop() {
1153 self.blocks[id as usize] = Some(blocks);
1154 self.refs[id as usize] = 1;
1155 return id;
1156 }
1157 self.blocks.push(Some(blocks));
1158 self.refs.push(1);
1159 (self.blocks.len() - 1) as u32
1160 }
1161
1162 fn retain(&mut self, id: u32) {
1163 let refs = &mut self.refs[id as usize];
1164 assert!(*refs > 0, "cannot retain group {id}, which has no holders");
1165 *refs += 1;
1166 }
1167
1168 /// Drops one holder, returning the blocks to free only when the
1169 /// last one goes.
1170 fn release(&mut self, id: u32) -> Option<Vec<usize>> {
1171 let refs = &mut self.refs[id as usize];
1172 assert!(*refs > 0, "double free of group {id}");
1173 *refs -= 1;
1174 if *refs > 0 {
1175 return None;
1176 }
1177 // The id is reusable now, but only after the blocks are out:
1178 // handing the id back while it still named blocks would let a
1179 // later `acquire_group` believe it owns them too.
1180 let blocks = self.blocks[id as usize]
1181 .take()
1182 .expect("a group with holders always has blocks");
1183 self.free_ids.push(id);
1184 Some(blocks)
1185 }
1186
1187 fn blocks(&self, id: u32) -> &[usize] {
1188 self.blocks[id as usize]
1189 .as_deref()
1190 .expect("group has no blocks; it was already released")
1191 }
1192
1193 fn refs(&self, id: u32) -> u32 {
1194 self.refs.get(id as usize).copied().unwrap_or(0)
1195 }
1196}
1197
1198#[cfg(test)]
1199mod tests {
1200 use super::*;
1201
1202 /// `blocks_needed_for` is the reservation the whole no-partial-write
1203 /// guarantee rests on, and it is wrong in two opposite directions
1204 /// that fail very differently.
1205 ///
1206 /// UNDER-counting is the dangerous one: `append_contiguous` reserves
1207 /// on this answer and then pushes with an `expect`, so too small a
1208 /// number panics part-way through a layer -- exactly the corrupted
1209 /// state the reservation exists to prevent. Over-counting merely
1210 /// refuses a request that would have fitted.
1211 ///
1212 /// Both mistakes are one edit away. Flooring instead of ceiling
1213 /// under-counts whenever the append does not land on a block
1214 /// boundary; ignoring the part-full tail over-counts whenever a
1215 /// sequence is mid-block, which after the first token is almost
1216 /// always. Neither shows up when the numbers happen to divide
1217 /// evenly, so the cases here are chosen so that they do not.
1218 #[test]
1219 fn blocks_needed_for_accounts_for_the_part_full_tail_block() {
1220 let mut store = PagedKvStore::new(/* block_size = */ 4, 64, 1, 1);
1221 let mut cache = PagedKvCache::new();
1222 let row = [1.0f32];
1223 // Real pushes rather than poking `seq_len`: the count is
1224 // against blocks this sequence HOLDS, so a length with no
1225 // blocks behind it is a state that cannot occur and would only
1226 // let the test agree with an arithmetic nothing produces.
1227 let advance = |cache: &mut PagedKvCache, store: &mut PagedKvStore, n: usize| {
1228 for _ in 0..n {
1229 cache.push(store, &row, &row).unwrap();
1230 }
1231 };
1232
1233 // Empty: a whole-block boundary, and a remainder that a floor
1234 // would round away.
1235 assert_eq!(cache.blocks_needed_for(&store, 0), 0);
1236 assert_eq!(cache.blocks_needed_for(&store, 1), 1);
1237 assert_eq!(cache.blocks_needed_for(&store, 4), 1);
1238 assert_eq!(cache.blocks_needed_for(&store, 5), 2, "5 into 4s needs 2");
1239
1240 // One position in: three slots free in the tail, so appending up
1241 // to three costs NOTHING. Ignoring the tail would say 1.
1242 advance(&mut cache, &mut store, 1);
1243 assert_eq!(cache.blocks_needed_for(&store, 3), 0, "fits in the tail");
1244 assert_eq!(cache.blocks_needed_for(&store, 4), 1);
1245 assert_eq!(cache.blocks_needed_for(&store, 8), 2);
1246
1247 // The awkward case: 1 free in the tail, 6 to append. 5 spill
1248 // over 4-wide blocks, so 2. A floor gives 1 and a tail-blind
1249 // ceil gives 2 for the wrong reason, so this pins the shape.
1250 advance(&mut cache, &mut store, 2); // seq_len = 3
1251 assert_eq!(cache.blocks_needed_for(&store, 6), 2);
1252 assert_eq!(cache.blocks_needed_for(&store, 5), 1);
1253
1254 // Tail exactly full: no free slots, so this behaves like empty.
1255 advance(&mut cache, &mut store, 1); // seq_len = 4
1256 assert_eq!(cache.blocks_needed_for(&store, 1), 1);
1257 assert_eq!(cache.blocks_needed_for(&store, 4), 1);
1258
1259 // A RESERVED block is capacity this sequence already holds, so
1260 // it must not be asked for twice. Counting from `seq_len` alone
1261 // would say 1 here and take a second block for positions the
1262 // reservation already covers.
1263 cache.reserve(&mut store, 4).unwrap();
1264 assert_eq!(
1265 cache.blocks_needed_for(&store, 4),
1266 0,
1267 "a reserved block is already held"
1268 );
1269 assert_eq!(cache.blocks_needed_for(&store, 5), 1);
1270 }
1271
1272 /// A group takes one block from every layer, and gives them all
1273 /// back together.
1274 ///
1275 /// All-or-nothing is the point: a group holding blocks in some
1276 /// layers and not others cannot answer "which block holds position
1277 /// p in layer l", which is the only question it exists for.
1278 #[test]
1279 fn a_group_takes_one_block_from_every_layer_and_returns_them_together() {
1280 let kv = SharedPagedKv::new(3, 2, 4, 1, 1);
1281 assert_eq!(kv.free_groups(), 4);
1282
1283 let g = kv.acquire_group().expect("4 groups available");
1284 let blocks = kv.group_blocks(g);
1285 assert_eq!(blocks.len(), 3, "one block per layer");
1286 for l in 0..3 {
1287 assert_eq!(kv.free_blocks(l), 3, "layer {l} gave up exactly one");
1288 }
1289 assert_eq!(kv.free_groups(), 3);
1290
1291 assert!(kv.release_group(g), "sole holder, so this frees it");
1292 for l in 0..3 {
1293 assert_eq!(kv.free_blocks(l), 4, "layer {l} got its block back");
1294 }
1295 assert_eq!(kv.free_groups(), 4);
1296 }
1297
1298 /// A group survives until its LAST holder releases it.
1299 ///
1300 /// This is what makes prefix sharing safe. Two sequences off one
1301 /// system prompt hold the same pages; if the first to finish freed
1302 /// them, the second would keep attending over blocks the store had
1303 /// already handed to somebody else -- surfacing as another
1304 /// conversation's tokens, not as a crash.
1305 #[test]
1306 fn a_group_shared_by_two_holders_survives_the_first_release() {
1307 let kv = SharedPagedKv::new(2, 2, 2, 1, 1);
1308 let g = kv.acquire_group().unwrap();
1309 let blocks = kv.group_blocks(g);
1310 kv.retain_group(g);
1311 assert_eq!(kv.group_refs(g), 2);
1312
1313 assert!(
1314 !kv.release_group(g),
1315 "one holder remains, so nothing is freed"
1316 );
1317 assert_eq!(kv.group_refs(g), 1);
1318 assert_eq!(kv.free_blocks(0), 1, "the blocks are still held");
1319 assert_eq!(kv.group_blocks(g), blocks, "and still name the same blocks");
1320
1321 assert!(kv.release_group(g), "last holder frees it");
1322 assert_eq!(kv.group_refs(g), 0);
1323 assert_eq!(kv.free_blocks(0), 2);
1324 }
1325
1326 /// Exhaustion is per group, bounded by the tightest layer.
1327 ///
1328 /// A layer with one block left caps the whole pool at one more
1329 /// group however much room the others have, because a group needs
1330 /// one block from each.
1331 #[test]
1332 fn group_capacity_is_bounded_by_the_layer_with_the_fewest_blocks() {
1333 let kv = SharedPagedKv::from_stores(vec![
1334 PagedKvStore::new(2, 5, 1, 1),
1335 PagedKvStore::new(2, 1, 1, 1),
1336 ]);
1337 assert_eq!(kv.free_groups(), 1, "layer 1 has only one block");
1338
1339 let g = kv.acquire_group().expect("one group fits");
1340 assert_eq!(kv.free_groups(), 0);
1341 assert!(
1342 kv.acquire_group().is_none(),
1343 "layer 1 is empty, so no group can be formed"
1344 );
1345 // The refused attempt must not have taken layer 0's block.
1346 assert_eq!(kv.free_blocks(0), 4, "a refused group leaks nothing");
1347 kv.release_group(g);
1348 assert_eq!(kv.free_blocks(0), 5);
1349 }
1350
1351 /// A released id is reused, with a refcount that starts over.
1352 #[test]
1353 fn a_released_group_id_is_reused_with_a_fresh_refcount() {
1354 let kv = SharedPagedKv::new(1, 2, 2, 1, 1);
1355 let first = kv.acquire_group().unwrap();
1356 kv.retain_group(first);
1357 assert_eq!(kv.group_refs(first), 2);
1358 kv.release_group(first);
1359 kv.release_group(first);
1360 assert_eq!(kv.group_refs(first), 0, "gone, not merely decremented");
1361
1362 let second = kv.acquire_group().unwrap();
1363 assert_eq!(second, first, "the id is reused");
1364 assert_eq!(
1365 kv.group_refs(second),
1366 1,
1367 "a reused id must not inherit the old count"
1368 );
1369 assert_eq!(kv.group_blocks(second).len(), 1);
1370 assert_eq!(kv.free_blocks(0), 1);
1371 }
1372
1373 /// Reading a group after its last holder released it PANICS rather
1374 /// than answering with stale blocks.
1375 ///
1376 /// This is the observable half of clearing the entry on release,
1377 /// and the reason it is `take` rather than `clone`: a caller still
1378 /// holding a `PageGroup` after releasing it is exactly the bug
1379 /// refcounting exists to prevent, and blocks that now belong to
1380 /// somebody else are the worst possible answer -- the caller reads
1381 /// another sequence's KV and nothing says so.
1382 ///
1383 /// Written after sabotage showed the previous test here passed with
1384 /// `clone` in place of `take`: `insert` overwrites the entry on
1385 /// reuse, so a stale entry was never reachable through the path
1386 /// that test took. This one reaches it.
1387 #[test]
1388 #[should_panic(expected = "already released")]
1389 fn reading_a_released_group_panics_rather_than_returning_stale_blocks() {
1390 let kv = SharedPagedKv::new(2, 2, 2, 1, 1);
1391 let g = kv.acquire_group().unwrap();
1392 assert!(kv.release_group(g));
1393 let _ = kv.group_blocks(g);
1394 }
1395
1396 /// Releasing a group nobody holds is a bug, not a no-op.
1397 ///
1398 /// Silently ignoring it would let a double release return the same
1399 /// blocks to the store twice, after which two sequences are handed
1400 /// the same page and both write it.
1401 #[test]
1402 #[should_panic(expected = "double free of group")]
1403 fn releasing_a_group_twice_panics_rather_than_freeing_it_twice() {
1404 let kv = SharedPagedKv::new(1, 2, 2, 1, 1);
1405 let g = kv.acquire_group().unwrap();
1406 assert!(kv.release_group(g));
1407 kv.release_group(g);
1408 }
1409
1410 /// A recycled block backs a later position without the store ever
1411 /// being asked for another one, and the later position's writes are
1412 /// what a read at that position returns.
1413 ///
1414 /// This is the whole sliding-window mechanism in miniature. Blocks
1415 /// of two, four positions, and only two blocks in the store: without
1416 /// recycling, position 2 has nowhere to go.
1417 #[test]
1418 fn a_recycled_block_backs_a_later_position_without_touching_the_store() {
1419 let mut store = PagedKvStore::new(2, 2, 1, 2);
1420 let mut cache = PagedKvCache::new();
1421 cache.push(&mut store, &[1.0, 1.0], &[1.0, 1.0]).unwrap();
1422 cache.push(&mut store, &[2.0, 2.0], &[2.0, 2.0]).unwrap();
1423 assert_eq!(store.free_block_count(), 1, "one block per position pair");
1424
1425 // Positions 0..2 have fallen behind a window of two. Their block
1426 // backs positions 2..4 instead, and the store is untouched.
1427 let recycled = cache.block_table()[0];
1428 cache.append_block(recycled);
1429 assert_eq!(
1430 store.free_block_count(),
1431 1,
1432 "recycling must not take a block from the store"
1433 );
1434 cache.push(&mut store, &[3.0, 3.0], &[3.0, 3.0]).unwrap();
1435 assert_eq!(cache.seq_len(), 3);
1436 assert_eq!(
1437 cache.block_table(),
1438 &[recycled, recycled],
1439 "the same block at the stale index and the live one"
1440 );
1441
1442 // Reading position 2 sees the new row. Position 0's row is gone,
1443 // which is exactly what "behind the window" means -- the kernel
1444 // never indexes it.
1445 let flat = cache.to_contiguous(&store);
1446 assert_eq!(&flat.k[4..6], &[3.0, 3.0], "position 2 reads its own row");
1447 assert_eq!(
1448 &flat.k[0..2],
1449 &[3.0, 3.0],
1450 "position 0 now reads the recycled row, and nothing may read it"
1451 );
1452 }
1453
1454 /// Releasing an aliased table hands each block back ONCE.
1455 ///
1456 /// Per index instead of per distinct block would put the recycled id
1457 /// on the free list twice, and the next two acquisitions would hand
1458 /// two sequences the same memory -- which does not fail, it
1459 /// interleaves two conversations' KV.
1460 #[test]
1461 fn releasing_a_recycled_table_gives_each_block_back_once() {
1462 // Exactly one block in the store, so "handed back twice" is
1463 // observable as a second acquisition succeeding.
1464 let mut store = PagedKvStore::new(2, 1, 1, 2);
1465 let mut cache = PagedKvCache::new();
1466 cache.push(&mut store, &[1.0, 1.0], &[1.0, 1.0]).unwrap();
1467 let held = cache.block_table()[0];
1468 cache.append_block(held);
1469 cache.append_block(held);
1470
1471 let free_before = store.free_block_count();
1472 cache.release(&mut store);
1473 assert_eq!(
1474 store.free_block_count(),
1475 free_before + 1,
1476 "three table entries naming one block are one block back"
1477 );
1478 // And the store agrees: it can hand out that block once.
1479 assert!(store.acquire_block().is_some());
1480 assert!(store.acquire_block().is_none());
1481 }
1482
1483 #[test]
1484 fn a_gathered_sequence_round_trips_through_the_store() {
1485 let mut store = PagedKvStore::new(2, 8, 2, 2);
1486 let mut cache = PagedKvCache::new();
1487 // Five positions over blocks of two: the tail block is half
1488 // full, which is where an off-by-one in the gather shows up.
1489 let rows: Vec<[f32; 4]> = (0..5)
1490 .map(|i| {
1491 let b = i as f32 * 10.0;
1492 [b + 1.0, b + 2.0, b + 3.0, b + 4.0]
1493 })
1494 .collect();
1495 for r in &rows {
1496 cache.push(&mut store, r, r).unwrap();
1497 }
1498
1499 let flat = cache.to_contiguous(&store);
1500 assert_eq!(flat.positions(), 5);
1501 assert_eq!(flat.k.len(), 5 * 4);
1502 for (i, r) in rows.iter().enumerate() {
1503 assert_eq!(&flat.k[i * 4..(i + 1) * 4], r, "position {i} k");
1504 assert_eq!(&flat.v[i * 4..(i + 1) * 4], r, "position {i} v");
1505 }
1506
1507 // And appending those same rows back onto a fresh sequence
1508 // reproduces the store's view of them exactly.
1509 let mut rebuilt = PagedKvCache::new();
1510 let mut store2 = PagedKvStore::new(2, 8, 2, 2);
1511 rebuilt
1512 .append_contiguous(&mut store2, &flat.k, &flat.v, 5)
1513 .unwrap();
1514 let again = rebuilt.to_contiguous(&store2);
1515 assert_eq!(again.k, flat.k);
1516 assert_eq!(again.v, flat.v);
1517 assert_eq!(again.positions(), flat.positions());
1518 }
1519
1520 #[test]
1521 fn push_grows_seq_len_and_stores_values() {
1522 let mut cache = KvCache::new(2, 2);
1523 cache
1524 .push(&[1.0, 2.0, 3.0, 4.0], &[5.0, 6.0, 7.0, 8.0])
1525 .unwrap();
1526 assert_eq!(cache.positions(), 1);
1527 cache
1528 .push(&[9.0, 10.0, 11.0, 12.0], &[13.0, 14.0, 15.0, 16.0])
1529 .unwrap();
1530 assert_eq!(cache.positions(), 2);
1531 assert_eq!(cache.k.len(), 2 * 2 * 2);
1532 assert_eq!(cache.k[4], 9.0);
1533 }
1534
1535 #[test]
1536 #[should_panic]
1537 fn push_wrong_size_panics() {
1538 let mut cache = KvCache::new(2, 2);
1539 let _ = cache.push(&[1.0, 2.0], &[1.0, 2.0]); // too short
1540 }
1541
1542 #[test]
1543 fn clear_resets_state() {
1544 let mut cache = KvCache::new(1, 1);
1545 cache.push(&[1.0], &[2.0]).unwrap();
1546 cache.clear();
1547 assert_eq!(cache.positions(), 0);
1548 assert!(cache.k.is_empty());
1549 }
1550
1551 #[test]
1552 fn truncate_rolls_back_to_exact_length_preserving_earlier_data() {
1553 let mut cache = KvCache::new(2, 2);
1554 cache
1555 .push(&[1.0, 2.0, 3.0, 4.0], &[10.0, 20.0, 30.0, 40.0])
1556 .unwrap();
1557 cache
1558 .push(&[5.0, 6.0, 7.0, 8.0], &[50.0, 60.0, 70.0, 80.0])
1559 .unwrap();
1560 cache
1561 .push(&[9.0, 9.0, 9.0, 9.0], &[90.0, 90.0, 90.0, 90.0])
1562 .unwrap();
1563 assert_eq!(cache.positions(), 3);
1564
1565 cache.truncate(1);
1566 assert_eq!(cache.positions(), 1);
1567 assert_eq!(cache.k, vec![1.0, 2.0, 3.0, 4.0]);
1568 assert_eq!(cache.v, vec![10.0, 20.0, 30.0, 40.0]);
1569 }
1570
1571 #[test]
1572 fn truncate_to_current_length_is_a_no_op() {
1573 let mut cache = KvCache::new(1, 2);
1574 cache.push(&[1.0, 2.0], &[3.0, 4.0]).unwrap();
1575 cache.truncate(1);
1576 assert_eq!(cache.positions(), 1);
1577 assert_eq!(cache.k, vec![1.0, 2.0]);
1578 }
1579
1580 #[test]
1581 fn truncate_to_zero_empties_the_cache() {
1582 let mut cache = KvCache::new(1, 2);
1583 cache.push(&[1.0, 2.0], &[3.0, 4.0]).unwrap();
1584 cache.truncate(0);
1585 assert_eq!(cache.positions(), 0);
1586 assert!(cache.k.is_empty());
1587 assert!(cache.v.is_empty());
1588 }
1589
1590 #[test]
1591 #[should_panic]
1592 fn truncate_beyond_current_length_panics() {
1593 let mut cache = KvCache::new(1, 2);
1594 cache.push(&[1.0, 2.0], &[3.0, 4.0]).unwrap();
1595 cache.truncate(5);
1596 }
1597
1598 #[test]
1599 fn push_after_truncate_continues_correctly() {
1600 let mut cache = KvCache::new(1, 1);
1601 cache.push(&[1.0], &[10.0]).unwrap();
1602 cache.push(&[2.0], &[20.0]).unwrap();
1603 cache.push(&[3.0], &[30.0]).unwrap(); // this one will be "rejected"
1604 cache.truncate(2);
1605 cache.push(&[99.0], &[990.0]).unwrap(); // real continuation after rejection
1606 assert_eq!(cache.positions(), 3);
1607 assert_eq!(cache.k, vec![1.0, 2.0, 99.0]);
1608 assert_eq!(cache.v, vec![10.0, 20.0, 990.0]);
1609 }
1610
1611 #[test]
1612 fn with_capacity_preallocates_and_never_reallocates_within_plan() {
1613 let n_kv_heads = 4;
1614 let head_dim = 8;
1615 let max_seq_len = 16;
1616 let mut cache = KvCache::with_capacity(n_kv_heads, head_dim, max_seq_len);
1617
1618 let expected_elems = max_seq_len * n_kv_heads * head_dim;
1619 assert!(cache.k.capacity() >= expected_elems);
1620 assert!(cache.v.capacity() >= expected_elems);
1621
1622 let step = vec![0.5f32; n_kv_heads * head_dim];
1623 let k_ptr_before = cache.k.as_ptr();
1624 for _ in 0..max_seq_len {
1625 cache.push(&step, &step).unwrap();
1626 }
1627 let k_ptr_after = cache.k.as_ptr();
1628 assert_eq!(
1629 k_ptr_before, k_ptr_after,
1630 "pushing exactly up to the planned capacity must not reallocate"
1631 );
1632 assert!(cache.is_within_planned_capacity());
1633 }
1634
1635 #[test]
1636 fn allocated_bytes_reflects_preallocated_capacity_not_just_used_length() {
1637 let cache = KvCache::with_capacity(4, 8, 100);
1638 // 100 positions * 4 kv_heads * 8 head_dim * 2 (k+v) * 4 bytes/f32
1639 let expected_min = 100 * 4 * 8 * 2 * 4;
1640 assert!(
1641 cache.allocated_bytes() >= expected_min,
1642 "allocated_bytes={} expected_min={expected_min}",
1643 cache.allocated_bytes()
1644 );
1645 // Nothing has been pushed yet, but the memory is already reserved.
1646 assert_eq!(cache.positions(), 0);
1647 }
1648
1649 #[test]
1650 fn grow_as_you_go_cache_reports_not_within_planned_capacity() {
1651 let mut cache = KvCache::new(2, 2);
1652 cache
1653 .push(&[1.0, 2.0, 3.0, 4.0], &[5.0, 6.0, 7.0, 8.0])
1654 .unwrap();
1655 assert!(
1656 !cache.is_within_planned_capacity(),
1657 "a cache built with `new` has no plan to be within"
1658 );
1659 }
1660
1661 #[test]
1662 fn with_pool_acquires_one_block_and_reports_it_in_free_blocks() {
1663 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 10)));
1664 let cache = KvCache::with_pool(2, 2, pool.clone(), 0).unwrap();
1665 assert_eq!(pool.lock().unwrap().free_blocks(), 9);
1666 assert_eq!(cache.positions(), 0);
1667 }
1668
1669 #[test]
1670 fn with_pool_fails_without_mutating_the_pool_when_exhausted() {
1671 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 0)));
1672 let result = KvCache::with_pool(2, 2, pool.clone(), 0);
1673 assert!(result.is_err());
1674 assert_eq!(pool.lock().unwrap().free_blocks(), 0);
1675 }
1676
1677 #[test]
1678 fn push_acquires_additional_blocks_as_the_cache_crosses_block_boundaries() {
1679 let block_size = 2;
1680 let pool = Arc::new(Mutex::new(KvBlockPool::new(block_size, 10)));
1681 let mut cache = KvCache::with_pool(1, 1, pool.clone(), 0).unwrap();
1682 assert_eq!(pool.lock().unwrap().free_blocks(), 9);
1683
1684 // First block holds `block_size` = 2 positions; pushing them
1685 // must not need a second block.
1686 cache.push(&[1.0], &[1.0]).unwrap();
1687 cache.push(&[2.0], &[2.0]).unwrap();
1688 assert_eq!(
1689 pool.lock().unwrap().free_blocks(),
1690 9,
1691 "filling exactly the first block must not acquire a second one"
1692 );
1693
1694 // The third position crosses into a second block.
1695 cache.push(&[3.0], &[3.0]).unwrap();
1696 assert_eq!(pool.lock().unwrap().free_blocks(), 8);
1697 assert_eq!(cache.positions(), 3);
1698 assert_eq!(cache.k, vec![1.0, 2.0, 3.0]);
1699 }
1700
1701 #[test]
1702 fn push_returns_pool_exhausted_and_leaves_state_unchanged_when_no_blocks_remain() {
1703 let block_size = 1;
1704 let pool = Arc::new(Mutex::new(KvBlockPool::new(block_size, 1)));
1705 let mut cache = KvCache::with_pool(1, 1, pool.clone(), 0).unwrap();
1706 assert_eq!(pool.lock().unwrap().free_blocks(), 0);
1707
1708 cache.push(&[1.0], &[1.0]).unwrap(); // fills the one held block
1709
1710 let before_k = cache.k.clone();
1711 let result = cache.push(&[2.0], &[2.0]);
1712 assert_eq!(result, Err(KvPoolExhausted));
1713 assert_eq!(
1714 cache.positions(),
1715 1,
1716 "a failed push must not change seq_len"
1717 );
1718 assert_eq!(cache.k, before_k, "a failed push must not append data");
1719 }
1720
1721 #[test]
1722 fn dropping_a_pooled_cache_returns_its_blocks_to_the_pool() {
1723 let pool = Arc::new(Mutex::new(KvBlockPool::new(1, 2)));
1724 {
1725 let mut cache = KvCache::with_pool(1, 1, pool.clone(), 0).unwrap();
1726 cache.push(&[1.0], &[1.0]).unwrap(); // fills the first (only held) block
1727 cache.push(&[2.0], &[2.0]).unwrap(); // crosses into a second block
1728 assert_eq!(pool.lock().unwrap().free_blocks(), 0);
1729 }
1730 assert_eq!(
1731 pool.lock().unwrap().free_blocks(),
1732 2,
1733 "both blocks held by the dropped cache must return to the pool"
1734 );
1735 }
1736
1737 #[test]
1738 fn release_to_pool_is_explicit_and_idempotent() {
1739 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 5)));
1740 let mut cache = KvCache::with_pool(1, 1, pool.clone(), 0).unwrap();
1741 assert_eq!(pool.lock().unwrap().free_blocks(), 4);
1742
1743 cache.release_to_pool();
1744 assert_eq!(pool.lock().unwrap().free_blocks(), 5);
1745
1746 cache.release_to_pool(); // no-op, must not over-release
1747 assert_eq!(pool.lock().unwrap().free_blocks(), 5);
1748
1749 drop(cache); // must not release again either
1750 assert_eq!(pool.lock().unwrap().free_blocks(), 5);
1751 }
1752
1753 #[test]
1754 fn two_pooled_caches_share_one_bounded_budget() {
1755 let pool = Arc::new(Mutex::new(KvBlockPool::new(1, 1)));
1756 let cache_a = KvCache::with_pool(1, 1, pool.clone(), 0).unwrap();
1757 let cache_b = KvCache::with_pool(1, 1, pool.clone(), 0);
1758 assert!(
1759 cache_b.is_err(),
1760 "a second concurrent request must not be admitted when the shared budget is full"
1761 );
1762
1763 drop(cache_a);
1764 let cache_c = KvCache::with_pool(1, 1, pool, 0);
1765 assert!(
1766 cache_c.is_ok(),
1767 "once the first request's cache is dropped, its budget must become available again"
1768 );
1769 }
1770
1771 #[test]
1772 fn cloning_a_pooled_cache_detaches_the_clone_from_pool_accounting() {
1773 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 3)));
1774 let original = KvCache::with_pool(1, 1, pool.clone(), 0).unwrap();
1775 assert_eq!(pool.lock().unwrap().free_blocks(), 2);
1776
1777 let clone = original.clone();
1778 assert_eq!(
1779 pool.lock().unwrap().free_blocks(),
1780 2,
1781 "cloning must not acquire additional blocks"
1782 );
1783 assert_eq!(clone.k, original.k);
1784
1785 drop(clone);
1786 assert_eq!(
1787 pool.lock().unwrap().free_blocks(),
1788 2,
1789 "dropping a detached clone must not release the original's blocks"
1790 );
1791
1792 drop(original);
1793 assert_eq!(
1794 pool.lock().unwrap().free_blocks(),
1795 3,
1796 "dropping the original must release its blocks exactly once"
1797 );
1798 }
1799
1800 /// A resize is arithmetic, and the one rule that is not: shrinking
1801 /// past what is held is refused, and the pool is left exactly as it
1802 /// was.
1803 ///
1804 /// Clamping to zero instead would silently over-promise -- the
1805 /// caches holding those blocks do not give them back, so every
1806 /// later acquire would decide against a budget that does not
1807 /// describe the memory in use. This test fails under that clamp.
1808 #[test]
1809 fn a_pool_refuses_to_shrink_below_what_is_already_held() {
1810 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 10)));
1811 let held = KvCache::with_pool(2, 4, Arc::clone(&pool), 24).expect("blocks");
1812 let in_use = {
1813 let p = pool.lock().unwrap();
1814 p.total_blocks() - p.free_blocks()
1815 };
1816 assert!(in_use > 0, "the fixture must actually hold blocks");
1817
1818 let mut p = pool.lock().unwrap();
1819 assert_eq!(p.resize(in_use - 1), Err(in_use));
1820 assert_eq!(p.total_blocks(), 10, "a refused resize changes nothing");
1821 assert_eq!(p.free_blocks(), 10 - in_use);
1822
1823 // Down to exactly what is held is legal, and leaves nothing free.
1824 assert_eq!(p.resize(in_use), Ok(()));
1825 assert_eq!(p.free_blocks(), 0);
1826 drop(p);
1827 drop(held);
1828 }
1829
1830 /// Growing hands the new blocks to the free list without disturbing
1831 /// what is held, which is the whole point of a live re-split.
1832 #[test]
1833 fn growing_a_pool_adds_to_what_is_free_and_not_to_what_is_held() {
1834 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 8)));
1835 let held = KvCache::with_pool(2, 4, Arc::clone(&pool), 16).expect("blocks");
1836 let mut p = pool.lock().unwrap();
1837 let in_use = p.total_blocks() - p.free_blocks();
1838
1839 assert_eq!(p.resize(32), Ok(()));
1840 assert_eq!(p.total_blocks(), 32);
1841 assert_eq!(p.free_blocks(), 32 - in_use);
1842 drop(p);
1843 drop(held);
1844 }
1845
1846 /// Positions and rows agree for every store that does not evict,
1847 /// and this pins that they are DERIVED separately rather than one
1848 /// being an alias for the other.
1849 ///
1850 /// The point of the split is #61: a windowed layer will drop rows
1851 /// behind its window while its position count keeps climbing. Until
1852 /// then the two are equal, so this test cannot prove much about
1853 /// eviction. What it CAN prove, and what matters, is that `rows`
1854 /// reads the buffer rather than the counter: set the counter to a
1855 /// lie and `rows` still reports the truth.
1856 #[test]
1857 fn rows_is_read_from_the_buffer_and_positions_from_the_counter() {
1858 let mut cache = KvCache::new(2, 4);
1859 for i in 0..5 {
1860 let step = vec![i as f32; 8];
1861 cache.push(&step, &step).expect("unbounded growth");
1862 }
1863 assert_eq!(cache.positions(), 5);
1864 assert_eq!(cache.rows(), 5, "nothing evicts, so they agree");
1865
1866 // The counter lies; the buffer does not.
1867 cache.force_positions_for_test(99);
1868 assert_eq!(cache.positions(), 99);
1869 assert_eq!(
1870 cache.rows(),
1871 5,
1872 "rows must come from k/v, or it is just a second name for the counter"
1873 );
1874 }
1875
1876 /// `truncate` is measured in POSITIONS, and it moves both, because
1877 /// nothing evicts yet. Written down because it is the first thing
1878 /// eviction changes: a truncate target will stop being a row index.
1879 #[test]
1880 fn truncate_moves_positions_and_rows_together_while_nothing_evicts() {
1881 let mut cache = KvCache::new(1, 2);
1882 for i in 0..4 {
1883 let step = vec![i as f32; 2];
1884 cache.push(&step, &step).expect("unbounded growth");
1885 }
1886 cache.truncate(2);
1887 assert_eq!(cache.positions(), 2);
1888 assert_eq!(cache.rows(), 2);
1889 assert_eq!(cache.k.len(), 4, "two positions of two elements each");
1890 }
1891
1892 /// The store's rule must be *the* rule, not a second copy of it.
1893 ///
1894 /// `KvWindow::rows_after` is what `ferrox_models::kv_budget` prices
1895 /// against. If the store drained to anything else the budget would
1896 /// be describing a cache that does not exist, which is #33 again.
1897 #[test]
1898 fn a_windowed_cache_holds_exactly_what_the_rule_says_it_holds() {
1899 let window = KvWindow::new(8, 3).expect("positive window");
1900 let mut cache = KvCache::new(2, 4);
1901 cache.arm_window(window);
1902 let step = vec![1.0f32; 8];
1903 for p in 1..=200usize {
1904 cache.push(&step, &step).expect("unbounded growth");
1905 cache.evict_behind_window();
1906 assert_eq!(cache.positions(), p);
1907 assert_eq!(
1908 cache.rows(),
1909 window.rows_after(p),
1910 "at {p} positions the store and the rule disagree"
1911 );
1912 }
1913 }
1914
1915 /// The point of the issue: positions keep counting, rows do not.
1916 #[test]
1917 fn a_windowed_layers_resident_rows_stop_growing() {
1918 let window = KvWindow::with_default_slack(16).expect("positive window");
1919 let mut cache = KvCache::new(2, 4);
1920 cache.arm_window(window);
1921 let step = vec![0.5f32; 8];
1922 for _ in 0..2000 {
1923 cache.push(&step, &step).expect("unbounded growth");
1924 cache.evict_behind_window();
1925 }
1926 assert_eq!(cache.positions(), 2000);
1927 assert!(
1928 cache.rows() <= window.max_rows(),
1929 "{} rows resident after 2000 positions",
1930 cache.rows()
1931 );
1932 // And the bytes really went back, not just the length: `drain`
1933 // frees nothing, and memory is the whole point.
1934 assert!(
1935 cache.allocated_bytes() <= window.max_rows() * 8 * 2 * 4 * 2,
1936 "capacity was never handed back: {} bytes",
1937 cache.allocated_bytes()
1938 );
1939 }
1940
1941 /// **The correctness argument, measured.**
1942 ///
1943 /// Eviction is only allowed to be token-identical because the rows a
1944 /// windowed kernel reads -- the last `window` of them -- are the
1945 /// same bytes whether or not anything behind them was dropped. This
1946 /// pushes distinguishable rows into an evicting cache and a plain
1947 /// one and compares exactly that slice.
1948 #[test]
1949 fn the_rows_a_windowed_kernel_reads_are_identical_with_and_without_eviction() {
1950 let window = KvWindow::new(6, 2).expect("positive window");
1951 let (n_kv_heads, head_dim) = (2usize, 4usize);
1952 let per = n_kv_heads * head_dim;
1953 let mut evicting = KvCache::new(n_kv_heads, head_dim);
1954 evicting.arm_window(window);
1955 let mut plain = KvCache::new(n_kv_heads, head_dim);
1956
1957 for p in 0..120usize {
1958 // A row nothing else could produce, so a misplaced row is
1959 // not merely a different number but an identifiable one.
1960 let k: Vec<f32> = (0..per).map(|i| (p * 100 + i) as f32).collect();
1961 let v: Vec<f32> = k.iter().map(|x| -x).collect();
1962 evicting.push(&k, &v).expect("unbounded growth");
1963 evicting.evict_behind_window();
1964 plain.push(&k, &v).expect("unbounded growth");
1965
1966 let read = window.window().min(p + 1);
1967 let e_start = (evicting.rows() - read) * per;
1968 let p_start = (plain.rows() - read) * per;
1969 assert_eq!(
1970 &evicting.k[e_start..],
1971 &plain.k[p_start..],
1972 "K read set diverged at position {p}"
1973 );
1974 assert_eq!(
1975 &evicting.v[e_start..],
1976 &plain.v[p_start..],
1977 "V read set diverged at position {p}"
1978 );
1979 assert_eq!(evicting.positions(), plain.positions());
1980 }
1981 }
1982
1983 /// An unarmed cache is the cache this engine has always had.
1984 #[test]
1985 fn an_unarmed_cache_never_drops_a_row() {
1986 let mut cache = KvCache::new(2, 4);
1987 let step = vec![1.0f32; 8];
1988 for _ in 0..64 {
1989 cache.push(&step, &step).expect("unbounded growth");
1990 assert_eq!(cache.evict_behind_window(), 0);
1991 }
1992 assert_eq!(cache.rows(), 64);
1993 assert_eq!(cache.positions(), 64);
1994 assert!(cache.window().is_none());
1995 }
1996
1997 /// Speculative decoding rolls the cache back by the number of
1998 /// rejected draft tokens. That is representable while the target is
1999 /// still resident.
2000 #[test]
2001 fn truncate_still_works_inside_the_resident_window() {
2002 let window = KvWindow::new(8, 3).expect("positive window");
2003 let mut cache = KvCache::new(1, 2);
2004 cache.arm_window(window);
2005 let step = vec![1.0f32; 2];
2006 for _ in 0..50 {
2007 cache.push(&step, &step).expect("unbounded growth");
2008 cache.evict_behind_window();
2009 }
2010 let rows_before = cache.rows();
2011 cache.truncate(46);
2012 assert_eq!(cache.positions(), 46);
2013 assert_eq!(cache.rows(), rows_before - 4);
2014 }
2015
2016 /// ...and stops, loudly, when it is not. A cache that rolled back to
2017 /// the oldest row it happened to have would answer the next token
2018 /// out of a history with a hole in it.
2019 #[test]
2020 #[should_panic(expected = "rows are resident")]
2021 fn truncate_refuses_to_roll_back_past_what_the_window_kept() {
2022 let window = KvWindow::new(4, 1).expect("positive window");
2023 let mut cache = KvCache::new(1, 2);
2024 cache.arm_window(window);
2025 let step = vec![1.0f32; 2];
2026 for _ in 0..50 {
2027 cache.push(&step, &step).expect("unbounded growth");
2028 cache.evict_behind_window();
2029 }
2030 cache.truncate(3);
2031 }
2032
2033 /// A clone of an evicting cache is still an evicting cache. Reset
2034 /// the window on clone and `positions` becomes a row count again in
2035 /// whatever reads the clone.
2036 #[test]
2037 fn a_clone_carries_the_window() {
2038 let window = KvWindow::new(4, 1).expect("positive window");
2039 let mut cache = KvCache::new(1, 2);
2040 cache.arm_window(window);
2041 let step = vec![1.0f32; 2];
2042 for _ in 0..20 {
2043 cache.push(&step, &step).expect("unbounded growth");
2044 cache.evict_behind_window();
2045 }
2046 let copy = cache.clone();
2047 assert_eq!(copy.window(), Some(window));
2048 assert_eq!(copy.rows(), cache.rows());
2049 assert_eq!(copy.positions(), cache.positions());
2050 }
2051
2052 /// A pool-backed cache grows when its BUFFER is full, not when its
2053 /// position counter reaches the buffer's size. Those are the same
2054 /// number until something evicts, and an evicting pooled cache keyed
2055 /// on positions would acquire a block per token forever and exhaust
2056 /// the pool.
2057 #[test]
2058 fn a_pooled_windowed_cache_stops_acquiring_blocks() {
2059 let pool = Arc::new(Mutex::new(KvBlockPool::new(4, 8)));
2060 let mut cache =
2061 KvCache::with_pool(1, 2, Arc::clone(&pool), 8).expect("the pool was sized for this");
2062 cache.arm_window(KvWindow::new(4, 1).expect("positive window"));
2063 let step = vec![1.0f32; 2];
2064 for _ in 0..64 {
2065 cache
2066 .push(&step, &step)
2067 .expect("the buffer never fills again");
2068 cache.evict_behind_window();
2069 }
2070 assert_eq!(cache.positions(), 64);
2071 assert!(cache.rows() <= 5);
2072 }
2073
2074 /// **The sibling above passes for the wrong reason on its own
2075 /// sizing, so this is the same property where the trap is armed.**
2076 ///
2077 /// Eviction hands surplus CAPACITY back with `shrink_to`, which is
2078 /// where its memory saving actually comes from. For a pool-backed
2079 /// cache that is a bug rather than a saving: the pool has already
2080 /// promised these blocks and does not take them back, while `push`
2081 /// decides it needs another block by comparing rows against
2082 /// `k.capacity()`. Shrink that capacity to a window and the
2083 /// condition is true again every `slack + 1` tokens, forever -- so
2084 /// the cache draws a fresh block from the shared pool on a cadence,
2085 /// never releases one, and runs the pool dry underneath every OTHER
2086 /// request in the process. It surfaces where `push` is documented
2087 /// infallible, so the caller panics mid-answer.
2088 ///
2089 /// The sibling's cache is 8 positions against a 5-row ceiling,
2090 /// which is under the 2x threshold `shrink_to` is gated on, so it
2091 /// never reaches the shrink at all. This one is 64 positions
2092 /// against the same ceiling, and the pool is given spare blocks so
2093 /// the leak shows up as blocks quietly gone rather than only as the
2094 /// error at the end of them.
2095 #[test]
2096 fn an_evicting_pooled_cache_never_hands_its_capacity_back_to_reacquire_it() {
2097 let pool = Arc::new(Mutex::new(KvBlockPool::new(8, 24)));
2098 let mut cache =
2099 KvCache::with_pool(1, 2, Arc::clone(&pool), 64).expect("8 of the 24 blocks");
2100 let free_after_construction = pool.lock().unwrap().free_blocks();
2101 assert_eq!(free_after_construction, 16, "8 blocks cover 64 positions");
2102
2103 cache.arm_window(KvWindow::new(4, 1).expect("positive window"));
2104 let step = vec![1.0f32; 2];
2105 for _ in 0..512 {
2106 cache
2107 .push(&step, &step)
2108 .expect("a reservation made up front must not be re-made per token");
2109 cache.evict_behind_window();
2110 }
2111
2112 assert_eq!(
2113 pool.lock().unwrap().free_blocks(),
2114 free_after_construction,
2115 "the cache drew more blocks from the shared pool while evicting"
2116 );
2117 assert_eq!(cache.positions(), 512);
2118 assert!(cache.rows() <= 5);
2119 }
2120}