update to tokio 1.0

Author: Devdutt Shenoi

Cargo.lock

@@ -8,21 +8,16 @@ members = [
 ]
 
 [workspace.dependencies]
-anyhow = "1.0.31"
-env_logger = "0.7.1"
-futures = "0.3.5"
-log = "0.4.8"
-rand = "0.7.3"
-serde = "1.0.114"
-serde_json = "1.0.55"
-tokio = { version = "0.2.21", features = [
-    "blocking",
-    "time",
-    "io-util",
-    "macros",
-] }
-tokio-rustls = { version = "0.14.0", features = ["dangerous_configuration"] }
-tokio-util = { version = "0.3.1", features = ["compat"] }
-webpki = "0.21.3"
-webpki-roots = "0.20.0"
+anyhow = "1"
+env_logger = "0.11.5"
+futures = "0.3.30"
+log = "0.4.22"
+rand = "0.8.5"
+serde = "1"
+serde_json = "1"
+tokio = { version = "1.40.0", features = ["time", "io-util", "macros"] }
+tokio-rustls = { version = "0.24.1" }
+tokio-util = { version = "0.7.12", features = ["compat"] }
+webpki = "0.22.4"
+webpki-roots = "0.26.6"
 lazy_static = "1.4.0"

Cargo.toml

@@ -43,16 +43,14 @@ async-tungstenite = { version = "0.8.0", features = ["tokio-runtime"] } #no-wasm
 crossterm = { version = "0.23.2" }
 libc = "0.2.71"
 tokio = { workspace = true, features = [
-    "rt-threaded",
-    "dns",
+    "rt-multi-thread",
     "io-std",
-    "tcp",
-    "udp",
+    "io-util",
+    "net",
     "sync",
     "process",
     "signal",
     "fs",
-    "uds",
 ] } #no-wasm
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]

tunshell-client/Cargo.toml

@@ -7,10 +7,9 @@ use anyhow::{Error, Result};
 use futures::{future, stream::StreamExt, FutureExt};
 use log::*;
 use std::sync::{Arc, Mutex};
-use tokio_util::compat::*;
 use tunshell_shared::*;
 
-pub type ClientMessageStream = MessageStream<ClientMessage, ServerMessage, Compat<ServerStream>>;
+pub type ClientMessageStream = MessageStream<ClientMessage, ServerMessage, ServerStream>;
 
 pub struct Client {
     config: Config,
@@ -33,7 +32,7 @@ impl Client {
         self.println("Connecting to relay server...").await;
         let relay_socket = ServerStream::connect(&self.config).await?;
 
-        let mut message_stream = ClientMessageStream::new(relay_socket.compat());
+        let mut message_stream = ClientMessageStream::new(relay_socket);
 
         self.send_key(&mut message_stream).await?;
 
@@ -164,7 +163,7 @@ impl Client {
         assert!(peer_info.session_nonce.len() > 10);
 
         let stream = AesStream::new(
-            stream.compat(),
+            stream,
             peer_info.session_nonce.as_bytes(),
             self.config.encryption_key().as_bytes(),
         )

tunshell-client/src/client.rs

@@ -4,18 +4,19 @@ use anyhow::{Error, Result};
 use async_trait::async_trait;
 use futures::future::pending;
 use futures::TryFutureExt;
+use futures::{AsyncRead, AsyncWrite};
 use log::*;
 use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
 use std::pin::Pin;
-use std::{time::Duration, task::{Context, Poll}};
-use tokio::io::{AsyncRead, AsyncWrite};
+use std::task::{Context, Poll};
 use tokio::net::{TcpListener, TcpStream};
+use tokio_util::compat::{Compat, TokioAsyncReadCompatExt};
 use tunshell_shared::PeerJoinedPayload;
 
 pub struct TcpConnection {
     peer_info: PeerJoinedPayload,
     listener: Option<TcpListener>,
-    socket: Option<TcpStream>,
+    socket: Option<Compat<TcpStream>>,
 }
 
 impl AsyncRead for TcpConnection {
@@ -24,7 +25,9 @@ impl AsyncRead for TcpConnection {
         cx: &mut Context<'_>,
         buff: &mut [u8],
     ) -> Poll<std::result::Result<usize, std::io::Error>> {
-        Pin::new(&mut self.socket.as_mut().unwrap()).poll_read(cx, buff)
+        Pin::new(&mut self.socket.as_mut().unwrap())
+            .poll_read(cx, buff)
+            .map(|c| c.map(|_| 0))
     }
 }
 
@@ -44,11 +47,11 @@ impl AsyncWrite for TcpConnection {
         Pin::new(&mut self.socket.as_mut().unwrap()).poll_flush(cx)
     }
 
-    fn poll_shutdown(
+    fn poll_close(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<std::result::Result<(), std::io::Error>> {
-        Pin::new(&mut self.socket.as_mut().unwrap()).poll_shutdown(cx)
+        Pin::new(&mut self.socket.as_mut().unwrap()).poll_close(cx)
     }
 }
 
@@ -106,8 +109,8 @@ impl P2PConnection for TcpConnection {
             let connected_ip = self.peer_info.peer_ip_address.parse::<IpAddr>().unwrap();
 
             if peer_addr.ip() == connected_ip {
-                socket.set_keepalive(Some(Duration::from_secs(30)))?;
-                self.socket.replace(socket);
+                // socket.set_keepalive(Some(Duration::from_secs(30)))?;
+                self.socket.replace(socket.compat());
                 return Ok(());
             } else {
                 error!("received connection from unknown ip address: {}", peer_addr);
@@ -121,15 +124,15 @@ impl P2PConnection for TcpConnection {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use futures::{AsyncReadExt, AsyncWriteExt, FutureExt};
     use std::time::Duration;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::{runtime::Runtime, time::delay_for};
-    use futures::FutureExt;
+    use tokio::{runtime::Runtime, time::sleep};
+    use tokio_util::compat::TokioAsyncReadCompatExt;
 
     #[test]
     fn test_connect_via_connect() {
         Runtime::new().unwrap().block_on(async {
-            let mut listener = TcpListener::bind("0.0.0.0:22335".to_owned())
+            let listener = TcpListener::bind("0.0.0.0:22335".to_owned())
                 .await
                 .expect("failed listen for connection");
 
@@ -145,10 +148,11 @@ mod tests {
                 .await
                 .expect("failed to connect");
 
-            let (mut socket, _) = listener
+            let (socket, _) = listener
                 .accept()
                 .await
                 .expect("failed to accept connection");
+            let mut socket = socket.compat();
 
             connection1
                 .write("hello".as_bytes())
@@ -189,12 +193,13 @@ mod tests {
 
             let port = connection1.bind().await.expect("failed to bind");
 
-            let socket = delay_for(Duration::from_millis(100))
+            let socket = sleep(Duration::from_millis(100))
                 .then(|_| TcpStream::connect(format!("127.0.0.1:{}", port)))
                 .or_else(|err| futures::future::err(Error::new(err)));
 
-            let (_, mut socket) = futures::try_join!(connection1.connect(22444, false), socket)
+            let (_, socket) = futures::try_join!(connection1.connect(22444, false), socket)
                 .expect("failed to connect");
+            let mut socket = socket.compat();
 
             socket.write("hello".as_bytes()).await.unwrap();
 

tunshell-client/src/p2p/tcp.rs

@@ -1,4 +1,4 @@
-use super::{UdpConnectionVars, UdpPacket, MAX_PACKET_SIZE, SequenceNumber, UdpPacketType};
+use super::{SequenceNumber, UdpConnectionVars, UdpPacket, UdpPacketType, MAX_PACKET_SIZE};
 use log::*;
 use std::cmp;
 use std::sync::{Arc, Mutex};
@@ -18,7 +18,7 @@ impl UdpConnectionVars {
                 self.peer_ack_number,
                 self.peer_window
             );
-            
+
             self.window_wakers
                 .push((cx.waker().clone(), packet.end_sequence_number()));
             return Poll::Pending;
@@ -31,7 +31,7 @@ impl UdpConnectionVars {
         sequence_number < self.max_send_sequence_number()
     }
 
-    pub(super) fn max_send_sequence_number(&self ) -> SequenceNumber {
+    pub(super) fn max_send_sequence_number(&self) -> SequenceNumber {
         self.peer_ack_number + SequenceNumber(cmp::min(self.peer_window, self.transit_window))
     }
 
@@ -40,7 +40,8 @@ impl UdpConnectionVars {
     }
 
     pub(super) fn wait_until_decongested(&mut self, waker: Waker) {
-        self.window_wakers.push((waker, self.sequence_number + SequenceNumber(1)));
+        self.window_wakers
+            .push((waker, self.sequence_number + SequenceNumber(1)));
     }
 
     pub(super) fn increase_transit_window_after_send(&mut self) {
@@ -71,7 +72,8 @@ impl UdpConnectionVars {
         let max_sequence_number = self.max_send_sequence_number();
 
         // Determine the number of packets we can now send form those are waiting.
-        let (ready_wakers, pending_wakers) = self.window_wakers
+        let (ready_wakers, pending_wakers) = self
+            .window_wakers
             .drain(..)
             .partition(|(_, end_sequence_number)| end_sequence_number < &max_sequence_number);
 
@@ -218,10 +220,10 @@ mod tests {
             let con = Arc::from(Mutex::from(con));
 
             let packet = UdpPacket::data(SequenceNumber(0), SequenceNumber(0), 0, &[]);
-            
+
             let packet = tokio::select! {
                 packet = wait_until_can_send(Arc::clone(&con), packet) => packet,
-                _ = tokio::time::delay_for(Duration::from_millis(1)) => panic!("should return from future immediately if there is sufficient window")
+                _ = tokio::time::sleep(Duration::from_millis(1)) => panic!("should return from future immediately if there is sufficient window")
             };
 
             let con = con.lock().unwrap();
@@ -237,7 +239,7 @@ mod tests {
         if std::env::var("CI").is_ok() {
             return;
         }
-        
+
         Runtime::new().unwrap().block_on(async {
             let mut con = UdpConnectionVars::new(UdpConnectionConfig::default());
 
@@ -247,14 +249,14 @@ mod tests {
             let con = Arc::from(Mutex::from(con));
 
             let packet = UdpPacket::data(SequenceNumber(0), SequenceNumber(0), 0, &[]);
-            
+
             let wait_for_send = tokio::spawn(wait_until_can_send(Arc::clone(&con), packet.clone()));
 
-            tokio::time::delay_for(Duration::from_millis(100)).await;
+            tokio::time::sleep(Duration::from_millis(100)).await;
 
             {
                 let con = con.lock().unwrap();
-                
+
                 assert_eq!(con.window_wakers.len(), 1);
                 assert_eq!(con.window_wakers[0].1, packet.end_sequence_number());
             }
@@ -263,18 +265,18 @@ mod tests {
             {
                 let mut con = con.lock().unwrap();
                 con.update_peer_window(MAX_PACKET_SIZE as u32);
-                
+
                 assert_eq!(con.window_wakers.len(), 0);
             }
 
-            // Task should now complete      
+            // Task should now complete
             let packet = tokio::select! {
                 packet = wait_for_send => packet.unwrap(),
-                _ = tokio::time::delay_for(Duration::from_millis(1)) => panic!("should return from future immediately if there is sufficient window")
+                _ = tokio::time::sleep(Duration::from_millis(1)) => panic!("should return from future immediately if there is sufficient window")
             };
 
             let con = con.lock().unwrap();
-            
+
             assert_eq!(packet.sequence_number, SequenceNumber(0));
             assert_eq!(con.window_wakers.len(), 0);
         });

tunshell-client/src/p2p/udp/congestion.rs

@@ -3,14 +3,14 @@ use super::{
     UdpConnectionState, UdpConnectionVars,
 };
 use anyhow::{Error, Result};
+use futures::{AsyncRead, AsyncWrite};
 use log::*;
 use std::io;
 use std::mem;
 use std::net::IpAddr;
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio::net::UdpSocket;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 
@@ -189,15 +189,15 @@ impl UdpConnection {
         Ok(())
     }
 
-    /// Closes the connection
-    #[allow(dead_code)]
-    pub async fn close(&mut self) -> Result<()> {
-        if !self.is_connected() {
-            return Err(Error::msg("Connection must be in CONNECTED state"));
-        }
+    // /// Closes the connection
+    // #[allow(dead_code)]
+    // pub async fn close(&mut self) -> Result<()> {
+    //     if !self.is_connected() {
+    //         return Err(Error::msg("Connection must be in CONNECTED state"));
+    //     }
 
-        self.shutdown().await.map_err(Error::from)
-    }
+    //     self.close().await.map_err(Error::from)
+    // }
 }
 
 impl AsyncRead for UdpConnection {
@@ -281,7 +281,7 @@ impl AsyncWrite for UdpConnection {
         Poll::Ready(Ok(()))
     }
 
-    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         if let State::Disconnecting(running) = &self.state {
             if !self.is_connected() {
                 self.state = State::Disconnected;
@@ -321,12 +321,12 @@ impl AsyncWrite for UdpConnection {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use futures::{AsyncReadExt, AsyncWriteExt};
     use lazy_static::lazy_static;
     use std::net::SocketAddr;
     use std::time::Duration;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio::runtime::Runtime;
-    use tokio::time::delay_for;
+    use tokio::time::sleep;
 
     lazy_static! {
         static ref UDP_PORT_NUMBER: Mutex<u16> = Mutex::from(27660);
@@ -467,7 +467,7 @@ mod tests {
             assert_eq!(con2.is_disconnected(), true);
 
             // Wait for close packet to be sent and process
-            delay_for(Duration::from_millis(500)).await;
+            sleep(Duration::from_millis(500)).await;
 
             assert_eq!(con1.is_new(), false);
             assert_eq!(con1.is_connected(), false);