kevy_rt/runtime.rs
1//! The public entry point: configure and run the thread-per-core server.
2
3use crate::Commands;
4use kevy_persist::Fsync;
5use std::path::PathBuf;
6
7/// Default slots in each per-core-pair SPSC ring. A full ring spills
8/// to a local backlog (see [`Shard`]), so this only bounds the
9/// lock-free fast path, not capacity. Overridable via the
10/// `[advanced] ring_capacity` config field threaded through
11/// [`Runtime::with_advanced`].
12const DEFAULT_RING_CAPACITY: usize = 1024;
13
14/// The public entry point: configure and run the thread-per-core server.
15pub struct Runtime<C: Commands> {
16 pub(crate) ip: [u8; 4],
17 pub(crate) port: u16,
18 pub(crate) nshards: usize,
19 pub(crate) commands: C,
20 /// Directory for per-shard snapshot files (`dump-<id>.rdb`) and AOF logs.
21 pub(crate) data_dir: PathBuf,
22 /// Whether the append-only log is enabled.
23 pub(crate) enable_aof: bool,
24 /// fsync policy for the AOF. Default `EverySec` matches Redis.
25 pub(crate) appendfsync: Fsync,
26 /// auto-trigger BGREWRITEAOF when AOF grew this many % above the size
27 /// at the previous rewrite. `0` disables. Default `100` (matches Redis).
28 pub(crate) auto_aof_rewrite_pct: u32,
29 pub(crate) auto_aof_rewrite_bytes: u64,
30 pub(crate) auto_aof_rewrite_interval_secs: u64,
31 pub(crate) replay_resync: bool,
32 /// Floor below which auto-rewrite is skipped. Default `64 MiB`.
33 pub(crate) auto_aof_rewrite_min_size: u64,
34 /// Reactor SPSC ring slot count. See [`DEFAULT_RING_CAPACITY`].
35 pub(crate) ring_capacity: usize,
36 /// Reactor busy-poll iter limit before parking. Stored as `u32`
37 /// for the per-shard counter; the [`Shard`] field carries it
38 /// forward into the loop.
39 pub(crate) spin_limit: u32,
40 /// `Some(N)` = only shards `0..N` arm accept SQE. `None`
41 /// = every shard accepts (the default; byte-identical to the
42 /// pre-flag behaviour).
43 pub(crate) accept_shards: Option<usize>,
44 /// Total cap on active client conns. `0` = unlimited.
45 pub(crate) max_clients: usize,
46 /// Reactor blocking-wait timeout in ms when parked.
47 pub(crate) park_timeout_ms: u32,
48 /// Wall-clock-read throttle for the tick check (TTL reaper / live
49 /// config refresh / auto-AOF-rewrite).
50 pub(crate) tick_check_every: u32,
51 /// `[slowlog].slower_than_micros`. Default: `-1` (OFF — zero
52 /// hot-path cost: every command would otherwise pay an
53 /// `Instant::now()` pair around dispatch). Set to `10_000` to match
54 /// Redis's default 10 ms threshold; see [`Self::with_slowlog`] /
55 /// `CONFIG SET slowlog-log-slower-than 10000`.
56 pub(crate) slowlog_slower_than_micros: i64,
57 /// `[slowlog].max_len`. Per-shard cap.
58 pub(crate) slowlog_max_len: u32,
59 /// Single-node cluster mode: slot-based key routing (CRC16 `{hashtag}`
60 /// → contiguous ranges) + one deterministic extra listener per shard at
61 /// `cluster_port_base + id`. `None` = off (default, zero change).
62 pub(crate) cluster_port_base: Option<u16>,
63 /// Replication: when `true`, each shard runs a
64 /// `ReplicationSource` with `replication_buffer_size` byte budget;
65 /// every applied mutation is pushed to the backlog. This wires
66 /// only the producer side — the TCP listener + streaming loop are
67 /// gated separately by `replication_port_base`. Default `false`.
68 pub(crate) enable_replication: bool,
69 /// FEED.* consumer surface. When set, every shard keeps a
70 /// backlog (even with no replicas) and persists the (generation,
71 /// offset) cursor via the feed sidecars.
72 pub(crate) feed_enabled: bool,
73 /// Per-shard backlog byte budget for the feed (`[feed]
74 /// feed_buffer_size`); the effective budget is
75 /// `max(replication_buffer_size, feed_buffer_size)` when both
76 /// features are on (one backlog, two readers).
77 pub(crate) feed_buffer_size: u64,
78 /// Per-shard backlog byte budget when `enable_replication` is set.
79 /// Fed from `[replication] replication_buffer_size`. Default
80 /// `256 MiB` (matches the kevy-config default).
81 pub(crate) replication_buffer_size: u64,
82 /// v3-cluster replication listener: shard `i` binds at
83 /// `replication_port_base + i` (mirrors cluster listener pattern;
84 /// per Issue Ledger I2). `None` = no listener (producer side runs
85 /// without a network surface, backlog accumulates and evicts —
86 /// useful for benchmarks). Default `None`.
87 pub(crate) replication_port_base: Option<u16>,
88 /// Per-shard SlotTable reconnect-window in ms. After a
89 /// streaming replica disconnects, its `(replica_id, sent_offset)`
90 /// is recorded in the shard's `slots` map; slots past this age
91 /// are reaped on the next shard tick. Default `60_000` (60 s)
92 /// matches the kevy-config default.
93 pub(crate) replication_reconnect_window_ms: u32,
94 /// Per-shard replica inboxes installed by
95 /// [`Self::with_replica_inboxes`]. Each entry is consumed
96 /// (via `Option::take`) when its shard is constructed, so the
97 /// receiver flows from this Vec to the matching `Shard.replica_inbox`.
98 /// Empty when no replica mode is configured.
99 pub(crate) replica_inboxes: Vec<Option<crate::replica_inbox::ReplicaInboxReceiver>>,
100 /// UDS: when `Some(path)`, ALSO bind a Unix-domain stream
101 /// listener at `path` on shard 0 (single global socket, like valkey's
102 /// `unixsocket` config). Lets benches/local clients skip TCP loopback
103 /// overhead. TCP listener stays bound regardless.
104 #[allow(dead_code)] // consumed during run() via take-into-Shard
105 pub(crate) unix_socket_path: Option<PathBuf>,
106 /// Transparent-tiering RAM budget for the WHOLE process, in
107 /// resolved bytes (the server resolves auto/percent forms before
108 /// building the runtime). Split evenly across shards at
109 /// construction. `None` = check the minimal `KEVY_TIER_BUDGET`
110 /// plain-bytes env knob (back-compat), else tiering off.
111 pub(crate) tier_budget: Option<u64>,
112 /// Cold-tier spill dir override (`[tiering] spill_dir`). `None` =
113 /// `<data_dir>/tier/`.
114 pub(crate) tier_dir: Option<PathBuf>,
115}
116
117impl<C: Commands> Runtime<C> {
118 /// Start configuring a runtime for `commands`. `Runtime` is its own
119 /// builder: chain [`Self::bind`] / [`Self::shards`] / the `with_*`
120 /// setters, then call `run`. Defaults: bind `127.0.0.1:6004`, one
121 /// shard, AOF on (`EverySec`), data dir `"."`.
122 #[must_use]
123 pub fn builder(commands: C) -> Self {
124 Runtime {
125 ip: [127, 0, 0, 1],
126 port: 6004,
127 nshards: 1,
128 commands,
129 data_dir: PathBuf::from("."),
130 enable_aof: true,
131 appendfsync: Fsync::EverySec,
132 auto_aof_rewrite_pct: 100,
133 auto_aof_rewrite_bytes: 0,
134 auto_aof_rewrite_interval_secs: 0,
135 replay_resync: false,
136 auto_aof_rewrite_min_size: 64 * 1024 * 1024,
137 ring_capacity: DEFAULT_RING_CAPACITY,
138 spin_limit: 256,
139 accept_shards: None,
140 max_clients: 10_000,
141 park_timeout_ms: 50,
142 tick_check_every: 256,
143 slowlog_slower_than_micros: -1,
144 slowlog_max_len: 128,
145 cluster_port_base: None,
146 enable_replication: false,
147 feed_enabled: false,
148 feed_buffer_size: 64 * 1024 * 1024,
149 replica_inboxes: Vec::new(),
150 replication_buffer_size: 256 * 1024 * 1024,
151 replication_port_base: None,
152 replication_reconnect_window_ms: 60_000,
153 unix_socket_path: None,
154 tier_budget: None,
155 tier_dir: None,
156 }
157 }
158
159 /// The process-level tiering budget in bytes (`None` = env-knob
160 /// fallback / off) and per-shard slice of it. The vlog dir root is
161 /// [`Self::tier_root`].
162 pub(crate) fn resolved_tier_budget(&self) -> Option<u64> {
163 self.tier_budget
164 .or_else(|| std::env::var("KEVY_TIER_BUDGET").ok().and_then(|v| v.parse::<u64>().ok()))
165 }
166
167 /// One shard's slice of the process tiering budget (even split,
168 /// floored at 1 byte so a tiny budget still tiers rather than
169 /// silently disabling).
170 pub(crate) fn per_shard_tier_budget(total: u64, nshards: usize) -> u64 {
171 (total / nshards.max(1) as u64).max(1)
172 }
173
174 /// The cold-tier root dir: the `[tiering] spill_dir` override, or
175 /// `<data_dir>/tier/`.
176 pub(crate) fn tier_root(&self) -> PathBuf {
177 self.tier_dir.clone().unwrap_or_else(|| self.data_dir.join("tier"))
178 }
179
180 /// Listen address for the client TCP listener (every shard binds it
181 /// via SO_REUSEPORT). Default `127.0.0.1:6004`.
182 #[must_use]
183 pub fn bind(mut self, ip: [u8; 4], port: u16) -> Self {
184 self.ip = ip;
185 self.port = port;
186 self
187 }
188
189 /// Shard (reactor thread) count. Clamped to at least 1. Default 1.
190 #[must_use]
191 pub fn shards(mut self, n: usize) -> Self {
192 self.nshards = n.max(1);
193 self
194 }
195
196 /// Spawn one thread per shard and run until `stop` is set.
197 /// UDS: also bind a Unix-domain stream listener at `path`. Lets
198 /// local clients (and benchmarks) skip the TCP loopback round-trip.
199 /// Bound on shard 0 only (no SO_REUSEPORT for AF_UNIX, single global
200 /// socket like valkey's `unixsocket` config). TCP listener stays
201 /// bound at the configured `port` regardless.
202 #[must_use]
203 pub fn with_unix_socket(mut self, path: PathBuf) -> Self {
204 self.unix_socket_path = Some(path);
205 self
206 }
207
208 // `run` (and its per-stage build helpers) lives in
209 // [`crate::runtime_run`] — same `impl<C: Commands> Runtime<C>`,
210 // split out so this file stays under the 500-LOC house rule.
211}