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