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