kevy_store/
bio_drop.rs

1//! Bio-drop batching: heavy `Value`s displaced by SET
2//! overwrites are shipped to the runtime's bio thread in per-iteration
3//! batches instead of dropping inline on the reactor.
4
5use crate::value::{self, Value};
6use crate::Store;
7
8/// Maximum [`Store::pending_drops`] depth before forcing a flush
9/// inside `maybe_offload_drop` (rather than waiting for the reactor's
10/// per-iter `flush_pending_drops`). Caps memory held in the batch
11/// buffer at ≤ 64 × sizeof(Box<Value>) (≤ 512 B of pointers + whatever
12/// the boxed payloads weigh — which we WANT to ship anyway, since
13/// holding the bio-bound batch defeats the point of off-reactor frees).
14/// 64 picked as: amortises mpsc send cost (~few hundred ns) across
15/// enough drops that per-drop overhead is ≤ 10 ns, while staying small
16/// enough that worst-case bunch-up latency at the bio thread is bounded.
17pub(crate) const MAX_PENDING_DROPS: usize = 64;
18
19impl Store {
20    /// Install the runtime's bio-drop channel. Called
21    /// once from `kevy-rt::Runtime::run` per shard before the reactor
22    /// loop starts. After install, [`Self::maybe_offload_drop`] (invoked
23    /// from the SET overwrite fast path) accumulates oversize `Value`s
24    /// into a per-shard batch; the reactor calls
25    /// [`Self::flush_pending_drops`] at the end of every iter to ship
26    /// the batch in one mpsc send. Bounds the 10 KB-SET p999/max
27    /// latency blow-up that synchronous `Box::<[u8]>::drop` of an
28    /// allocator large-class slot causes (see `kevy_rt::bio`).
29    #[inline]
30    pub fn set_bio_drop_sender(&mut self, sender: value::BioDropSender) {
31        self.bio_drop_sender = Some(sender);
32    }
33
34    /// Accumulate `old` into the per-shard bio-drop batch buffer
35    /// ([`Store::pending_drops`]) if it's heap-heavy AND a bio channel
36    /// is installed. Otherwise drop inline. The hot path is one branch
37    /// on `bio_drop_sender.is_none()` followed by the variant-cheap
38    /// [`Value::is_heap_heavy`] check; for the `Value::Str(SmallBytes)`
39    /// steady state of typical bench shapes the inline-drop path is
40    /// preserved unchanged.
41    ///
42    /// **Batch model**: per-send mpsc cost (atomic +
43    /// cross-thread cacheline) is amortised across the batch by
44    /// [`Self::flush_pending_drops`], which the reactor calls once per
45    /// iter. Force-flushes here when the buffer hits
46    /// [`MAX_PENDING_DROPS`] to bound RAM in-flight.
47    #[inline]
48    pub(crate) fn maybe_offload_drop(&mut self, old: Value) {
49        if self.bio_drop_sender.is_none() {
50            // No channel (bare Store / embedded reaper / tests): the
51            // Value falls out of scope and drops inline.
52            drop(old);
53            return;
54        }
55        if !old.is_heap_heavy() {
56            // Under-threshold: jemalloc small-class free is sub-µs.
57            // The Vec::push + force-flush branch costs more than the
58            // inline free for this size — leave it inline.
59            drop(old);
60            return;
61        }
62        self.pending_drops.push(old);
63        if self.pending_drops.len() >= MAX_PENDING_DROPS {
64            self.flush_pending_drops();
65        }
66    }
67
68    /// Ship the per-shard bio-drop batch buffer to the bio thread in
69    /// one mpsc send. Called from `kevy-rt`'s reactor loop at the end
70    /// of every iteration (both the epoll `Shard::run` and the io_uring
71    /// `Shard::run_uring` paths, just before the AOF fsync window so a
72    /// pending fsync stall doesn't pin a batch-ful of heavy values in
73    /// per-shard memory).
74    ///
75    /// Empty-buffer fast path: zero work, predictable not-taken
76    /// branch. Reactor calls this unconditionally per iter; the steady-
77    /// state cost for a no-SET-overwrite iter is one length check.
78    ///
79    /// `SendError` here means the bio thread has exited (shutdown
80    /// territory — `Runtime::run` has dropped its sender AFTER the
81    /// shard threads joined). Drop the batch inline; the `SendError`
82    /// payload carries the `Vec` back so its `Box<Value>`s run their
83    /// Drop here, preserving correctness.
84    #[inline]
85    pub fn flush_pending_drops(&mut self) {
86        if self.pending_drops.is_empty() {
87            return;
88        }
89        let tx = match self.bio_drop_sender.as_ref() {
90            Some(tx) => tx,
91            // Shouldn't happen — caller (`maybe_offload_drop`) only
92            // pushes when the sender exists. Defensive: if a future
93            // refactor invokes `flush_pending_drops` from somewhere
94            // unconditional, drop the batch inline.
95            None => {
96                self.pending_drops.clear();
97                return;
98            }
99        };
100        let batch = core::mem::take(&mut self.pending_drops);
101        if let Err(_send_err) = tx.send(batch) {
102            // Bio thread is gone (shutdown). The SendError carries
103            // the Vec, which drops here — every Box<Value> runs its
104            // Drop inline. Benign one-time stall during tear-down.
105        }
106    }
107}
kevy_store/bio_drop.rs

kevy_store/
bio_drop.rs
