Merge #820

820: Change AioCb to primarily use Bytes instead of Rc<[u8]> r=Susurrus a=asomers

`Rc<[u8]>` isn't a very good buffer type to use for aio.  For one thing, it lacks interior mutability.  For another, a single `Rc<[u8]>` can't be carved up into smaller buffers of the same type.  `Bytes` and `BytesMut` fix both problems.  This PR removes the ability to construct an `AioCb` from `Rc<[u8]>` and adds the ability to construct one from `Bytes`, `BytesMut`, or raw pointers (for consumers who need even more flexibility).  At this stage, the PR has the following warts:

1. A hack is necessary to force small `Bytes` buffers to allocate on the heap.  I plan to fix this with an enhancement to the bytes crate.
2. The `AioCb::buffer` method is necessary due to a deficiency in the tokio-core crate.  Once I fix that, then only `AioCb::into_buffer`will need to be public.

1 files changed, 182 insertions, 32 deletions

diff --git a/test/sys/test_aio.rs b/test/sys/test_aio.rs
index 1f807585..f88fc268 100644
--- a/test/sys/test_aio.rs
+++ b/test/sys/test_aio.rs
@@ -1,4 +1,5 @@
-use libc::c_int;
+use bytes::{Bytes, BytesMut};
+use libc::{c_int, c_void};
 use nix::{Error, Result};
 use nix::errno::*;
 use nix::sys::aio::*;
@@ -7,7 +8,6 @@ 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::rc::Rc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::{thread, time};
 use tempfile::tempfile;
@@ -43,8 +43,8 @@ fn test_accessors() {
     assert_eq!(99, sev.sigev_value.sival_ptr as i64);
 }
 
-// Tests AioCb.cancel.  We aren't trying to test the OS's implementation, only our
-// bindings.  So it's sufficient to check that AioCb.cancel returned any
+// Tests AioCb.cancel.  We aren't trying to test the OS's implementation, only
+// our bindings.  So it's sufficient to check that AioCb.cancel returned any
 // AioCancelStat value.
 #[test]
 #[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
@@ -110,7 +110,7 @@ fn test_fsync() {
     aiocb.aio_return().unwrap();
 }
 
-/// `AioCb::fsync` should not modify the `AioCb` object if libc::aio_fsync returns
+/// `AioCb::fsync` should not modify the `AioCb` object if `libc::aio_fsync` returns
 /// an error
 // Skip on Linux, because Linux's AIO implementation can't detect errors
 // synchronously
@@ -135,9 +135,10 @@ fn test_fsync_error() {
 #[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
 fn test_aio_suspend() {
     const INITIAL: &[u8] = b"abcdef123456";
-    const WBUF: &[u8] = b"CDEF";
+    const WBUF: &[u8] = b"CDEFG";
     let timeout = TimeSpec::seconds(10);
-    let rbuf = Rc::new(vec![0; 4].into_boxed_slice());
+    let mut rbuf = vec![0; 4];
+    let rlen = rbuf.len();
     let mut f = tempfile().unwrap();
     f.write_all(INITIAL).unwrap();
 
@@ -148,9 +149,9 @@ fn test_aio_suspend() {
                            SigevNotify::SigevNone,
                            LioOpcode::LIO_WRITE);
 
-    let mut rcb = AioCb::from_boxed_slice( f.as_raw_fd(),
+    let mut rcb = AioCb::from_mut_slice( f.as_raw_fd(),
                             8,   //offset
-                            Rc::clone(&rbuf),
+                            &mut rbuf,
                             0,   //priority
                             SigevNotify::SigevNone,
                             LioOpcode::LIO_READ);
@@ -168,7 +169,7 @@ fn test_aio_suspend() {
     }
 
     assert!(wcb.aio_return().unwrap() as usize == WBUF.len());
-    assert!(rcb.aio_return().unwrap() as usize == WBUF.len());
+    assert!(rcb.aio_return().unwrap() as usize == rlen);
 }
 
 // Test a simple aio operation with no completion notification.  We must poll
@@ -177,14 +178,14 @@ fn test_aio_suspend() {
 #[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
 fn test_read() {
     const INITIAL: &[u8] = b"abcdef123456";
-    let rbuf = Rc::new(vec![0; 4].into_boxed_slice());
+    let mut rbuf = vec![0; 4];
     const EXPECT: &[u8] = b"cdef";
     let mut f = tempfile().unwrap();
     f.write_all(INITIAL).unwrap();
     {
-        let mut aiocb = AioCb::from_boxed_slice( f.as_raw_fd(),
+        let mut aiocb = AioCb::from_mut_slice( f.as_raw_fd(),
                                2,   //offset
-                               Rc::clone(&rbuf),
+                               &mut rbuf,
                                0,   //priority
                                SigevNotify::SigevNone,
                                LioOpcode::LIO_NOP);
@@ -198,20 +199,20 @@ fn test_read() {
     assert!(EXPECT == rbuf.deref().deref());
 }
 
-/// `AioCb::read` should not modify the `AioCb` object if libc::aio_read returns
-/// an error
+/// `AioCb::read` should not modify the `AioCb` object if `libc::aio_read`
+/// returns an error
 // Skip on Linux, because Linux's AIO implementation can't detect errors
 // synchronously
 #[test]
 #[cfg(any(target_os = "freebsd", target_os = "macos"))]
 fn test_read_error() {
     const INITIAL: &[u8] = b"abcdef123456";
-    let rbuf = Rc::new(vec![0; 4].into_boxed_slice());
+    let mut rbuf = vec![0; 4];
     let mut f = tempfile().unwrap();
     f.write_all(INITIAL).unwrap();
-    let mut aiocb = AioCb::from_boxed_slice( f.as_raw_fd(),
+    let mut aiocb = AioCb::from_mut_slice( f.as_raw_fd(),
                            -1,   //an invalid offset
-                           Rc::clone(&rbuf),
+                           &mut rbuf,
                            0,   //priority
                            SigevNotify::SigevNone,
                            LioOpcode::LIO_NOP);
@@ -244,6 +245,36 @@ fn test_read_into_mut_slice() {
     assert!(rbuf == EXPECT);
 }
 
+// Tests from_ptr
+#[test]
+#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
+fn test_read_into_pointer() {
+    const INITIAL: &[u8] = b"abcdef123456";
+    let mut rbuf = vec![0; 4];
+    const EXPECT: &[u8] = b"cdef";
+    let mut f = tempfile().unwrap();
+    f.write_all(INITIAL).unwrap();
+    {
+        // Safety: ok because rbuf lives until after poll_aio
+        let mut aiocb = unsafe {
+            AioCb::from_mut_ptr( f.as_raw_fd(),
+                                 2,   //offset
+                                 rbuf.as_mut_ptr() as *mut c_void,
+                                 rbuf.len(),
+                                 0,   //priority
+                                 SigevNotify::SigevNone,
+                                 LioOpcode::LIO_NOP)
+        };
+        aiocb.read().unwrap();
+
+        let err = poll_aio(&mut aiocb);
+        assert!(err == Ok(()));
+        assert!(aiocb.aio_return().unwrap() as usize == EXPECT.len());
+    }
+
+    assert!(rbuf == EXPECT);
+}
+
 // Test reading into an immutable buffer.  It should fail
 // FIXME: This test fails to panic on Linux/musl
 #[test]
@@ -292,8 +323,124 @@ fn test_write() {
     assert!(rbuf == EXPECT);
 }
 
-/// `AioCb::write` should not modify the `AioCb` object if libc::aio_write returns
-/// an error
+// Tests `AioCb::from_bytes`
+#[test]
+#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
+fn test_write_bytes() {
+    const INITIAL: &[u8] = b"abcdef123456";
+    let wbuf = Bytes::from(&b"CDEF"[..]);
+    let mut rbuf = Vec::new();
+    const EXPECT: &[u8] = b"abCDEF123456";
+
+    let mut f = tempfile().unwrap();
+    f.write_all(INITIAL).unwrap();
+    let mut aiocb = AioCb::from_bytes( f.as_raw_fd(),
+                           2,   //offset
+                           wbuf.clone(),
+                           0,   //priority
+                           SigevNotify::SigevNone,
+                           LioOpcode::LIO_NOP);
+    aiocb.write().unwrap();
+
+    let err = poll_aio(&mut aiocb);
+    assert!(err == Ok(()));
+    assert!(aiocb.aio_return().unwrap() as usize == wbuf.len());
+
+    f.seek(SeekFrom::Start(0)).unwrap();
+    let len = f.read_to_end(&mut rbuf).unwrap();
+    assert!(len == EXPECT.len());
+    assert!(rbuf == EXPECT);
+}
+
+// Tests `AioCb::from_bytes_mut`
+#[test]
+#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
+fn test_read_bytes_mut_big() {
+    const INITIAL: &[u8] = b"abcdefgh12345678abcdefgh12345678abcdefgh12345678abcdefgh12345678abcdefgh12345678";
+    // rbuf needs to be larger than 32 bytes (64 on 32-bit systems) so
+    // BytesMut::clone is implemented by reference.
+    let rbuf = BytesMut::from(vec![0; 70]);
+    const EXPECT: &[u8] = b"cdefgh12345678abcdefgh12345678abcdefgh12345678abcdefgh12345678abcdefgh";
+    let mut f = tempfile().unwrap();
+    f.write_all(INITIAL).unwrap();
+
+    let mut aiocb = AioCb::from_bytes_mut( 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.into_buffer();
+    assert_eq!(buffer.bytes_mut().unwrap(), EXPECT);
+}
+
+// Tests reallocation in `AioCb::from_bytes_mut`
+#[test]
+#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
+fn test_read_bytes_mut_small() {
+    const INITIAL: &[u8] = b"abcdef";
+    // rbuf needs to be no more than 32 bytes (64 on 32-bit systems) so
+    // BytesMut::clone is implemented inline.
+    let rbuf = 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_bytes_mut( 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.into_buffer();
+    assert_eq!(buffer.bytes_mut().unwrap(), EXPECT);
+}
+
+// Tests `AioCb::from_ptr`
+#[test]
+#[cfg_attr(all(target_env = "musl", target_arch = "x86_64"), ignore)]
+fn test_write_from_pointer() {
+    const INITIAL: &[u8] = b"abcdef123456";
+    let wbuf = "CDEF".to_string().into_bytes();
+    let mut rbuf = Vec::new();
+    const EXPECT: &[u8] = b"abCDEF123456";
+
+    let mut f = tempfile().unwrap();
+    f.write_all(INITIAL).unwrap();
+    // Safety: ok because aiocb outlives poll_aio
+    let mut aiocb = unsafe {
+        AioCb::from_ptr( f.as_raw_fd(),
+                         2,   //offset
+                         wbuf.as_ptr() as *const c_void,
+                         wbuf.len(),
+                         0,   //priority
+                         SigevNotify::SigevNone,
+                         LioOpcode::LIO_NOP)
+    };
+    aiocb.write().unwrap();
+
+    let err = poll_aio(&mut aiocb);
+    assert!(err == Ok(()));
+    assert!(aiocb.aio_return().unwrap() as usize == wbuf.len());
+
+    f.seek(SeekFrom::Start(0)).unwrap();
+    let len = f.read_to_end(&mut rbuf).unwrap();
+    assert!(len == EXPECT.len());
+    assert!(rbuf == EXPECT);
+}
+
+/// `AioCb::write` should not modify the `AioCb` object if `libc::aio_write`
+/// returns an error
 // Skip on Linux, because Linux's AIO implementation can't detect errors
 // synchronously
 #[test]
@@ -366,7 +513,8 @@ fn test_write_sigev_signal() {
 fn test_lio_listio_wait() {
     const INITIAL: &[u8] = b"abcdef123456";
     const WBUF: &[u8] = b"CDEF";
-    let rbuf = Rc::new(vec![0; 4].into_boxed_slice());
+    let mut rbuf = vec![0; 4];
+    let rlen = rbuf.len();
     let mut rbuf2 = Vec::new();
     const EXPECT: &[u8] = b"abCDEF123456";
     let mut f = tempfile().unwrap();
@@ -381,9 +529,9 @@ fn test_lio_listio_wait() {
                                SigevNotify::SigevNone,
                                LioOpcode::LIO_WRITE);
 
-        let mut rcb = AioCb::from_boxed_slice( f.as_raw_fd(),
+        let mut rcb = AioCb::from_mut_slice( f.as_raw_fd(),
                                 8,   //offset
-                                Rc::clone(&rbuf),
+                                &mut rbuf,
                                 0,   //priority
                                 SigevNotify::SigevNone,
                                 LioOpcode::LIO_READ);
@@ -391,7 +539,7 @@ fn test_lio_listio_wait() {
         err.expect("lio_listio failed");
 
         assert!(wcb.aio_return().unwrap() as usize == WBUF.len());
-        assert!(rcb.aio_return().unwrap() as usize == WBUF.len());
+        assert!(rcb.aio_return().unwrap() as usize == rlen);
     }
     assert!(rbuf.deref().deref() == b"3456");
 
@@ -409,7 +557,8 @@ fn test_lio_listio_wait() {
 fn test_lio_listio_nowait() {
     const INITIAL: &[u8] = b"abcdef123456";
     const WBUF: &[u8] = b"CDEF";
-    let rbuf = Rc::new(vec![0; 4].into_boxed_slice());
+    let mut rbuf = vec![0; 4];
+    let rlen = rbuf.len();
     let mut rbuf2 = Vec::new();
     const EXPECT: &[u8] = b"abCDEF123456";
     let mut f = tempfile().unwrap();
@@ -424,9 +573,9 @@ fn test_lio_listio_nowait() {
                                SigevNotify::SigevNone,
                                LioOpcode::LIO_WRITE);
 
-        let mut rcb = AioCb::from_boxed_slice( f.as_raw_fd(),
+        let mut rcb = AioCb::from_mut_slice( f.as_raw_fd(),
                                 8,   //offset
-                                Rc::clone(&rbuf),
+                                &mut rbuf,
                                 0,   //priority
                                 SigevNotify::SigevNone,
                                 LioOpcode::LIO_READ);
@@ -436,7 +585,7 @@ fn test_lio_listio_nowait() {
         poll_aio(&mut wcb).unwrap();
         poll_aio(&mut rcb).unwrap();
         assert!(wcb.aio_return().unwrap() as usize == WBUF.len());
-        assert!(rcb.aio_return().unwrap() as usize == WBUF.len());
+        assert!(rcb.aio_return().unwrap() as usize == rlen);
     }
     assert!(rbuf.deref().deref() == b"3456");
 
@@ -457,7 +606,8 @@ fn test_lio_listio_signal() {
     let m = ::SIGNAL_MTX.lock().expect("Mutex got poisoned by another test");
     const INITIAL: &[u8] = b"abcdef123456";
     const WBUF: &[u8] = b"CDEF";
-    let rbuf = Rc::new(vec![0; 4].into_boxed_slice());
+    let mut rbuf = vec![0; 4];
+    let rlen = rbuf.len();
     let mut rbuf2 = Vec::new();
     const EXPECT: &[u8] = b"abCDEF123456";
     let mut f = tempfile().unwrap();
@@ -477,9 +627,9 @@ fn test_lio_listio_signal() {
                                SigevNotify::SigevNone,
                                LioOpcode::LIO_WRITE);
 
-        let mut rcb = AioCb::from_boxed_slice( f.as_raw_fd(),
+        let mut rcb = AioCb::from_mut_slice( f.as_raw_fd(),
                                 8,   //offset
-                                Rc::clone(&rbuf),
+                                &mut rbuf,
                                 0,   //priority
                                 SigevNotify::SigevNone,
                                 LioOpcode::LIO_READ);
@@ -492,7 +642,7 @@ fn test_lio_listio_signal() {
         }
 
         assert!(wcb.aio_return().unwrap() as usize == WBUF.len());
-        assert!(rcb.aio_return().unwrap() as usize == WBUF.len());
+        assert!(rcb.aio_return().unwrap() as usize == rlen);
     }
     assert!(rbuf.deref().deref() == b"3456");