From a967f083cbd5eef649696b413baa0a79c7929a92 Mon Sep 17 00:00:00 2001 From: soypat Date: Sun, 25 May 2025 18:30:55 -0300 Subject: [PATCH] add TCPConn Read+Write and ancilliary methods; looks like IP mux is broken --- examples/stackbasic/main.go | 20 ++++++ internal/backoff.go | 59 ++++++++++++++++ internet/basicstack.go | 1 + internet/tcpconn.go | 135 +++++++++++++++++++++++++++++++++++- tcp/handler.go | 19 ++++- 5 files changed, 232 insertions(+), 2 deletions(-) create mode 100644 internal/backoff.go diff --git a/examples/stackbasic/main.go b/examples/stackbasic/main.go index 433326e..eaa5704 100644 --- a/examples/stackbasic/main.go +++ b/examples/stackbasic/main.go @@ -14,9 +14,11 @@ import ( "github.com/soypat/lneto" "github.com/soypat/lneto/arp" "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/http/httpraw" "github.com/soypat/lneto/internal" "github.com/soypat/lneto/internal/ltesto" "github.com/soypat/lneto/internet" + "github.com/soypat/lneto/tcp" ) const ( @@ -57,6 +59,7 @@ func main() { fmt.Println("hosting server at ", addrPort.String()) var buf [mtu]byte + var hdr httpraw.Header for { nread, err := tap.Read(buf[:]) if err != nil { @@ -81,6 +84,23 @@ func main() { slogger.info("write", slog.Int("plen", nw)) } } + if handler.State() == tcp.StateEstablished { + data := handler.BufferedInput() + if data > 0 { + n, err := handler.Read(buf[:]) + if err != nil { + slogger.error("tcp-read", slog.String("err", err.Error())) + } else { + hdr.Reset(buf[:n]) + err = hdr.Parse(false) + if err != nil { + slogger.error("http-parse", slog.String("err", err.Error())) + } else { + fmt.Println(hdr.String()) + } + } + } + } if nread == 0 && nw == 0 { time.Sleep(5 * time.Millisecond) } diff --git a/internal/backoff.go b/internal/backoff.go new file mode 100644 index 0000000..ae58d97 --- /dev/null +++ b/internal/backoff.go @@ -0,0 +1,59 @@ +package internal + +import "time" + +type BackoffFlags uint8 + +const ( + BackoffHasPriority BackoffFlags = 1 << iota + BackoffCriticalPath +) + +func NewBackoff(priority BackoffFlags) Backoff { + if priority&BackoffCriticalPath != 0 { + return Backoff{ + maxWait: uint32(1 * time.Millisecond), + } + } + return Backoff{ + maxWait: uint32(time.Second) >> (priority & BackoffHasPriority), + } +} + +// A Backoff with a non-zero MaxWait is ready for use. +type Backoff struct { + // wait defines the amount of time that Miss will wait on next call. + wait uint32 + // Maximum allowable value for Wait. + maxWait uint32 + // startWait is the value that Wait takes after a call to Hit. + startWait uint32 + // expMinusOne is the shift performed on Wait minus one, so the zero value performs a shift of 1. + expMinusOne uint32 +} + +// Hit sets eb.Wait to the StartWait value. +func (eb *Backoff) Hit() { + if eb.maxWait == 0 { + panic("MaxWait cannot be zero") + } + eb.wait = eb.startWait +} + +// Miss sleeps for eb.Wait and increases eb.Wait exponentially. +func (eb *Backoff) Miss() { + const k = 1 + wait := eb.wait + maxWait := eb.maxWait + exp := eb.expMinusOne + 1 + if maxWait == 0 { + panic("MaxWait cannot be zero") + } + time.Sleep(time.Duration(wait)) + wait |= k + wait <<= exp + if wait > maxWait { + wait = maxWait + } + eb.wait = wait +} diff --git a/internet/basicstack.go b/internet/basicstack.go index 197661e..0b6f07b 100644 --- a/internet/basicstack.go +++ b/internet/basicstack.go @@ -63,6 +63,7 @@ func (sb *StackBasic) Recv(frame []byte) error { h := &sb.handlers[i] proto := ifrm.Protocol() if h.proto == proto { + sb.info("iprecv", slog.String("ipproto", proto.String()), slog.Int("plen", int(totalLen))) return h.recv(frame[:totalLen], off) } } diff --git a/internet/tcpconn.go b/internet/tcpconn.go index fe7456a..9b3a0ed 100644 --- a/internet/tcpconn.go +++ b/internet/tcpconn.go @@ -4,24 +4,35 @@ import ( "bytes" "errors" "log/slog" + "net" "net/netip" + "os" + "runtime" "time" + "github.com/soypat/lneto/internal" "github.com/soypat/lneto/ipv4" "github.com/soypat/lneto/ipv6" "github.com/soypat/lneto/tcp" ) +var ( + errDeadlineExceeded = os.ErrDeadlineExceeded +) + type TCPConn struct { h tcp.Handler remoteAddr []byte - logger rdead time.Time wdead time.Time lastTx time.Time lastRx time.Time + + abortErr error + logger } + type TCPConnConfig struct { RxBuf []byte TxBuf []byte @@ -80,6 +91,11 @@ func (conn *TCPConn) OpenListen(localPort uint16, iss tcp.Value) error { return nil } +func (conn *TCPConn) Close() error { + conn.trace("TCPConn.Close") + return conn.h.Close() +} + func (conn *TCPConn) RecvIP(buf []byte, off int) (err error) { conn.trace("tcpconn.Recv:start") if off >= len(buf) { @@ -102,6 +118,83 @@ func (conn *TCPConn) RecvIP(buf []byte, off int) (err error) { return nil } +// Write writes argument data to the TCPConns's output buffer which is queued to be sent. +func (conn *TCPConn) 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.BackoffHasPriority) + 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 *TCPConn) 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.BackoffHasPriority) + for conn.h.BufferedInput() == 0 && conn.State() == tcp.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 *TCPConn) checkPipeOpen() error { + if conn.abortErr != nil { + return conn.abortErr + } + state := conn.State() + if state.IsClosed() { + return net.ErrClosed + } + return nil +} + func (conn *TCPConn) HandleIP(buf []byte, off int) (n int, err error) { if len(conn.remoteAddr) == 0 { return 0, errors.New("unset IP address") @@ -184,3 +277,43 @@ func (conn *TCPConn) reset(h tcp.Handler) { 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 *TCPConn) 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 *TCPConn) 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 *TCPConn) SetWriteDeadline(t time.Time) error { + conn.trace("TCPConn.SetWriteDeadline:start") + err := conn.checkPipeOpen() + if err == nil { + conn.wdead = t + } + return err +} + +func (conn *TCPConn) deadlineExceeded(deadline time.Time) bool { + return !deadline.IsZero() && time.Since(deadline) > 0 +} diff --git a/tcp/handler.go b/tcp/handler.go index 397d816..234b2f3 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -30,7 +30,7 @@ type Handler struct { // connid is a conenction counter that is incremented each time a new // connection is established via Open calls. This disambiguate's whether // Read and Write calls belong to the current connection. - connid uint8 + connid uint16 closing bool } @@ -39,6 +39,11 @@ func (h *Handler) SetLoggers(handler, scb *slog.Logger) { h.scb.logger.log = scb } +// ConnectionID returns the connection identifier which is incremented every time the connection is closed or open. +func (h *Handler) ConnectionID() int { + return int(h.connid) +} + // State returns the state of the TCP state machine as per RFC9293. See [State]. func (h *Handler) State() State { return h.scb.State() } @@ -99,6 +104,13 @@ func (h *Handler) OpenListen(localPort uint16, iss Value) error { return nil } +// Abort forcibly terminates all state associated to current connection. +// After a call to abort no more data can be sent nor received over the connection. +func (h *Handler) Abort() { + h.scb.Abort() + h.reset(0, 0, 0) +} + func (h *Handler) reset(localPort, remotePort uint16, iss Value) { *h = Handler{ scb: h.scb, @@ -177,6 +189,11 @@ func (h *Handler) Recv(incomingPacket []byte) error { return nil } +func (h *Handler) Close() error { + h.trace("tcp.Handler.Close") + return h.scb.Close() +} + // Send writes TCP frame to be sent over the network to the remote peer to `b`. // It does no IP interfacing or CRC calculation of packet, which is left to the caller to perform. // The returned integer is the length written to the argument buffer.