Stop asking the file where it is, and write a run of pages in one call

Cargo.toml

@@ -3,7 +3,7 @@ members = ["codegen", "tools/create-data-file", "tools/dump-data-file"]
 
 [package]
 name = "data_bucket"
-version = "0.5.7"
+version = "0.6.0"
 edition = "2021"
 authors = ["Handy-caT"]
 license = "MIT"
@@ -13,7 +13,6 @@ description = "DataBucket is container for WorkTable's data"
 [dependencies]
 data_bucket_derive = { path = "codegen", version = "^0.3" }
 
-eyre = "0.6.12"
 derive_more = { version = "1.0.0", features = ["from", "error", "display", "into"] }
 rkyv = { version = "0.8.17", features = ["uuid-1"] }
 uuid = { version = "1.11.0", features = ["v4"] }

examples/write-pages-sweep.rs

@@ -0,0 +1,84 @@
+//! Where batching starts to matter, as a function of how many pages a caller
+//! actually hands over at once.
+//!
+//! The headline number for `persist_pages_batch` is measured at 6,400 pages in
+//! one call. The question this answers is whether anything reaches that, and
+//! what the two paths cost at the sizes a caller really passes.
+
+use data_bucket::page::{persist_page, persist_pages_batch};
+use data_bucket::{DataPage, GeneralHeader, GeneralPage, PageType, DATA_VERSION, INNER_PAGE_SIZE};
+use std::time::Instant;
+
+const SIZES: [usize; 8] = [1, 2, 4, 16, 64, 256, 1024, 6400];
+const REPS: usize = 9;
+
+fn pages(count: usize) -> Vec<GeneralPage<DataPage<INNER_PAGE_SIZE>>> {
+    (0..count as u32)
+        .map(|id| {
+            let mut data = [0u8; INNER_PAGE_SIZE];
+            for (n, byte) in data.iter_mut().enumerate() {
+                *byte = (n % 251) as u8;
+            }
+            GeneralPage {
+                header: GeneralHeader {
+                    data_version: DATA_VERSION,
+                    space_id: 1.into(),
+                    page_id: id.into(),
+                    previous_id: 0.into(),
+                    next_id: 0.into(),
+                    page_type: PageType::Data,
+                    data_length: 0,
+                },
+                inner: DataPage { length: INNER_PAGE_SIZE as u32, data },
+            }
+        })
+        .collect()
+}
+
+async fn fresh(path: &std::path::Path) -> tokio::fs::File {
+    tokio::fs::OpenOptions::new()
+        .read(true).write(true).create(true).truncate(true)
+        .open(path).await.unwrap()
+}
+
+fn median(mut v: Vec<f64>) -> f64 {
+    v.sort_by(|a, b| a.partial_cmp(b).unwrap());
+    v[v.len() / 2]
+}
+
+#[tokio::main]
+async fn main() {
+    let path = std::env::var("SCRATCH")
+        .map(std::path::PathBuf::from)
+        .unwrap_or_else(|_| std::env::temp_dir())
+        .join("data_bucket_sweep.wt");
+
+    println!("  pages        MB   one-at-a-time      batched      gain");
+    for count in SIZES {
+        let bytes = count * data_bucket::PAGE_SIZE;
+        let (mut ones, mut many) = (Vec::new(), Vec::new());
+        // One untimed pass of each, so neither pays for the file appearing.
+        { let mut f = fresh(&path).await; persist_pages_batch(pages(count), &mut f).await.unwrap(); f.sync_all().await.unwrap(); }
+        for _ in 0..REPS {
+            let mut all = pages(count);
+            let mut file = fresh(&path).await;
+            let at = Instant::now();
+            for page in &mut all { persist_page(page, &mut file).await.unwrap(); }
+            file.sync_all().await.unwrap();
+            ones.push(at.elapsed().as_secs_f64());
+
+            let all = pages(count);
+            let mut file = fresh(&path).await;
+            let at = Instant::now();
+            persist_pages_batch(all, &mut file).await.unwrap();
+            file.sync_all().await.unwrap();
+            many.push(at.elapsed().as_secs_f64());
+        }
+        let (one, batch) = (median(ones), median(many));
+        println!(
+            "  {count:>5}   {:>7.2}      {:>7.2} ms    {:>7.2} ms    {:>5.2}x",
+            bytes as f64 / 1e6, one * 1e3, batch * 1e3, one / batch
+        );
+    }
+    let _ = std::fs::remove_file(&path);
+}

examples/write-pages.rs

@@ -0,0 +1,197 @@
+//! What persisting a file's worth of pages costs, on the real path.
+//!
+//! Not a model of it: this calls `persist_page` and `persist_pages_batch`
+//! themselves, on the async file handles they take.
+
+use data_bucket::page::{persist_page, persist_pages_batch};
+use data_bucket::{DataPage, GeneralHeader, GeneralPage, PageType, DATA_VERSION, INNER_PAGE_SIZE};
+use std::time::Instant;
+
+const PAGES: u32 = 6_400;
+const REPS: usize = 5;
+
+fn pages() -> Vec<GeneralPage<DataPage<INNER_PAGE_SIZE>>> {
+    (0..PAGES)
+        .map(|id| {
+            let mut data = [0u8; INNER_PAGE_SIZE];
+            for (n, byte) in data.iter_mut().enumerate() {
+                *byte = (n % 251) as u8;
+            }
+            GeneralPage {
+                header: GeneralHeader {
+                    data_version: DATA_VERSION,
+                    space_id: 1.into(),
+                    page_id: id.into(),
+                    previous_id: 0.into(),
+                    next_id: 0.into(),
+                    page_type: PageType::Data,
+                    data_length: 0,
+                },
+                inner: DataPage {
+                    length: INNER_PAGE_SIZE as u32,
+                    data,
+                },
+            }
+        })
+        .collect()
+}
+
+/// Pages at every `stride`-th id, which is the shape of an update to a file
+/// that already exists: the ids are not consecutive, so nothing coalesces.
+fn scattered(stride: u32) -> Vec<GeneralPage<DataPage<INNER_PAGE_SIZE>>> {
+    pages()
+        .into_iter()
+        .enumerate()
+        .filter(|(id, _)| *id as u32 % stride == 0)
+        .map(|(_, page)| page)
+        .collect()
+}
+
+async fn existing(path: &std::path::Path) -> tokio::fs::File {
+    tokio::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .open(path)
+        .await
+        .unwrap()
+}
+
+async fn fresh(path: &std::path::Path) -> tokio::fs::File {
+    tokio::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(true)
+        .open(path)
+        .await
+        .unwrap()
+}
+
+#[tokio::main]
+async fn main() {
+    let path = std::env::var("SCRATCH")
+        .map(std::path::PathBuf::from)
+        .unwrap_or_else(|_| std::env::temp_dir())
+        .join("data_bucket_write_pages.wt");
+    let bytes = PAGES as usize * data_bucket::PAGE_SIZE;
+
+    println!(
+        "{PAGES} pages, {:.1} MB, median of {REPS}\n",
+        bytes as f64 / 1e6
+    );
+
+    // **The order of the arms is a variable, so it is one that can be set.**
+    // Run in a fixed order, whichever arm goes second inherits a file the first
+    // arm just wrote and looks faster for it. `REVERSE=1` runs them the other
+    // way round; the two orders agreeing is what makes either number mean
+    // anything.
+    let reverse = std::env::var("REVERSE").is_ok();
+    let mut one_at_a_time = Vec::new();
+    let mut batched = Vec::new();
+
+    let mut run_one = async |timings: &mut Vec<f64>| {
+        let mut all = pages();
+        let mut file = fresh(&path).await;
+        let at = Instant::now();
+        for page in &mut all {
+            persist_page(page, &mut file).await.unwrap();
+        }
+        file.sync_all().await.unwrap();
+        timings.push(at.elapsed().as_secs_f64());
+    };
+    let mut run_batch = async |timings: &mut Vec<f64>| {
+        let all = pages();
+        let mut file = fresh(&path).await;
+        let at = Instant::now();
+        persist_pages_batch(all, &mut file).await.unwrap();
+        file.sync_all().await.unwrap();
+        timings.push(at.elapsed().as_secs_f64());
+    };
+
+    for _ in 0..REPS {
+        if reverse {
+            run_batch(&mut batched).await;
+            run_one(&mut one_at_a_time).await;
+        } else {
+            run_one(&mut one_at_a_time).await;
+            run_batch(&mut batched).await;
+        }
+    }
+
+    let median = |mut v: Vec<f64>| {
+        v.sort_by(|a, b| a.partial_cmp(b).unwrap());
+        v[v.len() / 2]
+    };
+    let (one, many) = (median(one_at_a_time), median(batched));
+    println!(
+        "  persist_page, one at a time     {:>8.1} ms   {:>6.0} MB/s",
+        one * 1e3,
+        bytes as f64 / 1e6 / one
+    );
+    println!(
+        "  persist_pages_batch             {:>8.1} ms   {:>6.0} MB/s   {:>5.2}x",
+        many * 1e3,
+        bytes as f64 / 1e6 / many,
+        one / many
+    );
+
+    // ---- the case that is not a whole file
+    //
+    // Everything above rewrites the file from empty, so the ids run 0..PAGES
+    // with no gaps and the batch path sees one enormous consecutive run. That
+    // is its best case and a database's rarest one. Updating scattered pages
+    // in a file that already exists breaks the run at every page, so the batch
+    // path falls back to one write per page and can only win by what it saves
+    // per page, not by joining anything up.
+    const STRIDE: u32 = 10;
+    let touched = scattered(STRIDE).len();
+    let touched_bytes = touched * data_bucket::PAGE_SIZE;
+
+    // Lay the whole file down once, outside the clock, so the updates land in
+    // a file that is already the right length.
+    {
+        let mut file = fresh(&path).await;
+        persist_pages_batch(pages(), &mut file).await.unwrap();
+        file.sync_all().await.unwrap();
+    }
+
+    let mut one_scattered = Vec::new();
+    let mut batch_scattered = Vec::new();
+    for _ in 0..REPS {
+        let mut some = scattered(STRIDE);
+        let mut file = existing(&path).await;
+        let at = Instant::now();
+        for page in &mut some {
+            persist_page(page, &mut file).await.unwrap();
+        }
+        file.sync_all().await.unwrap();
+        one_scattered.push(at.elapsed().as_secs_f64());
+
+        let some = scattered(STRIDE);
+        let mut file = existing(&path).await;
+        let at = Instant::now();
+        persist_pages_batch(some, &mut file).await.unwrap();
+        file.sync_all().await.unwrap();
+        batch_scattered.push(at.elapsed().as_secs_f64());
+    }
+
+    let (one_s, many_s) = (median(one_scattered), median(batch_scattered));
+    println!(
+        "\nevery {STRIDE}th page of an existing file, {touched} pages, {:.1} MB",
+        touched_bytes as f64 / 1e6
+    );
+    println!(
+        "  persist_page, one at a time     {:>8.1} ms   {:>6.0} MB/s",
+        one_s * 1e3,
+        touched_bytes as f64 / 1e6 / one_s
+    );
+    println!(
+        "  persist_pages_batch             {:>8.1} ms   {:>6.0} MB/s   {:>5.2}x",
+        many_s * 1e3,
+        touched_bytes as f64 / 1e6 / many_s,
+        one_s / many_s
+    );
+
+    let _ = std::fs::remove_file(&path);
+}