diff --git a/BENCHMARKS.md b/BENCHMARKS.md index c105698..c4e1705 100644 --- a/BENCHMARKS.md +++ b/BENCHMARKS.md @@ -528,7 +528,11 @@ What this shows: (about 880 MB/s) while h5py processes reach 4424 MB/s. Hyperslab reads, which bypass the `File`'s chunk cache, keep scaling, so the cache (one mutex and one 16 MiB budget per `File`, thrashed by 64 MiB - datasets) is the suspect. + datasets) is the suspect. The cause of the `--decode-threads 1` + ceiling was not the cache: every full read queued its chunks for the + pool's single rayon worker. That case was fixed after these + measurements (2026-09-26, not yet re-measured here). With the default + pool the gap to h5py processes remains (see `docs/known-issues.md`). - *Contiguous reads are slow*: 2.5 GB/s for a single-threaded full read against h5py's 9.8 GB/s (0.25x), and 0.12x for 256 x 256 hyperslabs. Threads close the gap (about 1.0x h5py at 16), but single-thread diff --git a/CHANGELOG.md b/CHANGELOG.md index ef4bd2f..9030c3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## Unreleased +### Concurrent reads (2026-09-26) +- **Full reads of chunked datasets scale with threads again when rayon's + pool has one thread.** Each full read handed its chunks to rayon to + decode; with a one-thread pool (`RAYON_NUM_THREADS=1`, or + `concurrent_read --decode-threads 1`) every thread reading through a + `File` queued behind that single worker, so N readers decoded on one core + and throughput stopped at about 2x one thread. Such reads, and + `verify_provenance`'s uncached reader, now decode on the calling thread + (`clawhdf5_format::parallel_read::pool_can_parallelise`). The `File`'s + chunk cache, the suspect in `docs/known-issues.md`, was not the cause: + reads of datasets larger than its budget already skipped inserting, and + its lookups cost a few percent at 16 threads. Throughput with the default + pool is unchanged, and still short of an h5py process pool. + ### Plugin filters (2026-09-26) - **LZF, bitshuffle, bzip2 and Blosc read and write, in pure Rust.** Files written by h5py with `compression="lzf"`, or with hdf5plugin's diff --git a/crates/clawhdf5-format/src/chunked_read.rs b/crates/clawhdf5-format/src/chunked_read.rs index cbbaa69..9169ab7 100644 --- a/crates/clawhdf5-format/src/chunked_read.rs +++ b/crates/clawhdf5-format/src/chunked_read.rs @@ -40,6 +40,7 @@ fn decompress_all_chunks( { if let Some(pl) = pipeline && parallel_read::should_use_parallel(chunks.len()) + && parallel_read::pool_can_parallelise() { // Seed from the first chunk's address and count for determinism. let seed = chunks.first().map(|c| c.address).unwrap_or(0) ^ (chunks.len() as u64); @@ -1056,7 +1057,9 @@ pub fn read_chunked_data_cached( // Decompress what the cache didn't have, a bounded batch at a time — in // parallel with the `parallel` feature (this path, the one the facade - // uses, was sequential; only the uncached reader was parallel). Chunks are + // uses, was sequential; only the uncached reader was parallel), unless the + // pool has one thread: then every reading thread would queue behind that + // one worker, so each decodes its own chunks instead. Chunks are // cached only when the whole dataset fits: pushing a larger dataset // through the cache just evicts each chunk moments after inserting it. let cache_them = total_bytes <= cache.max_bytes(); @@ -1073,12 +1076,13 @@ pub fn read_chunked_data_cached( }; for batch in misses.chunks(DECODE_BATCH) { #[cfg(feature = "parallel")] - let decoded: Vec, FormatError>> = if batch.len() >= 4 { - use rayon::prelude::*; - batch.par_iter().map(decode).collect() - } else { - batch.iter().map(decode).collect() - }; + let decoded: Vec, FormatError>> = + if batch.len() >= 4 && parallel_read::pool_can_parallelise() { + use rayon::prelude::*; + batch.par_iter().map(decode).collect() + } else { + batch.iter().map(decode).collect() + }; #[cfg(not(feature = "parallel"))] let decoded: Vec, FormatError>> = batch.iter().map(decode).collect(); diff --git a/crates/clawhdf5-format/src/parallel_read.rs b/crates/clawhdf5-format/src/parallel_read.rs index 6593132..1b940c6 100644 --- a/crates/clawhdf5-format/src/parallel_read.rs +++ b/crates/clawhdf5-format/src/parallel_read.rs @@ -27,6 +27,20 @@ pub fn should_use_parallel(chunk_count: usize) -> bool { chunk_count > PARALLEL_THRESHOLD } +/// Whether handing a read's chunks to rayon can decode them faster than the +/// calling thread would alone. +/// +/// `false` when the pool the work would go to (the current pool inside a +/// rayon worker, else the global one) has a single thread. Handing work to +/// that pool is then worse than useless: the caller blocks while the one +/// worker decodes, and every other thread reading at the same time queues +/// behind the same worker, so N reader threads decode on one core. (That is +/// how full reads with `--decode-threads 1` stopped scaling at about 2x in +/// the `concurrent_read` benchmark.) +pub fn pool_can_parallelise() -> bool { + rayon::current_num_threads() > 1 +} + /// Decompress chunks in parallel using lane-partitioned assignment. /// /// Instead of naive `par_iter`, chunks are deterministically assigned to lanes diff --git a/crates/clawhdf5/tests/single_thread_decode_pool.rs b/crates/clawhdf5/tests/single_thread_decode_pool.rs new file mode 100644 index 0000000..455ed90 --- /dev/null +++ b/crates/clawhdf5/tests/single_thread_decode_pool.rs @@ -0,0 +1,90 @@ +//! With a one-thread rayon pool, full reads of chunked datasets must decode +//! on the calling thread. +//! +//! Handing a read's chunks to a one-worker pool made every reading thread +//! queue behind that single worker: N threads reading through one `File` +//! decoded on one core, and full reads stopped scaling at about 2x in the +//! `concurrent_read` benchmark with `--decode-threads 1` (see +//! `docs/known-issues.md`). The test makes that queueing observable: it keeps +//! the pool's only worker busy and requires reads to finish anyway. +//! +//! One test in its own binary: it configures the process-wide rayon pool. + +#![cfg(feature = "parallel")] + +use std::sync::mpsc; +use std::time::Duration; + +use clawhdf5::{File, FileBuilder}; + +const N: usize = 4096; // 64 chunks of 64 elements + +fn values() -> Vec { + (0..N).map(|i| i as f64 * 0.5).collect() +} + +fn build() -> File { + let mut b = FileBuilder::new(); + b.create_dataset("data") + .with_f64_data(&values()) + .with_shape(&[N as u64]) + .with_chunks(&[64]) + .with_deflate(1) + .with_provenance("test-suite", "2026-09-26T00:00:00Z", None); + File::from_bytes(b.finish().unwrap()).unwrap() +} + +/// Run `f` on a fresh thread; `None` if it has not finished within `limit`. +fn finishes_within( + limit: Duration, + f: impl FnOnce() -> T + Send + 'static, +) -> Option { + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(f()); + }); + rx.recv_timeout(limit).ok() +} + +#[test] +fn full_reads_do_not_wait_for_a_busy_one_thread_pool() { + rayon::ThreadPoolBuilder::new() + .num_threads(1) + .build_global() + .expect("this test binary configures the global pool first"); + + // Built first: the writer compresses on the pool too. + let file = std::sync::Arc::new(build()); + let file2 = std::sync::Arc::clone(&file); + + // Occupy the pool's only worker until the reads are done. + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel::<()>(); + rayon::spawn(move || { + started_tx.send(()).unwrap(); + let _ = release_rx.recv(); + }); + started_rx.recv().unwrap(); + + let limit = Duration::from_secs(20); + // Cached full read (the path `read_*` uses), then the uncached reader + // behind `verify_provenance`. + let read = finishes_within(limit, move || { + file.dataset("data").unwrap().read_f64().unwrap() + }); + let verified = finishes_within(limit, move || { + file2.dataset("data").unwrap().verify_provenance().unwrap() + }); + // Free the worker before asserting, so a failure does not hang the + // blocked reader threads forever. + release_tx.send(()).unwrap(); + + assert_eq!( + read.expect("a full read waited for the busy one-thread rayon pool"), + values() + ); + assert_eq!( + verified.expect("verify_provenance waited for the busy one-thread rayon pool"), + clawhdf5::provenance::VerifyResult::Ok + ); +} diff --git a/docs/known-issues.md b/docs/known-issues.md index bf8f77a..f8d7798 100644 --- a/docs/known-issues.md +++ b/docs/known-issues.md @@ -9,12 +9,29 @@ deleting it. ## Concurrent and contiguous read performance (measured 2026-09-26) -**Status:** open. Measured on tank with `concurrent_read` against h5py -3.16 / HDF5 2.0 (`BENCHMARKS.md`, "Concurrent reads"): -- Full reads of chunked datasets from several threads through one `File` - stop scaling at about 4 threads (880 MB/s on deflate data vs 4424 MB/s - for 16 h5py processes). Hyperslab reads, which skip the chunk cache, - scale to 1244 MB/s, so the `File`'s shared chunk cache is the suspect. +**Status:** open; one cause of the first bullet fixed (2026-09-26). Measured on +tank with `concurrent_read` against h5py 3.16 / HDF5 2.0 (`BENCHMARKS.md`, +"Concurrent reads"): +- **Partly fixed 2026-09-26.** Full reads of chunked datasets from several threads + through one `File` stop scaling at about 4 threads (880 MB/s on deflate + data vs 4424 MB/s for 16 h5py processes). Hyperslab reads, which skip the + chunk cache, scale to 1244 MB/s, so the `File`'s shared chunk cache is the + suspect. *Cause:* not the cache. Those numbers were taken with + `--decode-threads 1`, a one-thread rayon pool, and every full read handed + its chunks to that pool, so all reader threads queued behind its single + worker (per-thread CPU time: one thread did all the decoding, the 16 + readers almost none). Hyperslab reads touch one chunk each and never used + the pool. Reads now decode on the calling thread when the pool has one + thread (`tests/single_thread_decode_pool.rs`). **Still open:** this + fixes only a one-thread pool. With the default pool, 16 reader threads + ran at about 2900 MB/s before and after the change, still short of 16 + h5py processes (4424 MB/s); with a small pool (2-4 threads) readers + outside it still wait on its workers. Datasets larger than the + cache's budget were already read without inserting into it, and skipping + its lookups entirely gained only a few percent at 16 threads. Remaining + per-read overhead, not yet addressed: each full `read_f32` of a chunked + dataset faults in about three times its size in fresh pages (the output, + the `f32` copy of it, and a new buffer per decoded chunk). - Contiguous datasets read 4x slower than h5py on one thread (2.5 vs 9.8 GB/s full, 0.12x for 256 x 256 hyperslabs). Values are correct; this is speed only.