Migrate from async_std to tokio

Cargo.toml

@@ -38,7 +38,6 @@ dynamic_plugin = []
 default = ["dynamic_plugin"]
 
 [dependencies]
-async-std = "=1.12.0"
 async-trait = "0.1.66"
 dunce = "1.0.3"
 git-version = "0.3.5"
@@ -49,6 +48,7 @@ regex = "1.7.1"
 rocksdb = "0.22.0"
 serde_json = "1.0.117"
 tempfile = "3.4.0"
+tokio = { version = "1.35.1", default-features = false } # Default features are disabled due to some crates' requirements
 tracing = "0.1"
 uhlc = "0.5.2"
 walkdir = "2.3.2"
@@ -57,7 +57,6 @@ zenoh = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "dev/1.0.
     "internal",
     "plugins"
 ] }
-
 zenoh_backend_traits = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "dev/1.0.0" }
 zenoh-plugin-trait = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "dev/1.0.0" }
 

src/data_info_mgt.rs

@@ -14,11 +14,12 @@
 use std::{
     borrow::Cow,
     path::{Path, PathBuf},
+    sync::Arc,
     time::Duration,
 };
 
-use async_std::sync::{Arc, Mutex};
 use rocksdb::DB;
+use tokio::sync::Mutex;
 use tracing::trace;
 use zenoh::{
     bytes::{Encoding, ZBytes},

src/files_mgt.rs

@@ -22,7 +22,6 @@ use std::{
     time::{SystemTime, UNIX_EPOCH},
 };
 
-use async_std::task;
 use tracing::{debug, trace, warn};
 use walkdir::{IntoIter, WalkDir};
 use zenoh::{
@@ -37,7 +36,7 @@ use zenoh::{
     Result as ZResult,
 };
 
-use crate::data_info_mgt::*;
+use crate::{blockon_runtime, data_info_mgt::*};
 
 pub const CONFLICT_SUFFIX: &str = ".##z";
 
@@ -378,7 +377,7 @@ impl Drop for FilesMgr {
         match self.on_closure {
             OnClosure::DeleteAll => {
                 // Close data_info_mgr at first
-                task::block_on(async move {
+                blockon_runtime(async move {
                     self.data_info_mgr
                         .close()
                         .await

src/lib.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 
-use std::{collections::HashMap, fs::DirBuilder, io::prelude::*, path::PathBuf};
+use std::{collections::HashMap, fs::DirBuilder, future::Future, io::prelude::*, path::PathBuf};
 
 use async_trait::async_trait;
 use tempfile::tempfile_in;
@@ -35,6 +35,32 @@ mod data_info_mgt;
 mod files_mgt;
 use files_mgt::*;
 
+const WORKER_THREAD_NUM: usize = 2;
+const MAX_BLOCK_THREAD_NUM: usize = 50;
+lazy_static::lazy_static! {
+    // The global runtime is used in the dynamic plugins, which we can't get the current runtime
+    static ref TOKIO_RUNTIME: tokio::runtime::Runtime = tokio::runtime::Builder::new_multi_thread()
+               .worker_threads(WORKER_THREAD_NUM)
+               .max_blocking_threads(MAX_BLOCK_THREAD_NUM)
+               .enable_all()
+               .build()
+               .expect("Unable to create runtime");
+}
+#[inline(always)]
+fn blockon_runtime<F: Future>(task: F) -> F::Output {
+    // Check whether able to get the current runtime
+    match tokio::runtime::Handle::try_current() {
+        Ok(rt) => {
+            // Able to get the current runtime (standalone binary), spawn on the current runtime
+            tokio::task::block_in_place(|| rt.block_on(task))
+        }
+        Err(_) => {
+            // Unable to get the current runtime (dynamic plugins), spawn on the global runtime
+            tokio::task::block_in_place(|| TOKIO_RUNTIME.block_on(task))
+        }
+    }
+}
+
 /// The environement variable used to configure the root of all storages managed by this FileSystemBackend.
 pub const SCOPE_ENV_VAR: &str = "ZENOH_BACKEND_FS_ROOT";