From 51ef82c12bfbe027754f6a40b5a539709fcfc55d Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Fri, 3 Jul 2026 10:45:43 +0200 Subject: [PATCH 1/5] perf(engine): coalesce detach-queue wakeups on WS broadcast fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The io_uring and epoll engines serialise off-event-loop writes (WebSocket frames, SSE events) through a per-loop detachQueue: an off-loop goroutine appends the conn and writes a wakeup eventfd so the event loop drains the queue and flushes each conn's writeBuf. Under a Hub broadcast fan-out the write closure runs once per (message x connection) from the GOMAXPROCS*4 dispatch goroutines, and every call issued its own eventfd wakeup write. At 1024 connections that is a storm of redundant wakeup syscalls — the loop needs only one wakeup per drain cycle, because a single drain already flushes every queued conn. Coalesce with an edge-triggered wakeup: only the enqueue that takes the detach queue empty->non-empty writes the eventfd, gated by `wasEmpty := detachQPending.Swap(1) == 0` under detachQMu. drainDetachQueue clears the flag (Store(0)) under the same lock *before* it swaps the queue out, so a racing enqueue is either captured by the drain's swap or observes pending==0 and re-arms the eventfd — never both missed. No wakeup is dropped. The detachMu-guarded writeBuf mutation is untouched, so the WS-write-vs-flushWrites ordering invariant (celeris#284) is preserved. Applied uniformly to every off-loop enqueue site in both engines (WS write, recv pause/resume backpressure, async-detach setup); loop-thread enqueues are unchanged since they run on the drain thread and never race the flag. Benchmarks (ws-hub-broadcast, 1024 conns, amd64, median of 3 x 12s): io_uring 773,823 -> 918,111 rps (+18.6%) epoll 718,965 -> 838,918 rps (+16.7%) go test -race passes on both engine/iouring and engine/epoll. --- engine/epoll/loop.go | 31 +++++++++++++++++++++++-------- engine/iouring/worker.go | 32 ++++++++++++++++++++++++-------- 2 files changed, 47 insertions(+), 16 deletions(-) diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index f94da89d..e573f0a2 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -1404,9 +1404,18 @@ func (l *Loop) initProtocol(cs *connState) { // from this goroutine — dirtyHead is event-loop-local. l.detachQMu.Lock() l.detachQueue = append(l.detachQueue, cs) - l.detachQPending.Store(1) + // Edge-triggered wakeup: only the enqueue that takes the detach + // queue empty->non-empty writes the wakeup eventfd. This Swap and + // the drain's detachQPending.Store(0) both run under detachQMu, so + // a racing enqueue is either captured by the drain's swap or + // observes pending==0 and re-arms — never both missed (see + // drainDetachQueue). Coalesces the per-message wakeup-syscall storm + // under a hot broadcast fan-out. The detachMu-guarded writeBuf + // mutation is untouched, so the WS-write-vs-flushWrites ordering + // invariant (celeris#284) is preserved. + wasEmpty := l.detachQPending.Swap(1) == 0 l.detachQMu.Unlock() - if l.eventFD >= 0 { + if wasEmpty && l.eventFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(l.eventFD, val[:]) @@ -1427,9 +1436,11 @@ func (l *Loop) initProtocol(cs *connState) { } l.detachQMu.Lock() l.detachQueue = append(l.detachQueue, cs) - l.detachQPending.Store(1) + // Coalesce the wakeup on the detach queue's empty->non-empty + // edge — see the write closure above and drainDetachQueue. + wasEmpty := l.detachQPending.Swap(1) == 0 l.detachQMu.Unlock() - if l.eventFD >= 0 { + if wasEmpty && l.eventFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(l.eventFD, val[:]) @@ -1441,9 +1452,11 @@ func (l *Loop) initProtocol(cs *connState) { } l.detachQMu.Lock() l.detachQueue = append(l.detachQueue, cs) - l.detachQPending.Store(1) + // Coalesce the wakeup on the detach queue's empty->non-empty + // edge — see the write closure above and drainDetachQueue. + wasEmpty := l.detachQPending.Swap(1) == 0 l.detachQMu.Unlock() - if l.eventFD >= 0 { + if wasEmpty && l.eventFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(l.eventFD, val[:]) @@ -1506,9 +1519,11 @@ func (l *Loop) initProtocol(cs *connState) { if l.async { l.detachQMu.Lock() l.detachQueue = append(l.detachQueue, cs) - l.detachQPending.Store(1) + // Coalesce the wakeup on the detach queue's empty->non-empty + // edge — see the write closure above and drainDetachQueue. + wasEmpty := l.detachQPending.Swap(1) == 0 l.detachQMu.Unlock() - if l.eventFD >= 0 { + if wasEmpty && l.eventFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(l.eventFD, val[:]) diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index 00063449..2ea6fe25 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -1383,9 +1383,19 @@ func (w *Worker) initProtocol(cs *connState) { // from this goroutine — dirtyHead is worker-local. w.detachQMu.Lock() w.detachQueue = append(w.detachQueue, cs) - w.detachQPending.Store(1) + // Edge-triggered wakeup: only the enqueue that takes the detach + // queue empty->non-empty writes the wakeup eventfd. This Swap and + // the drain's detachQPending.Store(0) both run under detachQMu, so + // a racing enqueue is either captured by the drain's swap or + // observes pending==0 and re-arms the wakeup — never both missed + // (see drainDetachQueue). Coalesces the per-message wakeup-syscall + // storm under a hot broadcast fan-out — previously one unix.Write + // per message per connection. The detachMu-guarded writeBuf + // mutation is untouched, so the WS-write-vs-flushSend ordering + // invariant (celeris#284) is fully preserved. + wasEmpty := w.detachQPending.Swap(1) == 0 w.detachQMu.Unlock() - if wakeupFD >= 0 { + if wasEmpty && wakeupFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(wakeupFD, val[:]) @@ -1407,9 +1417,11 @@ func (w *Worker) initProtocol(cs *connState) { } w.detachQMu.Lock() w.detachQueue = append(w.detachQueue, cs) - w.detachQPending.Store(1) + // Coalesce the wakeup on the detach queue's empty->non-empty + // edge — see the write closure above and drainDetachQueue. + wasEmpty := w.detachQPending.Swap(1) == 0 w.detachQMu.Unlock() - if wakeupFD >= 0 { + if wasEmpty && wakeupFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(wakeupFD, val[:]) @@ -1421,9 +1433,11 @@ func (w *Worker) initProtocol(cs *connState) { } w.detachQMu.Lock() w.detachQueue = append(w.detachQueue, cs) - w.detachQPending.Store(1) + // Coalesce the wakeup on the detach queue's empty->non-empty + // edge — see the write closure above and drainDetachQueue. + wasEmpty := w.detachQPending.Swap(1) == 0 w.detachQMu.Unlock() - if wakeupFD >= 0 { + if wasEmpty && wakeupFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(wakeupFD, val[:]) @@ -1480,9 +1494,11 @@ func (w *Worker) initProtocol(cs *connState) { if w.async { w.detachQMu.Lock() w.detachQueue = append(w.detachQueue, cs) - w.detachQPending.Store(1) + // Coalesce the wakeup on the detach queue's empty->non-empty + // edge — see the write closure above and drainDetachQueue. + wasEmpty := w.detachQPending.Swap(1) == 0 w.detachQMu.Unlock() - if wakeupFD >= 0 { + if wasEmpty && wakeupFD >= 0 { var val [8]byte val[0] = 1 _, _ = unix.Write(wakeupFD, val[:]) From 68c87ba74bbaa0c606b82d86c1251f005f9a232f Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Fri, 3 Jul 2026 12:31:19 +0200 Subject: [PATCH 2/5] perf(epoll): inline WS/SSE egress on the dispatch goroutine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detached WS/SSE writes funneled through the single event-loop thread (append to writeBuf under detachMu, enqueue to the per-loop detachQueue, wake the loop, which did the write). On a broadcast fan-out that serializes 1024 sends across N loop threads (N/thread), while the std engine does them inline on GOMAXPROCS goroutines across all cores — the ~2x broadcast gap. Issue the send inline on the dispatch goroutine when the conn is clean: flushWrites(cs,false) runs inside the same detachMu critical section that already guards orig() and that the loop-thread dirty-flush + closeConn take, so it can neither race the loop's flush nor touch a closed fd; writeBuf is one ordered buffer flushed from writePos, so no reorder. On full drain we skip the loop handoff entirely; on partial/EAGAIN/error we fall through to the existing enqueue path (surfacing OnError on I/O failure first). ws-hub-broadcast @1024 conns (epoll, amd64): +43-48% (638K -> ~930-960K), loop-thread CPU 50%% -> 0.3%% (write moves to the dispatch goroutines). -race clean on engine/epoll + middleware/websocket. --- engine/epoll/loop.go | 38 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index e573f0a2..b3fadf26 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -1399,9 +1399,43 @@ func (l *Loop) initProtocol(cs *connState) { return } orig(data) + // Inline egress fast path (WS/SSE): issue the send on THIS + // dispatch goroutine instead of funnelling every detached-conn + // write through the single event-loop thread. detachMu (held) + // is the SAME lock the loop-thread dirty-flush takes around + // flushWrites, and closeConn takes it before tearing the fd + // down (loop.go:2371) — so the write here can neither race the + // loop's flush nor touch a closed fd. writeBuf is one ordered + // buffer flushed from writePos, so dispatch-side and loop-side + // flushWrites can never reorder a conn's bytes. Reconcile + // pendingBytes exactly as the dirty-flush does (loop.go:568/587). + // This parallelises the write(2) across all cores like the std + // engine, lifting the single-loop-thread broadcast ceiling. On + // full drain we return WITHOUT enqueuing — the loop never touches + // the conn. On partial/EAGAIN/error we fall through to the + // existing enqueue path so the loop finishes the remainder + // (detached WS conns stay on the dirty list) or tears it down. + if err := l.flushWrites(cs, false); err != nil { + // Surface the specific I/O error (EPIPE/ECONNRESET/…) to the + // detached middleware before teardown, matching the loop-thread + // dirty-flush (loop.go:558) and handleWritable. Without this the + // handler would see a generic io.EOF/ErrWriteClosed from the + // asyncClosed→OnDetachClose path instead of the real errno. + if cs.h1State != nil && cs.h1State.OnError != nil { + cs.h1State.OnError(err) + } + cs.asyncClosed.Store(true) // teardown via drainDetachQueue + } else if !csWritePending(cs) { + cs.pendingBytes = 0 + mu.Unlock() + return + } else { + cs.pendingBytes = csPendingBytes(cs) + } mu.Unlock() - // Signal the event loop to flush. Do NOT call markDirty - // from this goroutine — dirtyHead is event-loop-local. + // Signal the event loop to flush the remainder / tear down. Do + // NOT call markDirty from this goroutine — dirtyHead is + // event-loop-local. l.detachQMu.Lock() l.detachQueue = append(l.detachQueue, cs) // Edge-triggered wakeup: only the enqueue that takes the detach From 0af8831f914ddae047e21d50fcaa70bdf27a30d2 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Fri, 3 Jul 2026 12:59:47 +0200 Subject: [PATCH 3/5] perf(iouring): inline WS/SSE egress on the dispatch goroutine (SINGLE_ISSUER-safe) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the epoll inline-egress ceiling-breaker to io_uring. SINGLE_ISSUER forbids the dispatch goroutine from submitting a ring SEND, but a raw unix.Write(2) on the socket fd is legal iff no ring SEND is in-flight for the conn — else the two writes interleave on the wire. detachMu gates it: every ring SEND is submitted under detachMu with cs.sending set, and completeSend now CLEARS cs.sending under detachMu too (was cleared before the lock — a cross-thread read race the inline path would hit). Gated on !cs.fixedFile: under ACCEPT_DIRECT cs.fd is a ring file-table index, not a syscall'able fd, so those conns keep using the ring (same guard hijack uses). Also gates on !zcNotifPending + empty sendBuf/bodyBuf. ws-hub-broadcast @1024 conns (io_uring, amd64, fixed_files=false): +43.2% (663K -> ~950K). -race clean on engine/iouring (102s) + middleware/websocket. --- engine/iouring/worker.go | 84 ++++++++++++++++++++++++++++++++++------ 1 file changed, 73 insertions(+), 11 deletions(-) diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index 2ea6fe25..b01676ce 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -1378,9 +1378,41 @@ func (w *Worker) initProtocol(cs *connState) { return } orig(data) + // Inline egress fast path (io_uring, SINGLE_ISSUER-safe): the + // ring may only be driven by the worker thread, but a raw + // unix.Write(2) on the socket fd is legal from any goroutine iff + // NO ring SEND is in-flight for this conn (else the two writes + // interleave on the wire). detachMu (held) gates it: the worker + // submits every SEND under detachMu and sets cs.sending, and + // completeSend now clears it under detachMu too, so with + // cs.sending==false && !zcNotifPending && sendBuf/bodyBuf empty no + // SEND is outstanding and none can start while we hold the lock — + // writeBuf is the only pending data, so the raw write is exclusive + // and correctly ordered. Requires a REAL fd: under ACCEPT_DIRECT + // (fixedFile) cs.fd is a ring file-table index, not a syscall'able + // fd, so those conns skip the fast path and use the ring (same + // guard hijack uses). This parallelises WS/SSE egress across the + // dispatch goroutines like the std engine instead of funnelling + // every send through the single worker thread. On full drain we + // skip the worker handoff; on partial we compact the remainder to + // the front (the worker's flushSend swaps writeBuf→sendBuf next); + // on EAGAIN/error we leave writeBuf and fall through — the worker + // ring-sends the rest and surfaces any I/O error via completeSend. + if !cs.fixedFile && !cs.sending && !cs.zcNotifPending && + len(cs.sendBuf) == 0 && len(cs.bodyBuf) == 0 && len(cs.writeBuf) > 0 { + if n, werr := unix.Write(cs.fd, cs.writeBuf); werr == nil { + w.bytesWritten.Add(uint64(n)) + if n >= len(cs.writeBuf) { + cs.writeBuf = cs.writeBuf[:0] + mu.Unlock() + return + } + cs.writeBuf = cs.writeBuf[:copy(cs.writeBuf, cs.writeBuf[n:])] + } + } mu.Unlock() - // Signal the event loop to flush. Do NOT call markDirty - // from this goroutine — dirtyHead is worker-local. + // Signal the event loop to flush the remainder. Do NOT call + // markDirty from this goroutine — dirtyHead is worker-local. w.detachQMu.Lock() w.detachQueue = append(w.detachQueue, cs) // Edge-triggered wakeup: only the enqueue that takes the detach @@ -2109,7 +2141,16 @@ func (w *Worker) handleSend(c *completionEntry, fd int, now int64) { // SEND_ZC notification CQE: the NIC has finished DMA-reading the buffer. // Now safe to modify/reuse sendBuf. Process the deferred result. if cqeIsNotif(c.Flags) { - cs.zcNotifPending = false + // zcNotifPending is read by the inline-egress guard on the dispatch + // goroutine under detachMu; clear it under the lock (completeSend + // re-acquires detachMu, so release first). + if mu := cs.detachMu; mu != nil { + mu.Lock() + cs.zcNotifPending = false + mu.Unlock() + } else { + cs.zcNotifPending = false + } w.completeSend(cs, fd, int(cs.zcSentBytes), now) return } @@ -2117,6 +2158,12 @@ func (w *Worker) handleSend(c *completionEntry, fd int, now int64) { // SEND_ZC first CQE: result is ready but buffer is still in DMA. // Store the result and wait for the notification before touching sendBuf. if w.sendZC && cqeHasMore(c.Flags) { + // cs.sending / cs.zcNotifPending are read by the inline-egress guard on + // the dispatch goroutine under detachMu; mutate them under the lock. + if mu := cs.detachMu; mu != nil { + mu.Lock() + defer mu.Unlock() + } if c.Res < 0 { cs.sending = false cs.zcNotifPending = true @@ -2137,24 +2184,37 @@ func (w *Worker) handleSend(c *completionEntry, fd int, now int64) { // SEND_ZC EINVAL fallback: kernel does not support the opcode. // Disable ZC for this worker and retry the send with regular SEND. if c.Res == -22 && w.sendZC { - cs.sending = false w.sendZC = false w.logger.Warn("SEND_ZC not supported (EINVAL), falling back to regular SEND", "worker", w.id) + // cs.sending is read by the inline-egress guard under detachMu; clear it + // and re-flush under the lock (flushSend for a detached conn is always + // called under detachMu, as in the dirty-flush loop). + mu := cs.detachMu + if mu != nil { + mu.Lock() + } + cs.sending = false if w.flushSend(cs) { w.markDirty(cs) } + if mu != nil { + mu.Unlock() + } return } if c.Res < 0 { - cs.sending = false w.errCount.Add(1) - cs.sendBuf = cs.sendBuf[:0] + // cs.sending / cs.sendBuf are read by the inline-egress guard under + // detachMu; reset them (and writeBuf) inside the lock rather than before + // it, so the dispatch-goroutine read never races this error completion. mu := cs.detachMu if mu != nil { mu.Lock() } + cs.sending = false + cs.sendBuf = cs.sendBuf[:0] cs.writeBuf = cs.writeBuf[:0] if cs.h1State != nil && cs.h1State.OnError != nil { cs.h1State.OnError(errIORingSend(c.Res)) @@ -2182,15 +2242,17 @@ func (w *Worker) handleSend(c *completionEntry, fd int, now int64) { // exists, otherwise the goroutine read races the event-loop write — // observed via -race in TestNativeEngineLargePayload/io_uring. func (w *Worker) completeSend(cs *connState, fd int, sent int, now int64) { - cs.sending = false - - // Take the lock up-front for detached connections so the entire - // state mutation (sendBuf truncate / writeBuf reset / OnError fire) - // is serialized against the goroutine writeFn path. + // Take the lock up-front for detached connections so the entire state + // mutation (cs.sending clear / sendBuf truncate / writeBuf reset / OnError + // fire) is serialized against the goroutine writeFn path. The inline-egress + // fast path (the initProtocol guarded closure) reads cs.sending under + // detachMu to decide whether a ring SEND is in-flight, so the clear MUST be + // inside the lock — otherwise that read races this completion. if mu := cs.detachMu; mu != nil { mu.Lock() defer mu.Unlock() } + cs.sending = false if sent < 0 { w.errCount.Add(1) From b487fc566d218f4aa0e9c2829c45ae26cc2091f3 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Fri, 3 Jul 2026 13:19:51 +0200 Subject: [PATCH 4/5] test(websocket): concurrent large-frame broadcast regression for inline egress Broadcasts 8KB frames (>= sendZCMinBytes, so io_uring uses SEND_ZC) to 32 detached conns while dispatch goroutines issue inline writes, asserting every received frame is byte-intact and per-conn in order. Run under -race in CI this covers the SEND_ZC-completion-vs-inline-write interaction the 64B ws-hub benchmark never reaches. Passes -race x3 on both epoll + io_uring. --- ...broadcast_egress_correctness_linux_test.go | 152 ++++++++++++++++++ 1 file changed, 152 insertions(+) create mode 100644 middleware/websocket/broadcast_egress_correctness_linux_test.go diff --git a/middleware/websocket/broadcast_egress_correctness_linux_test.go b/middleware/websocket/broadcast_egress_correctness_linux_test.go new file mode 100644 index 00000000..a3b9a26d --- /dev/null +++ b/middleware/websocket/broadcast_egress_correctness_linux_test.go @@ -0,0 +1,152 @@ +//go:build linux + +package websocket + +// Regression test for the inline-egress fast path (engine/{epoll,iouring} +// detached guarded writeFn) and its ring/worker fallback. Broadcasts large +// (>= sendZCMinBytes = 4096B, so io_uring uses SEND_ZC) frames to many detached +// conns from a publisher while the dispatch goroutines issue inline writes, +// and asserts every RECEIVED frame is byte-intact and per-conn in order — a +// wire interleave between a raw unix.Write and a ring SEND (or a reorder vs the +// loop-thread flush) would corrupt the payload or the sequence. Run under +// -race in CI this covers the SEND_ZC-completion-vs-inline-write interaction +// that the 64-byte ws-hub-broadcast benchmark never reaches (below the 4096B +// ZC threshold). Frames may be DROPPED under backpressure (HubPolicyDrop) — +// that is allowed; corruption and reorder are not. + +import ( + "bufio" + "encoding/binary" + "io" + "sync" + "sync/atomic" + "testing" + "time" +) + +// readServerFrameNonFatal reads one unmasked server frame without calling +// t.Fatal (safe to run in a background goroutine at teardown). +func readServerFrameNonFatal(br *bufio.Reader) (fin bool, op Opcode, payload []byte, err error) { + var h [2]byte + if _, err = io.ReadFull(br, h[:]); err != nil { + return + } + fin = h[0]&0x80 != 0 + op = Opcode(h[0] & 0x0F) + length := int64(h[1] & 0x7F) + switch length { + case 126: + var ext [2]byte + if _, err = io.ReadFull(br, ext[:]); err != nil { + return + } + length = int64(binary.BigEndian.Uint16(ext[:])) + case 127: + var ext [8]byte + if _, err = io.ReadFull(br, ext[:]); err != nil { + return + } + length = int64(binary.BigEndian.Uint64(ext[:])) + } + payload = make([]byte, length) + _, err = io.ReadFull(br, payload) + return +} + +func TestNativeEngineHubBroadcastInlineEgress(t *testing.T) { + for _, kind := range engineKinds(t) { + kind := kind + t.Run(kind.String(), func(t *testing.T) { + const ( + nconns = 32 + nframes = 400 + frameSize = 8192 // >= sendZCMinBytes(4096): io_uring exercises SEND_ZC + ) + + hub := NewHub(HubConfig{ + OnSlowConn: func(*Conn, error) HubPolicy { return HubPolicyDrop }, + }) + cfg := Config{Handler: func(c *Conn) { + unreg := hub.Register(c) + defer unreg() + for { + if _, _, err := c.ReadMessage(); err != nil { + return + } + } + }} + addr, stop := startNativeServer(t, kind, cfg) + defer stop() + + var wg sync.WaitGroup + var corrupt, reorder int64 + clients := make([]*testWSClient, nconns) + for i := 0; i < nconns; i++ { + c := dialRaw(t, addr) + c.upgrade(t, "/ws") + clients[i] = c + wg.Add(1) + go func(c *testWSClient) { + defer wg.Done() + var lastSeq int64 = -1 + for { + _ = c.conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + fin, op, payload, err := readServerFrameNonFatal(c.br) + if err != nil { + return + } + // A well-formed broadcast frame: final binary frame of the + // exact size, seq in the first 8 bytes, fill = byte(seq)+byte(j). + if !fin || op != OpBinary || len(payload) != frameSize { + atomic.AddInt64(&corrupt, 1) + return + } + seq := int64(binary.BigEndian.Uint64(payload[:8])) + for j := 8; j < len(payload); j++ { + if payload[j] != byte(seq)+byte(j) { + atomic.AddInt64(&corrupt, 1) + return + } + } + if seq <= lastSeq { // per-conn FIFO: seq must strictly increase + atomic.AddInt64(&reorder, 1) + } + lastSeq = seq + } + }(c) + } + + for hub.Len() < nconns { + time.Sleep(5 * time.Millisecond) + } + + for seq := 0; seq < nframes; seq++ { + payload := make([]byte, frameSize) + binary.BigEndian.PutUint64(payload[:8], uint64(seq)) + for j := 8; j < frameSize; j++ { + payload[j] = byte(seq) + byte(j) + } + pm, err := NewPreparedMessage(OpBinary, payload) + if err != nil { + t.Fatal(err) + } + if _, err := hub.BroadcastPrepared(pm); err != nil { + t.Fatal(err) + } + } + + time.Sleep(300 * time.Millisecond) // let the tail drain + for _, c := range clients { + c.close() + } + wg.Wait() + + if n := atomic.LoadInt64(&corrupt); n != 0 { + t.Errorf("%s: %d corrupted frames — inline egress interleaved with a ring/loop send", kind, n) + } + if n := atomic.LoadInt64(&reorder); n != 0 { + t.Errorf("%s: %d out-of-order frames — per-conn FIFO violated", kind, n) + } + }) + } +} From ba3887aa2bc58776d7c8298fdafcb8bfd3fe19fc Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Fri, 3 Jul 2026 17:19:30 +0200 Subject: [PATCH 5/5] fix(websocket): close pre-existing upgrade-vs-close races + writeErr panic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three pre-existing (v1.5.6) bugs in the engine WS detach/close path, surfaced by a -race build under aggressive peer-RST-mid-upgrade: 1. WSReady upgrade-completion barrier. The WS middleware installs the detached callbacks (RawWriteFn, pause/resume, idle-deadline, OnDetachClose LAST) on the async goroutine AFTER Detach releases detachMu (celeris#273/#309), so they were not lock-serialised against closeConn reading OnDetachClose / PauseRecv while tearing the conn down on a peer RST. H1State.WSReady is Stored(true, release) as the final wiring step; closeConn Loads it (acquire) before touching those callbacks, so it sees a fully-wired conn or skips WS teardown entirely (conn still closed via fd/read path). Gated in both engines. 2. writeErr atomic.Value inconsistent-type panic (== the validator's I-LIVENESS code=2 crash). OnError stores errors of varying concrete types (errPeerClosed, syscall errors, …); atomic.Value.Store panics on the second differing type. Boxed via storedWriteErr so the stored dynamic type is constant; also skip storing a nil error. -race clean on ./middleware/websocket, ./engine/{epoll,iouring}, ./internal/conn. --- engine/epoll/loop.go | 22 +++++++++++++++++----- engine/iouring/worker.go | 22 +++++++++++++++++----- internal/conn/h1.go | 20 ++++++++++++++++++++ middleware/websocket/conn.go | 12 ++++++++++-- middleware/websocket/websocket.go | 4 +++- middleware/websocket/websocket_test.go | 5 +++-- 6 files changed, 70 insertions(+), 15 deletions(-) diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go index b3fadf26..f154d0e9 100644 --- a/engine/epoll/loop.go +++ b/engine/epoll/loop.go @@ -2404,14 +2404,23 @@ func (l *Loop) closeConn(fd int) { // goroutine is mid-write, we block until it finishes. cs.detachMu.Lock() cs.detachClosed = true - if cs.h1State != nil && cs.h1State.OnDetachClose != nil { + // Acquire barrier: only invoke OnDetachClose once the WS upgrade has + // fully wired the conn (WSReady). Otherwise the read of OnDetachClose — + // and the ws.Close() it calls — races the upgrade installing it and the + // rest of the ws state on the async goroutine after Detach released + // detachMu (peer RST mid-upgrade). Not-yet-wired conns are still torn + // down via the fd close + read path below. + if cs.h1State != nil && cs.h1State.WSReady.Load() && cs.h1State.OnDetachClose != nil { cs.h1State.OnDetachClose() cs.h1State.OnDetachClose = nil } cs.detachMu.Unlock() - // Drop callbacks once the engine relinquishes the conn so any - // late goroutine references resolve to no-ops without crashing. - if cs.h1State != nil { + // Drop callbacks once the engine relinquishes the conn so any late + // goroutine references resolve to no-ops without crashing. Same acquire + // barrier as OnDetachClose above: only drop them once the WS upgrade has + // finished reading them (via WSReadPauser) and published WSReady — before + // that, the async upgrade goroutine is still reading PauseRecv/ResumeRecv. + if cs.h1State != nil && cs.h1State.WSReady.Load() { cs.h1State.PauseRecv = nil cs.h1State.ResumeRecv = nil } @@ -2529,7 +2538,10 @@ func (l *Loop) shutdown() { } cs.detachMu.Lock() cs.detachClosed = true - if cs.h1State != nil && cs.h1State.OnDetachClose != nil { + // Acquire barrier — see closeConn: skip OnDetachClose until the WS + // upgrade has fully wired the conn (WSReady) to avoid racing the + // post-Detach wiring on the async goroutine. + if cs.h1State != nil && cs.h1State.WSReady.Load() && cs.h1State.OnDetachClose != nil { cs.h1State.OnDetachClose() cs.h1State.OnDetachClose = nil } diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go index b01676ce..bca4d777 100644 --- a/engine/iouring/worker.go +++ b/engine/iouring/worker.go @@ -2364,14 +2364,23 @@ func (w *Worker) closeConn(fd int) { // Signal the detached goroutine's writeFn to stop writing. cs.detachMu.Lock() cs.detachClosed = true - if cs.h1State != nil && cs.h1State.OnDetachClose != nil { + // Acquire barrier: only invoke OnDetachClose once the WS upgrade has + // fully wired the conn (WSReady). Otherwise the read of OnDetachClose — + // and the ws.Close() it calls — races the upgrade installing it and the + // rest of the ws state on the async goroutine after Detach released + // detachMu (peer RST mid-upgrade). Not-yet-wired conns are still torn + // down via the fd close + read path below. + if cs.h1State != nil && cs.h1State.WSReady.Load() && cs.h1State.OnDetachClose != nil { cs.h1State.OnDetachClose() cs.h1State.OnDetachClose = nil } cs.detachMu.Unlock() - // Drop callbacks once the engine relinquishes the conn so any - // late goroutine references resolve to no-ops without crashing. - if cs.h1State != nil { + // Drop callbacks once the engine relinquishes the conn so any late + // goroutine references resolve to no-ops without crashing. Same acquire + // barrier as OnDetachClose above: only drop them once the WS upgrade has + // finished reading them (via WSReadPauser) and published WSReady — before + // that, the async upgrade goroutine is still reading PauseRecv/ResumeRecv. + if cs.h1State != nil && cs.h1State.WSReady.Load() { cs.h1State.PauseRecv = nil cs.h1State.ResumeRecv = nil } @@ -3766,7 +3775,10 @@ func (w *Worker) shutdown() { } cs.detachMu.Lock() cs.detachClosed = true - if cs.h1State != nil && cs.h1State.OnDetachClose != nil { + // Acquire barrier — see the primary close path: skip OnDetachClose + // until the WS upgrade has fully wired the conn (WSReady) to avoid + // racing the post-Detach wiring on the async goroutine. + if cs.h1State != nil && cs.h1State.WSReady.Load() && cs.h1State.OnDetachClose != nil { cs.h1State.OnDetachClose() cs.h1State.OnDetachClose = nil } diff --git a/internal/conn/h1.go b/internal/conn/h1.go index 1f5a3136..f620060e 100644 --- a/internal/conn/h1.go +++ b/internal/conn/h1.go @@ -140,6 +140,21 @@ type H1State struct { // the celeris public API: changes require a major version bump. OnDetachClose func() + // WSReady is the WebSocket-upgrade completion barrier. The WS middleware + // installs the detached-conn callbacks (RawWriteFn, pause/resume, + // idle-deadline, and OnDetachClose LAST) on the async-handler goroutine + // AFTER Context.Detach — which releases cs.detachMu (celeris#273/#309, so + // the guarded write path can re-lock it) — so those writes are NOT + // lock-serialised against the engine's closeConn, which reads OnDetachClose + // (and calls ws.Close through it) while tearing the conn down when a peer + // RSTs mid-upgrade. WSReady is Stored(true) as the final wiring step (in the + // OnWSDetachClose setter), publishing every prior write with release + // semantics; closeConn Loads it with acquire semantics before invoking + // OnDetachClose, so it either observes a fully-wired connection or skips WS + // teardown entirely (the conn is still closed via the fd/read path). Fresh + // per connection (NewH1State), so it starts false. + WSReady atomic.Bool + // OnError is called by the engine when an I/O failure occurs on a // detached connection (read error, write error, EPIPE, ECONNRESET, etc). // The WebSocket middleware uses this to surface engine-side errors @@ -942,6 +957,11 @@ func populateCachedStream(state *H1State, req *h1.Request, body []byte) *stream. } s.OnWSDetachClose = func(closeFn func()) { state.OnDetachClose = closeFn + // Release barrier: OnDetachClose is the LAST detached-conn callback + // the WS upgrade installs, so publishing WSReady here makes every + // prior wiring write (RawWriteFn, pause/resume, idle-deadline, and + // this OnDetachClose) visible to a closeConn that observes WSReady. + state.WSReady.Store(true) } s.OnWSSetError = func(errFn func(error)) { state.OnError = errFn diff --git a/middleware/websocket/conn.go b/middleware/websocket/conn.go index f25bb83e..a601a3a9 100644 --- a/middleware/websocket/conn.go +++ b/middleware/websocket/conn.go @@ -164,9 +164,17 @@ type engineWriter struct { conn *Conn } +// storedWriteErr boxes the engine-reported error so Conn.writeErr (an +// atomic.Value) always holds ONE concrete type. The engine surfaces errors of +// varying concrete types via OnError (errPeerClosed, syscall errors, +// ErrWriteClosed, …); storing them directly panics atomic.Value with "store of +// inconsistently typed value" on the second differing type. Boxing keeps the +// stored dynamic type constant. +type storedWriteErr struct{ err error } + func (w *engineWriter) Write(p []byte) (int, error) { - if e := w.conn.writeErr.Load(); e != nil { - return 0, e.(error) + if v := w.conn.writeErr.Load(); v != nil { + return 0, v.(storedWriteErr).err } if w.conn.closed.Load() { return 0, ErrWriteClosed diff --git a/middleware/websocket/websocket.go b/middleware/websocket/websocket.go index 0a5863f7..5bdc9061 100644 --- a/middleware/websocket/websocket.go +++ b/middleware/websocket/websocket.go @@ -229,7 +229,9 @@ func tryEngineUpgrade(c *celeris.Context, acceptKey, subproto string, ws.subprotocol = subproto c.SetWSErrorHandler(func(err error) { - ws.writeErr.Store(err) + if err != nil { + ws.writeErr.Store(storedWriteErr{err}) + } reader.closeWith(err) }) diff --git a/middleware/websocket/websocket_test.go b/middleware/websocket/websocket_test.go index ddc3f891..68d008df 100644 --- a/middleware/websocket/websocket_test.go +++ b/middleware/websocket/websocket_test.go @@ -2312,9 +2312,10 @@ func TestEngineWriteErrorPropagation(t *testing.T) { t.Fatalf("first write failed: %v", err) } - // Simulate engine reporting an EPIPE. + // Simulate engine reporting an EPIPE. Stored boxed (storedWriteErr) exactly + // as the OnError handler does, so atomic.Value holds one concrete type. want := errors.New("synthetic EPIPE") - ws.writeErr.Store(want) + ws.writeErr.Store(storedWriteErr{want}) // Subsequent writes return the engine error verbatim. err := ws.WriteMessage(TextMessage, []byte("second"))