feat(io)!: zero-copy send for large writes on io_uring

h1/src/conn.rs

@@ -144,44 +144,20 @@ impl Outbox {
         self.bufs.len() >= MAX_WRITE_BUFFERS
     }
 
-    /// Writes everything queued.
+    /// Writes everything queued. The buffers go to the writer, which may
+    /// keep them until the kernel has sent them (a zero-copy send); a writer
+    /// that hands them back has their allocation reused for the next message.
     pub(crate) async fn flush<W: Write>(&mut self, writer: &mut W) -> Result<(), Error> {
-        let mut bufs = mem::take(&mut self.bufs);
-        while !bufs.is_empty() {
-            let BufResult(result, returned) = writer.write_vectored(bufs).await;
-            bufs = returned;
-            let mut written = match result {
-                Ok(0) => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::WriteZero,
-                        "stream accepted no bytes",
-                    )
-                    .into());
-                }
-                Ok(n) => n,
-                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
-                Err(error) => return Err(error.into()),
-            };
-            let done = bufs
-                .iter()
-                .take_while(|buf| {
-                    if written >= buf.len() {
-                        written -= buf.len();
-                        true
-                    } else {
-                        false
-                    }
-                })
-                .count();
-            bufs.drain(..done);
-            if written > 0 {
-                let first = &mut bufs[0];
-                *first = Bytes::slice(first, written..);
-            }
+        if self.bufs.is_empty() {
+            return Ok(());
+        }
+        let bufs = mem::take(&mut self.bufs);
+        let BufResult(result, returned) = writer.write_all_owned(bufs).await;
+        if let Some(mut bufs) = returned {
+            bufs.clear();
+            self.bufs = bufs;
         }
-        // Hand the allocation back for the next message.
-        self.bufs = bufs;
-        Ok(())
+        result.map_err(Error::from)
     }
 }
 

h1/src/lib.rs

@@ -72,19 +72,20 @@
 //!
 //! # fn main() -> std::io::Result<()> {
 //! compio::runtime::Runtime::new()?.block_on(async {
-//!     let listener = Compio.bind_tcp("127.0.0.1:0".parse().unwrap())?;
+//!     let io = Compio::default();
+//!     let listener = io.bind_tcp("127.0.0.1:0".parse().unwrap())?;
 //!     let addr = listener.local_addr()?;
 //!     let service = |request: Request| async move {
 //!         Response::new(Body::from_bytes(request.uri().path().to_owned()))
 //!     };
 //!     let server = async {
 //!         let (stream, _) = listener.accept().await?;
-//!         let connection = ServerConnection::from_stream(stream, Compio, ServerConfig::default());
+//!         let connection = ServerConnection::from_stream(stream, io.clone(), ServerConfig::default());
 //!         connection.serve(&service).await.map_err(std::io::Error::other)
 //!     };
 //!     let client = async {
-//!         let stream = Compio.connect(addr).await?;
-//!         let (send, connection) = handshake_stream(stream, Compio, ClientConfig::default());
+//!         let stream = io.connect(addr).await?;
+//!         let (send, connection) = handshake_stream(stream, io.clone(), ClientConfig::default());
 //!         let request = async move {
 //!             let mut request = Request::new(Body::empty());
 //!             *request.uri_mut() = format!("http://{addr}/hello").parse().unwrap();

h1/tests/net.rs

@@ -333,7 +333,7 @@ mod compio {
             .block_on(future)
     }
 
-    suite!(http_kit_io::compio::Compio, run);
+    suite!(http_kit_io::compio::Compio::default(), run);
 }
 
 mod tokio {

h1/tests/splice.rs

@@ -947,7 +947,7 @@ mod compio {
             .block_on(future)
     }
 
-    suite!(http_kit_io::compio::Compio, run);
+    suite!(http_kit_io::compio::Compio::default(), run);
 }
 
 mod tokio {

h1/tests/timeouts.rs

@@ -674,7 +674,7 @@ mod compio {
             .block_on(future)
     }
 
-    suite!(http_kit_io::compio::Compio, run);
+    suite!(http_kit_io::compio::Compio::default(), run);
 }
 
 mod tokio {

h1/tests/unix.rs

@@ -141,7 +141,7 @@ mod compio {
             .block_on(future)
     }
 
-    suite!(http_kit_io::compio::Compio, run);
+    suite!(http_kit_io::compio::Compio::default(), run);
 }
 
 mod tokio {

h3/src/lib.rs

@@ -63,8 +63,9 @@
 //! let identity = Identity::from_pem(std::fs::read("cert.pem")?, std::fs::read("key.pem")?);
 //! let config = ServerConfig::new(identity, 64 * 1024);
 //! compio::runtime::Runtime::new()?.block_on(async {
-//!     let socket = Compio.bind_udp("[::]:443".parse().unwrap()).await?;
-//!     let server = Server::new(socket, Compio, &config)?;
+//!     let io = Compio::default();
+//!     let socket = io.bind_udp("[::]:443".parse().unwrap()).await?;
+//!     let server = Server::new(socket, io, &config)?;
 //!     let service = |request: Request| async move {
 //!         Response::new(Body::from_bytes(request.uri().path().to_owned()))
 //!     };

h3/tests/curl.rs

@@ -54,7 +54,7 @@ fn answers_curl_http3() {
     on_compio(|io| async move {
         let socket = io.bind_udp(loopback()).await.unwrap();
         let port = socket.local_addr().unwrap().port();
-        let server = Server::new(socket, io, &server_config(&pki)).unwrap();
+        let server = Server::new(socket, io.clone(), &server_config(&pki)).unwrap();
         let (stop, stopped) = async_channel::bounded::<()>(1);
         let served = server.serve_until(&echo, async move {
             stopped.recv().await.ok();

h3/tests/support/mod.rs

@@ -163,7 +163,7 @@ pub async fn within<I: Timer, F: Future>(io: &I, limit: Duration, test: F) -> F:
 pub fn on_compio<F: Future>(test: impl FnOnce(http_kit_io::compio::Compio) -> F) -> F::Output {
     compio::runtime::Runtime::new()
         .unwrap()
-        .block_on(test(http_kit_io::compio::Compio))
+        .block_on(test(http_kit_io::compio::Compio::default()))
 }
 
 /// Runs a test body on a current-thread tokio runtime.