cros_async/
io_source.rs

1// Copyright 2023 The ChromiumOS Authors
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use std::sync::Arc;
6
7use base::AsRawDescriptor;
8
9#[cfg(any(target_os = "android", target_os = "linux"))]
10use crate::sys::linux::PollSource;
11#[cfg(any(target_os = "android", target_os = "linux"))]
12use crate::sys::linux::UringSource;
13#[cfg(feature = "tokio")]
14use crate::sys::platform::tokio_source::TokioSource;
15#[cfg(windows)]
16use crate::sys::windows::HandleSource;
17#[cfg(windows)]
18use crate::sys::windows::OverlappedSource;
19use crate::AsyncResult;
20use crate::BackingMemory;
21use crate::MemRegion;
22
23/// Associates an IO object `F` with cros_async's runtime and exposes an API to perform async IO on
24/// that object's descriptor.
25pub enum IoSource<F: base::AsRawDescriptor> {
26    #[cfg(any(target_os = "android", target_os = "linux"))]
27    Uring(UringSource<F>),
28    #[cfg(any(target_os = "android", target_os = "linux"))]
29    Epoll(PollSource<F>),
30    #[cfg(windows)]
31    Handle(HandleSource<F>),
32    #[cfg(windows)]
33    Overlapped(OverlappedSource<F>),
34    #[cfg(feature = "tokio")]
35    Tokio(TokioSource<F>),
36}
37
38static_assertions::assert_impl_all!(IoSource<std::fs::File>: Send, Sync);
39
40/// Invoke a method on the underlying source type and await the result.
41///
42/// `await_on_inner(io_source, method, ...)` => `inner_source.method(...).await`
43macro_rules! await_on_inner {
44    ($x:ident, $method:ident $(, $args:expr)*) => {
45        match $x {
46            #[cfg(any(target_os = "android", target_os = "linux"))]
47            IoSource::Uring(x) => UringSource::$method(x, $($args),*).await,
48            #[cfg(any(target_os = "android", target_os = "linux"))]
49            IoSource::Epoll(x) => PollSource::$method(x, $($args),*).await,
50            #[cfg(windows)]
51            IoSource::Handle(x) => HandleSource::$method(x, $($args),*).await,
52            #[cfg(windows)]
53            IoSource::Overlapped(x) => OverlappedSource::$method(x, $($args),*).await,
54            #[cfg(feature = "tokio")]
55            IoSource::Tokio(x) => TokioSource::$method(x, $($args),*).await,
56        }
57    };
58}
59
60/// Invoke a method on the underlying source type.
61///
62/// `on_inner(io_source, method, ...)` => `inner_source.method(...)`
63macro_rules! on_inner {
64    ($x:ident, $method:ident $(, $args:expr)*) => {
65        match $x {
66            #[cfg(any(target_os = "android", target_os = "linux"))]
67            IoSource::Uring(x) => UringSource::$method(x, $($args),*),
68            #[cfg(any(target_os = "android", target_os = "linux"))]
69            IoSource::Epoll(x) => PollSource::$method(x, $($args),*),
70            #[cfg(windows)]
71            IoSource::Handle(x) => HandleSource::$method(x, $($args),*),
72            #[cfg(windows)]
73            IoSource::Overlapped(x) => OverlappedSource::$method(x, $($args),*),
74            #[cfg(feature = "tokio")]
75            IoSource::Tokio(x) => TokioSource::$method(x, $($args),*),
76        }
77    };
78}
79
80/// Options for IO operations.
81#[derive(Clone, Copy, Debug, Default)]
82pub struct IoOptions {
83    /// Bypasses the page cache for I/O operations.
84    ///
85    /// Currently only implemented for the Linux `PollSource` backend using `RWF_DONTCACHE`.
86    /// Other backends (like `io_uring`, `tokio`, or Windows) silently ignore this option.
87    ///
88    /// On `PollSource`, if the kernel doesn't support `RWF_DONTCACHE` (e.g. older kernels),
89    /// the backend will transparently fallback to cached I/O. However, if the filesystem
90    /// doesn't support it (returning `EOPNOTSUPP`), the operation will fail.
91    pub dontcache: bool,
92}
93
94impl<F: AsRawDescriptor> IoSource<F> {
95    /// Reads at `file_offset` and fills the given `vec`.
96    pub async fn read_to_vec(
97        &self,
98        file_offset: Option<u64>,
99        vec: Vec<u8>,
100        options: IoOptions,
101    ) -> AsyncResult<(usize, Vec<u8>)> {
102        await_on_inner!(self, read_to_vec, file_offset, vec, options)
103    }
104
105    /// Reads to the given `mem` at the given offsets from the file starting at `file_offset`.
106    pub async fn read_to_mem(
107        &self,
108        file_offset: Option<u64>,
109        mem: Arc<dyn BackingMemory + Send + Sync>,
110        mem_offsets: impl IntoIterator<Item = MemRegion>,
111        options: IoOptions,
112    ) -> AsyncResult<usize> {
113        await_on_inner!(self, read_to_mem, file_offset, mem, mem_offsets, options)
114    }
115
116    /// Waits for the object to be readable.
117    pub async fn wait_readable(&self) -> AsyncResult<()> {
118        await_on_inner!(self, wait_readable)
119    }
120
121    /// Writes from the given `vec` to the file starting at `file_offset`.
122    pub async fn write_from_vec(
123        &self,
124        file_offset: Option<u64>,
125        vec: Vec<u8>,
126        options: IoOptions,
127    ) -> AsyncResult<(usize, Vec<u8>)> {
128        await_on_inner!(self, write_from_vec, file_offset, vec, options)
129    }
130
131    /// Writes from the given `mem` at the given offsets to the file starting at `file_offset`.
132    pub async fn write_from_mem(
133        &self,
134        file_offset: Option<u64>,
135        mem: Arc<dyn BackingMemory + Send + Sync>,
136        mem_offsets: impl IntoIterator<Item = MemRegion>,
137        options: IoOptions,
138    ) -> AsyncResult<usize> {
139        await_on_inner!(self, write_from_mem, file_offset, mem, mem_offsets, options)
140    }
141
142    /// Deallocates the given range of a file.
143    pub async fn punch_hole(&self, file_offset: u64, len: u64) -> AsyncResult<()> {
144        await_on_inner!(self, punch_hole, file_offset, len)
145    }
146
147    /// Fills the given range with zeroes.
148    pub async fn write_zeroes_at(&self, file_offset: u64, len: u64) -> AsyncResult<()> {
149        await_on_inner!(self, write_zeroes_at, file_offset, len)
150    }
151
152    /// Sync all completed write operations to the backing storage.
153    pub async fn fsync(&self) -> AsyncResult<()> {
154        await_on_inner!(self, fsync)
155    }
156
157    /// Sync all data of completed write operations to the backing storage, avoiding updating extra
158    /// metadata. Note that an implementation may simply implement fsync for fdatasync.
159    pub async fn fdatasync(&self) -> AsyncResult<()> {
160        await_on_inner!(self, fdatasync)
161    }
162
163    /// Yields the underlying IO source.
164    pub fn into_source(self) -> F {
165        on_inner!(self, into_source)
166    }
167
168    /// Provides a ref to the underlying IO source.
169    pub fn as_source(&self) -> &F {
170        on_inner!(self, as_source)
171    }
172
173    /// Provides a mutable ref to the underlying IO source.
174    pub fn as_source_mut(&mut self) -> &mut F {
175        on_inner!(self, as_source_mut)
176    }
177
178    /// Waits on a waitable handle.
179    ///
180    /// Needed for Windows currently, and subject to a potential future upstream.
181    #[cfg(windows)]
182    pub async fn wait_for_handle(&self) -> AsyncResult<()> {
183        await_on_inner!(self, wait_for_handle)
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use std::fs::File;
190    use std::io::Read;
191    use std::io::Seek;
192    use std::io::SeekFrom;
193    use std::io::Write;
194    use std::sync::Arc;
195
196    use tempfile::tempfile;
197
198    use super::*;
199    use crate::mem::VecIoWrapper;
200    #[cfg(any(target_os = "android", target_os = "linux"))]
201    use crate::sys::linux::uring_executor::is_uring_stable;
202    use crate::sys::ExecutorKindSys;
203    use crate::Executor;
204    use crate::ExecutorKind;
205    use crate::MemRegion;
206
207    #[cfg(any(target_os = "android", target_os = "linux"))]
208    fn all_kinds() -> Vec<ExecutorKind> {
209        let mut kinds = vec![ExecutorKindSys::Fd.into()];
210        if is_uring_stable() {
211            kinds.push(ExecutorKindSys::Uring.into());
212        }
213        kinds
214    }
215    #[cfg(windows)]
216    fn all_kinds() -> Vec<ExecutorKind> {
217        // TODO: Test OverlappedSource. It requires files to be opened specially, so this test
218        // fixture needs to be refactored first.
219        vec![ExecutorKindSys::Handle.into()]
220    }
221
222    fn tmpfile_with_contents(bytes: &[u8]) -> File {
223        let mut f = tempfile().unwrap();
224        f.write_all(bytes).unwrap();
225        f.flush().unwrap();
226        f.seek(SeekFrom::Start(0)).unwrap();
227        f
228    }
229
230    #[test]
231    fn readvec() {
232        for kind in all_kinds() {
233            async fn go<F: AsRawDescriptor>(async_source: IoSource<F>) {
234                let v = vec![0x55u8; 32];
235                let v_ptr = v.as_ptr();
236                let (n, v) = async_source
237                    .read_to_vec(None, v, Default::default())
238                    .await
239                    .unwrap();
240                assert_eq!(v_ptr, v.as_ptr());
241                assert_eq!(n, 4);
242                assert_eq!(&v[..4], "data".as_bytes());
243            }
244
245            let f = tmpfile_with_contents("data".as_bytes());
246            let ex = Executor::with_executor_kind(kind).unwrap();
247            let source = ex.async_from(f).unwrap();
248            ex.run_until(go(source)).unwrap();
249        }
250    }
251
252    #[test]
253    fn writevec() {
254        for kind in all_kinds() {
255            async fn go<F: AsRawDescriptor>(async_source: IoSource<F>) {
256                let v = "data".as_bytes().to_vec();
257                let v_ptr = v.as_ptr();
258                let (n, v) = async_source
259                    .write_from_vec(None, v, Default::default())
260                    .await
261                    .unwrap();
262                assert_eq!(n, 4);
263                assert_eq!(v_ptr, v.as_ptr());
264            }
265
266            let mut f = tmpfile_with_contents(&[]);
267            let ex = Executor::with_executor_kind(kind).unwrap();
268            let source = ex.async_from(f.try_clone().unwrap()).unwrap();
269            ex.run_until(go(source)).unwrap();
270
271            f.rewind().unwrap();
272            assert_eq!(std::io::read_to_string(f).unwrap(), "data");
273        }
274    }
275
276    #[test]
277    fn readmem() {
278        for kind in all_kinds() {
279            async fn go<F: AsRawDescriptor>(async_source: IoSource<F>) {
280                let mem = Arc::new(VecIoWrapper::from(vec![b' '; 10]));
281                let n = async_source
282                    .read_to_mem(
283                        None,
284                        Arc::<VecIoWrapper>::clone(&mem),
285                        [
286                            MemRegion { offset: 0, len: 2 },
287                            MemRegion { offset: 4, len: 1 },
288                        ],
289                        Default::default(),
290                    )
291                    .await
292                    .unwrap();
293                assert_eq!(n, 3);
294                let vec: Vec<u8> = match Arc::try_unwrap(mem) {
295                    Ok(v) => v.into(),
296                    Err(_) => panic!("Too many vec refs"),
297                };
298                assert_eq!(std::str::from_utf8(&vec).unwrap(), "da  t     ");
299            }
300
301            let f = tmpfile_with_contents("data".as_bytes());
302            let ex = Executor::with_executor_kind(kind).unwrap();
303            let source = ex.async_from(f).unwrap();
304            ex.run_until(go(source)).unwrap();
305        }
306    }
307
308    #[test]
309    fn writemem() {
310        for kind in all_kinds() {
311            async fn go<F: AsRawDescriptor>(async_source: IoSource<F>) {
312                let mem = Arc::new(VecIoWrapper::from("data".as_bytes().to_vec()));
313                let ret = async_source
314                    .write_from_mem(
315                        None,
316                        Arc::<VecIoWrapper>::clone(&mem),
317                        [
318                            MemRegion { offset: 0, len: 1 },
319                            MemRegion { offset: 2, len: 2 },
320                        ],
321                        Default::default(),
322                    )
323                    .await
324                    .unwrap();
325                assert_eq!(ret, 3);
326            }
327
328            let mut f = tmpfile_with_contents(&[]);
329            let ex = Executor::with_executor_kind(kind).unwrap();
330            let source = ex.async_from(f.try_clone().unwrap()).unwrap();
331            ex.run_until(go(source)).unwrap();
332
333            f.rewind().unwrap();
334            assert_eq!(std::io::read_to_string(f).unwrap(), "dta");
335        }
336    }
337
338    #[test]
339    fn fsync() {
340        for kind in all_kinds() {
341            async fn go<F: AsRawDescriptor>(source: IoSource<F>) {
342                let v = vec![0x55u8; 32];
343                let v_ptr = v.as_ptr();
344                let ret = source
345                    .write_from_vec(None, v, Default::default())
346                    .await
347                    .unwrap();
348                assert_eq!(ret.0, 32);
349                let ret_v = ret.1;
350                assert_eq!(v_ptr, ret_v.as_ptr());
351                source.fsync().await.unwrap();
352            }
353
354            let f = tempfile::tempfile().unwrap();
355            let ex = Executor::with_executor_kind(kind).unwrap();
356            let source = ex.async_from(f).unwrap();
357
358            ex.run_until(go(source)).unwrap();
359        }
360    }
361
362    #[test]
363    fn readmulti() {
364        for kind in all_kinds() {
365            async fn go<F: AsRawDescriptor>(source: IoSource<F>) {
366                let v = vec![0x55u8; 32];
367                let v2 = vec![0x55u8; 32];
368                let (ret, ret2) = futures::future::join(
369                    source.read_to_vec(None, v, Default::default()),
370                    source.read_to_vec(Some(32), v2, Default::default()),
371                )
372                .await;
373
374                let (count, v) = ret.unwrap();
375                let (count2, v2) = ret2.unwrap();
376
377                assert!(v.iter().take(count).all(|&b| b == 0xAA));
378                assert!(v2.iter().take(count2).all(|&b| b == 0xBB));
379            }
380
381            let mut f = tempfile::tempfile().unwrap();
382            f.write_all(&[0xAA; 32]).unwrap();
383            f.write_all(&[0xBB; 32]).unwrap();
384            f.rewind().unwrap();
385
386            let ex = Executor::with_executor_kind(kind).unwrap();
387            let source = ex.async_from(f).unwrap();
388
389            ex.run_until(go(source)).unwrap();
390        }
391    }
392
393    #[test]
394    fn writemulti() {
395        for kind in all_kinds() {
396            async fn go<F: AsRawDescriptor>(source: IoSource<F>) {
397                let v = vec![0x55u8; 32];
398                let v2 = vec![0x55u8; 32];
399                let (r, r2) = futures::future::join(
400                    source.write_from_vec(None, v, Default::default()),
401                    source.write_from_vec(Some(32), v2, Default::default()),
402                )
403                .await;
404                assert_eq!(32, r.unwrap().0);
405                assert_eq!(32, r2.unwrap().0);
406            }
407
408            let f = tempfile::tempfile().unwrap();
409            let ex = Executor::with_executor_kind(kind).unwrap();
410            let source = ex.async_from(f).unwrap();
411
412            ex.run_until(go(source)).unwrap();
413        }
414    }
415
416    #[test]
417    fn read_current_file_position() {
418        for kind in all_kinds() {
419            async fn go<F: AsRawDescriptor>(source: IoSource<F>) {
420                let (count1, verify1) = source
421                    .read_to_vec(None, vec![0u8; 32], Default::default())
422                    .await
423                    .unwrap();
424                let (count2, verify2) = source
425                    .read_to_vec(None, vec![0u8; 32], Default::default())
426                    .await
427                    .unwrap();
428                assert_eq!(count1, 32);
429                assert_eq!(count2, 32);
430                assert_eq!(verify1, [0x55u8; 32]);
431                assert_eq!(verify2, [0xffu8; 32]);
432            }
433
434            let mut f = tempfile::tempfile().unwrap();
435            f.write_all(&[0x55u8; 32]).unwrap();
436            f.write_all(&[0xffu8; 32]).unwrap();
437            f.rewind().unwrap();
438
439            let ex = Executor::with_executor_kind(kind).unwrap();
440            let source = ex.async_from(f).unwrap();
441
442            ex.run_until(go(source)).unwrap();
443        }
444    }
445
446    #[test]
447    fn write_current_file_position() {
448        for kind in all_kinds() {
449            async fn go<F: AsRawDescriptor>(source: IoSource<F>) {
450                let count1 = source
451                    .write_from_vec(None, vec![0x55u8; 32], Default::default())
452                    .await
453                    .unwrap()
454                    .0;
455                assert_eq!(count1, 32);
456                let count2 = source
457                    .write_from_vec(None, vec![0xffu8; 32], Default::default())
458                    .await
459                    .unwrap()
460                    .0;
461                assert_eq!(count2, 32);
462            }
463
464            let mut f = tempfile::tempfile().unwrap();
465            let ex = Executor::with_executor_kind(kind).unwrap();
466            let source = ex.async_from(f.try_clone().unwrap()).unwrap();
467
468            ex.run_until(go(source)).unwrap();
469
470            f.rewind().unwrap();
471            let mut verify1 = [0u8; 32];
472            let mut verify2 = [0u8; 32];
473            f.read_exact(&mut verify1).unwrap();
474            f.read_exact(&mut verify2).unwrap();
475            assert_eq!(verify1, [0x55u8; 32]);
476            assert_eq!(verify2, [0xffu8; 32]);
477        }
478    }
479
480    #[test]
481    fn read_dontcache() {
482        for kind in all_kinds() {
483            async fn go<F: AsRawDescriptor>(source: IoSource<F>) {
484                let v = vec![0u8; 4];
485                let options = IoOptions { dontcache: true };
486                let res = source.read_to_vec(None, v, options).await;
487                match res {
488                    Ok((n, _)) => assert_eq!(n, 4),
489                    Err(e) => {
490                        let io_err: std::io::Error = e.into();
491                        #[cfg(unix)]
492                        assert_eq!(io_err.raw_os_error(), Some(libc::EOPNOTSUPP));
493                        #[cfg(windows)]
494                        panic!("Expected success on Windows, got error: {io_err:?}");
495                    }
496                }
497            }
498
499            let f = tmpfile_with_contents("data".as_bytes());
500            let ex = Executor::with_executor_kind(kind).unwrap();
501            let source = ex.async_from(f).unwrap();
502            ex.run_until(go(source)).unwrap();
503        }
504    }
505}
cros_async/io_source.rs

cros_async/
io_source.rs
