ntex_rt/
driver.rs

1//! The platform-specified driver.
2use std::{any::Any, cell::RefCell, fmt, io, panic, pin::Pin, rc::Rc};
3
4use crate::rt::Runtime;
5
6pub type BlockFuture = Pin<Box<dyn Future<Output = ()>>>;
7
8#[derive(Copy, Clone, Debug, PartialEq, Eq)]
9pub enum DriverType {
10    Poll,
11    IoUring,
12    Iocp,
13}
14
15impl DriverType {
16    pub const fn name(&self) -> &'static str {
17        match self {
18            DriverType::Poll => "polling",
19            DriverType::IoUring => "io-uring",
20            DriverType::Iocp => "iocp",
21        }
22    }
23
24    pub const fn is_polling(&self) -> bool {
25        matches!(self, &DriverType::Poll)
26    }
27}
28
29pub trait Runner: Send + Sync + 'static {
30    /// Runs the future to completion.
31    ///
32    /// A panic of the event loop may be returned as an error or may unwind,
33    /// both are handled the same way.
34    fn block_on(&self, fut: BlockFuture) -> Result<(), Box<dyn Any + Send>>;
35}
36
37pub trait Driver: 'static {
38    fn handle(&self) -> Box<dyn Notify>;
39
40    fn run(&self, rt: &Runtime) -> io::Result<()>;
41
42    fn clear(&self) {}
43}
44
45#[derive(Copy, Clone, Debug, PartialEq, Eq)]
46pub enum PollResult {
47    Ready,
48    Pending,
49    PollAgain,
50}
51
52/// Wakes a runtime driver, retaining the wakeup if it has not blocked yet.
53///
54/// Internal scheduler notifications retry [`io::ErrorKind::Interrupted`] and
55/// log and abort the process on any other error or if `notify` panics, since
56/// queued work may otherwise remain stranded. Explicit notifications through
57/// [`Runtime::handle`] return errors and propagate panics to their caller.
58pub trait Notify: Send + Sync + fmt::Debug + 'static {
59    fn notify(&self) -> io::Result<()>;
60}
61
62#[cfg(windows)]
63#[macro_export]
64macro_rules! syscall {
65    (BOOL, $e:expr) => {
66        $crate::syscall!($e, == 0)
67    };
68    (SOCKET, $e:expr) => {
69        $crate::syscall!($e, != 0)
70    };
71    (HANDLE, $e:expr) => {
72        $crate::syscall!($e, == ::windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE)
73    };
74    ($e:expr, $op: tt $rhs: expr) => {{
75        #[allow(unused_unsafe)]
76        let res = unsafe { $e };
77        if res $op $rhs {
78            Err(::std::io::Error::last_os_error())
79        } else {
80            Ok(res)
81        }
82    }};
83}
84
85/// Helper macro to execute a system call
86#[cfg(unix)]
87#[macro_export]
88macro_rules! syscall {
89    (break $e:expr) => {
90        loop {
91            match $crate::syscall!($e) {
92                Ok(fd) => break ::std::task::Poll::Ready(Ok(fd as usize)),
93                Err(e) if e.kind() == ::std::io::ErrorKind::WouldBlock || e.raw_os_error() == Some(::libc::EINPROGRESS)
94                    => break ::std::task::Poll::Pending,
95                Err(e) if e.kind() == ::std::io::ErrorKind::Interrupted => {},
96                Err(e) => break ::std::task::Poll::Ready(Err(e)),
97            }
98        }
99    };
100    ($e:expr, $f:ident($fd:expr)) => {
101        match $crate::syscall!(break $e) {
102            ::std::task::Poll::Pending => Ok($crate::sys::Decision::$f($fd)),
103            ::std::task::Poll::Ready(Ok(res)) => Ok($crate::sys::Decision::Completed(res)),
104            ::std::task::Poll::Ready(Err(e)) => Err(e),
105        }
106    };
107    ($e:expr) => {{
108        #[allow(unused_unsafe)]
109        let res = unsafe { $e };
110        if res == -1 {
111            Err(::std::io::Error::last_os_error())
112        } else {
113            Ok(res)
114        }
115    }};
116}
117
118/// Execute a future with custom `block_on` method and wait for result
119pub(super) fn block_on<F, R>(run: &dyn Runner, fut: F) -> Result<R, Box<dyn Any + Send>>
120where
121    F: Future<Output = R> + 'static,
122    R: 'static,
123{
124    // run loop
125    let result = Rc::new(RefCell::new(None));
126    let result_inner = result.clone();
127
128    ntex_error::set_backtrace_start_alt("ntex/ntex-rt/src/driver.rs", 0);
129    // finalizes the thread's arbiter once its event loop has exited,
130    // including when the runner returns an error or a shutdown callback panics
131    let finalize = FinalizeArbiter;
132    let fut = Box::pin(async move {
133        let r = fut.await;
134        *result_inner.borrow_mut() = Some(r);
135    });
136    // a panic escaping the runner is returned like a runner error, so the
137    // caller cleans up the same way for unwinding and catching runners
138    let res =
139        panic::catch_unwind(panic::AssertUnwindSafe(|| run.block_on(fut))).unwrap_or_else(Err);
140    if let Err(e) = res {
141        // the event loop panicked, a panic of `on_stop` aborts the process
142        crate::abort_on_panic(|| drop(finalize));
143        return Err(e);
144    }
145
146    crate::arbiter::run_shutdown_callbacks();
147    unsafe {
148        crate::remove_all_items();
149    }
150    Ok(result.borrow_mut().take().unwrap())
151}
152
153struct FinalizeArbiter;
154
155impl Drop for FinalizeArbiter {
156    fn drop(&mut self) {
157        if let Some(arb) = crate::Arbiter::try_current() {
158            arb.finalize();
159        }
160    }
161}
162
163pub(super) fn block_on_panic<F, R>(run: &dyn Runner, fut: F) -> R
164where
165    F: Future<Output = R> + 'static,
166    R: 'static,
167{
168    match block_on(run, fut) {
169        Ok(v) => v,
170        Err(e) => panic::resume_unwind(e),
171    }
172}
173
174#[cfg(test)]
175mod tests {
176    use super::DriverType;
177
178    #[test]
179    fn driver_type() {
180        assert_eq!(DriverType::Poll.name(), "polling");
181        assert_eq!(DriverType::IoUring.name(), "io-uring");
182        assert_eq!(DriverType::Iocp.name(), "iocp");
183        assert!(DriverType::Poll.is_polling());
184        assert!(!DriverType::IoUring.is_polling());
185        assert!(!DriverType::Iocp.is_polling());
186    }
187}
ntex_rt/driver.rs

ntex_rt/
driver.rs
