Upgrade to Tokio 1.0.0 ecosystem (#530)

* Upgrade Tonic to Tokio 1.0

Work in progress for updating Tonic to Tokio 1.0. Since tower has not
been released to crates.io, a git dependency is taken instead.

* Upgrade Tonic to Tokio 1.0 phase 2

* tonic: remove tower-* deps

* Apply suggestions from code review

Co-authored-by: Ed Marshall <[email protected]>
Co-authored-by: Lucio Franco <[email protected]>

tonic/Cargo.toml

@@ -30,8 +30,6 @@ transport = [
     "hyper",
     "tokio",
     "tower",
-    "tower-balance",
-    "tower-load",
     "tracing-futures",
 ]
 tls = ["transport", "tokio-rustls"]
@@ -43,45 +41,43 @@ prost = ["prost1", "prost-derive"]
 # harness = false
 
 [dependencies]
-bytes = "0.5"
+bytes = "1.0"
 futures-core = { version = "0.3", default-features = false }
 futures-util = { version = "0.3", default-features = false }
 tracing = "0.1"
 http = "0.2"
-base64 = "0.12"
+base64 = "0.13"
 
-percent-encoding = "2.0"
+percent-encoding = "2.1"
 tower-service = "0.3"
-tokio-util = { version = "0.3", features = ["codec"] }
-async-stream = "0.2"
-http-body = "0.3"
-pin-project = "0.4.17"
+tokio-util = { version = "0.6", features = ["codec"] }
+async-stream = "0.3"
+http-body = "0.4"
+pin-project = "1.0"
 
 # prost
-prost1 = { package = "prost", version = "0.6", optional = true }
-prost-derive = { version = "0.6", optional = true }
+prost1 = { package = "prost", version = "0.7", optional = true }
+prost-derive = { version = "0.7", optional = true }
 
 # codegen
 async-trait = { version = "0.1.13", optional = true }
 
 # transport
-h2    = { version = "0.2.2", optional = true }
-hyper = { version = "0.13.4", features = ["stream"], optional = true }
-tokio = { version = "0.2.13", features = ["tcp"], optional = true }
-tower = { version = "0.3", optional = true}
-tower-make = { version = "0.3", features = ["connect"] }
-tower-balance =  { version = "0.3", optional = true }
-tower-load = { version = "0.3", optional = true }
+h2    = { version = "0.3", optional = true }
+hyper = { version = "0.14.2", features = ["full"], optional = true }
+tokio = { version = "1.0.1", features = ["net"], optional = true }
+tokio-stream = "0.1"
+tower = { version = "0.4", features = ["balance", "buffer", "discover", "limit", "load", "make", "timeout", "util"], optional = true}
 tracing-futures = { version = "0.2", optional = true }
 
 # rustls
-tokio-rustls = { version = "0.14", optional = true }
-rustls-native-certs = { version = "0.4", optional = true }
+tokio-rustls = { version = "0.22", optional = true }
+rustls-native-certs = { version = "0.5", optional = true }
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["rt-core", "macros"] }
+tokio = { version = "1.0", features = ["rt", "macros"] }
 static_assertions = "1.0"
-rand = "0.7"
+rand = "0.8"
 bencher = "0.1.5"
 
 [package.metadata.docs.rs]
@@ -91,4 +87,3 @@ rustdoc-args = ["--cfg", "docsrs"]
 [[bench]]
 name = "decode"
 harness = false
-

tonic/benches/decode.rs

@@ -11,8 +11,7 @@ use tonic::{codec::DecodeBuf, codec::Decoder, Status, Streaming};
 macro_rules! bench {
     ($name:ident, $message_size:expr, $chunk_size:expr, $message_count:expr) => {
         fn $name(b: &mut Bencher) {
-            let mut rt = tokio::runtime::Builder::new()
-                .basic_scheduler()
+            let rt = tokio::runtime::Builder::new_multi_thread()
                 .build()
                 .expect("runtime");
 
@@ -102,7 +101,7 @@ impl Decoder for MockDecoder {
     type Error = Status;
 
     fn decode(&mut self, buf: &mut DecodeBuf<'_>) -> Result<Option<Self::Item>, Self::Error> {
-        let out = Vec::from(buf.bytes());
+        let out = Vec::from(buf.chunk());
         buf.advance(self.message_size);
         Ok(Some(out))
     }

tonic/src/body.rs

@@ -160,7 +160,7 @@ where
             Pin::new_unchecked(&mut me.0).poll_data(cx)
         };
         match futures_util::ready!(v) {
-            Some(Ok(mut i)) => Poll::Ready(Some(Ok(i.to_bytes()))),
+            Some(Ok(mut i)) => Poll::Ready(Some(Ok(i.copy_to_bytes(i.remaining())))),
             Some(Err(e)) => {
                 let err = Status::map_error(e.into());
                 Poll::Ready(Some(Err(err)))

tonic/src/codec/buffer.rs

@@ -1,5 +1,5 @@
+use bytes::buf::UninitSlice;
 use bytes::{Buf, BufMut, BytesMut};
-use std::mem::MaybeUninit;
 
 /// A specialized buffer to decode gRPC messages from.
 #[derive(Debug)]
@@ -27,8 +27,8 @@ impl Buf for DecodeBuf<'_> {
     }
 
     #[inline]
-    fn bytes(&self) -> &[u8] {
-        let ret = self.buf.bytes();
+    fn chunk(&self) -> &[u8] {
+        let ret = self.buf.chunk();
 
         if ret.len() > self.len {
             &ret[..self.len]
@@ -63,7 +63,7 @@ impl EncodeBuf<'_> {
     }
 }
 
-impl BufMut for EncodeBuf<'_> {
+unsafe impl BufMut for EncodeBuf<'_> {
     #[inline]
     fn remaining_mut(&self) -> usize {
         self.buf.remaining_mut()
@@ -75,8 +75,8 @@ impl BufMut for EncodeBuf<'_> {
     }
 
     #[inline]
-    fn bytes_mut(&mut self) -> &mut [MaybeUninit<u8>] {
-        self.buf.bytes_mut()
+    fn chunk_mut(&mut self) -> &mut UninitSlice {
+        self.buf.chunk_mut()
     }
 }
 
@@ -92,7 +92,7 @@ mod tests {
 
         assert_eq!(buf.len, 20);
         assert_eq!(buf.remaining(), 20);
-        assert_eq!(buf.bytes().len(), 20);
+        assert_eq!(buf.chunk().len(), 20);
 
         buf.advance(10);
         assert_eq!(buf.remaining(), 10);
@@ -100,9 +100,9 @@ mod tests {
         let mut out = [0; 5];
         buf.copy_to_slice(&mut out);
         assert_eq!(buf.remaining(), 5);
-        assert_eq!(buf.bytes().len(), 5);
+        assert_eq!(buf.chunk().len(), 5);
 
-        assert_eq!(buf.to_bytes().len(), 5);
+        assert_eq!(buf.copy_to_bytes(5).len(), 5);
         assert!(!buf.has_remaining());
     }
 

tonic/src/transport/channel/endpoint.rs

@@ -15,7 +15,7 @@ use std::{
     fmt,
     time::Duration,
 };
-use tower_make::MakeConnection;
+use tower::make::MakeConnection;
 
 /// Channel builder.
 ///

tonic/src/transport/channel/mod.rs

@@ -29,13 +29,13 @@ use tokio::{
     sync::mpsc::{channel, Sender},
 };
 
+use tower::balance::p2c::Balance;
 use tower::{
     buffer::{self, Buffer},
     discover::{Change, Discover},
     util::{BoxService, Either},
     Service,
 };
-use tower_balance::p2c::Balance;
 
 type Svc = Either<Connection, BoxService<Request<BoxBody>, Response<hyper::Body>, crate::Error>>;
 
@@ -109,7 +109,7 @@ impl Channel {
     /// This creates a [`Channel`] that will load balance accross all the
     /// provided endpoints.
     pub fn balance_list(list: impl Iterator<Item = Endpoint>) -> Self {
-        let (channel, mut tx) = Self::balance_channel(DEFAULT_BUFFER_SIZE);
+        let (channel, tx) = Self::balance_channel(DEFAULT_BUFFER_SIZE);
         list.for_each(|endpoint| {
             tx.try_send(Change::Insert(endpoint.uri.clone(), endpoint))
                 .unwrap();
@@ -166,9 +166,9 @@ impl Channel {
     where
         D: Discover<Service = Connection> + Unpin + Send + 'static,
         D::Error: Into<crate::Error>,
-        D::Key: Send + Clone,
+        D::Key: Hash + Send + Clone,
     {
-        let svc = Balance::from_entropy(discover);
+        let svc = Balance::new(discover);
 
         let svc = BoxService::new(svc);
         let svc = Buffer::new(Either::B(svc), buffer_size);

tonic/src/transport/server/incoming.rs

@@ -105,8 +105,8 @@ where
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<std::io::Result<usize>> {
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
         use std::future::Future;
 
         let pin = self.get_mut();

tonic/src/transport/server/mod.rs

@@ -163,7 +163,7 @@ impl Server {
     /// ```
     /// # use tonic::transport::Server;
     /// # use tower_service::Service;
-    /// # let mut builder = Server::builder();
+    /// # let builder = Server::builder();
     /// builder.concurrency_limit_per_connection(32);
     /// ```
     pub fn concurrency_limit_per_connection(self, limit: usize) -> Self {

tonic/src/transport/service/connection.rs

@@ -10,6 +10,7 @@ use std::{
     task::{Context, Poll},
 };
 use tokio::io::{AsyncRead, AsyncWrite};
+use tower::load::Load;
 use tower::{
     layer::Layer,
     limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer},
@@ -17,7 +18,6 @@ use tower::{
     util::BoxService,
     ServiceBuilder, ServiceExt,
 };
-use tower_load::Load;
 use tower_service::Service;
 
 pub(crate) type Request = http::Request<BoxBody>;

tonic/src/transport/service/connector.rs

@@ -4,7 +4,7 @@ use super::io::BoxedIo;
 use super::tls::TlsConnector;
 use http::Uri;
 use std::task::{Context, Poll};
-use tower_make::MakeConnection;
+use tower::make::MakeConnection;
 use tower_service::Service;
 
 #[cfg(not(feature = "tls"))]

tonic/src/transport/service/discover.rs

@@ -7,9 +7,10 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio::{stream::Stream, sync::mpsc::Receiver};
+use tokio::sync::mpsc::Receiver;
 
-use tower::discover::{Change, Discover};
+use tokio_stream::Stream;
+use tower::discover::Change;
 
 type DiscoverResult<K, S, E> = Result<Change<K, S>, E>;
 
@@ -23,17 +24,12 @@ impl<K: Hash + Eq + Clone> DynamicServiceStream<K> {
     }
 }
 
-impl<K: Hash + Eq + Clone> Discover for DynamicServiceStream<K> {
-    type Key = K;
-    type Service = Connection;
-    type Error = crate::Error;
+impl<K: Hash + Eq + Clone> Stream for DynamicServiceStream<K> {
+    type Item = DiscoverResult<K, Connection, crate::Error>;
 
-    fn poll_discover(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<DiscoverResult<Self::Key, Self::Service, Self::Error>> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let c = &mut self.changes;
-        match Pin::new(&mut *c).poll_next(cx) {
+        match Pin::new(&mut *c).poll_recv(cx) {
             Poll::Pending | Poll::Ready(None) => Poll::Pending,
             Poll::Ready(Some(change)) => match change {
                 Change::Insert(k, endpoint) => {
@@ -48,9 +44,9 @@ impl<K: Hash + Eq + Clone> Discover for DynamicServiceStream<K> {
                     let connector = service::connector(http);
                     let connection = Connection::lazy(connector, endpoint);
                     let change = Ok(Change::Insert(k, connection));
-                    Poll::Ready(change)
+                    Poll::Ready(Some(change))
                 }
-                Change::Remove(k) => Poll::Ready(Ok(Change::Remove(k))),
+                Change::Remove(k) => Poll::Ready(Some(Ok(Change::Remove(k)))),
             },
         }
     }

tonic/src/transport/service/io.rs

@@ -4,7 +4,7 @@ use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
 pub(in crate::transport) trait Io:
     AsyncRead + AsyncWrite + Send + 'static
@@ -33,8 +33,8 @@ impl AsyncRead for BoxedIo {
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_read(cx, buf)
     }
 }
@@ -83,8 +83,8 @@ impl AsyncRead for ServerIo {
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_read(cx, buf)
     }
 }

tonic/src/transport/service/layer.rs

@@ -1,8 +1,5 @@
-use tower::{
-    layer::{Layer, Stack},
-    util::Either,
-    ServiceBuilder,
-};
+use tower::layer::util::Stack;
+use tower::{layer::Layer, util::Either, ServiceBuilder};
 
 pub(crate) trait ServiceBuilderExt<L> {
     fn layer_fn<F: Fn(S) -> Out, S, Out>(self, f: F) -> ServiceBuilder<Stack<LayerFn<F>, L>>;

tonic/src/transport/service/reconnect.rs

@@ -6,7 +6,7 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use tower_make::MakeService;
+use tower::make::MakeService;
 use tower_service::Service;
 use tracing::trace;