Split pooled oneshot to separate module

ntex/CHANGES.md

@@ -1,11 +1,13 @@
 # Changes
 
-## [0.1.16] - 2020-xx-xx
+## [0.1.16] - 2020-05-10
 
 * ntex::http: Remove redundant BodySize::Sized64
 
 * ntex::http: Do not check h1 keep-alive during response processing
 
+* ntex::channel: Split pooled oneshot to separate module
+
 ## [0.1.15] - 2020-05-03
 
 * ntex::util: Refactor stream dispatcher

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "0.1.15"
+version = "0.1.16"
 authors = ["Nikolay Kim <[email protected]>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -39,7 +39,7 @@ cookie = ["coo-kie", "coo-kie/percent-encode"]
 ntex-codec = "0.1.2"
 ntex-rt = "0.1.1"
 ntex-rt-macros = "0.1"
-ntex-router = "0.3.4"
+ntex-router = "0.3.5"
 ntex-service = "0.1.1"
 ntex-macros = "0.1"
 

ntex/src/channel/mod.rs

@@ -4,3 +4,4 @@ mod cell;
 pub mod condition;
 pub mod mpsc;
 pub mod oneshot;
+pub mod pool;

ntex/src/channel/oneshot.rs

@@ -4,11 +4,23 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 pub use futures::channel::oneshot::Canceled;
-use slab::Slab;
 
 use super::cell::Cell;
 use crate::task::LocalWaker;
 
+#[doc(hidden)]
+#[deprecated(since = "0.1.16", note = "Use pool::create instead")]
+pub use super::pool::new as pool;
+#[doc(hidden)]
+#[deprecated(since = "0.1.16", note = "Use pool::Pool instead")]
+pub use super::pool::Pool;
+#[doc(hidden)]
+#[deprecated(since = "0.1.16", note = "Use pool::Receiver instead")]
+pub use super::pool::Receiver as PReceiver;
+#[doc(hidden)]
+#[deprecated(since = "0.1.16", note = "Use pool::Sender instead")]
+pub use super::pool::Sender as PSender;
+
 /// Creates a new futures-aware, one-shot channel.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     let inner = Cell::new(Inner {
@@ -22,11 +34,6 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     (tx, rx)
 }
 
-/// Creates a new futures-aware, pool of one-shot's.
-pub fn pool<T>() -> Pool<T> {
-    Pool(Cell::new(Slab::new()))
-}
-
 /// Represents the completion half of a oneshot through which the result of a
 /// computation is signaled.
 #[derive(Debug)]
@@ -108,167 +115,6 @@ impl<T> Future for Receiver<T> {
     }
 }
 
-/// Futures-aware, pool of one-shot's.
-pub struct Pool<T>(Cell<Slab<PoolInner<T>>>);
-
-bitflags::bitflags! {
-    struct Flags: u8 {
-        const SENDER = 0b0000_0001;
-        const RECEIVER = 0b0000_0010;
-    }
-}
-
-#[derive(Debug)]
-struct PoolInner<T> {
-    flags: Flags,
-    value: Option<T>,
-    tx_waker: LocalWaker,
-    rx_waker: LocalWaker,
-}
-
-impl<T> Pool<T> {
-    pub fn channel(&self) -> (PSender<T>, PReceiver<T>) {
-        let token = self.0.get_mut().insert(PoolInner {
-            flags: Flags::all(),
-            value: None,
-            tx_waker: LocalWaker::default(),
-            rx_waker: LocalWaker::default(),
-        });
-
-        (
-            PSender {
-                token,
-                inner: self.0.clone(),
-            },
-            PReceiver {
-                token,
-                inner: self.0.clone(),
-            },
-        )
-    }
-}
-
-impl<T> Clone for Pool<T> {
-    fn clone(&self) -> Self {
-        Pool(self.0.clone())
-    }
-}
-
-/// Represents the completion half of a oneshot through which the result of a
-/// computation is signaled.
-#[derive(Debug)]
-pub struct PSender<T> {
-    token: usize,
-    inner: Cell<Slab<PoolInner<T>>>,
-}
-
-/// A future representing the completion of a computation happening elsewhere in
-/// memory.
-#[derive(Debug)]
-#[must_use = "futures do nothing unless polled"]
-pub struct PReceiver<T> {
-    token: usize,
-    inner: Cell<Slab<PoolInner<T>>>,
-}
-
-#[allow(clippy::mut_from_ref)]
-fn get_inner<T>(inner: &Cell<Slab<PoolInner<T>>>, token: usize) -> &mut PoolInner<T> {
-    unsafe { inner.get_mut().get_unchecked_mut(token) }
-}
-
-// The oneshots do not ever project Pin to the inner T
-impl<T> Unpin for PReceiver<T> {}
-impl<T> Unpin for PSender<T> {}
-
-impl<T> PSender<T> {
-    /// Completes this oneshot with a successful result.
-    ///
-    /// This function will consume `self` and indicate to the other end, the
-    /// `Receiver`, that the error provided is the result of the computation this
-    /// represents.
-    ///
-    /// If the value is successfully enqueued for the remote end to receive,
-    /// then `Ok(())` is returned. If the receiving end was dropped before
-    /// this function was called, however, then `Err` is returned with the value
-    /// provided.
-    pub fn send(self, val: T) -> Result<(), T> {
-        let inner = get_inner(&self.inner, self.token);
-        if inner.flags.contains(Flags::RECEIVER) {
-            inner.value = Some(val);
-            inner.rx_waker.wake();
-            Ok(())
-        } else {
-            Err(val)
-        }
-    }
-
-    /// Tests to see whether this `Sender`'s corresponding `Receiver`
-    /// has gone away.
-    pub fn is_canceled(&self) -> bool {
-        !get_inner(&self.inner, self.token)
-            .flags
-            .contains(Flags::RECEIVER)
-    }
-
-    /// Polls the channel to determine if receiving path is dropped
-    pub fn poll_canceled(&self, cx: &mut Context<'_>) -> Poll<()> {
-        let inner = get_inner(&self.inner, self.token);
-
-        if inner.flags.contains(Flags::RECEIVER) {
-            inner.tx_waker.register(cx.waker());
-            Poll::Pending
-        } else {
-            Poll::Ready(())
-        }
-    }
-}
-
-impl<T> Drop for PSender<T> {
-    fn drop(&mut self) {
-        let inner = get_inner(&self.inner, self.token);
-        if inner.flags.contains(Flags::RECEIVER) {
-            inner.rx_waker.wake();
-            inner.flags.remove(Flags::SENDER);
-        } else {
-            self.inner.get_mut().remove(self.token);
-        }
-    }
-}
-
-impl<T> Drop for PReceiver<T> {
-    fn drop(&mut self) {
-        let inner = get_inner(&self.inner, self.token);
-        if inner.flags.contains(Flags::SENDER) {
-            inner.tx_waker.wake();
-            inner.flags.remove(Flags::RECEIVER);
-        } else {
-            self.inner.get_mut().remove(self.token);
-        }
-    }
-}
-
-impl<T> Future for PReceiver<T> {
-    type Output = Result<T, Canceled>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.get_mut();
-        let inner = get_inner(&this.inner, this.token);
-
-        // If we've got a value, then skip the logic below as we're done.
-        if let Some(val) = inner.value.take() {
-            return Poll::Ready(Ok(val));
-        }
-
-        // Check if sender is dropped and return error if it is.
-        if !inner.flags.contains(Flags::SENDER) {
-            Poll::Ready(Err(Canceled))
-        } else {
-            inner.rx_waker.register(cx.waker());
-            Poll::Pending
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -300,46 +146,4 @@ mod tests {
         drop(tx);
         assert!(rx.await.is_err());
     }
-
-    #[ntex_rt::test]
-    async fn test_pool() {
-        let p = pool();
-        let (tx, rx) = p.channel();
-        tx.send("test").unwrap();
-        assert_eq!(rx.await.unwrap(), "test");
-
-        let p2 = p.clone();
-        let (tx, rx) = p2.channel();
-        assert!(!tx.is_canceled());
-        drop(rx);
-        assert!(tx.is_canceled());
-        assert!(tx.send("test").is_err());
-
-        let (tx, rx) = pool::<&'static str>().channel();
-        drop(tx);
-        assert!(rx.await.is_err());
-
-        let (tx, mut rx) = pool::<&'static str>().channel();
-        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
-        tx.send("test").unwrap();
-        assert_eq!(rx.await.unwrap(), "test");
-
-        let (tx, mut rx) = pool::<&'static str>().channel();
-        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
-        drop(tx);
-        assert!(rx.await.is_err());
-
-        let (mut tx, rx) = pool::<&'static str>().channel();
-        assert!(!tx.is_canceled());
-        assert_eq!(
-            lazy(|cx| Pin::new(&mut tx).poll_canceled(cx)).await,
-            Poll::Pending
-        );
-        drop(rx);
-        assert!(tx.is_canceled());
-        assert_eq!(
-            lazy(|cx| Pin::new(&mut tx).poll_canceled(cx)).await,
-            Poll::Ready(())
-        );
-    }
 }