Skip to content

Commit

Permalink
fix(client): prevent a checkout loop of pooled connections that aren'…
Browse files Browse the repository at this point in the history
…t ready yet
  • Loading branch information
seanmonstar committed Apr 30, 2018
1 parent 11c92ff commit 5e3b43a
Show file tree
Hide file tree
Showing 4 changed files with 30 additions and 29 deletions.
2 changes: 1 addition & 1 deletion src/client/dispatch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ impl<T, U> Sender<T, U> {

impl<T, U> UnboundedSender<T, U> {
pub fn is_ready(&self) -> bool {
self.giver.is_wanting()
!self.giver.is_canceled()
}

pub fn is_closed(&self) -> bool {
Expand Down
22 changes: 19 additions & 3 deletions src/client/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -318,6 +318,15 @@ where C: Connect + Sync + 'static,
Ok(())
})
);
} else {
// There's no body to delay, but the connection isn't
// ready yet. Only re-insert when it's ready
executor.execute(
future::poll_fn(move || {
pooled.poll_ready()
})
.then(|_| Ok(()))
);
}
Ok(res)
});
Expand Down Expand Up @@ -441,6 +450,13 @@ impl<B> PoolClient<B> {
PoolTx::Http2(ref tx) => tx.is_ready(),
}
}

fn is_closed(&self) -> bool {
match self.tx {
PoolTx::Http1(ref tx) => tx.is_closed(),
PoolTx::Http2(ref tx) => tx.is_closed(),
}
}
}

impl<B: Payload + 'static> PoolClient<B> {
Expand All @@ -460,10 +476,10 @@ impl<B> Poolable for PoolClient<B>
where
B: 'static,
{
fn is_closed(&self) -> bool {
fn is_open(&self) -> bool {
match self.tx {
PoolTx::Http1(ref tx) => tx.is_closed(),
PoolTx::Http2(ref tx) => tx.is_closed(),
PoolTx::Http1(ref tx) => tx.is_ready(),
PoolTx::Http2(ref tx) => tx.is_ready(),
}
}

Expand Down
33 changes: 9 additions & 24 deletions src/client/pool.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ pub(super) struct Pool<T> {
//
// See https://github.com/hyperium/hyper/issues/1429
pub(super) trait Poolable: Sized {
fn is_closed(&self) -> bool;
fn is_open(&self) -> bool;
/// Reserve this connection.
///
/// Allows for HTTP/2 to return a shared reservation.
Expand Down Expand Up @@ -236,7 +236,7 @@ impl<'a, T: Poolable + 'a> IdlePopper<'a, T> {
while let Some(entry) = self.list.pop() {
// If the connection has been closed, or is older than our idle
// timeout, simply drop it and keep looking...
if entry.value.is_closed() {
if !entry.value.is_open() {
trace!("removing closed connection for {:?}", self.key);
continue;
}
Expand Down Expand Up @@ -377,7 +377,7 @@ impl<T: Poolable> PoolInner<T> {

self.idle.retain(|key, values| {
values.retain(|entry| {
if entry.value.is_closed() {
if !entry.value.is_open() {
trace!("idle interval evicting closed for {:?}", key);
return false;
}
Expand Down Expand Up @@ -475,7 +475,7 @@ impl<T: Poolable> DerefMut for Pooled<T> {
impl<T: Poolable> Drop for Pooled<T> {
fn drop(&mut self) {
if let Some(value) = self.value.take() {
if value.is_closed() {
if !value.is_open() {
// If we *already* know the connection is done here,
// it shouldn't be re-inserted back into the pool.
return;
Expand Down Expand Up @@ -519,7 +519,7 @@ impl<T: Poolable> Checkout<T> {
if let Some(mut rx) = self.waiter.take() {
match rx.poll() {
Ok(Async::Ready(value)) => {
if !value.is_closed() {
if value.is_open() {
Ok(Async::Ready(Some(self.pool.reuse(&self.key, value))))
} else {
Err(::Error::new_canceled(Some(CANCELED)))
Expand Down Expand Up @@ -662,30 +662,15 @@ mod tests {
struct Uniq<T>(T);

impl<T> Poolable for Uniq<T> {
fn is_closed(&self) -> bool {
false
fn is_open(&self) -> bool {
true
}

fn reserve(self) -> Reservation<Self> {
Reservation::Unique(self)
}
}

/*
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
struct Share<T>(T);
impl<T> Poolable for Share<T> {
fn is_closed(&self) -> bool {
false
}
fn reserve(self) -> Reservation<Self> {
Reservation::Shared(self.clone(), self)
}
}
*/

fn c<T: Poolable>(key: Key) -> Connecting<T> {
Connecting {
key,
Expand Down Expand Up @@ -817,8 +802,8 @@ mod tests {
}

impl Poolable for CanClose {
fn is_closed(&self) -> bool {
self.closed
fn is_open(&self) -> bool {
!self.closed
}

fn reserve(self) -> Reservation<Self> {
Expand Down
2 changes: 1 addition & 1 deletion src/client/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ fn conn_reset_after_write() {
}

// sleep to allow some time for the connection to return to the pool
thread::sleep(Duration::from_secs(1));
thread::sleep(Duration::from_millis(50));

let req = Request::builder()
.uri("http://mock.local/a")
Expand Down

0 comments on commit 5e3b43a

Please sign in to comment.