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 // New-key event capture: the owned key copy is only paid when
20 // the capture flag is on (server with `n` notifications).
21 let new_key_copy =
22 (self.notify_capture & crate::notify::CAPTURE_NEW != 0).then(|| key.to_vec());
23 // A wholesale value replacement (type change / RESTORE)
24 // discards any per-field hash TTLs; a fresh create is a no-op.
25 self.clear_hash_key_ttls(key.as_slice());
26 let key_heap = key.heap_bytes() as u64;
27 entry.set_weight(key_heap + entry.value.weight());
28 if self.clock_on() {
29 self.tick_clock();
30 entry.set_lru_clock(self.clock_counter as u32);
31 }
32 let new_w = entry.weight();
33 let new_has_ttl = entry.expire_at_ns.is_some();
34 let prev = self.map.insert(key, entry);
35 match &prev {
36 Some(old) => {
37 // A displaced cold stub's vlog record dies with it.
38 self.tier_note_dead(key_heap, &old.value);
39 self.used_memory =
40 self.used_memory.saturating_sub(old.weight()).saturating_add(new_w);
41 }
42 None => {
43 self.used_memory = self.used_memory.saturating_add(new_w + ENTRY_OVERHEAD);
44 }
45 }
46 let old_has_ttl = prev.as_ref().is_some_and(|o| o.expire_at_ns.is_some());
47 self.adjust_expires(i64::from(new_has_ttl) - i64::from(old_has_ttl));
48 self.update_peak();
49 if prev.is_none()
50 && let Some(k) = new_key_copy
51 {
52 self.notify_events.push((crate::notify::KeyspaceEvent::New, k));
53 }
54 prev
55 }
56
57 /// Remove a key, returning the displaced entry (`None` if absent).
58 /// Frees the entry's cached weight + [`ENTRY_OVERHEAD`]. This is the
59 /// DISCARD form: a cold stub's vlog record is credited dead. A
60 /// caller re-homing the entry intact (RENAME) uses
61 /// [`Self::take_entry_keepalive`] instead.
62 pub(crate) fn remove_entry(&mut self, key: &[u8]) -> Option<Entry> {
63 let old = self.take_entry_keepalive(key)?;
64 self.tier_note_dead(key_heap_bytes_for(key), &old.value);
65 Some(old)
66 }
67
68 /// [`Self::remove_entry`] minus the cold-record dead-credit — for
69 /// moves that keep the entry (and any [`crate::value::ColdRef`] in
70 /// it) alive under another key. Same accounting/hfttl behaviour.
71 pub(crate) fn take_entry_keepalive(&mut self, key: &[u8]) -> Option<Entry> {
72 self.clear_hash_key_ttls(key);
73 let old = self.map.remove(key)?;
74 self.used_memory = self.used_memory.saturating_sub(old.weight() + ENTRY_OVERHEAD);
75 if old.expire_at_ns.is_some() {
76 self.adjust_expires(-1);
77 }
78 Some(old)
79 }
80
81 /// Apply a signed weight delta to `key`'s cached `Entry::weight` AND to
82 /// the shard-wide `used_memory`. Used by in-place collection mutators
83 /// (HSET adding a field, LPUSH adding an item, …) so we account in O(1)
84 /// without re-walking the container.
85 pub(crate) fn account_delta(&mut self, key: &[u8], delta: i64) {
86 if delta == 0 {
87 return;
88 }
89 if let Some(e) = self.map.get_mut(key) {
90 e.add_to_weight(delta);
91 }
92 apply_delta(&mut self.used_memory, delta);
93 if delta > 0 {
94 self.update_peak();
95 }
96 }
97
98 /// Recompute `weight` for the entry at `key` from its current value +
99 /// key, then propagate the delta to `used_memory`. Use after a wholesale
100 /// in-place value swap (SET / APPEND / INCRBYFLOAT) where the prior
101 /// `Value`'s weight was already cached on the entry.
102 pub(crate) fn reweigh_entry(&mut self, key: &[u8]) {
103 let key_heap = key_heap_bytes_for(key);
104 let Some(e) = self.map.get_mut(key) else {
105 return;
106 };
107 let new_w = key_heap + e.value.weight();
108 let delta = new_w as i64 - e.weight() as i64;
109 e.set_weight(new_w);
110 apply_delta(&mut self.used_memory, delta);
111 if delta > 0 {
112 self.update_peak();
113 }
114 }
115
116 /// Advance the global access ordinal by one tick. Only invoked under
117 /// `maxmemory > 0` so the wrapping_add cost stays out of the unlimited
118 /// fast path.
119 #[inline]
120 pub(crate) fn tick_clock(&mut self) {
121 self.clock_counter = self.clock_counter.wrapping_add(1);
122 }
123
124 #[inline]
125 fn update_peak(&mut self) {
126 if self.used_memory > self.used_memory_peak {
127 self.used_memory_peak = self.used_memory;
128 }
129 }
130
131 /// Apply a weight delta computed in-place by a caller that already held
132 /// `&mut Entry` (overwrite-SET fast path) — same arithmetic as
133 /// [`Self::reweigh_entry`] but WITHOUT the second hash + map probe that
134 /// `reweigh_entry(key)` pays to re-find the entry it just mutated.
135 #[inline]
136 pub(crate) fn apply_weight_delta(&mut self, delta: i64) {
137 apply_delta(&mut self.used_memory, delta);
138 if delta > 0 {
139 self.update_peak();
140 }
141 }
142
143 /// Hint the CPU to fetch the bucket cache line for `key` into L1. Called
144 /// by the reactor's parse loop on command N+1 while command N is still
145 /// being dispatched — by the time N+1 actually probes the table, the
146 /// metadata line is hot. No-op when the table is empty. Cheap when not.
147 #[inline]
148 pub fn prefetch_for_key(&self, key: &[u8]) {
149 let hash = key.kevy_hash();
150 self.map.prefetch_for_hash(hash);
151 }
152
153 pub(crate) fn expired(&self, key: &[u8], now: u64) -> bool {
154 match self.map.get(key) {
155 Some(e) => e.is_expired_at(now),
156 None => false,
157 }
158 }
159
160 /// Drop `key` if expired; returns whether it is live afterwards. `now` is
161 /// monotonic ns since epoch (from [`crate::now_ns`]).
162 pub(crate) fn reap(&mut self, key: &[u8], now: u64) -> bool {
163 if self.expired(key, now) {
164 self.note_expired(key);
165 self.remove_entry(key);
166 self.expired_keys_total = self.expired_keys_total.saturating_add(1);
167 false
168 } else {
169 self.map.contains_key(key)
170 }
171 }
172
173 /// Single-lookup lazy-expiring read: the live `Entry` for `key`, or `None` if
174 /// absent or expired (expired keys are dropped here, as `reap` would).
175 ///
176 /// Two wins over the old `reap(now)`-then-`get` read path: (1) the clock is
177 /// read **only when the entry actually carries a TTL** — most keys don't, so
178 /// the common hit skips `Instant::now()` (~20–40 ns); (2) one fewer keyspace
179 /// lookup on hits (was peek-expiry + `contains_key` + `get` = 3; now peek +
180 /// `get` = 2). The two-phase shape (decide, then mutate/fetch) keeps the
181 /// borrow checker happy without an owning key clone.
182 pub(crate) fn live_entry(&mut self, key: &[u8]) -> Option<&Entry> {
183 // TTL-free fast path. Read cached clock fields
184 // ONLY when the entry actually carries a TTL — most keys don't,
185 // and a prior implementation paid two field reads + a pass
186 // through `is_expired` (which itself short-circuits on None)
187 // unconditionally. Saves ~5 ns / hot lookup across every
188 // collection / string read path.
189 let needs_check = self.map.get(key)?.expire_at_ns.is_some();
190 if needs_check {
191 let (uc, cn) = (self.cached_clock, self.cached_ns);
192 let expired = self.map.get(key).is_some_and(|e| e.is_expired(uc, cn));
193 if expired {
194 self.note_expired(key);
195 self.remove_entry(key);
196 self.expired_keys_total = self.expired_keys_total.saturating_add(1);
197 return None;
198 }
199 }
200 if self.clock_on() {
201 self.tick_clock();
202 let c = self.clock_counter as u32;
203 let policy = self.touch_policy();
204 let e = self.map.get_mut(key)?;
205 evict::touch_on_access(e, policy, c);
206 return Some(&*e);
207 }
208 self.map.get(key)
209 }
210
211 /// Mutable [`live_entry`](Self::live_entry): the live `Entry` for `key` by
212 /// `&mut`, or `None` if absent/expired (expired dropped). Same wins — clock
213 /// read only on TTL'd keys, one fewer lookup than `reap`-then-`get_mut`.
214 /// Read-modify commands (INCR/APPEND/…) get the entry once and mutate in
215 /// place, preserving any TTL on it.
216 pub(crate) fn live_entry_mut(&mut self, key: &[u8]) -> Option<&mut Entry> {
217 // See `live_entry` doc — TTL-free fast path.
218 let needs_check = self.map.get(key)?.expire_at_ns.is_some();
219 if needs_check {
220 let (uc, cn) = (self.cached_clock, self.cached_ns);
221 let expired = self.map.get(key).is_some_and(|e| e.is_expired(uc, cn));
222 if expired {
223 self.note_expired(key);
224 self.remove_entry(key);
225 self.expired_keys_total = self.expired_keys_total.saturating_add(1);
226 return None;
227 }
228 }
229 if self.clock_on() {
230 self.tick_clock();
231 let c = self.clock_counter as u32;
232 let policy = self.touch_policy();
233 let e = self.map.get_mut(key)?;
234 evict::touch_on_access(e, policy, c);
235 return Some(e);
236 }
237 self.map.get_mut(key)
238 }
239}