crates/clawhdf5-format/src/parallel_read.rs

@@ -41,6 +41,126 @@ pub fn pool_can_parallelise() -> bool {
     rayon::current_num_threads() > 1
 }
 
+/// How many rayon workers [`run_with_helpers`] should ask to help with
+/// `items` work items, given that the calling thread works too: the pool's
+/// other threads (all of them when the caller is not one), at most one per
+/// item beyond the caller's first.
+pub(crate) fn helper_count(items: usize) -> usize {
+    let pool = rayon::current_num_threads();
+    let others = if rayon::current_thread_index().is_some() {
+        pool.saturating_sub(1)
+    } else {
+        pool
+    };
+    others.min(items.saturating_sub(1))
+}
+
+/// Run `body` on the calling thread and on up to `helpers` rayon workers at
+/// once, returning when the caller's call has finished and every worker that
+/// started one has too. `body` shares its work out itself (typically by
+/// claiming items from an atomic counter until none are left).
+///
+/// The caller never waits for a worker to *become* free: helpers are queued
+/// on the pool, and one that only gets to run after the caller has finished
+/// returns without calling `body`. So a busy or small pool can only fail to
+/// speed a read up, never hold it back — with `par_iter`, the calling thread
+/// (not a pool worker) handed all the work to the pool and slept, and N
+/// threads reading through a 2-worker pool decoded on 2 cores.
+///
+/// A panic in `body`, on any thread, is resumed on the caller once every
+/// helper that started has stopped.
+pub(crate) fn run_with_helpers(helpers: usize, body: &(dyn Fn() + Sync)) {
+    use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
+    use std::sync::{Arc, Condvar, Mutex, PoisonError};
+
+    if helpers == 0 {
+        body();
+        return;
+    }
+
+    type Body = dyn Fn() + Sync + 'static;
+    struct Shared {
+        /// `body`, its lifetime erased. Only dereferenced by a helper that
+        /// registered in `state` while it was open (see below).
+        body: *const Body,
+        /// (closed, helpers inside `body`).
+        state: Mutex<(bool, usize)>,
+        idle: Condvar,
+        panic: Mutex<Option<Box<dyn core::any::Any + Send>>>,
+    }
+    // SAFETY: `body` points to a `Sync` closure, so calling it from other
+    // threads is allowed; the pointer is only used under the protocol below,
+    // which keeps it from outliving the closure.
+    unsafe impl Send for Shared {}
+    unsafe impl Sync for Shared {}
+
+    fn help(shared: &Shared) {
+        {
+            let mut state = shared.state.lock().unwrap_or_else(PoisonError::into_inner);
+            if state.0 {
+                return;
+            }
+            state.1 += 1;
+        }
+        // SAFETY: registered while open, so the caller of `run_with_helpers`
+        // is still inside it (it closes, then waits until no helper is
+        // registered, before returning), and `body` is alive.
+        let body = unsafe { &*shared.body };
+        if let Err(payload) = catch_unwind(AssertUnwindSafe(body)) {
+            shared
+                .panic
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner)
+                .get_or_insert(payload);
+        }
+        let mut state = shared.state.lock().unwrap_or_else(PoisonError::into_inner);
+        state.1 -= 1;
+        if state.1 == 0 {
+            shared.idle.notify_all();
+        }
+    }
+
+    let body_ptr: *const (dyn Fn() + Sync + '_) = body;
+    // SAFETY: only the lifetime changes (same fat-pointer layout). The
+    // pointer is dereferenced only while this function is running: see
+    // `help` and the wait below.
+    let body_ptr: *const Body = unsafe { core::mem::transmute(body_ptr) };
+    let shared = Arc::new(Shared {
+        body: body_ptr,
+        state: Mutex::new((false, 0)),
+        idle: Condvar::new(),
+        panic: Mutex::new(None),
+    });
+    for _ in 0..helpers {
+        let shared = Arc::clone(&shared);
+        rayon::spawn(move || help(&shared));
+    }
+    let caller = catch_unwind(AssertUnwindSafe(body));
+    {
+        // Close, then wait for the helpers inside `body`; later ones return
+        // at once. This must happen even if `body` panicked on this thread.
+        let mut state = shared.state.lock().unwrap_or_else(PoisonError::into_inner);
+        state.0 = true;
+        while state.1 > 0 {
+            state = shared
+                .idle
+                .wait(state)
+                .unwrap_or_else(PoisonError::into_inner);
+        }
+    }
+    if let Err(payload) = caller {
+        resume_unwind(payload);
+    }
+    let helper_panic = shared
+        .panic
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner)
+        .take();
+    if let Some(payload) = helper_panic {
+        resume_unwind(payload);
+    }
+}
+
 /// Decompress chunks in parallel using lane-partitioned assignment.
 ///
 /// Instead of naive `par_iter`, chunks are deterministically assigned to lanes
@@ -258,6 +378,60 @@ mod tests {
         (file, infos)
     }
 
+    /// Every item is processed exactly once, whatever mix of caller and
+    /// helpers ends up doing it.
+    #[test]
+    fn run_with_helpers_shares_all_work() {
+        use core::sync::atomic::{AtomicUsize, Ordering};
+        for helpers in [0, 1, 3, 16] {
+            let n = 1000;
+            let next = AtomicUsize::new(0);
+            let done: Vec<AtomicUsize> = (0..n).map(|_| AtomicUsize::new(0)).collect();
+            run_with_helpers(helpers, &|| {
+                loop {
+                    let i = next.fetch_add(1, Ordering::Relaxed);
+                    if i >= n {
+                        break;
+                    }
+                    done[i].fetch_add(1, Ordering::Relaxed);
+                }
+            });
+            assert!(done.iter().all(|d| d.load(Ordering::Relaxed) == 1));
+        }
+    }
+
+    /// A panic in the shared body reaches the caller whichever thread it
+    /// happened on, and only after the helpers inside the body have left it
+    /// (they borrow the caller's stack).
+    #[test]
+    fn run_with_helpers_propagates_panics() {
+        use core::sync::atomic::{AtomicUsize, Ordering};
+        use std::panic::{AssertUnwindSafe, catch_unwind};
+        let caller = std::thread::current().id();
+        for panic_on_caller in [true, false] {
+            let inside = AtomicUsize::new(0);
+            let calls = AtomicUsize::new(0);
+            let result = catch_unwind(AssertUnwindSafe(|| {
+                run_with_helpers(4, &|| {
+                    inside.fetch_add(1, Ordering::SeqCst);
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    let on_caller = std::thread::current().id() == caller;
+                    std::thread::sleep(std::time::Duration::from_millis(20));
+                    inside.fetch_sub(1, Ordering::SeqCst);
+                    if on_caller == panic_on_caller {
+                        panic!("boom");
+                    }
+                });
+            }));
+            // A helper may never have run (the pool was slow to start it),
+            // in which case nothing panicked when `panic_on_caller` is false.
+            if panic_on_caller || calls.load(Ordering::SeqCst) > 1 {
+                assert!(result.is_err());
+            }
+            assert_eq!(inside.load(Ordering::SeqCst), 0);
+        }
+    }
+
     /// Every parallel decoder refuses a chunk that decodes short, naming it.
     #[test]
     fn short_decoded_chunk_is_an_error() {

crates/clawhdf5/tests/busy_decode_pool.rs

@@ -0,0 +1,112 @@
+//! Full reads of chunked datasets must not wait for a busy rayon pool of
+//! any size.
+//!
+//! A full read handed its chunks to the rayon pool (`par_iter`) and the
+//! calling thread — not a pool worker — slept until the pool had decoded
+//! them. With a small pool (2-4 threads) and more reading threads than
+//! workers, every reader queued behind the same few workers
+//! (`docs/known-issues.md`, "Concurrent and contiguous read performance").
+//! Now the calling thread decodes too, and pool workers only help when they
+//! are free. The test keeps both workers of a two-thread pool busy and
+//! requires reads to finish anyway, with the right values.
+//!
+//! 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.25 - 7.0).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_small_pool() {
+    const WORKERS: usize = 2;
+    rayon::ThreadPoolBuilder::new()
+        .num_threads(WORKERS)
+        .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());
+
+    // Occupy every worker of the pool until the reads are done.
+    let (started_tx, started_rx) = mpsc::channel();
+    let (release_tx, release_rx) = mpsc::channel::<()>();
+    let release_rx = std::sync::Arc::new(std::sync::Mutex::new(release_rx));
+    for _ in 0..WORKERS {
+        let (started_tx, release_rx) = (started_tx.clone(), release_rx.clone());
+        rayon::spawn(move || {
+            started_tx.send(()).unwrap();
+            let _ = release_rx.lock().unwrap().recv();
+        });
+    }
+    for _ in 0..WORKERS {
+        started_rx.recv().unwrap();
+    }
+
+    let limit = Duration::from_secs(20);
+    // Several readers at once, as in the `concurrent_read` benchmark: the
+    // cached full read (`read_*`), the typed one and the uncached reader
+    // behind `verify_provenance`.
+    let readers: Vec<_> = (0..4)
+        .map(|_| {
+            let file = std::sync::Arc::clone(&file);
+            std::thread::spawn(move || {
+                finishes_within(limit, move || {
+                    let ds = file.dataset("data").unwrap();
+                    (
+                        ds.read_f64().unwrap(),
+                        ds.read_f32().unwrap(),
+                        ds.verify_provenance().unwrap(),
+                    )
+                })
+            })
+        })
+        .collect();
+    let results: Vec<_> = readers.into_iter().map(|h| h.join().unwrap()).collect();
+    // Free the workers before asserting, so a failure does not hang the
+    // blocked reader threads forever.
+    for _ in 0..WORKERS {
+        release_tx.send(()).unwrap();
+    }
+
+    let want = values();
+    let want_f32: Vec<f32> = want.iter().map(|&v| v as f32).collect();
+    for result in results {
+        let (f64s, f32s, verified) =
+            result.expect("a full read waited for the busy two-thread rayon pool");
+        assert_eq!(f64s, want);
+        assert_eq!(f32s, want_f32);
+        assert_eq!(verified, clawhdf5::provenance::VerifyResult::Ok);
+    }
+}