From c97c5e17cf4e516c85bcfd0b20ab4b47ba36d326 Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Wed, 29 Jul 2026 11:32:26 -0300 Subject: [PATCH] remove backoff assumption from Router --- examples/http-linux/main-httplinux.go | 7 ++----- http/httphi/example_test.go | 2 -- http/httphi/mux.go | 6 ++++++ http/httphi/router.go | 23 ++++++++--------------- http/httphi/router_test.go | 8 -------- 5 files changed, 16 insertions(+), 30 deletions(-) diff --git a/examples/http-linux/main-httplinux.go b/examples/http-linux/main-httplinux.go index 2b260da..3af8176 100644 --- a/examples/http-linux/main-httplinux.go +++ b/examples/http-linux/main-httplinux.go @@ -52,11 +52,8 @@ func run() error { RequestNumHeaderKVCap: numHeaderFields, ResponseHeaderMinBufferSize: bufferSizes, MaxAwaitingConns: 256, - Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) { - return min(time.Second, time.Millisecond*time.Duration(consecutiveBackoffs)) - }, - Mux: &mux, - Logger: slog.Default(), + Mux: &mux, + Logger: slog.Default(), }) if err != nil { return err diff --git a/http/httphi/example_test.go b/http/httphi/example_test.go index e1503f0..24e92d8 100644 --- a/http/httphi/example_test.go +++ b/http/httphi/example_test.go @@ -6,7 +6,6 @@ import ( "log" "log/slog" "os" - "time" "github.com/soypat/lneto/http/httphi" "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. RequestNumHeaderKVCap: numHeaderKV, NormalizeOutgoingKeys: true, - Backoff: func(uint) time.Duration { return time.Millisecond }, Mux: &mux, Logger: slog.Default(), }) diff --git a/http/httphi/mux.go b/http/httphi/mux.go index 0e6e4e4..e6c8950 100644 --- a/http/httphi/mux.go +++ b/http/httphi/mux.go @@ -1,6 +1,7 @@ package httphi import ( + "io" "strings" "unsafe" @@ -12,6 +13,8 @@ import ( // 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. // 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 { if !exch.acquired.Load() { return lneto.ErrBadState @@ -26,6 +29,9 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { exch.handleError(err) return err } else if n == 0 { + if backoff == nil { + return io.ErrNoProgress + } backoff.Do(consecutiveBackoffs) consecutiveBackoffs++ continue diff --git a/http/httphi/router.go b/http/httphi/router.go index 1c21ef4..f46d36a 100644 --- a/http/httphi/router.go +++ b/http/httphi/router.go @@ -59,8 +59,7 @@ type Router struct { exchs []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. @@ -92,9 +91,6 @@ type RouterConfig struct { // must be non-zero when running a fixed number of goroutines, unused otherwise. 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 Mux // Logger receives failed exchanges. Optional, nil disables logging. @@ -127,8 +123,6 @@ func (cfg RouterConfig) Validate() error { cfg.ResponseHeaderMinBufferSize > maxExchangeBuffer, cfg.RequestHeaderBufferSize > maxExchangeBuffer-cfg.ResponseHeaderMinBufferSize: return lneto.ErrInvalidConfig - case cfg.Backoff == nil: - return lneto.ErrMissingHALConfig } if workerMode { // 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. r.freeList = nil if !workerMode { - r.backoff = cfg.Backoff r.numGoro = 0 r.pendingConns = nil return nil @@ -225,7 +218,7 @@ func (r *Router) Configure(cfg RouterConfig) error { NormalizeOutgoingKeys: cfg.NormalizeOutgoingKeys, 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.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 // under the same lock: [Router.Configure] may run concurrently. 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. if numGoro > 0 && r.pendingConns == nil { // Goroutines torn down: refuse before claiming an exchange. @@ -284,7 +277,7 @@ func (r *Router) Handle(conn io.ReadWriteCloser) error { return lneto.ErrExhausted } else if numGoro == 0 { r.mu.Unlock() - go r.goroHandle(gen, exch, backoff, mux) + go r.goroHandle(gen, exch, mux) return nil } // Enqueue under the lock: [Router.Configure] closes pendingConns while @@ -305,7 +298,7 @@ func (r *Router) Handle(conn io.ReadWriteCloser) error { 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 { exch := job.exch if exch == nil { @@ -316,13 +309,13 @@ func (r *Router) goroWorker(gen uint32, queue chan job, backoff lneto.BackoffStr exch.Release() 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) - err := Handle(exch, mux, backoff) + err := Handle(exch, mux, nil) if err != nil { if exch.readErr != nil { r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error())) diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 50a625c..242883c 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -9,8 +9,6 @@ import ( "sync" "testing" "time" - - "github.com/soypat/lneto" ) // 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, RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: bufferSize, - Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) { - return lneto.BackoffFlagNop - }, }) if err != nil { t.Fatal(err) @@ -393,7 +388,6 @@ func TestRouterHandleAfterTeardown(t *testing.T) { RequestHeaderBufferSize: 512, RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: 512, - Backoff: nopBackoff, }) if err != nil { t.Fatal(err) @@ -427,7 +421,6 @@ func TestRouterTeardownReleasesQueuedConns(t *testing.T) { RequestHeaderBufferSize: 512, RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: 512, - Backoff: nopBackoff, } // Handing connections over and tearing down immediately leaves them queued // for workers that will never serve them. Rounds bound the scheduling luck @@ -475,7 +468,6 @@ func TestRouterConfigureDuringWorkerHandle(t *testing.T) { RequestHeaderBufferSize: 512, RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: 512, - Backoff: nopBackoff, } if err := router.Configure(cfg); err != nil { t.Fatal(err)