Adapt aio to the world of async/await, and fix some potential unsoundness.

* libc::aiocb must not be moved while the kernel has a pointer to it.
  This change enforces that requirement by using std::pin.

* Split LioCbBuilder out of LioCb.  struct LioCb relied on the
  (incorrect) assumption that a Vec's elements have a stable location in
  memory.  That's not true; they can be moved during Vec::push.  The
  solution is to use a Vec in the new Builder struct, but finalize it to
  a boxed slice (which doesn't support push) before allowing it to be
  submitted to the kernel.

* Eliminate owned buffer types.  mio-aio no longer uses owned buffers
  with nix::aio.  There's little need for it in the world of
  async/await.  I'm not aware of any other consumers.  This
  substantially simplifies the code.

2 files changed, 81 insertions, 125 deletions

diff --git a/test/sys/test_aio.rs b/test/sys/test_aio.rs
index 3878da94..753f1182 100644
--- a/test/sys/test_aio.rs
+++ b/test/sys/test_aio.rs
@@ -1,4 +1,3 @@
-use bytes::{Bytes, BytesMut};
 use libc::{c_int, c_void};
 use nix::{Error, Result};
 use nix::errno::*;
@@ -8,12 +7,13 @@ use nix::sys::time::{TimeSpec, TimeValLike};
 use std::io::{Write, Read, Seek, SeekFrom};
 use std::ops::Deref;
 use std::os::unix::io::AsRawFd;
+use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::{thread, time};
 use tempfile::tempfile;
 
 // Helper that polls an AioCb for completion or error
-fn poll_aio(aiocb: &mut AioCb) -> Result<()> {
+fn poll_aio(aiocb: &mut Pin<Box<AioCb>>) -> Result<()> {
     loop {
         let err = aiocb.error();
         if err != Err(Error::from(Errno::EINPROGRESS)) { return err; };
@@ -21,6 +21,16 @@ fn poll_aio(aiocb: &mut AioCb) -> Result<()> {
     }
 }
 
+// Helper that polls a component of an LioCb for completion or error
+#[cfg(not(any(target_os = "ios", target_os = "macos")))]
+fn poll_lio(liocb: &mut LioCb, i: usize) -> Result<()> {
+    loop {
+        let err = liocb.error(i);
+        if err != Err(Error::from(Errno::EINPROGRESS)) { return err; };
+        thread::sleep(time::Duration::from_millis(10));
+    }
+}
+
 #[test]
 fn test_accessors() {
     let mut rbuf = vec![0; 4];
@@ -166,7 +176,7 @@ fn test_aio_suspend() {
     rcb.read().unwrap();
     loop {
         {
-            let cbbuf = [&wcb, &rcb];
+            let cbbuf = [wcb.as_ref(), rcb.as_ref()];
             let r = aio_suspend(&cbbuf[..], Some(timeout));
             match r {
                 Err(Error::Sys(Errno::EINTR)) => continue,
@@ -335,61 +345,6 @@ fn test_write() {
     assert_eq!(rbuf, EXPECT);
 }
 
-// Tests `AioCb::from_boxed_slice` with `Bytes`
-#[test]
-#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
-fn test_write_bytes() {
-    const INITIAL: &[u8] = b"abcdef123456";
-    let wbuf = Box::new(Bytes::from(&b"CDEF"[..]));
-    let mut rbuf = Vec::new();
-    const EXPECT: &[u8] = b"abCDEF123456";
-    let expected_len = wbuf.len();
-
-    let mut f = tempfile().unwrap();
-    f.write_all(INITIAL).unwrap();
-    let mut aiocb = AioCb::from_boxed_slice( f.as_raw_fd(),
-                           2,   //offset
-                           wbuf,
-                           0,   //priority
-                           SigevNotify::SigevNone,
-                           LioOpcode::LIO_NOP);
-    aiocb.write().unwrap();
-
-    let err = poll_aio(&mut aiocb);
-    assert_eq!(err, Ok(()));
-    assert_eq!(aiocb.aio_return().unwrap() as usize, expected_len);
-
-    f.seek(SeekFrom::Start(0)).unwrap();
-    let len = f.read_to_end(&mut rbuf).unwrap();
-    assert_eq!(len, EXPECT.len());
-    assert_eq!(rbuf, EXPECT);
-}
-
-// Tests `AioCb::from_boxed_mut_slice` with `BytesMut`
-#[test]
-#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
-fn test_read_bytes_mut_small() {
-    const INITIAL: &[u8] = b"abcdef";
-    let rbuf = Box::new(BytesMut::from(vec![0; 4]));
-    const EXPECT: &[u8] = b"cdef";
-    let mut f = tempfile().unwrap();
-    f.write_all(INITIAL).unwrap();
-
-    let mut aiocb = AioCb::from_boxed_mut_slice( f.as_raw_fd(),
-                           2,   //offset
-                           rbuf,
-                           0,   //priority
-                           SigevNotify::SigevNone,
-                           LioOpcode::LIO_NOP);
-    aiocb.read().unwrap();
-
-    let err = poll_aio(&mut aiocb);
-    assert_eq!(err, Ok(()));
-    assert_eq!(aiocb.aio_return().unwrap() as usize, EXPECT.len());
-    let buffer = aiocb.boxed_mut_slice().unwrap();
-    assert_eq!(buffer.borrow(), EXPECT);
-}
-
 // Tests `AioCb::from_ptr`
 #[test]
 #[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
@@ -505,22 +460,22 @@ fn test_liocb_listio_wait() {
     f.write_all(INITIAL).unwrap();
 
     {
-        let wcb = AioCb::from_slice( f.as_raw_fd(),
-                               2,   //offset
-                               WBUF,
-                               0,   //priority
-                               SigevNotify::SigevNone,
-                               LioOpcode::LIO_WRITE);
-
-        let rcb = AioCb::from_mut_slice( f.as_raw_fd(),
-                                8,   //offset
-                                &mut rbuf,
-                                0,   //priority
-                                SigevNotify::SigevNone,
-                                LioOpcode::LIO_READ);
-        let mut liocb = LioCb::with_capacity(2);
-        liocb.aiocbs.push(wcb);
-        liocb.aiocbs.push(rcb);
+        let mut liocb = LioCbBuilder::with_capacity(2)
+            .emplace_slice(
+                f.as_raw_fd(),
+                2,   //offset
+                WBUF,
+                0,   //priority
+                SigevNotify::SigevNone,
+                LioOpcode::LIO_WRITE
+            ).emplace_mut_slice(
+                f.as_raw_fd(),
+                8,   //offset
+                &mut rbuf,
+                0,   //priority
+                SigevNotify::SigevNone,
+                LioOpcode::LIO_READ
+            ).finish();
         let err = liocb.listio(LioMode::LIO_WAIT, SigevNotify::SigevNone);
         err.expect("lio_listio");
 
@@ -552,29 +507,29 @@ fn test_liocb_listio_nowait() {
     f.write_all(INITIAL).unwrap();
 
     {
-        let wcb = AioCb::from_slice( f.as_raw_fd(),
-                               2,   //offset
-                               WBUF,
-                               0,   //priority
-                               SigevNotify::SigevNone,
-                               LioOpcode::LIO_WRITE);
-
-        let rcb = AioCb::from_mut_slice( f.as_raw_fd(),
-                                8,   //offset
-                                &mut rbuf,
-                                0,   //priority
-                                SigevNotify::SigevNone,
-                                LioOpcode::LIO_READ);
-        let mut liocb = LioCb::with_capacity(2);
-        liocb.aiocbs.push(wcb);
-        liocb.aiocbs.push(rcb);
+        let mut liocb = LioCbBuilder::with_capacity(2)
+            .emplace_slice(
+                f.as_raw_fd(),
+                2,   //offset
+                WBUF,
+                0,   //priority
+                SigevNotify::SigevNone,
+                LioOpcode::LIO_WRITE
+            ).emplace_mut_slice(
+                f.as_raw_fd(),
+                8,   //offset
+                &mut rbuf,
+                0,   //priority
+                SigevNotify::SigevNone,
+                LioOpcode::LIO_READ
+            ).finish();
         let err = liocb.listio(LioMode::LIO_NOWAIT, SigevNotify::SigevNone);
         err.expect("lio_listio");
 
-        poll_aio(&mut liocb.aiocbs[0]).unwrap();
-        poll_aio(&mut liocb.aiocbs[1]).unwrap();
-        assert_eq!(liocb.aiocbs[0].aio_return().unwrap() as usize, WBUF.len());
-        assert_eq!(liocb.aiocbs[1].aio_return().unwrap() as usize, rlen);
+        poll_lio(&mut liocb, 0).unwrap();
+        poll_lio(&mut liocb, 1).unwrap();
+        assert_eq!(liocb.aio_return(0).unwrap() as usize, WBUF.len());
+        assert_eq!(liocb.aio_return(1).unwrap() as usize, rlen);
     }
     assert_eq!(rbuf.deref().deref(), b"3456");
 
@@ -608,22 +563,22 @@ fn test_liocb_listio_signal() {
     f.write_all(INITIAL).unwrap();
 
     {
-        let wcb = AioCb::from_slice( f.as_raw_fd(),
-                               2,   //offset
-                               WBUF,
-                               0,   //priority
-                               SigevNotify::SigevNone,
-                               LioOpcode::LIO_WRITE);
-
-        let rcb = AioCb::from_mut_slice( f.as_raw_fd(),
-                                8,   //offset
-                                &mut rbuf,
-                                0,   //priority
-                                SigevNotify::SigevNone,
-                                LioOpcode::LIO_READ);
-        let mut liocb = LioCb::with_capacity(2);
-        liocb.aiocbs.push(wcb);
-        liocb.aiocbs.push(rcb);
+        let mut liocb = LioCbBuilder::with_capacity(2)
+            .emplace_slice(
+                f.as_raw_fd(),
+                2,   //offset
+                WBUF,
+                0,   //priority
+                SigevNotify::SigevNone,
+                LioOpcode::LIO_WRITE
+            ).emplace_mut_slice(
+                f.as_raw_fd(),
+                8,   //offset
+                &mut rbuf,
+                0,   //priority
+                SigevNotify::SigevNone,
+                LioOpcode::LIO_READ
+            ).finish();
         SIGNALED.store(false, Ordering::Relaxed);
         unsafe { sigaction(Signal::SIGUSR2, &sa) }.unwrap();
         let err = liocb.listio(LioMode::LIO_NOWAIT, sigev_notify);
@@ -632,8 +587,8 @@ fn test_liocb_listio_signal() {
             thread::sleep(time::Duration::from_millis(10));
         }
 
-        assert_eq!(liocb.aiocbs[0].aio_return().unwrap() as usize, WBUF.len());
-        assert_eq!(liocb.aiocbs[1].aio_return().unwrap() as usize, rlen);
+        assert_eq!(liocb.aio_return(0).unwrap() as usize, WBUF.len());
+        assert_eq!(liocb.aio_return(1).unwrap() as usize, rlen);
     }
     assert_eq!(rbuf.deref().deref(), b"3456");
 
@@ -654,13 +609,14 @@ fn test_liocb_listio_read_immutable() {
     let f = tempfile().unwrap();
 
 
-    let mut liocb = LioCb::from(vec![
-        AioCb::from_slice( f.as_raw_fd(),
+    let mut liocb = LioCbBuilder::with_capacity(1)
+        .emplace_slice(
+            f.as_raw_fd(),
             2,   //offset
             rbuf,
             0,   //priority
             SigevNotify::SigevNone,
-            LioOpcode::LIO_READ)
-    ]);
+            LioOpcode::LIO_READ
+        ).finish();
     let _ = liocb.listio(LioMode::LIO_NOWAIT, SigevNotify::SigevNone);
 }
diff --git a/test/sys/test_lio_listio_resubmit.rs b/test/sys/test_lio_listio_resubmit.rs
index 0795370b..6b33aa4f 100644
--- a/test/sys/test_lio_listio_resubmit.rs
+++ b/test/sys/test_lio_listio_resubmit.rs
@@ -20,7 +20,7 @@ const BYTES_PER_OP: usize = 512;
 /// Attempt to collect final status for all of `liocb`'s operations, freeing
 /// system resources
 fn finish_liocb(liocb: &mut LioCb) {
-    for j in 0..liocb.aiocbs.len() {
+    for j in 0..liocb.len() {
         loop {
             let e = liocb.error(j);
             match e {
@@ -70,17 +70,17 @@ fn test_lio_listio_resubmit() {
     }).collect::<Vec<_>>();
 
     let mut liocbs = (0..num_listios).map(|i| {
-        let mut liocb = LioCb::with_capacity(ops_per_listio);
+        let mut builder = LioCbBuilder::with_capacity(ops_per_listio);
         for j in 0..ops_per_listio {
             let offset = (BYTES_PER_OP * (i * ops_per_listio + j)) as off_t;
-            let wcb = AioCb::from_slice( f.as_raw_fd(),
-                                   offset,
-                                   &buffer_set[i][j][..],
-                                   0,   //priority
-                                   SigevNotify::SigevNone,
-                                   LioOpcode::LIO_WRITE);
-            liocb.aiocbs.push(wcb);
+            builder = builder.emplace_slice(f.as_raw_fd(),
+                                offset,
+                                &buffer_set[i][j][..],
+                                0,   //priority
+                                SigevNotify::SigevNone,
+                                LioOpcode::LIO_WRITE);
         }
+        let mut liocb = builder.finish();
         let mut err = liocb.listio(LioMode::LIO_NOWAIT, SigevNotify::SigevNone);
         while err == Err(Error::Sys(Errno::EIO)) ||
               err == Err(Error::Sys(Errno::EAGAIN)) ||