pal_async/
driver.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! Driver trait.
5
6// UNSAFETY: Needed to define and implement the unsafe new_dyn_overlapped_file method.
7#![cfg_attr(any(windows, target_os = "linux"), expect(unsafe_code))]
8
9#[cfg(unix)]
10use crate::fd::FdReadyDriver;
11#[cfg(unix)]
12use crate::fd::PollFdReady;
13#[cfg(target_os = "linux")]
14use crate::io_uring::IoUringDriver;
15#[cfg(target_os = "macos")]
16use crate::process::macos::PollProcessWait;
17#[cfg(target_os = "macos")]
18use crate::process::macos::ProcessWaitDriver;
19use crate::socket::PollSocketReady;
20use crate::socket::SocketReadyDriver;
21#[cfg(windows)]
22use crate::sys::overlapped::IoOverlapped;
23#[cfg(windows)]
24use crate::sys::overlapped::OverlappedIoDriver;
25use crate::task::Spawn;
26use crate::timer::PollTimer;
27use crate::timer::TimerDriver;
28use crate::wait::PollWait;
29use crate::wait::WaitDriver;
30use smallbox::SmallBox;
31use smallbox::space::S4;
32use std::io;
33#[cfg(unix)]
34use std::os::unix::prelude::*;
35#[cfg(windows)]
36use std::os::windows::prelude::*;
37use std::sync::Arc;
38
39/// A generic `Box`-like container of one of the polled types.
40pub type PollImpl<T> = SmallBox<T, S4>;
41
42/// A driver that supports polled IO.
43pub trait Driver: 'static + Send + Sync {
44    /// Returns a new timer.
45    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer>;
46
47    /// Returns a new object for polling file descriptor readiness.
48    #[cfg(unix)]
49    fn new_dyn_fd_ready(&self, fd: RawFd) -> io::Result<PollImpl<dyn PollFdReady>>;
50
51    /// Creates a new object for polling socket readiness.
52    #[cfg(windows)]
53    fn new_dyn_socket_ready(&self, socket: RawSocket) -> io::Result<PollImpl<dyn PollSocketReady>>;
54
55    /// Creates a new object for polling socket readiness.
56    #[cfg(unix)]
57    fn new_dyn_socket_ready(&self, socket: RawFd) -> io::Result<PollImpl<dyn PollSocketReady>>;
58
59    /// Creates a new wait.
60    #[cfg(windows)]
61    fn new_dyn_wait(&self, handle: RawHandle) -> io::Result<PollImpl<dyn PollWait>>;
62
63    /// Creates a new wait.
64    ///
65    /// Signals will be consumed using reads of `read_size` bytes, with 8-byte
66    /// buffer alignment. `read_size` must be at most
67    /// [`MAXIMUM_WAIT_READ_SIZE`](super::wait::MAXIMUM_WAIT_READ_SIZE) bytes.
68    #[cfg(unix)]
69    fn new_dyn_wait(&self, fd: RawFd, read_size: usize) -> io::Result<PollImpl<dyn PollWait>>;
70
71    /// Creates a new process wait from a process ID.
72    #[cfg(target_os = "macos")]
73    fn new_dyn_process_wait(&self, pid: i32) -> io::Result<PollImpl<dyn PollProcessWait>>;
74
75    /// Creates a new overlapped file handler.
76    ///
77    /// # Safety
78    /// The caller must ensure that they exclusively own `handle`, and that
79    /// `handle` stays alive until the new handler is dropped.
80    #[cfg(windows)]
81    unsafe fn new_dyn_overlapped_file(
82        &self,
83        handle: RawHandle,
84    ) -> io::Result<PollImpl<dyn IoOverlapped>>;
85
86    /// Returns whether the given opcode is supported by the ring.
87    #[cfg(target_os = "linux")]
88    fn io_uring_probe(&self, opcode: u8) -> bool;
89
90    /// Submits an io-uring SQE for asynchronous execution.
91    ///
92    /// Returns a future that completes with the IO result. The future **aborts
93    /// the process** if dropped while the IO is in flight, since there is no
94    /// way to synchronously cancel an in-flight io-uring operation.
95    ///
96    /// # Safety
97    ///
98    /// All memory referenced by the SQE must remain valid for the lifetime of
99    /// the returned future.
100    ///
101    /// This can be hard to do safely; in particular, if this future can be
102    /// leaked (via [`std::mem::forget`] or otherwise) then the caller must
103    /// ensure that any referenced memory also leaks. The easiest way to do that
104    /// is to ensure that the future is `await`ed in an async function or block
105    /// that owns the underlying memory. So, this is safe:
106    ///
107    /// ```rust,ignore
108    /// async fn write(driver: &impl Driver, file: &File, buf: Vec<u8>) -> io::Result<usize> {
109    ///     let sqe = opcode::Write::new(
110    ///         types::Fd(file.as_raw_fd()), buf.as_ptr(), buf.len() as u32,
111    ///     ).build();
112    ///     // SAFETY: `buf` is owned by this async function's state machine.
113    ///     // If the outer future is leaked, `buf` leaks with it, so the
114    ///     // memory remains valid for the io-uring operation.
115    ///     unsafe { driver.io_uring_submit(sqe).await? };
116    ///     Ok(buf.len())
117    /// }
118    /// ```
119    ///
120    /// But this is not:
121    ///
122    /// ```rust,ignore
123    /// async fn write(driver: &impl Driver, file: &File, buf: &[u8]) -> io::Result<usize> {
124    ///     let sqe = opcode::Write::new(
125    ///         types::Fd(file.as_raw_fd()), buf.as_ptr(), buf.len() as u32,
126    ///     ).build();
127    ///     // NOT SAFE: `buf` is a borrow. If the outer future is leaked,
128    ///     // the referent can be freed while the io-uring operation is
129    ///     // still in flight.
130    ///     unsafe { driver.io_uring_submit(sqe).await? };
131    ///     Ok(buf.len())
132    /// }
133    /// ```
134    #[cfg(target_os = "linux")]
135    unsafe fn io_uring_submit(
136        &self,
137        sqe: crate::io_uring::Entry,
138    ) -> std::pin::Pin<Box<dyn Future<Output = io::Result<i32>> + Send + '_>>;
139}
140
141#[cfg(target_os = "macos")]
142impl<T> Driver for T
143where
144    T: 'static
145        + Send
146        + Sync
147        + FdReadyDriver
148        + TimerDriver
149        + SocketReadyDriver
150        + WaitDriver
151        + ProcessWaitDriver,
152{
153    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
154        smallbox::smallbox!(self.new_timer())
155    }
156
157    fn new_dyn_fd_ready(&self, fd: RawFd) -> io::Result<PollImpl<dyn PollFdReady>> {
158        Ok(smallbox::smallbox!(self.new_fd_ready(fd)?))
159    }
160
161    fn new_dyn_socket_ready(&self, socket: RawFd) -> io::Result<PollImpl<dyn PollSocketReady>> {
162        Ok(smallbox::smallbox!(self.new_socket_ready(socket)?))
163    }
164
165    fn new_dyn_wait(&self, fd: RawFd, read_size: usize) -> io::Result<PollImpl<dyn PollWait>> {
166        Ok(smallbox::smallbox!(self.new_wait(fd, read_size)?))
167    }
168
169    fn new_dyn_process_wait(&self, pid: i32) -> io::Result<PollImpl<dyn PollProcessWait>> {
170        Ok(smallbox::smallbox!(self.new_process_wait_pid(pid)?))
171    }
172}
173
174#[cfg(target_os = "linux")]
175impl<T> Driver for T
176where
177    T: 'static
178        + Send
179        + Sync
180        + FdReadyDriver
181        + TimerDriver
182        + SocketReadyDriver
183        + WaitDriver
184        + IoUringDriver,
185{
186    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
187        smallbox::smallbox!(self.new_timer())
188    }
189
190    fn new_dyn_fd_ready(&self, fd: RawFd) -> io::Result<PollImpl<dyn PollFdReady>> {
191        Ok(smallbox::smallbox!(self.new_fd_ready(fd)?))
192    }
193
194    fn new_dyn_socket_ready(&self, socket: RawFd) -> io::Result<PollImpl<dyn PollSocketReady>> {
195        Ok(smallbox::smallbox!(self.new_socket_ready(socket)?))
196    }
197
198    fn new_dyn_wait(&self, fd: RawFd, read_size: usize) -> io::Result<PollImpl<dyn PollWait>> {
199        Ok(smallbox::smallbox!(self.new_wait(fd, read_size)?))
200    }
201
202    fn io_uring_probe(&self, opcode: u8) -> bool {
203        use crate::io_uring::IoUringSubmit as _;
204
205        self.io_uring_submitter()
206            .is_some_and(|submitter| submitter.probe(opcode))
207    }
208
209    unsafe fn io_uring_submit(
210        &self,
211        sqe: crate::io_uring::Entry,
212    ) -> std::pin::Pin<Box<dyn Future<Output = io::Result<i32>> + Send + '_>> {
213        use crate::io_uring::IoUringSubmit as _;
214
215        Box::pin(async move {
216            // SAFETY: caller guarantees contract
217            unsafe {
218                self.io_uring_submitter()
219                    .ok_or(io::ErrorKind::Unsupported)?
220                    .submit(sqe)
221            }
222            .await
223        })
224    }
225}
226
227#[cfg(windows)]
228impl<T> Driver for T
229where
230    T: 'static + Send + Sync + TimerDriver + SocketReadyDriver + WaitDriver + OverlappedIoDriver,
231{
232    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
233        smallbox::smallbox!(self.new_timer())
234    }
235
236    fn new_dyn_socket_ready(&self, socket: RawSocket) -> io::Result<PollImpl<dyn PollSocketReady>> {
237        Ok(smallbox::smallbox!(self.new_socket_ready(socket)?))
238    }
239
240    fn new_dyn_wait(&self, handle: RawHandle) -> io::Result<PollImpl<dyn PollWait>> {
241        Ok(smallbox::smallbox!(self.new_wait(handle)?))
242    }
243
244    unsafe fn new_dyn_overlapped_file(
245        &self,
246        handle: RawHandle,
247    ) -> io::Result<PollImpl<dyn IoOverlapped>> {
248        // SAFETY: caller guarantees contract
249        Ok(smallbox::smallbox!(unsafe {
250            self.new_overlapped_file(handle)
251        }?))
252    }
253}
254
255#[cfg(unix)]
256impl Driver for Box<dyn Driver> {
257    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
258        self.as_ref().new_dyn_timer()
259    }
260
261    fn new_dyn_fd_ready(&self, fd: RawFd) -> io::Result<PollImpl<dyn PollFdReady>> {
262        self.as_ref().new_dyn_fd_ready(fd)
263    }
264
265    fn new_dyn_socket_ready(&self, socket: RawFd) -> io::Result<PollImpl<dyn PollSocketReady>> {
266        self.as_ref().new_dyn_socket_ready(socket)
267    }
268
269    fn new_dyn_wait(&self, fd: RawFd, read_size: usize) -> io::Result<PollImpl<dyn PollWait>> {
270        self.as_ref().new_dyn_wait(fd, read_size)
271    }
272
273    #[cfg(target_os = "macos")]
274    fn new_dyn_process_wait(&self, pid: i32) -> io::Result<PollImpl<dyn PollProcessWait>> {
275        self.as_ref().new_dyn_process_wait(pid)
276    }
277
278    #[cfg(target_os = "linux")]
279    fn io_uring_probe(&self, opcode: u8) -> bool {
280        self.as_ref().io_uring_probe(opcode)
281    }
282
283    #[cfg(target_os = "linux")]
284    unsafe fn io_uring_submit(
285        &self,
286        sqe: crate::io_uring::Entry,
287    ) -> std::pin::Pin<Box<dyn Future<Output = io::Result<i32>> + Send + '_>> {
288        // SAFETY: caller guarantees contract
289        unsafe { self.as_ref().io_uring_submit(sqe) }
290    }
291}
292
293#[cfg(windows)]
294impl Driver for Box<dyn Driver> {
295    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
296        self.as_ref().new_dyn_timer()
297    }
298
299    fn new_dyn_socket_ready(&self, socket: RawSocket) -> io::Result<PollImpl<dyn PollSocketReady>> {
300        self.as_ref().new_dyn_socket_ready(socket)
301    }
302
303    fn new_dyn_wait(&self, handle: RawHandle) -> io::Result<PollImpl<dyn PollWait>> {
304        self.as_ref().new_dyn_wait(handle)
305    }
306
307    unsafe fn new_dyn_overlapped_file(
308        &self,
309        handle: RawHandle,
310    ) -> io::Result<PollImpl<dyn IoOverlapped>> {
311        // SAFETY: caller guarantees contract
312        unsafe { self.as_ref().new_dyn_overlapped_file(handle) }
313    }
314}
315
316#[cfg(unix)]
317impl Driver for Arc<dyn Driver> {
318    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
319        self.as_ref().new_dyn_timer()
320    }
321
322    fn new_dyn_fd_ready(&self, fd: RawFd) -> io::Result<PollImpl<dyn PollFdReady>> {
323        self.as_ref().new_dyn_fd_ready(fd)
324    }
325
326    fn new_dyn_socket_ready(&self, socket: RawFd) -> io::Result<PollImpl<dyn PollSocketReady>> {
327        self.as_ref().new_dyn_socket_ready(socket)
328    }
329
330    fn new_dyn_wait(&self, fd: RawFd, read_size: usize) -> io::Result<PollImpl<dyn PollWait>> {
331        self.as_ref().new_dyn_wait(fd, read_size)
332    }
333
334    #[cfg(target_os = "macos")]
335    fn new_dyn_process_wait(&self, pid: i32) -> io::Result<PollImpl<dyn PollProcessWait>> {
336        self.as_ref().new_dyn_process_wait(pid)
337    }
338
339    #[cfg(target_os = "linux")]
340    fn io_uring_probe(&self, opcode: u8) -> bool {
341        self.as_ref().io_uring_probe(opcode)
342    }
343
344    #[cfg(target_os = "linux")]
345    unsafe fn io_uring_submit(
346        &self,
347        sqe: crate::io_uring::Entry,
348    ) -> std::pin::Pin<Box<dyn Future<Output = io::Result<i32>> + Send + '_>> {
349        // SAFETY: caller guarantees contract
350        unsafe { self.as_ref().io_uring_submit(sqe) }
351    }
352}
353
354#[cfg(windows)]
355impl Driver for Arc<dyn Driver> {
356    fn new_dyn_timer(&self) -> PollImpl<dyn PollTimer> {
357        self.as_ref().new_dyn_timer()
358    }
359
360    fn new_dyn_socket_ready(&self, socket: RawSocket) -> io::Result<PollImpl<dyn PollSocketReady>> {
361        self.as_ref().new_dyn_socket_ready(socket)
362    }
363
364    fn new_dyn_wait(&self, handle: RawHandle) -> io::Result<PollImpl<dyn PollWait>> {
365        self.as_ref().new_dyn_wait(handle)
366    }
367
368    unsafe fn new_dyn_overlapped_file(
369        &self,
370        handle: RawHandle,
371    ) -> io::Result<PollImpl<dyn IoOverlapped>> {
372        // SAFETY: caller guarantees contract
373        unsafe { self.as_ref().new_dyn_overlapped_file(handle) }
374    }
375}
376
377/// Trait for [`Driver`]s that also implement [`Spawn`].
378pub trait SpawnDriver: Spawn + Driver {}
379
380impl<T: Spawn + Driver> SpawnDriver for T {}
pal_async/driver.rs

pal_async/
driver.rs
