diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 9446aaa..ec73e7c 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -13,6 +13,9 @@ pub(super) const RTO: std::time::Duration = std::time::Duration::from_secs(1); /// Maximum count of retransmissions before dropping the packet pub(super) const MAX_RETRANSMIT_COUNT: usize = 3; +/// Maximum window scale shift count, which RFC 7323 §2.3 limits to 14 for a maximum window of 1 GiB +const MAX_WINDOW_SHIFT: u8 = 14; + #[derive(Debug, PartialEq, Clone, Copy)] pub(crate) enum TcpState { // Init, /* Since we always act as a server, it starts from `Listen`, so we don't use states Init & SynSent. */ @@ -44,13 +47,19 @@ pub(super) enum PacketType { /// - `unordered_packets` is the bytes stream received from the lower device, /// which can be acknowledged and extracted by `consume_unordered_packets` method /// then can be read by upstream application via `Tcp::poll_read` method. +/// - `send_window_shift` is the peer's window scale, applied to every window the peer advertises, +/// and `recv_window_shift` is this stack's own, applied to every window this stack advertises. +/// Both are settled by the SYN exchange, and `recv_window_shift` is `None` when the peer's SYN +/// carried no window scale option, which leaves both directions unscaled. #[derive(Debug, Clone)] pub(crate) struct Tcb { seq: SeqNum, ack: SeqNum, mtu: u16, last_received_ack: SeqNum, - send_window: u16, + send_window: u32, + send_window_shift: u8, + recv_window_shift: Option, state: TcpState, inflight_packets: BTreeMap, unordered_packets: BTreeMap>, @@ -64,8 +73,11 @@ pub(crate) struct Tcb { } impl Tcb { + #[allow(clippy::too_many_arguments)] pub(super) fn new( ack: SeqNum, + peer_window: u16, + peer_window_shift: Option, mtu: u16, max_unacked_bytes: u32, read_buffer_size: usize, @@ -77,12 +89,32 @@ impl Tcb { let seq = 100; #[cfg(not(debug_assertions))] let seq = rand::RngExt::random::(&mut rand::rng()); + let send_window_shift = peer_window_shift.map_or(0, |shift| { + if shift > MAX_WINDOW_SHIFT { + log::warn!("Peer window scale shift count {shift} is too large, limiting it to {MAX_WINDOW_SHIFT}"); + MAX_WINDOW_SHIFT + } else { + shift + } + }); + // The stack scales its own receive window by the smallest shift that expresses the whole + // read buffer in the 16-bit window field. + let recv_window_shift = peer_window_shift.map(|_| { + (0..=MAX_WINDOW_SHIFT) + .find(|&shift| read_buffer_size >> shift <= u16::MAX as usize) + .unwrap_or_else(|| { + log::warn!("Read buffer size {read_buffer_size} is too large to scale, limiting the shift count to {MAX_WINDOW_SHIFT}"); + MAX_WINDOW_SHIFT + }) + }); Tcb { seq: seq.into(), ack, mtu, last_received_ack: seq.into(), - send_window: u16::MAX, + send_window: peer_window as u32, + send_window_shift, + recv_window_shift, state: TcpState::Listen, inflight_packets: BTreeMap::new(), unordered_packets: BTreeMap::new(), @@ -209,15 +241,29 @@ impl Tcb { pub(super) fn get_state(&self) -> TcpState { self.state } - pub(super) fn update_send_window(&mut self, window: u16) { - self.send_window = window; + fn honoured_window(&self, tcp_header: &TcpHeader) -> u32 { + // RFC 7323 §2.3: SND.WND = SEG.WND << Snd.Wind.Shift, except on a segment carrying SYN, + // whose window field is never scaled. + let window = tcp_header.window_size as u32; + if tcp_header.syn { window } else { window << self.send_window_shift } } - pub(super) fn get_send_window(&self) -> u16 { + pub(super) fn update_send_window(&mut self, tcp_header: &TcpHeader) { + self.send_window = self.honoured_window(tcp_header); + } + pub(super) fn get_send_window(&self) -> u32 { self.send_window } pub(super) fn get_recv_window(&self) -> u16 { self.get_available_read_buffer_size().try_into().unwrap_or(u16::MAX) } + pub(super) fn get_scaled_recv_window(&self) -> u16 { + // RFC 7323 §2.3: SEG.WND = RCV.WND >> Rcv.Wind.Shift + let window = self.get_available_read_buffer_size() >> self.recv_window_shift.unwrap_or(0); + window.try_into().unwrap_or(u16::MAX) + } + pub(super) fn get_recv_window_shift(&self) -> Option { + self.recv_window_shift + } // #[inline(always)] // pub(super) fn buffer_size(&self, payload_len: u16) -> u16 { // match MAX_UNACK - self.inflight_packets.len() as u32 { @@ -234,7 +280,7 @@ impl Tcb { pub(super) fn check_pkt_type(&self, tcp_header: &TcpHeader, payload: &[u8]) -> PacketType { let rcvd_ack = SeqNum(tcp_header.acknowledgment_number); let rcvd_seq = SeqNum(tcp_header.sequence_number); - let rcvd_window = tcp_header.window_size; + let rcvd_window = self.honoured_window(tcp_header); let len = payload.len(); let res = if rcvd_ack > self.seq { PacketType::Invalid @@ -339,7 +385,7 @@ impl Tcb { 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) - self.seq.distance(self.get_last_received_ack()) >= self.max_unacked_bytes.min(self.get_send_window() as u32) + self.seq.distance(self.get_last_received_ack()) >= self.max_unacked_bytes.min(self.get_send_window()) } } @@ -391,6 +437,8 @@ mod tests { fn test_get_unordered_packets_with_max_bytes() { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -431,6 +479,8 @@ mod tests { fn test_add_unordered_packet_enforces_read_buffer() { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -456,6 +506,8 @@ mod tests { fn test_consume_trims_overlapping_head_entry() { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -480,6 +532,8 @@ mod tests { fn test_update_inflight_packet_queue() { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -512,6 +566,8 @@ mod tests { fn test_update_inflight_packet_queue_cumulative_ack() { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -535,6 +591,8 @@ mod tests { fn test_retransmit_with_exponential_backoff() { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -563,4 +621,76 @@ mod tests { assert!(packets.is_empty()); assert!(tcb.inflight_packets.is_empty()); } + + /// A peer shift of 7 is applied to every window the peer advertises after the handshake. + #[test] + fn test_peer_window_shift_is_taken_from_the_syn() { + let mut tcb = window_tcb(4000, Some(7), READ_BUFFER_SIZE); + assert_eq!(tcb.get_send_window(), 4000); // the SYN's own window is unscaled + + tcb.update_send_window(&TcpHeader::new(1, 2, 1000, 40_000)); + assert_eq!(tcb.get_send_window(), 40_000 << 7); + } + + /// A peer that offers no window scale option has its windows honoured as they stand. + #[test] + fn test_no_window_scale_option_leaves_the_window_unshifted() { + let mut tcb = window_tcb(4000, None, READ_BUFFER_SIZE); + assert_eq!(tcb.get_send_window(), 4000); + + tcb.update_send_window(&TcpHeader::new(1, 2, 1000, 40_000)); + assert_eq!(tcb.get_send_window(), 40_000); + assert_eq!(tcb.get_recv_window_shift(), None); + } + + /// RFC 7323 §2.3 limits the shift count to 14, so a larger one is used as 14. + #[test] + fn test_peer_window_shift_above_the_maximum_is_clamped() { + let mut tcb = window_tcb(4000, Some(15), READ_BUFFER_SIZE); + + tcb.update_send_window(&TcpHeader::new(1, 2, 1000, 40_000)); + assert_eq!(tcb.get_send_window(), 40_000 << MAX_WINDOW_SHIFT); + } + + /// A peer opening with a closed window is held to it, whatever shift the same SYN offers. + #[test] + fn test_zero_peer_window_is_honoured() { + assert_eq!(window_tcb(0, None, READ_BUFFER_SIZE).get_send_window(), 0); + assert_eq!(window_tcb(0, Some(7), READ_BUFFER_SIZE).get_send_window(), 0); + } + + /// The announced shift is the smallest expressing the read buffer in 16 bits, the advertised + /// window is the free space shifted down by it, and a peer shift of 0 still enables scaling. + #[test] + fn test_advertised_window_is_derived_from_the_read_buffer() { + assert_eq!(window_tcb(4000, Some(0), 16 * 1024).get_recv_window_shift(), Some(0)); + assert_eq!(window_tcb(4000, Some(0), 64 * 1024).get_recv_window_shift(), Some(1)); + + // 1.2 MiB needs a shift of 5, which expresses the window only in multiples of 32, so the + // advertised value rounds down and withholds the remainder rather than overstating the room + let buffer = 1_258_291; + let mut tcb = window_tcb(4000, Some(0), buffer); + assert_eq!(tcb.get_recv_window_shift(), Some(5)); + assert_eq!(usize::from(tcb.get_scaled_recv_window()), buffer >> 5); + assert!(usize::from(tcb.get_scaled_recv_window()) << 5 < buffer); + + // the advertised window shrinks as the buffer fills + tcb.add_unordered_packet(SeqNum(1000), vec![0; buffer - 1000]); + assert_eq!(usize::from(tcb.get_scaled_recv_window()), 1000 >> 5); + } + + // A `Tcb` opened by a SYN advertising `peer_window` under `peer_window_shift`. + fn window_tcb(peer_window: u16, peer_window_shift: Option, read_buffer_size: usize) -> Tcb { + Tcb::new( + SeqNum(1000), + peer_window, + peer_window_shift, + 1500, + MAX_UNACK, + read_buffer_size, + MAX_COUNT_FOR_DUP_ACK, + RTO, + MAX_RETRANSMIT_COUNT, + ) + } } diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index d766f16..3e1c41d 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -185,8 +185,20 @@ impl IpStackTcpStream { destroy_messenger: Option<::tokio::sync::oneshot::Sender<()>>, config: Arc, ) -> Result { + // RFC 7323 §2.2: a window scale option is read only from a segment carrying SYN. + let peer_window_shift = tcp + .syn + .then(|| { + tcp.options_iterator().flatten().find_map(|option| match option { + TcpOptionElement::WindowScale(shift) => Some(shift), + _ => None, + }) + }) + .flatten(); let tcb = Tcb::new( SeqNum(tcp.sequence_number), + tcp.window_size, + peer_window_shift, mtu, config.max_unacked_bytes, config.read_buffer_size, @@ -677,7 +689,6 @@ async fn tcp_main_logic_loop( let flags = tcp_header_flags(tcp_header); let incoming_ack: SeqNum = tcp_header.acknowledgment_number.into(); let incoming_seq: SeqNum = tcp_header.sequence_number.into(); - let incoming_win = tcp_header.window_size; let mut tcb = tcb.lock().unwrap(); @@ -727,6 +738,7 @@ async fn tcp_main_logic_loop( extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; } tcb.change_state(TcpState::Established); + write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } TcpState::Established => { if flags == ACK { @@ -813,6 +825,7 @@ async fn tcp_main_logic_loop( tcb.add_unordered_packet(incoming_seq, payload); extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; } + write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } else { // unnormal case, we do nothing here log::trace!("{network_tuple} {state:?}: {l_info}, {pkt_type:?}, unnormal case, we do nothing here"); @@ -920,7 +933,7 @@ async fn tcp_main_logic_loop( } // end of match state tcb.update_last_received_ack(incoming_ack); - tcb.update_send_window(incoming_win); + tcb.update_send_window(tcp_header); } // end of loop Ok::<(), std::io::Error>(()) } @@ -978,9 +991,29 @@ pub(crate) fn write_packet_to_device( // Silly-window-syndrome avoidance: advertise a real window only when a full segment fits, // otherwise advertise zero so the peer enters persist mode until the reader frees space. let recv_window = tcb.get_recv_window(); - let window_size = if recv_window >= tcb.get_mtu() { recv_window } else { 0 }; + // RFC 7323 §2.2: the window field of a segment carrying SYN is not scaled. + let advertised = if flags & SYN != 0 { + recv_window + } else { + tcb.get_scaled_recv_window() + }; + let window_size = if recv_window >= tcb.get_mtu() { advertised } else { 0 }; let ack = tcb.get_ack().0; let (src, dst) = (tuple.dst, tuple.src); // Note: The address is reversed here + let mut tcp_options = Vec::new(); + // Note: Instead of an `if let Some ... ` a loop is used as the enum is #[non_exhaustive] + for option in options.into_iter().flatten() { + match option { + TcpOptions::MaximumSegmentSize(mss) => tcp_options.push(TcpOptionElement::MaximumSegmentSize(*mss)), + } + } + // RFC 7323 §2.2: the window scale option travels only on a segment carrying SYN, and only when + // the peer's SYN carried one. + if flags & SYN != 0 + && let Some(shift) = tcb.get_recv_window_shift() + { + tcp_options.push(TcpOptionElement::WindowScale(shift)); + } let calc = |ip_header_len: usize, tcp_header_len: usize| tcb.calculate_payload_max_len(ip_header_len, tcp_header_len); let packet = create_raw_packet( src, @@ -992,7 +1025,7 @@ pub(crate) fn write_packet_to_device( ack, window_size, payload.unwrap_or_default(), - options, + &tcp_options, )?; let len = packet.payload.as_ref().map(|p| p.len()).unwrap_or(0); up_packet_sender.send(packet).map_err(|e| Error::new(UnexpectedEof, e))?; @@ -1010,7 +1043,7 @@ pub(crate) fn create_raw_packet( ack: u32, win: u16, mut payload: Vec, - options: Option<&Vec>, + options: &[TcpOptionElement], ) -> std::io::Result { let mut tcp_header = etherparse::TcpHeader::new(src_addr.port(), dst_addr.port(), seq, win); tcp_header.acknowledgment_number = ack; @@ -1020,17 +1053,8 @@ pub(crate) fn create_raw_packet( tcp_header.fin = flags & FIN != 0; tcp_header.psh = flags & PSH != 0; - if let Some(opts) = options { - let mut tcp_options = Vec::new(); - for opt in opts { - match opt { - TcpOptions::MaximumSegmentSize(mss) => tcp_options.push(TcpOptionElement::MaximumSegmentSize(*mss)), - } - } - tcp_header - .set_options(&tcp_options) - .map_err(|e| std::io::Error::new(InvalidInput, e))?; - } + tcp_header.set_options(options).map_err(|e| std::io::Error::new(InvalidInput, e))?; + let ip_header = match (src_addr.ip(), dst_addr.ip()) { (std::net::IpAddr::V4(src), std::net::IpAddr::V4(dst)) => { let mut ip_h = @@ -1086,6 +1110,18 @@ mod tests { use super::*; use crate::stream::tcb::{MAX_COUNT_FOR_DUP_ACK, MAX_RETRANSMIT_COUNT, MAX_UNACK, READ_BUFFER_SIZE, RTO}; + /// A waker that records whether it was woken, and the flag it sets. + fn recording_waker() -> (Arc, Waker) { + struct Recording(Arc); + impl std::task::Wake for Recording { + fn wake(self: Arc) { + self.0.store(true, std::sync::atomic::Ordering::SeqCst); + } + } + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + (woken.clone(), Waker::from(Arc::new(Recording(woken)))) + } + #[tokio::test] async fn extract_reserves_before_consuming() { let (up_tx, _up_rx) = tokio::sync::mpsc::unbounded_channel::(); @@ -1095,6 +1131,8 @@ mod tests { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -1123,4 +1161,113 @@ mod tests { assert_eq!(tcb.get_ack(), SeqNum(2500)); assert_eq!(tcb.get_unordered_packets_total_len(), 0); } + + /// Opens a connection with a SYN advertising `window` and `syn_options`, returning the stream + /// and the SYN-ACK the stack emitted. + async fn open(window: u16, syn_options: &[TcpOptionElement]) -> (IpStackTcpStream, TcpHeader, PacketReceiver) { + let (up_tx, mut up_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (src, dst) = (SocketAddr::from(([10, 0, 0, 2], 40000)), SocketAddr::from(([10, 0, 0, 1], 80))); + + let mut syn = TcpHeader::new(src.port(), dst.port(), 1000, window); + syn.syn = true; + syn.set_options(syn_options).unwrap(); + + let stream = IpStackTcpStream::new(src, dst, syn, up_tx, 1500, None, Arc::new(TcpConfig::default())).unwrap(); + let packet = up_rx.recv().await.expect("the stack emitted no SYN-ACK"); + let TransportHeader::Tcp(syn_ack) = packet.transport_header() else { + panic!("the emitted SYN-ACK is not a TCP packet"); + }; + (stream, syn_ack.clone(), up_rx) + } + + /// Feeds `stream` a segment advertising `window`, carrying `payload` at `seq`. + fn feed(stream: &IpStackTcpStream, syn_ack: &TcpHeader, seq: SeqNum, window: u16, payload: &[u8]) { + let builder = etherparse::PacketBuilder::ipv4([10, 0, 0, 2], [10, 0, 0, 1], TTL) + .tcp(40000, 80, seq.0, window) + .ack((SeqNum(syn_ack.sequence_number) + 1).0) + .psh(); + let mut buf = Vec::new(); + builder.write(&mut buf, payload).unwrap(); + stream.stream_sender().send(NetworkPacket::parse(&buf).unwrap()).unwrap(); + } + + /// Returns the shift count of the window scale option in `header`, if it carries one. + fn window_scale_of(header: &TcpHeader) -> Option { + header.options_iterator().flatten().find_map(|option| match option { + TcpOptionElement::WindowScale(shift) => Some(shift), + _ => None, + }) + } + + /// A peer offering a shift of 7 gets this stack's shift of 0 back and has its windows honoured + /// shifted by 7. + #[tokio::test] + async fn syn_with_window_scale_is_answered_and_honoured() { + let (stream, syn_ack, mut up_rx) = open(4000, &[TcpOptionElement::WindowScale(7)]).await; + assert_eq!( + window_scale_of(&syn_ack), + Some(0), + "the SYN-ACK does not carry this stack's window scale" + ); + + let peer_seq = SeqNum(syn_ack.acknowledgment_number); + feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]); + feed(&stream, &syn_ack, peer_seq + 4, 40_000, &[2; 4]); + up_rx.recv().await.unwrap(); + up_rx.recv().await.unwrap(); + + assert_eq!(stream.tcb.lock().unwrap().get_send_window(), 40_000 << 7); + } + + /// A peer offering a shift above the RFC 7323 §2.3 maximum has its windows honoured shifted by 14. + #[tokio::test] + async fn syn_with_a_malformed_window_scale_is_clamped() { + let (stream, syn_ack, mut up_rx) = open(4000, &[TcpOptionElement::WindowScale(20)]).await; + + let peer_seq = SeqNum(syn_ack.acknowledgment_number); + feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]); + feed(&stream, &syn_ack, peer_seq + 4, 40_000, &[2; 4]); + up_rx.recv().await.unwrap(); + up_rx.recv().await.unwrap(); + + let tcb = stream.tcb.lock().unwrap(); + assert_eq!(tcb.get_state(), TcpState::Established); + assert_eq!(tcb.get_send_window(), 40_000 << 14); + } + + /// A closed peer window holds the writer, and the segment that reopens it wakes the writer, both + /// when it completes the handshake and when it arrives on an established connection. + #[tokio::test] + async fn zero_peer_window_holds_the_writer_until_it_reopens() { + let (mut stream, syn_ack, mut up_rx) = open(0, &[]).await; + let peer_seq = SeqNum(syn_ack.acknowledgment_number); + + // the segment completing the handshake reopens the window + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(Pin::new(&mut stream).poll_write(&mut cx, b"held").is_pending()); + feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]); + up_rx.recv().await.unwrap(); + assert!( + woken.load(std::sync::atomic::Ordering::SeqCst), + "the handshake reopened the window without waking the writer" + ); + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"held"), Poll::Ready(Ok(4)))); + assert_eq!(up_rx.recv().await.unwrap().payload.as_deref(), Some(&b"held"[..])); + + // a data segment closes the window and a later one reopens it + feed(&stream, &syn_ack, peer_seq + 4, 0, &[2; 4]); + up_rx.recv().await.unwrap(); + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(Pin::new(&mut stream).poll_write(&mut cx, b"sent").is_pending()); + feed(&stream, &syn_ack, peer_seq + 8, 40_000, &[3; 4]); + up_rx.recv().await.unwrap(); + assert!( + woken.load(std::sync::atomic::Ordering::SeqCst), + "a data segment reopened the window without waking the writer" + ); + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"sent"), Poll::Ready(Ok(4)))); + assert_eq!(up_rx.recv().await.unwrap().payload.as_deref(), Some(&b"sent"[..])); + } }