fix(read): decode on the calling thread when rayon's pool has one thread
Full reads of chunked datasets handed their chunks to rayon. With a one-thread pool (concurrent_read --decode-threads 1, RAYON_NUM_THREADS=1) every thread reading through a File queued behind that single worker, so 16 readers decoded on one core: per-thread CPU time showed one thread doing all the decoding and the readers almost none, and full reads stopped at about 2x one thread. The cached full-read path and the uncached reader behind verify_provenance now decode inline when the pool cannot parallelise (parallel_read::pool_can_parallelise). The File's chunk cache was the suspect but not the cause: datasets over its budget were already read without inserting, and skipping its lookups gained only a few percent at 16 threads. The regression test keeps a one-thread global pool's worker busy and requires a full read and verify_provenance to finish anyway; before the fix both waited for the worker (timed out). Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
+5
-1
@@ -528,7 +528,11 @@ What this shows:
|
|||||||
(about 880 MB/s) while h5py processes reach 4424 MB/s. Hyperslab
|
(about 880 MB/s) while h5py processes reach 4424 MB/s. Hyperslab
|
||||||
reads, which bypass the `File`'s chunk cache, keep scaling, so the
|
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
|
cache (one mutex and one 16 MiB budget per `File`, thrashed by 64 MiB
|
||||||
datasets) is the suspect.
|
datasets) is the suspect. **Fixed after these measurements
|
||||||
|
(2026-09-26); the table above predates the fix and has not been
|
||||||
|
re-measured.** The cause was not the cache: with `--decode-threads 1`
|
||||||
|
every full read queued its chunks for the pool's single rayon worker;
|
||||||
|
see `docs/known-issues.md`.
|
||||||
- *Contiguous reads are slow*: 2.5 GB/s for a single-threaded full read
|
- *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.
|
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
|
Threads close the gap (about 1.0x h5py at 16), but single-thread
|
||||||
|
|||||||
@@ -2,6 +2,19 @@
|
|||||||
|
|
||||||
## Unreleased
|
## 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.
|
||||||
|
|
||||||
### Plugin filters (2026-09-26)
|
### Plugin filters (2026-09-26)
|
||||||
- **LZF, bitshuffle, bzip2 and Blosc read and write, in pure Rust.** Files
|
- **LZF, bitshuffle, bzip2 and Blosc read and write, in pure Rust.** Files
|
||||||
written by h5py with `compression="lzf"`, or with hdf5plugin's
|
written by h5py with `compression="lzf"`, or with hdf5plugin's
|
||||||
|
|||||||
@@ -40,6 +40,7 @@ fn decompress_all_chunks(
|
|||||||
{
|
{
|
||||||
if let Some(pl) = pipeline
|
if let Some(pl) = pipeline
|
||||||
&& parallel_read::should_use_parallel(chunks.len())
|
&& parallel_read::should_use_parallel(chunks.len())
|
||||||
|
&& parallel_read::pool_can_parallelise()
|
||||||
{
|
{
|
||||||
// Seed from the first chunk's address and count for determinism.
|
// 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);
|
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
|
// 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
|
// 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
|
// cached only when the whole dataset fits: pushing a larger dataset
|
||||||
// through the cache just evicts each chunk moments after inserting it.
|
// through the cache just evicts each chunk moments after inserting it.
|
||||||
let cache_them = total_bytes <= cache.max_bytes();
|
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) {
|
for batch in misses.chunks(DECODE_BATCH) {
|
||||||
#[cfg(feature = "parallel")]
|
#[cfg(feature = "parallel")]
|
||||||
let decoded: Vec<Result<Vec<u8>, FormatError>> = if batch.len() >= 4 {
|
let decoded: Vec<Result<Vec<u8>, FormatError>> =
|
||||||
use rayon::prelude::*;
|
if batch.len() >= 4 && parallel_read::pool_can_parallelise() {
|
||||||
batch.par_iter().map(decode).collect()
|
use rayon::prelude::*;
|
||||||
} else {
|
batch.par_iter().map(decode).collect()
|
||||||
batch.iter().map(decode).collect()
|
} else {
|
||||||
};
|
batch.iter().map(decode).collect()
|
||||||
|
};
|
||||||
#[cfg(not(feature = "parallel"))]
|
#[cfg(not(feature = "parallel"))]
|
||||||
let decoded: Vec<Result<Vec<u8>, FormatError>> = batch.iter().map(decode).collect();
|
let decoded: Vec<Result<Vec<u8>, FormatError>> = batch.iter().map(decode).collect();
|
||||||
|
|
||||||
|
|||||||
@@ -27,6 +27,20 @@ pub fn should_use_parallel(chunk_count: usize) -> bool {
|
|||||||
chunk_count > PARALLEL_THRESHOLD
|
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.
|
/// Decompress chunks in parallel using lane-partitioned assignment.
|
||||||
///
|
///
|
||||||
/// Instead of naive `par_iter`, chunks are deterministically assigned to lanes
|
/// Instead of naive `par_iter`, chunks are deterministically assigned to lanes
|
||||||
|
|||||||
@@ -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<f64> {
|
||||||
|
(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<T: Send + 'static>(
|
||||||
|
limit: Duration,
|
||||||
|
f: impl FnOnce() -> T + Send + 'static,
|
||||||
|
) -> Option<T> {
|
||||||
|
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
|
||||||
|
);
|
||||||
|
}
|
||||||
+19
-6
@@ -9,12 +9,25 @@ deleting it.
|
|||||||
|
|
||||||
## Concurrent and contiguous read performance (measured 2026-09-26)
|
## Concurrent and contiguous read performance (measured 2026-09-26)
|
||||||
|
|
||||||
**Status:** open. Measured on tank with `concurrent_read` against h5py
|
**Status:** first bullet fixed (2026-09-26), second open. Measured on
|
||||||
3.16 / HDF5 2.0 (`BENCHMARKS.md`, "Concurrent reads"):
|
tank with `concurrent_read` against h5py 3.16 / HDF5 2.0 (`BENCHMARKS.md`,
|
||||||
- Full reads of chunked datasets from several threads through one `File`
|
"Concurrent reads"):
|
||||||
stop scaling at about 4 threads (880 MB/s on deflate data vs 4424 MB/s
|
- **Fixed 2026-09-26.** Full reads of chunked datasets from several threads
|
||||||
for 16 h5py processes). Hyperslab reads, which skip the chunk cache,
|
through one `File` stop scaling at about 4 threads (880 MB/s on deflate
|
||||||
scale to 1244 MB/s, so the `File`'s shared chunk cache is the suspect.
|
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`). 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
|
- 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).
|
9.8 GB/s full, 0.12x for 256 x 256 hyperslabs).
|
||||||
Values are correct; this is speed only.
|
Values are correct; this is speed only.
|
||||||
|
|||||||
Reference in New Issue
Block a user