cros_async/sys/linux/
poll_source.rs

1// Copyright 2020 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::io;
6use std::os::fd::AsFd;
7use std::os::fd::AsRawFd;
8use std::sync::atomic::AtomicBool;
9use std::sync::atomic::Ordering;
10use std::sync::Arc;
11
12use base::handle_eintr_errno;
13use base::linux::preadv2;
14use base::linux::pwritev2;
15use base::sys::fallocate;
16use base::sys::FallocateMode;
17use base::AsRawDescriptor;
18use base::IoBufMut;
19use base::VolatileSlice;
20use remain::sorted;
21use thiserror::Error as ThisError;
22
23use super::fd_executor;
24use super::fd_executor::EpollReactor;
25use super::fd_executor::RegisteredSource;
26use crate::common_executor::RawExecutor;
27use crate::mem::BackingMemory;
28use crate::AsyncError;
29use crate::AsyncResult;
30use crate::IoOptions;
31use crate::MemRegion;
32
33#[sorted]
34#[derive(ThisError, Debug)]
35pub enum Error {
36    /// An error occurred attempting to register a waker with the executor.
37    #[error("An error occurred attempting to register a waker with the executor: {0}.")]
38    AddingWaker(fd_executor::Error),
39    /// Failed to discard a block
40    #[error("Failed to discard a block: {0}")]
41    Discard(base::Error),
42    /// An executor error occurred.
43    #[error("An executor error occurred: {0}")]
44    Executor(fd_executor::Error),
45    /// An error occurred when executing fallocate synchronously.
46    #[error("An error occurred when executing fallocate synchronously: {0}")]
47    Fallocate(base::Error),
48    /// An error occurred when executing fdatasync synchronously.
49    #[error("An error occurred when executing fdatasync synchronously: {0}")]
50    Fdatasync(base::Error),
51    /// An error occurred when executing fsync synchronously.
52    #[error("An error occurred when executing fsync synchronously: {0}")]
53    Fsync(base::Error),
54    /// An error occurred when reading the FD.
55    #[error("An error occurred when reading the FD: {0}.")]
56    Read(base::Error),
57    /// Can't seek file.
58    #[error("An error occurred when seeking the FD: {0}.")]
59    Seeking(base::Error),
60    /// An error occurred when writing the FD.
61    #[error("An error occurred when writing the FD: {0}.")]
62    Write(base::Error),
63}
64pub type Result<T> = std::result::Result<T, Error>;
65
66impl From<Error> for io::Error {
67    fn from(e: Error) -> Self {
68        use Error::*;
69        match e {
70            AddingWaker(e) => e.into(),
71            Executor(e) => e.into(),
72            Discard(e) => e.into(),
73            Fallocate(e) => e.into(),
74            Fdatasync(e) => e.into(),
75            Fsync(e) => e.into(),
76            Read(e) => e.into(),
77            Seeking(e) => e.into(),
78            Write(e) => e.into(),
79        }
80    }
81}
82
83impl From<Error> for AsyncError {
84    fn from(e: Error) -> AsyncError {
85        AsyncError::SysVariants(e.into())
86    }
87}
88
89static KERNEL_SUPPORTS_DONTCACHE: AtomicBool = AtomicBool::new(true);
90
91/// Async wrapper for an IO source that uses the FD executor to drive async operations.
92pub struct PollSource<F> {
93    registered_source: RegisteredSource<F>,
94}
95
96impl<F: AsRawDescriptor> PollSource<F> {
97    /// Create a new `PollSource` from the given IO source.
98    pub fn new(f: F, ex: &Arc<RawExecutor<EpollReactor>>) -> Result<Self> {
99        RegisteredSource::new(ex, f)
100            .map({
101                |f| PollSource {
102                    registered_source: f,
103                }
104            })
105            .map_err(Error::Executor)
106    }
107}
108
109impl<F: AsRawDescriptor> PollSource<F> {
110    /// Reads from the iosource at `file_offset` and fill the given `vec`.
111    pub async fn read_to_vec(
112        &self,
113        file_offset: Option<u64>,
114        mut vec: Vec<u8>,
115        options: IoOptions,
116    ) -> AsyncResult<(usize, Vec<u8>)> {
117        let mut use_dontcache =
118            options.dontcache && KERNEL_SUPPORTS_DONTCACHE.load(Ordering::Relaxed);
119        loop {
120            let res = handle_eintr_errno!(if use_dontcache {
121                let offset = file_offset.map(|o| o as libc::off_t).unwrap_or(-1);
122                let mut iovecs = [IoBufMut::new(&mut vec)];
123                preadv2(
124                    self.registered_source.duped_fd.as_fd(),
125                    &mut iovecs,
126                    offset,
127                    libc::RWF_DONTCACHE,
128                )
129            } else if let Some(offset) = file_offset {
130                // SAFETY:
131                // Safe because we trust the kernel not to write past the length given and the
132                // pointers/buffers are guaranteed to be valid for the duration of the call.
133                unsafe {
134                    libc::pread64(
135                        self.registered_source.duped_fd.as_raw_fd(),
136                        vec.as_mut_ptr() as *mut libc::c_void,
137                        vec.len(),
138                        offset as libc::off64_t,
139                    )
140                }
141            } else {
142                // SAFETY:
143                // Safe because we trust the kernel not to write past the length given and the
144                // pointers/buffers are guaranteed to be valid for the duration of the call.
145                unsafe {
146                    libc::read(
147                        self.registered_source.duped_fd.as_raw_fd(),
148                        vec.as_mut_ptr() as *mut libc::c_void,
149                        vec.len(),
150                    )
151                }
152            });
153
154            if res >= 0 {
155                return Ok((res as usize, vec));
156            }
157
158            let err = base::Error::last();
159            match err.errno() {
160                libc::ENOSYS | libc::EINVAL if use_dontcache => {
161                    // EINVAL can be returned if the kernel supports preadv2 but not the
162                    // RWF_DONTCACHE flag. ENOSYS is returned if preadv2 is not supported at all.
163                    base::warn!(
164                        "crosvm: preadv2 or RWF_DONTCACHE read not supported, falling back to v1!"
165                    );
166                    KERNEL_SUPPORTS_DONTCACHE.store(false, Ordering::Relaxed);
167                    use_dontcache = false;
168                    continue;
169                }
170                libc::EWOULDBLOCK => {
171                    let op = self
172                        .registered_source
173                        .wait_readable()
174                        .map_err(Error::AddingWaker)?;
175                    op.await.map_err(Error::Executor)?;
176                }
177                _ => return Err(Error::Read(err).into()),
178            }
179        }
180    }
181
182    /// Reads to the given `mem` at the given offsets from the file starting at `file_offset`.
183    pub async fn read_to_mem(
184        &self,
185        file_offset: Option<u64>,
186        mem: Arc<dyn BackingMemory + Send + Sync>,
187        mem_offsets: impl IntoIterator<Item = MemRegion>,
188        options: IoOptions,
189    ) -> AsyncResult<usize> {
190        let mut iovecs = mem_offsets
191            .into_iter()
192            .filter_map(|mem_range| mem.get_volatile_slice(mem_range).ok())
193            .collect::<Vec<VolatileSlice>>();
194        let mut use_dontcache =
195            options.dontcache && KERNEL_SUPPORTS_DONTCACHE.load(Ordering::Relaxed);
196        loop {
197            let res = handle_eintr_errno!(if use_dontcache {
198                let offset = file_offset.map(|o| o as libc::off_t).unwrap_or(-1);
199                preadv2(
200                    self.registered_source.duped_fd.as_fd(),
201                    VolatileSlice::as_iobufs_mut(&mut iovecs),
202                    offset,
203                    libc::RWF_DONTCACHE,
204                )
205            } else if let Some(offset) = file_offset {
206                // SAFETY:
207                // Safe because we trust the kernel not to write past the length given and the
208                // volatile slices are guaranteed to be valid for the duration of the call.
209                unsafe {
210                    libc::preadv64(
211                        self.registered_source.duped_fd.as_raw_fd(),
212                        iovecs.as_mut_ptr() as *mut _,
213                        iovecs.len() as i32,
214                        offset as libc::off64_t,
215                    )
216                }
217            } else {
218                // SAFETY:
219                // Safe because we trust the kernel not to write past the length given and the
220                // volatile slices are guaranteed to be valid for the duration of the call.
221                unsafe {
222                    libc::readv(
223                        self.registered_source.duped_fd.as_raw_fd(),
224                        iovecs.as_mut_ptr() as *mut _,
225                        iovecs.len() as i32,
226                    )
227                }
228            });
229
230            if res >= 0 {
231                return Ok(res as usize);
232            }
233
234            let err = base::Error::last();
235            match err.errno() {
236                libc::ENOSYS | libc::EINVAL if use_dontcache => {
237                    // EINVAL can be returned if the kernel supports preadv2 but not the
238                    // RWF_DONTCACHE flag. ENOSYS is returned if preadv2 is not supported at all.
239                    base::warn!(
240                        "crosvm: preadv2 or RWF_DONTCACHE read not supported, falling back to v1!"
241                    );
242                    KERNEL_SUPPORTS_DONTCACHE.store(false, Ordering::Relaxed);
243                    use_dontcache = false;
244                    continue;
245                }
246                libc::EWOULDBLOCK => {
247                    let op = self
248                        .registered_source
249                        .wait_readable()
250                        .map_err(Error::AddingWaker)?;
251                    op.await.map_err(Error::Executor)?;
252                }
253                _ => return Err(Error::Read(err).into()),
254            }
255        }
256    }
257
258    /// Wait for the FD of `self` to be readable.
259    pub async fn wait_readable(&self) -> AsyncResult<()> {
260        let op = self
261            .registered_source
262            .wait_readable()
263            .map_err(Error::AddingWaker)?;
264        op.await.map_err(Error::Executor)?;
265        Ok(())
266    }
267
268    /// Writes from the given `vec` to the file starting at `file_offset`.
269    pub async fn write_from_vec(
270        &self,
271        file_offset: Option<u64>,
272        mut vec: Vec<u8>,
273        options: IoOptions,
274    ) -> AsyncResult<(usize, Vec<u8>)> {
275        let mut use_dontcache =
276            options.dontcache && KERNEL_SUPPORTS_DONTCACHE.load(Ordering::Relaxed);
277        loop {
278            let res = handle_eintr_errno!(if use_dontcache {
279                let offset = file_offset.map(|o| o as libc::off_t).unwrap_or(-1);
280                let iovecs = [IoBufMut::new(&mut vec)];
281                pwritev2(
282                    self.registered_source.duped_fd.as_fd(),
283                    IoBufMut::as_iobufs(&iovecs),
284                    offset,
285                    libc::RWF_DONTCACHE,
286                )
287            } else if let Some(offset) = file_offset {
288                // SAFETY:
289                // Safe because we only read from the passed buffer and the pointers/buffers
290                // are guaranteed to be valid for the duration of the call.
291                unsafe {
292                    libc::pwrite64(
293                        self.registered_source.duped_fd.as_raw_fd(),
294                        vec.as_ptr() as *const libc::c_void,
295                        vec.len(),
296                        offset as libc::off64_t,
297                    )
298                }
299            } else {
300                // SAFETY:
301                // Safe because we only read from the passed buffer and the pointers/buffers
302                // are guaranteed to be valid for the duration of the call.
303                unsafe {
304                    libc::write(
305                        self.registered_source.duped_fd.as_raw_fd(),
306                        vec.as_ptr() as *const libc::c_void,
307                        vec.len(),
308                    )
309                }
310            });
311
312            if res >= 0 {
313                return Ok((res as usize, vec));
314            }
315
316            let err = base::Error::last();
317            match err.errno() {
318                libc::ENOSYS | libc::EINVAL if use_dontcache => {
319                    // EINVAL can be returned if the kernel supports pwritev2 but not the
320                    // RWF_DONTCACHE flag. ENOSYS is returned if pwritev2 is not supported at all.
321                    base::warn!(
322                        "crosvm: pwritev2 or RWF_DONTCACHE write not supported, falling back to v1!"
323                    );
324                    KERNEL_SUPPORTS_DONTCACHE.store(false, Ordering::Relaxed);
325                    use_dontcache = false;
326                    continue;
327                }
328                libc::EWOULDBLOCK => {
329                    let op = self
330                        .registered_source
331                        .wait_writable()
332                        .map_err(Error::AddingWaker)?;
333                    op.await.map_err(Error::Executor)?;
334                }
335                _ => return Err(Error::Write(err).into()),
336            }
337        }
338    }
339
340    /// Writes from the given `mem` from the given offsets to the file starting at `file_offset`.
341    pub async fn write_from_mem(
342        &self,
343        file_offset: Option<u64>,
344        mem: Arc<dyn BackingMemory + Send + Sync>,
345        mem_offsets: impl IntoIterator<Item = MemRegion>,
346        options: IoOptions,
347    ) -> AsyncResult<usize> {
348        let iovecs = mem_offsets
349            .into_iter()
350            .map(|mem_range| mem.get_volatile_slice(mem_range))
351            .filter_map(|r| r.ok())
352            .collect::<Vec<VolatileSlice>>();
353        let mut use_dontcache =
354            options.dontcache && KERNEL_SUPPORTS_DONTCACHE.load(Ordering::Relaxed);
355        loop {
356            let res = handle_eintr_errno!(if use_dontcache {
357                let offset = file_offset.map(|o| o as libc::off_t).unwrap_or(-1);
358                pwritev2(
359                    self.registered_source.duped_fd.as_fd(),
360                    IoBufMut::as_iobufs(VolatileSlice::as_iobufs(&iovecs)),
361                    offset,
362                    libc::RWF_DONTCACHE,
363                )
364            } else if let Some(offset) = file_offset {
365                // SAFETY:
366                // Safe because we only read from the passed volatile slices and they are
367                // guaranteed to be valid for the duration of the call.
368                unsafe {
369                    libc::pwritev64(
370                        self.registered_source.duped_fd.as_raw_fd(),
371                        iovecs.as_ptr() as *mut _,
372                        iovecs.len() as i32,
373                        offset as libc::off64_t,
374                    )
375                }
376            } else {
377                // SAFETY:
378                // Safe because we only read from the passed volatile slices and they are
379                // guaranteed to be valid for the duration of the call.
380                unsafe {
381                    libc::writev(
382                        self.registered_source.duped_fd.as_raw_fd(),
383                        iovecs.as_ptr() as *mut _,
384                        iovecs.len() as i32,
385                    )
386                }
387            });
388
389            if res >= 0 {
390                return Ok(res as usize);
391            }
392
393            let err = base::Error::last();
394            match err.errno() {
395                libc::ENOSYS | libc::EINVAL if use_dontcache => {
396                    // EINVAL can be returned if the kernel supports pwritev2 but not the
397                    // RWF_DONTCACHE flag. ENOSYS is returned if pwritev2 is not supported at all.
398                    base::warn!(
399                        "crosvm: pwritev2 or RWF_DONTCACHE write not supported, falling back to v1!"
400                    );
401                    KERNEL_SUPPORTS_DONTCACHE.store(false, Ordering::Relaxed);
402                    use_dontcache = false;
403                    continue;
404                }
405                libc::EWOULDBLOCK => {
406                    let op = self
407                        .registered_source
408                        .wait_writable()
409                        .map_err(Error::AddingWaker)?;
410                    op.await.map_err(Error::Executor)?;
411                }
412                _ => return Err(Error::Write(err).into()),
413            }
414        }
415    }
416
417    /// # Safety
418    ///
419    /// Sync all completed write operations to the backing storage.
420    pub async fn fsync(&self) -> AsyncResult<()> {
421        // SAFETY: the duped_fd is valid and return value is checked.
422        let ret = handle_eintr_errno!(unsafe {
423            libc::fsync(self.registered_source.duped_fd.as_raw_fd())
424        });
425        if ret == 0 {
426            Ok(())
427        } else {
428            Err(Error::Fsync(base::Error::last()).into())
429        }
430    }
431
432    /// punch_hole
433    pub async fn punch_hole(&self, file_offset: u64, len: u64) -> AsyncResult<()> {
434        Ok(fallocate(
435            &self.registered_source.duped_fd,
436            FallocateMode::PunchHole,
437            file_offset,
438            len,
439        )
440        .map_err(Error::Fallocate)?)
441    }
442
443    /// write_zeroes_at
444    pub async fn write_zeroes_at(&self, file_offset: u64, len: u64) -> AsyncResult<()> {
445        Ok(fallocate(
446            &self.registered_source.duped_fd,
447            FallocateMode::ZeroRange,
448            file_offset,
449            len,
450        )
451        .map_err(Error::Fallocate)?)
452    }
453
454    /// Sync all data of completed write operations to the backing storage, avoiding updating extra
455    /// metadata.
456    pub async fn fdatasync(&self) -> AsyncResult<()> {
457        // SAFETY: the duped_fd is valid and return value is checked.
458        let ret = handle_eintr_errno!(unsafe {
459            libc::fdatasync(self.registered_source.duped_fd.as_raw_fd())
460        });
461        if ret == 0 {
462            Ok(())
463        } else {
464            Err(Error::Fdatasync(base::Error::last()).into())
465        }
466    }
467
468    /// Yields the underlying IO source.
469    pub fn into_source(self) -> F {
470        self.registered_source.source
471    }
472
473    /// Provides a mutable ref to the underlying IO source.
474    pub fn as_source_mut(&mut self) -> &mut F {
475        &mut self.registered_source.source
476    }
477
478    /// Provides a ref to the underlying IO source.
479    pub fn as_source(&self) -> &F {
480        &self.registered_source.source
481    }
482}
483
484// NOTE: Prefer adding tests to io_source.rs if not backend specific.
485#[cfg(test)]
486mod tests {
487    use std::fs::File;
488
489    use super::*;
490    use crate::ExecutorTrait;
491
492    #[test]
493    fn memory_leak() {
494        // This test needs to run under ASAN to detect memory leaks.
495
496        async fn owns_poll_source(source: PollSource<File>) {
497            let _ = source.wait_readable().await;
498        }
499
500        let (rx, _tx) = base::pipe().unwrap();
501        let ex = RawExecutor::<EpollReactor>::new().unwrap();
502        let source = PollSource::new(rx, &ex).unwrap();
503        ex.spawn_local(owns_poll_source(source)).detach();
504
505        // Drop `ex` without running. This would cause a memory leak if PollSource owned a strong
506        // reference to the executor because it owns a reference to the future that owns PollSource
507        // (via its Runnable). The strong reference prevents the drop impl from running, which would
508        // otherwise poll the future and have it return with an error.
509    }
510}
cros_async/sys/linux/poll_source.rs

cros_async/sys/linux/
poll_source.rs
