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

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)
 		}
 	}()

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 => ../../../.
+replace github.com/soypat/lneto => ../../../.
+
+// Local cyw43439 with the poll-based EthPoll API.
+replace github.com/soypat/cyw43439 => ../../../../cyw43439

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=

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)

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
+}

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 {

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
 }

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)
+	}
+}

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)
 }