diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 9d322a2..0818ca0 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 { @@ -453,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) @@ -544,6 +557,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( diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index 28e06b2..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 { @@ -773,8 +776,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 +809,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 +869,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 +936,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 +973,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 +1023,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 +1067,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 +1080,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 +1270,77 @@ 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]); + } + } + + /// 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];