From 99e9d90a6075f2a1e6f0dadfdf64bceaa6aa69fc Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Thu, 9 Jul 2026 14:04:10 -0300 Subject: [PATCH] Netdev revamp (#145) * begin working on netdev solution * need to roll back some assumptions in next commit * dual poll/async mode for netdev.Runner * add wake on rx semantics * remove TODO * more reworking of Runner * work on applying @MDr164 suggestions and a couple extra revamps * add newline to end of test file --- .../_import_examples/espradio-netdev/main.go | 12 +- examples/_import_examples/picow-netdev/go.mod | 11 +- examples/_import_examples/picow-netdev/go.sum | 4 - .../_import_examples/picow-netdev/main.go | 30 +- x/netdev/buffer.go | 146 ++++ x/netdev/interface.go | 57 +- x/netdev/runner.go | 495 +++++++++++--- x/netdev/runner_test.go | 644 ++++++++++++++++++ x/xnet/stack-async.go | 6 +- 9 files changed, 1288 insertions(+), 117 deletions(-) create mode 100644 x/netdev/buffer.go create mode 100644 x/netdev/runner_test.go diff --git a/examples/_import_examples/espradio-netdev/main.go b/examples/_import_examples/espradio-netdev/main.go index ac85a7e..8ef7308 100644 --- a/examples/_import_examples/espradio-netdev/main.go +++ b/examples/_import_examples/espradio-netdev/main.go @@ -83,7 +83,17 @@ func main() { var runner netdev.Runner[espradio.STAConfig] go func() { - if err := runner.Run(context.Background(), iface, &stack, backoff); err != nil { + // EthPoll drains the C ring buffer and delivers frames through the + // receive handler, so the device is async but still needs the pump. + err := runner.Configure(netdev.RunnerConfig[espradio.STAConfig]{ + Buffers: iface.RunnerBuffers(2), + Backoff: backoff, + Flags: netdev.RunnerInterfaceAsync | netdev.RunnerInterfacePoll, + }) + if err != nil { + failIfErr("runnerconfig", err) + } + if err := runner.Run(context.Background(), &iface, &stack); err != nil { failIfErr("runner", err) } }() diff --git a/examples/_import_examples/picow-netdev/go.mod b/examples/_import_examples/picow-netdev/go.mod index cd78555..4bd4918 100644 --- a/examples/_import_examples/picow-netdev/go.mod +++ b/examples/_import_examples/picow-netdev/go.mod @@ -2,10 +2,12 @@ module piconetdev go 1.25.7 -require github.com/soypat/cyw43439 v0.1.1 +require ( + github.com/soypat/cyw43439 v0.1.1 + github.com/soypat/lneto v0.1.1-0.20260425023453-aa77403a2b32 +) require ( - github.com/soypat/lneto v0.1.0 // indirect github.com/soypat/seqs v0.0.0-20250124201400-0d65bc7c1710 // indirect github.com/tinygo-org/pio v0.2.0 // indirect golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa // indirect @@ -13,4 +15,7 @@ require ( // This is an example taken grom github.com/soypat/lneto // Remove this replace directive when using as own program. -replace github.com/soypat/lneto => ../../../. \ No newline at end of file +replace github.com/soypat/lneto => ../../../. + +// Local cyw43439 with the poll-based EthPoll API. +replace github.com/soypat/cyw43439 => ../../../../cyw43439 diff --git a/examples/_import_examples/picow-netdev/go.sum b/examples/_import_examples/picow-netdev/go.sum index 5dddaa6..fdd278c 100644 --- a/examples/_import_examples/picow-netdev/go.sum +++ b/examples/_import_examples/picow-netdev/go.sum @@ -1,7 +1,3 @@ -github.com/soypat/cyw43439 v0.1.1 h1:vcaTiVzfuz3keK7lJpVxStZ6tV8HCw7Ugzsh1k4mneE= -github.com/soypat/cyw43439 v0.1.1/go.mod h1:R2uSILRwSPmcmmKy5Z0FtK4ypgiPf5YqK+F+IKmXqxc= -github.com/soypat/lneto v0.1.0 h1:VAHCJ33hvC3wDqhM0Vm7w0k6vwNsOCAsQ8XTrXJpS7I= -github.com/soypat/lneto v0.1.0/go.mod h1:g/8Lk+hIsMZydyWDJjK2YfsCuG6jA5mWCO6U+4S7w1U= github.com/soypat/seqs v0.0.0-20250124201400-0d65bc7c1710 h1:Y9fBuiR/urFY/m76+SAZTxk2xAOS2n85f+H1CugajeA= github.com/soypat/seqs v0.0.0-20250124201400-0d65bc7c1710/go.mod h1:oCVCNGCHMKoBj97Zp9znLbQ1nHxpkmOY9X+UAGzOxc8= github.com/tinygo-org/pio v0.2.0 h1:vo3xa6xDZ2rVtxrks/KcTZHF3qq4lyWOntvEvl2pOhU= diff --git a/examples/_import_examples/picow-netdev/main.go b/examples/_import_examples/picow-netdev/main.go index 62adcfb..1218ee4 100644 --- a/examples/_import_examples/picow-netdev/main.go +++ b/examples/_import_examples/picow-netdev/main.go @@ -70,14 +70,25 @@ func main() { failIfErr("init iface", err) go func() { - if err := runner.Run(context.Background(), iface, &stack, backoff); err != nil { + err := runner.Configure(netdev.RunnerConfig[ConnectParams]{ + Buffers: iface.RunnerBuffers(1), + Backoff: backoff, + Flags: netdev.RunnerInterfaceAsync | netdev.RunnerInterfacePoll, + }) + if err != nil { + failIfErr("runnerconfig", err) + } + if err := runner.Run(context.Background(), &iface, &stack); err != nil { failIfErr("runner", err) } }() assigned, gatewayRt, subnetBits, err := stack.EnableDHCP(context.Background(), true, netip.Addr{}) failIfErr("enable dhcp", err) println("assigned=", assigned.String(), "gateway=", gatewayRt.String(), "subnet", subnetBits) - select {} + for { + runner.PrintDebug() + time.Sleep(2 * time.Second) + } } // compile-time guarantee of interface implementation. @@ -142,23 +153,24 @@ func (d *Netdev) SendOffsetEthFrame(offsetTxEthFrame []byte) error { } // SetRecvHandler implements [netdev.DevEthernet]. +// +// The cyw43439 delivers received Ethernet frames through this callback whenever +// the bus is serviced (including ioctl/control transactions outside EthPoll), so +// the runner must drive it in async mode. EthPoll is still used to pump the device. func (d *Netdev) SetEthRecvHandler(handler func(rxEthframe []byte)) { - d.dev.RecvEthHandle(func(pkt []byte) error { - handler(pkt) - return nil - }) + d.dev.RecvEthHandle(handler) } // EthPoll implements [netdev.DevEthernet]. func (d *Netdev) EthPoll(buf []byte) (ethFrameOff, ethernetBytes int, err error) { - _, err = d.dev.PollOne() - return 0, 0, err + return d.dev.EthPoll(buf) } // MaxFrameSizeAndOffset implements [netdev.DevEthernet]. func (d *Netdev) MaxFrameSizeAndOffset() (maxFrameSize int, frameOff int) { - return cyw43439.MaxFrameSize, 0 + return 2048, 0 } + func failIfErr(msg string, err error) { if err != nil { fail(msg, err) diff --git a/x/netdev/buffer.go b/x/netdev/buffer.go new file mode 100644 index 0000000..ebfdbed --- /dev/null +++ b/x/netdev/buffer.go @@ -0,0 +1,146 @@ +package netdev + +import ( + "sync/atomic" + + "github.com/soypat/lneto/internal" +) + +// TODO(soypat): True Zero Copy (TZC) +// TODO(soypat): TZC acheived on redesigning [DevEthernet] to not own any buffers and ask the networking stack for buffers in the callback path. TZC already acheived for Polling path. +// TODO(soypat): TZC in callback path requires redesign of bufferSelect and [Runner] likely. + +// lenClaimed marks a slot claimed by putRx before its frame copy completes. +// The slot is published by storing the frame length, which must be the +// claimant's last write so getRx never observes a partially copied frame. +const lenClaimed = -1 + +// bufferSelect is a fixed pool of frame buffers shared between the runner +// goroutine and the device's receive handler goroutine. Slot ownership is +// arbitrated exclusively through CAS on lenAcquire: +// - 0: slot free. +// - lenClaimed(<0): slot claimed by putRx, frame copy in progress. +// - n>0: slot owned; if isRx is set the slot holds a published Rx frame. +// +// Only goroPutRx may be called concurrently with the other methods; all other +// methods must be called from a single goroutine (the runner's). +type bufferSelect struct { + // nextSeq generates arrival-order sequence numbers for Rx frames so getRx + // yields frames in the order they were received, not in slot order. + nextSeq atomic.Uint32 + missedAcquire atomic.Uint32 + bufs []struct { + lenAcquire atomic.Int32 + isRx atomic.Bool + seq uint32 + buf []byte + } +} + +func (bs *bufferSelect) reset(bufs [][]byte) { + internal.SliceReuse(&bs.bufs, len(bufs)) + bs.bufs = bs.bufs[:len(bufs)] + for i := range bs.bufs { + bs.bufs[i].buf = bufs[i] + } + bs.releaseAll() +} + +func (bs *bufferSelect) releaseAll() { + bs.nextSeq.Store(0) + bs.missedAcquire.Store(0) + for i := range bs.bufs { + bs.bufs[i].isRx.Store(false) + bs.bufs[i].lenAcquire.Store(0) + } +} + +// acquire claims a free slot for exclusive use by the caller and returns it +// sized to len. Returns nil if no slot is free or len exceeds slot size. +func (bs *bufferSelect) acquire(len int) []byte { + if len == 0 { + bs.missedAcquire.Add(1) + return nil + } + for i := range bs.bufs { + if len > cap(bs.bufs[i].buf) { + break // Length too long, would need allocation. + } + if bs.bufs[i].lenAcquire.CompareAndSwap(0, int32(len)) { + return bs.bufs[i].buf[:len] + } + } + bs.missedAcquire.Add(1) + return nil +} + +// goroPutRx copies an incoming frame into a free slot and publishes it for getRx. +// It is the only method safe to call concurrently with the runner goroutine. +// Returns false if the frame is empty, oversize or no slot is free. +func (bs *bufferSelect) goroPutRx(frame []byte) bool { + n := len(frame) + if n == 0 { + return false + } + for i := range bs.bufs { + if n > cap(bs.bufs[i].buf) { + return false + } + if bs.bufs[i].lenAcquire.CompareAndSwap(0, lenClaimed) { + bs.bufs[i].isRx.Store(true) + bs.bufs[i].seq = bs.nextSeq.Add(1) + copy(bs.bufs[i].buf, frame) + bs.bufs[i].lenAcquire.Store(int32(n)) // publish: must be last write. + return true + } + } + return false +} + +func (bs *bufferSelect) numFree() (numFree int) { + for i := range bs.bufs { + if bs.bufs[i].lenAcquire.Load() == 0 { + numFree++ + } + } + return numFree +} + +// getRx returns the oldest published Rx frame, or nil if none is pending. +func (bs *bufferSelect) getRx() []byte { + oldest := -1 + var oldestSeq uint32 + for i := range bs.bufs { + n := bs.bufs[i].lenAcquire.Load() + if n > 0 && bs.bufs[i].isRx.Load() && + (oldest < 0 || lessThan(bs.bufs[i].seq, oldestSeq)) { + oldest = i + oldestSeq = bs.bufs[i].seq + } + } + if oldest < 0 { + return nil + } + return bs.bufs[oldest].buf[:bs.bufs[oldest].lenAcquire.Load()] +} + +func (bs *bufferSelect) release(buf []byte) { + ptr := &buf[0] + for i := range bs.bufs { + if &bs.bufs[i].buf[0] == ptr { + // Clear isRx while still owning the slot so the next claimant + // never inherits a stale Rx mark. + bs.bufs[i].isRx.Store(false) + len := bs.bufs[i].lenAcquire.Load() + if len > 0 && bs.bufs[i].lenAcquire.CompareAndSwap(len, 0) { + return + } + panic("bs:race to release") + } + } + panic("bs:buffer not exist or bad offset") +} + +func lessThan(aIsLessThan, b uint32) bool { + return int32(aIsLessThan-b) < 0 +} diff --git a/x/netdev/interface.go b/x/netdev/interface.go index 655b422..349ec65 100644 --- a/x/netdev/interface.go +++ b/x/netdev/interface.go @@ -4,6 +4,7 @@ import ( "context" "errors" "net/netip" + "unsafe" "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" @@ -30,42 +31,48 @@ type DevEthernet interface { // HardwareAddr6 returns the device's 6-byte MAC address. // For PHY-only devices, returns the MAC provided at configuration. HardwareAddr6() ([6]byte, error) - // SendEthFrameOffset transmits a complete Ethernet frame at offset given by [DevEthernet.MaxFrameSizeAndOffset]. + // SendOffsetEthFrame transmits a complete Ethernet frame at offset given by [DevEthernet.MaxFrameSizeAndOffset]. // The frame includes the Ethernet header but NOT the FCS/CRC // trailer (device or stack handles CRC as appropriate). - // SendEthFrameOffset blocks until the transmission is queued succesfully + // SendOffsetEthFrame blocks until the transmission is queued succesfully // or finished sending. Should not be called concurrently // unless user is sure the driver supports it. SendOffsetEthFrame(offsetTxEthFrame []byte) error - // SetRecvHandler registers the function called when an Ethernet + // SetEthRecvHandler registers the function called when an Ethernet // frame is received. Buffers needed by the device to operate efficiently - // should be allocated on its side. This function is mutually exclusive with EthPoll: - // use on or the other to receive data. + // should be allocated on its side. + // + // Frames may be delivered via this handler, via EthPoll's buffer, or both: + // - Handler unset: EthPoll writes received frames into its argument buffer. + // - Handler set: received frames are delivered to the handler. EthPoll must + // not write to its argument buffer; it is called with a nil buffer purely + // to pump devices that need explicit servicing to drive the handler. + // + // Quiescence guarantee: SetEthRecvHandler(nil) must not return while a + // previously installed handler is executing on another goroutine, and after + // it returns the old handler must not be invoked again (analogous to Linux + // synchronize_irq semantics). Callers rely on this to safely reuse the + // buffers a handler writes into. SetEthRecvHandler(handler func(rxEthframe []byte)) // EthPoll services the device. For poll-based devices (e.g. CYW43439 // over SPI), reads from the bus and invokes the handler for each - // received frame. This method is mutually exclusive with SetEthRecvHandler: - // use one or the other to receive data but not return data via both channels. + // received frame. + // + // Behavior depends on whether a handler is set via SetEthRecvHandler: + // - No handler: writes a received frame into buf, returning its offset/length. + // - Handler set: buf is nil and must not be written to; EthPoll only pumps + // the device so frames are delivered through the handler. Return values are ignored. EthPoll(buf []byte) (ethFrameOff, ethernetBytes int, err error) // MaxFrameSizeAndOffset returns the max complete device frame size - // (including headers and any overhead) for buffer allocation. + // (including headers and any overhead) for Ethernet Rx buffer allocation. // The second value returned is the offset at which the ethernet frame // should be stored when being passed to [DevEthernet.SendOffsetEthFrame]. - // Buffers allocated should be maxEthernetFrameSize+frameOff where maxEthernetFrameSize - // is usually 1500 but less or equal to maxFrameSize-frameOff. - // MTU can be calculated doing: - // // mfu-(14+4+4) for: - // // ethernet header+ethernet CRC if present+ethernet VLAN overhead for VLAN support. - // mtu := dev.MaxFrameSizeAndOffset() - ethernet.MaxOverheadSize - MaxFrameSizeAndOffset() (maxFrameSize int, frameOff int) + // Buffers allocated for Rx should be maxFrameSize. + MaxFrameSizeAndOffset() (maxFrameSize int, sendEthFrameOff int) } // Stack is an abstraction for a networking stack. type Stack interface { - // Configure configures this Stack with the argument mac, ip and gateway addresses. - // The Stack must resolve the gateway hardware address if set. - // Configure(mac net.HardwareAddr, ip netip.Prefix, gw netip.Addr) error - // EnableICMP enables responding/sending ICMP echo frames. EnableICMP(enabled bool) error // EnableDHCP enables DHCP on the device if enabled=true and performs a DHCP request. @@ -116,6 +123,18 @@ type InterfaceConfig struct { MTU uint16 } +// RunnerBuffers returns 32-bit aligned contiguous buffers for using with [RunnerConfig]. +func (iface *Interface[C]) RunnerBuffers(n int) [][]byte { + frmlen32 := (iface.frameSize + 3) / 4 + rawBuf32 := make([]uint32, n*frmlen32) // ensure memory aligned + bufs := make([][]byte, n) + for i := range n { + buf32 := rawBuf32[i*frmlen32 : (i+1)*frmlen32] + bufs[i] = unsafe.Slice((*byte)(unsafe.Pointer(&buf32[0])), iface.frameSize) + } + return bufs +} + // Init initializes the interface from scratch with a netlink and device. If Init fails all methods on Interface are unsafe to call (panic). func (iface *Interface[C]) Init(netlink Netlink[C], dev DevEthernet, cfg InterfaceConfig) (err error) { if netlink == nil || dev == nil { diff --git a/x/netdev/runner.go b/x/netdev/runner.go index 7b45a0f..1d43ffb 100644 --- a/x/netdev/runner.go +++ b/x/netdev/runner.go @@ -3,91 +3,328 @@ package netdev import ( "context" "errors" + "runtime" "sync/atomic" + "time" "github.com/soypat/lneto" ) -// Runner orchestrates an Interface and a Stack asynchronously. -type Runner[C any] struct { - running atomic.Uint32 - // buflen stores the length of data inside buf. It is used as a buffer acquisition synchronizing primitive. - buflen atomic.Uint32 - // pktlost is incremented each time an incoming packet is lost due to insufficient buffer size. - pktlost atomic.Uint64 - tx, rx atomic.Uint64 - // buf stores actual data. - buf []byte - // bufsaux is used as ana argument to stack processing so that no allocations are performed - bufsaux [1][]byte - sizesaux [1]int - handlerTriggered bool - deviceIsPollOnly bool +var ( + errRunnerAcquired = errors.New("runner currently running") + errNotDriven = errors.New("runner needs to be poll/async driven") + errWakeNeedsAsync = errors.New("runner wake needs to be async driven") + errNoBackoffNeedsWake = errors.New("wake mechanic not set for omitting backoff") + errAsyncHandlingWithRun = errors.New("incompatible use of EnableAsyncHandling with Run- use with RunOnce") + errEgressInvalidWrite = errors.New("EgressPackets returned invalid written data given frameOffset and argument buffer size") +) + +// RunnerFlags selects how a [Runner] drives its [Interface]. See [RunnerConfig.Flags]. +type RunnerFlags uint32 + +const ( + // Signals Interface needs to be driven via [DevEthernet.EthPoll]. + // [RunnerAsync] can be set too to signal packets received via callback instead of written to EthPoll buffer. + RunnerInterfacePoll RunnerFlags = 1 << iota + // Interface data channel driven exclusively by callback passed to [DevEthernet.SetEthRecvHandler]. + // If set [DevEthernet.EthPoll] must not write data to argument buffer. + RunnerInterfaceAsync + // Runner backs off but also wakes up on receiving data asynchronously. + // Needs [RunnerInterfaceAsync] to be set to be effective. + RunnerAsyncWakeOnRx + // Runner will not use the backoff to queue timer wakeups. When setting this option + // the user is responsible for waking up the stack so that it can transmit even when not receiving data. + // Needs [RunnerAsyncWakeOnRx] to be set to be effective. + RunnerNoBackoff +) + +// HasAll reports whether every bit in query is set. +func (rf RunnerFlags) HasAll(query RunnerFlags) bool { return rf&query == query } + +// HasAny reports whether at least one bit in query is set. +func (rf RunnerFlags) HasAny(query RunnerFlags) bool { return rf&query != 0 } + +// Validate reports whether the flag combination is a usable [Runner] configuration. +func (rf RunnerFlags) Validate() error { + driven := rf.HasAny(RunnerInterfaceAsync | RunnerInterfacePoll) + wake := rf.HasAny(RunnerAsyncWakeOnRx) + if !driven { + return errNotDriven + } else if wake && !rf.HasAll(RunnerInterfaceAsync) { + return errWakeNeedsAsync + } else if rf.HasAny(RunnerNoBackoff) && !wake { + return errNoBackoffNeedsWake + } + return nil } -func (r *Runner[C]) Run(ctx context.Context, iface Interface[C], stack Stack, backoff lneto.BackoffStrategy) error { - if stack == nil || backoff == nil { - return errors.New("nil arguments to Run") +// Runner orchestrates an Interface and a Stack asynchronously. +// The complexity of Runner stems from it needing to handle three types of devices, see [RunnerConfig.Flags] documentation. +type Runner[C any] struct { + running atomic.Uint32 + bufs bufferSelect + backoff lneto.BackoffStrategy + reconnect *C + // pktlost is incremented each time an incoming packet is lost due to insufficient buffer size. + // Does not include packets dropped by the stack ([lneto.ErrPacketDrop]); the stack counts those itself. + pktlost atomic.Uint64 + // rx includes ALL data received, even dropped data. xnet.StackAsync keeps track of actual processed data. + rx atomic.Uint64 + // rxStackErrs/rxPollErrs/txStackErrs/txSendErrs count receive/transmit path + // errors which are intentionally not propagated out of the run loop nor + // printed in the datapath. See [RunnerStatistics] for per-counter semantics. + rxStackErrs, rxPollErrs atomic.Uint64 + txStackErrs, txSendErrs atomic.Uint64 + // bufsaux is used as an argument to stack processing so that no allocations are performed + bufsaux [1][]byte + sizesaux [1]int + // flags is atomic since [Runner.Wake] reads it from arbitrary goroutines + // concurrently with Configure. Configure stores it last so a visible + // RunnerAsyncWakeOnRx bit guarantees a non-nil wake channel. + flags atomic.Uint32 + wake chan struct{} + waketimer *time.Timer + asyncH *Interface[C] +} + +// RunnerConfig configures a [Runner]. Used in [Runner.Configure]. +type RunnerConfig[C any] struct { + // Buffers are the Rx/Tx packet buffers. At least one is required. Use + // [Interface.RunnerBuffers] to get correctly sized, aligned buffers. + Buffers [][]byte + // ReconnectParams is stored for use during link reconnection. Optional. + ReconnectParams *C + // Backoff is the idle wait strategy between loop iterations. Required. + Backoff lneto.BackoffStrategy + // Flags must be set to be Async, Poll driven, or both. + // This selects the kind of device operation: + // - [RunnerInterfacePoll]: Entirely poll driven Rx [DevEthernet] i.e: ESP32 family. + // - [RunnerInterfaceAsync]: Entirely async driven Rx (IRQ) [DevEthernet] i.e: LAN8720. + // - [RunnerInterfacePoll]|[RunnerInterfaceAsync]: Hybrid Rx [DevEthernet] that are poll driven but can receive data outside EthPoll method. i.e: CYW43439. + Flags RunnerFlags +} + +// RunnerStatistics keeps track of send/receive statistics. +// It is an incomplete view that should be combined with the likes of xnet.StackAsync.ReadStatistics. +// It also does not have a view into packets dropped by the [DevEthernet] because of insufficient/slow polling. +type RunnerStatistics struct { + // Rx total bytes received from interface including packets dropped. + Rx uint64 + // RxPacketsDropped incremented by 1 each time there is not enough resources to process an incoming packet. + // This does NOT include packets dropped by the Stack with [lneto.ErrPacketDrop]. + RxPacketsDropped uint64 + // RxPollErrs increments by 1 each time polling the [Interface] returns an error. + // Is irrelevant for callback driven [DevEthernet]. + RxPollErrs uint64 + // RxStackErrs increments by 1 each time stack returns a non [lneto.ErrPacketDrop] error. + RxStackErrs uint64 + // TxStackErrs increments by 1 each time stack returns an error generating egress packets. + TxStackErrs uint64 + // TxSendErrs increments by 1 each time sending a frame over [Interface] returns error. + TxSendErrs uint64 + // BufAcquireFail increments by 1 each time a buffer is unable to be acquired for Rx/Tx. + BufAcquireFail uint64 +} + +// ReadStatistics reads statistics of Runner into [RunnerStatistics]. +func (r *Runner[C]) ReadStatistics(stats *RunnerStatistics) { + *stats = RunnerStatistics{ + Rx: r.rx.Load(), + RxPacketsDropped: r.pktlost.Load(), + RxPollErrs: r.rxPollErrs.Load(), + RxStackErrs: r.rxStackErrs.Load(), + TxStackErrs: r.txStackErrs.Load(), + TxSendErrs: r.txSendErrs.Load(), + BufAcquireFail: uint64(r.bufs.missedAcquire.Load()), + } +} + +func (r *Runner[C]) getFlags() RunnerFlags { return RunnerFlags(r.flags.Load()) } + +// Wake unblocks a [Runner] sleeping in [RunnerAsyncWakeOnRx] mode so it services the +// stack immediately instead of waiting out the backoff. Signals coalesce and never block. +// Safe to call from any goroutine. +// Returns an error if the Runner is not configured for wake mode. +func (r *Runner[C]) Wake() error { + if !r.getFlags().HasAny(RunnerAsyncWakeOnRx) { + return lneto.ErrInvalidConfig + } + select { + case r.wake <- struct{}{}: // signal waiting runner + default: // already pending — coalesce, never block + } + return nil +} + +// Configure validates cfg and applies it to the Runner. Call before [Runner.Run]. +// Returns an error on invalid flags, missing buffers/backoff, or while the Runner is running. +func (r *Runner[C]) Configure(cfg RunnerConfig[C]) error { + if err := cfg.Flags.Validate(); err != nil { + return err + } + if len(cfg.Buffers) < 1 { + return lneto.ErrInvalidConfig + } else if cfg.Backoff == nil { + return lneto.ErrMissingHALConfig } if !r.acquire() { - return errors.New("runner currently running.") + return errRunnerAcquired } - defer func() { - iface.dev.SetEthRecvHandler(nil) - r.release() - }() - r.rx.Store(0) - r.tx.Store(0) - r.buflen.Store(0) - r.pktlost.Store(0) - r.handlerTriggered = false - r.deviceIsPollOnly = false + defer r.release() + r.flags.Store(0) // Disable Wake during reconfiguration. + r.teardownAsync() + r.bufs.reset(cfg.Buffers) + r.backoff = cfg.Backoff + r.reconnect = cfg.ReconnectParams + if cfg.Flags.HasAny(RunnerAsyncWakeOnRx) && r.wake == nil { + r.wake = make(chan struct{}, 1) + r.waketimer = time.NewTimer(24 * time.Hour) + } + r.flags.Store(uint32(cfg.Flags)) // Publish flags last; see field comment. + return nil +} + +// RunOnce performs a single Rx-then-Tx service cycle and returns the bytes received +// and transmitted. It does no backoff, wake wait, or state reset: the caller controls +// pacing and must [Runner.Configure] (with a non-nil stack) before the first call. +// Returns an error if a [Runner.Run] or another RunOnce is already in progress. +// +// Unlike [Runner.Run], RunOnce does not install the async receive handler. For a +// poll-driven interface ([RunnerInterfacePoll]) it works as-is; for an async interface +// ([RunnerInterfaceAsync]) call [Runner.EnableAsyncHandling] once beforehand so delivered +// frames are captured. +func (r *Runner[C]) RunOnce(iface *Interface[C], stack Stack) (nrx, ntx int, err error) { + if stack == nil { + return 0, 0, lneto.ErrInvalidConfig + } + if !r.acquire() { + return 0, 0, errRunnerAcquired + } + defer r.release() + flags := r.getFlags() + async := flags.HasAny(RunnerInterfaceAsync) + poll := flags.HasAny(RunnerInterfacePoll) bufsize := iface.bufsize() - if cap(r.buf) < bufsize { - r.buf = make([]byte, bufsize) + nrx, ntx, err = r.service(iface, stack, bufsize, poll, async) + return nrx, ntx, err +} + +// EnableAsyncHandling installs the Runner's async receive handler on iface so that +// frames delivered via [DevEthernet.SetEthRecvHandler] are captured into the buffer +// pool. Use it to drive an async interface with [Runner.RunOnce], which (unlike +// [Runner.Run]) does not install the handler itself. [Runner.Run] manages the handler +// on its own and does not need this. +// +// Returns [lneto.ErrUnsupported] if the Runner is not configured async +// ([RunnerInterfaceAsync]), or an error if a Run/RunOnce is in progress. +func (r *Runner[C]) EnableAsyncHandling(iface *Interface[C]) error { + if !r.acquire() { + return errRunnerAcquired } - r.buf = r.buf[:bufsize] + defer r.release() + if !r.getFlags().HasAny(RunnerInterfaceAsync) { + return lneto.ErrUnsupported + } + r.asyncH = iface iface.dev.SetEthRecvHandler(r.recvEthHandler) + return nil +} + +// DisableAsyncHandling removes the receive handler installed by [Runner.EnableAsyncHandling], +// stopping async frame delivery into the buffer pool. Call before reconfiguring or tearing +// down the Runner. No-op if async handling was not enabled. +func (r *Runner[C]) DisableAsyncHandling() error { + if !r.acquire() { + return errRunnerAcquired + } + defer r.release() + r.teardownAsync() + return nil +} + +func (r *Runner[C]) teardownAsync() { + if r.asyncH != nil { + r.asyncH.dev.SetEthRecvHandler(nil) + r.asyncH = nil + } +} + +// Run drives iface and stack until ctx is cancelled, doing one Rx then Tx per iteration +// and backing off when idle. Only one Run (and not concurrent with Configure) may execute +// at a time. Returns ctx.Err(). +func (r *Runner[C]) Run(ctx context.Context, iface *Interface[C], stack Stack) error { + if stack == nil { + return lneto.ErrInvalidConfig + } + if !r.acquire() { + return errRunnerAcquired + } + defer r.release() + if r.asyncH != nil { + return errAsyncHandlingWithRun + } + r.bufs.releaseAll() + r.rx.Store(0) + r.pktlost.Store(0) + r.rxStackErrs.Store(0) + r.rxPollErrs.Store(0) + r.txStackErrs.Store(0) + r.txSendErrs.Store(0) + bufsize := iface.bufsize() + flags := r.getFlags() + async := flags.HasAny(RunnerInterfaceAsync) + poll := flags.HasAny(RunnerInterfacePoll) + wake := flags.HasAny(RunnerAsyncWakeOnRx) + backoffEnabled := !flags.HasAny(RunnerNoBackoff) + if wake { + r.waketimer.Stop() + } + backoff := r.backoff + if async { + iface.dev.SetEthRecvHandler(r.recvEthHandler) + // Only tear down the handler Run itself installed. In poll-only mode + // any handler on the device is not Run's to clear. + defer iface.dev.SetEthRecvHandler(nil) + } // backoffs stores number of consecutive times no data was sent/received. var backoffs uint for ctx.Err() == nil { - n1, _ := r.processRx(stack, 0) - eoff, efrm, err := iface.dev.EthPoll(r.buf) - n2, _ := r.processRx(stack, 0) - if efrm > 0 && n2 == 0 { - r.deviceIsPollOnly = true - r.buflen.Store(uint32(eoff + efrm)) - r.processRx(stack, eoff) - } else if efrm > 0 && n2 > 0 { - return errors.New("device both returns nonzero poll read and calls, choose one") - } else if err != nil { - println("err EthPoll:", err.Error()) - } - - // Now do Tx, but first acquire buffer. - if !r.buflen.CompareAndSwap(0, 1) { - continue // Oh no, async data received, go back to Rx processing. - } - r.bufsaux = [1][]byte{r.buf} - - err = stack.EgressPackets(r.bufsaux[:], r.sizesaux[:], iface.frameOff) - n := r.sizesaux[0] + nrx, ntx, err := r.service(iface, stack, bufsize, poll, async) if err != nil { - println("err EgressPackets:", err.Error()) - } else if n > 0 { - if n+iface.frameOff > len(r.buf) { - return errors.New("EgressPackets returned invalid written data given frameOffset and argument buffer size") - } - err = iface.dev.SendOffsetEthFrame(r.bufsaux[0][:n+iface.frameOff]) - r.tx.Add(uint64(n + iface.frameOff)) - if err != nil { - println("err SendOffsetEthFrame:", err.Error()) - } + return err } - r.buflen.Store(0) // Release buffer. - if n1 > 0 || n2 > 0 || efrm > 0 || n > 0 { + if nrx > 0 || ntx > 0 { backoffs = 0 + } else if wake { + if backoffEnabled { + d := backoff(backoffs) + backoffs++ + switch d { + case lneto.BackoffFlagGosched: + runtime.Gosched() + fallthrough + case lneto.BackoffFlagNop: + continue + default: + d = max(d, 100*time.Microsecond) + } + // Claude say: + // Reset without draining waketimer.C assumes Go 1.23+ timer + // semantics (stale expiries do not linger in the channel). On + // runtimes with older semantics (e.g. TinyGo) worst case is a + // single spurious early wakeup, which is benign here. + r.waketimer.Reset(d) + } + select { + case <-r.wake: + backoffs = 0 // woke early on data. + case <-ctx.Done(): + case <-r.waketimer.C: + } + if backoffEnabled { + r.waketimer.Stop() + } } else { backoff.Do(backoffs) backoffs++ @@ -96,20 +333,68 @@ func (r *Runner[C]) Run(ctx context.Context, iface Interface[C], stack Stack, ba return ctx.Err() } +func (r *Runner[C]) service(iface *Interface[C], stack Stack, bufsize int, poll, async bool) (nrx, ntx int, err error) { + nrx, err = r.doRx(iface, stack, poll, async) + + // Now do Tx, but first acquire buffer. + txbuf := r.bufs.acquire(bufsize) + if txbuf == nil { + // We got blocked by Rx. Try draining rx. + for range len(r.bufs.bufs) { + n, err := r.doRx(iface, stack, poll, async) + if err != nil { + + break + } + nrx += n + } + txbuf = r.bufs.acquire(bufsize) + if txbuf == nil { + // Buffers still held by in-flight Rx. Skip Tx this cycle rather + // than steal a slot the receive handler may be writing into. + return nrx, 0, nil + } + } + r.bufsaux = [1][]byte{txbuf} + err = stack.EgressPackets(r.bufsaux[:], r.sizesaux[:], iface.frameOff) + ntx = r.sizesaux[0] + if err != nil { + r.txStackErrs.Add(1) + } else if ntx > 0 { + if ntx+iface.frameOff > len(txbuf) { + r.bufs.release(txbuf) + return nrx, 0, errEgressInvalidWrite + } + err = iface.dev.SendOffsetEthFrame(r.bufsaux[0][:ntx+iface.frameOff]) + if err != nil { + r.txSendErrs.Add(1) + } + } + r.bufs.release(txbuf) // Release buffer. + return nrx, ntx, nil +} + // PrintDebug // // Deprecated: Might be given other shape in future, but this is not how we do debugging. use freely meanwhile. func (r *Runner[C]) PrintDebug() { - print("RUNNER: tx|rx:", r.tx.Load(), "|", r.rx.Load(), - " devpollonly:", r.deviceIsPollOnly, " pktlost:", r.pktlost.Load(), - " handles:", r.handlerTriggered, " bufsize:", len(r.buf), + flags := r.getFlags() + print("RUNNER: rx:", r.rx.Load(), + " pktlost:", r.pktlost.Load(), + " rxErrs:", r.rxStackErrs.Load(), + " txErrs:", r.txSendErrs.Load(), + " devPoll:", flags.HasAny(RunnerInterfacePoll), + " devAsync:", flags.HasAny(RunnerInterfaceAsync), + " devWakeRx:", flags.HasAny(RunnerAsyncWakeOnRx), "\n") } +// acquire takes the single-use lock, returning false if already held. func (r *Runner[C]) acquire() bool { return r.running.CompareAndSwap(0, 1) } +// release frees the lock taken by acquire. Panics if not held. func (r *Runner[C]) release() { if r.running.Load()&1 == 0 { panic("release of unacquired resource") @@ -119,23 +404,73 @@ func (r *Runner[C]) release() { // recvEthHandler is called asynchronously. Should be as fast as possible. Do not block inside. func (r *Runner[C]) recvEthHandler(incomingEthernet []byte) { - if !r.buflen.CompareAndSwap(0, uint32(len(incomingEthernet))) { - // Failed to acquire buffer, packet dropped. + r.rx.Add(uint64(len(incomingEthernet))) // rx includes dropped data. xnet.StackAsync keeps track of actual received statistics. + if !r.bufs.goroPutRx(incomingEthernet) { + // No free buffer or frame oversize, packet dropped. r.pktlost.Add(1) return } - copy(r.buf, incomingEthernet) + r.Wake() } -// processRx is called after a packet is received asynchronously and compied to buffer via recvEthHandler -func (r *Runner[C]) processRx(stack Stack, ethFrameOff int) (int, error) { - r.handlerTriggered = true - n := r.buflen.Load() - if n == 0 { +// doRx services one Rx cycle. In poll mode it reads a frame from the device into a buffer +// and ingresses it. In async mode it drains frames delivered by recvEthHandler, pumping a +// poll-driven device with EthPoll(nil) when buffers are free. Device errors are counted +// in rxPollErrs; the returned error is the first stack ingress error encountered. +func (r *Runner[C]) doRx(iface *Interface[C], stack Stack, poll, async bool) (n int, gerr error) { + if !async { // poll guaranteed to be set as per RunnerFlags.Validate. + // Poll-only doRx. + buf := r.bufs.acquire(iface.frameSize) + if buf == nil { + return 0, nil + } + eoff, efrm, err := iface.dev.EthPoll(buf) + if err != nil { + r.rxPollErrs.Add(1) + } + if efrm > 0 { + r.bufsaux = [1][]byte{buf[:eoff+efrm]} + gerr = stack.IngressPackets(r.bufsaux[:], eoff) + if gerr != nil && gerr != lneto.ErrPacketDrop { + r.rxStackErrs.Add(1) + } + } + r.bufs.release(buf) + r.rx.Add(uint64(efrm)) + return efrm, gerr + } + + // Async branch. + n, gerr = r.processAsyncRx(stack) + if poll && r.bufs.numFree() > 0 { + // Manual polling required by device. + // Data not transmitted via this channel as per RunnerFlags documentation. + _, _, err := iface.dev.EthPoll(nil) + if err != nil { + r.rxPollErrs.Add(1) + } + } else { + return n, gerr + } + n2, err := r.processAsyncRx(stack) + if gerr == nil { + gerr = err + } + return n + n2, gerr +} + +// processAsyncRx ingresses one frame previously copied to a buffer by recvEthHandler. +// Returns the frame length, or 0 if none is pending. +func (r *Runner[C]) processAsyncRx(stack Stack) (int, error) { + buf := r.bufs.getRx() + if buf == nil { return 0, nil } - r.rx.Add(uint64(n)) - defer r.buflen.Store(0) - r.bufsaux = [1][]byte{r.buf[:n]} - return int(n), stack.IngressPackets(r.bufsaux[:], ethFrameOff) + r.bufsaux = [1][]byte{buf} + err := stack.IngressPackets(r.bufsaux[:], 0) + if err != nil && err != lneto.ErrPacketDrop { + r.rxStackErrs.Add(1) + } + r.bufs.release(buf) + return len(buf), err } diff --git a/x/netdev/runner_test.go b/x/netdev/runner_test.go new file mode 100644 index 0000000..3c5d498 --- /dev/null +++ b/x/netdev/runner_test.go @@ -0,0 +1,644 @@ +package netdev_test + +import ( + "context" + "encoding/binary" + "errors" + "net/netip" + "runtime" + "sync" + "testing" + "time" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/x/netdev" +) + +// mockDev is a DevEthernet test double. The mutex makes SetEthRecvHandler +// honor the quiescence guarantee with respect to deliver and EthPoll. +type mockDev struct { + mu sync.Mutex + handler func([]byte) + sent [][]byte + rxq [][]byte // frames pending delivery via EthPoll (poll mode or pump). + pumped int + frameSize int + frameOff int + pollErr error + sendErr error +} + +func (d *mockDev) HardwareAddr6() ([6]byte, error) { + return [6]byte{0xde, 0xad, 0xbe, 0xef, 0, 1}, nil +} + +func (d *mockDev) SendOffsetEthFrame(f []byte) error { + d.mu.Lock() + defer d.mu.Unlock() + if d.sendErr != nil { + return d.sendErr + } + d.sent = append(d.sent, append([]byte(nil), f...)) + return nil +} + +func (d *mockDev) SetEthRecvHandler(h func(rxEthFrame []byte)) { + d.mu.Lock() + d.handler = h + d.mu.Unlock() +} + +// deliver invokes the installed receive handler as a driver goroutine would. +// Returns false if no handler is installed. +func (d *mockDev) deliver(frame []byte) bool { + d.mu.Lock() + defer d.mu.Unlock() + if d.handler == nil { + return false + } + d.handler(frame) + return true +} + +func (d *mockDev) handlerInstalled() bool { + d.mu.Lock() + defer d.mu.Unlock() + return d.handler != nil +} + +func (d *mockDev) numSent() int { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.sent) +} + +func (d *mockDev) queueRx(frame []byte) { + d.mu.Lock() + defer d.mu.Unlock() + d.rxq = append(d.rxq, append([]byte(nil), frame...)) +} + +func (d *mockDev) EthPoll(buf []byte) (int, int, error) { + d.mu.Lock() + defer d.mu.Unlock() + if d.handler != nil { + // Pump mode: frames go through the handler, buf must not be written. + if buf != nil { + return 0, 0, errors.New("EthPoll got non-nil buf with handler set") + } + d.pumped++ + for _, f := range d.rxq { + d.handler(f) + } + d.rxq = nil + return 0, 0, d.pollErr + } + if len(d.rxq) == 0 { + return 0, 0, d.pollErr + } + f := d.rxq[0] + d.rxq = d.rxq[1:] + n := copy(buf[d.frameOff:], f) + return d.frameOff, n, nil +} + +func (d *mockDev) MaxFrameSizeAndOffset() (int, int) { return d.frameSize, d.frameOff } + +type mockNetlink struct{} + +func (mockNetlink) LinkConnect(_ struct{}) error { return nil } +func (mockNetlink) LinkDisconnect() {} +func (mockNetlink) LinkNotify(_ netdev.NotifyCallback[struct{}]) {} + +// mockStack records ingressed frames and emits queued egress frames. +type mockStack struct { + mu sync.Mutex + ingress [][]byte + egressq [][]byte + ingressErr error + egressErr error +} + +func (s *mockStack) EnableICMP(bool) error { return nil } + +func (s *mockStack) EnableDHCP(context.Context, bool, netip.Addr) (netip.Addr, netip.Addr, int, error) { + return netip.Addr{}, netip.Addr{}, 0, nil +} + +func (s *mockStack) Socket(context.Context, string, int, int, netip.AddrPort, netip.AddrPort) (any, error) { + return nil, nil +} + +func (s *mockStack) EgressPackets(bufs [][]byte, sizes []int, offset int) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.egressErr != nil { + return s.egressErr + } + for i := range bufs { + sizes[i] = 0 + if len(s.egressq) == 0 { + continue + } + sizes[i] = copy(bufs[i][offset:], s.egressq[0]) + s.egressq = s.egressq[1:] + } + return nil +} + +func (s *mockStack) IngressPackets(bufs [][]byte, offset int) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.ingressErr != nil { + return s.ingressErr + } + for _, b := range bufs { + s.ingress = append(s.ingress, append([]byte(nil), b[offset:]...)) + } + return nil +} + +func (s *mockStack) queueEgress(frame []byte) { + s.mu.Lock() + defer s.mu.Unlock() + s.egressq = append(s.egressq, append([]byte(nil), frame...)) +} + +func (s *mockStack) numIngress() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.ingress) +} + +func newIface(t *testing.T, dev *mockDev) *netdev.Interface[struct{}] { + t.Helper() + if dev.frameSize == 0 { + dev.frameSize = 1514 + dev.frameOff + } + var iface netdev.Interface[struct{}] + err := iface.Init(mockNetlink{}, dev, netdev.InterfaceConfig{}) + if err != nil { + t.Fatal(err) + } + return &iface +} + +func newRunner(t *testing.T, iface *netdev.Interface[struct{}], nbufs int, flags netdev.RunnerFlags, backoff lneto.BackoffStrategy) *netdev.Runner[struct{}] { + t.Helper() + if backoff == nil { + backoff = func(uint) time.Duration { return time.Millisecond } + } + var r netdev.Runner[struct{}] + err := r.Configure(netdev.RunnerConfig[struct{}]{ + Buffers: iface.RunnerBuffers(nbufs), + Backoff: backoff, + Flags: flags, + }) + if err != nil { + t.Fatal(err) + } + return &r +} + +// testFrame returns a frame whose payload encodes and repeats seq for +// integrity checking with checkFrame. +func testFrame(seq uint32, size int) []byte { + f := make([]byte, size) + binary.LittleEndian.PutUint32(f, seq) + for i := 4; i < size; i++ { + f[i] = byte(seq) + } + return f +} + +func checkFrame(t *testing.T, f []byte, size int) uint32 { + t.Helper() + if len(f) != size { + t.Fatalf("frame length %d, want %d", len(f), size) + } + seq := binary.LittleEndian.Uint32(f) + for i := 4; i < len(f); i++ { + if f[i] != byte(seq) { + t.Fatalf("frame seq %d corrupt at byte %d: got %#x want %#x", seq, i, f[i], byte(seq)) + } + } + return seq +} + +func TestRunnerFlagsValidate(t *testing.T) { + for _, tc := range []struct { + flags netdev.RunnerFlags + ok bool + }{ + {flags: 0, ok: false}, + {flags: netdev.RunnerInterfacePoll, ok: true}, + {flags: netdev.RunnerInterfaceAsync, ok: true}, + {flags: netdev.RunnerInterfacePoll | netdev.RunnerInterfaceAsync, ok: true}, + {flags: netdev.RunnerAsyncWakeOnRx, ok: false}, + {flags: netdev.RunnerInterfacePoll | netdev.RunnerAsyncWakeOnRx, ok: false}, + {flags: netdev.RunnerInterfaceAsync | netdev.RunnerAsyncWakeOnRx, ok: true}, + {flags: netdev.RunnerInterfaceAsync | netdev.RunnerNoBackoff, ok: false}, + {flags: netdev.RunnerInterfaceAsync | netdev.RunnerAsyncWakeOnRx | netdev.RunnerNoBackoff, ok: true}, + } { + err := tc.flags.Validate() + if (err == nil) != tc.ok { + t.Errorf("flags %#b: got err=%v, want ok=%v", tc.flags, err, tc.ok) + } + } +} + +func TestRunOncePollOnly(t *testing.T) { + dev := &mockDev{frameOff: 4} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 2, netdev.RunnerInterfacePoll, nil) + + const fsize = 64 + dev.queueRx(testFrame(1, fsize)) + stack.queueEgress(testFrame(2, fsize)) + + nrx, ntx, err := r.RunOnce(iface, stack) + if err != nil { + t.Fatal(err) + } + if nrx != fsize { + t.Errorf("nrx=%d, want %d", nrx, fsize) + } + if ntx != fsize { + t.Errorf("ntx=%d, want %d", ntx, fsize) + } + if stack.numIngress() != 1 { + t.Fatalf("ingress=%d, want 1", stack.numIngress()) + } + if got := checkFrame(t, stack.ingress[0], fsize); got != 1 { + t.Errorf("ingress seq=%d, want 1", got) + } + if dev.numSent() != 1 { + t.Fatalf("sent=%d, want 1", dev.numSent()) + } + // Sent frame includes the device frame offset prefix. + if got := checkFrame(t, dev.sent[0][dev.frameOff:], fsize); got != 2 { + t.Errorf("sent seq=%d, want 2", got) + } +} + +// TestRunOnceAsyncOrdering checks frames ingress in arrival order, not in +// buffer slot order (review: getRx scans slots in index order; reordering +// TCP segments triggers dup-ACK/retransmit churn). +func TestRunOnceAsyncOrdering(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 4, netdev.RunnerInterfaceAsync, nil) + err := r.EnableAsyncHandling(iface) + if err != nil { + t.Fatal(err) + } + const fsize = 64 + for seq := uint32(1); seq <= 3; seq++ { + if !dev.deliver(testFrame(seq, fsize)) { + t.Fatal("handler not installed") + } + } + for range 3 { + _, _, err := r.RunOnce(iface, stack) + if err != nil { + t.Fatal(err) + } + } + if stack.numIngress() != 3 { + t.Fatalf("ingress=%d, want 3", stack.numIngress()) + } + for i, f := range stack.ingress { + if got := checkFrame(t, f, fsize); got != uint32(i+1) { + t.Errorf("ingress[%d] seq=%d, want %d: frames reordered", i, got, i+1) + } + } +} + +// TestRunOnceAsyncPollPumpSingleBuffer exercises the async+poll pump path with +// a single buffer over several cycles. Review (MDr164, buffer.go inline): reset +// and release clear lenAcquire but not isRx, so numFree undercounts and the +// EthPoll pump is wrongly skipped. +func TestRunOnceAsyncPollPumpSingleBuffer(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 1, netdev.RunnerInterfaceAsync|netdev.RunnerInterfacePoll, nil) + err := r.EnableAsyncHandling(iface) + if err != nil { + t.Fatal(err) + } + const fsize = 64 + const cycles = 3 + for seq := uint32(1); seq <= cycles; seq++ { + dev.queueRx(testFrame(seq, fsize)) + _, _, err := r.RunOnce(iface, stack) + if err != nil { + t.Fatal(err) + } + } + if stack.numIngress() != cycles { + t.Fatalf("ingress=%d, want %d: EthPoll pump skipped", stack.numIngress(), cycles) + } + for i, f := range stack.ingress { + if got := checkFrame(t, f, fsize); got != uint32(i+1) { + t.Errorf("ingress[%d] seq=%d, want %d", i, got, i+1) + } + } + // Deliver a frame that stays pending in the pool, then reconfigure: reset + // must fully clear slot state. A stale isRx mark from the abandoned frame + // makes numFree undercount and skip the pump. + if !dev.deliver(testFrame(cycles+1, fsize)) { + t.Fatal("handler not installed") + } + err = r.Configure(netdev.RunnerConfig[struct{}]{ + Buffers: iface.RunnerBuffers(1), + Backoff: func(uint) time.Duration { return time.Millisecond }, + Flags: netdev.RunnerInterfaceAsync | netdev.RunnerInterfacePoll, + }) + if err != nil { + t.Fatal(err) + } + err = r.EnableAsyncHandling(iface) + if err != nil { + t.Fatal(err) + } + dev.queueRx(testFrame(cycles+2, fsize)) + _, _, err = r.RunOnce(iface, stack) + if err != nil { + t.Fatal(err) + } + // The frame in flight at reconfigure time is legitimately dropped; the + // queued frame must still arrive through the pump. + if stack.numIngress() != cycles+1 { + t.Fatalf("ingress=%d, want %d: pump skipped after reconfigure (stale isRx)", stack.numIngress(), cycles+1) + } +} + +// TestRunAfterEnableAsyncHandlingFails checks Run rejects a Runner set up via +// EnableAsyncHandling WITHOUT tearing down the installed handler. Review issue +// 4: the teardown defer is installed before the asyncH check, silently +// uninstalling the handler so subsequent RunOnce drops all async frames. +func TestRunAfterEnableAsyncHandlingFails(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 2, netdev.RunnerInterfaceAsync, nil) + err := r.EnableAsyncHandling(iface) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + err = r.Run(ctx, iface, stack) + if err == nil || errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Run after EnableAsyncHandling: got %v, want immediate config error", err) + } + // The rejected Run must not tear down the handler EnableAsyncHandling installed. + if !dev.handlerInstalled() { + t.Fatal("Run teardown removed handler it did not install") + } + if !dev.deliver(testFrame(1, 64)) { + t.Fatal("handler not installed") + } + nrx, _, err := r.RunOnce(iface, stack) + if err != nil || nrx != 64 { + t.Fatalf("RunOnce after rejected Run: nrx=%d err=%v", nrx, err) + } +} + +// TestOversizeFrameDropped checks a frame larger than the pool buffers is +// dropped instead of panicking. Review: acquireNext slices buf[:len] without a +// bounds check — a slice-bounds panic inside the receive path. +func TestOversizeFrameDropped(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 2, netdev.RunnerInterfaceAsync, nil) + err := r.EnableAsyncHandling(iface) + if err != nil { + t.Fatal(err) + } + func() { + defer func() { + if recovered := recover(); recovered != nil { + t.Fatalf("receive handler panicked on oversize frame: %v", recovered) + } + }() + dev.deliver(make([]byte, dev.frameSize+1)) + }() + nrx, _, err := r.RunOnce(iface, stack) + if err != nil || nrx != 0 || stack.numIngress() != 0 { + t.Fatalf("oversize frame ingressed: nrx=%d ingress=%d err=%v", nrx, stack.numIngress(), err) + } +} + +// TestRunBackoffEscalates checks the backoff counter escalates past a +// Gosched/Nop opening move. Review issue 5 (MDr164, runner.go inline): continue +// skips backoffs++, so a strategy returning Gosched at 0 busy-spins forever. +func TestRunBackoffEscalates(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + var mu sync.Mutex + var recorded []uint + backoff := func(consecutive uint) time.Duration { + mu.Lock() + recorded = append(recorded, consecutive) + mu.Unlock() + if consecutive == 0 { + return lneto.BackoffFlagGosched + } + return time.Millisecond + } + r := newRunner(t, iface, 2, netdev.RunnerInterfaceAsync|netdev.RunnerAsyncWakeOnRx, backoff) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + err := r.Run(ctx, iface, stack) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatal(err) + } + mu.Lock() + defer mu.Unlock() + if len(recorded) == 0 { + t.Fatal("backoff never called") + } + escalated := false + for _, c := range recorded { + if c > 0 { + escalated = true + break + } + } + if !escalated { + t.Fatalf("backoff counter pinned at 0 across %d idle iterations", len(recorded)) + } +} + +// TestRunAsyncStress hammers the receive handler from a producer goroutine +// while Run services the stack under Tx pressure, forcing buffer pool +// contention. Run with -race. Review issue 1: forceAcquireTx steals a slot the +// receive handler may be concurrently copying into, so partially constructed +// frames land on the wire and double release panics the runner. Checks frames +// are never corrupted (ingress AND egress) and never reordered. +func TestRunAsyncStress(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 3, netdev.RunnerInterfaceAsync|netdev.RunnerAsyncWakeOnRx, nil) + + const fsize = 64 + const nframes = 2000 + const egressSeq = 0xffff + ctx, cancel := context.WithCancel(context.Background()) + runDone := make(chan error, 1) + go func() { runDone <- r.Run(ctx, iface, stack) }() + + // Egress traffic keeps the Tx path contending with Rx for buffers. + for range 200 { + stack.queueEgress(testFrame(egressSeq, fsize)) + } + for seq := uint32(1); seq <= nframes; seq++ { + for !dev.deliver(testFrame(seq, fsize)) { + runtime.Gosched() + } + } + time.Sleep(10 * time.Millisecond) // let the runner drain pending frames. + cancel() + err := <-runDone + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + stack.mu.Lock() + if len(stack.ingress) == 0 { + t.Fatal("no frames ingressed") + } + prev := uint32(0) + for _, f := range stack.ingress { + seq := checkFrame(t, f, fsize) + if seq <= prev { + t.Fatalf("reordered: seq %d after %d", seq, prev) + } + prev = seq + } + numIngress := len(stack.ingress) + stack.mu.Unlock() + dev.mu.Lock() + for _, f := range dev.sent { + if seq := checkFrame(t, f, fsize); seq != egressSeq { + t.Fatalf("egress frame corrupted: seq=%#x", seq) + } + } + numSent := len(dev.sent) + dev.mu.Unlock() + t.Logf("ingressed %d/%d frames, sent %d", numIngress, nframes, numSent) +} + +// TestRunnerStatisticsCounting checks each RunnerStatistics counter increments +// per its documented trigger: device poll errors -> RxPollErrs, stack ingress +// errors -> RxStackErrs except ErrPacketDrop which -> RxPacketsDropped, stack +// egress errors -> TxStackErrs, and device send errors -> TxSendErrs. +func TestRunnerStatisticsCounting(t *testing.T) { + const fsize = 64 + errBoom := errors.New("boom") + for _, tc := range []struct { + name string + setup func(dev *mockDev, stack *mockStack) + want netdev.RunnerStatistics + }{ + { + // Successful Rx+Tx must not bump any error counter (regression: + // missing nil check counted every successful ingress as RxStackErrs). + name: "no errors", + setup: func(dev *mockDev, stack *mockStack) { + dev.queueRx(testFrame(1, fsize)) + stack.queueEgress(testFrame(2, fsize)) + }, + want: netdev.RunnerStatistics{Rx: fsize}, + }, + { + name: "poll error", + setup: func(dev *mockDev, stack *mockStack) { dev.pollErr = errBoom }, + want: netdev.RunnerStatistics{RxPollErrs: 1}, + }, + { + // ErrPacketDrop is counted by the stack itself, not the Runner: + // neither RxPacketsDropped nor RxStackErrs may increment. + name: "ingress packet drop", + setup: func(dev *mockDev, stack *mockStack) { + stack.ingressErr = lneto.ErrPacketDrop + dev.queueRx(testFrame(1, fsize)) + }, + want: netdev.RunnerStatistics{Rx: fsize}, + }, + { + name: "ingress stack error", + setup: func(dev *mockDev, stack *mockStack) { + stack.ingressErr = errBoom + dev.queueRx(testFrame(1, fsize)) + }, + want: netdev.RunnerStatistics{Rx: fsize, RxStackErrs: 1}, + }, + { + name: "egress stack error", + setup: func(dev *mockDev, stack *mockStack) { stack.egressErr = errBoom }, + want: netdev.RunnerStatistics{TxStackErrs: 1}, + }, + { + name: "send error", + setup: func(dev *mockDev, stack *mockStack) { + dev.sendErr = errBoom + stack.queueEgress(testFrame(1, fsize)) + }, + want: netdev.RunnerStatistics{TxSendErrs: 1}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 2, netdev.RunnerInterfacePoll, nil) + tc.setup(dev, stack) + _, _, err := r.RunOnce(iface, stack) + if err != nil { + t.Fatal(err) + } + var stats netdev.RunnerStatistics + r.ReadStatistics(&stats) + if stats != tc.want { + t.Errorf("stats=%+v, want %+v", stats, tc.want) + } + }) + } +} + +// TestRunResetsStatistics checks Run zeroes all counters accumulated by a +// previous session, including RxPollErrs. +func TestRunResetsStatistics(t *testing.T) { + dev := &mockDev{} + iface := newIface(t, dev) + stack := &mockStack{} + r := newRunner(t, iface, 2, netdev.RunnerInterfacePoll, nil) + dev.pollErr = errors.New("boom") + if _, _, err := r.RunOnce(iface, stack); err != nil { + t.Fatal(err) + } + var stats netdev.RunnerStatistics + r.ReadStatistics(&stats) + if stats == (netdev.RunnerStatistics{}) { + t.Fatal("expected nonzero statistics before Run") + } + dev.mu.Lock() + dev.pollErr = nil + dev.mu.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if err := r.Run(ctx, iface, stack); !errors.Is(err, context.DeadlineExceeded) { + t.Fatal(err) + } + r.ReadStatistics(&stats) + if stats != (netdev.RunnerStatistics{}) { + t.Errorf("Run did not reset statistics: %+v", stats) + } +} diff --git a/x/xnet/stack-async.go b/x/xnet/stack-async.go index c0a34d8..170e212 100644 --- a/x/xnet/stack-async.go +++ b/x/xnet/stack-async.go @@ -550,8 +550,12 @@ func (s *StackAsync) RegisterUDP4(node lneto.StackNode, remoteAddr [4]byte, remo if idx >= cap(s.userUDPs) { return lneto.ErrExhausted } + raddr := remoteAddr[:] + if remoteAddr == [4]byte{} { + raddr = nil + } s.userUDPs = s.userUDPs[:idx+1] - s.userUDPs[idx].SetStackNode(node, remoteAddr[:], remotePort) + s.userUDPs[idx].SetStackNode(node, raddr, remotePort) return s.udps.RegisterMACFiltered(&s.userUDPs[idx], nil) }