completion: refactor to use references instead of Arc

Author: jcsp

src/ceph.rs

@@ -960,15 +960,15 @@ impl IoCtx {
 
     /// Async variant of rados_object_write
     pub async fn rados_async_object_write(
-        self: &Arc<Self>,
+        &self,
         object_name: &str,
         buffer: &[u8],
         offset: u64,
     ) -> RadosResult<i32> {
         self.ioctx_guard()?;
         let obj_name_str = CString::new(object_name)?;
 
-        with_completion(self.clone(), |c| unsafe {
+        with_completion(self, |c| unsafe {
             rados_aio_write(
                 self.ioctx,
                 obj_name_str.as_ptr(),
@@ -977,67 +977,67 @@ impl IoCtx {
                 buffer.len(),
                 offset,
             )
-        })
+        })?
         .await
     }
 
     /// Async variant of rados_object_append
     pub async fn rados_async_object_append(
-        self: &Arc<Self>,
+        &self,
         object_name: &str,
         buffer: &[u8],
     ) -> RadosResult<i32> {
         self.ioctx_guard()?;
         let obj_name_str = CString::new(object_name)?;
 
-        with_completion(self.clone(), |c| unsafe {
+        with_completion(self, |c| unsafe {
             rados_aio_append(
                 self.ioctx,
                 obj_name_str.as_ptr(),
                 c,
                 buffer.as_ptr() as *const ::libc::c_char,
                 buffer.len(),
             )
-        })
+        })?
         .await
     }
 
     /// Async variant of rados_object_write_full
     pub async fn rados_async_object_write_full(
-        self: &Arc<Self>,
+        &self,
         object_name: &str,
         buffer: &[u8],
     ) -> RadosResult<i32> {
         self.ioctx_guard()?;
         let obj_name_str = CString::new(object_name)?;
 
-        with_completion(self.clone(), |c| unsafe {
+        with_completion(self, |c| unsafe {
             rados_aio_write_full(
                 self.ioctx,
                 obj_name_str.as_ptr(),
                 c,
                 buffer.as_ptr() as *const ::libc::c_char,
                 buffer.len(),
             )
-        })
+        })?
         .await
     }
 
     /// Async variant of rados_object_remove
-    pub async fn rados_async_object_remove(self: &Arc<Self>, object_name: &str) -> RadosResult<()> {
+    pub async fn rados_async_object_remove(&self, object_name: &str) -> RadosResult<()> {
         self.ioctx_guard()?;
         let object_name_str = CString::new(object_name)?;
 
-        with_completion(self.clone(), |c| unsafe {
+        with_completion(self, |c| unsafe {
             rados_aio_remove(self.ioctx, object_name_str.as_ptr() as *const c_char, c)
-        })
+        })?
         .await
         .map(|_r| ())
     }
 
     /// Async variant of rados_object_read
     pub async fn rados_async_object_read(
-        self: &Arc<Self>,
+        &self,
         object_name: &str,
         fill_buffer: &mut Vec<u8>,
         read_offset: u64,
@@ -1049,7 +1049,7 @@ impl IoCtx {
             fill_buffer.reserve_exact(DEFAULT_READ_BYTES);
         }
 
-        with_completion(self.clone(), |c| unsafe {
+        with_completion(self, |c| unsafe {
             rados_aio_read(
                 self.ioctx,
                 obj_name_str.as_ptr(),
@@ -1058,7 +1058,7 @@ impl IoCtx {
                 fill_buffer.capacity(),
                 read_offset,
             )
-        })
+        })?
         .await
     }
 

src/completion.rs

@@ -14,7 +14,7 @@
 
 use std::ffi::c_void;
 use std::pin::Pin;
-use std::sync::{Arc, Mutex};
+use std::sync::Mutex;
 use std::task::{Context, Poll, Waker};
 
 use crate::ceph::IoCtx;
@@ -25,7 +25,7 @@ use crate::rados::{
     rados_completion_t,
 };
 
-struct Completion {
+pub(crate) struct Completion<'a> {
     inner: rados_completion_t,
 
     // Box to provide a stable address for completion_complete callback
@@ -34,15 +34,15 @@ struct Completion {
 
     // A reference to the IOCtx is required to issue a cancel on
     // the operation if we are dropped before ready.  This needs
-    // to be an Arc rather than a raw rados_ioctx_t because otherwise
+    // to be a Rust reference rather than a raw rados_ioctx_t because otherwise
     // there would be nothing to stop the rados_ioctx_t being invalidated
     // during the lifetime of this Completion.
     // (AioCompletionImpl does hold a reference to IoCtxImpl for writes, but
     //  not for reads.)
-    ioctx: Arc<IoCtx>,
+    ioctx: &'a IoCtx,
 }
 
-unsafe impl Send for Completion {}
+unsafe impl Send for Completion<'_> {}
 
 #[no_mangle]
 pub extern "C" fn completion_complete(_cb: rados_completion_t, arg: *mut c_void) -> () {
@@ -58,7 +58,7 @@ pub extern "C" fn completion_complete(_cb: rados_completion_t, arg: *mut c_void)
     }
 }
 
-impl Drop for Completion {
+impl Drop for Completion<'_> {
     fn drop(&mut self) {
         // Ensure that after dropping the Completion, the AIO callback
         // will not be called on our dropped waker Box.  Only necessary
@@ -67,7 +67,11 @@ impl Drop for Completion {
         let am_complete = unsafe { rados_aio_is_complete(self.inner) } != 0;
         if !am_complete {
             unsafe {
-                rados_aio_cancel(self.ioctx.ioctx, self.inner);
+                let cancel_r = rados_aio_cancel(self.ioctx.ioctx, self.inner);
+
+                // It is unsound to proceed if the Objecter op is still in flight
+                assert!(cancel_r == 0 || cancel_r == -libc::ENOENT);
+
                 rados_aio_wait_for_complete_and_cb(self.inner);
             }
         }
@@ -78,7 +82,7 @@ impl Drop for Completion {
     }
 }
 
-impl std::future::Future for Completion {
+impl std::future::Future for Completion<'_> {
     type Output = crate::error::RadosResult<i32>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
@@ -97,7 +101,10 @@ impl std::future::Future for Completion {
     }
 }
 
-fn with_completion_impl<F>(ioctx: Arc<IoCtx>, f: F) -> RadosResult<Completion>
+/// Completions are only created via this wrapper, in order to ensure
+/// that the Completion struct is only constructed around 'armed' rados_completion_t
+/// instances (i.e. those that have been used to start an I/O).
+pub(crate) fn with_completion<F>(ioctx: &IoCtx, f: F) -> RadosResult<Completion<'_>>
 where
     F: FnOnce(rados_completion_t) -> libc::c_int,
 {
@@ -134,16 +141,3 @@ where
         })
     }
 }
-/// Completions are only created via this wrapper, in order to ensure
-/// that the Completion struct is only constructed around 'armed' rados_completion_t
-/// instances (i.e. those that have been used to start an I/O).
-pub async fn with_completion<F>(ioctx: Arc<IoCtx>, f: F) -> RadosResult<i32>
-where
-    F: FnOnce(rados_completion_t) -> libc::c_int,
-{
-    // Hide c_void* temporaries in a non-async function so that the future generated
-    // by this function isn't encumbered by their non-Send-ness.
-    let completion = with_completion_impl(ioctx, f)?;
-
-    completion.await
-}