linkrs_sync/
runtime.rs

1//! Runtime-aware bridging from synchronous code to asynchronous operations.
2//!
3//! Operators and managers run on arbitrary threads (per-query worker threads,
4//! session threads, maintenance threads). A bare executor-agnostic
5//! `block_on` cannot drive futures that depend on the tokio reactor or
6//! timers (network clients, timeouts), so callers should route through this
7//! helper, which reuses the ambient runtime when one is present.
8
9/// Error raised when no runtime can be established to drive a future.
10#[derive(Debug)]
11pub struct RuntimeBridgeError(String);
12
13impl std::fmt::Display for RuntimeBridgeError {
14    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
15        write!(f, "runtime bridge failed: {}", self.0)
16    }
17}
18
19impl std::error::Error for RuntimeBridgeError {}
20
21/// Block the calling thread until `future` completes.
22///
23/// Behavior by ambient context:
24/// - inside a multi-thread runtime: `block_in_place` on the current worker;
25/// - inside a current-thread runtime: drive the future on a dedicated
26///   transient runtime thread (the worker itself must stay free);
27/// - outside any runtime: drive the future on a transient current-thread
28///   runtime created here.
29pub fn block_on_ambient<F, T>(future: F) -> Result<T, RuntimeBridgeError>
30where
31    F: std::future::Future<Output = T> + Send,
32    T: Send,
33{
34    if let Ok(handle) = tokio::runtime::Handle::try_current() {
35        return match handle.runtime_flavor() {
36            tokio::runtime::RuntimeFlavor::MultiThread => {
37                Ok(tokio::task::block_in_place(|| handle.block_on(future)))
38            }
39            tokio::runtime::RuntimeFlavor::CurrentThread => std::thread::scope(|scope| {
40                scope
41                    .spawn(|| {
42                        let runtime = tokio::runtime::Builder::new_current_thread()
43                            .enable_all()
44                            .build()
45                            .map_err(|error| RuntimeBridgeError(error.to_string()))?;
46                        Ok(runtime.block_on(future))
47                    })
48                    .join()
49                    .map_err(|_| {
50                        RuntimeBridgeError("transient runtime thread panicked".to_string())
51                    })?
52            }),
53            _ => Ok(handle.block_on(future)),
54        };
55    }
56
57    let runtime = tokio::runtime::Builder::new_current_thread()
58        .enable_all()
59        .build()
60        .map_err(|error| RuntimeBridgeError(error.to_string()))?;
61    Ok(runtime.block_on(future))
62}
63
64#[cfg(test)]
65mod tests {
66    use super::*;
67
68    #[test]
69    fn completes_future_outside_any_runtime() {
70        let value = block_on_ambient(async { 41 + 1 }).expect("bridge must succeed");
71        assert_eq!(value, 42);
72    }
73
74    #[test]
75    fn completes_future_inside_multi_thread_runtime() {
76        let runtime = tokio::runtime::Builder::new_multi_thread()
77            .worker_threads(2)
78            .build()
79            .expect("test runtime must build");
80        let value = runtime.block_on(async move {
81            let inner = tokio::task::spawn_blocking(|| {
82                block_on_ambient(async { std::string::String::from("from-blocking") })
83                    .expect("bridge must succeed")
84            })
85            .await
86            .expect("spawn_blocking must succeed");
87            assert_eq!(inner, "from-blocking");
88            block_on_ambient(async { true }).expect("bridge must succeed")
89        });
90        assert!(value);
91    }
92}
linkrs_sync/runtime.rs

linkrs_sync/
runtime.rs
