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
722pub struct LatentTailCapture {
733 pub len: usize,
735 pub width: usize,
737 pub index_width: usize,
738 pub index_pool: usize,
740 pub index_pools_ready: usize,
743 pub index_tail: Option<CudaSlice<f32>>,
747}
748
749impl LatentTailCapture {
750 pub fn bytes(&self) -> usize {
752 self.index_tail.as_ref().map_or(0, CudaSlice::len) * std::mem::size_of::<f32>()
753 }
754}
755
756impl LatentKvLayer {
757 pub fn snapshot_plane(
765 &self,
766 e: &impl KvDev,
767 ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
768 let (len, width) = (self.len, self.width);
769 if len == 0 {
770 return Err("latent snapshot at len 0 (record the layer as absent instead)".into());
771 }
772 if self.rows.len() < len * width {
773 return Err(format!(
774 "latent plane holds {} f32 but len {len} x width {width} requires {}",
775 self.rows.len(),
776 len * width,
777 )
778 .into());
779 }
780 let mut rows = e.uninit(len * width)?;
781 e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
782 if self.index_width == 0 {
783 return Ok(LatentPlaneSnapshot {
784 rows,
785 width,
786 len,
787 index_width: 0,
788 index_pool: 0,
789 index_tail: None,
790 index_pool_keys: None,
791 index_pools_ready: 0,
792 });
793 }
794 let pool = self.index_pool;
795 if pool == 0 {
796 return Err(format!(
797 "latent snapshot: index plane (width {}) has an unresolved pool — no indexer \
798 call ran against this layer, so its derived state cannot be validated",
799 self.index_width,
800 )
801 .into());
802 }
803 let d = self.index_width / 2;
804 let pools_ready = self.index_pools_ready;
805 if pools_ready != len / pool {
806 return Err(format!(
807 "latent snapshot: index_pools_ready {pools_ready} != len/pool {} (len {len}, \
808 pool {pool}); a capture must sit at a drained call boundary or its keys \
809 violate the append-only finality invariant",
810 len / pool,
811 )
812 .into());
813 }
814 let index_pool_keys = if pools_ready > 0 {
815 let src = self
816 .index_pool_keys
817 .as_ref()
818 .ok_or("latent snapshot: pools are ready but the resident key plane is gone")?;
819 if src.len() < pools_ready * d {
820 return Err(format!(
821 "latent snapshot: resident key plane holds {} f32 but {pools_ready} pools \
822 x d {d} require {}",
823 src.len(),
824 pools_ready * d,
825 )
826 .into());
827 }
828 let mut keys = e.uninit(pools_ready * d)?;
829 e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
830 Some(keys)
831 } else {
832 None
833 };
834 let tail_rows = len - pools_ready * pool;
835 let index_tail = if tail_rows > 0 {
836 let src = self
837 .index_rows
838 .as_ref()
839 .ok_or("latent snapshot: index_width > 0 but the state plane is gone")?;
840 let ring = self.index_ring_rows.unwrap_or(0);
841 let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
844 let want = (phys + tail_rows) * self.index_width;
845 if src.len() < want {
846 return Err(format!(
847 "latent snapshot: index plane holds {} f32 but the live tail window \
848 requires {want}",
849 src.len(),
850 )
851 .into());
852 }
853 let mut tail = e.uninit(tail_rows * self.index_width)?;
854 e.copy_range_into(
855 &mut tail,
856 0,
857 src,
858 phys * self.index_width,
859 tail_rows * self.index_width,
860 )?;
861 Some(tail)
862 } else {
863 None
864 };
865 Ok(LatentPlaneSnapshot {
866 rows,
867 width,
868 len,
869 index_width: self.index_width,
870 index_pool: pool,
871 index_tail,
872 index_pool_keys,
873 index_pools_ready: pools_ready,
874 })
875 }
876
877 pub fn snapshot_tail(
883 &self,
884 e: &impl KvDev,
885 ) -> Result<LatentTailCapture, Box<dyn std::error::Error>> {
886 let (len, width) = (self.len, self.width);
887 if len == 0 {
888 return Err("latent tail capture at len 0 (record the layer as absent instead)".into());
889 }
890 if self.index_width == 0 {
891 return Ok(LatentTailCapture {
892 len,
893 width,
894 index_width: 0,
895 index_pool: 0,
896 index_pools_ready: 0,
897 index_tail: None,
898 });
899 }
900 let pool = self.index_pool;
901 if pool == 0 {
902 return Err(format!(
903 "latent tail capture: index plane (width {}) has an unresolved pool — no \
904 indexer call ran against this layer, so its derived state cannot be validated",
905 self.index_width,
906 )
907 .into());
908 }
909 let pools_ready = self.index_pools_ready;
910 if pools_ready != len / pool {
911 return Err(format!(
912 "latent tail capture: index_pools_ready {pools_ready} != len/pool {} (len \
913 {len}, pool {pool}); a capture must sit at a drained call boundary",
914 len / pool,
915 )
916 .into());
917 }
918 let tail_rows = len - pools_ready * pool;
919 let index_tail = if tail_rows > 0 {
920 let src = self
921 .index_rows
922 .as_ref()
923 .ok_or("latent tail capture: index_width > 0 but the state plane is gone")?;
924 let ring = self.index_ring_rows.unwrap_or(0);
925 let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
926 let want = (phys + tail_rows) * self.index_width;
927 if src.len() < want {
928 return Err(format!(
929 "latent tail capture: index plane holds {} f32 but the live tail window \
930 requires {want}",
931 src.len(),
932 )
933 .into());
934 }
935 let mut tail = e.uninit(tail_rows * self.index_width)?;
936 e.copy_range_into(
937 &mut tail,
938 0,
939 src,
940 phys * self.index_width,
941 tail_rows * self.index_width,
942 )?;
943 Some(tail)
944 } else {
945 None
946 };
947 Ok(LatentTailCapture {
948 len,
949 width,
950 index_width: self.index_width,
951 index_pool: pool,
952 index_pools_ready: pools_ready,
953 index_tail,
954 })
955 }
956
957 pub fn snapshot_plane_at(
966 &self,
967 e: &impl KvDev,
968 cap: LatentTailCapture,
969 ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
970 let (len, width) = (cap.len, cap.width);
971 if len == 0 {
972 return Err("latent boundary publish at len 0".into());
973 }
974 if width != self.width {
975 return Err(format!(
976 "latent boundary publish: captured width {width} != live width {}",
977 self.width,
978 )
979 .into());
980 }
981 if self.len < len {
982 return Err(format!(
983 "latent boundary publish: live len {} < boundary {len} — the plane was \
984 truncated below the capture boundary",
985 self.len,
986 )
987 .into());
988 }
989 if self.rows.len() < len * width {
990 return Err(format!(
991 "latent boundary publish: live plane holds {} f32 but boundary {len} x width \
992 {width} requires {}",
993 self.rows.len(),
994 len * width,
995 )
996 .into());
997 }
998 let mut rows = e.uninit(len * width)?;
999 e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
1000 if cap.index_width != self.index_width {
1001 return Err(format!(
1002 "latent boundary publish: captured index_width {} != live {}",
1003 cap.index_width, self.index_width,
1004 )
1005 .into());
1006 }
1007 if cap.index_width == 0 {
1008 return Ok(LatentPlaneSnapshot {
1009 rows,
1010 width,
1011 len,
1012 index_width: 0,
1013 index_pool: 0,
1014 index_tail: None,
1015 index_pool_keys: None,
1016 index_pools_ready: 0,
1017 });
1018 }
1019 if cap.index_pool != self.index_pool {
1020 return Err(format!(
1021 "latent boundary publish: captured pool {} != live pool {}",
1022 cap.index_pool, self.index_pool,
1023 )
1024 .into());
1025 }
1026 let d = cap.index_width / 2;
1027 let pools_ready = cap.index_pools_ready;
1028 if self.index_pools_ready < pools_ready {
1029 return Err(format!(
1030 "latent boundary publish: live index_pools_ready {} < boundary {pools_ready} \
1031 — the key plane was clamped below the capture boundary",
1032 self.index_pools_ready,
1033 )
1034 .into());
1035 }
1036 let index_pool_keys = if pools_ready > 0 {
1037 let src = self
1038 .index_pool_keys
1039 .as_ref()
1040 .ok_or("latent boundary publish: pools are ready but the key plane is gone")?;
1041 if src.len() < pools_ready * d {
1042 return Err(format!(
1043 "latent boundary publish: key plane holds {} f32 but {pools_ready} pools \
1044 x d {d} require {}",
1045 src.len(),
1046 pools_ready * d,
1047 )
1048 .into());
1049 }
1050 let mut keys = e.uninit(pools_ready * d)?;
1051 e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
1052 Some(keys)
1053 } else {
1054 None
1055 };
1056 Ok(LatentPlaneSnapshot {
1057 rows,
1058 width,
1059 len,
1060 index_width: cap.index_width,
1061 index_pool: cap.index_pool,
1062 index_tail: cap.index_tail,
1063 index_pool_keys,
1064 index_pools_ready: pools_ready,
1065 })
1066 }
1067
1068 pub fn validate_restore(
1072 &self,
1073 snap: &LatentPlaneSnapshot,
1074 max_ctx: usize,
1075 ) -> Result<(), String> {
1076 if self.len != 0 {
1077 return Err("restore destination latent plane is not fresh".into());
1078 }
1079 if self.width != snap.width {
1080 return Err(format!(
1081 "snapshot width {} != destination width {}",
1082 snap.width, self.width,
1083 ));
1084 }
1085 if snap.len == 0 || snap.len > max_ctx {
1086 return Err(format!("snapshot len {} outside [1,{max_ctx}]", snap.len));
1087 }
1088 if snap.rows.len() < snap.len * snap.width {
1089 return Err(format!(
1090 "snapshot rows plane holds {} f32 but len {} x width {} requires {} \
1091 (truncated capture)",
1092 snap.rows.len(),
1093 snap.len,
1094 snap.width,
1095 snap.len * snap.width,
1096 ));
1097 }
1098 if self.rows.len() < snap.len * self.width {
1099 return Err(format!(
1100 "destination latent plane holds {} f32 but the restore requires {}",
1101 self.rows.len(),
1102 snap.len * self.width,
1103 ));
1104 }
1105 if self.index_width != snap.index_width {
1106 return Err(format!(
1107 "snapshot index width {} != destination {}",
1108 snap.index_width, self.index_width,
1109 ));
1110 }
1111 if snap.index_width == 0 {
1112 return Ok(());
1113 }
1114 let pool = snap.index_pool;
1115 if pool == 0 {
1116 return Err("snapshot carries an index plane with an unresolved pool".into());
1117 }
1118 if self.index_pool != 0 && self.index_pool != pool {
1119 return Err(format!(
1120 "snapshot pool {pool} != destination resident pool {}",
1121 self.index_pool,
1122 ));
1123 }
1124 let d = snap.index_width / 2;
1125 if snap.index_pools_ready != snap.len / pool {
1126 return Err(format!(
1127 "snapshot index_pools_ready {} != len/pool {} (len {}, pool {pool}): the \
1128 append-only finality invariant does not hold, so its keys are stale",
1129 snap.index_pools_ready,
1130 snap.len / pool,
1131 snap.len,
1132 ));
1133 }
1134 match (&snap.index_pool_keys, snap.index_pools_ready) {
1135 (Some(keys), ready @ 1..) => {
1136 if keys.len() < ready * d {
1137 return Err(format!(
1138 "snapshot key plane holds {} f32 but {ready} pools x d {d} require {}",
1139 keys.len(),
1140 ready * d,
1141 ));
1142 }
1143 }
1144 (None, 0) => {}
1145 (Some(_), 0) => return Err("snapshot carries keys for zero ready pools".into()),
1146 (None, ready) => {
1147 return Err(format!(
1148 "snapshot claims {ready} ready pools but carries no keys"
1149 ));
1150 }
1151 }
1152 let tail_rows = snap.len - snap.index_pools_ready * pool;
1153 match (&snap.index_tail, tail_rows) {
1154 (Some(tail), rows @ 1..) => {
1155 if tail.len() < rows * snap.index_width {
1156 return Err(format!(
1157 "snapshot tail holds {} f32 but {rows} rows x index width {} require {}",
1158 tail.len(),
1159 snap.index_width,
1160 rows * snap.index_width,
1161 ));
1162 }
1163 }
1164 (None, 0) => {}
1165 (Some(_), 0) => return Err("snapshot carries a tail at a pool-aligned boundary".into()),
1166 (None, rows) => {
1167 return Err(format!(
1168 "snapshot owes {rows} live tail rows but carries none"
1169 ));
1170 }
1171 }
1172 if self.index_rows.is_none() {
1173 return Err("destination declares an index plane but allocated none".into());
1174 }
1175 if tail_rows > 0 {
1176 let ring = self.index_ring_rows.unwrap_or(0);
1177 let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
1178 let want = (phys + tail_rows) * self.index_width;
1179 let have = self.index_rows.as_ref().map_or(0, CudaSlice::len);
1180 if have < want {
1181 return Err(format!(
1182 "destination index plane holds {have} f32 but the tail window requires \
1183 {want}",
1184 ));
1185 }
1186 }
1187 Ok(())
1188 }
1189
1190 pub fn restore_plane(
1198 &mut self,
1199 e: &impl KvDev,
1200 snap: &LatentPlaneSnapshot,
1201 max_ctx: usize,
1202 ) -> Result<(), Box<dyn std::error::Error>> {
1203 self.validate_restore(snap, max_ctx)?;
1204 e.copy_range_into(&mut self.rows, 0, &snap.rows, 0, snap.len * snap.width)?;
1205 if snap.index_width > 0 {
1206 let pool = snap.index_pool;
1207 let d = snap.index_width / 2;
1208 let mut keys = e.zeros(((max_ctx / pool) * d).max(1))?;
1211 if let Some(src) = &snap.index_pool_keys {
1212 e.copy_range_into(&mut keys, 0, src, 0, snap.index_pools_ready * d)?;
1213 }
1214 self.index_pool_keys = Some(keys);
1215 self.index_pools_ready = snap.index_pools_ready;
1216 self.index_pool = pool;
1217 if let Some(tail) = &snap.index_tail {
1218 let tail_rows = snap.len - snap.index_pools_ready * pool;
1219 let ring = self.index_ring_rows.unwrap_or(0);
1220 let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
1221 let dst = self
1222 .index_rows
1223 .as_mut()
1224 .ok_or("destination index plane vanished after validation")?;
1225 e.copy_range_into(
1226 dst,
1227 phys * self.index_width,
1228 tail,
1229 0,
1230 tail_rows * self.index_width,
1231 )?;
1232 }
1233 }
1234 self.len = snap.len;
1235 let len_i32 = i32::try_from(snap.len).map_err(|_| "latent length exceeds i32 mirror")?;
1236 e.set_i32_one(&mut self.len_d, len_i32)?;
1237 Ok(())
1238 }
1239}
1240
1241pub struct RecurLayer {
1245 pub conv_state: CudaSlice<f32>, pub ssm_state: CudaSlice<f32>, pub ssm_state_alt: CudaSlice<f32>,
1257}
1258
1259pub struct ResidentTpKvCacheRank {
1260 k: CudaSlice<u8>,
1261 v: CudaSlice<u8>,
1262 len_d: CudaSlice<i32>,
1263 base_d: Option<CudaSlice<i32>>,
1268}
1269
1270impl ResidentTpKvCacheRank {
1271 pub fn new(k: CudaSlice<u8>, v: CudaSlice<u8>, len_d: CudaSlice<i32>) -> Self {
1272 Self {
1273 k,
1274 v,
1275 len_d,
1276 base_d: None,
1277 }
1278 }
1279
1280 pub fn base_d(&self) -> Option<&CudaSlice<i32>> {
1281 self.base_d.as_ref()
1282 }
1283
1284 pub fn base_d_mut(&mut self) -> Option<&mut CudaSlice<i32>> {
1285 self.base_d.as_mut()
1286 }
1287
1288 pub fn arm_base_d(&mut self, buf: CudaSlice<i32>) {
1289 self.base_d = Some(buf);
1290 }
1291
1292 pub fn k(&self) -> &CudaSlice<u8> {
1293 &self.k
1294 }
1295
1296 pub fn v(&self) -> &CudaSlice<u8> {
1297 &self.v
1298 }
1299
1300 pub fn len_d(&self) -> &CudaSlice<i32> {
1301 &self.len_d
1302 }
1303
1304 pub fn k_mut(&mut self) -> &mut CudaSlice<u8> {
1305 &mut self.k
1306 }
1307
1308 pub fn v_mut(&mut self) -> &mut CudaSlice<u8> {
1309 &mut self.v
1310 }
1311
1312 pub fn planes_mut(&mut self) -> (&mut CudaSlice<u8>, &mut CudaSlice<u8>) {
1313 (&mut self.k, &mut self.v)
1314 }
1315
1316 #[allow(clippy::type_complexity)]
1318 pub fn planes_and_counters_mut(
1319 &mut self,
1320 ) -> (
1321 &mut CudaSlice<u8>,
1322 &mut CudaSlice<u8>,
1323 &CudaSlice<i32>,
1324 Option<&CudaSlice<i32>>,
1325 ) {
1326 (&mut self.k, &mut self.v, &self.len_d, self.base_d.as_ref())
1327 }
1328
1329 pub fn len_d_mut(&mut self) -> &mut CudaSlice<i32> {
1330 &mut self.len_d
1331 }
1332}
1333
1334#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1335pub struct TpKvTransaction {
1336 generation: u64,
1337 base_len: usize,
1338}
1339
1340impl TpKvTransaction {
1341 pub fn generation(self) -> u64 {
1342 self.generation
1343 }
1344
1345 pub fn base_len(self) -> usize {
1346 self.base_len
1347 }
1348}
1349
1350#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1351pub struct TpKvAppendPlan {
1352 transaction: TpKvTransaction,
1353 target: usize,
1354 write_row: usize,
1355 ring_append: Option<KvRingAppend>,
1356}
1357
1358impl TpKvAppendPlan {
1359 pub fn target(self) -> usize {
1360 self.target
1361 }
1362
1363 pub fn write_row(self) -> usize {
1364 self.write_row
1365 }
1366
1367 pub fn ring_append(self) -> Option<KvRingAppend> {
1368 self.ring_append
1369 }
1370}
1371
1372#[derive(Debug, PartialEq, Eq)]
1373pub struct TpKvGrowPlan {
1374 rows: usize,
1375 source_row: usize,
1376 copy_rows: usize,
1377 target_base: usize,
1378 k_bytes: usize,
1379 v_bytes: usize,
1380 source_capacity: usize,
1381 target_capacity: usize,
1382 ring_window: Option<usize>,
1383 target_physical_rows: usize,
1384 kv_dim_k: usize,
1385 kv_dim_v: usize,
1386 k_tok_bytes: usize,
1387 v_tok_bytes: usize,
1388 ranks: usize,
1389 next_generation: u64,
1390}
1391
1392impl TpKvGrowPlan {
1393 pub fn rows(&self) -> usize {
1394 self.rows
1395 }
1396
1397 pub fn source_row(&self) -> usize {
1398 self.source_row
1399 }
1400
1401 pub fn copy_rows(&self) -> usize {
1402 self.copy_rows
1403 }
1404
1405 pub fn k_bytes(&self) -> usize {
1406 self.k_bytes
1407 }
1408
1409 pub fn v_bytes(&self) -> usize {
1410 self.v_bytes
1411 }
1412}
1413
1414#[derive(Clone, Debug, PartialEq, Eq)]
1415struct TpKvTransactionState {
1416 committed_len: usize,
1417 staged_len: usize,
1418 next_generation: u64,
1419 active: Option<TpKvTransaction>,
1420}
1421
1422impl TpKvTransactionState {
1423 fn new() -> Self {
1424 Self {
1425 committed_len: 0,
1426 staged_len: 0,
1427 next_generation: 1,
1428 active: None,
1429 }
1430 }
1431
1432 fn begin(&mut self) -> Result<TpKvTransaction, String> {
1433 if let Some(active) = self.active {
1434 return Err(format!(
1435 "TP KV transaction generation {} is already active at base {}",
1436 active.generation, active.base_len
1437 ));
1438 }
1439 if self.staged_len != self.committed_len {
1440 return Err(format!(
1441 "TP KV cache is half-committed: staged {} != committed {}",
1442 self.staged_len, self.committed_len
1443 ));
1444 }
1445 let transaction = TpKvTransaction {
1446 generation: self.next_generation,
1447 base_len: self.committed_len,
1448 };
1449 self.next_generation = self
1450 .next_generation
1451 .checked_add(1)
1452 .ok_or("TP KV transaction generation overflow")?;
1453 self.active = Some(transaction);
1454 Ok(transaction)
1455 }
1456
1457 fn validate(&self, transaction: TpKvTransaction) -> Result<(), String> {
1458 if self.active != Some(transaction) {
1459 return Err(format!(
1460 "stale TP KV transaction generation {} at base {}",
1461 transaction.generation, transaction.base_len
1462 ));
1463 }
1464 if transaction.base_len != self.committed_len {
1465 return Err(format!(
1466 "TP KV transaction base {} != committed length {}",
1467 transaction.base_len, self.committed_len
1468 ));
1469 }
1470 Ok(())
1471 }
1472
1473 fn append_target(
1474 &self,
1475 transaction: TpKvTransaction,
1476 rows: usize,
1477 capacity: usize,
1478 ) -> Result<usize, String> {
1479 self.validate(transaction)?;
1480 if rows == 0 {
1481 return Err("TP KV append must contain at least one row".into());
1482 }
1483 let target = self
1484 .staged_len
1485 .checked_add(rows)
1486 .ok_or("TP KV staged length overflow")?;
1487 if target > capacity {
1488 return Err(format!(
1489 "TP KV append exceeds capacity: {target} > {capacity}"
1490 ));
1491 }
1492 Ok(target)
1493 }
1494
1495 fn publish_append(
1496 &mut self,
1497 transaction: TpKvTransaction,
1498 target: usize,
1499 ) -> Result<(), String> {
1500 self.validate(transaction)?;
1501 if target <= self.staged_len {
1502 return Err(format!(
1503 "TP KV append target {target} must exceed staged length {}",
1504 self.staged_len
1505 ));
1506 }
1507 self.staged_len = target;
1508 Ok(())
1509 }
1510
1511 fn commit_target(
1512 &self,
1513 transaction: TpKvTransaction,
1514 accepted_rows: usize,
1515 ) -> Result<usize, String> {
1516 self.validate(transaction)?;
1517 let staged_rows = self
1518 .staged_len
1519 .checked_sub(transaction.base_len)
1520 .ok_or("TP KV staged length precedes its transaction base")?;
1521 if accepted_rows > staged_rows {
1522 return Err(format!(
1523 "TP KV commit accepts {accepted_rows} rows from a {staged_rows}-row transaction"
1524 ));
1525 }
1526 transaction
1527 .base_len
1528 .checked_add(accepted_rows)
1529 .ok_or_else(|| "TP KV committed length overflow".to_string())
1530 }
1531
1532 fn publish_finalize(
1533 &mut self,
1534 transaction: TpKvTransaction,
1535 target: usize,
1536 ) -> Result<(), String> {
1537 self.validate(transaction)?;
1538 if target < transaction.base_len || target > self.staged_len {
1539 return Err(format!(
1540 "TP KV finalize target {target} outside transaction range {}..={}",
1541 transaction.base_len, self.staged_len
1542 ));
1543 }
1544 self.committed_len = target;
1545 self.staged_len = target;
1546 self.active = None;
1547 Ok(())
1548 }
1549
1550 fn rewind(&mut self, target: usize, capacity: usize) -> Result<(), String> {
1551 if target > capacity {
1552 return Err(format!(
1553 "TP KV rewind target {target} exceeds capacity {capacity}"
1554 ));
1555 }
1556 self.committed_len = target;
1557 self.staged_len = target;
1558 self.active = None;
1559 Ok(())
1560 }
1561}
1562
1563pub struct ResidentTpKvCache {
1564 ranks: Vec<ResidentTpKvCacheRank>,
1565 kv_dim_k: usize,
1566 kv_dim_v: usize,
1567 k_tok_bytes: usize,
1568 v_tok_bytes: usize,
1569 capacity: usize,
1570 ring: Option<KvRing>,
1571 state: TpKvTransactionState,
1572 external_rows: bool,
1579}
1580
1581impl ResidentTpKvCache {
1582 #[allow(clippy::too_many_arguments)]
1583 pub fn new(
1584 ranks: Vec<ResidentTpKvCacheRank>,
1585 kv_dim_k: usize,
1586 kv_dim_v: usize,
1587 k_tok_bytes: usize,
1588 v_tok_bytes: usize,
1589 capacity: usize,
1590 ) -> Self {
1591 Self::new_inner(
1592 ranks,
1593 kv_dim_k,
1594 kv_dim_v,
1595 k_tok_bytes,
1596 v_tok_bytes,
1597 capacity,
1598 None,
1599 )
1600 }
1601
1602 #[allow(clippy::too_many_arguments)]
1603 pub fn new_swa(
1604 ranks: Vec<ResidentTpKvCacheRank>,
1605 kv_dim_k: usize,
1606 kv_dim_v: usize,
1607 k_tok_bytes: usize,
1608 v_tok_bytes: usize,
1609 capacity: usize,
1610 window: usize,
1611 ) -> Self {
1612 Self::new_inner(
1613 ranks,
1614 kv_dim_k,
1615 kv_dim_v,
1616 k_tok_bytes,
1617 v_tok_bytes,
1618 capacity,
1619 Some(KvRing::new(swa_ring_rows(window, capacity), window)),
1620 )
1621 }
1622
1623 #[allow(clippy::too_many_arguments)]
1624 fn new_inner(
1625 ranks: Vec<ResidentTpKvCacheRank>,
1626 kv_dim_k: usize,
1627 kv_dim_v: usize,
1628 k_tok_bytes: usize,
1629 v_tok_bytes: usize,
1630 capacity: usize,
1631 ring: Option<KvRing>,
1632 ) -> Self {
1633 Self {
1634 ranks,
1635 kv_dim_k,
1636 kv_dim_v,
1637 k_tok_bytes,
1638 v_tok_bytes,
1639 capacity,
1640 ring,
1641 state: TpKvTransactionState::new(),
1642 external_rows: false,
1643 }
1644 }
1645
1646 pub fn begin_transaction(&mut self) -> Result<TpKvTransaction, String> {
1647 self.state.begin()
1648 }
1649
1650 pub fn rows_external(&self) -> bool {
1653 self.external_rows
1654 }
1655
1656 pub fn mark_rows_external(&mut self, external: bool) {
1657 self.external_rows = external;
1658 }
1659
1660 pub fn committed_len(&self) -> usize {
1661 self.state.committed_len
1662 }
1663
1664 pub fn staged_len(&self) -> usize {
1665 self.state.staged_len
1666 }
1667
1668 pub fn capacity(&self) -> usize {
1669 self.capacity
1670 }
1671
1672 pub fn physical_capacity(&self) -> usize {
1673 self.ring
1674 .as_ref()
1675 .map(KvRing::rows)
1676 .unwrap_or(self.capacity)
1677 }
1678
1679 pub fn ring_window(&self) -> Option<usize> {
1680 self.ring.as_ref().map(KvRing::window)
1681 }
1682
1683 pub fn ring_base(&self) -> Option<usize> {
1684 self.ring.as_ref().map(KvRing::base)
1685 }
1686
1687 pub fn physical_range(
1688 &self,
1689 start: usize,
1690 end: usize,
1691 ) -> Result<std::ops::Range<usize>, String> {
1692 match &self.ring {
1693 Some(ring) => ring.physical_range(start, end),
1694 None => {
1695 if end < start || end > self.capacity {
1696 return Err(format!(
1697 "TP KV linear view [{start},{end}) exceeds capacity {}",
1698 self.capacity
1699 ));
1700 }
1701 Ok(start..end)
1702 }
1703 }
1704 }
1705
1706 pub fn can_rewind_to(&self, target: usize) -> bool {
1707 target <= self.capacity
1708 && self
1709 .ring
1710 .as_ref()
1711 .is_none_or(|ring| ring.can_rewind_to(target))
1712 }
1713
1714 pub fn kv_dim_k(&self) -> usize {
1715 self.kv_dim_k
1716 }
1717
1718 pub fn kv_dim_v(&self) -> usize {
1719 self.kv_dim_v
1720 }
1721
1722 pub fn k_tok_bytes(&self) -> usize {
1723 self.k_tok_bytes
1724 }
1725
1726 pub fn v_tok_bytes(&self) -> usize {
1727 self.v_tok_bytes
1728 }
1729
1730 pub fn ranks_len(&self) -> usize {
1731 self.ranks.len()
1732 }
1733
1734 pub fn rank(&self, rank: usize) -> Option<&ResidentTpKvCacheRank> {
1735 self.ranks.get(rank)
1736 }
1737
1738 pub fn rank_mut(&mut self, rank: usize) -> Option<&mut ResidentTpKvCacheRank> {
1739 self.ranks.get_mut(rank)
1740 }
1741
1742 pub fn ranks(&self) -> &[ResidentTpKvCacheRank] {
1743 &self.ranks
1744 }
1745
1746 pub fn ranks_mut(&mut self) -> &mut [ResidentTpKvCacheRank] {
1747 &mut self.ranks
1748 }
1749
1750 pub fn prepare_grow(
1751 &self,
1752 target_capacity: usize,
1753 rows: usize,
1754 ) -> Result<TpKvGrowPlan, String> {
1755 if let Some(active) = self.state.active {
1756 return Err(format!(
1757 "TP KV grow refuses active transaction generation {} at base {}",
1758 active.generation, active.base_len
1759 ));
1760 }
1761 if self.state.staged_len != self.state.committed_len {
1762 return Err(format!(
1763 "TP KV grow requires quiescent state, got committed/staged={}/{}",
1764 self.state.committed_len, self.state.staged_len
1765 ));
1766 }
1767 if target_capacity <= self.capacity {
1768 return Err(format!(
1769 "TP KV grow target capacity {target_capacity} must exceed source capacity {}",
1770 self.capacity
1771 ));
1772 }
1773 if target_capacity > i32::MAX as usize {
1774 return Err(format!(
1775 "TP KV grow target capacity {target_capacity} exceeds i32 device mirrors"
1776 ));
1777 }
1778 if rows > self.state.committed_len {
1779 return Err(format!(
1780 "TP KV grow rows {rows} exceed committed length {}",
1781 self.state.committed_len
1782 ));
1783 }
1784 let (source_row, copy_rows, target_base, ring_window, target_physical_rows) =
1785 match &self.ring {
1786 Some(ring) => {
1787 let raw = rows.saturating_sub(ring.window().saturating_sub(1));
1788 let target_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1789 let physical = ring.physical_range(target_base, rows)?;
1790 (
1791 physical.start,
1792 physical.len(),
1793 target_base,
1794 Some(ring.window()),
1795 swa_ring_rows(ring.window(), target_capacity),
1796 )
1797 }
1798 None => (0, rows, 0, None, target_capacity),
1799 };
1800 let k_bytes = copy_rows
1801 .checked_mul(self.k_tok_bytes)
1802 .ok_or("TP KV grow K byte extent overflow")?;
1803 let v_bytes = copy_rows
1804 .checked_mul(self.v_tok_bytes)
1805 .ok_or("TP KV grow V byte extent overflow")?;
1806 Ok(TpKvGrowPlan {
1807 rows,
1808 source_row,
1809 copy_rows,
1810 target_base,
1811 k_bytes,
1812 v_bytes,
1813 source_capacity: self.capacity,
1814 target_capacity,
1815 ring_window,
1816 target_physical_rows,
1817 kv_dim_k: self.kv_dim_k,
1818 kv_dim_v: self.kv_dim_v,
1819 k_tok_bytes: self.k_tok_bytes,
1820 v_tok_bytes: self.v_tok_bytes,
1821 ranks: self.ranks.len(),
1822 next_generation: self.state.next_generation,
1823 })
1824 }
1825
1826 pub fn publish_grow(&mut self, plan: TpKvGrowPlan) -> Result<(), String> {
1827 if self.state != TpKvTransactionState::new() {
1828 return Err(format!(
1829 "TP KV grow target must be fresh, got committed/staged={}/{} active={}",
1830 self.state.committed_len,
1831 self.state.staged_len,
1832 self.state.active.is_some()
1833 ));
1834 }
1835 if self.capacity != plan.target_capacity
1836 || self.capacity <= plan.source_capacity
1837 || self.kv_dim_k != plan.kv_dim_k
1838 || self.kv_dim_v != plan.kv_dim_v
1839 || self.k_tok_bytes != plan.k_tok_bytes
1840 || self.v_tok_bytes != plan.v_tok_bytes
1841 || self.ranks.len() != plan.ranks
1842 || self.ring.as_ref().map(KvRing::window) != plan.ring_window
1843 || self.physical_capacity() != plan.target_physical_rows
1844 {
1845 return Err("TP KV grow target layout does not match its source plan".into());
1846 }
1847 if plan.rows > self.capacity {
1848 return Err(format!(
1849 "TP KV grow rows {} exceed target capacity {}",
1850 plan.rows, self.capacity
1851 ));
1852 }
1853 if let Some(ring) = self.ring.as_mut() {
1854 let mut target_ring = *ring;
1855 target_ring.apply_rebase(plan.target_base);
1856 if !target_ring.can_rewind_to(plan.rows) {
1857 return Err(format!(
1858 "TP KV grow target ring base {} cannot expose committed length {}",
1859 target_ring.base(),
1860 plan.rows
1861 ));
1862 }
1863 *ring = target_ring;
1864 }
1865 self.state.committed_len = plan.rows;
1866 self.state.staged_len = plan.rows;
1867 self.state.next_generation = plan.next_generation;
1868 self.state.active = None;
1869 Ok(())
1870 }
1871
1872 pub fn prepare_append(
1873 &self,
1874 transaction: TpKvTransaction,
1875 rows: usize,
1876 ) -> Result<TpKvAppendPlan, String> {
1877 let target = self.state.append_target(transaction, rows, self.capacity)?;
1878 let ring_append = self
1879 .ring
1880 .as_ref()
1881 .map(|ring| {
1882 let staged_retain =
1883 target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1884 let rollback_retain = transaction
1885 .base_len
1886 .saturating_sub(ring.window().saturating_sub(1))
1887 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1888 ring.append_plan(
1889 self.state.staged_len,
1890 staged_retain.min(rollback_retain),
1891 rows,
1892 )
1893 })
1894 .transpose()?;
1895 let write_row = match ring_append {
1896 Some(KvRingAppend::Contiguous { write_row })
1897 | Some(KvRingAppend::Rebase { write_row, .. }) => write_row,
1898 None => self.state.staged_len,
1899 };
1900 Ok(TpKvAppendPlan {
1901 transaction,
1902 target,
1903 write_row,
1904 ring_append,
1905 })
1906 }
1907
1908 pub fn peek_append_ring(&self, rows: usize) -> Result<(usize, bool), String> {
1912 let target = self.state.staged_len + rows;
1913 let plan = self
1914 .ring
1915 .as_ref()
1916 .map(|ring| {
1917 let staged_retain =
1918 target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1919 let rollback_retain = self
1920 .state
1921 .staged_len
1922 .saturating_sub(ring.window().saturating_sub(1))
1923 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1924 ring.append_plan(
1925 self.state.staged_len,
1926 staged_retain.min(rollback_retain),
1927 rows,
1928 )
1929 })
1930 .transpose()?;
1931 Ok(match plan {
1932 Some(KvRingAppend::Contiguous { write_row }) => (write_row, false),
1933 Some(KvRingAppend::Rebase { write_row, .. }) => (write_row, true),
1934 None => (self.state.staged_len, false),
1935 })
1936 }
1937
1938 pub fn publish_append_rebase(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1939 self.state.validate(plan.transaction)?;
1940 match (self.ring.as_mut(), plan.ring_append) {
1941 (
1942 Some(ring),
1943 Some(KvRingAppend::Rebase {
1944 new_base,
1945 keep_rows,
1946 ..
1947 }),
1948 ) => {
1949 if keep_rows > ring.rows() {
1950 return Err(format!(
1951 "TP KV ring rebase keeps {keep_rows} rows in {} physical rows",
1952 ring.rows()
1953 ));
1954 }
1955 let mut target_ring = *ring;
1956 target_ring.apply_rebase(new_base);
1957 if !target_ring.can_rewind_to(plan.transaction.base_len) {
1958 return Err(format!(
1959 "TP KV ring rebase to {new_base} laps transaction base {}",
1960 plan.transaction.base_len
1961 ));
1962 }
1963 *ring = target_ring;
1964 Ok(())
1965 }
1966 (Some(_), Some(KvRingAppend::Contiguous { .. })) | (None, None) => Ok(()),
1967 _ => Err("TP KV append plan does not match cache ring layout".into()),
1968 }
1969 }
1970
1971 pub fn publish_append_plan(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1972 if let Some(KvRingAppend::Rebase { new_base, .. }) = plan.ring_append {
1973 if self.ring.as_ref().map(KvRing::base) != Some(new_base) {
1974 return Err(format!(
1975 "TP KV append rebase {new_base} was not published before its state"
1976 ));
1977 }
1978 }
1979 self.state.publish_append(plan.transaction, plan.target)
1980 }
1981
1982 pub fn publish_hydration(
1983 &mut self,
1984 logical_len: usize,
1985 resident_start: usize,
1986 ) -> Result<(), Box<dyn std::error::Error>> {
1987 if self.state != TpKvTransactionState::new() {
1988 return Err("TP KV hydration target must be fresh".into());
1989 }
1990 if resident_start > logical_len || logical_len > self.capacity {
1991 return Err(format!(
1992 "TP KV hydration range [{resident_start},{logical_len}) exceeds capacity {}",
1993 self.capacity
1994 )
1995 .into());
1996 }
1997 match self.ring.as_mut() {
1998 Some(ring) => {
1999 let rows = logical_len - resident_start;
2000 if rows > ring.rows() {
2001 return Err(format!(
2002 "TP KV hydration requires {rows} rows in a {}-row ring",
2003 ring.rows()
2004 )
2005 .into());
2006 }
2007 let mut hydrated_ring = *ring;
2008 hydrated_ring.apply_rebase(resident_start);
2009 if !hydrated_ring.can_rewind_to(logical_len) {
2010 return Err(format!(
2011 "TP KV hydration base {resident_start} cannot expose logical length \
2012 {logical_len}"
2013 )
2014 .into());
2015 }
2016 *ring = hydrated_ring;
2017 }
2018 None if resident_start != 0 => {
2019 return Err("linear TP KV hydration must start at absolute row zero".into());
2020 }
2021 None => {}
2022 }
2023 self.rewind_to(logical_len)
2024 }
2025
2026 pub fn append_target(
2027 &self,
2028 transaction: TpKvTransaction,
2029 rows: usize,
2030 ) -> Result<usize, String> {
2031 self.state.append_target(transaction, rows, self.capacity)
2032 }
2033
2034 pub fn publish_append(
2035 &mut self,
2036 transaction: TpKvTransaction,
2037 target: usize,
2038 ) -> Result<(), String> {
2039 self.state.publish_append(transaction, target)
2040 }
2041
2042 pub fn commit_target(
2043 &self,
2044 transaction: TpKvTransaction,
2045 accepted_rows: usize,
2046 ) -> Result<usize, String> {
2047 self.state.commit_target(transaction, accepted_rows)
2048 }
2049
2050 pub fn validate_transaction(&self, transaction: TpKvTransaction) -> Result<(), String> {
2051 self.state.validate(transaction)
2052 }
2053
2054 pub fn publish_finalize(
2055 &mut self,
2056 transaction: TpKvTransaction,
2057 target: usize,
2058 ) -> Result<(), String> {
2059 if !self.can_rewind_to(target) {
2060 return Err(format!(
2061 "TP KV finalize target {target} is outside the resident cache window/capacity"
2062 ));
2063 }
2064 self.state.publish_finalize(transaction, target)
2065 }
2066
2067 pub fn rewind_to(&mut self, target: usize) -> Result<(), Box<dyn std::error::Error>> {
2068 if !self.can_rewind_to(target) {
2069 return Err(format!(
2070 "TP KV rewind target {target} is outside the resident cache window/capacity"
2071 )
2072 .into());
2073 }
2074 let target_i32 =
2075 i32::try_from(target).map_err(|_| "TP KV length exceeds i32 device mirror")?;
2076 for rank in &mut self.ranks {
2077 let stream = rank.len_d.stream().clone();
2078 stream.memcpy_htod(&[target_i32], &mut rank.len_d)?;
2079 }
2080 self.state.rewind(target, self.capacity)?;
2081 Ok(())
2082 }
2083
2084 pub fn publish_device_rewind(
2088 &mut self,
2089 target: usize,
2090 ) -> Result<(), Box<dyn std::error::Error>> {
2091 if !self.can_rewind_to(target) {
2092 return Err(format!(
2093 "TP KV device rewind target {target} is outside the resident cache window/capacity"
2094 )
2095 .into());
2096 }
2097 self.state.rewind(target, self.capacity)?;
2098 Ok(())
2099 }
2100}
2101
2102pub struct Cache {
2103 pub kv: Vec<Option<KvLayer>>,
2104 pub recur: Vec<Option<RecurLayer>>,
2105 pub latent: Vec<Option<LatentKvLayer>>,
2108 pub tp_kv: Vec<Option<ResidentTpKvCache>>,
2111 pub glm5_tp_recur: Vec<Option<Vec<RecurLayer>>>,
2121 pub glm5_tp_latent_peer: Vec<Option<Vec<LatentKvLayer>>>,
2126 pub pos: usize,
2127 pub max_ctx: usize,
2128 pub tainted: bool,
2131 pub last_logits_dev: Option<CudaSlice<f32>>,
2139 pub dflash_taps: Option<DflashTapSink>,
2144 pub hc_taps: Option<HcTapSink>,
2152 pub glm5_decode_graph: Option<Box<dyn std::any::Any + Send>>,
2163}
2164
2165#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2169enum FullAttentionClass {
2170 Ordinary,
2171 GemmaGlobal,
2172 GemmaWindowed,
2173}
2174
2175fn full_attention_class(plan: &ModelPlan, il: u32) -> FullAttentionClass {
2176 let layer = plan
2177 .layers
2178 .iter()
2179 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2180 .find(|layer| layer.index == il)
2181 .unwrap_or_else(|| panic!("ModelPlan has no layer {il}"));
2182 if !matches!(layer.residual, ResidualTopology::Gemma { .. }) {
2183 return FullAttentionClass::Ordinary;
2184 }
2185 match layer.state {
2186 StatePlan::SlidingKvCache { .. } => FullAttentionClass::GemmaWindowed,
2187 StatePlan::KvCache { .. } => FullAttentionClass::GemmaGlobal,
2188 _ => panic!("Gemma layer {il} does not declare a KV-cache state"),
2189 }
2190}
2191
2192fn full_attention_kv_layout(
2193 cfg: &ModelConfig,
2194 plan: &ModelPlan,
2195 il: u32,
2196) -> (usize, usize, usize, usize) {
2197 debug_assert_eq!(cfg.layer_kind(il), LayerKind::FullAttention);
2198 let class = full_attention_class(plan, il);
2199 let n_head_kv = cfg.n_head_kv as usize;
2200 let (kv_dim_k, kv_dim_v) = match class {
2201 FullAttentionClass::GemmaGlobal | FullAttentionClass::GemmaWindowed => {
2202 let g = cfg
2203 .gemma4
2204 .as_ref()
2205 .expect("Gemma ModelPlan layer requires Gemma cache geometry");
2206 let hd = match class {
2207 FullAttentionClass::GemmaWindowed => g.key_length_swa,
2208 FullAttentionClass::GemmaGlobal => g.key_length_global,
2209 FullAttentionClass::Ordinary => unreachable!(),
2210 } as usize;
2211 let d = match g.head_count_kv.get(il as usize) {
2220 Some(n) => hd * *n as usize,
2221 None => hd * n_head_kv,
2222 };
2223 (d, d)
2224 }
2225 FullAttentionClass::Ordinary => (
2226 cfg.head_dim_k as usize * n_head_kv,
2227 cfg.head_dim_v as usize * n_head_kv,
2228 ),
2229 };
2230 assert!(
2231 kv_dim_k % 32 == 0 && kv_dim_v % 32 == 0,
2232 "KVQUANT requires per-layer kv_dim_k%32==0 && kv_dim_v%32==0 \
2233 (layer {il}: k={kv_dim_k} v={kv_dim_v})"
2234 );
2235 let (kbb, vbb) = kv_blk_bytes();
2236 let g4_global_fp8 = gkv_on() && class == FullAttentionClass::GemmaGlobal;
2237 let g4_windowed_fp8 = wkv_on() && class == FullAttentionClass::GemmaWindowed;
2238 let qwen_fp8 = kv_fp8_on() && class == FullAttentionClass::Ordinary;
2239 let (kbb_l, vbb_l) = if g4_global_fp8 || g4_windowed_fp8 || qwen_fp8 {
2240 (32, 32)
2241 } else {
2242 (kbb, vbb)
2243 };
2244 (kv_dim_k, kv_dim_v, kbb_l, vbb_l)
2245}
2246
2247fn kv_plane_allocation_bytes(rows: usize, token_bytes: usize) -> usize {
2248 rows * token_bytes + 8
2249}
2250
2251pub fn cache_bytes_per_token(cfg: &ModelConfig) -> usize {
2258 cache_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2259}
2260
2261pub fn cache_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2265 let plan = ModelPlan::compile(cfg).expect("cache sizing requires a compilable ModelPlan");
2266 cache_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2267}
2268
2269pub fn cache_bytes_per_token_for_plan(
2270 cfg: &ModelConfig,
2271 plan: &ModelPlan,
2272 lo: usize,
2273 hi: usize,
2274) -> usize {
2275 assert!(
2276 lo <= hi && hi <= cfg.n_layer as usize,
2277 "cache layer range out of bounds"
2278 );
2279 let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2280 let full_attn: usize = (lo as u32..hi as u32)
2281 .filter(|&il| cfg.layer_kind(il) == LayerKind::FullAttention)
2282 .filter(|&il| shared == 0 || il < cfg.n_layer - shared)
2283 .map(|il| {
2284 let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, il);
2285 (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2286 })
2287 .sum();
2288 full_attn + latent_kv_bytes_per_token_for_plan(cfg, plan, lo, hi)
2289}
2290
2291pub fn latent_kv_bytes_per_token_for_plan(
2321 cfg: &ModelConfig,
2322 plan: &ModelPlan,
2323 lo: usize,
2324 hi: usize,
2325) -> usize {
2326 let ring_disabled = std::env::var("MEMRA_DSA_INDEX_RING")
2327 .ok()
2328 .and_then(|v| v.trim().parse::<usize>().ok())
2329 == Some(0);
2330 plan.layers
2331 .iter()
2332 .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2333 .map(|layer| match layer.state {
2334 StatePlan::LatentKvCache { width, index_width } => {
2335 let latent = width as usize * std::mem::size_of::<f32>();
2336 let index_width = index_width as usize;
2337 let pool = cfg
2338 .glm5
2339 .as_ref()
2340 .map(|g| g.index_kpool as usize)
2341 .filter(|&p| p > 0)
2342 .unwrap_or(1);
2343 let pool_keys = if index_width > 0 {
2345 (index_width / 2) * std::mem::size_of::<f32>() / pool
2346 } else {
2347 0
2348 };
2349 let flat_plane = if index_width > 0 && ring_disabled {
2350 index_width * std::mem::size_of::<f32>()
2351 } else {
2352 0
2353 };
2354 latent + pool_keys + flat_plane
2355 }
2356 _ => 0,
2357 })
2358 .sum()
2359}
2360
2361pub fn cache_ring_bytes_per_token(cfg: &ModelConfig) -> usize {
2364 cache_ring_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2365}
2366
2367pub fn cache_ring_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2369 assert!(
2370 lo <= hi && hi <= cfg.n_layer as usize,
2371 "cache layer range out of bounds"
2372 );
2373 let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2374 return 0;
2375 };
2376 cache_ring_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2377}
2378
2379pub fn cache_ring_bytes_per_token_for_plan(
2380 cfg: &ModelConfig,
2381 plan: &ModelPlan,
2382 lo: usize,
2383 hi: usize,
2384) -> usize {
2385 let total = plan.layers.len() + plan.mtp_blocks.len();
2386 assert!(
2387 lo <= hi && hi <= total,
2388 "cache plan layer range out of bounds"
2389 );
2390 if !swa_ring_on() {
2391 return 0;
2392 }
2393 let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2394 plan.layers
2395 .iter()
2396 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2397 .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2398 .filter(|layer| {
2399 matches!(
2400 layer.state,
2401 memra_gguf::model_plan::StatePlan::SlidingKvCache { .. }
2402 )
2403 })
2404 .filter(|layer| shared == 0 || layer.index < cfg.n_layer - shared)
2405 .map(|layer| {
2406 let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, layer.index);
2407 (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2408 })
2409 .sum()
2410}
2411
2412pub fn cache_ring_row_cap(cfg: &ModelConfig) -> usize {
2414 let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2415 return 0;
2416 };
2417 cache_ring_row_cap_for_plan(&plan)
2418}
2419
2420pub fn cache_ring_row_cap_for_plan(plan: &memra_gguf::model_plan::ModelPlan) -> usize {
2421 if !swa_ring_on() {
2422 return 0;
2423 }
2424 plan.layers
2425 .iter()
2426 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2427 .filter_map(|layer| match layer.state {
2428 memra_gguf::model_plan::StatePlan::SlidingKvCache { window, .. } => {
2429 Some(window as usize)
2430 }
2431 _ => None,
2432 })
2433 .map(|window| swa_ring_rows(window, usize::MAX))
2434 .max()
2435 .unwrap_or(0)
2436}
2437
2438pub struct DflashTapSink {
2441 pub layer_ids: Vec<usize>,
2442 pub buf: CudaSlice<f32>,
2443 pub hidden: usize,
2444 pub t: usize,
2445 pub base: usize,
2448}
2449
2450pub struct HcTapSink {
2457 pub layer_ids: Vec<usize>,
2459 pub rows: Vec<f32>,
2461 pub hidden: usize,
2462 pub t: usize,
2464 pub base: usize,
2467 pub origin: usize,
2473 pub dev: Vec<Option<CudaSlice<f32>>>,
2481 pub device_stage: bool,
2489 pub dtoh_ns: u64,
2492 pub ingest_state: Option<Box<dyn std::any::Any + Send>>,
2499}
2500
2501impl HcTapSink {
2502 pub fn new(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2503 let n_taps = layer_ids.len();
2504 Self {
2505 layer_ids,
2506 rows: vec![0.0; t * n_taps * hidden],
2507 hidden,
2508 t,
2509 base: 0,
2510 origin: 0,
2511 dev: (0..n_taps).map(|_| None).collect(),
2512 device_stage: false,
2513 dtoh_ns: 0,
2514 ingest_state: None,
2515 }
2516 }
2517
2518 pub fn new_at(layer_ids: Vec<usize>, hidden: usize, t: usize, origin: usize) -> Self {
2521 Self {
2522 origin,
2523 ..Self::new(layer_ids, hidden, t)
2524 }
2525 }
2526
2527 pub fn new_device_staged(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2530 Self {
2531 device_stage: true,
2532 ..Self::new(layer_ids, hidden, t)
2533 }
2534 }
2535
2536 pub fn new_device_staged_at(
2542 layer_ids: Vec<usize>,
2543 hidden: usize,
2544 t: usize,
2545 origin: usize,
2546 ) -> Self {
2547 let n_taps = layer_ids.len();
2548 Self {
2549 layer_ids,
2550 rows: Vec::new(),
2551 hidden,
2552 t,
2553 base: 0,
2554 origin,
2555 dev: (0..n_taps).map(|_| None).collect(),
2556 device_stage: true,
2557 dtoh_ns: 0,
2558 ingest_state: None,
2559 }
2560 }
2561}
2562
2563pub struct CacheSnapshot {
2593 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>>>,
2597 pub pos: usize,
2598}
2599
2600impl Cache {
2601 pub fn ensure_usable(&self, path: &str) -> Result<(), Box<dyn std::error::Error>> {
2602 if self.tainted {
2603 return Err(format!(
2604 "{path}: cache was tainted by a failed pipeline wave and cannot be reused"
2605 )
2606 .into());
2607 }
2608 Ok(())
2609 }
2610
2611 pub fn mark_tainted(&mut self) {
2612 self.tainted = true;
2613 self.last_logits_dev = None;
2614 self.dflash_taps = None;
2615 }
2616
2617 pub fn new(
2619 e: &impl KvDev,
2620 cfg: &ModelConfig,
2621 max_ctx: usize,
2622 ) -> Result<Self, Box<dyn std::error::Error>> {
2623 Self::new_inner(&|_| e, cfg, None, max_ctx)
2624 }
2625
2626 pub fn new_planned(
2627 e: &impl KvDev,
2628 cfg: &ModelConfig,
2629 plan: &memra_gguf::model_plan::ModelPlan,
2630 max_ctx: usize,
2631 ) -> Result<Self, Box<dyn std::error::Error>> {
2632 Self::new_inner(&|_| e, cfg, Some(plan), max_ctx)
2633 }
2634
2635 pub fn new_pp2(
2640 dev0: &dyn KvDev,
2641 dev1: &dyn KvDev,
2642 split: usize,
2643 cfg: &ModelConfig,
2644 max_ctx: usize,
2645 ) -> Result<Self, Box<dyn std::error::Error>> {
2646 Self::new_inner(
2647 &|il| if il < split { dev0 } else { dev1 },
2648 cfg,
2649 None,
2650 max_ctx,
2651 )
2652 }
2653
2654 pub fn new_ppn(
2660 devs: &[&dyn KvDev],
2661 fence: &[usize],
2662 cfg: &ModelConfig,
2663 max_ctx: usize,
2664 ) -> Result<Self, Box<dyn std::error::Error>> {
2665 assert_eq!(
2666 devs.len() + 1,
2667 fence.len(),
2668 "ppn cache: devs vs fence mismatch"
2669 );
2670 let pick = |il: usize| -> &dyn KvDev {
2671 let s = match fence[1..fence.len() - 1].binary_search(&il) {
2672 Ok(k) => k + 1,
2673 Err(k) => k,
2674 };
2675 devs[s.min(devs.len() - 1)]
2676 };
2677 Self::new_inner(&pick, cfg, None, max_ctx)
2678 }
2679
2680 pub fn new_ppn_planned(
2681 devs: &[&dyn KvDev],
2682 fence: &[usize],
2683 cfg: &ModelConfig,
2684 plan: &memra_gguf::model_plan::ModelPlan,
2685 max_ctx: usize,
2686 ) -> Result<Self, Box<dyn std::error::Error>> {
2687 assert_eq!(
2688 devs.len() + 1,
2689 fence.len(),
2690 "ppn cache: devs vs fence mismatch"
2691 );
2692 let pick = |il: usize| -> &dyn KvDev {
2693 let stage = match fence[1..fence.len() - 1].binary_search(&il) {
2694 Ok(index) => index + 1,
2695 Err(index) => index,
2696 };
2697 devs[stage.min(devs.len() - 1)]
2698 };
2699 Self::new_inner(&pick, cfg, Some(plan), max_ctx)
2700 }
2701
2702 fn new_inner<'a>(
2705 pick: &dyn Fn(usize) -> &'a dyn KvDev,
2706 cfg: &ModelConfig,
2707 plan: Option<&memra_gguf::model_plan::ModelPlan>,
2708 max_ctx: usize,
2709 ) -> Result<Self, Box<dyn std::error::Error>> {
2710 let fallback_plan = if plan.is_none() {
2711 Some(ModelPlan::compile(cfg)?)
2712 } else {
2713 None
2714 };
2715 let plan = plan
2716 .or(fallback_plan.as_ref())
2717 .expect("cache allocation requires a ModelPlan");
2718 let n = cfg.n_layer as usize;
2719 let mut kv = Vec::with_capacity(n);
2720 let mut recur = Vec::with_capacity(n);
2721 let mut latent = Vec::with_capacity(n);
2722 let head_dim_k = cfg.head_dim_k as usize;
2723 let head_dim_v = cfg.head_dim_v as usize;
2724 for il in 0..cfg.n_layer {
2725 let e = pick(il as usize);
2727 let layer = plan
2728 .layers
2729 .iter()
2730 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2731 .find(|layer| layer.index == il)
2732 .ok_or_else(|| format!("cache ModelPlan has no layer {il}"))?;
2733 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2738 if g4_shared > 0 && il >= cfg.n_layer - g4_shared {
2739 kv.push(None);
2740 recur.push(None);
2741 latent.push(None);
2742 continue;
2743 }
2744 match layer.state {
2745 StatePlan::KvCache { .. } | StatePlan::SlidingKvCache { .. } => {
2746 assert!(
2751 head_dim_k.is_multiple_of(32) && head_dim_v.is_multiple_of(32),
2752 "KVQUANT requires head_dim_k%32==0 && head_dim_v%32==0 \
2753 (layer {il}: k={head_dim_k} v={head_dim_v})"
2754 );
2755 let (kv_dim_k, kv_dim_v, kbb_l, vbb_l) =
2758 full_attention_kv_layout(cfg, plan, il);
2759 let k_tok_bytes = (kv_dim_k / 32) * kbb_l;
2760 let v_tok_bytes = (kv_dim_v / 32) * vbb_l;
2761 let planned_window = plan
2762 .layers
2763 .iter()
2764 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2765 .find(|layer| layer.index == il)
2766 .and_then(|layer| match layer.state {
2767 StatePlan::SlidingKvCache { window, .. } => Some(window),
2768 _ => None,
2769 });
2770 let ring = if swa_ring_on() {
2771 planned_window.map(|window| {
2772 let window = window as usize;
2773 KvRing::new(swa_ring_rows(window, max_ctx), window)
2774 })
2775 } else {
2776 None
2777 };
2778 let alloc_rows = ring.as_ref().map(KvRing::rows).unwrap_or(max_ctx);
2779 kv.push(Some(KvLayer {
2780 k: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, k_tok_bytes))?,
2784 v: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, v_tok_bytes))?,
2785 kv_dim_k,
2786 kv_dim_v,
2787 k_tok_bytes,
2788 v_tok_bytes,
2789 len: 0,
2790 ring,
2791 len_d: e.htod_i32(&[0])?,
2792 base_d: None,
2793 }));
2794 recur.push(None);
2795 latent.push(None);
2796 }
2797 StatePlan::Recurrent {
2798 conv_width,
2799 conv_kernel,
2800 state_width,
2801 } => {
2802 kv.push(None);
2803 recur.push(Some(RecurLayer {
2804 conv_state: e.zeros(
2805 conv_width as usize * (conv_kernel as usize).saturating_sub(1),
2806 )?,
2807 ssm_state: e.zeros(state_width as usize)?,
2808 ssm_state_alt: e.zeros(state_width as usize)?,
2809 }));
2810 latent.push(None);
2811 }
2812 StatePlan::LatentKvCache { width, index_width } => {
2813 let width = width as usize;
2817 assert!(
2818 width > 0,
2819 "layer {il}: LatentKvCache width must be positive"
2820 );
2821 kv.push(None);
2822 recur.push(None);
2823 let index_width = index_width as usize;
2824 let index_ring = if index_width == 0 {
2828 None
2829 } else {
2830 index_ring_rows(max_ctx)
2831 };
2832 let index_rows = match index_width {
2833 0 => None,
2834 w => Some(e.zeros(index_ring.unwrap_or(max_ctx) * w)?),
2835 };
2836 latent.push(Some(LatentKvLayer {
2837 rows: e.zeros(max_ctx * width)?,
2838 width,
2839 len: 0,
2840 len_d: e.htod_i32(&[0])?,
2841 index_rows,
2842 index_width,
2843 index_ring_rows: index_ring,
2844 index_pool_keys: None,
2847 index_pools_ready: 0,
2848 index_pool: 0,
2849 }));
2850 }
2851 ref state => {
2852 return Err(format!(
2853 "native cache allocator has no implementation for layer {il} state {state:?}"
2854 )
2855 .into());
2856 }
2857 }
2858 }
2859 Ok(Cache {
2860 kv,
2861 recur,
2862 latent,
2863 tp_kv: (0..n).map(|_| None).collect(),
2864 glm5_tp_recur: (0..n).map(|_| None).collect(),
2865 glm5_tp_latent_peer: (0..n).map(|_| None).collect(),
2866 pos: 0,
2867 max_ctx,
2868 tainted: false,
2869 dflash_taps: None,
2870 hc_taps: None,
2871 glm5_decode_graph: None,
2872 last_logits_dev: None,
2873 })
2874 }
2875
2876 pub fn has_swa_ring(&self) -> bool {
2877 self.kv.iter().flatten().any(|layer| layer.ring.is_some())
2878 || self
2879 .tp_kv
2880 .iter()
2881 .flatten()
2882 .any(|layer| layer.ring_window().is_some())
2883 }
2884
2885 pub fn can_rollback(&self, snap: &CacheSnapshot, accept_len: usize) -> bool {
2886 let local = self
2887 .kv
2888 .iter()
2889 .zip(&snap.kv_len)
2890 .all(|(layer, saved)| match (layer, saved) {
2891 (Some(layer), Some(saved)) => layer
2892 .ring
2893 .as_ref()
2894 .is_none_or(|ring| ring.can_rewind_to(saved + accept_len)),
2895 _ => true,
2896 });
2897 let tensor = self
2898 .tp_kv
2899 .iter()
2900 .zip(&snap.tp_kv_len)
2901 .all(|(layer, saved)| match (layer, saved) {
2902 (Some(layer), Some(saved)) => saved
2903 .checked_add(accept_len)
2904 .is_some_and(|target| layer.can_rewind_to(target)),
2905 _ => true,
2906 });
2907 local && tensor
2908 }
2909
2910 pub fn snapshot(&self, e: &impl KvDev) -> Result<CacheSnapshot, Box<dyn std::error::Error>> {
2914 if self.glm5_tp_recur.iter().any(Option::is_some)
2918 || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2919 {
2920 return Err(
2921 "cache snapshot is unwired for glm5 TP rank state (MEMRA_GLM5_TP): \
2922 per-rank planes are not carried by CacheSnapshot"
2923 .into(),
2924 );
2925 }
2926 self.ensure_usable("cache snapshot")?;
2927 let n = self.kv.len();
2928 let mut kv_len = Vec::with_capacity(n);
2929 let mut tp_kv_len = Vec::with_capacity(n);
2930 let mut conv = Vec::with_capacity(n);
2931 let mut ssm = Vec::with_capacity(n);
2932 for il in 0..n {
2933 match &self.kv[il] {
2934 Some(kvl) => kv_len.push(Some(kvl.len)),
2935 None => kv_len.push(None),
2936 }
2937 tp_kv_len.push(
2938 self.tp_kv[il]
2939 .as_ref()
2940 .map(ResidentTpKvCache::committed_len),
2941 );
2942 match &self.recur[il] {
2943 Some(rl) => {
2944 conv.push(Some(e.clone_dtod(&rl.conv_state)?));
2945 ssm.push(Some(e.clone_dtod(&rl.ssm_state)?));
2946 }
2947 None => {
2948 conv.push(None);
2949 ssm.push(None);
2950 }
2951 }
2952 }
2953 Ok(CacheSnapshot {
2954 kv_len,
2955 tp_kv_len,
2956 conv,
2957 ssm,
2958 pos: self.pos,
2959 })
2960 }
2961
2962 pub fn snapshot_into(
2967 &self,
2968 e: &impl KvDev,
2969 snap: &mut CacheSnapshot,
2970 ) -> Result<(), Box<dyn std::error::Error>> {
2971 if self.glm5_tp_recur.iter().any(Option::is_some)
2972 || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2973 {
2974 return Err("cache snapshot_into is unwired for glm5 TP rank state \
2975 (MEMRA_GLM5_TP): per-rank planes are not carried by CacheSnapshot"
2976 .into());
2977 }
2978 self.ensure_usable("cache snapshot refresh")?;
2979 let n = self.kv.len();
2980 for il in 0..n {
2981 snap.kv_len[il] = self.kv[il].as_ref().map(|kvl| kvl.len);
2982 snap.tp_kv_len[il] = self.tp_kv[il]
2983 .as_ref()
2984 .map(ResidentTpKvCache::committed_len);
2985 if let Some(rl) = &self.recur[il] {
2986 let dc = snap.conv[il]
2987 .as_mut()
2988 .expect("snapshot_into: shape mismatch (conv)");
2989 let ds = snap.ssm[il]
2990 .as_mut()
2991 .expect("snapshot_into: shape mismatch (ssm)");
2992 let (cn, sn) = (rl.conv_state.len(), rl.ssm_state.len());
2993 e.copy_into(dc, 0, &rl.conv_state, cn)?;
2994 e.copy_into(ds, 0, &rl.ssm_state, sn)?;
2995 }
2996 }
2997 snap.pos = self.pos;
2998 Ok(())
2999 }
3000
3001 pub fn rollback(
3009 &mut self,
3010 e: &impl KvDev,
3011 snap: &CacheSnapshot,
3012 accept_len: usize,
3013 ) -> Result<(), Box<dyn std::error::Error>> {
3014 self.ensure_usable("cache rollback")?;
3015 if !self.can_rollback(snap, accept_len) {
3016 return Err(
3017 "SWA ring rewind checkpoint has been lapped; full re-prime required".into(),
3018 );
3019 }
3020 for il in 0..self.kv.len() {
3021 if let (Some(kvl), Some(saved)) = (self.kv[il].as_mut(), snap.kv_len[il]) {
3022 kvl.len = saved + accept_len;
3023 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
3028 }
3029 if let (Some(kvl), Some(saved)) = (self.tp_kv[il].as_mut(), snap.tp_kv_len[il]) {
3030 kvl.rewind_to(saved + accept_len)?;
3031 }
3032 if let Some(rl) = self.recur[il].as_mut() {
3033 if let Some(c) = &snap.conv[il] {
3034 e.copy_into(&mut rl.conv_state, 0, c, c.len())?;
3035 }
3036 if let Some(s) = &snap.ssm[il] {
3037 e.copy_into(&mut rl.ssm_state, 0, s, s.len())?;
3038 }
3039 }
3040 }
3041 self.pos = snap.pos;
3042 Ok(())
3043 }
3044}
3045
3046#[cfg(test)]
3047mod tp_transaction_tests {
3048 use super::{
3049 Cache, INDEX_RING_WORKING_ROWS, KvRingAppend, ResidentTpKvCache, TpKvTransactionState,
3050 index_ring_default_rows, index_ring_rows_for, index_ring_take, tp_kv_rank_allocation_shape,
3051 };
3052
3053 const GLM5_NEXT_POOL: usize = 4;
3056 const GLM5_NEXT_STATE_ROW_BYTES: usize = 2 * 128 * 4;
3058 const RING_BYTES_PER_LAYER: usize = INDEX_RING_WORKING_ROWS * GLM5_NEXT_STATE_ROW_BYTES;
3061
3062 #[test]
3081 fn the_derived_ring_serves_a_monolithic_prime_of_the_whole_configured_context() {
3082 for max_ctx in [8192usize, 262_144, 1 << 20] {
3085 let rows = index_ring_default_rows(max_ctx).unwrap_or_else(|| {
3086 panic!("the ring must engage at max_ctx {max_ctx}: it is where the saving is")
3087 });
3088 let ring = rows / GLM5_NEXT_POOL * GLM5_NEXT_POOL;
3091
3092 let mut cur = 0usize;
3094 let mut pools_ready = 0usize;
3095 let mut steps = 0usize;
3096 while cur < max_ctx {
3097 let take = index_ring_take(ring, GLM5_NEXT_POOL, pools_ready, cur, max_ctx - cur)
3098 .unwrap_or_else(|| {
3099 panic!(
3100 "MEMRA_CTX={max_ctx}: the {ring}-row ring refused a monolithic prime \
3101 after {cur} of {max_ctx} tokens ({}% of the configured context). \
3102 USABLE CONTEXT MUST BE AT LEAST CONFIGURED CONTEXT. This is the \
3103 4630-of-8192 regression, in arithmetic.",
3104 cur * 100 / max_ctx
3105 )
3106 });
3107 assert!(
3108 take > 0,
3109 "MEMRA_CTX={max_ctx}: the drain made no progress at row {cur}: a zero take \
3110 is an infinite loop in the engine, not a refusal"
3111 );
3112 cur += take;
3113 pools_ready = cur / GLM5_NEXT_POOL;
3114 steps += 1;
3115 assert!(
3116 steps <= max_ctx,
3117 "MEMRA_CTX={max_ctx}: the drain did not terminate"
3118 );
3119 }
3120 assert_eq!(cur, max_ctx, "the whole prompt must be appended");
3121
3122 let ring_bytes = rows * GLM5_NEXT_STATE_ROW_BYTES;
3128 let flat_bytes = max_ctx * GLM5_NEXT_STATE_ROW_BYTES;
3129 assert_eq!(
3130 ring_bytes, RING_BYTES_PER_LAYER,
3131 "MEMRA_CTX={max_ctx}: the ring books {rows} rows, not the context-independent \
3132 {INDEX_RING_WORKING_ROWS}. A plane that tracks max_ctx is the flat plane \
3133 wearing a modulus"
3134 );
3135 assert!(
3136 rows < max_ctx,
3137 "MEMRA_CTX={max_ctx}: a ring of {rows} rows is not shorter than the flat plane \
3138 it replaces, so it would not engage at all"
3139 );
3140 assert!(
3144 ring_bytes <= 16 << 20,
3145 "MEMRA_CTX={max_ctx}: {} MiB per MLA layer, over glm5_next's 12 of them. The ring \
3146 exists to delete 11.94 GiB; a working set this large is not paying for itself",
3147 ring_bytes >> 20
3148 );
3149 println!(
3150 "MEMRA_CTX={max_ctx}: ring {rows} rows (effective {ring}), monolithic prime of \
3151 {max_ctx} tokens admitted in {steps} drain step(s); plane {} MiB/layer vs flat \
3152 {} MiB/layer",
3153 ring_bytes >> 20,
3154 flat_bytes >> 20
3155 );
3156 }
3157 }
3158
3159 fn empty_tp_cache(capacity: usize) -> ResidentTpKvCache {
3160 ResidentTpKvCache::new(Vec::new(), 128, 128, 136, 96, capacity)
3161 }
3162
3163 #[test]
3164 fn step_tp8_rank_allocation_matches_the_official_kv_geometry() {
3165 let shape = tp_kv_rank_allocation_shape(8 * 128, 8 * 128, 8).unwrap();
3166 assert_eq!((shape.kv_dim_k, shape.kv_dim_v), (128, 128));
3167 assert_eq!((shape.k_token_bytes, shape.v_token_bytes), (136, 96));
3168 assert_eq!(shape.bytes_per_token(), 232);
3169 assert_eq!(shape.fixed_bytes, 20);
3170 assert_eq!(shape.allocation_bytes(262_144), 232 * 262_144 + 20);
3171 }
3172
3173 #[test]
3174 fn tp_rank_allocation_refuses_non_divisible_and_non_block_aligned_shards() {
3175 assert!(tp_kv_rank_allocation_shape(1024, 1024, 3).is_err());
3176 assert!(tp_kv_rank_allocation_shape(1024, 1024, 64).is_err());
3177 assert!(tp_kv_rank_allocation_shape(0, 1024, 8).is_err());
3178 }
3179
3180 #[test]
3181 fn partial_commit_publishes_only_the_accepted_prefix() {
3182 let mut state = TpKvTransactionState::new();
3183 let transaction = state.begin().unwrap();
3184 let staged = state.append_target(transaction, 3, 8).unwrap();
3185 state.publish_append(transaction, staged).unwrap();
3186 assert_eq!(state.committed_len, 0);
3187 assert_eq!(state.staged_len, 3);
3188
3189 let committed = state.commit_target(transaction, 2).unwrap();
3190 state.publish_finalize(transaction, committed).unwrap();
3191 assert_eq!(state.committed_len, 2);
3192 assert_eq!(state.staged_len, 2);
3193 assert!(state.active.is_none());
3194 assert!(state.validate(transaction).is_err());
3195 }
3196
3197 #[test]
3198 fn rollback_restores_the_committed_boundary() {
3199 let mut state = TpKvTransactionState::new();
3200 let first = state.begin().unwrap();
3201 let staged = state.append_target(first, 1, 8).unwrap();
3202 state.publish_append(first, staged).unwrap();
3203 let committed = state.commit_target(first, 1).unwrap();
3204 state.publish_finalize(first, committed).unwrap();
3205
3206 let speculative = state.begin().unwrap();
3207 let staged = state.append_target(speculative, 2, 8).unwrap();
3208 state.publish_append(speculative, staged).unwrap();
3209 assert_eq!(state.committed_len, 1);
3210 assert_eq!(state.staged_len, 3);
3211 state
3212 .publish_finalize(speculative, speculative.base_len)
3213 .unwrap();
3214 assert_eq!(state.committed_len, 1);
3215 assert_eq!(state.staged_len, 1);
3216 assert!(state.validate(speculative).is_err());
3217 }
3218
3219 #[test]
3220 fn index_ring_sizing_is_pure_and_carries_no_per_call_t() {
3221 let rows = INDEX_RING_WORKING_ROWS;
3224 assert_eq!(index_ring_rows_for(None, 1 << 20), Some(rows));
3225 assert_eq!(index_ring_default_rows(1 << 20), Some(rows));
3226 assert_eq!(index_ring_rows_for(None, 4096), None);
3229 assert_eq!(index_ring_rows_for(None, rows), None);
3230 assert_eq!(index_ring_rows_for(None, rows + 1), Some(rows));
3231
3232 for max_ctx in [8192usize, 262_144, 1 << 20] {
3237 assert_eq!(
3238 index_ring_rows_for(None, max_ctx),
3239 Some(INDEX_RING_WORKING_ROWS),
3240 "the derived ring must not vary with the configured context"
3241 );
3242 }
3243
3244 assert_eq!(index_ring_rows_for(Some(0), 1 << 20), None);
3247 assert_eq!(index_ring_rows_for(Some(16), 64), Some(16));
3248 assert_eq!(index_ring_rows_for(Some(64), 64), None);
3249 }
3250
3251 #[test]
3253 fn index_ring_take_drains_instead_of_bounding_the_call() {
3254 const POOL: usize = GLM5_NEXT_POOL;
3255 assert_eq!(index_ring_take(0, POOL, 0, 0, 1 << 20), Some(1 << 20));
3257 assert_eq!(index_ring_take(64, POOL, 0, 0, 1024), Some(64));
3260 for cur in 0..64usize {
3263 let ready = cur / POOL;
3264 let take = index_ring_take(64, POOL, ready, cur, 1024).expect("never lapses");
3265 assert!(
3266 (64 - POOL + 1..=64).contains(&take),
3267 "cur {cur}: take {take} outside the guaranteed progress band"
3268 );
3269 }
3270 assert_eq!(index_ring_take(64, POOL, 4, 16, 1), Some(1));
3272 assert_eq!(index_ring_take(16, POOL, 0, 64, 1), None);
3275 assert_eq!(index_ring_take(16, POOL, 0, 16, 1), None);
3276 assert_eq!(index_ring_take(16, POOL, 0, 15, 1), Some(1));
3277 }
3278
3279 #[test]
3280 fn rejects_nested_stale_and_out_of_range_actions() {
3281 let mut state = TpKvTransactionState::new();
3282 let transaction = state.begin().unwrap();
3283 assert!(state.begin().is_err());
3284 assert!(state.append_target(transaction, 0, 2).is_err());
3285 assert!(state.append_target(transaction, 3, 2).is_err());
3286 let staged = state.append_target(transaction, 2, 2).unwrap();
3287 state.publish_append(transaction, staged).unwrap();
3288 assert!(state.commit_target(transaction, 3).is_err());
3289 state.publish_finalize(transaction, 0).unwrap();
3290 assert!(state.publish_append(transaction, 1).is_err());
3291 }
3292
3293 #[test]
3294 fn rewind_resets_visibility_and_invalidates_an_active_transaction() {
3295 let mut state = TpKvTransactionState::new();
3296 let transaction = state.begin().unwrap();
3297 let staged = state.append_target(transaction, 3, 8).unwrap();
3298 state.publish_append(transaction, staged).unwrap();
3299 state.rewind(1, 8).unwrap();
3300 assert_eq!(state.committed_len, 1);
3301 assert_eq!(state.staged_len, 1);
3302 assert!(state.active.is_none());
3303 assert!(state.validate(transaction).is_err());
3304 assert!(state.rewind(9, 8).is_err());
3305 }
3306
3307 #[test]
3308 fn device_rewind_updates_host_visibility_after_external_rank_writes() {
3309 let mut cache = empty_tp_cache(8);
3310 let transaction = cache.begin_transaction().unwrap();
3311 let staged = cache.append_target(transaction, 5).unwrap();
3312 cache.publish_append(transaction, staged).unwrap();
3313 let committed = cache.commit_target(transaction, 5).unwrap();
3314 cache.publish_finalize(transaction, committed).unwrap();
3315 cache.publish_device_rewind(3).unwrap();
3316 assert_eq!(cache.committed_len(), 3);
3317 assert_eq!(cache.staged_len(), 3);
3318 assert!(cache.publish_device_rewind(9).is_err());
3319 }
3320
3321 #[test]
3322 fn grow_preserves_generation_and_publishes_only_the_checkpoint_prefix() {
3323 let mut source = empty_tp_cache(8);
3324 let first = source.begin_transaction().unwrap();
3325 let staged = source.append_target(first, 5).unwrap();
3326 source.publish_append(first, staged).unwrap();
3327 let committed = source.commit_target(first, 5).unwrap();
3328 source.publish_finalize(first, committed).unwrap();
3329
3330 let rolled_back = source.begin_transaction().unwrap();
3331 source
3332 .publish_finalize(rolled_back, rolled_back.base_len())
3333 .unwrap();
3334 let plan = source.prepare_grow(16, 3).unwrap();
3335 assert_eq!(plan.rows(), 3);
3336 assert_eq!(plan.k_bytes(), 3 * 136);
3337 assert_eq!(plan.v_bytes(), 3 * 96);
3338
3339 let mut target = empty_tp_cache(16);
3340 target.publish_grow(plan).unwrap();
3341 assert_eq!(target.committed_len(), 3);
3342 assert_eq!(target.staged_len(), 3);
3343 assert_eq!(target.capacity(), 16);
3344 let next = target.begin_transaction().unwrap();
3345 assert_eq!(next.generation(), rolled_back.generation() + 1);
3346 assert_eq!(next.base_len(), 3);
3347 }
3348
3349 #[test]
3350 fn grow_refuses_active_source_and_invalid_target_state_or_layout() {
3351 let mut active = empty_tp_cache(8);
3352 active.begin_transaction().unwrap();
3353 assert!(active.prepare_grow(16, 0).is_err());
3354
3355 let mut source = empty_tp_cache(8);
3356 source.rewind_to(5).unwrap();
3357 assert!(source.prepare_grow(8, 5).is_err());
3358 assert!(source.prepare_grow(16, 6).is_err());
3359 let plan = source.prepare_grow(16, 4).unwrap();
3360
3361 let mut wrong_layout = ResidentTpKvCache::new(Vec::new(), 128, 128, 144, 96, 16);
3362 assert!(wrong_layout.publish_grow(plan).is_err());
3363
3364 let plan = source.prepare_grow(16, 4).unwrap();
3365 let mut dirty_target = empty_tp_cache(16);
3366 dirty_target.rewind_to(1).unwrap();
3367 assert!(dirty_target.publish_grow(plan).is_err());
3368 }
3369
3370 #[test]
3371 fn swa_transaction_rebase_preserves_the_rollback_window() {
3372 let mut cache = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3373 assert_eq!(cache.physical_capacity(), 32 + 4096 + 512 + 31);
3374 cache.publish_hydration(8762, 4096).unwrap();
3378 assert_eq!(cache.ring_base(), Some(4096));
3379
3380 let transaction = cache.begin_transaction().unwrap();
3381 let plan = cache.prepare_append(transaction, 10).unwrap();
3382 assert_eq!(plan.target(), 8772);
3383 assert_eq!(plan.write_row(), 58);
3384 assert_eq!(
3385 plan.ring_append(),
3386 Some(KvRingAppend::Rebase {
3387 src_row: 4608,
3388 keep_rows: 58,
3389 new_base: 8704,
3390 write_row: 58,
3391 })
3392 );
3393 cache.publish_append_rebase(plan).unwrap();
3394 cache.publish_append_plan(plan).unwrap();
3395 assert_eq!(cache.ring_base(), Some(8704));
3396 assert_eq!(cache.physical_range(8740, 8772).unwrap(), 36..68);
3397
3398 let rollback = cache.commit_target(transaction, 0).unwrap();
3399 cache.publish_finalize(transaction, rollback).unwrap();
3400 assert_eq!((cache.committed_len(), cache.staged_len()), (8762, 8762));
3401 assert!(cache.rewind_to(8200).is_err());
3402 }
3403
3404 #[test]
3405 fn swa_grow_normalizes_only_the_live_prefix_and_preserves_generation() {
3406 let mut source = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3407 source.publish_hydration(8250, 4096).unwrap();
3408 let transaction = source.begin_transaction().unwrap();
3409 source
3410 .publish_finalize(transaction, transaction.base_len())
3411 .unwrap();
3412
3413 let plan = source.prepare_grow(20_000, 8250).unwrap();
3414 assert_eq!(plan.source_row(), 4096);
3415 assert_eq!(plan.copy_rows(), 58);
3416 assert_eq!(plan.k_bytes(), 58 * 136);
3417 assert_eq!(plan.v_bytes(), 58 * 96);
3418
3419 let mut target = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 20_000, 32);
3420 target.publish_grow(plan).unwrap();
3421 assert_eq!(target.ring_base(), Some(8192));
3422 assert_eq!((target.committed_len(), target.staged_len()), (8250, 8250));
3423 assert_eq!(target.physical_range(8192, 8250).unwrap(), 0..58);
3424 let next = target.begin_transaction().unwrap();
3425 assert_eq!(next.generation(), transaction.generation() + 1);
3426 }
3427
3428 #[test]
3429 fn cache_reports_a_materialized_distributed_swa_ring() {
3430 let mut cache = Cache {
3431 kv: Vec::new(),
3432 recur: Vec::new(),
3433 latent: Vec::new(),
3434 tp_kv: vec![None],
3435 glm5_tp_recur: vec![None],
3436 glm5_tp_latent_peer: vec![None],
3437 pos: 0,
3438 max_ctx: 10_000,
3439 tainted: false,
3440 dflash_taps: None,
3441 hc_taps: None,
3442 glm5_decode_graph: None,
3443 last_logits_dev: None,
3444 };
3445 assert!(!cache.has_swa_ring());
3446 cache.tp_kv[0] = Some(ResidentTpKvCache::new_swa(
3447 Vec::new(),
3448 128,
3449 128,
3450 136,
3451 96,
3452 10_000,
3453 512,
3454 ));
3455 assert!(cache.has_swa_ring());
3456 }
3457}
3458
3459#[cfg(test)]
3460mod swa_ring_tests {
3461 use super::{
3462 KvRing, KvRingAppend, PRIME_CHUNK_MAX_TOKENS, SWA_REWIND_SLACK_ROWS,
3463 SWA_VIEW_ALIGNMENT_ROWS, kv_plane_allocation_bytes, swa_retain_from, swa_ring_rows,
3464 };
3465
3466 #[test]
3467 fn allocation_rows_cover_window_max_prime_and_alignment_slack() {
3468 assert_eq!(swa_ring_rows(512, 262_144), 512 + 4096 + 512 + 31);
3469 assert_eq!(swa_ring_rows(512, 4096), 4096);
3470 assert_eq!(
3471 kv_plane_allocation_bytes(5151, 1088),
3472 5151 * 1088 + 8,
3473 "the Step35 session plane allocates ring rows plus the existing tail pad",
3474 );
3475 }
3476
3477 #[test]
3490 fn retain_grants_rewind_slack_but_never_asks_below_base() {
3491 const WINDOW: usize = 512;
3492 let rows = swa_ring_rows(WINDOW, 262_144);
3493 let len = rows;
3494 let aligned = |pos: usize| (pos - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3495
3496 let retain = swa_retain_from(len, WINDOW, 0);
3499 assert!(retain <= aligned(len) - SWA_REWIND_SLACK_ROWS);
3500 let mut ring = KvRing::new(rows, WINDOW);
3501 ring.apply_rebase(retain);
3502 assert!(
3503 ring.can_rewind_to(len - 1),
3504 "a one-token rewind must survive the rebase"
3505 );
3506 assert!(ring.can_rewind_to(len - SWA_REWIND_SLACK_ROWS));
3507
3508 assert!(len - retain + PRIME_CHUNK_MAX_TOKENS <= rows);
3510
3511 for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3515 assert!(
3516 ring.can_rewind_to(len - depth),
3517 "a {depth}-row rewind must be resident, not clamped away",
3518 );
3519 }
3520
3521 let base = ring.base();
3528 for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3529 let window_start = (len - depth - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3530 assert!(
3531 window_start >= base,
3532 "after a {depth}-row rewind the window starts at {window_start}, below base \
3533 {base} — clamping here would serve rows the ring no longer holds (the pos-8661 \
3534 all-NaN case)",
3535 );
3536 assert!(swa_retain_from(len - depth, WINDOW, base) >= base);
3537 }
3538
3539 let past = len - (SWA_REWIND_SLACK_ROWS + WINDOW);
3542 let past_start = (past - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3543 assert!(
3544 past_start < base,
3545 "beyond the headroom the window must fall below base"
3546 );
3547 assert!(
3548 !ring.can_rewind_to(past),
3549 "and can_rewind_to must refuse it"
3550 );
3551 }
3552
3553 #[test]
3554 fn ring_matches_flat_bytes_before_wrap() {
3555 let ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3556 let flat: Vec<u32> = (0..1024).collect();
3557 let mut physical = vec![u32::MAX; ring.rows()];
3558 let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, flat.len()).unwrap()
3559 else {
3560 panic!("first append unexpectedly wrapped")
3561 };
3562 physical[write_row..write_row + flat.len()].copy_from_slice(&flat);
3563 let view = ring.physical_range(0, flat.len()).unwrap();
3564 assert_eq!(&physical[view], flat.as_slice());
3565 }
3566
3567 #[test]
3568 fn wrap_rebases_the_exact_aligned_prime_view() {
3569 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3570 let flat: Vec<u32> = (0..8192).collect();
3571 let mut physical = vec![u32::MAX; ring.rows()];
3572 let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, 4096).unwrap() else {
3573 panic!("first prime chunk unexpectedly wrapped")
3574 };
3575 physical[write_row..write_row + 4096].copy_from_slice(&flat[..4096]);
3576
3577 let off = (4096usize - (512 - 1)) & !31usize;
3578 let KvRingAppend::Rebase {
3579 src_row,
3580 keep_rows,
3581 new_base,
3582 write_row,
3583 } = ring.append_plan(4096, off, 4096).unwrap()
3584 else {
3585 panic!("second prime chunk did not wrap")
3586 };
3587 let retained = physical[src_row..src_row + keep_rows].to_vec();
3588 physical[..keep_rows].copy_from_slice(&retained);
3589 ring.apply_rebase(new_base);
3590 physical[write_row..write_row + 4096].copy_from_slice(&flat[4096..8192]);
3591
3592 let view = ring.physical_range(off, 8192).unwrap();
3593 assert_eq!(&physical[view], &flat[off..8192]);
3594 assert_eq!(ring.base(), off);
3595 }
3596
3597 #[test]
3598 fn rewind_declines_once_the_required_window_was_lapped() {
3599 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3600 let KvRingAppend::Rebase { new_base, .. } = ring.append_plan(4096, 3584, 4096).unwrap()
3601 else {
3602 panic!("expected wrap")
3603 };
3604 ring.apply_rebase(new_base);
3605 assert!(ring.can_rewind_to(4095));
3606 assert!(!ring.can_rewind_to(4094));
3607 assert!(!ring.can_rewind_to(0));
3608 }
3609
3610 #[test]
3615 fn restore_plan_copies_the_window_not_the_absolute_length() {
3616 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3617 let (base, phys) = ring.restore_plan(400).unwrap();
3619 assert_eq!((base, phys), (0, 0..400));
3620
3621 let mut live = 0usize;
3624 while live < 40_960 {
3625 let retain = swa_retain_from(live, 512, ring.base());
3626 if let KvRingAppend::Rebase { new_base, .. } =
3627 ring.append_plan(live, retain, 4096).unwrap()
3628 {
3629 ring.apply_rebase(new_base);
3630 }
3631 live += 4096;
3632 }
3633 assert!(ring.base() > 0, "a 40k walk must have lapped the ring");
3634 let (base, phys) = ring.restore_plan(live).unwrap();
3635 assert_eq!(base, (live - (512 - 1)) & !31usize);
3636 assert!(
3637 base >= ring.base(),
3638 "the plan must stay above the ring floor"
3639 );
3640 assert_eq!(phys.len(), live - base);
3641 assert!(
3642 phys.end <= ring.rows(),
3643 "the copy must fit the physical buffer ({} rows), got {:?}",
3644 ring.rows(),
3645 phys
3646 );
3647
3648 assert!(ring.restore_plan(400).is_err());
3650 }
3651}