mirror of
https://github.com/soypat/lneto.git
synced 2026-08-08 17:03:40 +00:00
add proactive window update (#39)
* add proactive window update * fix xnet tests due to runt padding bug
This commit is contained in:
@@ -323,6 +323,9 @@ func (h *Handler) Read(b []byte) (n int, err error) {
|
||||
if h.bufRx.Buffered() > 0 {
|
||||
n, err = h.bufRx.Read(b)
|
||||
}
|
||||
if n > 0 {
|
||||
h.maybeQueueWindowUpdate()
|
||||
}
|
||||
if n == 0 && err == nil {
|
||||
state := h.State()
|
||||
if state.IsClosed() {
|
||||
@@ -334,6 +337,26 @@ func (h *Handler) Read(b []byte) (n int, err error) {
|
||||
return n, err
|
||||
}
|
||||
|
||||
// maybeQueueWindowUpdate queues a window update ACK if the receive window has
|
||||
// opened significantly since it was last advertised. This prevents zero-window
|
||||
// deadlocks where the remote peer cannot send data because it thinks our window
|
||||
// is still 0 after we've Read() data from the buffer.
|
||||
//
|
||||
// Per RFC 9293 §3.8.6.2.2 (SWS avoidance), the window is updated when freed
|
||||
// space >= min(bufferSize/2, MSS). Since we don't track MSS, we use bufferSize/2.
|
||||
// Zero-window openings always trigger an update.
|
||||
func (h *Handler) maybeQueueWindowUpdate() {
|
||||
currentFree := Size(h.bufRx.Free())
|
||||
lastAdvertised := h.scb.RecvWindow()
|
||||
if currentFree <= lastAdvertised {
|
||||
return // Window hasn't grown.
|
||||
}
|
||||
bufSize := Size(h.bufRx.Size())
|
||||
if lastAdvertised == 0 || currentFree-lastAdvertised >= bufSize/2 {
|
||||
h.scb.pending[0] |= FlagACK
|
||||
}
|
||||
}
|
||||
|
||||
// BufferedInput returns amount of bytes buffered in receive(input) buffer and ready to read
|
||||
// with a [Handler.Read] call.
|
||||
func (h *Handler) BufferedInput() int {
|
||||
|
||||
@@ -267,6 +267,207 @@ func TestTxBufferFreedOnACK(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestWindowUpdateAfterRead verifies that after the application reads data from
|
||||
// a full receive buffer (Window=0), the TCP stack queues a window update ACK
|
||||
// so the remote peer can resume sending. This is a regression test for a
|
||||
// zero-window deadlock: without proactive window updates, the remote peer stays
|
||||
// stuck at Window=0 indefinitely after the app frees buffer space via Read().
|
||||
func TestWindowUpdateAfterRead(t *testing.T) {
|
||||
const rxBufSize = 256
|
||||
const mtu = 1500
|
||||
const maxpackets = 4
|
||||
rng := rand.New(rand.NewSource(99))
|
||||
|
||||
client := new(Handler)
|
||||
server := new(Handler)
|
||||
// Server gets a small RX buffer so we can fill it easily.
|
||||
err := client.SetBuffers(make([]byte, mtu), make([]byte, mtu), maxpackets)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = server.SetBuffers(make([]byte, mtu), make([]byte, rxBufSize), maxpackets)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = server.OpenListen(uint16(rng.Uint32()), 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = client.OpenActive(uint16(rng.Uint32()), server.LocalPort(), 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var rawbuf [mtu]byte
|
||||
establish(t, client, server, rawbuf[:])
|
||||
|
||||
// Fill the server's RX buffer completely (without reading).
|
||||
fillData := make([]byte, server.FreeRx())
|
||||
n, err := client.Write(fillData)
|
||||
if err != nil {
|
||||
t.Fatal("client write:", err)
|
||||
} else if n != len(fillData) {
|
||||
t.Fatal("short write")
|
||||
}
|
||||
clear(rawbuf[:])
|
||||
n, err = client.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal("client send:", err)
|
||||
}
|
||||
err = server.Recv(rawbuf[:n])
|
||||
if err != nil {
|
||||
t.Fatal("server recv:", err)
|
||||
}
|
||||
if server.FreeRx() != 0 {
|
||||
t.Fatalf("expected server RX buffer full, got %d free", server.FreeRx())
|
||||
}
|
||||
|
||||
// Server sends ACK — should advertise Window=0.
|
||||
clear(rawbuf[:])
|
||||
n, err = server.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal("server send ACK:", err)
|
||||
}
|
||||
if n == 0 {
|
||||
t.Fatal("expected server to send ACK for received data")
|
||||
}
|
||||
zeroWndFrm, _ := NewFrame(rawbuf[:n])
|
||||
if wnd := zeroWndFrm.WindowSize(); wnd != 0 {
|
||||
t.Fatalf("expected Window=0 in ACK, got %d", wnd)
|
||||
}
|
||||
|
||||
// Verify no pending segment before Read (nothing to send).
|
||||
clear(rawbuf[:])
|
||||
n, err = server.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatal("expected no pending segment before Read")
|
||||
}
|
||||
|
||||
// App reads ALL data from server, freeing the entire buffer.
|
||||
readBuf := make([]byte, rxBufSize)
|
||||
n, err = server.Read(readBuf)
|
||||
if err != nil {
|
||||
t.Fatal("server read:", err)
|
||||
}
|
||||
if n != len(fillData) {
|
||||
t.Fatalf("read %d, expected %d", n, len(fillData))
|
||||
}
|
||||
|
||||
// Server should now have a pending window update ACK.
|
||||
clear(rawbuf[:])
|
||||
n, err = server.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal("server send window update:", err)
|
||||
}
|
||||
if n == 0 {
|
||||
t.Fatal("BUG: no window update sent after Read() freed buffer space from Window=0")
|
||||
}
|
||||
wndFrm, _ := NewFrame(rawbuf[:n])
|
||||
if wnd := wndFrm.WindowSize(); wnd == 0 {
|
||||
t.Fatal("BUG: window update ACK still has Window=0")
|
||||
}
|
||||
t.Logf("window update sent: Window=%d (buffer free=%d)", wndFrm.WindowSize(), server.FreeRx())
|
||||
}
|
||||
|
||||
// TestWindowUpdateSWSAvoidance verifies that small reads that free less than
|
||||
// half the buffer do NOT trigger a window update (Silly Window Syndrome avoidance).
|
||||
func TestWindowUpdateSWSAvoidance(t *testing.T) {
|
||||
const rxBufSize = 256
|
||||
const mtu = 1500
|
||||
const maxpackets = 4
|
||||
rng := rand.New(rand.NewSource(77))
|
||||
|
||||
client := new(Handler)
|
||||
server := new(Handler)
|
||||
err := client.SetBuffers(make([]byte, mtu), make([]byte, mtu), maxpackets)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = server.SetBuffers(make([]byte, mtu), make([]byte, rxBufSize), maxpackets)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = server.OpenListen(uint16(rng.Uint32()), 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = client.OpenActive(uint16(rng.Uint32()), server.LocalPort(), 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var rawbuf [mtu]byte
|
||||
establish(t, client, server, rawbuf[:])
|
||||
|
||||
// Fill most of the server's RX buffer (leave a tiny amount free).
|
||||
fillSize := server.FreeRx() - 10
|
||||
fillData := make([]byte, fillSize)
|
||||
for i := range fillData {
|
||||
fillData[i] = byte(i)
|
||||
}
|
||||
n, err := client.Write(fillData)
|
||||
if err != nil {
|
||||
t.Fatal("client write:", err)
|
||||
} else if n != len(fillData) {
|
||||
t.Fatal("short write")
|
||||
}
|
||||
clear(rawbuf[:])
|
||||
n, err = client.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal("client send:", err)
|
||||
}
|
||||
err = server.Recv(rawbuf[:n])
|
||||
if err != nil {
|
||||
t.Fatal("server recv:", err)
|
||||
}
|
||||
|
||||
// Server sends ACK with small window.
|
||||
clear(rawbuf[:])
|
||||
n, err = server.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n == 0 {
|
||||
t.Fatal("expected ACK")
|
||||
}
|
||||
// Client receives the ACK so its send window is updated.
|
||||
err = client.Recv(rawbuf[:n])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// App reads a small amount (less than half the buffer).
|
||||
smallRead := make([]byte, rxBufSize/4)
|
||||
n, err = server.Read(smallRead)
|
||||
if err != nil {
|
||||
t.Fatal("server read:", err)
|
||||
}
|
||||
if n == 0 {
|
||||
t.Fatal("expected to read data")
|
||||
}
|
||||
|
||||
// Because freed space < bufSize/2, no window update should be queued.
|
||||
clear(rawbuf[:])
|
||||
n, err = server.Send(rawbuf[:])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Logf("NOTE: window update sent after small read (freed %d of %d buffer)", len(smallRead), rxBufSize)
|
||||
// This is acceptable if the threshold is met, but for SWS avoidance
|
||||
// we expect no update when the freed increment is < bufSize/2.
|
||||
freeAfterRead := Size(server.FreeRx())
|
||||
if freeAfterRead < Size(rxBufSize/2) {
|
||||
t.Fatalf("SWS violation: window update sent when free=%d < bufSize/2=%d", freeAfterRead, rxBufSize/2)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestBufferNotClearedOnPassiveClose tests that data remains readable after
|
||||
// the TCP connection is closed by the remote peer. This is a regression test
|
||||
// for a bug where the receive buffer was cleared when the connection transitioned
|
||||
|
||||
+18
-1
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/soypat/lneto/arp"
|
||||
"github.com/soypat/lneto/ethernet"
|
||||
"github.com/soypat/lneto/internet/pcap"
|
||||
"github.com/soypat/lneto/ipv4"
|
||||
"github.com/soypat/lneto/tcp"
|
||||
)
|
||||
|
||||
@@ -589,7 +590,23 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) {
|
||||
}
|
||||
|
||||
func (tst *tester) getTCPFrame() tcp.Frame {
|
||||
data := tst.getPayload(ethernet.TypeIPv4)
|
||||
tst.t.Helper()
|
||||
// Find the IP frame's position in the captured packet buffer.
|
||||
ipFrm := getProtoFrame(tst.frmbuf, ethernet.TypeIPv4)
|
||||
if ipFrm == nil {
|
||||
tst.t.Fatal("no IP frame in capture")
|
||||
}
|
||||
ipStart := ipFrm.PacketBitOffset / 8
|
||||
// Use IP TotalLength to correctly bound the frame, stripping any
|
||||
// Ethernet runt-frame padding (802.3 §3.2.7). This mirrors what
|
||||
// StackIP.Demux does before passing data to TCP.
|
||||
ifrm, err := ipv4.NewFrame(tst.buf[ipStart:])
|
||||
if err != nil {
|
||||
tst.t.Fatal("parsing IP frame:", err)
|
||||
}
|
||||
totalLen := int(ifrm.TotalLength())
|
||||
ihl := ifrm.HeaderLength()
|
||||
data := tst.buf[ipStart+ihl : ipStart+totalLen]
|
||||
frame, err := tcp.NewFrame(data)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
|
||||
Reference in New Issue
Block a user