fix awc pool leak

actix-http/src/client/pool.rs

@@ -2,7 +2,7 @@ use std::cell::RefCell;
 use std::collections::VecDeque;
 use std::future::Future;
 use std::pin::Pin;
-use std::rc::{Rc, Weak};
+use std::rc::Rc;
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 
@@ -66,7 +66,7 @@ where
         // start support future
         actix_rt::spawn(ConnectorPoolSupport {
             connector: connector_rc.clone(),
-            inner: Rc::downgrade(&inner_rc),
+            inner: inner_rc.clone(),
         });
 
         ConnectionPool(connector_rc, inner_rc)
@@ -82,6 +82,13 @@ where
     }
 }
 
+impl<T, Io> Drop for ConnectionPool<T, Io> {
+    fn drop(&mut self) {
+        // We wake up the ConnectorPoolSupport when dropping so it can exit properly.
+        self.1.borrow().waker.wake();
+    }
+}
+
 impl<T, Io> Service for ConnectionPool<T, Io>
 where
     Io: AsyncRead + AsyncWrite + Unpin + 'static,
@@ -421,7 +428,7 @@ where
     Io: AsyncRead + AsyncWrite + Unpin + 'static,
 {
     connector: T,
-    inner: Weak<RefCell<Inner<Io>>>,
+    inner: Rc<RefCell<Inner<Io>>>,
 }
 
 impl<T, Io> Future for ConnectorPoolSupport<T, Io>
@@ -435,8 +442,13 @@ where
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
 
-        if let Some(this_inner) = this.inner.upgrade() {
-            let mut inner = this_inner.as_ref().borrow_mut();
+        if Rc::strong_count(this.inner) == 1 {
+            // If we are last copy of Inner<Io> it means the ConnectionPool is already gone
+            // and we are safe to exit.
+            return Poll::Ready(());
+        }
+
+        let mut inner = this.inner.borrow_mut();
         inner.waker.register(cx.waker());
 
         // check waiters
@@ -459,7 +471,7 @@ where
                     if let Err(conn) = tx.send(Ok(IoConnection::new(
                         io,
                         created,
-                            Some(Acquired(key.clone(), Some(this_inner.clone()))),
+                        Some(Acquired(key.clone(), Some(this.inner.clone()))),
                     ))) {
                         let (io, created) = conn.unwrap().into_inner();
                         inner.release_conn(&key, io, created);
@@ -471,7 +483,7 @@ where
                     OpenWaitingConnection::spawn(
                         key.clone(),
                         tx,
-                            this_inner.clone(),
+                        this.inner.clone(),
                         this.connector.call(connect),
                         inner.config.clone(),
                     );
@@ -481,9 +493,6 @@ where
         }
 
         Poll::Pending
-        } else {
-            Poll::Ready(())
-        }
     }
 }