From 0382410f0286f390c2740f491b2bf7cc668c69a0 Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Fri, 10 Apr 2026 00:34:37 -0300 Subject: [PATCH] begin adding udp conn (#69) * begin adding udp conn * udp tests passing * bugfixes for udp Send,Abort,Encapsulate methods * document UDP methods --- udp/conn.go | 231 +++++++++++++++++++++++++++++++++++++++++++ udp/conn_test.go | 240 +++++++++++++++++++++++++++++++++++++++++++++ udp/definitions.go | 5 - udp/frame.go | 2 + udp/handler.go | 214 ++++++++++++++++++++++++++++++++++++++++ x/xnet/stack-go.go | 45 ++++++++- 6 files changed, 731 insertions(+), 6 deletions(-) create mode 100644 udp/conn.go create mode 100644 udp/conn_test.go delete mode 100644 udp/definitions.go create mode 100644 udp/handler.go diff --git a/udp/conn.go b/udp/conn.go new file mode 100644 index 0000000..bfc1881 --- /dev/null +++ b/udp/conn.go @@ -0,0 +1,231 @@ +package udp + +import ( + "net" + "os" + "sync" + "time" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/internal" +) + +var _ lneto.StackNode = (*Conn)(nil) + +// Conn implements a UDP datagram socket with SOCK_DGRAM semantics. +// Each [Conn.Write] enqueues one datagram and each [Conn.Read] dequeues one complete datagram. +type Conn struct { + mu sync.Mutex + h Handler + + remoteAddr []byte + + rdead time.Time + wdead time.Time + + ipID uint16 +} + +// ConnConfig configures a [Conn] or [Handler] with pre-allocated buffers and queue sizes. +type ConnConfig struct { + // RxBuf is the buffer for incoming datagrams. + RxBuf []byte + // TxBuf is the buffer for outgoing datagrams. + TxBuf []byte + // RxQueueSize is the maximum number of incoming datagrams that can be queued. + RxQueueSize int + // TxQueueSize is the maximum number of outgoing datagrams that can be queued. + TxQueueSize int +} + +// Configure initializes the connection with the given buffer and queue configuration. +// Must be called before [Conn.Open]. Calling Configure on an active connection aborts it. +func (conn *Conn) Configure(cfg ConnConfig) error { + conn.mu.Lock() + defer conn.mu.Unlock() + conn.abort() + err := conn.h.Configure(cfg) + if err != nil { + return err + } + return nil +} + +// Abort resets the connection, discarding all buffered data and clearing deadlines. +func (conn *Conn) Abort() { + conn.mu.Lock() + defer conn.mu.Unlock() + conn.h.Abort() + conn.abort() +} + +func (conn *Conn) abort() { + conn.rdead = time.Time{} + conn.wdead = time.Time{} + conn.remoteAddr = conn.remoteAddr[:0] +} + +// Open sets the local port and remote address for the connection. +func (conn *Conn) Open(localPort, remotePort uint16, remoteAddr []byte) error { + conn.mu.Lock() + defer conn.mu.Unlock() + conn.abort() + err := conn.h.SetPorts(localPort, remotePort) + if err != nil { + return err + } + conn.remoteAddr = append(conn.remoteAddr[:0], remoteAddr...) + return nil +} + +// LocalPort returns the local port set by [Conn.Open]. +func (conn *Conn) LocalPort() uint16 { + conn.mu.Lock() + defer conn.mu.Unlock() + return conn.h.lport +} + +// RemotePort returns the remote port set by [Conn.Open]. +func (conn *Conn) RemotePort() uint16 { return conn.h.rport } + +// RemoteAddr returns the remote address set by [Conn.Open]. +func (conn *Conn) RemoteAddr() []byte { + conn.mu.Lock() + defer conn.mu.Unlock() + return conn.remoteAddr +} + +// Protocol returns [lneto.IPProtoUDP]. +func (conn *Conn) Protocol() uint64 { return uint64(lneto.IPProtoUDP) } + +// ConnectionID returns a pointer to the connection ID. The value changes on +// each [Conn.Configure] or [Conn.Abort] call, signaling to the stack that the +// previous registration is no longer valid. +func (conn *Conn) ConnectionID() *uint64 { return &conn.h.connid } + +// Write enqueues a single datagram to be sent. The entire payload is queued atomically. +func (conn *Conn) Write(b []byte) (int, error) { + if len(b) == 0 { + return 0, nil + } + backoff := internal.NewBackoff(internal.BackoffTCPConn) + for { + if conn.deadlineExceeded(&conn.wdead) { + return 0, os.ErrDeadlineExceeded + } + conn.mu.Lock() + if conn.h.closeCalled { + conn.mu.Unlock() + return 0, net.ErrClosed + } + n, err := conn.h.Write(b) + conn.mu.Unlock() + if n > 0 { + return n, err + } + backoff.Miss() + } +} + +// Read dequeues a single datagram. If the buffer is smaller than the datagram, +// the remaining bytes are discarded (SOCK_DGRAM semantics). +func (conn *Conn) Read(b []byte) (int, error) { + backoff := internal.NewBackoff(internal.BackoffTCPConn) + for { + if conn.deadlineExceeded(&conn.rdead) { + return 0, os.ErrDeadlineExceeded + } + conn.mu.Lock() + if conn.h.closeCalled && conn.h.BufferedInput() == 0 { + conn.mu.Unlock() + return 0, net.ErrClosed + } + n, err := conn.h.Read(b) + conn.mu.Unlock() + if n > 0 { + return n, err + } + backoff.Miss() + } +} + +// Close marks the connection as closed. Subsequent calls to [Conn.Write] and +// [Conn.Demux] return [net.ErrClosed]. [Conn.Read] continues to return buffered +// data until exhausted, then returns [net.ErrClosed]. +func (conn *Conn) Close() error { + conn.mu.Lock() + defer conn.mu.Unlock() + conn.h.Close() + return nil +} + +// Demux receives an incoming UDP payload into the rx ring buffer. +func (conn *Conn) Demux(carrierData []byte, frameOffset int) error { + conn.mu.Lock() + defer conn.mu.Unlock() + if conn.h.closeCalled { + return net.ErrClosed + } + return conn.h.Recv(carrierData[frameOffset:]) +} + +// Encapsulate writes a queued outgoing datagram into the carrier buffer. +func (conn *Conn) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { + conn.mu.Lock() + defer conn.mu.Unlock() + if conn.h.closeCalled { + return 0, net.ErrClosed + } + n, err := conn.h.Send(carrierData[offsetToFrame:]) + if err != nil || n == 0 { + return 0, err + } + if offsetToIP >= 0 && len(conn.remoteAddr) > 0 { + err = internal.SetIPAddrs(carrierData[offsetToIP:], conn.ipID, nil, conn.remoteAddr) + if err != nil { + return 0, err + } + conn.ipID++ + } + return n, nil +} + +// SetDeadline sets both the read and write deadlines. A zero value disables the deadline. +func (conn *Conn) SetDeadline(t time.Time) error { + conn.mu.Lock() + defer conn.mu.Unlock() + if conn.h.closeCalled { + return net.ErrClosed + } + conn.rdead = t + conn.wdead = t + return nil +} + +// SetReadDeadline sets the read deadline. A zero value disables the deadline. +func (conn *Conn) SetReadDeadline(t time.Time) error { + conn.mu.Lock() + defer conn.mu.Unlock() + if conn.h.closeCalled { + return net.ErrClosed + } + conn.rdead = t + return nil +} + +// SetWriteDeadline sets the write deadline. A zero value disables the deadline. +func (conn *Conn) SetWriteDeadline(t time.Time) error { + conn.mu.Lock() + defer conn.mu.Unlock() + if conn.h.closeCalled { + return net.ErrClosed + } + conn.wdead = t + return nil +} + +func (conn *Conn) deadlineExceeded(deadline *time.Time) bool { + conn.mu.Lock() + defer conn.mu.Unlock() + return !deadline.IsZero() && time.Since(*deadline) > 0 +} diff --git a/udp/conn_test.go b/udp/conn_test.go new file mode 100644 index 0000000..2cab73c --- /dev/null +++ b/udp/conn_test.go @@ -0,0 +1,240 @@ +package udp + +import ( + "encoding/binary" + "testing" + + "github.com/soypat/lneto/internal" +) + +// makeUDPFrame builds a minimal UDP frame with the given ports and payload. +func makeUDPFrame(src, dst uint16, payload []byte) []byte { + buf := make([]byte, 8+len(payload)) + binary.BigEndian.PutUint16(buf[0:2], src) + binary.BigEndian.PutUint16(buf[2:4], dst) + binary.BigEndian.PutUint16(buf[4:6], uint16(8+len(payload))) + copy(buf[8:], payload) + return buf +} + +func newTestConn(t *testing.T) *Conn { + t.Helper() + var conn Conn + err := conn.Configure(ConnConfig{ + RxBuf: make([]byte, 256), + TxBuf: make([]byte, 256), + RxQueueSize: 4, + TxQueueSize: 4, + }) + if err != nil { + t.Fatal(err) + } + err = conn.Open(1234, 8080, []byte{10, 0, 0, 1}) + if err != nil { + t.Fatal(err) + } + return &conn +} + +func TestConn_WriteEncapsulateRoundtrip(t *testing.T) { + conn := newTestConn(t) + payload := []byte("hello udp") + n, err := conn.Write(payload) + if err != nil { + t.Fatal(err) + } + if n != len(payload) { + t.Fatalf("wrote %d, want %d", n, len(payload)) + } + + var buf [128]byte + n, err = conn.Encapsulate(buf[:], -1, 0) + if err != nil { + t.Fatal(err) + } + wantLen := 8 + len(payload) // UDP header + payload + if n != wantLen { + t.Fatalf("encapsulated %d, want %d", n, wantLen) + } + ufrm, err := NewFrame(buf[:n]) + if err != nil { + t.Fatal(err) + } + if !internal.BytesEqual(ufrm.Payload(), payload) { + t.Fatalf("encapsulated payload mismatch") + } + + // No more data pending. + n, err = conn.Encapsulate(buf[:], -1, 0) + if err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("expected no pending data, got %d", n) + } +} + +func TestConn_DemuxReadRoundtrip(t *testing.T) { + conn := newTestConn(t) + payload := []byte("incoming datagram") + frame := makeUDPFrame(8080, 1234, payload) // remote:8080 -> local:1234 + err := conn.Demux(frame, 0) + if err != nil { + t.Fatal(err) + } + + var buf [64]byte + n, err := conn.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + if n != len(payload) { + t.Fatalf("read %d, want %d", n, len(payload)) + } + if !internal.BytesEqual(buf[:n], payload) { + t.Fatal("read data mismatch") + } +} + +func TestConn_MultipleDatagrams(t *testing.T) { + conn := newTestConn(t) + messages := []string{"first", "second", "third"} + for _, msg := range messages { + frame := makeUDPFrame(8080, 1234, []byte(msg)) + err := conn.Demux(frame, 0) + if err != nil { + t.Fatal(err) + } + } + + // Read back in order. + var buf [64]byte + for _, want := range messages { + n, err := conn.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + got := string(buf[:n]) + if got != want { + t.Fatalf("got %q, want %q", got, want) + } + } +} + +func TestConn_ReadTruncates(t *testing.T) { + conn := newTestConn(t) + payload := []byte("a_longer_datagram") + frame := makeUDPFrame(8080, 1234, payload) + err := conn.Demux(frame, 0) + if err != nil { + t.Fatal(err) + } + + // Read into small buffer: truncates, discards remainder. + var buf [4]byte + n, err := conn.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + if n != len(buf) { + t.Fatalf("read %d, want %d", n, len(buf)) + } + if !internal.BytesEqual(buf[:], payload[:4]) { + t.Fatal("truncated data mismatch") + } + + // Next Demux+Read should work cleanly after truncation. + payload2 := []byte("ok") + frame2 := makeUDPFrame(8080, 1234, payload2) + err = conn.Demux(frame2, 0) + if err != nil { + t.Fatal(err) + } + var buf2 [64]byte + n, err = conn.Read(buf2[:]) + if err != nil { + t.Fatal(err) + } + if !internal.BytesEqual(buf2[:n], payload2) { + t.Fatal("post-truncation read mismatch") + } +} + +func TestConn_DemuxExhausted(t *testing.T) { + conn := newTestConn(t) // queue size 4 + for i := 0; i < 4; i++ { + frame := makeUDPFrame(8080, 1234, []byte{byte(i)}) + err := conn.Demux(frame, 0) + if err != nil { + t.Fatal(err) + } + } + // 5th should fail. + frame := makeUDPFrame(8080, 1234, []byte{0xff}) + err := conn.Demux(frame, 0) + if err == nil { + t.Fatal("expected error on exhausted rx queue") + } +} + +func TestConn_ClosedBehavior(t *testing.T) { + conn := newTestConn(t) + conn.Close() + + _, err := conn.Write([]byte("data")) + if err == nil { + t.Fatal("expected error writing to closed conn") + } + + err = conn.Demux([]byte("data"), 0) + if err == nil { + t.Fatal("expected error demuxing to closed conn") + } +} + +func TestConn_EncapsulateMultiple(t *testing.T) { + conn := newTestConn(t) + msgs := []string{"aaa", "bbb"} + for _, msg := range msgs { + _, err := conn.Write([]byte(msg)) + if err != nil { + t.Fatal(err) + } + } + + var buf [128]byte + for _, want := range msgs { + n, err := conn.Encapsulate(buf[:], -1, 0) + if err != nil { + t.Fatal(err) + } + ufrm, err := NewFrame(buf[:n]) + if err != nil { + t.Fatal(err) + } + got := string(ufrm.Payload()) + if got != want { + t.Fatalf("got %q, want %q", got, want) + } + } +} + +func TestConn_FrameOffset(t *testing.T) { + conn := newTestConn(t) + // Demux with an offset simulating IP header before the UDP frame. + udpFrame := makeUDPFrame(8080, 1234, []byte("hi")) + carrier := make([]byte, 8+len(udpFrame)) // 8 bytes of "IP header" prefix + copy(carrier[8:], udpFrame) + err := conn.Demux(carrier, 8) + if err != nil { + t.Fatal(err) + } + var buf [8]byte + n, err := conn.Read(buf[:]) + if err != nil { + t.Fatal(err) + } + if string(buf[:n]) != "hi" { + t.Fatalf("got %q, want %q", string(buf[:n]), "hi") + } +} diff --git a/udp/definitions.go b/udp/definitions.go deleted file mode 100644 index 8d00951..0000000 --- a/udp/definitions.go +++ /dev/null @@ -1,5 +0,0 @@ -package udp - -const ( - sizeHeader = 8 -) diff --git a/udp/frame.go b/udp/frame.go index fbb7d61..7f04655 100644 --- a/udp/frame.go +++ b/udp/frame.go @@ -6,6 +6,8 @@ import ( "github.com/soypat/lneto" ) +const sizeHeader = 8 + // NewFrame returns a new udp.Frame with data set to buf. // An error is returned if the buffer size is smaller than 8. // Users should still call [Frame.ValidateSize] before working diff --git a/udp/handler.go b/udp/handler.go new file mode 100644 index 0000000..39a63eb --- /dev/null +++ b/udp/handler.go @@ -0,0 +1,214 @@ +package udp + +import ( + "fmt" + "net" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/internal" +) + +// Handler implements the stateless UDP frame processing logic. It manages +// rx/tx ring buffers and datagram queues without locking or deadlines. +// [Conn] wraps Handler to provide a goroutine-safe socket API. +type Handler struct { + connid uint64 + rxRing internal.Ring + rxDgrams []struct { + length uint16 + } + + txRing internal.Ring + txDgrams []struct { + length uint16 + } + closeCalled bool + lport uint16 + rport uint16 +} + +// Configure initializes the handler with the given buffer and queue configuration. +// Increments the connection ID, invalidating any prior stack registration. +func (h *Handler) Configure(cfg ConnConfig) error { + if len(cfg.RxBuf) < sizeHeader || len(cfg.TxBuf) < sizeHeader || cfg.RxQueueSize <= 0 || cfg.TxQueueSize <= 0 { + return lneto.ErrInvalidConfig + } + h.connid++ + h.rxRing = internal.Ring{Buf: cfg.RxBuf} + h.txRing = internal.Ring{Buf: cfg.TxBuf} + internal.SliceReuse(&h.rxDgrams, cfg.RxQueueSize) + internal.SliceReuse(&h.txDgrams, cfg.TxQueueSize) + h.closeCalled = false + h.lport = 0 + h.rport = 0 + return nil +} + +// SetPorts sets the local and remote ports for the connection. +// Both ports must be non-zero. +func (h *Handler) SetPorts(localPort, remotePort uint16) error { + if localPort == 0 { + return lneto.ErrZeroSource + } else if remotePort == 0 { + return lneto.ErrZeroDestination + } + h.lport = localPort + h.rport = remotePort + return nil +} + +// LocalPort returns the local port set by [Handler.SetPorts]. +func (h *Handler) LocalPort() uint16 { + return h.lport +} + +// Recv parses a UDP frame from buf, validates the ports and length fields, +// and enqueues the payload into the rx ring buffer. Returns [lneto.ErrMismatch] +// if source/destination ports don't match the configured ports. +func (h *Handler) Recv(buf []byte) error { + if h.closeCalled { + return net.ErrClosed + } + ufrm, err := NewFrame(buf) + if err != nil { + return err + } else if ufrm.DestinationPort() != h.lport || ufrm.SourcePort() != h.rport { + return lneto.ErrMismatch + } + // Header size validation. + // No CRC validation at this level. + ul := ufrm.Length() + if ul < sizeHeader { + return lneto.ErrInvalidLengthField + } else if int(ul) > len(ufrm.RawData()) { + return lneto.ErrTruncatedFrame + } + + free := cap(h.rxDgrams) - len(h.rxDgrams) + if free == 0 { + return lneto.ErrExhausted + } + payload := ufrm.Payload() + _, err = h.rxRing.Write(payload) + if err != nil { + return err + } + dgram := internal.SliceReclaim(&h.rxDgrams) + dgram.length = uint16(len(payload)) + return nil +} + +// Send dequeues the next pending datagram and writes a complete UDP frame +// (header + payload) into buf. Returns 0, nil if no datagrams are queued. +func (h *Handler) Send(buf []byte) (int, error) { + if h.closeCalled { + return 0, net.ErrClosed + } else if len(h.txDgrams) == 0 { + return 0, nil + } + ufrm, err := NewFrame(buf) + if err != nil { + return 0, err + } + avail := len(buf) - 8 + if avail < int(h.txDgrams[0].length) { + return 0, lneto.ErrShortBuffer + } + dgram := internal.SliceDequeueFront(&h.txDgrams) + n, err := h.txRing.Read(buf[8 : 8+dgram.length]) + if err != nil || n != int(dgram.length) { + panic(fmt.Sprintf("udp send handler failure %d %s", n, err)) + } + ufrm.SetSourcePort(h.lport) + ufrm.SetDestinationPort(h.rport) + ufrm.SetLength(8 + dgram.length) + return int(8 + dgram.length), nil +} + +// Write enqueues a datagram payload for later transmission via [Handler.Send]. +// Returns [lneto.ErrExhausted] if the tx datagram queue is full. +func (h *Handler) Write(b []byte) (int, error) { + free := cap(h.txDgrams) - len(h.txDgrams) + if free == 0 { + return 0, lneto.ErrExhausted + } + _, err := h.txRing.Write(b) + if err != nil { + return 0, err + } + dgram := internal.SliceReclaim(&h.txDgrams) + dgram.length = uint16(len(b)) + return len(b), nil +} + +// Read dequeues the next received datagram into b. If b is smaller than the +// datagram, the remaining bytes are discarded (SOCK_DGRAM semantics). +// Returns 0, nil if no datagrams are available. +func (h *Handler) Read(b []byte) (int, error) { + if len(h.rxDgrams) == 0 { + return 0, nil + } + // SOCK_DGRAM semantics. Read up to len(b) bytes and discard unread portion of datagram. + dgram := internal.SliceDequeueFront(&h.rxDgrams) + n, err := h.rxRing.Read(b[:min(len(b), int(dgram.length))]) + if err != nil { + panic(fmt.Sprintf("udp read handler failure %d %s", n, err)) + } + discard := int(dgram.length) - len(b) + if discard > 0 { + err = h.rxRing.ReadDiscard(discard) + if err != nil { + panic(fmt.Sprintf("udp readdiscard handler failure %d %s", n, err)) + } + } + return n, nil +} + +// Close closes the connection. Calls to [Handler.Send] and [Handler.Recv] will +// return [net.ErrClosed] after Close is called. +func (h *Handler) Close() { + h.closeCalled = true +} + +// Abort resets the handler, discarding all buffered data and incrementing the +// connection ID. Buffers are retained for reuse. +func (h *Handler) Abort() { + *h = Handler{ + connid: h.connid + 1, + rxRing: h.rxRing, + rxDgrams: h.rxDgrams[:0], + txRing: h.txRing, + txDgrams: h.txDgrams[:0], + } + h.txRing.Reset() + h.rxRing.Reset() +} + +// BufferedInputNext returns the size of the next datagram to read. A call +// to [Handler.Read] will read up to this amount of bytes. +func (h *Handler) BufferedInputNext() int { + if len(h.rxDgrams) == 0 { + return 0 + } + return int(h.rxDgrams[0].length) +} + +// BufferedInput returns the number of unread bytes in the receive buffer. +func (h *Handler) BufferedInput() int { + return h.rxRing.Buffered() +} + +// BufferedUnsent returns the number of written but unsent bytes in the transmit buffer. +func (h *Handler) BufferedOutput() int { + return h.txRing.Buffered() +} + +// SizeInput returns the total size of the receive ring buffer. +func (h *Handler) SizeInput() int { + return h.rxRing.Size() +} + +// SizeOutput returns the total size of the transmit ring buffer. +func (h *Handler) SizeOutput() int { + return h.txRing.Size() +} diff --git a/x/xnet/stack-go.go b/x/xnet/stack-go.go index e78e16e..d1e50a7 100644 --- a/x/xnet/stack-go.go +++ b/x/xnet/stack-go.go @@ -9,11 +9,13 @@ import ( "github.com/soypat/lneto" "github.com/soypat/lneto/tcp" + "github.com/soypat/lneto/udp" ) // Socket types const ( sockSTREAM = 0x1 + sockDGRAM = 0x2 ) type StackGoConfig struct { @@ -81,7 +83,37 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype } switch network { case "udp", "udp4": - return nil, lneto.ErrUnsupported + if sotype != sockDGRAM { + return nil, lneto.ErrUnsupported + } + if !raddr.IsValid() || raddr.Addr() == netip.IPv4Unspecified() { + return nil, lneto.ErrZeroDestination + } + var conn udp.Conn + err = conn.Configure(udp.ConnConfig{ + TxBuf: make([]byte, s.plcfg.TxBufSize), + RxBuf: make([]byte, s.plcfg.RxBufSize), + TxQueueSize: s.plcfg.QueueSize, + RxQueueSize: s.plcfg.QueueSize, + }) + if err != nil { + return nil, err + } + raddr4 := raddr.Addr().As4() + err = conn.Open(laddr.Port(), raddr.Port(), raddr4[:]) + if err != nil { + return nil, err + } + err = s.blk.async.RegisterUDP(&conn, raddr4[:], raddr.Port()) + if err != nil { + return nil, err + } + uc := udpconn{ + Conn: &conn, + localAddr: net.UDPAddrFromAddrPort(laddr), + raddr: net.UDPAddrFromAddrPort(raddr), + } + return uc, nil case "tcp", "tcp4": if sotype != sockSTREAM { return nil, lneto.ErrUnsupported @@ -206,3 +238,14 @@ func (c tcpconn) RemoteAddr() net.Addr { Port: int(c.Conn.RemotePort()), } } + +type udpconn struct { + *udp.Conn + localAddr net.Addr + raddr net.Addr +} + +var _ net.Conn = udpconn{} + +func (c udpconn) LocalAddr() net.Addr { return c.localAddr } +func (c udpconn) RemoteAddr() net.Addr { return c.raddr }