forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "refactor(lib): convert to futures 0.2.0-beta (#1470)"

Browse Source 
This reverts commit a12f7beed9.

Much sadness 😢.
This commit is contained in:
Sean McArthur
2018-04-07 10:19:50 -07:00
parent 72e02d6ac8
commit 625e4daaa1
34 changed files with 1368 additions and 1386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/client/conn.rs
View File

@@ -11,10 +11,9 @@ use std::fmt;
use std::marker::PhantomData;
use bytes::Bytes;
use futures::{Async, Future, FutureExt, Poll};
use futures::{Async, Future, Poll};
use futures::future::{self, Either};
use futures::task;
use futures::io::{AsyncRead, AsyncWrite};
use tokio_io::{AsyncRead, AsyncWrite};
use proto;
use proto::body::Entity;
@@ -124,8 +123,8 @@ impl<B> SendRequest<B>
/// Polls to determine whether this sender can be used yet for a request.
///
/// If the associated connection is closed, this returns an Error.
pub fn poll_ready(&mut self, cx: &mut task::Context) -> Poll<(), ::Error> {
self.dispatch.poll_ready(cx)
pub fn poll_ready(&mut self) -> Poll<(), ::Error> {
self.dispatch.poll_ready()
}
pub(super) fn is_ready(&self) -> bool {
@@ -167,7 +166,7 @@ where
/// # use http::header::HOST;
/// # use hyper::client::conn::SendRequest;
/// # use hyper::Body;
/// use futures::FutureExt;
/// use futures::Future;
/// use hyper::Request;
///
/// # fn doc(mut tx: SendRequest<Body>) {
@@ -191,19 +190,19 @@ where
pub fn send_request(&mut self, req: Request<B>) -> ResponseFuture {
let inner = match self.dispatch.send(req) {
Ok(rx) => {
rx.then(move |res| {
Either::A(rx.then(move |res| {
match res {
Ok(Ok(res)) => Ok(res),
Ok(Err(err)) => Err(err),
// this is definite bug if it happens, but it shouldn't happen!
Err(_) => panic!("dispatch dropped without returning error"),
}
}).left()
}))
},
Err(_req) => {
debug!("connection was not ready");
let err = ::Error::new_canceled(Some("connection was not ready"));
future::err(err).right()
Either::B(future::err(err))
}
};
@@ -219,7 +218,7 @@ where
{
let inner = match self.dispatch.try_send(req) {
Ok(rx) => {
Either::Left(rx.then(move |res| {
Either::A(rx.then(move |res| {
match res {
Ok(Ok(res)) => Ok(res),
Ok(Err(err)) => Err(err),
@@ -231,7 +230,7 @@ where
Err(req) => {
debug!("connection was not ready");
let err = ::Error::new_canceled(Some("connection was not ready"));
Either::Right(future::err((err, Some(req))))
Either::B(future::err((err, Some(req))))
}
};
Box::new(inner)
@@ -282,8 +281,8 @@ where
/// upgrade. Once the upgrade is completed, the connection would be "done",
/// but it is not desired to actally shutdown the IO object. Instead you
/// would take it back using `into_parts`.
pub fn poll_without_shutdown(&mut self, cx: &mut task::Context) -> Poll<(), ::Error> {
self.inner.poll_without_shutdown(cx)
pub fn poll_without_shutdown(&mut self) -> Poll<(), ::Error> {
self.inner.poll_without_shutdown()
}
}
@@ -295,8 +294,8 @@ where
type Item = ();
type Error = ::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
self.inner.poll(cx)
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
}
}
@@ -368,8 +367,8 @@ where
type Item = (SendRequest<B>, Connection<T, B>);
type Error = ::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
self.inner.poll(cx)
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
.map(|async| {
async.map(|(tx, dispatch)| {
(tx, Connection { inner: dispatch })
@@ -399,8 +398,8 @@ where
>);
type Error = ::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
self.inner.poll(cx)
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
}
}
@@ -422,7 +421,7 @@ where
>);
type Error = ::Error;
fn poll(&mut self, _cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let io = self.io.take().expect("polled more than once");
let (tx, rx) = dispatch::channel();
let mut conn = proto::Conn::new(io);
@@ -446,8 +445,8 @@ impl Future for ResponseFuture {
type Error = ::Error;
#[inline]
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
self.inner.poll(cx)
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
}
}

98
src/client/connect.rs
View File

@@ -8,21 +8,24 @@
use std::error::Error as StdError;
use std::fmt;
use std::io;
use std::mem;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use futures::{Future, Never, Poll, Async};
use futures::executor::{Executor, SpawnError, ThreadPoolBuilder};
use futures::task;
use futures::io::{AsyncRead, AsyncWrite};
use futures::{Future, Poll, Async};
use futures::future::{Executor, ExecuteError};
use futures::sync::oneshot;
use futures_cpupool::{Builder as CpuPoolBuilder};
use http::Uri;
use http::uri::Scheme;
use net2::TcpBuilder;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio::reactor::Handle;
use tokio::net::{TcpStream, ConnectFuture};
use executor::CloneBoxedExecutor;
use super::dns;
use self::http_connector::HttpConnectorBlockingTask;
/// Connect to a destination, returning an IO transport.
///
@@ -171,7 +174,7 @@ impl HttpConnector {
/// Takes number of DNS worker threads.
#[inline]
pub fn new(threads: usize, handle: &Handle) -> HttpConnector {
let pool = ThreadPoolBuilder::new()
let pool = CpuPoolBuilder::new()
.name_prefix("hyper-dns")
.pool_size(threads)
.create();
@@ -183,10 +186,10 @@ impl HttpConnector {
/// Takes an executor to run blocking tasks on.
#[inline]
pub fn new_with_executor<E: 'static>(executor: E, handle: &Handle) -> HttpConnector
where E: Executor + Clone + Send + Sync
where E: Executor<HttpConnectorBlockingTask> + Send + Sync
{
HttpConnector {
executor: HttpConnectExecutor(Box::new(executor)),
executor: HttpConnectExecutor(Arc::new(executor)),
enforce_http: true,
handle: handle.clone(),
keep_alive_timeout: None,
@@ -295,7 +298,7 @@ pub struct HttpConnecting {
enum State {
Lazy(HttpConnectExecutor, String, u16),
Resolving(dns::Resolving),
Resolving(oneshot::SpawnHandle<dns::IpAddrs, io::Error>),
Connecting(ConnectingTcp),
Error(Option<io::Error>),
}
@@ -304,11 +307,11 @@ impl Future for HttpConnecting {
type Item = (TcpStream, Connected);
type Error = io::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
let state;
match self.state {
State::Lazy(ref mut executor, ref mut host, port) => {
State::Lazy(ref executor, ref mut host, port) => {
// If the host is already an IP addr (v4 or v6),
// skip resolving the dns and start connecting right away.
if let Some(addrs) = dns::IpAddrs::try_parse(host, port) {
@@ -317,19 +320,24 @@ impl Future for HttpConnecting {
current: None
})
} else {
let host = ::std::mem::replace(host, String::new());
state = State::Resolving(dns::Resolving::spawn(host, port, executor));
let host = mem::replace(host, String::new());
let work = dns::Work::new(host, port);
state = State::Resolving(oneshot::spawn(work, executor));
}
},
State::Resolving(ref mut future) => {
let addrs = try_ready!(future.poll(cx));
state = State::Connecting(ConnectingTcp {
addrs: addrs,
current: None,
});
match try!(future.poll()) {
Async::NotReady => return Ok(Async::NotReady),
Async::Ready(addrs) => {
state = State::Connecting(ConnectingTcp {
addrs: addrs,
current: None,
})
}
};
},
State::Connecting(ref mut c) => {
let sock = try_ready!(c.poll(cx, &self.handle));
let sock = try_ready!(c.poll(&self.handle));
if let Some(dur) = self.keep_alive_timeout {
sock.set_keepalive(Some(dur))?;
@@ -357,11 +365,11 @@ struct ConnectingTcp {
impl ConnectingTcp {
// not a Future, since passing a &Handle to poll
fn poll(&mut self, cx: &mut task::Context, handle: &Handle) -> Poll<TcpStream, io::Error> {
fn poll(&mut self, handle: &Handle) -> Poll<TcpStream, io::Error> {
let mut err = None;
loop {
if let Some(ref mut current) = self.current {
match current.poll(cx) {
match current.poll() {
Ok(ok) => return Ok(ok),
Err(e) => {
trace!("connect error {:?}", e);
@@ -384,19 +392,37 @@ impl ConnectingTcp {
}
}
#[derive(Clone)]
struct HttpConnectExecutor(Box<CloneBoxedExecutor>);
impl Executor for HttpConnectExecutor {
fn spawn(
&mut self,
f: Box<Future<Item = (), Error = Never> + 'static + Send>
) -> Result<(), SpawnError> {
self.0.spawn(f)
// Make this Future unnameable outside of this crate.
mod http_connector {
use super::*;
// Blocking task to be executed on a thread pool.
pub struct HttpConnectorBlockingTask {
pub(super) work: oneshot::Execute<dns::Work>
}
fn status(&self) -> Result<(), SpawnError> {
self.0.status()
impl fmt::Debug for HttpConnectorBlockingTask {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.pad("HttpConnectorBlockingTask")
}
}
impl Future for HttpConnectorBlockingTask {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<(), ()> {
self.work.poll()
}
}
}
#[derive(Clone)]
struct HttpConnectExecutor(Arc<Executor<HttpConnectorBlockingTask> + Send + Sync>);
impl Executor<oneshot::Execute<dns::Work>> for HttpConnectExecutor {
fn execute(&self, future: oneshot::Execute<dns::Work>) -> Result<(), ExecuteError<oneshot::Execute<dns::Work>>> {
self.0.execute(HttpConnectorBlockingTask { work: future })
.map_err(|err| ExecuteError::new(err.kind(), err.into_future().work))
}
}
@@ -404,7 +430,7 @@ impl Executor for HttpConnectExecutor {
mod tests {
#![allow(deprecated)]
use std::io;
use futures::executor::block_on;
use futures::Future;
use tokio::runtime::Runtime;
use super::{Connect, Destination, HttpConnector};
@@ -417,7 +443,7 @@ mod tests {
};
let connector = HttpConnector::new(1, runtime.handle());
assert_eq!(block_on(connector.connect(dst)).unwrap_err().kind(), io::ErrorKind::InvalidInput);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
}
#[test]
@@ -429,7 +455,7 @@ mod tests {
};
let connector = HttpConnector::new(1, runtime.handle());
assert_eq!(block_on(connector.connect(dst)).unwrap_err().kind(), io::ErrorKind::InvalidInput);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
}
@@ -442,6 +468,6 @@ mod tests {
};
let connector = HttpConnector::new(1, runtime.handle());
assert_eq!(block_on(connector.connect(dst)).unwrap_err().kind(), io::ErrorKind::InvalidInput);
assert_eq!(connector.connect(dst).wait().unwrap_err().kind(), io::ErrorKind::InvalidInput);
}
}

57
src/client/dispatch.rs
View File

@@ -1,8 +1,9 @@
use futures::{Async, Never, Poll, Stream};
use futures::channel::{mpsc, oneshot};
use futures::task;
use futures::{Async, Poll, Stream};
use futures::sync::{mpsc, oneshot};
use want;
use common::Never;
//pub type Callback<T, U> = oneshot::Sender<Result<U, (::Error, Option<T>)>>;
pub type RetryPromise<T, U> = oneshot::Receiver<Result<U, (::Error, Option<T>)>>;
pub type Promise<T> = oneshot::Receiver<Result<T, ::Error>>;
@@ -32,15 +33,15 @@ pub struct Sender<T, U> {
}
impl<T, U> Sender<T, U> {
pub fn poll_ready(&mut self, cx: &mut task::Context) -> Poll<(), ::Error> {
match self.inner.poll_ready(cx) {
pub fn poll_ready(&mut self) -> Poll<(), ::Error> {
match self.inner.poll_ready() {
Ok(Async::Ready(())) => {
// there's room in the queue, but does the Connection
// want a message yet?
self.giver.poll_want(cx)
self.giver.poll_want()
.map_err(|_| ::Error::Closed)
},
Ok(Async::Pending) => Ok(Async::Pending),
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(_) => Err(::Error::Closed),
}
}
@@ -78,15 +79,16 @@ impl<T, U> Stream for Receiver<T, U> {
type Item = (T, Callback<T, U>);
type Error = Never;
fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
match self.inner.poll_next(cx)? {
Async::Ready(item) => Ok(Async::Ready(item.map(|mut env| {
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
match self.inner.poll() {
Ok(Async::Ready(item)) => Ok(Async::Ready(item.map(|mut env| {
env.0.take().expect("envelope not dropped")
}))),
Async::Pending => {
Ok(Async::NotReady) => {
self.taker.want();
Ok(Async::Pending)
Ok(Async::NotReady)
}
Err(()) => unreachable!("mpsc never errors"),
}
}
}
@@ -109,11 +111,11 @@ impl<T, U> Drop for Receiver<T, U> {
// This poll() is safe to call in `Drop`, because we've
// called, `close`, which promises that no new messages
// will arrive, and thus, once we reach the end, we won't
// see a `Pending` (and try to park), but a Ready(None).
// see a `NotReady` (and try to park), but a Ready(None).
//
// All other variants:
// - Ready(None): the end. we want to stop looping
// - Pending: unreachable
// - NotReady: unreachable
// - Err: unreachable
while let Ok(Async::Ready(Some((val, cb)))) = self.inner.poll() {
let _ = cb.send(Err((::Error::new_canceled(None::<::Error>), Some(val))));
@@ -139,10 +141,10 @@ pub enum Callback<T, U> {
}
impl<T, U> Callback<T, U> {
pub fn poll_cancel(&mut self, cx: &mut task::Context) -> Poll<(), Never> {
pub fn poll_cancel(&mut self) -> Poll<(), ()> {
match *self {
Callback::Retry(ref mut tx) => tx.poll_cancel(cx),
Callback::NoRetry(ref mut tx) => tx.poll_cancel(cx),
Callback::Retry(ref mut tx) => tx.poll_cancel(),
Callback::NoRetry(ref mut tx) => tx.poll_cancel(),
}
}
@@ -164,8 +166,7 @@ mod tests {
#[cfg(feature = "nightly")]
extern crate test;
use futures::{future, FutureExt};
use futures::executor::block_on;
use futures::{future, Future};
#[cfg(feature = "nightly")]
use futures::{Stream};
@@ -174,7 +175,7 @@ mod tests {
fn drop_receiver_sends_cancel_errors() {
let _ = pretty_env_logger::try_init();
block_on(future::lazy(|_| {
future::lazy(|| {
#[derive(Debug)]
struct Custom(i32);
let (mut tx, rx) = super::channel::<Custom, ()>();
@@ -191,7 +192,7 @@ mod tests {
Ok::<(), ()>(())
})
})).unwrap();
}).wait().unwrap();
}
#[cfg(feature = "nightly")]
@@ -200,18 +201,18 @@ mod tests {
let (mut tx, mut rx) = super::channel::<i32, ()>();
b.iter(move || {
block_on(future::lazy(|cx| {
::futures::future::lazy(|| {
let _ = tx.send(1).unwrap();
loop {
let async = rx.poll_next(cx).unwrap();
if async.is_pending() {
let async = rx.poll().unwrap();
if async.is_not_ready() {
break;
}
}
Ok::<(), ()>(())
})).unwrap();
}).wait().unwrap();
})
}
@@ -221,11 +222,11 @@ mod tests {
let (_tx, mut rx) = super::channel::<i32, ()>();
b.iter(move || {
block_on(future::lazy(|cx| {
assert!(rx.poll_next(cx).unwrap().is_pending());
::futures::future::lazy(|| {
assert!(rx.poll().unwrap().is_not_ready());
Ok::<(), ()>(())
})).unwrap();
}).wait().unwrap();
})
}

41
src/client/dns.rs
View File

@@ -6,44 +6,27 @@ use std::net::{
};
use std::vec;
use futures::{Async, Future, Poll};
use futures::task;
use futures::future::lazy;
use futures::executor::Executor;
use futures::channel::oneshot;
use ::futures::{Async, Future, Poll};
pub struct Resolving {
receiver: oneshot::Receiver<Result<IpAddrs, io::Error>>
pub struct Work {
host: String,
port: u16
}
impl Resolving {
pub fn spawn(host: String, port: u16, executor: &mut Executor) -> Resolving {
let (sender, receiver) = oneshot::channel();
// The `Resolving` future will return an error when the sender is dropped,
// so we can just ignore the spawn error here
executor.spawn(Box::new(lazy(move |_| {
debug!("resolving host={:?}, port={:?}", host, port);
let result = (host.as_ref(), port).to_socket_addrs()
.map(|i| IpAddrs { iter: i });
sender.send(result).ok();
Ok(())
}))).ok();
Resolving { receiver }
impl Work {
pub fn new(host: String, port: u16) -> Work {
Work { host: host, port: port }
}
}
impl Future for Resolving {
impl Future for Work {
type Item = IpAddrs;
type Error = io::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<IpAddrs, io::Error> {
match self.receiver.poll(cx) {
Ok(Async::Pending) => Ok(Async::Pending),
Ok(Async::Ready(Ok(ips))) => Ok(Async::Ready(ips)),
Ok(Async::Ready(Err(err))) => Err(err),
Err(_) =>
Err(io::Error::new(io::ErrorKind::Other, "dns task was cancelled"))
}
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
debug!("resolving host={:?}, port={:?}", self.host, self.port);
(&*self.host, self.port).to_socket_addrs()
.map(|i| Async::Ready(IpAddrs { iter: i }))
}
}

235
src/client/mod.rs
View File

@@ -6,22 +6,23 @@ use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
use futures::{Async, Future, FutureExt, Never, Poll};
use futures::channel::oneshot;
use futures::future;
use futures::task;
use futures::{Async, Future, Poll};
use futures::future::{self, Executor};
use futures::sync::oneshot;
use http::{Method, Request, Response, Uri, Version};
use http::header::{Entry, HeaderValue, HOST};
use http::uri::Scheme;
use tokio::reactor::Handle;
use tokio_executor::spawn;
pub use tokio_service::Service;
pub use service::Service;
use proto::body::{Body, Entity};
use proto;
use self::pool::Pool;
pub use self::connect::{Connect, HttpConnector};
use self::background::{bg, Background};
use self::connect::Destination;
pub mod conn;
@@ -35,6 +36,7 @@ mod tests;
/// A Client to make outgoing HTTP requests.
pub struct Client<C, B = proto::Body> {
connector: Arc<C>,
executor: Exec,
h1_writev: bool,
pool: Pool<PoolClient<B>>,
retry_canceled_requests: bool,
@@ -81,9 +83,10 @@ impl Client<HttpConnector, proto::Body> {
impl<C, B> Client<C, B> {
#[inline]
fn configured(config: Config<C, B>) -> Client<C, B> {
fn configured(config: Config<C, B>, exec: Exec) -> Client<C, B> {
Client {
connector: Arc::new(config.connector),
executor: exec,
h1_writev: config.h1_writev,
pool: Pool::new(config.keep_alive, config.keep_alive_timeout),
retry_canceled_requests: config.retry_canceled_requests,
@@ -123,6 +126,14 @@ where C: Connect<Error=io::Error> + Sync + 'static,
/// Send a constructed Request using this Client.
pub fn request(&self, mut req: Request<B>) -> FutureResponse {
// TODO(0.12): do this at construction time.
//
// It cannot be done in the constructor because the Client::configured
// does not have `B: 'static` bounds, which are required to spawn
// the interval. In 0.12, add a static bounds to the constructor,
// and move this.
self.schedule_pool_timer();
match req.version() {
Version::HTTP_10 |
Version::HTTP_11 => (),
@@ -165,6 +176,7 @@ where C: Connect<Error=io::Error> + Sync + 'static,
}
}
let client = self.clone();
let uri = req.uri().clone();
let fut = RetryableSendRequest {
@@ -181,6 +193,7 @@ where C: Connect<Error=io::Error> + Sync + 'static,
let url = req.uri().clone();
let checkout = self.pool.checkout(domain);
let connect = {
let executor = self.executor.clone();
let pool = self.pool.clone();
let pool_key = Arc::new(domain.to_string());
let h1_writev = self.h1_writev;
@@ -188,39 +201,36 @@ where C: Connect<Error=io::Error> + Sync + 'static,
let dst = Destination {
uri: url,
};
future::lazy(move |_| {
future::lazy(move || {
connector.connect(dst)
.err_into()
.from_err()
.and_then(move |(io, connected)| {
conn::Builder::new()
.h1_writev(h1_writev)
.handshake_no_upgrades(io)
.and_then(move |(tx, conn)| {
future::lazy(move |cx| {
execute(conn.recover(|e| {
debug!("client connection error: {}", e);
}), cx)?;
Ok(pool.pooled(pool_key, PoolClient {
is_proxied: connected.is_proxied,
tx: tx,
}))
})
executor.execute(conn.map_err(|e| debug!("client connection error: {}", e)))?;
Ok(pool.pooled(pool_key, PoolClient {
is_proxied: connected.is_proxied,
tx: tx,
}))
})
})
})
};
let race = checkout.select(connect).map(|either| {
either.either(|(pooled, _)| pooled, |(pooled, _)| pooled)
}).map_err(|either| {
// the Pool Checkout cannot error, so the only error
// is from the Connector
// XXX: should wait on the Checkout? Problem is
// that if the connector is failing, it may be that we
// never had a pooled stream at all
ClientError::Normal(either.either(|(e, _)| e, |(e, _)| e))
});
let race = checkout.select(connect)
.map(|(pooled, _work)| pooled)
.map_err(|(e, _checkout)| {
// the Pool Checkout cannot error, so the only error
// is from the Connector
// XXX: should wait on the Checkout? Problem is
// that if the connector is failing, it may be that we
// never had a pooled stream at all
ClientError::Normal(e)
});
let executor = self.executor.clone();
let resp = race.and_then(move |mut pooled| {
let conn_reused = pooled.is_reused();
set_relative_uri(req.uri_mut(), pooled.is_proxied);
@@ -237,35 +247,34 @@ where C: Connect<Error=io::Error> + Sync + 'static,
}
})
.and_then(move |mut res| {
future::lazy(move |cx| {
// when pooled is dropped, it will try to insert back into the
// pool. To delay that, spawn a future that completes once the
// sender is ready again.
//
// This *should* only be once the related `Connection` has polled
// for a new request to start.
//
// It won't be ready if there is a body to stream.
if pooled.tx.is_ready() {
drop(pooled);
} else if !res.body().is_empty() {
let (delayed_tx, delayed_rx) = oneshot::channel();
res.body_mut().delayed_eof(delayed_rx);
// If the executor doesn't have room, oh well. Things will likely
// be blowing up soon, but this specific task isn't required.
let fut = future::poll_fn(move |cx| {
pooled.tx.poll_ready(cx)
// when pooled is dropped, it will try to insert back into the
// pool. To delay that, spawn a future that completes once the
// sender is ready again.
//
// This *should* only be once the related `Connection` has polled
// for a new request to start.
//
// It won't be ready if there is a body to stream.
if pooled.tx.is_ready() {
drop(pooled);
} else if !res.body().is_empty() {
let (delayed_tx, delayed_rx) = oneshot::channel();
res.body_mut().delayed_eof(delayed_rx);
// If the executor doesn't have room, oh well. Things will likely
// be blowing up soon, but this specific task isn't required.
let _ = executor.execute(
future::poll_fn(move || {
pooled.tx.poll_ready()
})
.then(move |_| {
// At this point, `pooled` is dropped, and had a chance
// to insert into the pool (if conn was idle)
drop(delayed_tx);
Ok(())
});
execute(fut, cx).ok();
}
Ok(res)
})
.then(move |_| {
// At this point, `pooled` is dropped, and had a chance
// to insert into the pool (if conn was idle)
drop(delayed_tx);
Ok(())
})
);
}
Ok(res)
});
@@ -274,6 +283,10 @@ where C: Connect<Error=io::Error> + Sync + 'static,
Box::new(resp)
}
fn schedule_pool_timer(&self) {
self.pool.spawn_expired_interval(&self.executor);
}
}
impl<C, B> Service for Client<C, B>
@@ -296,6 +309,7 @@ impl<C, B> Clone for Client<C, B> {
fn clone(&self) -> Client<C, B> {
Client {
connector: self.connector.clone(),
executor: self.executor.clone(),
h1_writev: self.h1_writev,
pool: self.pool.clone(),
retry_canceled_requests: self.retry_canceled_requests,
@@ -325,8 +339,8 @@ impl Future for FutureResponse {
type Item = Response<Body>;
type Error = ::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
self.0.poll(cx)
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.0.poll()
}
}
@@ -347,11 +361,11 @@ where
type Item = Response<Body>;
type Error = ::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
match self.future.poll(cx) {
match self.future.poll() {
Ok(Async::Ready(resp)) => return Ok(Async::Ready(resp)),
Ok(Async::Pending) => return Ok(Async::Pending),
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(ClientError::Normal(err)) => return Err(err),
Err(ClientError::Canceled {
connection_reused,
@@ -559,7 +573,18 @@ where C: Connect<Error=io::Error>,
/// Construct the Client with this configuration.
#[inline]
pub fn build(self) -> Client<C, B> {
Client::configured(self)
Client::configured(self, Exec::Default)
}
/// Construct a Client with this configuration and an executor.
///
/// The executor will be used to spawn "background" connection tasks
/// to drive requests and responses.
pub fn executor<E>(self, executor: E) -> Client<C, B>
where
E: Executor<Background> + Send + Sync + 'static,
{
Client::configured(self, Exec::new(executor))
}
}
@@ -577,6 +602,21 @@ where
}
self.connector(connector).build()
}
/// Construct a Client with this configuration and an executor.
///
/// The executor will be used to spawn "background" connection tasks
/// to drive requests and responses.
pub fn build_with_executor<E>(self, handle: &Handle, executor: E) -> Client<HttpConnector, B>
where
E: Executor<Background> + Send + Sync + 'static,
{
let mut connector = HttpConnector::new(4, handle);
if self.keep_alive {
connector.set_keepalive(self.keep_alive_timeout);
}
self.connector(connector).executor(executor)
}
}
impl<C, B> fmt::Debug for Config<C, B> {
@@ -601,15 +641,68 @@ impl<C: Clone, B> Clone for Config<C, B> {
}
fn execute<F>(fut: F, cx: &mut task::Context) -> Result<(), ::Error>
where F: Future<Item=(), Error=Never> + Send + 'static,
{
if let Some(executor) = cx.executor() {
executor.spawn(Box::new(fut)).map_err(|err| {
debug!("executor error: {:?}", err);
::Error::Executor
})
} else {
Err(::Error::Executor)
// ===== impl Exec =====
#[derive(Clone)]
enum Exec {
Default,
Executor(Arc<Executor<Background> + Send + Sync>),
}
impl Exec {
pub(crate) fn new<E: Executor<Background> + Send + Sync + 'static>(executor: E) -> Exec {
Exec::Executor(Arc::new(executor))
}
fn execute<F>(&self, fut: F) -> io::Result<()>
where
F: Future<Item=(), Error=()> + Send + 'static,
{
match *self {
Exec::Default => spawn(fut),
Exec::Executor(ref e) => {
e.execute(bg(Box::new(fut)))
.map_err(|err| {
debug!("executor error: {:?}", err.kind());
io::Error::new(
io::ErrorKind::Other,
"executor error",
)
})?
},
}
Ok(())
}
}
// ===== impl Background =====
// The types inside this module are not exported out of the crate,
// so they are in essence un-nameable.
mod background {
use futures::{Future, Poll};
// This is basically `impl Future`, since the type is un-nameable,
// and only implementeds `Future`.
#[allow(missing_debug_implementations)]
pub struct Background {
inner: Box<Future<Item=(), Error=()> + Send>,
}
pub fn bg(fut: Box<Future<Item=(), Error=()> + Send>) -> Background {
Background {
inner: fut,
}
}
impl Future for Background {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
}
}
}

133
src/client/pool.rs
View File

@@ -4,13 +4,15 @@ use std::ops::{Deref, DerefMut};
use std::sync::{Arc, Mutex, Weak};
use std::time::{Duration, Instant};
use futures::{Future, Async, Never, Poll, Stream};
use futures::channel::oneshot;
use futures::task;
use futures::{Future, Async, Poll, Stream};
use futures::sync::oneshot;
use futures_timer::Interval;
pub(super) struct Pool<T> {
inner: Arc<Mutex<PoolInner<T>>>
use common::Never;
use super::Exec;
pub struct Pool<T> {
inner: Arc<Mutex<PoolInner<T>>>,
}
// Before using a pooled connection, make sure the sender is not dead.
@@ -44,7 +46,7 @@ struct PoolInner<T> {
}
impl<T> Pool<T> {
pub(super) fn new(enabled: bool, timeout: Option<Duration>) -> Pool<T> {
pub fn new(enabled: bool, timeout: Option<Duration>) -> Pool<T> {
Pool {
inner: Arc::new(Mutex::new(PoolInner {
enabled: enabled,
@@ -52,7 +54,7 @@ impl<T> Pool<T> {
idle_interval_ref: None,
parked: HashMap::new(),
timeout: timeout,
}))
})),
}
}
}
@@ -63,7 +65,6 @@ impl<T: Closed> Pool<T> {
key: Arc::new(key.to_owned()),
pool: self.clone(),
parked: None,
spawned_expired_interval: false
}
}
@@ -218,16 +219,16 @@ impl<T: Closed> PoolInner<T> {
impl<T: Closed + Send + 'static> Pool<T> {
fn spawn_expired_interval(&mut self, cx: &mut task::Context) -> Result<(), ::Error> {
pub(super) fn spawn_expired_interval(&self, exec: &Exec) {
let (dur, rx) = {
let mut inner = self.inner.lock().unwrap();
if !inner.enabled {
return Ok(());
return;
}
if inner.idle_interval_ref.is_some() {
return Ok(());
return;
}
if let Some(dur) = inner.timeout {
@@ -235,23 +236,23 @@ impl<T: Closed + Send + 'static> Pool<T> {
inner.idle_interval_ref = Some(tx);
(dur, rx)
} else {
return Ok(());
return
}
};
let interval = Interval::new(dur);
super::execute(IdleInterval {
exec.execute(IdleInterval {
interval: interval,
pool: Arc::downgrade(&self.inner),
pool_drop_notifier: rx,
}, cx)
}).unwrap();
}
}
impl<T> Clone for Pool<T> {
fn clone(&self) -> Pool<T> {
Pool {
inner: self.inner.clone()
inner: self.inner.clone(),
}
}
}
@@ -321,23 +322,22 @@ pub struct Checkout<T> {
key: Arc<String>,
pool: Pool<T>,
parked: Option<oneshot::Receiver<T>>,
spawned_expired_interval: bool
}
struct NotParked;
impl<T: Closed> Checkout<T> {
fn poll_parked(&mut self, cx: &mut task::Context) -> Poll<Pooled<T>, NotParked> {
fn poll_parked(&mut self) -> Poll<Pooled<T>, NotParked> {
let mut drop_parked = false;
if let Some(ref mut rx) = self.parked {
match rx.poll(cx) {
match rx.poll() {
Ok(Async::Ready(value)) => {
if !value.is_closed() {
return Ok(Async::Ready(self.pool.reuse(&self.key, value)));
}
drop_parked = true;
},
Ok(Async::Pending) => return Ok(Async::Pending),
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(_canceled) => drop_parked = true,
}
}
@@ -347,27 +347,22 @@ impl<T: Closed> Checkout<T> {
Err(NotParked)
}
fn park(&mut self, cx: &mut task::Context) {
fn park(&mut self) {
if self.parked.is_none() {
let (tx, mut rx) = oneshot::channel();
let _ = rx.poll(cx); // park this task
let _ = rx.poll(); // park this task
self.pool.park(self.key.clone(), tx);
self.parked = Some(rx);
}
}
}
impl<T: Closed + Send + 'static> Future for Checkout<T> {
impl<T: Closed> Future for Checkout<T> {
type Item = Pooled<T>;
type Error = ::Error;
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
if !self.spawned_expired_interval {
self.pool.spawn_expired_interval(cx)?;
self.spawned_expired_interval = true;
}
match self.poll_parked(cx) {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.poll_parked() {
Ok(async) => return Ok(async),
Err(_not_parked) => (),
}
@@ -377,8 +372,8 @@ impl<T: Closed + Send + 'static> Future for Checkout<T> {
if let Some(pooled) = entry {
Ok(Async::Ready(pooled))
} else {
self.park(cx);
Ok(Async::Pending)
self.park();
Ok(Async::NotReady)
}
}
}
@@ -418,20 +413,20 @@ struct IdleInterval<T> {
impl<T: Closed + 'static> Future for IdleInterval<T> {
type Item = ();
type Error = Never;
type Error = ();
fn poll(&mut self, cx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
match self.pool_drop_notifier.poll(cx) {
match self.pool_drop_notifier.poll() {
Ok(Async::Ready(n)) => match n {},
Ok(Async::Pending) => (),
Ok(Async::NotReady) => (),
Err(_canceled) => {
trace!("pool closed, canceling idle interval");
return Ok(Async::Ready(()));
}
}
try_ready!(self.interval.poll_next(cx).map_err(|_| unreachable!("interval cannot error")));
try_ready!(self.interval.poll().map_err(|_| unreachable!("interval cannot error")));
if let Some(inner) = self.pool.upgrade() {
if let Ok(mut inner) = inner.lock() {
@@ -448,10 +443,9 @@ impl<T: Closed + 'static> Future for IdleInterval<T> {
mod tests {
use std::sync::Arc;
use std::time::Duration;
use futures::{Async, Future, FutureExt};
use futures::{Async, Future};
use futures::future;
use futures::executor::block_on;
use super::{Closed, Pool};
use super::{Closed, Pool, Exec};
impl Closed for i32 {
fn is_closed(&self) -> bool {
@@ -461,37 +455,34 @@ mod tests {
#[test]
fn test_pool_checkout_smoke() {
block_on(future::lazy(|cx| {
let pool = Pool::new(true, Some(Duration::from_secs(5)));
let key = Arc::new("foo".to_string());
let pooled = pool.pooled(key.clone(), 41);
let pool = Pool::new(true, Some(Duration::from_secs(5)));
let key = Arc::new("foo".to_string());
let pooled = pool.pooled(key.clone(), 41);
drop(pooled);
drop(pooled);
match pool.checkout(&key).poll(cx).unwrap() {
Async::Ready(pooled) => assert_eq!(*pooled, 41),
_ => panic!("not ready"),
}
Ok::<_, ()>(())
})).unwrap();
match pool.checkout(&key).poll().unwrap() {
Async::Ready(pooled) => assert_eq!(*pooled, 41),
_ => panic!("not ready"),
}
}
#[test]
fn test_pool_checkout_returns_none_if_expired() {
block_on(future::lazy(|cx| {
future::lazy(|| {
let pool = Pool::new(true, Some(Duration::from_millis(100)));
let key = Arc::new("foo".to_string());
let pooled = pool.pooled(key.clone(), 41);
drop(pooled);
::std::thread::sleep(pool.inner.lock().unwrap().timeout.unwrap());
assert!(pool.checkout(&key).poll(cx).unwrap().is_pending());
Ok::<_, ()>(())
})).unwrap();
assert!(pool.checkout(&key).poll().unwrap().is_not_ready());
::futures::future::ok::<(), ()>(())
}).wait().unwrap();
}
#[test]
fn test_pool_checkout_removes_expired() {
block_on(future::lazy(|cx| {
future::lazy(|| {
let pool = Pool::new(true, Some(Duration::from_millis(100)));
let key = Arc::new("foo".to_string());
@@ -503,20 +494,20 @@ mod tests {
::std::thread::sleep(pool.inner.lock().unwrap().timeout.unwrap());
// checkout.poll() should clean out the expired
pool.checkout(&key).poll(cx).unwrap();
pool.checkout(&key).poll().unwrap();
assert!(pool.inner.lock().unwrap().idle.get(&key).is_none());
Ok::<_, ()>(())
})).unwrap();
Ok::<(), ()>(())
}).wait().unwrap();
}
#[test]
fn test_pool_timer_removes_expired() {
let mut pool = Pool::new(true, Some(Duration::from_millis(100)));
let runtime = ::tokio::runtime::Runtime::new().unwrap();
let pool = Pool::new(true, Some(Duration::from_millis(100)));
block_on(future::lazy(|cx| {
pool.spawn_expired_interval(cx)
})).unwrap();
let executor = runtime.executor();
pool.spawn_expired_interval(&Exec::new(executor));
let key = Arc::new("foo".to_string());
pool.pooled(key.clone(), 41);
@@ -525,9 +516,9 @@ mod tests {
assert_eq!(pool.inner.lock().unwrap().idle.get(&key).map(|entries| entries.len()), Some(3));
block_on(::futures_timer::Delay::new(
::futures_timer::Delay::new(
Duration::from_millis(400) // allow for too-good resolution
)).unwrap();
).wait().unwrap();
assert!(pool.inner.lock().unwrap().idle.get(&key).is_none());
}
@@ -538,7 +529,7 @@ mod tests {
let key = Arc::new("foo".to_string());
let pooled = pool.pooled(key.clone(), 41);
let checkout = pool.checkout(&key).join(future::lazy(move |_| {
let checkout = pool.checkout(&key).join(future::lazy(move || {
// the checkout future will park first,
// and then this lazy future will be polled, which will insert
// the pooled back into the pool
@@ -547,12 +538,12 @@ mod tests {
drop(pooled);
Ok(())
})).map(|(entry, _)| entry);
assert_eq!(*block_on(checkout).unwrap(), 41);
assert_eq!(*checkout.wait().unwrap(), 41);
}
#[test]
fn test_pool_checkout_drop_cleans_up_parked() {
block_on(future::lazy(|cx| {
future::lazy(|| {
let pool = Pool::<i32>::new(true, Some(Duration::from_secs(10)));
let key = Arc::new("localhost:12345".to_string());
@@ -560,9 +551,9 @@ mod tests {
let mut checkout2 = pool.checkout(&key);
// first poll needed to get into Pool's parked
checkout1.poll(cx).unwrap();
checkout1.poll().unwrap();
assert_eq!(pool.inner.lock().unwrap().parked.get(&key).unwrap().len(), 1);
checkout2.poll(cx).unwrap();
checkout2.poll().unwrap();
assert_eq!(pool.inner.lock().unwrap().parked.get(&key).unwrap().len(), 2);
// on drop, clean up Pool
@@ -572,7 +563,7 @@ mod tests {
drop(checkout2);
assert!(pool.inner.lock().unwrap().parked.get(&key).is_none());
Ok::<_, ()>(())
})).unwrap();
::futures::future::ok::<(), ()>(())
}).wait().unwrap();
}
}

37
src/client/tests.rs
View File

@@ -5,7 +5,6 @@ use std::time::Duration;
use futures::Async;
use futures::future::poll_fn;
use futures::executor::block_on;
use tokio::executor::thread_pool::{Builder as ThreadPoolBuilder};
use mock::MockConnector;
@@ -23,7 +22,7 @@ fn retryable_request() {
let client = Client::configure()
.connector(connector)
.build();
.executor(executor.sender().clone());
{
@@ -31,13 +30,13 @@ fn retryable_request() {
.uri("http://mock.local/a")
.body(Default::default())
.unwrap();
let res1 = client.request(req).with_executor(executor.sender().clone());
let srv1 = poll_fn(|cx| {
try_ready!(sock1.read(cx, &mut [0u8; 512]));
let res1 = client.request(req);
let srv1 = poll_fn(|| {
try_ready!(sock1.read(&mut [0u8; 512]));
try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(()))
});
block_on(res1.join(srv1)).expect("res1");
res1.join(srv1).wait().expect("res1");
}
drop(sock1);
@@ -45,17 +44,17 @@ fn retryable_request() {
.uri("http://mock.local/b")
.body(Default::default())
.unwrap();
let res2 = client.request(req).with_executor(executor.sender().clone())
let res2 = client.request(req)
.map(|res| {
assert_eq!(res.status().as_u16(), 222);
});
let srv2 = poll_fn(|cx| {
try_ready!(sock2.read(cx, &mut [0u8; 512]));
let srv2 = poll_fn(|| {
try_ready!(sock2.read(&mut [0u8; 512]));
try_ready!(sock2.write(b"HTTP/1.1 222 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(()))
});
block_on(res2.join(srv2)).expect("res2");
res2.join(srv2).wait().expect("res2");
}
#[test]
@@ -69,7 +68,7 @@ fn conn_reset_after_write() {
let client = Client::configure()
.connector(connector)
.build();
.executor(executor.sender().clone());
{
let req = Request::builder()
@@ -78,13 +77,13 @@ fn conn_reset_after_write() {
.header("content-length", "0")
.body(Default::default())
.unwrap();
let res1 = client.request(req).with_executor(executor.sender().clone());
let srv1 = poll_fn(|cx| {
try_ready!(sock1.read(cx, &mut [0u8; 512]));
let res1 = client.request(req);
let srv1 = poll_fn(|| {
try_ready!(sock1.read(&mut [0u8; 512]));
try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
Ok(Async::Ready(()))
});
block_on(res1.join(srv1)).expect("res1");
res1.join(srv1).wait().expect("res1");
}
// sleep to allow some time for the connection to return to the pool
@@ -94,20 +93,20 @@ fn conn_reset_after_write() {
.uri("http://mock.local/a")
.body(Default::default())
.unwrap();
let res2 = client.request(req).with_executor(executor.sender().clone());
let res2 = client.request(req);
let mut sock1 = Some(sock1);
let srv2 = poll_fn(|cx| {
let srv2 = poll_fn(|| {
// We purposefully keep the socket open until the client
// has written the second request, and THEN disconnect.
//
// Not because we expect servers to be jerks, but to trigger
// state where we write on an assumedly good connetion, and
// only reset the close AFTER we wrote bytes.
try_ready!(sock1.as_mut().unwrap().read(cx, &mut [0u8; 512]));
try_ready!(sock1.as_mut().unwrap().read(&mut [0u8; 512]));
sock1.take();
Ok(Async::Ready(()))
});
let err = block_on(res2.join(srv2)).expect_err("res2");
let err = res2.join(srv2).wait().expect_err("res2");
match err {
::Error::Incomplete => (),
other => panic!("expected Incomplete, found {:?}", other)