update the proxy to use Tokio 0.3

Cargo.toml

@@ -63,4 +63,19 @@ debug = false
 
 [patch.crates-io]
 webpki = { git = "https://github.com/linkerd/webpki", branch = "cert-dns-names-0.21", rev = "b2c3bb3" }
-tower = { version = "0.3", git = "https://github.com/tower-rs/tower", rev = "ad348d8" }
+# TODO(eliza): when the Tokio 0.3 versions of our various dependencies are
+# published, remove these patches...
+tower = { version = "0.4", git = "https://github.com/tower-rs/tower", rev = "450fa3d2be2b43850ceb125009d636d1d8629ad7" }
+hyper = { git = "https://github.com/hyperium/hyper", rev = "d6aadb830072959497f414c01bcdba4c8e681088" }
+# this isn't pinned to a rev because it needs to be consistent with Hyper's git
+# dep (which is not pinned).
+h2 = { git = "https://github.com/hyperium/h2" }
+# this isn't pinned to a rev because it needs to be consistent with Hyper's git
+# dep (which is `branch = "master"`).
+http-body = { git = "https://github.com/hyperium/http-body", branch = "master" }
+tokio-rustls = { git = "https://github.com/tokio-rs/tls", rev = "e6ef54641b911cfcf23b77a8c4826ae0f8e9870e" }
+tonic = { git = "https://github.com/hawkw/tonic", branch = "eliza/tokio-0.3" }
+tonic-build = { git = "https://github.com/hawkw/tonic", branch = "eliza/tokio-0.3" }
+prost = { git = "https://github.com/danburkert/prost" }
+prost-build = { git = "https://github.com/danburkert/prost" }
+prost-types = { git = "https://github.com/danburkert/prost" }

hyper-balance/Cargo.toml

@@ -8,10 +8,10 @@ publish = false
 [dependencies]
 futures = "0.3"
 http = "0.2"
-hyper = "0.13.7"
+hyper = "0.14.0-dev"
 pin-project = "0.4"
-tower = { version = "0.3", default-features = false, features = ["load"]}
-tokio = { version = "0.2", features = ["macros"]}
+tower = { version = "0.4", default-features = false, features = ["load"]}
+tokio = { version = "0.3", features = ["macros"]}
 
 [dev-dependencies]
 tokio-test = "0.2"

linkerd/app/Cargo.toml

@@ -24,8 +24,8 @@ linkerd2-app-outbound = { path = "./outbound" }
 linkerd2-opencensus = { path = "../opencensus" }
 linkerd2-error = { path = "../error" }
 regex = "1.0.0"
-tokio = { version = "0.2", features = ["rt-util"] }
-tonic = { version = "0.3", default-features = false }
-tower = "0.3"
-tracing = "0.1.19"
-tracing-futures = { version = "0.2", features = ["std-future"]}
+tokio = { version = "0.3", features = ["rt"] }
+tonic = { version = "0.3", default-features = false, features = ["prost"] }
+tower = "0.4"
+tracing = "0.1.22"
+tracing-futures = { version = "0.2", features = ["std-future"]}

linkerd/app/core/Cargo.toml

@@ -15,10 +15,10 @@ independently of the inbound and outbound proxy logic.
 mock-orig-dst  = ["linkerd2-proxy-transport/mock-orig-dst"]
 
 [dependencies]
-bytes = "0.5"
+bytes = "0.6"
 http = "0.2"
-http-body = "0.3"
-hyper = "0.13.7"
+http-body = "0.4"
+hyper = "0.14.0-dev"
 futures = "0.3"
 indexmap = "1.0"
 ipnet = "1.0"
@@ -40,7 +40,7 @@ linkerd2-http-metrics = { path = "../../http-metrics" }
 linkerd2-metrics = { path = "../../metrics" }
 linkerd2-opencensus = { path = "../../opencensus" }
 linkerd2-proxy-core = { path = "../../proxy/core" }
-linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.15" }
+linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", rev = "259628840ba613c2e5673fc6a39b946e1b06f09a" }
 linkerd2-proxy-api-resolve = { path = "../../proxy/api-resolve" }
 linkerd2-proxy-discover = { path = "../../proxy/discover" }
 linkerd2-proxy-identity = { path = "../../proxy/identity" }
@@ -60,16 +60,16 @@ linkerd2-stack-metrics = { path = "../../stack/metrics" }
 linkerd2-stack-tracing = { path = "../../stack/tracing" }
 linkerd2-trace-context = { path = "../../trace-context" }
 regex = "1.0.0"
-tokio = { version = "0.2.22", features = ["macros", "sync", "parking_lot"]}
+tokio = { version = "0.3", features = ["macros", "sync", "parking_lot"]}
 tokio-timer = "0.2"
-tower-request-modifier = { git = "https://github.com/tower-rs/tower-http" }
+tower-request-modifier = { git = "https://github.com/tower-rs/tower-http", rev = "bd7a4654bdc4e2b5363572e9f66b4dbbc7c0e1ea" }
 tonic = { version = "0.3", default-features = false, features = ["prost"] }
-tracing = "0.1.19"
+tracing = "0.1.22"
 tracing-futures = { version = "0.2" }
 pin-project = "0.4"
 
 [dependencies.tower]
-version = "0.3"
+version = "0.4"
 # disable tower's tracing `log` integration for performance reasons, since we
 # will consume tower's traces as traces.
 default-features = false
@@ -85,5 +85,5 @@ libc = "0.2"
 procinfo = "0.4.2"
 
 [dev-dependencies]
-linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.15", features = ["arbitrary"] }
+linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", rev = "259628840ba613c2e5673fc6a39b946e1b06f09a", features = ["arbitrary"] }
 prost-types = "0.6.0"

linkerd/app/core/src/control.rs

@@ -34,7 +34,7 @@ impl fmt::Display for ControlAddr {
 }
 
 type BalanceBody =
-    http::balance::PendingUntilFirstDataBody<tower::load::peak_ewma::Handle, http::Payload>;
+    http::balance::PendingUntilFirstDataBody<tower::load::peak_ewma::Handle, hyper::Body>;
 
 type RspBody = linkerd2_http_metrics::requests::ResponseBody<BalanceBody, classify::Eos>;
 

linkerd/app/gateway/Cargo.toml

@@ -12,11 +12,12 @@ indexmap = "1.0"
 linkerd2-app-core = { path = "../core" }
 linkerd2-app-inbound = { path = "../inbound" }
 linkerd2-app-outbound = { path = "../outbound" }
-tower = { version = "0.3", default-features = false }
-tracing = "0.1.19"
+tower = { version = "0.4", default-features = false }
+tracing = "0.1.22"
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["rt-core", "macros"] }
+tokio = { version = "0.3", features = ["rt", "macros"] }
 tokio-test = "0.2"
-tower = { version = "0.3", default-features = false, features = ["util"] }
+tower = { version = "0.4", default-features = false, features = ["util"] }
 tower-test = "0.3"
+linkerd2-app-test = { path = "../test" }

linkerd/app/gateway/src/lib.rs

@@ -19,6 +19,7 @@ mod test {
         Error, NameAddr, NameMatch, Never,
     };
     use linkerd2_app_inbound::endpoint as inbound;
+    use linkerd2_app_test as support;
     use std::{net::SocketAddr, str::FromStr};
     use tower::util::{service_fn, ServiceExt};
     use tower_test::mock;
@@ -133,11 +134,10 @@ mod test {
             >();
             let mut make_gateway = {
                 let profiles = service_fn(move |na: NameAddr| async move {
-                    let (mut tx, rx) = tokio::sync::watch::channel(profiles::Profile {
+                    let rx = support::profile::only(profiles::Profile {
                         name: Some(na.name().clone()),
                         ..profiles::Profile::default()
                     });
-                    tokio::spawn(async move { tx.closed().await });
                     Ok::<_, Never>(Some(rx))
                 });
                 let allow_discovery = NameMatch::new(Some(dns::Suffix::from_str(suffix).unwrap()));

linkerd/app/inbound/Cargo.toml

@@ -9,16 +9,16 @@ Configures and runs the inbound proxy
 """
 
 [dependencies]
-bytes = "0.5"
+bytes = "0.6"
 http = "0.2"
 futures = { version = "0.3" }
 indexmap = "1.0"
 linkerd2-app-core = { path = "../core" }
-tokio = { version = "0.2", features = ["sync"] }
-tracing = "0.1.19"
+tokio = { version = "0.3", features = ["sync"] }
+tracing = "0.1.22"
 
 [dependencies.tower]
-version = "0.3"
+version = "0.4"
 # disable tower's tracing `log` integration for performance reasons, since we
 # will consume tower's traces as traces.
 default-features = false

linkerd/app/integration/Cargo.toml

@@ -18,25 +18,25 @@ flaky_tests = []
 nyi = []
 
 [dependencies]
-bytes = "0.5"
-futures = "0.3" 
-h2 = "0.2.6"
+bytes = "0.6"
+futures = "0.3"
+h2 = "0.3"
 http = "0.2"
-http-body = "0.3"
-hyper = "0.13.7"
+http-body = "0.4"
+hyper = "0.14.0-dev"
 linkerd2-app = { path = "..", features = ["mock-orig-dst"] }
 linkerd2-app-core = { path = "../core", features = ["mock-orig-dst"] }
 linkerd2-metrics = { path = "../../metrics", features = ["test_util"] }
-linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.15", features = ["arbitrary"] }
+linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", rev = "259628840ba613c2e5673fc6a39b946e1b06f09a", features = ["arbitrary"] }
 linkerd2-app-test = { path = "../test" }
 regex = "0.1"
 socket2 = "0.3.12"
 rustls = "0.18"
-tokio = { version = "0.2", features = ["io-util", "net", "rt-core"]}
-tokio-rustls = "0.14.1"
-tower = { version = "0.3", default-features = false}
+tokio = { version = "0.3", features = ["io-util", "net", "rt", "macros"]}
+tokio-rustls = "0.20"
+tower = { version = "0.4", default-features = false}
 tonic = { version = "0.3", default-features = false }
-tracing = "0.1.19"
+tracing = "0.1.22"
 tracing-futures = { version = "0.2", features = ["std-future"] }
 webpki = "0.21"
 

linkerd/app/integration/src/client.rs

@@ -1,4 +1,5 @@
 use super::*;
+use hyper::body::Buf;
 use linkerd2_app_core::proxy::http::trace;
 use rustls::ClientConfig;
 use std::io;
@@ -121,7 +122,7 @@ impl Client {
         );
         let stream = res.into_parts().1;
         let mut body = hyper::body::aggregate(stream).await.expect("wait body");
-        std::str::from_utf8(body.to_bytes().as_ref())
+        std::str::from_utf8(body.copy_to_bytes(body.remaining()).as_ref())
             .unwrap()
             .to_string()
     }

linkerd/app/integration/src/controller.rs

@@ -362,7 +362,7 @@ where
     let task = tokio::spawn(
         cancelable(drain.clone(), async move {
             // Start listening on the socket.
-            let mut listener = crate::listen(sock);
+            let listener = crate::listen(sock);
             let mut listening_tx = Some(listening_tx);
 
             if let Some(delay) = delay {

linkerd/app/integration/src/lib.rs

@@ -30,7 +30,7 @@ use std::pin::Pin;
 pub use std::sync::Arc;
 use std::task::{Context, Poll};
 pub use std::time::Duration;
-pub use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+pub use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
 use tokio::net::TcpListener;
 pub use tokio::stream::{Stream, StreamExt};
 pub use tokio::sync::oneshot;
@@ -69,7 +69,7 @@ macro_rules! assert_eventually {
             use std::{env, u64};
             use std::time::{Instant, Duration};
             use std::str::FromStr;
-            use tracing_futures::Instrument as _;
+            use tracing::Instrument as _;
             // TODO: don't do this *every* time eventually is called (lazy_static?)
             let patience = env::var($crate::ENV_TEST_PATIENCE_MS).ok()
                 .map(|s| {
@@ -96,7 +96,7 @@ macro_rules! assert_eventually {
                         )
                     } else {
                         tracing::trace!("waiting...");
-                        tokio::time::delay_for(patience).await;
+                        tokio::time::sleep(patience).await;
                         std::thread::yield_now();
                         tracing::trace!("done");
                     }
@@ -165,14 +165,10 @@ impl AsyncRead for RunningIo {
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         self.as_mut().io.as_mut().poll_read(cx, buf)
     }
-
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
-        self.io.prepare_uninitialized_buffer(buf)
-    }
 }
 
 impl AsyncWrite for RunningIo {
@@ -297,6 +293,8 @@ pub(crate) fn bind_ephemeral() -> (Socket, SocketAddr) {
 pub(crate) fn listen(sock: Socket) -> TcpListener {
     sock.listen(1024)
         .expect("socket should be able to start listening");
-    TcpListener::from_std(sock.into_tcp_listener())
-        .expect("socket should be able to set nonblocking")
+    let sock = sock.into_tcp_listener();
+    sock.set_nonblocking(true)
+        .expect("socket should be able to set nonblocking");
+    TcpListener::from_std(sock).expect("socket should seem okay to tokio")
 }

linkerd/app/integration/src/proxy.rs

@@ -313,9 +313,8 @@ async fn run(proxy: Proxy, mut env: TestEnv, random_ports: bool) -> Listening {
                 let span = info_span!("proxy", test = %thread_name());
                 let _enter = span.enter();
 
-                tokio::runtime::Builder::new()
+                tokio::runtime::Builder::new_current_thread()
                     .enable_all()
-                    .basic_scheduler()
                     .build()
                     .expect("proxy")
                     .block_on(async move {

linkerd/app/integration/src/server.rs

@@ -169,7 +169,7 @@ impl Server {
         self.route_async(path, move |_| {
             let resp = resp.clone();
             async move {
-                tokio::time::delay_for(latency).await;
+                tokio::time::sleep(latency).await;
                 Ok::<_, BoxError>(
                     http::Response::builder()
                         .status(200)
@@ -223,7 +223,7 @@ impl Server {
                 }
 
                 // After the delay, start listening on the socket.
-                let mut listener = crate::listen(sock);
+                let listener = crate::listen(sock);
 
                 if let Some(listening_tx) = listening_tx {
                     let _ = listening_tx.send(());

linkerd/app/integration/src/tcp.rs

@@ -212,7 +212,10 @@ async fn run_server(tcp: TcpServer) -> server::Listening {
     let task = tokio::spawn(
         cancelable(drain.clone(), async move {
             let mut accepts = tcp.accepts;
-            let mut listener = TcpListener::from_std(std_listener).expect("TcpListener::from_std");
+            std_listener
+                .set_nonblocking(true)
+                .expect("socket must be able to set nonblocking");
+            let listener = TcpListener::from_std(std_listener).expect("TcpListener::from_std");
 
             let _ = started_tx.send(());
             loop {

linkerd/app/integration/src/tests/discovery.rs

@@ -213,12 +213,13 @@ macro_rules! generate_tests {
             .await
         }
 
-        #[tokio::test]
+        #[tokio::test(flavor = "current_thread")]
         async fn outbound_destinations_reset_on_reconnect_followed_by_dne() {
             outbound_destinations_reset_on_reconnect(controller::destination_does_not_exist()).await
         }
 
         async fn outbound_destinations_reset_on_reconnect(up: pb::destination::Update) {
+            let _trace = trace_init();
             let env = TestEnv::new();
             let srv = $make_server().route("/", "hello").run().await;
             let ctrl = controller::new();
@@ -250,7 +251,7 @@ macro_rules! generate_tests {
             dst_tx1.send(up);
 
             // Wait for the reconnect to happen. TODO: Replace this flaky logic.
-            tokio::time::delay_for(Duration::from_millis(1000)).await;
+            tokio::time::sleep(Duration::from_millis(1000)).await;
 
             let rsp = initially_exists
                 .request(initially_exists.request_builder("/"))
@@ -320,12 +321,12 @@ macro_rules! generate_tests {
                 .await;
 
             // Allow the control client to notice a connection error
-            tokio::time::delay_for(Duration::from_millis(500)).await;
+            tokio::time::sleep(Duration::from_millis(500)).await;
 
             // Allow our controller to start accepting connections,
             // and then wait a little bit so the client tries again.
             drop(tx);
-            tokio::time::delay_for(Duration::from_millis(500)).await;
+            tokio::time::sleep(Duration::from_millis(500)).await;
 
             let client = $make_client(proxy.outbound, "disco.test.svc.cluster.local");
 

linkerd/app/integration/src/tests/identity.rs

@@ -232,7 +232,7 @@ async fn refresh() {
     let expiry = expiry_rx.await.expect("wait for expiry");
     let how_long = expiry.duration_since(SystemTime::now()).unwrap();
 
-    tokio::time::delay_for(how_long).await;
+    tokio::time::sleep(how_long).await;
 
     assert_eventually!(refreshed.load(Ordering::SeqCst) == true);
 }

linkerd/app/integration/src/tests/profile_dst_overrides.rs

@@ -61,7 +61,7 @@ async fn wait_for_profile_stage(client: &client::Client, metrics: &client::Clien
             break;
         }
 
-        tokio::time::delay_for(std::time::Duration::from_millis(200)).await;
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
     }
 }
 

linkerd/app/integration/src/tests/profiles.rs

@@ -122,7 +122,7 @@ macro_rules! profile_test {
                 break;
             }
 
-            tokio::time::delay_for(std::time::Duration::from_millis(200)).await;
+            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
         }
 
         $with_client(client).await;