Skip to main content

sva_engine/cache/
memory.rs

1// Concern: the memory tier: each resident value and its node under one cap, what it evicts and writes back, its misses | Non-concern: the disk | IO: (key) -> held, answered; (key, samples, node) -> kept
2
3#[cfg(test)]
4mod chains;
5
6use std::cell::RefCell;
7use std::collections::{HashMap, HashSet};
8use std::sync::{Arc, Mutex, MutexGuard};
9
10use sva_formula::Hash;
11use sva_samples::{Buffer, Extent, Label};
12
13use super::stats::{Outcome, Recording};
14use super::stored::{Header, Samples};
15use super::{Entry, Expected, Payload, PayloadKind, Run, Stored, joined};
16
17pub const DEFAULT_CACHE_BYTES: u64 = 2 << 30;
18
19/// Samples between two states a run keeps, so a reader resumes from one at most this far back.
20pub const DEFAULT_MARK_EVERY: usize = 16_384;
21
22/// What passed between memory and the disk beneath it, and what memory let go.
23#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
24pub struct Counters {
25    pub disk_lookups: u64,
26    pub disk_reads: u64,
27    pub disk_read_bytes: u64,
28    /// Disk answers made resident: a header looked up, or samples read.
29    pub promotions: u64,
30    pub writebacks: u64,
31    /// Every entry's hits, summed: each answer of a node or a value memory held.
32    pub hits: u64,
33    pub probation_evictions: u64,
34    pub protected_evictions: u64,
35}
36
37impl Counters {
38    pub fn since(self, then: Counters) -> Counters {
39        Counters {
40            disk_lookups: self.disk_lookups - then.disk_lookups,
41            disk_reads: self.disk_reads - then.disk_reads,
42            disk_read_bytes: self.disk_read_bytes - then.disk_read_bytes,
43            promotions: self.promotions - then.promotions,
44            writebacks: self.writebacks - then.writebacks,
45            hits: self.hits - then.hits,
46            probation_evictions: self.probation_evictions - then.probation_evictions,
47            protected_evictions: self.protected_evictions - then.protected_evictions,
48        }
49    }
50
51    pub fn evictions(self) -> u64 {
52        self.probation_evictions + self.protected_evictions
53    }
54}
55
56/// What a keep states of the node its samples answer: whether it is the target, whether two
57/// or more values read it, and whether a stateful run wrote it.
58#[derive(Clone, Copy, Debug)]
59pub(crate) struct Facts {
60    pub(crate) target: bool,
61    pub(crate) shared: bool,
62    pub(crate) stateful: bool,
63}
64
65/// What one keep sends memory: the samples a value computed, and the node they answer.
66pub(crate) struct Keep<'a> {
67    pub(crate) samples: Option<Payload>,
68    pub(crate) label: Option<&'a Label>,
69    pub(crate) slot: Option<Hash>,
70    pub(crate) node: Option<(Stored, Offered, Facts)>,
71}
72
73/// The protected segment holds at most this share of the cap, so probation always has a
74/// quarter for a new entry to earn its first hit in.
75const PROTECTED: (u64, u64) = (3, 4);
76
77#[derive(Clone, Copy, Debug, PartialEq, Eq)]
78pub(crate) enum Kept {
79    Held,
80    Replaced,
81    Refused,
82}
83
84/// What memory answers a node's key with: a miss holds for the round it was met in and later
85/// ones begun before it, so another holder's commit is found once a new round begins.
86pub(crate) enum Known {
87    Hit(Arc<Stored>),
88    Miss,
89    Unknown,
90}
91
92/// Where a kept node's sample `n` is.
93#[derive(Clone, Debug, PartialEq)]
94pub(crate) enum Offered {
95    /// The value held under the node's own key: a run's through each segment before it.
96    Own,
97    Moves {
98        of: Hash,
99        by: i64,
100    },
101    /// Samples no value holds, the node's own.
102    Held(Vec<Arc<Buffer>>),
103}
104
105pub(crate) struct Writeback {
106    pub(crate) key: Hash,
107    pub(crate) head: Header,
108    pub(crate) parts: Vec<Arc<Buffer>>,
109}
110
111/// A value's samples, and the node they were kept or read off the disk as.
112struct Item {
113    payload: Option<Payload>,
114    label: Option<Label>,
115    slot: Option<Hash>,
116    node: Option<Node>,
117}
118
119struct Node {
120    source: Source,
121    /// Holds what the disk lacks: each keep of more samples sets it again.
122    dirty: bool,
123    bound: bool,
124}
125
126enum Source {
127    Disk {
128        head: Box<Header>,
129        chunks: Vec<Arc<Buffer>>,
130        /// The node as a value told it since: where the samples the disk lacks of it are.
131        more: Option<(Box<Stored>, Offered)>,
132        /// Whether memory sent the disk more of it, so its header names a layout now gone.
133        rewritten: bool,
134    },
135    Offered {
136        stored: Box<Stored>,
137        offered: Offered,
138    },
139}
140
141impl Node {
142    fn stored(&self) -> &Stored {
143        match &self.source {
144            Source::Disk { head, .. } => head.stored(),
145            Source::Offered { stored, .. } => stored,
146        }
147    }
148
149    fn offered(&self) -> Option<&Offered> {
150        match &self.source {
151            Source::Offered { offered, .. } => Some(offered),
152            Source::Disk { .. } => None,
153        }
154    }
155
156    fn head(&self) -> Option<&Header> {
157        match &self.source {
158            Source::Disk { head, .. } => Some(head),
159            Source::Offered { .. } => None,
160        }
161    }
162
163    fn chunks(&self) -> &[Arc<Buffer>] {
164        match &self.source {
165            Source::Disk { chunks, .. }
166            | Source::Offered {
167                offered: Offered::Held(chunks),
168                ..
169            } => chunks,
170            Source::Offered { .. } => &[],
171        }
172    }
173}
174
175struct Held {
176    item: Item,
177    read: u64,
178    since: u64,
179    hit_round: u64,
180    protected: bool,
181}
182
183impl Held {
184    fn admitted(item: Item, at: u64) -> Held {
185        Held {
186            item,
187            read: at,
188            since: at,
189            hit_round: 0,
190            protected: false,
191        }
192    }
193
194    fn bytes(&self) -> u64 {
195        let value = self.item.payload.as_ref().map_or(0, |p| p.bytes() as u64);
196        let node = self.item.node.as_ref().map_or(&[][..], Node::chunks);
197        value + node.iter().map(|c| planes(c)).sum::<u64>()
198    }
199}
200
201fn planes(b: &Buffer) -> u64 {
202    (b.len() * b.width() * size_of::<f64>()) as u64
203}
204
205#[derive(Default)]
206struct State {
207    entries: HashMap<Hash, Held>,
208    slots: HashMap<Hash, Hash>,
209    misses: HashMap<Hash, u64>,
210    pending: Vec<Writeback>,
211    bytes: u64,
212    max_bytes: u64,
213    clock: u64,
214    round: u64,
215    mark_every: usize,
216    counters: Counters,
217    disk: bool,
218    /// Writes that failed, and why the latest did: none is tried again until a persist.
219    failed: (u64, Option<String>),
220    links: RefCell<Links>,
221    #[cfg(test)]
222    walked: std::cell::Cell<u64>,
223}
224
225/// Where a chain of moved nodes ends and the shift there; a chain that loops ends nowhere.
226#[derive(Clone, Copy)]
227struct Link {
228    end: Hash,
229    by: i64,
230    unbound: bool,
231    looped: bool,
232}
233
234/// Each moved node's link, a cache every change to `entries` evicts what it found through.
235#[derive(Default)]
236struct Links {
237    /// Each link, with the key it was found through.
238    found: HashMap<Hash, (Link, Hash)>,
239    through: HashMap<Hash, HashSet<Hash>>,
240}
241
242impl Links {
243    fn keep(&mut self, at: Hash, next: Hash, link: Link) {
244        self.found.insert(at, (link, next));
245        self.through.entry(next).or_default().insert(at);
246    }
247
248    fn evict(&mut self, key: Hash) {
249        let mut open = vec![key];
250        while let Some(at) = open.pop() {
251            if let Some((_, next)) = self.found.remove(&at)
252                && let Some(readers) = self.through.get_mut(&next)
253            {
254                readers.remove(&at);
255                if readers.is_empty() {
256                    self.through.remove(&next);
257                }
258            }
259            open.extend(self.through.remove(&at).into_iter().flatten());
260        }
261    }
262}
263
264impl State {
265    fn tick(&mut self) -> u64 {
266        self.clock += 1;
267        self.clock
268    }
269
270    fn node(&self, key: Hash) -> Option<&Node> {
271        self.entries.get(&key)?.item.node.as_ref()
272    }
273
274    fn source_mut(&mut self, key: Hash) -> Option<&mut Source> {
275        let node = self.entries.get_mut(&key)?.item.node.as_mut()?;
276        Some(&mut node.source)
277    }
278
279    fn stands(&self, key: Hash) -> bool {
280        let bound = |at: &Hash| self.node(*at).is_some_and(|node| node.bound);
281        self.doomed(key).iter().any(bound)
282    }
283
284    fn on_disk(&self, key: Hash, payload: &Payload) -> bool {
285        let Some(head) = self.node(key).and_then(Node::head) else {
286            return false;
287        };
288        let stored = head.stored();
289        match payload {
290            Payload::Segments(parts) => parts.iter().all(|part| stored.holds(part.extent())),
291            Payload::Run(run) => stored.holds(run.samples.extent()),
292            Payload::Frames(_) => false,
293        }
294    }
295
296    /// A run's segment, then each before it, ending where the one after it starts.
297    fn segments(&self, key: Hash) -> Vec<(Hash, Extent)> {
298        let (mut out, mut at, mut end) = (Vec::new(), Some(key), i64::MAX);
299        while let Some(k) = at {
300            out.push((k, Extent::new(i64::MIN, end)));
301            let held = self
302                .entries
303                .get(&k)
304                .and_then(|held| held.item.payload.as_ref());
305            let Some(Payload::Run(run)) = held else {
306                break;
307            };
308            (at, end) = (run.parent, run.samples.start);
309        }
310        out
311    }
312
313    fn reads(&self, key: Hash) -> Vec<Hash> {
314        match self.node(key).and_then(Node::offered) {
315            None | Some(Offered::Held(_)) => Vec::new(),
316            Some(Offered::Own) => self
317                .segments(key)
318                .into_iter()
319                .skip(1)
320                .map(|(k, _)| k)
321                .collect(),
322            Some(Offered::Moves { of, .. }) => vec![*of],
323        }
324    }
325
326    fn doomed(&self, key: Hash) -> Vec<Hash> {
327        let mut readers: HashMap<Hash, Vec<Hash>> = HashMap::new();
328        for at in self.entries.keys() {
329            for read in self.reads(*at) {
330                readers.entry(read).or_default().push(*at);
331            }
332        }
333        let mut out = vec![key];
334        let mut k = 0;
335        while k < out.len() {
336            for at in readers.get(&out[k]).into_iter().flatten() {
337                if !out.contains(at) {
338                    out.push(*at);
339                }
340            }
341            k += 1;
342        }
343        out
344    }
345
346    /// `key` and every node reading its samples gone, each dirty node flushed first: a later
347    /// keep brings back what it keeps on computing.
348    fn remove(&mut self, key: Hash) -> bool {
349        if !self.entries.contains_key(&key) {
350            return false;
351        }
352        let doomed = self.doomed(key);
353        let extending = self.extending(&doomed);
354        for at in doomed.iter().chain(&extending) {
355            self.flush(*at);
356        }
357        for at in extending {
358            if let Some(Source::Disk { more, .. }) = self.source_mut(at) {
359                *more = None;
360            }
361        }
362        for at in doomed {
363            self.discard(at);
364        }
365        true
366    }
367
368    /// Each node the disk holds whose samples past it are any of `going`'s, moved: it stays, as
369    /// the disk holds it.
370    fn extending(&self, going: &[Hash]) -> Vec<Hash> {
371        let moves = |held: &Held| match &held.item.node.as_ref()?.source {
372            Source::Disk {
373                more: Some((_, Offered::Moves { of, .. })),
374                ..
375            } => Some(*of),
376            _ => None,
377        };
378        let reading = self.entries.iter().filter(|(at, held)| {
379            !going.contains(at) && moves(held).is_some_and(|of| going.contains(&of))
380        });
381        reading.map(|(at, _)| *at).collect()
382    }
383
384    fn discard(&mut self, key: Hash) {
385        let Some(gone) = self.entries.remove(&key) else {
386            return;
387        };
388        self.links.get_mut().evict(key);
389        self.bytes -= gone.bytes();
390        if let Some(slot) = gone.item.slot
391            && self.slots.get(&slot) == Some(&key)
392        {
393            self.slots.remove(&slot);
394        }
395    }
396
397    fn unnoded(&mut self, key: Hash) {
398        let Some(held) = self.entries.get_mut(&key) else {
399            return;
400        };
401        if held.item.payload.is_none() {
402            return self.discard(key);
403        }
404        let before = held.bytes();
405        held.item.node = None;
406        self.bytes = self.bytes - before + held.bytes();
407        self.links.get_mut().evict(key);
408    }
409
410    /// A dirty node's header and what memory holds of it, on their way to the disk.
411    fn flush(&mut self, key: Hash) {
412        let Some(Node {
413            dirty: dirty @ true,
414            ..
415        }) = self
416            .entries
417            .get_mut(&key)
418            .and_then(|held| held.item.node.as_mut())
419        else {
420            return;
421        };
422        *dirty = false;
423        let back = self.written(key);
424        if let (Some(_), Some(Source::Disk { rewritten, .. })) = (&back, self.source_mut(key)) {
425            *rewritten = true;
426        }
427        if let Some(back) = back {
428            self.pending.push(back);
429        }
430    }
431
432    /// A node no disk holds, whole; of one the disk holds, only the samples it lacks, and
433    /// nothing where it lacks none or refers to them where they are.
434    fn written(&self, key: Hash) -> Option<Writeback> {
435        let (stored, offered, on_disk) = match &self.node(key)?.source {
436            Source::Offered { stored, offered } => (stored, offered, None),
437            Source::Disk {
438                head,
439                more: Some((stored, offered)),
440                ..
441            } => (stored, offered, Some(head)),
442            Source::Disk { more: None, .. } => return None,
443        };
444        let head = |samples| Header::new((**stored).clone(), samples);
445        let referred = match offered {
446            Offered::Moves { of, by } => match self.referred(*of, *by) {
447                Some(found) => Some(found),
448                None if self.unbound(*of) => None,
449                None => return None,
450            },
451            _ => None,
452        };
453        let refers = |to: (Hash, i64)| match on_disk.map(|head| head.samples()) {
454            Some(Samples::Entry { file, shift, .. }) => (*file, -*shift) == to,
455            _ => false,
456        };
457        let own = on_disk.is_some_and(|head| match head.samples() {
458            Samples::Entry { file, .. } | Samples::Staged { file, .. } => *file == key,
459            Samples::None | Samples::Of { .. } => false,
460        });
461        match referred {
462            Some(to) if refers(to) => None,
463            Some((of, by)) if !own => Some(Writeback {
464                key,
465                head: head(Samples::Of { key: of, by }),
466                parts: Vec::new(),
467            }),
468            _ => {
469                let parts = self.offered_parts(key, offered, Extent::EVERYWHERE);
470                let mut parts = parts.unwrap_or_default();
471                if let Some(on_disk) = on_disk {
472                    parts.retain(|part| !on_disk.stored().holds(part.extent()));
473                    if parts.is_empty() {
474                        return None;
475                    }
476                }
477                Some(Writeback {
478                    key,
479                    head: head(Samples::None),
480                    parts,
481                })
482            }
483        }
484    }
485
486    fn moving(&self, key: Hash) -> Option<(Hash, i64, bool)> {
487        let node = self.node(key)?;
488        match node.offered()? {
489            Offered::Moves { of, by } => Some((*of, *by, node.bound)),
490            _ => None,
491        }
492    }
493
494    /// Walked from `key` to the first link already found, each link walked kept.
495    fn link(&self, key: Hash) -> Link {
496        let (mut walked, mut seen, mut at) = (Vec::new(), HashSet::new(), key);
497        let mut link = loop {
498            if let Some((held, _)) = self.links.borrow().found.get(&at) {
499                break *held;
500            }
501            let looped = !seen.insert(at);
502            match self.moving(at) {
503                Some((of, by, bound)) if !looped => {
504                    #[cfg(test)]
505                    self.walked.set(self.walked.get() + 1);
506                    walked.push((at, of, by, !bound));
507                    at = of;
508                }
509                moving => {
510                    break Link {
511                        end: at,
512                        by: 0,
513                        unbound: false,
514                        looped: moving.is_some(),
515                    };
516                }
517            }
518        };
519        for (at, of, by, unbound) in walked.into_iter().rev() {
520            link.by += by;
521            link.unbound |= unbound;
522            if !link.looped {
523                self.links.borrow_mut().keep(at, of, link);
524            }
525        }
526        link
527    }
528
529    /// The node `stored` names, as `offered` says, in place of the last one under its key, and,
530    /// `sole`, with no samples to keep, under its slot; whether memory writes it to the disk. A
531    /// node the disk holds stays as the disk holds it, and is written what it lacks.
532    fn noded(
533        &mut self,
534        stored: Stored,
535        offered: Offered,
536        (slot, sole, facts): (Option<Hash>, bool, Facts),
537    ) -> bool {
538        let key = stored.key;
539        if let Some(node) = self
540            .entries
541            .get_mut(&key)
542            .and_then(|h| h.item.node.as_mut())
543            && let Source::Disk { more, .. } = &mut node.source
544        {
545            *more = Some((Box::new(stored), offered));
546            node.dirty = true;
547            return true;
548        }
549        let bound = self.disk && writes(&stored, facts);
550        let read = self.tick();
551        let same = self.node(key).is_some_and(|node| match &node.source {
552            Source::Offered {
553                stored: had,
554                offered: was,
555            } => **had == stored && *was == offered,
556            Source::Disk { .. } => false,
557        });
558        let node = Node {
559            source: Source::Offered {
560                stored: Box::new(stored),
561                offered,
562            },
563            dirty: bound,
564            bound,
565        };
566        match self.entries.get_mut(&key) {
567            Some(held) if same => {
568                let node = held.item.node.as_mut().expect("the node held");
569                node.dirty |= node.bound;
570                held.read = read;
571            }
572            Some(held) => {
573                let before = held.bytes();
574                held.item.node = Some(node);
575                held.item.slot = slot;
576                held.read = read;
577                let after = held.bytes();
578                self.bytes = self.bytes - before + after;
579                self.links.get_mut().evict(key);
580            }
581            None => {
582                let item = Item {
583                    payload: None,
584                    label: None,
585                    slot,
586                    node: Some(node),
587                };
588                let held = Held::admitted(item, read);
589                self.bytes += held.bytes();
590                self.admit(key, held);
591            }
592        }
593        self.misses.remove(&key);
594        if let Some(last) = slot
595            .filter(|_| sole)
596            .and_then(|slot| self.slots.insert(slot, key))
597            && last != key
598        {
599            self.remove(last);
600        }
601        bound
602    }
603
604    fn admit(&mut self, key: Hash, held: Held) -> Option<Held> {
605        self.links.get_mut().evict(key);
606        self.entries.insert(key, held)
607    }
608
609    fn foot(&self, key: Hash) -> Option<(Hash, i64)> {
610        let link = self.link(key);
611        (!link.looped).then_some((link.end, link.by))
612    }
613
614    /// A link or its end is no node the disk is written, or is only a value.
615    fn unbound(&self, key: Hash) -> bool {
616        let link = self.link(key);
617        let end = self.entries.get(&link.end).map(|held| &held.item.node);
618        link.unbound || matches!(end, Some(None | Some(Node { bound: false, .. })))
619    }
620
621    fn referred(&self, of: Hash, by: i64) -> Option<(Hash, i64)> {
622        let link = self.link(of);
623        if link.looped || link.unbound {
624            return None;
625        }
626        let more = link.by;
627        let node = self.node(link.end)?;
628        match (node.head().map(Header::samples), node.bound) {
629            (Some(Samples::Entry { file, shift, .. } | Samples::Staged { file, shift, .. }), _) => {
630                Some((*file, by + more - shift))
631            }
632            (Some(_), _) | (None, false) => None,
633            (None, true) => Some((link.end, by + more)),
634        }
635    }
636
637    fn parts(&self, key: Hash) -> Option<Vec<(Arc<Buffer>, Extent)>> {
638        self.entries.get(&key)?;
639        let mut out = Vec::new();
640        for (k, within) in self.segments(key) {
641            let parts = match self
642                .entries
643                .get(&k)
644                .and_then(|held| held.item.payload.as_ref())
645            {
646                Some(Payload::Segments(parts)) => parts.clone(),
647                Some(Payload::Run(run)) => vec![Arc::clone(&run.samples)],
648                _ => Vec::new(),
649            };
650            let met = |part: Arc<Buffer>| {
651                let met = part.extent().intersect(within);
652                (!met.is_empty()).then_some((part, met))
653            };
654            out.extend(parts.into_iter().filter_map(met));
655        }
656        Some(out)
657    }
658
659    fn own(&self, key: Hash, over: Extent) -> Option<Vec<Arc<Buffer>>> {
660        let met = self.parts(key)?.into_iter();
661        let met = met.filter(|(_, held)| !held.intersect(over).is_empty());
662        Some(met.map(|(part, held)| clipped(&part, held)).collect())
663    }
664
665    fn offered_parts(
666        &self,
667        key: Hash,
668        offered: &Offered,
669        over: Extent,
670    ) -> Option<Vec<Arc<Buffer>>> {
671        let (parts, by): (Vec<Arc<Buffer>>, i64) = match offered {
672            Offered::Own => (self.own(key, over)?, 0),
673            Offered::Moves { of, by } => {
674                let (foot, more) = self.foot(*of)?;
675                (self.resident(foot, over.shifted(by + more))?.0, by + more)
676            }
677            Offered::Held(parts) => (parts.clone(), 0),
678        };
679        let over = over.shifted(by);
680        let meets = |b: &&Arc<Buffer>| !b.extent().intersect(over).is_empty();
681        Some(parts.iter().filter(meets).map(|p| moved(p, -by)).collect())
682    }
683
684    /// What memory holds of `key` meeting `over`, and for a node off the disk, what of `over`
685    /// the disk holds that memory lacks, with its layout.
686    fn resident(&self, key: Hash, over: Extent) -> Option<Resident> {
687        let source = match &self.entries.get(&key)?.item.node {
688            None => return Some((self.own(key, over)?, None)),
689            Some(node) => &node.source,
690        };
691        match source {
692            Source::Offered { offered, .. } => {
693                Some((self.offered_parts(key, offered, over)?, None))
694            }
695            Source::Disk { head, chunks, .. } => {
696                let meets = |b: &&Arc<Buffer>| !b.extent().intersect(over).is_empty();
697                let parts: Vec<Arc<Buffer>> = chunks.iter().filter(meets).cloned().collect();
698                let mut lacks = Vec::new();
699                for held in head.stored().extents() {
700                    let mut asked = vec![held.intersect(over)];
701                    for part in chunks {
702                        asked = asked
703                            .into_iter()
704                            .flat_map(|a| minus(a, part.extent()))
705                            .collect();
706                    }
707                    lacks.extend(asked.into_iter().filter(|a| !a.is_empty()));
708                }
709                let hull = lacks.iter().fold(Extent::NOWHERE, |h, e| h.hull(*e));
710                Some((parts, (!hull.is_empty()).then(|| ((**head).clone(), hull))))
711            }
712        }
713    }
714
715    fn coverage(&self, key: Hash) -> Option<Vec<Extent>> {
716        let (foot, by) = self.foot(key)?;
717        let node = self.entries.get(&foot)?.item.node.as_ref();
718        let held = match (node.and_then(Node::head), node.and_then(Node::offered)) {
719            (Some(head), _) => head.stored().extents().to_vec(),
720            (None, None | Some(Offered::Own)) => {
721                let parts = self.parts(foot)?.into_iter();
722                parts.map(|(_, held)| held).collect()
723            }
724            (None, Some(Offered::Held(parts))) => parts.iter().map(|p| p.extent()).collect(),
725            (None, Some(Offered::Moves { .. })) => return None,
726        };
727        Some(held.into_iter().map(|e| e.shifted(-by)).collect())
728    }
729
730    /// A node off the disk keeps its header.
731    fn evict(&mut self, key: Hash) {
732        if matches!(self.source_mut(key), Some(Source::Disk { .. })) {
733            self.flush(key);
734        }
735        let Some(held) = self.entries.get_mut(&key) else {
736            return;
737        };
738        let protected = held.protected;
739        let before = held.bytes();
740        let gone = match held.item.node.as_mut().map(|node| &mut node.source) {
741            Some(Source::Disk { chunks, .. }) => {
742                chunks.clear();
743                held.item.payload = None;
744                self.bytes -= before;
745                true
746            }
747            _ => self.remove(key),
748        };
749        match (gone, protected) {
750            (false, _) => {}
751            (true, false) => self.counters.probation_evictions += 1,
752            (true, true) => self.counters.protected_evictions += 1,
753        }
754    }
755
756    /// A hit on `key` and on every entry whose samples it answers with: each moves to the
757    /// protected segment, whose least recent move back to probation past its share.
758    fn hit(&mut self, key: Hash, round: Option<u64>) {
759        let tick = self.tick();
760        let mut promoted = false;
761        for at in self.under(key) {
762            let Some(held) = self.entries.get_mut(&at) else {
763                continue;
764            };
765            held.read = tick;
766            if round.is_some_and(|round| held.hit_round == round) {
767                continue;
768            }
769            held.hit_round = round.unwrap_or(0);
770            promoted |= !std::mem::replace(&mut held.protected, true);
771            self.counters.hits += 1;
772        }
773        if promoted {
774            self.shared();
775        }
776    }
777
778    fn under(&self, key: Hash) -> Vec<Hash> {
779        let mut out = vec![key];
780        let mut k = 0;
781        while k < out.len() {
782            for at in self.reads(out[k]) {
783                if !out.contains(&at) {
784                    out.push(at);
785                }
786            }
787            k += 1;
788        }
789        out
790    }
791
792    /// Protected within its share: its least recent move back to probation, as its newest.
793    fn shared(&mut self) {
794        let most =
795            (u128::from(self.max_bytes) * u128::from(PROTECTED.0) / u128::from(PROTECTED.1)) as u64;
796        let mut protected: Vec<(u64, Hash, u64)> = self
797            .entries
798            .iter()
799            .filter(|(_, held)| held.protected)
800            .map(|(key, held)| (held.read, *key, held.bytes()))
801            .collect();
802        let mut bytes: u64 = protected.iter().map(|(_, _, b)| b).sum();
803        protected.sort_unstable();
804        for (_, key, held) in protected {
805            if bytes <= most {
806                break;
807            }
808            let tick = self.tick();
809            if let Some(entry) = self.entries.get_mut(&key) {
810                entry.protected = false;
811                entry.since = tick;
812                bytes -= held;
813            }
814        }
815    }
816
817    /// Under the cap: an entry too large for it goes first, then probation oldest first,
818    /// then protected least recently read.
819    fn bounded(&mut self) {
820        if self.bytes <= self.max_bytes {
821            return;
822        }
823        self.shared();
824        let max = self.max_bytes;
825        let mut order: Vec<(u8, u64, Hash)> = self
826            .entries
827            .iter()
828            .filter(|(_, held)| held.bytes() > 0)
829            .map(|(key, held)| match (held.bytes() > max, held.protected) {
830                (true, _) => (0, held.since, *key),
831                (false, false) => (1, held.since, *key),
832                (false, true) => (2, held.read, *key),
833            })
834            .collect();
835        order.sort_unstable();
836        for (_, _, key) in order {
837            if self.bytes <= self.max_bytes {
838                break;
839            }
840            self.evict(key);
841        }
842    }
843}
844
845type Resident = (Vec<Arc<Buffer>>, Option<(Header, Extent)>);
846
847fn minus(e: Extent, cut: Extent) -> Vec<Extent> {
848    let met = e.intersect(cut);
849    if met.is_empty() {
850        return vec![e];
851    }
852    vec![Extent::new(e.start, met.start), Extent::new(met.end, e.end)]
853}
854
855fn clipped(part: &Arc<Buffer>, met: Extent) -> Arc<Buffer> {
856    let held = part.extent();
857    match met == held {
858        true => Arc::clone(part),
859        false => Arc::new(part.over(met, held)),
860    }
861}
862
863fn moved(part: &Arc<Buffer>, by: i64) -> Arc<Buffer> {
864    if by == 0 {
865        return Arc::clone(part);
866    }
867    let mut out = (**part).clone();
868    out.start += by;
869    Arc::new(out)
870}
871
872/// The memory tier: the one owner of every value and node held resident, and of what memory
873/// knows a disk beneath it holds or lacks. A clone is a handle on the same memory.
874#[derive(Clone)]
875pub(crate) struct Memory {
876    state: Arc<Mutex<State>>,
877}
878
879impl Default for Memory {
880    fn default() -> Memory {
881        Memory::holding(DEFAULT_CACHE_BYTES)
882    }
883}
884
885impl Memory {
886    pub(crate) fn holding(max_bytes: u64) -> Memory {
887        Memory {
888            state: Arc::new(Mutex::new(State {
889                max_bytes,
890                mark_every: DEFAULT_MARK_EVERY,
891                ..State::default()
892            })),
893        }
894    }
895
896    /// Over a disk: each node a value keeps is written back before it goes.
897    pub(crate) fn over_disk(max_bytes: u64) -> Memory {
898        let memory = Memory::holding(max_bytes);
899        memory.locked().disk = true;
900        memory
901    }
902
903    fn locked(&self) -> MutexGuard<'_, State> {
904        self.state.lock().unwrap_or_else(|poisoned| {
905            let mut state = poisoned.into_inner();
906            state.entries.clear();
907            *state.links.get_mut() = Links::default();
908            state.slots.clear();
909            state.bytes = 0;
910            self.state.clear_poison();
911            state
912        })
913    }
914
915    pub(crate) fn max_bytes(&self) -> u64 {
916        self.locked().max_bytes
917    }
918
919    pub(crate) fn set_max_bytes(&self, max_bytes: u64) {
920        let mut state = self.locked();
921        state.max_bytes = max_bytes;
922        state.bounded();
923    }
924
925    pub(crate) fn bytes(&self) -> u64 {
926        self.locked().bytes
927    }
928
929    pub(crate) fn entries(&self) -> usize {
930        self.locked().entries.len()
931    }
932
933    pub(crate) fn holds(&self, key: Hash) -> bool {
934        self.locked().entries.contains_key(&key)
935    }
936
937    pub(crate) fn counters(&self) -> Counters {
938        self.locked().counters
939    }
940
941    pub(crate) fn count(&self, by: impl FnOnce(&mut Counters)) {
942        by(&mut self.locked().counters);
943    }
944
945    pub(crate) fn mark_every(&self) -> usize {
946        self.locked().mark_every
947    }
948
949    pub(crate) fn set_mark_every(&self, samples: usize) {
950        self.locked().mark_every = samples.max(1);
951    }
952
953    /// Whether memory takes a value computed: a memory of no bytes takes none.
954    pub(crate) fn keeps(&self) -> bool {
955        self.locked().max_bytes > 0
956    }
957
958    /// A new round of lookups: what any earlier round missed is asked again.
959    pub(crate) fn begin(&self) -> u64 {
960        let mut state = self.locked();
961        state.round += 1;
962        let round = state.round;
963        state.misses.retain(|_, met| *met + 1 >= round);
964        round
965    }
966
967    pub(crate) fn answer(&self, key: Hash, round: u64) -> Known {
968        let mut state = self.locked();
969        if state.node(key).is_some() {
970            let covered = state.coverage(key);
971            if let Some(held) = covered.clone().filter(|held| !held.is_empty()) {
972                state.hit(key, Some(round));
973                let node = state.node(key).expect("a node held");
974                return Known::Hit(Arc::new(node.stored().holding(held)));
975            }
976            match covered {
977                None => {
978                    state.remove(key);
979                }
980                Some(_) => return Known::Miss,
981            }
982        }
983        match state.misses.get(&key) {
984            Some(met) if *met >= round => Known::Miss,
985            _ => Known::Unknown,
986        }
987    }
988
989    /// `answer`, told to `seen` once a hit or a miss.
990    pub(crate) fn answered(
991        &self,
992        (key, round): (Hash, u64),
993        (node, seen): (&str, &mut Recording),
994    ) -> Known {
995        let known = self.answer(key, round);
996        let outcome = match &known {
997            Known::Hit(_) => Outcome::Hit,
998            Known::Miss => Outcome::ComputedNotStored,
999            Known::Unknown => return known,
1000        };
1001        let state = self.locked();
1002        let held = state
1003            .entries
1004            .get(&key)
1005            .and_then(|held| held.item.payload.as_ref());
1006        let kind = held.map_or(PayloadKind::Segments, Payload::kind);
1007        drop(state);
1008        seen.answered(key, node, kind, outcome);
1009        known
1010    }
1011
1012    pub(crate) fn miss(&self, key: Hash, round: u64) {
1013        let mut state = self.locked();
1014        let met = state.misses.entry(key).or_insert(round);
1015        *met = (*met).max(round);
1016    }
1017
1018    /// A header the disk answered, resident from now on, unless memory took the node meanwhile.
1019    pub(crate) fn promote(&self, head: Header) {
1020        let mut state = self.locked();
1021        let (key, read) = (head.stored().key, state.tick());
1022        if state.node(key).is_some() {
1023            return;
1024        }
1025        state.misses.remove(&key);
1026        state.counters.promotions += 1;
1027        let node = Node {
1028            source: Source::Disk {
1029                head: Box::new(head),
1030                chunks: Vec::new(),
1031                more: None,
1032                rewritten: false,
1033            },
1034            dirty: false,
1035            bound: true,
1036        };
1037        match state.entries.get_mut(&key) {
1038            Some(held) => held.item.node = Some(node),
1039            None => {
1040                let item = Item {
1041                    payload: None,
1042                    label: None,
1043                    slot: None,
1044                    node: Some(node),
1045                };
1046                state.admit(key, Held::admitted(item, read));
1047            }
1048        }
1049        state.links.get_mut().evict(key);
1050    }
1051
1052    pub(crate) fn promote_samples(&self, key: Hash, read: Vec<Buffer>) -> Vec<Arc<Buffer>> {
1053        let read: Vec<Arc<Buffer>> = read.into_iter().map(Arc::new).collect();
1054        let mut state = self.locked();
1055        let tick = state.tick();
1056        let Some(held) = state.entries.get_mut(&key) else {
1057            return read;
1058        };
1059        let Some(Node {
1060            source: Source::Disk { chunks, .. },
1061            ..
1062        }) = &mut held.item.node
1063        else {
1064            return read;
1065        };
1066        let mut added = 0;
1067        for part in &read {
1068            let covered = chunks
1069                .iter()
1070                .any(|c| c.extent().intersect(part.extent()) == part.extent());
1071            if !covered {
1072                added += planes(part);
1073                chunks.push(Arc::clone(part));
1074            }
1075        }
1076        chunks.sort_by_key(|c| c.start);
1077        held.read = tick;
1078        state.bytes += added;
1079        state.counters.promotions += 1;
1080        state.bounded();
1081        read
1082    }
1083
1084    /// What memory holds of `key` over `over`, and what it lacks there that the disk holds.
1085    pub(crate) fn resident(&self, key: Hash, over: Extent) -> Resident {
1086        let mut state = self.locked();
1087        let tick = state.tick();
1088        if let Some(held) = state.entries.get_mut(&key) {
1089            held.read = tick;
1090        }
1091        state.resident(key, over).unwrap_or_default()
1092    }
1093
1094    pub(crate) fn forget(&self, key: Hash) {
1095        self.locked().remove(key);
1096    }
1097
1098    /// A value's samples, joined to what memory holds under `key`, and the node they answer,
1099    /// in place of the last one under its key or slot: the node first, so samples a node over
1100    /// the disk stands on are kept to be written back. A node holding none of its samples that
1101    /// memory will write nowhere goes. What became of the samples is told to `seen`.
1102    pub(crate) fn keep(&self, key: Hash, keep: Keep, seen: &mut Recording) {
1103        let Keep {
1104            samples,
1105            label,
1106            slot,
1107            node,
1108        } = keep;
1109        let sole = samples.is_none();
1110        let noded = node.map(|(stored, offered, facts)| {
1111            let key = stored.key;
1112            let bound = self.locked().noded(stored, offered, (slot, sole, facts));
1113            (key, bound)
1114        });
1115        let kept = samples.map(|samples| self.merge(key, samples, label, slot));
1116        if let Some(kept) = kept {
1117            seen.kept(key, kept);
1118        }
1119        if let Some((key, bound)) = noded {
1120            let mut state = self.locked();
1121            let covered = state.coverage(key).is_some_and(|held| !held.is_empty());
1122            if !covered && !bound {
1123                state.unnoded(key);
1124            }
1125            state.bounded();
1126        }
1127    }
1128
1129    /// The nodes read after `since`, least recent first, and the clock now.
1130    pub(crate) fn read_since(&self, since: u64) -> (Vec<Hash>, u64) {
1131        let state = self.locked();
1132        let mut read: Vec<(u64, Hash)> = state
1133            .entries
1134            .iter()
1135            .filter(|(_, held)| held.read > since && held.item.node.is_some())
1136            .map(|(key, held)| (held.read, *key))
1137            .collect();
1138        read.sort_unstable();
1139        (read.into_iter().map(|(_, key)| key).collect(), state.clock)
1140    }
1141
1142    /// Every node not yet on the disk, on its way there.
1143    pub(crate) fn flush(&self) {
1144        let mut state = self.locked();
1145        let keys: Vec<Hash> = state.entries.keys().copied().collect();
1146        for key in keys {
1147            state.flush(key);
1148        }
1149        state.failed.1 = None;
1150    }
1151
1152    /// The nodes on their way to the disk; none while a write that failed waits for a persist.
1153    pub(crate) fn pending(&self) -> Vec<Writeback> {
1154        let mut state = self.locked();
1155        match state.failed.1 {
1156            Some(_) => Vec::new(),
1157            None => std::mem::take(&mut state.pending),
1158        }
1159    }
1160
1161    /// `left` still on its way, after a write failed for `why`.
1162    pub(crate) fn failed(&self, why: String, left: Vec<Writeback>) {
1163        let mut state = self.locked();
1164        state.failed.0 += 1;
1165        state.failed.1 = Some(why);
1166        let mut more = std::mem::take(&mut state.pending);
1167        state.pending = left;
1168        state.pending.append(&mut more);
1169    }
1170
1171    pub(crate) fn written(&self) {
1172        self.locked().counters.writebacks += 1;
1173    }
1174
1175    pub(crate) fn failures(&self) -> (u64, Option<String>) {
1176        self.locked().failed.clone()
1177    }
1178
1179    pub(crate) fn blocked(&self) -> bool {
1180        self.locked().failed.1.is_some()
1181    }
1182
1183    /// The disk committed what was staged: a header that named staged samples, or an entry
1184    /// memory sent more of since, names nothing now.
1185    pub(crate) fn committed(&self) {
1186        let mut state = self.locked();
1187        let staged: Vec<Hash> = state
1188            .entries
1189            .iter()
1190            .filter(|(_, held)| {
1191                let node = held.item.node.as_ref();
1192                let staged = node.and_then(Node::head);
1193                let staged =
1194                    staged.is_some_and(|head| matches!(head.samples(), Samples::Staged { .. }));
1195                let rewritten = |node: &Node| {
1196                    matches!(
1197                        node.source,
1198                        Source::Disk {
1199                            rewritten: true,
1200                            ..
1201                        }
1202                    )
1203                };
1204                staged || node.is_some_and(rewritten)
1205            })
1206            .map(|(key, _)| *key)
1207            .collect();
1208        for key in staged {
1209            state.unnoded(key);
1210        }
1211    }
1212
1213    /// What `key` holds, shared, never copied, told to `seen`.
1214    pub(crate) fn load(
1215        &self,
1216        key: Hash,
1217        expected: Expected,
1218        (node, seen): (&str, &mut Recording),
1219    ) -> Option<Entry> {
1220        let entry = self.entry(key, expected);
1221        let outcome = match entry {
1222            Some(_) => Outcome::Hit,
1223            None => Outcome::ComputedNotStored,
1224        };
1225        seen.answered(key, node, expected.kind(), outcome);
1226        entry
1227    }
1228
1229    /// A run's segments held under `keys` in turn, each handed to `take` until it takes none or
1230    /// asks no more, told to `seen` as one lookup of the last: a prefix where it took fewer.
1231    pub(crate) fn runs(
1232        &self,
1233        keys: &[Hash],
1234        expected: Expected,
1235        (node, seen): (&str, &mut Recording),
1236        mut take: impl FnMut(usize, Arc<Run>) -> (bool, bool),
1237    ) {
1238        let mut taken = 0;
1239        for (k, key) in keys.iter().enumerate() {
1240            let Some(run) = self.entry(*key, expected).and_then(|e| e.payload.run()) else {
1241                break;
1242            };
1243            let (took, more) = take(k, run);
1244            taken += usize::from(took);
1245            if !(took && more) {
1246                break;
1247            }
1248        }
1249        let outcome = match taken {
1250            0 => Outcome::ComputedNotStored,
1251            n if n == keys.len() => Outcome::Hit,
1252            _ => Outcome::Prefix,
1253        };
1254        let last = *keys.last().expect("a run of a segment or more");
1255        seen.answered(last, node, PayloadKind::Run, outcome);
1256    }
1257
1258    fn entry(&self, key: Hash, expected: Expected) -> Option<Entry> {
1259        let mut state = self.locked();
1260        let item = &state.entries.get(&key)?.item;
1261        let entry = Entry {
1262            payload: item.payload.clone()?,
1263            label: item.label.clone(),
1264        };
1265        if !entry.payload.answers(expected) {
1266            state.remove(key);
1267            return None;
1268        }
1269        state.hit(key, None);
1270        Some(entry)
1271    }
1272
1273    /// A value's segments join those held under `key`, and a run continuing the one held
1274    /// there extends it, each in place; anything else replaces what `key` held.
1275    fn merge(
1276        &self,
1277        key: Hash,
1278        payload: Payload,
1279        label: Option<&Label>,
1280        slot: Option<Hash>,
1281    ) -> Kept {
1282        let mut state = self.locked();
1283        let tick = state.tick();
1284        let joined = match state.entries.get_mut(&key) {
1285            Some(held) if held.item.slot == slot && held.item.payload.is_some() => {
1286                let before = held.bytes();
1287                let had = held.item.payload.as_mut().expect("a value held");
1288                let payload = match (had, payload) {
1289                    (Payload::Segments(parts), Payload::Segments(more)) => {
1290                        joined(parts, more);
1291                        None
1292                    }
1293                    (Payload::Run(run), Payload::Run(more)) if overlaps(run, &more) => {
1294                        let from = (run.end() - more.samples.start).max(0) as usize;
1295                        let run = Arc::make_mut(run);
1296                        let samples = Arc::make_mut(&mut run.samples);
1297                        for (held, more) in samples.planes.iter_mut().zip(&more.samples.planes) {
1298                            held.extend_from_slice(&more[from.min(more.len())..]);
1299                        }
1300                        run.marks
1301                            .extend(more.marks.iter().map(|(at, m)| (*at, m.clone())));
1302                        None
1303                    }
1304                    (_, payload) => Some(payload),
1305                };
1306                match payload {
1307                    None => {
1308                        held.read = tick;
1309                        let after = held.bytes();
1310                        Ok((before, after))
1311                    }
1312                    Some(payload) => Err(payload),
1313                }
1314            }
1315            _ => Err(payload),
1316        };
1317        match joined {
1318            Ok((before, after)) => {
1319                state.bytes = state.bytes - before + after;
1320                state.bounded();
1321                Kept::Held
1322            }
1323            Err(payload) => {
1324                drop(state);
1325                self.store(key, payload, label, slot)
1326            }
1327        }
1328    }
1329
1330    /// A value the disk holds under `key` is held there alone. One too large to stay is
1331    /// refused, unless a node over the disk stands on it: then it is kept only to be evicted,
1332    /// and so written back.
1333    fn store(
1334        &self,
1335        key: Hash,
1336        payload: Payload,
1337        label: Option<&Label>,
1338        slot: Option<Hash>,
1339    ) -> Kept {
1340        let mut state = self.locked();
1341        if state.on_disk(key, &payload) {
1342            return Kept::Held;
1343        }
1344        let bytes = payload.bytes() as u64;
1345        if bytes > state.max_bytes && !state.stands(key) {
1346            return Kept::Refused;
1347        }
1348        let read = state.tick();
1349        let replaced = match slot.and_then(|slot| state.slots.insert(slot, key)) {
1350            Some(last) if last != key => state.remove(last),
1351            _ => false,
1352        };
1353        let old = state.entries.remove(&key);
1354        if let Some(old) = &old {
1355            state.bytes -= old.bytes();
1356        }
1357        let (node, earned) = match old {
1358            Some(old) => (
1359                old.item.node,
1360                Some((old.read, old.since, old.hit_round, old.protected)),
1361            ),
1362            None => (None, None),
1363        };
1364        let item = Item {
1365            payload: Some(payload),
1366            label: label.cloned(),
1367            slot,
1368            node,
1369        };
1370        let held = match earned {
1371            Some((read, since, hit_round, protected)) if item.node.is_some() => Held {
1372                item,
1373                read,
1374                since,
1375                hit_round,
1376                protected,
1377            },
1378            _ => Held::admitted(item, read),
1379        };
1380        state.bytes += held.bytes();
1381        state.admit(key, held);
1382        state.bounded();
1383        match replaced {
1384            true => Kept::Replaced,
1385            false => Kept::Held,
1386        }
1387    }
1388}
1389
1390/// The target, and a node a later render is answered by that two or more values read or a
1391/// stateful run wrote: what one reader alone recomputes from its own reads is never written.
1392fn writes(stored: &Stored, facts: Facts) -> bool {
1393    facts.target || (stored.readable && (facts.shared || facts.stateful))
1394}
1395
1396/// A run that starts inside or at the end of the one held continues it: what it holds past
1397/// that one's end is laid on, the samples both hold being the same.
1398fn overlaps(held: &super::Run, more: &super::Run) -> bool {
1399    let (a, b) = (held.samples.start, held.end());
1400    a <= more.samples.start && more.samples.start <= b
1401}
1402
1403#[cfg(test)]
1404mod tests {
1405    use super::*;
1406
1407    /// Only a colliding key reaches this, so no render can: the entry is a miss, and goes.
1408    #[test]
1409    fn an_entry_that_does_not_answer_what_was_asked_is_a_miss_and_goes() {
1410        let memory = Memory::default();
1411        let key = Hash(7, 11);
1412        let four = Payload::Segments(vec![Arc::new(Buffer::mono(8_000, vec![0.25; 4]))]);
1413        for (rate, width) in [(48_000, 1), (8_000, 2)] {
1414            memory.store(key, four.clone(), None, None);
1415            let asked = Expected::Segments { rate, width };
1416            assert!(memory.entry(key, asked).is_none());
1417            assert!(!memory.holds(key));
1418            assert_eq!(memory.bytes(), 0);
1419        }
1420    }
1421
1422    #[test]
1423    fn a_load_shares_the_samples_it_holds() {
1424        let memory = Memory::default();
1425        let key = Hash(3, 5);
1426        let part = Arc::new(Buffer::mono(8_000, vec![0.5; 64]));
1427        memory.store(key, Payload::Segments(vec![Arc::clone(&part)]), None, None);
1428        let asked = Expected::Segments {
1429            rate: 8_000,
1430            width: 1,
1431        };
1432        for _ in 0..2 {
1433            let loaded = memory.entry(key, asked).expect("a hit");
1434            let Payload::Segments(parts) = loaded.payload else {
1435                panic!("segments were stored");
1436            };
1437            assert!(Arc::ptr_eq(&parts[0], &part), "the stored part itself");
1438        }
1439    }
1440}