A cloud store can be used from inside an async runtime c9369ad8b1b1
A store blocks on a runtime of its own, which a thread already driving async tasks (a server's handler) may not do: such a call now runs on a thread of its own, and the store's runtime shuts down in the background when dropped. Found when the hub made a repository in a bucket. refs #68
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-10-05 01:33 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
c9369ad8b1b153b2493c3fafb5ab2a404dc86e0a356b1694469753c9bdd6a230- Depends on
- 7026bd7dc5b8 8d80356371c5
- Depended on by
- 767ac6cd0545 87c948720343 8b53b935dd03 2ddd471651cd
- In bookmarks
- bare-hashes carried-not-removed carry-identical doctor-trust identical-sides keyless-tags main merge-remote-bookmark only-help-103 paths-for-resolutions-and-amend push-notes show-stat subrepo-pin-state switch-says
- In tags
- v0.10.0 v0.11.0 v0.4.1 v0.4.2 v0.4.3 v0.4.4 v0.4.5 v0.4.6 v0.5.0 v0.5.1 v0.6.0 v0.7.0 v0.8.0 v0.9.0
- Edits
- 7 atoms, 0 tree operations, 1 file
Browse the tree at this change
modified crates/oxy-strata/src/lib.rs +44 −4
| … | 255 unchanged lines |
||
| 256 | 256 | prefix: String, |
|
| 257 | 257 | name: String, |
|
| 258 | 258 | hint: &'static str, |
|
| 259 | - | runtime: tokio::runtime::Runtime, |
|
| 259 | + | /// Always there until the store is dropped. |
|
| 260 | + | runtime: Option<tokio::runtime::Runtime>, |
|
| 260 | 261 | /// One call at a time: the runtime runs on the calling thread. |
|
| 261 | 262 | turn: Mutex<()>, |
|
| 262 | 263 | } |
|
| 263 | 264 | ||
| 265 | + | impl Drop for CloudStore { |
|
| 266 | + | /// Without waiting, so a store can be dropped inside a runtime too. |
|
| 267 | + | fn drop(&mut self) { |
|
| 268 | + | if let Some(rt) = self.runtime.take() { |
|
| 269 | + | rt.shutdown_background(); |
|
| 270 | + | } |
|
| 271 | + | } |
|
| 272 | + | } |
|
| 273 | + | ||
| 264 | 274 | impl CloudStore { |
|
| 265 | 275 | /// Open the store a URL names. Nothing is sent yet: credentials are |
|
| 266 | 276 | /// resolved here, but the bucket is first spoken to on the first call. |
|
| … | 73 unchanged lines |
||
| 340 | 350 | prefix, |
|
| 341 | 351 | name: parsed.plain(), |
|
| 342 | 352 | hint, |
|
| 343 | - | runtime, |
|
| 353 | + | runtime: Some(runtime), |
|
| 344 | 354 | turn: Mutex::new(()), |
|
| 345 | 355 | }) |
|
| 346 | 356 | } |
|
| 347 | 357 | ||
| 348 | - | fn block<F: Future>(&self, f: F) -> F::Output { |
|
| 358 | + | /// Run a request to completion. Called from a thread that is itself |
|
| 359 | + | /// driving async tasks (a server's handler), where blocking on another |
|
| 360 | + | /// runtime is not allowed, the request runs on a thread of its own. |
|
| 361 | + | fn block<F>(&self, f: F) -> F::Output |
|
| 362 | + | where |
|
| 363 | + | F: Future + Send, |
|
| 364 | + | F::Output: Send, |
|
| 365 | + | { |
|
| 349 | 366 | let _turn = self.turn.lock().unwrap_or_else(|p| p.into_inner()); |
|
| 350 | - | self.runtime.block_on(f) |
|
| 367 | + | let runtime = self |
|
| 368 | + | .runtime |
|
| 369 | + | .as_ref() |
|
| 370 | + | .expect("the runtime lives as long as the store"); |
|
| 371 | + | if tokio::runtime::Handle::try_current().is_ok() { |
|
| 372 | + | return std::thread::scope(|s| { |
|
| 373 | + | s.spawn(|| runtime.block_on(f)) |
|
| 374 | + | .join() |
|
| 375 | + | .unwrap_or_else(|p| std::panic::resume_unwind(p)) |
|
| 376 | + | }); |
|
| 377 | + | } |
|
| 378 | + | runtime.block_on(f) |
|
| 351 | 379 | } |
|
| 352 | 380 | ||
| 353 | 381 | fn path(&self, key: &str) -> Result<ObjectPath> { |
|
| … | 220 unchanged lines |
||
| 574 | 602 | ||
| 575 | 603 | #[cfg(test)] |
|
| 576 | 604 | mod tests { |
|
| 605 | + | #[tokio::test] |
|
| 606 | + | async fn a_store_can_be_used_from_inside_a_runtime() { |
|
| 607 | + | // As a server's handler would: blocking on the store's own runtime |
|
| 608 | + | // from here must neither panic nor deadlock, and nor must dropping it. |
|
| 609 | + | let store = CloudStore::open( |
|
| 610 | + | "s3://oxy-no-such-bucket/p?region=eu-west-2&endpoint=http://127.0.0.1:9", |
|
| 611 | + | ) |
|
| 612 | + | .unwrap(); |
|
| 613 | + | assert_eq!(store.block(async { 7 }), 7); |
|
| 614 | + | drop(store); |
|
| 615 | + | } |
|
| 616 | + | ||
| 577 | 617 | #[test] |
|
| 578 | 618 | fn credentials_round_trip_and_never_show_their_secret() { |
|
| 579 | 619 | let aws = Credentials::Aws { |
|
| … | 71 unchanged lines |