TCP Read Optimization

[Orvaxis]

The current implementation shares contiguous bytes with an AsyncRead consumer through a channel. However, this approach prevents backpressure, which leads to significant memory consumption if the sending rate is too fast. Furthermore, because the buffered data is immediately sent to the channel, the TCP receive window does not shrink. This disables TCP-level flow control, further exacerbating memory accumulation.

The solution I've come up with is to create an ordered list of packets within the TCB. The consumption method, each time it's called, will move the contiguous packets into this list.
The window calculation will also be modified to determine the actual window size by accounting for the sizes of both the in-order (ordered) and out-of-order packets.
Then, in the AsyncRead implementation, it will get the TCB, pop the first packet from the ordered collection, and write it into the read buffer.

[SajjadPourali]

Do you mean this part?

ipstack/src/stream/tcb.rs

Lines 54 to 55 in 92e5a87

	inflight_packets: BTreeMap<SeqNum, InflightPacket>,
	unordered_packets: BTreeMap<SeqNum, Vec<u8>>,

[Orvaxis]

Here, every time a new data packet is received, it is sent through the channel, which frees up the receive window. However, the data in the channel might not be consumed in time by the upstream TCP Stream, causing it to accumulate here.

ipstack/src/stream/tcp.rs

Lines 846 to 853 in 92e5a87

	if let Some(data) = tcb.consume_unordered_packets(8192) {
	let hint = if state == TcpState::Established { "normally" } else { "still" };
	log::trace!("{network_tuple} {state:?}: {l_info} {hint} receiving data, len = {}", data.len());
	data_tx.send(data).map_err(|e| std::io::Error::new(BrokenPipe, e))?;
	read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
	write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?;
	}
	Ok(())

Recently, I discovered a similar issue here, which causes traffic to accumulate in bursts during high-throughput uploads.

ipstack/src/lib.rs

Lines 13 to 15 in 92e5a87

	pub(crate) type PacketSender = UnboundedSender<NetworkPacket>;
	pub(crate) type PacketReceiver = UnboundedReceiver<NetworkPacket>;
	pub(crate) type SessionCollection = AHashMap<NetworkTuple, PacketSender>;

ipstack/src/lib.rs

Lines 177 to 181 in 92e5a87

	match sessions.entry(network_tuple) {
	std::collections::hash_map::Entry::Occupied(entry) => {
	let len = packet.payload.as_ref().map(|p| p.len()).unwrap_or(0);
	log::trace!("packet sent to stream: {network_tuple} len {len}");
	entry.get().send(packet).map_err(std::io::Error::other)?;