osobhandClaude Opus 5.5
a5e41c1a53
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]>
91 lines
3.0 KiB
Rust
91 lines
3.0 KiB
Rust
//! 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
|
|
);
|
|
}
|