use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
use super::WeightBytes;
#[derive(Clone)]
pub struct MapId {
map: std::sync::Weak<memmap2::Mmap>,
id: usize,
offset: usize,
}
impl MapId {
pub(super) fn of(mmap: &Arc<memmap2::Mmap>, offset: usize) -> Self {
MapId {
map: Arc::downgrade(mmap),
id: Arc::as_ptr(mmap) as usize,
offset,
}
}
fn matches(&self, other: &MapId) -> bool {
self.id == other.id
&& self.offset == other.offset
&& self
.map
.upgrade()
.is_some_and(|m| Arc::as_ptr(&m) as usize == other.id)
}
fn key(&self, format: Format, rows: usize, cols: usize) -> RepackKey {
(format, self.id, self.offset, rows, cols)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum Format {
Q4Kx8,
Q5Kx8,
Q6Kx8,
Q8_0x4,
Q4_0x4,
}
type RepackKey = (Format, usize, usize, usize, usize);
struct Entry {
id: MapId,
packed: Arc<[u8]>,
last_used: u64,
}
#[derive(Default)]
struct Cache {
entries: HashMap<RepackKey, Entry>,
resident_bytes: usize,
clock: u64,
}
impl Cache {
fn evict(&mut self, key: &RepackKey) {
if let Some(entry) = self.entries.remove(key) {
self.resident_bytes = self.resident_bytes.saturating_sub(entry.packed.len());
}
}
fn lru(&self) -> Option<RepackKey> {
self.entries
.iter()
.min_by_key(|(_, e)| e.last_used)
.map(|(k, _)| *k)
}
fn take_hit(&mut self, key: &RepackKey, id: &MapId) -> Option<Arc<[u8]>> {
match self.entries.get_mut(key) {
Some(entry) if entry.id.matches(id) => {
self.clock += 1;
entry.last_used = self.clock;
Some(Arc::clone(&entry.packed))
}
Some(_) => {
self.evict(key);
None
}
None => None,
}
}
fn insert_within_budget(&mut self, key: RepackKey, id: MapId, packed: Arc<[u8]>) {
let budget = budget_bytes();
let size = packed.len();
if size > budget {
return;
}
while self.resident_bytes + size > budget {
let Some(victim) = self.lru() else { break };
self.evict(&victim);
}
if self.resident_bytes + size > budget {
return;
}
self.clock += 1;
self.resident_bytes += size;
self.entries.insert(
key,
Entry {
id,
packed,
last_used: self.clock,
},
);
}
}
fn cache() -> &'static Mutex<Cache> {
static CACHE: OnceLock<Mutex<Cache>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(Cache::default()))
}
fn lock() -> MutexGuard<'static, Cache> {
cache().lock().unwrap_or_else(|e| e.into_inner())
}
fn budget_bytes() -> usize {
#[cfg(test)]
{
if let Some(bytes) = tests::budget_override() {
return bytes;
}
}
static BYTES: OnceLock<usize> = OnceLock::new();
*BYTES.get_or_init(|| {
if let Some(explicit) = std::env::var("FERROX_REPACK_CACHE_BYTES")
.ok()
.and_then(|v| v.trim().parse::<u64>().ok())
{
return usize::try_from(explicit).unwrap_or(usize::MAX);
}
let derived = crate::host_memory::derived_copy_budget(
crate::host_memory::available_bytes(),
crate::host_memory::FIT_HEADROOM_BYTES,
crate::expert_store::committed_expert_bytes(),
);
usize::try_from(derived).unwrap_or(usize::MAX)
})
}
fn get_or_repack(
format: Format,
id: Option<MapId>,
rows: usize,
cols: usize,
repack: impl FnOnce() -> Vec<u8>,
) -> Arc<[u8]> {
let Some(id) = id else {
return Arc::from(repack().into_boxed_slice());
};
let key = id.key(format, rows, cols);
if let Some(hit) = lock().take_hit(&key, &id) {
return hit;
}
let arc: Arc<[u8]> = Arc::from(repack().into_boxed_slice());
let mut cache = lock();
match cache.take_hit(&key, &id) {
Some(hit) => hit,
None => {
cache.insert_within_budget(key, id, Arc::clone(&arc));
arc
}
}
}
#[cfg(test)]
fn is_cached(format: Format, data: &WeightBytes, rows: usize, cols: usize) -> bool {
let Some(id) = data.map_id() else {
return false;
};
lock()
.entries
.get(&id.key(format, rows, cols))
.is_some_and(|e| e.id.matches(&id))
}
#[cfg(test)]
fn resident_bytes() -> usize {
lock().resident_bytes
}
#[cfg(test)]
fn clear() {
let mut cache = lock();
cache.entries.clear();
cache.resident_bytes = 0;
}
pub(super) fn get_or_repack_q4k(data: &WeightBytes, rows: usize, cols: usize) -> Arc<[u8]> {
get_or_repack(Format::Q4Kx8, data.map_id(), rows, cols, || {
ferrox_quant::pack_q4_k_matrix_x8(
data.as_slice(),
rows,
cols,
ferrox_quant::q4_kx8_interleave(),
)
})
}
pub(super) fn get_or_repack_q5k(data: &WeightBytes, rows: usize, cols: usize) -> Arc<[u8]> {
get_or_repack(Format::Q5Kx8, data.map_id(), rows, cols, || {
ferrox_quant::pack_q5_k_matrix_x8(
data.as_slice(),
rows,
cols,
ferrox_quant::q5_kx8_interleave(),
)
})
}
pub(super) fn get_or_repack_q6k(data: &WeightBytes, rows: usize, cols: usize) -> Arc<[u8]> {
get_or_repack(Format::Q6Kx8, data.map_id(), rows, cols, || {
ferrox_quant::pack_q6_k_matrix_x8(
data.as_slice(),
rows,
cols,
ferrox_quant::q6_kx8_interleave(),
)
})
}
pub(super) fn get_or_repack_q8x4(data: &WeightBytes, rows: usize, cols: usize) -> Arc<[u8]> {
get_or_repack(Format::Q8_0x4, data.map_id(), rows, cols, || {
ferrox_quant::pack_q8_0_matrix_x4(
data.as_slice(),
rows,
cols,
ferrox_quant::q8_0x4_interleave(),
)
})
}
pub(super) fn get_or_repack_q4_0x4(data: &WeightBytes, rows: usize, cols: usize) -> Arc<[u8]> {
get_or_repack(Format::Q4_0x4, data.map_id(), rows, cols, || {
ferrox_quant::pack_q4_0_matrix_x4(
data.as_slice(),
rows,
cols,
ferrox_quant::q4_0x4_interleave(),
)
})
}
#[cfg(test)]
pub(super) fn q8x4_is_cached(data: &WeightBytes, rows: usize, cols: usize) -> bool {
is_cached(Format::Q8_0x4, data, rows, cols)
}
#[cfg(test)]
mod tests {
use super::super::tests::{f16_le, ForceIntDot};
use super::super::{QuantKind, WeightMatrix};
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
static BUDGET_OVERRIDE: AtomicUsize = AtomicUsize::new(usize::MAX);
pub(super) fn budget_override() -> Option<usize> {
match BUDGET_OVERRIDE.load(Ordering::Acquire) {
usize::MAX => None,
bytes => Some(bytes),
}
}
pub(super) struct ForceBudget {
_lock: std::sync::MutexGuard<'static, ()>,
}
impl ForceBudget {
fn new(bytes: usize) -> Self {
static LOCK: Mutex<()> = Mutex::new(());
let lock = LOCK.lock().unwrap_or_else(|e| e.into_inner());
BUDGET_OVERRIDE.store(bytes, Ordering::Release);
clear();
ForceBudget { _lock: lock }
}
fn generous() -> Self {
Self::new(1 << 20)
}
}
impl Drop for ForceBudget {
fn drop(&mut self) {
clear();
BUDGET_OVERRIDE.store(usize::MAX, Ordering::Release);
}
}
fn mapped(tag: &str, bytes: &[u8]) -> (Arc<memmap2::Mmap>, WeightBytes) {
let path = std::env::temp_dir().join(format!(
"ferrox_repack_{tag}_{}_{:?}.bin",
std::process::id(),
std::thread::current().id()
));
std::fs::write(&path, bytes).expect("write fixture");
let file = std::fs::File::open(&path).expect("open fixture");
let mmap = Arc::new(unsafe { memmap2::Mmap::map(&file).expect("map fixture") });
let _ = std::fs::remove_file(&path);
let view = WeightBytes::Mapped {
mmap: Arc::clone(&mmap),
range: 0..bytes.len(),
};
(mmap, view)
}
fn q8_0_matrix_bytes(rows: usize, cols: usize, seed: u32) -> Vec<u8> {
let mut state = seed | 1;
let mut next = move || {
state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
(state >> 24) as u8
};
let mut data = Vec::with_capacity(rows * (cols / 32) * 34);
for _ in 0..rows * (cols / 32) {
data.extend_from_slice(&f16_le(0.02 + f32::from(next()) * 0.0004));
for _ in 0..32 {
data.push(next());
}
}
data
}
#[test]
fn map_id_stops_matching_once_its_mapping_is_dropped() {
let (mmap, view) = mapped("live", &q8_0_matrix_bytes(4, 32, 7));
let id = view.map_id().expect("Mapped bytes must have an identity");
let held = id.clone();
assert!(
held.matches(&id),
"a live mapping must match its own identity"
);
drop(view);
drop(mmap);
assert!(
!held.matches(&id),
"an identity whose mapping is dead must not match, or the \
cache will trust an address the kernel has already reissued"
);
}
#[test]
fn stale_repack_entry_is_replaced_not_served() {
let _budget = ForceBudget::generous();
let (rows, cols) = (8usize, 64usize);
let bytes = q8_0_matrix_bytes(rows, cols, 11);
let (_mmap, view) = mapped("stale", &bytes);
let id = view.map_id().expect("Mapped bytes must have an identity");
let poison = vec![0xABu8; 16];
{
let mut cache = lock();
cache.insert_within_budget(
id.key(Format::Q8_0x4, rows, cols),
MapId {
map: std::sync::Weak::new(),
id: id.id,
offset: id.offset,
},
Arc::from(poison.clone().into_boxed_slice()),
);
}
let got = get_or_repack_q8x4(&view, rows, cols);
let want = ferrox_quant::pack_q8_0_matrix_x4(
view.as_slice(),
rows,
cols,
ferrox_quant::q8_0x4_interleave(),
);
assert_ne!(&got[..], &poison[..], "served a dead mapping's bytes");
assert_eq!(&got[..], &want[..], "stale entry was not repacked");
let cache = lock();
let entry = cache
.entries
.get(&id.key(Format::Q8_0x4, rows, cols))
.expect("the live packing should now be cached");
assert!(
entry.id.matches(&id),
"the replacement entry must carry the LIVE identity"
);
}
#[test]
fn repack_key_separates_two_widths_at_one_address() {
let _budget = ForceBudget::generous();
let rows = 8usize;
let narrow = q8_0_matrix_bytes(rows, 32, 3);
let wide = q8_0_matrix_bytes(rows, 64, 5);
let (_mmap, view) = mapped("widths", &narrow);
let id = view.map_id().expect("Mapped bytes must have an identity");
let il = ferrox_quant::q8_0x4_interleave();
let a = get_or_repack(Format::Q8_0x4, Some(id.clone()), rows, 32, || {
ferrox_quant::pack_q8_0_matrix_x4(&narrow, rows, 32, il)
});
let b = get_or_repack(Format::Q8_0x4, Some(id.clone()), rows, 64, || {
ferrox_quant::pack_q8_0_matrix_x4(&wide, rows, 64, il)
});
assert_eq!(
&a[..],
&ferrox_quant::pack_q8_0_matrix_x4(&narrow, rows, 32, il)[..]
);
assert_eq!(
&b[..],
&ferrox_quant::pack_q8_0_matrix_x4(&wide, rows, 64, il)[..],
"the wider matrix was served the narrower one's packing"
);
assert!(b.len() > a.len(), "widths must not share a cache entry");
}
#[test]
fn map_id_is_none_for_owned_and_shared_bytes() {
let owned = WeightBytes::Owned(q8_0_matrix_bytes(4, 32, 9));
assert!(owned.map_id().is_none(), "an owned Vec's address is reused");
let buf = Arc::new(q8_0_matrix_bytes(4, 32, 13));
let leased = WeightBytes::Shared {
buf,
range: 0..34 * 4,
};
assert!(
leased.map_id().is_none(),
"an expert lease keeps its address across a content swap"
);
}
#[test]
fn a_mapped_matrix_is_packed_once_and_an_owned_one_never_cached() {
let _budget = ForceBudget::generous();
let (rows, cols) = (8usize, 64usize);
let bytes = q8_0_matrix_bytes(rows, cols, 17);
let (_mmap, view) = mapped("once", &bytes);
assert!(!q8x4_is_cached(&view, rows, cols));
let first = get_or_repack_q8x4(&view, rows, cols);
assert!(q8x4_is_cached(&view, rows, cols));
let second = get_or_repack_q8x4(&view, rows, cols);
assert!(
Arc::ptr_eq(&first, &second),
"a second lookup of a live mapping must be a cache hit"
);
let owned = WeightBytes::Owned(bytes);
let a = get_or_repack_q8x4(&owned, rows, cols);
let b = get_or_repack_q8x4(&owned, rows, cols);
assert!(!q8x4_is_cached(&owned, rows, cols));
assert!(
!Arc::ptr_eq(&a, &b),
"owned bytes have no identity and must repack every time"
);
}
#[test]
fn apply_cpu_q8_caches_the_packing_of_a_mapped_matrix() {
let _force = ForceIntDot::new(true);
let _budget = ForceBudget::generous();
if !super::super::cpu_int_dot_for(super::super::IntDotShape::Matvec) {
return;
}
let (rows, cols) = (8usize, 64usize);
let bytes = q8_0_matrix_bytes(rows, cols, 23);
let (_mmap, view) = mapped("apply_q8", &bytes);
let m = WeightMatrix::Quantized {
data: view,
rows,
cols,
kind: QuantKind::Q8_0,
};
let WeightMatrix::Quantized { data, .. } = &m else {
unreachable!()
};
assert!(!q8x4_is_cached(data, rows, cols), "fresh mapping");
let x: Vec<f32> = (0..cols).map(|i| (i as f32) * 0.01 - 0.3).collect();
let act = ferrox_quant::quantize_activations_q8(&x);
let out = m
.apply_cpu_q8(&act)
.expect("Q8_0 with int-dot on takes the interleaved path");
assert_eq!(out.len(), rows);
assert!(
q8x4_is_cached(data, rows, cols),
"apply_cpu_q8 repacked a mapped matrix without caching it: \
that is a full copy of the matrix per token (#128)"
);
}
fn il() -> usize {
ferrox_quant::q8_0x4_interleave()
}
#[test]
fn the_cache_never_exceeds_its_budget() {
let (rows, cols) = (8usize, 64usize);
let one_packing =
ferrox_quant::pack_q8_0_matrix_x4(&q8_0_matrix_bytes(rows, cols, 1), rows, cols, il())
.len();
let budget = one_packing * 3;
let _guard = ForceBudget::new(budget);
let mut held = Vec::new();
for seed in 0..10u32 {
let bytes = q8_0_matrix_bytes(rows, cols, seed + 1);
let (mmap, view) = mapped(&format!("budget{seed}"), &bytes);
let got = get_or_repack_q8x4(&view, rows, cols);
assert_eq!(
&got[..],
&ferrox_quant::pack_q8_0_matrix_x4(&bytes, rows, cols, il())[..],
"an evicting cache must still answer correctly"
);
assert!(
resident_bytes() <= budget,
"cache grew past its budget at matrix {seed}: {} > {budget}",
resident_bytes()
);
held.push((mmap, view));
}
assert!(resident_bytes() <= budget);
assert!(
resident_bytes() >= one_packing,
"a budget that fits three packings must be holding some"
);
let (_, last) = held.last().expect("ten matrices were packed");
assert!(
q8x4_is_cached(last, rows, cols),
"the most recent matrix must be cached: a cache that stops \
caching once full leaves every later matrix repacking per \
token, which is the #128 defect for all but the first few"
);
let (_, first) = &held[0];
assert!(
!q8x4_is_cached(first, rows, cols),
"ten packings into a three-packing budget must have evicted \
the least recently used one"
);
}
#[test]
fn the_batch_path_answers_the_same_with_the_cache_full_and_disabled() {
let _force = ForceIntDot::new(true);
if !super::super::cpu_int_dot_for(super::super::IntDotShape::BatchGemm) {
return;
}
let (rows, cols, batch) = (19usize, 64usize, 6usize);
let bytes = q8_0_matrix_bytes(rows, cols, 71);
let (_mmap, view) = mapped("batch_budget", &bytes);
let m = WeightMatrix::Quantized {
data: view,
rows,
cols,
kind: QuantKind::Q8_0,
};
let x: Vec<f32> = (0..batch * cols)
.map(|i| ((i as f32) * 0.013 - 0.7).sin() * 1.4)
.collect();
let cached = {
let _budget = ForceBudget::generous();
let cold = m.apply_batch(&x, batch);
assert!(
resident_bytes() > 0,
"a generous budget retained nothing, so the second call \
below would miss too and the hit path would go untested"
);
let warm = m.apply_batch(&x, batch);
assert_eq!(
cold.iter().map(|v| v.to_bits()).collect::<Vec<_>>(),
warm.iter().map(|v| v.to_bits()).collect::<Vec<_>>(),
"the retained packing served a different answer than the \
one that built it"
);
warm
};
let uncached = {
let _budget = ForceBudget::new(0);
let out = m.apply_batch(&x, batch);
assert_eq!(resident_bytes(), 0, "a zero budget retained something");
out
};
assert_eq!(
cached.iter().map(|v| v.to_bits()).collect::<Vec<_>>(),
uncached.iter().map(|v| v.to_bits()).collect::<Vec<_>>(),
"the repack budget changed the answer, not just where the \
interleaved bytes live"
);
}
#[test]
fn a_zero_budget_retains_nothing_and_matches_the_pre_cache_behaviour() {
let _guard = ForceBudget::new(0);
let (rows, cols) = (8usize, 64usize);
let bytes = q8_0_matrix_bytes(rows, cols, 29);
let (_mmap, view) = mapped("zero_budget", &bytes);
let first = get_or_repack_q8x4(&view, rows, cols);
let second = get_or_repack_q8x4(&view, rows, cols);
assert_eq!(resident_bytes(), 0, "a zero budget retained something");
assert!(!q8x4_is_cached(&view, rows, cols));
assert!(
!Arc::ptr_eq(&first, &second),
"a zero budget must repack every call, as the engine did \
before this cache existed"
);
let want = ferrox_quant::pack_q8_0_matrix_x4(&bytes, rows, cols, il());
assert_eq!(&first[..], &want[..]);
assert_eq!(&second[..], &want[..], "same bytes, different allocation");
}
#[test]
fn every_format_spends_one_budget() {
let (rows, cols) = (8usize, 64usize);
let bytes = q8_0_matrix_bytes(rows, cols, 31);
let q8_len = ferrox_quant::pack_q8_0_matrix_x4(&bytes, rows, cols, il()).len();
let _guard = ForceBudget::new(q8_len);
let (_mmap, view) = mapped("one_budget_q8", &bytes);
let _ = get_or_repack_q8x4(&view, rows, cols);
assert!(q8x4_is_cached(&view, rows, cols), "the budget holds one");
let q4_bytes = vec![7u8; rows * (cols / 32) * 18];
let (_mmap4, view4) = mapped("one_budget_q4", &q4_bytes);
let _ = get_or_repack_q4_0x4(&view4, rows, cols);
assert!(
resident_bytes() <= q8_len,
"the two formats spent one budget, not two"
);
assert!(
!q8x4_is_cached(&view, rows, cols),
"the Q4_0 packing must have displaced the Q8_0 one"
);
}
}