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
//! io_uring setup face split from `uring_reactor.rs` (500-LOC rule):
//! ring capacities, the global availability probe, the per-shard
//! ring builder the spawn site uses for the auto-mode epoll fallback,
//! and the shard's pre-loop preparation.
use std::io;
use std::sync::Arc;
use kevy_persist::{load_snapshot, replay_aof};
use kevy_uring::IoUring;
use crate::Commands;
use crate::shard::Shard;
impl<C: Commands> Shard<C> {
/// Everything `run_uring` does before its first loop iteration:
/// the embedder's per-shard start hook (Lua / cross-shard registry
/// setup lives behind it — dropping it broke EVAL across shards on
/// the uring path only), the replica inbox's waker (see
/// replica_inbox.rs's wake contract), and the snapshot + AOF
/// restore, same as the readiness path.
pub(crate) fn prepare_uring_shard(&mut self) -> io::Result<()> {
self.announce_to_commands();
if let Some(rx) = &self.replica_inbox {
rx.attach_waker(Arc::clone(&self.waker));
}
let segs_dir = kevy_persist::layout::segs_dir(&self.data_dir, self.id);
self.store.enable_seg_rows(&segs_dir).map_err(std::io::Error::other)?;
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;
// In-replay demotion — same K-frame watermark
// drain as the readiness path's replay.
let mut frames: u64 = 0;
let mut torn: Option<String> = None;
let apply = |args: kevy_persist::Argv| {
if let Some(f) = kevy_persist::segmented_frame(&args) {
// Same stitch handling as the readiness path: a
// missing manifest entry is a named startup refusal.
if let Err(e) = kevy_store::apply_segmented(store, &segs_dir, f) {
torn.get_or_insert(e);
}
return;
}
crate::shard_run::replay_dispatch(commands, store, &args);
frames += 1;
if frames.is_multiple_of(kevy_persist::REPLAY_DEMOTE_INTERVAL) {
store.demote_to_watermark();
}
};
let report = if self.replay_resync {
kevy_persist::replay_aof_resync(&aof_path, apply)?
} else {
replay_aof(&aof_path, apply)?
};
if let Some(e) = torn {
return Err(std::io::Error::other(format!("shard {}: {e}", self.id)));
}
self.commands.on_replay_report(report.dropped_bytes, report.corrupt);
}
self.store.sweep_orphan_row_segs();
self.store.demote_to_watermark();
Ok(())
}
}
/// SQ/CQ depth per-shard. Paired with `PBUF_ENTRIES` — both were bumped
/// to fix the c=10 000 cliff (deco-axis-k-c10000).
pub(crate) const URING_ENTRIES: u32 = 2048;
// The nap rung was removed (see the idle-ladder comment in `run_uring`).
// URING_NAP_LIMIT / URING_NAP_MICROS / `uring_nap` are gone; spin →
// park is the whole ladder now.
/// Shared provided-buffer ring: 4096 × 16K = 64 MiB/shard. Linux multishot
/// recv terminates on ENOBUFS — must size for max conns (deco-axis-k-c10000).
pub(crate) const PBUF_ENTRIES: u16 = 4096;
pub(crate) const PBUF_SIZE: u32 = 16 * 1024;
pub(crate) const PBUF_GROUP: u16 = 0;
/// Probe whether this host can build the io_uring + provided-buffer ring that
/// [`Shard::run_uring`] needs: `io_uring_setup` not blocked by seccomp (Docker's
/// default profile blocks it) and a kernel new enough for the buf ring (5.19+).
/// Builds and immediately drops a real ring with the same parameters, so a
/// success here means `run_uring` will start. [`crate::Runtime`] calls this once
/// before spawning shards to auto-select io_uring with a graceful epoll fallback
/// — so an unavailable io_uring degrades to epoll instead of failing startup.
pub(crate) fn io_uring_available() -> bool {
match IoUring::new(URING_ENTRIES) {
Ok(ring) => ring.register_buf_ring(PBUF_ENTRIES, PBUF_SIZE, PBUF_GROUP).is_ok(),
Err(_) => false,
}
}
/// Build the per-shard ring pair (SQ/CQ ring + provided-buffer ring).
///
/// Split out of [`Shard::run_uring`] so the spawn site can
/// attempt it BEFORE committing the shard to the io_uring path — a
/// per-shard setup failure (ENOMEM under memory pressure with many
/// shards, rlimit exhaustion) in auto mode then falls back to the
/// epoll reactor for that shard instead of killing its thread. The
/// global [`io_uring_available`] probe only proves ONE ring can be
/// built; N shards need N rings.
///
/// `KEVY_SQPOLL=1` opts the ring into kernel-side SQ polling
/// (`IORING_SETUP_SQPOLL`, idle 1000 ms). Measurement-only switch:
/// SQPOLL spawns one kernel poll thread per shard competing for the
/// same core set as the shard threads, so it loses badly on a fully
/// subscribed box — it exists so the A/B stays reproducible whenever
/// the tradeoff is re-judged (spare-core layouts, kernel changes).
pub(crate) fn build_uring() -> io::Result<(IoUring, kevy_uring::ProvidedBufRing)> {
let sqpoll = matches!(
std::env::var("KEVY_SQPOLL").ok().as_deref(),
Some(v) if !v.is_empty() && v != "0" && v != "off" && v != "no" && v != "false"
);
let ring = if sqpoll {
IoUring::new_sqpoll(URING_ENTRIES, 1000, None)?
} else {
IoUring::new(URING_ENTRIES)?
};
let pbuf = ring.register_buf_ring(PBUF_ENTRIES, PBUF_SIZE, PBUF_GROUP)?;
Ok((ring, pbuf))
}