CTTY commented on code in PR #2308:
URL: https://github.com/apache/iceberg-rust/pull/2308#discussion_r3076021970
##########
crates/iceberg/src/table.rs:
##########
Review Comment:
I think we should pass runtime here as well, need to revisit
##########
crates/iceberg/src/runtime/mod.rs:
##########
@@ -39,37 +42,147 @@ 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))
+/// Stable runtime wrapper for spawning async tasks and blocking operations.
+///
+/// Wraps `Arc<tokio::runtime::Runtime>` internally. Cloning is cheap (Arc
clone).
+/// This is the public API boundary — internals can evolve without breaking
consumers.
+#[derive(Clone)]
+pub struct Runtime {
+ /// We keep an optional owned runtime to prevent it from being dropped.
+ /// When created via `Runtime::new()`, this holds the runtime alive.
+ /// When created via `Default` inside an existing tokio context, this is
None
+ /// and we only use the handle.
+ _owned: Option<Arc<tokio::runtime::Runtime>>,
+ handle: tokio::runtime::Handle,
}
-#[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 fmt::Debug for Runtime {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Runtime").finish()
+ }
+}
+
+impl Runtime {
+ /// Create a new Runtime wrapping the given tokio Runtime.
+ pub fn new(runtime: Arc<tokio::runtime::Runtime>) -> Self {
+ let handle = runtime.handle().clone();
+ Self {
+ _owned: Some(runtime),
+ handle,
+ }
+ }
+
+ /// Spawn an async task on the wrapped runtime.
+ 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 on the wrapped runtime.
+ 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))
+ }
+}
+
+impl Default for Runtime {
+ fn default() -> Self {
+ // Try to use the current tokio runtime handle if we're already inside
one.
+ // This avoids creating a new runtime that would panic when dropped in
an
+ // async context.
+ if let Ok(handle) = tokio::runtime::Handle::try_current() {
+ return Self {
+ _owned: None,
+ handle,
+ };
+ }
+
+ let rt = Arc::new(
Review Comment:
The fallback here seems very implicit, a warning log may help
##########
crates/iceberg/src/runtime/mod.rs:
##########
@@ -39,37 +42,147 @@ 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))
+/// Stable runtime wrapper for spawning async tasks and blocking operations.
+///
+/// Wraps `Arc<tokio::runtime::Runtime>` internally. Cloning is cheap (Arc
clone).
+/// This is the public API boundary — internals can evolve without breaking
consumers.
+#[derive(Clone)]
+pub struct Runtime {
Review Comment:
Need to think a bit about compute/io separation use cases
--
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]
