Add round-robin implementation (#66)

* add handler.encapsulateNode

* implement round robin handler approach

* simplify round robin implementation

* leave TODO

* internet.node touch up

* fix conflict resolve f-up

* replace certain ErrShortBuffer with ErrTruncatedFrame error

arp/definitions.go

@@ -19,8 +19,7 @@ var (
 	errQueryNotFound = errors.New("arp: query not found")
 
 	// errGeneric aliases for common ARP errors.
-	errARPBufferFull  = lneto.ErrBufferFull
-	errShortARP       = lneto.ErrShortBuffer
+	errShortARP       = lneto.ErrTruncatedFrame
 	errARPUnsupported = lneto.ErrUnsupported
 	errLargeSizes     = lneto.ErrPacketDrop
 )

arp/frame.go

@@ -10,13 +10,13 @@ import (
 	"github.com/soypat/lneto/ethernet"
 )
 
-// NewARPFrame returns a ARPFrame with data set to buf.
+// NewFrame returns a Frame with data set to buf.
 // An error is returned if the buffer size is smaller than 28 (IPv4 min size).
-// Users should still call [ARPFrame.ValidateSize] before working
+// Users should still call [Frame.ValidateSize] before working
 // with payload/options of frames to avoid panics.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeaderv4 {
-		return Frame{buf: nil}, lneto.ErrShortBuffer
+		return Frame{buf: nil}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }

arp/handler.go

@@ -149,7 +149,7 @@ func (h *Handler) StartQuery(dstHWAddr, proto []byte) error {
 	if len(h.queries) == cap(h.queries) {
 		h.compactQueries()
 		if len(h.queries) == cap(h.queries) {
-			return lneto.ErrBufferFull
+			return lneto.ErrExhausted
 		}
 	}
 	if len(proto) != len(h.ourProtoAddr) {
@@ -214,7 +214,7 @@ func (h *Handler) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int)
 
 func (h *Handler) Demux(ethFrame []byte, frameOffset int) error {
 	if len(h.pendingResponse) == cap(h.pendingResponse) {
-		return errARPBufferFull
+		return lneto.ErrExhausted
 	}
 
 	b := ethFrame[frameOffset:]

dhcpv4/frame.go

@@ -26,7 +26,7 @@ const (
 // An error is returned if the buffer size is smaller than 240.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < OptionsOffset {
-		return Frame{}, lneto.ErrShortBuffer
+		return Frame{}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }
@@ -125,7 +125,7 @@ func (frm Frame) ForEachOption(fn func(off int, opt OptNum, data []byte) error)
 	// Parse DHCP options.
 	ptr := OptionsOffset
 	if ptr > len(frm.buf) {
-		return lneto.ErrShortBuffer
+		return lneto.ErrTruncatedFrame
 	} else if len(frm.buf[ptr:]) == 0 {
 		return lneto.ErrInvalidField
 	}

dns/definitions.go

@@ -16,7 +16,6 @@ var (
 	errNoNullTerm     = errors.New("DNS name missing null terminator")
 	errCalcLen        = errors.New("DNS calculated name label length exceeds remaining buffer length")
 	errCantAddLabel   = errors.New("long/empty/zterm/escape DNS label or not enough space")
-	errBaseLen        = lneto.ErrShortBuffer
 	errReserved       = errors.New("segment prefix is reserved")
 	errTooManyPtr     = errors.New("too many pointers (>10)")
 	errInvalidPtr     = errors.New("invalid pointer")
@@ -27,10 +26,10 @@ var (
 	errZeroSegLen     = errors.New("zero length segment")
 	errResTooLong     = errors.New("resource length too long")
 
-	errTooManyQuestions   = lneto.ErrBufferFull
-	errTooManyAnswers     = lneto.ErrBufferFull
-	errTooManyAuthorities = lneto.ErrBufferFull
-	errTooManyAdditionals = lneto.ErrBufferFull
+	errTooManyQuestions   = lneto.ErrExhausted
+	errTooManyAnswers     = lneto.ErrExhausted
+	errTooManyAuthorities = lneto.ErrExhausted
+	errTooManyAdditionals = lneto.ErrExhausted
 
 	errNonCanonicalName = errors.New("name is not in canonical format (it must end with a .)")
 	errStringTooLong    = errors.New("character string exceeds maximum length (255)")
@@ -49,7 +48,7 @@ type Frame struct {
 
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < SizeHeader {
-		return Frame{}, errBaseLen
+		return Frame{}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }

dns/dns.go

@@ -8,6 +8,7 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/internal"
 )
 
@@ -196,7 +197,7 @@ func skipQuestion(msg []byte, off uint16) (_ uint16, err error) {
 		return off, err
 	}
 	if off+4 > uint16(len(msg)) {
-		return off, errBaseLen
+		return off, lneto.ErrTruncatedFrame
 	}
 	return off + 4, nil
 }
@@ -210,7 +211,7 @@ func skipResource(msg []byte, off uint16) (_ uint16, err error) {
 	datalen := binary.BigEndian.Uint16(msg[off+8:])
 	off += datalen + 10
 	if off > uint16(len(msg)) {
-		return off, errBaseLen
+		return off, lneto.ErrTruncatedFrame
 	}
 	return off, nil
 }
@@ -610,7 +611,7 @@ func visitAllLabels(msg []byte, off uint16, fn func(b []byte), allowCompression
 LOOP:
 	for {
 		if currOff >= uint16(len(msg)) {
-			return off, errBaseLen
+			return off, lneto.ErrTruncatedFrame
 		}
 		c := uint16(msg[currOff])
 		currOff++

errors.go

@@ -21,6 +21,7 @@ const (
 	ErrInvalidLengthField                   // invalid length field
 	ErrExhausted                            // resource exhausted
 	ErrAlreadyRegistered                    // protocol already registered
+	ErrTruncatedFrame                       // truncated frame
 
 	// Below are potentially good future error additions
 	// based on one or two encountered use cases, example use case included.

ethernet/frame.go

@@ -12,7 +12,7 @@ import (
 // with payload/options of frames to avoid panics.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeaderNoVLAN {
-		return Frame{buf: nil}, lneto.ErrShortBuffer
+		return Frame{buf: nil}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }
@@ -122,6 +122,6 @@ func (efrm Frame) ValidateSize(v *lneto.Validator) {
 		v.AddError(lneto.ErrInvalidLengthField)
 	}
 	if sz == TypeVLAN && len(efrm.buf) < 18 {
-		v.AddError(lneto.ErrShortBuffer)
+		v.AddError(lneto.ErrTruncatedFrame)
 	}
 }

internet/definitions.go

@@ -11,24 +11,27 @@ import (
 
 // node is a concrete StackNode as stored in Stacks. Methods are devirtualized for performance benefits, especially on TinyGo.
 type node struct {
+	// currConnID stores the stack node *connID value on registration.
 	currConnID uint64
-	connID     *uint64
+	// connID is StackNode.ConnectionID() return value.
+	connID *uint64
 	// cbnode has different definitions in tinygo and normal Go compiled programs
 	// for performance and heap control reasons.
 	callbacks cbnode
-	// demux       func([]byte, int) error
-	// encapsulate func([]byte, int, int) (int, error)
-	proto uint16
-	port  uint16
 	// remoteAddr will be set on active(outbound) port connections
 	// that require an ARP to set the remoteAddr beforehand.
 	remoteAddr []byte
+	proto      uint16 // StackNode.Protocol()
+	lport      uint16 // StackNode.LocalPort()
 }
 
 type handlers struct {
+	nodes []node
+	// encapsIdx stores the index of next node to check for encapsulation.
+	encapsIdx int
+
 	context string
 	logger
-	nodes []node
 }
 
 func (h *handlers) reset(context string, maxNodes int) {
@@ -53,7 +56,7 @@ func (h *handlers) registerByPortProto(n node) error {
 	if err != nil {
 		return err
 	}
-	if h.nodeByPortProto(n.port, n.proto) != nil {
+	if h.nodeByPortProto(n.lport, n.proto) != nil {
 		return lneto.ErrAlreadyRegistered
 	}
 	h.nodes = append(h.nodes, n)
@@ -64,7 +67,7 @@ func (h *handlers) prepAdd() error {
 	if h.full() {
 		h.compact()
 		if h.full() {
-			return lneto.ErrBufferFull
+			return lneto.ErrExhausted
 		}
 	}
 	return nil
@@ -104,7 +107,7 @@ func (h *handlers) nodeByProto(proto uint16) *node {
 func (h *handlers) nodeByPort(port uint16) *node {
 	for i := range h.nodes {
 		node := &h.nodes[i]
-		if node.port == port && !node.IsInvalid() {
+		if node.lport == port && !node.IsInvalid() {
 			return node
 		}
 	}
@@ -114,7 +117,7 @@ func (h *handlers) nodeByPort(port uint16) *node {
 func (h *handlers) nodeByPortProto(port uint16, protocol uint16) *node {
 	for i := range h.nodes {
 		node := &h.nodes[i]
-		if node.port == port && node.proto == protocol && !node.IsInvalid() {
+		if node.lport == port && node.proto == protocol && !node.IsInvalid() {
 			return node
 		}
 	}
@@ -147,24 +150,37 @@ func (h *handlers) demuxByPort(buf []byte, offset int, port uint16) (*node, erro
 	return node, err
 }
 
+func (h *handlers) encapsulateNode(node *node, buf []byte, offsetIP, offsetThisFrame int) (n int, err error) {
+	if node.IsInvalid() {
+		return 0, nil
+	}
+	n, err = node.callbacks.Encapsulate(buf, offsetIP, offsetThisFrame)
+	if h.tryHandleError(node, err) {
+		err = nil  // CLOSE error handled gracefully by deleting node.
+		node = nil // Node is destroyed in tryHandleError and invalidated.
+	}
+	if n > 0 {
+		return n, err
+	} else if err != nil {
+		// Make sure not to hang on one handler that keeps returning an error.
+		h.error("handlers:encapsulate", slog.String("func", "encapsulateAny"), slog.String("ctx", h.context), slog.String("err", err.Error()))
+	}
+	return 0, nil
+}
+
 // encapsulateAny finds a node suitable to write and encapsulates the package.
 // If no data is sent it returns the last error encountered.
-func (h *handlers) encapsulateAny(buf []byte, offsetIP, offsetThisFrame int) (_ *node, n int, err error) {
-	for i := range h.nodes {
-		node := &h.nodes[i]
-		if node.IsInvalid() {
-			continue
-		}
-		n, err = node.callbacks.Encapsulate(buf, offsetIP, offsetThisFrame)
-		if h.tryHandleError(node, err) {
-			err = nil  // CLOSE error handled gracefully by deleting node.
-			node = nil // Node is destroyed in tryHandleError and invalidated.
-		}
-		if n > 0 {
-			return node, n, err
-		} else if err != nil {
-			// Make sure not to hang on one handler that keeps returning an error.
-			h.error("handlers:encapsulate", slog.String("func", "encapsulateAny"), slog.String("ctx", h.context), slog.String("err", err.Error()))
+func (h *handlers) encapsulateAny(buf []byte, offsetIP, offsetThisFrame int) (hn *node, n int, err error) {
+	// Round robin approach to encapsulation.
+	// TODO(soypat): benchmark impact of round robin. Consider removing fields from handlers to make it more lean and potentially get perf improvements that way.
+	i := h.encapsIdx
+	for range h.nodes {
+		hn := &h.nodes[i]
+		n, err = h.encapsulateNode(hn, buf, offsetIP, offsetThisFrame)
+		i = incLim(i, len(h.nodes))
+		if n > 0 || err != nil {
+			h.encapsIdx = i
+			return hn, n, err
 		}
 	}
 	return nil, 0, err // Return last written error.
@@ -196,7 +212,7 @@ func nodeFromStackNode(s lneto.StackNode, port uint16, protocol uint64, remoteAd
 		connID:     connIDPtr,
 		callbacks:  makecbnode(s),
 		proto:      uint16(protocol),
-		port:       port,
+		lport:      port,
 		remoteAddr: remoteAddr, // SHARED MEMORY- used to signal.
 	}
 }
@@ -205,3 +221,11 @@ func nodeFromStackNode(s lneto.StackNode, port uint16, protocol uint64, remoteAd
 func (n *node) destroy() {
 	*n = node{}
 }
+
+func incLim(v, max int) int {
+	v++
+	if v == max {
+		v = 0
+	}
+	return v
+}

internet/stack-udpport.go

@@ -24,7 +24,7 @@ func (sudp *StackUDPPort) SetStackNode(node lneto.StackNode, raddr []byte, rmpor
 
 func (sudp *StackUDPPort) Protocol() uint64 { return uint64(lneto.IPProtoUDP) }
 
-func (sudp *StackUDPPort) LocalPort() uint16 { return sudp.h.port }
+func (sudp *StackUDPPort) LocalPort() uint16 { return sudp.h.lport }
 
 func (sudp *StackUDPPort) ConnectionID() *uint64 { return sudp.h.connID }
 
@@ -42,7 +42,7 @@ func (sudp *StackUDPPort) Demux(carrierData []byte, frameOffset int) error {
 		return sudp.vld.ErrPop()
 	}
 	dst := ufrm.DestinationPort()
-	if dst != sudp.h.port {
+	if dst != sudp.h.lport {
 		return lneto.ErrPacketDrop // Not meant for us.
 	}
 	// TODO remote ip address handling.
@@ -70,7 +70,7 @@ func (sudp *StackUDPPort) Encapsulate(carrierData []byte, offsetToIP, offsetToFr
 	if err != nil {
 		return 0, err
 	}
-	ufrm.SetSourcePort(sudp.h.port)
+	ufrm.SetSourcePort(sudp.h.lport)
 	ufrm.SetDestinationPort(sudp.rmport)
 	if len(sudp.raddr) > 0 && offsetToIP >= 0 {
 		err = internal.SetIPAddrs(carrierData[offsetToIP:], 0, nil, sudp.raddr)

ipv4/frame.go

@@ -8,13 +8,13 @@ import (
 	"github.com/soypat/lneto"
 )
 
-// NewIPv4Frame returns a new IPv4Frame with data set to buf.
+// NewFrame returns a new [Frame] with data set to buf.
 // An error is returned if the buffer size is smaller than 20.
-// Users should still call [IPv4Frame.ValidateSize] before working
+// Users should still call [Frame.ValidateSize] before working
 // with payload/options of frames to avoid panics.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeader {
-		return Frame{buf: nil}, lneto.ErrShortBuffer
+		return Frame{buf: nil}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }
@@ -218,7 +218,7 @@ func (ifrm Frame) ValidateSize(v *lneto.Validator) {
 		v.AddError(lneto.ErrInvalidLengthField)
 	}
 	if int(tl) > len(ifrm.RawData()) {
-		v.AddError(lneto.ErrShortBuffer)
+		v.AddError(lneto.ErrTruncatedFrame)
 	}
 	if ihl < 5 || uint16(ihl)*4 > tl {
 		v.AddError(lneto.ErrInvalidLengthField)

ipv4/icmpv4/client.go

@@ -54,6 +54,7 @@ func (client *Client) Configure(cfg ClientConfig) error {
 	}
 	client.connid++
 	internal.SliceReuse(&client.outgoingEcho, cfg.ResponseQueueLimit)
+	internal.SliceReuse(&client.incomingEcho, cfg.ResponseQueueLimit)
 	client.responseRing = internal.Ring{Buf: cfg.ResponseQueueBuffer}
 	client.magic = cfg.HashSeed
 	client.id = cfg.ID
@@ -101,6 +102,10 @@ func (client *Client) Demux(carrierData []byte, frameOffset int) error {
 	}
 	switch tp {
 	case TypeEcho:
+		free := cap(client.incomingEcho) - len(client.incomingEcho)
+		if free == 0 {
+			return lneto.ErrExhausted
+		}
 		// We received a ping request; not handled client-side.
 		efrm := FrameEcho{Frame: ifrm}
 		data := efrm.Data()
@@ -227,6 +232,10 @@ func (client *Client) PingStart(remoteAddr [4]byte, pattern []byte, size uint16)
 	} else if remoteAddr == [4]byte{} {
 		return 0, lneto.ErrZeroDestination
 	}
+	free := cap(client.outgoingEcho) - len(client.outgoingEcho)
+	if free == 0 {
+		return 0, lneto.ErrExhausted
+	}
 	key = client.magichash(pattern, int(size)) & keyHashBits
 	v := internal.SliceReclaim(&client.outgoingEcho)
 	v.key = key

ipv4/icmpv4/icmpv4.go

@@ -59,7 +59,7 @@ const (
 
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeader {
-		return Frame{}, lneto.ErrShortBuffer
+		return Frame{}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }

ipv6/frame.go

@@ -6,13 +6,13 @@ import (
 	"github.com/soypat/lneto"
 )
 
-// NewIPv6Frame returns a new IPv6Frame with data set to buf.
+// NewFrame returns a new [Frame] with data set to buf.
 // An error is returned if the buffer size is smaller than 40.
-// Users should still call [IPv6Frame.ValidateSize] before working
+// Users should still call [Frame.ValidateSize] before working
 // with payload/options of frames to avoid panics.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeader {
-		return Frame{buf: nil}, lneto.ErrShortBuffer
+		return Frame{buf: nil}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }

ntp/ntp.go

@@ -27,7 +27,7 @@ const (
 
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < SizeHeader {
-		return Frame{buf: nil}, lneto.ErrShortBuffer
+		return Frame{buf: nil}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }

stringers.go

@@ -210,11 +210,12 @@ func _() {
 	_ = x[ErrInvalidLengthField-14]
 	_ = x[ErrExhausted-15]
 	_ = x[ErrAlreadyRegistered-16]
+	_ = x[ErrTruncatedFrame-17]
 }
 
-const _errGeneric_name = "lneto-bug(use build tag \"debugheaplog\")packet droppedincorrect checksumzero source(port/addr)zero destination(port/addr)short bufferbuffer fullinvalid addressunsupportedmismatchmismatched lengthinvalid configurationinvalid fieldinvalid length fieldresource exhaustedprotocol already registered"
+const _errGeneric_name = "lneto-bug(use build tag \"debugheaplog\")packet droppedincorrect checksumzero source(port/addr)zero destination(port/addr)short bufferbuffer fullinvalid addressunsupportedmismatchmismatched lengthinvalid configurationinvalid fieldinvalid length fieldresource exhaustedprotocol already registeredtruncated frame"
 
-var _errGeneric_index = [...]uint16{0, 39, 53, 71, 93, 120, 132, 143, 158, 169, 177, 194, 215, 228, 248, 266, 293}
+var _errGeneric_index = [...]uint16{0, 39, 53, 71, 93, 120, 132, 143, 158, 169, 177, 194, 215, 228, 248, 266, 293, 308}
 
 func (i errGeneric) String() string {
 	i -= 1

tcp/conn.go

@@ -329,7 +329,7 @@ func (conn *Conn) Demux(buf []byte, off int) (err error) {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
 	if off >= len(buf) {
-		return lneto.ErrShortBuffer
+		return lneto.ErrTruncatedFrame // TODO: this check is bad.
 	}
 	raddr, _, id, _, err := internal.GetIPAddr(buf[:off])
 	if err != nil {

tcp/frame.go

@@ -12,13 +12,13 @@ const (
 	sizeHeaderTCP = 20
 )
 
-// NewFrame returns a new TCPFrame with data set to buf.
+// NewFrame returns a new [Frame] with data set to buf.
 // An error is returned if the buffer size is smaller than 20.
 // Users should still call [Frame.ValidateSize] before working
 // with payload/options of frames to avoid panics.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeaderTCP {
-		return Frame{buf: nil}, lneto.ErrShortBuffer
+		return Frame{buf: nil}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }
@@ -190,7 +190,7 @@ func (tfrm Frame) ValidateSize(v *lneto.Validator) {
 		v.AddError(lneto.ErrInvalidLengthField)
 	}
 	if off > len(tfrm.RawData()) {
-		v.AddError(lneto.ErrShortBuffer)
+		v.AddError(lneto.ErrTruncatedFrame)
 	}
 }
 

tcp/options.go

@@ -111,13 +111,13 @@ func (op OptionCodec) ForEachOption(opts []byte, fn func(OptionKind, []byte) err
 			continue
 		}
 		if len(opts[off:]) < 1 {
-			return lneto.ErrShortBuffer
+			return lneto.ErrTruncatedFrame
 		}
 		size := int(opts[off]) // Total option length including kind and length bytes.
 		off++
 		dataLen := size - 2 // Data bytes after kind and length.
 		if dataLen < 0 || len(opts[off:]) < dataLen {
-			return lneto.ErrShortBuffer
+			return lneto.ErrTruncatedFrame
 		}
 
 		if !skipSizeValidation {

udp/frame.go

@@ -12,7 +12,7 @@ import (
 // with payload/options of frames to avoid panics.
 func NewFrame(buf []byte) (Frame, error) {
 	if len(buf) < sizeHeader {
-		return Frame{buf: buf}, lneto.ErrShortBuffer
+		return Frame{buf: buf}, lneto.ErrTruncatedFrame
 	}
 	return Frame{buf: buf}, nil
 }
@@ -97,6 +97,6 @@ func (ufrm Frame) ValidateSize(v *lneto.Validator) {
 		v.AddError(lneto.ErrInvalidLengthField)
 	}
 	if int(ul) > len(ufrm.RawData()) {
-		v.AddError(lneto.ErrShortBuffer)
+		v.AddError(lneto.ErrTruncatedFrame)
 	}
 }

x/xnet/stack-async.go

@@ -406,7 +406,7 @@ func (s *StackAsync) RegisterUDP(node lneto.StackNode, remoteAddr []byte, remote
 	defer s.mu.Unlock()
 	idx := len(s.userUDPs)
 	if idx >= cap(s.userUDPs) {
-		return lneto.ErrBufferFull
+		return lneto.ErrExhausted
 	}
 	s.userUDPs = s.userUDPs[:idx+1]
 	s.userUDPs[idx].SetStackNode(node, remoteAddr, remotePort)