Restore off-runtime creation of fixed buffer collections (tokio-rs#200)

tokio-rs#196 broke the ability to populate a collection of fixed buffers
off-runtime and then have it registered inside a runtime.

There is no need for the driver when a fixed buffer collection is
populated. The driver comes in when registering and unregistering the
buffers, and then as well, we don't currently need the collection to be
affine to the driver. The driver always compares the pointer passed in
to prevent unregistering the wrong buffers, so we can just grab the
driver from the context.

Change the doc examples to show and exercise off-runtime creation of
`FixedBufPool` and `FixedBufRegistry`.

Author: mzabaluev

src/buf/fixed/pool.rs

@@ -1,7 +1,6 @@
 use super::handle::CheckedOutBuf;
 use super::{FixedBuf, FixedBuffers};
 
-use crate::runtime::driver::WeakHandle;
 use crate::runtime::CONTEXT;
 use libc::{iovec, UIO_MAXIOV};
 use std::cell::RefCell;
@@ -84,7 +83,6 @@ use std::slice;
 #[derive(Clone)]
 pub struct FixedBufPool {
     inner: Rc<RefCell<Inner>>,
-    driver: WeakHandle,
 }
 
 impl FixedBufPool {
@@ -116,10 +114,11 @@ impl FixedBufPool {
     /// # let (memlock_limit, _) = getrlimit(Resource::RLIMIT_MEMLOCK)?;
     /// # let NUM_BUFFERS = std::cmp::max(memlock_limit as usize / 4096 / 8, 1);
     /// # let BUF_SIZE = 4096;
+    /// let pool = FixedBufPool::new(
+    ///     iter::repeat(Vec::with_capacity(BUF_SIZE)).take(NUM_BUFFERS)
+    /// );
+    ///
     /// tokio_uring::start(async {
-    ///     let pool = FixedBufPool::new(
-    ///         iter::repeat(Vec::with_capacity(BUF_SIZE)).take(NUM_BUFFERS)
-    ///     );
     ///     pool.register()?;
     ///     // ...
     ///     Ok(())
@@ -139,10 +138,11 @@ impl FixedBufPool {
     /// # let (memlock_limit, _) = getrlimit(Resource::RLIMIT_MEMLOCK)?;
     /// # let NUM_BUFFERS = std::cmp::max(memlock_limit as usize / 4096 / 8, 1);
     /// # let BUF_SIZE = 4096;
+    /// let pool = FixedBufPool::new(
+    ///     iter::repeat_with(|| Vec::with_capacity(BUF_SIZE)).take(NUM_BUFFERS)
+    /// );
+    ///
     /// tokio_uring::start(async {
-    ///     let pool = FixedBufPool::new(
-    ///         iter::repeat_with(|| Vec::with_capacity(BUF_SIZE)).take(NUM_BUFFERS)
-    ///     );
     ///     pool.register()?;
     ///     // ...
     ///     Ok(())
@@ -152,7 +152,6 @@ impl FixedBufPool {
     pub fn new(bufs: impl IntoIterator<Item = Vec<u8>>) -> Self {
         FixedBufPool {
             inner: Rc::new(RefCell::new(Inner::new(bufs.into_iter()))),
-            driver: CONTEXT.with(|x| x.weak().expect("Not in a runtime context")),
         }
     }
 
@@ -176,10 +175,12 @@ impl FixedBufPool {
     /// of the `tokio-uring` runtime this call is made in, the function returns
     /// an error.
     pub fn register(&self) -> io::Result<()> {
-        self.driver
-            .upgrade()
-            .expect("Runtime context is no longer present")
-            .register_buffers(Rc::clone(&self.inner) as _)
+        CONTEXT.with(|x| {
+            x.handle()
+                .as_ref()
+                .expect("Not in a runtime context")
+                .register_buffers(Rc::clone(&self.inner) as _)
+        })
     }
 
     /// Unregisters this collection of buffers.
@@ -198,10 +199,12 @@ impl FixedBufPool {
     /// an error. Calling `unregister` when no `FixedBufPool` is currently
     /// registered on this runtime also returns an error.
     pub fn unregister(&self) -> io::Result<()> {
-        self.driver
-            .upgrade()
-            .expect("Runtime context is no longer present")
-            .unregister_buffers(Rc::clone(&self.inner) as _)
+        CONTEXT.with(|x| {
+            x.handle()
+                .as_ref()
+                .expect("Not in a runtime context")
+                .unregister_buffers(Rc::clone(&self.inner) as _)
+        })
     }
 
     /// Returns a buffer of requested capacity from this pool

src/buf/fixed/registry.rs

@@ -1,7 +1,6 @@
 use super::handle::CheckedOutBuf;
 use super::{FixedBuf, FixedBuffers};
 
-use crate::runtime::driver::WeakHandle;
 use crate::runtime::CONTEXT;
 use libc::{iovec, UIO_MAXIOV};
 use std::cell::RefCell;
@@ -37,7 +36,6 @@ use std::slice;
 #[derive(Clone)]
 pub struct FixedBufRegistry {
     inner: Rc<RefCell<Inner>>,
-    driver: WeakHandle,
 }
 
 impl FixedBufRegistry {
@@ -93,10 +91,11 @@ impl FixedBufRegistry {
     /// # let (memlock_limit, _) = getrlimit(Resource::RLIMIT_MEMLOCK)?;
     /// # let NUM_BUFFERS = std::cmp::max(memlock_limit as usize / 4096 / 8, 1);
     /// # let BUF_SIZE = 4096;
+    /// let registry = FixedBufRegistry::new(
+    ///     iter::repeat_with(|| Vec::with_capacity(BUF_SIZE)).take(NUM_BUFFERS)
+    /// );
+    ///
     /// tokio_uring::start(async {
-    ///     let registry = FixedBufRegistry::new(
-    ///         iter::repeat_with(|| Vec::with_capacity(BUF_SIZE)).take(NUM_BUFFERS)
-    ///     );
     ///     registry.register()?;
     ///     // ...
     ///     Ok(())
@@ -106,7 +105,6 @@ impl FixedBufRegistry {
     pub fn new(bufs: impl IntoIterator<Item = Vec<u8>>) -> Self {
         FixedBufRegistry {
             inner: Rc::new(RefCell::new(Inner::new(bufs.into_iter()))),
-            driver: CONTEXT.with(|x| x.weak().expect("Not in a runtime context")),
         }
     }
 
@@ -130,10 +128,12 @@ impl FixedBufRegistry {
     /// of the `tokio-uring` runtime this call is made in, the function returns
     /// an error.
     pub fn register(&self) -> io::Result<()> {
-        self.driver
-            .upgrade()
-            .expect("Runtime context is no longer present")
-            .register_buffers(Rc::clone(&self.inner) as _)
+        CONTEXT.with(|x| {
+            x.handle()
+                .as_ref()
+                .expect("Not in a runtime context")
+                .register_buffers(Rc::clone(&self.inner) as _)
+        })
     }
 
     /// Unregisters this collection of buffers.
@@ -152,10 +152,12 @@ impl FixedBufRegistry {
     /// an error. Calling `unregister` when no `FixedBufRegistry` is currently
     /// registered on this runtime also returns an error.
     pub fn unregister(&self) -> io::Result<()> {
-        self.driver
-            .upgrade()
-            .expect("Runtime context is no longer present")
-            .unregister_buffers(Rc::clone(&self.inner) as _)
+        CONTEXT.with(|x| {
+            x.handle()
+                .as_ref()
+                .expect("Not in a runtime context")
+                .unregister_buffers(Rc::clone(&self.inner) as _)
+        })
     }
 
     /// Returns a buffer identified by the specified index for use by the

src/runtime/context.rs

@@ -44,6 +44,7 @@ impl RuntimeContext {
         self.driver.borrow().clone()
     }
 
+    #[allow(dead_code)]
     pub(crate) fn weak(&self) -> Option<WeakHandle> {
         self.driver.borrow().as_ref().map(Into::into)
     }