utils: introduce per-thread tokio current thread Runtime

app/Cargo.toml

@@ -2,7 +2,7 @@
 name = "nydus-app"
 version = "0.2.0"
 authors = ["The Nydus Developers"]
-description = "Application framework and utilities for Nydus"
+description = "Application framework for Nydus Image Service"
 readme = "README.md"
 repository = "https://github.com/dragonflyoss/image-service"
 license = "Apache-2.0 OR BSD-3-Clause"

storage/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "nydus-storage"
 version = "0.5.0"
-description = "The core storage subsystem for Nydus Image Service"
+description = "Storage subsystem for Nydus Image Service"
 authors = ["The Nydus Developers"]
 license = "Apache-2.0 OR BSD-3-Clause"
 homepage = "https://nydus.dev/"

storage/src/cache/worker.rs

@@ -18,6 +18,7 @@ use nydus_utils::metrics::{BlobcacheMetrics, Metric};
 use tokio::runtime::Runtime;
 
 use nydus_api::http::BlobPrefetchConfig;
+use nydus_utils::async_helper::with_runtime;
 use nydus_utils::mpmc::Channel;
 
 use crate::cache::{BlobCache, BlobIoRange};
@@ -199,11 +200,9 @@ impl AsyncWorkerMgr {
                         .prefetch_workers
                         .fetch_add(1, Ordering::Relaxed);
 
-                    let rt = tokio::runtime::Builder::new_current_thread()
-                        .enable_all()
-                        .build()
-                        .expect("storage: failed to create tokio runtime for current thread");
-                    rt.block_on(Self::handle_prefetch_requests(mgr2.clone(), &rt));
+                    with_runtime(|rt| {
+                        rt.block_on(Self::handle_prefetch_requests(mgr2.clone(), rt));
+                    });
 
                     mgr2.metrics
                         .prefetch_workers

utils/CHANGELOG.md

@@ -1,6 +1,9 @@
 # Changelog
 ## [Unreleased]
 
+## [v0.3.1]
+- Add async io helpers
+
 ## [v0.3]
 
 ### Added

utils/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "nydus-utils"
-version = "0.3.0"
-description = "Compression/encryption/digest utilities for Nydus Image Service"
+version = "0.3.1"
+description = "Utilities and helpers for Nydus Image Service"
 authors = ["The Nydus Developers"]
 license = "Apache-2.0 OR BSD-3-Clause"
 homepage = "https://nydus.dev/"

utils/README.md

@@ -2,9 +2,11 @@
 
 A collection of utilities to support [Nydus Image Service](https://nydus.dev/).
 It provides:
-- LZ4 and zstd compression algorithms
+- Asynchronous Multi-Producer Multi-Consumer channel
 - Blake3 and SHA256 message digest algorithms
+- LZ4 and zstd compression algorithms
 - `InodeBitmap`: a bitmap implementation to manage inode numbers
+- Per-thread async runtime of type tokio current thread Runtime.
 - exec() helper
 - metric helpers
 

utils/src/async_helper.rs

@@ -0,0 +1,33 @@
+// Copyright (C) 2022 Alibaba Cloud. All rights reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+use tokio::runtime::{Builder, Runtime};
+
+std::thread_local! {
+    static CURRENT_THREAD_RT: Runtime = Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("utils: failed to create tokio runtime for current thread");
+}
+
+/// Run the callback with a tokio current thread runtime instance.
+pub fn with_runtime<F, R>(f: F) -> R
+where
+    F: FnOnce(&Runtime) -> R,
+{
+    CURRENT_THREAD_RT.with(f)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_with_runtime() {
+        let res = with_runtime(|rt| rt.block_on(async { 1 }));
+        assert_eq!(res, 1);
+
+        let res = with_runtime(|rt| rt.block_on(async { 3 }));
+        assert_eq!(res, 3);
+    }
+}

utils/src/lib.rs

@@ -17,6 +17,7 @@ pub use self::exec::*;
 pub use self::inode_bitmap::InodeBitmap;
 pub use self::types::*;
 
+pub mod async_helper;
 pub mod compress;
 pub mod digest;
 pub mod exec;