From d644adebbcd4f3ccf18d40f070b5813355480f1f Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 8 Jul 2026 01:26:55 +0700 Subject: [PATCH 1/2] perf(persistence): coalesce each AOF group-commit batch into one write The monoio AOF writer was write-syscall-bound under appendfsync always with pipelining. strace -c on aof-writer-0 during an 8s always-P16 SET window (GCE c3-standard-8, 2 shards): write: 127,812 calls, 1.25s (~1 syscall per record, 9us each) fdatasync: 2,144 calls, 0.20s The writer spent 6x more time issuing write(2) than fsyncing: both monoio paths wrote each record individually on the raw unbuffered File -- the TopLevel path one write per message via commit_group_commit_batch, the PerShard framed path a header+body write PAIR per record. Redis batches ~120 records into one write via aof_buf (its main thread did 32,791 writes total for MORE throughput), which is where the remaining 0.85x always-P16 gap lived: each ~60-record batch paid ~540us of serialized write syscalls before its fsync could start. Fix, both monoio paths: - commit_group_commit_batch (TopLevel): multi-message batches coalesce into one contiguous buffer and ONE sink.write_all; single-message batches keep the zero-copy direct write. - PerShard framed loop: records are framed into a reusable batch_buf ([u64 lsn][u32 len] header + body, channel order) and written with ONE write_all; buffer high-water capped at 1MB. - tokio writer loops already amortize via BufWriter -- untouched. Failure semantics unchanged: a failed batch write acks every AppendSync waiter WriteFailed (none Synced) and engages the torn-stream latch; the single per-batch fsync still runs strictly after all bytes are written, so fsync-before-ack (H1) is preserved. The everysec path shares the same write code and gets the same syscall reduction. wal_group_commit sink tests updated to pin the coalesced contract: one write per batch, bytes concatenated in channel order, fsync after it, whole-batch write failure (no torn half-batch acks). Validation: - wal_group_commit 9/9 green (contract tests updated). - crash_matrix_per_shard_aof --ignored 3/3 green (SIGKILL durability contract intact, including appendfsync always). - fmt, clippy x2, cargo test --lib x2 (monoio 3868 / tokio 3143), unwrap ratchet green. - GCE always-P16 A/B vs Redis to be posted on the PR before merge. author: Tin Dang --- CHANGELOG.md | 17 +++++++++++++ src/persistence/aof/group_commit.rs | 27 +++++++++++++++++--- src/persistence/aof/writer_task.rs | 39 ++++++++++++++++++----------- tests/wal_group_commit.rs | 36 ++++++++++++++++++++------ 4 files changed, 94 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 49d24e14..c0b968ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed — AOF writer coalesces each group-commit batch into one write (PR #TBD) + +- The two monoio AOF writer paths (TopLevel via `commit_group_commit_batch`, + PerShard framed loop) wrote each record with its own `write(2)` on the raw + unbuffered `File` (the PerShard path even used a header+body pair). strace + during always-P16 SET: **127,812 write calls / 1.25 s** of an 8 s window vs + 2,144 fdatasyncs / 0.2 s — the writer thread was write-syscall-bound, not + fsync-bound, while Redis batches ~120 records per write via `aof_buf`. Each + batch's records now coalesce into one contiguous buffer (reusable in the + PerShard loop, capped at 1 MB high-water) and are written with ONE + `write_all` before the single per-batch fsync. Single-message batches keep + the zero-copy direct write. Failure semantics unchanged: a failed batch + write acks every waiter `WriteFailed` and engages the torn-stream latch. +- tokio writer loops already amortize via `BufWriter` — untouched. +- `wal_group_commit` sink tests updated to the coalesced contract (one write, + channel-order bytes, fsync after it). + ### Changed — AOF writer polls park-free under everysec/no (PR #TBD) - The two std-thread AOF writer loops (monoio TopLevel + PerShard) no longer diff --git a/src/persistence/aof/group_commit.rs b/src/persistence/aof/group_commit.rs index f1491ee0..48c9b315 100644 --- a/src/persistence/aof/group_commit.rs +++ b/src/persistence/aof/group_commit.rs @@ -230,10 +230,29 @@ pub fn commit_group_commit_batch( batch: &mut GroupCommitBatch, do_fsync: bool, ) -> CommitOutcome { - // Step 1 — write every data message's bytes in channel order. - for msg in &batch.data { - if sink.write_all(msg_body(msg)).is_err() { - return ack_batch(batch, BatchAck::WriteFailed); + // Step 1 — write the batch's bytes in channel order. Multi-message + // batches are coalesced into ONE contiguous `write_all` (the sink is a + // raw unbuffered File on the monoio paths, so per-message writes meant + // one write(2) PER RECORD — measured as the dominant writer-thread cost + // under always-P16: 127,812 write calls / 1.25s of an 8s strace window + // vs 2,144 fdatasyncs / 0.2s; Redis batches ~120 records per write via + // aof_buf). Single-message batches keep the zero-copy direct write. + match batch.data.len() { + 0 => {} + 1 => { + if sink.write_all(msg_body(&batch.data[0])).is_err() { + return ack_batch(batch, BatchAck::WriteFailed); + } + } + _ => { + let total: usize = batch.data.iter().map(|m| msg_body(m).len()).sum(); + let mut buf = Vec::with_capacity(total); + for msg in &batch.data { + buf.extend_from_slice(msg_body(msg)); + } + if sink.write_all(&buf).is_err() { + return ack_batch(batch, BatchAck::WriteFailed); + } } } // Step 2 — exactly ONE fsync, AFTER all writes, only under Always. diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs index 4afa32be..1af32663 100644 --- a/src/persistence/aof/writer_task.rs +++ b/src/persistence/aof/writer_task.rs @@ -1437,6 +1437,9 @@ pub async fn per_shard_aof_writer_task( let mut write_error = false; let mut _dbg_processed: u64 = 0; let _dbg_start = Instant::now(); + // Reusable frame-coalescing buffer: one contiguous write_all per + // group-commit batch instead of a header+body write pair per record. + let mut batch_buf: Vec = Vec::new(); // Idle-adaptive channel-poll wake cadence (RSS/CPU wave 5, item B) — // see `IdleWait` docs near the top of this file. let mut idle_wait = IdleWait::new(); @@ -1510,9 +1513,15 @@ pub async fn per_shard_aof_writer_task( // waiter so callers error instead of acking a corrupt write. let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed); } else { - // Write each record framed, header + body, in channel order; - // the single fsync below covers them all. - let mut write_failed = false; + // Frame each record (`[u64 lsn][u32 len]` header + body, in + // channel order) into the reusable batch buffer and issue + // ONE write_all for the whole batch; the single fsync below + // covers it all. The old per-record header+body write_all + // pair on the raw File was the dominant writer-thread cost + // under always-P16 (measured 127,812 write(2) calls / 1.25s + // of an 8s strace window vs 2,144 fdatasyncs / 0.2s — Redis + // batches ~120 records per write via aof_buf). + batch_buf.clear(); for msg in &batch.data { let (lsn, data) = match msg { AofMessage::Append { lsn, bytes } @@ -1529,23 +1538,25 @@ pub async fn per_shard_aof_writer_task( let mut header = [0u8; 12]; header[..8].copy_from_slice(&lsn.to_le_bytes()); header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes()); - if let Err(e) = file.write_all(&header) { - error!( - "AOF header write failed shard {} (seq {}): {}. Persistence degraded.", - shard_id, manifest.seq, e - ); - write_failed = true; - break; - } - if let Err(e) = file.write_all(data) { + batch_buf.extend_from_slice(&header); + batch_buf.extend_from_slice(data); + } + let mut write_failed = false; + if !batch_buf.is_empty() { + if let Err(e) = file.write_all(&batch_buf) { error!( - "AOF write failed shard {} (seq {}): {}. Persistence degraded.", + "AOF batch write failed shard {} (seq {}): {}. Persistence degraded.", shard_id, manifest.seq, e ); write_failed = true; - break; } } + // Cap the reusable buffer's high-water mark: a rare burst of + // large values (up to AOF_GROUP_COMMIT_MAX_BYTES) must not + // pin megabytes on the writer thread forever. + if batch_buf.capacity() > 1 << 20 { + batch_buf = Vec::new(); + } let do_fsync = matches!(fsync, FsyncPolicy::Always); let verdict = if write_failed { diff --git a/tests/wal_group_commit.rs b/tests/wal_group_commit.rs index 3290eab4..89578ef8 100644 --- a/tests/wal_group_commit.rs +++ b/tests/wal_group_commit.rs @@ -207,11 +207,21 @@ fn commit_one_fsync_many_acks() { let outcome = commit_group_commit_batch(&mut sink, &mut batch, true); assert_eq!(sink.sync_calls, 1, "exactly ONE fsync for the whole batch"); - assert_eq!(sink.writes.len(), 3, "all three payloads written"); + // Multi-message batches are coalesced into ONE contiguous write (the + // always-P16 write-syscall fix); channel order must be preserved inside it. + assert_eq!( + sink.writes.len(), + 1, + "one coalesced write for the whole batch" + ); + assert_eq!( + sink.writes[0], b"abc", + "all three payloads written in channel order" + ); assert_eq!( sink.synced_at_write_count, - Some(3), - "the fsync runs AFTER all writes (ack-after-fsync ordering)" + Some(1), + "the fsync runs AFTER the batch write (ack-after-fsync ordering)" ); assert_eq!( outcome, @@ -237,7 +247,9 @@ fn commit_write_fail_acks_write_failed() { deferred_control: None, }; let mut sink = CountingSink::new(); - sink.fail_write_at = Some(1); // the 2nd write fails + // Multi-message batches coalesce into a single write — fail THAT write + // (index 0). The whole batch fails together: no torn half-batch acks. + sink.fail_write_at = Some(0); let outcome = commit_group_commit_batch(&mut sink, &mut batch, true); @@ -304,7 +316,8 @@ fn commit_everysec_no_fsync() { sink.sync_calls, 0, "everysec (do_fsync=false) must not fsync per batch" ); - assert_eq!(sink.writes.len(), 2, "bytes are still written in order"); + assert_eq!(sink.writes.len(), 1, "one coalesced write for the batch"); + assert_eq!(sink.writes[0], b"xy", "bytes are still written in order"); assert_eq!( outcome.synced, 0, "no AppendSync waiters in an everysec batch" @@ -330,10 +343,19 @@ fn commit_barrier_covers_preceding_appends() { sink.sync_calls, 1, "one fsync covers the appends + the barrier" ); + assert_eq!( + sink.writes.len(), + 1, + "the appends + zero-length barrier coalesce into one write" + ); + assert_eq!( + sink.writes[0], b"onetwo", + "append bytes precede the fsync in channel order" + ); assert_eq!( sink.synced_at_write_count, - Some(3), - "the fsync runs after the two appends AND the barrier are written" + Some(1), + "the fsync runs after the appends AND the barrier are written" ); assert_eq!(outcome.synced, 1, "the barrier AppendSync is acked"); assert_eq!(rx.try_recv(), Ok(AofAck::Synced)); From 7c1b911c71174aca8d672750a9dce1f2408d745a Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 8 Jul 2026 02:13:49 +0700 Subject: [PATCH 2/2] fix(server): cfg-gate accept-backoff libc errnos for Windows is_resource_exhaustion (accept-loop backoff, PR #230) matched libc::EMFILE/ENFILE/ENOBUFS/ENOMEM without a cfg guard; libc is not linked on Windows targets, so every main-push Windows check has failed E0433 since the merge (PR CI never caught it -- the Windows job is skipped on PRs). Split the function: unix keeps the errno match, Windows matches WSAEMFILE (10024) / WSAENOBUFS (10055) and ErrorKind::OutOfMemory. author: Tin Dang --- CHANGELOG.md | 9 +++++++++ src/server/accept_backoff.rs | 10 ++++++++++ 2 files changed, 19 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c0b968ec..ea9ee5f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — Windows build: `accept_backoff` used unix-only `libc` errnos (PR #TBD) + +- `is_resource_exhaustion` (accept-loop backoff, PR #230) referenced + `libc::EMFILE`/`ENFILE`/`ENOBUFS`/`ENOMEM` unguarded; `libc` is not linked + on Windows, breaking the main-push Windows check (PRs never caught it — + Windows CI is skipped on PRs). Now cfg-split: unix keeps the errno match, + Windows matches WSAEMFILE (10024) / WSAENOBUFS (10055) / + `ErrorKind::OutOfMemory`. + ### Changed — AOF writer coalesces each group-commit batch into one write (PR #TBD) - The two monoio AOF writer paths (TopLevel via `commit_group_commit_batch`, diff --git a/src/server/accept_backoff.rs b/src/server/accept_backoff.rs index 6f6644a8..06da1008 100644 --- a/src/server/accept_backoff.rs +++ b/src/server/accept_backoff.rs @@ -87,6 +87,7 @@ fn backoff_for(consecutive: u32) -> Duration { /// Errors that indicate the process/system is out of a resource needed to /// accept — retrying immediately would just spin. Everything else is treated /// as a transient per-connection error (logged, loop continues, no sleep). +#[cfg(unix)] fn is_resource_exhaustion(err: &std::io::Error) -> bool { matches!( err.raw_os_error(), @@ -94,6 +95,15 @@ fn is_resource_exhaustion(err: &std::io::Error) -> bool { ) } +/// Windows variant — `libc` is not linked there. WSAEMFILE (10024) and +/// WSAENOBUFS (10055) are winsock's fd/buffer exhaustion; `OutOfMemory` +/// covers the ENOMEM-alikes the kind mapping recognizes. +#[cfg(not(unix))] +fn is_resource_exhaustion(err: &std::io::Error) -> bool { + matches!(err.raw_os_error(), Some(10024) | Some(10055)) + || matches!(err.kind(), std::io::ErrorKind::OutOfMemory) +} + #[cfg(test)] mod tests { use super::*;