Code and formater fix

Author: Lucas Paixão

src/buf/fixed/buffers.rs

@@ -1,7 +1,8 @@
 use libc::iovec;
+use std::any::Any;
 
 // Abstracts management of fixed buffers in a buffer registry.
-pub(crate) trait FixedBuffers {
+pub(crate) trait FixedBuffers: Any {
   // Provides access to the raw buffers as a slice of iovec.
   fn iovecs(&self) -> &[iovec];
 

src/buf/fixed/handle.rs

@@ -1,5 +1,7 @@
 use super::FixedBuffers;
 use crate::buf::{IoBuf, IoBufMut};
+use std::borrow::BorrowMut;
+use std::cell::RefCell;
 
 use libc::iovec;
 use std::fmt::{self, Debug};
@@ -30,19 +32,17 @@ pub(crate) struct CheckedOutBuf {
 /// [`FixedBufPool`]:     super::FixedBufPool
 ///
 pub struct FixedBuf {
-  registry: Rc<dyn FixedBuffers>,
+  registry: Rc<RefCell<dyn FixedBuffers>>,
   buf: CheckedOutBuf,
 }
 
 impl Drop for FixedBuf {
   fn drop(&mut self) {
-    let registry = Rc::get_mut(&mut self.registry);
+    let borrow_mut = self.registry.borrow_mut();
 
-    if registry.is_none() {
-      return;
-    }
+    let mut item = borrow_mut.as_ref().borrow_mut();
 
-    let registry = registry.unwrap();
+    let registry = item.borrow_mut();
     // Safety: the length of the initialized data in the buffer has been
     // maintained accordingly to the safety contracts on
     // Self::new and IoBufMut.
@@ -57,7 +57,7 @@ impl FixedBuf {
   // - the array will not be deallocated until the buffer is checked in;
   // - the data in the array must be initialized up to the number of bytes
   //   given in init_len.
-  pub(super) unsafe fn new(registry: Rc<dyn FixedBuffers>, buf: CheckedOutBuf) -> Self {
+  pub(super) unsafe fn new(registry: Rc<RefCell<dyn FixedBuffers>>, buf: CheckedOutBuf) -> Self {
     FixedBuf { registry, buf }
   }
 

src/buf/fixed/mod.rs

@@ -27,4 +27,6 @@ pub mod pool;
 pub use pool::FixedBufPool;
 
 mod registry;
+mod shared;
+
 pub use registry::FixedBufRegistry;

src/buf/fixed/plumbing/registry.rs

@@ -5,14 +5,15 @@ use libc::{iovec, UIO_MAXIOV};
 use std::cmp;
 use std::mem;
 use std::ptr;
+use std::ptr::NonNull;
 use std::slice;
 
 // Internal state shared by FixedBufRegistry and FixedBuf handles.
 pub(crate) struct Registry<T: IoBufMut> {
   // Pointer to an allocated array of iovec records referencing
   // the allocated buffers. The number of initialized records is the
   // same as the length of the states array.
-  raw_bufs: ptr::NonNull<iovec>,
+  raw_bufs: NonNull<iovec>,
   // Original capacity of raw_bufs as a Vec.
   orig_cap: usize,
   // State information on the buffers. Indices in this array correspond to
@@ -22,6 +23,17 @@ pub(crate) struct Registry<T: IoBufMut> {
   buffers: Vec<T>,
 }
 
+impl<T: IoBufMut> Default for Registry<T> {
+  fn default() -> Self {
+    Self {
+      raw_bufs: NonNull::<iovec>::dangling(),
+      orig_cap: 0,
+      states: vec![],
+      buffers: Vec::with_capacity(0),
+    }
+  }
+}
+
 // State information of a buffer in the registry,
 enum BufState {
   // The buffer is not in use.
@@ -128,7 +140,10 @@ impl<T: IoBufMut> Drop for Registry<T> {
           // from Registry ownership, rather than deallocate.
           unsafe { self.buffers[i].set_init(*init_len) };
         }
-        BufState::CheckedOut => unreachable!("all buffers must be checked in"),
+        BufState::CheckedOut => { 
+          //unreachable!("all buffers must be checked in")
+          println!("Bug");
+        },
       }
     }
 

src/buf/fixed/pool.rs

@@ -7,12 +7,14 @@
 //! by [`FixedBufPool`].
 //!
 //! [`FixedBufPool`]: self::FixedBufPool
-
 use super::plumbing;
 use super::FixedBuf;
 use crate::buf::IoBufMut;
-use crate::runtime::CONTEXT;
+use std::cell::RefCell;
 
+use crate::buf::fixed::handle::CheckedOutBuf;
+use crate::buf::fixed::plumbing::Pool;
+use crate::buf::fixed::shared::{process, register, unregister};
 use std::io;
 use std::rc::Rc;
 use std::sync::Arc;
@@ -93,7 +95,7 @@ use tokio::sync::Notify;
 /// ```
 #[derive(Clone)]
 pub struct FixedBufPool<T: IoBufMut> {
-  inner: Rc<plumbing::Pool<T>>,
+  inner: RefCell<Rc<RefCell<plumbing::Pool<T>>>>,
 }
 
 impl<T: IoBufMut> FixedBufPool<T> {
@@ -162,7 +164,7 @@ impl<T: IoBufMut> FixedBufPool<T> {
   /// ```
   pub fn new(bufs: impl IntoIterator<Item = T>) -> Self {
     FixedBufPool {
-      inner: Rc::new(plumbing::Pool::new(bufs.into_iter())),
+      inner: RefCell::new(Rc::new(RefCell::new(Pool::new(bufs.into_iter())))),
     }
   }
 
@@ -186,7 +188,7 @@ impl<T: IoBufMut> FixedBufPool<T> {
   /// of the `tokio-uring` runtime this call is made in, the function returns
   /// an error.
   pub fn register(&self) -> io::Result<()> {
-    CONTEXT.with(|x| x.handle().register_buffers(Rc::clone(&self.inner) as _))
+    register(self.inner.borrow().clone())
   }
 
   /// Unregisters this collection of buffers.
@@ -204,8 +206,9 @@ impl<T: IoBufMut> FixedBufPool<T> {
   /// of the `tokio-uring` runtime this call is made in, the function returns
   /// an error. Calling `unregister` when no `FixedBufPool` is currently
   /// registered on this runtime also returns an error.
+  #[allow(private_interfaces)]
   pub fn unregister(&self) -> io::Result<()> {
-    CONTEXT.with(|x| x.handle().unregister_buffers(Rc::clone(&self.inner) as _))
+    unregister()
   }
 
   /// Returns a buffer of requested capacity from this pool
@@ -220,14 +223,13 @@ impl<T: IoBufMut> FixedBufPool<T> {
   /// An application should not rely on any particular order
   /// in which available buffers are retrieved.
   pub fn try_next(&mut self, cap: usize) -> Option<FixedBuf> {
-    let mut inner = Rc::get_mut(&mut self.inner);
-
-    inner.as_mut()?.try_next(cap).map(|data| {
-      let pool = Rc::clone(&self.inner);
-      // Safety: the validity of buffer data is ensured by
-      // plumbing::Pool::try_next
-      unsafe { FixedBuf::new(pool, data) }
-    })
+    let cob = {
+      let inner = self.inner.borrow_mut();
+      let mut inner = inner.borrow_mut();
+      inner.try_next(cap)
+    };
+    let current = self.inner.borrow().clone();
+    process(current, cob)
   }
 
   /// Resolves to a buffer of requested capacity
@@ -243,12 +245,11 @@ impl<T: IoBufMut> FixedBufPool<T> {
   pub async fn next(&mut self, cap: usize) -> FixedBuf {
     // Fast path: get the buffer if it's already available
     let notify = {
-      let inner = Rc::get_mut(&mut self.inner).unwrap();
+      let item = self.inner.borrow();
+      let mut inner = item.borrow_mut();
       if let Some(data) = inner.try_next(cap) {
-        // Safety: the validity of buffer data is ensured by
-        // plumbing::Pool::try_next
-        let buf = unsafe { FixedBuf::new(Rc::clone(&self.inner) as _, data) };
-        return buf;
+        let current = self.inner.borrow().clone();
+        return Self::get_fixed_buf(current, data);
       }
       inner.notify_on_next(cap)
     };
@@ -257,6 +258,12 @@ impl<T: IoBufMut> FixedBufPool<T> {
     self.next_when_notified(cap, notify).await
   }
 
+  fn get_fixed_buf(inner: Rc<RefCell<Pool<T>>>, data: CheckedOutBuf) -> FixedBuf {
+    // Safety: the validity of buffer data is ensured by
+    // plumbing::Pool::try_next
+    unsafe { FixedBuf::new(inner, data) }
+  }
+
   #[cold]
   async fn next_when_notified(&mut self, cap: usize, notify: Arc<Notify>) -> FixedBuf {
     let notified = notify.notified();
@@ -266,11 +273,12 @@ impl<T: IoBufMut> FixedBufPool<T> {
       // between us calling `try_next` and here, so we can't miss a wakeup.
       notified.as_mut().await;
 
-      if let Some(data) = Rc::get_mut(&mut self.inner).unwrap().try_next(cap) {
-        // Safety: the validity of buffer data is ensured by
-        // plumbing::Pool::try_next
-        let buf = unsafe { FixedBuf::new(Rc::clone(&self.inner) as _, data) };
-        return buf;
+      let inner = self.inner.borrow_mut();
+      let mut inner = inner.borrow_mut();
+
+      if let Some(data) = inner.try_next(cap) {
+        let current = self.inner.borrow().clone();
+        return Self::get_fixed_buf(current, data);
       }
 
       // It's possible that the task did not get a buffer from `try_next`.

src/buf/fixed/registry.rs

@@ -1,8 +1,9 @@
-use super::plumbing;
 use super::FixedBuf;
+use super::plumbing;
+use std::cell::RefCell;
 
+use crate::buf::fixed::shared::{process, register, unregister};
 use crate::buf::IoBufMut;
-use crate::runtime::CONTEXT;
 use std::io;
 use std::rc::Rc;
 
@@ -30,7 +31,7 @@ use std::rc::Rc;
 /// [`Runtime`]: crate::Runtime
 #[derive(Clone)]
 pub struct FixedBufRegistry<T: IoBufMut> {
-  inner: Rc<plumbing::Registry<T>>,
+  inner: RefCell<Rc<RefCell<plumbing::Registry<T>>>>,
 }
 
 impl<T: IoBufMut> FixedBufRegistry<T> {
@@ -99,7 +100,9 @@ impl<T: IoBufMut> FixedBufRegistry<T> {
   /// ```
   pub fn new(bufs: impl IntoIterator<Item = T>) -> Self {
     FixedBufRegistry {
-      inner: Rc::new(plumbing::Registry::new(bufs.into_iter())),
+      inner: RefCell::new(Rc::new(RefCell::new(plumbing::Registry::new(
+        bufs.into_iter(),
+      )))),
     }
   }
 
@@ -123,7 +126,7 @@ impl<T: IoBufMut> FixedBufRegistry<T> {
   /// of the `tokio-uring` runtime this call is made in, the function returns
   /// an error.
   pub fn register(&self) -> io::Result<()> {
-    CONTEXT.with(|x| x.handle().register_buffers(Rc::clone(&self.inner) as _))
+    register(self.inner.borrow().clone())
   }
 
   /// Unregisters this collection of buffers.
@@ -142,7 +145,7 @@ impl<T: IoBufMut> FixedBufRegistry<T> {
   /// an error. Calling `unregister` when no `FixedBufRegistry` is currently
   /// registered on this runtime also returns an error.
   pub fn unregister(&self) -> io::Result<()> {
-    CONTEXT.with(|x| x.handle().unregister_buffers(Rc::clone(&self.inner) as _))
+    unregister()
   }
 
   /// Returns a buffer identified by the specified index for use by the
@@ -153,12 +156,12 @@ impl<T: IoBufMut> FixedBufRegistry<T> {
   /// using the buffer takes ownership of it and returns it once completed,
   /// preventing shared use of the buffer while the operation is in flight.
   pub fn check_out(&mut self, index: usize) -> Option<FixedBuf> {
-    let inner = Rc::get_mut(&mut self.inner).unwrap();
-    inner.check_out(index).map(|data| {
-      let registry = Rc::clone(&self.inner);
-      // Safety: the validity of buffer data is ensured by
-      // plumbing::Registry::check_out
-      unsafe { FixedBuf::new(registry, data) }
-    })
+    let cob = {
+      let inner = self.inner.borrow_mut();
+      let mut inner = inner.borrow_mut();
+      inner.check_out(index)
+    };
+    let current = self.inner.borrow().clone();
+    process(current, cob)
   }
 }

src/buf/fixed/shared.rs

@@ -0,0 +1,25 @@
+use crate::buf::fixed::handle::CheckedOutBuf;
+use crate::buf::fixed::{FixedBuf, FixedBuffers};
+use crate::runtime::CONTEXT;
+use std::cell::RefCell;
+use std::io;
+use std::rc::Rc;
+
+pub(crate) fn process(
+  this: Rc<RefCell<dyn FixedBuffers>>,
+  cob: Option<CheckedOutBuf>,
+) -> Option<FixedBuf> {
+  cob.map(|data| {
+    // Safety: the validity of buffer data is ensured by
+    // plumbing::Pool::try_next
+    unsafe { FixedBuf::new(this, data) }
+  })
+}
+
+pub(crate) fn register(this: Rc<RefCell<dyn FixedBuffers>>) -> io::Result<()> {
+  CONTEXT.with(|x| x.handle().register_buffers(this))
+}
+
+pub(crate) fn unregister() -> io::Result<()> {
+  CONTEXT.with(|x| x.handle().unregister_buffers())
+}

src/fs/file.rs

@@ -465,7 +465,7 @@ impl File {
   /// use std::iter;
   ///
   /// tokio_uring::start(async {
-  ///     let registry = FixedBufRegistry::new(iter::repeat(vec![0; 10]).take(10));
+  ///     let mut registry = FixedBufRegistry::new(iter::repeat(vec![0; 10]).take(10));
   ///     registry.register()?;
   ///
   ///     let f = File::open("foo.txt").await?;
@@ -656,7 +656,7 @@ impl File {
   /// use tokio_uring::buf::BoundedBuf;
   ///
   /// tokio_uring::start(async {
-  ///     let registry = FixedBufRegistry::new([b"some bytes".to_vec()]);
+  ///     let mut registry = FixedBufRegistry::new([b"some bytes".to_vec()]);
   ///     registry.register()?;
   ///
   ///     let file = File::create("foo.txt").await?;