1use std::{
27 hash::BuildHasherDefault,
28 mem::size_of_val,
29 sync::{
30 Arc,
31 atomic::{AtomicU64, Ordering},
32 },
33};
34
35use aho_corasick::AhoCorasick;
36use moka::sync::Cache;
37use rustc_hash::FxHasher;
38
39use crate::{TokenIdType, traits::Encoder};
40
41type Blake3Hash = [u8; 32];
43
44type PrefixHasher = BuildHasherDefault<FxHasher>;
47
48type PrefixCache = Cache<Blake3Hash, Arc<[TokenIdType]>, PrefixHasher>;
50
51fn boundaries_with(text: &str, matcher: &AhoCorasick) -> Vec<usize> {
62 let mut boundaries: Vec<usize> = matcher
63 .find_overlapping_iter(text)
64 .map(|m| m.end())
65 .filter(|&end| end < text.len())
66 .collect();
67 boundaries.sort_unstable();
68 boundaries.dedup();
69 boundaries
70}
71
72#[cfg(test)]
75fn find_special_token_boundaries(text: &str, special_tokens: &[&str]) -> Vec<usize> {
76 if special_tokens.is_empty() {
77 return Vec::new();
78 }
79 let matcher = AhoCorasick::new(special_tokens)
80 .expect("special tokens form a valid Aho-Corasick automaton");
81 boundaries_with(text, &matcher)
82}
83
84pub type CacheEventFn = Arc<dyn Fn() + Send + Sync>;
88
89pub struct L1Cache {
93 cache: PrefixCache,
95 matcher: Option<AhoCorasick>,
98 hits: AtomicU64,
99 misses: AtomicU64,
100 on_hit: Option<CacheEventFn>,
101 on_miss: Option<CacheEventFn>,
102}
103
104impl L1Cache {
105 pub fn new(max_memory: usize, special_tokens: Vec<String>) -> Self {
108 let cache = Cache::builder()
112 .max_capacity(max_memory as u64)
113 .weigher(|_k: &Blake3Hash, tokens: &Arc<[TokenIdType]>| -> u32 {
114 size_of_val(tokens.as_ref()).min(u32::MAX as usize) as u32
115 })
116 .build_with_hasher(PrefixHasher::default());
117
118 let matcher = (!special_tokens.is_empty()).then(|| {
120 AhoCorasick::new(&special_tokens)
121 .expect("special tokens form a valid Aho-Corasick automaton")
122 });
123
124 Self {
125 cache,
126 matcher,
127 hits: AtomicU64::new(0),
128 misses: AtomicU64::new(0),
129 on_hit: None,
130 on_miss: None,
131 }
132 }
133
134 pub fn set_observer(&mut self, on_hit: CacheEventFn, on_miss: CacheEventFn) {
136 self.on_hit = Some(on_hit);
137 self.on_miss = Some(on_miss);
138 }
139
140 fn boundaries(&self, text: &str) -> Vec<usize> {
144 match &self.matcher {
145 Some(matcher) => boundaries_with(text, matcher),
146 None => Vec::new(),
147 }
148 }
149
150 pub fn longest_prefix_match(&self, input: &str) -> Option<(Arc<[TokenIdType]>, usize, usize)> {
157 let boundaries = self.boundaries(input);
158
159 if boundaries.is_empty() {
160 self.misses.fetch_add(1, Ordering::Relaxed);
161 if let Some(cb) = &self.on_miss {
162 cb();
163 }
164 return None;
165 }
166
167 let deepest_boundary = *boundaries.last().expect("boundaries is non-empty here");
170
171 let mut hasher = blake3::Hasher::new();
173 let mut prefix_hashes = Vec::with_capacity(boundaries.len());
174 let mut last_pos = 0;
175 let bytes = input.as_bytes();
176 for &boundary_pos in &boundaries {
177 hasher.update(&bytes[last_pos..boundary_pos]);
178 prefix_hashes.push((boundary_pos, *hasher.finalize().as_bytes()));
180 last_pos = boundary_pos;
181 }
182
183 for (boundary_pos, hash_bytes) in prefix_hashes.into_iter().rev() {
186 if let Some(tokens) = self.cache.get(&hash_bytes) {
187 self.hits.fetch_add(1, Ordering::Relaxed);
188 if let Some(cb) = &self.on_hit {
189 cb();
190 }
191 return Some((tokens, boundary_pos, deepest_boundary));
195 }
196 }
197
198 self.misses.fetch_add(1, Ordering::Relaxed);
199 if let Some(cb) = &self.on_miss {
200 cb();
201 }
202 None
203 }
204
205 pub fn insert_at_boundaries<E: Encoder + ?Sized>(
213 &self,
214 input: &str,
215 tokenizer: &E,
216 ) -> anyhow::Result<()> {
217 let boundaries = self.boundaries(input);
218 if boundaries.is_empty() {
219 return Ok(());
220 }
221 self.populate_boundaries(input, &boundaries, tokenizer)?;
222 Ok(())
223 }
224
225 pub fn populate_and_encode<E: Encoder + ?Sized>(
234 &self,
235 input: &str,
236 tokenizer: &E,
237 ) -> anyhow::Result<Vec<TokenIdType>> {
238 let boundaries = self.boundaries(input);
239 if boundaries.is_empty() {
240 return Ok(tokenizer.encode(input)?.token_ids().to_vec());
242 }
243
244 let mut running = self.populate_boundaries(input, &boundaries, tokenizer)?;
246
247 let tail_start = *boundaries.last().expect("boundaries is non-empty here");
250 let tail = tokenizer.encode(&input[tail_start..])?;
251 running.extend_from_slice(tail.token_ids());
252 Ok(running)
253 }
254
255 fn populate_boundaries<E: Encoder + ?Sized>(
259 &self,
260 input: &str,
261 boundaries: &[usize],
262 tokenizer: &E,
263 ) -> anyhow::Result<Vec<TokenIdType>> {
264 let mut hasher = blake3::Hasher::new();
265 let mut running_tokens: Vec<TokenIdType> = Vec::new();
266 let mut last_pos = 0;
267 let bytes = input.as_bytes();
268
269 for &boundary_pos in boundaries {
270 hasher.update(&bytes[last_pos..boundary_pos]);
272 let hash_bytes: Blake3Hash = *hasher.finalize().as_bytes();
273
274 let seg = tokenizer.encode(&input[last_pos..boundary_pos])?;
278 running_tokens.extend_from_slice(seg.token_ids());
279
280 let prefix_tokens: Arc<[TokenIdType]> = running_tokens.as_slice().into();
283 self.cache.insert(hash_bytes, prefix_tokens);
284
285 last_pos = boundary_pos;
286 }
287
288 Ok(running_tokens)
289 }
290
291 pub fn extend_after_match<E: Encoder + ?Sized>(
308 &self,
309 input: &str,
310 prefix_tokens: Arc<[TokenIdType]>,
311 prefix_len: usize,
312 deepest_boundary: usize,
313 tokenizer: &E,
314 ) -> anyhow::Result<Vec<TokenIdType>> {
315 let deepest = (deepest_boundary > prefix_len).then_some(deepest_boundary);
321
322 let Some(deepest) = deepest else {
323 let suffix_enc = tokenizer.encode(&input[prefix_len..])?;
327 let mut merged = Vec::with_capacity(prefix_tokens.len() + suffix_enc.token_ids().len());
328 merged.extend_from_slice(&prefix_tokens);
329 merged.extend_from_slice(suffix_enc.token_ids());
330 return Ok(merged);
331 };
332
333 let seg_a = tokenizer.encode(&input[prefix_len..deepest])?;
340 let seg_b = tokenizer.encode(&input[deepest..])?;
341 let mut cumulative = Vec::with_capacity(
342 prefix_tokens.len() + seg_a.token_ids().len() + seg_b.token_ids().len(),
343 );
344 cumulative.extend_from_slice(&prefix_tokens);
345 cumulative.extend_from_slice(seg_a.token_ids());
346
347 let mut hasher = blake3::Hasher::new();
351 hasher.update(&input.as_bytes()[..deepest]);
352 let hash_bytes: Blake3Hash = *hasher.finalize().as_bytes();
353
354 let tokens: Arc<[TokenIdType]> = cumulative.as_slice().into();
357 self.cache.insert(hash_bytes, tokens);
358
359 cumulative.extend_from_slice(seg_b.token_ids());
362 Ok(cumulative)
363 }
364
365 pub fn len(&self) -> usize {
368 self.cache.run_pending_tasks();
369 self.cache.entry_count() as usize
370 }
371
372 pub fn is_empty(&self) -> bool {
373 self.len() == 0
374 }
375
376 pub fn stats(&self) -> L1CacheStats {
377 self.cache.run_pending_tasks();
379 let hits = self.hits.load(Ordering::Relaxed);
380 let misses = self.misses.load(Ordering::Relaxed);
381 let total_requests = hits + misses;
382
383 L1CacheStats {
384 hits,
385 misses,
386 entries: self.cache.entry_count() as usize,
387 memory_bytes: self.cache.weighted_size() as usize,
388 hit_rate: if total_requests > 0 {
389 hits as f64 / total_requests as f64
390 } else {
391 0.0
392 },
393 }
394 }
395
396 pub fn clear(&self) {
397 self.cache.invalidate_all();
398 self.cache.run_pending_tasks();
399 self.hits.store(0, Ordering::Relaxed);
400 self.misses.store(0, Ordering::Relaxed);
401 }
402}
403
404#[derive(Debug, Clone)]
405pub struct L1CacheStats {
406 pub hits: u64,
407 pub misses: u64,
408 pub entries: usize,
409 pub memory_bytes: usize,
410 pub hit_rate: f64,
411}
412
413#[cfg(test)]
414mod tests {
415 use std::sync::Arc;
416
417 use super::*;
418 use crate::{HuggingFaceTokenizer, traits::Tokenizer};
419
420 const TINYLLAMA_PATH: &str = concat!(
423 env!("CARGO_MANIFEST_DIR"),
424 "/tests/data/sample-models/TinyLlama_v1.1/tokenizer.json"
425 );
426
427 const SPECIALS: &[&str] = &["<s>", "</s>"];
428
429 fn load_tokenizer() -> Arc<dyn Tokenizer> {
430 Arc::new(HuggingFaceTokenizer::from_file(TINYLLAMA_PATH).expect("load TinyLlama"))
431 }
432
433 fn test_cache(max_memory: usize) -> L1Cache {
435 L1Cache::new(
436 max_memory,
437 SPECIALS.iter().map(|s| (*s).to_string()).collect(),
438 )
439 }
440
441 #[test]
442 fn boundaries_are_after_each_special_token_occurrence() {
443 let input = "<s>system\nHi</s><s>user\nHello</s>";
444 let bounds = find_special_token_boundaries(input, SPECIALS);
445 assert_eq!(bounds.len(), 3);
447 for w in bounds.windows(2) {
448 assert!(w[0] < w[1], "boundaries must be strictly increasing");
449 }
450 assert!(bounds.iter().all(|&b| b < input.len()));
451 }
452
453 #[test]
454 fn no_special_tokens_yields_no_boundaries() {
455 assert!(find_special_token_boundaries("plain text", &[]).is_empty());
456 }
457
458 #[test]
459 fn insert_then_lookup_finds_shared_prefix() {
460 let cache = test_cache(1024 * 1024);
461 let tokenizer = load_tokenizer();
462
463 let warm = "<s>system\nYou are helpful.</s><s>user\nHi</s>";
464 cache
465 .insert_at_boundaries(warm, tokenizer.as_ref())
466 .unwrap();
467 assert!(!cache.is_empty());
468
469 let target = "<s>system\nYou are helpful.</s><s>user\nDifferent question</s>";
470 let (tokens, offset, _deepest) = cache
471 .longest_prefix_match(target)
472 .expect("shared prefix should match");
473 assert!(offset > 0);
474 assert!(!tokens.is_empty());
475 }
476
477 #[test]
478 fn miss_increments_misses_counter() {
479 let cache = test_cache(1024 * 1024);
480 assert!(
481 cache
482 .longest_prefix_match("plain text no specials")
483 .is_none()
484 );
485 assert_eq!(cache.stats().misses, 1);
486 }
487
488 #[test]
489 fn hit_increments_hits_counter() {
490 let cache = test_cache(1024 * 1024);
491 let tokenizer = load_tokenizer();
492 let warm = "<s>system\nA.</s><s>user\nB</s>";
493 cache
494 .insert_at_boundaries(warm, tokenizer.as_ref())
495 .unwrap();
496 let _ = cache.longest_prefix_match(warm);
497 assert!(cache.stats().hits >= 1);
498 }
499
500 #[test]
501 fn merge_invariant_holds_against_uncached_encode() {
502 let cache = test_cache(1024 * 1024);
506 let tokenizer = load_tokenizer();
507
508 let template = "<s>system\nYou are helpful.</s><s>user\n";
509 let warm = format!("{template}First.</s>");
510 cache
511 .insert_at_boundaries(&warm, tokenizer.as_ref())
512 .unwrap();
513
514 let target = format!("{template}A completely different second question.</s>");
515 let (prefix_tokens, prefix_len, _deepest) = cache
516 .longest_prefix_match(&target)
517 .expect("should find prefix");
518
519 let suffix = &target[prefix_len..];
520 let suffix_enc = tokenizer.encode(suffix).unwrap();
521 let mut merged = prefix_tokens.to_vec();
523 merged.extend_from_slice(suffix_enc.token_ids());
524
525 let plain = tokenizer.encode(&target).unwrap();
526 assert_eq!(
527 merged,
528 plain.token_ids(),
529 "merged tokens must equal plain encode"
530 );
531 }
532
533 #[test]
534 fn eviction_respects_memory_budget() {
535 let cache = test_cache(4 * 1024);
537 let tokenizer = load_tokenizer();
538 for i in 0..50 {
539 let input =
540 format!("<s>system\nPersona {i} chatty.</s><s>user\nTurn {i} content here.</s>");
541 cache
542 .insert_at_boundaries(&input, tokenizer.as_ref())
543 .unwrap();
544 }
545 let stats = cache.stats();
546 assert!(
547 stats.memory_bytes <= 4 * 1024,
548 "memory_bytes={} exceeds budget",
549 stats.memory_bytes
550 );
551 }
552
553 #[test]
554 fn concurrent_inserts_and_lookups_do_not_corrupt() {
555 use std::thread;
556
557 let cache = Arc::new(test_cache(1024 * 1024));
558 let tokenizer = load_tokenizer();
559
560 let mut handles = vec![];
561 for i in 0..10 {
562 let cache_c = cache.clone();
563 let tok = tokenizer.clone();
564 handles.push(thread::spawn(move || {
565 let input = format!("<s>system\nThread {i}.</s><s>user\nThread {i} body.</s>");
566 cache_c.insert_at_boundaries(&input, tok.as_ref()).unwrap();
567 let r = cache_c.longest_prefix_match(&input);
568 assert!(r.is_some(), "thread {i} expected match after insert");
569 }));
570 }
571 for h in handles {
572 h.join().unwrap();
573 }
574 assert!(cache.stats().memory_bytes > 0);
575 assert!(cache.stats().hits >= 10);
576 }
577
578 fn growing_chat_turns(n: usize) -> Vec<String> {
584 let mut convo = String::from("<s>system\nYou are a helpful assistant.</s>");
585 let mut turns = Vec::with_capacity(n);
586 for i in 0..n {
587 convo.push_str(&format!(
588 "<s>user\nQuestion {i} please answer it.</s><s>assistant\nDetailed answer {i} follows here.</s>"
589 ));
590 turns.push(format!("{convo}<s>user\nFollow-up {i}"));
591 }
592 turns
593 }
594
595 #[test]
596 fn extend_on_hit_advances_match_depth_each_turn() {
597 let tok = load_tokenizer();
601 let turns = growing_chat_turns(5);
602
603 let off = test_cache(8 * 1024 * 1024);
605 off.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
606 let pinned = off.longest_prefix_match(&turns[1]).expect("hit").1;
607 for t in &turns[1..] {
608 let (_toks, offset, _deepest) = off.longest_prefix_match(t).expect("hit");
609 assert_eq!(
610 offset, pinned,
611 "extend-off offset must stay pinned at turn-1 depth"
612 );
613 }
614
615 let on = test_cache(8 * 1024 * 1024);
617 on.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
618 let mut prev = 0usize;
619 for (i, t) in turns.iter().enumerate().skip(1) {
620 let (prefix_tokens, offset, deepest) = on.longest_prefix_match(t).expect("hit");
621 assert!(
622 offset > prev,
623 "turn {i}: extend-on offset {offset} must exceed previous {prev}"
624 );
625 prev = offset;
626
627 let merged = on
629 .extend_after_match(t, prefix_tokens, offset, deepest, tok.as_ref())
630 .unwrap();
631 let plain = tok.encode(t).unwrap();
632 assert_eq!(
633 merged,
634 plain.token_ids(),
635 "turn {i}: extend merge must equal plain encode"
636 );
637 }
638
639 assert!(
640 prev > pinned,
641 "extend-on frontier ({prev}) must reach deeper than pinned extend-off depth ({pinned})"
642 );
643 }
644
645 #[test]
646 fn extend_on_hit_respects_budget_and_stays_correct() {
647 let tok = load_tokenizer();
650 let cache = test_cache(4 * 1024);
651 let turns = growing_chat_turns(20);
652 cache.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
653
654 for t in &turns[1..] {
655 let merged = match cache.longest_prefix_match(t) {
656 Some((prefix_tokens, offset, deepest)) => cache
657 .extend_after_match(t, prefix_tokens, offset, deepest, tok.as_ref())
658 .unwrap(),
659 None => {
660 let enc = tok.encode(t).unwrap();
662 cache.insert_at_boundaries(t, tok.as_ref()).unwrap();
663 enc.token_ids().to_vec()
664 }
665 };
666 let plain = tok.encode(t).unwrap();
667 assert_eq!(
668 merged,
669 plain.token_ids(),
670 "encode must stay correct under eviction pressure"
671 );
672 assert!(
673 cache.stats().memory_bytes <= 4 * 1024,
674 "memory_bytes={} exceeds budget",
675 cache.stats().memory_bytes
676 );
677 }
678 }
679
680 #[test]
681 fn concurrent_extend_on_hit_does_not_corrupt() {
682 use std::thread;
683
684 let tok = load_tokenizer();
685 let cache = Arc::new(test_cache(8 * 1024 * 1024));
686 let turns = growing_chat_turns(8);
687 cache.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
689
690 let mut handles = vec![];
691 for _ in 0..8 {
692 let cache_c = cache.clone();
693 let tok_c = tok.clone();
694 let turns_c = turns.clone();
695 handles.push(thread::spawn(move || {
696 for t in &turns_c[1..] {
697 if let Some((prefix_tokens, offset, deepest)) = cache_c.longest_prefix_match(t)
698 {
699 let merged = cache_c
700 .extend_after_match(t, prefix_tokens, offset, deepest, tok_c.as_ref())
701 .unwrap();
702 let plain = tok_c.encode(t).unwrap();
703 assert_eq!(
704 merged,
705 plain.token_ids(),
706 "concurrent extend must stay correct"
707 );
708 }
709 }
710 }));
711 }
712 for h in handles {
713 h.join().unwrap();
714 }
715 assert!(cache.stats().memory_bytes > 0);
716 }
717
718 #[test]
719 fn extend_after_match_persists_correct_deepest_entry() {
720 let tok = load_tokenizer();
726 let turns = growing_chat_turns(3);
727
728 let cache = test_cache(8 * 1024 * 1024);
729 cache.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
730
731 let (prefix_tokens, prefix_len, deepest_boundary) = cache
732 .longest_prefix_match(&turns[1])
733 .expect("partial hit on turns[1]");
734 let entries_before = cache.stats().entries;
735
736 let _merged = cache
737 .extend_after_match(
738 &turns[1],
739 prefix_tokens,
740 prefix_len,
741 deepest_boundary,
742 tok.as_ref(),
743 )
744 .unwrap();
745
746 assert_eq!(
747 cache.stats().entries,
748 entries_before + 1,
749 "extend must persist exactly one (deepest) entry"
750 );
751
752 let deepest = find_special_token_boundaries(&turns[1], SPECIALS)
755 .into_iter()
756 .rev()
757 .find(|&b| b > prefix_len)
758 .expect("a deeper boundary must exist in the appended turn");
759 assert_eq!(
760 deepest_boundary, deepest,
761 "longest_prefix_match must return the deepest boundary used by extend"
762 );
763
764 let (saved_tokens, saved_offset, _deepest) = cache
767 .longest_prefix_match(&turns[1])
768 .expect("hit after extend");
769 assert_eq!(
770 saved_offset, deepest,
771 "lookup must now hit at the just-saved deepest boundary"
772 );
773 let expected = tok.encode(&turns[1][..deepest]).unwrap();
774 assert_eq!(
775 &*saved_tokens,
776 expected.token_ids(),
777 "persisted entry tokens must equal the uncached encode of the cached prefix"
778 );
779 }
780
781 #[test]
782 fn boundaries_detected_for_multibyte_deepseek_tool_tokens() {
783 let specials = &["<|tool▁calls▁begin|>", "<|tool▁call▁end|>"];
788 let text = "<|tool▁calls▁begin|>payload<|tool▁call▁end|>tail";
789 let bounds = find_special_token_boundaries(text, specials);
790
791 let after_begin = "<|tool▁calls▁begin|>".len();
792 let after_end = text.find("<|tool▁call▁end|>").unwrap() + "<|tool▁call▁end|>".len();
793 assert_eq!(bounds, vec![after_begin, after_end]);
794 for &b in &bounds {
795 assert!(
796 text.is_char_boundary(b),
797 "boundary {b} is not a char boundary"
798 );
799 let _ = &text[..b]; }
801 }
802
803 #[test]
804 fn populate_and_encode_matches_uncached_and_seeds_cache() {
805 let tok = load_tokenizer();
808 let cache = test_cache(8 * 1024 * 1024);
809 let input = "<s>system\nYou are helpful.</s><s>user\nHello there, friend.</s>";
810
811 let got = cache.populate_and_encode(input, tok.as_ref()).unwrap();
812 let plain = tok.encode(input).unwrap();
813 assert_eq!(
814 got,
815 plain.token_ids(),
816 "fused miss encode must equal uncached encode"
817 );
818
819 assert!(
821 !cache.is_empty(),
822 "miss path must populate boundary entries"
823 );
824 let (_t, offset, _d) = cache
825 .longest_prefix_match(input)
826 .expect("hit after populate");
827 assert!(offset > 0, "follow-up lookup should hit a cached boundary");
828 }
829
830 #[test]
831 fn populate_and_encode_handles_inputs_without_special_tokens() {
832 let tok = load_tokenizer();
835 let cache = test_cache(8 * 1024 * 1024);
836 let input = "plain text with no special tokens at all";
837
838 let got = cache.populate_and_encode(input, tok.as_ref()).unwrap();
839 let plain = tok.encode(input).unwrap();
840 assert_eq!(got, plain.token_ids());
841 assert!(cache.is_empty(), "nothing cacheable without boundaries");
842 }
843
844 #[test]
845 fn populate_and_encode_handles_trailing_special_token() {
846 let tok = load_tokenizer();
850 let cache = test_cache(8 * 1024 * 1024);
851 let input = "<s>system\nDone.</s>";
852
853 let got = cache.populate_and_encode(input, tok.as_ref()).unwrap();
854 let plain = tok.encode(input).unwrap();
855 assert_eq!(
856 got,
857 plain.token_ids(),
858 "tail-segment assembly must be exact"
859 );
860 }
861}