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
//! 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.commands.on_shard_start(self.id);
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))
}