use std::{
hash::BuildHasherDefault,
mem::size_of_val,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
};
use aho_corasick::AhoCorasick;
use moka::sync::Cache;
use rustc_hash::FxHasher;
use crate::{TokenIdType, traits::Encoder};
type Blake3Hash = [u8; 32];
type PrefixHasher = BuildHasherDefault<FxHasher>;
type PrefixCache = Cache<Blake3Hash, Arc<[TokenIdType]>, PrefixHasher>;
fn boundaries_with(text: &str, matcher: &AhoCorasick) -> Vec<usize> {
let mut boundaries: Vec<usize> = matcher
.find_overlapping_iter(text)
.map(|m| m.end())
.filter(|&end| end < text.len())
.collect();
boundaries.sort_unstable();
boundaries.dedup();
boundaries
}
#[cfg(test)]
fn find_special_token_boundaries(text: &str, special_tokens: &[&str]) -> Vec<usize> {
if special_tokens.is_empty() {
return Vec::new();
}
let matcher = AhoCorasick::new(special_tokens)
.expect("special tokens form a valid Aho-Corasick automaton");
boundaries_with(text, &matcher)
}
pub type CacheEventFn = Arc<dyn Fn() + Send + Sync>;
pub struct L1Cache {
cache: PrefixCache,
matcher: Option<AhoCorasick>,
hits: AtomicU64,
misses: AtomicU64,
on_hit: Option<CacheEventFn>,
on_miss: Option<CacheEventFn>,
}
impl L1Cache {
pub fn new(max_memory: usize, special_tokens: Vec<String>) -> Self {
let cache = Cache::builder()
.max_capacity(max_memory as u64)
.weigher(|_k: &Blake3Hash, tokens: &Arc<[TokenIdType]>| -> u32 {
size_of_val(tokens.as_ref()).min(u32::MAX as usize) as u32
})
.build_with_hasher(PrefixHasher::default());
let matcher = (!special_tokens.is_empty()).then(|| {
AhoCorasick::new(&special_tokens)
.expect("special tokens form a valid Aho-Corasick automaton")
});
Self {
cache,
matcher,
hits: AtomicU64::new(0),
misses: AtomicU64::new(0),
on_hit: None,
on_miss: None,
}
}
pub fn set_observer(&mut self, on_hit: CacheEventFn, on_miss: CacheEventFn) {
self.on_hit = Some(on_hit);
self.on_miss = Some(on_miss);
}
fn boundaries(&self, text: &str) -> Vec<usize> {
match &self.matcher {
Some(matcher) => boundaries_with(text, matcher),
None => Vec::new(),
}
}
pub fn longest_prefix_match(&self, input: &str) -> Option<(Arc<[TokenIdType]>, usize, usize)> {
let boundaries = self.boundaries(input);
if boundaries.is_empty() {
self.misses.fetch_add(1, Ordering::Relaxed);
if let Some(cb) = &self.on_miss {
cb();
}
return None;
}
let deepest_boundary = *boundaries.last().expect("boundaries is non-empty here");
let mut hasher = blake3::Hasher::new();
let mut prefix_hashes = Vec::with_capacity(boundaries.len());
let mut last_pos = 0;
let bytes = input.as_bytes();
for &boundary_pos in &boundaries {
hasher.update(&bytes[last_pos..boundary_pos]);
prefix_hashes.push((boundary_pos, *hasher.finalize().as_bytes()));
last_pos = boundary_pos;
}
for (boundary_pos, hash_bytes) in prefix_hashes.into_iter().rev() {
if let Some(tokens) = self.cache.get(&hash_bytes) {
self.hits.fetch_add(1, Ordering::Relaxed);
if let Some(cb) = &self.on_hit {
cb();
}
return Some((tokens, boundary_pos, deepest_boundary));
}
}
self.misses.fetch_add(1, Ordering::Relaxed);
if let Some(cb) = &self.on_miss {
cb();
}
None
}
pub fn insert_at_boundaries<E: Encoder + ?Sized>(
&self,
input: &str,
tokenizer: &E,
) -> anyhow::Result<()> {
let boundaries = self.boundaries(input);
if boundaries.is_empty() {
return Ok(());
}
self.populate_boundaries(input, &boundaries, tokenizer)?;
Ok(())
}
pub fn populate_and_encode<E: Encoder + ?Sized>(
&self,
input: &str,
tokenizer: &E,
) -> anyhow::Result<Vec<TokenIdType>> {
let boundaries = self.boundaries(input);
if boundaries.is_empty() {
return Ok(tokenizer.encode(input)?.token_ids().to_vec());
}
let mut running = self.populate_boundaries(input, &boundaries, tokenizer)?;
let tail_start = *boundaries.last().expect("boundaries is non-empty here");
let tail = tokenizer.encode(&input[tail_start..])?;
running.extend_from_slice(tail.token_ids());
Ok(running)
}
fn populate_boundaries<E: Encoder + ?Sized>(
&self,
input: &str,
boundaries: &[usize],
tokenizer: &E,
) -> anyhow::Result<Vec<TokenIdType>> {
let mut hasher = blake3::Hasher::new();
let mut running_tokens: Vec<TokenIdType> = Vec::new();
let mut last_pos = 0;
let bytes = input.as_bytes();
for &boundary_pos in boundaries {
hasher.update(&bytes[last_pos..boundary_pos]);
let hash_bytes: Blake3Hash = *hasher.finalize().as_bytes();
let seg = tokenizer.encode(&input[last_pos..boundary_pos])?;
running_tokens.extend_from_slice(seg.token_ids());
let prefix_tokens: Arc<[TokenIdType]> = running_tokens.as_slice().into();
self.cache.insert(hash_bytes, prefix_tokens);
last_pos = boundary_pos;
}
Ok(running_tokens)
}
pub fn extend_after_match<E: Encoder + ?Sized>(
&self,
input: &str,
prefix_tokens: Arc<[TokenIdType]>,
prefix_len: usize,
deepest_boundary: usize,
tokenizer: &E,
) -> anyhow::Result<Vec<TokenIdType>> {
let deepest = (deepest_boundary > prefix_len).then_some(deepest_boundary);
let Some(deepest) = deepest else {
let suffix_enc = tokenizer.encode(&input[prefix_len..])?;
let mut merged = Vec::with_capacity(prefix_tokens.len() + suffix_enc.token_ids().len());
merged.extend_from_slice(&prefix_tokens);
merged.extend_from_slice(suffix_enc.token_ids());
return Ok(merged);
};
let seg_a = tokenizer.encode(&input[prefix_len..deepest])?;
let seg_b = tokenizer.encode(&input[deepest..])?;
let mut cumulative = Vec::with_capacity(
prefix_tokens.len() + seg_a.token_ids().len() + seg_b.token_ids().len(),
);
cumulative.extend_from_slice(&prefix_tokens);
cumulative.extend_from_slice(seg_a.token_ids());
let mut hasher = blake3::Hasher::new();
hasher.update(&input.as_bytes()[..deepest]);
let hash_bytes: Blake3Hash = *hasher.finalize().as_bytes();
let tokens: Arc<[TokenIdType]> = cumulative.as_slice().into();
self.cache.insert(hash_bytes, tokens);
cumulative.extend_from_slice(seg_b.token_ids());
Ok(cumulative)
}
pub fn len(&self) -> usize {
self.cache.run_pending_tasks();
self.cache.entry_count() as usize
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn stats(&self) -> L1CacheStats {
self.cache.run_pending_tasks();
let hits = self.hits.load(Ordering::Relaxed);
let misses = self.misses.load(Ordering::Relaxed);
let total_requests = hits + misses;
L1CacheStats {
hits,
misses,
entries: self.cache.entry_count() as usize,
memory_bytes: self.cache.weighted_size() as usize,
hit_rate: if total_requests > 0 {
hits as f64 / total_requests as f64
} else {
0.0
},
}
}
pub fn clear(&self) {
self.cache.invalidate_all();
self.cache.run_pending_tasks();
self.hits.store(0, Ordering::Relaxed);
self.misses.store(0, Ordering::Relaxed);
}
}
#[derive(Debug, Clone)]
pub struct L1CacheStats {
pub hits: u64,
pub misses: u64,
pub entries: usize,
pub memory_bytes: usize,
pub hit_rate: f64,
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use super::*;
use crate::{HuggingFaceTokenizer, traits::Tokenizer};
const TINYLLAMA_PATH: &str = concat!(
env!("CARGO_MANIFEST_DIR"),
"/../llm/tests/data/sample-models/TinyLlama_v1.1/tokenizer.json"
);
const SPECIALS: &[&str] = &["<s>", "</s>"];
fn load_tokenizer() -> Arc<dyn Tokenizer> {
Arc::new(HuggingFaceTokenizer::from_file(TINYLLAMA_PATH).expect("load TinyLlama"))
}
fn test_cache(max_memory: usize) -> L1Cache {
L1Cache::new(
max_memory,
SPECIALS.iter().map(|s| (*s).to_string()).collect(),
)
}
#[test]
fn boundaries_are_after_each_special_token_occurrence() {
let input = "<s>system\nHi</s><s>user\nHello</s>";
let bounds = find_special_token_boundaries(input, SPECIALS);
assert_eq!(bounds.len(), 3);
for w in bounds.windows(2) {
assert!(w[0] < w[1], "boundaries must be strictly increasing");
}
assert!(bounds.iter().all(|&b| b < input.len()));
}
#[test]
fn no_special_tokens_yields_no_boundaries() {
assert!(find_special_token_boundaries("plain text", &[]).is_empty());
}
#[test]
fn insert_then_lookup_finds_shared_prefix() {
let cache = test_cache(1024 * 1024);
let tokenizer = load_tokenizer();
let warm = "<s>system\nYou are helpful.</s><s>user\nHi</s>";
cache
.insert_at_boundaries(warm, tokenizer.as_ref())
.unwrap();
assert!(!cache.is_empty());
let target = "<s>system\nYou are helpful.</s><s>user\nDifferent question</s>";
let (tokens, offset, _deepest) = cache
.longest_prefix_match(target)
.expect("shared prefix should match");
assert!(offset > 0);
assert!(!tokens.is_empty());
}
#[test]
fn miss_increments_misses_counter() {
let cache = test_cache(1024 * 1024);
assert!(
cache
.longest_prefix_match("plain text no specials")
.is_none()
);
assert_eq!(cache.stats().misses, 1);
}
#[test]
fn hit_increments_hits_counter() {
let cache = test_cache(1024 * 1024);
let tokenizer = load_tokenizer();
let warm = "<s>system\nA.</s><s>user\nB</s>";
cache
.insert_at_boundaries(warm, tokenizer.as_ref())
.unwrap();
let _ = cache.longest_prefix_match(warm);
assert!(cache.stats().hits >= 1);
}
#[test]
fn merge_invariant_holds_against_uncached_encode() {
let cache = test_cache(1024 * 1024);
let tokenizer = load_tokenizer();
let template = "<s>system\nYou are helpful.</s><s>user\n";
let warm = format!("{template}First.</s>");
cache
.insert_at_boundaries(&warm, tokenizer.as_ref())
.unwrap();
let target = format!("{template}A completely different second question.</s>");
let (prefix_tokens, prefix_len, _deepest) = cache
.longest_prefix_match(&target)
.expect("should find prefix");
let suffix = &target[prefix_len..];
let suffix_enc = tokenizer.encode(suffix).unwrap();
let mut merged = prefix_tokens.to_vec();
merged.extend_from_slice(suffix_enc.token_ids());
let plain = tokenizer.encode(&target).unwrap();
assert_eq!(
merged,
plain.token_ids(),
"merged tokens must equal plain encode"
);
}
#[test]
fn eviction_respects_memory_budget() {
let cache = test_cache(4 * 1024);
let tokenizer = load_tokenizer();
for i in 0..50 {
let input =
format!("<s>system\nPersona {i} chatty.</s><s>user\nTurn {i} content here.</s>");
cache
.insert_at_boundaries(&input, tokenizer.as_ref())
.unwrap();
}
let stats = cache.stats();
assert!(
stats.memory_bytes <= 4 * 1024,
"memory_bytes={} exceeds budget",
stats.memory_bytes
);
}
#[test]
fn concurrent_inserts_and_lookups_do_not_corrupt() {
use std::thread;
let cache = Arc::new(test_cache(1024 * 1024));
let tokenizer = load_tokenizer();
let mut handles = vec![];
for i in 0..10 {
let cache_c = cache.clone();
let tok = tokenizer.clone();
handles.push(thread::spawn(move || {
let input = format!("<s>system\nThread {i}.</s><s>user\nThread {i} body.</s>");
cache_c.insert_at_boundaries(&input, tok.as_ref()).unwrap();
let r = cache_c.longest_prefix_match(&input);
assert!(r.is_some(), "thread {i} expected match after insert");
}));
}
for h in handles {
h.join().unwrap();
}
assert!(cache.stats().memory_bytes > 0);
assert!(cache.stats().hits >= 10);
}
fn growing_chat_turns(n: usize) -> Vec<String> {
let mut convo = String::from("<s>system\nYou are a helpful assistant.</s>");
let mut turns = Vec::with_capacity(n);
for i in 0..n {
convo.push_str(&format!(
"<s>user\nQuestion {i} please answer it.</s><s>assistant\nDetailed answer {i} follows here.</s>"
));
turns.push(format!("{convo}<s>user\nFollow-up {i}"));
}
turns
}
#[test]
fn extend_on_hit_advances_match_depth_each_turn() {
let tok = load_tokenizer();
let turns = growing_chat_turns(5);
let off = test_cache(8 * 1024 * 1024);
off.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
let pinned = off.longest_prefix_match(&turns[1]).expect("hit").1;
for t in &turns[1..] {
let (_toks, offset, _deepest) = off.longest_prefix_match(t).expect("hit");
assert_eq!(
offset, pinned,
"extend-off offset must stay pinned at turn-1 depth"
);
}
let on = test_cache(8 * 1024 * 1024);
on.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
let mut prev = 0usize;
for (i, t) in turns.iter().enumerate().skip(1) {
let (prefix_tokens, offset, deepest) = on.longest_prefix_match(t).expect("hit");
assert!(
offset > prev,
"turn {i}: extend-on offset {offset} must exceed previous {prev}"
);
prev = offset;
let merged = on
.extend_after_match(t, prefix_tokens, offset, deepest, tok.as_ref())
.unwrap();
let plain = tok.encode(t).unwrap();
assert_eq!(
merged,
plain.token_ids(),
"turn {i}: extend merge must equal plain encode"
);
}
assert!(
prev > pinned,
"extend-on frontier ({prev}) must reach deeper than pinned extend-off depth ({pinned})"
);
}
#[test]
fn extend_on_hit_respects_budget_and_stays_correct() {
let tok = load_tokenizer();
let cache = test_cache(4 * 1024);
let turns = growing_chat_turns(20);
cache.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
for t in &turns[1..] {
let merged = match cache.longest_prefix_match(t) {
Some((prefix_tokens, offset, deepest)) => cache
.extend_after_match(t, prefix_tokens, offset, deepest, tok.as_ref())
.unwrap(),
None => {
let enc = tok.encode(t).unwrap();
cache.insert_at_boundaries(t, tok.as_ref()).unwrap();
enc.token_ids().to_vec()
}
};
let plain = tok.encode(t).unwrap();
assert_eq!(
merged,
plain.token_ids(),
"encode must stay correct under eviction pressure"
);
assert!(
cache.stats().memory_bytes <= 4 * 1024,
"memory_bytes={} exceeds budget",
cache.stats().memory_bytes
);
}
}
#[test]
fn concurrent_extend_on_hit_does_not_corrupt() {
use std::thread;
let tok = load_tokenizer();
let cache = Arc::new(test_cache(8 * 1024 * 1024));
let turns = growing_chat_turns(8);
cache.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
let mut handles = vec![];
for _ in 0..8 {
let cache_c = cache.clone();
let tok_c = tok.clone();
let turns_c = turns.clone();
handles.push(thread::spawn(move || {
for t in &turns_c[1..] {
if let Some((prefix_tokens, offset, deepest)) = cache_c.longest_prefix_match(t)
{
let merged = cache_c
.extend_after_match(t, prefix_tokens, offset, deepest, tok_c.as_ref())
.unwrap();
let plain = tok_c.encode(t).unwrap();
assert_eq!(
merged,
plain.token_ids(),
"concurrent extend must stay correct"
);
}
}
}));
}
for h in handles {
h.join().unwrap();
}
assert!(cache.stats().memory_bytes > 0);
}
#[test]
fn extend_after_match_persists_correct_deepest_entry() {
let tok = load_tokenizer();
let turns = growing_chat_turns(3);
let cache = test_cache(8 * 1024 * 1024);
cache.insert_at_boundaries(&turns[0], tok.as_ref()).unwrap();
let (prefix_tokens, prefix_len, deepest_boundary) = cache
.longest_prefix_match(&turns[1])
.expect("partial hit on turns[1]");
let entries_before = cache.stats().entries;
let _merged = cache
.extend_after_match(
&turns[1],
prefix_tokens,
prefix_len,
deepest_boundary,
tok.as_ref(),
)
.unwrap();
assert_eq!(
cache.stats().entries,
entries_before + 1,
"extend must persist exactly one (deepest) entry"
);
let deepest = find_special_token_boundaries(&turns[1], SPECIALS)
.into_iter()
.rev()
.find(|&b| b > prefix_len)
.expect("a deeper boundary must exist in the appended turn");
assert_eq!(
deepest_boundary, deepest,
"longest_prefix_match must return the deepest boundary used by extend"
);
let (saved_tokens, saved_offset, _deepest) = cache
.longest_prefix_match(&turns[1])
.expect("hit after extend");
assert_eq!(
saved_offset, deepest,
"lookup must now hit at the just-saved deepest boundary"
);
let expected = tok.encode(&turns[1][..deepest]).unwrap();
assert_eq!(
&*saved_tokens,
expected.token_ids(),
"persisted entry tokens must equal the uncached encode of the cached prefix"
);
}
#[test]
fn boundaries_detected_for_multibyte_deepseek_tool_tokens() {
let specials = &["<|tool▁calls▁begin|>", "<|tool▁call▁end|>"];
let text = "<|tool▁calls▁begin|>payload<|tool▁call▁end|>tail";
let bounds = find_special_token_boundaries(text, specials);
let after_begin = "<|tool▁calls▁begin|>".len();
let after_end = text.find("<|tool▁call▁end|>").unwrap() + "<|tool▁call▁end|>".len();
assert_eq!(bounds, vec![after_begin, after_end]);
for &b in &bounds {
assert!(
text.is_char_boundary(b),
"boundary {b} is not a char boundary"
);
let _ = &text[..b]; }
}
#[test]
fn populate_and_encode_matches_uncached_and_seeds_cache() {
let tok = load_tokenizer();
let cache = test_cache(8 * 1024 * 1024);
let input = "<s>system\nYou are helpful.</s><s>user\nHello there, friend.</s>";
let got = cache.populate_and_encode(input, tok.as_ref()).unwrap();
let plain = tok.encode(input).unwrap();
assert_eq!(
got,
plain.token_ids(),
"fused miss encode must equal uncached encode"
);
assert!(
!cache.is_empty(),
"miss path must populate boundary entries"
);
let (_t, offset, _d) = cache
.longest_prefix_match(input)
.expect("hit after populate");
assert!(offset > 0, "follow-up lookup should hit a cached boundary");
}
#[test]
fn populate_and_encode_handles_inputs_without_special_tokens() {
let tok = load_tokenizer();
let cache = test_cache(8 * 1024 * 1024);
let input = "plain text with no special tokens at all";
let got = cache.populate_and_encode(input, tok.as_ref()).unwrap();
let plain = tok.encode(input).unwrap();
assert_eq!(got, plain.token_ids());
assert!(cache.is_empty(), "nothing cacheable without boundaries");
}
#[test]
fn populate_and_encode_handles_trailing_special_token() {
let tok = load_tokenizer();
let cache = test_cache(8 * 1024 * 1024);
let input = "<s>system\nDone.</s>";
let got = cache.populate_and_encode(input, tok.as_ref()).unwrap();
let plain = tok.encode(input).unwrap();
assert_eq!(
got,
plain.token_ids(),
"tail-segment assembly must be exact"
);
}
}