From 55025b2f4ee240bb1ad8c1b5d74e0570e1a13153 Mon Sep 17 00:00:00 2001 From: IntellyCode <47359527+IntellyCode@users.noreply.github.com> Date: Mon, 14 Sep 2026 12:18:17 +0200 Subject: [PATCH 1/2] Negotiate window scaling so a peer's window reaches past 65535 The SYN exchange settles a shift for each direction. A peer whose SYN carries a window scale option gets one back in the SYN-ACK, and every window it advertises afterwards is honoured shifted left by its count, limited to the 14 the RFC allows and the 1 GiB that implies. The stack's own shift is the smallest that expresses its read buffer in the 16-bit field, zero for the 16 KiB default, and every window it advertises is shifted down by it. The window field of a segment carrying SYN travels unscaled in both directions. A peer that offers no scaling leaves both shifts off and sees the raw 16-bit window it saw before. The options a segment carries are assembled where its flags are known, so create_raw_packet writes the list it is handed. --- src/stream/tcb.rs | 139 +++++++++++++++++++++++++++++++++++++++++--- src/stream/tcp.rs | 145 +++++++++++++++++++++++++++++++++++++++++----- 2 files changed, 261 insertions(+), 23 deletions(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 9446aaa..fa7f329 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,29 @@ 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(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 +238,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 +277,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 +382,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 +434,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 +476,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 +503,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 +529,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 +563,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 +588,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 +618,74 @@ 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); + + 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); + + 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..c9a275d 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(); @@ -920,7 +931,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 +989,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 +1023,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 +1041,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 +1051,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 = @@ -1095,6 +1117,8 @@ mod tests { let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -1123,4 +1147,93 @@ 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 the option 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!(window_scale_of(&syn_ack).is_some(), "the SYN-ACK carries no window scale option"); + + 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 peer opening with a closed window holds the writer until a later segment reopens it. + #[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); + + let mut cx = Context::from_waker(Waker::noop()); + assert!( + Pin::new(&mut stream).poll_write(&mut cx, b"held").is_pending(), + "a closed peer window let the write through" + ); + + feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]); + up_rx.recv().await.unwrap(); + assert_eq!(stream.tcb.lock().unwrap().get_send_window(), 40_000); + + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"sent"), Poll::Ready(Ok(4)))); + let packet = up_rx.recv().await.unwrap(); + assert_eq!(packet.payload.as_deref(), Some(&b"sent"[..])); + } } From 97f856bff610150359fe41908de607fdbcd6cc02 Mon Sep 17 00:00:00 2001 From: IntellyCode <47359527+IntellyCode@users.noreply.github.com> Date: Tue, 15 Sep 2026 10:21:05 +0200 Subject: [PATCH 2/2] Wake a writer held by a closed peer window and pin the SYN window A writer held by a closed peer window is woken by the segment that completes the handshake and by an established connection's data segments, as every other segment that can reopen the window already does. The window tests assert the SYN's window before any later segment replaces it, the handshake test asserts the shift the SYN-ACK carries, and the zero-window test uses a recording waker so it fails when a reopening segment leaves the writer asleep. A read buffer too large for any shift is limited to the maximum, with a warning. --- src/stream/tcb.rs | 7 +++++- src/stream/tcp.rs | 56 +++++++++++++++++++++++++++++++++++++---------- 2 files changed, 51 insertions(+), 12 deletions(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index fa7f329..ec73e7c 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -102,7 +102,10 @@ impl Tcb { let recv_window_shift = peer_window_shift.map(|_| { (0..=MAX_WINDOW_SHIFT) .find(|&shift| read_buffer_size >> shift <= u16::MAX as usize) - .unwrap_or(MAX_WINDOW_SHIFT) + .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(), @@ -623,6 +626,7 @@ mod tests { #[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); @@ -632,6 +636,7 @@ mod tests { #[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); diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index c9a275d..3e1c41d 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -738,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 { @@ -824,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"); @@ -1108,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::(); @@ -1185,11 +1199,16 @@ mod tests { }) } - /// A peer offering a shift of 7 gets the option back and has its windows honoured shifted by 7. + /// 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!(window_scale_of(&syn_ack).is_some(), "the SYN-ACK carries no window scale option"); + 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]); @@ -1216,24 +1235,39 @@ mod tests { assert_eq!(tcb.get_send_window(), 40_000 << 14); } - /// A peer opening with a closed window holds the writer until a later segment reopens it. + /// 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); - let mut cx = Context::from_waker(Waker::noop()); + // 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!( - Pin::new(&mut stream).poll_write(&mut cx, b"held").is_pending(), - "a closed peer window let the write through" + 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"[..])); - feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]); + // 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(); - assert_eq!(stream.tcb.lock().unwrap().get_send_window(), 40_000); - + 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)))); - let packet = up_rx.recv().await.unwrap(); - assert_eq!(packet.payload.as_deref(), Some(&b"sent"[..])); + assert_eq!(up_rx.recv().await.unwrap().payload.as_deref(), Some(&b"sent"[..])); } }