Repository: eclipse-zenoh-flow/zenoh-flow (language: rust)

Issue:
Remove blocking tasks
In Zenoh Flow runtime each graph node is spawned in a blocking task, while this ensures some isolation between the nodes it does not scale, as each blocking task may require a blocking thread.

The code where this happens is: https://github.com/eclipse-zenoh/zenoh-flow/blob/master/zenoh-flow/src/runtime/dataflow/instance/runners/mod.rs#L250 

So ideally, those tasks should be spawned as normal async tasks. 



Discussion / hints:

Fix patch:
diff --git a/cargo-zenoh-flow/Cargo.toml b/cargo-zenoh-flow/Cargo.toml
index ff5a8610..2d99a553 100644
--- a/cargo-zenoh-flow/Cargo.toml
+++ b/cargo-zenoh-flow/Cargo.toml
@@ -16,7 +16,7 @@ readme = "README.md"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-std = { version = "=1.9.0", features = ["attributes"] }
+async-std = { version = "=1.10.0", features = ["attributes"] }
 zenoh = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "master", optional = true}
 zenoh-util = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "master", optional = true }
 zenoh-flow = {path = "../zenoh-flow"}
diff --git a/zenoh-flow-ctl/Cargo.toml b/zenoh-flow-ctl/Cargo.toml
index 5c6403fe..01b8fc3b 100644
--- a/zenoh-flow-ctl/Cargo.toml
+++ b/zenoh-flow-ctl/Cargo.toml
@@ -20,7 +20,7 @@ zenoh = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "master"
 zenoh-util = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "master" }
 zrpc = { git = "https://github.com/atolab/zenoh-rpc.git", branch = "master" }
 znrpc-macros = { git = "https://github.com/atolab/zenoh-rpc.git", branch = "master" }
-async-std = { version = "=1.9.0", features = ["attributes"] }
+async-std = { version = "=1.10.0", features = ["attributes"] }
 structopt = "0.3.13"
 clap = "2.33"
 exitfailure = "0.5.1"
diff --git a/zenoh-flow-daemon/Cargo.toml b/zenoh-flow-daemon/Cargo.toml
index f1831730..11907c7d 100644
--- a/zenoh-flow-daemon/Cargo.toml
+++ b/zenoh-flow-daemon/Cargo.toml
@@ -22,7 +22,7 @@ serde = { version = "1.0", features = ["derive", "rc"] }
 serde_yaml = {version = "0.8"}
 serde_json = "1.0"
 log = "0.4"
-async-std = { version = "=1.9.0", features = ["attributes"] }
+async-std = { version = "=1.10.0", features = ["attributes"] }
 uuid = { version = "0.8", features = ["serde", "v4"] }
 uhlc = "0.4"
 zenoh = { git = "https://github.com/eclipse-zenoh/zenoh.git", branch = "master"}
diff --git a/zenoh-flow-derive/Cargo.toml b/zenoh-flow-derive/Cargo.toml
index ce671ef3..1919ab00 100644
--- a/zenoh-flow-derive/Cargo.toml
+++ b/zenoh-flow-derive/Cargo.toml
@@ -16,7 +16,7 @@ readme = "README.md"
 # To build with debug on macros: RUSTFLAGS="-Z macro-backtrace"
 
 [dependencies]
-async-std = { version = "=1.9.0", features = ["attributes"] }
+async-std = { version = "=1.10.0", features = ["attributes"] }
 futures = "0.3.5"
 syn-serde = { version = "0.2", features = ["json"] }
 syn = { version = "1.0.11", features = ["full"] }
diff --git a/zenoh-flow/Cargo.toml b/zenoh-flow/Cargo.toml
index dcb0139a..540ac251 100644
--- a/zenoh-flow/Cargo.toml
+++ b/zenoh-flow/Cargo.toml
@@ -15,7 +15,8 @@ readme = "README.md"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-std = { version = "=1.9.0", features = ["attributes"] }
+async-std = { version = "=1.10.0", features = ["attributes"] }
+async-lock = "2.4.0"
 async-trait = "0.1.50"
 base64 = "0.13.0"
 bincode = { version = "1.3"}
diff --git a/zenoh-flow/src/runtime/dataflow/instance/runners/mod.rs b/zenoh-flow/src/runtime/dataflow/instance/runners/mod.rs
index 2160cb5d..1aa0c1eb 100644
--- a/zenoh-flow/src/runtime/dataflow/instance/runners/mod.rs
+++ b/zenoh-flow/src/runtime/dataflow/instance/runners/mod.rs
@@ -27,13 +27,13 @@ use crate::runtime::message::Message;
 use crate::runtime::InstanceContext;
 use crate::types::{NodeId, ZFResult};
 use crate::{PortId, PortType, ZFError};
+use async_lock::Barrier;
 use async_trait::async_trait;
 use futures_lite::future::FutureExt;
 use std::collections::HashMap;
 use std::ops::Deref;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use zenoh_sync::Signal;
 
 /// Type of the Runner.
 ///
@@ -50,7 +50,7 @@ pub enum RunnerKind {
 ///
 /// The runner manager is used to send commands to the runner.
 pub struct RunnerManager {
-    stopper: Signal,
+    stopper: Arc<Barrier>,
     handler: JoinHandle<ZFResult<()>>,
     runner: Arc<dyn Runner>,
     ctx: InstanceContext,
@@ -61,7 +61,7 @@ impl RunnerManager {
     ///
     /// It is able to communicate via the `stopper` and the `handler`.
     pub fn new(
-        stopper: Signal,
+        stopper: Arc<Barrier>,
         handler: JoinHandle<ZFResult<()>>,
         runner: Arc<dyn Runner>,
         ctx: InstanceContext,
@@ -82,7 +82,8 @@ impl RunnerManager {
         if self.runner.is_recording().await {
             self.runner.stop_recording().await?;
         }
-        self.stopper.trigger();
+        log::info!("RunnerManager triggering stop to {}", self.get_id());
+        self.stopper.wait().await;
         Ok(())
     }
 
@@ -235,50 +236,61 @@ impl NodeRunner {
     ///
     ///  # Errors
     /// An error variant is returned in case the run returns an error.
-    fn run_stoppable(&self, signal: Signal) -> ZFResult<()> {
-        async fn run(runner: &NodeRunner) -> RunAction {
-            match runner.run().await {
-                Ok(_) => RunAction::Stop,
-                Err(e) => RunAction::RestartRun(Some(e)),
-            }
-        }
-
-        async fn stop(signal: Signal) -> RunAction {
-            signal.wait().await;
-            RunAction::Stop
-        }
-        async_std::task::block_on(async move {
-            loop {
-                let cloned_signal = signal.clone();
-                match stop(cloned_signal).race(run(self)).await {
-                    RunAction::RestartRun(e) => {
-                        log::error!(
-                            "[Node: {}] The run loop exited with {:?}, restarting…",
-                            self.get_id(),
-                            e
-                        );
-                    }
-                    RunAction::Stop => {
-                        log::trace!(
-                            "[Node: {}] Received kill command, killing runner",
-                            self.get_id()
-                        );
-                        self.stop().await;
-                        return Ok(());
-                    }
+    async fn run_stoppable(&self, signal: Arc<Barrier>) -> ZFResult<()> {
+        loop {
+            log::info!("NodeRunner {} starting run!", self.get_id());
+
+            let my_id = self.get_id();
+            let future_stop = async {
+                signal.wait().await;
+                log::info!("NodeRunner {} received stop trigger", my_id);
+                RunAction::Stop
+            };
+
+            let future_run = async {
+                log::info!("NodeRunner {} running", self.get_id());
+                match self.run().await {
+                    Ok(_) => RunAction::Stop,
+                    Err(e) => RunAction::RestartRun(Some(e)),
+                }
+            };
+
+            match future_stop.race(future_run).await {
+                RunAction::RestartRun(e) => {
+                    log::error!(
+                        "[Node: {}] The run loop exited with {:?}, restarting…",
+                        self.get_id(),
+                        e
+                    );
+                }
+                RunAction::Stop => {
+                    log::trace!(
+                        "[Node: {}] Received kill command, killing runner",
+                        self.get_id()
+                    );
+                    self.stop().await;
+                    return Ok(());
                 }
             }
-        })
+        }
     }
 
     /// Starts the node, returning the `RunnerManager` to stop it.
     pub fn start(&self) -> RunnerManager {
-        let signal = Signal::new();
+        // Using barrier to wait for the stop notification.
+        // A Barrier(N) block N-1 futures that call the wait().
+        // In this case the future we want to block is the one defined in
+        // line 244. That future will block when calling the wait().
+        // When the next future, the one defined in line 86, calls the wait()
+        // both gets unlocked and continue execution, winning the race
+        // in line 258 and stop the execution.
+        // This is why this is a Barrier(2).
+        let signal = Arc::new(Barrier::new(2));
+
         let cloned_self = self.clone();
         let cloned_signal = signal.clone();
-
-        let h = async_std::task::spawn_blocking(move || cloned_self.run_stoppable(cloned_signal));
-
+        let h =
+            async_std::task::spawn(async move { cloned_self.run_stoppable(cloned_signal).await });
         RunnerManager::new(signal, h, self.inner.clone(), self.ctx.clone())
     }
 }


Test patch: