diff --git a/CHANGELOG.md b/CHANGELOG.md index 69941c3..344afbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,9 +32,10 @@ backoff; bodies are asked for with `Accept-Encoding: identity` and an encoded one is refused. The ranges of one call are fetched in parallel. - **`ObjectStoreStorage`** reads one object of any `object_store` store, - pinned by ETag (else version or modification time) and size. It blocks - on a small tokio runtime it owns; called from inside another runtime it - refuses (`RemoteError::Usage`) instead of blocking a worker. + pinned by ETag (else version or modification time) and size. Each read + runs on a small tokio runtime the storage owns while the caller waits, + so it works from any thread, `spawn_blocking` and other runtimes + included. `open_object(store, path, options)` opens a file through a block cache. - Counted on the conformance corpus (tank, 2026-09-26, `CLAWHDF5_REMOTE_CORPUS=conformance/.cache/corpus CLAWHDF5_REMOTE_REPORT=1 diff --git a/crates/clawhdf5-remote/README.md b/crates/clawhdf5-remote/README.md index b4a487d..c64c7c6 100644 --- a/crates/clawhdf5-remote/README.md +++ b/crates/clawhdf5-remote/README.md @@ -62,11 +62,11 @@ checks both). ## Object stores `object_store` is async; `Storage` is synchronous (parsing is CPU work). -`ObjectStoreStorage` owns a small tokio runtime (two worker threads) and -blocks the calling thread on it for each read, so it is read from ordinary -threads, several at once. From inside an async runtime it refuses -(`RemoteError::Usage`) rather than block a worker: read in -`tokio::task::spawn_blocking`. The object is pinned by its ETag +`ObjectStoreStorage` owns a small tokio runtime (two worker threads): each +read runs there while the calling thread waits, so it works from any +thread, several at once โ€” including `tokio::task::spawn_blocking` and code +inside another runtime (where `spawn_blocking` is still the better place, +since a read blocks the thread it is called on). The object is pinned by its ETag (`If-Match`, and compared on every response), else its version or modification time, and its size. The ranges of one read are fetched concurrently (up to 8). diff --git a/crates/clawhdf5-remote/src/object.rs b/crates/clawhdf5-remote/src/object.rs index c8c5664..ecfb7ac 100644 --- a/crates/clawhdf5-remote/src/object.rs +++ b/crates/clawhdf5-remote/src/object.rs @@ -3,12 +3,11 @@ //! //! `object_store` is async and [`Storage`] is synchronous (parsing is CPU //! work; `docs/design/range-reads.md` ยง3 (a)). The storage owns a small -//! multi-threaded tokio runtime (two worker threads) and blocks the calling -//! thread on it for each read, so it can be used from ordinary threads โ€” -//! several at once. Calling it from inside another tokio runtime would -//! block that runtime's worker, so it refuses with -//! [`RemoteError::Usage`]: from async code, read in -//! `tokio::task::spawn_blocking`. +//! multi-threaded tokio runtime (two worker threads): each read is spawned +//! on it and the calling thread waits for the result, so it can be used +//! from any thread โ€” several at once, and from async code too. From async +//! code prefer `tokio::task::spawn_blocking` (a read blocks the thread it +//! is called on, which inside a runtime is one of its workers). //! //! The object is pinned when the storage is made: its size, and its ETag //! (sent as `If-Match` with every read, and compared with every response) @@ -105,19 +104,23 @@ impl ObjectStoreStorage { ) } - fn block_on( + /// Run `fut` on the storage's own runtime and wait for it. The future + /// never runs on the caller's thread, so the caller's context does not + /// matter: a plain thread, `spawn_blocking`, or even inside another + /// runtime (whose thread is then blocked for the duration of the read, + /// as by any blocking call, but nothing deadlocks or panics). + fn block_on( &self, - fut: impl std::future::Future>, + fut: impl std::future::Future> + Send + 'static, ) -> Result { - if tokio::runtime::Handle::try_current().is_ok() { - return Err(RemoteError::Usage( - "ObjectStoreStorage blocks on its own runtime and cannot be read from inside \ - an async runtime; read in tokio::task::spawn_blocking" - .into(), - )); - } let rt = self.runtime.as_ref().expect("runtime lives until drop"); - rt.block_on(fut) + let (tx, rx) = std::sync::mpsc::sync_channel(1); + rt.spawn(async move { + let _ = tx.send(fut.await); + }); + rx.recv().map_err(|_| { + RemoteError::ObjectStore("the object store task ended without a result".into()) + })? } fn options(&self, range: Range) -> GetOptions { diff --git a/crates/clawhdf5-remote/tests/object_store.rs b/crates/clawhdf5-remote/tests/object_store.rs index 0d2f5fa..a3208ba 100644 --- a/crates/clawhdf5-remote/tests/object_store.rs +++ b/crates/clawhdf5-remote/tests/object_store.rs @@ -109,10 +109,51 @@ fn missing_objects_and_async_callers_are_clean_errors() { assert!(ObjectStoreStorage::new(store.clone(), ObjectPath::from("nope.h5")).is_err()); put(store.as_ref(), "m.h5", multi_block_file()); let storage = ObjectStoreStorage::new(store, ObjectPath::from("m.h5")).unwrap(); - // From inside a runtime: refused, not a panic or a deadlock. + // From inside a current-thread runtime: works (the read runs on the + // storage's own runtime), no panic, no deadlock. let rt = tokio_rt(); let r = rt.block_on(async { storage.read_at(0, 10).map(|b| b.len()) }); - assert!(r.unwrap_err().to_string().contains("spawn_blocking")); + assert_eq!(r.unwrap(), 10); // Dropping the storage inside a runtime does not panic. rt.block_on(async move { drop(storage) }); } + +/// The advice for async callers works: a read in `spawn_blocking` of a +/// multi-threaded runtime (where `Handle::try_current` is Ok), and a read +/// straight inside a current-thread runtime's task. +#[test] +fn object_stores_read_from_spawn_blocking_and_inside_runtimes() { + let store: Arc = Arc::new(InMemory::new()); + let bytes = multi_block_file(); + put(store.as_ref(), "m.h5", bytes.clone()); + let want = File::from_bytes(bytes) + .unwrap() + .dataset("big") + .unwrap() + .read_f64() + .unwrap(); + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .build() + .unwrap(); + let (s, w) = (store.clone(), want.clone()); + let got = rt + .block_on(async move { + tokio::task::spawn_blocking(move || { + assert!(tokio::runtime::Handle::try_current().is_ok()); + let (f, _) = + open_object(s, "m.h5", &Options::default()).map_err(|e| e.to_string())?; + let v = f.dataset("big").unwrap().read_f64().unwrap(); + Ok::<_, String>(v == w) + }) + .await + }) + .unwrap(); + assert_eq!(got, Ok(true)); + let rt = tokio_rt(); + let got = rt.block_on(async { + let (f, _) = open_object(store, "m.h5", &Options::default()).unwrap(); + f.dataset("big").unwrap().read_f64().unwrap() + }); + assert_eq!(got, want); +} diff --git a/docs/design/range-reads.md b/docs/design/range-reads.md index 2c7d1a4..123dc3b 100644 --- a/docs/design/range-reads.md +++ b/docs/design/range-reads.md @@ -471,9 +471,10 @@ fast path within benchmark noise. with ring. S3/GCS/Azure go through `object_store` (`ObjectStoreStorage`, features `s3`/`gcs`/`azure`, opt-in because of aws-lc-rs); the `object-store` feature alone (in-memory, local files, a store you - build) is pure Rust. object_store is async: the storage blocks on a - two-thread tokio runtime of its own, and refuses to run inside another - runtime. + build) is pure Rust. object_store is async: each read runs on a + two-thread tokio runtime of the storage's own while the caller waits, + so the caller's context (a plain thread, `spawn_blocking`, another + runtime) does not matter. - `BlockCache` (any `Storage`): 1 MiB blocks and a 64 MiB LRU budget by default, the first block fetched at open (for HTTP by the request that learns the length), the missing blocks of one read fetched as