refactor(core): Replace Tokio I/O abstraction (#909)

Author: 0x676e67

Cargo.toml

@@ -55,7 +55,7 @@ stream = ["tokio/fs", "dep:tokio-util", "dep:sync_wrapper"]
 socks = ["dep:tokio-socks"]
 
 # Enable WebSocket support.
-ws = ["dep:async-tungstenite", "futures-util/std", "futures-util/io"]
+ws = ["dep:tokio-tungstenite"]
 
 # Enable webpki-roots for TLS certificate validation.
 webpki-roots = ["dep:webpki-root-certs"]
@@ -92,20 +92,17 @@ socket2 = { version = "0.6.0", features = ["all"] }
 ipnet = "2.11.0"
 schnellru = { version = "0.2.4", default-features = false }
 ahash = { version = "0.8.12", default-features = false }
-
-## runtime
+boring2 = { version = "5.0.0-alpha.4", features = ["pq-experimental"] }
+tokio-boring2 = { version = "5.0.0-alpha.4", features = ["pq-experimental"] }
+brotli = "8.0.1"
+flate2 = "1.1.2"
+zstd = "0.13.3"
 tokio = { version = "1.47.1", default-features = false, features = [
     "net",
     "time",
     "rt",
 ] }
 
-## tls
-boring2 = { version = "5.0.0-alpha.4", features = ["pq-experimental"] }
-tokio-boring2 = { version = "5.0.0-alpha.4", features = ["pq-experimental"] }
-brotli = "8.0.1"
-flate2 = "1.1.2"
-zstd = "0.13.3"
 
 # Optional deps...
 
@@ -139,7 +136,9 @@ tokio-util = { version = "0.7.16", default-features = false, optional = true }
 tokio-socks = { version = "0.5.2", optional = true }
 
 ## websocket
-async-tungstenite = { version = "0.31.0", optional = true }
+tokio-tungstenite = { version = "0.27.0", default-features = false, features = [
+    "handshake",
+], optional = true }
 
 ## hickory-dns
 hickory-resolver = { version = "0.25.2", optional = true }
@@ -162,12 +161,12 @@ system-configuration = { version = "0.6.1", optional = true }
 libc = "0.2.173"
 
 [dev-dependencies]
-hyper = { version = "1.1.0", default-features = false, features = [
+hyper = { version = "1.7.0", default-features = false, features = [
     "http1",
     "http2",
     "server",
 ] }
-hyper-util = { version = "0.1.13", features = [
+hyper-util = { version = "0.1.16", features = [
     "http1",
     "http2",
     "server-auto",

src/client/http/connect/conn.rs

@@ -8,17 +8,14 @@ use pin_project_lite::pin_project;
 #[cfg(unix)]
 use tokio::net::UnixStream;
 use tokio::{
-    io::{AsyncRead, AsyncWrite},
+    io::{AsyncRead, AsyncWrite, ReadBuf},
     net::TcpStream,
 };
 use tokio_boring2::SslStream;
 
 use super::{AsyncConnWithInfo, TlsInfoFactory};
 use crate::{
-    core::{
-        client::connect::{Connected, Connection},
-        rt::{Read, ReadBufCursor, TokioIo, Write},
-    },
+    core::client::connect::{Connected, Connection},
     tls::{MaybeHttpsStream, TlsInfo},
 };
 
@@ -43,7 +40,7 @@ pin_project! {
     /// It is mainly used internally to abstract over different connection types.
     pub struct TlsConn<T> {
         #[pin]
-        inner: TokioIo<SslStream<T>>,
+        inner: SslStream<T>,
     }
 }
 
@@ -65,48 +62,49 @@ impl Connection for Conn {
     }
 }
 
-impl Read for Conn {
+impl AsyncRead for Conn {
+    #[inline]
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
-        buf: ReadBufCursor<'_>,
+        buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        let this = self.project();
-        Read::poll_read(this.inner, cx, buf)
+        AsyncRead::poll_read(self.project().inner, cx, buf)
     }
 }
 
-impl Write for Conn {
+impl AsyncWrite for Conn {
+    #[inline]
     fn poll_write(
         self: Pin<&mut Self>,
         cx: &mut Context,
         buf: &[u8],
     ) -> Poll<Result<usize, io::Error>> {
-        let this = self.project();
-        Write::poll_write(this.inner, cx, buf)
+        AsyncWrite::poll_write(self.project().inner, cx, buf)
     }
 
+    #[inline]
     fn poll_write_vectored(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
     ) -> Poll<Result<usize, io::Error>> {
-        let this = self.project();
-        Write::poll_write_vectored(this.inner, cx, bufs)
+        AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
     }
 
+    #[inline]
     fn is_write_vectored(&self) -> bool {
         self.inner.is_write_vectored()
     }
 
+    #[inline]
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
-        let this = self.project();
-        Write::poll_flush(this.inner, cx)
+        AsyncWrite::poll_flush(self.project().inner, cx)
     }
 
+    #[inline]
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
-        let this = self.project();
-        Write::poll_shutdown(this.inner, cx)
+        AsyncWrite::poll_shutdown(self.project().inner, cx)
     }
 }
 
@@ -119,29 +117,27 @@ where
     /// Creates a new `TlsConn` wrapping the provided `SslStream`.
     #[inline(always)]
     pub fn new(inner: SslStream<T>) -> Self {
-        Self {
-            inner: TokioIo::new(inner),
-        }
+        Self { inner }
     }
 }
 
 // ===== impl TcpStream =====
 
 impl Connection for TlsConn<TcpStream> {
     fn connected(&self) -> Connected {
-        let connected = self.inner.inner().get_ref().connected();
-        if self.inner.inner().ssl().selected_alpn_protocol() == Some(b"h2") {
+        let connected = self.inner.get_ref().connected();
+        if self.inner.ssl().selected_alpn_protocol() == Some(b"h2") {
             connected.negotiated_h2()
         } else {
             connected
         }
     }
 }
 
-impl Connection for TlsConn<TokioIo<MaybeHttpsStream<TcpStream>>> {
+impl Connection for TlsConn<MaybeHttpsStream<TcpStream>> {
     fn connected(&self) -> Connected {
-        let connected = self.inner.inner().get_ref().connected();
-        if self.inner.inner().ssl().selected_alpn_protocol() == Some(b"h2") {
+        let connected = self.inner.get_ref().connected();
+        if self.inner.ssl().selected_alpn_protocol() == Some(b"h2") {
             connected.negotiated_h2()
         } else {
             connected
@@ -154,8 +150,8 @@ impl Connection for TlsConn<TokioIo<MaybeHttpsStream<TcpStream>>> {
 #[cfg(unix)]
 impl Connection for TlsConn<UnixStream> {
     fn connected(&self) -> Connected {
-        let connected = self.inner.inner().get_ref().connected();
-        if self.inner.inner().ssl().selected_alpn_protocol() == Some(b"h2") {
+        let connected = self.inner.get_ref().connected();
+        if self.inner.ssl().selected_alpn_protocol() == Some(b"h2") {
             connected.negotiated_h2()
         } else {
             connected
@@ -164,65 +160,66 @@ impl Connection for TlsConn<UnixStream> {
 }
 
 #[cfg(unix)]
-impl Connection for TlsConn<TokioIo<MaybeHttpsStream<UnixStream>>> {
+impl Connection for TlsConn<MaybeHttpsStream<UnixStream>> {
     fn connected(&self) -> Connected {
-        let connected = self.inner.inner().get_ref().connected();
-        if self.inner.inner().ssl().selected_alpn_protocol() == Some(b"h2") {
+        let connected = self.inner.get_ref().connected();
+        if self.inner.ssl().selected_alpn_protocol() == Some(b"h2") {
             connected.negotiated_h2()
         } else {
             connected
         }
     }
 }
 
-impl<T: AsyncRead + AsyncWrite + Unpin> Read for TlsConn<T> {
+impl<T: AsyncRead + AsyncWrite + Unpin> AsyncRead for TlsConn<T> {
+    #[inline]
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
-        buf: ReadBufCursor<'_>,
+        buf: &mut ReadBuf<'_>,
     ) -> Poll<tokio::io::Result<()>> {
-        let this = self.project();
-        Read::poll_read(this.inner, cx, buf)
+        AsyncRead::poll_read(self.project().inner, cx, buf)
     }
 }
 
-impl<T: AsyncRead + AsyncWrite + Unpin> Write for TlsConn<T> {
+impl<T: AsyncRead + AsyncWrite + Unpin> AsyncWrite for TlsConn<T> {
+    #[inline]
     fn poll_write(
         self: Pin<&mut Self>,
         cx: &mut Context,
         buf: &[u8],
     ) -> Poll<Result<usize, tokio::io::Error>> {
-        let this = self.project();
-        Write::poll_write(this.inner, cx, buf)
+        AsyncWrite::poll_write(self.project().inner, cx, buf)
     }
 
+    #[inline]
     fn poll_write_vectored(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
     ) -> Poll<Result<usize, io::Error>> {
-        let this = self.project();
-        Write::poll_write_vectored(this.inner, cx, bufs)
+        AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
     }
 
+    #[inline]
     fn is_write_vectored(&self) -> bool {
         self.inner.is_write_vectored()
     }
 
+    #[inline]
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), tokio::io::Error>> {
-        let this = self.project();
-        Write::poll_flush(this.inner, cx)
+        AsyncWrite::poll_flush(self.project().inner, cx)
     }
 
+    #[inline]
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), tokio::io::Error>> {
-        let this = self.project();
-        Write::poll_shutdown(this.inner, cx)
+        AsyncWrite::poll_shutdown(self.project().inner, cx)
     }
 }
 
 impl<T> TlsInfoFactory for TlsConn<T>
 where
-    TokioIo<SslStream<T>>: TlsInfoFactory,
+    SslStream<T>: TlsInfoFactory,
 {
     fn tls_info(&self) -> Option<TlsInfo> {
         self.inner.tls_info()

src/client/http/connect/connector.rs

@@ -9,6 +9,7 @@ use std::{
 };
 
 use http::{Uri, uri::Scheme};
+use tokio::io::{AsyncRead, AsyncWrite};
 use tower::{
     Service, ServiceBuilder, ServiceExt,
     timeout::TimeoutLayer,
@@ -24,12 +25,9 @@ use super::{
 #[cfg(unix)]
 use crate::core::client::connect::UnixConnector;
 use crate::{
-    core::{
-        client::{
-            ConnectExtra, ConnectRequest,
-            connect::{Connection, proxy},
-        },
-        rt::{Read, TokioIo, Write},
+    core::client::{
+        ConnectExtra, ConnectRequest,
+        connect::{Connection, proxy},
     },
     dns::DynResolver,
     error::{BoxError, TimedOut, map_timeout_to_connector_error},
@@ -411,8 +409,6 @@ impl ConnectorService {
                     // The tunnel connector will first establish a CONNECT tunnel,
                     // then perform the TLS handshake over the tunneled stream.
                     let tunneled = tunnel.call(uri).await?;
-                    let tunneled = TokioIo::new(tunneled);
-                    let tunneled = TokioIo::new(tunneled);
 
                     // Wrap the established tunneled stream with TLS.
                     let established_conn = EstablishedConn::new(req, tunneled);
@@ -448,8 +444,6 @@ impl ConnectorService {
                     // The tunnel connector will first establish a CONNECT tunnel,
                     // then perform the TLS handshake over the tunneled stream.
                     let tunneled = tunnel.call(uri).await?;
-                    let tunneled = TokioIo::new(tunneled);
-                    let tunneled = TokioIo::new(tunneled);
 
                     // Wrap the established tunneled stream with TLS.
                     let established_conn = EstablishedConn::new(req, tunneled);
@@ -521,7 +515,7 @@ impl ConnectorService {
         S: Service<Uri, Response = T> + Send,
         S::Error: Into<BoxError>,
         S::Future: Unpin + Send + 'static,
-        T: Read + Write + Connection + Unpin + std::fmt::Debug + Sync + Send + 'static,
+        T: AsyncRead + AsyncWrite + Connection + Unpin + std::fmt::Debug + Sync + Send + 'static,
     {
         // Prefer TLS options from metadata, fallback to default
         let tls = extra

src/client/http/connect/mod.rs

@@ -3,11 +3,10 @@ mod connector;
 mod tls_info;
 mod verbose;
 
+use tokio::io::{AsyncRead, AsyncWrite};
+
 pub(super) use self::{conn::Conn, connector::Connector, tls_info::TlsInfoFactory};
-use crate::core::{
-    client::{ConnectRequest, connect::Connection},
-    rt::{Read, Write},
-};
+use crate::core::client::{ConnectRequest, connect::Connection};
 
 /// A wrapper type for [`ConnectRequest`] used to erase its concrete type.
 ///
@@ -22,14 +21,14 @@ pub struct Unnameable(pub(super) ConnectRequest);
 /// - [`Read`] + [`Write`]: For I/O operations
 /// - [`Connection`]: For connection metadata
 /// - [`Send`] + [`Sync`] + [`Unpin`] + `'static`: For async/await compatibility
-trait AsyncConn: Read + Write + Connection + Send + Sync + Unpin + 'static {}
+trait AsyncConn: AsyncRead + AsyncWrite + Connection + Send + Sync + Unpin + 'static {}
 
 /// An async connection that can also provide TLS information.
 ///
 /// This extends [`AsyncConn`] with the ability to extract TLS certificate information
 /// when available. Useful for connections that may be either plain TCP or TLS-encrypted.
 trait AsyncConnWithInfo: AsyncConn + TlsInfoFactory {}
 
-impl<T> AsyncConn for T where T: Read + Write + Connection + Send + Sync + Unpin + 'static {}
+impl<T> AsyncConn for T where T: AsyncRead + AsyncWrite + Connection + Send + Sync + Unpin + 'static {}
 
 impl<T> AsyncConnWithInfo for T where T: AsyncConn + TlsInfoFactory {}