handle noncontiguous send buffers (#14)

handle noncontiguous send buffers to support chunks_vectored with zero copy concatenated buffers

Author: PDXKimani

.github/workflows/pr.yaml

@@ -2,7 +2,9 @@ name: PR
 
 on:
   pull_request:
-    branches: [ main ]
+    types:
+      - opened
+      - reopened
 
 env:
   CARGO_TERM_COLOR: always

protosocket-connection/src/connection.rs

@@ -378,18 +378,27 @@ impl<
                 // 16 is the lowest I've seen mention of, and I've seen 1024 more commonly.
                 const UIO_MAXIOV: usize = 256;
 
-                let buffers: Vec<IoSlice> = self
-                    .send_buffer
-                    .iter()
-                    .take(UIO_MAXIOV)
-                    .map(|v| IoSlice::new(v.chunk()))
-                    .collect();
+                // Use chunks_vectored rather than chunk() to correctly handle
+                // Buf implementations that aren't backed by a single contiguous slice.
+                let mut stack_buffers = [IoSlice::new(&[]); UIO_MAXIOV];
+                let mut filled = 0;
+                for buf in self.send_buffer.iter() {
+                    if UIO_MAXIOV <= filled {
+                        break;
+                    }
+                    let n = buf.chunks_vectored(&mut stack_buffers[filled..]);
+                    if n == 0 {
+                        break;
+                    }
+                    filled += n;
+                }
+                let buffers = &stack_buffers[..filled];
 
                 #[cfg(feature = "tracing")]
                 let span = tracing::span!(tracing::Level::INFO, "writing", buffers = buffers.len());
                 #[cfg(feature = "tracing")]
                 let span_guard = span.enter();
-                let poll = pin!(&mut self.stream).poll_write_vectored(context, &buffers);
+                let poll = pin!(&mut self.stream).poll_write_vectored(context, buffers);
                 #[cfg(feature = "tracing")]
                 drop(span_guard);
                 match poll {
@@ -423,7 +432,7 @@ impl<
                             "error while writing to tcp stream: {err:?}, buffers: {}, {}b: {:?}",
                             buffers.len(),
                             buffers.iter().map(|b| b.len()).sum::<usize>(),
-                            buffers.into_iter().map(|b| b.len()).collect::<Vec<_>>()
+                            buffers.iter().map(|b| b.len()).collect::<Vec<_>>()
                         );
                         Err(err)
                     }