improve locking and acquisition of Exchanges in reconfiguring

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)
+	}
+}

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
+	Backoff lneto.BackoffStrategy
-	Logger                *slog.Logger
+	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.
+			// Exchange buffers below are reused: the previous generation must be
-			time.Sleep(5 * time.Millisecond)
+			// 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 {
+		if needMore {
-			// Done!
+			continue // Request header split across reads, accumulate the rest.
-			break
 		} 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
+	numGoro, backoff, mux := r.numGoro, r.backoff, r.mux
-	r.mu.Unlock()
 	if exch == nil {
+		r.mu.Unlock()
 		return lneto.ErrExhausted
-	}
+	} else if numGoro == 0 {
-
+		r.mu.Unlock()
-	if numGoro == 0 {
 		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}:
+	case r.pendingConns <- job{exch: exch}:
-		return nil
+		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
+		exch.nextFree = nil // Freelist full, exchange is dropped.
-		depth := 0
-		for depth < freelistMaxDepth && node.nextFree != nil {
-			node = node.nextFree
-			depth++
-		}
-		node.nextFree = exch
 	}
 	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 {
+	if exch.respErr != nil {
-		exch.FlushHeader()
+		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) {

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()
+}