diff --git a/definitions.go b/definitions.go index 9ad6b75..c4e226d 100644 --- a/definitions.go +++ b/definitions.go @@ -1,5 +1,7 @@ package lneto +import "time" + // StackNode is an abstraction of a packet exchanging protocol controller. This is the building block for all protocols, // from Ethernet to IP to TCP, practically any protocol can be expressed as a StackNode and function completely. // Today protocols represented by StackNode also include NTP, DNS, DHCP, ARP, ICMP, UDP, mDNS. @@ -37,12 +39,18 @@ type StackNode interface { // SetFlagPending(flagPending func(numPendingEncapsulations int)) } -//go:generate stringer -type=IPProto,errGeneric -linecomment -output stringers.go . +// BackoffStrategy is the abstraction of a backoff strategy for retrying an operation. +// It returns the amount of time to sleep for. If returned value is 0 then runtime.Gosched is called. +// A negative value results in no action taken. +// consecutiveBackoffs starts at 1 and increments by 1 every time the operation is retried. +type BackoffStrategy func(consecutiveBackoffs int) (sleep time.Duration) // IPProto represents the IP protocol number. type IPProto uint8 // IP protocol numbers. +// +//go:generate stringer -type=IPProto,errGeneric -linecomment -output stringers.go . const ( IPProtoHopByHop IPProto = 0 // IPv6 Hop-by-Hop Option [RFC8200] IPProtoICMP IPProto = 1 // ICMP [RFC792] diff --git a/internal/backoff.go b/internal/backoff.go index 061282d..1cb6081 100644 --- a/internal/backoff.go +++ b/internal/backoff.go @@ -2,61 +2,16 @@ package internal import "time" -type BackoffFlags uint8 - -const ( - BackoffHasPriority BackoffFlags = 1 << iota - BackoffCriticalPath - BackoffTCPConn -) - -const backoffMinWait = time.Microsecond - -func backoffMaxWait(priority BackoffFlags) time.Duration { - switch { - case priority&BackoffCriticalPath != 0: - return 1 * time.Millisecond - case priority&BackoffTCPConn != 0: - return 5 * time.Millisecond - default: - return time.Second >> (priority & BackoffHasPriority) - } -} - -func NewBackoff(priority BackoffFlags) Backoff { - return Backoff{ - wait: uint32(backoffMinWait), - maxWait: uint32(backoffMaxWait(priority)), - startWait: uint32(backoffMinWait), - } -} - -// A Backoff with a non-zero MaxWait is ready for use. -type Backoff struct { - // wait defines the amount of time that Miss will wait on next call. - wait uint32 - // Maximum allowable value for Wait. - maxWait uint32 - // startWait is the intial Wait value, as well as the value that Wait takes after a call to Hit. - startWait uint32 -} - -// Hit sets eb.Wait to the StartWait value. -func (eb *Backoff) Hit() { - if eb.maxWait == 0 { - panic("MaxWait cannot be zero") - } - eb.wait = eb.startWait -} - -// Miss sleeps for eb.Wait and increases eb.Wait exponentially. -func (eb *Backoff) Miss() { - if eb.maxWait == 0 { - panic("MaxWait cannot be zero") - } - time.Sleep(time.Duration(eb.wait)) - eb.wait *= 2 - if eb.wait > eb.maxWait { - eb.wait = eb.maxWait +// ConnRWBackoff 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) { + const ( + minWait = time.Microsecond >> 1 + maxWait = 5 * time.Millisecond + ) + wait := minWait << consecutiveBackoffs + if wait > maxWait { + wait = maxWait } + time.Sleep(wait) } diff --git a/tcp/conn.go b/tcp/conn.go index 639a919..dee7b6c 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -29,6 +29,7 @@ type Conn struct { h Handler remoteAddr []byte + _backoff lneto.BackoffStrategy rdead time.Time wdead time.Time abortErr error @@ -54,7 +55,10 @@ type ConnConfig struct { RxBuf []byte TxBuf []byte TxPacketQueueSize int - Logger *slog.Logger + // 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]. + RWBackoff lneto.BackoffStrategy + Logger *slog.Logger } func (conn *Conn) Configure(config ConnConfig) (err error) { @@ -64,6 +68,7 @@ func (conn *Conn) Configure(config ConnConfig) (err error) { if err != nil { return err } + conn._backoff = config.RWBackoff conn.logger.log = config.Logger return nil } @@ -206,8 +211,8 @@ func (conn *Conn) Write(b []byte) (int, error) { } else if plen == 0 { return 0, nil } - backoff := internal.NewBackoff(internal.BackoffTCPConn) n := 0 + backoffs := 0 for { if err := conn.checkPipe(connid, &conn.wdead); err != nil { return 0, err @@ -221,10 +226,11 @@ func (conn *Conn) Write(b []byte) (int, error) { if (err != nil && err != internal.ErrRingBufferFull) || n == plen { break } else if ngot > 0 { - backoff.Hit() + backoffs = 0 runtime.Gosched() // Do a little yield since we won't have data for sure otherwise. } else { - backoff.Miss() + backoffs++ + conn.backoff(backoffs) } conn.trace("TCPConn.Write:insuf-buf", slog.Int("missing", plen-n), slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport))) if conn.deadlineExceeded(&conn.wdead) { @@ -239,17 +245,13 @@ func (conn *Conn) Flush() error { if err != nil { return err } - if conn.deadlineExceeded(&conn.wdead) { - return errDeadlineExceeded - } else if conn.BufferedUnsent() == 0 { - return nil - } - backoff := internal.NewBackoff(internal.BackoffTCPConn) + backoffs := 0 for conn.BufferedUnsent() != 0 { if err := conn.checkPipe(connid, &conn.wdead); err != nil { return err } - backoff.Miss() + backoffs++ + conn.backoff(backoffs) } return nil } @@ -268,7 +270,7 @@ func (conn *Conn) Read(b []byte) (int, error) { lport := conn.LocalPort() rport := conn.RemotePort() conn.trace("TCPConn.Read:start", slog.Uint64("lport", uint64(lport)), slog.Uint64("rport", uint64(rport))) - backoff := internal.NewBackoff(internal.BackoffTCPConn) + backoffs := 0 for conn.BufferedInput() == 0 { state := conn.State() if !state.RxDataOpen() { @@ -280,7 +282,8 @@ func (conn *Conn) Read(b []byte) (int, error) { } return 0, err } - backoff.Miss() + backoffs++ + conn.backoff(backoffs) } return conn.handlerRead(b) } @@ -446,3 +449,11 @@ func (conn *Conn) deadlineExceeded(deadline *time.Time) bool { func (conn *Conn) ConnectionID() *uint64 { return conn.h.ConnectionID() } + +func (conn *Conn) backoff(consecutiveBackoffs int) { + if conn._backoff != nil { + conn._backoff(consecutiveBackoffs) + } else { + internal.ConnRWBackoff(consecutiveBackoffs) + } +} diff --git a/udp/conn.go b/udp/conn.go index 7b0227f..811cadd 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -22,8 +22,9 @@ type Conn struct { remoteAddr []byte - rdead time.Time - wdead time.Time + _backoff lneto.BackoffStrategy + rdead time.Time + wdead time.Time ipID uint16 } @@ -38,6 +39,10 @@ type ConnConfig struct { RxQueueSize int // TxQueueSize is the maximum number of outgoing datagrams that can be queued. 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]. + RWBackoff lneto.BackoffStrategy } // Configure initializes the connection with the given buffer and queue configuration. @@ -50,6 +55,7 @@ func (conn *Conn) Configure(cfg ConnConfig) error { if err != nil { return err } + conn._backoff = cfg.RWBackoff return nil } @@ -120,13 +126,14 @@ func (conn *Conn) Write(b []byte) (int, error) { if len(b) == 0 { return 0, nil } - backoff := internal.NewBackoff(internal.BackoffTCPConn) + connID, err := conn.lockPipeConnID() + if err != nil { + return 0, err + } + backoffs := 0 for { - if conn.deadlineExceeded(&conn.wdead) { - return 0, os.ErrDeadlineExceeded - } conn.mu.Lock() - if conn.h.closeCalled { + if conn.h.closeCalled || connID != conn.h.connid { conn.mu.Unlock() return 0, net.ErrClosed } @@ -135,20 +142,25 @@ func (conn *Conn) Write(b []byte) (int, error) { if n > 0 { return n, err } - backoff.Miss() + if conn.deadlineExceeded(&conn.wdead) { + return 0, os.ErrDeadlineExceeded + } + backoffs++ + conn.backoff(backoffs) } } // Read dequeues a single datagram. If the buffer is smaller than the datagram, // the remaining bytes are discarded (SOCK_DGRAM semantics). func (conn *Conn) Read(b []byte) (int, error) { - backoff := internal.NewBackoff(internal.BackoffTCPConn) + connID, err := conn.lockPipeConnID() + if err != nil { + return 0, err + } + backoffs := 0 for { - if conn.deadlineExceeded(&conn.rdead) { - return 0, os.ErrDeadlineExceeded - } conn.mu.Lock() - if conn.h.closeCalled && conn.h.BufferedInput() == 0 { + if conn.h.closeCalled && conn.h.BufferedInput() == 0 || connID != conn.h.connid { conn.mu.Unlock() return 0, net.ErrClosed } @@ -157,7 +169,11 @@ func (conn *Conn) Read(b []byte) (int, error) { if n > 0 { return n, err } - backoff.Miss() + if conn.deadlineExceeded(&conn.rdead) { + return 0, os.ErrDeadlineExceeded + } + backoffs++ + conn.backoff(backoffs) } } @@ -284,3 +300,20 @@ func (conn *Conn) FreeInput() int { defer conn.mu.Unlock() return conn.h.FreeInput() } + +func (conn *Conn) backoff(consecutiveBackoffs int) { + if conn._backoff != nil { + conn._backoff(consecutiveBackoffs) + } else { + internal.ConnRWBackoff(consecutiveBackoffs) + } +} + +func (conn *Conn) lockPipeConnID() (uint64, error) { + conn.mu.Lock() + defer conn.mu.Unlock() + if conn.h.closeCalled && len(conn.h.rxDgrams) == 0 { + return 0, net.ErrClosed + } + return conn.h.connid, nil +} diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go index 9d220f1..e794254 100644 --- a/x/xnet/tcppool.go +++ b/x/xnet/tcppool.go @@ -53,6 +53,9 @@ type TCPPoolConfig struct { ClosingTimeout time.Duration // NewUserData is used to create user data used for each individual TCP connection and returned on GetTCP. NewUserData func() any + // NewBackoff returns the backoff to use for every newly configured TCP connection. + // This should always return a static(non-method) function unless you know what you are doing. + NewBackoff func() lneto.BackoffStrategy } func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { @@ -76,12 +79,16 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { for i := range pool.conns { bufoff := i * allocPerConn txOff := bufoff + cfg.RxBufSize - err := pool.conns[i].Configure(tcp.ConnConfig{ + conncfg := tcp.ConnConfig{ RxBuf: bufSpace[bufoff:txOff], TxBuf: bufSpace[txOff : txOff+cfg.TxBufSize], TxPacketQueueSize: cfg.QueueSize, Logger: cfg.ConnLogger, - }) + } + if cfg.NewBackoff != nil { + conncfg.RWBackoff = cfg.NewBackoff() + } + err := pool.conns[i].Configure(conncfg) if err != nil { return nil, err }