diff --git a/protocol/h2/stream/pool_starvation_test.go b/protocol/h2/stream/pool_starvation_test.go new file mode 100644 index 00000000..715e4cb0 --- /dev/null +++ b/protocol/h2/stream/pool_starvation_test.go @@ -0,0 +1,123 @@ +package stream + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +// blockingPoolHandler blocks every stream handler until release is closed and +// reports how many handlers actually started. +type blockingPoolHandler struct { + started atomic.Int32 + entered chan struct{} + release chan struct{} +} + +func (h *blockingPoolHandler) HandleStream(_ context.Context, _ *Stream) error { + h.started.Add(1) + select { + case h.entered <- struct{}{}: + default: + } + <-h.release + return nil +} + +func (h *blockingPoolHandler) RouteAsync(_, _ string) bool { return true } +func (h *blockingPoolHandler) HasAsyncRoutes() bool { return true } + +// TestH2Pool_StreamingHandlersDoNotStarveLaterStreams pins celeris#520. +// +// An H2 stream handler is not guaranteed to return: an SSE stream, a long +// poll or any streaming handler holds its worker until the peer goes away. +// The pool used to queue onto a buffered channel whenever the channel had +// room, so once `size` such handlers were running every later stream sat in +// the buffer behind workers that would never come back -- and because the +// pool is process-global, that starved every H2 connection in the binary, +// not just the one that filled it. +// +// The pool is deliberately tiny here: the failure needs more concurrent +// streaming handlers than workers, which on a 4-vCPU CI runner (16 workers) +// is 17 SSE clients and on a 32-core host is 129. Sizing the pool down is +// what makes the bug reproducible on any machine. +func TestH2Pool_StreamingHandlersDoNotStarveLaterStreams(t *testing.T) { + const size = 2 + const streams = size * 4 + + p := newH2WorkerPool(size) + h := &blockingPoolHandler{entered: make(chan struct{}, streams), release: make(chan struct{})} + defer close(h.release) + + var wg sync.WaitGroup + for i := 0; i < streams; i++ { + proc := NewProcessor(h, newTestFrameWriter(), newTestResponseWriter()) + s := proc.manager.CreateStream(uint32(2*i + 1)) + if s == nil { + t.Fatalf("stream %d: CreateStream returned nil", i) + } + wg.Add(1) + go func() { + defer wg.Done() + p.Submit(proc, s) + }() + } + wg.Wait() + + deadline := time.Now().Add(5 * time.Second) + for h.started.Load() < int32(streams) && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if got := h.started.Load(); got != int32(streams) { + t.Fatalf("%d/%d streaming handlers started on a %d-worker pool: the rest are queued behind "+ + "workers that never return, and every later H2 stream in the process is stuck with them "+ + "(celeris#520)", got, streams, size) + } +} + +// TestH2Pool_IdleCreditTracksParkedWorkers pins the invariant Submit relies +// on: idle == (workers parked in the receive) - (tasks queued in work). With +// no work submitted every worker is parked, so idle must equal the pool size; +// it must never go negative. +func TestH2Pool_IdleCreditTracksParkedWorkers(t *testing.T) { + const size = 4 + p := newH2WorkerPool(size) + + deadline := time.Now().Add(2 * time.Second) + for p.idle.Load() != size && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + if got := p.idle.Load(); got != size { + t.Fatalf("idle=%d with an empty queue, want %d (every worker parked)", got, size) + } + + // Run a burst of handlers that DO return: the credit must come back. + h := &blockingPoolHandler{entered: make(chan struct{}, 64), release: make(chan struct{})} + close(h.release) // handlers return immediately + for i := 0; i < 64; i++ { + proc := NewProcessor(h, newTestFrameWriter(), newTestResponseWriter()) + s := proc.manager.CreateStream(uint32(2*i + 1)) + if s == nil { + t.Fatalf("stream %d: CreateStream returned nil", i) + } + p.Submit(proc, s) + if got := p.idle.Load(); got < 0 { + t.Fatalf("idle went negative (%d) after %d submits: a task was queued with no parked worker", got, i+1) + } + } + deadline = time.Now().Add(5 * time.Second) + for h.started.Load() < 64 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if got := h.started.Load(); got != 64 { + t.Fatalf("%d/64 returning handlers ran", got) + } + for p.idle.Load() != size && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if got := p.idle.Load(); got != size { + t.Fatalf("idle=%d after the burst drained, want %d: credits leak", got, size) + } +} diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go index 1e1128c4..f81241e5 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -28,8 +28,26 @@ var headersSlicePoolIn = sync.Pool{New: func() any { s := make([][2]string, 0, 1 // h2WorkerPool is a global goroutine pool for executing H2 stream handlers. // A single pool is shared across all connections to avoid per-connection // goroutine overhead (8 goroutines × N connections was catastrophic). +// +// A task is only ever QUEUED when a worker is parked waiting for one. That +// rule is the whole design: an H2 handler is not guaranteed to return. A +// Server-Sent Events stream, a long poll or any handler that streams for the +// life of its client holds its worker until the peer goes away, so a pool +// that queues into a buffered channel whenever the channel has room wedges as +// soon as `size` such handlers are running -- every later stream sits in the +// buffer behind workers that will never come back, on EVERY connection in the +// process, because the pool is global (celeris#520). Work that finds no +// parked worker runs on its own goroutine instead, which is what net/http +// does for every stream unconditionally. type h2WorkerPool struct { work chan h2Task + // idle is (workers parked in the receive) - (tasks sitting in work). + // A worker credits it before blocking on the channel; Submit claims a + // credit before it is allowed to queue. It is therefore positive only + // when a parked worker will pick the task up promptly, and drops to + // zero the moment the queue catches up with the parked workers -- see + // the invariant proof in TestH2Pool_IdleCreditTracksParkedWorkers. + idle atomic.Int32 } type h2Task struct { @@ -49,19 +67,57 @@ func newH2WorkerPool(size int) *h2WorkerPool { return p } +// run is one pool worker. func (p *h2WorkerPool) run() { - for task := range p.work { + for { + // Credit the pool before parking. Submit spends exactly one + // credit per queued task, so the counter stays equal to + // parked-workers minus queued-tasks no matter which worker + // ends up taking which task. + p.idle.Add(1) + task, ok := <-p.work + if !ok { + p.idle.Add(-1) + return + } task.proc.executeHandler(task.stream) } } -// Submit tries to dispatch to a pooled worker. If all workers are busy -// and the channel is full, falls back to a one-shot goroutine to avoid -// blocking the event loop (which would stall frame processing). +// Submit dispatches a stream handler. It queues onto the shared pool while a +// worker is parked waiting for work, and otherwise runs the handler on its +// own goroutine. +// +// It never blocks the event loop: blocking there would stall frame +// processing for every stream on the connection. +// +// Growing the pool instead of spawning a one-shot goroutine was measured and +// rejected: a surplus worker that joins the pool and retires on an idle TTL +// made the dispatch benchmark SLOWER (865 ns/op vs 793) while adding a +// window in which a traffic spike is held as parked goroutines. The spawn is +// what net/http pays for every stream, unconditionally. func (p *h2WorkerPool) Submit(proc *Processor, s *Stream) { + for { + n := p.idle.Load() + if n <= 0 { + // No parked worker. Queueing here is what used to wedge the + // pool (celeris#520): a streaming handler holds its worker + // for the life of the stream, so a queued task would wait + // behind handlers that never return. + go proc.executeHandler(s) + return + } + if p.idle.CompareAndSwap(n, n-1) { + break + } + } select { case p.work <- h2Task{proc, s}: default: + // Unreachable while the buffer is size*16 and we only queue + // against a parked worker, but a dropped task would hang a + // stream forever, so hand it a goroutine rather than trust that. + p.idle.Add(1) go proc.executeHandler(s) } }