Stack backoff rework (#83)

* huge Stack backoff rework

* lneto bump: huge Stack backoff rework

backoff.go

@@ -17,13 +17,14 @@ const (
 //     Useful for when the BackoffStrategy implements its own yield.
 //   - Returns [BackoffFlagGosched]: Signal [runtime.Gosched] should be called.
 //
-// consecutiveBackoffs starts at 1 and increments by 1 every time the operation is retried.
+// Despite the name(changes welcome) consecutiveBackoffs starts at 0 and
+// increments by 1 every time the operation is retried.
 // See internal/backoff.go for a implementation example.
-type BackoffStrategy func(consecutiveBackoffs int) (sleepOrFlag time.Duration)
+type BackoffStrategy func(consecutiveBackoffs uint) (sleepOrFlag time.Duration)
 
 // Do applies the backoff strategy by calling backoff(consecutiveBackoffs)
 // and then the corresponding yield function for the returned value. See [BackoffStrategy].
-func (backoff BackoffStrategy) Do(consecutiveBackoffs int) {
+func (backoff BackoffStrategy) Do(consecutiveBackoffs uint) {
 	sleep := backoff(consecutiveBackoffs)
 	switch sleep {
 	case BackoffFlagNop:

examples/berkeley-listener/berkeley_server.go

@@ -199,7 +199,7 @@ func run() error {
 	}()
 
 	// Create blocking + Berkeley stack
-	blocking := stack.StackBlocking(5 * time.Millisecond)
+	blocking := stack.StackBlocking(stackBackoff)
 	berkeley := blocking.StackGo(xnet.StackGoConfig{
 		ListenerPoolConfig: xnet.TCPPoolConfig{
 			PoolSize:           uint16(flagPoolSize),
@@ -212,7 +212,7 @@ func run() error {
 	})
 
 	// Perform DHCP to get address.
-	rstack := stack.StackRetrying(5 * time.Millisecond)
+	rstack := stack.StackRetrying(stackBackoff)
 	const dhcpTimeout = 6 * time.Second
 	const dhcpRetries = 2
 	results, err := rstack.DoDHCPv4([4]byte{192, 168, 1, 96}, dhcpTimeout, dhcpRetries)
@@ -399,3 +399,10 @@ func mockClient(stack *xnet.StackAsync, port uint16, subnet netip.Prefix) {
 	fmt.Println("mockclient: received response:\n", string(page))
 	mockConn.Close()
 }
+
+func stackBackoff(consecutiveBackoffs uint) time.Duration {
+	if consecutiveBackoffs < 10 {
+		return time.Millisecond
+	}
+	return 10 * time.Millisecond
+}

examples/httpserver/main.go

@@ -190,7 +190,7 @@ func run() (err error) {
 		}
 	}()
 
-	rstack := stack.StackRetrying(5 * time.Millisecond)
+	rstack := stack.StackRetrying(stackBackoff)
 
 	const (
 		dhcpTimeout = 6 * time.Second
@@ -354,3 +354,10 @@ func tryPoll(iface ltesto.Interface, poll time.Duration) (dataMayBeReady bool, _
 	dataMayBeReady = true
 	return dataMayBeReady, nil
 }
+
+func stackBackoff(consecutiveBackoffs uint) time.Duration {
+	if consecutiveBackoffs < 10 {
+		return time.Millisecond
+	}
+	return 10 * time.Millisecond
+}

examples/min-working-example/main-mwe.go

@@ -79,7 +79,7 @@ func run(ctx context.Context, stack *xnet.StackAsync) error {
 	// Other option is to instead use async API which leads to more verbose
 	// and more stateful code.
 	go stackLoop(ctx, stack)
-	rstack := stack.StackRetrying(pollTime)
+	rstack := stack.StackRetrying(stackBackoff)
 	results, err := rstack.DoDHCPv4([4]byte{}, protoTimeout, protoRetries)
 	if err != nil {
 		return fmt.Errorf("doing DHCP: %w", err)
@@ -93,7 +93,7 @@ func run(ctx context.Context, stack *xnet.StackAsync) error {
 		return fmt.Errorf("resolving router MAC: %w", err)
 	}
 	stack.SetGateway6(gateway)
-	berkstack := stack.StackBlocking(pollTime).StackGo(xnet.StackGoConfig{
+	berkstack := stack.StackBlocking(stackBackoff).StackGo(xnet.StackGoConfig{
 		ListenerPoolConfig: xnet.TCPPoolConfig{
 			PoolSize:           tcpConnPoolSize,
 			QueueSize:          tcpPacketQueueSize,
@@ -171,3 +171,10 @@ func must(err error) {
 		panic(err)
 	}
 }
+
+func stackBackoff(consecutiveBackoffs uint) time.Duration {
+	if consecutiveBackoffs < 10 {
+		return time.Millisecond
+	}
+	return 10 * time.Millisecond
+}

examples/xcurl/main.go

@@ -212,7 +212,7 @@ func run() (err error) {
 		}
 	}()
 
-	rstack := stack.StackRetrying(5 * time.Millisecond)
+	rstack := stack.StackRetrying(stackBackoff)
 
 	const (
 		dhcpTimeout = 6 * time.Second
@@ -382,3 +382,10 @@ func tryPoll(iface ltesto.Interface, poll time.Duration) (dataMayBeReady bool, _
 	dataMayBeReady = true
 	return dataMayBeReady, nil
 }
+
+func stackBackoff(consecutiveBackoffs uint) time.Duration {
+	if consecutiveBackoffs < 10 {
+		return time.Millisecond
+	}
+	return 10 * time.Millisecond
+}

internal/backoff.go

@@ -4,13 +4,31 @@ import (
 	"time"
 )
 
-// ConnRWBackoff implements exponential backoff suitable for TCP connection
+// BackoffConnRW implements exponential backoff suitable for TCP connection
 // read/write polling. It starts at 1us and caps at 5ms, doubling on each consecutive backoff.
-func ConnRWBackoff(consecutiveBackoffs int) {
+func BackoffConnRW(consecutiveBackoffs uint) {
 	const (
-		minWait        = uint32(time.Microsecond) >> 1
+		minWait        = uint32(time.Microsecond)
 		maxWait        = 5 * uint32(time.Millisecond)
-		maxShift       = 23
+		maxShift       = 22
+		_overflowCheck = minWait << maxShift
+	)
+	wait := minWait << min(consecutiveBackoffs, maxShift)
+	if wait > maxWait {
+		wait = maxWait
+	}
+	time.Sleep(time.Duration(wait))
+}
+
+// BackoffStackProto implements exponential backoff suitable for stack-level
+// protocol processing polling. It starts at 1us and caps at 100ms, doubling on each consecutive backoff.
+func BackoffStackProto(consecutiveBackoffs uint) {
+	const (
+		minWait = uint32(time.Microsecond)
+		maxWait = 100 * uint32(time.Millisecond)
+
+		// Statically calculated numbers below.
+		maxShift       = 22
 		_overflowCheck = minWait << maxShift
 	)
 	wait := minWait << min(consecutiveBackoffs, maxShift)

internal/ring.go

@@ -271,20 +271,6 @@ func (r *Ring) addOff(a, b int) int {
 	return result
 }
 
-func max(a, b int) int {
-	if a > b {
-		return a
-	}
-	return b
-}
-
-func min(a, b int) int {
-	if a < b {
-		return a
-	}
-	return b
-}
-
 func (r *Ring) string() string {
 	var b bytes.Buffer
 	r2 := *r

tcp/conn.go

@@ -56,7 +56,7 @@ type ConnConfig struct {
 	TxBuf             []byte
 	TxPacketQueueSize int
 	// RWBackoff sets the backoff policy for backoff when data unavailable on Read or buffer full on Write.
-	// If not set a default backoff strategy will be used. See [internal.ConnRWBackoff].
+	// If not set a default backoff strategy will be used. See [internal.BackoffConnRW].
 	RWBackoff lneto.BackoffStrategy
 	Logger    *slog.Logger
 }
@@ -212,7 +212,7 @@ func (conn *Conn) Write(b []byte) (int, error) {
 		return 0, nil
 	}
 	n := 0
-	backoffs := 0
+	var backoffs uint
 	for len(b) > 0 {
 		if err := conn.checkPipe(connid, &conn.wdead); err != nil {
 			return n, err
@@ -238,8 +238,8 @@ func (conn *Conn) Write(b []byte) (int, error) {
 			if conn.deadlineExceeded(&conn.wdead) {
 				return n, errDeadlineExceeded
 			}
-			backoffs++
 			conn.backoff(backoffs)
+			backoffs++
 		}
 	}
 	return n, err
@@ -250,13 +250,13 @@ func (conn *Conn) Flush() error {
 	if err != nil {
 		return err
 	}
-	backoffs := 0
+	var backoffs uint
 	for conn.BufferedUnsent() != 0 {
 		if err := conn.checkPipe(connid, &conn.wdead); err != nil {
 			return err
 		}
-		backoffs++
 		conn.backoff(backoffs)
+		backoffs++
 	}
 	return nil
 }
@@ -271,7 +271,7 @@ func (conn *Conn) Read(b []byte) (int, error) {
 	rport := conn.h.remotePort
 	conn.mu.Unlock()
 	conn.trace("TCPConn.Read:start", slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport)))
-	backoffs := 0
+	var backoffs uint
 	n := 0
 	for len(b) > 0 {
 		conn.mu.Lock()
@@ -302,8 +302,8 @@ func (conn *Conn) Read(b []byte) (int, error) {
 			} else if conn.deadlineExceeded(&conn.rdead) {
 				return n, errDeadlineExceeded
 			}
-			backoffs++
 			conn.backoff(backoffs)
+			backoffs++
 		}
 	}
 	return n, nil
@@ -465,10 +465,10 @@ func (conn *Conn) ConnectionID() *uint64 {
 	return conn.h.ConnectionID()
 }
 
-func (conn *Conn) backoff(consecutiveBackoffs int) {
+func (conn *Conn) backoff(consecutiveBackoffs uint) {
 	if conn._backoff != nil {
 		conn._backoff.Do(consecutiveBackoffs)
 	} else {
-		internal.ConnRWBackoff(consecutiveBackoffs)
+		internal.BackoffConnRW(consecutiveBackoffs)
 	}
 }

udp/conn.go

@@ -41,7 +41,7 @@ type ConnConfig struct {
 	TxQueueSize int
 	// RWBackoff sets the backoff policy for backoff when data unavailable on Read or buffer full on Write.
 	// This field is ineffective on configuring a [Handler].
-	// If not set a default backoff strategy will be used. See [internal.ConnRWBackoff].
+	// If not set a default backoff strategy will be used. See [internal.BackoffConnRW].
 	RWBackoff lneto.BackoffStrategy
 }
 
@@ -130,7 +130,7 @@ func (conn *Conn) Write(b []byte) (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	backoffs := 0
+	var backoffs uint
 	for {
 		conn.mu.Lock()
 		if conn.h.closeCalled || connID != conn.h.connid {
@@ -145,8 +145,8 @@ func (conn *Conn) Write(b []byte) (int, error) {
 		if conn.deadlineExceeded(&conn.wdead) {
 			return 0, os.ErrDeadlineExceeded
 		}
-		backoffs++
 		conn.backoff(backoffs)
+		backoffs++
 	}
 }
 
@@ -157,7 +157,7 @@ func (conn *Conn) Read(b []byte) (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	backoffs := 0
+	var backoffs uint
 	for {
 		conn.mu.Lock()
 		if conn.h.closeCalled && conn.h.BufferedInput() == 0 || connID != conn.h.connid {
@@ -172,8 +172,8 @@ func (conn *Conn) Read(b []byte) (int, error) {
 		if conn.deadlineExceeded(&conn.rdead) {
 			return 0, os.ErrDeadlineExceeded
 		}
-		backoffs++
 		conn.backoff(backoffs)
+		backoffs++
 	}
 }
 
@@ -301,11 +301,11 @@ func (conn *Conn) FreeInput() int {
 	return conn.h.FreeInput()
 }
 
-func (conn *Conn) backoff(consecutiveBackoffs int) {
+func (conn *Conn) backoff(consecutiveBackoffs uint) {
 	if conn._backoff != nil {
 		conn._backoff.Do(consecutiveBackoffs)
 	} else {
-		internal.ConnRWBackoff(consecutiveBackoffs)
+		internal.BackoffConnRW(consecutiveBackoffs)
 	}
 }
 

x/xnet/stack-blocking.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/dhcpv4"
+	"github.com/soypat/lneto/internal"
 	"github.com/soypat/lneto/tcp"
 )
 
@@ -19,22 +20,16 @@ var (
 	errDeadlineExceed = errors.New("cywnet: deadline exceeded")
 )
 
-func (s *StackAsync) StackBlocking(loopSleep time.Duration) StackBlocking {
-	if loopSleep < 0 {
-		panic("invalid sleep")
-	} else if loopSleep > 3*time.Second {
-		// loopSleep should be a very small amount of time for stack to remain responsive.
-		panic("StackBlocking sleep too large")
-	}
+func (s *StackAsync) StackBlocking(stackProtoBackoff lneto.BackoffStrategy) StackBlocking {
 	return StackBlocking{
-		async:     s,
-		loopSleep: loopSleep,
+		async:    s,
+		_backoff: stackProtoBackoff,
 	}
 }
 
 type StackBlocking struct {
-	async     *StackAsync
-	loopSleep time.Duration
+	async    *StackAsync
+	_backoff lneto.BackoffStrategy
 }
 
 func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPResults, error) {
@@ -42,20 +37,31 @@ func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPRe
 	if err != nil {
 		return nil, err
 	}
-	sleep := s.loopSleep
+	var backoffs uint
 	deadline := time.Now().Add(timeout)
 	requested := false
+	var lastState dhcpv4.ClientState
 	for i := 0; i < maxIter; i++ {
+		s.async.mu.Lock()
 		state := s.async.dhcp.State()
-		requested = requested || state > dhcpv4.StateInit
-		if requested && state == dhcpv4.StateInit {
-			return nil, errors.New("DHCP NACK")
-		} else if state == dhcpv4.StateBound {
-			break // DHCP done succesfully.
-		} else if err = s.checkDeadline(deadline); err != nil {
-			return nil, err
+		s.async.mu.Unlock()
+		if state == lastState {
+			if err = s.checkDeadline(deadline); err != nil {
+				return nil, err
+			}
+			s.backoff(backoffs)
+			backoffs++
+		} else {
+			// State change indicates something happened.
+			backoffs = 0
+			lastState = state
+			requested = requested || state > dhcpv4.StateInit
+			if requested && state == dhcpv4.StateInit {
+				return nil, errors.New("DHCP NACK")
+			} else if state == dhcpv4.StateBound {
+				break // DHCP done succesfully.
+			}
 		}
-		time.Sleep(sleep)
 	}
 	return s.async.ResultDHCP()
 }
@@ -65,24 +71,30 @@ func (s StackBlocking) DoPing(hostAddr netip.Addr, timeout time.Duration) (round
 		return 0, lneto.ErrInvalidAddr
 	}
 	var buf [16]byte
+	s.async.mu.Lock()
 	s.async.prandRead(buf[:])
 	key, err := s.async.icmp.PingStart(hostAddr.As4(), buf[:], 56) // size=56 so ICMP size is 64, like linux.
+	s.async.mu.Unlock()
 	if err != nil {
 		return 0, err
 	}
 	start := time.Now()
-	sleep := timeout / maxIter
+	var backoffs uint
 	for i := 0; i < maxIter; i++ {
-		time.Sleep(sleep)
-		elapsed := time.Since(start)
+		s.async.mu.Lock()
 		completed, exists := s.async.icmp.PingPop(key)
+		s.async.mu.Unlock()
 		if !exists {
 			return 0, net.ErrClosed // lneto.ErrAborted
-		} else if completed {
+		}
+		elapsed := time.Since(start)
+		if completed {
 			return elapsed, nil
 		} else if elapsed > timeout {
 			break
 		}
+		s.backoff(backoffs)
+		backoffs++
 	}
 	return 0, errDeadlineExceed
 }
@@ -92,9 +104,10 @@ func (s StackBlocking) DoNTP(hostAddr netip.Addr, timeout time.Duration) (offset
 	if err != nil {
 		return -1, err
 	}
-	sleep := s.loopSleep
+
 	deadline := time.Now().Add(timeout)
 	var done bool
+	var backoffs uint
 	for i := 0; i < maxIter; i++ {
 		offset, done = s.async.ResultNTPOffset()
 		if done {
@@ -102,7 +115,8 @@ func (s StackBlocking) DoNTP(hostAddr netip.Addr, timeout time.Duration) (offset
 		} else if err = s.checkDeadline(deadline); err != nil {
 			return -1, err
 		}
-		time.Sleep(sleep)
+		s.backoff(backoffs)
+		backoffs++
 	}
 	return -1, errDeadlineExceed
 }
@@ -112,7 +126,7 @@ func (s StackBlocking) DoResolveHardwareAddress6(addr netip.Addr, timeout time.D
 	if err != nil {
 		return hw, err
 	}
-	sleep := s.loopSleep
+	var backoffs uint
 	deadline := time.Now().Add(timeout)
 	for i := 0; i < maxIter; i++ {
 		hw, err = s.async.ResultResolveHardwareAddress6(addr)
@@ -121,7 +135,8 @@ func (s StackBlocking) DoResolveHardwareAddress6(addr netip.Addr, timeout time.D
 		} else if err = s.checkDeadline(deadline); err != nil {
 			break
 		}
-		time.Sleep(sleep)
+		s.backoff(backoffs)
+		backoffs++
 		err = errDeadlineExceed // Ensure that if iterations done error is returned.
 	}
 	ip4 := addr.As4()
@@ -134,8 +149,9 @@ func (s StackBlocking) DoLookupIP(host string, timeout time.Duration) (addrs []n
 	if err != nil {
 		return nil, err
 	}
-	sleep := s.loopSleep
+
 	deadline := time.Now().Add(timeout)
+	var backoffs uint
 	for i := 0; i < maxIter; i++ {
 		addrs, completed, err := s.async.ResultLookupIP(host)
 		if completed {
@@ -143,7 +159,8 @@ func (s StackBlocking) DoLookupIP(host string, timeout time.Duration) (addrs []n
 		} else if err = s.checkDeadline(deadline); err != nil {
 			return nil, err
 		}
-		time.Sleep(sleep)
+		s.backoff(backoffs)
+		backoffs++
 	}
 	return nil, errDeadlineExceed
 }
@@ -155,8 +172,8 @@ func (s StackBlocking) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.A
 	if err != nil {
 		return err
 	}
-	sleep := s.loopSleep
 	deadline := time.Now().Add(timeout)
+	var backoffs uint
 	for i := 0; i < maxIter; i++ {
 		state := conn.State()
 		if state == tcp.StateEstablished {
@@ -166,12 +183,13 @@ func (s StackBlocking) DoDialTCP(conn *tcp.Conn, localPort uint16, addrp netip.A
 				conn.Abort()
 				return err
 			}
-			time.Sleep(sleep)
 		} else {
 			// Unexpected state, abort and terminate connection.
 			conn.Abort()
 			return errTCPFailedToConnect
 		}
+		s.backoff(backoffs)
+		backoffs++
 	}
 	return errDeadlineExceed
 }
@@ -182,3 +200,15 @@ func (s StackBlocking) checkDeadline(deadline time.Time) error {
 	}
 	return nil
 }
+
+func (s StackBlocking) backoff(consecutiveBackoffs uint) {
+	backoff(s._backoff, consecutiveBackoffs)
+}
+
+func backoff(bo lneto.BackoffStrategy, consecutiveBackoffs uint) {
+	if bo != nil {
+		bo.Do(consecutiveBackoffs)
+	} else {
+		internal.BackoffStackProto(consecutiveBackoffs)
+	}
+}

x/xnet/stack-go.go

@@ -5,7 +5,6 @@ import (
 	"net"
 	"net/netip"
 	"syscall"
-	"time"
 
 	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/tcp"
@@ -22,8 +21,8 @@ type StackGoConfig struct {
 	ListenerPoolConfig TCPPoolConfig
 }
 
-func (s *StackAsync) StackGo(loopSleep time.Duration, cfg StackGoConfig) StackGo {
-	return s.StackBlocking(loopSleep).StackGo(cfg)
+func (s *StackAsync) StackGo(stackProtoBackoff lneto.BackoffStrategy, cfg StackGoConfig) StackGo {
+	return s.StackBlocking(stackProtoBackoff).StackGo(cfg)
 }
 
 func (s StackBlocking) StackGo(cfg StackGoConfig) StackGo {
@@ -131,8 +130,10 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
 			if err != nil {
 				return nil, err
 			}
+			var backoffs uint
 			for {
-				time.Sleep(s.blk.loopSleep)
+				s.blk.backoff(backoffs)
+				backoffs++
 				state := conn.State()
 				if state == tcp.StateEstablished {
 					tc := tcpconn{
@@ -159,7 +160,7 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
 			}
 			var l tcplistener
 			l.localAddr = net.TCPAddrFromAddrPort(laddr)
-			l.sleep = s.blk.loopSleep
+			l.sleep = s.blk._backoff
 			err = l.l.Reset(laddr.Port(), pool)
 			if err != nil {
 				return nil, err
@@ -177,7 +178,7 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
 type tcplistener struct {
 	l         tcp.Listener
 	closed    bool
-	sleep     time.Duration
+	sleep     lneto.BackoffStrategy
 	localAddr net.Addr
 }
 
@@ -191,12 +192,15 @@ func (l *tcplistener) Accept() (net.Conn, error) {
 	if l.closed {
 		return nil, net.ErrClosed
 	}
+	var backoffs uint
 	for {
 		n := l.l.NumberOfReadyToAccept()
 		if n == 0 {
-			time.Sleep(l.sleep)
+			backoff(l.sleep, backoffs)
+			backoffs++
 			continue
 		}
+		backoffs = 0
 		c, _, err := l.l.TryAccept()
 		if err != nil {
 			return nil, err

x/xnet/stack-retrying.go

@@ -5,12 +5,13 @@ import (
 	"net/netip"
 	"time"
 
+	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/tcp"
 )
 
-func (s *StackAsync) StackRetrying(loopSleep time.Duration) StackRetrying {
+func (s *StackAsync) StackRetrying(stackProtoBackoff lneto.BackoffStrategy) StackRetrying {
 	return StackRetrying{
-		block: s.StackBlocking(loopSleep),
+		block: s.StackBlocking(stackProtoBackoff),
 	}
 }
 

x/xnet/xnet_concurrent_test.go

@@ -432,6 +432,6 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn
 
 }
 
-func backoffGosched(consecutiveBackoffs int) (sleep time.Duration) {
+func backoffGosched(consecutiveBackoffs uint) (sleep time.Duration) {
 	return lneto.BackoffFlagGosched
 }