kevy_store/accounting.rs
1//! Internal accounting helpers on [`Store`]: per-entry weight bookkeeping,
2//! LRU/LFU clock advance, prefetch, and the lazy-expire `live_entry` /
3//! `live_entry_mut` lookups used by every typed accessor.
4//!
5//! Split out of [`crate`] for file-size hygiene. Nothing here is part of
6//! the public surface — all methods are `pub(crate)` and called by sibling
7//! modules (string/hash/list/set/zset/evict/expire/keyspace).
8
9use kevy_hash::KevyHash;
10
11use crate::value::ENTRY_OVERHEAD;
12use crate::{Entry, SmallBytes, Store, apply_delta, evict, key_heap_bytes_for};
13
14impl Store {
15 /// Insert a fresh entry, replacing any prior. Stamps `entry.weight` from
16 /// the live value and key, then updates `used_memory` for either the
17 /// new-key (charges [`ENTRY_OVERHEAD`]) or overwrite (weight swap) case.
18 pub(crate) fn insert_entry(&mut self, key: SmallBytes, mut entry: Entry) -> Option<Entry> {
19 // v2.4: a wholesale value replacement (type change / RESTORE)
20 // discards any per-field hash TTLs; a fresh create is a no-op.
21 self.clear_hash_key_ttls(key.as_slice());
22 entry.set_weight(key.heap_bytes() as u64 + entry.value.weight());
23 if self.maxmemory > 0 {
24 self.tick_clock();
25 entry.set_lru_clock(self.clock_counter as u32);
26 }
27 let new_w = entry.weight();
28 let new_has_ttl = entry.expire_at_ns.is_some();
29 let prev = self.map.insert(key, entry);
30 match &prev {
31 Some(old) => {
32 self.used_memory = self
33 .used_memory
34 .saturating_sub(old.weight())
35 .saturating_add(new_w);
36 }
37 None => {
38 self.used_memory = self.used_memory.saturating_add(new_w + ENTRY_OVERHEAD);
39 }
40 }
41 let old_has_ttl = prev.as_ref().is_some_and(|o| o.expire_at_ns.is_some());
42 self.adjust_expires(i64::from(new_has_ttl) - i64::from(old_has_ttl));
43 self.update_peak();
44 prev
45 }
46
47 /// Remove a key, returning the displaced entry (`None` if absent).
48 /// Frees the entry's cached weight + [`ENTRY_OVERHEAD`].
49 pub(crate) fn remove_entry(&mut self, key: &[u8]) -> Option<Entry> {
50 self.clear_hash_key_ttls(key);
51 let old = self.map.remove(key)?;
52 self.used_memory = self
53 .used_memory
54 .saturating_sub(old.weight() + ENTRY_OVERHEAD);
55 if old.expire_at_ns.is_some() {
56 self.adjust_expires(-1);
57 }
58 Some(old)
59 }
60
61 /// Apply a signed weight delta to `key`'s cached `Entry::weight` AND to
62 /// the shard-wide `used_memory`. Used by in-place collection mutators
63 /// (HSET adding a field, LPUSH adding an item, …) so we account in O(1)
64 /// without re-walking the container.
65 pub(crate) fn account_delta(&mut self, key: &[u8], delta: i64) {
66 if delta == 0 {
67 return;
68 }
69 if let Some(e) = self.map.get_mut(key) {
70 e.add_to_weight(delta);
71 }
72 apply_delta(&mut self.used_memory, delta);
73 if delta > 0 {
74 self.update_peak();
75 }
76 }
77
78 /// Recompute `weight` for the entry at `key` from its current value +
79 /// key, then propagate the delta to `used_memory`. Use after a wholesale
80 /// in-place value swap (SET / APPEND / INCRBYFLOAT) where the prior
81 /// `Value`'s weight was already cached on the entry.
82 pub(crate) fn reweigh_entry(&mut self, key: &[u8]) {
83 let key_heap = key_heap_bytes_for(key);
84 let Some(e) = self.map.get_mut(key) else {
85 return;
86 };
87 let new_w = key_heap + e.value.weight();
88 let delta = new_w as i64 - e.weight() as i64;
89 e.set_weight(new_w);
90 apply_delta(&mut self.used_memory, delta);
91 if delta > 0 {
92 self.update_peak();
93 }
94 }
95
96 /// Advance the global access ordinal by one tick. Only invoked under
97 /// `maxmemory > 0` so the wrapping_add cost stays out of the unlimited
98 /// fast path.
99 #[inline]
100 pub(crate) fn tick_clock(&mut self) {
101 self.clock_counter = self.clock_counter.wrapping_add(1);
102 }
103
104 #[inline]
105 fn update_peak(&mut self) {
106 if self.used_memory > self.used_memory_peak {
107 self.used_memory_peak = self.used_memory;
108 }
109 }
110
111 /// Apply a weight delta computed in-place by a caller that already held
112 /// `&mut Entry` (overwrite-SET fast path) — same arithmetic as
113 /// [`Self::reweigh_entry`] but WITHOUT the second hash + map probe that
114 /// `reweigh_entry(key)` pays to re-find the entry it just mutated.
115 #[inline]
116 pub(crate) fn apply_weight_delta(&mut self, delta: i64) {
117 apply_delta(&mut self.used_memory, delta);
118 if delta > 0 {
119 self.update_peak();
120 }
121 }
122
123 /// Hint the CPU to fetch the bucket cache line for `key` into L1. Called
124 /// by the reactor's parse loop on command N+1 while command N is still
125 /// being dispatched — by the time N+1 actually probes the table, the
126 /// metadata line is hot. No-op when the table is empty. Cheap when not.
127 #[inline]
128 pub fn prefetch_for_key(&self, key: &[u8]) {
129 let hash = key.kevy_hash();
130 self.map.prefetch_for_hash(hash);
131 }
132
133 pub(crate) fn expired(&self, key: &[u8], now: u64) -> bool {
134 match self.map.get(key) {
135 Some(e) => e.is_expired_at(now),
136 None => false,
137 }
138 }
139
140 /// Drop `key` if expired; returns whether it is live afterwards. `now` is
141 /// monotonic ns since epoch (from [`crate::now_ns`]).
142 pub(crate) fn reap(&mut self, key: &[u8], now: u64) -> bool {
143 if self.expired(key, now) {
144 self.remove_entry(key);
145 self.expired_keys_total = self.expired_keys_total.saturating_add(1);
146 false
147 } else {
148 self.map.contains_key(key)
149 }
150 }
151
152 /// Single-lookup lazy-expiring read: the live `Entry` for `key`, or `None` if
153 /// absent or expired (expired keys are dropped here, as `reap` would).
154 ///
155 /// Two wins over the old `reap(now)`-then-`get` read path: (1) the clock is
156 /// read **only when the entry actually carries a TTL** — most keys don't, so
157 /// the common hit skips `Instant::now()` (~20–40 ns); (2) one fewer keyspace
158 /// lookup on hits (was peek-expiry + `contains_key` + `get` = 3; now peek +
159 /// `get` = 2). The two-phase shape (decide, then mutate/fetch) keeps the
160 /// borrow checker happy without an owning key clone.
161 pub(crate) fn live_entry(&mut self, key: &[u8]) -> Option<&Entry> {
162 // G-A4 (v1.25): TTL-free fast path. Read cached clock fields
163 // ONLY when the entry actually carries a TTL — most keys don't,
164 // and the prior implementation paid two field reads + a pass
165 // through `is_expired` (which itself short-circuits on None)
166 // unconditionally. Saves ~5 ns / hot lookup across every
167 // collection / string read path.
168 let needs_check = match self.map.get(key) {
169 None => return None,
170 Some(e) => e.expire_at_ns.is_some(),
171 };
172 if needs_check {
173 let (uc, cn) = (self.cached_clock, self.cached_ns);
174 let expired = self.map.get(key).is_some_and(|e| e.is_expired(uc, cn));
175 if expired {
176 self.remove_entry(key);
177 self.expired_keys_total = self.expired_keys_total.saturating_add(1);
178 return None;
179 }
180 }
181 if self.maxmemory > 0 {
182 self.tick_clock();
183 let c = self.clock_counter as u32;
184 let e = self.map.get_mut(key)?;
185 evict::touch_on_access(e, self.eviction_policy, c);
186 return Some(&*e);
187 }
188 self.map.get(key)
189 }
190
191 /// Mutable [`live_entry`](Self::live_entry): the live `Entry` for `key` by
192 /// `&mut`, or `None` if absent/expired (expired dropped). Same wins — clock
193 /// read only on TTL'd keys, one fewer lookup than `reap`-then-`get_mut`.
194 /// Read-modify commands (INCR/APPEND/…) get the entry once and mutate in
195 /// place, preserving any TTL on it.
196 pub(crate) fn live_entry_mut(&mut self, key: &[u8]) -> Option<&mut Entry> {
197 // G-A4 (v1.25): see `live_entry` doc — TTL-free fast path.
198 let needs_check = match self.map.get(key) {
199 None => return None,
200 Some(e) => e.expire_at_ns.is_some(),
201 };
202 if needs_check {
203 let (uc, cn) = (self.cached_clock, self.cached_ns);
204 let expired = self.map.get(key).is_some_and(|e| e.is_expired(uc, cn));
205 if expired {
206 self.remove_entry(key);
207 self.expired_keys_total = self.expired_keys_total.saturating_add(1);
208 return None;
209 }
210 }
211 if self.maxmemory > 0 {
212 self.tick_clock();
213 let c = self.clock_counter as u32;
214 let e = self.map.get_mut(key)?;
215 evict::touch_on_access(e, self.eviction_policy, c);
216 return Some(e);
217 }
218 self.map.get_mut(key)
219 }
220}