OXYGENHUB
◆ run-one-job oxygenhub / crates / hub / src / repos / cache.rs
9.1 KiB · Rust · last changed Hub memory: two allocator arenas, freed memory returned, cache in /health 27 days ago source blame history raw
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::sync::{Arc, Mutex};
11
12
/// A state and its materialised snapshot.
13
pub struct Loaded {
14
    pub heads: Vec<ChangeHash>,
15
    pub state: State,
16
    pub snapshot: Snapshot,
17
}
18
19
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
20
struct Key {
21
    repo: [u8; 16],
22
    heads: Vec<ChangeHash>,
23
}
24
25
struct Entry {
26
    loaded: Arc<Loaded>,
27
    last_used: u64,
28
    /// Its estimated size in bytes (`weight`).
29
    weight: usize,
30
}
31
32
/// A loaded state's estimated size: its rendered text three times over (the
33
/// text, and the line graphs it was rendered from), plus a kilobyte per
34
/// change. Rough, but proportional: a repository of large files weighs
35
/// more than one of small ones.
36
fn weight(l: &Loaded) -> usize {
37
    let text: usize = l
38
        .snapshot
39
        .files
40
        .values()
41
        .map(|f| match &f.content {
42
            oxy_core::state::FileContent::Text { bytes, .. } => bytes.len(),
43
            _ => 64,
44
        })
45
        .sum();
46
    text.saturating_mul(3).saturating_add(l.state.len().saturating_mul(1024))
47
}
48
49
/// What the state cache may hold, in bytes: OXYHUB_STATE_CACHE_MB, else
50
/// 3 GiB. A count alone let a few large repositories' states grow the hub
51
/// to 17 GB.
52
pub fn budget() -> usize {
53
    std::env::var("OXYHUB_STATE_CACHE_MB").ok().and_then(|v| v.parse::<usize>().ok()).unwrap_or(3 * 1024).saturating_mul(1 << 20)
54
}
55
56
/// What the bookmark lists and a change's "in bookmarks" need of a state:
57
/// small, so many are kept, and computed without filling the state cache.
58
/// A repository with more bookmarks than the state cache holds otherwise
59
/// cycled through it on every such page, so nothing stayed cached
60
/// (metalgear's 128 bookmarks: 7.7 s per change page, and every page of
61
/// every repository slower after it).
62
pub struct Summary {
63
    pub changes: std::collections::HashSet<ChangeHash>,
64
    pub latest_ms: u64,
65
    pub conflicted: bool,
66
}
67
68
/// Summaries kept; each is the state's change set, a few hundred kilobytes
69
/// for a long history.
70
const SUMMARIES: usize = 2048;
71
72
/// Summaries by heads, each with when it was last used, and the clock.
73
type Summaries = (HashMap<Key, (Arc<Summary>, u64)>, u64);
74
75
/// A bounded least-recently-used cache of loaded states.
76
pub struct StateCache {
77
    capacity: usize,
78
    /// Bytes of estimated state it may hold (`budget`).
79
    budget: usize,
80
    inner: Mutex<(HashMap<Key, Entry>, u64)>,
81
    summaries: Mutex<Summaries>,
82
}
83
84
impl StateCache {
85
    pub fn new(capacity: usize) -> StateCache {
86
        StateCache::with_budget(capacity, budget())
87
    }
88
89
    pub fn with_budget(capacity: usize, budget: usize) -> StateCache {
90
        StateCache { capacity: capacity.max(1), budget, inner: Mutex::new((HashMap::new(), 0)), summaries: Mutex::new((HashMap::new(), 0)) }
91
    }
92
93
    /// The summary of the state with these heads: from the state cache when
94
    /// it holds the state, else loaded and dropped, the state cache left as
95
    /// it was.
96
    pub fn summary(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Summary>, oxy_core::repo::RepoError> {
97
        let mut sorted = heads.to_vec();
98
        sorted.sort();
99
        sorted.dedup();
100
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
101
        {
102
            let mut guard = self.summaries.lock().unwrap();
103
            let (map, tick) = &mut *guard;
104
            *tick += 1;
105
            if let Some(e) = map.get_mut(&key) {
106
                e.1 = *tick;
107
                return Ok(e.0.clone());
108
            }
109
        }
110
        let held = self.inner.lock().unwrap().0.get(&key).map(|e| e.loaded.clone());
111
        let summary = Arc::new(match held {
112
            Some(l) => summarise(&l.state, &l.snapshot),
113
            None => {
114
                let state = repo.load_state(&sorted)?;
115
                let snapshot = state.materialize();
116
                summarise(&state, &snapshot)
117
            }
118
        });
119
        let mut guard = self.summaries.lock().unwrap();
120
        let (map, tick) = &mut *guard;
121
        if map.len() >= SUMMARIES {
122
            if let Some(oldest) = map.iter().min_by_key(|(_, e)| e.1).map(|(k, _)| k.clone()) {
123
                map.remove(&oldest);
124
            }
125
        }
126
        map.insert(key, (summary.clone(), *tick));
127
        Ok(summary)
128
    }
129
130
    /// The state with these heads, loaded once per hub process.
131
    pub fn load(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Loaded>, oxy_core::repo::RepoError> {
132
        let mut sorted = heads.to_vec();
133
        sorted.sort();
134
        sorted.dedup();
135
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
136
        {
137
            let mut guard = self.inner.lock().unwrap();
138
            let (map, tick) = &mut *guard;
139
            *tick += 1;
140
            if let Some(e) = map.get_mut(&key) {
141
                e.last_used = *tick;
142
                return Ok(e.loaded.clone());
143
            }
144
        }
145
        let state = repo.load_state(&sorted)?;
146
        let snapshot = state.materialize();
147
        let loaded = Arc::new(Loaded { heads: sorted, state, snapshot });
148
        let w = weight(&loaded);
149
        let mut guard = self.inner.lock().unwrap();
150
        let (map, tick) = &mut *guard;
151
        // The oldest go until it fits by count and by size; the new one
152
        // stays even when it alone is over the budget.
153
        let budget = self.budget;
154
        let mut total: usize = map.values().map(|e| e.weight).sum::<usize>() + w;
155
        while !map.is_empty() && (map.len() >= self.capacity || total > budget) {
156
            let Some(oldest) = map.iter().min_by_key(|(_, e)| e.last_used).map(|(k, _)| k.clone()) else {
157
                break;
158
            };
159
            if let Some(e) = map.remove(&oldest) {
160
                total -= e.weight;
161
            }
162
        }
163
        map.insert(key, Entry { loaded: loaded.clone(), last_used: *tick, weight: w });
164
        Ok(loaded)
165
    }
166
167
    /// What it holds: loaded states, their estimated bytes, and summaries.
168
    pub fn stats(&self) -> (usize, usize, usize) {
169
        let (entries, weight) = {
170
            let guard = self.inner.lock().unwrap();
171
            (guard.0.len(), guard.0.values().map(|e| e.weight).sum())
172
        };
173
        (entries, weight, self.summaries.lock().unwrap().0.len())
174
    }
175
176
    pub fn len(&self) -> usize {
177
        self.inner.lock().unwrap().0.len()
178
    }
179
180
    pub fn is_empty(&self) -> bool {
181
        self.len() == 0
182
    }
183
}
184
185
fn summarise(state: &State, snapshot: &Snapshot) -> Summary {
186
    Summary {
187
        changes: state.changes().map(|(h, _)| *h).collect(),
188
        latest_ms: state.changes().map(|(_, m)| m.hlc.wall_ms).max().unwrap_or(0),
189
        conflicted: snapshot.is_conflicted(),
190
    }
191
}
192
193
#[cfg(test)]
194
mod tests {
195
    use super::*;
196
197
    #[test]
198
    fn caches_by_heads_and_evicts_the_oldest() {
199
        let dir = tempfile::tempdir().unwrap();
200
        let repo = Repo::init_bare(dir.path()).unwrap();
201
        let cache = StateCache::new(2);
202
        let a = cache.load(&repo, &[]).unwrap();
203
        let b = cache.load(&repo, &[]).unwrap();
204
        assert!(Arc::ptr_eq(&a, &b), "same heads, same entry");
205
        assert_eq!(cache.len(), 1);
206
        let h1 = oxy_core::hash::ObjectRef([1; 32]);
207
        let h2 = oxy_core::hash::ObjectRef([2; 32]);
208
        // Unknown heads fail to load and are not cached.
209
        assert!(cache.load(&repo, &[h1]).is_err());
210
        assert_eq!(cache.len(), 1);
211
        let _ = (h2, dir);
212
    }
213
214
    #[test]
215
    fn summaries_leave_the_state_cache_alone() {
216
        let dir = tempfile::tempdir().unwrap();
217
        let repo = Repo::init_bare(dir.path()).unwrap();
218
        let cache = StateCache::new(2);
219
        let a = cache.summary(&repo, &[]).unwrap();
220
        assert_eq!(cache.len(), 0, "a summary does not fill the state cache");
221
        assert!(Arc::ptr_eq(&a, &cache.summary(&repo, &[]).unwrap()), "kept");
222
        assert!(a.changes.is_empty() && !a.conflicted);
223
        assert!(cache.summary(&repo, &[oxy_core::hash::ObjectRef([3; 32])]).is_err());
224
    }
225
226
    #[test]
227
    fn a_size_budget_evicts_the_oldest_and_keeps_the_newest() {
228
        let dir = tempfile::tempdir().unwrap();
229
        let repo = Repo::init(dir.path()).unwrap();
230
        std::fs::write(dir.path().join("a.txt"), "a\n".repeat(1000)).unwrap();
231
        let one = repo.commit("one", &[]).unwrap();
232
        std::fs::write(dir.path().join("b.txt"), "b\n".repeat(1000)).unwrap();
233
        let two = repo.commit("two", &[]).unwrap();
234
        // Room for nothing: each load stays, the one before goes.
235
        let cache = StateCache::with_budget(64, 0);
236
        cache.load(&repo, &[one]).unwrap();
237
        assert_eq!(cache.len(), 1, "kept though over the budget");
238
        cache.load(&repo, &[two]).unwrap();
239
        assert_eq!(cache.len(), 1, "the older one went");
240
        // Room for plenty: both stay.
241
        let roomy = StateCache::with_budget(64, 1 << 30);
242
        roomy.load(&repo, &[one]).unwrap();
243
        roomy.load(&repo, &[two]).unwrap();
244
        assert_eq!(roomy.len(), 2);
245
    }
246
}