From d7ade0f82e46b4d1a137abeea2122c0d3f69bff9 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Sat, 11 Jul 2026 11:01:51 +0200 Subject: [PATCH 1/3] =?UTF-8?q?fix(h2):=20spread=20write-queue=20shards=20?= =?UTF-8?q?across=20all=204=20(streamID>>1)=20=E2=80=94=20celeris#406?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Client HTTP/2 request streams are always odd (RFC 9113 §5.1.1) and Celeris does no server push, so streamID%h2QueueShards (=4) only ever hit shards {1,3} and left 0,2 dead — the 4-shard queue was effectively 2 shards. Shift off the constant low bit so odd IDs 1,3,5,7,… spread across all shards. Pure function of streamID, so per-stream HEADERS-before-DATA FIFO + single-writer drain are unchanged; only the shard distribution improves. Adds a regression test (fails on %4: shards 0,2 get 0 bufs). --- internal/conn/h2.go | 9 ++++- internal/conn/h2_sharding_test.go | 58 +++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 internal/conn/h2_sharding_test.go diff --git a/internal/conn/h2.go b/internal/conn/h2.go index b32144ee..3b31a04f 100644 --- a/internal/conn/h2.go +++ b/internal/conn/h2.go @@ -104,7 +104,14 @@ type h2QueueShard struct { // Enqueue appends pre-encoded frame bytes to the write queue. // Called from handler goroutines. Shards by stream ID. func (q *h2ShardedQueue) Enqueue(streamID uint32, data *[]byte) { - shard := &q.shards[streamID%h2QueueShards] + // Client request streams are always odd (RFC 9113 §5.1.1) and Celeris does + // no server push, so streamID%h2QueueShards would only ever hit the odd + // residues (shards 1,3) and leave 0,2 dead — a 2-shard queue in disguise. + // Shift off the constant low bit first so odd IDs 1,3,5,7,… spread across + // ALL shards. Still a pure function of streamID → every frame of a stream + // maps to one shard (per-stream HEADERS-before-DATA FIFO + single-writer + // drain preserved). See celeris#406. + shard := &q.shards[(streamID>>1)%h2QueueShards] shard.mu.Lock() shard.bufs = append(shard.bufs, data) shard.mu.Unlock() diff --git a/internal/conn/h2_sharding_test.go b/internal/conn/h2_sharding_test.go new file mode 100644 index 00000000..5b3cceb1 --- /dev/null +++ b/internal/conn/h2_sharding_test.go @@ -0,0 +1,58 @@ +package conn + +import "testing" + +// TestH2ShardedQueue_OddStreamShardDistribution is a regression guard for +// celeris#406. HTTP/2 client request streams are always odd (RFC 9113 §5.1.1) +// and Celeris does no server push, so the old streamID%h2QueueShards collapsed +// every response onto the odd residues (shards 1,3) and left 0,2 permanently +// dead — a 4-shard queue that behaved like 2. The (streamID>>1) hash must +// spread odd IDs across ALL shards while keeping each stream affine to one. +func TestH2ShardedQueue_OddStreamShardDistribution(t *testing.T) { + var q h2ShardedQueue + q.wakeupFD = -1 // pure in-memory; no eventfd signaling + + // Enqueue one frame per odd stream ID (1,3,5,…), like real client streams. + const perShard = 4 + n := perShard * h2QueueShards + for i := 0; i < n; i++ { + q.Enqueue(uint32(2*i+1), getH2FrameBuf()) + } + + // Every shard must be exercised, evenly. Fails on the old %4 (shards 0,2 + // get 0 bufs). + for i := range q.shards { + if got := len(q.shards[i].bufs); got != perShard { + t.Fatalf("shard %d got %d bufs, want %d: odd stream IDs not spread across all %d shards (celeris#406)", + i, got, perShard, h2QueueShards) + } + } + + // Stream affinity: the same stream ID always lands in exactly one shard. + var q2 h2ShardedQueue + q2.wakeupFD = -1 + const affID = 5 + for i := 0; i < 3; i++ { + q2.Enqueue(affID, getH2FrameBuf()) + } + nonEmpty := 0 + for i := range q2.shards { + switch len(q2.shards[i].bufs) { + case 0: + case 3: + nonEmpty++ + default: + t.Fatalf("stream %d split across shards: shard %d has %d bufs", affID, i, len(q2.shards[i].bufs)) + } + } + if nonEmpty != 1 { + t.Fatalf("stream %d landed in %d shards, want exactly 1", affID, nonEmpty) + } + + // DrainTo must return every buffer, in order, from the single-writer drain. + got := 0 + q.DrainTo(func([]byte) { got++ }) + if got != n { + t.Fatalf("DrainTo wrote %d bufs, want %d", got, n) + } +} From bcdc8f1a53a5b997a86d2564dbf7bf06d8ea8830 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Sat, 11 Jul 2026 11:05:57 +0200 Subject: [PATCH 2/3] =?UTF-8?q?fix(overload):=20fire=20Reap=20GC=20once=20?= =?UTF-8?q?on=20entry,=20not=20every=20poll=20tick=20=E2=80=94=20celeris#4?= =?UTF-8?q?07?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StageReap forced a full runtime.GC() on every poll tick while the signal sat in the Reap band (default CPU 0.80-0.85), not once on the transition into Reap. Under sustained moderate load that runs a STW GC every PollInterval (default 1s) — repeated capacity reclaim precisely when the server is already loaded, doing more harm (GC CPU + STW pauses) than good. Track prevStage (goroutine-local; run() is the sole writer of the stage atomic) and fire runtime.GC() only on prev!=StageReap && new==StageReap. Re-entry (Reap->Reorder->Reap) correctly re-fires. Gated behind EnableReap (opt-in, default off). Also documents ReapAggressiveness==2 as reserved (it currently behaves identically to 1; no pool-drain primitive is wired). Adds a NumForcedGC-delta regression test (29 GCs while lingering pre-fix, <=1 post-fix). --- middleware/overload/config.go | 6 ++- middleware/overload/overload.go | 16 ++++++- middleware/overload/overload_reap_test.go | 52 +++++++++++++++++++++++ 3 files changed, 71 insertions(+), 3 deletions(-) create mode 100644 middleware/overload/overload_reap_test.go diff --git a/middleware/overload/config.go b/middleware/overload/config.go index 2fea1c22..ccb2a39c 100644 --- a/middleware/overload/config.go +++ b/middleware/overload/config.go @@ -158,8 +158,10 @@ type Config struct { // can measure the effect. EnableReap bool - // ReapAggressiveness: 1=GC hint, 2=GC + encourage pool drain. - // Default: 1. + // ReapAggressiveness gates the opt-in GC fired on entry into StageReap: + // any value >= 1 triggers a one-shot runtime.GC(). Level 2 ("GC + encourage + // pool drain") is reserved and currently behaves identically to level 1 — + // no pool-drain primitive is wired yet (celeris#407). Default: 1. ReapAggressiveness int // Skip defines a function to skip this middleware for certain diff --git a/middleware/overload/overload.go b/middleware/overload/overload.go index f21ad49a..8ddac91e 100644 --- a/middleware/overload/overload.go +++ b/middleware/overload/overload.go @@ -231,6 +231,12 @@ func run(ctx context.Context, cfg Config, defer close(stopped) t := time.NewTicker(cfg.PollInterval) defer t.Stop() + // prevStage tracks the stage applied on the previous tick so the opt-in + // Reap GC fires only on the transition INTO Reap, not on every tick the + // stage lingers in the band (celeris#407). Goroutine-local: run() is the + // sole writer of the `stage` atomic and the sole owner of prevStage, so + // this does not affect the stage value the hot path reads. + prevStage := StageNormal for { select { case <-ctx.Done(): @@ -256,11 +262,19 @@ func run(ctx context.Context, cfg Config, newStage = latencyStage } stage.Store(int32(newStage)) - if newStage == StageReap && cfg.EnableReap { + // Fire the opt-in GC only on ENTRY into Reap. A forced GC every + // PollInterval while the stage lingers in the Reap band (celeris#407) + // wrecks tail latency and defeats the stage's purpose — a Reap GC is + // a one-shot capacity reclaim, not a periodic sweep. Re-entry + // (Reap→Reorder→Reap) correctly re-fires. The col==nil / cpu<0 skip + // branches above `continue` before both stage.Store and this + // assignment, so a skipped tick advances neither. + if cfg.EnableReap && newStage == StageReap && prevStage != StageReap { if cfg.ReapAggressiveness >= 1 { runtime.GC() } } + prevStage = newStage _ = inFlight // surfaced through Controller; not needed in poll } } diff --git a/middleware/overload/overload_reap_test.go b/middleware/overload/overload_reap_test.go new file mode 100644 index 00000000..375fec7f --- /dev/null +++ b/middleware/overload/overload_reap_test.go @@ -0,0 +1,52 @@ +package overload + +import ( + "runtime" + "testing" + "time" +) + +// TestReapGCFiresOnceOnEntryNotPerTick is a regression guard for celeris#407: +// the opt-in Reap GC must fire on the transition INTO Reap, not on every poll +// tick the stage lingers in the Reap band. Pre-fix this forced a GC (STW) every +// PollInterval under sustained moderate load. +func TestReapGCFiresOnceOnEntryNotPerTick(t *testing.T) { + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + before := ms.NumForcedGC + + m := &mockCPU{} + m.set(0.82) // in the Reap band [Reap=0.80, Reorder=0.85) + poll := 5 * time.Millisecond + _, ctrl := NewWithController(Config{ + CollectorProvider: withMock(m), + PollInterval: poll, + EnableReap: true, + ReapAggressiveness: 1, + }) + defer ctrl.Stop() + + waitForStage(t, ctrl, StageReap) + // waitForStage returns the instant stage.Store publishes StageReap, but the + // entry runtime.GC() runs after that in the same tick — wait a few ticks so + // it (and only it) has completed before sampling. + time.Sleep(3 * poll) + + // Entering Reap must fire the opt-in GC exactly once. + runtime.ReadMemStats(&ms) + atEntry := ms.NumForcedGC + if atEntry <= before { + t.Fatalf("entering Reap did not fire the opt-in GC (NumForcedGC %d -> %d)", before, atEntry) + } + + // Linger in the Reap band for ~30 poll intervals. Pre-fix (celeris#407) + // forces a GC every tick (~30 more); post-fix fires 0 more. + time.Sleep(30 * poll) + if ctrl.Stage() != StageReap { + t.Fatalf("expected to stay in Reap, got %s", ctrl.Stage()) + } + runtime.ReadMemStats(&ms) + if lingering := ms.NumForcedGC - atEntry; lingering > 1 { + t.Fatalf("Reap GC fired %d extra times while lingering in the band; want <=1 (celeris#407: GC only on entry)", lingering) + } +} From 32e51d825159e1eb932074fd25a101591c699150 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Sat, 11 Jul 2026 11:13:33 +0200 Subject: [PATCH 3/3] =?UTF-8?q?perf(h2):=20realize=20the=20inline-response?= =?UTF-8?q?=20outBuf=20bypass=20safely=20=E2=80=94=20celeris#408?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The direct-to-outBuf inline path (h2InlineResponseAdapter / Processor.InlineWriter) was fully wired but never read: executeHandlerInline still set ResponseWriter = connWriter, so inline GET/HEAD END_STREAM responses paid the sharded-queue mutex, a pooled buffer, and an eventfd self-wake syscall (Enqueue on the same event-loop thread that is already awake), then drained back into outBuf. Route inline execution through InlineWriter so buffered responses (WriteResponse) append straight to outBuf, skipping all of that. Safe because inline runs ON the event-loop thread under H2State.mu — the same invariant control frames rely on. The naive swap would have broken SSE/chunked-over-H2: Context.StreamWriter() type-asserts ResponseWriter.(stream.Streamer) and 500s on nil, and the inline adapter was not a Streamer. Fix: h2InlineResponseAdapter now implements Streamer, delegating WriteHeader/Write/Flush/Close to the queue adapter (streaming may run on a detached goroutine, which must NOT write outBuf directly). So buffered -> direct outBuf (the win), streaming -> queue (safe). nil-guard falls back to connWriter for Processors built without the conn layer (unit tests). Adds an end-to-end regression test asserting the inline writer is both the direct-outBuf adapter AND a stream.Streamer. --- internal/conn/h2.go | 3 ++ internal/conn/h2_inline_test.go | 82 +++++++++++++++++++++++++++++++++ internal/conn/response.go | 32 +++++++++++++ protocol/h2/stream/processor.go | 18 ++++++-- 4 files changed, 131 insertions(+), 4 deletions(-) create mode 100644 internal/conn/h2_inline_test.go diff --git a/internal/conn/h2.go b/internal/conn/h2.go index 3b31a04f..699a3daa 100644 --- a/internal/conn/h2.go +++ b/internal/conn/h2.go @@ -526,6 +526,9 @@ func NewH2State(handler stream.Handler, cfg H2Config, write func([]byte), wakeup // peer's SETTINGS lands. s.adapter.manager = mgr s.inlineAdapter.manager = mgr + // The inline adapter delegates incremental streaming to the queue adapter + // so StreamWriter()-based responses stay safe on inline routes (celeris#408). + s.inlineAdapter.queue = &s.adapter p := frame.NewParser() p.InitReader(&s.inBuf) diff --git a/internal/conn/h2_inline_test.go b/internal/conn/h2_inline_test.go new file mode 100644 index 00000000..d1720a86 --- /dev/null +++ b/internal/conn/h2_inline_test.go @@ -0,0 +1,82 @@ +package conn + +import ( + "bytes" + "context" + "sync" + "testing" + + "golang.org/x/net/http2" + + "github.com/goceleris/celeris/protocol/h2/frame" + "github.com/goceleris/celeris/protocol/h2/stream" +) + +// inlineStreamerHandler records the writer it was handed: whether it is the +// direct-to-outBuf inline adapter (the #408 optimization) and whether it is a +// stream.Streamer (SSE-safety), then writes a buffered response. +type inlineStreamerHandler struct { + isInline *bool + streamerOK *bool +} + +func (h inlineStreamerHandler) HandleStream(_ context.Context, s *stream.Stream) error { + if s.ResponseWriter == nil { + return nil + } + _, *h.isInline = s.ResponseWriter.(*h2InlineResponseAdapter) + _, *h.streamerOK = s.ResponseWriter.(stream.Streamer) + return s.ResponseWriter.WriteResponse(s, 200, [][2]string{{"content-type", "text/plain"}}, []byte("ok")) +} + +// TestH2InlineResponse_DirectOutBuf_StaysStreamer is the regression guard for +// celeris#408. Realizing the inline-outBuf bypass must (a) actually take the +// direct-to-outBuf path (skip the sharded queue + eventfd self-wake), and +// (b) keep the inline ResponseWriter a stream.Streamer — otherwise +// Context.StreamWriter() returns nil and SSE/chunked-over-H2 hard-500s. +// +// old connWriter path: streamer=true, queue.pending=TRUE → fails (b)-check (not optimized) +// naive InlineWriter: streamer=FALSE, queue.pending=false → fails (a)-check (SSE broken) +// this fix: streamer=true, queue.pending=false → passes both +func TestH2InlineResponse_DirectOutBuf_StaysStreamer(t *testing.T) { + var isInline, streamerOK bool + h := inlineStreamerHandler{isInline: &isInline, streamerOK: &streamerOK} + var mu sync.Mutex + var writes []byte + write := func(b []byte) { mu.Lock(); writes = append(writes, b...); mu.Unlock() } + + state := NewH2State(h, H2Config{}, write, -1) // wakeupFD=-1: no eventfd + + // Client preface + empty SETTINGS + a single inline-eligible GET (END_STREAM). + hdr := encodeH2Headers(t, [][2]string{ + {":method", "GET"}, {":scheme", "http"}, {":path", "/"}, {":authority", "x"}, + }) + var in bytes.Buffer + in.WriteString(frame.ClientPreface) + fr := http2.NewFramer(&in, nil) + if err := fr.WriteSettings(); err != nil { + t.Fatalf("WriteSettings: %v", err) + } + if err := fr.WriteRawFrame(http2.FrameHeaders, + http2.FlagHeadersEndStream|http2.FlagHeadersEndHeaders, 1, hdr); err != nil { + t.Fatalf("WriteHeaders: %v", err) + } + if err := ProcessH2(context.Background(), in.Bytes(), state, h, write, H2Config{}); err != nil { + t.Fatalf("ProcessH2: %v", err) + } + + // (a) the optimization: the inline handler must be handed the direct-to-outBuf + // InlineWriter, not the queue-backed connWriter (else the Enqueue + eventfd + // self-wake the inline path was meant to skip still fire). + if !isInline { + t.Fatal("inline handler's ResponseWriter is not the direct-outBuf InlineWriter (celeris#408 optimization not applied)") + } + // (b) SSE-safety: the inline writer MUST be a stream.Streamer, else + // Context.StreamWriter() returns nil and SSE/chunked-over-H2 hard-500s. + if !streamerOK { + t.Fatal("inline handler's ResponseWriter is not a stream.Streamer — Context.StreamWriter()/SSE over H2 would 500 (celeris#408)") + } + if len(writes) == 0 { + t.Fatal("no bytes written for the inline GET response") + } +} diff --git a/internal/conn/response.go b/internal/conn/response.go index 28f27a03..664704b1 100644 --- a/internal/conn/response.go +++ b/internal/conn/response.go @@ -518,6 +518,13 @@ type h2InlineResponseAdapter struct { maxFrameSize uint32 // peer's advertised MAX_FRAME_SIZE (updated via Refresh) manager *stream.Manager // peer-SETTINGS-aware source of max frame size enc h2StreamEncoder // per-connection encoder (no sync.Pool needed for inline path) + // queue is this connection's sharded-queue adapter. Incremental streaming + // (the Streamer methods: WriteHeader/Write/Flush/Close) MUST delegate to it + // — a streaming response (SSE, chunked) may be produced by a detached/async + // goroutine that cannot write outBuf directly (outBuf is event-loop-thread + + // H2State.mu only). Only WriteResponse — the buffered inline GET/HEAD fast + // path, always on the event loop — writes straight to outBuf. See celeris#408. + queue *h2ResponseAdapter } // peerMaxFrame returns the peer's currently-advertised SETTINGS_MAX_FRAME_SIZE. @@ -629,6 +636,31 @@ func (a *h2InlineResponseAdapter) WriteRSTStreamPriority(_ uint32, _ http2.ErrCo func (a *h2InlineResponseAdapter) CloseConn() error { return nil } +// h2InlineResponseAdapter implements stream.Streamer by delegating incremental +// streaming to the sharded write queue (a.queue). This is what keeps +// StreamWriter()-based responses (SSE, chunked) working when an inline-eligible +// route streams: Context.StreamWriter() type-asserts ResponseWriter.(Streamer), +// so the inline writer MUST be a Streamer or streaming 500s. The queue path is +// safe from a detached/async streaming goroutine; direct-outBuf (WriteResponse) +// is not. See celeris#408. +func (a *h2InlineResponseAdapter) WriteHeader(s *stream.Stream, status int, headers [][2]string) error { + return a.queue.WriteHeader(s, status, headers) +} + +func (a *h2InlineResponseAdapter) Write(s *stream.Stream, data []byte) error { + return a.queue.Write(s, data) +} + +func (a *h2InlineResponseAdapter) Flush(s *stream.Stream) error { + return a.queue.Flush(s) +} + +func (a *h2InlineResponseAdapter) Close(s *stream.Stream) error { + return a.queue.Close(s) +} + +var _ stream.Streamer = (*h2InlineResponseAdapter)(nil) + type h2ResponseAdapter struct { write func([]byte) outBuf *bytes.Buffer diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go index 8eec8d2e..182d25b9 100644 --- a/protocol/h2/stream/processor.go +++ b/protocol/h2/stream/processor.go @@ -557,12 +557,22 @@ func (p *Processor) runHandler(stream *Stream) { } // executeHandlerInline runs the handler synchronously on the event loop. -// Response writes go directly to outBuf (via connWriter) instead of the -// async write queue, eliminating goroutine dispatch, shard mutex ops, -// eventfd syscall, and drain loop overhead. +// Because it runs ON the event-loop thread (under H2State.mu), a buffered +// response (WriteResponse — the common inline GET/HEAD END_STREAM path) goes +// straight to outBuf via InlineWriter, skipping the sharded write queue's mutex +// + pooled buffer and — the real win — the eventfd self-wake syscall Enqueue +// fires on every response. Incremental streaming (StreamWriter) still routes +// through the queue: InlineWriter is a stream.Streamer that delegates its +// streaming methods to connWriter, so SSE/chunked stays safe (celeris#408). +// Falls back to connWriter when InlineWriter is unset (Processors built without +// the conn layer, e.g. unit tests). func (p *Processor) executeHandlerInline(stream *Stream) { p.InlineCount++ - stream.ResponseWriter = p.connWriter + if p.InlineWriter != nil { + stream.ResponseWriter = p.InlineWriter + } else { + stream.ResponseWriter = p.connWriter + } if p.InlineCachedCtx != nil { stream.CachedCtx = p.InlineCachedCtx