use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::Result;
use crate::exploded::ExplodedStats;
use crate::exploded_arrow::{ExplodePolicy, ExplodedArchive};
use crate::object::GitObjectKind;
const MIN_SEGMENT: u64 = 16 << 20;
const MAX_SEGMENT: u64 = 1 << 30;
const SEGMENTS_PER_BUDGET: u64 = 16;
pub const CACHE_FLUSH_BYTES: u64 = 4 << 20;
thread_local! {
static SUPPRESS_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}
#[must_use = "the suppression lasts only as long as the guard is held"]
pub fn suppress_write_back() -> SuppressWriteBack {
SUPPRESS_DEPTH.with(|d| d.set(d.get().saturating_add(1)));
SuppressWriteBack { _private: () }
}
pub fn write_back_suppressed() -> bool {
SUPPRESS_DEPTH.with(|d| d.get()) > 0
}
pub struct SuppressWriteBack {
_private: (),
}
impl Drop for SuppressWriteBack {
fn drop(&mut self) {
SUPPRESS_DEPTH.with(|d| d.set(d.get().saturating_sub(1)));
}
}
pub const CHEAP_DEPTH: u32 = 2;
pub const MAX_CACHED_BYTES: u64 = 1 << 20;
pub fn rebuild_cost(depth: u32, inflated: u64, external_base: bool) -> u64 {
let links = depth as u64 + 1;
let base = links * inflated.max(1);
if external_base { base.saturating_mul(4) } else { base }
}
pub fn worth_caching(depth: u32, inflated: u64, external_base: bool) -> bool {
inflated <= MAX_CACHED_BYTES && (external_base || depth > CHEAP_DEPTH)
}
pub use crate::arms::ENV_EXPLODE_BUDGET as ENV_BUDGET;
pub const BUDGET_FRACTION_PERCENT: u64 = 10;
pub const BUDGET_CEILING: u64 = 8 << 30;
pub const MIN_BUDGET: u64 = 64 << 20;
pub fn default_budget(dir: &Path) -> u64 {
let total = filesystem_bytes(dir).unwrap_or(0);
if total == 0 {
return BUDGET_CEILING;
}
(total / 100 * BUDGET_FRACTION_PERCENT)
.min(BUDGET_CEILING)
.max(MIN_BUDGET)
}
pub fn default_global_line(dir: &Path) -> u64 {
filesystem_bytes(dir).unwrap_or(0) / 100 * BUDGET_FRACTION_PERCENT
}
fn filesystem_bytes(dir: &Path) -> Option<u64> {
use std::os::unix::ffi::OsStrExt;
let mut c = Vec::with_capacity(dir.as_os_str().len() + 1);
c.extend_from_slice(dir.as_os_str().as_bytes());
c.push(0);
let mut s: libc::statvfs = unsafe { std::mem::zeroed() };
let rc = unsafe { libc::statvfs(c.as_ptr().cast(), &mut s) };
(rc == 0).then(|| s.f_blocks as u64 * s.f_frsize as u64)
}
pub fn budget_from_env(dir: &Path) -> Result<u64> {
match crate::arms::read_env(crate::arms::ENV_EXPLODE_BUDGET) {
Some(v) => parse_budget(&v),
None => Ok(default_budget(dir)),
}
}
pub fn parse_budget(s: &str) -> Result<u64> {
let t = s.trim().to_ascii_lowercase();
if t == "unlimited" || t == "none" || t == "off" {
return Ok(0);
}
let (digits, mult) = match t.as_bytes().last() {
Some(b'k') => (&t[..t.len() - 1], 1u64 << 10),
Some(b'm') => (&t[..t.len() - 1], 1u64 << 20),
Some(b'g') => (&t[..t.len() - 1], 1u64 << 30),
Some(b't') => (&t[..t.len() - 1], 1u64 << 40),
_ => (t.as_str(), 1),
};
let n: u64 = digits
.trim()
.parse()
.map_err(|_| anyhow::anyhow!("unreadable explode budget {s:?}; expected 0, 512M, 8G, …"))?;
n.checked_mul(mult)
.ok_or_else(|| anyhow::anyhow!("explode budget {s:?} overflows"))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct CacheUse {
pub eager_bytes: u64,
pub lazy_bytes: u64,
pub budget: u64,
pub segments: u64,
pub evicted_segments: u64,
pub evicted_bytes: u64,
pub evicted_rows: u64,
pub written_back: u64,
pub direct_reads: u64,
pub suppressed_write_backs: u64,
}
impl CacheUse {
pub fn used(&self) -> u64 {
self.eager_bytes + self.lazy_bytes
}
pub fn band(&self) -> String {
if self.budget == 0 {
format!("{}/unlimited", human(self.used()))
} else {
format!("{}/{}", human(self.used()), human(self.budget))
}
}
pub fn over(&self) -> bool {
self.budget != 0 && self.used() > self.budget
}
}
pub fn human(bytes: u64) -> String {
const UNITS: [(u64, &str); 4] = [(1 << 40, "T"), (1 << 30, "G"), (1 << 20, "M"), (1 << 10, "K")];
for (scale, suffix) in UNITS {
if bytes >= scale {
let v = bytes as f64 / scale as f64;
return if v < 10.0 {
format!("{v:.1}{suffix}")
} else {
format!("{v:.0}{suffix}")
};
}
}
format!("{bytes}B")
}
#[derive(Debug, Clone, Copy, Default)]
struct SegStat {
rows: u64,
cost: u64,
touched: u64,
}
pub struct ExplodedCache {
eager: ExplodedArchive,
lazy: Mutex<Vec<std::sync::Arc<ExplodedArchive>>>,
base: PathBuf,
policy: ExplodePolicy,
budget: AtomicU64,
global_line: AtomicU64,
segment_bytes: AtomicU64,
next_seg: AtomicU64,
written_back: AtomicU64,
direct_reads: AtomicU64,
suppressed_write_backs: AtomicU64,
seg_cost: Mutex<Vec<SegStat>>,
evicted_segments: AtomicU64,
evicted_bytes: AtomicU64,
evicted_rows: AtomicU64,
dirty: AtomicU64,
root: String,
maintaining: std::sync::atomic::AtomicBool,
}
impl ExplodedCache {
pub fn open(base: &Path, policy: ExplodePolicy, budget: u64) -> Result<Self> {
let eager = ExplodedArchive::open_with_policy(base, policy)?;
let mut found: Vec<(u64, PathBuf)> = Vec::new();
if let Some(dir) = base.parent()
&& let Ok(rd) = std::fs::read_dir(dir)
{
let stem = base
.file_name()
.map(|s| s.to_string_lossy().into_owned())
.unwrap_or_default();
let prefix = format!("{stem}.c");
for e in rd.flatten() {
let name = e.file_name().to_string_lossy().into_owned();
if let Some(n) = name.strip_prefix(&prefix)
&& let Ok(n) = n.parse::<u64>()
{
found.push((n, e.path()));
}
}
}
found.sort_unstable_by_key(|(n, _)| *n);
let next = found.last().map(|(n, _)| n + 1).unwrap_or(1);
let seg_stats: Vec<SegStat> = (0..found.len())
.map(|i| SegStat { rows: 0, cost: 0, touched: i as u64 })
.collect();
let mut lazy = Vec::with_capacity(found.len());
for (_, p) in found {
lazy.push(std::sync::Arc::new(ExplodedArchive::open_with_policy(
&p,
ExplodePolicy::Full,
)?));
}
Ok(Self {
eager,
lazy: Mutex::new(lazy),
base: base.to_path_buf(),
policy,
budget: AtomicU64::new(budget),
global_line: AtomicU64::new(default_global_line(base)),
segment_bytes: AtomicU64::new(segment_for(budget)),
next_seg: AtomicU64::new(next),
written_back: AtomicU64::new(0),
direct_reads: AtomicU64::new(0),
suppressed_write_backs: AtomicU64::new(0),
seg_cost: Mutex::new(seg_stats),
evicted_segments: AtomicU64::new(0),
evicted_bytes: AtomicU64::new(0),
evicted_rows: AtomicU64::new(0),
dirty: AtomicU64::new(0),
root: base.parent().unwrap_or(base).display().to_string(),
maintaining: std::sync::atomic::AtomicBool::new(false),
})
}
pub fn registered(self: std::sync::Arc<Self>) -> std::sync::Arc<Self> {
registry()
.lock()
.unwrap_or_else(|e| e.into_inner())
.push(std::sync::Arc::downgrade(&self));
self
}
pub fn root(&self) -> &str {
&self.root
}
pub fn set_budget(&self, bytes: u64) {
self.budget.store(bytes, Ordering::Release);
self.segment_bytes.store(segment_for(bytes), Ordering::Release);
}
pub fn set_global_line(&self, bytes: u64) {
self.global_line.store(bytes, Ordering::Release);
}
fn over_global_share(&self) -> bool {
let line = self.global_line.load(Ordering::Acquire);
if line == 0 {
return false;
}
let all = use_all();
let live = all
.iter()
.filter(|(_, u)| u.eager_bytes + u.lazy_bytes > 0)
.count()
.max(1) as u64;
let sum: u64 = all.iter().map(|(_, u)| u.eager_bytes + u.lazy_bytes).sum();
if sum <= line {
return false;
}
let mine = self.use_now();
mine.eager_bytes + mine.lazy_bytes > line / live
}
pub fn budget(&self) -> u64 {
self.budget.load(Ordering::Acquire)
}
pub fn policy(&self) -> ExplodePolicy {
self.policy
}
pub fn path(&self) -> &Path {
&self.base
}
pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
self.eager.explode(oid, kind, payload)
}
pub fn explode_at(
&self,
oid: &[u8],
kind: GitObjectKind,
mode: Option<u32>,
path: Option<&str>,
payload: &[u8],
) -> Result<()> {
self.eager.explode_at(oid, kind, mode, path, payload)
}
pub fn write_back(
&self,
oid: &[u8],
kind: GitObjectKind,
payload: &[u8],
cost: crate::resolve::ReadCost,
) -> Result<bool> {
if write_back_suppressed() {
self.suppressed_write_backs.fetch_add(1, Ordering::Relaxed);
return Ok(false);
}
if !worth_caching(cost.depth, cost.inflated, cost.external_base) {
self.direct_reads.fetch_add(1, Ordering::Relaxed);
return Ok(false);
}
let seg = {
let mut l = self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
match l.last() {
Some(s) => s.clone(),
None => {
let s = self.new_segment()?;
l.push(s.clone());
s
}
}
};
seg.explode(oid, kind, payload)?;
{
let tick = self.written_back.load(Ordering::Relaxed);
let mut st = self
.seg_cost
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
if st.is_empty() {
st.push(SegStat::default());
}
if let Some(last) = st.last_mut() {
last.rows += 1;
last.cost += rebuild_cost(cost.depth, cost.inflated, cost.external_base);
last.touched = tick;
}
}
self.written_back.fetch_add(1, Ordering::Relaxed);
let n = self
.dirty
.fetch_add(payload.len() as u64, Ordering::AcqRel)
+ payload.len() as u64;
Ok(n >= CACHE_FLUSH_BYTES)
}
pub fn maintain(&self) -> Result<u64> {
let newest = {
let l = self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
l.last().cloned()
};
if let Some(seg) = newest {
seg.flush()?;
self.dirty.store(0, Ordering::Release);
if seg.disk_bytes() >= self.segment_bytes.load(Ordering::Acquire) {
let fresh = self.new_segment()?;
let mut l = self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
l.push(fresh);
if let Ok(mut st) = self.seg_cost.lock() {
st.push(SegStat {
rows: 0,
cost: 0,
touched: self.written_back.load(Ordering::Relaxed),
});
}
}
}
self.evict_to_budget()
}
fn evict_to_budget(&self) -> Result<u64> {
let budget = self.budget.load(Ordering::Acquire);
if budget == 0 && self.global_line.load(Ordering::Acquire) == 0 {
return Ok(0);
}
let eager = self.eager.disk_bytes();
let mut freed = 0u64;
loop {
let over_share = self.over_global_share();
let (used, victim, at) = {
let l = self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
let lazy: u64 = l.iter().map(|s| s.disk_bytes()).sum();
let within_own = budget == 0 || eager + lazy <= budget;
if (within_own && !over_share) || l.len() < 2 {
(eager + lazy, None, 0)
} else {
let at = self.cheapest_segment(l.len() - 1);
(eager + lazy, Some(l[at].clone()), at)
}
};
let Some(victim) = victim else {
let _ = used;
break;
};
let bytes = victim.disk_bytes();
let rows = victim.rows().unwrap_or(0);
let path = victim.path().to_path_buf();
{
let mut l = self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
if at < l.len() {
l.remove(at);
}
if let Ok(mut st) = self.seg_cost.lock()
&& at < st.len()
{
st.remove(at);
}
}
drop(victim);
std::fs::remove_file(&path).ok();
self.evicted_segments.fetch_add(1, Ordering::Relaxed);
self.evicted_bytes.fetch_add(bytes, Ordering::Relaxed);
self.evicted_rows.fetch_add(rows, Ordering::Relaxed);
freed += bytes;
}
Ok(freed)
}
fn cheapest_segment(&self, exclusive: usize) -> usize {
let Ok(st) = self.seg_cost.lock() else {
return 0;
};
let newest = st.last().map(|s| s.touched).unwrap_or(0).max(1);
let mut best = 0usize;
let mut best_score = u128::MAX;
for i in 0..exclusive.min(st.len()) {
let s = st[i];
let mean = if s.rows == 0 { 0 } else { s.cost / s.rows };
let recency = 1 + (s.touched as u128 * 1023) / newest as u128;
let score = (mean as u128 + 1) * recency;
if score < best_score {
best_score = score;
best = i;
}
}
best
}
fn new_segment(&self) -> Result<std::sync::Arc<ExplodedArchive>> {
let n = self.next_seg.fetch_add(1, Ordering::AcqRel);
let name = format!(
"{}.c{n}",
self.base.file_name().unwrap_or_default().to_string_lossy()
);
let path = self.base.with_file_name(name);
Ok(std::sync::Arc::new(ExplodedArchive::open_with_policy(
&path,
ExplodePolicy::Full,
)?))
}
pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
if let Some(hit) = self.eager.content(oid)? {
return Ok(Some(hit));
}
let segs = self.segments()?;
for s in segs.iter().rev() {
if let Some(hit) = s.content(oid)? {
return Ok(Some(hit));
}
}
Ok(None)
}
pub fn has(&self, oid: &[u8]) -> Result<bool> {
if self.eager.has(oid)? {
return Ok(true);
}
for s in self.segments()?.iter().rev() {
if s.has(oid)? {
return Ok(true);
}
}
Ok(false)
}
pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
if let Some(n) = self.eager.name(oid)? {
return Ok(Some(n));
}
for s in self.segments()?.iter().rev() {
if let Some(n) = s.name(oid)? {
return Ok(Some(n));
}
}
Ok(None)
}
pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
let mut out = self.eager.of_kind(kind)?;
let segs = self.segments()?;
if segs.is_empty() {
return Ok(out);
}
let mut seen: std::collections::HashSet<Vec<u8>> =
out.iter().map(|(o, _)| o.clone()).collect();
for s in segs.iter().rev() {
for (oid, payload) in s.of_kind(kind)? {
if seen.insert(oid.clone()) {
out.push((oid, payload));
}
}
}
Ok(out)
}
pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
let mut n = self.eager.retain(live)?;
for s in self.segments()? {
n += s.retain(live)?;
}
Ok(n)
}
pub fn drop_cache(&self) -> Result<u64> {
let victims = {
let mut l = self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?;
if let Ok(mut st) = self.seg_cost.lock() {
st.clear();
}
std::mem::take(&mut *l)
};
let mut freed = 0;
for s in victims {
let bytes = s.disk_bytes();
let rows = s.rows().unwrap_or(0);
let path = s.path().to_path_buf();
drop(s);
std::fs::remove_file(&path).ok();
self.evicted_segments.fetch_add(1, Ordering::Relaxed);
self.evicted_bytes.fetch_add(bytes, Ordering::Relaxed);
self.evicted_rows.fetch_add(rows, Ordering::Relaxed);
freed += bytes;
}
self.dirty.store(0, Ordering::Release);
Ok(freed)
}
pub fn flush(&self) -> Result<()> {
self.eager.flush()?;
for s in self.segments()? {
s.flush()?;
}
self.dirty.store(0, Ordering::Release);
Ok(())
}
pub fn drop_caches(&self) {
self.eager.drop_caches();
if let Ok(l) = self.lazy.lock() {
for s in l.iter() {
s.drop_caches();
}
}
}
pub fn rows(&self) -> Result<u64> {
let mut n = self.eager.rows()?;
for s in self.segments()? {
n += s.rows()?;
}
Ok(n)
}
pub fn eager_rows(&self) -> Result<u64> {
self.eager.rows()
}
pub fn disk_bytes(&self) -> u64 {
self.use_now().used()
}
pub fn use_now(&self) -> CacheUse {
let (lazy_bytes, segments) = match self.lazy.lock() {
Ok(l) => (l.iter().map(|s| s.disk_bytes()).sum(), l.len() as u64),
Err(_) => (0, 0),
};
CacheUse {
eager_bytes: self.eager.disk_bytes(),
lazy_bytes,
budget: self.budget.load(Ordering::Acquire),
segments,
evicted_segments: self.evicted_segments.load(Ordering::Relaxed),
evicted_bytes: self.evicted_bytes.load(Ordering::Relaxed),
evicted_rows: self.evicted_rows.load(Ordering::Relaxed),
written_back: self.written_back.load(Ordering::Relaxed),
direct_reads: self.direct_reads.load(Ordering::Relaxed),
suppressed_write_backs: self.suppressed_write_backs.load(Ordering::Relaxed),
}
}
pub fn note_rederived(&self) {
self.eager.note_rederived();
}
pub fn pread_served(&self) -> u64 {
self.eager.pread_served()
+ self
.segments()
.map(|s| s.iter().map(|x| x.pread_served()).sum::<u64>())
.unwrap_or(0)
}
pub fn skipped(&self) -> u64 {
self.eager.skipped()
}
pub fn engine_stats(&self) -> ExplodedStats {
let mut s = self.eager.engine_stats();
if let Ok(segs) = self.segments() {
for x in segs {
let e = x.engine_stats();
s.rows += e.rows;
s.written += e.written;
s.served += e.served;
s.absent += e.absent;
}
}
s
}
pub fn kick(self: &std::sync::Arc<Self>) {
if self
.maintaining
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
return;
}
let me = self.clone();
let spawned = std::thread::Builder::new()
.name("explode-writeback".into())
.stack_size(256 << 10)
.spawn(move || {
loop {
if let Err(e) = me.maintain() {
tracing::warn!(target: "znippy.git", error = %e, "explode cache maintenance");
break;
}
if me.dirty.load(Ordering::Acquire) < CACHE_FLUSH_BYTES {
break;
}
}
me.maintaining.store(false, Ordering::Release);
});
if spawned.is_err() {
self.maintaining.store(false, Ordering::Release);
}
}
fn segments(&self) -> Result<Vec<std::sync::Arc<ExplodedArchive>>> {
Ok(self
.lazy
.lock()
.map_err(|_| anyhow::anyhow!("exploded cache poisoned"))?
.clone())
}
}
impl crate::exploded::PayloadSink for ExplodedCache {
fn wants(&self, kind: GitObjectKind) -> bool {
crate::exploded::PayloadSink::wants(&self.eager, kind)
}
fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
ExplodedCache::explode(self, oid, kind, payload)
}
fn flush(&self) -> Result<()> {
self.eager.flush().map(|_| ())
}
}
fn registry() -> &'static Mutex<Vec<std::sync::Weak<ExplodedCache>>> {
static R: std::sync::OnceLock<Mutex<Vec<std::sync::Weak<ExplodedCache>>>> =
std::sync::OnceLock::new();
R.get_or_init(|| Mutex::new(Vec::new()))
}
pub fn use_all() -> Vec<(String, CacheUse)> {
let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
r.retain(|w| w.strong_count() > 0);
r.iter()
.filter_map(|w| w.upgrade())
.map(|c| (c.root.clone(), c.use_now()))
.collect()
}
pub fn set_budget_under(root: &str, bytes: u64) -> usize {
let mut n = 0;
let mut r = registry().lock().unwrap_or_else(|e| e.into_inner());
r.retain(|w| w.strong_count() > 0);
for c in r.iter().filter_map(|w| w.upgrade()) {
if c.root.starts_with(root) {
c.set_budget(bytes);
n += 1;
}
}
n
}
fn segment_for(budget: u64) -> u64 {
if budget == 0 {
return MAX_SEGMENT.min(1 << 28);
}
let floor = MIN_SEGMENT.min(budget / 4).max(64 << 10);
(budget / SEGMENTS_PER_BUDGET).clamp(floor, MAX_SEGMENT)
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("explode-cache-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d.join("objects.exploded")
}
fn noise(len: usize, seed: u8) -> Vec<u8> {
let mut x: u32 = 0x9E37_79B9 ^ (seed as u32).wrapping_mul(0x0101_0101);
(0..len)
.map(|_| {
x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
(x >> 24) as u8
})
.collect()
}
fn dear(payload: &[u8]) -> crate::resolve::ReadCost {
crate::resolve::ReadCost {
depth: CHEAP_DEPTH + 8,
inflated: payload.len() as u64,
external_base: false,
}
}
const _: () = assert!((64 << 10) < MAX_CACHED_BYTES as usize);
fn oid(n: u32) -> Vec<u8> {
let mut o = vec![0u8; 20];
o[..4].copy_from_slice(&n.to_be_bytes());
o
}
#[test]
fn the_tier_plateaus_at_its_budget_under_a_full_scan() {
let base = tmp("plateau");
let budget = 2 << 20;
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, budget).unwrap();
let payload = vec![7u8; 8 << 10];
let mut peak = 0u64;
for i in 0..2000u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear(&payload)).unwrap();
c.maintain().unwrap();
peak = peak.max(c.use_now().used());
}
let end = c.use_now();
let written = 2000u64 * payload.len() as u64;
assert!(
peak <= budget as u64 + segment_for(budget) + (1 << 20),
"the tier climbed to {peak} B against a {budget} B budget after {written} B of \
write-backs — it is a tier, not a cache"
);
assert!(
end.evicted_bytes > 0 && end.evicted_segments > 0,
"nothing was evicted, so the plateau is an accident of size: {end:?}"
);
}
#[test]
fn an_evicted_object_is_absent_and_never_wrong() {
let base = tmp("absent");
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
let payload = noise(64 << 10, 3);
for i in 0..200u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear(&payload)).unwrap();
c.maintain().unwrap();
}
let mut hits = 0;
for i in 0..200u32 {
match c.content(&oid(i)).unwrap() {
Some((k, b)) => {
assert_eq!(k, GitObjectKind::Blob);
assert_eq!(b, payload, "an evicting cache returned the wrong bytes");
hits += 1;
}
None => {}
}
}
assert!(hits > 0, "the cache evicted everything, including the newest segment");
assert!(hits < 200, "nothing was evicted, so this proves nothing");
}
#[test]
fn the_global_line_is_paid_by_the_cache_over_its_share() {
let base_a = tmp("global-a");
let base_b = tmp("global-b");
let a = std::sync::Arc::new(ExplodedCache::open(&base_a, ExplodePolicy::Commits, 4 << 20).unwrap())
.registered();
let b = std::sync::Arc::new(ExplodedCache::open(&base_b, ExplodePolicy::Commits, 4 << 20).unwrap())
.registered();
a.set_global_line(0);
b.set_global_line(0);
let big = noise(64 << 10, 5);
for i in 0..48u32 {
a.write_back(&oid(i), GitObjectKind::Blob, &big, dear(&big)).unwrap();
a.maintain().unwrap();
}
let small = noise(8 << 10, 6);
b.write_back(&oid(100), GitObjectKind::Blob, &small, dear(&small)).unwrap();
b.maintain().unwrap();
let a0 = a.use_now();
let b0 = b.use_now();
let line = 1u64 << 20;
assert!(a0.lazy_bytes > line, "precondition: A alone is over the line, has {}", a0.lazy_bytes);
assert!(a0.segments >= 3, "precondition: A rotated into segments, has {}", a0.segments);
assert_eq!(a0.evicted_segments + b0.evicted_segments, 0, "precondition: nothing evicted yet");
a.set_global_line(line);
b.set_global_line(line);
b.maintain().unwrap();
assert_eq!(b.use_now().evicted_segments, 0, "a cache under its fair share must not evict for the box");
a.maintain().unwrap();
let a1 = a.use_now();
assert!(a1.evicted_segments > 0, "the big pool must evict for the box");
assert!(
a1.lazy_bytes + a1.eager_bytes <= line,
"A must have shrunk under the line on its own: {} > {line}",
a1.lazy_bytes + a1.eager_bytes
);
assert!(a1.lazy_bytes < 4 << 20, "its own budget was never the reason");
assert_eq!(b.use_now().evicted_segments, 0, "B still untouched");
}
#[test]
fn eviction_never_takes_a_commit() {
let base = tmp("commits");
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
for i in 0..50u32 {
c.explode(&oid(1_000_000 + i), GitObjectKind::Commit, b"tree x\n")
.unwrap();
}
c.flush().unwrap();
let blob = noise(64 << 10, 1);
for i in 0..200u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
c.maintain().unwrap();
}
assert!(c.use_now().evicted_segments > 0, "nothing was evicted");
for i in 0..50u32 {
assert!(
c.content(&oid(1_000_000 + i)).unwrap().is_some(),
"a commit was evicted — graph and reach are now blind"
);
}
}
#[test]
fn dropping_the_cache_reclaims_the_lazy_pool_only() {
let base = tmp("gc");
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
c.explode(&oid(9), GitObjectKind::Commit, b"tree y\n").unwrap();
let blob = vec![2u8; 32 << 10];
for i in 0..100u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
}
c.maintain().unwrap();
let before = c.use_now();
assert!(before.lazy_bytes > 0);
let freed = c.drop_cache().unwrap();
assert!(freed > 0, "gc reclaimed nothing");
let after = c.use_now();
assert_eq!(after.lazy_bytes, 0, "the lazy pool survived gc");
assert_eq!(after.eager_bytes, before.eager_bytes, "gc took the graph too");
assert!(c.content(&oid(9)).unwrap().is_some());
}
#[test]
fn a_reopen_adopts_the_segments_it_left() {
let base = tmp("reopen");
let blob = vec![5u8; 32 << 10];
let bytes = {
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
for i in 0..40u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &blob, dear(&blob)).unwrap();
}
c.maintain().unwrap();
c.use_now().lazy_bytes
};
assert!(bytes > 0);
let again = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
assert_eq!(again.use_now().lazy_bytes, bytes, "the segments were not adopted");
assert_eq!(
again.content(&oid(7)).unwrap().map(|(_, b)| b),
Some(blob),
"an adopted segment does not answer"
);
}
#[test]
fn a_shallow_small_object_is_read_directly_and_never_cached() {
let base = tmp("cheap");
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 0).unwrap();
let small = vec![1u8; 1 << 10];
let big = vec![2u8; 64 << 10];
for i in 0..400u32 {
let took = c
.write_back(
&oid(i),
GitObjectKind::Blob,
&small,
crate::resolve::ReadCost {
depth: CHEAP_DEPTH,
inflated: small.len() as u64,
external_base: false,
},
)
.unwrap();
assert!(!took, "a cheap object asked for a flush");
}
c.maintain().unwrap();
let after_cheap = c.use_now();
assert_eq!(
after_cheap.lazy_bytes, 0,
"400 shallow 1 KiB objects put {} B in the cache; a direct read is \
microseconds and this is quota spent to save them",
after_cheap.lazy_bytes
);
assert_eq!(
after_cheap.direct_reads, 400,
"the policy declined {} of 400 cheap objects — it is not running",
after_cheap.direct_reads
);
assert_eq!(after_cheap.written_back, 0);
c.write_back(
&oid(1000),
GitObjectKind::Blob,
&small,
crate::resolve::ReadCost { depth: CHEAP_DEPTH + 1, inflated: small.len() as u64, external_base: false },
)
.unwrap();
c.write_back(
&oid(1002),
GitObjectKind::Blob,
&small,
crate::resolve::ReadCost { depth: 0, inflated: small.len() as u64, external_base: true },
)
.unwrap();
c.write_back(
&oid(1001),
GitObjectKind::Blob,
&big,
crate::resolve::ReadCost { depth: 40, inflated: (MAX_CACHED_BYTES + 1) as u64, external_base: true },
)
.unwrap();
c.maintain().unwrap();
let end = c.use_now();
assert_eq!(end.written_back, 2, "an expensive object was declined");
assert_eq!(
end.direct_reads, 401,
"an object over the size cap was cached instead of read directly"
);
for n in [1000u32, 1002] {
assert!(
c.content(&oid(n)).unwrap().is_some(),
"the expensive object {n} is not in the cache"
);
}
assert!(
c.content(&oid(1001)).unwrap().is_none(),
"an object over the size cap is in the cache"
);
assert!(
c.content(&oid(7)).unwrap().is_none(),
"a declined object is in the cache after all"
);
}
#[test]
fn eviction_takes_the_cheap_segment_before_the_old_expensive_one() {
let base = tmp("costrank");
let c = ExplodedCache::open(&base, ExplodePolicy::Commits, 1 << 20).unwrap();
let payload = noise(32 << 10, 9);
let dear = crate::resolve::ReadCost { depth: 40, inflated: payload.len() as u64, external_base: false };
let cheapish = crate::resolve::ReadCost {
depth: CHEAP_DEPTH + 1,
inflated: payload.len() as u64,
external_base: false,
};
for i in 0..16u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &payload, dear).unwrap();
}
c.maintain().unwrap();
for i in 100..160u32 {
c.write_back(&oid(i), GitObjectKind::Blob, &payload, cheapish).unwrap();
c.maintain().unwrap();
}
let end = c.use_now();
assert!(end.evicted_segments > 0, "nothing was evicted; this proves nothing");
let expensive_left = (0..16u32).filter(|i| c.content(&oid(*i)).unwrap().is_some()).count();
let cheap_left = (100..160u32).filter(|i| c.content(&oid(*i)).unwrap().is_some()).count();
assert!(
expensive_left > 0,
"every depth-40 object was evicted while {cheap_left} barely-admitted ones \
survived — eviction is ranking by age alone: {end:?}"
);
}
#[test]
fn a_budget_is_read_and_a_typo_is_refused() {
assert_eq!(parse_budget("8G").unwrap(), 8 << 30);
assert_eq!(parse_budget(" 512m ").unwrap(), 512 << 20);
assert_eq!(parse_budget("1048576").unwrap(), 1 << 20);
assert_eq!(parse_budget("unlimited").unwrap(), 0);
assert!(parse_budget("8 gigs").is_err());
assert!(parse_budget("").is_err());
}
#[test]
fn the_default_budget_is_the_smaller_of_a_tenth_and_the_ceiling() {
let d = default_budget(&std::env::temp_dir());
assert!(d >= MIN_BUDGET && d <= BUDGET_CEILING, "{d}");
}
#[test]
fn a_suppressed_thread_caches_nothing_and_an_unsuppressed_one_fills_the_pool() {
let payload = vec![7u8; 64 << 10];
let costly = crate::resolve::ReadCost {
depth: CHEAP_DEPTH + 8,
inflated: payload.len() as u64,
external_base: false,
};
let hot = ExplodedCache::open(&tmp("nosuppress"), ExplodePolicy::Commits, 0).unwrap();
for i in 0..200u32 {
hot.write_back(&oid(i), GitObjectKind::Tree, &payload, costly).unwrap();
}
hot.maintain().unwrap();
let hot_use = hot.use_now();
assert!(
hot_use.lazy_bytes > 0,
"without the guard these objects must be cached, or the GREEN half \
below is proving that `worth_caching` declined them and nothing else"
);
assert_eq!(hot_use.written_back, 200);
assert_eq!(hot_use.suppressed_write_backs, 0);
let cold = ExplodedCache::open(&tmp("suppress"), ExplodePolicy::Commits, 0).unwrap();
{
let _guard = suppress_write_back();
assert!(write_back_suppressed());
for i in 0..200u32 {
let took = cold.write_back(&oid(i), GitObjectKind::Tree, &payload, costly).unwrap();
assert!(!took, "a validation read asked for a flush");
}
}
cold.maintain().unwrap();
let cold_use = cold.use_now();
assert_eq!(
cold_use.lazy_bytes, 0,
"a receive-pack's connectivity walk put {} B in the cache",
cold_use.lazy_bytes
);
assert_eq!(cold_use.written_back, 0);
assert_eq!(
cold_use.suppressed_write_backs, 200,
"the gate must be what declined them — {} suppressed, not 200",
cold_use.suppressed_write_backs
);
assert_eq!(cold_use.direct_reads, 0);
}
#[test]
fn the_guard_is_released_on_drop_and_on_unwind_and_it_nests() {
assert!(!write_back_suppressed(), "a fresh thread starts unsuppressed");
{
let _a = suppress_write_back();
{
let _b = suppress_write_back();
assert!(write_back_suppressed());
}
assert!(
write_back_suppressed(),
"the inner guard ending reopened the cache while the outer scope \
is still validating"
);
}
assert!(!write_back_suppressed());
let unwound = std::panic::catch_unwind(|| {
let _g = suppress_write_back();
assert!(write_back_suppressed());
panic!("a connectivity walk died");
});
assert!(unwound.is_err());
assert!(
!write_back_suppressed(),
"a panic inside the guard left this thread never caching again"
);
}
}