Migrate to tokio

.github/workflows/ci.yml

@@ -83,13 +83,11 @@ jobs:
         run: cargo test --verbose
         env:
           CARGO_REGISTRIES_CRATES_IO_PROTOCOL: sparse
-          ASYNC_STD_THREAD_COUNT: 4
 
       - name: Run doctests
         run: cargo test --doc
         env:
           CARGO_REGISTRIES_CRATES_IO_PROTOCOL: sparse
-          ASYNC_STD_THREAD_COUNT: 4
 
   # NOTE: In GitHub repository settings, the "Require status checks to pass
   # before merging" branch protection rule ensures that commits are only merged

Cargo.toml

@@ -30,7 +30,6 @@ categories = ["network-programming"]
 
 [workspace.dependencies]
 atoi = "2.0.0"
-async-std = "=1.12.0"
 async-trait = "0.1"
 clap = "3.2.23"
 ctrlc = "3.2.5"
@@ -49,8 +48,11 @@ flume = "0.11"
 hex = "0.4.3"
 xml-rpc = "0.0.12"
 rustc_version = "0.4"
+test-case = { version = "3.3" }
+tokio = { version = "1.35.1", features = ["process"] }
 tracing = "0.1"
 zenoh = { version = "0.11.0-dev", git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "dev/1.0.0", features = [
+  "internal",
   "unstable",
   "plugins",
 ] }

DEFAULT_CONFIG.json5

@@ -118,6 +118,18 @@
       //// The string format is [0-9]+(ns|us|ms|[smhdwy])
       ////
       // ros_master_polling_interval: "100ms",
+
+      ////
+      //// work_thread_num: The number of worker thread in TOKIO runtime (default: 2)
+      //// The configuration only takes effect if running as a dynamic plugin, which can not reuse the current runtime.
+      ////
+      // work_thread_num: 2,
+
+      ////
+      //// max_block_thread_num: The number of blocking thread in TOKIO runtime (default: 50)
+      //// The configuration only takes effect if running as a dynamic plugin, which can not reuse the current runtime.
+      ////
+      // max_block_thread_num: 50,
     },
 
     ////

zenoh-bridge-ros1/Cargo.toml

@@ -24,12 +24,12 @@ description = "Zenoh bridge for ROS1"
 publish = false
 
 [dependencies]
-async-std = { workspace = true, features = ["unstable", "attributes"] }
 clap = { workspace = true }
 ctrlc = { workspace = true }
 tracing = { workspace = true }
 lazy_static = { workspace = true }
 serde_json = { workspace = true }
+tokio = { workspace = true }
 zenoh = { workspace = true }
 zenoh-plugin-trait = { workspace = true }
 zenoh-plugin-ros1 = { workspace = true }

zenoh-bridge-ros1/src/main.rs

@@ -13,8 +13,8 @@
 //
 use std::str::FromStr;
 
-use async_std::channel::unbounded;
 use clap::{App, Arg};
+use tokio::sync::mpsc::unbounded_channel;
 use zenoh::{
     config::{Config, ZenohId},
     internal::{plugins::PluginsManager, runtime::RuntimeBuilder},
@@ -187,6 +187,16 @@ Bridge polls ROS1 master to get information on local topics. This option is the
 Accepted value:'
 A string such as 100ms, 2s, 5m
 The string format is [0-9]+(ns|us|ms|[smhdwy])"#
+        ))
+        .arg(Arg::from_usage(
+r#"--work_thread_num=[usize] \
+'The number of worker thread in TOKIO runtime (default: 2)
+The configuration only takes effect if running as a dynamic plugin, which can not reuse the current runtime.'"#
+        ))
+        .arg(Arg::from_usage(
+r#"--max_block_thread_num=[usize] \
+'The number of blocking thread in TOKIO runtime (default: 50)
+The configuration only takes effect if running as a dynamic plugin, which can not reuse the current runtime.'"#
         ));
     let args = app.get_matches();
 
@@ -252,15 +262,11 @@ The string format is [0-9]+(ns|us|ms|[smhdwy])"#
     config
 }
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
-    let (sender, receiver) = unbounded();
-    ctrlc::set_handler(move || {
-        sender
-            .send_blocking(())
-            .expect("Error handling Ctrl+C signal")
-    })
-    .expect("Error setting Ctrl+C handler");
+    let (sender, mut receiver) = unbounded_channel();
+    ctrlc::set_handler(move || sender.send(()).expect("Error handling Ctrl+C signal"))
+        .expect("Error setting Ctrl+C handler");
 
     zenoh::init_log_from_env_or("z=info");
     tracing::info!(

zenoh-plugin-ros1/Cargo.toml

@@ -41,6 +41,8 @@ flume = { workspace = true }
 futures = { workspace = true }
 git-version = { workspace = true }
 lazy_static = { workspace = true }
+test-case = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
@@ -63,10 +65,6 @@ ctrlc = { workspace = true }
 # TODO: https://zettascale.atlassian.net/browse/ZEN-291
 # zenoh-plugin-ros1 = { path = ".", features = ["test"]}
 
-[dependencies.async-std]
-version = "=1.12.0"
-features = ["unstable", "attributes"]
-
 [build-dependencies]
 rustc_version = { workspace = true }
 

zenoh-plugin-ros1/examples/ros1_pub.rs

@@ -17,7 +17,7 @@ use zenoh_plugin_ros1::ros_to_zenoh_bridge::{
     environment::Environment, ros1_master_ctrl::Ros1MasterCtrl, Ros1ToZenohBridge,
 };
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
     // initiate logging
     zenoh::try_init_log_from_env();
@@ -80,5 +80,5 @@ async fn main() {
             std::thread::sleep(core::time::Duration::from_secs(1));
         }
     };
-    async_std::task::spawn_blocking(working_loop).await;
+    tokio::task::spawn_blocking(working_loop).await.unwrap();
 }

zenoh-plugin-ros1/examples/ros1_service.rs

@@ -16,7 +16,7 @@ use zenoh_plugin_ros1::ros_to_zenoh_bridge::{
     environment::Environment, ros1_master_ctrl::Ros1MasterCtrl, Ros1ToZenohBridge,
 };
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
     // initiate logging
     zenoh::try_init_log_from_env();
@@ -86,6 +86,6 @@ async fn main() {
                 println!("Zenoh got error: {}", e);
             }
         }
-        async_std::task::sleep(core::time::Duration::from_secs(1)).await;
+        tokio::time::sleep(core::time::Duration::from_secs(1)).await;
     }
 }

zenoh-plugin-ros1/examples/ros1_standalone_pub.rs

@@ -12,17 +12,15 @@
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
 
-use async_std::channel::unbounded;
+use tokio::sync::mpsc::unbounded_channel;
 use zenoh_plugin_ros1::ros_to_zenoh_bridge::environment::Environment;
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
-    let (sender, receiver) = unbounded();
+    let (sender, mut receiver) = unbounded_channel();
     ctrlc::set_handler(move || {
         tracing::info!("Catching Ctrl+C...");
-        sender
-            .send_blocking(())
-            .expect("Error handling Ctrl+C signal")
+        sender.send(()).expect("Error handling Ctrl+C signal")
     })
     .expect("Error setting Ctrl+C handler");
 
@@ -59,5 +57,5 @@ async fn main() {
         }
         tracing::info!("Caught Ctrl+C, stopping...");
     };
-    async_std::task::spawn_blocking(working_loop).await;
+    tokio::task::spawn_blocking(working_loop).await.unwrap();
 }

zenoh-plugin-ros1/examples/ros1_standalone_sub.rs

@@ -12,17 +12,15 @@
 //   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
 //
 
-use async_std::channel::unbounded;
+use tokio::sync::mpsc::unbounded_channel;
 use zenoh_plugin_ros1::ros_to_zenoh_bridge::environment::Environment;
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
-    let (sender, receiver) = unbounded();
+    let (sender, mut receiver) = unbounded_channel();
     ctrlc::set_handler(move || {
         tracing::info!("Catching Ctrl+C...");
-        sender
-            .send_blocking(())
-            .expect("Error handling Ctrl+C signal")
+        sender.send(()).expect("Error handling Ctrl+C signal")
     })
     .expect("Error setting Ctrl+C handler");
 

zenoh-plugin-ros1/examples/ros1_sub.rs

@@ -17,7 +17,7 @@ use zenoh_plugin_ros1::ros_to_zenoh_bridge::{
     environment::Environment, ros1_master_ctrl::Ros1MasterCtrl, Ros1ToZenohBridge,
 };
 
-#[async_std::main]
+#[tokio::main]
 async fn main() {
     // initiate logging
     zenoh::try_init_log_from_env();
@@ -73,6 +73,6 @@ async fn main() {
     loop {
         println!("Zenoh Publisher: publishing data...");
         zenoh_publisher.put(data.clone()).await.unwrap();
-        async_std::task::sleep(core::time::Duration::from_secs(1)).await;
+        tokio::time::sleep(core::time::Duration::from_secs(1)).await;
     }
 }