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] }