zenoh_flow/runtime/dataflow/instance/runners/
mod.rs

1//
2// Copyright (c) 2021 - 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14
15pub mod connector;
16
17use crate::traits::Node;
18use crate::zfresult::Error;
19use crate::Result as ZFResult;
20use async_std::task::JoinHandle;
21use futures::future::{AbortHandle, Abortable, Aborted};
22use std::sync::Arc;
23use std::time::Instant;
24
25/// Type of the Runner.
26///
27/// The runner is the one actually running the nodes.
28#[derive(Clone, Debug, PartialEq, Eq)]
29pub enum RunnerKind {
30    Source,
31    Operator,
32    Sink,
33    Connector,
34}
35
36/// Action to be taken depending on the result of the run.
37pub enum RunAction {
38    RestartRun(Option<Error>),
39    Stop,
40}
41
42/// A `Runner` takes care of running a `Node`.
43///
44/// It spawns an abortable task in which the `iteration` is called in a loop, indefinitely.
45pub(crate) struct Runner {
46    pub(crate) node: Arc<dyn Node>,
47    pub(crate) run_loop_handle: Option<JoinHandle<Result<Error, Aborted>>>,
48    pub(crate) run_loop_abort_handle: Option<AbortHandle>,
49}
50
51impl Runner {
52    pub(crate) fn new(node: Arc<dyn Node>) -> Self {
53        Self {
54            node,
55            run_loop_handle: None,
56            run_loop_abort_handle: None,
57        }
58    }
59
60    /// Start the `Runner`, spawning an abortable task.
61    ///
62    /// `start` is idempotent and will do nothing if the node is already running.
63    pub(crate) fn start(&mut self) {
64        if self.is_running() {
65            log::warn!("Called `start` while node is ALREADY running. Returning.");
66            return;
67        }
68
69        let node = self.node.clone();
70        let run_loop = async move {
71            let mut instant: Instant;
72            loop {
73                instant = Instant::now();
74                log::trace!("Iteration start: {:?}", instant);
75                if let Err(e) = node.iteration().await {
76                    log::error!("Iteration error: {:?}", e);
77                    return e;
78                }
79
80                log::trace!("iteration took: {}ms", instant.elapsed().as_millis());
81
82                async_std::task::yield_now().await;
83            }
84        };
85
86        let (abort_handle, abort_registration) = AbortHandle::new_pair();
87        let handle = async_std::task::spawn(Abortable::new(run_loop, abort_registration));
88
89        self.run_loop_handle = Some(handle);
90        self.run_loop_abort_handle = Some(abort_handle);
91    }
92
93    /// Stop the execution of a `Node`.
94    ///
95    /// We will call `abort` on the `AbortHandle` and then `await` the `JoinHandle`. As per its
96    /// documentation, `abort` will not forcefully interrupt an execution if the corresponding task
97    /// is being polled on another thread.
98    ///
99    /// `stop` is idempotent and will do nothing if the node is not running.
100    pub(crate) async fn stop(&mut self) -> ZFResult<()> {
101        if !self.is_running() {
102            log::warn!("Called `stop` while node is NOT running. Returning.");
103            return Ok(()); // TODO Return an error instead?
104        }
105
106        if let Some(abort_handle) = self.run_loop_abort_handle.take() {
107            abort_handle.abort();
108            if let Some(handle) = self.run_loop_handle.take() {
109                log::trace!("Handler finished with {:?}", handle.await);
110            }
111        }
112
113        Ok(())
114    }
115
116    /// Tell if the node is running.
117    ///
118    /// To do so we check if an `AbortHandle` was set. If so, then a task was spawned and the node
119    /// is indeed running.
120    pub(crate) fn is_running(&self) -> bool {
121        self.run_loop_handle.is_some()
122    }
123}
zenoh_flow/runtime/dataflow/instance/runners/mod.rs

zenoh_flow/runtime/dataflow/instance/runners/
mod.rs
