From cb8769e5bc7418ab747be901d50ab5785c866863 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 19:03:21 +0700 Subject: [PATCH 1/5] Stop copying the page list every time a page is appended `DataPages::pages` was a `Vec` behind an `ArcSwap`, and both append paths cloned the whole thing to add one page. Every page in it is an `Arc`, so the clone was one atomic increment per existing page, each touching a separately allocated page header: a cache miss apiece. Appending page N cost O(N), and filling a table cost O(N^2). It hid behind row size, because row size is what decides how many pages a table has. At 256 bytes a page holds sixty-odd rows and twenty thousand rows is three hundred pages, where the copy is invisible and per-row cost is flat. At 4 KiB a page holds three, the same rows are six thousand pages, and per-row cost climbed from 2.95 to 30.08 microseconds across the load. The pages now live in fixed-size chunks, so an append copies one chunk and the spine of chunk pointers rather than every page. Readers still take an `ArcSwap` snapshot and never block, and page access still goes through the directory first, so the read path is unchanged. Measured over twenty thousand 4 KiB rows, per-row cost goes flat - 1.53 to 1.95 microseconds against 2.95 to 30.08 - and inserting into an unpersisted table goes from 254 MB/s to 3062. A persisted table gains far less, 208 to 279 MB/s, because with the copy gone persistence is what the load now waits on. Co-Authored-By: Claude Opus 5 --- src/in_memory/pages.rs | 151 +++++++++++++++++++++++++++++------------ 1 file changed, 106 insertions(+), 45 deletions(-) diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index f118dff5..e7c1ad4a 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -67,6 +67,88 @@ const RECLAIM_BATCH_LIMIT: usize = 256; /// gain. const RECLAIM_BACKLOG_TRIGGER: usize = RECLAIM_BATCH_LIMIT; +/// Pages per chunk of [`PageList`]. +/// +/// The list is copy-on-write, so an append copies whatever the writer has to +/// replace. Chunking bounds that at one chunk plus the (much shorter) spine, +/// instead of the whole list. +const PAGE_LIST_CHUNK: usize = 256; + +/// Owns every page, and appends one without copying the ones already there. +/// +/// **This was a `Vec` behind an `ArcSwap`, and appending cloned all of it.** +/// Every existing page is an `Arc`, so the clone was one atomic increment per +/// page, each touching a separately allocated page header: a cache miss apiece. +/// Appending page N cost O(N) and filling a table cost O(N^2). It only showed +/// up with large rows, because those are what make pages plentiful - at 4 KiB a +/// row, three rows to a page, per-row insert cost grew tenfold over twenty +/// thousand rows while a 256-byte row stayed flat. +/// +/// Readers still take an `ArcSwap` snapshot and never block. +#[derive(Debug)] +struct PageList { + chunks: ArcSwap>>>>, +} + +impl PageList { + fn from_pages(pages: Vec>) -> Self { + let chunks = pages + .chunks(PAGE_LIST_CHUNK) + .map(|chunk| Arc::new(chunk.to_vec())) + .collect::>(); + Self { + chunks: ArcSwap::from_pointee(chunks), + } + } + + /// Append a page. Copies the last chunk, or starts a new one, plus the + /// spine of chunk pointers. + fn push(&self, page: Arc) { + let chunks = self.chunks.load_full(); + let mut next = (*chunks).clone(); + match next.last() { + // Every chunk but the last is full, so only the last can take one. + Some(last) if last.len() < PAGE_LIST_CHUNK => { + let mut grown = (**last).clone(); + grown.push(page); + *next.last_mut().expect("the branch matched on it") = Arc::new(grown); + } + _ => { + let mut chunk = Vec::with_capacity(PAGE_LIST_CHUNK); + chunk.push(page); + next.push(Arc::new(chunk)); + } + } + self.chunks.store(Arc::new(next)); + } + + fn len(&self) -> usize { + let chunks = self.chunks.load(); + match chunks.last() { + None => 0, + // Full but for the last, so its length is the only remainder. + Some(last) => (chunks.len() - 1) * PAGE_LIST_CHUNK + last.len(), + } + } + + fn get(&self, index: usize) -> Option> { + let chunks = self.chunks.load(); + chunks + .get(index / PAGE_LIST_CHUNK)? + .get(index % PAGE_LIST_CHUNK) + .cloned() + } + + fn for_each(&self, mut visit: impl FnMut(&Arc)) { + let chunks = self.chunks.load(); + for chunk in chunks.iter() { + for page in chunk.iter() { + visit(page); + } + } + } +} + #[derive(Debug)] struct PageDirectoryChunk { pages: [AtomicPtr; PAGE_DIRECTORY_CHUNK_SIZE], @@ -261,7 +343,7 @@ where /// Immutable page-directory snapshots. Reads load one snapshot without a /// shared read-modify-write; rare growth copies and swaps the short vector. - pages: ArcSwap::WrappedRow, DATA_LENGTH>>>>, + pages: PageList::WrappedRow, DATA_LENGTH>>, /// Stable pointers for point access without ArcSwap's shared snapshot /// accounting. The corresponding `Arc`s remain owned by `pages`. page_directory: PageDirectory::WrappedRow, DATA_LENGTH>>, @@ -303,11 +385,11 @@ where return Ok(page); } - let page = { - let pages = self.pages.load(); - pages.get(index).map(Arc::as_ptr) - } - .ok_or(ExecutionError::PageNotFound(page_id))?; + let page = self + .pages + .get(index) + .map(|page| Arc::as_ptr(&page)) + .ok_or(ExecutionError::PageNotFound(page_id))?; // SAFETY: as above, the current directory retains this allocation and // all future directory snapshots clone its Arc. @@ -578,7 +660,7 @@ where queued_page_retirements: AtomicUsize::new(0), // We are starting ID's from `1` because `0`'s page in file is info page. page_directory: PageDirectory::new(&pages), - pages: ArcSwap::from_pointee(pages), + pages: PageList::from_pages(pages), pages_write: Mutex::new(()), empty_links: EmptyLinkRegistry::::default(), empty_pages: Default::default(), @@ -602,7 +684,7 @@ where pending_retirements: AtomicUsize::new(0), queued_page_retirements: AtomicUsize::new(0), page_directory, - pages: ArcSwap::from_pointee(vec), + pages: PageList::from_pages(vec), pages_write: Mutex::new(()), empty_links: EmptyLinkRegistry::default(), empty_pages: Default::default(), @@ -743,18 +825,8 @@ where let _write = self.pages_write.lock(); if tried_page == page_id_mapper(self.current_page_id.load(Ordering::Acquire) as usize) { let index = self.last_page_id.fetch_add(1, Ordering::AcqRel) + 1; - let pages = self.pages.load_full(); - let mut next = (*pages).clone(); let page = Arc::new(Data::new(index.into())); - next.push(page.clone()); - debug_assert_eq!(next.len(), pages.len() + 1); - debug_assert!( - next[..pages.len()] - .iter() - .zip(pages.iter()) - .all(|(new, old)| Arc::ptr_eq(new, old)) - ); - self.pages.store(Arc::new(next)); + self.pages.push(page.clone()); self.publish_page(&page); self.current_page_id.store(index, Ordering::Release); } @@ -771,9 +843,11 @@ where }; if let Some(page_id) = page_id { - let pages = self.pages.load(); let index = page_id_mapper(page_id.into()); - let page = pages[index].clone(); + let page = self + .pages + .get(index) + .expect("an empty page id names a page that was allocated"); { let _page_guard = page.access.write(); page.reset(); @@ -785,17 +859,7 @@ where let _write = self.pages_write.lock(); let index = self.last_page_id.fetch_add(1, Ordering::AcqRel) + 1; let page = Arc::new(Data::new(index.into())); - let pages = self.pages.load_full(); - let mut next = (*pages).clone(); - next.push(page.clone()); - debug_assert_eq!(next.len(), pages.len() + 1); - debug_assert!( - next[..pages.len()] - .iter() - .zip(pages.iter()) - .all(|(new, old)| Arc::ptr_eq(new, old)) - ); - self.pages.store(Arc::new(next)); + self.pages.push(page.clone()); self.publish_page(&page); page @@ -842,12 +906,12 @@ where + Deserialize<::WrappedRow, HighDeserializer> + for<'a> rkyv::bytecheck::CheckBytes>, { - let pages = self.pages.load(); let page_id: usize = link.page_id.into(); let page_index = page_id .checked_sub(1) .ok_or(ExecutionError::PageNotFound(link.page_id))?; - let page = pages + let page = self + .pages .get(page_index) .ok_or(ExecutionError::PageNotFound(link.page_id))?; let wrapped = page.get_row_checked(link).map_err(ExecutionError::DataPageError)?; @@ -1119,9 +1183,7 @@ where } pub fn get_page(&self, page_id: PageId) -> Option::WrappedRow, DATA_LENGTH>>> { - let pages = self.pages.load(); - let page = pages.get(page_id_mapper(page_id.into()))?; - Some(page.clone()) + self.pages.get(page_id_mapper(page_id.into())) } /// Registers an already-indexed cell while rebuilding runtime metadata @@ -1170,11 +1232,10 @@ where /// Approximate under concurrency: a failing `save_row`'s transient /// reservation may be counted before its rollback. Metrics only. pub fn used_bytes(&self) -> u64 { - let pages = self.pages.load(); - pages - .iter() - .map(|p| u64::from(p.free_offset.load(Ordering::Relaxed))) - .sum() + let mut total = 0u64; + self.pages + .for_each(|page| total += u64::from(page.free_offset.load(Ordering::Relaxed))); + total } /// Copies a row to another page without exposing either mutable byte @@ -1230,7 +1291,7 @@ where } pub fn get_page_count(&self) -> usize { - self.pages.load().len() + self.pages.len() } pub fn get_empty_links(&self) -> Vec { @@ -1253,12 +1314,12 @@ where /// figure without it cannot be checked, because a sweep that never runs /// looks exactly like a sweep that is free. pub fn allocated_pages(&self) -> usize { - self.pages.load().len() + self.pages.len() } /// Heap bytes reserved by the fixed-size data-page allocations. pub fn allocated_bytes(&self) -> usize { - self.pages.load().len() * std::mem::size_of::::WrappedRow, DATA_LENGTH>>() + self.pages.len() * std::mem::size_of::::WrappedRow, DATA_LENGTH>>() } /// Pages allocated but currently on the empty list, so reusable without From 2c857379d752cb1546fe183e8a8ebc3b4977cd79 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 19:03:42 +0700 Subject: [PATCH 2/5] Let the page directory reach past 64 MiB, and measure the insert path The directory holds stable page pointers so a point access does not take an `ArcSwap` snapshot, and it reached 64 * 64 = 4,096 pages, which is 64 MiB at the default page size. Past that `publish` returned early and every access fell back to the snapshot. A table of 4 KiB rows crosses that after twelve thousand rows, which is not a large table. Raising it to 1,024 roots reaches 1 GiB for an 8 KiB array of pointers. On its own it made no measurable difference to insert cost, because the page list copy was the term that mattered; this moves a ceiling rather than removing a cost, and it is committed separately so the two are not confused. The tests are the harness the page list work was measured with: whether insert cost scales with row size, whether it grows as the table fills, what the floor is with few pages, and what the persistence path sustains end to end. They are all `#[ignore]`d, since they are measurements and not assertions. Co-Authored-By: Claude Opus 5 --- src/in_memory/pages.rs | 13 +- tests/persistence/insert_cost_shape.rs | 216 +++++++++++++++++++++ tests/persistence/local_write_bandwidth.rs | 201 +++++++++++++++++++ tests/persistence/mod.rs | 2 + 4 files changed, 431 insertions(+), 1 deletion(-) create mode 100644 tests/persistence/insert_cost_shape.rs create mode 100644 tests/persistence/local_write_bandwidth.rs diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index e7c1ad4a..a28d210d 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -37,7 +37,18 @@ fn page_id_mapper(page_id: usize) -> usize { } const PAGE_DIRECTORY_CHUNK_SIZE: usize = 64; -const PAGE_DIRECTORY_ROOTS: usize = 64; +/// Roots in the page directory, so its reach is `ROOTS * CHUNK_SIZE` pages. +/// +/// **This was 64, which reached 4,096 pages: 64 MiB at the default page size.** +/// Past that, `publish` returns early and every page access falls back to an +/// `ArcSwap` snapshot of the owning list. A table of 4 KiB rows, which fit +/// three to a page, crosses it after twelve thousand rows. +/// +/// Raising it did not measurably change insert cost on that fixture - the +/// copy-on-write page list dominated, and still does at these sizes - so this +/// is a ceiling being moved rather than a cost being removed. 1,024 roots +/// reach 65,536 pages, or 1 GiB, for an 8 KiB array of pointers. +const PAGE_DIRECTORY_ROOTS: usize = 1024; const GHOSTED: u8 = 1 << 0; const DELETED: u8 = 1 << 1; const VACUUMED: u8 = 1 << 2; diff --git a/tests/persistence/insert_cost_shape.rs b/tests/persistence/insert_cost_shape.rs new file mode 100644 index 00000000..72cc3ba9 --- /dev/null +++ b/tests/persistence/insert_cost_shape.rs @@ -0,0 +1,216 @@ +//! Where the time in an insert goes, since it is not the disk. +//! +//! WorkTable's bulk load runs at about 200 MB/s while the write path under it +//! does gigabytes, and removing persistence entirely changes nothing. So the +//! cost is in the insert. This asks the first question that splits the +//! candidates: does it scale with the size of the row, or is it a fixed price +//! per row? + +use worktable::prelude::*; +use worktable_codegen::worktable; + +worktable!( + name: InsertShape, + columns: { + id: u64 primary_key, + payload: String, + } +); + +#[test] +#[ignore = "a measurement, not an assertion"] +fn does_insert_cost_scale_with_row_size() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async { + const ROWS: u64 = 25_000; + println!(" payload rows/s us/row MB/s"); + for size in [8usize, 64, 256, 1024, 2048, 4096, 8192] { + let payload = "x".repeat(size); + // Warm: allocator and the table's first growth are not the subject. + { + let warm = InsertShapeWorkTable::default(); + for id in 0..1_000u64 { + warm.insert(InsertShapeRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + } + let table = InsertShapeWorkTable::default(); + let at = std::time::Instant::now(); + for id in 0..ROWS { + table + .insert(InsertShapeRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + let elapsed = at.elapsed().as_secs_f64(); + println!( + " {size:>7} {:>9.0} {:>8.2} {:>7.0}", + ROWS as f64 / elapsed, + elapsed * 1e6 / ROWS as f64, + (ROWS as usize * size) as f64 / 1e6 / elapsed, + ); + } + }); +} + +/// The same rows, with the table taken out of it: what the loop costs when all +/// it does is build the row and drop it. Anything the insert arm spends beyond +/// this is the table. +#[test] +#[ignore = "a measurement, not an assertion"] +fn what_the_loop_costs_without_the_table() { + const ROWS: u64 = 25_000; + println!(" payload rows/s us/row"); + for size in [8usize, 4096] { + let payload = "x".repeat(size); + let at = std::time::Instant::now(); + let mut sink = 0usize; + for id in 0..ROWS { + let row = InsertShapeRow { + id, + payload: payload.clone(), + }; + sink = sink.wrapping_add(row.payload.len()); + std::hint::black_box(&row); + } + let elapsed = at.elapsed().as_secs_f64(); + std::hint::black_box(sink); + println!( + " {size:>7} {:>9.0} {:>8.3}", + ROWS as f64 / elapsed, + elapsed * 1e6 / ROWS as f64, + ); + } +} + +/// A long run of the expensive case, so a sampling profiler has something to +/// look at. Not a measurement in itself. +#[test] +#[ignore = "for profiling only"] +fn keep_inserting_four_kilobyte_rows() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let payload = "x".repeat(4096); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20); + while std::time::Instant::now() < deadline { + let table = InsertShapeWorkTable::default(); + for id in 0..20_000u64 { + table + .insert(InsertShapeRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + } + }); +} + +/// Does an insert get slower as the table fills? +/// +/// A cost that scales with rows already present is O(n) per insert and O(n^2) +/// overall, which is what a super-linear response to row size would look like +/// if bigger rows simply reach any given page count sooner. +#[test] +#[ignore = "a measurement, not an assertion"] +fn does_insert_slow_down_as_the_table_fills() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async { + for size in [256usize, 4096] { + let payload = "x".repeat(size); + let table = InsertShapeWorkTable::default(); + const BLOCK: u64 = 2_000; + const BLOCKS: u64 = 10; + println!(" payload {size}: us/row by block of {BLOCK}"); + let mut line = String::new(); + for block in 0..BLOCKS { + let at = std::time::Instant::now(); + for n in 0..BLOCK { + let id = block * BLOCK + n; + table + .insert(InsertShapeRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + let per_row = at.elapsed().as_secs_f64() * 1e6 / BLOCK as f64; + line.push_str(&format!("{per_row:>8.2}")); + } + println!(" {line}"); + } + }); +} + +/// The per-row cost with the page-list clone nearly absent, which is the floor +/// a fix for it would approach. +/// +/// The clone is O(pages), so a table that has barely any pages barely pays it. +/// Timing small tables gives the cost of everything else: the row clone, the +/// rkyv serialize, and the copy into the page. +#[test] +#[ignore = "a measurement, not an assertion"] +fn the_floor_with_almost_no_pages() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async { + println!(" payload rows pages us/row MB/s"); + for size in [256usize, 1024, 4096] { + let payload = "x".repeat(size); + let per_page = (16356 / size).max(1); + for rows in [30u64, 120, 480] { + // Median of several fresh tables: a single small run is noise. + let mut samples = Vec::new(); + for _ in 0..25 { + let table = InsertShapeWorkTable::default(); + let at = std::time::Instant::now(); + for id in 0..rows { + table + .insert(InsertShapeRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + samples.push(at.elapsed().as_secs_f64() / rows as f64); + } + samples.sort_by(|a, b| a.partial_cmp(b).unwrap()); + let per_row = samples[samples.len() / 2]; + println!( + " {size:>7} {rows:>5} {:>5} {:>7.3} {:>7.0}", + rows as usize / per_page, + per_row * 1e6, + size as f64 / 1e6 / per_row, + ); + } + } + }); +} diff --git a/tests/persistence/local_write_bandwidth.rs b/tests/persistence/local_write_bandwidth.rs new file mode 100644 index 00000000..ab36e386 --- /dev/null +++ b/tests/persistence/local_write_bandwidth.rs @@ -0,0 +1,201 @@ +//! What WorkTable's local persistence path sustains, in bytes per second. +//! +//! Measured through `insert` and `wait_for_ops` rather than against +//! `persist_page` directly, so it counts everything the engine does to make a +//! write durable and not just the call at the bottom of it. + +use worktable::prelude::PersistedWorkTable; +use worktable::prelude::*; +use worktable_codegen::worktable; + +use crate::remove_dir_if_exists; + +worktable!( + name: WriteBandwidth, + persist: true, + columns: { + id: u64 primary_key, + payload: String, + } +); + +// The same table without persistence, so the cost of being a WorkTable can be +// told apart from the cost of writing to disk. Anything this arm spends is +// spent by the persisted one too, before any page is written. +worktable!( + name: WriteBandwidthMemory, + columns: { + id: u64 primary_key, + payload: String, + } +); + +/// A page of the on-disk format, which is the unit a write actually lands in. +const PAGE: usize = 4096 * 4; + +/// Per-page checksums of every file, so two snapshots say how many pages a +/// stretch of work really wrote. +fn page_checksums(dir: &str) -> Vec<(String, Vec)> { + fn walk(dir: &std::path::Path, into: &mut Vec) { + let Ok(entries) = std::fs::read_dir(dir) else { return }; + for entry in entries.filter_map(|e| e.ok()) { + let path = entry.path(); + if path.is_dir() { + walk(&path, into); + } else if path.is_file() { + into.push(path); + } + } + } + let mut paths = Vec::new(); + walk(std::path::Path::new(dir), &mut paths); + paths.sort(); + paths + .into_iter() + .map(|path| { + let bytes = std::fs::read(&path).expect("a table file"); + ( + path.to_string_lossy().into_owned(), + bytes.chunks(PAGE).map(crc32fast::hash).collect(), + ) + }) + .collect() +} + +/// Bytes that differ between two snapshots, counted a page at a time. +fn written_bytes(before: &[(String, Vec)], after: &[(String, Vec)]) -> u64 { + let mut pages = 0u64; + for (name, now) in after { + let then = before + .iter() + .find(|(n, _)| n == name) + .map(|(_, c)| c.as_slice()) + .unwrap_or(&[]); + pages += now + .iter() + .enumerate() + .filter(|(index, checksum)| then.get(*index) != Some(*checksum)) + .count() as u64; + } + pages * PAGE as u64 +} + +fn table_bytes(dir: &str) -> u64 { + fn walk(dir: &std::path::Path, total: &mut u64) { + let Ok(entries) = std::fs::read_dir(dir) else { return }; + for entry in entries.filter_map(|e| e.ok()) { + let path = entry.path(); + if path.is_dir() { + walk(&path, total); + } else if let Ok(meta) = entry.metadata() { + *total += meta.len(); + } + } + } + let mut total = 0; + walk(std::path::Path::new(dir), &mut total); + total +} + +#[test] +#[ignore = "a measurement, not an assertion"] +fn local_write_bandwidth() { + let dir = "tests/data/local_write_bandwidth"; + let config = DiskConfig::new_with_table_name( + dir, + WriteBandwidthWorkTable::name_snake_case(), + WriteBandwidthWorkTable::version(), + ); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_io() + .enable_time() + .build() + .unwrap(); + + runtime.block_on(async { + remove_dir_if_exists(dir.to_string()).await; + let engine = WriteBandwidthPersistenceEngine::new(config.clone()).await.unwrap(); + let table = WriteBandwidthWorkTable::load(engine).await.unwrap(); + let payload = "x".repeat(4096); + + // ---- bulk load: consecutive pages, which is the batch path's shape + const ROWS: u64 = 25_000; + let at = std::time::Instant::now(); + for id in 0..ROWS { + table + .insert(WriteBandwidthRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + table.wait_for_ops().await.expect("the queue drains"); + let bulk = at.elapsed().as_secs_f64(); + let bytes = table_bytes(dir); + + // ---- scattered updates into what already exists + const UPDATES: u64 = 2_000; + let replacement = "y".repeat(4096); + let pages_before = page_checksums(dir); + let at = std::time::Instant::now(); + for n in 0..UPDATES { + // Spread across the whole table rather than a contiguous run. + let id = (n * (ROWS / UPDATES)) % ROWS; + table + .update(WriteBandwidthRow { + id, + payload: replacement.clone(), + }) + .await + .unwrap(); + } + table.wait_for_ops().await.expect("the queue drains"); + let scattered = at.elapsed().as_secs_f64(); + let scattered_bytes = written_bytes(&pages_before, &page_checksums(dir)); + + println!("table {:.1} MB on disk", bytes as f64 / 1e6); + println!( + " bulk load, {ROWS} rows : {:>8.1} ms {:>7.0} MB/s {:>8.0} rows/s", + bulk * 1e3, + bytes as f64 / 1e6 / bulk, + ROWS as f64 / bulk, + ); + println!( + " scattered update, {UPDATES} rows : {:>8.1} ms {:>7.0} MB/s {:>8.0} rows/s {:.1} MB written", + scattered * 1e3, + scattered_bytes as f64 / 1e6 / scattered, + UPDATES as f64 / scattered, + scattered_bytes as f64 / 1e6, + ); + + // ---- the same inserts with nothing underneath them + let memory = WriteBandwidthMemoryWorkTable::default(); + let at = std::time::Instant::now(); + for id in 0..ROWS { + memory + .insert(WriteBandwidthMemoryRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + let in_memory = at.elapsed().as_secs_f64(); + println!( + " in memory, no persistence : {:>8.1} ms {:>7.0} MB/s {:>8.0} rows/s", + in_memory * 1e3, + bytes as f64 / 1e6 / in_memory, + ROWS as f64 / in_memory, + ); + println!( + " ^ persistence adds {:.1} ms on top of {:.1} ms of table work", + (bulk - in_memory) * 1e3, + in_memory * 1e3, + ); + + remove_dir_if_exists(dir.to_string()).await; + }); +} diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index e282cc9f..e07610a5 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -10,9 +10,11 @@ mod exact_boundary_load; mod failure; mod in_place_durability; mod index_page; +mod insert_cost_shape; mod insert_many; mod insert_many_bench; mod loaded_index_growth; +mod local_write_bandwidth; mod multi_row_backend_order; mod read; mod recovery_load; From 8b84a9c9189d6a4ff55ca983f8abd5f2a392ed17 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 19:15:56 +0700 Subject: [PATCH 3/5] Measure insert latency as two populations, not one distribution An insert that fits in the page already open and one that has to add a page are different operations with different costs, and averaging them hides the second behind the first. How much it hides depends on row size: at 256 bytes one insert in sixty allocates, so allocation is a tail event, while at 4 KiB one in three does and it is a third of all traffic. Split by whether the page count moved, the page list change reads as what it is - a tail-latency fix that leaves the common path alone. On 4 KiB rows the allocating population goes from 59.67 to 7.08 microseconds at p50, 133.38 to 14.46 at p99, and 451.54 to 42.33 at its worst, while the existing-page population does not move. On 256-byte rows the same change is worth about a third, on the one insert in sixty that pays it. Co-Authored-By: Claude Opus 5 --- tests/persistence/insert_latency.rs | 124 ++++++++++++++++++++++++++++ tests/persistence/mod.rs | 1 + 2 files changed, 125 insertions(+) create mode 100644 tests/persistence/insert_latency.rs diff --git a/tests/persistence/insert_latency.rs b/tests/persistence/insert_latency.rs new file mode 100644 index 00000000..00afe6a0 --- /dev/null +++ b/tests/persistence/insert_latency.rs @@ -0,0 +1,124 @@ +//! Per-insert latency, split by whether the insert had to allocate a page. +//! +//! These are two populations, not one distribution. An insert that fits in the +//! page already open is cheap; one that has to add a page pays for the page +//! list as well. Blending them hides the second behind the first, and how much +//! it hides depends on row size: at 4 KiB a page holds three rows, so a third +//! of all inserts allocate and the expensive population is not a tail at all. + +use worktable::prelude::PersistedWorkTable; +use worktable::prelude::*; +use worktable_codegen::worktable; + +use crate::remove_dir_if_exists; + +worktable!( + name: InsertLatency, + persist: true, + columns: { + id: u64 primary_key, + payload: String, + } +); + +worktable!( + name: InsertLatencyMemory, + columns: { + id: u64 primary_key, + payload: String, + } +); + +fn report(label: &str, mut us: Vec, of: usize) { + if us.is_empty() { + println!(" {label:<34} (none)"); + return; + } + us.sort_by(|a, b| a.partial_cmp(b).unwrap()); + let at = |q: f64| us[((us.len() as f64 * q) as usize).min(us.len() - 1)]; + println!( + " {label:<34} {:>5.1}% of inserts p50 {:>7.2} p99 {:>8.2} max {:>9.2} (us)", + 100.0 * us.len() as f64 / of as f64, + at(0.50), + at(0.99), + us[us.len() - 1], + ); +} + +#[test] +#[ignore = "a measurement, not an assertion"] +fn insert_latency_split_by_page_allocation() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async { + const ROWS: u64 = 20_000; + for size in [256usize, 4096] { + let payload = "x".repeat(size); + println!("payload {size} B, {ROWS} rows"); + + // ---- no persistence + let table = InsertLatencyMemoryWorkTable::default(); + let (mut same, mut fresh) = (Vec::new(), Vec::new()); + let mut pages = table.0.data.get_page_count(); + for id in 0..ROWS { + let at = std::time::Instant::now(); + table + .insert(InsertLatencyMemoryRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + let took = at.elapsed().as_secs_f64() * 1e6; + let now = table.0.data.get_page_count(); + if now == pages { + same.push(took) + } else { + fresh.push(took) + } + pages = now; + } + report("in memory, existing page", same, ROWS as usize); + report("in memory, allocated a page", fresh, ROWS as usize); + + // ---- persisted + let dir = "tests/data/insert_latency"; + remove_dir_if_exists(dir.to_string()).await; + let config = DiskConfig::new_with_table_name( + dir, + InsertLatencyWorkTable::name_snake_case(), + InsertLatencyWorkTable::version(), + ); + let engine = InsertLatencyPersistenceEngine::new(config).await.unwrap(); + let persisted = InsertLatencyWorkTable::load(engine).await.unwrap(); + let (mut same, mut fresh) = (Vec::new(), Vec::new()); + let mut pages = persisted.0.data.get_page_count(); + for id in 0..ROWS { + let at = std::time::Instant::now(); + persisted + .insert(InsertLatencyRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + let took = at.elapsed().as_secs_f64() * 1e6; + let now = persisted.0.data.get_page_count(); + if now == pages { + same.push(took) + } else { + fresh.push(took) + } + pages = now; + } + report("persisted, existing page", same, ROWS as usize); + report("persisted, allocated a page", fresh, ROWS as usize); + persisted.wait_for_ops().await.expect("the queue drains"); + remove_dir_if_exists(dir.to_string()).await; + } + }); +} diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index e07610a5..06864a9d 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -11,6 +11,7 @@ mod failure; mod in_place_durability; mod index_page; mod insert_cost_shape; +mod insert_latency; mod insert_many; mod insert_many_bench; mod loaded_index_growth; From df2566cc112951ba023ea4a543f842fc9af4d7dd Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 8 Sep 2026 20:52:57 +0700 Subject: [PATCH 4/5] Borrow the page on the link read instead of cloning its Arc The chunked page list handed back an owned `Arc` from `get`, which costs an atomic increment and the matching decrement on drop. The link-based read is a read path and needs the page only for the length of one call, so it paid both for nothing. It measured. A delete went from 665 to 751 ns, while a select by primary key - which goes through the page directory and never touches this - did not move. Borrowing through `with_page` puts delete back at 654 ns, and the criterion cases either side of it are unchanged or better: insert 442 -> 86 ns, select by primary key 21.0 -> 20.1 ns. Found by running the benchmark suite that already existed rather than the ad-hoc probes the rest of this work was measured with. Co-Authored-By: Claude Opus 5 --- src/in_memory/pages.rs | 41 +++++++++++----- tests/persistence/mod.rs | 1 + tests/persistence/persistence_is_what.rs | 60 ++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 12 deletions(-) create mode 100644 tests/persistence/persistence_is_what.rs diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index a28d210d..08cfeaa1 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -150,6 +150,21 @@ impl PageList { .cloned() } + /// Run `visit` against the page at `index`, borrowing it rather than + /// handing back an owned `Arc`. + /// + /// **`get` costs an atomic increment and the matching decrement on drop.** + /// A read that only needs the page for the length of one call pays both for + /// nothing, and it is measurable: routing the link-based read through `get` + /// moved a delete from 665 to 751 ns, while a select by primary key - which + /// goes through the page directory and never touches this - did not move at + /// all. + fn with_page(&self, index: usize, visit: impl FnOnce(&T) -> R) -> Option { + let chunks = self.chunks.load(); + let page = chunks.get(index / PAGE_LIST_CHUNK)?.get(index % PAGE_LIST_CHUNK)?; + Some(visit(page)) + } + fn for_each(&self, mut visit: impl FnMut(&Arc)) { let chunks = self.chunks.load(); for chunk in chunks.iter() { @@ -921,18 +936,20 @@ where let page_index = page_id .checked_sub(1) .ok_or(ExecutionError::PageNotFound(link.page_id))?; - let page = self - .pages - .get(page_index) - .ok_or(ExecutionError::PageNotFound(link.page_id))?; - let wrapped = page.get_row_checked(link).map_err(ExecutionError::DataPageError)?; - if wrapped.is_ghosted() { - return Err(ExecutionError::Ghosted); - } - if wrapped.is_deleted() { - return Err(ExecutionError::Deleted); - } - Ok(wrapped.get_inner()) + // Borrowed rather than cloned: this is a read path and an `Arc` bump + // here showed up as a 13% slower delete. + self.pages + .with_page(page_index, |page| { + let wrapped = page.get_row_checked(link).map_err(ExecutionError::DataPageError)?; + if wrapped.is_ghosted() { + return Err(ExecutionError::Ghosted); + } + if wrapped.is_deleted() { + return Err(ExecutionError::Deleted); + } + Ok(wrapped.get_inner()) + }) + .ok_or(ExecutionError::PageNotFound(link.page_id))? } pub fn select_non_vacuumed(&self, link: Link) -> Result diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index 06864a9d..9b2535a9 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -17,6 +17,7 @@ mod insert_many_bench; mod loaded_index_growth; mod local_write_bandwidth; mod multi_row_backend_order; +mod persistence_is_what; mod read; mod recovery_load; mod same_size_in_place; diff --git a/tests/persistence/persistence_is_what.rs b/tests/persistence/persistence_is_what.rs new file mode 100644 index 00000000..ca17484a --- /dev/null +++ b/tests/persistence/persistence_is_what.rs @@ -0,0 +1,60 @@ +//! Is the persistence path waiting, or working? +//! +//! Bulk load persists at about 310 MB/s while the disk under it does gigabytes +//! and DataBucket's own write path does 500+ MB/s single threaded. So something +//! between them is the limit. CPU time against wall time says which kind of +//! limit it is: near or above wall means it is computing, well under means it +//! is waiting. + +use worktable::prelude::PersistedWorkTable; +use worktable::prelude::*; +use worktable_codegen::worktable; + +use crate::remove_dir_if_exists; + +worktable!( + name: PersistShape, + persist: true, + columns: { id: u64 primary_key, payload: String } +); + +#[test] +#[ignore = "a measurement, not an assertion"] +fn is_persistence_waiting_or_working() { + let dir = "tests/data/persistence_is_what"; + let config = DiskConfig::new_with_table_name( + dir, + PersistShapeWorkTable::name_snake_case(), + PersistShapeWorkTable::version(), + ); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + + runtime.block_on(async { + remove_dir_if_exists(dir.to_string()).await; + let engine = PersistShapePersistenceEngine::new(config).await.unwrap(); + let table = PersistShapeWorkTable::load(engine).await.unwrap(); + let payload = "x".repeat(4096); + + // Marked so the times either side can be attributed to this and not to + // building the table or tearing it down. + println!("MARK begin"); + let at = std::time::Instant::now(); + for id in 0..25_000u64 { + table + .insert(PersistShapeRow { + id, + payload: payload.clone(), + }) + .await + .unwrap(); + } + table.wait_for_ops().await.expect("the queue drains"); + println!("MARK end {:.1} ms", at.elapsed().as_secs_f64() * 1e3); + + remove_dir_if_exists(dir.to_string()).await; + }); +} From dc0ae34fceb9ae119f432d96be69ed93fd5972f8 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 9 Sep 2026 01:33:06 +0700 Subject: [PATCH 5/5] Call the insert benchmark what it measures It was named "bulk load", which reads as loading a table from disk. It inserts 25,000 rows. Co-Authored-By: Claude Opus 5 --- tests/persistence/local_write_bandwidth.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/persistence/local_write_bandwidth.rs b/tests/persistence/local_write_bandwidth.rs index ab36e386..5d74f201 100644 --- a/tests/persistence/local_write_bandwidth.rs +++ b/tests/persistence/local_write_bandwidth.rs @@ -158,7 +158,7 @@ fn local_write_bandwidth() { println!("table {:.1} MB on disk", bytes as f64 / 1e6); println!( - " bulk load, {ROWS} rows : {:>8.1} ms {:>7.0} MB/s {:>8.0} rows/s", + " bulk insert, {ROWS} rows : {:>8.1} ms {:>7.0} MB/s {:>8.0} rows/s", bulk * 1e3, bytes as f64 / 1e6 / bulk, ROWS as f64 / bulk,