cache.rs 84.9%

241 of 284 lines run, 57 of 73 functions · run #532 on main · run missed

//! Loaded states cached by their heads (hub spec §16.3). A state is a pure
//! function of its heads, so entries never go stale; the cache is only
//! bounded. Shared across threads because `oxy_core::state::State` is
//! `Send + Sync`.

use oxy_core::hash::ChangeHash;
use oxy_core::repo::Repo;
use oxy_core::state::{Snapshot, State};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

/// A state and its materialised snapshot.
pub struct Loaded {
    pub heads: Vec<ChangeHash>,
    pub state: State,
    pub snapshot: Snapshot,
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct Key {
    repo: [u8; 16],
    heads: Vec<ChangeHash>,
}

struct Entry {
    loaded: Arc<Loaded>,
    last_used: u64,
    /// Its estimated size in bytes (`weight`).
    weight: usize,
}

/// A loaded state's estimated size (oxygenhub#246): five bytes per byte of
/// text and 384 per line (each file's line graph), and a kilobyte per change.
/// Measured on 2026-10-17 against what loading a state really adds to the
/// process (`measures_a_real_state`): oxygen's states ~85 MiB, oxygenhub's
/// ~70 MiB, which this puts at 88 and 75. The older "text three times over"
/// said 18 and 28, so a cache that thought it held 2.5 GB held 6-8 GB.
fn weight(l: &Loaded) -> usize {
    let (mut text, mut lines) = (0usize, 0usize);
    for f in l.snapshot.files.values() {
        match &f.content {
            oxy_core::state::FileContent::Text { bytes, .. } => {
                text += bytes.len();
                lines += bytes.iter().filter(|b| **b == b'\n').count() + 1;
            }
            _ => text += 64,
        }
    }
    text.saturating_mul(5).saturating_add(lines.saturating_mul(384)).saturating_add(l.state.len().saturating_mul(1024))
}

/// What the state cache may hold, in bytes: OXYHUB_STATE_CACHE_MB, else
/// 3 GiB. A count alone let a few large repositories' states grow the hub
/// to 17 GB.
pub fn budget() -> usize {
    std::env::var("OXYHUB_STATE_CACHE_MB").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(3 * 1024).saturating_mul(1 << 20)
}

/// What the bookmark lists and a change's "in bookmarks" need of a state:
/// small, so many are kept, and computed without filling the state cache.
/// A repository with more bookmarks than the state cache holds otherwise
/// cycled through it on every such page, so nothing stayed cached
/// (metalgear's 128 bookmarks: 7.7 s per change page, and every page of
/// every repository slower after it).
pub struct Summary {
    pub changes: std::collections::HashSet<ChangeHash>,
    pub latest_ms: u64,
    pub conflicted: bool,
}

impl Summary {
    pub fn brief(&self) -> Brief {
        Brief { changes: self.changes.len(), latest_ms: self.latest_ms, conflicted: self.conflicted }
    }
}

/// What a bookmark list and the overview show of a state (#145): how many
/// changes, the latest, and whether it is conflicted. A pure function of
/// the heads, so it is kept on disk as well and a restart costs nothing:
/// metalgear's 162 bookmarks took ~31 s to summarise again after one.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Brief {
    pub changes: usize,
    pub latest_ms: u64,
    pub conflicted: bool,
}

/// Summaries kept; each is the state's change set, a few hundred kilobytes
/// for a long history.
const SUMMARIES: usize = 2048;
/// Most summaries computed at once when a page needs many.
const PARALLEL: usize = 4;

/// Summaries by heads, each with when it was last used, and the clock.
type Summaries = (HashMap<Key, (Arc<Summary>, u64)>, u64);

/// A bounded least-recently-used cache of loaded states.
pub struct StateCache {
    capacity: usize,
    /// Bytes of estimated state it may hold (`budget`).
    budget: usize,
    inner: Mutex<(HashMap<Key, Entry>, u64)>,
    summaries: Mutex<Summaries>,
    /// Where briefs are kept on disk, when anywhere.
    briefs: Option<PathBuf>,
}

impl StateCache {
    pub fn new(capacity: usize) -> StateCache {
        StateCache::with_budget(capacity, budget())
    }

    pub fn with_budget(capacity: usize, budget: usize) -> StateCache {
        StateCache {
            capacity: capacity.max(1),
            budget,
            inner: Mutex::new((HashMap::new(), 0)),
            summaries: Mutex::new((HashMap::new(), 0)),
            briefs: None,
        }
    }

    /// Keep briefs on disk under `dir`, one small file per state.
    pub fn with_briefs(mut self, dir: PathBuf) -> StateCache {
        self.briefs = Some(dir);
        self
    }

    fn brief_path(&self, key: &Key) -> Option<PathBuf> {
        let dir = self.briefs.as_ref()?;
        let mut h = blake3::Hasher::new();
        for c in &key.heads {
            h.update(&c.0);
        }
        Some(dir.join(hex::encode(key.repo)).join(h.finalize().to_hex().as_str()))
    }

    fn known_brief(&self, key: &Key) -> Option<Brief> {
        if let Some((s, _)) = self.summaries.lock().unwrap().0.get(key) {
            return Some(s.brief());
        }
        read_brief(&self.brief_path(key)?)
    }

    /// The brief of the state with these heads: from memory, from disk, or
    /// worked out and kept.
    pub fn brief(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Brief, oxy_core::repo::RepoError> {
        Ok(self.briefs(repo, &[heads.to_vec()])?[0])
    }

    /// The briefs of many states, those not yet known worked out a few at
    /// a time in parallel.
    pub fn briefs(&self, repo: &Repo, all: &[Vec<ChangeHash>]) -> Result<Vec<Brief>, oxy_core::repo::RepoError> {
        let id = repo.id()?;
        let mut out: Vec<Option<Brief>> = all.iter().map(|h| self.known_brief(&Key { repo: id, heads: sorted(h) })).collect();
        let missing: Vec<usize> = (0..all.len()).filter(|&i| out[i].is_none()).collect();
        if missing.is_empty() {
            return Ok(out.into_iter().map(|b| b.expect("known")).collect());
        }
        let next = Mutex::new(missing.iter());
        let found: Mutex<Vec<(usize, Result<Brief, oxy_core::repo::RepoError>)>> = Mutex::new(Vec::new());
        let threads = PARALLEL.min(missing.len()).min(std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));
        // A `Repo` stays on its thread, so each worker opens its own.
        let root = repo.root().to_path_buf();
        std::thread::scope(|s| {
            for _ in 0..threads {
                s.spawn(|| {
                    let mine = Repo::open(&root);
                    loop {
                        let Some(&i) = next.lock().unwrap().next() else { break };
                        let b = match &mine {
                            Ok(r) => self.summary(r, &all[i]).map(|s| s.brief()),
                            Err(e) => Err(oxy_core::repo::RepoError::Msg(format!("open {}: {e}", root.display()))),
                        };
                        found.lock().unwrap().push((i, b));
                    }
                });
            }
        });
        for (i, b) in found.into_inner().unwrap() {
            out[i] = Some(b?);
        }
        Ok(out.into_iter().map(|b| b.expect("worked out")).collect())
    }

    /// The summary of the state with these heads: from the state cache when
    /// it holds the state, else loaded and dropped, the state cache left as
    /// it was.
    pub fn summary(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Summary>, oxy_core::repo::RepoError> {
        let sorted = sorted(heads);
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
        {
            let mut guard = self.summaries.lock().unwrap();
            let (map, tick) = &mut *guard;
            *tick += 1;
            if let Some(e) = map.get_mut(&key) {
                e.1 = *tick;
                return Ok(e.0.clone());
            }
        }
        let held = self.inner.lock().unwrap().0.get(&key).map(|e| e.loaded.clone());
        let summary = Arc::new(match held {
            Some(l) => summarise(&l.state, &l.snapshot),
            None => {
                let state = repo.load_state(&sorted)?;
                let snapshot = state.materialize();
                summarise(&state, &snapshot)
            }
        });
        if let Some(path) = self.brief_path(&key) {
            write_brief(&path, summary.brief());
        }
        let mut guard = self.summaries.lock().unwrap();
        let (map, tick) = &mut *guard;
        if map.len() >= SUMMARIES {
            if let Some(oldest) = map.iter().min_by_key(|(_, e)| e.1).map(|(k, _)| k.clone()) {
                map.remove(&oldest);
            }
        }
        map.insert(key, (summary.clone(), *tick));
        Ok(summary)
    }

    /// The state with these heads, loaded once per hub process.
    pub fn load(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Loaded>, oxy_core::repo::RepoError> {
        let mut sorted = heads.to_vec();
        sorted.sort();
        sorted.dedup();
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
        {
            let mut guard = self.inner.lock().unwrap();
            let (map, tick) = &mut *guard;
            *tick += 1;
            if let Some(e) = map.get_mut(&key) {
                e.last_used = *tick;
                return Ok(e.loaded.clone());
            }
        }
        let state = repo.load_state(&sorted)?;
        let snapshot = state.materialize();
        let loaded = Arc::new(Loaded { heads: sorted, state, snapshot });
        let w = weight(&loaded);
        let mut guard = self.inner.lock().unwrap();
        let (map, tick) = &mut *guard;
        // The oldest go until it fits by count and by size; the new one
        // stays even when it alone is over the budget.
        let budget = self.budget;
        let mut total: usize = map.values().map(|e| e.weight).sum::<usize>() + w;
        while !map.is_empty() && (map.len() >= self.capacity || total > budget) {
            let Some(oldest) = map.iter().min_by_key(|(_, e)| e.last_used).map(|(k, _)| k.clone()) else {
                break;
            };
            if let Some(e) = map.remove(&oldest) {
                total -= e.weight;
            }
        }
        map.insert(key, Entry { loaded: loaded.clone(), last_used: *tick, weight: w });
        Ok(loaded)
    }

    /// What it holds: loaded states, their estimated bytes, and summaries.
    pub fn stats(&self) -> (usize, usize, usize) {
        let (entries, weight) = {
            let guard = self.inner.lock().unwrap();
            (guard.0.len(), guard.0.values().map(|e| e.weight).sum())
        };
        (entries, weight, self.summaries.lock().unwrap().0.len())
    }

    pub fn len(&self) -> usize {
        self.inner.lock().unwrap().0.len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

fn sorted(heads: &[ChangeHash]) -> Vec<ChangeHash> {
    let mut v = heads.to_vec();
    v.sort();
    v.dedup();
    v
}

/// A brief on disk: `changes latest_ms conflicted`.
fn read_brief(path: &Path) -> Option<Brief> {
    let text = std::fs::read_to_string(path).ok()?;
    let mut f = text.split_whitespace();
    Some(Brief { changes: f.next()?.parse().ok()?, latest_ms: f.next()?.parse().ok()?, conflicted: f.next()? == "1" })
}

/// Kept when it can be; a brief that is not kept is only worked out again.
fn write_brief(path: &Path, b: Brief) {
    let Some(dir) = path.parent() else { return };
    if std::fs::create_dir_all(dir).is_err() {
        return;
    }
    let tmp = path.with_extension(format!("tmp{}", std::process::id()));
    if std::fs::write(&tmp, format!("{} {} {}", b.changes, b.latest_ms, u8::from(b.conflicted))).is_ok() {
        std::fs::rename(&tmp, path).ok();
    }
}

fn summarise(state: &State, snapshot: &Snapshot) -> Summary {
    Summary {
        changes: state.changes().map(|(h, _)| *h).collect(),
        latest_ms: state.changes().map(|(_, m)| m.hlc.wall_ms).max().unwrap_or(0),
        conflicted: snapshot.is_conflicted(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// What a loaded state really costs against `weight` (oxygenhub#246):
    /// `OXYHUB_MEASURE_REPO=path cargo test -p oxygenhub --lib
    /// measures_a_real_state -- --ignored --nocapture`.
    #[test]
    #[ignore]
    fn measures_a_real_state() {
        fn rss_kib() -> usize {
            let s = std::fs::read_to_string("/proc/self/status").unwrap();
            s.lines().find(|l| l.starts_with("VmRSS:")).and_then(|l| l.split_whitespace().nth(1)).unwrap().parse().unwrap()
        }
        let path = std::env::var("OXYHUB_MEASURE_REPO").expect("OXYHUB_MEASURE_REPO");
        let repo = Repo::open(std::path::Path::new(&path)).unwrap();
        let view = repo.view().unwrap();
        // Kept alive, so each load's memory is new, not reused.
        let cache = StateCache::with_budget(64, usize::MAX);
        let mut held = Vec::new();
        for bookmark in view.bookmarks.keys().take(6).cloned().collect::<Vec<_>>() {
            let heads = repo.heads_of(&view, &bookmark).unwrap();
            let before = rss_kib();
            let loaded = cache.load(&repo, &heads).unwrap();
            let after = rss_kib();
            let files = loaded.snapshot.files.len();
            let (mut text, mut lines) = (0usize, 0usize);
            for f in loaded.snapshot.files.values() {
                if let oxy_core::state::FileContent::Text { bytes, .. } = &f.content {
                    text += bytes.len();
                    lines += bytes.iter().filter(|b| **b == b'\n').count();
                }
            }
            println!(
                "{bookmark}: {} changes, {files} files, {} KiB text, {lines} lines: weight {} MiB, resident +{} MiB",
                loaded.state.len(),
                text >> 10,
                weight(&loaded) >> 20,
                (after.saturating_sub(before)) >> 10
            );
            held.push(loaded);
        }
    }

    #[test]
    fn caches_by_heads_and_evicts_the_oldest() {
        let dir = tempfile::tempdir().unwrap();
        let repo = Repo::init_bare(dir.path()).unwrap();
        let cache = StateCache::new(2);
        let a = cache.load(&repo, &[]).unwrap();
        let b = cache.load(&repo, &[]).unwrap();
        assert!(Arc::ptr_eq(&a, &b), "same heads, same entry");
        assert_eq!(cache.len(), 1);
        let h1 = oxy_core::hash::ObjectRef([1; 32]);
        let h2 = oxy_core::hash::ObjectRef([2; 32]);
        // Unknown heads fail to load and are not cached.
        assert!(cache.load(&repo, &[h1]).is_err());
        assert_eq!(cache.len(), 1);
        let _ = (h2, dir);
    }

    #[test]
    fn summaries_leave_the_state_cache_alone() {
        let dir = tempfile::tempdir().unwrap();
        let repo = Repo::init_bare(dir.path()).unwrap();
        let cache = StateCache::new(2);
        let a = cache.summary(&repo, &[]).unwrap();
        assert_eq!(cache.len(), 0, "a summary does not fill the state cache");
        assert!(Arc::ptr_eq(&a, &cache.summary(&repo, &[]).unwrap()), "kept");
        assert!(a.changes.is_empty() && !a.conflicted);
        assert!(cache.summary(&repo, &[oxy_core::hash::ObjectRef([3; 32])]).is_err());
    }

    #[test]
    fn a_size_budget_evicts_the_oldest_and_keeps_the_newest() {
        let dir = tempfile::tempdir().unwrap();
        let repo = Repo::init(dir.path()).unwrap();
        std::fs::write(dir.path().join("a.txt"), "a\n".repeat(1000)).unwrap();
        let one = repo.commit("one", &[]).unwrap();
        std::fs::write(dir.path().join("b.txt"), "b\n".repeat(1000)).unwrap();
        let two = repo.commit("two", &[]).unwrap();
        // Room for nothing: each load stays, the one before goes.
        let cache = StateCache::with_budget(64, 0);
        cache.load(&repo, &[one]).unwrap();
        assert_eq!(cache.len(), 1, "kept though over the budget");
        cache.load(&repo, &[two]).unwrap();
        assert_eq!(cache.len(), 1, "the older one went");
        // Room for plenty: both stay.
        let roomy = StateCache::with_budget(64, 1 << 30);
        roomy.load(&repo, &[one]).unwrap();
        roomy.load(&repo, &[two]).unwrap();
        assert_eq!(roomy.len(), 2);
    }

    #[test]
    fn briefs_are_worked_out_in_parallel_and_survive_a_restart() {
        let dir = tempfile::tempdir().unwrap();
        let repo = Repo::init(dir.path()).unwrap();
        std::fs::write(dir.path().join("a.txt"), "a\n").unwrap();
        let one = repo.commit("one", &[]).unwrap();
        std::fs::write(dir.path().join("b.txt"), "b\n").unwrap();
        let two = repo.commit("two", &[]).unwrap();
        let kept = tempfile::tempdir().unwrap();
        let cache = StateCache::new(8).with_briefs(kept.path().to_path_buf());
        let got = cache.briefs(&repo, &[vec![one], vec![one, two], vec![]]).unwrap();
        assert_eq!(got.iter().map(|b| b.changes).collect::<Vec<_>>(), vec![1, 2, 0]);
        assert!(got[1].latest_ms >= got[0].latest_ms && !got[1].conflicted);
        // A new cache (a restarted hub) reads them back without loading a
        // state: a repository that no longer has the changes still answers.
        let gone = tempfile::tempdir().unwrap();
        let empty = Repo::init(gone.path()).unwrap();
        let empty_id = empty.id().unwrap();
        for f in std::fs::read_dir(kept.path()).unwrap() {
            let f = f.unwrap().path();
            std::fs::rename(&f, kept.path().join(hex::encode(empty_id))).unwrap();
        }
        let restarted = StateCache::new(8).with_briefs(kept.path().to_path_buf());
        assert_eq!(restarted.brief(&empty, &[two, one]).unwrap(), got[1], "from disk, whatever the heads' order");
        assert!(StateCache::new(8).brief(&empty, &[one, two]).is_err(), "without the disk it would have to load");
        // Unknown heads are an error, not a guess.
        assert!(cache.briefs(&repo, &[vec![one], vec![oxy_core::hash::ObjectRef([9; 32])]]).is_err());
        // A bare repository, as the hub keeps them, works across threads too.
        let bare_dir = tempfile::tempdir().unwrap();
        let bare = Repo::init_bare(bare_dir.path()).unwrap();
        assert_eq!(cache.briefs(&bare, &[vec![], vec![]]).unwrap().len(), 2);
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! Loaded states cached by their heads (hub spec §16.3). A state is a pure
2		//! function of its heads, so entries never go stale; the cache is only
3		//! bounded. Shared across threads because `oxy_core::state::State` is
4		//! `Send + Sync`.
5
6		use oxy_core::hash::ChangeHash;
7		use oxy_core::repo::Repo;
8		use oxy_core::state::{Snapshot, State};
9		use std::collections::HashMap;
10		use std::path::{Path, PathBuf};
11		use std::sync::{Arc, Mutex};
12
13		/// A state and its materialised snapshot.
14		pub struct Loaded {
15		pub heads: Vec<ChangeHash>,
16		pub state: State,
17		pub snapshot: Snapshot,
18		}
19
20		#[derive(Clone, Debug, PartialEq, Eq, Hash)]
21		struct Key {
22		repo: [u8; 16],
23		heads: Vec<ChangeHash>,
24		}
25
26		struct Entry {
27		loaded: Arc<Loaded>,
28		last_used: u64,
29		/// Its estimated size in bytes (`weight`).
30		weight: usize,
31		}
32
33		/// A loaded state's estimated size (oxygenhub#246): five bytes per byte of
34		/// text and 384 per line (each file's line graph), and a kilobyte per change.
35		/// Measured on 2026-10-17 against what loading a state really adds to the
36		/// process (`measures_a_real_state`): oxygen's states ~85 MiB, oxygenhub's
37		/// ~70 MiB, which this puts at 88 and 75. The older "text three times over"
38		/// said 18 and 28, so a cache that thought it held 2.5 GB held 6-8 GB.
39	331×	fn weight(l: &Loaded) -> usize {
40	331×	let (mut text, mut lines) = (0usize, 0usize);
41	830×	for f in l.snapshot.files.values() {
42	830×	match &f.content {
43	613×	oxy_core::state::FileContent::Text { bytes, .. } => {
44	613×	text += bytes.len();
45	65229×	lines += bytes.iter().filter(\|b\| **b == b'\n').count() + 1;
46		}
47	217×	_ => text += 64,
48		}
49		}
50	331×	text.saturating_mul(5).saturating_add(lines.saturating_mul(384)).saturating_add(l.state.len().saturating_mul(1024))
51	331×	}
52
53		/// What the state cache may hold, in bytes: OXYHUB_STATE_CACHE_MB, else
54		/// 3 GiB. A count alone let a few large repositories' states grow the hub
55		/// to 17 GB.
56	426×	pub fn budget() -> usize {
57	426×	std::env::var("OXYHUB_STATE_CACHE_MB").ok().and_then(\|v\| v.parse::<usize>().ok()).unwrap_or(3 * 1024).saturating_mul(1 << 20)
58	426×	}
59
60		/// What the bookmark lists and a change's "in bookmarks" need of a state:
61		/// small, so many are kept, and computed without filling the state cache.
62		/// A repository with more bookmarks than the state cache holds otherwise
63		/// cycled through it on every such page, so nothing stayed cached
64		/// (metalgear's 128 bookmarks: 7.7 s per change page, and every page of
65		/// every repository slower after it).
66		pub struct Summary {
67		pub changes: std::collections::HashSet<ChangeHash>,
68		pub latest_ms: u64,
69		pub conflicted: bool,
70		}
71
72		impl Summary {
73	678×	pub fn brief(&self) -> Brief {
74	678×	Brief { changes: self.changes.len(), latest_ms: self.latest_ms, conflicted: self.conflicted }
75	678×	}
76		}
77
78		/// What a bookmark list and the overview show of a state (#145): how many
79		/// changes, the latest, and whether it is conflicted. A pure function of
80		/// the heads, so it is kept on disk as well and a restart costs nothing:
81		/// metalgear's 162 bookmarks took ~31 s to summarise again after one.
82		#[derive(Clone, Copy, Debug, PartialEq, Eq)]
83		pub struct Brief {
84		pub changes: usize,
85		pub latest_ms: u64,
86		pub conflicted: bool,
87		}
88
89		/// Summaries kept; each is the state's change set, a few hundred kilobytes
90		/// for a long history.
91		const SUMMARIES: usize = 2048;
92		/// Most summaries computed at once when a page needs many.
93		const PARALLEL: usize = 4;
94
95		/// Summaries by heads, each with when it was last used, and the clock.
96		type Summaries = (HashMap<Key, (Arc<Summary>, u64)>, u64);
97
98		/// A bounded least-recently-used cache of loaded states.
99		pub struct StateCache {
100		capacity: usize,
101		/// Bytes of estimated state it may hold (`budget`).
102		budget: usize,
103		inner: Mutex<(HashMap<Key, Entry>, u64)>,
104		summaries: Mutex<Summaries>,
105		/// Where briefs are kept on disk, when anywhere.
106		briefs: Option<PathBuf>,
107		}
108
109		impl StateCache {
110	417×	pub fn new(capacity: usize) -> StateCache {
111	417×	StateCache::with_budget(capacity, budget())
112	417×	}
113
114	419×	pub fn with_budget(capacity: usize, budget: usize) -> StateCache {
115	419×	StateCache {
116	419×	capacity: capacity.max(1),
117	419×	budget,
118	419×	inner: Mutex::new((HashMap::new(), 0)),
119	419×	summaries: Mutex::new((HashMap::new(), 0)),
120	419×	briefs: None,
121	419×	}
122	419×	}
123
124		/// Keep briefs on disk under `dir`, one small file per state.
125	414×	pub fn with_briefs(mut self, dir: PathBuf) -> StateCache {
126	414×	self.briefs = Some(dir);
127	414×	self
128	414×	}
129
130	333×	fn brief_path(&self, key: &Key) -> Option<PathBuf> {
131	333×	let dir = self.briefs.as_ref()?;
132	331×	let mut h = blake3::Hasher::new();
133	521×	for c in &key.heads {
134	521×	h.update(&c.0);
135	521×	}
136	331×	Some(dir.join(hex::encode(key.repo)).join(h.finalize().to_hex().as_str()))
137	333×	}
138
139	404×	fn known_brief(&self, key: &Key) -> Option<Brief> {
140	404×	if let Some((s, _)) = self.summaries.lock().unwrap().0.get(key) {
141	349×	return Some(s.brief());
142	55×	}
143	55×	read_brief(&self.brief_path(key)?)
144	404×	}
145
146		/// The brief of the state with these heads: from memory, from disk, or
147		/// worked out and kept.
148	2×	pub fn brief(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Brief, oxy_core::repo::RepoError> {
149	2×	Ok(self.briefs(repo, &[heads.to_vec()])?[0])
150	2×	}
151
152		/// The briefs of many states, those not yet known worked out a few at
153		/// a time in parallel.
154	333×	pub fn briefs(&self, repo: &Repo, all: &[Vec<ChangeHash>]) -> Result<Vec<Brief>, oxy_core::repo::RepoError> {
155	333×	let id = repo.id()?;
156	404×	let mut out: Vec<Option<Brief>> = all.iter().map(\|h\| self.known_brief(&Key { repo: id, heads: sorted(h) })).collect();
157	404×	let missing: Vec<usize> = (0..all.len()).filter(\|&i\| out[i].is_none()).collect();
158	333×	if missing.is_empty() {
159	311×	return Ok(out.into_iter().map(\|b\| b.expect("known")).collect());
160	50×	}
161	50×	let next = Mutex::new(missing.iter());
162	50×	let found: Mutex<Vec<(usize, Result<Brief, oxy_core::repo::RepoError>)>> = Mutex::new(Vec::new());
163	50×	let threads = PARALLEL.min(missing.len()).min(std::thread::available_parallelism().map(\|n\| n.get()).unwrap_or(1));
164		// A `Repo` stays on its thread, so each worker opens its own.
165	50×	let root = repo.root().to_path_buf();
166	50×	std::thread::scope(\|s\| {
167	50×	for _ in 0..threads {
168	54×	s.spawn(\|\| {
169	54×	let mine = Repo::open(&root);
170		loop {
171	108×	let Some(&i) = next.lock().unwrap().next() else { break };
172	54×	let b = match &mine {
173	54×	Ok(r) => self.summary(r, &all[i]).map(\|s\| s.brief()),
174	0×	Err(e) => Err(oxy_core::repo::RepoError::Msg(format!("open {}: {e}", root.display()))),
175		};
176	54×	found.lock().unwrap().push((i, b));
177		}
178	54×	});
179		}
180	50×	});
181	54×	for (i, b) in found.into_inner().unwrap() {
182	54×	out[i] = Some(b?);
183		}
184	90×	Ok(out.into_iter().map(\|b\| b.expect("worked out")).collect())
185	333×	}
186
187		/// The summary of the state with these heads: from the state cache when
188		/// it holds the state, else loaded and dropped, the state cache left as
189		/// it was.
190	498×	pub fn summary(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Summary>, oxy_core::repo::RepoError> {
191	498×	let sorted = sorted(heads);
192	498×	let key = Key { repo: repo.id()?, heads: sorted.clone() };
193		{
194	498×	let mut guard = self.summaries.lock().unwrap();
195	498×	let (map, tick) = &mut *guard;
196	498×	*tick += 1;
197	498×	if let Some(e) = map.get_mut(&key) {
198	217×	e.1 = *tick;
199	217×	return Ok(e.0.clone());
200	281×	}
201		}
202	281×	let held = self.inner.lock().unwrap().0.get(&key).map(\|e\| e.loaded.clone());
203	281×	let summary = Arc::new(match held {
204	33×	Some(l) => summarise(&l.state, &l.snapshot),
205		None => {
206	248×	let state = repo.load_state(&sorted)?;
207	245×	let snapshot = state.materialize();
208	245×	summarise(&state, &snapshot)
209		}
210		});
211	278×	if let Some(path) = self.brief_path(&key) {
212	277×	write_brief(&path, summary.brief());
213	277×	}
214	278×	let mut guard = self.summaries.lock().unwrap();
215	278×	let (map, tick) = &mut *guard;
216	278×	if map.len() >= SUMMARIES {
217	0×	if let Some(oldest) = map.iter().min_by_key(\|(_, e)\| e.1).map(\|(k, _)\| k.clone()) {
218	0×	map.remove(&oldest);
219	0×	}
220	278×	}
221	278×	map.insert(key, (summary.clone(), *tick));
222	278×	Ok(summary)
223	498×	}
224
225		/// The state with these heads, loaded once per hub process.
226	1631×	pub fn load(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Loaded>, oxy_core::repo::RepoError> {
227	1631×	let mut sorted = heads.to_vec();
228	1631×	sorted.sort();
229	1631×	sorted.dedup();
230	1631×	let key = Key { repo: repo.id()?, heads: sorted.clone() };
231		{
232	1630×	let mut guard = self.inner.lock().unwrap();
233	1630×	let (map, tick) = &mut *guard;
234	1630×	*tick += 1;
235	1630×	if let Some(e) = map.get_mut(&key) {
236	1298×	e.last_used = *tick;
237	1298×	return Ok(e.loaded.clone());
238	332×	}
239		}
240	332×	let state = repo.load_state(&sorted)?;
241	331×	let snapshot = state.materialize();
242	331×	let loaded = Arc::new(Loaded { heads: sorted, state, snapshot });
243	331×	let w = weight(&loaded);
244	331×	let mut guard = self.inner.lock().unwrap();
245	331×	let (map, tick) = &mut *guard;
246		// The oldest go until it fits by count and by size; the new one
247		// stays even when it alone is over the budget.
248	331×	let budget = self.budget;
249	331×	let mut total: usize = map.values().map(\|e\| e.weight).sum::<usize>() + w;
250	332×	while !map.is_empty() && (map.len() >= self.capacity \|\| total > budget) {
251	1×	let Some(oldest) = map.iter().min_by_key(\|(_, e)\| e.last_used).map(\|(k, _)\| k.clone()) else {
252	0×	break;
253		};
254	1×	if let Some(e) = map.remove(&oldest) {
255	1×	total -= e.weight;
256	1×	}
257		}
258	331×	map.insert(key, Entry { loaded: loaded.clone(), last_used: *tick, weight: w });
259	331×	Ok(loaded)
260	1631×	}
261
262		/// What it holds: loaded states, their estimated bytes, and summaries.
263	9×	pub fn stats(&self) -> (usize, usize, usize) {
264	9×	let (entries, weight) = {
265	9×	let guard = self.inner.lock().unwrap();
266	9×	(guard.0.len(), guard.0.values().map(\|e\| e.weight).sum())
267		};
268	9×	(entries, weight, self.summaries.lock().unwrap().0.len())
269	9×	}
270
271	6×	pub fn len(&self) -> usize {
272	6×	self.inner.lock().unwrap().0.len()
273	6×	}
274
275	0×	pub fn is_empty(&self) -> bool {
276	0×	self.len() == 0
277	0×	}
278		}
279
280	902×	fn sorted(heads: &[ChangeHash]) -> Vec<ChangeHash> {
281	902×	let mut v = heads.to_vec();
282	902×	v.sort();
283	902×	v.dedup();
284	902×	v
285	902×	}
286
287		/// A brief on disk: `changes latest_ms conflicted`.
288	54×	fn read_brief(path: &Path) -> Option<Brief> {
289	54×	let text = std::fs::read_to_string(path).ok()?;
290	1×	let mut f = text.split_whitespace();
291	1×	Some(Brief { changes: f.next()?.parse().ok()?, latest_ms: f.next()?.parse().ok()?, conflicted: f.next()? == "1" })
292	54×	}
293
294		/// Kept when it can be; a brief that is not kept is only worked out again.
295	277×	fn write_brief(path: &Path, b: Brief) {
296	277×	let Some(dir) = path.parent() else { return };
297	277×	if std::fs::create_dir_all(dir).is_err() {
298	0×	return;
299	277×	}
300	277×	let tmp = path.with_extension(format!("tmp{}", std::process::id()));
301	277×	if std::fs::write(&tmp, format!("{} {} {}", b.changes, b.latest_ms, u8::from(b.conflicted))).is_ok() {
302	277×	std::fs::rename(&tmp, path).ok();
303	277×	}
304	277×	}
305
306	278×	fn summarise(state: &State, snapshot: &Snapshot) -> Summary {
307		Summary {
308	278×	changes: state.changes().map(\|(h, _)\| *h).collect(),
309	278×	latest_ms: state.changes().map(\|(_, m)\| m.hlc.wall_ms).max().unwrap_or(0),
310	278×	conflicted: snapshot.is_conflicted(),
311		}
312	278×	}
313
314		#[cfg(test)]
315		mod tests {
316		use super::*;
317
318		/// What a loaded state really costs against `weight` (oxygenhub#246):
319		/// `OXYHUB_MEASURE_REPO=path cargo test -p oxygenhub --lib
320		/// measures_a_real_state -- --ignored --nocapture`.
321		#[test]
322		#[ignore]
323	0×	fn measures_a_real_state() {
324	0×	fn rss_kib() -> usize {
325	0×	let s = std::fs::read_to_string("/proc/self/status").unwrap();
326	0×	s.lines().find(\|l\| l.starts_with("VmRSS:")).and_then(\|l\| l.split_whitespace().nth(1)).unwrap().parse().unwrap()
327	0×	}
328	0×	let path = std::env::var("OXYHUB_MEASURE_REPO").expect("OXYHUB_MEASURE_REPO");
329	0×	let repo = Repo::open(std::path::Path::new(&path)).unwrap();
330	0×	let view = repo.view().unwrap();
331		// Kept alive, so each load's memory is new, not reused.
332	0×	let cache = StateCache::with_budget(64, usize::MAX);
333	0×	let mut held = Vec::new();
334	0×	for bookmark in view.bookmarks.keys().take(6).cloned().collect::<Vec<_>>() {
335	0×	let heads = repo.heads_of(&view, &bookmark).unwrap();
336	0×	let before = rss_kib();
337	0×	let loaded = cache.load(&repo, &heads).unwrap();
338	0×	let after = rss_kib();
339	0×	let files = loaded.snapshot.files.len();
340	0×	let (mut text, mut lines) = (0usize, 0usize);
341	0×	for f in loaded.snapshot.files.values() {
342	0×	if let oxy_core::state::FileContent::Text { bytes, .. } = &f.content {
343	0×	text += bytes.len();
344	0×	lines += bytes.iter().filter(\|b\| **b == b'\n').count();
345	0×	}
346		}
347	0×	println!(
348		"{bookmark}: {} changes, {files} files, {} KiB text, {lines} lines: weight {} MiB, resident +{} MiB",
349	0×	loaded.state.len(),
350	0×	text >> 10,
351	0×	weight(&loaded) >> 20,
352	0×	(after.saturating_sub(before)) >> 10
353		);
354	0×	held.push(loaded);
355		}
356	0×	}
357
358		#[test]
359	1×	fn caches_by_heads_and_evicts_the_oldest() {
360	1×	let dir = tempfile::tempdir().unwrap();
361	1×	let repo = Repo::init_bare(dir.path()).unwrap();
362	1×	let cache = StateCache::new(2);
363	1×	let a = cache.load(&repo, &[]).unwrap();
364	1×	let b = cache.load(&repo, &[]).unwrap();
365	1×	assert!(Arc::ptr_eq(&a, &b), "same heads, same entry");
366	1×	assert_eq!(cache.len(), 1);
367	1×	let h1 = oxy_core::hash::ObjectRef([1; 32]);
368	1×	let h2 = oxy_core::hash::ObjectRef([2; 32]);
369		// Unknown heads fail to load and are not cached.
370	1×	assert!(cache.load(&repo, &[h1]).is_err());
371	1×	assert_eq!(cache.len(), 1);
372	1×	let _ = (h2, dir);
373	1×	}
374
375		#[test]
376	1×	fn summaries_leave_the_state_cache_alone() {
377	1×	let dir = tempfile::tempdir().unwrap();
378	1×	let repo = Repo::init_bare(dir.path()).unwrap();
379	1×	let cache = StateCache::new(2);
380	1×	let a = cache.summary(&repo, &[]).unwrap();
381	1×	assert_eq!(cache.len(), 0, "a summary does not fill the state cache");
382	1×	assert!(Arc::ptr_eq(&a, &cache.summary(&repo, &[]).unwrap()), "kept");
383	1×	assert!(a.changes.is_empty() && !a.conflicted);
384	1×	assert!(cache.summary(&repo, &[oxy_core::hash::ObjectRef([3; 32])]).is_err());
385	1×	}
386
387		#[test]
388	1×	fn a_size_budget_evicts_the_oldest_and_keeps_the_newest() {
389	1×	let dir = tempfile::tempdir().unwrap();
390	1×	let repo = Repo::init(dir.path()).unwrap();
391	1×	std::fs::write(dir.path().join("a.txt"), "a\n".repeat(1000)).unwrap();
392	1×	let one = repo.commit("one", &[]).unwrap();
393	1×	std::fs::write(dir.path().join("b.txt"), "b\n".repeat(1000)).unwrap();
394	1×	let two = repo.commit("two", &[]).unwrap();
395		// Room for nothing: each load stays, the one before goes.
396	1×	let cache = StateCache::with_budget(64, 0);
397	1×	cache.load(&repo, &[one]).unwrap();
398	1×	assert_eq!(cache.len(), 1, "kept though over the budget");
399	1×	cache.load(&repo, &[two]).unwrap();
400	1×	assert_eq!(cache.len(), 1, "the older one went");
401		// Room for plenty: both stay.
402	1×	let roomy = StateCache::with_budget(64, 1 << 30);
403	1×	roomy.load(&repo, &[one]).unwrap();
404	1×	roomy.load(&repo, &[two]).unwrap();
405	1×	assert_eq!(roomy.len(), 2);
406	1×	}
407
408		#[test]
409	1×	fn briefs_are_worked_out_in_parallel_and_survive_a_restart() {
410	1×	let dir = tempfile::tempdir().unwrap();
411	1×	let repo = Repo::init(dir.path()).unwrap();
412	1×	std::fs::write(dir.path().join("a.txt"), "a\n").unwrap();
413	1×	let one = repo.commit("one", &[]).unwrap();
414	1×	std::fs::write(dir.path().join("b.txt"), "b\n").unwrap();
415	1×	let two = repo.commit("two", &[]).unwrap();
416	1×	let kept = tempfile::tempdir().unwrap();
417	1×	let cache = StateCache::new(8).with_briefs(kept.path().to_path_buf());
418	1×	let got = cache.briefs(&repo, &[vec![one], vec![one, two], vec![]]).unwrap();
419	1×	assert_eq!(got.iter().map(\|b\| b.changes).collect::<Vec<_>>(), vec![1, 2, 0]);
420	1×	assert!(got[1].latest_ms >= got[0].latest_ms && !got[1].conflicted);
421		// A new cache (a restarted hub) reads them back without loading a
422		// state: a repository that no longer has the changes still answers.
423	1×	let gone = tempfile::tempdir().unwrap();
424	1×	let empty = Repo::init(gone.path()).unwrap();
425	1×	let empty_id = empty.id().unwrap();
426	1×	for f in std::fs::read_dir(kept.path()).unwrap() {
427	1×	let f = f.unwrap().path();
428	1×	std::fs::rename(&f, kept.path().join(hex::encode(empty_id))).unwrap();
429	1×	}
430	1×	let restarted = StateCache::new(8).with_briefs(kept.path().to_path_buf());
431	1×	assert_eq!(restarted.brief(&empty, &[two, one]).unwrap(), got[1], "from disk, whatever the heads' order");
432	1×	assert!(StateCache::new(8).brief(&empty, &[one, two]).is_err(), "without the disk it would have to load");
433		// Unknown heads are an error, not a guess.
434	1×	assert!(cache.briefs(&repo, &[vec![one], vec![oxy_core::hash::ObjectRef([9; 32])]]).is_err());
435		// A bare repository, as the hub keeps them, works across threads too.
436	1×	let bare_dir = tempfile::tempdir().unwrap();
437	1×	let bare = Repo::init_bare(bare_dir.path()).unwrap();
438	1×	assert_eq!(cache.briefs(&bare, &[vec![], vec![]]).unwrap().len(), 2);
439	1×	}
440		}

Coverage / crates/hub/src/repos/cache.rs 84.9%

Keyboard