// vim: tw=80 //! POSIX Asynchronous I/O //! //! The POSIX AIO interface is used for asynchronous I/O on files and disk-like //! devices. It supports [`read`](struct.AioCb.html#method.read), //! [`write`](struct.AioCb.html#method.write), and //! [`fsync`](struct.AioCb.html#method.fsync) operations. Completion //! notifications can optionally be delivered via //! [signals](../signal/enum.SigevNotify.html#variant.SigevSignal), via the //! [`aio_suspend`](fn.aio_suspend.html) function, or via polling. Some //! platforms support other completion //! notifications, such as //! [kevent](../signal/enum.SigevNotify.html#variant.SigevKevent). //! //! Multiple operations may be submitted in a batch with //! [`lio_listio`](fn.lio_listio.html), though the standard does not guarantee //! that they will be executed atomically. //! //! Outstanding operations may be cancelled with //! [`cancel`](struct.AioCb.html#method.cancel) or //! [`aio_cancel_all`](fn.aio_cancel_all.html), though the operating system may //! not support this for all filesystems and devices. use crate::{Error, Result}; use crate::errno::Errno; use std::os::unix::io::RawFd; use libc::{c_void, off_t, size_t}; use std::fmt; use std::fmt::Debug; use std::marker::PhantomData; use std::mem; use std::pin::Pin; use std::ptr::{null, null_mut}; use crate::sys::signal::*; use std::thread; use crate::sys::time::TimeSpec; libc_enum! { /// Mode for `AioCb::fsync`. Controls whether only data or both data and /// metadata are synced. #[repr(i32)] pub enum AioFsyncMode { /// do it like `fsync` O_SYNC, /// on supported operating systems only, do it like `fdatasync` #[cfg(any(target_os = "ios", target_os = "linux", target_os = "macos", target_os = "netbsd", target_os = "openbsd"))] O_DSYNC } } libc_enum! { /// When used with [`lio_listio`](fn.lio_listio.html), determines whether a /// given `aiocb` should be used for a read operation, a write operation, or /// ignored. Has no effect for any other aio functions. #[repr(i32)] pub enum LioOpcode { LIO_NOP, LIO_WRITE, LIO_READ, } } libc_enum! { /// Mode for [`lio_listio`](fn.lio_listio.html) #[repr(i32)] pub enum LioMode { /// Requests that [`lio_listio`](fn.lio_listio.html) block until all /// requested operations have been completed LIO_WAIT, /// Requests that [`lio_listio`](fn.lio_listio.html) return immediately LIO_NOWAIT, } } /// Return values for [`AioCb::cancel`](struct.AioCb.html#method.cancel) and /// [`aio_cancel_all`](fn.aio_cancel_all.html) #[repr(i32)] #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum AioCancelStat { /// All outstanding requests were canceled AioCanceled = libc::AIO_CANCELED, /// Some requests were not canceled. Their status should be checked with /// `AioCb::error` AioNotCanceled = libc::AIO_NOTCANCELED, /// All of the requests have already finished AioAllDone = libc::AIO_ALLDONE, } /// Newtype that adds Send and Sync to libc::aiocb, which contains raw pointers #[repr(transparent)] struct LibcAiocb(libc::aiocb); unsafe impl Send for LibcAiocb {} unsafe impl Sync for LibcAiocb {} /// AIO Control Block. /// /// The basic structure used by all aio functions. Each `AioCb` represents one /// I/O request. pub struct AioCb<'a> { aiocb: LibcAiocb, /// Tracks whether the buffer pointed to by `libc::aiocb.aio_buf` is mutable mutable: bool, /// Could this `AioCb` potentially have any in-kernel state? in_progress: bool, _buffer: std::marker::PhantomData<&'a [u8]>, _pin: std::marker::PhantomPinned } impl<'a> AioCb<'a> { /// Returns the underlying file descriptor associated with the `AioCb` pub fn fd(&self) -> RawFd { self.aiocb.0.aio_fildes } /// Constructs a new `AioCb` with no associated buffer. /// /// The resulting `AioCb` structure is suitable for use with `AioCb::fsync`. /// /// # Parameters /// /// * `fd`: File descriptor. Required for all aio functions. /// * `prio`: If POSIX Prioritized IO is supported, then the /// operation will be prioritized at the process's /// priority level minus `prio`. /// * `sigev_notify`: Determines how you will be notified of event /// completion. /// /// # Examples /// /// Create an `AioCb` from a raw file descriptor and use it for an /// [`fsync`](#method.fsync) operation. /// /// ``` /// # use nix::errno::Errno; /// # use nix::Error; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify::SigevNone; /// # use std::{thread, time}; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// let f = tempfile().unwrap(); /// let mut aiocb = AioCb::from_fd( f.as_raw_fd(), 0, SigevNone); /// aiocb.fsync(AioFsyncMode::O_SYNC).expect("aio_fsync failed early"); /// while (aiocb.error() == Err(Error::from(Errno::EINPROGRESS))) { /// thread::sleep(time::Duration::from_millis(10)); /// } /// aiocb.aio_return().expect("aio_fsync failed late"); /// # } /// ``` pub fn from_fd(fd: RawFd, prio: libc::c_int, sigev_notify: SigevNotify) -> Pin>> { let mut a = AioCb::common_init(fd, prio, sigev_notify); a.0.aio_offset = 0; a.0.aio_nbytes = 0; a.0.aio_buf = null_mut(); Box::pin(AioCb { aiocb: a, mutable: false, in_progress: false, _buffer: PhantomData, _pin: std::marker::PhantomPinned }) } // Private helper #[cfg(not(any(target_os = "ios", target_os = "macos")))] fn from_mut_slice_unpinned(fd: RawFd, offs: off_t, buf: &'a mut [u8], prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> AioCb<'a> { let mut a = AioCb::common_init(fd, prio, sigev_notify); a.0.aio_offset = offs; a.0.aio_nbytes = buf.len() as size_t; a.0.aio_buf = buf.as_ptr() as *mut c_void; a.0.aio_lio_opcode = opcode as libc::c_int; AioCb { aiocb: a, mutable: true, in_progress: false, _buffer: PhantomData, _pin: std::marker::PhantomPinned } } /// Constructs a new `AioCb` from a mutable slice. /// /// The resulting `AioCb` will be suitable for both read and write /// operations, but only if the borrow checker can guarantee that the slice /// will outlive the `AioCb`. That will usually be the case if the `AioCb` /// is stack-allocated. /// /// # Parameters /// /// * `fd`: File descriptor. Required for all aio functions. /// * `offs`: File offset /// * `buf`: A memory buffer /// * `prio`: If POSIX Prioritized IO is supported, then the /// operation will be prioritized at the process's /// priority level minus `prio` /// * `sigev_notify`: Determines how you will be notified of event /// completion. /// * `opcode`: This field is only used for `lio_listio`. It /// determines which operation to use for this individual /// aiocb /// /// # Examples /// /// Create an `AioCb` from a mutable slice and read into it. /// /// ``` /// # use nix::errno::Errno; /// # use nix::Error; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::{thread, time}; /// # use std::io::Write; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// const INITIAL: &[u8] = b"abcdef123456"; /// const LEN: usize = 4; /// let mut rbuf = vec![0; LEN]; /// let mut f = tempfile().unwrap(); /// f.write_all(INITIAL).unwrap(); /// { /// let mut aiocb = AioCb::from_mut_slice( f.as_raw_fd(), /// 2, //offset /// &mut rbuf, /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_NOP); /// aiocb.read().unwrap(); /// while (aiocb.error() == Err(Error::from(Errno::EINPROGRESS))) { /// thread::sleep(time::Duration::from_millis(10)); /// } /// assert_eq!(aiocb.aio_return().unwrap() as usize, LEN); /// } /// assert_eq!(rbuf, b"cdef"); /// # } /// ``` pub fn from_mut_slice(fd: RawFd, offs: off_t, buf: &'a mut [u8], prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> Pin>> { let mut a = AioCb::common_init(fd, prio, sigev_notify); a.0.aio_offset = offs; a.0.aio_nbytes = buf.len() as size_t; a.0.aio_buf = buf.as_ptr() as *mut c_void; a.0.aio_lio_opcode = opcode as libc::c_int; Box::pin(AioCb { aiocb: a, mutable: true, in_progress: false, _buffer: PhantomData, _pin: std::marker::PhantomPinned }) } /// Constructs a new `AioCb` from a mutable raw pointer /// /// Unlike `from_mut_slice`, this method returns a structure suitable for /// placement on the heap. It may be used for both reads and writes. Due /// to its unsafety, this method is not recommended. It is most useful when /// heap allocation is required. /// /// # Parameters /// /// * `fd`: File descriptor. Required for all aio functions. /// * `offs`: File offset /// * `buf`: Pointer to the memory buffer /// * `len`: Length of the buffer pointed to by `buf` /// * `prio`: If POSIX Prioritized IO is supported, then the /// operation will be prioritized at the process's /// priority level minus `prio` /// * `sigev_notify`: Determines how you will be notified of event /// completion. /// * `opcode`: This field is only used for `lio_listio`. It /// determines which operation to use for this individual /// aiocb /// /// # Safety /// /// The caller must ensure that the storage pointed to by `buf` outlives the /// `AioCb`. The lifetime checker can't help here. pub unsafe fn from_mut_ptr(fd: RawFd, offs: off_t, buf: *mut c_void, len: usize, prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> Pin>> { let mut a = AioCb::common_init(fd, prio, sigev_notify); a.0.aio_offset = offs; a.0.aio_nbytes = len; a.0.aio_buf = buf; a.0.aio_lio_opcode = opcode as libc::c_int; Box::pin(AioCb { aiocb: a, mutable: true, in_progress: false, _buffer: PhantomData, _pin: std::marker::PhantomPinned, }) } /// Constructs a new `AioCb` from a raw pointer. /// /// Unlike `from_slice`, this method returns a structure suitable for /// placement on the heap. Due to its unsafety, this method is not /// recommended. It is most useful when heap allocation is required. /// /// # Parameters /// /// * `fd`: File descriptor. Required for all aio functions. /// * `offs`: File offset /// * `buf`: Pointer to the memory buffer /// * `len`: Length of the buffer pointed to by `buf` /// * `prio`: If POSIX Prioritized IO is supported, then the /// operation will be prioritized at the process's /// priority level minus `prio` /// * `sigev_notify`: Determines how you will be notified of event /// completion. /// * `opcode`: This field is only used for `lio_listio`. It /// determines which operation to use for this individual /// aiocb /// /// # Safety /// /// The caller must ensure that the storage pointed to by `buf` outlives the /// `AioCb`. The lifetime checker can't help here. pub unsafe fn from_ptr(fd: RawFd, offs: off_t, buf: *const c_void, len: usize, prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> Pin>> { let mut a = AioCb::common_init(fd, prio, sigev_notify); a.0.aio_offset = offs; a.0.aio_nbytes = len; // casting a const ptr to a mutable ptr here is ok, because we set the // AioCb's mutable field to false a.0.aio_buf = buf as *mut c_void; a.0.aio_lio_opcode = opcode as libc::c_int; Box::pin(AioCb { aiocb: a, mutable: false, in_progress: false, _buffer: PhantomData, _pin: std::marker::PhantomPinned }) } // Private helper fn from_slice_unpinned(fd: RawFd, offs: off_t, buf: &'a [u8], prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> AioCb { let mut a = AioCb::common_init(fd, prio, sigev_notify); a.0.aio_offset = offs; a.0.aio_nbytes = buf.len() as size_t; // casting an immutable buffer to a mutable pointer looks unsafe, // but technically its only unsafe to dereference it, not to create // it. a.0.aio_buf = buf.as_ptr() as *mut c_void; assert!(opcode != LioOpcode::LIO_READ, "Can't read into an immutable buffer"); a.0.aio_lio_opcode = opcode as libc::c_int; AioCb { aiocb: a, mutable: false, in_progress: false, _buffer: PhantomData, _pin: std::marker::PhantomPinned } } /// Like [`AioCb::from_mut_slice`], but works on constant slices rather than /// mutable slices. /// /// An `AioCb` created this way cannot be used with `read`, and its /// `LioOpcode` cannot be set to `LIO_READ`. This method is useful when /// writing a const buffer with `AioCb::write`, since `from_mut_slice` can't /// work with const buffers. /// /// # Examples /// /// Construct an `AioCb` from a slice and use it for writing. /// /// ``` /// # use nix::errno::Errno; /// # use nix::Error; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::{thread, time}; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// const WBUF: &[u8] = b"abcdef123456"; /// let mut f = tempfile().unwrap(); /// let mut aiocb = AioCb::from_slice( f.as_raw_fd(), /// 2, //offset /// WBUF, /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_NOP); /// aiocb.write().unwrap(); /// while (aiocb.error() == Err(Error::from(Errno::EINPROGRESS))) { /// thread::sleep(time::Duration::from_millis(10)); /// } /// assert_eq!(aiocb.aio_return().unwrap() as usize, WBUF.len()); /// # } /// ``` // Note: another solution to the problem of writing const buffers would be // to genericize AioCb for both &mut [u8] and &[u8] buffers. AioCb::read // could take the former and AioCb::write could take the latter. However, // then lio_listio wouldn't work, because that function needs a slice of // AioCb, and they must all be of the same type. pub fn from_slice(fd: RawFd, offs: off_t, buf: &'a [u8], prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> Pin> { Box::pin(AioCb::from_slice_unpinned(fd, offs, buf, prio, sigev_notify, opcode)) } fn common_init(fd: RawFd, prio: libc::c_int, sigev_notify: SigevNotify) -> LibcAiocb { // Use mem::zeroed instead of explicitly zeroing each field, because the // number and name of reserved fields is OS-dependent. On some OSes, // some reserved fields are used the kernel for state, and must be // explicitly zeroed when allocated. let mut a = unsafe { mem::zeroed::()}; a.aio_fildes = fd; a.aio_reqprio = prio; a.aio_sigevent = SigEvent::new(sigev_notify).sigevent(); LibcAiocb(a) } /// Update the notification settings for an existing `aiocb` pub fn set_sigev_notify(self: &mut Pin>, sigev_notify: SigevNotify) { // Safe because we don't move any of the data let selfp = unsafe { self.as_mut().get_unchecked_mut() }; selfp.aiocb.0.aio_sigevent = SigEvent::new(sigev_notify).sigevent(); } /// Cancels an outstanding AIO request. /// /// The operating system is not required to implement cancellation for all /// file and device types. Even if it does, there is no guarantee that the /// operation has not already completed. So the caller must check the /// result and handle operations that were not canceled or that have already /// completed. /// /// # Examples /// /// Cancel an outstanding aio operation. Note that we must still call /// `aio_return` to free resources, even though we don't care about the /// result. /// /// ``` /// # use nix::errno::Errno; /// # use nix::Error; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::{thread, time}; /// # use std::io::Write; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// let wbuf = b"CDEF"; /// let mut f = tempfile().unwrap(); /// let mut aiocb = AioCb::from_slice( f.as_raw_fd(), /// 2, //offset /// &wbuf[..], /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_NOP); /// aiocb.write().unwrap(); /// let cs = aiocb.cancel().unwrap(); /// if cs == AioCancelStat::AioNotCanceled { /// while (aiocb.error() == Err(Error::from(Errno::EINPROGRESS))) { /// thread::sleep(time::Duration::from_millis(10)); /// } /// } /// // Must call `aio_return`, but ignore the result /// let _ = aiocb.aio_return(); /// # } /// ``` /// /// # References /// /// [aio_cancel](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_cancel.html) pub fn cancel(self: &mut Pin>) -> Result { let r = unsafe { let selfp = self.as_mut().get_unchecked_mut(); libc::aio_cancel(selfp.aiocb.0.aio_fildes, &mut selfp.aiocb.0) }; match r { libc::AIO_CANCELED => Ok(AioCancelStat::AioCanceled), libc::AIO_NOTCANCELED => Ok(AioCancelStat::AioNotCanceled), libc::AIO_ALLDONE => Ok(AioCancelStat::AioAllDone), -1 => Err(Error::from(Errno::last())), _ => panic!("unknown aio_cancel return value") } } fn error_unpinned(&mut self) -> Result<()> { let r = unsafe { libc::aio_error(&mut self.aiocb.0 as *mut libc::aiocb) }; match r { 0 => Ok(()), num if num > 0 => Err(Error::from(Errno::from_i32(num))), -1 => Err(Error::from(Errno::last())), num => panic!("unknown aio_error return value {:?}", num) } } /// Retrieve error status of an asynchronous operation. /// /// If the request has not yet completed, returns `EINPROGRESS`. Otherwise, /// returns `Ok` or any other error. /// /// # Examples /// /// Issue an aio operation and use `error` to poll for completion. Polling /// is an alternative to `aio_suspend`, used by most of the other examples. /// /// ``` /// # use nix::errno::Errno; /// # use nix::Error; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::{thread, time}; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// const WBUF: &[u8] = b"abcdef123456"; /// let mut f = tempfile().unwrap(); /// let mut aiocb = AioCb::from_slice( f.as_raw_fd(), /// 2, //offset /// WBUF, /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_NOP); /// aiocb.write().unwrap(); /// while (aiocb.error() == Err(Error::from(Errno::EINPROGRESS))) { /// thread::sleep(time::Duration::from_millis(10)); /// } /// assert_eq!(aiocb.aio_return().unwrap() as usize, WBUF.len()); /// # } /// ``` /// /// # References /// /// [aio_error](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_error.html) pub fn error(self: &mut Pin>) -> Result<()> { // Safe because error_unpinned doesn't move the data let selfp = unsafe { self.as_mut().get_unchecked_mut() }; selfp.error_unpinned() } /// An asynchronous version of `fsync(2)`. /// /// # References /// /// [aio_fsync](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_fsync.html) pub fn fsync(self: &mut Pin>, mode: AioFsyncMode) -> Result<()> { // Safe because we don't move the libc::aiocb unsafe { let selfp = self.as_mut().get_unchecked_mut(); Errno::result({ let p: *mut libc::aiocb = &mut selfp.aiocb.0; libc::aio_fsync(mode as libc::c_int, p) }).map(|_| { selfp.in_progress = true; }) } } /// Returns the `aiocb`'s `LioOpcode` field /// /// If the value cannot be represented as an `LioOpcode`, returns `None` /// instead. pub fn lio_opcode(&self) -> Option { match self.aiocb.0.aio_lio_opcode { libc::LIO_READ => Some(LioOpcode::LIO_READ), libc::LIO_WRITE => Some(LioOpcode::LIO_WRITE), libc::LIO_NOP => Some(LioOpcode::LIO_NOP), _ => None } } /// Returns the requested length of the aio operation in bytes /// /// This method returns the *requested* length of the operation. To get the /// number of bytes actually read or written by a completed operation, use /// `aio_return` instead. pub fn nbytes(&self) -> usize { self.aiocb.0.aio_nbytes } /// Returns the file offset stored in the `AioCb` pub fn offset(&self) -> off_t { self.aiocb.0.aio_offset } /// Returns the priority of the `AioCb` pub fn priority(&self) -> libc::c_int { self.aiocb.0.aio_reqprio } /// Asynchronously reads from a file descriptor into a buffer /// /// # References /// /// [aio_read](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_read.html) pub fn read(self: &mut Pin>) -> Result<()> { assert!(self.mutable, "Can't read into an immutable buffer"); // Safe because we don't move anything let selfp = unsafe { self.as_mut().get_unchecked_mut() }; Errno::result({ let p: *mut libc::aiocb = &mut selfp.aiocb.0; unsafe { libc::aio_read(p) } }).map(|_| { selfp.in_progress = true; }) } /// Returns the `SigEvent` stored in the `AioCb` pub fn sigevent(&self) -> SigEvent { SigEvent::from(&self.aiocb.0.aio_sigevent) } fn aio_return_unpinned(&mut self) -> Result { unsafe { let p: *mut libc::aiocb = &mut self.aiocb.0; self.in_progress = false; Errno::result(libc::aio_return(p)) } } /// Retrieve return status of an asynchronous operation. /// /// Should only be called once for each `AioCb`, after `AioCb::error` /// indicates that it has completed. The result is the same as for the /// synchronous `read(2)`, `write(2)`, of `fsync(2)` functions. /// /// # References /// /// [aio_return](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_return.html) // Note: this should be just `return`, but that's a reserved word pub fn aio_return(self: &mut Pin>) -> Result { // Safe because aio_return_unpinned does not move the data let selfp = unsafe { self.as_mut().get_unchecked_mut() }; selfp.aio_return_unpinned() } /// Asynchronously writes from a buffer to a file descriptor /// /// # References /// /// [aio_write](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_write.html) pub fn write(self: &mut Pin>) -> Result<()> { // Safe because we don't move anything let selfp = unsafe { self.as_mut().get_unchecked_mut() }; Errno::result({ let p: *mut libc::aiocb = &mut selfp.aiocb.0; unsafe{ libc::aio_write(p) } }).map(|_| { selfp.in_progress = true; }) } } /// Cancels outstanding AIO requests for a given file descriptor. /// /// # Examples /// /// Issue an aio operation, then cancel all outstanding operations on that file /// descriptor. /// /// ``` /// # use nix::errno::Errno; /// # use nix::Error; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::{thread, time}; /// # use std::io::Write; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// let wbuf = b"CDEF"; /// let mut f = tempfile().unwrap(); /// let mut aiocb = AioCb::from_slice( f.as_raw_fd(), /// 2, //offset /// &wbuf[..], /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_NOP); /// aiocb.write().unwrap(); /// let cs = aio_cancel_all(f.as_raw_fd()).unwrap(); /// if cs == AioCancelStat::AioNotCanceled { /// while (aiocb.error() == Err(Error::from(Errno::EINPROGRESS))) { /// thread::sleep(time::Duration::from_millis(10)); /// } /// } /// // Must call `aio_return`, but ignore the result /// let _ = aiocb.aio_return(); /// # } /// ``` /// /// # References /// /// [`aio_cancel`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_cancel.html) pub fn aio_cancel_all(fd: RawFd) -> Result { match unsafe { libc::aio_cancel(fd, null_mut()) } { libc::AIO_CANCELED => Ok(AioCancelStat::AioCanceled), libc::AIO_NOTCANCELED => Ok(AioCancelStat::AioNotCanceled), libc::AIO_ALLDONE => Ok(AioCancelStat::AioAllDone), -1 => Err(Error::from(Errno::last())), _ => panic!("unknown aio_cancel return value") } } /// Suspends the calling process until at least one of the specified `AioCb`s /// has completed, a signal is delivered, or the timeout has passed. /// /// If `timeout` is `None`, `aio_suspend` will block indefinitely. /// /// # Examples /// /// Use `aio_suspend` to block until an aio operation completes. /// /// ``` /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// const WBUF: &[u8] = b"abcdef123456"; /// let mut f = tempfile().unwrap(); /// let mut aiocb = AioCb::from_slice( f.as_raw_fd(), /// 2, //offset /// WBUF, /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_NOP); /// aiocb.write().unwrap(); /// aio_suspend(&[aiocb.as_ref()], None).expect("aio_suspend failed"); /// assert_eq!(aiocb.aio_return().unwrap() as usize, WBUF.len()); /// # } /// ``` /// # References /// /// [`aio_suspend`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/aio_suspend.html) pub fn aio_suspend(list: &[Pin<&AioCb>], timeout: Option) -> Result<()> { let plist = list as *const [Pin<&AioCb>] as *const [*const libc::aiocb]; let p = plist as *const *const libc::aiocb; let timep = match timeout { None => null::(), Some(x) => x.as_ref() as *const libc::timespec }; Errno::result(unsafe { libc::aio_suspend(p, list.len() as i32, timep) }).map(drop) } impl<'a> Debug for AioCb<'a> { fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { fmt.debug_struct("AioCb") .field("aiocb", &self.aiocb.0) .field("mutable", &self.mutable) .field("in_progress", &self.in_progress) .finish() } } impl<'a> Drop for AioCb<'a> { /// If the `AioCb` has no remaining state in the kernel, just drop it. /// Otherwise, dropping constitutes a resource leak, which is an error fn drop(&mut self) { assert!(thread::panicking() || !self.in_progress, "Dropped an in-progress AioCb"); } } /// LIO Control Block. /// /// The basic structure used to issue multiple AIO operations simultaneously. #[cfg(not(any(target_os = "ios", target_os = "macos")))] pub struct LioCb<'a> { /// A collection of [`AioCb`]s. All of these will be issued simultaneously /// by the [`listio`] method. /// /// [`AioCb`]: struct.AioCb.html /// [`listio`]: #method.listio // Their locations in memory must be fixed once they are passed to the // kernel. So this field must be non-public so the user can't swap. aiocbs: Box<[AioCb<'a>]>, /// The actual list passed to `libc::lio_listio`. /// /// It must live for as long as any of the operations are still being /// processesed, because the aio subsystem uses its address as a unique /// identifier. list: Vec<*mut libc::aiocb>, /// A partial set of results. This field will get populated by /// `listio_resubmit` when an `LioCb` is resubmitted after an error results: Vec>> } /// LioCb can't automatically impl Send and Sync just because of the raw /// pointers in list. But that's stupid. There's no reason that raw pointers /// should automatically be non-Send #[cfg(not(any(target_os = "ios", target_os = "macos")))] unsafe impl<'a> Send for LioCb<'a> {} #[cfg(not(any(target_os = "ios", target_os = "macos")))] unsafe impl<'a> Sync for LioCb<'a> {} #[cfg(not(any(target_os = "ios", target_os = "macos")))] impl<'a> LioCb<'a> { pub fn is_empty(&self) -> bool { self.aiocbs.is_empty() } /// Return the number of individual [`AioCb`]s contained. pub fn len(&self) -> usize { self.aiocbs.len() } /// Submits multiple asynchronous I/O requests with a single system call. /// /// They are not guaranteed to complete atomically, and the order in which /// the requests are carried out is not specified. Reads, writes, and /// fsyncs may be freely mixed. /// /// This function is useful for reducing the context-switch overhead of /// submitting many AIO operations. It can also be used with /// `LioMode::LIO_WAIT` to block on the result of several independent /// operations. Used that way, it is often useful in programs that /// otherwise make little use of AIO. /// /// # Examples /// /// Use `listio` to submit an aio operation and wait for its completion. In /// this case, there is no need to use [`aio_suspend`] to wait or /// [`AioCb::error`] to poll. /// /// ``` /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::os::unix::io::AsRawFd; /// # use tempfile::tempfile; /// # fn main() { /// const WBUF: &[u8] = b"abcdef123456"; /// let mut f = tempfile().unwrap(); /// let mut liocb = LioCbBuilder::with_capacity(1) /// .emplace_slice( /// f.as_raw_fd(), /// 2, //offset /// WBUF, /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_WRITE /// ).finish(); /// liocb.listio(LioMode::LIO_WAIT, /// SigevNotify::SigevNone).unwrap(); /// assert_eq!(liocb.aio_return(0).unwrap() as usize, WBUF.len()); /// # } /// ``` /// /// # References /// /// [`lio_listio`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/lio_listio.html) /// /// [`aio_suspend`]: fn.aio_suspend.html /// [`AioCb::error`]: struct.AioCb.html#method.error pub fn listio(&mut self, mode: LioMode, sigev_notify: SigevNotify) -> Result<()> { let sigev = SigEvent::new(sigev_notify); let sigevp = &mut sigev.sigevent() as *mut libc::sigevent; self.list.clear(); for a in &mut self.aiocbs.iter_mut() { a.in_progress = true; self.list.push(a as *mut AioCb<'a> as *mut libc::aiocb); } let p = self.list.as_ptr(); Errno::result(unsafe { libc::lio_listio(mode as i32, p, self.list.len() as i32, sigevp) }).map(drop) } /// Resubmits any incomplete operations with [`lio_listio`]. /// /// Sometimes, due to system resource limitations, an `lio_listio` call will /// return `EIO`, or `EAGAIN`. Or, if a signal is received, it may return /// `EINTR`. In any of these cases, only a subset of its constituent /// operations will actually have been initiated. `listio_resubmit` will /// resubmit any operations that are still uninitiated. /// /// After calling `listio_resubmit`, results should be collected by /// [`LioCb::aio_return`]. /// /// # Examples /// ```no_run /// # use nix::Error; /// # use nix::errno::Errno; /// # use nix::sys::aio::*; /// # use nix::sys::signal::SigevNotify; /// # use std::os::unix::io::AsRawFd; /// # use std::{thread, time}; /// # use tempfile::tempfile; /// # fn main() { /// const WBUF: &[u8] = b"abcdef123456"; /// let mut f = tempfile().unwrap(); /// let mut liocb = LioCbBuilder::with_capacity(1) /// .emplace_slice( /// f.as_raw_fd(), /// 2, //offset /// WBUF, /// 0, //priority /// SigevNotify::SigevNone, /// LioOpcode::LIO_WRITE /// ).finish(); /// let mut err = liocb.listio(LioMode::LIO_WAIT, SigevNotify::SigevNone); /// while err == Err(Error::from(Errno::EIO)) || /// err == Err(Error::from(Errno::EAGAIN)) { /// thread::sleep(time::Duration::from_millis(10)); /// err = liocb.listio_resubmit(LioMode::LIO_WAIT, SigevNotify::SigevNone); /// } /// assert_eq!(liocb.aio_return(0).unwrap() as usize, WBUF.len()); /// # } /// ``` /// /// # References /// /// [`lio_listio`](https://pubs.opengroup.org/onlinepubs/9699919799/functions/lio_listio.html) /// /// [`lio_listio`]: https://pubs.opengroup.org/onlinepubs/9699919799/functions/lio_listio.html /// [`LioCb::aio_return`]: struct.LioCb.html#method.aio_return // Note: the addresses of any EINPROGRESS or EOK aiocbs _must_ not be // changed by this method, because the kernel relies on their addresses // being stable. // Note: aiocbs that are Ok(()) must be finalized by aio_return, or else the // sigev_notify will immediately refire. pub fn listio_resubmit(&mut self, mode:LioMode, sigev_notify: SigevNotify) -> Result<()> { let sigev = SigEvent::new(sigev_notify); let sigevp = &mut sigev.sigevent() as *mut libc::sigevent; self.list.clear(); while self.results.len() < self.aiocbs.len() { self.results.push(None); } for (i, a) in self.aiocbs.iter_mut().enumerate() { if self.results[i].is_some() { // Already collected final status for this operation continue; } match a.error_unpinned() { Ok(()) => { // aiocb is complete; collect its status and don't resubmit self.results[i] = Some(a.aio_return_unpinned()); }, Err(Errno::EAGAIN) => { self.list.push(a as *mut AioCb<'a> as *mut libc::aiocb); }, Err(Errno::EINPROGRESS) => { // aiocb is was successfully queued; no need to do anything }, Err(Errno::EINVAL) => panic!( "AioCb was never submitted, or already finalized"), _ => unreachable!() } } let p = self.list.as_ptr(); Errno::result(unsafe { libc::lio_listio(mode as i32, p, self.list.len() as i32, sigevp) }).map(drop) } /// Collect final status for an individual `AioCb` submitted as part of an /// `LioCb`. /// /// This is just like [`AioCb::aio_return`], except it takes into account /// operations that were restarted by [`LioCb::listio_resubmit`] /// /// [`AioCb::aio_return`]: struct.AioCb.html#method.aio_return /// [`LioCb::listio_resubmit`]: #method.listio_resubmit pub fn aio_return(&mut self, i: usize) -> Result { if i >= self.results.len() || self.results[i].is_none() { self.aiocbs[i].aio_return_unpinned() } else { self.results[i].unwrap() } } /// Retrieve error status of an individual `AioCb` submitted as part of an /// `LioCb`. /// /// This is just like [`AioCb::error`], except it takes into account /// operations that were restarted by [`LioCb::listio_resubmit`] /// /// [`AioCb::error`]: struct.AioCb.html#method.error /// [`LioCb::listio_resubmit`]: #method.listio_resubmit pub fn error(&mut self, i: usize) -> Result<()> { if i >= self.results.len() || self.results[i].is_none() { self.aiocbs[i].error_unpinned() } else { Ok(()) } } } #[cfg(not(any(target_os = "ios", target_os = "macos")))] impl<'a> Debug for LioCb<'a> { fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { fmt.debug_struct("LioCb") .field("aiocbs", &self.aiocbs) .finish() } } /// Used to construct `LioCb` // This must be a separate class from LioCb due to pinning constraints. LioCb // must use a boxed slice of AioCbs so they will have stable storage, but // LioCbBuilder must use a Vec to make construction possible when the final size // is unknown. #[cfg(not(any(target_os = "ios", target_os = "macos")))] #[derive(Debug)] pub struct LioCbBuilder<'a> { /// A collection of [`AioCb`]s. /// /// [`AioCb`]: struct.AioCb.html pub aiocbs: Vec>, } #[cfg(not(any(target_os = "ios", target_os = "macos")))] impl<'a> LioCbBuilder<'a> { /// Initialize an empty `LioCb` pub fn with_capacity(capacity: usize) -> LioCbBuilder<'a> { LioCbBuilder { aiocbs: Vec::with_capacity(capacity), } } /// Add a new operation on an immutable slice to the [`LioCb`] under /// construction. /// /// Arguments are the same as for [`AioCb::from_slice`] /// /// [`LioCb`]: struct.LioCb.html /// [`AioCb::from_slice`]: struct.AioCb.html#method.from_slice pub fn emplace_slice(mut self, fd: RawFd, offs: off_t, buf: &'a [u8], prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> Self { self.aiocbs.push(AioCb::from_slice_unpinned(fd, offs, buf, prio, sigev_notify, opcode)); self } /// Add a new operation on a mutable slice to the [`LioCb`] under /// construction. /// /// Arguments are the same as for [`AioCb::from_mut_slice`] /// /// [`LioCb`]: struct.LioCb.html /// [`AioCb::from_mut_slice`]: struct.AioCb.html#method.from_mut_slice pub fn emplace_mut_slice(mut self, fd: RawFd, offs: off_t, buf: &'a mut [u8], prio: libc::c_int, sigev_notify: SigevNotify, opcode: LioOpcode) -> Self { self.aiocbs.push(AioCb::from_mut_slice_unpinned(fd, offs, buf, prio, sigev_notify, opcode)); self } /// Finalize this [`LioCb`]. /// /// Afterwards it will be possible to issue the operations with /// [`LioCb::listio`]. Conversely, it will no longer be possible to add new /// operations with [`LioCbBuilder::emplace_slice`] or /// [`LioCbBuilder::emplace_mut_slice`]. /// /// [`LioCb::listio`]: struct.LioCb.html#method.listio /// [`LioCb::from_mut_slice`]: struct.LioCb.html#method.from_mut_slice /// [`LioCb::from_slice`]: struct.LioCb.html#method.from_slice pub fn finish(self) -> LioCb<'a> { let len = self.aiocbs.len(); LioCb { aiocbs: self.aiocbs.into(), list: Vec::with_capacity(len), results: Vec::with_capacity(len) } } } #[cfg(not(any(target_os = "ios", target_os = "macos")))] #[cfg(test)] mod t { use super::*; // It's important that `LioCb` be `UnPin`. The tokio-file crate relies on // it. #[test] fn liocb_is_unpin() { use assert_impl::assert_impl; assert_impl!(Unpin: LioCb); } }