From 531ad8c3ff953ba00b2075db77cb3cacc1eca342 Mon Sep 17 00:00:00 2001 From: WYCLIFF001 Date: Wed, 30 Sep 2026 02:10:36 +0300 Subject: [PATCH 1/3] Answer an out-of-order segment with a duplicate ACK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A data segment is now always acknowledged. One that fills no gap, because it arrived out of order or was already received, used to leave nothing for the reader and so produced no ACK at all, which left the peer's retransmission timer as the only way past a lost segment. RFC 5681 §4.2 asks for an immediate duplicate ACK instead, and three of them are what drive the peer's fast retransmit. A segment carrying PSH is handled like any other: out of order, it used to be dropped without an ACK, so the peer resent it on its timer and backed off with every loss. --- src/stream/tcp.rs | 136 +++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 121 insertions(+), 15 deletions(-) diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index 28e06b2..33be8c7 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -773,8 +773,15 @@ async fn tcp_main_logic_loop( match state { TcpState::SynReceived if flags & ACK == ACK => { if len > 0 { - tcb.add_unordered_packet(incoming_seq, payload); - extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; + receive_segment( + &up_packet_sender, + &mut tcb, + network_tuple, + incoming_seq, + payload, + &data_tx, + &read_notify, + )?; } tcb.change_state(TcpState::Established); write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); @@ -799,9 +806,8 @@ async fn tcp_main_logic_loop( } } PacketType::NewPacket => { - tcb.add_unordered_packet(incoming_seq, payload); let nt = network_tuple; - extract_data_n_write_upstream(&up_packet_sender, &mut tcb, nt, &data_tx, &read_notify)?; + receive_segment(&up_packet_sender, &mut tcb, nt, incoming_seq, payload, &data_tx, &read_notify)?; write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } PacketType::Ack => { @@ -860,9 +866,18 @@ async fn tcp_main_logic_loop( )); } } else if flags == (ACK | PSH) && pkt_type == PacketType::NewPacket { - if !payload.is_empty() && tcb.get_ack() == incoming_seq { - tcb.add_unordered_packet(incoming_seq, payload); - extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; + // PSH changes nothing about where a segment belongs: one that arrives out of order + // is kept and answered like any other. + if !payload.is_empty() { + receive_segment( + &up_packet_sender, + &mut tcb, + network_tuple, + incoming_seq, + payload, + &data_tx, + &read_notify, + )?; } write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } else { @@ -918,8 +933,15 @@ async fn tcp_main_logic_loop( tcb.change_state(TcpState::FinWait2); if len > 0 { // if the other side is still sending data, we need to deal with it like PacketStatus::NewPacket - tcb.add_unordered_packet(incoming_seq, payload); - extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; + receive_segment( + &up_packet_sender, + &mut tcb, + network_tuple, + incoming_seq, + payload, + &data_tx, + &read_notify, + )?; write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } let new_state = tcb.get_state(); @@ -948,8 +970,15 @@ async fn tcp_main_logic_loop( write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?; } else { // if the other side is still sending data, we need to deal with it like PacketStatus::NewPacket - tcb.add_unordered_packet(incoming_seq, payload); - extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; + receive_segment( + &up_packet_sender, + &mut tcb, + network_tuple, + incoming_seq, + payload, + &data_tx, + &read_notify, + )?; write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } if flags & FIN == FIN && tcb.get_ack() == incoming_seq + len as u32 { @@ -991,18 +1020,41 @@ async fn wait_retransmission_deadline(tcb: &TcbPtr, arm_notify: &tokio::sync::No } } -fn extract_data_n_write_upstream( +/// Stores a data segment that arrived from the peer and acknowledges it. +/// +/// The ACK goes out even when no new data reaches the reader: a segment that arrives out of order, +/// or one already received, is answered with a duplicate ACK at once (RFC 5681 §4.2), which is +/// what lets the peer fast-retransmit the segment in the gap instead of waiting out its timer. +fn receive_segment( up_packet_sender: &PacketSender, tcb: &mut Tcb, network_tuple: NetworkTuple, + seq: SeqNum, + payload: Vec, data_tx: &tokio::sync::mpsc::Sender>, read_notify: &std::sync::Arc>>, ) -> std::io::Result<()> { + tcb.add_unordered_packet(seq, payload); + if !extract_data_n_write_upstream(up_packet_sender, tcb, network_tuple, data_tx, read_notify)? { + write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?; + } + Ok(()) +} + +/// Hands acknowledged data to the reader. Returns whether an ACK was written, or none is due +/// because the session is closed. +fn extract_data_n_write_upstream( + up_packet_sender: &PacketSender, + tcb: &mut Tcb, + network_tuple: NetworkTuple, + data_tx: &tokio::sync::mpsc::Sender>, + read_notify: &std::sync::Arc>>, +) -> std::io::Result { let (state, seq, ack) = (tcb.get_state(), tcb.get_seq(), tcb.get_ack()); let l_info = format!("local {{ seq: {seq}, ack: {ack} }}"); if state == TcpState::Closed { log::debug!("{network_tuple} {state:?}: {l_info} session closed, exiting \"data extraction task\"..."); - return Ok(()); + return Ok(true); } tcb.acknowledge_unordered_packets(); @@ -1012,7 +1064,7 @@ fn extract_data_n_write_upstream( Ok(permit) => permit, Err(tokio::sync::mpsc::error::TrySendError::Full(())) => { write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?; - return Ok(()); + return Ok(true); } Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => { return Err(std::io::Error::new(BrokenPipe, "data channel closed")); @@ -1025,8 +1077,9 @@ fn extract_data_n_write_upstream( permit.send(data); 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)?; + return Ok(true); } - Ok(()) + Ok(false) } /// Send a TCP packet to the downstream device, with the specified flags, sequence number, and payload. @@ -1214,6 +1267,59 @@ mod tests { create_raw_packet(src, dst, |_, _| usize::MAX, ACK, TTL, seq, ack, u16::MAX, Vec::new(), &[]).unwrap() } + /// A segment from the peer carrying `payload` at `seq`, with `flags` and a raw `window`. + fn peer_segment(seq: u32, ack: u32, flags: u8, window: u16, payload: &[u8]) -> NetworkPacket { + let (src, dst) = (PEER.parse().unwrap(), LOCAL.parse().unwrap()); + create_raw_packet(src, dst, |_, _| usize::MAX, flags, TTL, seq, ack, window, payload.to_vec(), &[]).unwrap() + } + + /// Drains what the stack wrote to the device so far. + async fn written(up_rx: &mut PacketReceiver) -> Vec { + tokio::task::yield_now().await; + let mut packets = Vec::new(); + while let Ok(packet) = up_rx.try_recv() { + packets.push(packet); + } + packets + } + + /// RFC 5681 §4.2: an out-of-order segment is answered at once with a duplicate ACK, whether or + /// not it carries PSH, and is kept for when the gap fills. + #[tokio::test(start_paused = true)] + async fn out_of_order_segment_is_answered_with_a_duplicate_ack() { + for flags in [ACK, ACK | PSH] { + let (mut stream, mut up_rx, local_seq) = established().await; + + // Bytes 1001..1009 are missing; 1009..1013 arrive first. + stream + .stream_sender() + .send(peer_segment(1009, local_seq, flags, u16::MAX, &[3; 4])) + .unwrap(); + let acks: Vec = written(&mut up_rx) + .await + .iter() + .map(|p| tcp_header_of(p).acknowledgment_number) + .collect(); + assert_eq!(acks, [1001], "flags {flags:#x}: the out-of-order segment got no duplicate ACK"); + + // The gap fills: the ACK covers both segments and the reader sees them in order. + stream + .stream_sender() + .send(peer_segment(1001, local_seq, flags, u16::MAX, &[1; 8])) + .unwrap(); + let acks: Vec = written(&mut up_rx) + .await + .iter() + .map(|p| tcp_header_of(p).acknowledgment_number) + .collect(); + assert_eq!(acks.last(), Some(&1013), "flags {flags:#x}: the buffered segment was not kept"); + let mut buf = [0u8; 12]; + tokio::io::AsyncReadExt::read_exact(&mut stream, &mut buf).await.unwrap(); + assert_eq!(buf[..8], [1; 8]); + assert_eq!(buf[8..], [3; 4]); + } + } + /// The intervals RFC 6298 §5.5 produces from a one-second timeout doubling to the §2.5 ceiling. const BACKOFF: [u64; MAX_RETRANSMIT_COUNT] = [1, 2, 4, 8, 16, 32, 60]; From ae044e9d9cecfdaed4f44336418a7f901f87506e Mon Sep 17 00:00:00 2001 From: WYCLIFF001 Date: Wed, 30 Sep 2026 02:11:02 +0300 Subject: [PATCH 2/3] Keep the new part of a retransmission that straddles the ack MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A segment starting below the ack was dropped whole, even when it ran past it. That happens when the peer re-segments a retransmission after an ACK it missed: the bytes past the ack were then lost with the rest, and the peer had to send them yet again. The segment is now trimmed to the part that is new (RFC 9293 §3.10.7.4); one lying entirely below the ack is still dropped as a duplicate. --- src/stream/tcb.rs | 43 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 39 insertions(+), 4 deletions(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 9d322a2..d3ee437 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -164,11 +164,17 @@ impl Tcb { self.duplicate_ack_count >= self.max_count_for_dup_ack } - pub(super) fn add_unordered_packet(&mut self, seq: SeqNum, mut buf: Vec) { + pub(super) fn add_unordered_packet(&mut self, mut seq: SeqNum, mut buf: Vec) { if seq < self.ack { - #[rustfmt::skip] - log::warn!("{:?}: Received packet seq {seq} < self ack {}, len = {}", self.state, self.ack, buf.len()); - return; + // A retransmission re-segmented across the ack: keep the bytes past it. + let received = self.ack.distance(seq) as usize; + if received >= buf.len() { + #[rustfmt::skip] + log::warn!("{:?}: Received packet seq {seq} < self ack {}, len = {}", self.state, self.ack, buf.len()); + return; + } + buf.drain(..received); + seq = self.ack; } // A segment starting beyond a closed receive window is dropped for the peer's RTO to resend. if seq != self.ack && self.get_available_read_buffer_size() == 0 { @@ -544,6 +550,35 @@ mod tests { assert!(data.is_none()); } + /// A retransmission re-segmented across the ack keeps the bytes past it (RFC 9293 §3.10.7.4 + /// trims a segment to the part that lies in the window), instead of being dropped whole. + #[test] + fn test_add_unordered_packet_keeps_the_new_part_of_a_straddling_segment() { + let mut tcb = Tcb::new( + SeqNum(1000), + u16::MAX, + None, + 1500, + MAX_UNACK, + READ_BUFFER_SIZE, + MAX_COUNT_FOR_DUP_ACK, + RTO, + MAX_RETRANSMIT_COUNT, + ); + tcb.add_unordered_packet(SeqNum(1000), vec![1; 100]); + tcb.consume_unordered_packets(usize::MAX).unwrap(); + assert_eq!(tcb.ack, SeqNum(1100)); + + // 1050..1150: the first 50 bytes were already received, the last 50 are new. + tcb.add_unordered_packet(SeqNum(1050), vec![2; 100]); + assert_eq!(tcb.consume_unordered_packets(usize::MAX), Some(vec![2; 50])); + assert_eq!(tcb.ack, SeqNum(1150)); + + // Entirely below the ack: a pure duplicate, nothing is stored. + tcb.add_unordered_packet(SeqNum(1000), vec![3; 100]); + assert_eq!(tcb.consume_unordered_packets(usize::MAX), None); + } + #[test] fn test_add_unordered_packet_enforces_read_buffer() { let mut tcb = Tcb::new( From bc2cab85c9cb2e0d2b91ff890cefd857e3a0a56e Mon Sep 17 00:00:00 2001 From: WYCLIFF001 Date: Wed, 30 Sep 2026 02:12:18 +0300 Subject: [PATCH 3/3] Keep each segment inside the peer's window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A write was held only once the data in flight had reached the peer's window, and was then sent as a whole segment, so the last segment before the window filled could run up to a segment past it. RFC 9293 §3.8.6.1 keeps what is in flight within the window. A segment now carries no more than the room the window has left. The write also copies only the bytes the segment carries. It used to copy the caller's whole buffer and send one segment of it, so a large write was copied again for every segment it produced. --- src/stream/tcb.rs | 7 +++++++ src/stream/tcp.rs | 23 ++++++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index d3ee437..0818ca0 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -459,6 +459,13 @@ impl Tcb { self.inflight_packets.values().collect::>() } + /// Bytes that may still be sent: the peer's window, capped by the unacknowledged limit, less + /// what is already in flight. + pub(super) fn send_room(&self) -> usize { + let limit = self.max_unacked_bytes.min(self.get_send_window()); + limit.saturating_sub(self.seq.distance(self.get_last_received_ack())) as usize + } + pub fn is_send_buffer_full(&self) -> bool { // To respect the receiver's window (remote_window) size and avoid sending too many unacknowledged packets, which may cause packet loss // Simplified version: min(cwnd, rwnd) diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index 33be8c7..305a842 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -391,8 +391,11 @@ impl AsyncWrite for IpStackTcpStream { return Poll::Pending; } + // A segment carries no more than the peer's window has room for (RFC 9293 §3.8.6.1), and only + // that much of `buf` is copied: the rest is the caller's to offer again. + let take = buf.len().min(tcb.send_room()).min(tcb.get_mtu() as usize); let sender = &self.up_packet_sender; - let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf.to_vec()))?; + let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf[..take].to_vec()))?; let was_empty = tcb.is_inflight_queue_empty(); tcb.add_inflight_packet(buf[..payload_len].to_vec())?; if was_empty { @@ -1320,6 +1323,24 @@ mod tests { } } + /// RFC 9293 §3.8.6.1: a sender keeps the data in flight within the peer's window, so the last + /// segment before the window fills is cut to the room left rather than sent whole. + #[tokio::test] + async fn segments_stay_inside_the_peers_window() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream.stream_sender().send(peer_segment(1001, local_seq, ACK, 3000, &[])).unwrap(); + written(&mut up_rx).await; + + // More than the window: the write parks once the window is full. + let _ = tokio::time::timeout(Duration::from_millis(100), stream.write_all(&[7; 4000])).await; + let sent: Vec = written(&mut up_rx) + .await + .iter() + .filter_map(|p| p.payload.as_ref().map(Vec::len)) + .collect(); + assert_eq!(sent.iter().sum::(), 3000, "sent {sent:?} into a 3000-byte window"); + } + /// The intervals RFC 6298 §5.5 produces from a one-second timeout doubling to the §2.5 ceiling. const BACKOFF: [u64; MAX_RETRANSMIT_COUNT] = [1, 2, 4, 8, 16, 32, 60];