Tcp buffer bug fix (#79)

* start working on tracking down tcp buffer bug

* mtu refactor

* modularize test

* more precise testing

* tests fail, but is it the failure we are looking for?

* fix typo in espradio link (#76)

* implement a new backoff abstraction (#75)

* rewrite backoff api

* rewrite tcp.Conn.Write

* keep fixing small things

* much better Conn.Read implementation

* fix critical overflow bug in internal.ConnRWBackoff

---------

Co-authored-by: Joel Wetzell <jwetzell@yahoo.com>

tcp/conn.go

@@ -2,11 +2,11 @@ package tcp
 
 import (
 	"errors"
+	"io"
 	"log/slog"
 	"net"
 	"net/netip"
 	"os"
-	"runtime"
 	"sync"
 	"time"
 
@@ -213,29 +213,34 @@ func (conn *Conn) Write(b []byte) (int, error) {
 	}
 	n := 0
 	backoffs := 0
-	for {
+	for len(b) > 0 {
 		if err := conn.checkPipe(connid, &conn.wdead); err != nil {
-			return 0, err
+			return n, err
 		}
 		conn.mu.Lock()
+		towrite := min(conn.h.FreeOutput(), len(b))
 		var ngot int
-		ngot, err = conn.h.Write(b)
-		conn.mu.Unlock()
-		n += ngot
-		b = b[ngot:]
-		if (err != nil && err != internal.ErrRingBufferFull) || n == plen {
-			break
-		} else if ngot > 0 {
+		if towrite > 0 {
+			ngot, err = conn.h.Write(b[:towrite])
+			conn.mu.Unlock()
+			if err != nil && err != internal.ErrRingBufferFull {
+				break
+			} else if ngot != towrite {
+				panic("unreachable")
+			}
+			n += ngot
+			b = b[ngot:]
 			backoffs = 0
-			runtime.Gosched() // Do a little yield since we won't have data for sure otherwise.
 		} else {
+			// No data can be written.
+			conn.mu.Unlock()
+			conn.trace("TCPConn.Write:insuf-buf", slog.Int("missing", plen-n), slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport)))
+			if conn.deadlineExceeded(&conn.wdead) {
+				return n, errDeadlineExceeded
+			}
 			backoffs++
 			conn.backoff(backoffs)
 		}
-		conn.trace("TCPConn.Write:insuf-buf", slog.Int("missing", plen-n), slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport)))
-		if conn.deadlineExceeded(&conn.wdead) {
-			return n, errDeadlineExceeded
-		}
 	}
 	return n, err
 }
@@ -260,38 +265,48 @@ func (conn *Conn) Flush() error {
 // Read will block until data is available or connection closes.
 // Returns io.EOF when the remote has closed the connection and all buffered data has been read.
 func (conn *Conn) Read(b []byte) (int, error) {
-	connid, err := conn.lockPipeConnID()
-	if err != nil {
-		if conn.BufferedInput() > 0 {
-			return conn.handlerRead(b) // Ensure remaining buffered data is read.
-		}
-		return 0, err
-	}
-	lport := conn.LocalPort()
-	rport := conn.RemotePort()
+	conn.mu.Lock()
+	connID := conn.h.connid
+	lport := conn.h.localPort
+	rport := conn.h.remotePort
+	conn.mu.Unlock()
 	conn.trace("TCPConn.Read:start", slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport)))
 	backoffs := 0
-	for conn.BufferedInput() == 0 {
-		state := conn.State()
-		if !state.RxDataOpen() {
-			// No use waiting for data, jump to read and return corresponding error from there.
-			break
-		} else if err := conn.checkPipe(connid, &conn.rdead); err != nil {
-			if conn.BufferedInput() > 0 {
-				return conn.handlerRead(b) // Ensure remaining buffered data is read.
-			}
-			return 0, err
+	n := 0
+	for len(b) > 0 {
+		conn.mu.Lock()
+		if connID != conn.h.connid {
+			conn.mu.Unlock()
+			return n, net.ErrClosed
+		}
+		avail := conn.h.BufferedInput()
+		if avail > 0 {
+			// Read branch.
+			ngot, err := conn.h.Read(b)
+			conn.mu.Unlock()
+			n += ngot
+			if err != nil {
+				return n, err
+			}
+			b = b[ngot:]
+		} else if n > 0 {
+			conn.mu.Unlock()
+			break
+		} else {
+			state := conn.h.State()
+			conn.mu.Unlock()
+			if state.IsClosed() {
+				return n, net.ErrClosed
+			} else if !state.RxDataOpen() {
+				return n, io.EOF
+			} else if conn.deadlineExceeded(&conn.rdead) {
+				return n, errDeadlineExceeded
+			}
+			backoffs++
+			conn.backoff(backoffs)
 		}
-		backoffs++
-		conn.backoff(backoffs)
 	}
-	return conn.handlerRead(b)
-}
-
-func (conn *Conn) handlerRead(b []byte) (int, error) {
-	conn.mu.Lock()
-	defer conn.mu.Unlock()
-	return conn.h.Read(b)
+	return n, nil
 }
 
 func (conn *Conn) lockPipeConnID() (uint64, error) {
@@ -452,7 +467,7 @@ func (conn *Conn) ConnectionID() *uint64 {
 
 func (conn *Conn) backoff(consecutiveBackoffs int) {
 	if conn._backoff != nil {
-		conn._backoff(consecutiveBackoffs)
+		conn._backoff.Do(consecutiveBackoffs)
 	} else {
 		internal.ConnRWBackoff(consecutiveBackoffs)
 	}

tcp/handler_test.go

@@ -5,10 +5,12 @@ import (
 	"fmt"
 	"math/rand"
 	"testing"
+
+	"github.com/soypat/lneto/ethernet"
 )
 
 func TestHandler(t *testing.T) {
-	const mtu = 1500
+	const mtu = ethernet.MaxMTU
 	const maxpackets = 3
 	rng := rand.New(rand.NewSource(0))
 	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
@@ -147,7 +149,7 @@ func establish(t *testing.T, client, server *Handler, packetBuf []byte) {
 // buffer for its SYN (advertising MSS=100), and the server should not send
 // segments with more than 100 bytes of payload.
 func TestHandler_MSSHonored(t *testing.T) {
-	const mtu = 1500
+	const mtu = ethernet.MaxMTU
 	rng := rand.New(rand.NewSource(0))
 	client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
 	setupClientServer(t, rng, client, server)
@@ -356,7 +358,7 @@ func TestTxBufferFreedOnACK(t *testing.T) {
 // 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 mtu = ethernet.MaxMTU
 	const maxpackets = 4
 	rng := rand.New(rand.NewSource(99))
 
@@ -459,7 +461,7 @@ func TestWindowUpdateAfterRead(t *testing.T) {
 // 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 mtu = ethernet.MaxMTU
 	const maxpackets = 4
 	rng := rand.New(rand.NewSource(77))
 
@@ -566,7 +568,7 @@ func TestWindowUpdateSWSAvoidance(t *testing.T) {
 //  6. Handler.Send() called again → same thing → AddPacket panics because
 //     off=0 but lastPkt.end=0 != bufsize
 func TestWriteAfterRemoteFIN(t *testing.T) {
-	const mtu = 1500
+	const mtu = ethernet.MaxMTU
 	const maxpackets = 3
 	rng := rand.New(rand.NewSource(11))
 	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
@@ -635,7 +637,7 @@ func TestWriteAfterRemoteFIN(t *testing.T) {
 // This is a regression test for a bug where RST segments in non-synchronized
 // states were blocked by errRequireSequential, causing connection pool leaks.
 func TestRSTinSynReceived(t *testing.T) {
-	const mtu = 1500
+	const mtu = ethernet.MaxMTU
 	const maxpackets = 3
 	rng := rand.New(rand.NewSource(2))
 	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
@@ -717,7 +719,7 @@ func TestRSTinSynReceived(t *testing.T) {
 //
 // The bug was that reset() cleared bufRx when state became CLOSED.
 func TestBufferNotClearedOnPassiveClose(t *testing.T) {
-	const mtu = 1500
+	const mtu = ethernet.MaxMTU
 	const maxpackets = 3
 	rng := rand.New(rand.NewSource(1))
 	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)
@@ -881,7 +883,7 @@ func TestBufferNotClearedOnPassiveClose(t *testing.T) {
 //  5. Handler.Send() called again → AddPacket panics because off=0 but
 //     lastPkt.end=0 != bufsize
 func TestChallengeACKWithBufferedData(t *testing.T) {
-	const mtu = 1500
+	const mtu = ethernet.MaxMTU
 	const maxpackets = 3
 	rng := rand.New(rand.NewSource(42))
 	client, server := newHandler(t, mtu, maxpackets), newHandler(t, mtu, maxpackets)