From 8a259a8d6cd02aec6a78ac6e4ef22a2ce06a8460 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Tue, 25 Aug 2026 03:32:22 +0700 Subject: [PATCH] perf(observability): count what the pipeline ordering guard costs (#513) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #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 MGETs with no preceding writes measure 0. A preceding foreign READ counts too: the E2 read fast path is disabled, so 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 head of the next pass runs inline whatever its shape. Together those explain why 64 interleavings produce fewer than 64 deferrals, and fewer at --shards 2 (48) than --shards 4 (55): with two shards more of the preceding SETs land locally and never reach remote_groups. The counts are shape- and placement-specific, not constants -- the same shape on a different key set gave 59. Add `total_pipeline_remote_defer` to INFO stats (and the Prometheus counter `moon_pipeline_remote_defer_total`), recorded at the two sites that set `deferred_tail_from` for `must_wait_for_pending_remote` — the monoio and sharded handlers. This is NOT the #438 blocking-command site, which defers for an unrelated reason. Measured on moon-dev (aarch64, 6 vCPU), one connection, 9 reps alternating leg order, median of 60 flushes: pipeline shape shards=1 shards=2 shards=4 MGET after every 2 SETs 1,296,360 (0) 49,203 (48) 38,856 (55) 128 SETs then one MGET 1,156,693 (0) 659,436 (1) 539,996 (1) SET,SET,GET (own key) 1,700,287 (0) 1,122,573 (0) 993,784 (0) Parenthesised numbers are deferral counts from the new counter. They 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. The counter also keeps a #513 fix honest. `pco12` in tests/pipeline_cross_shard_ordering.rs asserts the interleaved shape triggers the guard at --shards 4 and that a single-key control does not; when #513 lands, that assertion flips from `> 0` to `== 0` rather than being deleted. Verified by mutation: unwiring the recorder fails pco12 and nothing else. Refs #513, #512, #507 author: Tin Dang --- CHANGELOG.md | 36 ++++++++++ src/admin/metrics_setup/memory.rs | 12 +++- src/admin/metrics_setup/mod.rs | 6 ++ src/admin/metrics_setup/recorders.rs | 24 ++++++- src/command/connection.rs | 2 + src/server/conn/handler_monoio/mod.rs | 1 + src/server/conn/handler_sharded/mod.rs | 1 + tests/pipeline_cross_shard_ordering.rs | 95 ++++++++++++++++++++++++++ 8 files changed, 174 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 19080794d..578a12dd0 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 5cfbae5d1..ebb92372c 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 81182002e..5c62f44ba 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 20b2270b9..f66fc9102 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 b0cbf8f2d..b96814c5f 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 e02d4d5d2..d7c6815a7 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 fb733e176..3bbba278d 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 e10a942b5..9b7b382a9 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);