1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
//! [`Shard::run`] — the epoll/kqueue readiness reactor loop — plus the
//! small path/routing helpers every reactor variant shares. Same
//! `impl<C: Commands> Shard<C>` as [`crate::shard`] (which owns the
//! struct); split out so that file stays under the 500-LOC house rule.
use crate::Commands;
use crate::shard::Shard;
use kevy_persist::{load_snapshot, replay_aof};
use std::io;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering, fence};
use std::time::{Duration, Instant};
impl<C: Commands> Shard<C> {
/// Owning shard of `key` under this server's routing scheme.
#[inline]
pub(crate) fn shard_of(&self, key: &[u8]) -> usize {
crate::reduce::shard_of(key, self.nshards, self.cluster.is_some())
}
/// This shard's snapshot file: `<data_dir>/dump-<id>.rdb`.
pub(crate) fn snapshot_path(&self) -> PathBuf {
kevy_persist::layout::snapshot_path(&self.data_dir, self.id)
}
/// This shard's append-only log: `<data_dir>/aof-<id>.aof`.
pub(crate) fn aof_path(&self) -> PathBuf {
kevy_persist::layout::aof_path(&self.data_dir, self.id)
}
// Busy-poll reactor main loop — per-iter overhead is the proven
// perf-sensitive surface here (perf-vs-foss §8 v1.30: per-iter
// amortization moves throughput where per-op µs shaving does not);
// stage extraction risks codegen change for zero readability win.
// LOC-WAIVER: busy-poll reactor main loop (per-iter perf-sensitive).
pub(crate) fn run(mut self, stop: Arc<AtomicBool>) -> io::Result<()> {
self.commands.on_shard_start(self.id);
// Restore: snapshot (state as of last SAVE) then replay the AOF (writes
// since that SAVE). The AOF is truncated at each SAVE, so this never
// double-applies. Replay goes straight to the store (no re-logging).
let snap = self.snapshot_path();
if snap.exists()
&& let Err(e) = load_snapshot(&mut self.store, &snap)
{
eprintln!(
"kevy: shard {} failed to load {}: {e}",
self.id,
snap.display()
);
}
if self.aof.is_some() {
let aof_path = self.aof_path();
let commands = &self.commands;
let store = &mut self.store;
replay_aof(&aof_path, |args| {
commands.dispatch(store, &args);
})?;
}
// v1.30 — off-accept-set shards have no listener (None); skip register.
let listener_fd = if let Some(l) = &self.listener {
l.set_nonblocking()?;
self.poller.add(l.raw(), true, false)?;
l.raw()
} else {
-1
};
self.poller.add(self.waker.read_fd(), true, false)?;
// -1 never matches an event fd, so the cluster-off loop below pays
// one dead integer compare per event and nothing else.
let mut cluster_fd = -1;
if let Some(cl) = &self.cluster_listener {
cl.set_nonblocking()?;
if self.arms_accept { self.poller.add(cl.raw(), true, false)?; }
cluster_fd = cl.raw();
}
// Same "fd or -1" trick for the replication listener (per Issue
// Ledger I2 — per-shard, deterministic ports). Replication-off
// pays one dead integer compare per event and nothing more.
let mut replication_fd = -1;
if let Some(rl) = &self.replication_listener {
rl.set_nonblocking()?;
self.poller.add(rl.raw(), true, false)?;
replication_fd = rl.raw();
}
let waker_fd = self.waker.read_fd();
let me = self.id;
let mut tick_interval = match self.commands.shard_tick_interval_ms() {
0 => None,
ms => Some(Duration::from_millis(ms)),
};
let mut last_tick = Instant::now();
let mut tick_check_counter: u32 = 0;
let mut idle_spins: u32 = 0;
while !stop.load(Ordering::Relaxed) {
// Busy-poll while there's recent work — a cross-core hop then costs
// no syscall. Park (blocking wait) once we've been idle a while.
let spinning = idle_spins < self.spin_limit;
let timeout = if spinning {
Some(0)
} else {
self.parked[me].store(true, Ordering::SeqCst);
// Close the park/wake race: the SeqCst fence pairs with
// the matching fence in `flush_wakes` on every other
// shard, so any push that lands BEFORE this drain on the
// peer's side is either (a) seen by `drain_inbound` here
// OR (b) the peer's parked-load saw `true` and a wake
// syscall is on the way. Without the fence, the lost-wake
// window was bounded by `PARK_TIMEOUT_MS` (50 ms) — the
// blocking wait below is now defense-in-depth (covers a
// missed eventfd write, OS scheduling glitch, etc.).
// Loom-verified by `tests/loom.rs::park_wake_fence_*`.
fence(Ordering::SeqCst);
if self.drain_inbound()? {
self.parked[me].store(false, Ordering::SeqCst);
self.flush_backlog();
self.flush_dirty()?;
self.flush_wakes();
idle_spins = 0;
continue;
}
Some(self.park_timeout_ms)
};
self.poller.wait(&mut self.events, timeout)?;
if !spinning {
self.parked[me].store(false, Ordering::SeqCst);
}
let mut did_work = !self.events.is_empty();
if did_work {
// Redis-style `updateCachedTime`: refresh the store's coarse
// clock once per batch, so the per-command read path's lazy
// expiry skips its own `Instant::now()` (amortized over the
// whole batch of events processed below).
self.store.refresh_clock();
// mem::take only when there's actually work, avoids two Vec
// moves per empty iter (timeout=Some(0) often returns 0).
let events = std::mem::take(&mut self.events);
for ev in &events {
if ev.fd == listener_fd {
self.accept_ready(false)?;
} else if ev.fd == cluster_fd {
self.accept_ready(true)?;
} else if ev.fd == replication_fd {
self.accept_ready_replication()?;
} else if ev.fd == waker_fd {
self.waker.drain();
} else if let Some(&conn_id) = self.fd_to_conn.get(&ev.fd) {
if ev.readable || ev.hup {
self.conn_readable(conn_id)?;
} else if ev.writable {
self.flush_conn(conn_id)?;
}
} else if let Some(idx) = self.replica_index_by_fd(ev.fd) {
if ev.readable || ev.hup {
self.replica_readable(idx)?;
}
// A handshake `+ACK` is small (≤ 30 B) and
// usually fits in the first non-blocking write,
// so try the drain unconditionally before
// requesting write-readiness. If it short-writes,
// `replica_writable` is a no-op until the poller
// signals writability (T1.14 wires the
// write-readiness re-arm; v1.18.0 ships with the
// assumption that `+ACK` drains in one syscall —
// which it does on every OS we test).
self.replica_writable(idx)?;
}
}
self.events = events;
// Drop conns that hit Closed mid-event (handshake
// error / peer EOF / `+ACK` drained in this batch's
// terminal state). Reaping before the next poll
// prevents a closed fd from re-firing on epoll level-
// triggered backends. E9: standalone shards skip even
// the gate inside the function.
if !self.replicas.is_empty() {
self.reap_closed_replicas();
}
}
// Messages from other cores (forwarded requests + replies to ours).
if self.drain_inbound()? {
did_work = true;
}
// Re-push anything that overflowed a full ring last iteration.
self.flush_backlog();
// Send this iteration's batched single-key dispatches (one per target).
self.flush_requests();
// Send this iteration's batched pub/sub deliveries (one per target).
self.flush_publish();
// Flush subscribers a PUBLISH wrote to this iteration.
self.flush_dirty()?;
// One wakeup per touched (and parked) target this iteration.
self.flush_wakes();
// v1.25 A.2: ship the per-shard bio-drop batch to the bio
// thread BEFORE the AOF fsync window. Same rationale as the
// io_uring path: don't let a pending fsync stall pin the
// batch in RSS, and bound the per-iter drop latency
// window. Empty-buffer fast path = predicted-not-taken
// length check, sub-ns on iters that did no overwrite.
self.store.flush_pending_drops();
// Honor the EverySec AOF fsync window.
if let Some(aof) = &mut self.aof {
let _ = aof.maybe_sync();
}
// Active TTL reaper / shard housekeeping. Skip the wall-clock
// read on most iters: in busy-poll the tick fires at 10 Hz
// with negligible overhead (counter saturates in ~us, then
// checks elapsed). In park mode each iter is already ≥ 1 ms
// so the throttle would delay the tick by 256 iters × 50 ms
// = ~12 s on a fully-idle server — bypass the counter when
// we just came back from a parking wait so the tick fires
// at every park iteration regardless of recent traffic.
if let Some(iv) = tick_interval {
tick_check_counter = tick_check_counter.wrapping_add(1);
if tick_check_counter >= self.tick_check_every || !spinning {
tick_check_counter = 0;
let now = Instant::now();
// BLOCK reactor: fire timeouts every tick gate (not gated
// by `iv`), so a `BLPOP k 0.5` resolves on the next 50ms
// park instead of the next user-level shard tick.
self.tick_blocked_timeouts();
self.tick_xshard_timeouts();
// v3.16: WAIT / REPL.WAIT deadline sweep — same
// cadence as the BLOCK timeout reactor above.
self.tick_repl_waiters();
if now.duration_since(last_tick) >= iv {
self.commands.on_shard_tick(&mut self.store);
self.apply_live_runtime_config(&mut tick_interval);
self.tick_persist();
// v3-cluster replication slot expiry (T1.15):
// drop slots whose reconnect window has passed.
// No-op short-circuits when replication is off or
// no slot has been recorded yet.
self.tick_replication_slots(now);
// v3-cluster ROLE / INFO replication (T1.28):
// publish master_repl_offset + connected_replicas
// count to the embedder. No-op when replication
// is off.
self.tick_replication_view();
// v3-cluster backlog watermark (T1.22.5): drop
// frames every consumer has moved past so the
// backlog reclaims space proactively. No-op
// when replication is off / no consumers yet.
self.tick_replication_watermark();
// v3-cluster server-as-replica (T1.29): drain
// events from the replica runner thread and
// apply them. No-op (one Option check) when
// this shard isn't a replica.
self.drain_replica_inbox();
last_tick = now;
}
}
}
// v3-cluster replication producer pump (Issue Ledger I2 +
// T1.14). OUTSIDE the did_work block since v3.14: the
// heartbeat (1s) and the ACK drain must run on idle iters
// too — a parked-and-woken shard with zero events still
// owes its replicas a pulse. Cost when replication is off
// stays one branch (E9 gating preserved).
if self.replicate.is_some() || !self.replicas.is_empty() {
self.pump_replication()?;
}
// A non-empty backlog means a peer ring is full: keep spinning so we
// re-attempt the flush (and keep draining inbound to unblock peers).
let has_backlog = self.backlog.iter().any(|b| !b.is_empty());
// v3.4: stay-hot-while-inflight, epoll symmetry — the
// uring reactor gained this in the v2.2 campaign (commit
// 65b7515): with forwarded cross-shard requests
// outstanding, replies land within ~one RTT, so hold the
// spin rung instead of paying park+wake per reply batch.
// Bounded: inflight only drains (owner answers) or the
// conn dies.
idle_spins = if did_work || has_backlog || self.xshard_inflight > 0 {
0
} else {
idle_spins.saturating_add(1)
};
}
// v1.25.x SAVE migration: drain any in-flight bg persist job
// before exit so a `Op::Save` that returned `+OK` to a client
// still lands its `dump-{i}.rdb` rename + AOF reset (the
// commit phase otherwise runs on the next tick, which won't
// happen after `stop=true`). See
// [`Self::drain_persist_on_shutdown`].
self.drain_persist_on_shutdown();
self.write_feed_shutdown_marker();
Ok(())
}
// `apply_live_runtime_config` + `maybe_auto_rewrite_aof` (the
// per-tick housekeeping) live in [`crate::shard_tick`] — same
// `impl<C: Commands> Shard<C>`, split out so this file stays under
// the 500-LOC house rule.
// The outbound transport half (`flush_wakes` / `flush_dirty` /
// `send_to` / `flush_backlog` / `flush_conn`) lives in
// [`crate::shard_flush`] — same `impl<C: Commands> Shard<C>`, split
// out so this file stays under the 500-LOC house rule.
}