diff --git a/CHANGELOG.md b/CHANGELOG.md index 49d24e14..ea9ee5f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,32 @@ 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`, + 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/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::*; 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));