Merge pull request #66 from jonhoo/no-block-reactor

Reimplement Pool to be fully asynchronous

Author: blackbeam

Cargo.toml

@@ -15,20 +15,22 @@ edition = "2018"
 bit-vec = "0.5"
 byteorder = "1"
 bytes = "0.4"
+crossbeam = "0.7"
 failure = "0.1"
 failure_derive = "0.1"
 fnv = "1"
-futures = "^0.1.18"
+futures = "0.1.18"
 mysql_common = "0.17"
 native-tls = { version = "0.2", optional = true }
 regex = "1"
 serde = "1"
 serde_json = "1"
-tokio = "^0.1.9"
+tokio = "0.1.9"
 tokio-codec = "0.1"
 tokio-io = "0.1"
 tokio-named-pipes = "0.1"
 tokio-uds = "0.2"
+tokio-sync = "0.1"
 twox-hash = "1"
 url = "1"
 

src/conn/mod.rs

@@ -144,6 +144,11 @@ impl Conn {
         self.get_affected_rows()
     }
 
+    fn close(mut self) -> impl MyFuture<()> {
+        self.inner.disconnected = true;
+        self.cleanup().and_then(Conn::disconnect)
+    }
+
     fn is_secure(&self) -> bool {
         if let Some(ref stream) = self.inner.stream {
             stream.is_secure()

src/conn/pool/futures/disconnect_pool.rs

@@ -6,40 +6,40 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{
-    Async::{NotReady, Ready},
-    Future, Poll,
+use futures::{task, Async, Future, Poll};
+
+use crate::{
+    conn::pool::{Inner, Pool},
+    error::Error,
 };
 
-use crate::{conn::pool::Pool, error::*};
+use std::sync::{atomic, Arc};
 
 /// Future that disconnects this pool from server and resolves to `()`.
 ///
 /// Active connections taken from this pool should be disconnected manually.
 /// Also all pending and new `GetConn`'s will resolve to error.
 pub struct DisconnectPool {
-    pool: Pool,
+    pool_inner: Arc<Inner>,
 }
 
 pub fn new(pool: Pool) -> DisconnectPool {
-    DisconnectPool { pool }
+    DisconnectPool {
+        pool_inner: pool.inner,
+    }
 }
 
 impl Future for DisconnectPool {
     type Item = ();
     type Error = Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        self.pool.handle_futures()?;
-
-        let (new_len, queue_len) = self
-            .pool
-            .with_inner(|inner| (inner.new.len(), inner.queue.len()));
+        self.pool_inner.wake.push(task::current());
 
-        if (new_len, queue_len) == (0, 0) {
-            Ok(Ready(()))
+        if self.pool_inner.closed.load(atomic::Ordering::Acquire) {
+            Ok(Async::Ready(()))
         } else {
-            Ok(NotReady)
+            Ok(Async::NotReady)
         }
     }
 }

src/conn/pool/futures/get_conn.rs

@@ -6,27 +6,84 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
-use futures::{Future, Poll};
+use futures::{try_ready, Async, Future, Poll};
 
 use crate::{
     conn::{pool::Pool, Conn},
     error::*,
+    MyFuture,
 };
 
+pub(crate) enum GetConnInner {
+    New,
+    Done(Option<Conn>),
+    // TODO: one day this should be an existential
+    // TODO: impl Drop?
+    Connecting(Box<dyn MyFuture<Conn>>),
+}
+
 /// This future will take connection from a pool and resolve to `Conn`.
 pub struct GetConn {
-    pool: Pool,
+    pub(crate) pool: Option<Pool>,
+    pub(crate) inner: GetConnInner,
 }
 
 pub fn new(pool: &Pool) -> GetConn {
-    GetConn { pool: pool.clone() }
+    GetConn {
+        pool: Some(pool.clone()),
+        inner: GetConnInner::New,
+    }
 }
 
 impl Future for GetConn {
     type Item = Conn;
     type Error = Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        self.pool.poll()
+        loop {
+            match self.inner {
+                GetConnInner::New => match try_ready!(self
+                    .pool
+                    .as_mut()
+                    .expect("GetConn::poll polled after returning Async::Ready")
+                    .poll_new_conn())
+                .inner
+                {
+                    GetConnInner::Done(Some(conn)) => {
+                        self.inner = GetConnInner::Done(Some(conn));
+                    }
+                    GetConnInner::Connecting(conn_fut) => {
+                        self.inner = GetConnInner::Connecting(conn_fut);
+                    }
+                    GetConnInner::Done(None) => unreachable!(
+                        "Pool::poll_new_conn never gives out already-consumed GetConns"
+                    ),
+                    GetConnInner::New => {
+                        unreachable!("Pool::poll_new_conn never gives out GetConnInner::New")
+                    }
+                },
+                GetConnInner::Done(ref mut c @ Some(_)) => {
+                    let mut c = c.take().unwrap();
+                    c.inner.pool = Some(
+                        self.pool
+                            .take()
+                            .expect("GetConn::poll polled after returning Async::Ready"),
+                    );
+                    return Ok(Async::Ready(c));
+                }
+                GetConnInner::Done(None) => {
+                    unreachable!("GetConn::poll polled after returning Async::Ready");
+                }
+                GetConnInner::Connecting(ref mut f) => {
+                    let mut c = try_ready!(f.poll());
+                    c.inner.pool = Some(
+                        self.pool
+                            .take()
+                            .expect("GetConn::poll polled after returning Async::Ready"),
+                    );
+                    return Ok(Async::Ready(c));
+                }
+            }
+        }
     }
 }

src/conn/pool/futures/mod.rs

@@ -6,6 +6,7 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
+pub(super) use self::get_conn::GetConnInner;
 pub use self::{
     disconnect_pool::{new as new_disconnect_pool, DisconnectPool},
     get_conn::{new as new_get_conn, GetConn},