Repository: tower-rs/tower
Pull request #379 (2019-12-05 05:25:28)

Update the rest of the crates and upgrade ready cache to `std::f…
This PR adds the rest of the crates and also converts `tower-ready-cache` to `std::future::Future`.

I tried to somewhat break up the commits into pieces that should be reviewed together. Mostly, around the ready cache upgrade.

After this PR I plan to do a pass of the crates to clean things up.

LucioFranco: I'd appreciate if we could get a review of this today, so I can finish up the tonic work :) <3

Patch:
diff --git a/Cargo.toml b/Cargo.toml
index 7921d2f..6a82d11 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -2,20 +2,20 @@
 
 members = [
   "tower",
-  # "tower-balance",
+  "tower-balance",
   "tower-buffer",
-  # "tower-discover",
-  # "tower-filter",
-  # "tower-hedge",
+  "tower-discover",
+  "tower-filter",
+  "tower-hedge",
   "tower-layer",
   "tower-limit",
-  # "tower-load",
-  # "tower-load-shed",
-  # "tower-ready-cache",
+  "tower-load",
+  "tower-load-shed",
+  "tower-ready-cache",
   "tower-reconnect",
   "tower-retry",
   "tower-service",
-  # "tower-spawn-ready",
+  "tower-spawn-ready",
   "tower-test",
   "tower-timeout",
   "tower-make",
diff --git a/tower-balance/CHANGELOG.md b/tower-balance/CHANGELOG.md
index 19df3db..1aa198f 100644
--- a/tower-balance/CHANGELOG.md
+++ b/tower-balance/CHANGELOG.md
@@ -1,3 +1,9 @@
+# 0.3.0 (December 4, 2019)
+
+- Update to `tower-service 0.3`
+- Update to `tower-ready-cache 0.3`
+- Update to `futures 0.3`
+
 # 0.3.0-alpha.2 (September 30, 2019)
 
 - Move to `futures-*-preview 0.3.0-alpha.19`
diff --git a/tower-balance/Cargo.toml b/tower-balance/Cargo.toml
index 1e71432..dac2f27 100644
--- a/tower-balance/Cargo.toml
+++ b/tower-balance/Cargo.toml
@@ -8,13 +8,13 @@ name = "tower-balance"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tower-rs/tower"
 homepage = "https://github.com/tower-rs/tower"
-documentation = "https://docs.rs/tower-balance/0.3.0-alpha.2"
+documentation = "https://docs.rs/tower-balance/0.3.0"
 description = """
 Balance load across a set of uniform services.
 """
@@ -26,29 +26,28 @@ log = ["tracing/log"]
 default = ["log"]
 
 [dependencies]
-futures-util-preview = "=0.3.0-alpha.19"
-futures-core-preview = "=0.3.0-alpha.19"
+futures-util = "0.3"
+futures-core = "0.3"
 pin-project = "0.4"
 indexmap = "1.0.2"
 tracing = "0.1"
 rand = "0.6.5"
-tokio-sync = "=0.2.0-alpha.6"
-tokio-timer = "=0.3.0-alpha.6"
-tower-discover = { version = "=0.3.0-alpha.2", path = "../tower-discover" }
-tower-layer = { version = "=0.3.0-alpha.2", path = "../tower-layer" }
-tower-load = { version = "=0.3.0-alpha.2", path = "../tower-load" }
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
-tower-make = { version = "=0.3.0-alpha.2a", path = "../tower-make" }
+tokio = { version = "0.2", features = ["sync", "time"] }
+tower-discover = { version = "0.3", path = "../tower-discover" }
+tower-layer = { version = "0.3", path = "../tower-layer" }
+tower-load = { version = "0.3", path = "../tower-load" }
+tower-service = { version = "0.3", path = "../tower-service" }
+tower-ready-cache = { version = "0.3", path = "../tower-ready-cache" }
+tower-make = { version = "0.3", path = "../tower-make" }
 slab = "0.4"
 
 [dev-dependencies]
 tracing-subscriber = "0.1.1"
 hdrhistogram = "6.0"
 quickcheck = { version = "0.6", default-features = false }
-tokio = "=0.2.0-alpha.6"
-tokio-executor = "=0.2.0-alpha.6"
-tokio-test = "=0.2.0-alpha.6"
-tower-buffer = { version = "=0.3.0-alpha.2", path = "../tower-buffer" }
-tower-limit = { version = "=0.3.0-alpha.2", path = "../tower-limit" }
-tower-test = { version = "=0.3.0-alpha.2", path = "../tower-test" }
-tower = { version = "=0.3.0-alpha.2", path = "../tower" }
+tokio = { version = "0.2", features = ["macros"] } 
+tokio-test = "0.2"
+tower-buffer = { version = "0.3", path = "../tower-buffer" }
+tower-limit = { version = "0.3", path = "../tower-limit" }
+tower-test = { version = "0.3", path = "../tower-test" }
+tower = { version = "0.3", path = "../tower" }
diff --git a/tower-balance/examples/demo.rs b/tower-balance/examples/demo.rs
index 4244639..88a6996 100644
--- a/tower-balance/examples/demo.rs
+++ b/tower-balance/examples/demo.rs
@@ -1,16 +1,16 @@
 //! Exercises load balancers with mocked services.
 
 use futures_core::TryStream;
-use futures_util::{stream, stream::StreamExt, try_stream::TryStreamExt};
+use futures_util::{stream, stream::StreamExt, stream::TryStreamExt};
 use hdrhistogram::Histogram;
 use pin_project::pin_project;
 use rand::{self, Rng};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio::timer;
+use tokio::time::{self, Instant};
 use tower::util::ServiceExt;
 use tower_balance as lb;
 use tower_discover::{Change, Discover};
@@ -116,7 +116,7 @@ fn gen_disco() -> impl Discover<
                     let latency = Duration::from_millis(rand::thread_rng().gen_range(0, maxms));
 
                     async move {
-                        timer::delay(start + latency).await;
+                        time::delay_until(start + latency).await;
                         let latency = start.elapsed();
                         Ok(Rsp { latency, instance })
                     }
@@ -140,7 +140,7 @@ where
 {
     println!("{}", name);
 
-    let requests = stream::repeat(Req).take(REQUESTS as u64);
+    let requests = stream::repeat(Req).take(REQUESTS);
     let service = ConcurrencyLimit::new(lb, CONCURRENCY);
     let responses = service.call_all(requests).unordered();
 
diff --git a/tower-balance/src/lib.rs b/tower-balance/src/lib.rs
index 7981753..2c0006f 100644
--- a/tower-balance/src/lib.rs
+++ b/tower-balance/src/lib.rs
@@ -1,6 +1,6 @@
 //! Load balancing middlewares.
 
-#![doc(html_root_url = "https://docs.rs/tower-balance/0.3.0-alpha.2")]
+#![doc(html_root_url = "https://docs.rs/tower-balance/0.3.0")]
 #![warn(
     missing_debug_implementations,
     missing_docs,
diff --git a/tower-balance/src/p2c/layer.rs b/tower-balance/src/p2c/layer.rs
index a90747a..7a92c62 100644
--- a/tower-balance/src/p2c/layer.rs
+++ b/tower-balance/src/p2c/layer.rs
@@ -1,5 +1,5 @@
 use super::BalanceMake;
-use rand::{rngs::SmallRng, Rng, SeedableRng};
+use rand::{rngs::SmallRng, FromEntropy, Rng, SeedableRng};
 use std::{fmt, marker::PhantomData};
 use tower_layer::Layer;
 
diff --git a/tower-balance/src/p2c/make.rs b/tower-balance/src/p2c/make.rs
index 23a1b00..68c9028 100644
--- a/tower-balance/src/p2c/make.rs
+++ b/tower-balance/src/p2c/make.rs
@@ -1,4 +1,5 @@
 use super::Balance;
+use crate::error;
 use futures_core::ready;
 use pin_project::pin_project;
 use rand::{rngs::SmallRng, FromEntropy};
@@ -73,6 +74,7 @@ where
     F: Future<Output = Result<T, E>>,
     T: Discover,
     <T as Discover>::Service: Service<Req>,
+    <<T as Discover>::Service as Service<Req>>::Error: Into<error::Error>,
 {
     type Output = Result<Balance<T, Req>, E>;
 
diff --git a/tower-balance/src/p2c/service.rs b/tower-balance/src/p2c/service.rs
index 170fac5..7b0cbff 100644
--- a/tower-balance/src/p2c/service.rs
+++ b/tower-balance/src/p2c/service.rs
@@ -1,7 +1,6 @@
 use crate::error;
-use futures_core::{ready, Stream};
-use futures_util::{stream, try_future, try_future::TryFutureExt};
-use indexmap::IndexMap;
+use futures_core::ready;
+use futures_util::future::{self, TryFutureExt};
 use pin_project::pin_project;
 use rand::{rngs::SmallRng, FromEntropy};
 use std::marker::PhantomData;
@@ -11,7 +10,7 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio_sync::oneshot;
+use tokio::sync::oneshot;
 use tower_discover::{Change, Discover};
 use tower_load::Load;
 use tower_ready_cache::{error::Failed, ReadyCache};
@@ -54,14 +53,12 @@ where
     D: fmt::Debug,
     D::Key: fmt::Debug,
     D::Service: fmt::Debug,
+    Req: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Balance")
             .field("discover", &self.discover)
-            .field("ready_services", &self.ready_services)
-            .field("unready_services", &self.unready_services)
-            .field("cancelations", &self.cancelations)
-            .field("next_ready_index", &self.next_ready_index)
+            .field("services", &self.services)
             .finish()
     }
 }
@@ -96,10 +93,8 @@ where
         Self {
             rng,
             discover,
-            ready_services: IndexMap::default(),
-            cancelations: IndexMap::default(),
-            unready_services: stream::FuturesUnordered::new(),
-            next_ready_index: None,
+            services: ReadyCache::default(),
+            ready_index: None,
 
             _req: PhantomData,
         }
@@ -114,10 +109,6 @@ where
     pub fn len(&self) -> usize {
         self.services.len()
     }
-
-    pub(crate) fn discover_mut(&mut self) -> &mut D {
-        &mut self.discover
-    }
 }
 
 impl<D, Req> Balance<D, Req>
@@ -132,7 +123,10 @@ where
     /// Polls `discover` for updates, adding new items to `not_ready`.
     ///
     /// Removals may alter the order of either `ready` or `not_ready`.
-    fn poll_discover(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), error::Discover>> {
+    fn update_pending_from_discover(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), error::Discover>> {
         debug!("updating from discover");
         loop {
             match ready!(Pin::new(&mut self.discover).poll_discover(cx))
@@ -142,7 +136,7 @@ where
                     trace!("remove");
                     self.services.evict(&key);
                 }
-                Async::Ready(Change::Insert(key, svc)) => {
+                Change::Insert(key, svc) => {
                     trace!("insert");
                     // If this service already existed in the set, it will be
                     // replaced as the new one becomes ready.
@@ -152,47 +146,23 @@ where
         }
     }
 
-    fn push_unready(&mut self, key: D::Key, svc: D::Service) {
-        let (tx, rx) = oneshot::channel();
-        self.cancelations.insert(key.clone(), tx);
-        self.unready_services.push(UnreadyService {
-            key: Some(key),
-            service: Some(svc),
-            cancel: rx,
-            _req: PhantomData,
-        });
-    }
-
-    fn evict(&mut self, key: &D::Key) {
-        // Update the ready index to account for reordering of ready.
-        if let Some((idx, _, _)) = self.ready_services.swap_remove_full(key) {
-            self.next_ready_index = self
-                .next_ready_index
-                .and_then(|i| Self::repair_index(i, idx, self.ready_services.len()));
-            debug_assert!(!self.cancelations.contains_key(key));
-        } else if let Some(cancel) = self.cancelations.swap_remove(key) {
-            let _ = cancel.send(());
-        }
-    }
-
-    fn poll_unready(&mut self, cx: &mut Context<'_>) {
+    fn promote_pending_to_ready(&mut self, cx: &mut Context<'_>) {
         loop {
-            match Pin::new(&mut self.unready_services).poll_next(cx) {
-                Poll::Pending | Poll::Ready(None) => return,
-                Poll::Ready(Some(Ok((key, svc)))) => {
-                    trace!("endpoint ready");
-                    let _cancel = self.cancelations.swap_remove(&key);
-                    debug_assert!(_cancel.is_some(), "missing cancelation");
-                    self.ready_services.insert(key, svc);
+            match self.services.poll_pending(cx) {
+                Poll::Ready(Ok(())) => {
+                    // There are no remaining pending services.
+                    debug_assert_eq!(self.services.pending_len(), 0);
+                    break;
                 }
-                Poll::Ready(Some(Err((key, Error::Canceled)))) => {
-                    debug_assert!(!self.cancelations.contains_key(&key))
+                Poll::Pending => {
+                    // None of the pending services are ready.
+                    debug_assert!(self.services.pending_len() > 0);
+                    break;
                 }
-                Poll::Ready(Some(Err((key, Error::Inner(e))))) => {
-                    let error = e.into();
-                    debug!({ %error }, "dropping failed endpoint");
-                    let _cancel = self.cancelations.swap_remove(&key);
-                    debug_assert!(_cancel.is_some());
+                Poll::Ready(Err(error)) => {
+                    // An individual service was lost; continue processing
+                    // pending services.
+                    debug!(%error, "dropping failed endpoint");
                 }
             }
         }
@@ -240,37 +210,8 @@ where
         svc.load()
     }
 
-    fn poll_ready_index_or_evict(
-        &mut self,
-        cx: &mut Context<'_>,
-        index: usize,
-    ) -> Poll<Result<(), ()>> {
-        let (_, svc) = self
-            .ready_services
-            .get_index_mut(index)
-            .expect("invalid index");
-
-        match svc.poll_ready(cx) {
-            Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
-            Poll::Pending => {
-                // became unready; so move it back there.
-                let (key, svc) = self
-                    .ready_services
-                    .swap_remove_index(index)
-                    .expect("invalid ready index");
-                self.push_unready(key, svc);
-                Poll::Pending
-            }
-            Poll::Ready(Err(e)) => {
-                // failed, so drop it.
-                let error = e.into();
-                debug!({ %error }, "evicting failed endpoint");
-                self.ready_services
-                    .swap_remove_index(index)
-                    .expect("invalid ready index");
-                Poll::Ready(Err(()))
-            }
-        }
+    pub(crate) fn discover_mut(&mut self) -> &mut D {
+        &mut self.discover
     }
 }
 
@@ -285,59 +226,60 @@ where
 {
     type Response = <D::Service as Service<Req>>::Response;
     type Error = error::Error;
-    type Future = try_future::MapErr<
+    type Future = future::MapErr<
         <D::Service as Service<Req>>::Future,
         fn(<D::Service as Service<Req>>::Error) -> error::Error,
     >;
 
-    /// Prepares the balancer to process a request.
-    ///
-    /// When `Async::Ready` is returned, `ready_index` is set with a valid index
-    /// into `ready` referring to a `Service` that is ready to disptach a request.
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        // First and foremost, process discovery updates. This removes or updates a
-        // previously-selected `ready_index` if appropriate.
-        let _ = self.poll_discover(cx)?;
-
-        // Drive new or busy services to readiness.
-        self.poll_unready(cx);
-        trace!({ nready = self.ready_services.len(), nunready = self.unready_services.len() }, "poll_ready");
+        // `ready_index` may have already been set by a prior invocation. These
+        // updates cannot disturb the order of existing ready services.
+        let _ = self.update_pending_from_discover(cx)?;
+        self.promote_pending_to_ready(cx);
 
         loop {
             // If a service has already been selected, ensure that it is ready.
             // This ensures that the underlying service is ready immediately
-            // before a request is dispatched to it. If, e.g., a failure
-            // detector has changed the state of the service, it may be evicted
-            // from the ready set so that P2C can be performed again.
-            if let Some(index) = self.next_ready_index {
-                trace!({ next.idx = index }, "preselected ready_index");
-                debug_assert!(index < self.ready_services.len());
-
-                if let Poll::Ready(Ok(())) = self.poll_ready_index_or_evict(cx, index) {
-                    return Poll::Ready(Ok(()));
+            // before a request is dispatched to it (i.e. in the same task
+            // invocation). If, e.g., a failure detector has changed the state
+            // of the service, it may be evicted from the ready set so that
+            // another service can be selected.
+            if let Some(index) = self.ready_index.take() {
+                match self.services.check_ready_index(cx, index) {
+                    Ok(true) => {
+                        // The service remains ready.
+                        self.ready_index = Some(index);
+                        return Poll::Ready(Ok(()));
+                    }
+                    Ok(false) => {
+                        // The service is no longer ready. Try to find a new one.
+                        trace!("ready service became unavailable");
+                    }
+                    Err(Failed(_, error)) => {
+                        // The ready endpoint failed, so log the error and try
+                        // to find a new one.
+                        debug!(%error, "endpoint failed");
+                    }
                 }
             }
 
-            self.next_ready_index = self.p2c_next_ready_index();
-            if self.next_ready_index.is_none() {
-                debug_assert!(self.ready_services.is_empty());
+            // Select a new service by comparing two at random and using the
+            // lesser-loaded service.
+            self.ready_index = self.p2c_ready_index();
+            if self.ready_index.is_none() {
+                debug_assert_eq!(self.services.ready_len(), 0);
+                // We have previously registered interest in updates from
+                // discover and pending services.
                 return Poll::Pending;
             }
         }
     }
 
     fn call(&mut self, request: Req) -> Self::Future {
-        let index = self.next_ready_index.take().expect("not ready");
-        let (key, mut svc) = self
-            .ready_services
-            .swap_remove_index(index)
-            .expect("invalid ready index");
-        // no need to repair since the ready_index has been cleared.
-
-        let fut = svc.call(request);
-        self.push_unready(key, svc);
-
-        fut.map_err(Into::into)
+        let index = self.ready_index.take().expect("called before ready");
+        self.services
+            .call_ready_index(index, request)
+            .map_err(Into::into)
     }
 }
 
diff --git a/tower-balance/src/pool/mod.rs b/tower-balance/src/pool/mod.rs
index a1f98f0..46efb14 100644
--- a/tower-balance/src/pool/mod.rs
+++ b/tower-balance/src/pool/mod.rs
@@ -56,9 +56,9 @@ where
     target: Target,
     load: Level,
     services: Slab<()>,
-    died_tx: tokio_sync::mpsc::UnboundedSender<usize>,
+    died_tx: tokio::sync::mpsc::UnboundedSender<usize>,
     #[pin]
-    died_rx: tokio_sync::mpsc::UnboundedReceiver<usize>,
+    died_rx: tokio::sync::mpsc::UnboundedReceiver<usize>,
     limit: Option<usize>,
 }
 
@@ -282,7 +282,7 @@ impl Builder {
         MS::Error: Into<error::Error>,
         Target: Clone,
     {
-        let (died_tx, died_rx) = tokio_sync::mpsc::unbounded_channel();
+        let (died_tx, died_rx) = tokio::sync::mpsc::unbounded_channel();
         let d = PoolDiscoverer {
             maker: make_service,
             making: None,
@@ -323,6 +323,7 @@ where
     MS::Error: Into<error::Error>,
     Target: Clone + fmt::Debug,
     MS::Service: fmt::Debug,
+    Request: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Pool")
@@ -433,12 +434,12 @@ where
 pub struct DropNotifyService<Svc> {
     svc: Svc,
     id: usize,
-    notify: tokio_sync::mpsc::UnboundedSender<usize>,
+    notify: tokio::sync::mpsc::UnboundedSender<usize>,
 }
 
 impl<Svc> Drop for DropNotifyService<Svc> {
     fn drop(&mut self) {
-        let _ = self.notify.try_send(self.id).is_ok();
+        let _ = self.notify.send(self.id).is_ok();
     }
 }
 
diff --git a/tower-discover/CHANGELOG.md b/tower-discover/CHANGELOG.md
index 633a866..4c08e82 100644
--- a/tower-discover/CHANGELOG.md
+++ b/tower-discover/CHANGELOG.md
@@ -1,3 +1,8 @@
+# 0.3.0 (December 4, 2019)
+
+- Update to `tower-service 0.3`
+- Update to `futures-core 0.3`
+
 # 0.3.0-alpha.2 (September 30, 2019)
 
 - Move to `futures-*-preview 0.3.0-alpha.19`
diff --git a/tower-discover/Cargo.toml b/tower-discover/Cargo.toml
index 5a7d261..5c3c24a 100644
--- a/tower-discover/Cargo.toml
+++ b/tower-discover/Cargo.toml
@@ -8,13 +8,13 @@ name = "tower-discover"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tower-rs/tower"
 homepage = "https://github.com/tower-rs/tower"
-documentation = "https://docs.rs/tower-discover/0.3.0-alpha.2"
+documentation = "https://docs.rs/tower-discover/0.3.0"
 description = """
 Abstracts over service discovery strategies.
 """
@@ -22,6 +22,6 @@ categories = ["asynchronous", "network-programming"]
 edition = "2018"
 
 [dependencies]
-futures-core-preview = "=0.3.0-alpha.19"
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
+futures-core = "0.3"
+tower-service = { version = "0.3", path = "../tower-service" }
 pin-project = "0.4"
diff --git a/tower-discover/src/lib.rs b/tower-discover/src/lib.rs
index fb59a3d..27d8fc3 100644
--- a/tower-discover/src/lib.rs
+++ b/tower-discover/src/lib.rs
@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tower-discover/0.3.0-alpha.2")]
+#![doc(html_root_url = "https://docs.rs/tower-discover/0.3.0")]
 #![warn(
     missing_debug_implementations,
     missing_docs,
diff --git a/tower-filter/Cargo.toml b/tower-filter/Cargo.toml
index 516d47b..5b711dc 100644
--- a/tower-filter/Cargo.toml
+++ b/tower-filter/Cargo.toml
@@ -8,7 +8,7 @@ name = "tower-filter"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
@@ -24,13 +24,12 @@ edition = "2018"
 publish = false
 
 [dependencies]
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
-tower-layer = { version = "=0.3.0-alpha.2", path = "../tower-layer" }
+tower-service = { version = "0.3", path = "../tower-service" }
+tower-layer = { version = "0.3", path = "../tower-layer" }
 pin-project = "0.4"
-futures-core-preview = "=0.3.0-alpha.19"
+futures-core = "0.3"
 
 [dev-dependencies]
-tower-test = { version = "=0.3.0-alpha.2", path = "../tower-test" }
-tokio-test = "=0.2.0-alpha.6"
-tokio = "=0.2.0-alpha.6"
-futures-util-preview = "=0.3.0-alpha.19"
+tower-test = { version = "0.3", path = "../tower-test" }
+futures-util = "0.3"
+tokio = { version = "0.2", features = ["macros", "sync"] }
diff --git a/tower-hedge/CHANGELOG.md b/tower-hedge/CHANGELOG.md
new file mode 100644
index 0000000..362ff4f
--- /dev/null
+++ b/tower-hedge/CHANGELOG.md
@@ -0,0 +1,3 @@
+# 0.3.0 (December 4, 2019)
+
+- Initial release
\ No newline at end of file
diff --git a/tower-hedge/Cargo.toml b/tower-hedge/Cargo.toml
index 0df8099..3480f8b 100644
--- a/tower-hedge/Cargo.toml
+++ b/tower-hedge/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "tower-hedge"
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Alex Leong <adlleong@gmail.com>"]
 edition = "2018"
 publish = false
@@ -8,13 +8,13 @@ publish = false
 [dependencies]
 hdrhistogram = "6.0"
 log = "0.4.1"
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
-tower-filter = { version = "0.3.0-alpha.2", path = "../tower-filter" }
-tokio-timer = "0.3.0-alpha.4"
-futures-util-preview = "0.3.0-alpha.19"
+tower-service = { version = "0.3", path = "../tower-service" }
+tower-filter = { version = "0.3", path = "../tower-filter" }
+tokio = { version = "0.2", features = ["time"] }
+futures-util = "0.3"
 pin-project = "0.4"
 
 [dev-dependencies]
-tower-test = { version = "0.3.0-alpha.2", path = "../tower-test" }
-tokio-test = "0.2.0-alpha.4"
-tokio-executor = "0.2.0-alpha.4"
+tower-test = { version = "0.3", path = "../tower-test" }
+tokio-test = "0.2"
+tokio = { version = "0.2", features = ["macros", "test-util"] }
\ No newline at end of file
diff --git a/tower-hedge/src/delay.rs b/tower-hedge/src/delay.rs
index a26df92..25356d1 100644
--- a/tower-hedge/src/delay.rs
+++ b/tower-hedge/src/delay.rs
@@ -32,7 +32,7 @@ pub struct ResponseFuture<Request, S, F> {
 #[pin_project]
 #[derive(Debug)]
 enum State<Request, F> {
-    Delaying(#[pin] tokio_timer::Delay, Option<Request>),
+    Delaying(#[pin] tokio::time::Delay, Option<Request>),
     Called(#[pin] F),
 }
 
@@ -62,7 +62,7 @@ where
     }
 
     fn call(&mut self, request: Request) -> Self::Future {
-        let deadline = tokio_timer::clock::now() + self.policy.delay(&request);
+        let deadline = tokio::time::Instant::now() + self.policy.delay(&request);
         let mut cloned = self.service.clone();
         // Pass the original service to the ResponseFuture and keep the cloned service on self.
         let orig = {
@@ -71,7 +71,7 @@ where
         };
         ResponseFuture {
             service: orig,
-            state: State::Delaying(tokio_timer::delay(deadline), Some(request)),
+            state: State::Delaying(tokio::time::delay_until(deadline), Some(request)),
         }
     }
 }
diff --git a/tower-hedge/src/latency.rs b/tower-hedge/src/latency.rs
index 2a9ef7c..43a4d47 100644
--- a/tower-hedge/src/latency.rs
+++ b/tower-hedge/src/latency.rs
@@ -1,12 +1,12 @@
 use futures_util::ready;
 use pin_project::pin_project;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use std::{
     future::Future,
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio_timer::clock;
+use tokio::time::Instant;
 use tower_service::Service;
 
 /// Record is the interface for accepting request latency measurements.  When
@@ -62,7 +62,7 @@ where
 
     fn call(&mut self, request: Request) -> Self::Future {
         ResponseFuture {
-            start: clock::now(),
+            start: Instant::now(),
             rec: self.rec.clone(),
             inner: self.service.call(request),
         }
@@ -81,7 +81,7 @@ where
         let this = self.project();
 
         let rsp = ready!(this.inner.poll(cx))?;
-        let duration = clock::now() - *this.start;
+        let duration = Instant::now() - *this.start;
         this.rec.record(duration);
         Poll::Ready(Ok(rsp))
     }
diff --git a/tower-hedge/src/rotating_histogram.rs b/tower-hedge/src/rotating_histogram.rs
index 87fb8a7..87f371b 100644
--- a/tower-hedge/src/rotating_histogram.rs
+++ b/tower-hedge/src/rotating_histogram.rs
@@ -1,7 +1,7 @@
 use hdrhistogram::Histogram;
 use log::trace;
-use std::time::{Duration, Instant};
-use tokio_timer::clock;
+use std::time::Duration;
+use tokio::time::Instant;
 
 /// This represents a "rotating" histogram which stores two histogram, one which
 /// should be read and one which should be written to.  Every period, the read
@@ -23,7 +23,7 @@ impl RotatingHistogram {
                 .expect("Invalid histogram params"),
             write: Histogram::<u64>::new_with_bounds(1, 10_000, 3)
                 .expect("Invalid histogram params"),
-            last_rotation: clock::now(),
+            last_rotation: Instant::now(),
             period,
         }
     }
@@ -39,7 +39,7 @@ impl RotatingHistogram {
     }
 
     fn maybe_rotate(&mut self) {
-        let delta = clock::now() - self.last_rotation;
+        let delta = Instant::now() - self.last_rotation;
         // TODO: replace with delta.duration_div when it becomes stable.
         let rotations = (nanos(delta) / nanos(self.period)) as u32;
         if rotations >= 2 {
diff --git a/tower-load-shed/CHANGELOG.md b/tower-load-shed/CHANGELOG.md
index 633a866..7223835 100644
--- a/tower-load-shed/CHANGELOG.md
+++ b/tower-load-shed/CHANGELOG.md
@@ -1,3 +1,8 @@
+# 0.3.0 (December 4, 2019)
+
+- Update to `tower-service 0.3`
+- Update to `futures 0.3`
+
 # 0.3.0-alpha.2 (September 30, 2019)
 
 - Move to `futures-*-preview 0.3.0-alpha.19`
diff --git a/tower-load-shed/Cargo.toml b/tower-load-shed/Cargo.toml
index bed4d71..6c94b68 100644
--- a/tower-load-shed/Cargo.toml
+++ b/tower-load-shed/Cargo.toml
@@ -8,13 +8,13 @@ name = "tower-load-shed"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tower-rs/tower"
 homepage = "https://github.com/tower-rs/tower"
-documentation = "https://docs.rs/tower-load-shed/0.3.0-alpha.2"
+documentation = "https://docs.rs/tower-load-shed/0.3.0"
 description = """
 Immediately reject requests if the inner service is not ready. This is also
 known as load-shedding.
@@ -23,12 +23,12 @@ categories = ["asynchronous", "network-programming"]
 edition = "2018"
 
 [dependencies]
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
-tower-layer = { version = "=0.3.0-alpha.2", path = "../tower-layer" }
+tower-service = { version = "0.3", path = "../tower-service" }
+tower-layer = { version = "0.3", path = "../tower-layer" }
 pin-project = "0.4"
-futures-core-preview = "=0.3.0-alpha.19"
+futures-core = "0.3"
 
 [dev-dependencies]
-tokio-test = "=0.2.0-alpha.6"
-tower-test = { version = "=0.3.0-alpha.2", path = "../tower-test" }
-futures-util-preview = "=0.3.0-alpha.19"
+tokio-test = "0.2"
+tower-test = { version = "0.3", path = "../tower-test" }
+tokio = { version = "0.2", features = ["macros"] }
\ No newline at end of file
diff --git a/tower-load-shed/src/lib.rs b/tower-load-shed/src/lib.rs
index c5559b1..1a8cb2b 100644
--- a/tower-load-shed/src/lib.rs
+++ b/tower-load-shed/src/lib.rs
@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tower-load-shed/0.3.0-alpha.2")]
+#![doc(html_root_url = "https://docs.rs/tower-load-shed/0.3.0")]
 #![warn(
     missing_debug_implementations,
     missing_docs,
diff --git a/tower-load/CHANGELOG.md b/tower-load/CHANGELOG.md
index 19df3db..1ec0e91 100644
--- a/tower-load/CHANGELOG.md
+++ b/tower-load/CHANGELOG.md
@@ -1,3 +1,8 @@
+# 0.3.0 (December 4, 2019)
+
+- Update to `tower-service 0.3`
+- Update to `tower-discover 0.3`
+
 # 0.3.0-alpha.2 (September 30, 2019)
 
 - Move to `futures-*-preview 0.3.0-alpha.19`
diff --git a/tower-load/Cargo.toml b/tower-load/Cargo.toml
index f4961b8..6b3dda4 100644
--- a/tower-load/Cargo.toml
+++ b/tower-load/Cargo.toml
@@ -8,7 +8,7 @@ name = "tower-load"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
@@ -22,13 +22,14 @@ categories = ["asynchronous", "network-programming"]
 edition = "2018"
 
 [dependencies]
-futures-core-preview = "=0.3.0-alpha.19"
+futures-core = "0.3"
 log = "0.4.1"
-tokio-timer = "=0.3.0-alpha.6"
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
-tower-discover = { version = "=0.3.0-alpha.2", path = "../tower-discover" }
+tokio = { version = "0.2", features = ["time"] }
+tower-service = { version = "0.3", path = "../tower-service" }
+tower-discover = { version = "0.3", path = "../tower-discover" }
 pin-project = "0.4"
 
 [dev-dependencies]
-tokio-test = "=0.2.0-alpha.6"
-futures-util-preview = "=0.3.0-alpha.19"
+tokio-test = "0.2"
+tokio = { version = "0.2", features = ["macros", "test-util" ] }
+futures-util = "0.3"
diff --git a/tower-load/src/lib.rs b/tower-load/src/lib.rs
index 55a1cce..c951621 100644
--- a/tower-load/src/lib.rs
+++ b/tower-load/src/lib.rs
@@ -1,6 +1,6 @@
 //! Abstractions and utilties for measuring a service's load.
 
-#![doc(html_root_url = "https://docs.rs/tower-load/0.3.0-alpha.2")]
+#![doc(html_root_url = "https://docs.rs/tower-load/0.3.0")]
 #![warn(
     missing_debug_implementations,
     missing_docs,
diff --git a/tower-load/src/peak_ewma.rs b/tower-load/src/peak_ewma.rs
index 0e5539f..7e3bb85 100644
--- a/tower-load/src/peak_ewma.rs
+++ b/tower-load/src/peak_ewma.rs
@@ -11,9 +11,9 @@ use std::{
 };
 use std::{
     sync::{Arc, Mutex},
-    time::{Duration, Instant},
+    time::Duration,
 };
-use tokio_timer::clock;
+use tokio::time::Instant;
 use tower_discover::{Change, Discover};
 use tower_service::Service;
 
@@ -156,7 +156,7 @@ impl<S, I> PeakEwma<S, I> {
     fn handle(&self) -> Handle {
         Handle {
             decay_ns: self.decay_ns,
-            sent_at: clock::now(),
+            sent_at: Instant::now(),
             rtt_estimate: self.rtt_estimate.clone(),
         }
     }
@@ -219,14 +219,14 @@ impl RttEstimate {
         debug_assert!(0.0 < rtt_ns, "rtt must be positive");
         Self {
             rtt_ns,
-            update_at: clock::now(),
+            update_at: Instant::now(),
         }
     }
 
     /// Decays the RTT estimate with a decay period of `decay_ns`.
     fn decay(&mut self, decay_ns: f64) -> f64 {
         // Updates with a 0 duration so that the estimate decays towards 0.
-        let now = clock::now();
+        let now = Instant::now();
         self.update(now, now, decay_ns)
     }
 
@@ -242,7 +242,7 @@ impl RttEstimate {
         );
         let rtt = nanos(recv_at - sent_at);
 
-        let now = clock::now();
+        let now = Instant::now();
         debug_assert!(
             self.update_at <= now,
             "update_at={:?} in the future",
@@ -285,7 +285,7 @@ impl RttEstimate {
 
 impl Drop for Handle {
     fn drop(&mut self) {
-        let recv_at = clock::now();
+        let recv_at = Instant::now();
 
         if let Ok(mut rtt) = self.rtt_estimate.lock() {
             rtt.update(self.sent_at, recv_at, self.decay_ns);
@@ -309,16 +309,9 @@ fn nanos(d: Duration) -> f64 {
 #[cfg(test)]
 mod tests {
     use futures_util::future;
-    use std::{
-        future::Future,
-        task::{Context, Poll},
-    };
-    use std::{
-        sync::{Arc, Mutex},
-        time::{Duration, Instant},
-    };
-    use tokio_test::{assert_ready, assert_ready_ok};
-    use tokio_timer::clock;
+    use std::time::Duration;
+    use tokio::time;
+    use tokio_test::{assert_ready, assert_ready_ok, task};
 
     use super::*;
 
@@ -337,81 +330,66 @@ mod tests {
         }
     }
 
-    struct Now(Arc<Mutex<Instant>>);
-    impl clock::Now for Now {
-        fn now(&self) -> Instant {
-            *self.0.lock().expect("now")
-        }
-    }
-
     /// The default RTT estimate decays, so that new nodes are considered if the
     /// default RTT is too high.
-    #[test]
-    fn default_decay() {
-        let time = Arc::new(Mutex::new(Instant::now()));
-        let clock = clock::Clock::new_with_now(Now(time.clone()));
-
-        clock::with_default(&clock, || {
-            let svc = PeakEwma::new(
-                Svc,
-                Duration::from_millis(10),
-                NANOS_PER_MILLI * 1_000.0,
-                NoInstrument,
-            );
-            let Cost(load) = svc.load();
-            assert_eq!(load, 10.0 * NANOS_PER_MILLI);
+    #[tokio::test]
+    async fn default_decay() {
+        time::pause();
+
+        let svc = PeakEwma::new(
+            Svc,
+            Duration::from_millis(10),
+            NANOS_PER_MILLI * 1_000.0,
+            NoInstrument,
+        );
+        let Cost(load) = svc.load();
+        assert_eq!(load, 10.0 * NANOS_PER_MILLI);
 
-            *time.lock().unwrap() += Duration::from_millis(100);
-            let Cost(load) = svc.load();
-            assert!(9.0 * NANOS_PER_MILLI < load && load < 10.0 * NANOS_PER_MILLI);
+        time::advance(Duration::from_millis(100)).await;
+        let Cost(load) = svc.load();
+        assert!(9.0 * NANOS_PER_MILLI < load && load < 10.0 * NANOS_PER_MILLI);
 
-            *time.lock().unwrap() += Duration::from_millis(100);
-            let Cost(load) = svc.load();
-            assert!(8.0 * NANOS_PER_MILLI < load && load < 9.0 * NANOS_PER_MILLI);
-        });
+        time::advance(Duration::from_millis(100)).await;
+        let Cost(load) = svc.load();
+        assert!(8.0 * NANOS_PER_MILLI < load && load < 9.0 * NANOS_PER_MILLI);
     }
 
-    /// The default RTT estimate decays, so that new nodes are considered if the
-    /// default RTT is too high.
-    #[test]
-    fn compound_decay() {
-        let time = Arc::new(Mutex::new(Instant::now()));
-        let clock = clock::Clock::new_with_now(Now(time.clone()));
-
-        clock::with_default(&clock, || {
-            tokio_test::task::mock(|cx| {
-                let mut svc = PeakEwma::new(
-                    Svc,
-                    Duration::from_millis(20),
-                    NANOS_PER_MILLI * 1_000.0,
-                    NoInstrument,
-                );
-                assert_eq!(svc.load(), Cost(20.0 * NANOS_PER_MILLI));
+    // /// The default RTT estimate decays, so that new nodes are considered if the
+    // /// default RTT is too high.
+    #[tokio::test]
+    async fn compound_decay() {
+        time::pause();
+
+        let mut svc = PeakEwma::new(
+            Svc,
+            Duration::from_millis(20),
+            NANOS_PER_MILLI * 1_000.0,
+            NoInstrument,
+        );
+        assert_eq!(svc.load(), Cost(20.0 * NANOS_PER_MILLI));
 
-                *time.lock().unwrap() += Duration::from_millis(100);
-                let mut rsp0 = svc.call(());
-                assert!(svc.load() > Cost(20.0 * NANOS_PER_MILLI));
+        time::advance(Duration::from_millis(100)).await;
+        let mut rsp0 = task::spawn(svc.call(()));
+        assert!(svc.load() > Cost(20.0 * NANOS_PER_MILLI));
 
-                *time.lock().unwrap() += Duration::from_millis(100);
-                let mut rsp1 = svc.call(());
-                assert!(svc.load() > Cost(40.0 * NANOS_PER_MILLI));
+        time::advance(Duration::from_millis(100)).await;
+        let mut rsp1 = task::spawn(svc.call(()));
+        assert!(svc.load() > Cost(40.0 * NANOS_PER_MILLI));
 
-                *time.lock().unwrap() += Duration::from_millis(100);
-                let () = assert_ready_ok!(Pin::new(&mut rsp0).poll(cx));
-                assert_eq!(svc.load(), Cost(400_000_000.0));
+        time::advance(Duration::from_millis(100)).await;
+        let () = assert_ready_ok!(rsp0.poll());
+        assert_eq!(svc.load(), Cost(400_000_000.0));
 
-                *time.lock().unwrap() += Duration::from_millis(100);
-                let () = assert_ready_ok!(Pin::new(&mut rsp1).poll(cx));
-                assert_eq!(svc.load(), Cost(200_000_000.0));
+        time::advance(Duration::from_millis(100)).await;
+        let () = assert_ready_ok!(rsp1.poll());
+        assert_eq!(svc.load(), Cost(200_000_000.0));
 
-                // Check that values decay as time elapses
-                *time.lock().unwrap() += Duration::from_secs(1);
-                assert!(svc.load() < Cost(100_000_000.0));
+        // Check that values decay as time elapses
+        time::advance(Duration::from_secs(1)).await;
+        assert!(svc.load() < Cost(100_000_000.0));
 
-                *time.lock().unwrap() += Duration::from_secs(10);
-                assert!(svc.load() < Cost(100_000.0));
-            })
-        });
+        time::advance(Duration::from_secs(10)).await;
+        assert!(svc.load() < Cost(100_000.0));
     }
 
     #[test]
diff --git a/tower-ready-cache/Cargo.toml b/tower-ready-cache/Cargo.toml
index 4403bde..21959d8 100644
--- a/tower-ready-cache/Cargo.toml
+++ b/tower-ready-cache/Cargo.toml
@@ -8,7 +8,7 @@ name = "tower-ready-cache"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.0"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
@@ -20,16 +20,17 @@ Caches a set of services
 """
 categories = ["asynchronous", "network-programming"]
 edition = "2018"
-publish = false
 
 [dependencies]
-futures = "0.1.26"
+futures-core = "0.3"
+futures-util = "0.3"
 indexmap = "1.0.2"
 log = "0.4.1"
-tokio-sync = "0.1.3"
-tower-service = "0.2.0"
+tokio = { version = "0.2", features = ["sync"] } 
+tower-service = { version = "0.3", path = "../tower-service" } 
 tower-util = "0.1.0"
 
 [dev-dependencies]
-tokio-executor = "0.1.7"
-tower-test = { version = "0.1.0", path = "../tower-test" }
+tower-test = { version = "0.3", path = "../tower-test" }
+tokio-test = "0.2"
+
diff --git a/tower-ready-cache/src/cache.rs b/tower-ready-cache/src/cache.rs
index 9236dcc..169aa5f 100644
--- a/tower-ready-cache/src/cache.rs
+++ b/tower-ready-cache/src/cache.rs
@@ -1,12 +1,16 @@
 //! A cache of services.
 
 use crate::error;
-use futures::{stream, Async, Future, Poll, Stream};
+use futures_core::Stream;
+use futures_util::stream::FuturesUnordered;
 pub use indexmap::Equivalent;
 use indexmap::IndexMap;
 use log::{debug, trace};
+use std::future::Future;
 use std::hash::Hash;
-use tokio_sync::oneshot;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::sync::oneshot;
 use tower_service::Service;
 
 /// Drives readiness over a set of services.
@@ -37,7 +41,7 @@ use tower_service::Service;
 ///
 /// When `ReadyCache::check_ready*` returns `false`, it indicates that the
 /// specified service is _not_ ready. If an error is returned, this indicats that
-/// the server failed nad has been removed from the cache entirely.
+/// the server failed and has been removed from the cache entirely.
 ///
 /// `ReadyCache::evict` can be used to remove a service from the cache (by key),
 /// though the service may not be dropped (if it is currently pending) until
@@ -55,7 +59,7 @@ where
     K: Eq + Hash,
 {
     /// A stream of services that are not yet ready.
-    pending: stream::FuturesUnordered<Pending<K, S, Req>>,
+    pending: FuturesUnordered<Pending<K, S, Req>>,
     /// An index of cancelation handles for pending streams.
     pending_cancel_txs: IndexMap<K, CancelTx>,
 
@@ -68,6 +72,9 @@ where
     ready: IndexMap<K, (S, CancelPair)>,
 }
 
+// Safety: This is safe because we do not use `Pin::new_unchecked`.
+impl<S, K: Eq + Hash, Req> Unpin for ReadyCache<K, S, Req> {}
+
 type CancelRx = oneshot::Receiver<()>;
 type CancelTx = oneshot::Sender<()>;
 type CancelPair = (CancelTx, CancelRx);
@@ -85,7 +92,8 @@ enum PendingError<K, E> {
 struct Pending<K, S, Req> {
     key: Option<K>,
     cancel: Option<CancelRx>,
-    ready: tower_util::Ready<S, Req>,
+    ready: Option<S>,
+    _pd: std::marker::PhantomData<Req>,
 }
 
 // === ReadyCache ===
@@ -98,7 +106,7 @@ where
     fn default() -> Self {
         Self {
             ready: IndexMap::default(),
-            pending: stream::FuturesUnordered::new(),
+            pending: FuturesUnordered::new(),
             pending_cancel_txs: IndexMap::default(),
         }
     }
@@ -203,7 +211,8 @@ where
         self.pending.push(Pending {
             key: Some(key),
             cancel: Some(cancel_rx),
-            ready: tower_util::Ready::new(svc),
+            ready: Some(svc),
+            _pd: std::marker::PhantomData,
         });
     }
 
@@ -217,12 +226,12 @@ where
     /// ready (and has been removed from the cache). In such a case,
     /// `poll_pending` should typically be called again to continue driving
     /// pending services to readiness.
-    pub fn poll_pending(&mut self) -> Poll<(), error::Failed<K>> {
+    pub fn poll_pending(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), error::Failed<K>>> {
         loop {
-            match self.pending.poll() {
-                Ok(Async::NotReady) => return Ok(Async::NotReady),
-                Ok(Async::Ready(None)) => return Ok(Async::Ready(())),
-                Ok(Async::Ready(Some((key, svc, cancel_rx)))) => {
+            match Pin::new(&mut self.pending).poll_next(cx) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(None) => return Poll::Ready(Ok(())),
+                Poll::Ready(Some(Ok((key, svc, cancel_rx)))) => {
                     trace!("endpoint ready");
                     let cancel_tx = self
                         .pending_cancel_txs
@@ -232,16 +241,16 @@ where
                     // recreated after the service is used.
                     self.ready.insert(key, (svc, (cancel_tx, cancel_rx)));
                 }
-                Err(PendingError::Canceled(_)) => {
+                Poll::Ready(Some(Err(PendingError::Canceled(_)))) => {
                     debug!("endpoint canceled");
                     // The cancellation for this service was removed in order to
                     // cause this cancellation.
                 }
-                Err(PendingError::Inner(key, e)) => {
+                Poll::Ready(Some(Err(PendingError::Inner(key, e)))) => {
                     self.pending_cancel_txs
                         .swap_remove(&key)
                         .expect("missing cancelation");
-                    return Err(error::Failed(key, e.into()));
+                    return Err(error::Failed(key, e.into())).into();
                 }
             }
         }
@@ -253,10 +262,11 @@ where
     /// returned if the endpoint fails.
     pub fn check_ready<Q: Hash + Equivalent<K>>(
         &mut self,
+        cx: &mut Context<'_>,
         key: &Q,
     ) -> Result<bool, error::Failed<K>> {
         match self.ready.get_full_mut(key) {
-            Some((index, _, _)) => self.check_ready_index(index),
+            Some((index, _, _)) => self.check_ready_index(cx, index),
             None => Ok(false),
         }
     }
@@ -267,14 +277,18 @@ where
     /// and `false` is returned.
     ///
     /// If the service errors, it is removed and dropped and the error is returned.
-    pub fn check_ready_index(&mut self, index: usize) -> Result<bool, error::Failed<K>> {
+    pub fn check_ready_index(
+        &mut self,
+        cx: &mut Context<'_>,
+        index: usize,
+    ) -> Result<bool, error::Failed<K>> {
         let svc = match self.ready.get_index_mut(index) {
             None => return Ok(false),
             Some((_, (svc, _))) => svc,
         };
-        match svc.poll_ready() {
-            Ok(Async::Ready(())) => Ok(true),
-            Ok(Async::NotReady) => {
+        match svc.poll_ready(cx) {
+            Poll::Ready(Ok(())) => Ok(true),
+            Poll::Pending => {
                 // became unready; so move it back there.
                 let (key, (svc, cancel)) = self
                     .ready
@@ -289,7 +303,7 @@ where
 
                 Ok(false)
             }
-            Err(e) => {
+            Poll::Ready(Err(e)) => {
                 // failed, so drop it.
                 let (key, _) = self
                     .ready
@@ -338,36 +352,37 @@ where
 
 // === Pending ===
 
+// Safety: No use unsafe access therefore this is safe.
+impl<K, S, Req> Unpin for Pending<K, S, Req> {}
+
 impl<K, S, Req> Future for Pending<K, S, Req>
 where
     S: Service<Req>,
 {
-    type Item = (K, S, CancelRx);
-    type Error = PendingError<K, S::Error>;
+    type Output = Result<(K, S, CancelRx), PendingError<K, S::Error>>;
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        if self
-            .cancel
-            .as_mut()
-            .expect("polled after complete")
-            .poll()
-            .expect("cancel sender lost")
-            .is_ready()
-        {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut fut = self.cancel.as_mut().expect("polled after complete");
+        if let Poll::Ready(Ok(_)) = Pin::new(&mut fut).poll(cx) {
             let key = self.key.take().expect("polled after complete");
-            return Err(PendingError::Canceled(key));
+            return Err(PendingError::Canceled(key)).into();
         }
 
-        match self.ready.poll() {
-            Ok(Async::NotReady) => Ok(Async::NotReady),
-            Ok(Async::Ready(svc)) => {
+        match self
+            .ready
+            .as_mut()
+            .expect("polled after ready")
+            .poll_ready(cx)
+        {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Ok(())) => {
                 let key = self.key.take().expect("polled after complete");
                 let cancel = self.cancel.take().expect("polled after complete");
-                Ok((key, svc, cancel).into())
+                Ok((key, self.ready.take().expect("polled after ready"), cancel)).into()
             }
-            Err(e) => {
+            Poll::Ready(Err(e)) => {
                 let key = self.key.take().expect("polled after compete");
-                Err(PendingError::Inner(key, e))
+                Err(PendingError::Inner(key, e)).into()
             }
         }
     }
diff --git a/tower-ready-cache/src/lib.rs b/tower-ready-cache/src/lib.rs
index 0792404..58e10a2 100644
--- a/tower-ready-cache/src/lib.rs
+++ b/tower-ready-cache/src/lib.rs
@@ -4,7 +4,6 @@
 #![deny(missing_docs)]
 #![deny(rust_2018_idioms)]
 #![allow(elided_lifetimes_in_paths)]
-#![deny(warnings)]
 
 pub mod cache;
 pub mod error;
diff --git a/tower-spawn-ready/CHANGELOG.md b/tower-spawn-ready/CHANGELOG.md
index 645b0b4..9ad907c 100644
--- a/tower-spawn-ready/CHANGELOG.md
+++ b/tower-spawn-ready/CHANGELOG.md
@@ -1,3 +1,9 @@
+# 0.3.0 (December 4, 2019)
+
+- Update to `tower-service 0.3`
+- Update to `tokio 0.2`
+- Remove `Executor` usage
+
 # 0.3.0-alpha.2 (September 30, 2019)
 
 - Move to `futures-*-preview 0.3.0-alpha.19`
diff --git a/tower-spawn-ready/Cargo.toml b/tower-spawn-ready/Cargo.toml
index 3fc94fa..99dfcc1 100644
--- a/tower-spawn-ready/Cargo.toml
+++ b/tower-spawn-ready/Cargo.toml
@@ -8,7 +8,7 @@ name = "tower-spawn-ready"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.3.0-alpha.2"
+version = "0.3.0"
 authors = ["Tower Maintainers <team@tower-rs.com>"]
 license = "MIT"
 readme = "README.md"
@@ -22,14 +22,14 @@ categories = ["asynchronous", "network-programming"]
 edition = "2018"
 
 [dependencies]
-futures-core-preview = "=0.3.0-alpha.19"
-futures-util-preview = "=0.3.0-alpha.19"
+futures-core = "0.3"
+futures-util = "0.3"
 pin-project = "0.4"
-tower-service = { version = "=0.3.0-alpha.2", path = "../tower-service" }
-tower-layer = { version = "=0.3.0-alpha.2", path = "../tower-layer" }
-tokio-executor = "=0.2.0-alpha.6"
-tokio-sync = "=0.2.0-alpha.6"
+tower-service = { version = "0.3", path = "../tower-service" }
+tower-layer = { version = "0.3", path = "../tower-layer" }
+tokio = { version = "0.2", features = ["sync"] }
 
 [dev-dependencies]
-tower-test = { version = "=0.3.0-alpha.2", path = "../tower-test" }
-tokio-test = "=0.2.0-alpha.6"
+tower-test = { version = "0.3", path = "../tower-test" }
+tokio-test = "0.2"
+tokio = { version = "0.2", features = ["macros"] }
\ No newline at end of file
diff --git a/tower-spawn-ready/src/error.rs b/tower-spawn-ready/src/error.rs
deleted file mode 100644
index 5958cf4..0000000
--- a/tower-spawn-ready/src/error.rs
+++ /dev/null
@@ -1,33 +0,0 @@
-//! Error types
-
-use std::fmt;
-use tokio_executor;
-
-/// Error produced when spawning the worker fails
-#[derive(Debug)]
-pub struct SpawnError {
-    inner: tokio_executor::SpawnError,
-}
-
-/// Errors produced by `SpawnReady`.
-pub(crate) type Error = Box<dyn std::error::Error + Send + Sync>;
-
-// ===== impl SpawnError =====
-
-impl SpawnError {
-    pub(crate) fn new(inner: tokio_executor::SpawnError) -> Self {
-        Self { inner }
-    }
-}
-
-impl fmt::Display for SpawnError {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        self.inner.fmt(f)
-    }
-}
-
-impl std::error::Error for SpawnError {
-    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-        Some(&self.inner)
-    }
-}
diff --git a/tower-spawn-ready/src/future.rs b/tower-spawn-ready/src/future.rs
index cffacc2..676058d 100644
--- a/tower-spawn-ready/src/future.rs
+++ b/tower-spawn-ready/src/future.rs
@@ -1,6 +1,6 @@
 //! Background readiness types
 
-use crate::error::Error;
+use crate::Error;
 use futures_core::ready;
 use pin_project::pin_project;
 use std::marker::PhantomData;
@@ -9,8 +9,7 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio_executor::TypedExecutor;
-use tokio_sync::oneshot;
+use tokio::sync::oneshot;
 use tower_service::Service;
 
 /// Drives a service to readiness.
@@ -22,24 +21,6 @@ pub struct BackgroundReady<T, Request> {
     _req: PhantomData<Request>,
 }
 
-/// This trait allows you to use either Tokio's threaded runtime's executor or
-/// the `current_thread` runtime's executor depending on if `T` is `Send` or
-/// `!Send`.
-pub trait BackgroundReadyExecutor<T, Request>: TypedExecutor<BackgroundReady<T, Request>>
-where
-    T: Service<Request>,
-    T::Error: Into<Error>,
-{
-}
-
-impl<T, Request, E> BackgroundReadyExecutor<T, Request> for E
-where
-    E: TypedExecutor<BackgroundReady<T, Request>>,
-    T: Service<Request>,
-    T::Error: Into<Error>,
-{
-}
-
 pub(crate) fn background_ready<T, Request>(
     service: T,
 ) -> (
diff --git a/tower-spawn-ready/src/lib.rs b/tower-spawn-ready/src/lib.rs
index 43ecb15..279a826 100644
--- a/tower-spawn-ready/src/lib.rs
+++ b/tower-spawn-ready/src/lib.rs
@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tower-spawn-ready/0.3.0-alpha.2")]
+#![doc(html_root_url = "https://docs.rs/tower-spawn-ready/0.3.0")]
 #![warn(
     missing_debug_implementations,
     missing_docs,
@@ -10,7 +10,6 @@
 //! When an underlying service is not ready, drive it to readiness on a
 //! background task.
 
-pub mod error;
 pub mod future;
 mod layer;
 mod make;
@@ -19,3 +18,6 @@ mod service;
 pub use crate::layer::SpawnReadyLayer;
 pub use crate::make::{MakeFuture, MakeSpawnReady};
 pub use crate::service::SpawnReady;
+
+/// Errors produced by `SpawnReady`.
+pub(crate) type Error = Box<dyn std::error::Error + Send + Sync>;
diff --git a/tower-spawn-ready/src/service.rs b/tower-spawn-ready/src/service.rs
index d2880e0..63ac8c9 100644
--- a/tower-spawn-ready/src/service.rs
+++ b/tower-spawn-ready/src/service.rs
@@ -1,16 +1,12 @@
-use crate::{
-    error::{Error, SpawnError},
-    future::{background_ready, BackgroundReadyExecutor},
-};
+use crate::{future::background_ready, Error};
 use futures_core::ready;
-use futures_util::try_future::{MapErr, TryFutureExt};
+use futures_util::future::{MapErr, TryFutureExt};
 use std::{
     future::Future,
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio_executor::{DefaultExecutor, TypedExecutor};
-use tokio_sync::oneshot;
+use tokio::sync::oneshot;
 use tower_service::Service;
 
 /// Spawns tasks to drive an inner service to readiness.
@@ -38,9 +34,9 @@ impl<T> SpawnReady<T> {
 
 impl<T, Request> Service<Request> for SpawnReady<T>
 where
-    T: Service<Request> + Send,
+    T: Service<Request> + Send + 'static,
     T::Error: Into<Error>,
-    DefaultExecutor: BackgroundReadyExecutor<T, Request>,
+    Request: Send + 'static,
 {
     type Response = T::Response;
     type Error = Error;
@@ -55,9 +51,8 @@ where
                     }
 
                     let (bg, rx) = background_ready(svc.take().expect("illegal state"));
-                    DefaultExecutor::current()
-                        .spawn(bg)
-                        .map_err(SpawnError::new)?;
+
+                    tokio::spawn(bg);
 
                     Inner::Future(rx)
                 }
diff --git a/tower-util/src/ready.rs b/tower-util/src/ready.rs
index c102347..44cb38a 100644
--- a/tower-util/src/ready.rs
+++ b/tower-util/src/ready.rs
@@ -1,7 +1,6 @@
 use std::{fmt, marker::PhantomData};
 
 use futures_util::ready;
-use pin_project::pin_project;
 use std::{
     future::Future,
     pin::Pin,
@@ -12,12 +11,14 @@ use tower_service::Service;
 /// Future yielding a `Service` once the service is ready to process a request
 ///
 /// `Ready` values are produced by `ServiceExt::ready`.
-#[pin_project]
 pub struct Ready<'a, T, Request> {
     inner: &'a mut T,
     _p: PhantomData<fn() -> Request>,
 }
 
+// Safety: This is safe because `Services`'s are always `Unpin`.
+impl<'a, T, Request> Unpin for Ready<'a, T, Request> {}
+
 impl<'a, T, Request> Ready<'a, T, Request>
 where
     T: Service<Request>,
@@ -37,9 +38,8 @@ where
 {
     type Output = Result<(), T::Error>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-        ready!(this.inner.poll_ready(cx))?;
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        ready!(self.inner.poll_ready(cx))?;
 
         Poll::Ready(Ok(()))
     }
diff --git a/tower/Cargo.toml b/tower/Cargo.toml
index fb87276..5314b48 100644
--- a/tower/Cargo.toml
+++ b/tower/Cargo.toml
@@ -29,17 +29,17 @@ full = []
 
 [dependencies]
 tower-buffer = { version = "0.3", path = "../tower-buffer" }
-# tower-discover = { version = "0.3", path = "../tower-discover" }
+tower-discover = { version = "0.3", path = "../tower-discover" }
 tower-layer = { version = "0.3", path = "../tower-layer" }
 tower-limit = { version = "0.3", path = "../tower-limit" }
-# tower-load-shed = { version = "0.3", path = "../tower-load-shed" }
+tower-load-shed = { version = "0.3", path = "../tower-load-shed" }
 tower-retry = { version = "0.3", path = "../tower-retry" }
 tower-service = { version = "0.3", path = "../tower-service" }
 tower-timeout = { version = "0.3", path = "../tower-timeout" }
 tower-util = { version = "0.3", path = "../tower-util" }
 futures-core = "0.3"
 
-# [dev-dependencies]
+[dev-dependencies]
 # env_logger = { version = "0.5.3", default-features = false }
 futures-util = "0.3"
 tokio = { version = "0.2", features = ["macros"] }
diff --git a/tower/src/builder/mod.rs b/tower/src/builder/mod.rs
index fb2ce04..7896e9e 100644
--- a/tower/src/builder/mod.rs
+++ b/tower/src/builder/mod.rs
@@ -3,7 +3,7 @@
 use crate::{
     buffer::BufferLayer,
     limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer},
-    // load_shed::LoadShedLayer,
+    load_shed::LoadShedLayer,
     retry::RetryLayer,
     timeout::TimeoutLayer,
 };
@@ -146,9 +146,9 @@ impl<L> ServiceBuilder<L> {
     ///
     /// `load_shed` immediately responds with an error when the next layer is
     /// out of capacity.
-    // pub fn load_shed(self) -> ServiceBuilder<Stack<LoadShedLayer, L>> {
-    //     self.layer(LoadShedLayer::new())
-    // }
+    pub fn load_shed(self) -> ServiceBuilder<Stack<LoadShedLayer, L>> {
+        self.layer(LoadShedLayer::new())
+    }
 
     /// Limit requests to at most `num` per the given duration
     pub fn rate_limit(self, num: u64, per: Duration) -> ServiceBuilder<Stack<RateLimitLayer, L>> {
diff --git a/tower/src/lib.rs b/tower/src/lib.rs
index b0b3a65..eb2384d 100644
--- a/tower/src/lib.rs
+++ b/tower/src/lib.rs
@@ -16,12 +16,12 @@
 
 #[doc(inline)]
 pub use tower_buffer as buffer;
-// #[doc(inline)]
-// pub use tower_discover as discover;
+#[doc(inline)]
+pub use tower_discover as discover;
 #[doc(inline)]
 pub use tower_limit as limit;
 #[doc(inline)]
-// pub use tower_load_shed as load_shed;
+pub use tower_load_shed as load_shed;
 #[doc(inline)]
 pub use tower_retry as retry;
 #[doc(inline)]


Test patch:
diff --git a/tower-balance/src/p2c/test.rs b/tower-balance/src/p2c/test.rs
index a565957..b1eb39a 100644
--- a/tower-balance/src/p2c/test.rs
+++ b/tower-balance/src/p2c/test.rs
@@ -1,122 +1,125 @@
 use futures_util::pin_mut;
-use std::{future::Future, task::Poll};
+use std::task::Poll;
 use tokio_test::{assert_pending, assert_ready, assert_ready_ok, task};
 use tower_discover::ServiceList;
 use tower_load as load;
-use tower_service::Service;
 use tower_test::{assert_request_eq, mock};
 
 use super::*;
 
-#[test]
-fn empty() {
-    task::mock(|cx| {
-        let empty: Vec<load::Constant<mock::Mock<(), &'static str>, usize>> = vec![];
-        let disco = ServiceList::new(empty);
-        let mut svc = Balance::from_entropy(disco);
-        assert_pending!(svc.poll_ready(cx));
-    });
+#[tokio::test]
+async fn empty() {
+    let empty: Vec<load::Constant<mock::Mock<(), &'static str>, usize>> = vec![];
+    let disco = ServiceList::new(empty);
+    let mut svc = mock::Spawn::new(Balance::from_entropy(disco));
+    assert_pending!(svc.poll_ready());
 }
 
-#[test]
-fn single_endpoint() {
-    task::mock(|cx| {
-        let (mock, handle) = mock::pair();
-        pin_mut!(handle);
-
-        let mock = load::Constant::new(mock, 0);
-
+#[tokio::test]
+async fn single_endpoint() {
+    let (mut svc, mut handle) = mock::spawn_with(|s| {
+        let mock = load::Constant::new(s, 0);
         let disco = ServiceList::new(vec![mock].into_iter());
-        let mut svc = Balance::from_entropy(disco);
+        Balance::from_entropy(disco)
+    });
 
-        handle.allow(0);
-        assert_pending!(svc.poll_ready(cx));
-        assert_eq!(svc.len(), 1, "balancer must have discovered endpoint");
+    handle.allow(0);
+    assert_pending!(svc.poll_ready());
+    assert_eq!(
+        svc.get_ref().len(),
+        1,
+        "balancer must have discovered endpoint"
+    );
 
-        handle.allow(1);
-        assert_ready_ok!(svc.poll_ready(cx));
+    handle.allow(1);
+    assert_ready_ok!(svc.poll_ready());
 
-        let fut = svc.call(());
-        pin_mut!(fut);
+    let mut fut = task::spawn(svc.call(()));
 
-        assert_request_eq!(handle, ()).send_response(1);
+    assert_request_eq!(handle, ()).send_response(1);
 
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), 1);
-        handle.allow(1);
-        assert_ready_ok!(svc.poll_ready(cx));
+    assert_eq!(assert_ready_ok!(fut.poll()), 1);
+    handle.allow(1);
+    assert_ready_ok!(svc.poll_ready());
 
-        handle.send_error("endpoint lost");
-        assert_pending!(svc.poll_ready(cx));
-        assert!(svc.len() == 0, "balancer must drop failed endpoints");
-    });
+    handle.send_error("endpoint lost");
+    assert_pending!(svc.poll_ready());
+    assert!(
+        svc.get_ref().len() == 0,
+        "balancer must drop failed endpoints"
+    );
 }
 
-#[test]
-fn two_endpoints_with_equal_load() {
-    task::mock(|cx| {
-        let (mock_a, handle_a) = mock::pair();
-        let (mock_b, handle_b) = mock::pair();
-        let mock_a = load::Constant::new(mock_a, 1);
-        let mock_b = load::Constant::new(mock_b, 1);
-
-        pin_mut!(handle_a);
-        pin_mut!(handle_b);
-
-        let disco = ServiceList::new(vec![mock_a, mock_b].into_iter());
-        let mut svc = Balance::from_entropy(disco);
-
-        handle_a.allow(0);
-        handle_b.allow(0);
-        assert_pending!(svc.poll_ready(cx));
-        assert_eq!(svc.len(), 2, "balancer must have discovered both endpoints");
-
-        handle_a.allow(1);
-        handle_b.allow(0);
-        assert_ready_ok!(
-            svc.poll_ready(cx),
-            "must be ready when one of two services is ready"
-        );
-        {
-            let fut = svc.call(());
-            pin_mut!(fut);
-            assert_request_eq!(handle_a, ()).send_response("a");
-            assert_eq!(assert_ready_ok!(fut.poll(cx)), "a");
-        }
-
-        handle_a.allow(0);
-        handle_b.allow(1);
+#[tokio::test]
+async fn two_endpoints_with_equal_load() {
+    let (mock_a, handle_a) = mock::pair();
+    let (mock_b, handle_b) = mock::pair();
+    let mock_a = load::Constant::new(mock_a, 1);
+    let mock_b = load::Constant::new(mock_b, 1);
+
+    pin_mut!(handle_a);
+    pin_mut!(handle_b);
+
+    let disco = ServiceList::new(vec![mock_a, mock_b].into_iter());
+    let mut svc = mock::Spawn::new(Balance::from_entropy(disco));
+
+    handle_a.allow(0);
+    handle_b.allow(0);
+    assert_pending!(svc.poll_ready());
+    assert_eq!(
+        svc.get_ref().len(),
+        2,
+        "balancer must have discovered both endpoints"
+    );
+
+    handle_a.allow(1);
+    handle_b.allow(0);
+    assert_ready_ok!(
+        svc.poll_ready(),
+        "must be ready when one of two services is ready"
+    );
+    {
+        let mut fut = task::spawn(svc.call(()));
+        assert_request_eq!(handle_a, ()).send_response("a");
+        assert_eq!(assert_ready_ok!(fut.poll()), "a");
+    }
+
+    handle_a.allow(0);
+    handle_b.allow(1);
+    assert_ready_ok!(
+        svc.poll_ready(),
+        "must be ready when both endpoints are ready"
+    );
+    {
+        let mut fut = task::spawn(svc.call(()));
+        assert_request_eq!(handle_b, ()).send_response("b");
+        assert_eq!(assert_ready_ok!(fut.poll()), "b");
+    }
+
+    handle_a.allow(1);
+    handle_b.allow(1);
+    for _ in 0..2 {
         assert_ready_ok!(
-            svc.poll_ready(cx),
+            svc.poll_ready(),
             "must be ready when both endpoints are ready"
         );
-        {
-            let fut = svc.call(());
-            pin_mut!(fut);
-            assert_request_eq!(handle_b, ()).send_response("b");
-            assert_eq!(assert_ready_ok!(fut.poll(cx)), "b");
-        }
+        let mut fut = task::spawn(svc.call(()));
 
-        handle_a.allow(1);
-        handle_b.allow(1);
-        for _ in 0..2 {
-            assert_ready_ok!(
-                svc.poll_ready(cx),
-                "must be ready when both endpoints are ready"
-            );
-            let fut = svc.call(());
-            pin_mut!(fut);
-            for (ref mut h, c) in &mut [(&mut handle_a, "a"), (&mut handle_b, "b")] {
-                if let Poll::Ready(Some((_, tx))) = h.as_mut().poll_request(cx) {
-                    tracing::info!("using {}", c);
-                    tx.send_response(c);
-                    h.allow(0);
-                }
+        for (ref mut h, c) in &mut [(&mut handle_a, "a"), (&mut handle_b, "b")] {
+            if let Poll::Ready(Some((_, tx))) = h.as_mut().poll_request() {
+                tracing::info!("using {}", c);
+                tx.send_response(c);
+                h.allow(0);
             }
-            assert_ready_ok!(fut.as_mut().poll(cx));
         }
-
-        handle_a.send_error("endpoint lost");
-        assert_pending!(svc.poll_ready(cx));
-        assert_eq!(svc.len(), 1, "balancer must drop failed endpoints",);
-    });
+        assert_ready_ok!(fut.poll());
+    }
+
+    handle_a.send_error("endpoint lost");
+    assert_pending!(svc.poll_ready());
+    assert_eq!(
+        svc.get_ref().len(),
+        1,
+        "balancer must drop failed endpoints",
+    );
 }
diff --git a/tower-balance/src/pool/test.rs b/tower-balance/src/pool/test.rs
index a07d6a7..f9f6eac 100644
--- a/tower-balance/src/pool/test.rs
+++ b/tower-balance/src/pool/test.rs
@@ -1,200 +1,190 @@
 use futures_util::pin_mut;
 use tokio_test::{assert_pending, assert_ready, assert_ready_ok, task};
 use tower_load as load;
-use tower_service::Service;
 use tower_test::{assert_request_eq, mock};
 
 use super::*;
 
-#[test]
-fn basic() {
-    task::mock(|cx| {
-        // start the pool
-        let (mock, handle) =
-            mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
-        pin_mut!(handle);
-
-        let mut pool = Builder::new().build(mock, ());
-        assert_pending!(pool.poll_ready(cx));
-
-        // give the pool a backing service
-        let (svc1_m, svc1) = mock::pair();
-        pin_mut!(svc1);
-
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
-        assert_ready_ok!(pool.poll_ready(cx));
-
-        // send a request to the one backing service
-        let fut = pool.call(());
-        pin_mut!(fut);
-        assert_pending!(fut.as_mut().poll(cx));
-        assert_request_eq!(svc1, ()).send_response("foobar");
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), "foobar");
-    });
+#[tokio::test]
+async fn basic() {
+    // start the pool
+    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
+    pin_mut!(handle);
+
+    let mut pool = mock::Spawn::new(Builder::new().build(mock, ()));
+    assert_pending!(pool.poll_ready());
+
+    // give the pool a backing service
+    let (svc1_m, svc1) = mock::pair();
+    pin_mut!(svc1);
+
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
+    assert_ready_ok!(pool.poll_ready());
+
+    // send a request to the one backing service
+    let mut fut = task::spawn(pool.call(()));
+
+    assert_pending!(fut.poll());
+    assert_request_eq!(svc1, ()).send_response("foobar");
+    assert_eq!(assert_ready_ok!(fut.poll()), "foobar");
 }
 
-#[test]
-fn high_load() {
-    task::mock(|cx| {
-        // start the pool
-        let (mock, handle) =
-            mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
-        pin_mut!(handle);
-
-        let mut pool = Builder::new()
-            .urgency(1.0) // so _any_ Pending will add a service
-            .underutilized_below(0.0) // so no Ready will remove a service
-            .max_services(Some(2))
-            .build(mock, ());
-        assert_pending!(pool.poll_ready(cx));
-
-        // give the pool a backing service
-        let (svc1_m, svc1) = mock::pair();
-        pin_mut!(svc1);
-
-        svc1.allow(1);
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
-        assert_ready_ok!(pool.poll_ready(cx));
-
-        // make the one backing service not ready
-        let fut1 = pool.call(());
-        pin_mut!(fut1);
-
-        // if we poll_ready again, pool should notice that load is increasing
-        // since urgency == 1.0, it should immediately enter high load
-        assert_pending!(pool.poll_ready(cx));
-        // it should ask the maker for another service, so we give it one
-        let (svc2_m, svc2) = mock::pair();
-        pin_mut!(svc2);
-
-        svc2.allow(1);
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
-
-        // the pool should now be ready again for one more request
-        assert_ready_ok!(pool.poll_ready(cx));
-        let fut2 = pool.call(());
-        pin_mut!(fut2);
-        assert_pending!(pool.poll_ready(cx));
-
-        // the pool should _not_ try to add another service
-        // sicen we have max_services(2)
-        assert_pending!(handle.as_mut().poll_request(cx));
-
-        // let see that each service got one request
-        assert_request_eq!(svc1, ()).send_response("foo");
-        assert_request_eq!(svc2, ()).send_response("bar");
-        assert_eq!(assert_ready_ok!(fut1.poll(cx)), "foo");
-        assert_eq!(assert_ready_ok!(fut2.poll(cx)), "bar");
-    });
+#[tokio::test]
+async fn high_load() {
+    // start the pool
+    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
+    pin_mut!(handle);
+
+    let pool = Builder::new()
+        .urgency(1.0) // so _any_ Pending will add a service
+        .underutilized_below(0.0) // so no Ready will remove a service
+        .max_services(Some(2))
+        .build(mock, ());
+    let mut pool = mock::Spawn::new(pool);
+    assert_pending!(pool.poll_ready());
+
+    // give the pool a backing service
+    let (svc1_m, svc1) = mock::pair();
+    pin_mut!(svc1);
+
+    svc1.allow(1);
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
+    assert_ready_ok!(pool.poll_ready());
+
+    // make the one backing service not ready
+    let mut fut1 = task::spawn(pool.call(()));
+
+    // if we poll_ready again, pool should notice that load is increasing
+    // since urgency == 1.0, it should immediately enter high load
+    assert_pending!(pool.poll_ready());
+    // it should ask the maker for another service, so we give it one
+    let (svc2_m, svc2) = mock::pair();
+    pin_mut!(svc2);
+
+    svc2.allow(1);
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
+
+    // the pool should now be ready again for one more request
+    assert_ready_ok!(pool.poll_ready());
+    let mut fut2 = task::spawn(pool.call(()));
+
+    assert_pending!(pool.poll_ready());
+
+    // the pool should _not_ try to add another service
+    // sicen we have max_services(2)
+    assert_pending!(handle.as_mut().poll_request());
+
+    // let see that each service got one request
+    assert_request_eq!(svc1, ()).send_response("foo");
+    assert_request_eq!(svc2, ()).send_response("bar");
+    assert_eq!(assert_ready_ok!(fut1.poll()), "foo");
+    assert_eq!(assert_ready_ok!(fut2.poll()), "bar");
 }
 
-#[test]
-fn low_load() {
-    task::mock(|cx| {
-        // start the pool
-        let (mock, handle) =
-            mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
-        pin_mut!(handle);
-
-        let mut pool = Builder::new()
-            .urgency(1.0) // so any event will change the service count
-            .build(mock, ());
-        assert_pending!(pool.poll_ready(cx));
-
-        // give the pool a backing service
-        let (svc1_m, svc1) = mock::pair();
-        pin_mut!(svc1);
-
-        svc1.allow(1);
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
-        assert_ready_ok!(pool.poll_ready(cx));
-
-        // cycling a request should now work
-        let fut = pool.call(());
-        pin_mut!(fut);
-
-        assert_request_eq!(svc1, ()).send_response("foo");
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), "foo");
-        // and pool should now not be ready (since svc1 isn't ready)
-        // it should immediately try to add another service
-        // which we give it
-        assert_pending!(pool.poll_ready(cx));
-        let (svc2_m, svc2) = mock::pair();
-        pin_mut!(svc2);
-
-        svc2.allow(1);
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
-        // pool is now ready
-        // which (because of urgency == 1.0) should immediately cause it to drop a service
-        // it'll drop svc1, so it'll still be ready
-        assert_ready_ok!(pool.poll_ready(cx));
-        // and even with another ready, it won't drop svc2 since its now the only service
-        assert_ready_ok!(pool.poll_ready(cx));
-
-        // cycling a request should now work on svc2
-        let fut = pool.call(());
-        pin_mut!(fut);
-        assert_request_eq!(svc2, ()).send_response("foo");
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), "foo");
-
-        // and again (still svc2)
-        svc2.allow(1);
-        assert_ready_ok!(pool.poll_ready(cx));
-        let fut = pool.call(());
-        pin_mut!(fut);
-        assert_request_eq!(svc2, ()).send_response("foo");
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), "foo");
-    });
+#[tokio::test]
+async fn low_load() {
+    // start the pool
+    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
+    pin_mut!(handle);
+
+    let pool = Builder::new()
+        .urgency(1.0) // so any event will change the service count
+        .build(mock, ());
+
+    let mut pool = mock::Spawn::new(pool);
+
+    assert_pending!(pool.poll_ready());
+
+    // give the pool a backing service
+    let (svc1_m, svc1) = mock::pair();
+    pin_mut!(svc1);
+
+    svc1.allow(1);
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
+    assert_ready_ok!(pool.poll_ready());
+
+    // cycling a request should now work
+    let mut fut = task::spawn(pool.call(()));
+
+    assert_request_eq!(svc1, ()).send_response("foo");
+    assert_eq!(assert_ready_ok!(fut.poll()), "foo");
+    // and pool should now not be ready (since svc1 isn't ready)
+    // it should immediately try to add another service
+    // which we give it
+    assert_pending!(pool.poll_ready());
+    let (svc2_m, svc2) = mock::pair();
+    pin_mut!(svc2);
+
+    svc2.allow(1);
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
+    // pool is now ready
+    // which (because of urgency == 1.0) should immediately cause it to drop a service
+    // it'll drop svc1, so it'll still be ready
+    assert_ready_ok!(pool.poll_ready());
+    // and even with another ready, it won't drop svc2 since its now the only service
+    assert_ready_ok!(pool.poll_ready());
+
+    // cycling a request should now work on svc2
+    let mut fut = task::spawn(pool.call(()));
+
+    assert_request_eq!(svc2, ()).send_response("foo");
+    assert_eq!(assert_ready_ok!(fut.poll()), "foo");
+
+    // and again (still svc2)
+    svc2.allow(1);
+    assert_ready_ok!(pool.poll_ready());
+    let mut fut = task::spawn(pool.call(()));
+
+    assert_request_eq!(svc2, ()).send_response("foo");
+    assert_eq!(assert_ready_ok!(fut.poll()), "foo");
 }
 
-#[test]
-fn failing_service() {
-    task::mock(|cx| {
-        // start the pool
-        let (mock, handle) =
-            mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
-        pin_mut!(handle);
-
-        let mut pool = Builder::new()
-            .urgency(1.0) // so _any_ Pending will add a service
-            .underutilized_below(0.0) // so no Ready will remove a service
-            .build(mock, ());
-        assert_pending!(pool.poll_ready(cx));
-
-        // give the pool a backing service
-        let (svc1_m, svc1) = mock::pair();
-        pin_mut!(svc1);
-
-        svc1.allow(1);
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
-        assert_ready_ok!(pool.poll_ready(cx));
-
-        // one request-response cycle
-        let fut = pool.call(());
-        pin_mut!(fut);
-
-        assert_request_eq!(svc1, ()).send_response("foo");
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), "foo");
-
-        // now make svc1 fail, so it has to be removed
-        svc1.send_error("ouch");
-        // polling now should recognize the failed service,
-        // try to create a new one, and then realize the maker isn't ready
-        assert_pending!(pool.poll_ready(cx));
-        // then we release another service
-        let (svc2_m, svc2) = mock::pair();
-        pin_mut!(svc2);
-
-        svc2.allow(1);
-        assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
-
-        // the pool should now be ready again
-        assert_ready_ok!(pool.poll_ready(cx));
-        // and a cycle should work (and go through svc2)
-        let fut = pool.call(());
-        pin_mut!(fut);
-
-        assert_request_eq!(svc2, ()).send_response("bar");
-        assert_eq!(assert_ready_ok!(fut.poll(cx)), "bar");
-    });
+#[tokio::test]
+async fn failing_service() {
+    // start the pool
+    let (mock, handle) = mock::pair::<(), load::Constant<mock::Mock<(), &'static str>, usize>>();
+    pin_mut!(handle);
+
+    let pool = Builder::new()
+        .urgency(1.0) // so _any_ Pending will add a service
+        .underutilized_below(0.0) // so no Ready will remove a service
+        .build(mock, ());
+
+    let mut pool = mock::Spawn::new(pool);
+
+    assert_pending!(pool.poll_ready());
+
+    // give the pool a backing service
+    let (svc1_m, svc1) = mock::pair();
+    pin_mut!(svc1);
+
+    svc1.allow(1);
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc1_m, 0));
+    assert_ready_ok!(pool.poll_ready());
+
+    // one request-response cycle
+    let mut fut = task::spawn(pool.call(()));
+
+    assert_request_eq!(svc1, ()).send_response("foo");
+    assert_eq!(assert_ready_ok!(fut.poll()), "foo");
+
+    // now make svc1 fail, so it has to be removed
+    svc1.send_error("ouch");
+    // polling now should recognize the failed service,
+    // try to create a new one, and then realize the maker isn't ready
+    assert_pending!(pool.poll_ready());
+    // then we release another service
+    let (svc2_m, svc2) = mock::pair();
+    pin_mut!(svc2);
+
+    svc2.allow(1);
+    assert_request_eq!(handle, ()).send_response(load::Constant::new(svc2_m, 0));
+
+    // the pool should now be ready again
+    assert_ready_ok!(pool.poll_ready());
+    // and a cycle should work (and go through svc2)
+    let mut fut = task::spawn(pool.call(()));
+
+    assert_request_eq!(svc2, ()).send_response("bar");
+    assert_eq!(assert_ready_ok!(fut.poll()), "bar");
 }
diff --git a/tower-filter/tests/filter.rs b/tower-filter/tests/filter.rs
index d6bf15b..240ed6f 100644
--- a/tower-filter/tests/filter.rs
+++ b/tower-filter/tests/filter.rs
@@ -1,6 +1,5 @@
 use futures_util::{future::poll_fn, pin_mut};
-use std::{future::Future, thread};
-use tokio_test::{assert_ready, assert_ready_err, task};
+use std::future::Future;
 use tower_filter::{error::Error, Filter};
 use tower_service::Service;
 use tower_test::{assert_request_eq, mock};
@@ -9,10 +8,10 @@ use tower_test::{assert_request_eq, mock};
 async fn passthrough_sync() {
     let (mut service, handle) = new_service(|_| async { Ok(()) });
 
-    let th = thread::spawn(move || {
+    let th = tokio::spawn(async move {
         // Receive the requests and respond
         pin_mut!(handle);
-        for i in 0..10 {
+        for i in 0..10usize {
             assert_request_eq!(handle, format!("ping-{}", i)).send_response(format!("pong-{}", i));
         }
     });
@@ -33,18 +32,14 @@ async fn passthrough_sync() {
     }
 
     futures_util::future::join_all(responses).await;
-    th.join().unwrap();
+    th.await.unwrap();
 }
 
-#[test]
-fn rejected_sync() {
-    task::mock(|cx| {
-        let (mut service, _handle) = new_service(|_| async { Err(Error::rejected()) });
+#[tokio::test]
+async fn rejected_sync() {
+    let (mut service, _handle) = new_service(|_| async { Err(Error::rejected()) });
 
-        let fut = service.call("hello".into());
-        pin_mut!(fut);
-        assert_ready_err!(fut.poll(cx));
-    });
+    service.call("hello".into()).await.unwrap_err();
 }
 
 type Mock = mock::Mock<String, String>;
diff --git a/tower-hedge/tests/hedge.rs b/tower-hedge/tests/hedge.rs
index 99e29ab..f33c615 100644
--- a/tower-hedge/tests/hedge.rs
+++ b/tower-hedge/tests/hedge.rs
@@ -1,154 +1,135 @@
-use futures_util::pin_mut;
-use std::future::Future;
 use std::time::Duration;
-use tokio_test::{assert_pending, assert_ready, assert_ready_ok, clock, task};
+use tokio::time;
+use tokio_test::{assert_pending, assert_ready, assert_ready_ok, task};
 use tower_hedge::{Hedge, Policy};
-use tower_service::Service;
-use tower_test::assert_request_eq;
-
-#[test]
-fn hedge_orig_completes_first() {
-    task::mock(|cx| {
-        clock::mock(|time| {
-            let (mut service, handle) = new_service(TestPolicy);
-            pin_mut!(handle);
-
-            assert_ready_ok!(service.poll_ready(cx));
-            let fut = service.call("orig");
-            pin_mut!(fut);
-
-            // Check that orig request has been issued.
-            let req = assert_request_eq!(handle, "orig");
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-            time.advance(Duration::from_millis(10));
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-            // Check that the hedge has been issued.
-            let _hedge_req = assert_request_eq!(handle, "orig");
-
-            req.send_response("orig-done");
-            // Check that fut gets orig response.
-            assert_eq!(assert_ready_ok!(fut.as_mut().poll(cx)), "orig-done");
-        });
-    });
+use tower_test::{assert_request_eq, mock};
+
+#[tokio::test]
+async fn hedge_orig_completes_first() {
+    time::pause();
+
+    let (mut service, mut handle) = new_service(TestPolicy);
+
+    assert_ready_ok!(service.poll_ready());
+    let mut fut = task::spawn(service.call("orig"));
+
+    // Check that orig request has been issued.
+    let req = assert_request_eq!(handle, "orig");
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+    time::advance(Duration::from_millis(11)).await;
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+    // Check that the hedge has been issued.
+    let _hedge_req = assert_request_eq!(handle, "orig");
+
+    req.send_response("orig-done");
+    // Check that fut gets orig response.
+    assert_eq!(assert_ready_ok!(fut.poll()), "orig-done");
 }
 
-#[test]
-fn hedge_hedge_completes_first() {
-    task::mock(|cx| {
-        clock::mock(|time| {
-            let (mut service, handle) = new_service(TestPolicy);
-            pin_mut!(handle);
-
-            assert_ready_ok!(service.poll_ready(cx));
-            let fut = service.call("orig");
-            pin_mut!(fut);
-            // Check that orig request has been issued.
-            let _req = assert_request_eq!(handle, "orig");
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-            time.advance(Duration::from_millis(10));
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-
-            // Check that the hedge has been issued.
-            let hedge_req = assert_request_eq!(handle, "orig");
-            hedge_req.send_response("hedge-done");
-            // Check that fut gets hedge response.
-            assert_eq!(assert_ready_ok!(fut.as_mut().poll(cx)), "hedge-done");
-        });
-    });
+#[tokio::test]
+async fn hedge_hedge_completes_first() {
+    time::pause();
+
+    let (mut service, mut handle) = new_service(TestPolicy);
+
+    assert_ready_ok!(service.poll_ready());
+    let mut fut = task::spawn(service.call("orig"));
+
+    // Check that orig request has been issued.
+    let _req = assert_request_eq!(handle, "orig");
+
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+    time::advance(Duration::from_millis(11)).await;
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+
+    // Check that the hedge has been issued.
+    let hedge_req = assert_request_eq!(handle, "orig");
+    hedge_req.send_response("hedge-done");
+    // Check that fut gets hedge response.
+    assert_eq!(assert_ready_ok!(fut.poll()), "hedge-done");
 }
 
-#[test]
-fn completes_before_hedge() {
-    task::mock(|cx| {
-        clock::mock(|_| {
-            let (mut service, handle) = new_service(TestPolicy);
-            pin_mut!(handle);
-
-            assert_ready_ok!(service.poll_ready(cx));
-            let fut = service.call("orig");
-            pin_mut!(fut);
-            // Check that orig request has been issued.
-            let req = assert_request_eq!(handle, "orig");
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-
-            req.send_response("orig-done");
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-            // Check that fut gets orig response.
-            assert_eq!(assert_ready_ok!(fut.as_mut().poll(cx)), "orig-done");
-        });
-    });
+#[tokio::test]
+async fn completes_before_hedge() {
+    let (mut service, mut handle) = new_service(TestPolicy);
+
+    assert_ready_ok!(service.poll_ready());
+    let mut fut = task::spawn(service.call("orig"));
+
+    // Check that orig request has been issued.
+    let req = assert_request_eq!(handle, "orig");
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+
+    req.send_response("orig-done");
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+    // Check that fut gets orig response.
+    assert_eq!(assert_ready_ok!(fut.poll()), "orig-done");
 }
 
-#[test]
-fn request_not_retyable() {
-    task::mock(|cx| {
-        clock::mock(|time| {
-            let (mut service, handle) = new_service(TestPolicy);
-            pin_mut!(handle);
-
-            assert_ready_ok!(service.poll_ready(cx));
-            let fut = service.call(NOT_RETRYABLE);
-            pin_mut!(fut);
-            // Check that orig request has been issued.
-            let req = assert_request_eq!(handle, NOT_RETRYABLE);
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-            time.advance(Duration::from_millis(10));
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-
-            req.send_response("orig-done");
-            // Check that fut gets orig response.
-            assert_eq!(assert_ready_ok!(fut.as_mut().poll(cx)), "orig-done");
-        });
-    });
+#[tokio::test]
+async fn request_not_retyable() {
+    time::pause();
+
+    let (mut service, mut handle) = new_service(TestPolicy);
+
+    assert_ready_ok!(service.poll_ready());
+    let mut fut = task::spawn(service.call(NOT_RETRYABLE));
+
+    // Check that orig request has been issued.
+    let req = assert_request_eq!(handle, NOT_RETRYABLE);
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+    time::advance(Duration::from_millis(10)).await;
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+
+    req.send_response("orig-done");
+    // Check that fut gets orig response.
+    assert_eq!(assert_ready_ok!(fut.poll()), "orig-done");
 }
 
-#[test]
-fn request_not_clonable() {
-    task::mock(|cx| {
-        clock::mock(|time| {
-            let (mut service, handle) = new_service(TestPolicy);
-            pin_mut!(handle);
-
-            assert_ready_ok!(service.poll_ready(cx));
-            let fut = service.call(NOT_CLONABLE);
-            pin_mut!(fut);
-            // Check that orig request has been issued.
-            let req = assert_request_eq!(handle, NOT_CLONABLE);
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-            time.advance(Duration::from_millis(10));
-            // Check fut is not ready.
-            assert_pending!(fut.as_mut().poll(cx));
-            // Check hedge has not been issued.
-            assert_pending!(handle.as_mut().poll_request(cx));
-
-            req.send_response("orig-done");
-            // Check that fut gets orig response.
-            assert_eq!(assert_ready_ok!(fut.as_mut().poll(cx)), "orig-done");
-        });
-    });
+#[tokio::test]
+async fn request_not_clonable() {
+    time::pause();
+
+    let (mut service, mut handle) = new_service(TestPolicy);
+
+    assert_ready_ok!(service.poll_ready());
+    let mut fut = task::spawn(service.call(NOT_CLONABLE));
+
+    // Check that orig request has been issued.
+    let req = assert_request_eq!(handle, NOT_CLONABLE);
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+    time::advance(Duration::from_millis(10)).await;
+    // Check fut is not ready.
+    assert_pending!(fut.poll());
+    // Check hedge has not been issued.
+    assert_pending!(handle.poll_request());
+
+    req.send_response("orig-done");
+    // Check that fut gets orig response.
+    assert_eq!(assert_ready_ok!(fut.poll()), "orig-done");
 }
 
 type Req = &'static str;
@@ -176,7 +157,7 @@ impl tower_hedge::Policy<Req> for TestPolicy {
     }
 }
 
-fn new_service<P: Policy<Req> + Clone>(policy: P) -> (Hedge<Mock, P>, Handle) {
+fn new_service<P: Policy<Req> + Clone>(policy: P) -> (mock::Spawn<Hedge<Mock, P>>, Handle) {
     let (service, handle) = tower_test::mock::pair();
 
     let mock_latencies: [u64; 10] = [1, 1, 1, 1, 1, 1, 1, 1, 10, 10];
@@ -189,5 +170,6 @@ fn new_service<P: Policy<Req> + Clone>(policy: P) -> (Hedge<Mock, P>, Handle) {
         Duration::from_secs(60),
         &mock_latencies,
     );
-    (service, handle)
+
+    (mock::Spawn::new(service), handle)
 }
diff --git a/tower-load-shed/tests/load-shed.rs b/tower-load-shed/tests/load-shed.rs
index 3090332..de92418 100644
--- a/tower-load-shed/tests/load-shed.rs
+++ b/tower-load-shed/tests/load-shed.rs
@@ -1,49 +1,31 @@
-use futures_util::pin_mut;
-use std::future::Future;
-use tokio_test::{assert_ready_err, assert_ready_ok, task::mock};
-use tower_load_shed::{self, LoadShed};
-use tower_service::Service;
+use tokio_test::{assert_ready_err, assert_ready_ok, task};
+use tower_load_shed::LoadShedLayer;
 use tower_test::{assert_request_eq, mock};
 
-#[test]
-fn when_ready() {
-    mock(|cx| {
-        let (mut service, handle) = new_service();
-        pin_mut!(handle);
+#[tokio::test]
+async fn when_ready() {
+    let layer = LoadShedLayer::new();
+    let (mut service, mut handle) = mock::spawn_layer(layer);
 
-        assert_ready_ok!(service.poll_ready(cx), "overload always reports ready");
+    assert_ready_ok!(service.poll_ready(), "overload always reports ready");
 
-        let response = service.call("hello");
-        pin_mut!(response);
+    let mut response = task::spawn(service.call("hello"));
 
-        assert_request_eq!(handle, "hello").send_response("world");
-        assert_eq!(assert_ready_ok!(response.poll(cx)), "world");
-    });
+    assert_request_eq!(handle, "hello").send_response("world");
+    assert_eq!(assert_ready_ok!(response.poll()), "world");
 }
 
-#[test]
-fn when_not_ready() {
-    mock(|cx| {
-        let (mut service, handle) = new_service();
-        pin_mut!(handle);
+#[tokio::test]
+async fn when_not_ready() {
+    let layer = LoadShedLayer::new();
+    let (mut service, mut handle) = mock::spawn_layer::<_, (), _>(layer);
 
-        handle.allow(0);
+    handle.allow(0);
 
-        assert_ready_ok!(service.poll_ready(cx), "overload always reports ready");
+    assert_ready_ok!(service.poll_ready(), "overload always reports ready");
 
-        let fut = service.call("hello");
-        pin_mut!(fut);
+    let mut fut = task::spawn(service.call("hello"));
 
-        let err = assert_ready_err!(fut.poll(cx));
-        assert!(err.is::<tower_load_shed::error::Overloaded>());
-    });
-}
-
-type Mock = mock::Mock<&'static str, &'static str>;
-type Handle = mock::Handle<&'static str, &'static str>;
-
-fn new_service() -> (LoadShed<Mock>, Handle) {
-    let (service, handle) = mock::pair();
-    let service = LoadShed::new(service);
-    (service, handle)
+    let err = assert_ready_err!(fut.poll());
+    assert!(err.is::<tower_load_shed::error::Overloaded>());
 }
diff --git a/tower-ready-cache/tests/ready_cache.rs b/tower-ready-cache/tests/ready_cache.rs
index c172319..5d57bc1 100644
--- a/tower-ready-cache/tests/ready_cache.rs
+++ b/tower-ready-cache/tests/ready_cache.rs
@@ -1,17 +1,13 @@
-use futures::prelude::*;
-use tower_ready_cache::{error, ReadyCache};
+use tokio_test::{assert_pending, assert_ready, task};
+use tower_ready_cache::ReadyCache;
 use tower_test::mock;
 
-fn with_task<F: FnOnce() -> U, U>(f: F) -> U {
-    use futures::future::lazy;
-    lazy(|| Ok::<_, ()>(f())).wait().unwrap()
-}
-
 type Req = &'static str;
 type Mock = mock::Mock<Req, Req>;
 
 #[test]
 fn poll_ready_inner_failure() {
+    let mut task = task::spawn(());
     let mut cache = ReadyCache::<usize, Mock, Req>::default();
 
     let (service0, mut handle0) = mock::pair::<Req, Req>();
@@ -22,20 +18,17 @@ fn poll_ready_inner_failure() {
     handle1.allow(1);
     cache.push(1, service1);
 
-    with_task(|| {
-        let error::Failed(key, err) = cache
-            .poll_pending()
-            .err()
-            .expect("poll_ready should fail when exhausted");
-        assert_eq!(key, 0);
-        assert_eq!(format!("{}", err), "doom");
-    });
+    let failed = assert_ready!(task.enter(|cx, _| cache.poll_pending(cx))).unwrap_err();
+
+    assert_eq!(failed.0, 0);
+    assert_eq!(format!("{}", failed.1), "doom");
 
     assert_eq!(cache.len(), 1);
 }
 
 #[test]
 fn poll_ready_not_ready() {
+    let mut task = task::spawn(());
     let mut cache = ReadyCache::<usize, Mock, Req>::default();
 
     let (service0, mut handle0) = mock::pair::<Req, Req>();
@@ -46,9 +39,7 @@ fn poll_ready_not_ready() {
     handle1.allow(0);
     cache.push(1, service1);
 
-    with_task(|| {
-        assert!(cache.poll_pending().expect("must succeed").is_not_ready());
-    });
+    assert_pending!(task.enter(|cx, _| cache.poll_pending(cx)));
 
     assert_eq!(cache.ready_len(), 0);
     assert_eq!(cache.pending_len(), 2);
@@ -57,6 +48,7 @@ fn poll_ready_not_ready() {
 
 #[test]
 fn poll_ready_promotes_inner() {
+    let mut task = task::spawn(());
     let mut cache = ReadyCache::<usize, Mock, Req>::default();
 
     let (service0, mut handle0) = mock::pair::<Req, Req>();
@@ -71,9 +63,7 @@ fn poll_ready_promotes_inner() {
     assert_eq!(cache.pending_len(), 2);
     assert_eq!(cache.len(), 2);
 
-    with_task(|| {
-        assert!(cache.poll_pending().expect("must succeed").is_ready());
-    });
+    assert_ready!(task.enter(|cx, _| cache.poll_pending(cx))).unwrap();
 
     assert_eq!(cache.ready_len(), 2);
     assert_eq!(cache.pending_len(), 0);
diff --git a/tower-spawn-ready/tests/spawn_ready.rs b/tower-spawn-ready/tests/spawn_ready.rs
index 66dc4cd..bef3bef 100644
--- a/tower-spawn-ready/tests/spawn_ready.rs
+++ b/tower-spawn-ready/tests/spawn_ready.rs
@@ -1,91 +1,35 @@
-use futures_util::pin_mut;
-use std::{future::Future, thread, time::Duration};
-use tokio_executor::{Executor, SpawnError};
-use tokio_test::{assert_pending, assert_ready, assert_ready_err, assert_ready_ok, task};
-use tower_service::Service;
-use tower_spawn_ready::{error, SpawnReady};
+use std::{thread, time::Duration};
+use tokio_test::{assert_pending, assert_ready, assert_ready_err, assert_ready_ok};
+use tower_spawn_ready::SpawnReadyLayer;
 use tower_test::mock;
 
-#[test]
-fn when_inner_is_not_ready() {
-    let mut ex = Exec;
-    tokio_executor::with_default(&mut ex, || {
-        task::mock(|cx| {
-            let (mut service, handle) = new_service();
-            pin_mut!(handle);
+#[tokio::test]
+async fn when_inner_is_not_ready() {
+    let layer = SpawnReadyLayer::new();
+    let (mut service, mut handle) = mock::spawn_layer::<(), (), _>(layer);
 
-            // Make the service NotReady
-            handle.allow(0);
+    // Make the service NotReady
+    handle.allow(0);
 
-            assert_pending!(service.poll_ready(cx));
+    assert_pending!(service.poll_ready());
 
-            // Make the service is Ready
-            handle.allow(1);
-            thread::sleep(Duration::from_millis(100));
-            assert_ready_ok!(service.poll_ready(cx));
-        });
-    });
+    // Make the service is Ready
+    handle.allow(1);
+    thread::sleep(Duration::from_millis(100));
+    assert_ready_ok!(service.poll_ready());
 }
 
-#[test]
-fn when_inner_fails() {
-    task::mock(|cx| {
-        let (mut service, handle) = new_service();
-        pin_mut!(handle);
+#[tokio::test]
+async fn when_inner_fails() {
+    let layer = SpawnReadyLayer::new();
+    let (mut service, mut handle) = mock::spawn_layer::<(), (), _>(layer);
 
-        // Make the service NotReady
-        handle.allow(0);
-        handle.send_error("foobar");
+    // Make the service NotReady
+    handle.allow(0);
+    handle.send_error("foobar");
 
-        assert_eq!(
-            assert_ready_err!(service.poll_ready(cx)).to_string(),
-            "foobar"
-        );
-    });
-}
-
-#[test]
-fn when_spawn_fails() {
-    task::mock(|cx| {
-        let (service, handle) = mock::pair::<(), ()>();
-        pin_mut!(handle);
-
-        let mut service = SpawnReady::new(service);
-
-        // Make the service NotReady so a background task is spawned.
-        handle.allow(0);
-
-        let err = assert_ready_err!(service.poll_ready(cx));
-
-        assert!(
-            err.is::<error::SpawnError>(),
-            "should be a SpawnError: {:?}",
-            err
-        );
-    });
-}
-
-type Mock = mock::Mock<&'static str, &'static str>;
-type Handle = mock::Handle<&'static str, &'static str>;
-
-struct Exec;
-
-impl Executor for Exec {
-    fn spawn(
-        &mut self,
-        fut: std::pin::Pin<Box<dyn Future<Output = ()> + Send>>,
-    ) -> Result<(), SpawnError> {
-        thread::spawn(move || {
-            let mut mock = tokio_test::task::MockTask::new();
-            pin_mut!(fut);
-            while mock.poll(fut.as_mut()).is_pending() {}
-        });
-        Ok(())
-    }
-}
-
-fn new_service() -> (SpawnReady<Mock>, Handle) {
-    let (service, handle) = mock::pair();
-    let service = SpawnReady::new(service);
-    (service, handle)
+    assert_eq!(
+        assert_ready_err!(service.poll_ready()).to_string(),
+        "foobar"
+    );
 }