From 97c6936cf9ae95adbcc7c22d8786743522a30c46 Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Sat, 25 Jul 2026 21:31:01 -0300 Subject: [PATCH] improve locking and acquisition of Exchanges in reconfiguring --- http/httphi/exchange_test.go | 85 ++++++++++++++++++++++ http/httphi/httphi.go | 137 ++++++++++++++++++++++++++--------- http/httphi/router_test.go | 85 +++++++++++++++++++++- 3 files changed, 273 insertions(+), 34 deletions(-) diff --git a/http/httphi/exchange_test.go b/http/httphi/exchange_test.go index 58afac5..754057d 100644 --- a/http/httphi/exchange_test.go +++ b/http/httphi/exchange_test.go @@ -1,6 +1,8 @@ package httphi import ( + "context" + "errors" "strings" "testing" "time" @@ -330,3 +332,86 @@ func TestHandleLeavesConnOpen(t *testing.T) { } } } + +// Hijacking hands the connection to the handler, so Release must not close it. +// Ownership must not carry over: the next connection the exchange serves is +// the router's again and must be closed on Release. +func TestExchangeHijackOwnership(t *testing.T) { + var sm sliceMux + var hijackErr error + sm.Handle("GET /", func(ex *Exchange) { + _, _, hijackErr = ex.HijackRaw(nil) + }) + first := newConn("GET / HTTP/1.1\r\nHost: h\r\n\r\n") + first.Hangup() + exch := newExchange(t, first, 1024, false) + if err := Handle(exch, &sm, nopBackoff); err != nil { + t.Fatal(err) + } + if hijackErr != nil { + t.Fatal(hijackErr) + } + exch.Release() + if first.IsClosed() { + t.Error("hijacked connection must stay open after Release") + } + + second := newConn("") + if !exch.Acquire(second) { + t.Fatal("released exchange must be acquirable") + } + exch.Release() + if !second.IsClosed() { + t.Error("connection must be closed on Release: hijack of a previous request must not carry over") + } +} + +// Idle peer policy belongs to the connection: Handle keeps retrying an empty +// read until the conn itself reports failure, so a stalled peer ends the +// exchange through the conn's deadline instead of pinning the exchange. +func TestHandleIdlePeerEndsOnConnDeadline(t *testing.T) { + var sm sliceMux + sm.Handle("/", func(ex *Exchange) { t.Error("handler must not run on partial request") }) + conn := newConn("GET / HTTP") // Peer stalls mid request line, never hangs up. + conn.SetDeadline(time.Now().Add(10 * time.Millisecond)) + exch := newExchange(t, conn, 1024, false) + + done := make(chan error, 1) + go func() { done <- Handle(exch, &sm, nopBackoff) }() + select { + case err := <-done: + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("want connection deadline error, got %v", err) + } + case <-time.After(time.Second): + t.Fatal("Handle ignored the connection deadline") + } +} + +// A body must never reach the wire without its header: if flushing the header +// fails, Write must report the failure and send nothing. +func TestExchangeWriteHeaderFlushFails(t *testing.T) { + const body = "body" + conn := newConn("") + exch := newExchange(t, conn, 128, false) + conn.FailWrites(1) // Status line write fails, body write would succeed. + + n, err := exch.Write([]byte(body)) + if err == nil { + t.Error("want error when header flush fails, got nil") + } + if n != 0 { + t.Errorf("want 0 bytes written, got %d", n) + } + if got := conn.ViewWritten(); got != "" { + t.Errorf("want nothing on the wire, got %q", got) + } + // Writes after a failed header stay failed: the response is unrecoverable, + // a body without its header would corrupt the stream. + if _, err = exch.Write([]byte(body)); err == nil { + t.Error("want error on write after failed header flush, got nil") + } + if got := conn.ViewWritten(); got != "" { + t.Errorf("want nothing on the wire, got %q", got) + } +} diff --git a/http/httphi/httphi.go b/http/httphi/httphi.go index 7a01104..50c2046 100644 --- a/http/httphi/httphi.go +++ b/http/httphi/httphi.go @@ -17,7 +17,14 @@ import ( //go:generate stringer -type Method,status -linecomment -output stringers.go -var errNoRequestProto = errors.New("httphi: request line with no HTTP version") +// defaultReconfigureWait is how long [Router.Configure] waits on a busy +// previous generation when [RouterConfig.MaxReconfigureWait] is unset. +const defaultReconfigureWait = 100 * time.Millisecond + +var ( + errNoRequestProto = errors.New("httphi: request line with no HTTP version") + errBusyExchanges = errors.New("httphi: exchanges still serving, cannot reuse their buffers") +) type conn = io.ReadWriteCloser @@ -61,9 +68,10 @@ type RouterConfig struct { NormalizeOutgoingKeys bool MaxAwaitingConns int - Backoff lneto.BackoffStrategy - Mux Mux - Logger *slog.Logger + + Backoff lneto.BackoffStrategy + Mux Mux + Logger *slog.Logger } func (cfg RouterConfig) Validate() error { @@ -84,9 +92,19 @@ func (cfg RouterConfig) workerMode() bool { // Teardown stops fixed goroutines. func (r *Router) TeardownGoroutines() { + r.mu.Lock() + defer r.mu.Unlock() + r.teardownGoroutinesLocked() +} + +// teardownGoroutinesLocked closes the job queue fixed goroutines feed from. +// Requires r.mu held so that a concurrent [Router.Handle] cannot be enqueueing +// on the channel being closed. +func (r *Router) teardownGoroutinesLocked() { r.gen.Add(1) if r.pendingConns != nil { close(r.pendingConns) + r.pendingConns = nil } } @@ -96,7 +114,7 @@ func (r *Router) Configure(cfg RouterConfig) error { } r.mu.Lock() defer r.mu.Unlock() - r.TeardownGoroutines() + r.teardownGoroutinesLocked() gen := r.gen.Load() numgoro := cfg.FixedNumGoroutines workerMode := cfg.workerMode() @@ -113,9 +131,14 @@ func (r *Router) Configure(cfg RouterConfig) error { if workerMode { jobqueue := make(chan job, cfg.MaxAwaitingConns) if gen > 1 { - // Previously existing goroutine manager, wait a bit for it to close. - time.Sleep(5 * time.Millisecond) + // Exchange buffers below are reused: the previous generation must be + // done serving before they may be handed to the new one. + err := r.awaitIdleExchangesLocked(10 * time.Millisecond) + if err != nil { + return err + } } + r.freeList = nil // Freelist entries point into the buffers reused below. internal.SliceReuse(&r.exchs, numgoro) r.exchs = r.exchs[:numgoro] rawBuflen := cfg.RequestBufferSize + cfg.ResponseMinBufferSize @@ -132,6 +155,31 @@ func (r *Router) Configure(cfg RouterConfig) error { return nil } +// awaitIdleExchangesLocked waits up to maxWait for exchanges of the previous +// generation to finish serving so their buffers may be reused. Requires r.mu +// held; the lock is released while waiting since [Router.freeExch] needs it to +// free the exchanges being waited on. +func (r *Router) awaitIdleExchangesLocked(maxWait time.Duration) error { + const pollInterval = time.Millisecond + for waited := time.Duration(0); ; waited += pollInterval { + busy := false + for i := range r.exchs { + if r.exchs[i].used.Load() { + busy = true + break + } + } + if !busy { + return nil + } else if waited >= maxWait { + return errBusyExchanges + } + r.mu.Unlock() + time.Sleep(pollInterval) + r.mu.Lock() + } +} + // 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. @@ -152,12 +200,12 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { consecutiveBackoffs = 0 const asRequest = false needMore, err := reqhdr.TryParse(asRequest) - if !needMore && err == nil { - // Done! - break + if needMore { + continue // Request header split across reads, accumulate the rest. } else if err != nil { return err } + break // Done! } // Setup Exchange fields necessary for correct functioning. parsed := reqhdr.BufferParsed() @@ -189,23 +237,30 @@ func (r *Router) Handle(conn conn) error { // under the same lock: [Router.Configure] may run concurrently. r.mu.Lock() exch := r.getExchLocked(conn) - numGoro, backoff, mux, queue := r.numGoro, r.backoff, r.mux, r.pendingConns - r.mu.Unlock() + numGoro, backoff, mux := r.numGoro, r.backoff, r.mux if exch == nil { + r.mu.Unlock() return lneto.ErrExhausted - } - - if numGoro == 0 { + } else if numGoro == 0 { + r.mu.Unlock() go r.goroHandle(exch, backoff, mux) return nil } + // Enqueue under the lock: [Router.Configure] closes pendingConns while + // holding it, so an unlocked send could land on a closed channel. The send + // never blocks, so holding the lock cannot stall a worker. + var enqueued bool select { - case queue <- job{exch: exch}: - return nil + case r.pendingConns <- job{exch: exch}: + enqueued = true default: // pendingConns cannot store another Conn, we drop and return error. exch.used.Store(false) // release. } + r.mu.Unlock() + if enqueued { + return nil + } return lneto.ErrPacketDrop } @@ -236,16 +291,17 @@ func (r *Router) goroHandle(exch *Exchange, backoff lneto.BackoffStrategy, mux M func (r *Router) freeExch(exch *Exchange) { const freelistMaxDepth = 5 r.mu.Lock() - if r.freeList == nil { + depth := 0 + for node := r.freeList; node != nil && depth < freelistMaxDepth; node = node.nextFree { + depth++ + } + if depth < freelistMaxDepth { + // Push at head: appending at the tail would drop every node past the + // depth limit instead of dropping the exchange we cannot store. + exch.nextFree = r.freeList r.freeList = exch } else { - node := r.freeList - depth := 0 - for depth < freelistMaxDepth && node.nextFree != nil { - node = node.nextFree - depth++ - } - node.nextFree = exch + exch.nextFree = nil // Freelist full, exchange is dropped. } exch.Release() r.mu.Unlock() @@ -254,13 +310,14 @@ func (r *Router) freeExch(exch *Exchange) { // getExchLocked returns an exchange acquired on conn. Requires r.mu held. func (r *Router) getExchLocked(conn conn) (exch *Exchange) { if r.freeList != nil { + // Successor must be read before Acquire: Acquire clears nextFree, so + // popping afterwards would truncate the freelist to the popped node. + next := r.freeList.nextFree if r.freeList.Acquire(conn) { exch = r.freeList + r.freeList = next + return exch } - r.freeList = r.freeList.nextFree - } - if exch != nil { - return exch } for i := range r.exchs { if r.exchs[i].Acquire(conn) { @@ -303,6 +360,7 @@ type Exchange struct { rw conn respRemains int + respErr error // Sticky: response is unrecoverable once a write fails. headerWritten bool normalizeKeys bool nextFree *Exchange @@ -353,6 +411,8 @@ func (exch *Exchange) Acquire(conn conn) bool { return false } exch.readErr = nil + exch.respErr = nil + exch.hijacked = false exch.respTopWritten = 0 exch.respHeaderOff = 0 exch.respHeaderLen = 0 @@ -420,7 +480,9 @@ func (exch *Exchange) WriteHeader(code int) { } } func (exch *Exchange) FlushHeader() (int, error) { - if exch.headerWritten { + if exch.respErr != nil { + return 0, exch.respErr + } else if exch.headerWritten { return 0, nil } if exch.respTopWritten == 0 { @@ -429,6 +491,7 @@ func (exch *Exchange) FlushHeader() (int, error) { exch.headerWritten = true ng, err := exch.rw.Write(exch.respTopBuf[:exch.respTopWritten]) if err != nil { + exch.respErr = err return ng, err } off := int(exch.respHeaderOff) @@ -436,17 +499,25 @@ func (exch *Exchange) FlushHeader() (int, error) { headers[len(headers)-1] = '\n' headers[len(headers)-2] = '\r' ng2, err := exch.rw.Write(headers) + exch.respErr = err return ng + ng2, err } func (exch *Exchange) Write(buf []byte) (int, error) { - if !exch.headerWritten { - exch.FlushHeader() + if exch.respErr != nil { + return 0, exch.respErr + } else if !exch.headerWritten { + _, err := exch.FlushHeader() + if err != nil { + return 0, err // Body must not reach the wire without its header. + } } if len(buf) == 0 { return 0, nil } - return exch.rw.Write(buf) + n, err := exch.rw.Write(buf) + exch.respErr = err + return n, err } func (exch *Exchange) ReadBody(dst []byte) (n int, _ error) { diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 7b9940c..598af03 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -23,9 +23,11 @@ import ( type rwconn struct { mu sync.Mutex readable bytes.Buffer + segments []string written bytes.Buffer closed bool hangup bool + failWr int onClose chan struct{} deadline time.Time } @@ -38,6 +40,23 @@ func newConn(request string) *rwconn { return r } +// AddSegment queues data delivered on a later read, once everything already +// pending has been read. Models a request split over several TCP segments +// without depending on goroutine scheduling. +func (r *rwconn) AddSegment(b string) { + r.mu.Lock() + defer r.mu.Unlock() + r.segments = append(r.segments, b) +} + +// FailWrites makes the next n writes fail, as a conn refusing further data +// would. Later writes succeed. +func (r *rwconn) FailWrites(n int) { + r.mu.Lock() + defer r.mu.Unlock() + r.failWr = n +} + // Hangup makes reads past the pending data return [io.EOF], as a peer that // closed its side of the connection would. func (r *rwconn) Hangup() { @@ -76,6 +95,11 @@ func (r *rwconn) Read(b []byte) (int, error) { } else if r.deadlineExceeded() { return 0, context.DeadlineExceeded } else if r.readable.Len() == 0 { + if len(r.segments) > 0 { + r.readable.WriteString(r.segments[0]) + r.segments = r.segments[1:] + return r.readable.Read(b) + } if r.hangup { return 0, io.EOF } @@ -90,6 +114,9 @@ func (r *rwconn) Write(b []byte) (int, error) { return 0, net.ErrClosed } else if r.deadlineExceeded() { return 0, context.DeadlineExceeded + } else if r.failWr > 0 { + r.failWr-- + return 0, io.ErrShortWrite } return r.written.Write(b) } @@ -101,6 +128,14 @@ func (r *rwconn) AddReadable(b []byte) { defer r.mu.Unlock() r.readable.Write(b) } +// SetDeadline makes reads and writes past t fail, as a conn with a read +// deadline set would. +func (r *rwconn) SetDeadline(t time.Time) { + r.mu.Lock() + defer r.mu.Unlock() + r.deadline = t +} + // IsClosed reports whether the connection was closed by its handler. func (r *rwconn) IsClosed() bool { r.mu.Lock() @@ -281,10 +316,11 @@ func TestRouterSplitRequest(t *testing.T) { configSynchronousRouter(t, &router, bufferSize, &sm) conn := newConn("GET / HTTP/1.1\r\nHo") + conn.AddSegment("st: tinygo.org\r\n\r") + conn.AddSegment("\n") // Final CRLF lands in its own segment. if err := router.Handle(conn); err != nil { t.Fatal(err) } - conn.AddReadable([]byte("st: tinygo.org\r\n\r\n")) conn.AwaitClose(t, time.Second) if got := conn.ViewWritten(); !strings.HasSuffix(got, expectResponse) { @@ -329,3 +365,50 @@ func TestRouterConfigureHandleRace(t *testing.T) { }() wg.Wait() } + +// Reconfiguring a running router tears down the job queue that Handle may be +// sending a connection on. Connections may be dropped, but never panic. +func TestRouterConfigureDuringWorkerHandle(t *testing.T) { + var ( + sm sliceMux + router Router + ) + sm.Handle("GET /", staticPage(t, "ok")) + cfg := RouterConfig{ + FixedNumGoroutines: 2, + MaxAwaitingConns: 4, + Mux: &sm, + RequestBufferSize: 512, + ResponseMinBufferSize: 512, + Backoff: nopBackoff, + } + if err := router.Configure(cfg); err != nil { + t.Fatal(err) + } + defer router.TeardownGoroutines() + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for i := 0; i < 300; i++ { + conn := newConn("GET / HTTP/1.1\r\nHost: h\r\n\r\n") + conn.Hangup() + router.Handle(conn) // Drops are fine, panics are not. + } + }() + go func() { + defer wg.Done() + // Each Configure sleeps 5ms tearing down the previous generation, keep + // the count low and let the Handle loop supply the concurrency. + for i := 0; i < 20; i++ { + // errBusyExchanges is legitimate backpressure: the previous + // generation was still serving when the buffers were needed. + if err := router.Configure(cfg); err != nil && err != errBusyExchanges { + t.Error(err) + return + } + } + }() + wg.Wait() +}