Upgrade to tokio 0.3

Author: mzohreva

Cargo.lock

@@ -28,7 +28,7 @@ byteorder = "1.0.0"
 yasna = { version = "0.2", optional = true }
 block-modes = { version = "0.3", optional = true }
 rc2 = { version = "0.3", optional = true }
-tokio = { version = "0.2", optional = true }
+tokio = { version = "0.3", optional = true }
 
 [target.x86_64-fortanix-unknown-sgx.dependencies]
 rs-libc = "0.1.0"

mbedtls/Cargo.toml

@@ -12,14 +12,14 @@ use std::cell::Cell;
 use std::future::Future;
 use std::io::{self, Read, Write};
 use std::marker::Unpin;
-use std::mem::{self, MaybeUninit};
+use std::mem;
 use std::ops::{Deref, DerefMut};
 use std::pin::Pin;
 use std::ptr::null_mut;
 use std::rc::Rc;
 use std::task::{Context as TaskContext, Poll};
 
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
 use crate::ssl::{Context, HandshakeError, MidHandshake, Session, Version};
 use crate::x509::VerifyError;
@@ -81,7 +81,9 @@ impl<S: Unpin> IoAdapter<S> {
 
 impl<S: AsyncRead + Unpin> Read for IoAdapter<S> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        self.with_context(|ctx, stream| stream.poll_read(ctx, buf))
+        let mut buf = ReadBuf::new(buf);
+        self.with_context(|ctx, stream| stream.poll_read(ctx, &mut buf))?;
+        Ok(buf.filled().len())
     }
 }
 
@@ -211,16 +213,16 @@ impl Drop for AsyncSession<'_> {
 }
 
 impl AsyncRead for AsyncSession<'_> {
-    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [MaybeUninit<u8>]) -> bool {
-        false
-    }
-
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut TaskContext<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
-        self.with_context(cx, |s| s.read(buf))
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        self.with_context(cx, |s| {
+            let n = s.read(buf.initialize_unfilled())?;
+            buf.advance(n);
+            Ok(())
+        })
     }
 }
 

mbedtls/src/ssl/async_session.rs

@@ -179,7 +179,6 @@ where
 #[cfg(unix)]
 mod test {
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::TcpStream;
 
     #[tokio::test]
     async fn client_server_test() {
@@ -222,8 +221,7 @@ mod test {
                 continue;
             }
 
-            let (c, s) = crate::support::net::create_tcp_pair().unwrap();
-            let (c, s) = (TcpStream::from_std(c).unwrap(), TcpStream::from_std(s).unwrap());
+            let (c, s) = crate::support::net::create_tcp_pair_async().unwrap();
             let c = tokio::spawn(super::client(c, min_c, max_c, exp_ver.clone()));
             let s = tokio::spawn(super::server(s, min_s, max_s, exp_ver));
 
@@ -234,8 +232,7 @@ mod test {
 
     #[tokio::test]
     async fn shutdown1() {
-        let (c, s) = crate::support::net::create_tcp_pair().unwrap();
-        let (c, s) = (TcpStream::from_std(c).unwrap(), TcpStream::from_std(s).unwrap());
+        let (c, s) = crate::support::net::create_tcp_pair_async().unwrap();
 
         let c = tokio::spawn(super::with_client(c, |session| Box::pin(async move {
             session.shutdown().await.unwrap();
@@ -255,8 +252,7 @@ mod test {
 
     #[tokio::test]
     async fn shutdown2() {
-        let (c, s) = crate::support::net::create_tcp_pair().unwrap();
-        let (c, s) = (TcpStream::from_std(c).unwrap(), TcpStream::from_std(s).unwrap());
+        let (c, s) = crate::support::net::create_tcp_pair_async().unwrap();
 
         let c = tokio::spawn(super::with_client(c, |session| Box::pin(async move {
             let mut buf = [0u8; 5];
@@ -279,8 +275,7 @@ mod test {
 
     #[tokio::test]
     async fn shutdown3() {
-        let (c, s) = crate::support::net::create_tcp_pair().unwrap();
-        let (c, s) = (TcpStream::from_std(c).unwrap(), TcpStream::from_std(s).unwrap());
+        let (c, s) = crate::support::net::create_tcp_pair_async().unwrap();
 
         let c = tokio::spawn(super::with_client(c, |session| Box::pin(async move {
             session.shutdown().await

mbedtls/tests/async_session.rs

@@ -12,6 +12,9 @@ use std::io::{Error as IoError, Result as IoResult};
 use std::net::TcpStream;
 use std::os::unix::io::FromRawFd;
 
+#[cfg(feature = "tokio")]
+use tokio::net;
+
 pub fn create_tcp_pair() -> IoResult<(TcpStream, TcpStream)> {
     let mut fds: [libc::c_int; 2] = [0; 2];
     unsafe {
@@ -28,3 +31,14 @@ pub fn create_tcp_pair() -> IoResult<(TcpStream, TcpStream)> {
         }
     }
 }
+
+#[cfg(feature = "tokio")]
+pub fn create_tcp_pair_async() -> IoResult<(net::TcpStream, net::TcpStream)> {
+    let (c, s) = create_tcp_pair()?;
+    c.set_nonblocking(true)?;
+    s.set_nonblocking(true)?;
+    Ok((
+        net::TcpStream::from_std(c)?,
+        net::TcpStream::from_std(s)?,
+    ))
+}