log buffer overflows, add non-linux IO

blocking_client/Cargo.toml

@@ -6,7 +6,6 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-
 solana-sdk = { workspace = true }
 anyhow = { workspace = true }
 log = { workspace = true }
@@ -19,6 +18,10 @@ quiche = { workspace = true }
 mio = { workspace = true }
 mio_channel = { workspace = true }
 
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"
+nix = { version = "0.27", features = ["net", "socket", "uio"] }
+
 [dev-dependencies]
 rand = { workspace = true }
 tracing-subscriber = { workspace = true }

blocking_client/src/client.rs

@@ -26,8 +26,6 @@ impl Client {
             connection_parameters.ack_exponent,
         )?;
         let server_address: SocketAddr = parse_host_port(&server_address)?;
-        let socket_addr: SocketAddr =
-            parse_host_port("[::]:0").expect("Socket address should be returned");
         let is_connected = Arc::new(AtomicBool::new(false));
         let (filters_sender, rx_sent_queue) = mio_channel::channel();
         let (sx_recv_queue, client_rx_queue) = std::sync::mpsc::channel();
@@ -36,7 +34,6 @@ impl Client {
         let _client_loop_jh = std::thread::spawn(move || {
             if let Err(e) = client_loop(
                 config,
-                socket_addr,
                 server_address,
                 rx_sent_queue,
                 sx_recv_queue,

blocking_client/src/lib.rs

@@ -1,3 +1,4 @@
 pub mod client;
 pub mod configure_client;
 pub mod quiche_client_loop;
+mod sendto;

blocking_client/src/quiche_client_loop.rs

@@ -13,18 +13,21 @@ use quic_geyser_quiche_utils::{
     quiche_utils::{generate_cid_and_reset_token, get_next_unidi, StreamBufferMap},
 };
 
-use anyhow::bail;
+use crate::sendto::{detect_gso, send_to};
+use anyhow::{bail, Context};
 use ring::rand::{SecureRandom, SystemRandom};
 
+const ENABLE_PACING: bool = true;
+
 pub fn client_loop(
     mut config: quiche::Config,
-    socket_addr: SocketAddr,
     server_address: SocketAddr,
     mut message_send_queue: mio_channel::Receiver<Message>,
     message_recv_queue: std::sync::mpsc::Sender<Message>,
     is_connected: Arc<AtomicBool>,
 ) -> anyhow::Result<()> {
-    let mut socket = mio::net::UdpSocket::bind(socket_addr)?;
+    let mut socket = mio::net::UdpSocket::bind("0.0.0.0:0".parse().unwrap())?;
+    let enable_gso = detect_gso(&socket, MAX_DATAGRAM_SIZE);
     let mut poll = mio::Poll::new()?;
     let mut events = mio::Events::with_capacity(1024);
     let mut buf = [0; 65535];
@@ -46,6 +49,7 @@ pub fn client_loop(
         bail!("Error filling scid");
     }
     log::info!("connecing client with quiche");
+    info!("quic client: enable gso? {}", enable_gso);
 
     let scid = quiche::ConnectionId::from_ref(&scid);
     let local_addr = socket.local_addr()?;
@@ -54,7 +58,7 @@ pub fn client_loop(
 
     // sending initial connection request
     {
-        let (write, send_info) = conn.send(&mut out).expect("initial send failed");
+        let (write, send_info) = conn.send(&mut out).context("initial send failed")?;
 
         if let Err(e) = socket.send_to(&out[..write], send_info.to) {
             bail!("send() failed: {:?}", e);
@@ -71,19 +75,22 @@ pub fn client_loop(
 
     let mut instance = Instant::now();
     loop {
-        poll.poll(&mut events, conn.timeout()).unwrap();
+        poll.poll(&mut events, conn.timeout()).context("poll")?;
 
         if conn.is_established() && !conn.is_closed() {
             match message_send_queue.try_recv() {
                 Ok(message) => {
                     let binary_message = message.to_binary_stream();
                     log::debug!("send message : {message:?}");
-                    let _ = send_message(
+                    let send_result = send_message(
                         &mut conn,
                         &mut stream_sender_map,
                         send_stream_id,
                         binary_message,
                     );
+                    if let Err(e) = send_result {
+                        log::error!("Error sending message : {e}");
+                    }
                 }
                 Err(e) => {
                     match e {
@@ -229,7 +236,16 @@ pub fn client_loop(
                 }
             };
 
-            if let Err(e) = socket.send_to(&out[..write], send_info.to) {
+            let send_result = send_to(
+                &socket,
+                &out[..write],
+                &send_info,
+                MAX_DATAGRAM_SIZE,
+                ENABLE_PACING,
+                enable_gso,
+            );
+
+            if let Err(e) = send_result {
                 if e.kind() == std::io::ErrorKind::WouldBlock {
                     debug!("send() would block");
                     break;
@@ -378,11 +394,9 @@ mod tests {
         let _client_loop_jh = std::thread::spawn(move || {
             let client_config =
                 configure_client(maximum_concurrent_streams, 20_000_000, 1, 25, 3).unwrap();
-            let socket_addr: SocketAddr = parse_host_port("[::]:0").unwrap();
             let is_connected = Arc::new(AtomicBool::new(false));
             if let Err(e) = client_loop(
                 client_config,
-                socket_addr,
                 server_addr,
                 rx_sent_queue,
                 sx_recv_queue,

blocking_client/src/sendto.rs

@@ -0,0 +1,156 @@
+// Copyright (C) 2021, Cloudflare, Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright notice,
+//       this list of conditions and the following disclaimer.
+//
+//     * Redistributions in binary form must reproduce the above copyright
+//       notice, this list of conditions and the following disclaimer in the
+//       documentation and/or other materials provided with the distribution.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+// IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
+// THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
+// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use std::cmp;
+
+use std::io;
+
+/// For Linux, try to detect GSO is available.
+#[cfg(target_os = "linux")]
+pub fn detect_gso(socket: &mio::net::UdpSocket, segment_size: usize) -> bool {
+    use nix::sys::socket::setsockopt;
+    use nix::sys::socket::sockopt::UdpGsoSegment;
+    use std::os::unix::io::AsRawFd;
+
+    // mio::net::UdpSocket doesn't implement AsFd (yet?).
+    let fd = unsafe { std::os::fd::BorrowedFd::borrow_raw(socket.as_raw_fd()) };
+
+    setsockopt(&fd, UdpGsoSegment, &(segment_size as i32)).is_ok()
+}
+
+/// For non-Linux, there is no GSO support.
+#[cfg(not(target_os = "linux"))]
+pub fn detect_gso(_socket: &mio::net::UdpSocket, _segment_size: usize) -> bool {
+    false
+}
+
+/// Send packets using sendmsg() with GSO.
+#[cfg(target_os = "linux")]
+fn send_to_gso_pacing(
+    socket: &mio::net::UdpSocket,
+    buf: &[u8],
+    send_info: &quiche::SendInfo,
+    segment_size: usize,
+) -> io::Result<usize> {
+    use nix::sys::socket::sendmsg;
+    use nix::sys::socket::ControlMessage;
+    use nix::sys::socket::MsgFlags;
+    use nix::sys::socket::SockaddrStorage;
+    use std::io::IoSlice;
+    use std::os::unix::io::AsRawFd;
+
+    let iov = [IoSlice::new(buf)];
+    let segment_size = segment_size as u16;
+    let dst = SockaddrStorage::from(send_info.to);
+    let sockfd = socket.as_raw_fd();
+
+    // GSO option.
+    let cmsg_gso = ControlMessage::UdpGsoSegments(&segment_size);
+
+    // Pacing option.
+    let send_time = std_time_to_u64(&send_info.at);
+    let cmsg_txtime = ControlMessage::TxTime(&send_time);
+
+    match sendmsg(
+        sockfd,
+        &iov,
+        &[cmsg_gso, cmsg_txtime],
+        MsgFlags::empty(),
+        Some(&dst),
+    ) {
+        Ok(v) => Ok(v),
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// For non-Linux platforms.
+#[cfg(not(target_os = "linux"))]
+fn send_to_gso_pacing(
+    _socket: &mio::net::UdpSocket,
+    _buf: &[u8],
+    _send_info: &quiche::SendInfo,
+    _segment_size: usize,
+) -> io::Result<usize> {
+    panic!("send_to_gso() should not be called on non-linux platforms");
+}
+
+/// A wrapper function of send_to().
+///
+/// When GSO and SO_TXTIME are enabled, send packets using send_to_gso().
+/// Otherwise, send packets using socket.send_to().
+pub fn send_to(
+    socket: &mio::net::UdpSocket,
+    buf: &[u8],
+    send_info: &quiche::SendInfo,
+    segment_size: usize,
+    pacing: bool,
+    enable_gso: bool,
+) -> io::Result<usize> {
+    if pacing && enable_gso {
+        match send_to_gso_pacing(socket, buf, send_info, segment_size) {
+            Ok(v) => {
+                return Ok(v);
+            }
+            Err(e) => {
+                return Err(e);
+            }
+        }
+    }
+
+    let mut off = 0;
+    let mut left = buf.len();
+    let mut written = 0;
+
+    while left > 0 {
+        let pkt_len = cmp::min(left, segment_size);
+
+        match socket.send_to(&buf[off..off + pkt_len], send_info.to) {
+            Ok(v) => {
+                written += v;
+            }
+            Err(e) => return Err(e),
+        }
+
+        off += pkt_len;
+        left -= pkt_len;
+    }
+
+    Ok(written)
+}
+
+#[cfg(target_os = "linux")]
+fn std_time_to_u64(time: &std::time::Instant) -> u64 {
+    const NANOS_PER_SEC: u64 = 1_000_000_000;
+
+    const INSTANT_ZERO: std::time::Instant = unsafe { std::mem::transmute(std::time::UNIX_EPOCH) };
+
+    let raw_time = time.duration_since(INSTANT_ZERO);
+
+    let sec = raw_time.as_secs();
+    let nsec = raw_time.subsec_nanos();
+
+    sec * NANOS_PER_SEC + nsec as u64
+}

common/src/stream_manager.rs

@@ -65,6 +65,7 @@ impl<const BUFFER_LEN: usize> StreamBuffer<BUFFER_LEN> {
 mod tests {
     use circular_buffer::CircularBuffer;
     use itertools::Itertools;
+    use log::warn;
     use rand::{rngs::ThreadRng, Rng};
     use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey};
 
@@ -147,6 +148,8 @@ mod tests {
                 let message = create_random_message(&mut rng);
                 if buffer.append_bytes(&message.to_binary_stream()) {
                     messages_appended.push(message);
+                } else {
+                    warn!("Buffer full");
                 }
             } else {
                 let buf = buffer.as_slices();

examples/tester-client/src/main.rs

@@ -118,7 +118,7 @@ impl<T: Ord + Display + Default + Copy> Stats<T> {
 }
 
 fn blocking(args: Args, client_stats: ClientStats, break_thread: Arc<AtomicBool>) {
-    println!("Connecting");
+    println!("Connecting (blocking=quiche)");
     let (client, reciever) =
         quic_geyser_blocking_client::client::Client::new(args.url, ConnectionParameters::default())
             .unwrap();
@@ -223,7 +223,7 @@ fn blocking(args: Args, client_stats: ClientStats, break_thread: Arc<AtomicBool>
 }
 
 async fn non_blocking(args: Args, client_stats: ClientStats, break_thread: Arc<AtomicBool>) {
-    println!("Connecting");
+    println!("Connecting (non-blocking=quinn)");
     let (client, mut reciever, _tasks) = Client::new(
         args.url,
         ConnectionParameters {

quiche/src/quiche_reciever.rs

@@ -18,7 +18,10 @@ pub fn recv_message(
             match connection.stream_recv(stream_id, &mut buf) {
                 Ok((read, _)) => {
                     log::trace!("read {} on stream {}", read, stream_id);
-                    total_buf.append_bytes(&buf[..read]);
+                    let buf_result = total_buf.append_bytes(&buf[..read]);
+                    if !buf_result {
+                        bail!("buffer too short");
+                    }
                 }
                 Err(e) => match &e {
                     quiche::Error::Done => {
@@ -47,7 +50,10 @@ pub fn recv_message(
             match connection.stream_recv(stream_id, &mut buf) {
                 Ok((read, _)) => {
                     log::trace!("read {} on stream {}", read, stream_id);
-                    total_buf.append_bytes(&buf[..read]);
+                    let buf_result = total_buf.append_bytes(&buf[..read]);
+                    if !buf_result {
+                        bail!("buffer too short");
+                    }
                 }
                 Err(e) => match &e {
                     quiche::Error::Done => {

server/Cargo.toml

@@ -21,12 +21,14 @@ boring = { workspace = true }
 mio = { workspace = true }
 mio_channel = { workspace = true }
 
-libc = "0.2"
-nix = { version = "0.27", features = ["net", "socket", "uio"] }
-
 quic-geyser-common = { workspace = true }
 prometheus = { workspace = true }
 lazy_static = { workspace = true }
+arrayvec = "0.7.6"
+
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"
+nix = { version = "0.27", features = ["net", "socket", "uio"] }
 
 [dev-dependencies]
 rand = { workspace = true }

server/src/lib.rs

@@ -1,3 +1,4 @@
 pub mod configure_server;
 pub mod quic_server;
 pub mod quiche_server_loop;
+pub mod sendto;