do not rely on unspecified Go behaviour; fix TCP ring buffer bug (#16)

* do not rely on unspecified Go behaviour; fix TCP ring buffer bug

* even more stricter error handling on tcp connections

* stricter lock copying in tcp.Conn, even though likely not a problem

* add listener and tcp pool logging

* forgot reqAddr unused

* add logging to TCPConn and friends

tcp/conn.go

@@ -40,6 +40,19 @@ type Conn struct {
 	ipID uint16
 }
 
+// reset must be called while holding [Conn.mu].
+func (conn *Conn) reset(h Handler) {
+	// Reset fields individually - DO NOT copy the mutex (undefined behavior in Go).
+	// "A Mutex must not be copied after first use." - sync package docs.
+	// Copying a locked mutex causes corruption on multi-core systems.
+	conn.h = h
+	conn.remoteAddr = conn.remoteAddr[:0]
+	conn.rdead = time.Time{}
+	conn.wdead = time.Time{}
+	conn.abortErr = nil
+	conn.ipID = 0
+}
+
 type ConnConfig struct {
 	RxBuf             []byte
 	TxBuf             []byte
@@ -123,7 +136,8 @@ func (conn *Conn) OpenActive(localPort uint16, remote netip.AddrPort, iss Value)
 	if !remote.IsValid() {
 		return errInvalidIP
 	}
-	err := conn.h.OpenActive(localPort, remote.Port(), iss)
+	rport := remote.Port()
+	err := conn.h.OpenActive(localPort, rport, iss)
 	if err != nil {
 		return err
 	}
@@ -136,6 +150,7 @@ func (conn *Conn) OpenActive(localPort uint16, remote netip.AddrPort, iss Value)
 		addr6 := raddr.As16()
 		conn.remoteAddr = append(conn.remoteAddr[:0], addr6[:]...)
 	}
+	conn.debug("conn:dial", slog.Uint64("lport", uint64(localPort)), slog.Uint64("rport", uint64(rport)))
 	return nil
 }
 
@@ -149,13 +164,14 @@ func (conn *Conn) OpenListen(localPort uint16, iss Value) error {
 		return err
 	}
 	conn.reset(conn.h)
+	conn.debug("conn:listen", slog.Uint64("lport", uint64(localPort)))
 	return nil
 }
 
 func (conn *Conn) Close() error {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
-	conn.trace("TCPConn.Close")
+	conn.trace("TCPConn.Close", slog.Uint64("lport", uint64(conn.h.localPort)))
 	return conn.h.Close()
 }
 
@@ -163,14 +179,9 @@ func (conn *Conn) Close() error {
 func (conn *Conn) Abort() {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
+	conn.trace("TCPConn.Abort", slog.Uint64("lport", uint64(conn.h.localPort)))
 	conn.h.Abort()
-	*conn = Conn{
-		mu:         conn.mu,
-		h:          conn.h,
-		remoteAddr: conn.remoteAddr[:0],
-		logger:     conn.logger,
-		ipID:       conn.ipID,
-	}
+	conn.reset(conn.h)
 }
 
 // InternalHandler returns the internal [Handler] instance. The Handler contains lower level implementation logic for a TCP connection.
@@ -186,8 +197,10 @@ func (conn *Conn) Write(b []byte) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+	rport := conn.RemotePort()
 	plen := len(b)
-	conn.trace("TCPConn.Write:start")
+	lport := conn.LocalPort()
+	conn.trace("TCPConn.Write:start", slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport)))
 	if conn.deadlineExceeded(&conn.wdead) {
 		return 0, errDeadlineExceeded
 	} else if plen == 0 {
@@ -200,11 +213,12 @@ func (conn *Conn) Write(b []byte) (int, error) {
 			return 0, err
 		}
 		conn.mu.Lock()
-		ngot, _ := conn.h.Write(b)
+		var ngot int
+		ngot, err = conn.h.Write(b)
 		conn.mu.Unlock()
 		n += ngot
 		b = b[ngot:]
-		if n == plen {
+		if err != nil || n == plen {
 			break
 		} else if ngot > 0 {
 			backoff.Hit()
@@ -212,12 +226,12 @@ func (conn *Conn) Write(b []byte) (int, error) {
 		} else {
 			backoff.Miss()
 		}
-		conn.trace("TCPConn.Write:insuf-buf", slog.Int("missing", plen-n))
+		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, nil
+	return n, err
 }
 
 func (conn *Conn) Flush() error {
@@ -242,15 +256,22 @@ func (conn *Conn) Flush() error {
 
 // Read reads data from the socket's input buffer. If the buffer is empty,
 // 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 {
 		return 0, err
 	}
-	conn.trace("TCPConn.Read:start")
+	lport := conn.LocalPort()
+	rport := conn.RemotePort()
+	conn.trace("TCPConn.Read:start", slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport)))
 	backoff := internal.NewBackoff(internal.BackoffTCPConn)
-	for conn.BufferedInput() == 0 && conn.State() == StateEstablished {
-		if err := conn.checkPipe(connid, &conn.rdead); err != nil {
+	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 {
 			return 0, err
 		}
 		backoff.Miss()
@@ -281,7 +302,7 @@ func (conn *Conn) checkPipe(connID uint64, deadline *time.Time) (err error) {
 	} else if !deadline.IsZero() && time.Since(*deadline) > 0 {
 		err = errDeadlineExceeded
 	}
-	return nil
+	return err
 }
 
 func (conn *Conn) checkPipeOpen() error {
@@ -298,7 +319,6 @@ func (conn *Conn) checkPipeOpen() error {
 func (conn *Conn) Demux(buf []byte, off int) (err error) {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
-	conn.trace("tcpconn.Recv:start")
 	if off >= len(buf) {
 		return errors.New("bad offset in TCPConn.Recv")
 	}
@@ -309,6 +329,7 @@ func (conn *Conn) Demux(buf []byte, off int) (err error) {
 	if conn.isRaddrSet() && !bytes.Equal(conn.remoteAddr, raddr) {
 		return errors.New("IP addr mismatch on TCPConn")
 	}
+	conn.trace("tcpconn.Recv", slog.Uint64("lport", uint64(conn.h.LocalPort())), slog.Uint64("rport", uint64(conn.h.remotePort)))
 	err = conn.h.Recv(buf[off:])
 	if err != nil {
 		return err
@@ -336,6 +357,7 @@ func (conn *Conn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int)
 	} else if len(raddr) != len(conn.remoteAddr) {
 		return 0, errMismatchedIPVersion
 	}
+	conn.trace("TCPConn.encaps", slog.Uint64("lport", uint64(conn.h.LocalPort())), slog.Uint64("rport", uint64(conn.h.remotePort)))
 	n, err = conn.h.Send(carrierData[offsetToFrame:])
 	if err != nil {
 		return 0, err
@@ -356,18 +378,6 @@ func (conn *Conn) isRaddrSet() bool {
 	return len(conn.remoteAddr) != 0
 }
 
-func (conn *Conn) reset(h Handler) {
-	if conn.mu.TryLock() {
-		panic("reset must be called from within locked conn")
-	}
-	*conn = Conn{
-		h:          h,
-		mu:         conn.mu,
-		remoteAddr: conn.remoteAddr[:0],
-		logger:     conn.logger,
-	}
-}
-
 // SetDeadline sets the read and write deadlines associated
 // with the connection. It is equivalent to calling both
 // SetReadDeadline and SetWriteDeadline. Implements [net.Conn].

tcp/handler.go

@@ -304,20 +304,29 @@ func (h *Handler) SizeRx() int {
 // Write implements [io.Writer] by copying b to a internal buffer to be sent over the network on the next
 // [Handler.Send] call that can send data to remote peer. Use [Handler.Free] to know the maximum length the argument slice can be before erroring.
 func (h *Handler) Write(b []byte) (int, error) {
+	state := h.State()
 	if h.closing {
 		return 0, errConnectionClosing
-	} else if h.State().IsClosed() { // Reject write call if data cannot be sent.
+	} else if !state.TxDataOpen() { // Reject write call if data cannot be sent.
 		return 0, net.ErrClosed
 	}
 	return h.bufTx.Write(b)
 }
 
 // Read implements [io.Reader] by reading received data from remote peer in internal buffer.
-func (h *Handler) Read(b []byte) (int, error) {
-	if h.State().IsClosed() { // Reject read call if state is at StateClosed. Note this is less strict than Write call condition.
-		return 0, net.ErrClosed
+func (h *Handler) Read(b []byte) (n int, err error) {
+	if h.bufRx.Buffered() > 0 {
+		n, err = h.bufRx.Read(b)
 	}
-	return h.bufRx.Read(b)
+	if n == 0 && err == nil {
+		state := h.State()
+		if state.IsClosed() {
+			err = net.ErrClosed
+		} else if !state.RxDataOpen() {
+			err = io.EOF
+		}
+	}
+	return n, err
 }
 
 // BufferedInput returns amount of bytes buffered in receive(input) buffer and ready to read

tcp/listener.go

@@ -27,6 +27,22 @@ type Listener struct {
 	port       uint16
 	poolGet    func() (*Conn, Value)
 	poolReturn func(*Conn)
+	logger
+}
+
+func (listener *Listener) reset(port uint16, tcppool pool) {
+	listener.accepted = listener.accepted[:0]
+	listener.incoming = listener.incoming[:0]
+	listener.connID++
+	listener.port = port
+	listener.poolGet = tcppool.GetTCP
+	listener.poolReturn = tcppool.PutTCP
+}
+
+func (listener *Listener) SetLogger(logger *slog.Logger) {
+	listener.mu.Lock()
+	defer listener.mu.Unlock()
+	listener.logger.log = logger
 }
 
 // LocalPort implements [StackNode].
@@ -48,6 +64,7 @@ func (listener *Listener) Close() error {
 	if listener.isClosed() {
 		return errors.New("already closed")
 	}
+	listener.debug("listener:reset", slog.Uint64("port", uint64(listener.port)))
 	listener.connID++
 	listener.port = 0
 	return nil
@@ -61,15 +78,8 @@ func (listener *Listener) Reset(port uint16, pool pool) error {
 	}
 	listener.mu.Lock()
 	defer listener.mu.Unlock()
-	*listener = Listener{
-		mu:         listener.mu,
-		connID:     listener.connID + 1,
-		port:       port,
-		poolGet:    pool.GetTCP,
-		poolReturn: pool.PutTCP,
-		incoming:   listener.incoming[:0],
-		accepted:   listener.accepted[:0],
-	}
+	listener.debug("listener:reset", slog.Uint64("port", uint64(port)))
+	listener.reset(port, pool)
 	return nil
 }
 
@@ -95,6 +105,7 @@ func (listener *Listener) TryAccept() (*Conn, error) {
 	if listener.isClosed() {
 		return nil, net.ErrClosed
 	}
+	listener.debug("listener:tryaccept", slog.Uint64("port", uint64(listener.port)))
 	listener.maintainConns()
 	for i, conn := range listener.incoming {
 		if conn == nil || conn.State() != StateEstablished {
@@ -114,6 +125,7 @@ func (listener *Listener) Encapsulate(carrierData []byte, offsetToIP, offsetToFr
 	if listener.isClosed() {
 		return 0, net.ErrClosed
 	}
+	//listener.trace("listener:encaps", slog.Uint64("port", uint64(listener.port)))
 	// First try incoming connections (for handshake SYN-ACK).
 	for i, conn := range listener.incoming {
 		if conn == nil || conn.State() == StateEstablished {
@@ -127,6 +139,7 @@ func (listener *Listener) Encapsulate(carrierData []byte, offsetToIP, offsetToFr
 		if n == 0 {
 			continue
 		}
+		listener.debug("listener:encaps", slog.Uint64("port", uint64(listener.port)), slog.Int("plen", n), slog.String("list", "incoming"))
 		return n, err
 	}
 	// Then try accepted connections.
@@ -141,6 +154,7 @@ func (listener *Listener) Encapsulate(carrierData []byte, offsetToIP, offsetToFr
 		if n == 0 {
 			continue
 		}
+		listener.debug("listener:encaps", slog.Uint64("port", uint64(listener.port)), slog.Int("plen", n), slog.String("list", "accepted"))
 		return n, err
 	}
 	return 0, nil
@@ -166,15 +180,19 @@ func (listener *Listener) Demux(carrierData []byte, tcpFrameOffset int) error {
 		return errors.New("not our port")
 	}
 	src := tfrm.SourcePort()
+
 	// Try to demux in accepted:
+	accepted := true
 	demuxed, err := listener.tryDemux(listener.accepted, src, srcaddr, carrierData, tcpFrameOffset)
+	if !demuxed {
+		accepted = false
+		demuxed, err = listener.tryDemux(listener.incoming, src, srcaddr, carrierData, tcpFrameOffset)
+	}
 	if demuxed {
+		listener.debug("tcplistener:demux", slog.Uint64("lport", uint64(listener.port)), slog.Uint64("rport", uint64(src)), slog.Bool("accepted", accepted))
 		return err
 	}
-	demuxed, err = listener.tryDemux(listener.incoming, src, srcaddr, carrierData, tcpFrameOffset)
-	if demuxed {
-		return err
-	}
+
 	// Connection not in ready nor accepted.
 	_, flags := tfrm.OffsetAndFlags()
 	if flags != FlagSYN {
@@ -198,6 +216,7 @@ func (listener *Listener) Demux(carrierData []byte, tcpFrameOffset int) error {
 		return lneto.ErrPacketDrop
 	}
 	listener.incoming = append(listener.incoming, conn)
+	listener.debug("tcplistener:demux-new", slog.Uint64("lport", uint64(listener.port)), slog.Uint64("rport", uint64(src)))
 	return nil
 }
 
@@ -223,7 +242,8 @@ func (listener *Listener) maintainConns() {
 		if listener.incoming[i] == nil {
 			continue
 		}
-		if listener.incoming[i].State() > StateEstablished || listener.incoming[i].State().IsClosed() {
+		state := listener.incoming[i].State()
+		if state > StateEstablished || state.IsClosed() {
 			// Something went wrong in handshake or pool aborted/closed the connection.
 			listener.poolReturn(listener.incoming[i])
 			listener.incoming[i] = nil

tcp/txqueue.go

@@ -2,7 +2,6 @@ package tcp
 
 import (
 	"errors"
-	"slices"
 
 	"github.com/soypat/lneto/internal"
 )
@@ -63,6 +62,7 @@ func (rtx *ringTx) Reset(buf []byte, maxqueuedPackets int, iss Value) error {
 
 	*rtx = ringTx{
 		rawbuf: buf,
+		slist:  rtx.slist,
 	}
 	rtx.slist.Reset(maxqueuedPackets, iss)
 	rtx.iss = iss
@@ -260,8 +260,11 @@ type sentlist struct {
 	pkts []ringidx
 }
 
+// Reset clears the sent packet list and prepares it for reuse.
+// The packet queue capacity is set to exactly pktQueueSize.
+// The initial sequence number is set to iss.
 func (sl *sentlist) Reset(pktQueueSize int, iss Value) {
-	sl.pkts = slices.Grow(sl.pkts[:0], pktQueueSize)
+	internal.SliceReuse(&sl.pkts, pktQueueSize)
 	sl.ssn = iss
 }