mega refactor internet.TCPConn->tcp.Conn

2026-09-09 08:19:09 +00:00 · 2025-06-07 20:43:20 -03:00
parent dab7951cbd
commit 9a60642283
9 changed files with 318 additions and 150 deletions
@@ -0,0 +1,295 @@
+package tcp
+
+import (
+	"bytes"
+	"errors"
+	"log/slog"
+	"net"
+	"net/netip"
+	"os"
+	"runtime"
+	"time"
+
+	"github.com/soypat/lneto"
+	"github.com/soypat/lneto/internal"
+)
+
+var (
+	errDeadlineExceeded = os.ErrDeadlineExceeded
+)
+
+// Conn builds on the [Handler] abstraction and adds IP header knowledge, time management, and familiar user facing API
+// like Write and Read methods.
+//
+// Note that the complete emulation of [net.TCPConn] at this level of abstraction is yet a non-goal,
+// even though the functionality provided is similar.
+type Conn struct {
+	h          Handler
+	remoteAddr []byte
+
+	rdead  time.Time
+	wdead  time.Time
+	lastTx time.Time
+	lastRx time.Time
+
+	ipID     uint16
+	abortErr error
+	logger
+}
+
+type ConnConfig struct {
+	RxBuf             []byte
+	TxBuf             []byte
+	TxPacketQueueSize int
+	Logger            *slog.Logger
+}
+
+func (conn *Conn) Configure(config *ConnConfig) (err error) {
+	err = conn.h.SetBuffers(config.TxBuf, config.RxBuf, config.TxPacketQueueSize)
+	if err != nil {
+		return err
+	}
+	conn.logger.log = config.Logger
+	return nil
+}
+
+// LocalPort returns the local port on which the socket is listening or connected to.
+func (conn *Conn) LocalPort() uint16 { return conn.h.LocalPort() }
+
+// RemotePort returns the port of the incoming remote connection. Is non-zero if connection is established.
+func (conn *Conn) RemotePort() uint16 { return conn.h.RemotePort() }
+
+func (conn *Conn) RemoteAddr() []byte { return conn.remoteAddr }
+
+// State returns the TCP state of the socket.
+func (conn *Conn) State() State { return conn.h.State() }
+
+// BufferedInput returns the number of bytes in the socket's receive/input buffer.
+func (conn *Conn) BufferedInput() int { return conn.h.BufferedInput() }
+
+// OpenActive opens a connection to a remote peer with a known IP address and port combination.
+// iss is the initial send sequence number which is ideally a random number which is far away from the last sequence number used on a connection to the same host.
+func (conn *Conn) OpenActive(remote netip.AddrPort, localPort uint16, iss Value) error {
+	err := conn.h.OpenActive(localPort, remote.Port(), iss)
+	if err != nil {
+		return err
+	}
+	conn.reset(conn.h)
+	raddr := remote.Addr()
+	if raddr.Is4() {
+		addr4 := raddr.As4()
+		conn.remoteAddr = append(conn.remoteAddr[:0], addr4[:]...)
+	} else if raddr.Is6() {
+		addr6 := raddr.As16()
+		conn.remoteAddr = append(conn.remoteAddr[:0], addr6[:]...)
+	}
+	return nil
+}
+
+// OpenListen opens a passive connection which listens for the first SYN packet to be received on a local port.
+// iss is the initial send sequence number which is usually a randomly chosen number.
+func (conn *Conn) OpenListen(localPort uint16, iss Value) error {
+	err := conn.h.OpenListen(localPort, iss)
+	if err != nil {
+		return err
+	}
+	conn.reset(conn.h)
+	return nil
+}
+
+func (conn *Conn) Close() error {
+	conn.trace("TCPConn.Close")
+	return conn.h.Close()
+}
+
+// Abort terminates all state of the connection forcibly.
+func (conn *Conn) Abort() {
+	conn.h.Abort()
+	*conn = Conn{
+		h:          conn.h,
+		remoteAddr: conn.remoteAddr[:0],
+		logger:     conn.logger,
+	}
+}
+
+func (conn *Conn) Demux(buf []byte, off int) (err error) {
+	conn.trace("tcpconn.Recv:start")
+	if off >= len(buf) {
+		return errors.New("bad offset in TCPConn.Recv")
+	}
+	raddr, id, err := internal.GetIPSourceAddr(buf[:off])
+	if err != nil {
+		return err
+	}
+	if conn.isRaddrSet() && !bytes.Equal(conn.remoteAddr, raddr) {
+		return errors.New("IP addr mismatch on TCPConn")
+	}
+	err = conn.h.Recv(buf[off:])
+	if err != nil {
+		return err
+	}
+	if !conn.isRaddrSet() && conn.h.RemotePort() != 0 {
+		conn.remoteAddr = append(conn.remoteAddr[:0], raddr...)
+		conn.ipID = ^(id - 1)
+	}
+	return nil
+}
+
+// Write writes argument data to the TCPConns's output buffer which is queued to be sent.
+func (conn *Conn) Write(b []byte) (int, error) {
+	err := conn.checkPipeOpen()
+	if err != nil {
+		return 0, err
+	}
+	plen := len(b)
+	conn.trace("TCPConn.Write:start")
+	connid := conn.h.ConnectionID()
+	if conn.deadlineExceeded(conn.wdead) {
+		return 0, errDeadlineExceeded
+	} else if plen == 0 {
+		return 0, nil
+	}
+	backoff := internal.NewBackoff(internal.BackoffTCPConn)
+	n := 0
+	for {
+		if conn.abortErr != nil {
+			return n, conn.abortErr
+		} else if connid != conn.h.ConnectionID() {
+			return n, net.ErrClosed
+		}
+		ngot, _ := conn.h.Write(b)
+		n += ngot
+		b = b[ngot:]
+		if n == plen {
+			break
+		} else if ngot > 0 {
+			backoff.Hit()
+			runtime.Gosched() // Do a little yield since we won't have data for sure otherwise.
+		} else {
+			backoff.Miss()
+		}
+		conn.trace("TCPConn.Write:insuf-buf", slog.Int("missing", plen-n))
+		if conn.deadlineExceeded(conn.wdead) {
+			return n, errDeadlineExceeded
+		}
+	}
+	return n, nil
+}
+
+// Read reads data from the socket's input buffer. If the buffer is empty,
+// Read will block until data is available or connection closes.
+func (conn *Conn) Read(b []byte) (int, error) {
+	err := conn.checkPipeOpen()
+	if err != nil {
+		return 0, err
+	}
+	conn.trace("TCPConn.Read:start")
+	connid := conn.h.ConnectionID()
+	backoff := internal.NewBackoff(internal.BackoffTCPConn)
+	for conn.h.BufferedInput() == 0 && conn.State() == StateEstablished {
+		if conn.abortErr != nil {
+			return 0, conn.abortErr
+		} else if connid != conn.h.ConnectionID() {
+			return 0, net.ErrClosed
+		}
+		if conn.deadlineExceeded(conn.rdead) {
+			return 0, errDeadlineExceeded
+		}
+		backoff.Miss()
+	}
+	n, err := conn.h.Read(b)
+	return n, err
+}
+
+func (conn *Conn) checkPipeOpen() error {
+	if conn.abortErr != nil {
+		return conn.abortErr
+	}
+	state := conn.State()
+	if state.IsClosed() {
+		return net.ErrClosed
+	}
+	return nil
+}
+
+func (conn *Conn) Encapsulate(buf []byte, off int) (n int, err error) {
+	if len(conn.remoteAddr) == 0 {
+		return 0, errors.New("unset IP address")
+	}
+	raddr, _, err := internal.GetIPSourceAddr(buf[:off])
+	if err != nil {
+		return 0, err
+	} else if len(raddr) != len(conn.remoteAddr) {
+		return 0, errors.New("mismatched IP version")
+	}
+	n, err = conn.h.Send(buf[off:])
+	if err != nil {
+		return 0, err
+	}
+	err = internal.SetIPDestinationAddr(buf[:off], conn.ipID, conn.remoteAddr)
+	if err != nil {
+		return 0, err
+	}
+	conn.ipID++
+	return n, nil
+}
+
+func (conn *Conn) Protocol() uint64 {
+	return uint64(lneto.IPProtoTCP)
+}
+
+func (conn *Conn) isRaddrSet() bool {
+	return len(conn.remoteAddr) != 0
+}
+
+func (conn *Conn) reset(h Handler) {
+	*conn = Conn{
+		h:          h,
+		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].
+func (conn *Conn) SetDeadline(t time.Time) error {
+	err := conn.SetReadDeadline(t)
+	if err != nil {
+		return err
+	}
+	return conn.SetWriteDeadline(t)
+}
+
+// SetReadDeadline sets the deadline for future Read calls
+// and any currently-blocked Read call. A zero value for t means Read will not time out.
+func (conn *Conn) SetReadDeadline(t time.Time) error {
+	conn.trace("TCPConn.SetReadDeadline:start")
+	err := conn.checkPipeOpen()
+	if err == nil {
+		conn.rdead = t
+	}
+	return err
+}
+
+// SetWriteDeadline sets the deadline for future Write calls
+// and any currently-blocked Write call.
+// Even if write times out, it may return n > 0, indicating that
+// some of the data was successfully written.
+// A zero value for t means Write will not time out.
+func (conn *Conn) SetWriteDeadline(t time.Time) error {
+	conn.trace("TCPConn.SetWriteDeadline:start")
+	err := conn.checkPipeOpen()
+	if err == nil {
+		conn.wdead = t
+	}
+	return err
+}
+
+func (conn *Conn) deadlineExceeded(deadline time.Time) bool {
+	return !deadline.IsZero() && time.Since(deadline) > 0
+}
+
+func (conn *Conn) ConnectionID() *uint64 {
+	return conn.h.ConnectionID()
+}
@@ -19,6 +19,10 @@ import (
 // A ControlBlock's internal state is modified by the available "System Calls" as defined in
 // RFC9293, such as Close, Listen/Open, Send, and Receive.
 // Sent and received data is represented with the [Segment] struct type.
+//
+// Note that [ControlBlock] is the lowest level implementation of TCP and as such is missing most useful functionality.
+// See [Handler], which uses ControlBlock, for a higher level implementation. [Conn] is an even higher level implementation
+// which makes use of a [Handler].
 type ControlBlock struct {
 	// # Send Sequence Space
 	//
@@ -20,6 +20,8 @@ var (
 // related to data buffering, frame sequencing and connection state handling.
 // Does NOT implement IP related logic, so no CRC calculation/validation or pseudo header logic.
 // Does NOT implement connection lifetime handling, so NO deadlines, keepalives, backoffs or anything that requires use of time package.
+//
+// See [Conn] for a higher level abstraction of a TCP connection, and see [ControlBlock] for the lower level bits of a TCP connection.
 type Handler struct {
 	connid uint64
 	scb    ControlBlock
@@ -117,10 +119,10 @@ func (h *Handler) Abort() {

 func (h *Handler) reset(localPort, remotePort uint16, iss Value) {
 	*h = Handler{
+		connid:     h.connid + 1,
 		scb:        h.scb,
 		bufTx:      h.bufTx,
 		bufRx:      h.bufRx,
-		connid:     h.connid + 1,
 		localPort:  localPort,
 		remotePort: remotePort,
 		validator:  h.validator,
@@ -300,7 +302,12 @@ func (h *Handler) BufferedInput() int {
 	return h.bufRx.Buffered()
 }

-// AwaitingSynResponse checks if the Handler is waiting for a Syn to arrive.
+// InUse returns true if the connection has been initialized and is being used to reach a remote port or if it is awaiting a remote packet.
+func (h *Handler) InUse() bool {
+	return h.remotePort != 0 || !h.State().IsClosed()
+}
+
+// AwaitingSynResponse returns true if the Handler is an active client opened with [Handler.OpenActive] and has already sent out the first SYN packet to the remote client.
 func (h *Handler) AwaitingSynResponse() bool {
 	return h.remotePort != 0 && h.scb.State() == StateSynSent
 }