mirror of
https://github.com/soypat/lneto.git
synced 2026-09-10 08:39:30 +00:00
begin adding udp conn (#69)
* begin adding udp conn * udp tests passing * bugfixes for udp Send,Abort,Encapsulate methods * document UDP methods
This commit is contained in:
+231
@@ -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
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,5 +0,0 @@
|
|||||||
package udp
|
|
||||||
|
|
||||||
const (
|
|
||||||
sizeHeader = 8
|
|
||||||
)
|
|
||||||
@@ -6,6 +6,8 @@ import (
|
|||||||
"github.com/soypat/lneto"
|
"github.com/soypat/lneto"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const sizeHeader = 8
|
||||||
|
|
||||||
// NewFrame returns a new udp.Frame with data set to buf.
|
// NewFrame returns a new udp.Frame with data set to buf.
|
||||||
// An error is returned if the buffer size is smaller than 8.
|
// An error is returned if the buffer size is smaller than 8.
|
||||||
// Users should still call [Frame.ValidateSize] before working
|
// Users should still call [Frame.ValidateSize] before working
|
||||||
|
|||||||
+214
@@ -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()
|
||||||
|
}
|
||||||
+44
-1
@@ -9,11 +9,13 @@ import (
|
|||||||
|
|
||||||
"github.com/soypat/lneto"
|
"github.com/soypat/lneto"
|
||||||
"github.com/soypat/lneto/tcp"
|
"github.com/soypat/lneto/tcp"
|
||||||
|
"github.com/soypat/lneto/udp"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Socket types
|
// Socket types
|
||||||
const (
|
const (
|
||||||
sockSTREAM = 0x1
|
sockSTREAM = 0x1
|
||||||
|
sockDGRAM = 0x2
|
||||||
)
|
)
|
||||||
|
|
||||||
type StackGoConfig struct {
|
type StackGoConfig struct {
|
||||||
@@ -81,7 +83,37 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
|
|||||||
}
|
}
|
||||||
switch network {
|
switch network {
|
||||||
case "udp", "udp4":
|
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":
|
case "tcp", "tcp4":
|
||||||
if sotype != sockSTREAM {
|
if sotype != sockSTREAM {
|
||||||
return nil, lneto.ErrUnsupported
|
return nil, lneto.ErrUnsupported
|
||||||
@@ -206,3 +238,14 @@ func (c tcpconn) RemoteAddr() net.Addr {
|
|||||||
Port: int(c.Conn.RemotePort()),
|
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 }
|
||||||
|
|||||||
Reference in New Issue
Block a user