diff --git a/backoff.go b/backoff.go index bc8b586..faa5029 100644 --- a/backoff.go +++ b/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: diff --git a/examples/berkeley-listener/berkeley_server.go b/examples/berkeley-listener/berkeley_server.go index 50ad2e4..a6ef591 100644 --- a/examples/berkeley-listener/berkeley_server.go +++ b/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 +} diff --git a/examples/httpserver/main.go b/examples/httpserver/main.go index 2aa29a1..f889cc4 100644 --- a/examples/httpserver/main.go +++ b/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 +} diff --git a/examples/min-working-example/main-mwe.go b/examples/min-working-example/main-mwe.go index bd72ab4..135f377 100644 --- a/examples/min-working-example/main-mwe.go +++ b/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 +} diff --git a/examples/xcurl/main.go b/examples/xcurl/main.go index f69bff8..593d7f0 100644 --- a/examples/xcurl/main.go +++ b/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 +} diff --git a/internal/backoff.go b/internal/backoff.go index 409456a..8ce4e31 100644 --- a/internal/backoff.go +++ b/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) diff --git a/internal/ring.go b/internal/ring.go index bf73281..5918d66 100644 --- a/internal/ring.go +++ b/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 diff --git a/tcp/conn.go b/tcp/conn.go index 4dbc0a5..9e391df 100644 --- a/tcp/conn.go +++ b/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) } } diff --git a/udp/conn.go b/udp/conn.go index b65bbd7..594938f 100644 --- a/udp/conn.go +++ b/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) } } diff --git a/x/xnet/stack-blocking.go b/x/xnet/stack-blocking.go index f9eb5fd..e0a1f42 100644 --- a/x/xnet/stack-blocking.go +++ b/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) + } +} diff --git a/x/xnet/stack-go.go b/x/xnet/stack-go.go index 0d346e7..085c798 100644 --- a/x/xnet/stack-go.go +++ b/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 diff --git a/x/xnet/stack-retrying.go b/x/xnet/stack-retrying.go index 2e20e38..ea89587 100644 --- a/x/xnet/stack-retrying.go +++ b/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), } } diff --git a/x/xnet/xnet_concurrent_test.go b/x/xnet/xnet_concurrent_test.go index 7cb5391..f17a6c3 100644 --- a/x/xnet/xnet_concurrent_test.go +++ b/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 }