diff --git a/CHANGELOG.md b/CHANGELOG.md index 19080794..578a12dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`INFO stats` reports what the pipeline ordering guarantee costs** (`total_pipeline_remote_defer`, + and the `moon_pipeline_remote_defer_total` Prometheus counter) — groundwork for #513. + + #512 made a pipelined command that cannot route by its own single key wait for the batch's + pending cross-shard commands, which is what stopped the silent write loss of #507. It is also + expensive, and nothing said so: the only symptom was throughput that looked bad for no + visible reason. + + Two things bound the cost, and the counter is what made both checkable: + + - A deferral needs an **undispatched cross-shard command already in the batch** — the guard + is `!remote_groups.is_empty() && must_wait_for_pending_remote(..)`. A shard-spanning `MGET` + on its own never defers: 64 spread `MGET`s with no preceding writes measure **0**. A + preceding foreign *read* counts too, since the E2 read fast path is disabled and foreign + reads are slotted alongside writes. + - **At most one deferral per batch pass.** The cut re-parses the tail with `remote_groups` + cleared, so the command at the head of the next pass runs inline whatever its shape. + + Together those explain why 64 interleavings produce fewer than 64 deferrals, and fewer still + at `--shards 2` (48) than at `--shards 4` (55) — with two shards, more of the preceding `SET`s + land locally and never reach `remote_groups` at all. The counts are shape- and + placement-specific, not constants: the same shape re-measured on a different key set gave 59. + + Measured on moon-dev (aarch64, 6 vCPU), one connection, 9 reps alternating leg order, median: + + | pipeline shape | shards=1 | shards=2 | shards=4 | + |---|---|---|---| + | `MGET` after every 2 `SET`s | 1,296,360 ops/s (0 deferrals) | 49,203 (48) | 38,856 (55) | + | 128 `SET`s then one `MGET` | 1,156,693 (0) | 659,436 (1) | 539,996 (1) | + | `SET`,`SET`,`GET` — routes by its own key | 1,700,287 (0) | 1,122,573 (0) | 993,784 (0) | + + The deferral counts are the server's own, not inferred: reading the code suggested 64 for the + first shape and the counter says 48, which is exactly why it exists. The `--shards 1` column + is the control — `remote_groups` is always empty there, so the guard structurally cannot fire. + ### Changed - **Sharded pub/sub channels are no longer workspace-scoped** (#703, fallout of #668). The hand-rolled workspace key walker had `PUBLISH`/`SUBSCRIBE`/`PSUBSCRIBE` in its no-key diff --git a/src/admin/metrics_setup/memory.rs b/src/admin/metrics_setup/memory.rs index 5cfbae5d..ebb92372 100644 --- a/src/admin/metrics_setup/memory.rs +++ b/src/admin/metrics_setup/memory.rs @@ -9,7 +9,8 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use metrics::gauge; use crate::admin::metrics_setup::{ - DISPATCH_CROSS_READ_SPSC_TOTAL, METRICS_INITIALIZED, TOTAL_CONNECTIONS, total_commands_sum, + DISPATCH_CROSS_READ_SPSC_TOTAL, METRICS_INITIALIZED, PIPELINE_REMOTE_DEFER_TOTAL, + TOTAL_CONNECTIONS, total_commands_sum, }; // ── Memory metrics ────────────────────────────────────────────────────── @@ -226,6 +227,15 @@ pub fn total_dispatch_cross_spsc() -> u64 { DISPATCH_CROSS_READ_SPSC_TOTAL.load(Ordering::Relaxed) } +/// Total pipeline batches cut short by the moon#507 ordering guard (moon#513). +/// +/// One per extra dispatch/await boundary. Always accurate — does not require +/// Prometheus to be initialised. +#[inline] +pub fn total_pipeline_remote_defer() -> u64 { + PIPELINE_REMOTE_DEFER_TOTAL.load(Ordering::Relaxed) +} + /// Read process CPU usage via `getrusage(RUSAGE_SELF)`. /// /// Returns `(used_cpu_sys, used_cpu_user)` in seconds (f64). diff --git a/src/admin/metrics_setup/mod.rs b/src/admin/metrics_setup/mod.rs index 81182002..5c62f44b 100644 --- a/src/admin/metrics_setup/mod.rs +++ b/src/admin/metrics_setup/mod.rs @@ -318,6 +318,12 @@ static WAL_AGGRESSIVE_RECYCLE_BYTES_TOTAL: AtomicU64 = AtomicU64::new(0); // SPSC (when fast-path is off) and writes. INFO exposes the unified total // as `total_dispatch_cross_spsc`. static DISPATCH_CROSS_READ_SPSC_TOTAL: AtomicU64 = AtomicU64::new(0); +/// moon#513: pipeline batches cut short because a command could not execute +/// against shards whose earlier writes in the same batch were still pending. +/// Each increment is one extra dispatch/await boundary — measured at ~57us on +/// moon-dev — so this is the counter that says whether a slow pipeline is +/// paying the moon#512 ordering guarantee or something else entirely. +static PIPELINE_REMOTE_DEFER_TOTAL: AtomicU64 = AtomicU64::new(0); // ── INFO-readable counter accessors ───────────────────────────────────── diff --git a/src/admin/metrics_setup/recorders.rs b/src/admin/metrics_setup/recorders.rs index 20b2270b..f66fc910 100644 --- a/src/admin/metrics_setup/recorders.rs +++ b/src/admin/metrics_setup/recorders.rs @@ -11,8 +11,8 @@ use metrics::{counter, gauge, histogram}; use crate::admin::metrics_setup::{ CONNECTED_CLIENTS, DISPATCH_CROSS_READ_SPSC_TOTAL, EVICTED_KEYS, EXPIRING_SPILL_SKIPPED, - KEYSPACE_HITS, KEYSPACE_MISSES, METRICS_INITIALIZED, SPILLED_KEYS, TOTAL_CONNECTIONS, - WAL_AGGRESSIVE_RECYCLE_BYTES_TOTAL, WAL_AGGRESSIVE_RECYCLE_SEGMENTS_TOTAL, + KEYSPACE_HITS, KEYSPACE_MISSES, METRICS_INITIALIZED, PIPELINE_REMOTE_DEFER_TOTAL, SPILLED_KEYS, + TOTAL_CONNECTIONS, WAL_AGGRESSIVE_RECYCLE_BYTES_TOTAL, WAL_AGGRESSIVE_RECYCLE_SEGMENTS_TOTAL, }; // ── Connection metrics ────────────────────────────────────────────────── @@ -324,6 +324,26 @@ pub fn record_xshard_reply_timeout(kind: &'static str) { counter!("moon_xshard_reply_timeout_total", "kind" => kind).increment(1); } +/// A pipeline batch was cut short by the moon#507 ordering guard. +/// +/// Recorded at the two sites that set `deferred_tail_from` for +/// `must_wait_for_pending_remote`: the command and the unconsumed tail move to +/// the next loop iteration so the pending `remote_groups` resolve first. +/// +/// Exists because the cost is invisible otherwise. A client interleaving reads +/// between write groups at `--shards >= 2` pays one boundary per interleaving, +/// and nothing in `INFO` said so — the throughput just looked bad. It is also +/// what keeps a benchmark for moon#513 honest: a shape that claims to trigger +/// the guard has to be able to PROVE it did, and how often. +#[inline] +pub fn record_pipeline_remote_defer() { + PIPELINE_REMOTE_DEFER_TOTAL.fetch_add(1, Ordering::Relaxed); + if !METRICS_INITIALIZED.load(Ordering::Relaxed) { + return; + } + counter!("moon_pipeline_remote_defer_total").increment(1); +} + /// Batched variant of `record_dispatch_cross_spsc`. #[inline] pub fn record_dispatch_cross_spsc_batch(count: u64) { diff --git a/src/command/connection.rs b/src/command/connection.rs index b0cbf8f2..b96814c5 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -657,6 +657,7 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { "total_commands_processed:{}\r\n\ total_connections_received:{}\r\n\ total_dispatch_cross_spsc:{}\r\n\ + total_pipeline_remote_defer:{}\r\n\ spsc_notify_wakes:{}\r\n\ spsc_drain_renotify:{}\r\n\ spsc_notify_skipped:{}\r\n\ @@ -664,6 +665,7 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { crate::admin::metrics_setup::total_commands_processed(), crate::admin::metrics_setup::total_connections_received(), crate::admin::metrics_setup::total_dispatch_cross_spsc(), + crate::admin::metrics_setup::total_pipeline_remote_defer(), crate::admin::metrics_setup::spsc_notify_wakes(), crate::admin::metrics_setup::spsc_drain_renotify(), crate::admin::metrics_setup::spsc_notify_skipped(), diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index e02d4d5d..d7c6815a 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1678,6 +1678,7 @@ pub(crate) async fn handle_connection_sharded_monoio< && crate::server::conn::shared::must_wait_for_pending_remote(cmd, cmd_args) { frames[frame_idx - 1] = frame; + crate::admin::metrics_setup::record_pipeline_remote_defer(); deferred_tail_from = Some(frame_idx - 1); break; } diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index fb733e17..3bbba278 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -818,6 +818,7 @@ pub(crate) async fn handle_connection_sharded_inner< && crate::server::conn::shared::must_wait_for_pending_remote(cmd, cmd_args) { batch[frame_idx - 1] = frame; + crate::admin::metrics_setup::record_pipeline_remote_defer(); deferred_tail_from = Some(frame_idx - 1); break; } diff --git a/tests/pipeline_cross_shard_ordering.rs b/tests/pipeline_cross_shard_ordering.rs index e10a942b..9b7b382a 100644 --- a/tests/pipeline_cross_shard_ordering.rs +++ b/tests/pipeline_cross_shard_ordering.rs @@ -166,6 +166,101 @@ fn each_trial(port: u16, tag: &str, mut body: impl FnMut(&mut Conn, &str) -> Opt // --------------------------------------------------------------------------- /// moon#507 as reported: MGET returns nulls for keys its own batch just wrote. +/// The ordering guarantee has a PRICE, and `INFO` now names it — moon#513. +/// +/// Every other test in this file asserts the guarantee holds. This one asserts +/// the cost of holding it is observable, because it was not: a client +/// interleaving reads between write groups at `--shards >= 2` pays one extra +/// dispatch/await boundary per interleaving, and nothing reported it. The +/// throughput simply looked bad. Measured on moon-dev, `--shards 2`, one +/// connection: an `MGET` after every two `SET`s runs at 49,203 ops/s against +/// 1,122,573 for the same shape with single-key `GET`s. +/// +/// It is also the acceptance criterion for moon#513, stated as a number rather +/// than a description: letting multi-key commands join the slotted batch means +/// `interleaved` stops deferring. When that lands, the assertion below flips +/// from "> 0" to "== 0" and the test stays meaningful either way. +/// +/// The `--shards 1` leg is what keeps it honest. There `remote_groups` is +/// always empty, so the guard structurally cannot fire — a counter that moved +/// there would be counting something else, and every claim built on it would +/// be wrong. +#[test] +fn pco12_the_ordering_guard_reports_what_it_costs() { + fn defers(port: u16) -> u64 { + let mut c = Conn::open(port); + let info = c.send(&["INFO", "stats"]); + info.split("\r\n") + .find_map(|l| l.strip_prefix("total_pipeline_remote_defer:")) + .and_then(|v| v.trim().parse().ok()) + .unwrap_or_else(|| { + panic!("INFO stats has no total_pipeline_remote_defer field: {info:?}") + }) + } + + /// `MGET` after every two `SET`s — the shape that triggers the guard. + fn interleaved(c: &mut Conn, tag: &str, n: usize) { + for i in (0..n).step_by(2) { + let (a, b) = (format!("{tag}:{i}"), format!("{tag}:{}", i + 1)); + c.pipeline(&[&["SET", &a, "v"], &["SET", &b, "v"], &["MGET", &a, &b]]); + } + } + + /// The same shape with single-key reads, which route by their own key and + /// therefore never need to wait. + fn single_key(c: &mut Conn, tag: &str, n: usize) { + for i in (0..n).step_by(2) { + let (a, b) = (format!("{tag}:{i}"), format!("{tag}:{}", i + 1)); + c.pipeline(&[&["SET", &a, "v"], &["SET", &b, "v"], &["GET", &a]]); + } + } + + // --- shards = 1: the guard cannot fire, whatever the shape --- + { + let m = spawn_moon("1"); + let base = defers(m.port); + let mut c = Conn::open(m.port); + interleaved(&mut c, "pco12s1", 64); + drop(c); + assert_eq!( + defers(m.port) - base, + 0, + "at --shards 1 `remote_groups` is always empty, so the moon#507 \ + guard cannot fire — a non-zero count here means the counter is \ + measuring something other than the deferral" + ); + } + + // --- shards = 4: the interleaved shape defers, the single-key one does not --- + let m = spawn_moon(SHARDS); + + let base = defers(m.port); + let mut c = Conn::open(m.port); + single_key(&mut c, "pco12ctl", 64); + drop(c); + let control_defers = defers(m.port) - base; + assert_eq!( + control_defers, 0, + "a pipeline of single-key commands routes every command by its own key \ + and must never defer — {control_defers} deferrals means the guard is \ + firing on the fast path, which would be a throughput regression for \ + every pipelined client" + ); + + let base = defers(m.port); + let mut c = Conn::open(m.port); + interleaved(&mut c, "pco12int", 64); + drop(c); + let interleaved_defers = defers(m.port) - base; + assert!( + interleaved_defers > 0, + "an MGET interleaved between writes at --shards {SHARDS} must trigger \ + the moon#507 guard, and INFO must say so. Zero here means either the \ + counter is not wired to the deferral site, or moon#513 has landed — \ + in which case flip this assertion to `== 0` rather than deleting it" + ); +} + #[test] fn pco1_mget_sees_writes_from_its_own_batch() { let m = spawn_moon(SHARDS);