From a70c5a7505dd233b7e5c41f8afe2de35d84552db Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Sun, 26 Jul 2026 03:07:07 -0300 Subject: [PATCH] Router.Handle returns error after being torn down --- http/httphi/exchange.go | 8 +++--- http/httphi/router.go | 52 ++++++++++++++++++++++++++++++-------- http/httphi/router_test.go | 30 ++++++++++++++++++++++ 3 files changed, 76 insertions(+), 14 deletions(-) diff --git a/http/httphi/exchange.go b/http/httphi/exchange.go index 622985a..ffcb420 100644 --- a/http/httphi/exchange.go +++ b/http/httphi/exchange.go @@ -174,10 +174,10 @@ func (exch *Exchange) StageHeaderInt(key string, value int64, base int) (enoughM return true } -// StageWriteStatus prepares the status line for the given code without writing +// StageStatus prepares the status line for the given code without writing // it, i.e: "HTTP/1.1 404 Not Found". Codes with no [StatusText] get an empty // reason phrase. Has no effect once the header has been written. -func (exch *Exchange) StageWriteStatus(code int) { +func (exch *Exchange) StageStatus(code int) { if code >= 1000 || exch.headerWritten { return } else if code == 200 { @@ -200,7 +200,7 @@ func (exch *Exchange) StageWriteStatus(code int) { // fields. Only the first call reaches the wire, as in http.ResponseWriter. func (exch *Exchange) WriteHeader(code int) { if !exch.headerWritten { - exch.StageWriteStatus(code) + exch.StageStatus(code) exch.FlushHeader() } } @@ -215,7 +215,7 @@ func (exch *Exchange) FlushHeader() (int, error) { return 0, nil } if exch.respTopWritten == 0 { - exch.StageWriteStatus(200) + exch.StageStatus(200) } exch.headerWritten = true ng, err := exch.rw.Write(exch.respTopBuf[:exch.respTopWritten]) diff --git a/http/httphi/router.go b/http/httphi/router.go index 2f5b94e..11ebb5c 100644 --- a/http/httphi/router.go +++ b/http/httphi/router.go @@ -21,11 +21,24 @@ const reconfigureWait = 10 * time.Millisecond var ( errNoRequestProto = errors.New("httphi: request line with no HTTP version") errBusyExchanges = errors.New("httphi: exchanges still serving, cannot reuse their buffers") + errRouterTornDown = errors.New("httphi: router torn down, configure it before serving") ) type conn = io.ReadWriteCloser -// Router hosts concurrent safe data. +// Router serves HTTP connections handed to it with [Router.Handle], routing +// each request to a handler found through its [Mux]. It plays the part of +// http.Server minus the listening: accepting connections is the caller's job, +// which is what lets the same router run over a TCP stack, a socket or a test +// pipe. +// +// A Router owns the exchanges and goroutines that serve connections and sizes +// both at [Router.Configure] time, so serving load costs no allocation and +// bounded memory. Connections arriving with nothing left to serve them are +// refused rather than queued, see [Router.Handle]. +// +// Methods are safe for concurrent use. The zero value is not usable: configure +// it first. type Router struct { mu sync.Mutex gen atomic.Uint32 @@ -44,6 +57,7 @@ type Router struct { log *slog.Logger } +// job is a connection waiting on an exchange for a worker goroutine to serve it. type job struct { exch *Exchange } @@ -60,12 +74,21 @@ type RouterConfig struct { // Response buffer will reuse unused request memory so this is not a strict limit. ResponseMinBufferSize int + // NormalizeOutgoingKeys normalizes response header field keys as they are + // staged, i.e: "content-type" becomes "Content-Type". NormalizeOutgoingKeys bool - MaxAwaitingConns int + // MaxAwaitingConns is the depth of the queue connections wait in for a free + // goroutine. [Router.Handle] drops connections once it is full. Required and + // 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 Mux - Logger *slog.Logger + // Mux resolves each request's method and path to the handler serving it. Required. + Mux Mux + // Logger receives failed exchanges. Optional, nil disables logging. + Logger *slog.Logger } // Validate returns a non-nil error if the configuration cannot be used to @@ -86,7 +109,10 @@ func (cfg RouterConfig) workerMode() bool { return cfg.FixedNumGoroutines > 0 } -// Teardown stops fixed goroutines. +// TeardownGoroutines stops the router's fixed goroutines once they finish the +// exchanges they are serving. [Router.Configure] calls it before installing a +// new generation. A torn down router refuses connections with a non-nil error +// until it is configured again. func (r *Router) TeardownGoroutines() { r.mu.Lock() defer r.mu.Unlock() @@ -189,16 +215,22 @@ func (r *Router) awaitIdleExchangesLocked(maxWait time.Duration) error { // done. It does not block on the exchange: the connection is handed to a // goroutine and Handle returns immediately. // -// Handle returns [lneto.ErrExhausted] when no exchange is free and -// [lneto.ErrPacketDrop] when the job queue is full. Both leave conn untouched -// and unclosed for the caller to dispose of: dropping is how a router with -// fixed memory applies backpressure. +// Handle returns [lneto.ErrExhausted] when no exchange is free, +// [lneto.ErrPacketDrop] when the queue of connections awaiting a goroutine is +// full, and an error when the router's goroutines have been torn down. On every +// one of them conn is left untouched and unclosed for the caller to dispose of: +// refusing connections is how a router with fixed memory applies backpressure. func (r *Router) Handle(conn conn) 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() - exch := r.getExchLocked(conn) numGoro, backoff, mux := r.numGoro, r.backoff, r.mux + if numGoro > 0 && r.pendingConns == nil { + // Goroutines torn down: refuse before claiming an exchange. + r.mu.Unlock() + return errRouterTornDown + } + exch := r.getExchLocked(conn) if exch == nil { r.mu.Unlock() return lneto.ErrExhausted diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 3731d40..e556e51 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -334,6 +334,36 @@ func TestRouterConfigureHandleRace(t *testing.T) { // Reconfiguring a running router tears down the job queue that Handle may be // sending a connection on. Connections may be dropped, but never panic. +// A torn down router has nothing left to serve with: it must say so instead of +// dropping the connection as if it were merely busy. +func TestRouterHandleAfterTeardown(t *testing.T) { + var ( + sm MuxSlice + router Router + ) + sm.Handle("GET /", staticPage(t, "ok")) + err := router.Configure(RouterConfig{ + FixedNumGoroutines: 2, + MaxAwaitingConns: 4, + Mux: &sm, + RequestBufferSize: 512, + ResponseMinBufferSize: 512, + Backoff: nopBackoff, + }) + if err != nil { + t.Fatal(err) + } + router.TeardownGoroutines() + + conn := newConn("GET / HTTP/1.1\r\nHost: h\r\n\r\n") + if err = router.Handle(conn); err != errRouterTornDown { + t.Errorf("want errRouterTornDown, got %v", err) + } + if conn.IsClosed() { + t.Error("refused connection must be left for the caller to dispose of") + } +} + func TestRouterConfigureDuringWorkerHandle(t *testing.T) { var ( sm MuxSlice