ntex_rt/
lib.rs

1//! Single-threaded asynchronous runtime infrastructure for ntex.
2//!
3//! A [`System`] owns the runtime configuration and coordinates one or more
4//! [`Arbiter`] execution threads. Each arbiter runs local futures on its own
5//! thread, allowing tasks to use types that are not `Send`.
6//!
7//! Use [`System::build()`] to configure a system, [`spawn()`] to start a local
8//! task, and [`spawn_blocking()`] for blocking work.
9//!
10//! Local [`JoinHandle`]s are neither `Send` nor `Sync`, even for `Send` outputs:
11//!
12//! ```compile_fail
13//! fn send<T: Send>() {}
14//! send::<ntex_rt::JoinHandle<usize>>();
15//! ```
16//!
17//! ```compile_fail
18//! fn sync<T: Sync>() {}
19//! sync::<ntex_rt::JoinHandle<usize>>();
20//! ```
21//!
22//! [`Handle::spawn()`] submits a `Send` future to the handle's runtime and
23//! returns a [`RemoteJoinHandle`] for the calling thread. On the runtime's own thread
24//! the task is spawned directly. From other threads the future and later its
25//! `Send` output share one allocation, which is passed to the runtime thread:
26//!
27//! ```no_run
28//! # async fn example(handle: ntex_rt::Handle) {
29//! let task = handle.spawn(async { 42 });
30//! assert_eq!(task.await.unwrap(), 42);
31//! # }
32//! ```
33//!
34//! Dropping the handle detaches the task, [`RemoteJoinHandle::cancel()`] cancels it.
35//! A submitted future is polled and dropped on the runtime thread; shutdown or
36//! cancellation before producing a result yields [`JoinError`]. No task is
37//! guaranteed to finish. Rejected submissions are never polled and their
38//! futures may be dropped on the submitting thread.
39//!
40//! Use [`Handle::spawn_detached()`] for fire-and-forget work. Its output need
41//! not be `Send` and is discarded on the owner.
42//!
43//! Remote join handles also remain on their calling thread, even when the
44//! submission uses the runtime's local fast path:
45//!
46//! ```compile_fail
47//! fn send<T: Send>() {}
48//! send::<ntex_rt::RemoteJoinHandle<usize>>();
49//! ```
50//!
51//! ```compile_fail
52//! fn sync<T: Sync>() {}
53//! sync::<ntex_rt::RemoteJoinHandle<usize>>();
54//! ```
55//!
56//! ```compile_fail
57//! # fn example(handle: ntex_rt::Handle) {
58//! handle.spawn(async { std::rc::Rc::new(42) });
59//! # }
60//! ```
61//!
62//! [`LocalWaker`] owns a thread-confined wake handle. It specializes native
63//! task wakes on their owner thread and preserves standard waker behavior
64//! for other runtimes and wrapped wakers.
65//!
66//! Runtime backends are selected with Cargo features:
67//!
68//! - no runtime feature uses the native ntex runtime
69//! - `tokio` uses the Tokio-based runtime adapter
70//! - `compio` uses the Compio-based runtime adapter
71#![deny(clippy::pedantic)]
72#![allow(
73    clippy::missing_errors_doc,
74    clippy::missing_fields_in_debug,
75    clippy::must_use_candidate
76)]
77
78mod arbiter;
79mod builder;
80mod busy;
81#[cfg(target_os = "linux")]
82mod capture;
83mod driver;
84mod handle;
85mod local_waker;
86mod metrics;
87mod pool;
88mod remote;
89pub mod signals;
90mod system;
91mod task;
92#[cfg(test)]
93mod testing;
94mod watchdog;
95
96mod rt;
97pub mod rt_default;
98
99#[cfg(feature = "compio")]
100pub mod rt_compio;
101#[cfg(feature = "tokio")]
102pub mod rt_tokio;
103
104pub use self::arbiter::{Arbiter, get_item, remove_all_items, set_item, with_item};
105pub use self::builder::{Builder, SystemRunner};
106pub use self::driver::{BlockFuture, Driver, DriverType, Notify, PollResult, Runner};
107pub use self::local_waker::LocalWaker;
108pub use self::metrics::{HISTOGRAM_BUCKETS, Histogram, LatencyMetrics};
109pub use self::pool::{BlockingError, BlockingResult, PoolMetrics, ThreadPool, spawn_blocking};
110pub use self::rt::{Runtime, RuntimeBuilder};
111pub use self::system::{Id, PingRecord, System};
112pub use self::task::{task_callbacks, task_opt_callbacks};
113
114#[cfg(feature = "tokio")]
115pub use self::rt_tokio::*;
116
117#[cfg(all(feature = "compio", not(feature = "tokio")))]
118pub use self::rt_compio::*;
119
120#[cfg(all(not(feature = "tokio"), not(feature = "compio")))]
121pub use self::rt_default::*;
122
123pub(crate) type HashMap<K, V> = std::collections::HashMap<K, V, foldhash::fast::RandomState>;
124
125/// Runs `f` while an earlier panic is pending, the process aborts if `f`
126/// panics.
127pub(crate) fn abort_on_panic<R>(f: impl FnOnce() -> R) -> R {
128    struct Abort;
129
130    impl Drop for Abort {
131        fn drop(&mut self) {
132            abort_on_second_panic();
133        }
134    }
135
136    let guard = Abort;
137    let result = f();
138    std::mem::forget(guard);
139    result
140}
141
142/// Aborts the process on a panic raised while an earlier panic is unwinding
143/// or held to be resumed.
144#[cold]
145pub(crate) fn abort_on_second_panic() -> ! {
146    if let Err(payload) = std::panic::catch_unwind(|| {
147        log::error!("Panic during shutdown after an earlier panic; aborting");
148    }) {
149        std::mem::forget(payload);
150    }
151    std::process::abort();
152}
ntex_rt/lib.rs

ntex_rt/
lib.rs
