From c7e1da3f82ff90f26d85c4f64bab3a179411be98 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Fri, 11 Sep 2026 03:31:14 +0200 Subject: [PATCH] fix(engine): clear the hand-off queue slots so a drained conn can be collected Four reuse queues across the two loop engines truncate a slice of pointers with [:0] and reuse the backing array. Truncation does not clear the array, so every pointer in it stays reachable until some later drain happens to overwrite that slot -- each queue pins its own high-water mark worth of objects for the life of the worker. engine/iouring/worker.go detachQueue / detachQSpare []*connState engine/epoll/loop.go detachQueue / detachQSpare []*connState engine/epoll/adopt.go adoptQueue / adoptQSpare []adoptItem engine/iouring/driver.go driverActionQueue / ...Spare []driverAction The detach queues are the ones that matter. A detached connState carries its read and write buffers, its H1State, and on the WebSocket path whatever the middleware hung off that state, so one deep burst leaves megabytes reachable behind a queue that is nominally empty. drainPendingRelease in the same file already guards exactly this, with a comment saying why; these four did not. Retention is bounded by peak queue depth rather than unbounded, so this is hardening, not a fix for the sustained growth in #573. It is a sibling of #571 in shape: one guarded copy and several unguarded ones of the same idea. Each engine gets two guards. The first reads the spare slice out to its capacity and requires every slot past the new length to be nil. The second states the same thing as an outcome -- a cleanup on an enqueued conn must run once the caller drops its reference -- which is what the fix is actually for. Reverting the clear() alone fails both on both engines: drainDetachQueue left 64 of 64 connStates reachable in the reused backing array a connState the detach queue already drained is still reachable after 5 GC cycles runtime.KeepAlive in the second test is load-bearing. Without it the compiler may treat the worker as dead after its last use, the queues become collectable with it, and the test passes against the unfixed code -- which it did, before the KeepAlive went in. --- engine/epoll/adopt.go | 3 + engine/epoll/detach_queue_retention_test.go | 81 ++++++++++++++++ engine/epoll/loop.go | 5 + engine/iouring/detach_queue_retention_test.go | 96 +++++++++++++++++++ engine/iouring/driver.go | 3 + engine/iouring/worker.go | 7 ++ 6 files changed, 195 insertions(+) create mode 100644 engine/epoll/detach_queue_retention_test.go create mode 100644 engine/iouring/detach_queue_retention_test.go diff --git a/engine/epoll/adopt.go b/engine/epoll/adopt.go index 0f5db77e..9ed75b61 100644 --- a/engine/epoll/adopt.go +++ b/engine/epoll/adopt.go @@ -70,6 +70,9 @@ func (l *Loop) drainAdoptQueue(ctx context.Context, now int64) { for _, it := range l.adoptQSpare { l.attachAdoptedFD(ctx, it.fd, it.carry, now) } + // Same guard as the detach queue: adoptItem carries a Carryover with + // the peer address string, so stale slots pin it after the handoff. + clear(l.adoptQSpare) l.adoptQSpare = l.adoptQSpare[:0] } diff --git a/engine/epoll/detach_queue_retention_test.go b/engine/epoll/detach_queue_retention_test.go new file mode 100644 index 00000000..67474628 --- /dev/null +++ b/engine/epoll/detach_queue_retention_test.go @@ -0,0 +1,81 @@ +//go:build linux + +package epoll + +import ( + "runtime" + "testing" +) + +// TestDrainDetachQueueDropsSlotRefs is the epoll half of the same guard the +// io_uring worker carries. Truncating the hand-off queue with [:0] reuses the +// backing array but leaves every *connState in it reachable, so the queue +// pins its own high-water mark worth of conns — buffers, H1State and, on the +// WebSocket path, whatever the middleware hung off that state — until some +// later drain overwrites each slot. +// +// detachClosed short-circuits the drain body, so the queue can be exercised +// end to end without an epoll fd or a live connection. +func TestDrainDetachQueueDropsSlotRefs(t *testing.T) { + const depth = 64 + + l := &Loop{} + for range depth { + l.detachQueue = append(l.detachQueue, &connState{fd: -1, detachClosed: true}) + } + l.detachQPending.Store(1) + + l.drainDetachQueue() + + if got := len(l.detachQSpare); got != 0 { + t.Fatalf("spare queue not truncated: len = %d, want 0", got) + } + full := l.detachQSpare[:cap(l.detachQSpare)] + if len(full) < depth { + t.Fatalf("backing array shrank to %d, expected at least the %d enqueued slots", len(full), depth) + } + held := 0 + for _, cs := range full[:depth] { + if cs != nil { + held++ + } + } + if held != 0 { + t.Fatalf("drainDetachQueue left %d of %d connStates reachable in the reused backing array", held, depth) + } +} + +// TestDrainDetachQueueLetsDrainedConnsBeCollected states the same guarantee +// as an observable outcome: a conn the queue has already drained must become +// unreachable once the caller drops its own reference. +func TestDrainDetachQueueLetsDrainedConnsBeCollected(t *testing.T) { + l := &Loop{} + collected := make(chan struct{}) + + func() { + cs := &connState{fd: -1, detachClosed: true} + runtime.AddCleanup(cs, func(ch chan struct{}) { close(ch) }, collected) + l.detachQueue = append(l.detachQueue, cs) + for range 15 { + l.detachQueue = append(l.detachQueue, &connState{fd: -1, detachClosed: true}) + } + }() + l.detachQPending.Store(1) + l.drainDetachQueue() + + for range 5 { + runtime.GC() + select { + case <-collected: + runtime.KeepAlive(l) + return + default: + } + } + // KeepAlive is load-bearing: without it the compiler may treat l as dead + // after its last use, the Loop and its queues become collectable, and the + // cleanup runs regardless of what the queue did — passing against the + // unfixed code. + runtime.KeepAlive(l) + t.Fatal("a connState the detach queue already drained is still reachable after 5 GC cycles") +} diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index 3708decf..79f61995 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -2180,6 +2180,11 @@ func (l *Loop) drainDetachQueue() { } l.markDirty(cs) } + // Drop the strong refs before reusing the array (see the io_uring + // worker's drainDetachQueue and drainPendingRelease for the same + // guard): [:0] alone leaves every *connState reachable in the + // backing array until some later drain overwrites its slot. + clear(l.detachQSpare) l.detachQSpare = l.detachQSpare[:0] } diff --git a/engine/iouring/detach_queue_retention_test.go b/engine/iouring/detach_queue_retention_test.go new file mode 100644 index 00000000..b32e5e68 --- /dev/null +++ b/engine/iouring/detach_queue_retention_test.go @@ -0,0 +1,96 @@ +//go:build linux + +package iouring + +import ( + "runtime" + "testing" +) + +// TestDrainDetachQueueDropsSlotRefs guards the hand-off queue against the +// retention hazard drainPendingRelease already guards against: truncating a +// []*connState with [:0] reuses the backing array but leaves every pointer in +// it reachable, so the queue pins its own high-water mark worth of +// connStates until some later drain happens to overwrite each slot. +// +// That matters here more than it would for a plain buffer. A detached +// connState carries its read and write buffers, its H1State, and — on the +// WebSocket path — whatever the middleware hung off that state, so a queue +// that once peaked deep holds all of it for as long as the worker lives. +// +// The test enqueues a deep batch of already-closed conns (detachClosed short- +// circuits the drain body, so no ring is needed), drains, and then reads the +// spare slice out to its capacity. Every slot past the new length must be nil. +func TestDrainDetachQueueDropsSlotRefs(t *testing.T) { + const depth = 64 + + w := &Worker{} + for range depth { + w.detachQueue = append(w.detachQueue, &connState{fd: -1, detachClosed: true}) + } + w.detachQPending.Store(1) + + w.drainDetachQueue() + + if got := len(w.detachQSpare); got != 0 { + t.Fatalf("spare queue not truncated: len = %d, want 0", got) + } + full := w.detachQSpare[:cap(w.detachQSpare)] + if len(full) < depth { + t.Fatalf("backing array shrank to %d, expected it to keep at least the %d enqueued slots", len(full), depth) + } + held := 0 + for _, cs := range full[:depth] { + if cs != nil { + held++ + } + } + if held != 0 { + t.Fatalf("drainDetachQueue left %d of %d connStates reachable in the reused backing array; "+ + "the queue pins its high-water mark (celeris#571 sibling)", held, depth) + } +} + +// TestDrainDetachQueueLetsDrainedConnsBeCollected is the same guarantee stated +// as an observable outcome rather than as a property of the slice: once the +// queue has drained a conn and the caller drops its own reference, the +// connState must become unreachable. A finalizer on one of the enqueued +// conns proves it, where reading the backing array only proves the slot was +// cleared. +func TestDrainDetachQueueLetsDrainedConnsBeCollected(t *testing.T) { + w := &Worker{} + collected := make(chan struct{}) + + func() { + // Enqueued inside a function literal so the local goes out of scope + // before the GC below; a live stack slot would keep it reachable + // regardless of what the queue does. + cs := &connState{fd: -1, detachClosed: true} + runtime.AddCleanup(cs, func(ch chan struct{}) { close(ch) }, collected) + w.detachQueue = append(w.detachQueue, cs) + // Pad the queue so the drained conn is not the only slot; a + // single-element array is the easiest case to get right by accident. + for range 15 { + w.detachQueue = append(w.detachQueue, &connState{fd: -1, detachClosed: true}) + } + }() + w.detachQPending.Store(1) + w.drainDetachQueue() + + for range 5 { + runtime.GC() + select { + case <-collected: + runtime.KeepAlive(w) + return + default: + } + } + // KeepAlive is load-bearing, not defensive. Without it the compiler is + // free to treat w as dead after its last use above, the whole Worker + // (queues included) becomes collectable, and the cleanup runs no matter + // what the queue did -- the test then passes against the unfixed code. + runtime.KeepAlive(w) + t.Fatal("a connState the detach queue already drained is still reachable after 5 GC cycles: " + + "the queue is holding it in its reused backing array") +} diff --git a/engine/iouring/driver.go b/engine/iouring/driver.go index 8572e21d..f705b7fe 100644 --- a/engine/iouring/driver.go +++ b/engine/iouring/driver.go @@ -232,6 +232,9 @@ func (w *Worker) drainDriverActions() { w.attachAdoptedFD(a.adoptFD, a.adoptCarry) } } + // Same guard as the detach queue: a driverAction holds pointers, so + // stale slots keep them reachable after the action has been applied. + clear(w.driverActionSpare) w.driverActionSpare = w.driverActionSpare[:0] } diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index 6a6514b1..d4080409 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -3808,6 +3808,13 @@ func (w *Worker) drainDetachQueue() { } w.markDirty(cs) } + // Drop the strong refs before reusing the array. Truncating to [:0] + // leaves every *connState in the backing array reachable until some + // later drain overwrites that slot, so the queue pins its own + // high-water mark worth of connStates -- each one holding its buffers + // and, for a detached conn, its H1State and whatever the middleware + // hung off it. drainPendingRelease already guards the same hazard. + clear(w.detachQSpare) w.detachQSpare = w.detachQSpare[:0] }