blackmwk commented on code in PR #2308:
URL: https://github.com/apache/iceberg-rust/pull/2308#discussion_r3166465433
##########
crates/iceberg/src/runtime/mod.rs:
##########
@@ -39,37 +43,245 @@ impl<T: Send + 'static> Future for JoinHandle<T> {
}
}
-#[allow(dead_code)]
-pub fn spawn<F>(f: F) -> JoinHandle<F::Output>
-where
- F: std::future::Future + Send + 'static,
- F::Output: Send + 'static,
-{
- JoinHandle(task::spawn(f))
+/// Handle to a single tokio runtime. Holds an optional `Arc` to keep the
+/// runtime alive when we own it, and a `Handle` for spawning.
+#[derive(Clone)]
+pub struct RuntimeHandle {
+ /// Keeps the tokio runtime alive when we own it (`Runtime::new`).
+ /// `None` when borrowing an existing runtime via `Handle::try_current()`.
+ _owned: Option<Arc<tokio::runtime::Runtime>>,
Review Comment:
What's the problem if it's not complete? If you are worring about some
actions may fail in the middle of executions, then this is not the right way to
guarantee that. There are many ways to crash the program, and storing
`Arc<Runtime>` doesn't help much and our design should not rely on it.
The problem with this design is that it makes difficult for user to
understand the subtle difference between these two api. We should either always
ask user to provide `Arc<Runtime>`, or always always only store `Handle`.
##########
crates/iceberg/src/runtime/mod.rs:
##########
@@ -39,37 +44,262 @@ impl<T: Send + 'static> Future for JoinHandle<T> {
}
}
-#[allow(dead_code)]
-pub fn spawn<F>(f: F) -> JoinHandle<F::Output>
-where
- F: std::future::Future + Send + 'static,
- F::Output: Send + 'static,
-{
- JoinHandle(task::spawn(f))
+/// Handle to a single tokio runtime. Holds an optional `Arc` to keep the
+/// runtime alive when we own it, and a `Handle` for spawning.
+#[derive(Clone)]
+pub struct RuntimeHandle {
+ /// Keeps the tokio runtime alive when we own it (`Runtime::new`).
+ /// `None` when borrowing an existing runtime via `Handle::try_current()`.
+ _owned: Option<Arc<tokio::runtime::Runtime>>,
+ handle: tokio::runtime::Handle,
+}
+
+impl fmt::Debug for RuntimeHandle {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("RuntimeHandle").finish()
+ }
+}
+
+impl RuntimeHandle {
+ /// Create a handle that owns the given tokio runtime.
+ fn new(runtime: Arc<tokio::runtime::Runtime>) -> Self {
+ let handle = runtime.handle().clone();
+ Self {
+ _owned: Some(runtime),
+ handle,
+ }
+ }
+
+ /// Create a handle that borrows an existing tokio runtime via its handle.
+ fn from_tokio_handle(handle: tokio::runtime::Handle) -> Self {
+ Self {
+ _owned: None,
+ handle,
+ }
+ }
+
+ /// Spawn an async task.
+ pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+ where
+ F: Future + Send + 'static,
+ F::Output: Send + 'static,
+ {
+ JoinHandle(self.handle.spawn(future))
+ }
+
+ /// Spawn a blocking task.
+ pub fn spawn_blocking<F, T>(&self, f: F) -> JoinHandle<T>
+ where
+ F: FnOnce() -> T + Send + 'static,
+ T: Send + 'static,
+ {
+ JoinHandle(self.handle.spawn_blocking(f))
+ }
+}
+
+/// Iceberg's runtime abstraction.
+///
+/// Contains separate handles for IO-bound and CPU-bound work. When constructed
+/// with a single tokio runtime, both `io()` and `cpu()` route to the same one.
+/// Use `new_with_split` to provide dedicated runtimes for each category.
+///
+/// Cloning is cheap (Arc clones internally).
+#[derive(Clone)]
+pub struct Runtime {
+ io: RuntimeHandle,
+ cpu: RuntimeHandle,
+}
+
+impl fmt::Debug for Runtime {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Runtime").finish()
+ }
+}
+
+impl Runtime {
+ /// Create a Runtime backed by a single tokio runtime for all work.
+ pub fn new(runtime: Arc<tokio::runtime::Runtime>) -> Self {
+ let handle = RuntimeHandle::new(runtime);
+ Self {
+ io: handle.clone(),
+ cpu: handle,
+ }
+ }
+
+ /// Create a Runtime with separate tokio runtimes for IO and CPU work.
+ pub fn new_with_split(
+ io_runtime: Arc<tokio::runtime::Runtime>,
+ cpu_runtime: Arc<tokio::runtime::Runtime>,
+ ) -> Self {
+ Self {
+ io: RuntimeHandle::new(io_runtime),
+ cpu: RuntimeHandle::new(cpu_runtime),
+ }
+ }
+
+ /// Create a Runtime that borrows the tokio runtime the caller is currently
+ /// running in.
+ ///
+ /// Returns an error if this is not called from within a tokio runtime
+ /// context. This is the preferred path for callers who already manage a
+ /// tokio runtime (e.g. `#[tokio::main]`, datafusion, axum): iceberg will
+ /// reuse that runtime rather than spawning its own.
+ pub fn try_from_current() -> Result<Self> {
+ let handle = tokio::runtime::Handle::try_current().map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "no tokio runtime in context; call Runtime::try_from_current()
\
+ from within a tokio runtime, or construct a Runtime
explicitly \
+ via Runtime::new / Runtime::new_with_split",
+ )
+ .with_source(e)
+ })?;
+ let rh = RuntimeHandle::from_tokio_handle(handle);
+ Ok(Self {
+ io: rh.clone(),
+ cpu: rh,
+ })
+ }
+
+ /// Handle for IO-bound work (network fetches, file reads).
+ pub fn io(&self) -> &RuntimeHandle {
+ &self.io
+ }
+
+ /// Handle for CPU-bound work (decoding, predicate eval, projection).
+ pub fn cpu(&self) -> &RuntimeHandle {
+ &self.cpu
+ }
}
-#[allow(dead_code)]
-pub fn spawn_blocking<F, T>(f: F) -> JoinHandle<T>
-where
- F: FnOnce() -> T + Send + 'static,
- T: Send + 'static,
-{
- JoinHandle(task::spawn_blocking(f))
+impl Default for Runtime {
Review Comment:
No, we should not implement `Default`
##########
crates/iceberg/src/runtime/mod.rs:
##########
@@ -17,12 +17,16 @@
// This module contains the async runtime abstraction for iceberg.
+use std::fmt;
use std::future::Future;
use std::pin::Pin;
+use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::task;
+use tracing::warn;
+/// Wrapper around tokio's JoinHandle that panics on task failure.
pub struct JoinHandle<T>(task::JoinHandle<T>);
Review Comment:
In this case I think it's reasonable to keep it, but we should not panic
when error happens, we should convert it to iceberg's error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
