diff --git a/internal/conn/h2.go b/internal/conn/h2.go index b32144ee..699a3daa 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() @@ -519,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/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) + } +} 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/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) + } +} 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