sva_engine/cache/
tier.rs

1// Concern: memory over a disk or over nothing, and the one passage between them | Non-concern: what memory keeps, the disk's medium | IO: (keys, needs) -> answers, samples; persist() -> commits
2
3use std::sync::Arc;
4use std::sync::atomic::{AtomicU64, Ordering};
5
6use sva_formula::Hash;
7use sva_samples::{Buffer, Extent};
8
9use super::memory::{Counters, DEFAULT_CACHE_BYTES, Known, Memory};
10use super::persist::{Backend, Persisted, Store};
11use super::watch::{Call, EngineStats, Kind, Phase, Watch};
12
13/// No disk: memory over nothing never waits.
14pub enum Nothing {}
15
16impl Backend for Nothing {
17    type Lock = ();
18
19    async fn lock(&self) -> Result<(), String> {
20        match *self {}
21    }
22
23    async fn get(&self, _: &str) -> Result<Option<Vec<u8>>, String> {
24        match *self {}
25    }
26
27    async fn get_range(&self, _: &str, _: u64, _: u64) -> Result<Option<Vec<u8>>, String> {
28        match *self {}
29    }
30
31    async fn put(&self, _: &str, _: &[u8]) -> Result<(), String> {
32        match *self {}
33    }
34
35    async fn delete(&self, _: &str) -> Result<bool, String> {
36        match *self {}
37    }
38
39    async fn list(&self) -> Result<Vec<(String, u64)>, String> {
40        match *self {}
41    }
42
43    async fn staging(&self) -> Result<Nothing, String> {
44        match *self {}
45    }
46
47    async fn rename(&self, _: &str, _: &Nothing) -> Result<bool, String> {
48        match *self {}
49    }
50}
51
52/// The most disk reads one fetch makes.
53pub const FETCH_READS: usize = 4;
54
55/// What one fetch handed out, and the needs its reads did not reach.
56#[derive(Default)]
57pub(crate) struct Fetched {
58    pub(crate) handed: Vec<(Hash, Vec<Arc<Buffer>>)>,
59    pub(crate) left: Vec<(Hash, Extent)>,
60}
61
62/// Memory over a disk, or over nothing. Evaluation talks to memory alone; only memory decides
63/// whether a lookup, a read or a write passes through to the disk.
64pub struct Tier<B = Nothing> {
65    memory: Memory,
66    disk: Option<Store<B>>,
67    /// Memory's clock where the disk last heard of its reads.
68    reported: AtomicU64,
69    watch: Arc<Watch>,
70}
71
72impl Default for Tier {
73    fn default() -> Tier {
74        Tier::new(DEFAULT_CACHE_BYTES)
75    }
76}
77
78impl Tier {
79    pub fn new(max_bytes: u64) -> Tier {
80        Tier::alone(max_bytes)
81    }
82}
83
84impl<B: Backend> Tier<B> {
85    pub fn alone(max_bytes: u64) -> Tier<B> {
86        Tier::held(Memory::holding(max_bytes), None)
87    }
88
89    pub fn over(disk: Store<B>, max_bytes: u64) -> Tier<B> {
90        Tier::held(Memory::over_disk(max_bytes), Some(disk))
91    }
92
93    fn held(memory: Memory, disk: Option<Store<B>>) -> Tier<B> {
94        Tier {
95            watch: Arc::new(Watch::over(memory.clone())),
96            memory,
97            disk,
98            reported: AtomicU64::new(0),
99        }
100    }
101
102    /// In place of the last subscriber: told at once, then by each call, on its own thread.
103    pub fn subscribe(&self, hears: impl Fn(&EngineStats) + Send + Sync + 'static) {
104        self.watch.subscribe(Arc::new(hears));
105    }
106
107    pub fn unsubscribe(&self) {
108        self.watch.unsubscribe();
109    }
110
111    pub fn call(&self, kind: Kind, target: &str, phase: Phase) -> Call {
112        Call::begin(&self.watch, kind, target, phase)
113    }
114
115    pub fn disk(&self) -> Option<&Store<B>> {
116        self.disk.as_ref()
117    }
118
119    pub(crate) fn memory(&self) -> &Memory {
120        &self.memory
121    }
122
123    pub(crate) fn begin(&self) -> u64 {
124        self.memory.begin()
125    }
126
127    /// `key` answered: from memory, else off the disk and resident from then on.
128    pub(crate) async fn lookup(&self, key: Hash, round: u64) {
129        if !matches!(self.memory.answer(key, round), Known::Unknown) {
130            return;
131        }
132        let found = match &self.disk {
133            Some(disk) => {
134                self.memory.count(|c| c.disk_lookups += 1);
135                disk.lookup(key).await
136            }
137            None => None,
138        };
139        match found {
140            Some(head) => self.memory.promote(head, round),
141            None => self.memory.miss(key, round),
142        }
143    }
144
145    /// What each need asks of a node memory holds, and off the disk what memory lacks, at most
146    /// `FETCH_READS` reads a call, the rest left for the next. Memory first writes back.
147    pub(crate) async fn fetch(&self, needs: &[(Hash, Extent)]) -> Fetched {
148        self.write_back().await;
149        let mut out = Fetched::default();
150        let mut reads = 0;
151        for (key, over) in needs {
152            let (mut parts, lacks) = self.memory.resident(*key, *over);
153            if let (Some(disk), Some((head, gap))) = (&self.disk, lacks) {
154                if reads == FETCH_READS {
155                    out.left.push((*key, *over));
156                    continue;
157                }
158                reads += 1;
159                let read = disk.read(&head, gap).await;
160                let samples = read.iter().flatten().map(|b| b.len() * b.width());
161                let bytes = (samples.sum::<usize>() * size_of::<f64>()) as u64;
162                self.memory.count(|c| {
163                    c.disk_reads += 1;
164                    c.disk_read_bytes += bytes;
165                });
166                match read {
167                    Some(read) => parts.extend(self.memory.promote_samples(*key, read)),
168                    None => self.memory.forget(*key),
169                }
170            }
171            if !parts.is_empty() {
172                out.handed.push((*key, parts));
173            }
174        }
175        out
176    }
177
178    /// What memory let go of while not yet on the disk, written there; a failure stops every
179    /// write until the next persist.
180    async fn write_back(&self) {
181        if let Some(disk) = &self.disk {
182            let pending = self.memory.pending();
183            if let Err((why, left)) = self.written(disk, pending).await {
184                self.memory.failed(why, left);
185            }
186        }
187    }
188
189    async fn written(
190        &self,
191        disk: &Store<B>,
192        mut pending: Vec<super::memory::Writeback>,
193    ) -> Result<(), (String, Vec<super::memory::Writeback>)> {
194        pending.reverse();
195        while let Some(back) = pending.pop() {
196            let mut staged = Ok(());
197            for part in &back.parts {
198                staged = disk.stage(back.key, part).await;
199                if staged.is_err() {
200                    break;
201                }
202            }
203            let staged = match staged {
204                Ok(()) => disk.stage_meta(back.key, &back.head).await,
205                Err(why) => Err(why),
206            };
207            if let Err(why) = staged {
208                pending.push(back);
209                pending.reverse();
210                return Err((why, pending));
211            }
212            self.memory.written();
213        }
214        Ok(())
215    }
216
217    /// Every node memory holds that the disk lacks, committed; each it answered since, used.
218    pub async fn persist(&self) -> Result<Persisted, String> {
219        let Some(disk) = &self.disk else {
220            return Ok(Persisted::default());
221        };
222        let call = self.call(Kind::Persist, "", Phase::Storing);
223        let (read, now) = self
224            .memory
225            .read_since(self.reported.load(Ordering::Relaxed));
226        read.into_iter().for_each(|key| disk.used(key));
227        self.reported.fetch_max(now, Ordering::Relaxed);
228        self.memory.flush();
229        let pending = self.memory.pending();
230        if let Err((why, left)) = self.written(disk, pending).await {
231            self.memory.failed(why.clone(), left);
232            return Err(why);
233        }
234        let done = disk.persist().await?;
235        self.memory.committed();
236        call.end(Phase::Done);
237        Ok(done)
238    }
239
240    pub fn counters(&self) -> Counters {
241        self.memory.counters()
242    }
243
244    pub fn max_bytes(&self) -> u64 {
245        self.memory.max_bytes()
246    }
247
248    pub fn set_max_bytes(&self, max_bytes: u64) {
249        self.memory.set_max_bytes(max_bytes);
250    }
251
252    pub fn bytes(&self) -> u64 {
253        self.memory.bytes()
254    }
255
256    pub fn entries(&self) -> usize {
257        self.memory.entries()
258    }
259
260    pub fn evictions(&self) -> u64 {
261        self.memory.counters().evictions()
262    }
263
264    pub fn holds(&self, key: Hash) -> bool {
265        self.memory.holds(key)
266    }
267
268    pub fn mark_every(&self) -> usize {
269        self.memory.mark_every()
270    }
271
272    pub fn set_mark_every(&self, samples: usize) {
273        self.memory.set_mark_every(samples);
274    }
275}
276
277/// A future over memory alone, finished in its one poll.
278pub(crate) fn now<F: Future>(future: F) -> F::Output {
279    let mut future = std::pin::pin!(future);
280    let waker = std::task::Waker::noop();
281    match future
282        .as_mut()
283        .poll(&mut std::task::Context::from_waker(waker))
284    {
285        std::task::Poll::Ready(out) => out,
286        std::task::Poll::Pending => unreachable!("memory over nothing never waits"),
287    }
288}
sva_engine/cache/tier.rs

sva_engine/cache/
tier.rs
