1pub fn kv_blk_bytes() -> (usize, usize) {
17 (34, 24)
18}
19
20#[derive(Clone, Copy, Debug, PartialEq, Eq)]
26pub struct TpKvRankAllocationShape {
27 pub kv_dim_k: usize,
28 pub kv_dim_v: usize,
29 pub k_token_bytes: usize,
30 pub v_token_bytes: usize,
31 pub fixed_bytes: usize,
32}
33
34impl TpKvRankAllocationShape {
35 pub fn bytes_per_token(self) -> usize {
36 self.k_token_bytes.saturating_add(self.v_token_bytes)
37 }
38
39 pub fn allocation_bytes(self, capacity: usize) -> usize {
40 self.bytes_per_token()
41 .saturating_mul(capacity)
42 .saturating_add(self.fixed_bytes)
43 }
44}
45
46#[allow(clippy::manual_is_multiple_of)] pub fn tp_kv_rank_allocation_shape(
48 kv_dim_k: usize,
49 kv_dim_v: usize,
50 ranks: usize,
51) -> Result<TpKvRankAllocationShape, String> {
52 if ranks == 0 || kv_dim_k == 0 || kv_dim_v == 0 {
53 return Err(format!(
54 "TP KV dimensions and rank count must be nonzero: k={kv_dim_k} v={kv_dim_v} \
55 ranks={ranks}"
56 ));
57 }
58 if kv_dim_k % ranks != 0 || kv_dim_v % ranks != 0 {
59 return Err(format!(
60 "TP KV dimensions k={kv_dim_k} v={kv_dim_v} are not divisible by TP={ranks}"
61 ));
62 }
63 let local_k = kv_dim_k / ranks;
64 let local_v = kv_dim_v / ranks;
65 if !local_k.is_multiple_of(32) || !local_v.is_multiple_of(32) {
66 return Err(format!(
67 "TP KV local dimensions k={local_k} v={local_v} must be 32-aligned"
68 ));
69 }
70 let (k_block_bytes, v_block_bytes) = kv_blk_bytes();
71 let k_token_bytes = (local_k / 32)
72 .checked_mul(k_block_bytes)
73 .ok_or("TP KV K token-byte overflow")?;
74 let v_token_bytes = (local_v / 32)
75 .checked_mul(v_block_bytes)
76 .ok_or("TP KV V token-byte overflow")?;
77 Ok(TpKvRankAllocationShape {
78 kv_dim_k: local_k,
79 kv_dim_v: local_v,
80 k_token_bytes,
81 v_token_bytes,
82 fixed_bytes: 8 + 8 + std::mem::size_of::<i32>(),
85 })
86}
87
88pub fn gkv_on() -> bool {
91 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
92 *ON.get_or_init(|| {
93 std::env::var("MEMRA_GEMMA_GKV")
94 .map(|v| v != "0")
95 .unwrap_or(true)
96 })
97}
98
99pub fn wkv_on() -> bool {
103 static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
104 *ON.get_or_init(|| {
105 std::env::var("MEMRA_GEMMA_WKV")
106 .map(|v| v != "0")
107 .unwrap_or_else(|_| std::env::var("MEMRA_DRAFT").is_err())
108 })
109}
110
111static SWA_RING_DEFAULT: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
118
119pub fn set_swa_ring_default(on: bool) {
120 SWA_RING_DEFAULT.store(on, std::sync::atomic::Ordering::Relaxed);
121}
122
123pub fn swa_ring_on() -> bool {
124 static ENV: std::sync::OnceLock<Option<bool>> = std::sync::OnceLock::new();
125 match *ENV.get_or_init(|| match std::env::var("MEMRA_SWA_RING").ok().as_deref() {
126 Some("1") => Some(true),
127 Some("0") => Some(false),
128 _ => None,
129 }) {
130 Some(forced) => forced,
131 None => SWA_RING_DEFAULT.load(std::sync::atomic::Ordering::Relaxed),
132 }
133}
134
135pub const PRIME_CHUNK_MAX_TOKENS: usize = 4096;
138const SWA_VIEW_ALIGNMENT_ROWS: usize = 32;
139
140pub fn swa_ring_rows(window: usize, max_ctx: usize) -> usize {
143 max_ctx.min(
148 window + PRIME_CHUNK_MAX_TOKENS + SWA_REWIND_SLACK_ROWS + (SWA_VIEW_ALIGNMENT_ROWS - 1),
149 )
150}
151
152pub const SWA_REWIND_SLACK_ROWS: usize = 512;
162
163pub fn swa_retain_from(first_row: usize, window: usize, base: usize) -> usize {
173 let ideal = first_row
174 .saturating_sub(window.saturating_sub(1))
175 .saturating_sub(SWA_REWIND_SLACK_ROWS)
176 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
177 ideal.max(base)
178}
179
180pub const INDEX_RING_WORKING_ROWS: usize = 5120;
208
209pub fn index_ring_rows_for(explicit: Option<usize>, max_ctx: usize) -> Option<usize> {
219 let rows = match explicit {
220 Some(0) => return None,
221 Some(n) => n,
222 None => INDEX_RING_WORKING_ROWS,
223 };
224 (rows > 0 && rows < max_ctx).then_some(rows)
225}
226
227pub fn index_ring_default_rows(max_ctx: usize) -> Option<usize> {
231 index_ring_rows_for(None, max_ctx)
232}
233
234#[allow(clippy::manual_is_multiple_of)] pub fn index_ring_take(
255 ring: usize,
256 pool: usize,
257 pools_ready: usize,
258 cur: usize,
259 remaining: usize,
260) -> Option<usize> {
261 if ring == 0 {
262 return Some(remaining);
263 }
264 debug_assert!(
265 ring % pool == 0,
266 "the effective ring is a whole number of pools"
267 );
268 let live = cur.checked_sub(pools_ready.saturating_mul(pool))?;
269 (ring > live).then(|| (ring - live).min(remaining))
270}
271
272pub fn index_ring_rows(max_ctx: usize) -> Option<usize> {
276 let explicit = std::env::var("MEMRA_DSA_INDEX_RING")
277 .ok()
278 .and_then(|v| v.trim().parse::<usize>().ok());
279 index_ring_rows_for(explicit, max_ctx)
280}
281
282#[derive(Debug, Clone, Copy, PartialEq, Eq)]
283pub struct KvRing {
284 rows: usize,
285 window: usize,
286 base: usize,
287}
288
289#[derive(Debug, Clone, Copy, PartialEq, Eq)]
290pub enum KvRingAppend {
291 Contiguous {
292 write_row: usize,
293 },
294 Rebase {
295 src_row: usize,
296 keep_rows: usize,
297 new_base: usize,
298 write_row: usize,
299 },
300}
301
302impl KvRing {
303 pub fn new(rows: usize, window: usize) -> Self {
304 assert!(window > 0 && rows > 0, "invalid SWA ring geometry");
305 Self {
306 rows,
307 window,
308 base: 0,
309 }
310 }
311
312 pub fn rows(&self) -> usize {
313 self.rows
314 }
315 pub fn base(&self) -> usize {
316 self.base
317 }
318 pub fn window(&self) -> usize {
319 self.window
320 }
321
322 pub fn append_plan(
325 &self,
326 len: usize,
327 retain_from: usize,
328 append_rows: usize,
329 ) -> Result<KvRingAppend, String> {
330 if len < self.base || retain_from < self.base || retain_from > len {
331 return Err(format!(
332 "SWA ring lapped required rows (base {}, retain {retain_from}, len {len})",
333 self.base
334 ));
335 }
336 let used = len - self.base;
337 if used > self.rows {
338 return Err(format!(
339 "SWA ring state exceeds capacity ({used} > {})",
340 self.rows
341 ));
342 }
343 if used.saturating_add(append_rows) <= self.rows {
344 return Ok(KvRingAppend::Contiguous {
345 write_row: used % self.rows,
346 });
347 }
348
349 let keep_rows = len - retain_from;
350 if keep_rows.saturating_add(append_rows) > self.rows {
351 return Err(format!(
352 "SWA ring append does not fit (keep {keep_rows} + append {append_rows} > {})",
353 self.rows
354 ));
355 }
356 Ok(KvRingAppend::Rebase {
357 src_row: retain_from - self.base,
358 keep_rows,
359 new_base: retain_from,
360 write_row: keep_rows,
361 })
362 }
363
364 pub fn apply_rebase(&mut self, new_base: usize) {
365 debug_assert!(new_base >= self.base);
366 self.base = new_base;
367 }
368
369 pub fn physical_range(
370 &self,
371 start: usize,
372 end: usize,
373 ) -> Result<std::ops::Range<usize>, String> {
374 if start < self.base || end < start || end - self.base > self.rows {
375 return Err(format!(
376 "SWA ring view [{start},{end}) is outside resident [{},{})",
377 self.base,
378 self.base + self.rows
379 ));
380 }
381 let start_row = (start - self.base) % self.rows;
382 let len = end - start;
383 debug_assert!(
384 start_row + len <= self.rows,
385 "ring view must be contiguous after rebase"
386 );
387 Ok(start_row..start_row + len)
388 }
389
390 pub fn can_rewind_to(&self, len: usize) -> bool {
392 let raw = len.saturating_sub(self.window - 1);
393 let view_start = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
394 view_start >= self.base
395 }
396
397 pub fn restore_plan(&self, len: usize) -> Result<(usize, std::ops::Range<usize>), String> {
404 let raw = len.saturating_sub(self.window.saturating_sub(1));
405 let new_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
406 let physical = self.physical_range(new_base, len)?;
407 Ok((new_base, physical))
408 }
409}
410
411pub trait KvDev {
416 fn zeros(&self, n: usize) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
417 fn uninit(&self, n: usize) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
418 fn alloc_u8(&self, n: usize) -> Result<CudaSlice<u8>, Box<dyn std::error::Error>>;
419 fn htod_i32(&self, v: &[i32]) -> Result<CudaSlice<i32>, Box<dyn std::error::Error>>;
420 fn clone_dtod(
421 &self,
422 src: &CudaSlice<f32>,
423 ) -> Result<CudaSlice<f32>, Box<dyn std::error::Error>>;
424 fn copy_into(
425 &self,
426 dst: &mut CudaSlice<f32>,
427 off: usize,
428 src: &CudaSlice<f32>,
429 len: usize,
430 ) -> Result<(), Box<dyn std::error::Error>>;
431 fn copy_range_into(
435 &self,
436 dst: &mut CudaSlice<f32>,
437 dst_off: usize,
438 src: &CudaSlice<f32>,
439 src_off: usize,
440 len: usize,
441 ) -> Result<(), Box<dyn std::error::Error>>;
442 fn set_i32_one(&self, d: &mut CudaSlice<i32>, v: i32)
443 -> Result<(), Box<dyn std::error::Error>>;
444}
445
446use cudarc::driver::CudaSlice;
447use memra_gguf::config::{LayerKind, ModelConfig};
448use memra_gguf::model_plan::{ModelPlan, ResidualTopology, StatePlan};
449
450pub struct KvLayer {
455 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,
462 pub ring: Option<KvRing>,
465 pub len_d: CudaSlice<i32>,
469 pub base_d: Option<CudaSlice<i32>>,
478}
479
480impl KvLayer {
481 pub fn physical_rows(
482 &self,
483 start: usize,
484 end: usize,
485 ) -> Result<std::ops::Range<usize>, String> {
486 match &self.ring {
487 Some(ring) => ring.physical_range(start, end),
488 None => Ok(start..end),
489 }
490 }
491}
492
493pub struct LatentKvLayer {
506 pub rows: CudaSlice<f32>,
508 pub width: usize,
509 pub len: usize,
510 pub len_d: CudaSlice<i32>,
512 pub index_rows: Option<CudaSlice<f32>>,
562 pub index_width: usize,
563 pub index_ring_rows: Option<usize>,
567 pub index_pool_keys: Option<CudaSlice<f32>>,
592 pub index_pools_ready: usize,
594 pub index_pool: usize,
602}
603
604impl LatentKvLayer {
605 pub fn truncate_index_pool_keys(&mut self, pool: usize) {
609 if pool == 0 {
610 return;
611 }
612 self.index_pools_ready = self.index_pools_ready.min(self.len / pool);
613 }
614}
615
616pub fn index_plane_physical_row(ring_rows: usize, pool: usize, abs: usize) -> usize {
621 if ring_rows == 0 {
622 return abs;
623 }
624 debug_assert!(pool > 0, "index plane addressing requires a known pool");
625 let effective = ring_rows / pool * pool;
626 debug_assert!(effective > 0, "the effective ring holds at least one pool");
627 abs % effective
628}
629
630pub struct LatentPlaneSnapshot {
647 pub rows: CudaSlice<f32>,
649 pub width: usize,
650 pub len: usize,
651 pub index_width: usize,
653 pub index_pool: usize,
655 pub index_tail: Option<CudaSlice<f32>>,
658 pub index_pool_keys: Option<CudaSlice<f32>>,
661 pub index_pools_ready: usize,
662}
663
664impl LatentPlaneSnapshot {
665 pub fn bytes(&self) -> usize {
668 let tail = self.index_tail.as_ref().map_or(0, CudaSlice::len);
669 let keys = self.index_pool_keys.as_ref().map_or(0, CudaSlice::len);
670 (self.rows.len() + tail + keys) * std::mem::size_of::<f32>()
671 }
672}
673
674pub struct LatentTailCapture {
685 pub len: usize,
687 pub width: usize,
689 pub index_width: usize,
690 pub index_pool: usize,
692 pub index_pools_ready: usize,
695 pub index_tail: Option<CudaSlice<f32>>,
699}
700
701impl LatentTailCapture {
702 pub fn bytes(&self) -> usize {
704 self.index_tail.as_ref().map_or(0, CudaSlice::len) * std::mem::size_of::<f32>()
705 }
706}
707
708impl LatentKvLayer {
709 pub fn snapshot_plane(
717 &self,
718 e: &impl KvDev,
719 ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
720 let (len, width) = (self.len, self.width);
721 if len == 0 {
722 return Err("latent snapshot at len 0 (record the layer as absent instead)".into());
723 }
724 if self.rows.len() < len * width {
725 return Err(format!(
726 "latent plane holds {} f32 but len {len} x width {width} requires {}",
727 self.rows.len(),
728 len * width,
729 )
730 .into());
731 }
732 let mut rows = e.uninit(len * width)?;
733 e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
734 if self.index_width == 0 {
735 return Ok(LatentPlaneSnapshot {
736 rows,
737 width,
738 len,
739 index_width: 0,
740 index_pool: 0,
741 index_tail: None,
742 index_pool_keys: None,
743 index_pools_ready: 0,
744 });
745 }
746 let pool = self.index_pool;
747 if pool == 0 {
748 return Err(format!(
749 "latent snapshot: index plane (width {}) has an unresolved pool — no indexer \
750 call ran against this layer, so its derived state cannot be validated",
751 self.index_width,
752 )
753 .into());
754 }
755 let d = self.index_width / 2;
756 let pools_ready = self.index_pools_ready;
757 if pools_ready != len / pool {
758 return Err(format!(
759 "latent snapshot: index_pools_ready {pools_ready} != len/pool {} (len {len}, \
760 pool {pool}); a capture must sit at a drained call boundary or its keys \
761 violate the append-only finality invariant",
762 len / pool,
763 )
764 .into());
765 }
766 let index_pool_keys = if pools_ready > 0 {
767 let src = self
768 .index_pool_keys
769 .as_ref()
770 .ok_or("latent snapshot: pools are ready but the resident key plane is gone")?;
771 if src.len() < pools_ready * d {
772 return Err(format!(
773 "latent snapshot: resident key plane holds {} f32 but {pools_ready} pools \
774 x d {d} require {}",
775 src.len(),
776 pools_ready * d,
777 )
778 .into());
779 }
780 let mut keys = e.uninit(pools_ready * d)?;
781 e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
782 Some(keys)
783 } else {
784 None
785 };
786 let tail_rows = len - pools_ready * pool;
787 let index_tail = if tail_rows > 0 {
788 let src = self
789 .index_rows
790 .as_ref()
791 .ok_or("latent snapshot: index_width > 0 but the state plane is gone")?;
792 let ring = self.index_ring_rows.unwrap_or(0);
793 let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
796 let want = (phys + tail_rows) * self.index_width;
797 if src.len() < want {
798 return Err(format!(
799 "latent snapshot: index plane holds {} f32 but the live tail window \
800 requires {want}",
801 src.len(),
802 )
803 .into());
804 }
805 let mut tail = e.uninit(tail_rows * self.index_width)?;
806 e.copy_range_into(
807 &mut tail,
808 0,
809 src,
810 phys * self.index_width,
811 tail_rows * self.index_width,
812 )?;
813 Some(tail)
814 } else {
815 None
816 };
817 Ok(LatentPlaneSnapshot {
818 rows,
819 width,
820 len,
821 index_width: self.index_width,
822 index_pool: pool,
823 index_tail,
824 index_pool_keys,
825 index_pools_ready: pools_ready,
826 })
827 }
828
829 pub fn snapshot_tail(
835 &self,
836 e: &impl KvDev,
837 ) -> Result<LatentTailCapture, Box<dyn std::error::Error>> {
838 let (len, width) = (self.len, self.width);
839 if len == 0 {
840 return Err("latent tail capture at len 0 (record the layer as absent instead)".into());
841 }
842 if self.index_width == 0 {
843 return Ok(LatentTailCapture {
844 len,
845 width,
846 index_width: 0,
847 index_pool: 0,
848 index_pools_ready: 0,
849 index_tail: None,
850 });
851 }
852 let pool = self.index_pool;
853 if pool == 0 {
854 return Err(format!(
855 "latent tail capture: index plane (width {}) has an unresolved pool — no \
856 indexer call ran against this layer, so its derived state cannot be validated",
857 self.index_width,
858 )
859 .into());
860 }
861 let pools_ready = self.index_pools_ready;
862 if pools_ready != len / pool {
863 return Err(format!(
864 "latent tail capture: index_pools_ready {pools_ready} != len/pool {} (len \
865 {len}, pool {pool}); a capture must sit at a drained call boundary",
866 len / pool,
867 )
868 .into());
869 }
870 let tail_rows = len - pools_ready * pool;
871 let index_tail = if tail_rows > 0 {
872 let src = self
873 .index_rows
874 .as_ref()
875 .ok_or("latent tail capture: index_width > 0 but the state plane is gone")?;
876 let ring = self.index_ring_rows.unwrap_or(0);
877 let phys = index_plane_physical_row(ring, pool, pools_ready * pool);
878 let want = (phys + tail_rows) * self.index_width;
879 if src.len() < want {
880 return Err(format!(
881 "latent tail capture: index plane holds {} f32 but the live tail window \
882 requires {want}",
883 src.len(),
884 )
885 .into());
886 }
887 let mut tail = e.uninit(tail_rows * self.index_width)?;
888 e.copy_range_into(
889 &mut tail,
890 0,
891 src,
892 phys * self.index_width,
893 tail_rows * self.index_width,
894 )?;
895 Some(tail)
896 } else {
897 None
898 };
899 Ok(LatentTailCapture {
900 len,
901 width,
902 index_width: self.index_width,
903 index_pool: pool,
904 index_pools_ready: pools_ready,
905 index_tail,
906 })
907 }
908
909 pub fn snapshot_plane_at(
918 &self,
919 e: &impl KvDev,
920 cap: LatentTailCapture,
921 ) -> Result<LatentPlaneSnapshot, Box<dyn std::error::Error>> {
922 let (len, width) = (cap.len, cap.width);
923 if len == 0 {
924 return Err("latent boundary publish at len 0".into());
925 }
926 if width != self.width {
927 return Err(format!(
928 "latent boundary publish: captured width {width} != live width {}",
929 self.width,
930 )
931 .into());
932 }
933 if self.len < len {
934 return Err(format!(
935 "latent boundary publish: live len {} < boundary {len} — the plane was \
936 truncated below the capture boundary",
937 self.len,
938 )
939 .into());
940 }
941 if self.rows.len() < len * width {
942 return Err(format!(
943 "latent boundary publish: live plane holds {} f32 but boundary {len} x width \
944 {width} requires {}",
945 self.rows.len(),
946 len * width,
947 )
948 .into());
949 }
950 let mut rows = e.uninit(len * width)?;
951 e.copy_range_into(&mut rows, 0, &self.rows, 0, len * width)?;
952 if cap.index_width != self.index_width {
953 return Err(format!(
954 "latent boundary publish: captured index_width {} != live {}",
955 cap.index_width, self.index_width,
956 )
957 .into());
958 }
959 if cap.index_width == 0 {
960 return Ok(LatentPlaneSnapshot {
961 rows,
962 width,
963 len,
964 index_width: 0,
965 index_pool: 0,
966 index_tail: None,
967 index_pool_keys: None,
968 index_pools_ready: 0,
969 });
970 }
971 if cap.index_pool != self.index_pool {
972 return Err(format!(
973 "latent boundary publish: captured pool {} != live pool {}",
974 cap.index_pool, self.index_pool,
975 )
976 .into());
977 }
978 let d = cap.index_width / 2;
979 let pools_ready = cap.index_pools_ready;
980 if self.index_pools_ready < pools_ready {
981 return Err(format!(
982 "latent boundary publish: live index_pools_ready {} < boundary {pools_ready} \
983 — the key plane was clamped below the capture boundary",
984 self.index_pools_ready,
985 )
986 .into());
987 }
988 let index_pool_keys = if pools_ready > 0 {
989 let src = self
990 .index_pool_keys
991 .as_ref()
992 .ok_or("latent boundary publish: pools are ready but the key plane is gone")?;
993 if src.len() < pools_ready * d {
994 return Err(format!(
995 "latent boundary publish: key plane holds {} f32 but {pools_ready} pools \
996 x d {d} require {}",
997 src.len(),
998 pools_ready * d,
999 )
1000 .into());
1001 }
1002 let mut keys = e.uninit(pools_ready * d)?;
1003 e.copy_range_into(&mut keys, 0, src, 0, pools_ready * d)?;
1004 Some(keys)
1005 } else {
1006 None
1007 };
1008 Ok(LatentPlaneSnapshot {
1009 rows,
1010 width,
1011 len,
1012 index_width: cap.index_width,
1013 index_pool: cap.index_pool,
1014 index_tail: cap.index_tail,
1015 index_pool_keys,
1016 index_pools_ready: pools_ready,
1017 })
1018 }
1019
1020 pub fn validate_restore(
1024 &self,
1025 snap: &LatentPlaneSnapshot,
1026 max_ctx: usize,
1027 ) -> Result<(), String> {
1028 if self.len != 0 {
1029 return Err("restore destination latent plane is not fresh".into());
1030 }
1031 if self.width != snap.width {
1032 return Err(format!(
1033 "snapshot width {} != destination width {}",
1034 snap.width, self.width,
1035 ));
1036 }
1037 if snap.len == 0 || snap.len > max_ctx {
1038 return Err(format!("snapshot len {} outside [1,{max_ctx}]", snap.len));
1039 }
1040 if snap.rows.len() < snap.len * snap.width {
1041 return Err(format!(
1042 "snapshot rows plane holds {} f32 but len {} x width {} requires {} \
1043 (truncated capture)",
1044 snap.rows.len(),
1045 snap.len,
1046 snap.width,
1047 snap.len * snap.width,
1048 ));
1049 }
1050 if self.rows.len() < snap.len * self.width {
1051 return Err(format!(
1052 "destination latent plane holds {} f32 but the restore requires {}",
1053 self.rows.len(),
1054 snap.len * self.width,
1055 ));
1056 }
1057 if self.index_width != snap.index_width {
1058 return Err(format!(
1059 "snapshot index width {} != destination {}",
1060 snap.index_width, self.index_width,
1061 ));
1062 }
1063 if snap.index_width == 0 {
1064 return Ok(());
1065 }
1066 let pool = snap.index_pool;
1067 if pool == 0 {
1068 return Err("snapshot carries an index plane with an unresolved pool".into());
1069 }
1070 if self.index_pool != 0 && self.index_pool != pool {
1071 return Err(format!(
1072 "snapshot pool {pool} != destination resident pool {}",
1073 self.index_pool,
1074 ));
1075 }
1076 let d = snap.index_width / 2;
1077 if snap.index_pools_ready != snap.len / pool {
1078 return Err(format!(
1079 "snapshot index_pools_ready {} != len/pool {} (len {}, pool {pool}): the \
1080 append-only finality invariant does not hold, so its keys are stale",
1081 snap.index_pools_ready,
1082 snap.len / pool,
1083 snap.len,
1084 ));
1085 }
1086 match (&snap.index_pool_keys, snap.index_pools_ready) {
1087 (Some(keys), ready @ 1..) => {
1088 if keys.len() < ready * d {
1089 return Err(format!(
1090 "snapshot key plane holds {} f32 but {ready} pools x d {d} require {}",
1091 keys.len(),
1092 ready * d,
1093 ));
1094 }
1095 }
1096 (None, 0) => {}
1097 (Some(_), 0) => return Err("snapshot carries keys for zero ready pools".into()),
1098 (None, ready) => {
1099 return Err(format!(
1100 "snapshot claims {ready} ready pools but carries no keys"
1101 ));
1102 }
1103 }
1104 let tail_rows = snap.len - snap.index_pools_ready * pool;
1105 match (&snap.index_tail, tail_rows) {
1106 (Some(tail), rows @ 1..) => {
1107 if tail.len() < rows * snap.index_width {
1108 return Err(format!(
1109 "snapshot tail holds {} f32 but {rows} rows x index width {} require {}",
1110 tail.len(),
1111 snap.index_width,
1112 rows * snap.index_width,
1113 ));
1114 }
1115 }
1116 (None, 0) => {}
1117 (Some(_), 0) => return Err("snapshot carries a tail at a pool-aligned boundary".into()),
1118 (None, rows) => {
1119 return Err(format!(
1120 "snapshot owes {rows} live tail rows but carries none"
1121 ));
1122 }
1123 }
1124 if self.index_rows.is_none() {
1125 return Err("destination declares an index plane but allocated none".into());
1126 }
1127 if tail_rows > 0 {
1128 let ring = self.index_ring_rows.unwrap_or(0);
1129 let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
1130 let want = (phys + tail_rows) * self.index_width;
1131 let have = self.index_rows.as_ref().map_or(0, CudaSlice::len);
1132 if have < want {
1133 return Err(format!(
1134 "destination index plane holds {have} f32 but the tail window requires \
1135 {want}",
1136 ));
1137 }
1138 }
1139 Ok(())
1140 }
1141
1142 pub fn restore_plane(
1150 &mut self,
1151 e: &impl KvDev,
1152 snap: &LatentPlaneSnapshot,
1153 max_ctx: usize,
1154 ) -> Result<(), Box<dyn std::error::Error>> {
1155 self.validate_restore(snap, max_ctx)?;
1156 e.copy_range_into(&mut self.rows, 0, &snap.rows, 0, snap.len * snap.width)?;
1157 if snap.index_width > 0 {
1158 let pool = snap.index_pool;
1159 let d = snap.index_width / 2;
1160 let mut keys = e.zeros(((max_ctx / pool) * d).max(1))?;
1163 if let Some(src) = &snap.index_pool_keys {
1164 e.copy_range_into(&mut keys, 0, src, 0, snap.index_pools_ready * d)?;
1165 }
1166 self.index_pool_keys = Some(keys);
1167 self.index_pools_ready = snap.index_pools_ready;
1168 self.index_pool = pool;
1169 if let Some(tail) = &snap.index_tail {
1170 let tail_rows = snap.len - snap.index_pools_ready * pool;
1171 let ring = self.index_ring_rows.unwrap_or(0);
1172 let phys = index_plane_physical_row(ring, pool, snap.index_pools_ready * pool);
1173 let dst = self
1174 .index_rows
1175 .as_mut()
1176 .ok_or("destination index plane vanished after validation")?;
1177 e.copy_range_into(
1178 dst,
1179 phys * self.index_width,
1180 tail,
1181 0,
1182 tail_rows * self.index_width,
1183 )?;
1184 }
1185 }
1186 self.len = snap.len;
1187 let len_i32 = i32::try_from(snap.len).map_err(|_| "latent length exceeds i32 mirror")?;
1188 e.set_i32_one(&mut self.len_d, len_i32)?;
1189 Ok(())
1190 }
1191}
1192
1193pub struct RecurLayer {
1197 pub conv_state: CudaSlice<f32>, pub ssm_state: CudaSlice<f32>, pub ssm_state_alt: CudaSlice<f32>,
1209}
1210
1211pub struct ResidentTpKvCacheRank {
1212 k: CudaSlice<u8>,
1213 v: CudaSlice<u8>,
1214 len_d: CudaSlice<i32>,
1215 base_d: Option<CudaSlice<i32>>,
1220}
1221
1222impl ResidentTpKvCacheRank {
1223 pub fn new(k: CudaSlice<u8>, v: CudaSlice<u8>, len_d: CudaSlice<i32>) -> Self {
1224 Self {
1225 k,
1226 v,
1227 len_d,
1228 base_d: None,
1229 }
1230 }
1231
1232 pub fn base_d(&self) -> Option<&CudaSlice<i32>> {
1233 self.base_d.as_ref()
1234 }
1235
1236 pub fn base_d_mut(&mut self) -> Option<&mut CudaSlice<i32>> {
1237 self.base_d.as_mut()
1238 }
1239
1240 pub fn arm_base_d(&mut self, buf: CudaSlice<i32>) {
1241 self.base_d = Some(buf);
1242 }
1243
1244 pub fn k(&self) -> &CudaSlice<u8> {
1245 &self.k
1246 }
1247
1248 pub fn v(&self) -> &CudaSlice<u8> {
1249 &self.v
1250 }
1251
1252 pub fn len_d(&self) -> &CudaSlice<i32> {
1253 &self.len_d
1254 }
1255
1256 pub fn k_mut(&mut self) -> &mut CudaSlice<u8> {
1257 &mut self.k
1258 }
1259
1260 pub fn v_mut(&mut self) -> &mut CudaSlice<u8> {
1261 &mut self.v
1262 }
1263
1264 pub fn planes_mut(&mut self) -> (&mut CudaSlice<u8>, &mut CudaSlice<u8>) {
1265 (&mut self.k, &mut self.v)
1266 }
1267
1268 #[allow(clippy::type_complexity)]
1270 pub fn planes_and_counters_mut(
1271 &mut self,
1272 ) -> (
1273 &mut CudaSlice<u8>,
1274 &mut CudaSlice<u8>,
1275 &CudaSlice<i32>,
1276 Option<&CudaSlice<i32>>,
1277 ) {
1278 (&mut self.k, &mut self.v, &self.len_d, self.base_d.as_ref())
1279 }
1280
1281 pub fn len_d_mut(&mut self) -> &mut CudaSlice<i32> {
1282 &mut self.len_d
1283 }
1284}
1285
1286#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1287pub struct TpKvTransaction {
1288 generation: u64,
1289 base_len: usize,
1290}
1291
1292impl TpKvTransaction {
1293 pub fn generation(self) -> u64 {
1294 self.generation
1295 }
1296
1297 pub fn base_len(self) -> usize {
1298 self.base_len
1299 }
1300}
1301
1302#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1303pub struct TpKvAppendPlan {
1304 transaction: TpKvTransaction,
1305 target: usize,
1306 write_row: usize,
1307 ring_append: Option<KvRingAppend>,
1308}
1309
1310impl TpKvAppendPlan {
1311 pub fn target(self) -> usize {
1312 self.target
1313 }
1314
1315 pub fn write_row(self) -> usize {
1316 self.write_row
1317 }
1318
1319 pub fn ring_append(self) -> Option<KvRingAppend> {
1320 self.ring_append
1321 }
1322}
1323
1324#[derive(Debug, PartialEq, Eq)]
1325pub struct TpKvGrowPlan {
1326 rows: usize,
1327 source_row: usize,
1328 copy_rows: usize,
1329 target_base: usize,
1330 k_bytes: usize,
1331 v_bytes: usize,
1332 source_capacity: usize,
1333 target_capacity: usize,
1334 ring_window: Option<usize>,
1335 target_physical_rows: usize,
1336 kv_dim_k: usize,
1337 kv_dim_v: usize,
1338 k_tok_bytes: usize,
1339 v_tok_bytes: usize,
1340 ranks: usize,
1341 next_generation: u64,
1342}
1343
1344impl TpKvGrowPlan {
1345 pub fn rows(&self) -> usize {
1346 self.rows
1347 }
1348
1349 pub fn source_row(&self) -> usize {
1350 self.source_row
1351 }
1352
1353 pub fn copy_rows(&self) -> usize {
1354 self.copy_rows
1355 }
1356
1357 pub fn k_bytes(&self) -> usize {
1358 self.k_bytes
1359 }
1360
1361 pub fn v_bytes(&self) -> usize {
1362 self.v_bytes
1363 }
1364}
1365
1366#[derive(Clone, Debug, PartialEq, Eq)]
1367struct TpKvTransactionState {
1368 committed_len: usize,
1369 staged_len: usize,
1370 next_generation: u64,
1371 active: Option<TpKvTransaction>,
1372}
1373
1374impl TpKvTransactionState {
1375 fn new() -> Self {
1376 Self {
1377 committed_len: 0,
1378 staged_len: 0,
1379 next_generation: 1,
1380 active: None,
1381 }
1382 }
1383
1384 fn begin(&mut self) -> Result<TpKvTransaction, String> {
1385 if let Some(active) = self.active {
1386 return Err(format!(
1387 "TP KV transaction generation {} is already active at base {}",
1388 active.generation, active.base_len
1389 ));
1390 }
1391 if self.staged_len != self.committed_len {
1392 return Err(format!(
1393 "TP KV cache is half-committed: staged {} != committed {}",
1394 self.staged_len, self.committed_len
1395 ));
1396 }
1397 let transaction = TpKvTransaction {
1398 generation: self.next_generation,
1399 base_len: self.committed_len,
1400 };
1401 self.next_generation = self
1402 .next_generation
1403 .checked_add(1)
1404 .ok_or("TP KV transaction generation overflow")?;
1405 self.active = Some(transaction);
1406 Ok(transaction)
1407 }
1408
1409 fn validate(&self, transaction: TpKvTransaction) -> Result<(), String> {
1410 if self.active != Some(transaction) {
1411 return Err(format!(
1412 "stale TP KV transaction generation {} at base {}",
1413 transaction.generation, transaction.base_len
1414 ));
1415 }
1416 if transaction.base_len != self.committed_len {
1417 return Err(format!(
1418 "TP KV transaction base {} != committed length {}",
1419 transaction.base_len, self.committed_len
1420 ));
1421 }
1422 Ok(())
1423 }
1424
1425 fn append_target(
1426 &self,
1427 transaction: TpKvTransaction,
1428 rows: usize,
1429 capacity: usize,
1430 ) -> Result<usize, String> {
1431 self.validate(transaction)?;
1432 if rows == 0 {
1433 return Err("TP KV append must contain at least one row".into());
1434 }
1435 let target = self
1436 .staged_len
1437 .checked_add(rows)
1438 .ok_or("TP KV staged length overflow")?;
1439 if target > capacity {
1440 return Err(format!(
1441 "TP KV append exceeds capacity: {target} > {capacity}"
1442 ));
1443 }
1444 Ok(target)
1445 }
1446
1447 fn publish_append(
1448 &mut self,
1449 transaction: TpKvTransaction,
1450 target: usize,
1451 ) -> Result<(), String> {
1452 self.validate(transaction)?;
1453 if target <= self.staged_len {
1454 return Err(format!(
1455 "TP KV append target {target} must exceed staged length {}",
1456 self.staged_len
1457 ));
1458 }
1459 self.staged_len = target;
1460 Ok(())
1461 }
1462
1463 fn commit_target(
1464 &self,
1465 transaction: TpKvTransaction,
1466 accepted_rows: usize,
1467 ) -> Result<usize, String> {
1468 self.validate(transaction)?;
1469 let staged_rows = self
1470 .staged_len
1471 .checked_sub(transaction.base_len)
1472 .ok_or("TP KV staged length precedes its transaction base")?;
1473 if accepted_rows > staged_rows {
1474 return Err(format!(
1475 "TP KV commit accepts {accepted_rows} rows from a {staged_rows}-row transaction"
1476 ));
1477 }
1478 transaction
1479 .base_len
1480 .checked_add(accepted_rows)
1481 .ok_or_else(|| "TP KV committed length overflow".to_string())
1482 }
1483
1484 fn publish_finalize(
1485 &mut self,
1486 transaction: TpKvTransaction,
1487 target: usize,
1488 ) -> Result<(), String> {
1489 self.validate(transaction)?;
1490 if target < transaction.base_len || target > self.staged_len {
1491 return Err(format!(
1492 "TP KV finalize target {target} outside transaction range {}..={}",
1493 transaction.base_len, self.staged_len
1494 ));
1495 }
1496 self.committed_len = target;
1497 self.staged_len = target;
1498 self.active = None;
1499 Ok(())
1500 }
1501
1502 fn rewind(&mut self, target: usize, capacity: usize) -> Result<(), String> {
1503 if target > capacity {
1504 return Err(format!(
1505 "TP KV rewind target {target} exceeds capacity {capacity}"
1506 ));
1507 }
1508 self.committed_len = target;
1509 self.staged_len = target;
1510 self.active = None;
1511 Ok(())
1512 }
1513}
1514
1515pub struct ResidentTpKvCache {
1516 ranks: Vec<ResidentTpKvCacheRank>,
1517 kv_dim_k: usize,
1518 kv_dim_v: usize,
1519 k_tok_bytes: usize,
1520 v_tok_bytes: usize,
1521 capacity: usize,
1522 ring: Option<KvRing>,
1523 state: TpKvTransactionState,
1524 external_rows: bool,
1531}
1532
1533impl ResidentTpKvCache {
1534 #[allow(clippy::too_many_arguments)]
1535 pub fn new(
1536 ranks: Vec<ResidentTpKvCacheRank>,
1537 kv_dim_k: usize,
1538 kv_dim_v: usize,
1539 k_tok_bytes: usize,
1540 v_tok_bytes: usize,
1541 capacity: usize,
1542 ) -> Self {
1543 Self::new_inner(
1544 ranks,
1545 kv_dim_k,
1546 kv_dim_v,
1547 k_tok_bytes,
1548 v_tok_bytes,
1549 capacity,
1550 None,
1551 )
1552 }
1553
1554 #[allow(clippy::too_many_arguments)]
1555 pub fn new_swa(
1556 ranks: Vec<ResidentTpKvCacheRank>,
1557 kv_dim_k: usize,
1558 kv_dim_v: usize,
1559 k_tok_bytes: usize,
1560 v_tok_bytes: usize,
1561 capacity: usize,
1562 window: usize,
1563 ) -> Self {
1564 Self::new_inner(
1565 ranks,
1566 kv_dim_k,
1567 kv_dim_v,
1568 k_tok_bytes,
1569 v_tok_bytes,
1570 capacity,
1571 Some(KvRing::new(swa_ring_rows(window, capacity), window)),
1572 )
1573 }
1574
1575 #[allow(clippy::too_many_arguments)]
1576 fn new_inner(
1577 ranks: Vec<ResidentTpKvCacheRank>,
1578 kv_dim_k: usize,
1579 kv_dim_v: usize,
1580 k_tok_bytes: usize,
1581 v_tok_bytes: usize,
1582 capacity: usize,
1583 ring: Option<KvRing>,
1584 ) -> Self {
1585 Self {
1586 ranks,
1587 kv_dim_k,
1588 kv_dim_v,
1589 k_tok_bytes,
1590 v_tok_bytes,
1591 capacity,
1592 ring,
1593 state: TpKvTransactionState::new(),
1594 external_rows: false,
1595 }
1596 }
1597
1598 pub fn begin_transaction(&mut self) -> Result<TpKvTransaction, String> {
1599 self.state.begin()
1600 }
1601
1602 pub fn rows_external(&self) -> bool {
1605 self.external_rows
1606 }
1607
1608 pub fn mark_rows_external(&mut self, external: bool) {
1609 self.external_rows = external;
1610 }
1611
1612 pub fn committed_len(&self) -> usize {
1613 self.state.committed_len
1614 }
1615
1616 pub fn staged_len(&self) -> usize {
1617 self.state.staged_len
1618 }
1619
1620 pub fn capacity(&self) -> usize {
1621 self.capacity
1622 }
1623
1624 pub fn physical_capacity(&self) -> usize {
1625 self.ring
1626 .as_ref()
1627 .map(KvRing::rows)
1628 .unwrap_or(self.capacity)
1629 }
1630
1631 pub fn ring_window(&self) -> Option<usize> {
1632 self.ring.as_ref().map(KvRing::window)
1633 }
1634
1635 pub fn ring_base(&self) -> Option<usize> {
1636 self.ring.as_ref().map(KvRing::base)
1637 }
1638
1639 pub fn physical_range(
1640 &self,
1641 start: usize,
1642 end: usize,
1643 ) -> Result<std::ops::Range<usize>, String> {
1644 match &self.ring {
1645 Some(ring) => ring.physical_range(start, end),
1646 None => {
1647 if end < start || end > self.capacity {
1648 return Err(format!(
1649 "TP KV linear view [{start},{end}) exceeds capacity {}",
1650 self.capacity
1651 ));
1652 }
1653 Ok(start..end)
1654 }
1655 }
1656 }
1657
1658 pub fn can_rewind_to(&self, target: usize) -> bool {
1659 target <= self.capacity
1660 && self
1661 .ring
1662 .as_ref()
1663 .is_none_or(|ring| ring.can_rewind_to(target))
1664 }
1665
1666 pub fn kv_dim_k(&self) -> usize {
1667 self.kv_dim_k
1668 }
1669
1670 pub fn kv_dim_v(&self) -> usize {
1671 self.kv_dim_v
1672 }
1673
1674 pub fn k_tok_bytes(&self) -> usize {
1675 self.k_tok_bytes
1676 }
1677
1678 pub fn v_tok_bytes(&self) -> usize {
1679 self.v_tok_bytes
1680 }
1681
1682 pub fn ranks_len(&self) -> usize {
1683 self.ranks.len()
1684 }
1685
1686 pub fn rank(&self, rank: usize) -> Option<&ResidentTpKvCacheRank> {
1687 self.ranks.get(rank)
1688 }
1689
1690 pub fn rank_mut(&mut self, rank: usize) -> Option<&mut ResidentTpKvCacheRank> {
1691 self.ranks.get_mut(rank)
1692 }
1693
1694 pub fn ranks(&self) -> &[ResidentTpKvCacheRank] {
1695 &self.ranks
1696 }
1697
1698 pub fn ranks_mut(&mut self) -> &mut [ResidentTpKvCacheRank] {
1699 &mut self.ranks
1700 }
1701
1702 pub fn prepare_grow(
1703 &self,
1704 target_capacity: usize,
1705 rows: usize,
1706 ) -> Result<TpKvGrowPlan, String> {
1707 if let Some(active) = self.state.active {
1708 return Err(format!(
1709 "TP KV grow refuses active transaction generation {} at base {}",
1710 active.generation, active.base_len
1711 ));
1712 }
1713 if self.state.staged_len != self.state.committed_len {
1714 return Err(format!(
1715 "TP KV grow requires quiescent state, got committed/staged={}/{}",
1716 self.state.committed_len, self.state.staged_len
1717 ));
1718 }
1719 if target_capacity <= self.capacity {
1720 return Err(format!(
1721 "TP KV grow target capacity {target_capacity} must exceed source capacity {}",
1722 self.capacity
1723 ));
1724 }
1725 if target_capacity > i32::MAX as usize {
1726 return Err(format!(
1727 "TP KV grow target capacity {target_capacity} exceeds i32 device mirrors"
1728 ));
1729 }
1730 if rows > self.state.committed_len {
1731 return Err(format!(
1732 "TP KV grow rows {rows} exceed committed length {}",
1733 self.state.committed_len
1734 ));
1735 }
1736 let (source_row, copy_rows, target_base, ring_window, target_physical_rows) =
1737 match &self.ring {
1738 Some(ring) => {
1739 let raw = rows.saturating_sub(ring.window().saturating_sub(1));
1740 let target_base = raw & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1741 let physical = ring.physical_range(target_base, rows)?;
1742 (
1743 physical.start,
1744 physical.len(),
1745 target_base,
1746 Some(ring.window()),
1747 swa_ring_rows(ring.window(), target_capacity),
1748 )
1749 }
1750 None => (0, rows, 0, None, target_capacity),
1751 };
1752 let k_bytes = copy_rows
1753 .checked_mul(self.k_tok_bytes)
1754 .ok_or("TP KV grow K byte extent overflow")?;
1755 let v_bytes = copy_rows
1756 .checked_mul(self.v_tok_bytes)
1757 .ok_or("TP KV grow V byte extent overflow")?;
1758 Ok(TpKvGrowPlan {
1759 rows,
1760 source_row,
1761 copy_rows,
1762 target_base,
1763 k_bytes,
1764 v_bytes,
1765 source_capacity: self.capacity,
1766 target_capacity,
1767 ring_window,
1768 target_physical_rows,
1769 kv_dim_k: self.kv_dim_k,
1770 kv_dim_v: self.kv_dim_v,
1771 k_tok_bytes: self.k_tok_bytes,
1772 v_tok_bytes: self.v_tok_bytes,
1773 ranks: self.ranks.len(),
1774 next_generation: self.state.next_generation,
1775 })
1776 }
1777
1778 pub fn publish_grow(&mut self, plan: TpKvGrowPlan) -> Result<(), String> {
1779 if self.state != TpKvTransactionState::new() {
1780 return Err(format!(
1781 "TP KV grow target must be fresh, got committed/staged={}/{} active={}",
1782 self.state.committed_len,
1783 self.state.staged_len,
1784 self.state.active.is_some()
1785 ));
1786 }
1787 if self.capacity != plan.target_capacity
1788 || self.capacity <= plan.source_capacity
1789 || self.kv_dim_k != plan.kv_dim_k
1790 || self.kv_dim_v != plan.kv_dim_v
1791 || self.k_tok_bytes != plan.k_tok_bytes
1792 || self.v_tok_bytes != plan.v_tok_bytes
1793 || self.ranks.len() != plan.ranks
1794 || self.ring.as_ref().map(KvRing::window) != plan.ring_window
1795 || self.physical_capacity() != plan.target_physical_rows
1796 {
1797 return Err("TP KV grow target layout does not match its source plan".into());
1798 }
1799 if plan.rows > self.capacity {
1800 return Err(format!(
1801 "TP KV grow rows {} exceed target capacity {}",
1802 plan.rows, self.capacity
1803 ));
1804 }
1805 if let Some(ring) = self.ring.as_mut() {
1806 let mut target_ring = *ring;
1807 target_ring.apply_rebase(plan.target_base);
1808 if !target_ring.can_rewind_to(plan.rows) {
1809 return Err(format!(
1810 "TP KV grow target ring base {} cannot expose committed length {}",
1811 target_ring.base(),
1812 plan.rows
1813 ));
1814 }
1815 *ring = target_ring;
1816 }
1817 self.state.committed_len = plan.rows;
1818 self.state.staged_len = plan.rows;
1819 self.state.next_generation = plan.next_generation;
1820 self.state.active = None;
1821 Ok(())
1822 }
1823
1824 pub fn prepare_append(
1825 &self,
1826 transaction: TpKvTransaction,
1827 rows: usize,
1828 ) -> Result<TpKvAppendPlan, String> {
1829 let target = self.state.append_target(transaction, rows, self.capacity)?;
1830 let ring_append = self
1831 .ring
1832 .as_ref()
1833 .map(|ring| {
1834 let staged_retain =
1835 target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1836 let rollback_retain = transaction
1837 .base_len
1838 .saturating_sub(ring.window().saturating_sub(1))
1839 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1840 ring.append_plan(
1841 self.state.staged_len,
1842 staged_retain.min(rollback_retain),
1843 rows,
1844 )
1845 })
1846 .transpose()?;
1847 let write_row = match ring_append {
1848 Some(KvRingAppend::Contiguous { write_row })
1849 | Some(KvRingAppend::Rebase { write_row, .. }) => write_row,
1850 None => self.state.staged_len,
1851 };
1852 Ok(TpKvAppendPlan {
1853 transaction,
1854 target,
1855 write_row,
1856 ring_append,
1857 })
1858 }
1859
1860 pub fn peek_append_ring(&self, rows: usize) -> Result<(usize, bool), String> {
1864 let target = self.state.staged_len + rows;
1865 let plan = self
1866 .ring
1867 .as_ref()
1868 .map(|ring| {
1869 let staged_retain =
1870 target.saturating_sub(ring.window()) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1871 let rollback_retain = self
1872 .state
1873 .staged_len
1874 .saturating_sub(ring.window().saturating_sub(1))
1875 & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
1876 ring.append_plan(
1877 self.state.staged_len,
1878 staged_retain.min(rollback_retain),
1879 rows,
1880 )
1881 })
1882 .transpose()?;
1883 Ok(match plan {
1884 Some(KvRingAppend::Contiguous { write_row }) => (write_row, false),
1885 Some(KvRingAppend::Rebase { write_row, .. }) => (write_row, true),
1886 None => (self.state.staged_len, false),
1887 })
1888 }
1889
1890 pub fn publish_append_rebase(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1891 self.state.validate(plan.transaction)?;
1892 match (self.ring.as_mut(), plan.ring_append) {
1893 (
1894 Some(ring),
1895 Some(KvRingAppend::Rebase {
1896 new_base,
1897 keep_rows,
1898 ..
1899 }),
1900 ) => {
1901 if keep_rows > ring.rows() {
1902 return Err(format!(
1903 "TP KV ring rebase keeps {keep_rows} rows in {} physical rows",
1904 ring.rows()
1905 ));
1906 }
1907 let mut target_ring = *ring;
1908 target_ring.apply_rebase(new_base);
1909 if !target_ring.can_rewind_to(plan.transaction.base_len) {
1910 return Err(format!(
1911 "TP KV ring rebase to {new_base} laps transaction base {}",
1912 plan.transaction.base_len
1913 ));
1914 }
1915 *ring = target_ring;
1916 Ok(())
1917 }
1918 (Some(_), Some(KvRingAppend::Contiguous { .. })) | (None, None) => Ok(()),
1919 _ => Err("TP KV append plan does not match cache ring layout".into()),
1920 }
1921 }
1922
1923 pub fn publish_append_plan(&mut self, plan: TpKvAppendPlan) -> Result<(), String> {
1924 if let Some(KvRingAppend::Rebase { new_base, .. }) = plan.ring_append {
1925 if self.ring.as_ref().map(KvRing::base) != Some(new_base) {
1926 return Err(format!(
1927 "TP KV append rebase {new_base} was not published before its state"
1928 ));
1929 }
1930 }
1931 self.state.publish_append(plan.transaction, plan.target)
1932 }
1933
1934 pub fn publish_hydration(
1935 &mut self,
1936 logical_len: usize,
1937 resident_start: usize,
1938 ) -> Result<(), Box<dyn std::error::Error>> {
1939 if self.state != TpKvTransactionState::new() {
1940 return Err("TP KV hydration target must be fresh".into());
1941 }
1942 if resident_start > logical_len || logical_len > self.capacity {
1943 return Err(format!(
1944 "TP KV hydration range [{resident_start},{logical_len}) exceeds capacity {}",
1945 self.capacity
1946 )
1947 .into());
1948 }
1949 match self.ring.as_mut() {
1950 Some(ring) => {
1951 let rows = logical_len - resident_start;
1952 if rows > ring.rows() {
1953 return Err(format!(
1954 "TP KV hydration requires {rows} rows in a {}-row ring",
1955 ring.rows()
1956 )
1957 .into());
1958 }
1959 let mut hydrated_ring = *ring;
1960 hydrated_ring.apply_rebase(resident_start);
1961 if !hydrated_ring.can_rewind_to(logical_len) {
1962 return Err(format!(
1963 "TP KV hydration base {resident_start} cannot expose logical length \
1964 {logical_len}"
1965 )
1966 .into());
1967 }
1968 *ring = hydrated_ring;
1969 }
1970 None if resident_start != 0 => {
1971 return Err("linear TP KV hydration must start at absolute row zero".into());
1972 }
1973 None => {}
1974 }
1975 self.rewind_to(logical_len)
1976 }
1977
1978 pub fn append_target(
1979 &self,
1980 transaction: TpKvTransaction,
1981 rows: usize,
1982 ) -> Result<usize, String> {
1983 self.state.append_target(transaction, rows, self.capacity)
1984 }
1985
1986 pub fn publish_append(
1987 &mut self,
1988 transaction: TpKvTransaction,
1989 target: usize,
1990 ) -> Result<(), String> {
1991 self.state.publish_append(transaction, target)
1992 }
1993
1994 pub fn commit_target(
1995 &self,
1996 transaction: TpKvTransaction,
1997 accepted_rows: usize,
1998 ) -> Result<usize, String> {
1999 self.state.commit_target(transaction, accepted_rows)
2000 }
2001
2002 pub fn validate_transaction(&self, transaction: TpKvTransaction) -> Result<(), String> {
2003 self.state.validate(transaction)
2004 }
2005
2006 pub fn publish_finalize(
2007 &mut self,
2008 transaction: TpKvTransaction,
2009 target: usize,
2010 ) -> Result<(), String> {
2011 if !self.can_rewind_to(target) {
2012 return Err(format!(
2013 "TP KV finalize target {target} is outside the resident cache window/capacity"
2014 ));
2015 }
2016 self.state.publish_finalize(transaction, target)
2017 }
2018
2019 pub fn rewind_to(&mut self, target: usize) -> Result<(), Box<dyn std::error::Error>> {
2020 if !self.can_rewind_to(target) {
2021 return Err(format!(
2022 "TP KV rewind target {target} is outside the resident cache window/capacity"
2023 )
2024 .into());
2025 }
2026 let target_i32 =
2027 i32::try_from(target).map_err(|_| "TP KV length exceeds i32 device mirror")?;
2028 for rank in &mut self.ranks {
2029 let stream = rank.len_d.stream().clone();
2030 stream.memcpy_htod(&[target_i32], &mut rank.len_d)?;
2031 }
2032 self.state.rewind(target, self.capacity)?;
2033 Ok(())
2034 }
2035
2036 pub fn publish_device_rewind(
2040 &mut self,
2041 target: usize,
2042 ) -> Result<(), Box<dyn std::error::Error>> {
2043 if !self.can_rewind_to(target) {
2044 return Err(format!(
2045 "TP KV device rewind target {target} is outside the resident cache window/capacity"
2046 )
2047 .into());
2048 }
2049 self.state.rewind(target, self.capacity)?;
2050 Ok(())
2051 }
2052}
2053
2054pub struct Cache {
2055 pub kv: Vec<Option<KvLayer>>,
2056 pub recur: Vec<Option<RecurLayer>>,
2057 pub latent: Vec<Option<LatentKvLayer>>,
2060 pub tp_kv: Vec<Option<ResidentTpKvCache>>,
2063 pub glm5_tp_recur: Vec<Option<Vec<RecurLayer>>>,
2073 pub glm5_tp_latent_peer: Vec<Option<Vec<LatentKvLayer>>>,
2078 pub pos: usize,
2079 pub max_ctx: usize,
2080 pub tainted: bool,
2083 pub last_logits_dev: Option<CudaSlice<f32>>,
2091 pub dflash_taps: Option<DflashTapSink>,
2096 pub hc_taps: Option<HcTapSink>,
2104 pub glm5_decode_graph: Option<Box<dyn std::any::Any + Send>>,
2115}
2116
2117#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2121enum FullAttentionClass {
2122 Ordinary,
2123 GemmaGlobal,
2124 GemmaWindowed,
2125}
2126
2127fn full_attention_class(plan: &ModelPlan, il: u32) -> FullAttentionClass {
2128 let layer = plan
2129 .layers
2130 .iter()
2131 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2132 .find(|layer| layer.index == il)
2133 .unwrap_or_else(|| panic!("ModelPlan has no layer {il}"));
2134 if !matches!(layer.residual, ResidualTopology::Gemma { .. }) {
2135 return FullAttentionClass::Ordinary;
2136 }
2137 match layer.state {
2138 StatePlan::SlidingKvCache { .. } => FullAttentionClass::GemmaWindowed,
2139 StatePlan::KvCache { .. } => FullAttentionClass::GemmaGlobal,
2140 _ => panic!("Gemma layer {il} does not declare a KV-cache state"),
2141 }
2142}
2143
2144fn full_attention_kv_layout(
2145 cfg: &ModelConfig,
2146 plan: &ModelPlan,
2147 il: u32,
2148) -> (usize, usize, usize, usize) {
2149 debug_assert_eq!(cfg.layer_kind(il), LayerKind::FullAttention);
2150 let class = full_attention_class(plan, il);
2151 let n_head_kv = cfg.n_head_kv as usize;
2152 let (kv_dim_k, kv_dim_v) = match class {
2153 FullAttentionClass::GemmaGlobal | FullAttentionClass::GemmaWindowed => {
2154 let g = cfg
2155 .gemma4
2156 .as_ref()
2157 .expect("Gemma ModelPlan layer requires Gemma cache geometry");
2158 let hd = match class {
2159 FullAttentionClass::GemmaWindowed => g.key_length_swa,
2160 FullAttentionClass::GemmaGlobal => g.key_length_global,
2161 FullAttentionClass::Ordinary => unreachable!(),
2162 } as usize;
2163 let d = match g.head_count_kv.get(il as usize) {
2172 Some(n) => hd * *n as usize,
2173 None => hd * n_head_kv,
2174 };
2175 (d, d)
2176 }
2177 FullAttentionClass::Ordinary => (
2178 cfg.head_dim_k as usize * n_head_kv,
2179 cfg.head_dim_v as usize * n_head_kv,
2180 ),
2181 };
2182 assert!(
2183 kv_dim_k % 32 == 0 && kv_dim_v % 32 == 0,
2184 "KVQUANT requires per-layer kv_dim_k%32==0 && kv_dim_v%32==0 \
2185 (layer {il}: k={kv_dim_k} v={kv_dim_v})"
2186 );
2187 let (kbb, vbb) = kv_blk_bytes();
2188 let g4_global_fp8 = gkv_on() && class == FullAttentionClass::GemmaGlobal;
2189 let g4_windowed_fp8 = wkv_on() && class == FullAttentionClass::GemmaWindowed;
2190 let (kbb_l, vbb_l) = if g4_global_fp8 || g4_windowed_fp8 {
2191 (32, 32)
2192 } else {
2193 (kbb, vbb)
2194 };
2195 (kv_dim_k, kv_dim_v, kbb_l, vbb_l)
2196}
2197
2198fn kv_plane_allocation_bytes(rows: usize, token_bytes: usize) -> usize {
2199 rows * token_bytes + 8
2200}
2201
2202pub fn cache_bytes_per_token(cfg: &ModelConfig) -> usize {
2209 cache_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2210}
2211
2212pub fn cache_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2216 let plan = ModelPlan::compile(cfg).expect("cache sizing requires a compilable ModelPlan");
2217 cache_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2218}
2219
2220pub fn cache_bytes_per_token_for_plan(
2221 cfg: &ModelConfig,
2222 plan: &ModelPlan,
2223 lo: usize,
2224 hi: usize,
2225) -> usize {
2226 assert!(
2227 lo <= hi && hi <= cfg.n_layer as usize,
2228 "cache layer range out of bounds"
2229 );
2230 let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2231 let full_attn: usize = (lo as u32..hi as u32)
2232 .filter(|&il| cfg.layer_kind(il) == LayerKind::FullAttention)
2233 .filter(|&il| shared == 0 || il < cfg.n_layer - shared)
2234 .map(|il| {
2235 let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, il);
2236 (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2237 })
2238 .sum();
2239 full_attn + latent_kv_bytes_per_token_for_plan(cfg, plan, lo, hi)
2240}
2241
2242pub fn latent_kv_bytes_per_token_for_plan(
2272 cfg: &ModelConfig,
2273 plan: &ModelPlan,
2274 lo: usize,
2275 hi: usize,
2276) -> usize {
2277 let ring_disabled = std::env::var("MEMRA_DSA_INDEX_RING")
2278 .ok()
2279 .and_then(|v| v.trim().parse::<usize>().ok())
2280 == Some(0);
2281 plan.layers
2282 .iter()
2283 .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2284 .map(|layer| match layer.state {
2285 StatePlan::LatentKvCache { width, index_width } => {
2286 let latent = width as usize * std::mem::size_of::<f32>();
2287 let index_width = index_width as usize;
2288 let pool = cfg
2289 .glm5
2290 .as_ref()
2291 .map(|g| g.index_kpool as usize)
2292 .filter(|&p| p > 0)
2293 .unwrap_or(1);
2294 let pool_keys = if index_width > 0 {
2296 (index_width / 2) * std::mem::size_of::<f32>() / pool
2297 } else {
2298 0
2299 };
2300 let flat_plane = if index_width > 0 && ring_disabled {
2301 index_width * std::mem::size_of::<f32>()
2302 } else {
2303 0
2304 };
2305 latent + pool_keys + flat_plane
2306 }
2307 _ => 0,
2308 })
2309 .sum()
2310}
2311
2312pub fn cache_ring_bytes_per_token(cfg: &ModelConfig) -> usize {
2315 cache_ring_bytes_per_token_for_layers(cfg, 0, cfg.n_layer as usize)
2316}
2317
2318pub fn cache_ring_bytes_per_token_for_layers(cfg: &ModelConfig, lo: usize, hi: usize) -> usize {
2320 assert!(
2321 lo <= hi && hi <= cfg.n_layer as usize,
2322 "cache layer range out of bounds"
2323 );
2324 let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2325 return 0;
2326 };
2327 cache_ring_bytes_per_token_for_plan(cfg, &plan, lo, hi)
2328}
2329
2330pub fn cache_ring_bytes_per_token_for_plan(
2331 cfg: &ModelConfig,
2332 plan: &ModelPlan,
2333 lo: usize,
2334 hi: usize,
2335) -> usize {
2336 let total = plan.layers.len() + plan.mtp_blocks.len();
2337 assert!(
2338 lo <= hi && hi <= total,
2339 "cache plan layer range out of bounds"
2340 );
2341 if !swa_ring_on() {
2342 return 0;
2343 }
2344 let shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2345 plan.layers
2346 .iter()
2347 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2348 .filter(|layer| (lo..hi).contains(&(layer.index as usize)))
2349 .filter(|layer| {
2350 matches!(
2351 layer.state,
2352 memra_gguf::model_plan::StatePlan::SlidingKvCache { .. }
2353 )
2354 })
2355 .filter(|layer| shared == 0 || layer.index < cfg.n_layer - shared)
2356 .map(|layer| {
2357 let (kv_dim_k, kv_dim_v, kbb, vbb) = full_attention_kv_layout(cfg, plan, layer.index);
2358 (kv_dim_k / 32) * kbb + (kv_dim_v / 32) * vbb
2359 })
2360 .sum()
2361}
2362
2363pub fn cache_ring_row_cap(cfg: &ModelConfig) -> usize {
2365 let Ok(plan) = memra_gguf::model_plan::ModelPlan::compile(cfg) else {
2366 return 0;
2367 };
2368 cache_ring_row_cap_for_plan(&plan)
2369}
2370
2371pub fn cache_ring_row_cap_for_plan(plan: &memra_gguf::model_plan::ModelPlan) -> usize {
2372 if !swa_ring_on() {
2373 return 0;
2374 }
2375 plan.layers
2376 .iter()
2377 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2378 .filter_map(|layer| match layer.state {
2379 memra_gguf::model_plan::StatePlan::SlidingKvCache { window, .. } => {
2380 Some(window as usize)
2381 }
2382 _ => None,
2383 })
2384 .map(|window| swa_ring_rows(window, usize::MAX))
2385 .max()
2386 .unwrap_or(0)
2387}
2388
2389pub struct DflashTapSink {
2392 pub layer_ids: Vec<usize>,
2393 pub buf: CudaSlice<f32>,
2394 pub hidden: usize,
2395 pub t: usize,
2396 pub base: usize,
2399}
2400
2401pub struct HcTapSink {
2408 pub layer_ids: Vec<usize>,
2410 pub rows: Vec<f32>,
2412 pub hidden: usize,
2413 pub t: usize,
2415 pub base: usize,
2418 pub origin: usize,
2424 pub dev: Vec<Option<CudaSlice<f32>>>,
2432 pub device_stage: bool,
2440 pub dtoh_ns: u64,
2443 pub ingest_state: Option<Box<dyn std::any::Any + Send>>,
2450}
2451
2452impl HcTapSink {
2453 pub fn new(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2454 let n_taps = layer_ids.len();
2455 Self {
2456 layer_ids,
2457 rows: vec![0.0; t * n_taps * hidden],
2458 hidden,
2459 t,
2460 base: 0,
2461 origin: 0,
2462 dev: (0..n_taps).map(|_| None).collect(),
2463 device_stage: false,
2464 dtoh_ns: 0,
2465 ingest_state: None,
2466 }
2467 }
2468
2469 pub fn new_at(layer_ids: Vec<usize>, hidden: usize, t: usize, origin: usize) -> Self {
2472 Self {
2473 origin,
2474 ..Self::new(layer_ids, hidden, t)
2475 }
2476 }
2477
2478 pub fn new_device_staged(layer_ids: Vec<usize>, hidden: usize, t: usize) -> Self {
2481 Self {
2482 device_stage: true,
2483 ..Self::new(layer_ids, hidden, t)
2484 }
2485 }
2486
2487 pub fn new_device_staged_at(
2493 layer_ids: Vec<usize>,
2494 hidden: usize,
2495 t: usize,
2496 origin: usize,
2497 ) -> Self {
2498 let n_taps = layer_ids.len();
2499 Self {
2500 layer_ids,
2501 rows: Vec::new(),
2502 hidden,
2503 t,
2504 base: 0,
2505 origin,
2506 dev: (0..n_taps).map(|_| None).collect(),
2507 device_stage: true,
2508 dtoh_ns: 0,
2509 ingest_state: None,
2510 }
2511 }
2512}
2513
2514pub struct CacheSnapshot {
2544 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>>>,
2548 pub pos: usize,
2549}
2550
2551impl Cache {
2552 pub fn ensure_usable(&self, path: &str) -> Result<(), Box<dyn std::error::Error>> {
2553 if self.tainted {
2554 return Err(format!(
2555 "{path}: cache was tainted by a failed pipeline wave and cannot be reused"
2556 )
2557 .into());
2558 }
2559 Ok(())
2560 }
2561
2562 pub fn mark_tainted(&mut self) {
2563 self.tainted = true;
2564 self.last_logits_dev = None;
2565 self.dflash_taps = None;
2566 }
2567
2568 pub fn new(
2570 e: &impl KvDev,
2571 cfg: &ModelConfig,
2572 max_ctx: usize,
2573 ) -> Result<Self, Box<dyn std::error::Error>> {
2574 Self::new_inner(&|_| e, cfg, None, max_ctx)
2575 }
2576
2577 pub fn new_planned(
2578 e: &impl KvDev,
2579 cfg: &ModelConfig,
2580 plan: &memra_gguf::model_plan::ModelPlan,
2581 max_ctx: usize,
2582 ) -> Result<Self, Box<dyn std::error::Error>> {
2583 Self::new_inner(&|_| e, cfg, Some(plan), max_ctx)
2584 }
2585
2586 pub fn new_pp2(
2591 dev0: &dyn KvDev,
2592 dev1: &dyn KvDev,
2593 split: usize,
2594 cfg: &ModelConfig,
2595 max_ctx: usize,
2596 ) -> Result<Self, Box<dyn std::error::Error>> {
2597 Self::new_inner(
2598 &|il| if il < split { dev0 } else { dev1 },
2599 cfg,
2600 None,
2601 max_ctx,
2602 )
2603 }
2604
2605 pub fn new_ppn(
2611 devs: &[&dyn KvDev],
2612 fence: &[usize],
2613 cfg: &ModelConfig,
2614 max_ctx: usize,
2615 ) -> Result<Self, Box<dyn std::error::Error>> {
2616 assert_eq!(
2617 devs.len() + 1,
2618 fence.len(),
2619 "ppn cache: devs vs fence mismatch"
2620 );
2621 let pick = |il: usize| -> &dyn KvDev {
2622 let s = match fence[1..fence.len() - 1].binary_search(&il) {
2623 Ok(k) => k + 1,
2624 Err(k) => k,
2625 };
2626 devs[s.min(devs.len() - 1)]
2627 };
2628 Self::new_inner(&pick, cfg, None, max_ctx)
2629 }
2630
2631 pub fn new_ppn_planned(
2632 devs: &[&dyn KvDev],
2633 fence: &[usize],
2634 cfg: &ModelConfig,
2635 plan: &memra_gguf::model_plan::ModelPlan,
2636 max_ctx: usize,
2637 ) -> Result<Self, Box<dyn std::error::Error>> {
2638 assert_eq!(
2639 devs.len() + 1,
2640 fence.len(),
2641 "ppn cache: devs vs fence mismatch"
2642 );
2643 let pick = |il: usize| -> &dyn KvDev {
2644 let stage = match fence[1..fence.len() - 1].binary_search(&il) {
2645 Ok(index) => index + 1,
2646 Err(index) => index,
2647 };
2648 devs[stage.min(devs.len() - 1)]
2649 };
2650 Self::new_inner(&pick, cfg, Some(plan), max_ctx)
2651 }
2652
2653 fn new_inner<'a>(
2656 pick: &dyn Fn(usize) -> &'a dyn KvDev,
2657 cfg: &ModelConfig,
2658 plan: Option<&memra_gguf::model_plan::ModelPlan>,
2659 max_ctx: usize,
2660 ) -> Result<Self, Box<dyn std::error::Error>> {
2661 let fallback_plan = if plan.is_none() {
2662 Some(ModelPlan::compile(cfg)?)
2663 } else {
2664 None
2665 };
2666 let plan = plan
2667 .or(fallback_plan.as_ref())
2668 .expect("cache allocation requires a ModelPlan");
2669 let n = cfg.n_layer as usize;
2670 let mut kv = Vec::with_capacity(n);
2671 let mut recur = Vec::with_capacity(n);
2672 let mut latent = Vec::with_capacity(n);
2673 let head_dim_k = cfg.head_dim_k as usize;
2674 let head_dim_v = cfg.head_dim_v as usize;
2675 for il in 0..cfg.n_layer {
2676 let e = pick(il as usize);
2678 let layer = plan
2679 .layers
2680 .iter()
2681 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2682 .find(|layer| layer.index == il)
2683 .ok_or_else(|| format!("cache ModelPlan has no layer {il}"))?;
2684 let g4_shared = cfg.gemma4.as_ref().map(|g| g.shared_kv_layers).unwrap_or(0);
2689 if g4_shared > 0 && il >= cfg.n_layer - g4_shared {
2690 kv.push(None);
2691 recur.push(None);
2692 latent.push(None);
2693 continue;
2694 }
2695 match layer.state {
2696 StatePlan::KvCache { .. } | StatePlan::SlidingKvCache { .. } => {
2697 assert!(
2702 head_dim_k.is_multiple_of(32) && head_dim_v.is_multiple_of(32),
2703 "KVQUANT requires head_dim_k%32==0 && head_dim_v%32==0 \
2704 (layer {il}: k={head_dim_k} v={head_dim_v})"
2705 );
2706 let (kv_dim_k, kv_dim_v, kbb_l, vbb_l) =
2709 full_attention_kv_layout(cfg, plan, il);
2710 let k_tok_bytes = (kv_dim_k / 32) * kbb_l;
2711 let v_tok_bytes = (kv_dim_v / 32) * vbb_l;
2712 let planned_window = plan
2713 .layers
2714 .iter()
2715 .chain(plan.mtp_blocks.iter().map(|block| &block.layer))
2716 .find(|layer| layer.index == il)
2717 .and_then(|layer| match layer.state {
2718 StatePlan::SlidingKvCache { window, .. } => Some(window),
2719 _ => None,
2720 });
2721 let ring = if swa_ring_on() {
2722 planned_window.map(|window| {
2723 let window = window as usize;
2724 KvRing::new(swa_ring_rows(window, max_ctx), window)
2725 })
2726 } else {
2727 None
2728 };
2729 let alloc_rows = ring.as_ref().map(KvRing::rows).unwrap_or(max_ctx);
2730 kv.push(Some(KvLayer {
2731 k: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, k_tok_bytes))?,
2735 v: e.alloc_u8(kv_plane_allocation_bytes(alloc_rows, v_tok_bytes))?,
2736 kv_dim_k,
2737 kv_dim_v,
2738 k_tok_bytes,
2739 v_tok_bytes,
2740 len: 0,
2741 ring,
2742 len_d: e.htod_i32(&[0])?,
2743 base_d: None,
2744 }));
2745 recur.push(None);
2746 latent.push(None);
2747 }
2748 StatePlan::Recurrent {
2749 conv_width,
2750 conv_kernel,
2751 state_width,
2752 } => {
2753 kv.push(None);
2754 recur.push(Some(RecurLayer {
2755 conv_state: e.zeros(
2756 conv_width as usize * (conv_kernel as usize).saturating_sub(1),
2757 )?,
2758 ssm_state: e.zeros(state_width as usize)?,
2759 ssm_state_alt: e.zeros(state_width as usize)?,
2760 }));
2761 latent.push(None);
2762 }
2763 StatePlan::LatentKvCache { width, index_width } => {
2764 let width = width as usize;
2768 assert!(
2769 width > 0,
2770 "layer {il}: LatentKvCache width must be positive"
2771 );
2772 kv.push(None);
2773 recur.push(None);
2774 let index_width = index_width as usize;
2775 let index_ring = if index_width == 0 {
2779 None
2780 } else {
2781 index_ring_rows(max_ctx)
2782 };
2783 let index_rows = match index_width {
2784 0 => None,
2785 w => Some(e.zeros(index_ring.unwrap_or(max_ctx) * w)?),
2786 };
2787 latent.push(Some(LatentKvLayer {
2788 rows: e.zeros(max_ctx * width)?,
2789 width,
2790 len: 0,
2791 len_d: e.htod_i32(&[0])?,
2792 index_rows,
2793 index_width,
2794 index_ring_rows: index_ring,
2795 index_pool_keys: None,
2798 index_pools_ready: 0,
2799 index_pool: 0,
2800 }));
2801 }
2802 ref state => {
2803 return Err(format!(
2804 "native cache allocator has no implementation for layer {il} state {state:?}"
2805 )
2806 .into());
2807 }
2808 }
2809 }
2810 Ok(Cache {
2811 kv,
2812 recur,
2813 latent,
2814 tp_kv: (0..n).map(|_| None).collect(),
2815 glm5_tp_recur: (0..n).map(|_| None).collect(),
2816 glm5_tp_latent_peer: (0..n).map(|_| None).collect(),
2817 pos: 0,
2818 max_ctx,
2819 tainted: false,
2820 dflash_taps: None,
2821 hc_taps: None,
2822 glm5_decode_graph: None,
2823 last_logits_dev: None,
2824 })
2825 }
2826
2827 pub fn has_swa_ring(&self) -> bool {
2828 self.kv.iter().flatten().any(|layer| layer.ring.is_some())
2829 || self
2830 .tp_kv
2831 .iter()
2832 .flatten()
2833 .any(|layer| layer.ring_window().is_some())
2834 }
2835
2836 pub fn can_rollback(&self, snap: &CacheSnapshot, accept_len: usize) -> bool {
2837 let local = self
2838 .kv
2839 .iter()
2840 .zip(&snap.kv_len)
2841 .all(|(layer, saved)| match (layer, saved) {
2842 (Some(layer), Some(saved)) => layer
2843 .ring
2844 .as_ref()
2845 .is_none_or(|ring| ring.can_rewind_to(saved + accept_len)),
2846 _ => true,
2847 });
2848 let tensor = self
2849 .tp_kv
2850 .iter()
2851 .zip(&snap.tp_kv_len)
2852 .all(|(layer, saved)| match (layer, saved) {
2853 (Some(layer), Some(saved)) => saved
2854 .checked_add(accept_len)
2855 .is_some_and(|target| layer.can_rewind_to(target)),
2856 _ => true,
2857 });
2858 local && tensor
2859 }
2860
2861 pub fn snapshot(&self, e: &impl KvDev) -> Result<CacheSnapshot, Box<dyn std::error::Error>> {
2865 if self.glm5_tp_recur.iter().any(Option::is_some)
2869 || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2870 {
2871 return Err(
2872 "cache snapshot is unwired for glm5 TP rank state (MEMRA_GLM5_TP): \
2873 per-rank planes are not carried by CacheSnapshot"
2874 .into(),
2875 );
2876 }
2877 self.ensure_usable("cache snapshot")?;
2878 let n = self.kv.len();
2879 let mut kv_len = Vec::with_capacity(n);
2880 let mut tp_kv_len = Vec::with_capacity(n);
2881 let mut conv = Vec::with_capacity(n);
2882 let mut ssm = Vec::with_capacity(n);
2883 for il in 0..n {
2884 match &self.kv[il] {
2885 Some(kvl) => kv_len.push(Some(kvl.len)),
2886 None => kv_len.push(None),
2887 }
2888 tp_kv_len.push(
2889 self.tp_kv[il]
2890 .as_ref()
2891 .map(ResidentTpKvCache::committed_len),
2892 );
2893 match &self.recur[il] {
2894 Some(rl) => {
2895 conv.push(Some(e.clone_dtod(&rl.conv_state)?));
2896 ssm.push(Some(e.clone_dtod(&rl.ssm_state)?));
2897 }
2898 None => {
2899 conv.push(None);
2900 ssm.push(None);
2901 }
2902 }
2903 }
2904 Ok(CacheSnapshot {
2905 kv_len,
2906 tp_kv_len,
2907 conv,
2908 ssm,
2909 pos: self.pos,
2910 })
2911 }
2912
2913 pub fn snapshot_into(
2918 &self,
2919 e: &impl KvDev,
2920 snap: &mut CacheSnapshot,
2921 ) -> Result<(), Box<dyn std::error::Error>> {
2922 if self.glm5_tp_recur.iter().any(Option::is_some)
2923 || self.glm5_tp_latent_peer.iter().any(Option::is_some)
2924 {
2925 return Err("cache snapshot_into is unwired for glm5 TP rank state \
2926 (MEMRA_GLM5_TP): per-rank planes are not carried by CacheSnapshot"
2927 .into());
2928 }
2929 self.ensure_usable("cache snapshot refresh")?;
2930 let n = self.kv.len();
2931 for il in 0..n {
2932 snap.kv_len[il] = self.kv[il].as_ref().map(|kvl| kvl.len);
2933 snap.tp_kv_len[il] = self.tp_kv[il]
2934 .as_ref()
2935 .map(ResidentTpKvCache::committed_len);
2936 if let Some(rl) = &self.recur[il] {
2937 let dc = snap.conv[il]
2938 .as_mut()
2939 .expect("snapshot_into: shape mismatch (conv)");
2940 let ds = snap.ssm[il]
2941 .as_mut()
2942 .expect("snapshot_into: shape mismatch (ssm)");
2943 let (cn, sn) = (rl.conv_state.len(), rl.ssm_state.len());
2944 e.copy_into(dc, 0, &rl.conv_state, cn)?;
2945 e.copy_into(ds, 0, &rl.ssm_state, sn)?;
2946 }
2947 }
2948 snap.pos = self.pos;
2949 Ok(())
2950 }
2951
2952 pub fn rollback(
2960 &mut self,
2961 e: &impl KvDev,
2962 snap: &CacheSnapshot,
2963 accept_len: usize,
2964 ) -> Result<(), Box<dyn std::error::Error>> {
2965 self.ensure_usable("cache rollback")?;
2966 if !self.can_rollback(snap, accept_len) {
2967 return Err(
2968 "SWA ring rewind checkpoint has been lapped; full re-prime required".into(),
2969 );
2970 }
2971 for il in 0..self.kv.len() {
2972 if let (Some(kvl), Some(saved)) = (self.kv[il].as_mut(), snap.kv_len[il]) {
2973 kvl.len = saved + accept_len;
2974 e.set_i32_one(&mut kvl.len_d, kvl.len as i32)?;
2979 }
2980 if let (Some(kvl), Some(saved)) = (self.tp_kv[il].as_mut(), snap.tp_kv_len[il]) {
2981 kvl.rewind_to(saved + accept_len)?;
2982 }
2983 if let Some(rl) = self.recur[il].as_mut() {
2984 if let Some(c) = &snap.conv[il] {
2985 e.copy_into(&mut rl.conv_state, 0, c, c.len())?;
2986 }
2987 if let Some(s) = &snap.ssm[il] {
2988 e.copy_into(&mut rl.ssm_state, 0, s, s.len())?;
2989 }
2990 }
2991 }
2992 self.pos = snap.pos;
2993 Ok(())
2994 }
2995}
2996
2997#[cfg(test)]
2998mod tp_transaction_tests {
2999 use super::{
3000 Cache, INDEX_RING_WORKING_ROWS, KvRingAppend, ResidentTpKvCache, TpKvTransactionState,
3001 index_ring_default_rows, index_ring_rows_for, index_ring_take, tp_kv_rank_allocation_shape,
3002 };
3003
3004 const GLM5_NEXT_POOL: usize = 4;
3007 const GLM5_NEXT_STATE_ROW_BYTES: usize = 2 * 128 * 4;
3009 const RING_BYTES_PER_LAYER: usize = INDEX_RING_WORKING_ROWS * GLM5_NEXT_STATE_ROW_BYTES;
3012
3013 #[test]
3032 fn the_derived_ring_serves_a_monolithic_prime_of_the_whole_configured_context() {
3033 for max_ctx in [8192usize, 262_144, 1 << 20] {
3036 let rows = index_ring_default_rows(max_ctx).unwrap_or_else(|| {
3037 panic!("the ring must engage at max_ctx {max_ctx}: it is where the saving is")
3038 });
3039 let ring = rows / GLM5_NEXT_POOL * GLM5_NEXT_POOL;
3042
3043 let mut cur = 0usize;
3045 let mut pools_ready = 0usize;
3046 let mut steps = 0usize;
3047 while cur < max_ctx {
3048 let take = index_ring_take(ring, GLM5_NEXT_POOL, pools_ready, cur, max_ctx - cur)
3049 .unwrap_or_else(|| {
3050 panic!(
3051 "MEMRA_CTX={max_ctx}: the {ring}-row ring refused a monolithic prime \
3052 after {cur} of {max_ctx} tokens ({}% of the configured context). \
3053 USABLE CONTEXT MUST BE AT LEAST CONFIGURED CONTEXT. This is the \
3054 4630-of-8192 regression, in arithmetic.",
3055 cur * 100 / max_ctx
3056 )
3057 });
3058 assert!(
3059 take > 0,
3060 "MEMRA_CTX={max_ctx}: the drain made no progress at row {cur}: a zero take \
3061 is an infinite loop in the engine, not a refusal"
3062 );
3063 cur += take;
3064 pools_ready = cur / GLM5_NEXT_POOL;
3065 steps += 1;
3066 assert!(
3067 steps <= max_ctx,
3068 "MEMRA_CTX={max_ctx}: the drain did not terminate"
3069 );
3070 }
3071 assert_eq!(cur, max_ctx, "the whole prompt must be appended");
3072
3073 let ring_bytes = rows * GLM5_NEXT_STATE_ROW_BYTES;
3079 let flat_bytes = max_ctx * GLM5_NEXT_STATE_ROW_BYTES;
3080 assert_eq!(
3081 ring_bytes, RING_BYTES_PER_LAYER,
3082 "MEMRA_CTX={max_ctx}: the ring books {rows} rows, not the context-independent \
3083 {INDEX_RING_WORKING_ROWS}. A plane that tracks max_ctx is the flat plane \
3084 wearing a modulus"
3085 );
3086 assert!(
3087 rows < max_ctx,
3088 "MEMRA_CTX={max_ctx}: a ring of {rows} rows is not shorter than the flat plane \
3089 it replaces, so it would not engage at all"
3090 );
3091 assert!(
3095 ring_bytes <= 16 << 20,
3096 "MEMRA_CTX={max_ctx}: {} MiB per MLA layer, over glm5_next's 12 of them. The ring \
3097 exists to delete 11.94 GiB; a working set this large is not paying for itself",
3098 ring_bytes >> 20
3099 );
3100 println!(
3101 "MEMRA_CTX={max_ctx}: ring {rows} rows (effective {ring}), monolithic prime of \
3102 {max_ctx} tokens admitted in {steps} drain step(s); plane {} MiB/layer vs flat \
3103 {} MiB/layer",
3104 ring_bytes >> 20,
3105 flat_bytes >> 20
3106 );
3107 }
3108 }
3109
3110 fn empty_tp_cache(capacity: usize) -> ResidentTpKvCache {
3111 ResidentTpKvCache::new(Vec::new(), 128, 128, 136, 96, capacity)
3112 }
3113
3114 #[test]
3115 fn step_tp8_rank_allocation_matches_the_official_kv_geometry() {
3116 let shape = tp_kv_rank_allocation_shape(8 * 128, 8 * 128, 8).unwrap();
3117 assert_eq!((shape.kv_dim_k, shape.kv_dim_v), (128, 128));
3118 assert_eq!((shape.k_token_bytes, shape.v_token_bytes), (136, 96));
3119 assert_eq!(shape.bytes_per_token(), 232);
3120 assert_eq!(shape.fixed_bytes, 20);
3121 assert_eq!(shape.allocation_bytes(262_144), 232 * 262_144 + 20);
3122 }
3123
3124 #[test]
3125 fn tp_rank_allocation_refuses_non_divisible_and_non_block_aligned_shards() {
3126 assert!(tp_kv_rank_allocation_shape(1024, 1024, 3).is_err());
3127 assert!(tp_kv_rank_allocation_shape(1024, 1024, 64).is_err());
3128 assert!(tp_kv_rank_allocation_shape(0, 1024, 8).is_err());
3129 }
3130
3131 #[test]
3132 fn partial_commit_publishes_only_the_accepted_prefix() {
3133 let mut state = TpKvTransactionState::new();
3134 let transaction = state.begin().unwrap();
3135 let staged = state.append_target(transaction, 3, 8).unwrap();
3136 state.publish_append(transaction, staged).unwrap();
3137 assert_eq!(state.committed_len, 0);
3138 assert_eq!(state.staged_len, 3);
3139
3140 let committed = state.commit_target(transaction, 2).unwrap();
3141 state.publish_finalize(transaction, committed).unwrap();
3142 assert_eq!(state.committed_len, 2);
3143 assert_eq!(state.staged_len, 2);
3144 assert!(state.active.is_none());
3145 assert!(state.validate(transaction).is_err());
3146 }
3147
3148 #[test]
3149 fn rollback_restores_the_committed_boundary() {
3150 let mut state = TpKvTransactionState::new();
3151 let first = state.begin().unwrap();
3152 let staged = state.append_target(first, 1, 8).unwrap();
3153 state.publish_append(first, staged).unwrap();
3154 let committed = state.commit_target(first, 1).unwrap();
3155 state.publish_finalize(first, committed).unwrap();
3156
3157 let speculative = state.begin().unwrap();
3158 let staged = state.append_target(speculative, 2, 8).unwrap();
3159 state.publish_append(speculative, staged).unwrap();
3160 assert_eq!(state.committed_len, 1);
3161 assert_eq!(state.staged_len, 3);
3162 state
3163 .publish_finalize(speculative, speculative.base_len)
3164 .unwrap();
3165 assert_eq!(state.committed_len, 1);
3166 assert_eq!(state.staged_len, 1);
3167 assert!(state.validate(speculative).is_err());
3168 }
3169
3170 #[test]
3171 fn index_ring_sizing_is_pure_and_carries_no_per_call_t() {
3172 let rows = INDEX_RING_WORKING_ROWS;
3175 assert_eq!(index_ring_rows_for(None, 1 << 20), Some(rows));
3176 assert_eq!(index_ring_default_rows(1 << 20), Some(rows));
3177 assert_eq!(index_ring_rows_for(None, 4096), None);
3180 assert_eq!(index_ring_rows_for(None, rows), None);
3181 assert_eq!(index_ring_rows_for(None, rows + 1), Some(rows));
3182
3183 for max_ctx in [8192usize, 262_144, 1 << 20] {
3188 assert_eq!(
3189 index_ring_rows_for(None, max_ctx),
3190 Some(INDEX_RING_WORKING_ROWS),
3191 "the derived ring must not vary with the configured context"
3192 );
3193 }
3194
3195 assert_eq!(index_ring_rows_for(Some(0), 1 << 20), None);
3198 assert_eq!(index_ring_rows_for(Some(16), 64), Some(16));
3199 assert_eq!(index_ring_rows_for(Some(64), 64), None);
3200 }
3201
3202 #[test]
3204 fn index_ring_take_drains_instead_of_bounding_the_call() {
3205 const POOL: usize = GLM5_NEXT_POOL;
3206 assert_eq!(index_ring_take(0, POOL, 0, 0, 1 << 20), Some(1 << 20));
3208 assert_eq!(index_ring_take(64, POOL, 0, 0, 1024), Some(64));
3211 for cur in 0..64usize {
3214 let ready = cur / POOL;
3215 let take = index_ring_take(64, POOL, ready, cur, 1024).expect("never lapses");
3216 assert!(
3217 (64 - POOL + 1..=64).contains(&take),
3218 "cur {cur}: take {take} outside the guaranteed progress band"
3219 );
3220 }
3221 assert_eq!(index_ring_take(64, POOL, 4, 16, 1), Some(1));
3223 assert_eq!(index_ring_take(16, POOL, 0, 64, 1), None);
3226 assert_eq!(index_ring_take(16, POOL, 0, 16, 1), None);
3227 assert_eq!(index_ring_take(16, POOL, 0, 15, 1), Some(1));
3228 }
3229
3230 #[test]
3231 fn rejects_nested_stale_and_out_of_range_actions() {
3232 let mut state = TpKvTransactionState::new();
3233 let transaction = state.begin().unwrap();
3234 assert!(state.begin().is_err());
3235 assert!(state.append_target(transaction, 0, 2).is_err());
3236 assert!(state.append_target(transaction, 3, 2).is_err());
3237 let staged = state.append_target(transaction, 2, 2).unwrap();
3238 state.publish_append(transaction, staged).unwrap();
3239 assert!(state.commit_target(transaction, 3).is_err());
3240 state.publish_finalize(transaction, 0).unwrap();
3241 assert!(state.publish_append(transaction, 1).is_err());
3242 }
3243
3244 #[test]
3245 fn rewind_resets_visibility_and_invalidates_an_active_transaction() {
3246 let mut state = TpKvTransactionState::new();
3247 let transaction = state.begin().unwrap();
3248 let staged = state.append_target(transaction, 3, 8).unwrap();
3249 state.publish_append(transaction, staged).unwrap();
3250 state.rewind(1, 8).unwrap();
3251 assert_eq!(state.committed_len, 1);
3252 assert_eq!(state.staged_len, 1);
3253 assert!(state.active.is_none());
3254 assert!(state.validate(transaction).is_err());
3255 assert!(state.rewind(9, 8).is_err());
3256 }
3257
3258 #[test]
3259 fn device_rewind_updates_host_visibility_after_external_rank_writes() {
3260 let mut cache = empty_tp_cache(8);
3261 let transaction = cache.begin_transaction().unwrap();
3262 let staged = cache.append_target(transaction, 5).unwrap();
3263 cache.publish_append(transaction, staged).unwrap();
3264 let committed = cache.commit_target(transaction, 5).unwrap();
3265 cache.publish_finalize(transaction, committed).unwrap();
3266 cache.publish_device_rewind(3).unwrap();
3267 assert_eq!(cache.committed_len(), 3);
3268 assert_eq!(cache.staged_len(), 3);
3269 assert!(cache.publish_device_rewind(9).is_err());
3270 }
3271
3272 #[test]
3273 fn grow_preserves_generation_and_publishes_only_the_checkpoint_prefix() {
3274 let mut source = empty_tp_cache(8);
3275 let first = source.begin_transaction().unwrap();
3276 let staged = source.append_target(first, 5).unwrap();
3277 source.publish_append(first, staged).unwrap();
3278 let committed = source.commit_target(first, 5).unwrap();
3279 source.publish_finalize(first, committed).unwrap();
3280
3281 let rolled_back = source.begin_transaction().unwrap();
3282 source
3283 .publish_finalize(rolled_back, rolled_back.base_len())
3284 .unwrap();
3285 let plan = source.prepare_grow(16, 3).unwrap();
3286 assert_eq!(plan.rows(), 3);
3287 assert_eq!(plan.k_bytes(), 3 * 136);
3288 assert_eq!(plan.v_bytes(), 3 * 96);
3289
3290 let mut target = empty_tp_cache(16);
3291 target.publish_grow(plan).unwrap();
3292 assert_eq!(target.committed_len(), 3);
3293 assert_eq!(target.staged_len(), 3);
3294 assert_eq!(target.capacity(), 16);
3295 let next = target.begin_transaction().unwrap();
3296 assert_eq!(next.generation(), rolled_back.generation() + 1);
3297 assert_eq!(next.base_len(), 3);
3298 }
3299
3300 #[test]
3301 fn grow_refuses_active_source_and_invalid_target_state_or_layout() {
3302 let mut active = empty_tp_cache(8);
3303 active.begin_transaction().unwrap();
3304 assert!(active.prepare_grow(16, 0).is_err());
3305
3306 let mut source = empty_tp_cache(8);
3307 source.rewind_to(5).unwrap();
3308 assert!(source.prepare_grow(8, 5).is_err());
3309 assert!(source.prepare_grow(16, 6).is_err());
3310 let plan = source.prepare_grow(16, 4).unwrap();
3311
3312 let mut wrong_layout = ResidentTpKvCache::new(Vec::new(), 128, 128, 144, 96, 16);
3313 assert!(wrong_layout.publish_grow(plan).is_err());
3314
3315 let plan = source.prepare_grow(16, 4).unwrap();
3316 let mut dirty_target = empty_tp_cache(16);
3317 dirty_target.rewind_to(1).unwrap();
3318 assert!(dirty_target.publish_grow(plan).is_err());
3319 }
3320
3321 #[test]
3322 fn swa_transaction_rebase_preserves_the_rollback_window() {
3323 let mut cache = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3324 assert_eq!(cache.physical_capacity(), 32 + 4096 + 512 + 31);
3325 cache.publish_hydration(8762, 4096).unwrap();
3329 assert_eq!(cache.ring_base(), Some(4096));
3330
3331 let transaction = cache.begin_transaction().unwrap();
3332 let plan = cache.prepare_append(transaction, 10).unwrap();
3333 assert_eq!(plan.target(), 8772);
3334 assert_eq!(plan.write_row(), 58);
3335 assert_eq!(
3336 plan.ring_append(),
3337 Some(KvRingAppend::Rebase {
3338 src_row: 4608,
3339 keep_rows: 58,
3340 new_base: 8704,
3341 write_row: 58,
3342 })
3343 );
3344 cache.publish_append_rebase(plan).unwrap();
3345 cache.publish_append_plan(plan).unwrap();
3346 assert_eq!(cache.ring_base(), Some(8704));
3347 assert_eq!(cache.physical_range(8740, 8772).unwrap(), 36..68);
3348
3349 let rollback = cache.commit_target(transaction, 0).unwrap();
3350 cache.publish_finalize(transaction, rollback).unwrap();
3351 assert_eq!((cache.committed_len(), cache.staged_len()), (8762, 8762));
3352 assert!(cache.rewind_to(8200).is_err());
3353 }
3354
3355 #[test]
3356 fn swa_grow_normalizes_only_the_live_prefix_and_preserves_generation() {
3357 let mut source = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 10_000, 32);
3358 source.publish_hydration(8250, 4096).unwrap();
3359 let transaction = source.begin_transaction().unwrap();
3360 source
3361 .publish_finalize(transaction, transaction.base_len())
3362 .unwrap();
3363
3364 let plan = source.prepare_grow(20_000, 8250).unwrap();
3365 assert_eq!(plan.source_row(), 4096);
3366 assert_eq!(plan.copy_rows(), 58);
3367 assert_eq!(plan.k_bytes(), 58 * 136);
3368 assert_eq!(plan.v_bytes(), 58 * 96);
3369
3370 let mut target = ResidentTpKvCache::new_swa(Vec::new(), 128, 128, 136, 96, 20_000, 32);
3371 target.publish_grow(plan).unwrap();
3372 assert_eq!(target.ring_base(), Some(8192));
3373 assert_eq!((target.committed_len(), target.staged_len()), (8250, 8250));
3374 assert_eq!(target.physical_range(8192, 8250).unwrap(), 0..58);
3375 let next = target.begin_transaction().unwrap();
3376 assert_eq!(next.generation(), transaction.generation() + 1);
3377 }
3378
3379 #[test]
3380 fn cache_reports_a_materialized_distributed_swa_ring() {
3381 let mut cache = Cache {
3382 kv: Vec::new(),
3383 recur: Vec::new(),
3384 latent: Vec::new(),
3385 tp_kv: vec![None],
3386 glm5_tp_recur: vec![None],
3387 glm5_tp_latent_peer: vec![None],
3388 pos: 0,
3389 max_ctx: 10_000,
3390 tainted: false,
3391 dflash_taps: None,
3392 hc_taps: None,
3393 glm5_decode_graph: None,
3394 last_logits_dev: None,
3395 };
3396 assert!(!cache.has_swa_ring());
3397 cache.tp_kv[0] = Some(ResidentTpKvCache::new_swa(
3398 Vec::new(),
3399 128,
3400 128,
3401 136,
3402 96,
3403 10_000,
3404 512,
3405 ));
3406 assert!(cache.has_swa_ring());
3407 }
3408}
3409
3410#[cfg(test)]
3411mod swa_ring_tests {
3412 use super::{
3413 KvRing, KvRingAppend, PRIME_CHUNK_MAX_TOKENS, SWA_REWIND_SLACK_ROWS,
3414 SWA_VIEW_ALIGNMENT_ROWS, kv_plane_allocation_bytes, swa_retain_from, swa_ring_rows,
3415 };
3416
3417 #[test]
3418 fn allocation_rows_cover_window_max_prime_and_alignment_slack() {
3419 assert_eq!(swa_ring_rows(512, 262_144), 512 + 4096 + 512 + 31);
3420 assert_eq!(swa_ring_rows(512, 4096), 4096);
3421 assert_eq!(
3422 kv_plane_allocation_bytes(5151, 1088),
3423 5151 * 1088 + 8,
3424 "the Step35 session plane allocates ring rows plus the existing tail pad",
3425 );
3426 }
3427
3428 #[test]
3441 fn retain_grants_rewind_slack_but_never_asks_below_base() {
3442 const WINDOW: usize = 512;
3443 let rows = swa_ring_rows(WINDOW, 262_144);
3444 let len = rows;
3445 let aligned = |pos: usize| (pos - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3446
3447 let retain = swa_retain_from(len, WINDOW, 0);
3450 assert!(retain <= aligned(len) - SWA_REWIND_SLACK_ROWS);
3451 let mut ring = KvRing::new(rows, WINDOW);
3452 ring.apply_rebase(retain);
3453 assert!(
3454 ring.can_rewind_to(len - 1),
3455 "a one-token rewind must survive the rebase"
3456 );
3457 assert!(ring.can_rewind_to(len - SWA_REWIND_SLACK_ROWS));
3458
3459 assert!(len - retain + PRIME_CHUNK_MAX_TOKENS <= rows);
3461
3462 for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3466 assert!(
3467 ring.can_rewind_to(len - depth),
3468 "a {depth}-row rewind must be resident, not clamped away",
3469 );
3470 }
3471
3472 let base = ring.base();
3479 for depth in [1usize, 32, 256, SWA_REWIND_SLACK_ROWS] {
3480 let window_start = (len - depth - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3481 assert!(
3482 window_start >= base,
3483 "after a {depth}-row rewind the window starts at {window_start}, below base \
3484 {base} — clamping here would serve rows the ring no longer holds (the pos-8661 \
3485 all-NaN case)",
3486 );
3487 assert!(swa_retain_from(len - depth, WINDOW, base) >= base);
3488 }
3489
3490 let past = len - (SWA_REWIND_SLACK_ROWS + WINDOW);
3493 let past_start = (past - (WINDOW - 1)) & !(SWA_VIEW_ALIGNMENT_ROWS - 1);
3494 assert!(
3495 past_start < base,
3496 "beyond the headroom the window must fall below base"
3497 );
3498 assert!(
3499 !ring.can_rewind_to(past),
3500 "and can_rewind_to must refuse it"
3501 );
3502 }
3503
3504 #[test]
3505 fn ring_matches_flat_bytes_before_wrap() {
3506 let ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3507 let flat: Vec<u32> = (0..1024).collect();
3508 let mut physical = vec![u32::MAX; ring.rows()];
3509 let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, flat.len()).unwrap()
3510 else {
3511 panic!("first append unexpectedly wrapped")
3512 };
3513 physical[write_row..write_row + flat.len()].copy_from_slice(&flat);
3514 let view = ring.physical_range(0, flat.len()).unwrap();
3515 assert_eq!(&physical[view], flat.as_slice());
3516 }
3517
3518 #[test]
3519 fn wrap_rebases_the_exact_aligned_prime_view() {
3520 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3521 let flat: Vec<u32> = (0..8192).collect();
3522 let mut physical = vec![u32::MAX; ring.rows()];
3523 let KvRingAppend::Contiguous { write_row } = ring.append_plan(0, 0, 4096).unwrap() else {
3524 panic!("first prime chunk unexpectedly wrapped")
3525 };
3526 physical[write_row..write_row + 4096].copy_from_slice(&flat[..4096]);
3527
3528 let off = (4096usize - (512 - 1)) & !31usize;
3529 let KvRingAppend::Rebase {
3530 src_row,
3531 keep_rows,
3532 new_base,
3533 write_row,
3534 } = ring.append_plan(4096, off, 4096).unwrap()
3535 else {
3536 panic!("second prime chunk did not wrap")
3537 };
3538 let retained = physical[src_row..src_row + keep_rows].to_vec();
3539 physical[..keep_rows].copy_from_slice(&retained);
3540 ring.apply_rebase(new_base);
3541 physical[write_row..write_row + 4096].copy_from_slice(&flat[4096..8192]);
3542
3543 let view = ring.physical_range(off, 8192).unwrap();
3544 assert_eq!(&physical[view], &flat[off..8192]);
3545 assert_eq!(ring.base(), off);
3546 }
3547
3548 #[test]
3549 fn rewind_declines_once_the_required_window_was_lapped() {
3550 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3551 let KvRingAppend::Rebase { new_base, .. } = ring.append_plan(4096, 3584, 4096).unwrap()
3552 else {
3553 panic!("expected wrap")
3554 };
3555 ring.apply_rebase(new_base);
3556 assert!(ring.can_rewind_to(4095));
3557 assert!(!ring.can_rewind_to(4094));
3558 assert!(!ring.can_rewind_to(0));
3559 }
3560
3561 #[test]
3566 fn restore_plan_copies_the_window_not_the_absolute_length() {
3567 let mut ring = KvRing::new(swa_ring_rows(512, 262_144), 512);
3568 let (base, phys) = ring.restore_plan(400).unwrap();
3570 assert_eq!((base, phys), (0, 0..400));
3571
3572 let mut live = 0usize;
3575 while live < 40_960 {
3576 let retain = swa_retain_from(live, 512, ring.base());
3577 if let KvRingAppend::Rebase { new_base, .. } =
3578 ring.append_plan(live, retain, 4096).unwrap()
3579 {
3580 ring.apply_rebase(new_base);
3581 }
3582 live += 4096;
3583 }
3584 assert!(ring.base() > 0, "a 40k walk must have lapped the ring");
3585 let (base, phys) = ring.restore_plan(live).unwrap();
3586 assert_eq!(base, (live - (512 - 1)) & !31usize);
3587 assert!(
3588 base >= ring.base(),
3589 "the plan must stay above the ring floor"
3590 );
3591 assert_eq!(phys.len(), live - base);
3592 assert!(
3593 phys.end <= ring.rows(),
3594 "the copy must fit the physical buffer ({} rows), got {:?}",
3595 ring.rows(),
3596 phys
3597 );
3598
3599 assert!(ring.restore_plan(400).is_err());
3601 }
3602}