fix(http3): always qlog on send_buffer() (#1876)

* fix(http3): always qlog on send_buffer()

`neqo_http3::SendMessage` calls `qlog::h3_data_moved_down()` whenever it moves
data down to the QUIC layer. `SendMessage` moves data down to the QUIC layer
either directly via `self.stream.send_atomic` or indirectly buffered through `self.stream.send_buffer`.

Previously only one of the 3 calls to `self.stream.send_buffer` would thereafter
call `qlog::h3_data_moved_down()`.

This commit moves the `h3_data_moved_down` call into `self.stream.send_buffer`, thus
ensuring the function is always called when data is moved. In addition,
`self.stream.send_atomic` now as well does the qlog call, thus containing all
qlog logic in `buffered_send_stream.rs` instead of `send_message.rs`.

* Trigger benchmark run

* Don't qlog if buffer is empty

* Fix typo

* Use early return to keep indentation short

* Don't qlog if sent is 0

neqo-http3/src/buffered_send_stream.rs

@@ -7,7 +7,7 @@
 use neqo_common::qtrace;
 use neqo_transport::{Connection, StreamId};
 
-use crate::Res;
+use crate::{qlog, Res};
 
 #[derive(Debug, PartialEq, Eq)]
 pub enum BufferedStream {
@@ -38,7 +38,7 @@ impl BufferedStream {
 
     /// # Panics
     ///
-    /// If the `BufferedStream` is initialized more than one it will panic.
+    /// If the `BufferedStream` is initialized more than once, it will panic.
     pub fn init(&mut self, stream_id: StreamId) {
         debug_assert!(&Self::Uninitialized == self);
         *self = Self::Initialized {
@@ -63,19 +63,23 @@ impl BufferedStream {
     /// Returns `neqo_transport` errors.
     pub fn send_buffer(&mut self, conn: &mut Connection) -> Res<usize> {
         let label = ::neqo_common::log_subject!(::log::Level::Debug, self);
-        let mut sent = 0;
-        if let Self::Initialized { stream_id, buf } = self {
-            if !buf.is_empty() {
-                qtrace!([label], "sending data.");
-                sent = conn.stream_send(*stream_id, &buf[..])?;
-                if sent == buf.len() {
-                    buf.clear();
-                } else {
-                    let b = buf.split_off(sent);
-                    *buf = b;
-                }
-            }
+        let Self::Initialized { stream_id, buf } = self else {
+            return Ok(0);
+        };
+        if buf.is_empty() {
+            return Ok(0);
+        }
+        qtrace!([label], "sending data.");
+        let sent = conn.stream_send(*stream_id, &buf[..])?;
+        if sent == 0 {
+            return Ok(0);
+        } else if sent == buf.len() {
+            buf.clear();
+        } else {
+            let b = buf.split_off(sent);
+            *buf = b;
         }
+        qlog::h3_data_moved_down(conn.qlog_mut(), *stream_id, sent);
         Ok(sent)
     }
 
@@ -85,16 +89,17 @@ impl BufferedStream {
     pub fn send_atomic(&mut self, conn: &mut Connection, to_send: &[u8]) -> Res<bool> {
         // First try to send anything that is in the buffer.
         self.send_buffer(conn)?;
-        if let Self::Initialized { stream_id, buf } = self {
-            if buf.is_empty() {
-                let res = conn.stream_send_atomic(*stream_id, to_send)?;
-                Ok(res)
-            } else {
-                Ok(false)
-            }
-        } else {
-            Ok(false)
+        let Self::Initialized { stream_id, buf } = self else {
+            return Ok(false);
+        };
+        if !buf.is_empty() {
+            return Ok(false);
+        }
+        let res = conn.stream_send_atomic(*stream_id, to_send)?;
+        if res {
+            qlog::h3_data_moved_down(conn.qlog_mut(), *stream_id, to_send.len());
         }
+        Ok(res)
     }
 
     #[must_use]

neqo-http3/src/send_message.rs

@@ -13,7 +13,7 @@ use neqo_transport::{Connection, StreamId};
 use crate::{
     frames::HFrame,
     headers_checks::{headers_valid, is_interim, trailers_valid},
-    qlog, BufferedStream, CloseType, Error, Http3StreamInfo, Http3StreamType, HttpSendStream, Res,
+    BufferedStream, CloseType, Error, Http3StreamInfo, Http3StreamType, HttpSendStream, Res,
     SendStream, SendStreamEvents, Stream,
 };
 
@@ -216,7 +216,6 @@ impl SendStream for SendMessage {
             .send_atomic(conn, &buf[..to_send])
             .map_err(|e| Error::map_stream_send_errors(&e))?;
         debug_assert!(sent);
-        qlog::h3_data_moved_down(conn.qlog_mut(), self.stream_id(), to_send);
         Ok(to_send)
     }
 
@@ -243,7 +242,6 @@ impl SendStream for SendMessage {
     /// info that the stream has been closed.)
     fn send(&mut self, conn: &mut Connection) -> Res<()> {
         let sent = Error::map_error(self.stream.send_buffer(conn), Error::HttpInternal(5))?;
-        qlog::h3_data_moved_down(conn.qlog_mut(), self.stream_id(), sent);
 
         qtrace!([self], "{} bytes sent", sent);
         if !self.stream.has_buffered_data() {