From d924656c1944ced34bec14b7032966822448b229 Mon Sep 17 00:00:00 2001 From: Jepson Taylor <130782313+jepsontaylor@users.noreply.github.com> Date: Fri, 25 Sep 2026 19:32:41 +0000 Subject: [PATCH] Use positional I/O and write contiguous WAL records together. Page and WAL transfers no longer seek before each read or write. The WAL writer still splits a batch at a segment boundary, then issues one write for each contiguous run. An exact fill rotates before the next record, including when the cursor sits on the segment length and the next reserved LSN would otherwise compare equal. Record bytes and durability are unchanged. wal_pipeline stays off. Claim: AGENT_CHAT.md 2026-09-25T19:40Z perf/positional-wal-batch. Acceptance: cargo test -p redlinedb-kernel --offline --lib positional_reads; cargo test -p redlinedb-kernel --offline --test wal_tests --- AGENT_CHAT.md | 4 + crates/kernel/src/io/fs.rs | 64 ++++++++++++-- .../src/wal/manager/coordinator/helpers.rs | 4 +- .../src/wal/manager/coordinator/writer.rs | 10 +-- .../kernel/src/wal/manager/storage/write.rs | 84 ++++++++++++++++--- crates/kernel/tests/wal_tests.rs | 26 ++++++ 6 files changed, 166 insertions(+), 26 deletions(-) diff --git a/AGENT_CHAT.md b/AGENT_CHAT.md index 53fcd0c00..f287c0b03 100644 --- a/AGENT_CHAT.md +++ b/AGENT_CHAT.md @@ -1,5 +1,9 @@ # RedlineDB Agent Chat +## 2026-09-25T19:40Z perf/positional-wal-batch + +Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `perf/positional-wal-batch` from `origin/main` `4c622a4fad6211133cd2bcd962375bca38170d0d`. Page and WAL bytes use positional read/write. The WAL writer emits one write for a contiguous run of records and splits the run at a segment boundary. WAL record bytes and durability are unchanged. `wal_pipeline` stays off. No speed ratio. No worktree. + ## 2026-09-25T19:10Z perf/one-pass-scan Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `perf/one-pass-scan` from `origin/main` `1e21f23caa5842ca2469ec444a6edc4ac4bf36fa`. An unordered table scan keeps the row from the visibility load instead of decoding it again. Index probes stay rowid lookups. Join order and `LIMIT 1` stay as locked. No speed ratio. No worktree. diff --git a/crates/kernel/src/io/fs.rs b/crates/kernel/src/io/fs.rs index 66ed52b62..988aa6588 100644 --- a/crates/kernel/src/io/fs.rs +++ b/crates/kernel/src/io/fs.rs @@ -1,7 +1,10 @@ use std::fs::{self, File, OpenOptions}; -use std::io::{Read, Seek, SeekFrom, Write}; +use std::io::{self, ErrorKind}; use std::path::Path; +#[cfg(unix)] +use std::os::unix::fs::FileExt; + use crate::Result; pub trait FileHandle { @@ -75,15 +78,11 @@ impl FileHandle for StdFileHandle { } fn read_exact_at(&mut self, offset: u64, buf: &mut [u8]) -> Result<()> { - self.0.seek(SeekFrom::Start(offset))?; - self.0.read_exact(buf)?; - Ok(()) + read_exact_at(&self.0, offset, buf) } fn write_all_at(&mut self, offset: u64, buf: &[u8]) -> Result<()> { - self.0.seek(SeekFrom::Start(offset))?; - self.0.write_all(buf)?; - Ok(()) + write_all_at(&self.0, offset, buf) } fn sync_data(&self) -> Result<()> { @@ -96,3 +95,54 @@ impl FileHandle for StdFileHandle { Ok(()) } } + +#[cfg(unix)] +fn read_exact_at(file: &File, offset: u64, buf: &mut [u8]) -> Result<()> { + let mut filled = 0; + while filled < buf.len() { + let n = file.read_at(&mut buf[filled..], offset + filled as u64)?; + if n == 0 { + return Err( + io::Error::new(ErrorKind::UnexpectedEof, "failed to fill whole buffer").into(), + ); + } + filled += n; + } + Ok(()) +} + +#[cfg(unix)] +fn write_all_at(file: &File, offset: u64, buf: &[u8]) -> Result<()> { + let mut written = 0; + while written < buf.len() { + let n = file.write_at(&buf[written..], offset + written as u64)?; + if n == 0 { + return Err( + io::Error::new(ErrorKind::WriteZero, "failed to write whole buffer").into(), + ); + } + written += n; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn positional_reads_do_not_depend_on_write_order() { + let dir = tempfile::tempdir().expect("temp dir"); + let mut file = StdFileSystem + .open_rw_create(&dir.path().join("bytes")) + .expect("open"); + file.write_all_at(8, b"xyz").expect("tail"); + file.write_all_at(0, b"abcd").expect("head"); + let mut head = [0_u8; 4]; + let mut tail = [0_u8; 3]; + file.read_exact_at(0, &mut head).expect("read head"); + file.read_exact_at(8, &mut tail).expect("read tail"); + assert_eq!(&head, b"abcd"); + assert_eq!(&tail, b"xyz"); + } +} diff --git a/crates/kernel/src/wal/manager/coordinator/helpers.rs b/crates/kernel/src/wal/manager/coordinator/helpers.rs index d5150aade..feac2965a 100644 --- a/crates/kernel/src/wal/manager/coordinator/helpers.rs +++ b/crates/kernel/src/wal/manager/coordinator/helpers.rs @@ -136,12 +136,12 @@ pub(super) fn drain_until( } let mut last_written = Lsn::ZERO; - for record in batch { + for record in &batch { last_written = record.append.end_lsn; totals.records = totals.records.saturating_add(1); totals.bytes = totals.bytes.saturating_add(record.encoded.len() as u64); - wal.write_encoded(record.append, &record.encoded)?; } + wal.write_encoded_batch(&batch)?; if last_written != Lsn::ZERO { publish_written_lsn(shared, last_written); } diff --git a/crates/kernel/src/wal/manager/coordinator/writer.rs b/crates/kernel/src/wal/manager/coordinator/writer.rs index 24bf2d8eb..9965d8a5b 100644 --- a/crates/kernel/src/wal/manager/coordinator/writer.rs +++ b/crates/kernel/src/wal/manager/coordinator/writer.rs @@ -73,17 +73,17 @@ pub(super) fn wal_writer_loop( } let mut last_written = Lsn::ZERO; - for record in batch { + for record in &batch { last_written = record.append.end_lsn; // Lane GC: accumulate before the write so a torn-write // failpoint doesn't desync the counter from durable // state — the failure path returns immediately. group_records = group_records.saturating_add(1); group_bytes = group_bytes.saturating_add(record.encoded.len() as u64); - if let Err(_err) = wal.write_encoded(record.append, &record.encoded) { - publish_wal_failure(&shared); - return; - } + } + if let Err(_err) = wal.write_encoded_batch(&batch) { + publish_wal_failure(&shared); + return; } if last_written != Lsn::ZERO { publish_written_lsn(&shared, last_written); diff --git a/crates/kernel/src/wal/manager/storage/write.rs b/crates/kernel/src/wal/manager/storage/write.rs index 893731a4b..ab58ea857 100644 --- a/crates/kernel/src/wal/manager/storage/write.rs +++ b/crates/kernel/src/wal/manager/storage/write.rs @@ -190,23 +190,71 @@ impl WalManager { } pub fn write_encoded(&mut self, append: WalAppend, encoded: &[u8]) -> Result<()> { - // Lane E failpoint: armed before the WAL segment write so harnesses - // can inject torn-write or panic faults at the precise moment the - // record would land on disk. - crate::fail_point!("wal::write_encoded"); - if encoded.len() > self.config.segment_bytes as usize { - return Err(Error::CorruptWal("record larger than wal segment")); + self.write_encoded_batch(&[QueuedWalRecord { + append, + encoded: encoded.to_vec(), + }]) + } + + /// Write a drained queue in as few positional writes as possible. + /// A run stops at a segment boundary or a gap in LSN. Each run is one + /// `write_all_at` and one `wal::write_encoded` failpoint. Record bytes + /// are unchanged. + pub(crate) fn write_encoded_batch(&mut self, records: &[QueuedWalRecord]) -> Result<()> { + let mut index = 0; + while index < records.len() { + self.prepare_for_record(records[index].append, records[index].encoded.len())?; + let offset = self.active_offset; + let mut bytes = Vec::new(); + let mut last_start = records[index].append.start_lsn; + let mut last_end = records[index].append.end_lsn; + loop { + let record = &records[index]; + let at = offset + bytes.len() as u64; + let expected = Lsn((self.active_segment - 1) * self.config.segment_bytes + at); + if record.append.start_lsn != expected { + break; + } + if at > 0 && at + record.encoded.len() as u64 > self.config.segment_bytes { + break; + } + bytes.extend_from_slice(&record.encoded); + last_start = record.append.start_lsn; + last_end = record.append.end_lsn; + index += 1; + if index == records.len() { + break; + } + } + if bytes.is_empty() { + return Err(Error::CorruptWal( + "record lsn does not match write position", + )); + } + self.emit_wal_bytes(offset, &bytes, last_end, last_start)?; } + Ok(()) + } + fn prepare_for_record(&mut self, append: WalAppend, encoded_len: usize) -> Result<()> { + if encoded_len > self.config.segment_bytes as usize { + return Err(Error::CorruptWal("record larger than wal segment")); + } + // An exact fill leaves the cursor at `segment_bytes`. The next + // reserved LSN is the following segment, and that LSN compares + // equal to `(segment - 1) * size + offset` when offset == size, + // so the mismatch check below would not rotate. + if self.active_offset >= self.config.segment_bytes { + self.rotate_segment()?; + } let expected_lsn = Lsn((self.active_segment - 1) * self.config.segment_bytes + self.active_offset); if expected_lsn != append.start_lsn && self.active_offset > 0 - && self.active_offset + encoded.len() as u64 > self.config.segment_bytes + && self.active_offset + encoded_len as u64 > self.config.segment_bytes { self.rotate_segment()?; } - let expected_lsn = Lsn((self.active_segment - 1) * self.config.segment_bytes + self.active_offset); if expected_lsn != append.start_lsn { @@ -214,17 +262,29 @@ impl WalManager { "record lsn does not match write position", )); } + Ok(()) + } - self.active_file.write_all_at(self.active_offset, encoded)?; + fn emit_wal_bytes( + &mut self, + offset: u64, + bytes: &[u8], + end_lsn: Lsn, + last_start_lsn: Lsn, + ) -> Result<()> { + // Lane E failpoint: armed before the bytes land, once per + // physical write rather than once per queued record. + crate::fail_point!("wal::write_encoded"); + self.active_file.write_all_at(offset, bytes)?; // Lane BH P1 #7: count the pwrite-equivalent before bumping // the offset; the bench harness reads this through // `WalCoordinator::sync_counters_snapshot`. if let Some(counters) = &self.sync_counters { counters.bump_pwrite(); } - self.active_offset += encoded.len() as u64; - self.written_lsn = append.end_lsn; - self.prev_lsn = append.start_lsn; + self.active_offset = offset + bytes.len() as u64; + self.written_lsn = end_lsn; + self.prev_lsn = last_start_lsn; Ok(()) } diff --git a/crates/kernel/tests/wal_tests.rs b/crates/kernel/tests/wal_tests.rs index 197825b89..913f53f09 100644 --- a/crates/kernel/tests/wal_tests.rs +++ b/crates/kernel/tests/wal_tests.rs @@ -376,6 +376,32 @@ fn wal_coordinator_group_flushes_concurrent_targets() { assert_eq!(reader.scan().unwrap().len(), 8); } +#[test] +fn queued_exact_fill_stays_inside_the_segment() { + let temp = TempDir::new().unwrap(); + let segment_bytes = 128; + let config = WalConfig { + segment_bytes, + group_commit_delay_us: 0, + ..WalConfig::default() + }; + let coordinator = WalCoordinator::create(temp.path(), config.clone()).unwrap(); + // 48-byte header + 16-byte payload = 64. Two records fill the segment. + let mut last = Lsn::ZERO; + for _ in 0..3 { + last = coordinator + .append(WalRecordKind::PageDelta, TxId(1), vec![b'q'; 16]) + .unwrap() + .end_lsn; + } + assert!(coordinator.flush_all().unwrap() >= last); + let first = temp.path().join(format!("{:020}.wal", 1_u64)); + let len = std::fs::metadata(&first).unwrap().len(); + assert_eq!(len, segment_bytes, "queued segment grew to {len}"); + let mut reader = WalReader::new(temp.path(), config); + assert_eq!(reader.scan().unwrap().len(), 3); +} + #[test] fn wal_coordinator_flush_until_returns_when_already_durable() { let temp = TempDir::new().unwrap();