feat(logseg): encode a block's column pages on a caller's executor

changelog.d/3499.added.md

@@ -0,0 +1,10 @@
+- **The RLOG writer can encode a block's pages on the caller's threads**
+  (issue #3499, ADR-3492 decision 7). `RlogWriter::finish_on` and
+  `finish_with_stats_on` hand each block's per-column page encodes to a
+  `PageEncodeExecutor` as `Send + 'static` jobs that own their input, and
+  assemble the results in staging order whatever order they finish in; the
+  object is byte-identical to `finish`'s under both page choices. A block
+  staging fewer than `PAGE_JOB_FLOOR` (8) pages is one job. `finish` keeps
+  its signature and behaviour. The writer's bloom token insert no longer
+  allocates a buffer per token. No caller uses the new API yet; the ingest
+  wiring is a separate change.

crates/ravel-logseg/benches/common/page_pool.rs

@@ -0,0 +1,69 @@
+//! A multi-threaded page encode executor (ADR-3492 decision 7) on
+//! `std::thread::scope`, shared by the `logseg_encode` bench and the parallel
+//! encode tests.
+
+use std::sync::Mutex;
+use std::thread;
+
+use ravel_logseg::{LogSegError, PageEncodeDone, PageEncodeExecutor, PageEncodeJob};
+
+/// Runs each block's jobs on `threads` scoped threads. Every worker takes the
+/// last job still queued, so jobs start in reverse column order, and the
+/// results go back in the order they finished, not job order.
+pub struct ScopedPool {
+    threads: usize,
+    /// Calls that had more than one job.
+    pub split_calls: usize,
+    /// Calls whose results went back in an order other than job order.
+    pub out_of_order_calls: usize,
+}
+
+impl ScopedPool {
+    pub fn new(threads: usize) -> Self {
+        ScopedPool {
+            threads,
+            split_calls: 0,
+            out_of_order_calls: 0,
+        }
+    }
+}
+
+impl PageEncodeExecutor for ScopedPool {
+    fn execute(&mut self, jobs: Vec<PageEncodeJob>) -> Result<Vec<PageEncodeDone>, LogSegError> {
+        let count = jobs.len();
+        let queue = Mutex::new(jobs);
+        let finished: Mutex<Vec<(usize, PageEncodeDone)>> = Mutex::new(Vec::with_capacity(count));
+        thread::scope(|s| {
+            for _ in 0..self.threads.max(1) {
+                s.spawn(|| {
+                    loop {
+                        let job = match queue.lock() {
+                            Ok(mut q) => q.pop(),
+                            Err(_) => None,
+                        };
+                        let Some(job) = job else { break };
+                        let index = job.index();
+                        let done = job.run();
+                        if let Ok(mut f) = finished.lock() {
+                            f.push((index, done));
+                        }
+                    }
+                });
+            }
+        });
+        let finished = finished
+            .into_inner()
+            .map_err(|_| LogSegError::PageEncode("a worker panicked".into()))?;
+        if count > 1 {
+            self.split_calls += 1;
+            if finished
+                .iter()
+                .enumerate()
+                .any(|(pos, (index, _))| pos != *index)
+            {
+                self.out_of_order_calls += 1;
+            }
+        }
+        Ok(finished.into_iter().map(|(_, done)| done).collect())
+    }
+}

crates/ravel-logseg/benches/logseg_encode.rs

@@ -6,14 +6,16 @@
 //! thin), at three record-count totals.
 //!
 //! The `ingest_shape` group encodes one ingest flush of the ADR-3492 benchmark
-//! shape (`common/ingest_shape.rs`) four ways: the row writer and the columnar
-//! writer, each under both page choices. The columnar arms time
-//! `ColumnarLogBatch::try_from_records` per push batch too, the work
-//! `rows_to_batch` does at ingest. The writer is single-threaded, so the
-//! timings are CPU time. Before timing, each flush size's stored object bytes
-//! are printed per arm, and the run fails unless the two writers agree byte
-//! for byte under each choice and `Estimated` stores at most
-//! [`MAX_ESTIMATED_GROWTH`] more than `Exhaustive` (ADR-3492, exit criteria).
+//! shape (`common/ingest_shape.rs`) eight ways: the row writer and the
+//! columnar writer, each under both page choices, each finished serially and
+//! with its column pages encoded on [`THREADS`] threads (ADR-3492 decision 7).
+//! The columnar arms time `ColumnarLogBatch::try_from_records` per push batch
+//! too, the work `rows_to_batch` does at ingest. A serial arm's timing is CPU
+//! time; a threaded arm's is wall time. Before timing, each flush size's
+//! stored object bytes are printed per arm, and the run fails unless the two
+//! writers and the threaded encode agree byte for byte under each choice and
+//! `Estimated` stores at most [`MAX_ESTIMATED_GROWTH`] more than `Exhaustive`
+//! (ADR-3492, exit criteria).
 #![allow(clippy::expect_used)]
 
 use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
@@ -27,6 +29,13 @@ use support::{bench_config, bench_identity, build_corpus};
 mod ingest_shape;
 use ingest_shape::{PUSH_BATCH, ingest_records};
 
+#[path = "common/page_pool.rs"]
+mod page_pool;
+use page_pool::ScopedPool;
+
+/// Threads the threaded arms encode column pages on.
+const THREADS: usize = 4;
+
 /// Record-count totals for the sweep. Kept well under the 8192-record block
 /// target's multi-block territory but large enough to amortize per-object
 /// fixed costs (footer, directories); 200_000 mirrors segment_encode's
@@ -85,34 +94,61 @@ fn writer(choice: PageChoice) -> RlogWriter {
     RlogWriter::new(cfg, bench_identity())
 }
 
-fn encode_rows(records: Vec<LogRecord>, choice: PageChoice) -> Vec<u8> {
+/// Finishes `w` serially, or with its pages encoded on `pool`.
+fn finish(w: RlogWriter, pool: Option<&mut ScopedPool>) -> Vec<u8> {
+    match pool {
+        Some(pool) => w.finish_on(pool).expect("finish_on"),
+        None => w.finish().expect("finish"),
+    }
+}
+
+fn encode_rows(
+    records: Vec<LogRecord>,
+    choice: PageChoice,
+    pool: Option<&mut ScopedPool>,
+) -> Vec<u8> {
     let mut w = writer(choice);
     for r in records {
         w.push(r).expect("push");
     }
-    w.finish().expect("finish")
+    finish(w, pool)
 }
 
-fn encode_columnar(records: Vec<LogRecord>, choice: PageChoice) -> Vec<u8> {
+fn encode_columnar(
+    records: Vec<LogRecord>,
+    choice: PageChoice,
+    pool: Option<&mut ScopedPool>,
+) -> Vec<u8> {
     let mut w = writer(choice);
     for batch in records.chunks(PUSH_BATCH) {
         let batch = ColumnarLogBatch::try_from_records(batch).expect("batch");
         w.push_columnar(batch).expect("push_columnar");
     }
-    w.finish().expect("finish")
+    finish(w, pool)
 }
 
 /// Prints each arm's stored object bytes for a flush and fails the run when
 /// the writers disagree or `Estimated` grows past its bound.
 fn check_sizes(flush: usize, records: &[LogRecord]) {
     let mut sizes = [0usize; 2];
     for (k, &(name, choice)) in CHOICES.iter().enumerate() {
-        let row = encode_rows(records.to_vec(), choice);
-        let columnar = encode_columnar(records.to_vec(), choice);
+        let row = encode_rows(records.to_vec(), choice, None);
+        let columnar = encode_columnar(records.to_vec(), choice, None);
         assert!(
             row == columnar,
             "{flush} records, {name}: row and columnar bytes differ"
         );
+        let mut pool = ScopedPool::new(THREADS);
+        let row_threaded = encode_rows(records.to_vec(), choice, Some(&mut pool));
+        let columnar_threaded = encode_columnar(records.to_vec(), choice, Some(&mut pool));
+        assert!(
+            row_threaded == row && columnar_threaded == row,
+            "{flush} records, {name}: threaded page encode bytes differ from finish"
+        );
+        assert!(
+            pool.split_calls > 0 && pool.out_of_order_calls > 0,
+            "{flush} records, {name}: no block was split and returned out of order"
+        );
         println!(
             "ingest_shape/{flush} stored_bytes row_{name}={} columnar_{name}={}",
             row.len(),
@@ -142,14 +178,30 @@ fn bench_ingest_shape(c: &mut Criterion) {
             group.bench_function(format!("{flush}/row_{name}"), |b| {
                 b.iter_batched(
                     || records.clone(),
-                    |records| encode_rows(records, choice),
+                    |records| encode_rows(records, choice, None),
                     BatchSize::LargeInput,
                 );
             });
             group.bench_function(format!("{flush}/columnar_{name}"), |b| {
                 b.iter_batched(
                     || records.clone(),
-                    |records| encode_columnar(records, choice),
+                    |records| encode_columnar(records, choice, None),
+                    BatchSize::LargeInput,
+                );
+            });
+            group.bench_function(format!("{flush}/row_{name}_{THREADS}threads"), |b| {
+                let mut pool = ScopedPool::new(THREADS);
+                b.iter_batched(
+                    || records.clone(),
+                    |records| encode_rows(records, choice, Some(&mut pool)),
+                    BatchSize::LargeInput,
+                );
+            });
+            group.bench_function(format!("{flush}/columnar_{name}_{THREADS}threads"), |b| {
+                let mut pool = ScopedPool::new(THREADS);
+                b.iter_batched(
+                    || records.clone(),
+                    |records| encode_columnar(records, choice, Some(&mut pool)),
                     BatchSize::LargeInput,
                 );
             });