Merge pull request ikatson#443 from ikatson/librqbit-utp-vectored-read

[perf] librqbit-utp vectored read + vectored write

Author: ikatson

Cargo.lock

@@ -123,7 +123,7 @@ walkdir = "2.5.0"
 arc-swap = "1.7.1"
 intervaltree = "0.2.7"
 async-compression = { version = "0.4.18", features = ["tokio", "gzip"] }
-librqbit-utp = { version = "0.5.5", features = ["export-metrics"] }
+librqbit-utp = { version = "0.5.7", features = ["export-metrics"] }
 axum-extra = { version = "0.10.1", features = ["query"] }
 librqbit-dualstack-sockets = { version = "0.6.5", features = ["axum"] }
 socket2 = "0.5.10"

crates/librqbit/Cargo.toml

@@ -10,10 +10,7 @@ use tokio::io::AsyncWrite;
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 
-use crate::{
-    stream_connect::ConnectionKind,
-    vectored_traits::{AsyncReadVectored, AsyncReadVectoredIntoCompat},
-};
+use crate::{stream_connect::ConnectionKind, vectored_traits::AsyncReadVectored};
 
 pub(crate) struct ListenResult {
     pub tcp_socket: Option<TcpListener>,
@@ -140,7 +137,7 @@ pub(crate) trait Accept {
         SocketAddr,
         (
             impl AsyncReadVectored + Send + 'static,
-            (impl AsyncWrite + Unpin + Send + 'static),
+            impl AsyncWrite + Unpin + Send + 'static,
         ),
     )>;
 }
@@ -176,6 +173,6 @@ impl Accept for Arc<UtpSocketUdp> {
         let stream = self.accept().await.context("error accepting uTP")?;
         let addr = stream.remote_addr();
         let (read, write) = stream.split();
-        Ok((addr, (read.into_vectored_compat(), write)))
+        Ok((addr, (read, write)))
     }
 }

crates/librqbit/src/listen.rs

@@ -255,7 +255,7 @@ impl StreamConnector {
                     match utp_res {
                         Ok(Some(stream)) => {
                             let (r, w) = stream.split();
-                            return Ok((ConnectionKind::Utp, Box::new(r.into_vectored_compat()), Box::new(w)));
+                            return Ok((ConnectionKind::Utp, Box::new(r), Box::new(w)));
                         },
                         Ok(None) => {
                             utp_err = Some(None);

crates/librqbit/src/stream_connect.rs

@@ -59,6 +59,16 @@ impl AsyncReadVectored for tokio::net::tcp::OwnedReadHalf {
     }
 }
 
+impl AsyncReadVectored for librqbit_utp::UtpStreamReadHalf {
+    fn poll_read_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        vec: &mut [IoSliceMut<'_>],
+    ) -> Poll<std::io::Result<usize>> {
+        librqbit_utp::UtpStreamReadHalf::poll_read_vectored(self, cx, vec)
+    }
+}
+
 pub struct AsyncReadToVectoredCompat<T>(T);
 
 impl<T: AsyncRead + Unpin> AsyncRead for AsyncReadToVectoredCompat<T> {
@@ -72,14 +82,6 @@ impl<T: AsyncRead + Unpin> AsyncRead for AsyncReadToVectoredCompat<T> {
 }
 
 impl<T: AsyncRead + Unpin> AsyncReadVectored for AsyncReadToVectoredCompat<T> {
-    // async fn read_vectored(&mut self, vec: &mut [IoSliceMut<'_>]) -> std::io::Result<usize> {
-    //     let first_non_empty = match vec.iter_mut().find(|s| !s.is_empty()) {
-    //         Some(s) => &mut **s,
-    //         None => return Ok(0),
-    //     };
-    //     self.read(first_non_empty).await
-    // }
-
     fn poll_read_vectored(
         self: Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,