fix(http3): only add to stream_has_pending_data on has_data_to_send

[mxinden]

<SendMessage as Sendstream>::send_data attempts to send a slice of data down into the QUIC layer, more specifically neqo_transport::Connection::stream_send_atomic.

While it attempts to send any existing buffered data at the http3 layer first, it does not itself fill the http3 layer buffer, but instead only sends data, if the lower QUIC layer has capacity, i.e. only if it can send the data down to the QUIC layer right away.

neqo/neqo-http3/src/send_message.rs

Lines 168 to 221 in 5dfe106

	fn send_data(&mut self, conn: &mut Connection, buf: &[u8]) -> Res<usize> {
	qtrace!([self], "send_body: len={}", buf.len());
	self.state.new_data()?;
	self.stream.send_buffer(conn)?;
	if self.stream.has_buffered_data() {
	return Ok(0);
	}
	let available = conn
	.stream_avail_send_space(self.stream_id())
	.map_err(|e| Error::map_stream_send_errors(&e.into()))?;
	if available <= 2 {
	return Ok(0);
	}
	let to_send = if available <= MAX_DATA_HEADER_SIZE_2_LIMIT {
	// 63 + 3
	min(min(buf.len(), available - 2), MAX_DATA_HEADER_SIZE_2)
	} else if available <= MAX_DATA_HEADER_SIZE_3_LIMIT {
	// 16383 + 5
	min(min(buf.len(), available - 3), MAX_DATA_HEADER_SIZE_3)
	} else if available <= MAX_DATA_HEADER_SIZE_5 {
	// 1073741823 + 9
	min(min(buf.len(), available - 5), MAX_DATA_HEADER_SIZE_5_LIMIT)
	} else {
	min(buf.len(), available - 9)
	};
	qdebug!(
	[self],
	"send_request_body: available={} to_send={}.",
	available,
	to_send
	);
	let data_frame = HFrame::Data {
	len: to_send as u64,
	};
	let mut enc = Encoder::default();
	data_frame.encode(&mut enc);
	let sent_fh = self
	.stream
	.send_atomic(conn, enc.as_ref())
	.map_err(|e| Error::map_stream_send_errors(&e))?;
	debug_assert!(sent_fh);
	let sent = self
	.stream
	.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)
	}

<SendMessage as Sendstream>::send_data is called via Http3ServerHandler::send_data. The wrapper first marks the stream as stream_has_pending_data, marks itself as needs_processing and then calls down into <SendMessage as Sendstream>::send_data.

neqo/neqo-http3/src/connection_server.rs

Lines 51 to 74 in 5dfe106

	/// Supply a response for a request.
	///
	/// # Errors
	///
	/// `InvalidStreamId` if the stream does not exist,
	/// `AlreadyClosed` if the stream has already been closed.
	/// `TransportStreamDoesNotExist` if the transport stream does not exist (this may happen if
	/// `process_output` has not been called when needed, and HTTP3 layer has not picked up the
	/// info that the stream has been closed.) `InvalidInput` if an empty buffer has been
	/// supplied.
	pub(crate) fn send_data(
	&mut self,
	stream_id: StreamId,
	data: &[u8],
	conn: &mut Connection,
	) -> Res<usize> {
	self.base_handler.stream_has_pending_data(stream_id);
	self.needs_processing = true;
	self.base_handler
	.send_streams
	.get_mut(&stream_id)
	.ok_or(Error::InvalidStreamId)?
	.send_data(conn, data)
	}

Thus the latter always marks the former as stream_has_pending_data even though the former never writes into the buffer and thus might actually not have pending data.

Why is this problematic?

1. Say that the buffer of the BufferedStream of SendMessage is empty.

2. Say that the user attempts to write data via Http3ServerHandler::send_data. Despite not filling the http3 layer buffer, the stream is marked as stream_has_pending_data.

3. Say that next the user calls Http3Server::process, which will call Http3Server::process_http3, which will call
   Http3ServerHandler::process_http3, which will call Http3Connection::process_sending, which will call Http3Connection::send_non_control_streams.

Http3Connection::send_non_control_streams will attempt to flush all http3 layer buffers of streams marked via stream_has_pending_data, e.g. the stream from step (2). Thus it will call <SendMessage as SendStream>::send (note send and not the previous send_data). This function will attempt the stream's http3 layer buffer. In the case where the http3 layer stream buffer is empty, it will enqueue a DataWritable event for the user. Given that the buffer of our stream is empty (see (1)) such DataWritable event is always emitted.

neqo/neqo-http3/src/send_message.rs

Lines 236 to 264 in 5dfe106

	/// # Errors
	///
	/// `InternalError` if an unexpected error occurred.
	/// `InvalidStreamId` if the stream does not exist,
	/// `AlreadyClosed` if the stream has already been closed.
	/// `TransportStreamDoesNotExist` if the transport stream does not exist (this may happen if
	/// `process_output` has not been called when needed, and HTTP3 layer has not picked up the
	/// 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() {
	if self.state.done() {
	Error::map_error(
	conn.stream_close_send(self.stream_id()),
	Error::HttpInternal(6),
	)?;
	qtrace!([self], "done sending request");
	} else {
	// DataWritable is just a signal for an application to try to write more data,
	// if writing fails it is fine. Therefore we do not need to properly check
	// whether more credits are available on the transport layer.
	self.conn_events.data_writable(self.get_stream_info());
	}
	}
	Ok(())
	}

The user, on receiving the DataWritable event will attempt to write once more to the stream via Http3ServerHandler::send_data, back to step (2), thus closing the busy loop.

How to break the loop?

This commit adds an additional check to the stream_has_pending_data function to ensure it indeed does have pending data. This breaks the above busy loop. In addition, it renames the function to stream_might_have_pending_data.

---

This e.g. surfaces in the 1-conn/10_000-1b-seq-resp (aka. RPS) benchmark.

neqo/neqo-bin/benches/main.rs

Lines 39 to 44 in 5dfe106

	Benchmark {
	name: "1-conn/10_000-1b-seq-resp (aka. RPS)".to_string(),
	requests: vec![1; 10_000],
	download_in_series: false,
	sample_size: None,
	},

[github-actions]

Benchmark results

Performance differences relative to a65d945.

• coalesce_acked_from_zero 1+1 entries
  time: [194.81 ns 195.23 ns 195.70 ns]
  change: [-0.1684% +0.2125% +0.6373%] (p = 0.31 > 0.05)
  No change in performance detected.

• coalesce_acked_from_zero 3+1 entries
  time: [236.11 ns 236.77 ns 237.48 ns]
  change: [-0.5263% -0.0602% +0.3564%] (p = 0.80 > 0.05)
  No change in performance detected.

• coalesce_acked_from_zero 10+1 entries
  time: [234.55 ns 235.39 ns 236.38 ns]
  change: [-2.9907% -0.8746% +0.5296%] (p = 0.44 > 0.05)
  No change in performance detected.

• coalesce_acked_from_zero 1000+1 entries
  time: [218.46 ns 227.28 ns 247.01 ns]
  change: [-0.6486% +1.4472% +4.8915%] (p = 0.49 > 0.05)
  No change in performance detected.

• RxStreamOrderer::inbound_frame()
  time: [119.06 ms 119.12 ms 119.17 ms]
  change: [+0.0213% +0.1081% +0.1943%] (p = 0.02 < 0.05)
  Change within noise threshold.

• transfer/Run multiple transfers with varying seeds
  time: [119.78 ms 119.99 ms 120.20 ms]
  thrpt: [33.277 MiB/s 33.335 MiB/s 33.394 MiB/s]
  change:
  time: [-2.3395% -2.0718% -1.8112%] (p = 0.00 < 0.05)
  thrpt: [+1.8446% +2.1156% +2.3955%]
  Change within noise threshold.

• transfer/Run multiple transfers with the same seed
  time: [120.51 ms 120.69 ms 120.86 ms]
  thrpt: [33.095 MiB/s 33.143 MiB/s 33.192 MiB/s]
  change:
  time: [-2.0471% -1.8573% -1.6702%] (p = 0.00 < 0.05)
  thrpt: [+1.6986% +1.8924% +2.0898%]
  Change within noise threshold.

• 1-conn/1-100mb-resp (aka. Download)/client
  time: [1.0668 s 1.1002 s 1.1388 s]
  thrpt: [87.814 MiB/s 90.893 MiB/s 93.740 MiB/s]
  change:
  time: [-6.5125% -2.2417% +2.2868%] (p = 0.35 > 0.05)
  thrpt: [-2.2357% +2.2931% +6.9662%]
  No change in performance detected.

• 1-conn/10_000-parallel-1b-resp (aka. RPS)/client
  time: [381.83 ms 384.29 ms 386.79 ms]
  thrpt: [25.854 Kelem/s 26.022 Kelem/s 26.190 Kelem/s]
  change:
  time: [-1.7708% -0.7550% +0.1392%] (p = 0.13 > 0.05)
  thrpt: [-0.1390% +0.7608% +1.8027%]
  No change in performance detected.

• 1-conn/1-1b-resp (aka. HPS)/client
  time: [42.529 ms 42.676 ms 42.846 ms]
  thrpt: [23.339 elem/s 23.432 elem/s 23.513 elem/s]
  change:
  time: [-0.8145% -0.2510% +0.3028%] (p = 0.39 > 0.05)
  thrpt: [-0.3018% +0.2516% +0.8212%]
  No change in performance detected.

Client/server transfer results

Transfer of 134217728 bytes over loopback.

Client	Server	CC	Pacing	Mean [ms]	Min [ms]	Max [ms]	Relative
msquic	msquic			780.7 ± 282.3	447.6	1317.6	1.00
neqo	msquic	reno	on	1972.7 ± 160.0	1873.4	2414.9	1.00
neqo	msquic	reno		2052.0 ± 223.5	1910.9	2536.8	1.00
neqo	msquic	cubic	on	1912.2 ± 26.2	1865.3	1940.9	1.00
neqo	msquic	cubic		2057.8 ± 192.1	1874.4	2458.6	1.00
msquic	neqo	reno	on	3274.9 ± 110.0	3114.8	3494.0	1.00
msquic	neqo	reno		3467.1 ± 281.3	3232.0	3814.6	1.00
msquic	neqo	cubic	on	3310.8 ± 195.5	3170.9	3858.2	1.00
msquic	neqo	cubic		3291.8 ± 244.4	3129.9	3790.1	1.00
neqo	neqo	reno	on	3014.4 ± 179.4	2876.5	3509.5	1.00
neqo	neqo	reno		3033.8 ± 198.8	2814.6	3420.9	1.00
neqo	neqo	cubic	on	3242.0 ± 236.4	3057.6	3685.2	1.00
neqo	neqo	cubic		3166.8 ± 222.1	2972.8	3757.4	1.00

⬇️ Download logs

[mxinden]

Thanks @martinthomson for the review. Would you mind taking another look?

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 93.12%. Comparing base (a65d945) to head (5157e73).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #1793   +/-   ##
=======================================
  Coverage   93.12%   93.12%           
=======================================
  Files         116      116           
  Lines       36097    36100    +3     
=======================================
+ Hits        33614    33617    +3     
  Misses       2483     2483           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.