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

ntex_rt/
driver.rs
