remove backoff assumption from Router

This commit is contained in:
Patricio Whittingslow
2026-07-29 11:32:26 -03:00
parent 564f277962
commit c97c5e17cf
5 changed files with 16 additions and 30 deletions
+2 -5
View File
@@ -52,11 +52,8 @@ func run() error {
RequestNumHeaderKVCap: numHeaderFields, RequestNumHeaderKVCap: numHeaderFields,
ResponseHeaderMinBufferSize: bufferSizes, ResponseHeaderMinBufferSize: bufferSizes,
MaxAwaitingConns: 256, MaxAwaitingConns: 256,
Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) { Mux: &mux,
return min(time.Second, time.Millisecond*time.Duration(consecutiveBackoffs)) Logger: slog.Default(),
},
Mux: &mux,
Logger: slog.Default(),
}) })
if err != nil { if err != nil {
return err return err
-2
View File
@@ -6,7 +6,6 @@ import (
"log" "log"
"log/slog" "log/slog"
"os" "os"
"time"
"github.com/soypat/lneto/http/httphi" "github.com/soypat/lneto/http/httphi"
"github.com/soypat/lneto/http/httpraw" "github.com/soypat/lneto/http/httpraw"
@@ -30,7 +29,6 @@ func ExampleRouter_linux() {
ResponseHeaderMinBufferSize: 32, // Shared buffer with Request, not strictly necessary, especially if not sending headers. ResponseHeaderMinBufferSize: 32, // Shared buffer with Request, not strictly necessary, especially if not sending headers.
RequestNumHeaderKVCap: numHeaderKV, RequestNumHeaderKVCap: numHeaderKV,
NormalizeOutgoingKeys: true, NormalizeOutgoingKeys: true,
Backoff: func(uint) time.Duration { return time.Millisecond },
Mux: &mux, Mux: &mux,
Logger: slog.Default(), Logger: slog.Default(),
}) })
+6
View File
@@ -1,6 +1,7 @@
package httphi package httphi
import ( import (
"io"
"strings" "strings"
"unsafe" "unsafe"
@@ -12,6 +13,8 @@ import (
// Handle is a extremely low-level HTTP handling method used internally in [Router]. // Handle is a extremely low-level HTTP handling method used internally in [Router].
// Requires exchange to be acquired and configured. Will panic if any argument is nil. // Requires exchange to be acquired and configured. Will panic if any argument is nil.
// Handle does not close the connection on any outcome: the caller owns it. // Handle does not close the connection on any outcome: the caller owns it.
// backoff can be set for dealing with non-blocking connections. If backoff set to nil
// then a zero-length-read will result in Handle returning [io.ErrNoProgress].
func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error {
if !exch.acquired.Load() { if !exch.acquired.Load() {
return lneto.ErrBadState return lneto.ErrBadState
@@ -26,6 +29,9 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error {
exch.handleError(err) exch.handleError(err)
return err return err
} else if n == 0 { } else if n == 0 {
if backoff == nil {
return io.ErrNoProgress
}
backoff.Do(consecutiveBackoffs) backoff.Do(consecutiveBackoffs)
consecutiveBackoffs++ consecutiveBackoffs++
continue continue
+8 -15
View File
@@ -59,8 +59,7 @@ type Router struct {
exchs []Exchange exchs []Exchange
freeList *Exchange freeList *Exchange
backoff lneto.BackoffStrategy log *slog.Logger
log *slog.Logger
} }
// job is a connection waiting on an exchange for a worker goroutine to serve it. // job is a connection waiting on an exchange for a worker goroutine to serve it.
@@ -92,9 +91,6 @@ type RouterConfig struct {
// must be non-zero when running a fixed number of goroutines, unused otherwise. // must be non-zero when running a fixed number of goroutines, unused otherwise.
MaxAwaitingConns int MaxAwaitingConns int
// Backoff is consulted when a read off a connection yields no data, letting
// the caller decide whether to sleep, yield or spin. Required.
Backoff lneto.BackoffStrategy
// Mux resolves each request's method and path to the handler serving it. Required. // Mux resolves each request's method and path to the handler serving it. Required.
Mux Mux Mux Mux
// Logger receives failed exchanges. Optional, nil disables logging. // Logger receives failed exchanges. Optional, nil disables logging.
@@ -127,8 +123,6 @@ func (cfg RouterConfig) Validate() error {
cfg.ResponseHeaderMinBufferSize > maxExchangeBuffer, cfg.ResponseHeaderMinBufferSize > maxExchangeBuffer,
cfg.RequestHeaderBufferSize > maxExchangeBuffer-cfg.ResponseHeaderMinBufferSize: cfg.RequestHeaderBufferSize > maxExchangeBuffer-cfg.ResponseHeaderMinBufferSize:
return lneto.ErrInvalidConfig return lneto.ErrInvalidConfig
case cfg.Backoff == nil:
return lneto.ErrMissingHALConfig
} }
if workerMode { if workerMode {
// Buffer sizes are bounded by maxExchangeBuffer above so the sum cannot // Buffer sizes are bounded by maxExchangeBuffer above so the sum cannot
@@ -195,7 +189,6 @@ func (r *Router) Configure(cfg RouterConfig) error {
// would serve a request with buffer limits cfg never asked for. // would serve a request with buffer limits cfg never asked for.
r.freeList = nil r.freeList = nil
if !workerMode { if !workerMode {
r.backoff = cfg.Backoff
r.numGoro = 0 r.numGoro = 0
r.pendingConns = nil r.pendingConns = nil
return nil return nil
@@ -225,7 +218,7 @@ func (r *Router) Configure(cfg RouterConfig) error {
NormalizeOutgoingKeys: cfg.NormalizeOutgoingKeys, NormalizeOutgoingKeys: cfg.NormalizeOutgoingKeys,
NoRequestBufferGrowth: true, // Hard memory limit. NoRequestBufferGrowth: true, // Hard memory limit.
}) })
go r.goroWorker(gen, jobqueue, cfg.Backoff, cfg.Mux) go r.goroWorker(gen, jobqueue, cfg.Mux)
} }
r.pendingConns = jobqueue r.pendingConns = jobqueue
r.numGoro = numgoro r.numGoro = numgoro
@@ -271,7 +264,7 @@ func (r *Router) Handle(conn io.ReadWriteCloser) error {
// Exchange acquisition and the configuration it is served with must be read // Exchange acquisition and the configuration it is served with must be read
// under the same lock: [Router.Configure] may run concurrently. // under the same lock: [Router.Configure] may run concurrently.
r.mu.Lock() r.mu.Lock()
numGoro, backoff, mux := r.numGoro, r.backoff, r.mux numGoro, mux := r.numGoro, r.mux
gen := r.gen.Load() // Generation whose buffers the exchange below is sized by. gen := r.gen.Load() // Generation whose buffers the exchange below is sized by.
if numGoro > 0 && r.pendingConns == nil { if numGoro > 0 && r.pendingConns == nil {
// Goroutines torn down: refuse before claiming an exchange. // Goroutines torn down: refuse before claiming an exchange.
@@ -284,7 +277,7 @@ func (r *Router) Handle(conn io.ReadWriteCloser) error {
return lneto.ErrExhausted return lneto.ErrExhausted
} else if numGoro == 0 { } else if numGoro == 0 {
r.mu.Unlock() r.mu.Unlock()
go r.goroHandle(gen, exch, backoff, mux) go r.goroHandle(gen, exch, mux)
return nil return nil
} }
// Enqueue under the lock: [Router.Configure] closes pendingConns while // Enqueue under the lock: [Router.Configure] closes pendingConns while
@@ -305,7 +298,7 @@ func (r *Router) Handle(conn io.ReadWriteCloser) error {
return lneto.ErrPacketDrop return lneto.ErrPacketDrop
} }
func (r *Router) goroWorker(gen uint32, queue chan job, backoff lneto.BackoffStrategy, mux Mux) { func (r *Router) goroWorker(gen uint32, queue chan job, mux Mux) {
for job := range queue { for job := range queue {
exch := job.exch exch := job.exch
if exch == nil { if exch == nil {
@@ -316,13 +309,13 @@ func (r *Router) goroWorker(gen uint32, queue chan job, backoff lneto.BackoffStr
exch.Release() exch.Release()
continue continue
} }
r.goroHandle(gen, exch, backoff, mux) r.goroHandle(gen, exch, mux)
} }
} }
func (r *Router) goroHandle(gen uint32, exch *Exchange, backoff lneto.BackoffStrategy, mux Mux) { func (r *Router) goroHandle(gen uint32, exch *Exchange, mux Mux) {
defer r.freeExch(gen, exch) defer r.freeExch(gen, exch)
err := Handle(exch, mux, backoff) err := Handle(exch, mux, nil)
if err != nil { if err != nil {
if exch.readErr != nil { if exch.readErr != nil {
r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error())) r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error()))
-8
View File
@@ -9,8 +9,6 @@ import (
"sync" "sync"
"testing" "testing"
"time" "time"
"github.com/soypat/lneto"
) )
// rwconn is a in-memory conn. The router handles connections on another // rwconn is a in-memory conn. The router handles connections on another
@@ -158,9 +156,6 @@ func configSynchronousRouter(t *testing.T, router *Router, bufferSize int, mux M
RequestHeaderBufferSize: bufferSize, RequestHeaderBufferSize: bufferSize,
RequestNumHeaderKVCap: 16, RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: bufferSize, ResponseHeaderMinBufferSize: bufferSize,
Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) {
return lneto.BackoffFlagNop
},
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -393,7 +388,6 @@ func TestRouterHandleAfterTeardown(t *testing.T) {
RequestHeaderBufferSize: 512, RequestHeaderBufferSize: 512,
RequestNumHeaderKVCap: 16, RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: 512, ResponseHeaderMinBufferSize: 512,
Backoff: nopBackoff,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -427,7 +421,6 @@ func TestRouterTeardownReleasesQueuedConns(t *testing.T) {
RequestHeaderBufferSize: 512, RequestHeaderBufferSize: 512,
RequestNumHeaderKVCap: 16, RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: 512, ResponseHeaderMinBufferSize: 512,
Backoff: nopBackoff,
} }
// Handing connections over and tearing down immediately leaves them queued // Handing connections over and tearing down immediately leaves them queued
// for workers that will never serve them. Rounds bound the scheduling luck // for workers that will never serve them. Rounds bound the scheduling luck
@@ -475,7 +468,6 @@ func TestRouterConfigureDuringWorkerHandle(t *testing.T) {
RequestHeaderBufferSize: 512, RequestHeaderBufferSize: 512,
RequestNumHeaderKVCap: 16, RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: 512, ResponseHeaderMinBufferSize: 512,
Backoff: nopBackoff,
} }
if err := router.Configure(cfg); err != nil { if err := router.Configure(cfg); err != nil {
t.Fatal(err) t.Fatal(err)