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

cros_async/sys/linux/
poll_source.rs
