zenoh_runtime/
lib.rs

1//
2// Copyright (c) 2024 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
15//! ⚠️ WARNING ⚠️
16//!
17//! This crate is intended for Zenoh's internal use.
18//!
19//! [Click here for Zenoh's documentation](https://docs.rs/zenoh/latest/zenoh)
20use core::panic;
21use std::{
22    borrow::Borrow,
23    collections::HashMap,
24    env, fmt,
25    future::Future,
26    ops::Deref,
27    sync::{
28        atomic::{AtomicUsize, Ordering},
29        OnceLock,
30    },
31    time::Duration,
32};
33
34use lazy_static::lazy_static;
35use serde::Deserialize;
36use tokio::{
37    runtime::{Handle, Runtime, RuntimeFlavor},
38    task::JoinHandle,
39};
40use zenoh_macros::{GenericRuntimeParam, RegisterParam};
41use zenoh_result::ZResult as Result;
42
43pub const ZENOH_RUNTIME_ENV: &str = "ZENOH_RUNTIME";
44
45/// Available parameters to configure the ZRuntime.
46#[derive(Deserialize, Debug, GenericRuntimeParam)]
47#[serde(deny_unknown_fields, default)]
48pub struct RuntimeParam {
49    /// Number of async worker threads. At least one.
50    pub worker_threads: usize,
51    /// Number of maximal worker threads for blocking tasks. At least one.
52    pub max_blocking_threads: usize,
53    /// Hand over one ZRuntime to another one.
54    pub handover: Option<ZRuntime>,
55}
56
57impl Default for RuntimeParam {
58    fn default() -> Self {
59        Self {
60            worker_threads: 1,
61            max_blocking_threads: 50,
62            handover: None,
63        }
64    }
65}
66
67impl RuntimeParam {
68    pub fn build(&self, zrt: ZRuntime) -> Result<Runtime> {
69        let rt = tokio::runtime::Builder::new_multi_thread()
70            .worker_threads(self.worker_threads)
71            .max_blocking_threads(self.max_blocking_threads)
72            .enable_io()
73            .enable_time()
74            .thread_name_fn(move || {
75                let id = ZRUNTIME_INDEX
76                    .get(&zrt)
77                    .unwrap()
78                    .fetch_add(1, Ordering::SeqCst);
79                format!("{zrt}-{id}")
80            })
81            .build()?;
82        Ok(rt)
83    }
84}
85
86/// [`ZRuntime`], the access point of manipulate runtimes within zenoh.
87/// The runtime parameter can be configured by setting the environmental variable [`ZENOH_RUNTIME_ENV`].
88/// The parsing syntax use [RON](https://github.com/ron-rs/ron). An example configuration looks
89/// like
90///
91/// ```console
92/// ZENOH_RUNTIME='(
93///   rx: (handover: app),
94///   acc: (handover: app),
95///   app: (worker_threads: 2),
96///   tx: (max_blocking_threads: 1)
97/// )'
98/// ```
99/// Note: The runtime parameter takes effect at the beginning of the zenoh process and no longer be
100/// changed after the initialization.
101#[derive(Hash, Eq, PartialEq, Clone, Copy, Debug, RegisterParam, Deserialize)]
102#[param(RuntimeParam)]
103pub enum ZRuntime {
104    /// Renamed to app. Default param: worker_threads = 1.
105    #[serde(rename = "app")]
106    #[param(worker_threads = 1)]
107    Application,
108
109    /// Renamed to acc. Default param: worker_threads = 1.
110    #[serde(rename = "acc")]
111    #[param(worker_threads = 1)]
112    Acceptor,
113
114    /// Renamed to tx. Default param: worker_threads = 1.
115    #[serde(rename = "tx")]
116    #[param(worker_threads = 1)]
117    TX,
118
119    /// Renamed to rx. Default param: worker_threads = 2.
120    #[serde(rename = "rx")]
121    #[param(worker_threads = 2)]
122    RX,
123
124    /// Renamed to net. Default param: worker_threads = 1.
125    #[serde(rename = "net")]
126    #[param(worker_threads = 1)]
127    Net,
128}
129
130impl ZRuntime {
131    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
132    where
133        F: Future + Send + 'static,
134        F::Output: Send + 'static,
135    {
136        #[cfg(feature = "tracing-instrument")]
137        let future = tracing::Instrument::instrument(future, tracing::Span::current());
138
139        self.deref().spawn(future)
140    }
141
142    pub fn block_in_place<F, R>(&self, f: F) -> R
143    where
144        F: Future<Output = R>,
145    {
146        match Handle::try_current() {
147            Ok(handle) => {
148                if handle.runtime_flavor() == RuntimeFlavor::CurrentThread {
149                    panic!("Zenoh runtime doesn't support Tokio's current thread scheduler. Please use multi thread scheduler instead, e.g. a multi thread scheduler with one worker thread: `#[tokio::main(flavor = \"multi_thread\", worker_threads = 1)]`");
150                }
151            }
152            Err(e) => {
153                if e.is_thread_local_destroyed() {
154                    panic!("The Thread Local Storage inside Tokio is destroyed. This might happen when Zenoh API is called at process exit, e.g. in the atexit handler. Calling the Zenoh API at process exit is not supported and should be avoided.");
155                }
156            }
157        }
158
159        #[cfg(feature = "tracing-instrument")]
160        let f = tracing::Instrument::instrument(f, tracing::Span::current());
161
162        tokio::task::block_in_place(move || self.block_on(f))
163    }
164}
165
166impl Deref for ZRuntime {
167    type Target = Handle;
168    fn deref(&self) -> &Self::Target {
169        ZRUNTIME_POOL.get(self)
170    }
171}
172
173lazy_static! {
174    pub static ref ZRUNTIME_POOL: ZRuntimePool = ZRuntimePool::new();
175    pub static ref ZRUNTIME_INDEX: HashMap<ZRuntime, AtomicUsize> = ZRuntime::iter()
176        .map(|zrt| (zrt, AtomicUsize::new(0)))
177        .collect();
178}
179
180// A runtime guard used to explicitly drop the static variables that Rust doesn't drop by default
181#[derive(Debug)]
182pub struct ZRuntimePoolGuard;
183
184impl Drop for ZRuntimePoolGuard {
185    fn drop(&mut self) {
186        unsafe {
187            std::mem::drop((ZRUNTIME_POOL.deref() as *const ZRuntimePool).read());
188            std::mem::drop(
189                (ZRUNTIME_INDEX.deref() as *const HashMap<ZRuntime, AtomicUsize>).read(),
190            );
191        }
192    }
193}
194
195pub struct ZRuntimePool(HashMap<ZRuntime, OnceLock<Runtime>>);
196
197impl fmt::Debug for ZRuntimePool {
198    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
199        let initialized = self
200            .0
201            .iter()
202            .filter_map(|(runtime, cell)| cell.get().map(|_| runtime))
203            .collect::<Vec<_>>();
204        f.debug_struct("ZRuntimePool")
205            .field("initialized", &initialized)
206            .finish_non_exhaustive()
207    }
208}
209
210impl ZRuntimePool {
211    fn new() -> Self {
212        Self(ZRuntime::iter().map(|zrt| (zrt, OnceLock::new())).collect())
213    }
214
215    pub fn get(&self, zrt: &ZRuntime) -> &Handle {
216        // Although the ZRuntime is called to use `zrt`, it may be handed over to another one
217        // specified via the environmental variable.
218        let param: &RuntimeParam = zrt.borrow();
219        let zrt = match param.handover {
220            Some(handover) => handover,
221            None => *zrt,
222        };
223
224        self.0
225            .get(&zrt)
226            .unwrap_or_else(|| panic!("The hashmap should contains {zrt} after initialization"))
227            .get_or_init(|| {
228                zrt.init()
229                    .unwrap_or_else(|_| panic!("Failed to init {zrt}"))
230            })
231            .handle()
232    }
233}
234
235// If there are any blocking tasks spawned by ZRuntimes, the function will block until they return.
236impl Drop for ZRuntimePool {
237    fn drop(&mut self) {
238        let handles: Vec<_> = self
239            .0
240            .drain()
241            .filter_map(|(_name, mut rt)| {
242                rt.take()
243                    .map(|r| std::thread::spawn(move || r.shutdown_timeout(Duration::from_secs(1))))
244            })
245            .collect();
246
247        for hd in handles {
248            let _ = hd.join();
249        }
250    }
251}
252
253#[should_panic(expected = "Zenoh runtime doesn't support")]
254#[tokio::test]
255async fn block_in_place_fail_test() {
256    use crate::ZRuntime;
257    ZRuntime::TX.block_in_place(async { println!("Done") });
258}
zenoh_runtime/lib.rs

zenoh_runtime/
lib.rs
