Implement Drop for AioCb

If an AioCb has any in-kernel state, AioCb.drop will print a warning and
wait for it to complete.

Author: asomers

src/sys/aio.rs

@@ -2,6 +2,8 @@ use {Error, Errno, Result};
 use std::os::unix::io::RawFd;
 use libc::{c_void, off_t, size_t};
 use libc;
+use std::io::Write;
+use std::io::stderr;
 use std::marker::PhantomData;
 use std::mem;
 use std::ptr::{null, null_mut};
@@ -59,11 +61,12 @@ pub enum AioCancelStat {
 
 /// The basic structure used by all aio functions.  Each `aiocb` represents one
 /// I/O request.
-#[repr(C)]
 pub struct AioCb<'a> {
     aiocb: libc::aiocb,
     /// Tracks whether the buffer pointed to by aiocb.aio_buf is mutable
     mutable: bool,
+    /// Could this `AioCb` potentially have any in-kernel state?
+    in_progress: bool,
     phantom: PhantomData<&'a mut [u8]>
 }
 
@@ -83,7 +86,8 @@ impl<'a> AioCb<'a> {
         a.aio_nbytes = 0;
         a.aio_buf = null_mut();
 
-        let aiocb = AioCb { aiocb: a, mutable: false, phantom: PhantomData};
+        let aiocb = AioCb { aiocb: a, mutable: false, in_progress: false,
+                            phantom: PhantomData};
         aiocb
     }
 
@@ -109,7 +113,8 @@ impl<'a> AioCb<'a> {
         a.aio_buf = buf.as_ptr() as *mut c_void;
         a.aio_lio_opcode = opcode as ::c_int;
 
-        let aiocb = AioCb { aiocb: a, mutable: true, phantom: PhantomData};
+        let aiocb = AioCb { aiocb: a, mutable: true, in_progress: false,
+                            phantom: PhantomData};
         aiocb
     }
 
@@ -136,7 +141,8 @@ impl<'a> AioCb<'a> {
         assert!(opcode != LioOpcode::LIO_READ, "Can't read into an immutable buffer");
         a.aio_lio_opcode = opcode as ::c_int;
 
-        let aiocb = AioCb { aiocb: a, mutable: false, phantom: PhantomData};
+        let aiocb = AioCb { aiocb: a, mutable: false, in_progress: false,
+                            phantom: PhantomData};
         aiocb
     }
 
@@ -184,13 +190,15 @@ impl<'a> AioCb<'a> {
     /// An asynchronous version of `fsync`.
     pub fn fsync(&mut self, mode: AioFsyncMode) -> Result<()> {
         let p: *mut libc::aiocb = &mut self.aiocb;
+        self.in_progress = true;
         Errno::result(unsafe { libc::aio_fsync(mode as ::c_int, p) }).map(drop)
     }
 
     /// Asynchronously reads from a file descriptor into a buffer
     pub fn read(&mut self) -> Result<()> {
         assert!(self.mutable, "Can't read into an immutable buffer");
         let p: *mut libc::aiocb = &mut self.aiocb;
+        self.in_progress = true;
         Errno::result(unsafe { libc::aio_read(p) }).map(drop)
     }
 
@@ -200,12 +208,14 @@ impl<'a> AioCb<'a> {
     // Note: this should be just `return`, but that's a reserved word
     pub fn aio_return(&mut self) -> Result<isize> {
         let p: *mut libc::aiocb = &mut self.aiocb;
+        self.in_progress = false;
         Errno::result(unsafe { libc::aio_return(p) })
     }
 
     /// Asynchronously writes from a buffer to a file descriptor
     pub fn write(&mut self) -> Result<()> {
         let p: *mut libc::aiocb = &mut self.aiocb;
+        self.in_progress = true;
         Errno::result(unsafe { libc::aio_write(p) }).map(drop)
     }
 
@@ -259,3 +269,27 @@ pub fn lio_listio(mode: LioMode, list: &[&mut AioCb],
         libc::lio_listio(mode as i32, p, list.len() as i32, sigevp)
     }).map(drop)
 }
+
+impl<'a> Drop for AioCb<'a> {
+    /// If the `AioCb` has no remaining state in the kernel, just drop it.
+    /// Otherwise, collect its error and return values, so as not to leak
+    /// resources.
+    fn drop(&mut self) {
+        if self.in_progress {
+            // Well-written programs should never get here.  They should always
+            // wait for an AioCb to complete before dropping it
+            let _ = write!(stderr(), "WARNING: dropped an in-progress AioCb");
+            loop {
+                let ret = aio_suspend(&[&self], None);
+                match ret {
+                    Ok(()) => break,
+                    Err(Error::Sys(Errno::EINVAL)) => panic!(
+                        "Inconsistent AioCb.in_progress value"),
+                    Err(Error::Sys(Errno::EINTR)) => (),    // Retry interrupted syscall
+                    _ => panic!("Unexpected aio_suspend return value {:?}", ret)
+                };
+            }
+            let _ = self.aio_return();
+        }
+    }
+}

test/sys/test_aio.rs

@@ -393,3 +393,30 @@ fn test_lio_listio_read_immutable() {
                            LioOpcode::LIO_READ);
     let _ = lio_listio(LioMode::LIO_NOWAIT, &[&mut rcb], SigevNotify::SigevNone);
 }
+
+// Test dropping an AioCb that hasn't yet finished.  Behind the scenes, the
+// library should wait for the AioCb's completion.
+#[test]
+fn test_drop() {
+    const INITIAL: &'static [u8] = b"abcdef123456";
+    const WBUF: &'static [u8] = b"CDEF"; //"CDEF".to_string().into_bytes();
+    let mut rbuf = Vec::new();
+    const EXPECT: &'static [u8] = b"abCDEF123456";
+
+    let mut f = tempfile().unwrap();
+    f.write(INITIAL).unwrap();
+    {
+        let mut aiocb = AioCb::from_slice( f.as_raw_fd(),
+                               2,   //offset
+                               &WBUF,
+                               0,   //priority
+                               SigevNotify::SigevNone,
+                               LioOpcode::LIO_NOP);
+        aiocb.write().unwrap();
+    }
+
+    f.seek(SeekFrom::Start(0)).unwrap();
+    let len = f.read_to_end(&mut rbuf).unwrap();
+    assert!(len == EXPECT.len());
+    assert!(rbuf == EXPECT);
+}