Bump tokio from 0.2.11 to 0.3

Cargo.toml

@@ -28,12 +28,12 @@ http = "0.2"
 http-body = "0.3.1"
 httpdate = "0.3"
 httparse = "1.0"
-h2 = "0.2.2"
+h2 = { git = "https://github.com/hyperium/h2" }
 itoa = "0.4.1"
 tracing = { version = "0.1", default-features = false, features = ["log", "std"] }
 pin-project = "1.0"
 tower-service = "0.3"
-tokio = { version = "0.2.11", features = ["sync"] }
+tokio = { version = "0.3", features = ["sync"] }
 want = "0.3"
 
 # Optional
@@ -49,9 +49,9 @@ spmc = "0.3"
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
-tokio = { version = "0.2.2", features = ["fs", "macros", "io-std", "rt-util", "sync", "time", "test-util"] }
-tokio-test = "0.2"
-tokio-util = { version = "0.3", features = ["codec"] }
+tokio = { version = "0.3", features = ["fs", "macros", "io-std", "rt", "sync", "time", "test-util"] }
+tokio-test = "0.3"
+tokio-util = { version = "0.4", features = ["codec"] }
 tower-util = "0.3"
 url = "1.0"
 
@@ -65,12 +65,12 @@ default = [
 ]
 runtime = [
     "tcp",
-    "tokio/rt-core",
+    "tokio/rt",
 ]
 tcp = [
     "socket2",
-    "tokio/blocking",
-    "tokio/tcp",
+    "tokio/net",
+    "tokio/rt",
     "tokio/time",
 ]
 

src/client/connect/http.rs

@@ -13,7 +13,7 @@ use futures_util::future::Either;
 use http::uri::{Scheme, Uri};
 use pin_project::pin_project;
 use tokio::net::TcpStream;
-use tokio::time::Delay;
+use tokio::time::Sleep;
 
 use super::dns::{self, resolve, GaiResolver, Resolve};
 use super::{Connected, Connection};
@@ -510,7 +510,7 @@ impl ConnectingTcp {
                 local_addr_ipv6,
                 preferred: ConnectingTcpRemote::new(preferred_addrs, connect_timeout),
                 fallback: Some(ConnectingTcpFallback {
-                    delay: tokio::time::delay_for(fallback_timeout),
+                    delay: tokio::time::sleep(fallback_timeout),
                     remote: ConnectingTcpRemote::new(fallback_addrs, connect_timeout),
                 }),
                 reuse_address,
@@ -528,7 +528,7 @@ impl ConnectingTcp {
 }
 
 struct ConnectingTcpFallback {
-    delay: Delay,
+    delay: Sleep,
     remote: ConnectingTcpRemote,
 }
 
@@ -640,17 +640,7 @@ fn connect(
 
     let std_tcp = socket.into_tcp_stream();
 
-    Ok(async move {
-        let connect = TcpStream::connect_std(std_tcp, &addr);
-        match connect_timeout {
-            Some(dur) => match tokio::time::timeout(dur, connect).await {
-                Ok(Ok(s)) => Ok(s),
-                Ok(Err(e)) => Err(e),
-                Err(e) => Err(io::Error::new(io::ErrorKind::TimedOut, e)),
-            },
-            None => connect.await,
-        }
-    })
+    Ok(async move { TcpStream::from_std(std_tcp) })
 }
 
 impl ConnectingTcp {

src/client/pool.rs

@@ -850,7 +850,7 @@ mod tests {
         let pooled = pool.pooled(c(key.clone()), Uniq(41));
 
         drop(pooled);
-        tokio::time::delay_for(pool.locked().timeout.unwrap()).await;
+        tokio::time::sleep(pool.locked().timeout.unwrap()).await;
         let mut checkout = pool.checkout(key);
         let poll_once = PollOnce(&mut checkout);
         let is_not_ready = poll_once.await.is_none();
@@ -871,7 +871,7 @@ mod tests {
             pool.locked().idle.get(&key).map(|entries| entries.len()),
             Some(3)
         );
-        tokio::time::delay_for(pool.locked().timeout.unwrap()).await;
+        tokio::time::sleep(pool.locked().timeout.unwrap()).await;
 
         let mut checkout = pool.checkout(key.clone());
         let poll_once = PollOnce(&mut checkout);

src/common/io/rewind.rs

@@ -2,7 +2,7 @@ use std::marker::Unpin;
 use std::{cmp, io};
 
 use bytes::{Buf, Bytes};
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
 use crate::common::{task, Pin, Poll};
 
@@ -46,27 +46,22 @@ impl<T> AsyncRead for Rewind<T>
 where
     T: AsyncRead + Unpin,
 {
-    #[inline]
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
-        self.inner.prepare_uninitialized_buffer(buf)
-    }
-
     fn poll_read(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         if let Some(mut prefix) = self.pre.take() {
             // If there are no remaining bytes, let the bytes get dropped.
             if !prefix.is_empty() {
-                let copy_len = cmp::min(prefix.len(), buf.len());
-                prefix.copy_to_slice(&mut buf[..copy_len]);
+                let copy_len = cmp::min(prefix.len(), buf.remaining());
+                // TODO: There should be a way to do following two lines cleaner...
+                buf.put_slice(prefix.to_vec().as_slice());
+                prefix.advance(copy_len);
                 // Put back whats left
                 if !prefix.is_empty() {
                     self.pre = Some(prefix);
                 }
-
-                return Poll::Ready(Ok(copy_len));
             }
         }
         Pin::new(&mut self.inner).poll_read(cx, buf)
@@ -92,15 +87,6 @@ where
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
         Pin::new(&mut self.inner).poll_shutdown(cx)
     }
-
-    #[inline]
-    fn poll_write_buf<B: Buf>(
-        mut self: Pin<&mut Self>,
-        cx: &mut task::Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>> {
-        Pin::new(&mut self.inner).poll_write_buf(cx, buf)
-    }
 }
 
 #[cfg(test)]

src/proto/h2/ping.rs

@@ -33,7 +33,7 @@ use std::time::Instant;
 
 use h2::{Ping, PingPong};
 #[cfg(feature = "runtime")]
-use tokio::time::{Delay, Instant};
+use tokio::time::{Instant, Sleep};
 
 type WindowSize = u32;
 
@@ -60,7 +60,7 @@ pub(super) fn channel(ping_pong: PingPong, config: Config) -> (Recorder, Ponger)
         interval,
         timeout: config.keep_alive_timeout,
         while_idle: config.keep_alive_while_idle,
-        timer: tokio::time::delay_for(interval),
+        timer: tokio::time::sleep(interval),
         state: KeepAliveState::Init,
     });
 
@@ -156,7 +156,7 @@ struct KeepAlive {
     while_idle: bool,
 
     state: KeepAliveState,
-    timer: Delay,
+    timer: Sleep,
 }
 
 #[cfg(feature = "runtime")]

src/server/tcp.rs

@@ -4,7 +4,7 @@ use std::net::{SocketAddr, TcpListener as StdTcpListener};
 use std::time::Duration;
 
 use tokio::net::TcpListener;
-use tokio::time::Delay;
+use tokio::time::Sleep;
 
 use crate::common::{task, Future, Pin, Poll};
 
@@ -19,7 +19,7 @@ pub struct AddrIncoming {
     sleep_on_errors: bool,
     tcp_keepalive_timeout: Option<Duration>,
     tcp_nodelay: bool,
-    timeout: Option<Delay>,
+    timeout: Option<Sleep>,
 }
 
 impl AddrIncoming {
@@ -119,7 +119,7 @@ impl AddrIncoming {
                         error!("accept error: {}", e);
 
                         // Sleep 1s.
-                        let mut timeout = tokio::time::delay_for(Duration::from_secs(1));
+                        let mut timeout = tokio::time::sleep(Duration::from_secs(1));
 
                         match Pin::new(&mut timeout).poll(cx) {
                             Poll::Ready(()) => {
@@ -186,7 +186,7 @@ mod addr_stream {
     use std::net::SocketAddr;
     #[cfg(unix)]
     use std::os::unix::io::{AsRawFd, RawFd};
-    use tokio::io::{AsyncRead, AsyncWrite};
+    use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
     use tokio::net::TcpStream;
 
     use crate::common::{task, Pin, Poll};
@@ -231,30 +231,14 @@ mod addr_stream {
     }
 
     impl AsyncRead for AddrStream {
-        unsafe fn prepare_uninitialized_buffer(
-            &self,
-            buf: &mut [std::mem::MaybeUninit<u8>],
-        ) -> bool {
-            self.inner.prepare_uninitialized_buffer(buf)
-        }
-
         #[inline]
         fn poll_read(
-            mut self: Pin<&mut Self>,
+            self: Pin<&mut Self>,
             cx: &mut task::Context<'_>,
-            buf: &mut [u8],
-        ) -> Poll<io::Result<usize>> {
+            buf: &mut ReadBuf<'_>,
+        ) -> Poll<io::Result<()>> {
             Pin::new(&mut self.inner).poll_read(cx, buf)
         }
-
-        #[inline]
-        fn poll_read_buf<B: BufMut>(
-            mut self: Pin<&mut Self>,
-            cx: &mut task::Context<'_>,
-            buf: &mut B,
-        ) -> Poll<io::Result<usize>> {
-            Pin::new(&mut self.inner).poll_read_buf(cx, buf)
-        }
     }
 
     impl AsyncWrite for AddrStream {
@@ -267,15 +251,6 @@ mod addr_stream {
             Pin::new(&mut self.inner).poll_write(cx, buf)
         }
 
-        #[inline]
-        fn poll_write_buf<B: Buf>(
-            mut self: Pin<&mut Self>,
-            cx: &mut task::Context<'_>,
-            buf: &mut B,
-        ) -> Poll<io::Result<usize>> {
-            Pin::new(&mut self.inner).poll_write_buf(cx, buf)
-        }
-
         #[inline]
         fn poll_flush(self: Pin<&mut Self>, _cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
             // TCP flush is a noop

src/upgrade.rs

@@ -12,7 +12,7 @@ use std::io;
 use std::marker::Unpin;
 
 use bytes::{Buf, Bytes};
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::oneshot;
 
 use crate::common::io::Rewind;
@@ -105,15 +105,11 @@ impl Upgraded {
 }
 
 impl AsyncRead for Upgraded {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
-        self.io.prepare_uninitialized_buffer(buf)
-    }
-
     fn poll_read(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.io).poll_read(cx, buf)
     }
 }
@@ -127,14 +123,6 @@ impl AsyncWrite for Upgraded {
         Pin::new(&mut self.io).poll_write(cx, buf)
     }
 
-    fn poll_write_buf<B: Buf>(
-        mut self: Pin<&mut Self>,
-        cx: &mut task::Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>> {
-        Pin::new(self.io.get_mut()).poll_write_dyn_buf(cx, buf)
-    }
-
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
         Pin::new(&mut self.io).poll_flush(cx)
     }
@@ -247,15 +235,11 @@ impl dyn Io + Send {
 }
 
 impl<T: AsyncRead + Unpin> AsyncRead for ForwardsWriteBuf<T> {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [std::mem::MaybeUninit<u8>]) -> bool {
-        self.0.prepare_uninitialized_buffer(buf)
-    }
-
     fn poll_read(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_read(cx, buf)
     }
 }
@@ -269,14 +253,6 @@ impl<T: AsyncWrite + Unpin> AsyncWrite for ForwardsWriteBuf<T> {
         Pin::new(&mut self.0).poll_write(cx, buf)
     }
 
-    fn poll_write_buf<B: Buf>(
-        mut self: Pin<&mut Self>,
-        cx: &mut task::Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>> {
-        Pin::new(&mut self.0).poll_write_buf(cx, buf)
-    }
-
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_flush(cx)
     }
@@ -292,7 +268,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static> Io for ForwardsWriteBuf<T> {
         cx: &mut task::Context<'_>,
         mut buf: &mut dyn Buf,
     ) -> Poll<io::Result<usize>> {
-        Pin::new(&mut self.0).poll_write_buf(cx, &mut buf)
+        Pin::new(&mut self.0).poll_write(cx, buf.bytes())
     }
 }
 
@@ -326,8 +302,8 @@ mod tests {
         fn poll_read(
             self: Pin<&mut Self>,
             _cx: &mut task::Context<'_>,
-            _buf: &mut [u8],
-        ) -> Poll<io::Result<usize>> {
+            _buf: &mut ReadBuf<'_>,
+        ) -> Poll<io::Result<()>> {
             unreachable!("Mock::poll_read")
         }
     }