1pub fn kv_cache_formats() -> (&'static str, &'static str) {
15 static F: std::sync::OnceLock<(&'static str, &'static str)> = std::sync::OnceLock::new();
16 *F.get_or_init(|| {
17 let k = match std::env::var("MEMRA_KV_K").as_deref() {
18 Ok("fp8") => "fp8",
19 Ok("q8_0") | Ok("") | Err(_) => "q8_0",
20 Ok(o) => panic!("MEMRA_KV_K={o} unsupported (q8_0 | fp8)"),
21 };
22 let v = match std::env::var("MEMRA_KV_V").as_deref() {
23 Ok("q4_0") => "q4_0",
24 Ok("fp8") => "fp8",
25 Ok("q5_1") | Ok("") | Err(_) => "q5_1",
26 Ok(o) => panic!("MEMRA_KV_V={o} unsupported (q5_1 | q4_0 | fp8)"),
27 };
28 if (k, v) != ("q8_0", "q5_1") {
29 eprintln!("[memra] KV cache format: K={k} V={v} (non-default — new numeric config)");
30 }
31 (k, v)
32 })
33}
34
35pub fn kv_blk_bytes() -> (usize, usize) {
37 let (k, v) = kv_cache_formats();
38 let kb = match k {
39 "fp8" => 32,
40 _ => 34,
41 };
42 let vb = match v {
43 "q4_0" => 18,
44 "fp8" => 32,
45 _ => 24,
46 };
47 (kb, vb)
48}
49
50#[derive(Clone, Copy, Debug, PartialEq, Eq)]
56pub struct TpKvRankAllocationShape {
57 pub kv_dim_k: usize,
58 pub kv_dim_v: usize,
59 pub k_token_bytes: usize,
60 pub v_token_bytes: usize,
61 pub fixed_bytes: usize,
62}
63
64impl TpKvRankAllocationShape {
65 pub fn bytes_per_token(self) -> usize {
66 self.k_token_bytes.saturating_add(self.v_token_bytes)
67 }
68
69 pub fn allocation_bytes(self, capacity: usize) -> usize {
70 self.bytes_per_token()
71 .saturating_mul(capacity)
72 .saturating_add(self.fixed_bytes)
73 }
74}
75
76#[allow(clippy::manual_is_multiple_of)] pub fn tp_kv_rank_allocation_shape(
78 kv_dim_k: usize,
79 kv_dim_v: usize,
80 ranks: usize,
81) -> Result<TpKvRankAllocationShape, String> {
82 if ranks == 0 || kv_dim_k == 0 || kv_dim_v == 0 {
83 return Err(format!(
84 "TP KV dimensions and rank count must be nonzero: k={kv_dim_k} v={kv_dim_v} \
85 ranks={ranks}"
86 ));
87 }
88 if kv_dim_k % ranks != 0 || kv_dim_v % ranks != 0 {
89 return Err(format!(
90 "TP KV dimensions k={kv_dim_k} v={kv_dim_v} are not divisible by TP={ranks}"
91 ));
92 }
93 let local_k = kv_dim_k / ranks;
94 let local_v = kv_dim_v / ranks;
95 if !local_k.is_multiple_of(32) || !local_v.is_multiple_of(32) {
96 return Err(format!(
97 "TP KV local dimensions k={local_k} v={local_v} must be 32-aligned"
98 ));
99 }
100 let (k_block_bytes, v_block_bytes) = kv_blk_bytes();
101 let k_token_bytes = (local_k / 32)
102 .checked_mul(k_block_bytes)
103 .ok_or("TP KV K token-byte overflow")?;
104 let v_token_bytes = (local_v / 32)
105 .checked_mul(v_block_bytes)
106 .ok_or("TP KV V token-byte overflow")?;
107 Ok(TpKvRankAllocationShape {
108 kv_dim_k: local_k,
109 kv_dim_v: local_v,
110 k_token_bytes,
111 v_token_bytes,
112 fixed_bytes: 8 + 8 + std::mem::size_of::<i32>(),
115 })
116}
117
118pub fn gkv_on() -> bool {
121 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
122 *ON.get_or_init(|| {
123 std::env::var("MEMRA_GEMMA_GKV")
124 .map(|v| v != "0")
125 .unwrap_or(true)
126 })
127}
128
129pub fn wkv_on() -> bool {
133 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
134 *ON.get_or_init(|| {
135 std::env::var("MEMRA_GEMMA_WKV")
136 .map(|v| v != "0")
137 .unwrap_or_else(|_| std::env::var("MEMRA_DRAFT").is_err())
138 })
139}
140
141pub static KV_FP8_FORCE: std::sync::atomic::AtomicI8 = std::sync::atomic::AtomicI8::new(-1);
147
148pub fn kv_fp8_on() -> bool {
152 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
153 if let Some(v) = *ENV.get_or_init(|| std::env::var("MEMRA_KV_FP8").ok().map(|v| v == "1")) {
154 return v;
155 }
156 matches!(KV_FP8_FORCE.load(std::sync::atomic::Ordering::Relaxed), 1)
157}
158
159static SWA_RING_DEFAULT: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
166
167pub fn set_swa_ring_default(on: bool) {
168 SWA_RING_DEFAULT.store(on, std::sync::atomic::Ordering::Relaxed);
169}
170
171pub fn swa_ring_on() -> bool {
172 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
173 match *ENV.get_or_init(|| match std::env::var("MEMRA_SWA_RING").ok().as_deref() {
174 Some("1") => Some(true),
175 Some("0") => Some(false),
176 _ => None,
177 }) {
178 Some(forced) => forced,
179 None => SWA_RING_DEFAULT.load(std::sync::atomic::Ordering::Relaxed),
180 }
181}
182
183pub const PRIME_CHUNK_MAX_TOKENS: usize = 4096;
186const SWA_VIEW_ALIGNMENT_ROWS: usize = 32;
187
188pub fn swa_ring_rows(window: usize, max_ctx: usize) -> usize {
191 max_ctx.min(
196 window + PRIME_CHUNK_MAX_TOKENS + SWA_REWIND_SLACK_ROWS + (SWA_VIEW_ALIGNMENT_ROWS - 1),
197 )
198}
199
200pub const SWA_REWIND_SLACK_ROWS: usize = 512;
210
211pub fn swa_retain_from(first_row: usize, window: usize, base: usize) -> usize {
221 let ideal = first_row
222 .saturating_sub(window.saturating_sub(1))
223 .saturating_sub(SWA_REWIND_SLACK_ROWS)
224 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
225 ideal.max(base)
226}
227
228pub const INDEX_RING_WORKING_ROWS: usize = 5120;
256
257pub fn index_ring_rows_for(explicit: Option<usize>, max_ctx: usize) -> Option<usize> {
267 let rows = match explicit {
268 Some(0) => return None,
269 Some(n) => n,
270 None => INDEX_RING_WORKING_ROWS,
271 };
272 (rows > 0 && rows < max_ctx).then_some(rows)
273}
274
275pub fn index_ring_default_rows(max_ctx: usize) -> Option<usize> {
279 index_ring_rows_for(None, max_ctx)
280}
281
282#[allow(clippy::manual_is_multiple_of)] pub fn index_ring_take(
303 ring: usize,
304 pool: usize,
305 pools_ready: usize,
306 cur: usize,
307 remaining: usize,
308) -> Option<usize> {
309 if ring == 0 {
310 return Some(remaining);
311 }
312 debug_assert!(
313 ring % pool == 0,
314 "the effective ring is a whole number of pools"
315 );
316 let live = cur.checked_sub(pools_ready.saturating_mul(pool))?;
317 (ring > live).then(|| (ring - live).min(remaining))
318}
319
320pub fn index_ring_rows(max_ctx: usize) -> Option<usize> {
324 let explicit = std::env::var("MEMRA_DSA_INDEX_RING")
325 .ok()
326 .and_then(|v| v.trim().parse::<usize>().ok());
327 index_ring_rows_for(explicit, max_ctx)
328}
329
330#[derive(Debug, Clone, Copy, PartialEq, Eq)]
331pub struct KvRing {
332 rows: usize,
333 window: usize,
334 base: usize,
335}
336
337#[derive(Debug, Clone, Copy, PartialEq, Eq)]
338pub enum KvRingAppend {
339 Contiguous {
340 write_row: usize,
341 },
342 Rebase {
343 src_row: usize,
344 keep_rows: usize,
345 new_base: usize,
346 write_row: usize,
347 },
348}
349
350impl KvRing {
351 pub fn new(rows: usize, window: usize) -> Self {
352 assert!(window > 0 && rows > 0, "invalid SWA ring geometry");
353 Self {
354 rows,
355 window,
356 base: 0,
357 }
358 }
359
360 pub fn rows(&self) -> usize {
361 self.rows
362 }
363 pub fn base(&self) -> usize {
364 self.base
365 }
366 pub fn window(&self) -> usize {
367 self.window
368 }
369
370 pub fn append_plan(
373 &self,
374 len: usize,
375 retain_from: usize,
376 append_rows: usize,
377 ) -> Result<KvRingAppend, String> {
378 if len < self.base || retain_from < self.base || retain_from > len {
379 return Err(format!(
380 "SWA ring lapped required rows (base {}, retain {retain_from}, len {len})",
381 self.base
382 ));
383 }
384 let used = len - self.base;
385 if used > self.rows {
386 return Err(format!(
387 "SWA ring state exceeds capacity ({used} > {})",
388 self.rows
389 ));
390 }
391 if used.saturating_add(append_rows) <= self.rows {
392 return Ok(KvRingAppend::Contiguous {
393 write_row: used % self.rows,
394 });
395 }
396
397 let keep_rows = len - retain_from;
398 if keep_rows.saturating_add(append_rows) > self.rows {
399 return Err(format!(
400 "SWA ring append does not fit (keep {keep_rows} + append {append_rows} > {})",
401 self.rows
402 ));
403 }
404 Ok(KvRingAppend::Rebase {
405 src_row: retain_from - self.base,
406 keep_rows,
407 new_base: retain_from,
408 write_row: keep_rows,
409 })
410 }
411
412 pub fn apply_rebase(&mut self, new_base: usize) {
413 debug_assert!(new_base >= self.base);
414 self.base = new_base;
415 }
416
417 pub fn physical_range(
418 &self,
419 start: usize,
420 end: usize,
421 ) -> Result<std::ops::Range<usize>, String> {
422 if start < self.base || end < start || end - self.base > self.rows {
423 return Err(format!(
424 "SWA ring view [{start},{end}) is outside resident [{},{})",
425 self.base,
426 self.base + self.rows
427 ));
428 }
429 let start_row = (start - self.base) % self.rows;
430 let len = end - start;
431 debug_assert!(
432 start_row + len <= self.rows,
433 "ring view must be contiguous after rebase"
434 );
435 Ok(start_row..start_row + len)
436 }
437
438 pub fn can_rewind_to(&self, len: usize) -> bool {
440 let raw = len.saturating_sub(self.window - 1);
441 let view_start = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
442 view_start >= self.base
443 }
444
445 pub fn restore_plan(&self, len: usize) -> Result<(usize, std::ops::Range<usize>), String> {
452 let raw = len.saturating_sub(self.window.saturating_sub(1));
453 let new_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
454 let physical = self.physical_range(new_base, len)?;
455 Ok((new_base, physical))
456 }
457}
458
459pub trait KvDev {
464 fn zeros(&self, n: usize) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
465 fn uninit(&self, n: usize) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
466 fn alloc_u8(&self, n: usize) -> Result<CudaSlice<u8>, Box<dyn std::error::Error>>;
467 fn htod_i32(&self, v: &[i32]) -> Result<CudaSlice<i32>, Box<dyn std::error::Error>>;
468 fn clone_dtod(
469 &self,
470 src: &CudaSlice<f32>,
471 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
472 fn copy_into(
473 &self,
474 dst: &mut CudaSlice<f32>,
475 off: usize,
476 src: &CudaSlice<f32>,
477 len: usize,
478 ) -> Result<(), Box<dyn std::error::Error>>;
479 fn copy_range_into(
483 &self,
484 dst: &mut CudaSlice<f32>,
485 dst_off: usize,
486 src: &CudaSlice<f32>,
487 src_off: usize,
488 len: usize,
489 ) -> Result<(), Box<dyn std::error::Error>>;
490 fn set_i32_one(&self, d: &mut CudaSlice<i32>, v: i32)
491 -> Result<(), Box<dyn std::error::Error>>;
492}
493
494use cudarc::driver::CudaSlice;
495use memra_gguf::config::{LayerKind, ModelConfig};
496use memra_gguf::model_plan::{ModelPlan, ResidualTopology, StatePlan};
497
498pub struct KvLayer {
503 pub k: CudaSlice<u8>, pub v: CudaSlice<u8>, pub kv_dim_k: usize, pub kv_dim_v: usize, pub k_tok_bytes: usize, pub v_tok_bytes: usize, pub len: usize,
510 pub ring: Option<KvRing>,
513 pub len_d: CudaSlice<i32>,
517 pub base_d: Option<CudaSlice<i32>>,
526}
527
528impl KvLayer {
529 pub fn physical_rows(
530 &self,
531 start: usize,
532 end: usize,
533 ) -> Result<std::ops::Range<usize>, String> {
534 match &self.ring {
535 Some(ring) => ring.physical_range(start, end),
536 None => Ok(start..end),
537 }
538 }
539}
540
541pub struct LatentKvLayer {
554 pub rows: CudaSlice<f32>,
556 pub width: usize,
557 pub len: usize,
558 pub len_d: CudaSlice<i32>,
560 pub index_rows: Option<CudaSlice<f32>>,
610 pub index_width: usize,
611 pub index_ring_rows: Option<usize>,
615 pub index_pool_keys: Option<CudaSlice<f32>>,
640 pub index_pools_ready: usize,
642 pub index_pool: usize,
650}
651
652impl LatentKvLayer {
653 pub fn truncate_index_pool_keys(&mut self, pool: usize) {
657 if pool == 0 {
658 return;
659 }
660 self.index_pools_ready = self.index_pools_ready.min(self.len / pool);
661 }
662}
663
664pub fn index_plane_physical_row(ring_rows: usize, pool: usize, abs: usize) -> usize {
669 if ring_rows == 0 {
670 return abs;
671 }
672 debug_assert!(pool > 0, "index plane addressing requires a known pool");
673 let effective = ring_rows / pool * pool;
674 debug_assert!(effective > 0, "the effective ring holds at least one pool");
675 abs % effective
676}
677
678pub struct LatentPlaneSnapshot {
695 pub rows: CudaSlice<f32>,
697 pub width: usize,
698 pub len: usize,
699 pub index_width: usize,
701 pub index_pool: usize,
703 pub index_tail: Option<CudaSlice<f32>>,
706 pub index_pool_keys: Option<CudaSlice<f32>>,
709 pub index_pools_ready: usize,
710}
711
712impl LatentPlaneSnapshot {
713 pub fn bytes(&self) -> usize {
716 let tail = self.index_tail.as_ref().map_or(0, CudaSlice::len);
717 let keys = self.index_pool_keys.as_ref().map_or(0, CudaSlice::len);
718 (self.rows.len() + tail + keys) * std::mem::size_of::<f32>()
719 }
720}
721
722impl LatentKvLayer {
723 pub fn snapshot_plane(
731 &self,
732 e: &impl KvDev,
733 ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
734 let (len, width) = (self.len, self.width);
735 if len == 0 {
736 return Err("latent snapshot at len 0 (record the layer as absent instead)".into());
737 }
738 if self.rows.len() < len * width {
739 return Err(format!(
740 "latent plane holds {} f32 but len {len} x width {width} requires {}",
741 self.rows.len(),
742 len * width,
743 )
744 .into());
745 }
746 let mut rows = e.uninit(len * width)?;
747 e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
748 if self.index_width == 0 {
749 return Ok(LatentPlaneSnapshot {
750 rows,
751 width,
752 len,
753 index_width: 0,
754 index_pool: 0,
755 index_tail: None,
756 index_pool_keys: None,
757 index_pools_ready: 0,
758 });
759 }
760 let pool = self.index_pool;
761 if pool == 0 {
762 return Err(format!(
763 "latent snapshot: index plane (width {}) has an unresolved pool — no indexer \
764 call ran against this layer, so its derived state cannot be validated",
765 self.index_width,
766 )
767 .into());
768 }
769 let d = self.index_width / 2;
770 let pools_ready = self.index_pools_ready;
771 if pools_ready != len / pool {
772 return Err(format!(
773 "latent snapshot: index_pools_ready {pools_ready} != len/pool {} (len {len}, \
774 pool {pool}); a capture must sit at a drained call boundary or its keys \
775 violate the append-only finality invariant",
776 len / pool,
777 )
778 .into());
779 }
780 let index_pool_keys = if pools_ready > 0 {
781 let src = self
782 .index_pool_keys
783 .as_ref()
784 .ok_or("latent snapshot: pools are ready but the resident key plane is gone")?;
785 if src.len() < pools_ready * d {
786 return Err(format!(
787 "latent snapshot: resident key plane holds {} f32 but {pools_ready} pools \
788 x d {d} require {}",
789 src.len(),
790 pools_ready * d,
791 )
792 .into());
793 }
794 let mut keys = e.uninit(pools_ready * d)?;
795 e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
796 Some(keys)
797 } else {
798 None
799 };
800 let tail_rows = len - pools_ready * pool;
801 let index_tail = if tail_rows > 0 {
802 let src = self
803 .index_rows
804 .as_ref()
805 .ok_or("latent snapshot: index_width > 0 but the state plane is gone")?;
806 let ring = self.index_ring_rows.unwrap_or(0);
807 let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
810 let want = (phys + tail_rows) * self.index_width;
811 if src.len() < want {
812 return Err(format!(
813 "latent snapshot: index plane holds {} f32 but the live tail window \
814 requires {want}",
815 src.len(),
816 )
817 .into());
818 }
819 let mut tail = e.uninit(tail_rows * self.index_width)?;
820 e.copy_range_into(
821 &mut tail,
822 0,
823 src,
824 phys * self.index_width,
825 tail_rows * self.index_width,
826 )?;
827 Some(tail)
828 } else {
829 None
830 };
831 Ok(LatentPlaneSnapshot {
832 rows,
833 width,
834 len,
835 index_width: self.index_width,
836 index_pool: pool,
837 index_tail,
838 index_pool_keys,
839 index_pools_ready: pools_ready,
840 })
841 }
842
843 pub fn validate_restore(
847 &self,
848 snap: &LatentPlaneSnapshot,
849 max_ctx: usize,
850 ) -> Result<(), String> {
851 if self.len != 0 {
852 return Err("restore destination latent plane is not fresh".into());
853 }
854 if self.width != snap.width {
855 return Err(format!(
856 "snapshot width {} != destination width {}",
857 snap.width, self.width,
858 ));
859 }
860 if snap.len == 0 || snap.len > max_ctx {
861 return Err(format!("snapshot len {} outside [1,{max_ctx}]", snap.len));
862 }
863 if snap.rows.len() < snap.len * snap.width {
864 return Err(format!(
865 "snapshot rows plane holds {} f32 but len {} x width {} requires {} \
866 (truncated capture)",
867 snap.rows.len(),
868 snap.len,
869 snap.width,
870 snap.len * snap.width,
871 ));
872 }
873 if self.rows.len() < snap.len * self.width {
874 return Err(format!(
875 "destination latent plane holds {} f32 but the restore requires {}",
876 self.rows.len(),
877 snap.len * self.width,
878 ));
879 }
880 if self.index_width != snap.index_width {
881 return Err(format!(
882 "snapshot index width {} != destination {}",
883 snap.index_width, self.index_width,
884 ));
885 }
886 if snap.index_width == 0 {
887 return Ok(());
888 }
889 let pool = snap.index_pool;
890 if pool == 0 {
891 return Err("snapshot carries an index plane with an unresolved pool".into());
892 }
893 if self.index_pool != 0 && self.index_pool != pool {
894 return Err(format!(
895 "snapshot pool {pool} != destination resident pool {}",
896 self.index_pool,
897 ));
898 }
899 let d = snap.index_width / 2;
900 if snap.index_pools_ready != snap.len / pool {
901 return Err(format!(
902 "snapshot index_pools_ready {} != len/pool {} (len {}, pool {pool}): the \
903 append-only finality invariant does not hold, so its keys are stale",
904 snap.index_pools_ready,
905 snap.len / pool,
906 snap.len,
907 ));
908 }
909 match (&snap.index_pool_keys, snap.index_pools_ready) {
910 (Some(keys), ready @ 1..) => {
911 if keys.len() < ready * d {
912 return Err(format!(
913 "snapshot key plane holds {} f32 but {ready} pools x d {d} require {}",
914 keys.len(),
915 ready * d,
916 ));
917 }
918 }
919 (None, 0) => {}
920 (Some(_), 0) => return Err("snapshot carries keys for zero ready pools".into()),
921 (None, ready) => {
922 return Err(format!(
923 "snapshot claims {ready} ready pools but carries no keys"
924 ));
925 }
926 }
927 let tail_rows = snap.len - snap.index_pools_ready * pool;
928 match (&snap.index_tail, tail_rows) {
929 (Some(tail), rows @ 1..) => {
930 if tail.len() < rows * snap.index_width {
931 return Err(format!(
932 "snapshot tail holds {} f32 but {rows} rows x index width {} require {}",
933 tail.len(),
934 snap.index_width,
935 rows * snap.index_width,
936 ));
937 }
938 }
939 (None, 0) => {}
940 (Some(_), 0) => return Err("snapshot carries a tail at a pool-aligned boundary".into()),
941 (None, rows) => {
942 return Err(format!(
943 "snapshot owes {rows} live tail rows but carries none"
944 ));
945 }
946 }
947 if self.index_rows.is_none() {
948 return Err("destination declares an index plane but allocated none".into());
949 }
950 if tail_rows > 0 {
951 let ring = self.index_ring_rows.unwrap_or(0);
952 let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
953 let want = (phys + tail_rows) * self.index_width;
954 let have = self.index_rows.as_ref().map_or(0, CudaSlice::len);
955 if have < want {
956 return Err(format!(
957 "destination index plane holds {have} f32 but the tail window requires \
958 {want}",
959 ));
960 }
961 }
962 Ok(())
963 }
964
965 pub fn restore_plane(
973 &mut self,
974 e: &impl KvDev,
975 snap: &LatentPlaneSnapshot,
976 max_ctx: usize,
977 ) -> Result<(), Box<dyn std::error::Error>> {
978 self.validate_restore(snap, max_ctx)?;
979 e.copy_range_into(&mut self.rows, 0, &snap.rows, 0, snap.len * snap.width)?;
980 if snap.index_width > 0 {
981 let pool = snap.index_pool;
982 let d = snap.index_width / 2;
983 let mut keys = e.zeros(((max_ctx / pool) * d).max(1))?;
986 if let Some(src) = &snap.index_pool_keys {
987 e.copy_range_into(&mut keys, 0, src, 0, snap.index_pools_ready * d)?;
988 }
989 self.index_pool_keys = Some(keys);
990 self.index_pools_ready = snap.index_pools_ready;
991 self.index_pool = pool;
992 if let Some(tail) = &snap.index_tail {
993 let tail_rows = snap.len - snap.index_pools_ready * pool;
994 let ring = self.index_ring_rows.unwrap_or(0);
995 let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
996 let dst = self
997 .index_rows
998 .as_mut()
999 .ok_or("destination index plane vanished after validation")?;
1000 e.copy_range_into(
1001 dst,
1002 phys * self.index_width,
1003 tail,
1004 0,
1005 tail_rows * self.index_width,
1006 )?;
1007 }
1008 }
1009 self.len = snap.len;
1010 let len_i32 = i32::try_from(snap.len).map_err(|_| "latent length exceeds i32 mirror")?;
1011 e.set_i32_one(&mut self.len_d, len_i32)?;
1012 Ok(())
1013 }
1014}
1015
1016pub struct RecurLayer {
1020 pub conv_state: CudaSlice<f32>, pub ssm_state: CudaSlice<f32>, pub ssm_state_alt: CudaSlice<f32>,
1032}
1033
1034pub struct ResidentTpKvCacheRank {
1035 k: CudaSlice<u8>,
1036 v: CudaSlice<u8>,
1037 len_d: CudaSlice<i32>,
1038 base_d: Option<CudaSlice<i32>>,
1043}
1044
1045impl ResidentTpKvCacheRank {
1046 pub fn new(k: CudaSlice<u8>, v: CudaSlice<u8>, len_d: CudaSlice<i32>) -> Self {
1047 Self {
1048 k,
1049 v,
1050 len_d,
1051 base_d: None,
1052 }
1053 }
1054
1055 pub fn base_d(&self) -> Option<&CudaSlice<i32>> {
1056 self.base_d.as_ref()
1057 }
1058
1059 pub fn base_d_mut(&mut self) -> Option<&mut CudaSlice<i32>> {
1060 self.base_d.as_mut()
1061 }
1062
1063 pub fn arm_base_d(&mut self, buf: CudaSlice<i32>) {
1064 self.base_d = Some(buf);
1065 }
1066
1067 pub fn k(&self) -> &CudaSlice<u8> {
1068 &self.k
1069 }
1070
1071 pub fn v(&self) -> &CudaSlice<u8> {
1072 &self.v
1073 }
1074
1075 pub fn len_d(&self) -> &CudaSlice<i32> {
1076 &self.len_d
1077 }
1078
1079 pub fn k_mut(&mut self) -> &mut CudaSlice<u8> {
1080 &mut self.k
1081 }
1082
1083 pub fn v_mut(&mut self) -> &mut CudaSlice<u8> {
1084 &mut self.v
1085 }
1086
1087 pub fn planes_mut(&mut self) -> (&mut CudaSlice<u8>, &mut CudaSlice<u8>) {
1088 (&mut self.k, &mut self.v)
1089 }
1090
1091 #[allow(clippy::type_complexity)]
1093 pub fn planes_and_counters_mut(
1094 &mut self,
1095 ) -> (
1096 &mut CudaSlice<u8>,
1097 &mut CudaSlice<u8>,
1098 &CudaSlice<i32>,
1099 Option<&CudaSlice<i32>>,
1100 ) {
1101 (&mut self.k, &mut self.v, &self.len_d, self.base_d.as_ref())
1102 }
1103
1104 pub fn len_d_mut(&mut self) -> &mut CudaSlice<i32> {
1105 &mut self.len_d
1106 }
1107}
1108
1109#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1110pub struct TpKvTransaction {
1111 generation: u64,
1112 base_len: usize,
1113}
1114
1115impl TpKvTransaction {
1116 pub fn generation(self) -> u64 {
1117 self.generation
1118 }
1119
1120 pub fn base_len(self) -> usize {
1121 self.base_len
1122 }
1123}
1124
1125#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1126pub struct TpKvAppendPlan {
1127 transaction: TpKvTransaction,
1128 target: usize,
1129 write_row: usize,
1130 ring_append: Option<KvRingAppend>,
1131}
1132
1133impl TpKvAppendPlan {
1134 pub fn target(self) -> usize {
1135 self.target
1136 }
1137
1138 pub fn write_row(self) -> usize {
1139 self.write_row
1140 }
1141
1142 pub fn ring_append(self) -> Option<KvRingAppend> {
1143 self.ring_append
1144 }
1145}
1146
1147#[derive(Debug, PartialEq, Eq)]
1148pub struct TpKvGrowPlan {
1149 rows: usize,
1150 source_row: usize,
1151 copy_rows: usize,
1152 target_base: usize,
1153 k_bytes: usize,
1154 v_bytes: usize,
1155 source_capacity: usize,
1156 target_capacity: usize,
1157 ring_window: Option<usize>,
1158 target_physical_rows: usize,
1159 kv_dim_k: usize,
1160 kv_dim_v: usize,
1161 k_tok_bytes: usize,
1162 v_tok_bytes: usize,
1163 ranks: usize,
1164 next_generation: u64,
1165}
1166
1167impl TpKvGrowPlan {
1168 pub fn rows(&self) -> usize {
1169 self.rows
1170 }
1171
1172 pub fn source_row(&self) -> usize {
1173 self.source_row
1174 }
1175
1176 pub fn copy_rows(&self) -> usize {
1177 self.copy_rows
1178 }
1179
1180 pub fn k_bytes(&self) -> usize {
1181 self.k_bytes
1182 }
1183
1184 pub fn v_bytes(&self) -> usize {
1185 self.v_bytes
1186 }
1187}
1188
1189#[derive(Clone, Debug, PartialEq, Eq)]
1190struct TpKvTransactionState {
1191 committed_len: usize,
1192 staged_len: usize,
1193 next_generation: u64,
1194 active: Option<TpKvTransaction>,
1195}
1196
1197impl TpKvTransactionState {
1198 fn new() -> Self {
1199 Self {
1200 committed_len: 0,
1201 staged_len: 0,
1202 next_generation: 1,
1203 active: None,
1204 }
1205 }
1206
1207 fn begin(&mut self) -> Result<TpKvTransaction, String> {
1208 if let Some(active) = self.active {
1209 return Err(format!(
1210 "TP KV transaction generation {} is already active at base {}",
1211 active.generation, active.base_len
1212 ));
1213 }
1214 if self.staged_len != self.committed_len {
1215 return Err(format!(
1216 "TP KV cache is half-committed: staged {} != committed {}",
1217 self.staged_len, self.committed_len
1218 ));
1219 }
1220 let transaction = TpKvTransaction {
1221 generation: self.next_generation,
1222 base_len: self.committed_len,
1223 };
1224 self.next_generation = self
1225 .next_generation
1226 .checked_add(1)
1227 .ok_or("TP KV transaction generation overflow")?;
1228 self.active = Some(transaction);
1229 Ok(transaction)
1230 }
1231
1232 fn validate(&self, transaction: TpKvTransaction) -> Result<(), String> {
1233 if self.active != Some(transaction) {
1234 return Err(format!(
1235 "stale TP KV transaction generation {} at base {}",
1236 transaction.generation, transaction.base_len
1237 ));
1238 }
1239 if transaction.base_len != self.committed_len {
1240 return Err(format!(
1241 "TP KV transaction base {} != committed length {}",
1242 transaction.base_len, self.committed_len
1243 ));
1244 }
1245 Ok(())
1246 }
1247
1248 fn append_target(
1249 &self,
1250 transaction: TpKvTransaction,
1251 rows: usize,
1252 capacity: usize,
1253 ) -> Result<usize, String> {
1254 self.validate(transaction)?;
1255 if rows == 0 {
1256 return Err("TP KV append must contain at least one row".into());
1257 }
1258 let target = self
1259 .staged_len
1260 .checked_add(rows)
1261 .ok_or("TP KV staged length overflow")?;
1262 if target > capacity {
1263 return Err(format!(
1264 "TP KV append exceeds capacity: {target} > {capacity}"
1265 ));
1266 }
1267 Ok(target)
1268 }
1269
1270 fn publish_append(
1271 &mut self,
1272 transaction: TpKvTransaction,
1273 target: usize,
1274 ) -> Result<(), String> {
1275 self.validate(transaction)?;
1276 if target <= self.staged_len {
1277 return Err(format!(
1278 "TP KV append target {target} must exceed staged length {}",
1279 self.staged_len
1280 ));
1281 }
1282 self.staged_len = target;
1283 Ok(())
1284 }
1285
1286 fn commit_target(
1287 &self,
1288 transaction: TpKvTransaction,
1289 accepted_rows: usize,
1290 ) -> Result<usize, String> {
1291 self.validate(transaction)?;
1292 let staged_rows = self
1293 .staged_len
1294 .checked_sub(transaction.base_len)
1295 .ok_or("TP KV staged length precedes its transaction base")?;
1296 if accepted_rows > staged_rows {
1297 return Err(format!(
1298 "TP KV commit accepts {accepted_rows} rows from a {staged_rows}-row transaction"
1299 ));
1300 }
1301 transaction
1302 .base_len
1303 .checked_add(accepted_rows)
1304 .ok_or_else(|| "TP KV committed length overflow".to_string())
1305 }
1306
1307 fn publish_finalize(
1308 &mut self,
1309 transaction: TpKvTransaction,
1310 target: usize,
1311 ) -> Result<(), String> {
1312 self.validate(transaction)?;
1313 if target < transaction.base_len || target > self.staged_len {
1314 return Err(format!(
1315 "TP KV finalize target {target} outside transaction range {}..={}",
1316 transaction.base_len, self.staged_len
1317 ));
1318 }
1319 self.committed_len = target;
1320 self.staged_len = target;
1321 self.active = None;
1322 Ok(())
1323 }
1324
1325 fn rewind(&mut self, target: usize, capacity: usize) -> Result<(), String> {
1326 if target > capacity {
1327 return Err(format!(
1328 "TP KV rewind target {target} exceeds capacity {capacity}"
1329 ));
1330 }
1331 self.committed_len = target;
1332 self.staged_len = target;
1333 self.active = None;
1334 Ok(())
1335 }
1336}
1337
1338pub struct ResidentTpKvCache {
1339 ranks: Vec<ResidentTpKvCacheRank>,
1340 kv_dim_k: usize,
1341 kv_dim_v: usize,
1342 k_tok_bytes: usize,
1343 v_tok_bytes: usize,
1344 capacity: usize,
1345 ring: Option<KvRing>,
1346 state: TpKvTransactionState,
1347}
1348
1349impl ResidentTpKvCache {
1350 #[allow(clippy::too_many_arguments)]
1351 pub fn new(
1352 ranks: Vec<ResidentTpKvCacheRank>,
1353 kv_dim_k: usize,
1354 kv_dim_v: usize,
1355 k_tok_bytes: usize,
1356 v_tok_bytes: usize,
1357 capacity: usize,
1358 ) -> Self {
1359 Self::new_inner(
1360 ranks,
1361 kv_dim_k,
1362 kv_dim_v,
1363 k_tok_bytes,
1364 v_tok_bytes,
1365 capacity,
1366 None,
1367 )
1368 }
1369
1370 #[allow(clippy::too_many_arguments)]
1371 pub fn new_swa(
1372 ranks: Vec<ResidentTpKvCacheRank>,
1373 kv_dim_k: usize,
1374 kv_dim_v: usize,
1375 k_tok_bytes: usize,
1376 v_tok_bytes: usize,
1377 capacity: usize,
1378 window: usize,
1379 ) -> Self {
1380 Self::new_inner(
1381 ranks,
1382 kv_dim_k,
1383 kv_dim_v,
1384 k_tok_bytes,
1385 v_tok_bytes,
1386 capacity,
1387 Some(KvRing::new(swa_ring_rows(window, capacity), window)),
1388 )
1389 }
1390
1391 #[allow(clippy::too_many_arguments)]
1392 fn new_inner(
1393 ranks: Vec<ResidentTpKvCacheRank>,
1394 kv_dim_k: usize,
1395 kv_dim_v: usize,
1396 k_tok_bytes: usize,
1397 v_tok_bytes: usize,
1398 capacity: usize,
1399 ring: Option<KvRing>,
1400 ) -> Self {
1401 Self {
1402 ranks,
1403 kv_dim_k,
1404 kv_dim_v,
1405 k_tok_bytes,
1406 v_tok_bytes,
1407 capacity,
1408 ring,
1409 state: TpKvTransactionState::new(),
1410 }
1411 }
1412
1413 pub fn begin_transaction(&mut self) -> Result<TpKvTransaction, String> {
1414 self.state.begin()
1415 }
1416
1417 pub fn committed_len(&self) -> usize {
1418 self.state.committed_len
1419 }
1420
1421 pub fn staged_len(&self) -> usize {
1422 self.state.staged_len
1423 }
1424
1425 pub fn capacity(&self) -> usize {
1426 self.capacity
1427 }
1428
1429 pub fn physical_capacity(&self) -> usize {
1430 self.ring
1431 .as_ref()
1432 .map(KvRing::rows)
1433 .unwrap_or(self.capacity)
1434 }
1435
1436 pub fn ring_window(&self) -> Option<usize> {
1437 self.ring.as_ref().map(KvRing::window)
1438 }
1439
1440 pub fn ring_base(&self) -> Option<usize> {
1441 self.ring.as_ref().map(KvRing::base)
1442 }
1443
1444 pub fn physical_range(
1445 &self,
1446 start: usize,
1447 end: usize,
1448 ) -> Result<std::ops::Range<usize>, String> {
1449 match &self.ring {
1450 Some(ring) => ring.physical_range(start, end),
1451 None => {
1452 if end < start || end > self.capacity {
1453 return Err(format!(
1454 "TP KV linear view [{start},{end}) exceeds capacity {}",
1455 self.capacity
1456 ));
1457 }
1458 Ok(start..end)
1459 }
1460 }
1461 }
1462
1463 pub fn can_rewind_to(&self, target: usize) -> bool {
1464 target <= self.capacity
1465 && self
1466 .ring
1467 .as_ref()
1468 .is_none_or(|ring| ring.can_rewind_to(target))
1469 }
1470
1471 pub fn kv_dim_k(&self) -> usize {
1472 self.kv_dim_k
1473 }
1474
1475 pub fn kv_dim_v(&self) -> usize {
1476 self.kv_dim_v
1477 }
1478
1479 pub fn k_tok_bytes(&self) -> usize {
1480 self.k_tok_bytes
1481 }
1482
1483 pub fn v_tok_bytes(&self) -> usize {
1484 self.v_tok_bytes
1485 }
1486
1487 pub fn ranks_len(&self) -> usize {
1488 self.ranks.len()
1489 }
1490
1491 pub fn rank(&self, rank: usize) -> Option<&ResidentTpKvCacheRank> {
1492 self.ranks.get(rank)
1493 }
1494
1495 pub fn rank_mut(&mut self, rank: usize) -> Option<&mut ResidentTpKvCacheRank> {
1496 self.ranks.get_mut(rank)
1497 }
1498
1499 pub fn ranks(&self) -> &[ResidentTpKvCacheRank] {
1500 &self.ranks
1501 }
1502
1503 pub fn ranks_mut(&mut self) -> &mut [ResidentTpKvCacheRank] {
1504 &mut self.ranks
1505 }
1506
1507 pub fn prepare_grow(
1508 &self,
1509 target_capacity: usize,
1510 rows: usize,
1511 ) -> Result<TpKvGrowPlan, String> {
1512 if let Some(active) = self.state.active {
1513 return Err(format!(
1514 "TP KV grow refuses active transaction generation {} at base {}",
1515 active.generation, active.base_len
1516 ));
1517 }
1518 if self.state.staged_len != self.state.committed_len {
1519 return Err(format!(
1520 "TP KV grow requires quiescent state, got committed/staged={}/{}",
1521 self.state.committed_len, self.state.staged_len
1522 ));
1523 }
1524 if target_capacity <= self.capacity {
1525 return Err(format!(
1526 "TP KV grow target capacity {target_capacity} must exceed source capacity {}",
1527 self.capacity
1528 ));
1529 }
1530 if target_capacity > i32::MAX as usize {
1531 return Err(format!(
1532 "TP KV grow target capacity {target_capacity} exceeds i32 device mirrors"
1533 ));
1534 }
1535 if rows > self.state.committed_len {
1536 return Err(format!(
1537 "TP KV grow rows {rows} exceed committed length {}",
1538 self.state.committed_len
1539 ));
1540 }
1541 let (source_row, copy_rows, target_base, ring_window, target_physical_rows) =
1542 match &self.ring {
1543 Some(ring) => {
1544 let raw = rows.saturating_sub(ring.window().saturating_sub(1));
1545 let target_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1546 let physical = ring.physical_range(target_base, rows)?;
1547 (
1548 physical.start,
1549 physical.len(),
1550 target_base,
1551 Some(ring.window()),
1552 swa_ring_rows(ring.window(), target_capacity),
1553 )
1554 }
1555 None => (0, rows, 0, None, target_capacity),
1556 };
1557 let k_bytes = copy_rows
1558 .checked_mul(self.k_tok_bytes)
1559 .ok_or("TP KV grow K byte extent overflow")?;
1560 let v_bytes = copy_rows
1561 .checked_mul(self.v_tok_bytes)
1562 .ok_or("TP KV grow V byte extent overflow")?;
1563 Ok(TpKvGrowPlan {
1564 rows,
1565 source_row,
1566 copy_rows,
1567 target_base,
1568 k_bytes,
1569 v_bytes,
1570 source_capacity: self.capacity,
1571 target_capacity,
1572 ring_window,
1573 target_physical_rows,
1574 kv_dim_k: self.kv_dim_k,
1575 kv_dim_v: self.kv_dim_v,
1576 k_tok_bytes: self.k_tok_bytes,
1577 v_tok_bytes: self.v_tok_bytes,
1578 ranks: self.ranks.len(),
1579 next_generation: self.state.next_generation,
1580 })
1581 }
1582
1583 pub fn publish_grow(&mut self, plan: TpKvGrowPlan) -> Result<(), String> {
1584 if self.state != TpKvTransactionState::new() {
1585 return Err(format!(
1586 "TP KV grow target must be fresh, got committed/staged={}/{} active={}",
1587 self.state.committed_len,
1588 self.state.staged_len,
1589 self.state.active.is_some()
1590 ));
1591 }
1592 if self.capacity != plan.target_capacity
1593 || self.capacity <= plan.source_capacity
1594 || self.kv_dim_k != plan.kv_dim_k
1595 || self.kv_dim_v != plan.kv_dim_v
1596 || self.k_tok_bytes != plan.k_tok_bytes
1597 || self.v_tok_bytes != plan.v_tok_bytes
1598 || self.ranks.len() != plan.ranks
1599 || self.ring.as_ref().map(KvRing::window) != plan.ring_window
1600 || self.physical_capacity() != plan.target_physical_rows
1601 {
1602 return Err("TP KV grow target layout does not match its source plan".into());
1603 }
1604 if plan.rows > self.capacity {
1605 return Err(format!(
1606 "TP KV grow rows {} exceed target capacity {}",
1607 plan.rows, self.capacity
1608 ));
1609 }
1610 if let Some(ring) = self.ring.as_mut() {
1611 let mut target_ring = *ring;
1612 target_ring.apply_rebase(plan.target_base);
1613 if !target_ring.can_rewind_to(plan.rows) {
1614 return Err(format!(
1615 "TP KV grow target ring base {} cannot expose committed length {}",
1616 target_ring.base(),
1617 plan.rows
1618 ));
1619 }
1620 *ring = target_ring;
1621 }
1622 self.state.committed_len = plan.rows;
1623 self.state.staged_len = plan.rows;
1624 self.state.next_generation = plan.next_generation;
1625 self.state.active = None;
1626 Ok(())
1627 }
1628
1629 pub fn prepare_append(
1630 &self,
1631 transaction: TpKvTransaction,
1632 rows: usize,
1633 ) -> Result<TpKvAppendPlan, String> {
1634 let target = self.state.append_target(transaction, rows, self.capacity)?;
1635 let ring_append = self
1636 .ring
1637 .as_ref()
1638 .map(|ring| {
1639 let staged_retain =
1640 target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1641 let rollback_retain = transaction
1642 .base_len
1643 .saturating_sub(ring.window().saturating_sub(1))
1644 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1645 ring.append_plan(
1646 self.state.staged_len,
1647 staged_retain.min(rollback_retain),
1648 rows,
1649 )
1650 })
1651 .transpose()?;
1652 let write_row = match ring_append {
1653 Some(KvRingAppend::Contiguous { write_row })
1654 | Some(KvRingAppend::Rebase { write_row, .. }) => write_row,
1655 None => self.state.staged_len,
1656 };
1657 Ok(TpKvAppendPlan {
1658 transaction,
1659 target,
1660 write_row,
1661 ring_append,
1662 })
1663 }
1664
1665 pub fn peek_append_ring(&self, rows: usize) -> Result<(usize, bool), String> {
1669 let target = self.state.staged_len + rows;
1670 let plan = self
1671 .ring
1672 .as_ref()
1673 .map(|ring| {
1674 let staged_retain =
1675 target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1676 let rollback_retain = self
1677 .state
1678 .staged_len
1679 .saturating_sub(ring.window().saturating_sub(1))
1680 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1681 ring.append_plan(
1682 self.state.staged_len,
1683 staged_retain.min(rollback_retain),
1684 rows,
1685 )
1686 })
1687 .transpose()?;
1688 Ok(match plan {
1689 Some(KvRingAppend::Contiguous { write_row }) => (write_row, false),
1690 Some(KvRingAppend::Rebase { write_row, .. }) => (write_row, true),
1691 None => (self.state.staged_len, false),
1692 })
1693 }
1694
1695 pub fn publish_append_rebase(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1696 self.state.validate(plan.transaction)?;
1697 match (self.ring.as_mut(), plan.ring_append) {
1698 (
1699 Some(ring),
1700 Some(KvRingAppend::Rebase {
1701 new_base,
1702 keep_rows,
1703 ..
1704 }),
1705 ) => {
1706 if keep_rows > ring.rows() {
1707 return Err(format!(
1708 "TP KV ring rebase keeps {keep_rows} rows in {} physical rows",
1709 ring.rows()
1710 ));
1711 }
1712 let mut target_ring = *ring;
1713 target_ring.apply_rebase(new_base);
1714 if !target_ring.can_rewind_to(plan.transaction.base_len) {
1715 return Err(format!(
1716 "TP KV ring rebase to {new_base} laps transaction base {}",
1717 plan.transaction.base_len
1718 ));
1719 }
1720 *ring = target_ring;
1721 Ok(())
1722 }
1723 (Some(_), Some(KvRingAppend::Contiguous { .. })) | (None, None) => Ok(()),
1724 _ => Err("TP KV append plan does not match cache ring layout".into()),
1725 }
1726 }
1727
1728 pub fn publish_append_plan(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1729 if let Some(KvRingAppend::Rebase { new_base, .. }) = plan.ring_append {
1730 if self.ring.as_ref().map(KvRing::base) != Some(new_base) {
1731 return Err(format!(
1732 "TP KV append rebase {new_base} was not published before its state"
1733 ));
1734 }
1735 }
1736 self.state.publish_append(plan.transaction, plan.target)
1737 }
1738
1739 pub fn publish_hydration(
1740 &mut self,
1741 logical_len: usize,
1742 resident_start: usize,
1743 ) -> Result<(), Box<dyn std::error::Error>> {
1744 if self.state != TpKvTransactionState::new() {
1745 return Err("TP KV hydration target must be fresh".into());
1746 }
1747 if resident_start > logical_len || logical_len > self.capacity {
1748 return Err(format!(
1749 "TP KV hydration range [{resident_start},{logical_len}) exceeds capacity {}",
1750 self.capacity
1751 )
1752 .into());
1753 }
1754 match self.ring.as_mut() {
1755 Some(ring) => {
1756 let rows = logical_len - resident_start;
1757 if rows > ring.rows() {
1758 return Err(format!(
1759 "TP KV hydration requires {rows} rows in a {}-row ring",
1760 ring.rows()
1761 )
1762 .into());
1763 }
1764 let mut hydrated_ring = *ring;
1765 hydrated_ring.apply_rebase(resident_start);
1766 if !hydrated_ring.can_rewind_to(logical_len) {
1767 return Err(format!(
1768 "TP KV hydration base {resident_start} cannot expose logical length \
1769 {logical_len}"
1770 )
1771 .into());
1772 }
1773 *ring = hydrated_ring;
1774 }
1775 None if resident_start != 0 => {
1776 return Err("linear TP KV hydration must start at absolute row zero".into());
1777 }
1778 None => {}
1779 }
1780 self.rewind_to(logical_len)
1781 }
1782
1783 pub fn append_target(
1784 &self,
1785 transaction: TpKvTransaction,
1786 rows: usize,
1787 ) -> Result<usize, String> {
1788 self.state.append_target(transaction, rows, self.capacity)
1789 }
1790
1791 pub fn publish_append(
1792 &mut self,
1793 transaction: TpKvTransaction,
1794 target: usize,
1795 ) -> Result<(), String> {
1796 self.state.publish_append(transaction, target)
1797 }
1798
1799 pub fn commit_target(
1800 &self,
1801 transaction: TpKvTransaction,
1802 accepted_rows: usize,
1803 ) -> Result<usize, String> {
1804 self.state.commit_target(transaction, accepted_rows)
1805 }
1806
1807 pub fn validate_transaction(&self, transaction: TpKvTransaction) -> Result<(), String> {
1808 self.state.validate(transaction)
1809 }
1810
1811 pub fn publish_finalize(
1812 &mut self,
1813 transaction: TpKvTransaction,
1814 target: usize,
1815 ) -> Result<(), String> {
1816 if !self.can_rewind_to(target) {
1817 return Err(format!(
1818 "TP KV finalize target {target} is outside the resident cache window/capacity"
1819 ));
1820 }
1821 self.state.publish_finalize(transaction, target)
1822 }
1823
1824 pub fn rewind_to(&mut self, target: usize) -> Result<(), Box<dyn std::error::Error>> {
1825 if !self.can_rewind_to(target) {
1826 return Err(format!(
1827 "TP KV rewind target {target} is outside the resident cache window/capacity"
1828 )
1829 .into());
1830 }
1831 let target_i32 =
1832 i32::try_from(target).map_err(|_| "TP KV length exceeds i32 device mirror")?;
1833 for rank in &mut self.ranks {
1834 let stream = rank.len_d.stream().clone();
1835 stream.memcpy_htod(&[target_i32], &mut rank.len_d)?;
1836 }
1837 self.state.rewind(target, self.capacity)?;
1838 Ok(())
1839 }
1840}
1841
1842pub struct Cache {
1843 pub kv: Vec<Option<KvLayer>>,
1844 pub recur: Vec<Option<RecurLayer>>,
1845 pub latent: Vec<Option<LatentKvLayer>>,
1848 pub tp_kv: Vec<Option<ResidentTpKvCache>>,
1851 pub glm5_tp_recur: Vec<Option<Vec<RecurLayer>>>,
1858 pub glm5_tp_latent_peer: Vec<Option<Vec<LatentKvLayer>>>,
1863 pub pos: usize,
1864 pub max_ctx: usize,
1865 pub tainted: bool,
1868 pub last_logits_dev: Option<CudaSlice<f32>>,
1876 pub dflash_taps: Option<DflashTapSink>,
1881 pub hc_taps: Option<HcTapSink>,
1889}
1890
1891#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1895enum FullAttentionClass {
1896 Ordinary,
1897 GemmaGlobal,
1898 GemmaWindowed,
1899}
1900
1901fn full_attention_class(plan: &ModelPlan, il: u32) -> FullAttentionClass {
1902 let layer = plan
1903 .layers
1904 .iter()
1905 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
1906 .find(|layer| layer.index == il)
1907 .unwrap_or_else(|| panic!("ModelPlan has no layer {il}"));
1908 if !matches!(layer.residual, ResidualTopology::Gemma { .. }) {
1909 return FullAttentionClass::Ordinary;
1910 }
1911 match layer.state {
1912 StatePlan::SlidingKvCache { .. } => FullAttentionClass::GemmaWindowed,
1913 StatePlan::KvCache { .. } => FullAttentionClass::GemmaGlobal,
1914 _ => panic!("Gemma layer {il} does not declare a KV-cache state"),
1915 }
1916}
1917
1918fn full_attention_kv_layout(
1919 cfg: &ModelConfig,
1920 plan: &ModelPlan,
1921 il: u32,
1922) -> (usize, usize, usize, usize) {
1923 debug_assert_eq!(cfg.layer_kind(il), LayerKind::FullAttention);
1924 let class = full_attention_class(plan, il);
1925 let n_head_kv = cfg.n_head_kv as usize;
1926 let (kv_dim_k, kv_dim_v) = match class {
1927 FullAttentionClass::GemmaGlobal | FullAttentionClass::GemmaWindowed => {
1928 let g = cfg
1929 .gemma4
1930 .as_ref()
1931 .expect("Gemma ModelPlan layer requires Gemma cache geometry");
1932 let hd = match class {
1933 FullAttentionClass::GemmaWindowed => g.key_length_swa,
1934 FullAttentionClass::GemmaGlobal => g.key_length_global,
1935 FullAttentionClass::Ordinary => unreachable!(),
1936 } as usize;
1937 let d = match g.head_count_kv.get(il as usize) {
1946 Some(n) => hd * *n as usize,
1947 None => hd * n_head_kv,
1948 };
1949 (d, d)
1950 }
1951 FullAttentionClass::Ordinary => (
1952 cfg.head_dim_k as usize * n_head_kv,
1953 cfg.head_dim_v as usize * n_head_kv,
1954 ),
1955 };
1956 assert!(
1957 kv_dim_k % 32 == 0 && kv_dim_v % 32 == 0,
1958 "KVQUANT requires per-layer kv_dim_k%32==0 && kv_dim_v%32==0 \
1959 (layer {il}: k={kv_dim_k} v={kv_dim_v})"
1960 );
1961 let (kbb, vbb) = kv_blk_bytes();
1962 let g4_global_fp8 = gkv_on() && class == FullAttentionClass::GemmaGlobal;
1963 let g4_windowed_fp8 = wkv_on() && class == FullAttentionClass::GemmaWindowed;
1964 let qwen_fp8 = kv_fp8_on() && class == FullAttentionClass::Ordinary;
1965 let (kbb_l, vbb_l) = if g4_global_fp8 || g4_windowed_fp8 || qwen_fp8 {
1966 (32, 32)
1967 } else {
1968 (kbb, vbb)
1969 };
1970 (kv_dim_k, kv_dim_v, kbb_l, vbb_l)
1971}
1972
1973fn kv_plane_allocation_bytes(rows: usize, token_bytes: usize) -> usize {
1974 rows * token_bytes + 8
1975}
1976
1977pub fn cache_bytes_per_token(cfg: &ModelConfig) -> usize {
1984 cache_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
1985}
1986
1987pub fn cache_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
1991 let plan = ModelPlan::compile(cfg).expect("cache sizing requires a compilable ModelPlan");
1992 cache_bytes_per_token_for_plan(cfg, &plan, lo, hi)
1993}
1994
1995pub fn cache_bytes_per_token_for_plan(
1996 cfg: &ModelConfig,
1997 plan: &ModelPlan,
1998 lo: usize,
1999 hi: usize,
2000) -> usize {
2001 assert!(
2002 lo <= hi && hi <= cfg.n_layer as usize,
2003 "cache layer range out of bounds"
2004 );
2005 let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2006 let full_attn: usize = (lo as u32..hi as u32)
2007 .filter(|&il| cfg.layer_kind(il) == LayerKind::FullAttention)
2008 .filter(|&il| shared == 0 || il < cfg.n_layer - shared)
2009 .map(|il| {
2010 let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, il);
2011 (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2012 })
2013 .sum();
2014 full_attn + latent_kv_bytes_per_token_for_plan(cfg, plan, lo, hi)
2015}
2016
2017pub fn latent_kv_bytes_per_token_for_plan(
2047 cfg: &ModelConfig,
2048 plan: &ModelPlan,
2049 lo: usize,
2050 hi: usize,
2051) -> usize {
2052 let ring_disabled = std::env::var("MEMRA_DSA_INDEX_RING")
2053 .ok()
2054 .and_then(|v| v.trim().parse::<usize>().ok())
2055 == Some(0);
2056 plan.layers
2057 .iter()
2058 .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2059 .map(|layer| match layer.state {
2060 StatePlan::LatentKvCache { width, index_width } => {
2061 let latent = width as usize * std::mem::size_of::<f32>();
2062 let index_width = index_width as usize;
2063 let pool = cfg
2064 .glm5
2065 .as_ref()
2066 .map(|g| g.index_kpool as usize)
2067 .filter(|&p| p > 0)
2068 .unwrap_or(1);
2069 let pool_keys = if index_width > 0 {
2071 (index_width / 2) * std::mem::size_of::<f32>() / pool
2072 } else {
2073 0
2074 };
2075 let flat_plane = if index_width > 0 && ring_disabled {
2076 index_width * std::mem::size_of::<f32>()
2077 } else {
2078 0
2079 };
2080 latent + pool_keys + flat_plane
2081 }
2082 _ => 0,
2083 })
2084 .sum()
2085}
2086
2087pub fn cache_ring_bytes_per_token(cfg: &ModelConfig) -> usize {
2090 cache_ring_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2091}
2092
2093pub fn cache_ring_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2095 assert!(
2096 lo <= hi && hi <= cfg.n_layer as usize,
2097 "cache layer range out of bounds"
2098 );
2099 let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2100 return 0;
2101 };
2102 cache_ring_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2103}
2104
2105pub fn cache_ring_bytes_per_token_for_plan(
2106 cfg: &ModelConfig,
2107 plan: &ModelPlan,
2108 lo: usize,
2109 hi: usize,
2110) -> usize {
2111 let total = plan.layers.len() + plan.mtp_blocks.len();
2112 assert!(
2113 lo <= hi && hi <= total,
2114 "cache plan layer range out of bounds"
2115 );
2116 if !swa_ring_on() {
2117 return 0;
2118 }
2119 let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2120 plan.layers
2121 .iter()
2122 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2123 .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2124 .filter(|layer| {
2125 matches!(
2126 layer.state,
2127 memra_gguf::model_plan::StatePlan::SlidingKvCache { .. }
2128 )
2129 })
2130 .filter(|layer| shared == 0 || layer.index < cfg.n_layer - shared)
2131 .map(|layer| {
2132 let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, layer.index);
2133 (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2134 })
2135 .sum()
2136}
2137
2138pub fn cache_ring_row_cap(cfg: &ModelConfig) -> usize {
2140 let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2141 return 0;
2142 };
2143 cache_ring_row_cap_for_plan(&plan)
2144}
2145
2146pub fn cache_ring_row_cap_for_plan(plan: &memra_gguf::model_plan::ModelPlan) -> usize {
2147 if !swa_ring_on() {
2148 return 0;
2149 }
2150 plan.layers
2151 .iter()
2152 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2153 .filter_map(|layer| match layer.state {
2154 memra_gguf::model_plan::StatePlan::SlidingKvCache { window, .. } => {
2155 Some(window as usize)
2156 }
2157 _ => None,
2158 })
2159 .map(|window| swa_ring_rows(window, usize::MAX))
2160 .max()
2161 .unwrap_or(0)
2162}
2163
2164pub struct DflashTapSink {
2167 pub layer_ids: Vec<usize>,
2168 pub buf: CudaSlice<f32>,
2169 pub hidden: usize,
2170 pub t: usize,
2171 pub base: usize,
2174}
2175
2176pub struct HcTapSink {
2183 pub layer_ids: Vec<usize>,
2185 pub rows: Vec<f32>,
2187 pub hidden: usize,
2188 pub t: usize,
2190 pub base: usize,
2193 pub dev: Vec<Option<CudaSlice<f32>>>,
2201 pub device_stage: bool,
2206}
2207
2208impl HcTapSink {
2209 pub fn new(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2210 let n_taps = layer_ids.len();
2211 Self {
2212 layer_ids,
2213 rows: vec![0.0; t * n_taps * hidden],
2214 hidden,
2215 t,
2216 base: 0,
2217 dev: (0..n_taps).map(|_| None).collect(),
2218 device_stage: false,
2219 }
2220 }
2221
2222 pub fn new_device_staged(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2225 Self {
2226 device_stage: true,
2227 ..Self::new(layer_ids, hidden, t)
2228 }
2229 }
2230}
2231
2232pub struct CacheSnapshot {
2257 pub kv_len: Vec<Option<usize>>, pub tp_kv_len: Vec<Option<usize>>, pub conv: Vec<Option<CudaSlice<f32>>>, pub ssm: Vec<Option<CudaSlice<f32>>>,
2261 pub pos: usize,
2262}
2263
2264impl Cache {
2265 pub fn ensure_usable(&self, path: &str) -> Result<(), Box<dyn std::error::Error>> {
2266 if self.tainted {
2267 return Err(format!(
2268 "{path}: cache was tainted by a failed pipeline wave and cannot be reused"
2269 )
2270 .into());
2271 }
2272 Ok(())
2273 }
2274
2275 pub fn mark_tainted(&mut self) {
2276 self.tainted = true;
2277 self.last_logits_dev = None;
2278 self.dflash_taps = None;
2279 }
2280
2281 pub fn new(
2283 e: &impl KvDev,
2284 cfg: &ModelConfig,
2285 max_ctx: usize,
2286 ) -> Result<Self, Box<dyn std::error::Error>> {
2287 Self::new_inner(&|_| e, cfg, None, max_ctx)
2288 }
2289
2290 pub fn new_planned(
2291 e: &impl KvDev,
2292 cfg: &ModelConfig,
2293 plan: &memra_gguf::model_plan::ModelPlan,
2294 max_ctx: usize,
2295 ) -> Result<Self, Box<dyn std::error::Error>> {
2296 Self::new_inner(&|_| e, cfg, Some(plan), max_ctx)
2297 }
2298
2299 pub fn new_pp2(
2304 dev0: &dyn KvDev,
2305 dev1: &dyn KvDev,
2306 split: usize,
2307 cfg: &ModelConfig,
2308 max_ctx: usize,
2309 ) -> Result<Self, Box<dyn std::error::Error>> {
2310 Self::new_inner(
2311 &|il| if il < split { dev0 } else { dev1 },
2312 cfg,
2313 None,
2314 max_ctx,
2315 )
2316 }
2317
2318 pub fn new_ppn(
2324 devs: &[&dyn KvDev],
2325 fence: &[usize],
2326 cfg: &ModelConfig,
2327 max_ctx: usize,
2328 ) -> Result<Self, Box<dyn std::error::Error>> {
2329 assert_eq!(
2330 devs.len() + 1,
2331 fence.len(),
2332 "ppn cache: devs vs fence mismatch"
2333 );
2334 let pick = |il: usize| -> &dyn KvDev {
2335 let s = match fence[1..fence.len() - 1].binary_search(&il) {
2336 Ok(k) => k + 1,
2337 Err(k) => k,
2338 };
2339 devs[s.min(devs.len() - 1)]
2340 };
2341 Self::new_inner(&pick, cfg, None, max_ctx)
2342 }
2343
2344 pub fn new_ppn_planned(
2345 devs: &[&dyn KvDev],
2346 fence: &[usize],
2347 cfg: &ModelConfig,
2348 plan: &memra_gguf::model_plan::ModelPlan,
2349 max_ctx: usize,
2350 ) -> Result<Self, Box<dyn std::error::Error>> {
2351 assert_eq!(
2352 devs.len() + 1,
2353 fence.len(),
2354 "ppn cache: devs vs fence mismatch"
2355 );
2356 let pick = |il: usize| -> &dyn KvDev {
2357 let stage = match fence[1..fence.len() - 1].binary_search(&il) {
2358 Ok(index) => index + 1,
2359 Err(index) => index,
2360 };
2361 devs[stage.min(devs.len() - 1)]
2362 };
2363 Self::new_inner(&pick, cfg, Some(plan), max_ctx)
2364 }
2365
2366 fn new_inner<'a>(
2369 pick: &dyn Fn(usize) -> &'a dyn KvDev,
2370 cfg: &ModelConfig,
2371 plan: Option<&memra_gguf::model_plan::ModelPlan>,
2372 max_ctx: usize,
2373 ) -> Result<Self, Box<dyn std::error::Error>> {
2374 let fallback_plan = if plan.is_none() {
2375 Some(ModelPlan::compile(cfg)?)
2376 } else {
2377 None
2378 };
2379 let plan = plan
2380 .or(fallback_plan.as_ref())
2381 .expect("cache allocation requires a ModelPlan");
2382 let n = cfg.n_layer as usize;
2383 let mut kv = Vec::with_capacity(n);
2384 let mut recur = Vec::with_capacity(n);
2385 let mut latent = Vec::with_capacity(n);
2386 let head_dim_k = cfg.head_dim_k as usize;
2387 let head_dim_v = cfg.head_dim_v as usize;
2388 for il in 0..cfg.n_layer {
2389 let e = pick(il as usize);
2391 let layer = plan
2392 .layers
2393 .iter()
2394 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2395 .find(|layer| layer.index == il)
2396 .ok_or_else(|| format!("cache ModelPlan has no layer {il}"))?;
2397 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2402 if g4_shared > 0 && il >= cfg.n_layer - g4_shared {
2403 kv.push(None);
2404 recur.push(None);
2405 latent.push(None);
2406 continue;
2407 }
2408 match layer.state {
2409 StatePlan::KvCache { .. } | StatePlan::SlidingKvCache { .. } => {
2410 assert!(
2415 head_dim_k.is_multiple_of(32) && head_dim_v.is_multiple_of(32),
2416 "KVQUANT requires head_dim_k%32==0 && head_dim_v%32==0 \
2417 (layer {il}: k={head_dim_k} v={head_dim_v})"
2418 );
2419 let (kv_dim_k, kv_dim_v, kbb_l, vbb_l) =
2422 full_attention_kv_layout(cfg, plan, il);
2423 let k_tok_bytes = (kv_dim_k / 32) * kbb_l;
2424 let v_tok_bytes = (kv_dim_v / 32) * vbb_l;
2425 let planned_window = plan
2426 .layers
2427 .iter()
2428 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2429 .find(|layer| layer.index == il)
2430 .and_then(|layer| match layer.state {
2431 StatePlan::SlidingKvCache { window, .. } => Some(window),
2432 _ => None,
2433 });
2434 let ring = if swa_ring_on() {
2435 planned_window.map(|window| {
2436 let window = window as usize;
2437 KvRing::new(swa_ring_rows(window, max_ctx), window)
2438 })
2439 } else {
2440 None
2441 };
2442 let alloc_rows = ring.as_ref().map(KvRing::rows).unwrap_or(max_ctx);
2443 kv.push(Some(KvLayer {
2444 k: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, k_tok_bytes))?,
2448 v: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, v_tok_bytes))?,
2449 kv_dim_k,
2450 kv_dim_v,
2451 k_tok_bytes,
2452 v_tok_bytes,
2453 len: 0,
2454 ring,
2455 len_d: e.htod_i32(&[0])?,
2456 base_d: None,
2457 }));
2458 recur.push(None);
2459 latent.push(None);
2460 }
2461 StatePlan::Recurrent {
2462 conv_width,
2463 conv_kernel,
2464 state_width,
2465 } => {
2466 kv.push(None);
2467 recur.push(Some(RecurLayer {
2468 conv_state: e.zeros(
2469 conv_width as usize * (conv_kernel as usize).saturating_sub(1),
2470 )?,
2471 ssm_state: e.zeros(state_width as usize)?,
2472 ssm_state_alt: e.zeros(state_width as usize)?,
2473 }));
2474 latent.push(None);
2475 }
2476 StatePlan::LatentKvCache { width, index_width } => {
2477 let width = width as usize;
2481 assert!(
2482 width > 0,
2483 "layer {il}: LatentKvCache width must be positive"
2484 );
2485 kv.push(None);
2486 recur.push(None);
2487 let index_width = index_width as usize;
2488 let index_ring = if index_width == 0 {
2492 None
2493 } else {
2494 index_ring_rows(max_ctx)
2495 };
2496 let index_rows = match index_width {
2497 0 => None,
2498 w => Some(e.zeros(index_ring.unwrap_or(max_ctx) * w)?),
2499 };
2500 latent.push(Some(LatentKvLayer {
2501 rows: e.zeros(max_ctx * width)?,
2502 width,
2503 len: 0,
2504 len_d: e.htod_i32(&[0])?,
2505 index_rows,
2506 index_width,
2507 index_ring_rows: index_ring,
2508 index_pool_keys: None,
2511 index_pools_ready: 0,
2512 index_pool: 0,
2513 }));
2514 }
2515 ref state => {
2516 return Err(format!(
2517 "native cache allocator has no implementation for layer {il} state {state:?}"
2518 )
2519 .into());
2520 }
2521 }
2522 }
2523 Ok(Cache {
2524 kv,
2525 recur,
2526 latent,
2527 tp_kv: (0..n).map(|_| None).collect(),
2528 glm5_tp_recur: (0..n).map(|_| None).collect(),
2529 glm5_tp_latent_peer: (0..n).map(|_| None).collect(),
2530 pos: 0,
2531 max_ctx,
2532 tainted: false,
2533 dflash_taps: None,
2534 hc_taps: None,
2535 last_logits_dev: None,
2536 })
2537 }
2538
2539 pub fn has_swa_ring(&self) -> bool {
2540 self.kv.iter().flatten().any(|layer| layer.ring.is_some())
2541 || self
2542 .tp_kv
2543 .iter()
2544 .flatten()
2545 .any(|layer| layer.ring_window().is_some())
2546 }
2547
2548 pub fn can_rollback(&self, snap: &CacheSnapshot, accept_len: usize) -> bool {
2549 let local = self
2550 .kv
2551 .iter()
2552 .zip(&snap.kv_len)
2553 .all(|(layer, saved)| match (layer, saved) {
2554 (Some(layer), Some(saved)) => layer
2555 .ring
2556 .as_ref()
2557 .is_none_or(|ring| ring.can_rewind_to(saved + accept_len)),
2558 _ => true,
2559 });
2560 let tensor = self
2561 .tp_kv
2562 .iter()
2563 .zip(&snap.tp_kv_len)
2564 .all(|(layer, saved)| match (layer, saved) {
2565 (Some(layer), Some(saved)) => saved
2566 .checked_add(accept_len)
2567 .is_some_and(|target| layer.can_rewind_to(target)),
2568 _ => true,
2569 });
2570 local && tensor
2571 }
2572
2573 pub fn snapshot(&self, e: &impl KvDev) -> Result<CacheSnapshot, Box<dyn std::error::Error>> {
2577 if self.glm5_tp_recur.iter().any(Option::is_some)
2581 || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2582 {
2583 return Err(
2584 "cache snapshot is unwired for glm5 TP-2 rank state (MEMRA_GLM5_TP): \
2585 per-rank planes are not carried by CacheSnapshot"
2586 .into(),
2587 );
2588 }
2589 self.ensure_usable("cache snapshot")?;
2590 let n = self.kv.len();
2591 let mut kv_len = Vec::with_capacity(n);
2592 let mut tp_kv_len = Vec::with_capacity(n);
2593 let mut conv = Vec::with_capacity(n);
2594 let mut ssm = Vec::with_capacity(n);
2595 for il in 0..n {
2596 match &self.kv[il] {
2597 Some(kvl) => kv_len.push(Some(kvl.len)),
2598 None => kv_len.push(None),
2599 }
2600 tp_kv_len.push(
2601 self.tp_kv[il]
2602 .as_ref()
2603 .map(ResidentTpKvCache::committed_len),
2604 );
2605 match &self.recur[il] {
2606 Some(rl) => {
2607 conv.push(Some(e.clone_dtod(&rl.conv_state)?));
2608 ssm.push(Some(e.clone_dtod(&rl.ssm_state)?));
2609 }
2610 None => {
2611 conv.push(None);
2612 ssm.push(None);
2613 }
2614 }
2615 }
2616 Ok(CacheSnapshot {
2617 kv_len,
2618 tp_kv_len,
2619 conv,
2620 ssm,
2621 pos: self.pos,
2622 })
2623 }
2624
2625 pub fn snapshot_into(
2630 &self,
2631 e: &impl KvDev,
2632 snap: &mut CacheSnapshot,
2633 ) -> Result<(), Box<dyn std::error::Error>> {
2634 if self.glm5_tp_recur.iter().any(Option::is_some)
2635 || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2636 {
2637 return Err("cache snapshot_into is unwired for glm5 TP-2 rank state \
2638 (MEMRA_GLM5_TP): per-rank planes are not carried by CacheSnapshot"
2639 .into());
2640 }
2641 self.ensure_usable("cache snapshot refresh")?;
2642 let n = self.kv.len();
2643 for il in 0..n {
2644 snap.kv_len[il] = self.kv[il].as_ref().map(|kvl| kvl.len);
2645 snap.tp_kv_len[il] = self.tp_kv[il]
2646 .as_ref()
2647 .map(ResidentTpKvCache::committed_len);
2648 if let Some(rl) = &self.recur[il] {
2649 let dc = snap.conv[il]
2650 .as_mut()
2651 .expect("snapshot_into: shape mismatch (conv)");
2652 let ds = snap.ssm[il]
2653 .as_mut()
2654 .expect("snapshot_into: shape mismatch (ssm)");
2655 let (cn, sn) = (rl.conv_state.len(), rl.ssm_state.len());
2656 e.copy_into(dc, 0, &rl.conv_state, cn)?;
2657 e.copy_into(ds, 0, &rl.ssm_state, sn)?;
2658 }
2659 }
2660 snap.pos = self.pos;
2661 Ok(())
2662 }
2663
2664 pub fn rollback(
2672 &mut self,
2673 e: &impl KvDev,
2674 snap: &CacheSnapshot,
2675 accept_len: usize,
2676 ) -> Result<(), Box<dyn std::error::Error>> {
2677 self.ensure_usable("cache rollback")?;
2678 if !self.can_rollback(snap, accept_len) {
2679 return Err(
2680 "SWA ring rewind checkpoint has been lapped; full re-prime required".into(),
2681 );
2682 }
2683 for il in 0..self.kv.len() {
2684 if let (Some(kvl), Some(saved)) = (self.kv[il].as_mut(), snap.kv_len[il]) {
2685 kvl.len = saved + accept_len;
2686 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
2691 }
2692 if let (Some(kvl), Some(saved)) = (self.tp_kv[il].as_mut(), snap.tp_kv_len[il]) {
2693 kvl.rewind_to(saved + accept_len)?;
2694 }
2695 if let Some(rl) = self.recur[il].as_mut() {
2696 if let Some(c) = &snap.conv[il] {
2697 e.copy_into(&mut rl.conv_state, 0, c, c.len())?;
2698 }
2699 if let Some(s) = &snap.ssm[il] {
2700 e.copy_into(&mut rl.ssm_state, 0, s, s.len())?;
2701 }
2702 }
2703 }
2704 self.pos = snap.pos;
2705 Ok(())
2706 }
2707}
2708
2709#[cfg(test)]
2710mod tp_transaction_tests {
2711 use super::{
2712 Cache, INDEX_RING_WORKING_ROWS, KvRingAppend, ResidentTpKvCache, TpKvTransactionState,
2713 index_ring_default_rows, index_ring_rows_for, index_ring_take, tp_kv_rank_allocation_shape,
2714 };
2715
2716 const GLM5_NEXT_POOL: usize = 4;
2719 const GLM5_NEXT_STATE_ROW_BYTES: usize = 2 * 128 * 4;
2721 const RING_BYTES_PER_LAYER: usize = INDEX_RING_WORKING_ROWS * GLM5_NEXT_STATE_ROW_BYTES;
2724
2725 #[test]
2744 fn the_derived_ring_serves_a_monolithic_prime_of_the_whole_configured_context() {
2745 for max_ctx in [8192usize, 262_144, 1 << 20] {
2748 let rows = index_ring_default_rows(max_ctx).unwrap_or_else(|| {
2749 panic!("the ring must engage at max_ctx {max_ctx}: it is where the saving is")
2750 });
2751 let ring = rows / GLM5_NEXT_POOL * GLM5_NEXT_POOL;
2754
2755 let mut cur = 0usize;
2757 let mut pools_ready = 0usize;
2758 let mut steps = 0usize;
2759 while cur < max_ctx {
2760 let take = index_ring_take(ring, GLM5_NEXT_POOL, pools_ready, cur, max_ctx - cur)
2761 .unwrap_or_else(|| {
2762 panic!(
2763 "MEMRA_CTX={max_ctx}: the {ring}-row ring refused a monolithic prime \
2764 after {cur} of {max_ctx} tokens ({}% of the configured context). \
2765 USABLE CONTEXT MUST BE AT LEAST CONFIGURED CONTEXT. This is the \
2766 4630-of-8192 regression, in arithmetic.",
2767 cur * 100 / max_ctx
2768 )
2769 });
2770 assert!(
2771 take > 0,
2772 "MEMRA_CTX={max_ctx}: the drain made no progress at row {cur}: a zero take \
2773 is an infinite loop in the engine, not a refusal"
2774 );
2775 cur += take;
2776 pools_ready = cur / GLM5_NEXT_POOL;
2777 steps += 1;
2778 assert!(
2779 steps <= max_ctx,
2780 "MEMRA_CTX={max_ctx}: the drain did not terminate"
2781 );
2782 }
2783 assert_eq!(cur, max_ctx, "the whole prompt must be appended");
2784
2785 let ring_bytes = rows * GLM5_NEXT_STATE_ROW_BYTES;
2791 let flat_bytes = max_ctx * GLM5_NEXT_STATE_ROW_BYTES;
2792 assert_eq!(
2793 ring_bytes, RING_BYTES_PER_LAYER,
2794 "MEMRA_CTX={max_ctx}: the ring books {rows} rows, not the context-independent \
2795 {INDEX_RING_WORKING_ROWS}. A plane that tracks max_ctx is the flat plane \
2796 wearing a modulus"
2797 );
2798 assert!(
2799 rows < max_ctx,
2800 "MEMRA_CTX={max_ctx}: a ring of {rows} rows is not shorter than the flat plane \
2801 it replaces, so it would not engage at all"
2802 );
2803 assert!(
2807 ring_bytes <= 16 << 20,
2808 "MEMRA_CTX={max_ctx}: {} MiB per MLA layer, over glm5_next's 12 of them. The ring \
2809 exists to delete 11.94 GiB; a working set this large is not paying for itself",
2810 ring_bytes >> 20
2811 );
2812 println!(
2813 "MEMRA_CTX={max_ctx}: ring {rows} rows (effective {ring}), monolithic prime of \
2814 {max_ctx} tokens admitted in {steps} drain step(s); plane {} MiB/layer vs flat \
2815 {} MiB/layer",
2816 ring_bytes >> 20,
2817 flat_bytes >> 20
2818 );
2819 }
2820 }
2821
2822 fn empty_tp_cache(capacity: usize) -> ResidentTpKvCache {
2823 ResidentTpKvCache::new(Vec::new(), 128, 128, 136, 96, capacity)
2824 }
2825
2826 #[test]
2827 fn step_tp8_rank_allocation_matches_the_official_kv_geometry() {
2828 let shape = tp_kv_rank_allocation_shape(8 * 128, 8 * 128, 8).unwrap();
2829 assert_eq!((shape.kv_dim_k, shape.kv_dim_v), (128, 128));
2830 assert_eq!((shape.k_token_bytes, shape.v_token_bytes), (136, 96));
2831 assert_eq!(shape.bytes_per_token(), 232);
2832 assert_eq!(shape.fixed_bytes, 20);
2833 assert_eq!(shape.allocation_bytes(262_144), 232 * 262_144 + 20);
2834 }
2835
2836 #[test]
2837 fn tp_rank_allocation_refuses_non_divisible_and_non_block_aligned_shards() {
2838 assert!(tp_kv_rank_allocation_shape(1024, 1024, 3).is_err());
2839 assert!(tp_kv_rank_allocation_shape(1024, 1024, 64).is_err());
2840 assert!(tp_kv_rank_allocation_shape(0, 1024, 8).is_err());
2841 }
2842
2843 #[test]
2844 fn partial_commit_publishes_only_the_accepted_prefix() {
2845 let mut state = TpKvTransactionState::new();
2846 let transaction = state.begin().unwrap();
2847 let staged = state.append_target(transaction, 3, 8).unwrap();
2848 state.publish_append(transaction, staged).unwrap();
2849 assert_eq!(state.committed_len, 0);
2850 assert_eq!(state.staged_len, 3);
2851
2852 let committed = state.commit_target(transaction, 2).unwrap();
2853 state.publish_finalize(transaction, committed).unwrap();
2854 assert_eq!(state.committed_len, 2);
2855 assert_eq!(state.staged_len, 2);
2856 assert!(state.active.is_none());
2857 assert!(state.validate(transaction).is_err());
2858 }
2859
2860 #[test]
2861 fn rollback_restores_the_committed_boundary() {
2862 let mut state = TpKvTransactionState::new();
2863 let first = state.begin().unwrap();
2864 let staged = state.append_target(first, 1, 8).unwrap();
2865 state.publish_append(first, staged).unwrap();
2866 let committed = state.commit_target(first, 1).unwrap();
2867 state.publish_finalize(first, committed).unwrap();
2868
2869 let speculative = state.begin().unwrap();
2870 let staged = state.append_target(speculative, 2, 8).unwrap();
2871 state.publish_append(speculative, staged).unwrap();
2872 assert_eq!(state.committed_len, 1);
2873 assert_eq!(state.staged_len, 3);
2874 state
2875 .publish_finalize(speculative, speculative.base_len)
2876 .unwrap();
2877 assert_eq!(state.committed_len, 1);
2878 assert_eq!(state.staged_len, 1);
2879 assert!(state.validate(speculative).is_err());
2880 }
2881
2882 #[test]
2883 fn index_ring_sizing_is_pure_and_carries_no_per_call_t() {
2884 let rows = INDEX_RING_WORKING_ROWS;
2887 assert_eq!(index_ring_rows_for(None, 1 << 20), Some(rows));
2888 assert_eq!(index_ring_default_rows(1 << 20), Some(rows));
2889 assert_eq!(index_ring_rows_for(None, 4096), None);
2892 assert_eq!(index_ring_rows_for(None, rows), None);
2893 assert_eq!(index_ring_rows_for(None, rows + 1), Some(rows));
2894
2895 for max_ctx in [8192usize, 262_144, 1 << 20] {
2900 assert_eq!(
2901 index_ring_rows_for(None, max_ctx),
2902 Some(INDEX_RING_WORKING_ROWS),
2903 "the derived ring must not vary with the configured context"
2904 );
2905 }
2906
2907 assert_eq!(index_ring_rows_for(Some(0), 1 << 20), None);
2910 assert_eq!(index_ring_rows_for(Some(16), 64), Some(16));
2911 assert_eq!(index_ring_rows_for(Some(64), 64), None);
2912 }
2913
2914 #[test]
2916 fn index_ring_take_drains_instead_of_bounding_the_call() {
2917 const POOL: usize = GLM5_NEXT_POOL;
2918 assert_eq!(index_ring_take(0, POOL, 0, 0, 1 << 20), Some(1 << 20));
2920 assert_eq!(index_ring_take(64, POOL, 0, 0, 1024), Some(64));
2923 for cur in 0..64usize {
2926 let ready = cur / POOL;
2927 let take = index_ring_take(64, POOL, ready, cur, 1024).expect("never lapses");
2928 assert!(
2929 (64 - POOL + 1..=64).contains(&take),
2930 "cur {cur}: take {take} outside the guaranteed progress band"
2931 );
2932 }
2933 assert_eq!(index_ring_take(64, POOL, 4, 16, 1), Some(1));
2935 assert_eq!(index_ring_take(16, POOL, 0, 64, 1), None);
2938 assert_eq!(index_ring_take(16, POOL, 0, 16, 1), None);
2939 assert_eq!(index_ring_take(16, POOL, 0, 15, 1), Some(1));
2940 }
2941
2942 #[test]
2943 fn rejects_nested_stale_and_out_of_range_actions() {
2944 let mut state = TpKvTransactionState::new();
2945 let transaction = state.begin().unwrap();
2946 assert!(state.begin().is_err());
2947 assert!(state.append_target(transaction, 0, 2).is_err());
2948 assert!(state.append_target(transaction, 3, 2).is_err());
2949 let staged = state.append_target(transaction, 2, 2).unwrap();
2950 state.publish_append(transaction, staged).unwrap();
2951 assert!(state.commit_target(transaction, 3).is_err());
2952 state.publish_finalize(transaction, 0).unwrap();
2953 assert!(state.publish_append(transaction, 1).is_err());
2954 }
2955
2956 #[test]
2957 fn rewind_resets_visibility_and_invalidates_an_active_transaction() {
2958 let mut state = TpKvTransactionState::new();
2959 let transaction = state.begin().unwrap();
2960 let staged = state.append_target(transaction, 3, 8).unwrap();
2961 state.publish_append(transaction, staged).unwrap();
2962 state.rewind(1, 8).unwrap();
2963 assert_eq!(state.committed_len, 1);
2964 assert_eq!(state.staged_len, 1);
2965 assert!(state.active.is_none());
2966 assert!(state.validate(transaction).is_err());
2967 assert!(state.rewind(9, 8).is_err());
2968 }
2969
2970 #[test]
2971 fn grow_preserves_generation_and_publishes_only_the_checkpoint_prefix() {
2972 let mut source = empty_tp_cache(8);
2973 let first = source.begin_transaction().unwrap();
2974 let staged = source.append_target(first, 5).unwrap();
2975 source.publish_append(first, staged).unwrap();
2976 let committed = source.commit_target(first, 5).unwrap();
2977 source.publish_finalize(first, committed).unwrap();
2978
2979 let rolled_back = source.begin_transaction().unwrap();
2980 source
2981 .publish_finalize(rolled_back, rolled_back.base_len())
2982 .unwrap();
2983 let plan = source.prepare_grow(16, 3).unwrap();
2984 assert_eq!(plan.rows(), 3);
2985 assert_eq!(plan.k_bytes(), 3 * 136);
2986 assert_eq!(plan.v_bytes(), 3 * 96);
2987
2988 let mut target = empty_tp_cache(16);
2989 target.publish_grow(plan).unwrap();
2990 assert_eq!(target.committed_len(), 3);
2991 assert_eq!(target.staged_len(), 3);
2992 assert_eq!(target.capacity(), 16);
2993 let next = target.begin_transaction().unwrap();
2994 assert_eq!(next.generation(), rolled_back.generation() + 1);
2995 assert_eq!(next.base_len(), 3);
2996 }
2997
2998 #[test]
2999 fn grow_refuses_active_source_and_invalid_target_state_or_layout() {
3000 let mut active = empty_tp_cache(8);
3001 active.begin_transaction().unwrap();
3002 assert!(active.prepare_grow(16, 0).is_err());
3003
3004 let mut source = empty_tp_cache(8);
3005 source.rewind_to(5).unwrap();
3006 assert!(source.prepare_grow(8, 5).is_err());
3007 assert!(source.prepare_grow(16, 6).is_err());
3008 let plan = source.prepare_grow(16, 4).unwrap();
3009
3010 let mut wrong_layout = ResidentTpKvCache::new(Vec::new(), 128, 128, 144, 96, 16);
3011 assert!(wrong_layout.publish_grow(plan).is_err());
3012
3013 let plan = source.prepare_grow(16, 4).unwrap();
3014 let mut dirty_target = empty_tp_cache(16);
3015 dirty_target.rewind_to(1).unwrap();
3016 assert!(dirty_target.publish_grow(plan).is_err());
3017 }
3018
3019 #[test]
3020 fn swa_transaction_rebase_preserves_the_rollback_window() {
3021 let mut cache = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3022 assert_eq!(cache.physical_capacity(), 32 + 4096 + 512 + 31);
3023 cache.publish_hydration(8762, 4096).unwrap();
3027 assert_eq!(cache.ring_base(), Some(4096));
3028
3029 let transaction = cache.begin_transaction().unwrap();
3030 let plan = cache.prepare_append(transaction, 10).unwrap();
3031 assert_eq!(plan.target(), 8772);
3032 assert_eq!(plan.write_row(), 58);
3033 assert_eq!(
3034 plan.ring_append(),
3035 Some(KvRingAppend::Rebase {
3036 src_row: 4608,
3037 keep_rows: 58,
3038 new_base: 8704,
3039 write_row: 58,
3040 })
3041 );
3042 cache.publish_append_rebase(plan).unwrap();
3043 cache.publish_append_plan(plan).unwrap();
3044 assert_eq!(cache.ring_base(), Some(8704));
3045 assert_eq!(cache.physical_range(8740, 8772).unwrap(), 36..68);
3046
3047 let rollback = cache.commit_target(transaction, 0).unwrap();
3048 cache.publish_finalize(transaction, rollback).unwrap();
3049 assert_eq!((cache.committed_len(), cache.staged_len()), (8762, 8762));
3050 assert!(cache.rewind_to(8200).is_err());
3051 }
3052
3053 #[test]
3054 fn swa_grow_normalizes_only_the_live_prefix_and_preserves_generation() {
3055 let mut source = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3056 source.publish_hydration(8250, 4096).unwrap();
3057 let transaction = source.begin_transaction().unwrap();
3058 source
3059 .publish_finalize(transaction, transaction.base_len())
3060 .unwrap();
3061
3062 let plan = source.prepare_grow(20_000, 8250).unwrap();
3063 assert_eq!(plan.source_row(), 4096);
3064 assert_eq!(plan.copy_rows(), 58);
3065 assert_eq!(plan.k_bytes(), 58 * 136);
3066 assert_eq!(plan.v_bytes(), 58 * 96);
3067
3068 let mut target = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 20_000, 32);
3069 target.publish_grow(plan).unwrap();
3070 assert_eq!(target.ring_base(), Some(8192));
3071 assert_eq!((target.committed_len(), target.staged_len()), (8250, 8250));
3072 assert_eq!(target.physical_range(8192, 8250).unwrap(), 0..58);
3073 let next = target.begin_transaction().unwrap();
3074 assert_eq!(next.generation(), transaction.generation() + 1);
3075 }
3076
3077 #[test]
3078 fn cache_reports_a_materialized_distributed_swa_ring() {
3079 let mut cache = Cache {
3080 kv: Vec::new(),
3081 recur: Vec::new(),
3082 latent: Vec::new(),
3083 tp_kv: vec![None],
3084 glm5_tp_recur: vec![None],
3085 glm5_tp_latent_peer: vec![None],
3086 pos: 0,
3087 max_ctx: 10_000,
3088 tainted: false,
3089 dflash_taps: None,
3090 hc_taps: None,
3091 last_logits_dev: None,
3092 };
3093 assert!(!cache.has_swa_ring());
3094 cache.tp_kv[0] = Some(ResidentTpKvCache::new_swa(
3095 Vec::new(),
3096 128,
3097 128,
3098 136,
3099 96,
3100 10_000,
3101 512,
3102 ));
3103 assert!(cache.has_swa_ring());
3104 }
3105}
3106
3107#[cfg(test)]
3108mod swa_ring_tests {
3109 use super::{
3110 KvRing, KvRingAppend, PRIME_CHUNK_MAX_TOKENS, SWA_REWIND_SLACK_ROWS,
3111 SWA_VIEW_ALIGNMENT_ROWS, kv_plane_allocation_bytes, swa_retain_from, swa_ring_rows,
3112 };
3113
3114 #[test]
3115 fn allocation_rows_cover_window_max_prime_and_alignment_slack() {
3116 assert_eq!(swa_ring_rows(512, 262_144), 512 + 4096 + 512 + 31);
3117 assert_eq!(swa_ring_rows(512, 4096), 4096);
3118 assert_eq!(
3119 kv_plane_allocation_bytes(5151, 1088),
3120 5151 * 1088 + 8,
3121 "the Step35 session plane allocates ring rows plus the existing tail pad",
3122 );
3123 }
3124
3125 #[test]
3138 fn retain_grants_rewind_slack_but_never_asks_below_base() {
3139 const WINDOW: usize = 512;
3140 let rows = swa_ring_rows(WINDOW, 262_144);
3141 let len = rows;
3142 let aligned = |pos: usize| (pos - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3143
3144 let retain = swa_retain_from(len, WINDOW, 0);
3147 assert!(retain <= aligned(len) - SWA_REWIND_SLACK_ROWS);
3148 let mut ring = KvRing::new(rows, WINDOW);
3149 ring.apply_rebase(retain);
3150 assert!(
3151 ring.can_rewind_to(len - 1),
3152 "a one-token rewind must survive the rebase"
3153 );
3154 assert!(ring.can_rewind_to(len - SWA_REWIND_SLACK_ROWS));
3155
3156 assert!(len - retain + PRIME_CHUNK_MAX_TOKENS <= rows);
3158
3159 for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3163 assert!(
3164 ring.can_rewind_to(len - depth),
3165 "a {depth}-row rewind must be resident, not clamped away",
3166 );
3167 }
3168
3169 let base = ring.base();
3176 for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3177 let window_start = (len - depth - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3178 assert!(
3179 window_start >= base,
3180 "after a {depth}-row rewind the window starts at {window_start}, below base \
3181 {base} — clamping here would serve rows the ring no longer holds (the pos-8661 \
3182 all-NaN case)",
3183 );
3184 assert!(swa_retain_from(len - depth, WINDOW, base) >= base);
3185 }
3186
3187 let past = len - (SWA_REWIND_SLACK_ROWS + WINDOW);
3190 let past_start = (past - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3191 assert!(
3192 past_start < base,
3193 "beyond the headroom the window must fall below base"
3194 );
3195 assert!(
3196 !ring.can_rewind_to(past),
3197 "and can_rewind_to must refuse it"
3198 );
3199 }
3200
3201 #[test]
3202 fn ring_matches_flat_bytes_before_wrap() {
3203 let ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3204 let flat: Vec<u32> = (0..1024).collect();
3205 let mut physical = vec![u32::MAX; ring.rows()];
3206 let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, flat.len()).unwrap()
3207 else {
3208 panic!("first append unexpectedly wrapped")
3209 };
3210 physical[write_row..write_row + flat.len()].copy_from_slice(&flat);
3211 let view = ring.physical_range(0, flat.len()).unwrap();
3212 assert_eq!(&physical[view], flat.as_slice());
3213 }
3214
3215 #[test]
3216 fn wrap_rebases_the_exact_aligned_prime_view() {
3217 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3218 let flat: Vec<u32> = (0..8192).collect();
3219 let mut physical = vec![u32::MAX; ring.rows()];
3220 let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, 4096).unwrap() else {
3221 panic!("first prime chunk unexpectedly wrapped")
3222 };
3223 physical[write_row..write_row + 4096].copy_from_slice(&flat[..4096]);
3224
3225 let off = (4096usize - (512 - 1)) & !31usize;
3226 let KvRingAppend::Rebase {
3227 src_row,
3228 keep_rows,
3229 new_base,
3230 write_row,
3231 } = ring.append_plan(4096, off, 4096).unwrap()
3232 else {
3233 panic!("second prime chunk did not wrap")
3234 };
3235 let retained = physical[src_row..src_row + keep_rows].to_vec();
3236 physical[..keep_rows].copy_from_slice(&retained);
3237 ring.apply_rebase(new_base);
3238 physical[write_row..write_row + 4096].copy_from_slice(&flat[4096..8192]);
3239
3240 let view = ring.physical_range(off, 8192).unwrap();
3241 assert_eq!(&physical[view], &flat[off..8192]);
3242 assert_eq!(ring.base(), off);
3243 }
3244
3245 #[test]
3246 fn rewind_declines_once_the_required_window_was_lapped() {
3247 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3248 let KvRingAppend::Rebase { new_base, .. } = ring.append_plan(4096, 3584, 4096).unwrap()
3249 else {
3250 panic!("expected wrap")
3251 };
3252 ring.apply_rebase(new_base);
3253 assert!(ring.can_rewind_to(4095));
3254 assert!(!ring.can_rewind_to(4094));
3255 assert!(!ring.can_rewind_to(0));
3256 }
3257
3258 #[test]
3263 fn restore_plan_copies_the_window_not_the_absolute_length() {
3264 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3265 let (base, phys) = ring.restore_plan(400).unwrap();
3267 assert_eq!((base, phys), (0, 0..400));
3268
3269 let mut live = 0usize;
3272 while live < 40_960 {
3273 let retain = swa_retain_from(live, 512, ring.base());
3274 if let KvRingAppend::Rebase { new_base, .. } =
3275 ring.append_plan(live, retain, 4096).unwrap()
3276 {
3277 ring.apply_rebase(new_base);
3278 }
3279 live += 4096;
3280 }
3281 assert!(ring.base() > 0, "a 40k walk must have lapped the ring");
3282 let (base, phys) = ring.restore_plan(live).unwrap();
3283 assert_eq!(base, (live - (512 - 1)) & !31usize);
3284 assert!(
3285 base >= ring.base(),
3286 "the plan must stay above the ring floor"
3287 );
3288 assert_eq!(phys.len(), live - base);
3289 assert!(
3290 phys.end <= ring.rows(),
3291 "the copy must fit the physical buffer ({} rows), got {:?}",
3292 ring.rows(),
3293 phys
3294 );
3295
3296 assert!(ring.restore_plan(400).is_err());
3298 }
3299}