epoll: drainRead ends a read turn after 64 KiB and re-arms the conn; reclaim counts a recv once, the async gate is ordered with the h2c switch, the replay shares the inline flush (v1.6.0, Refs #885 #886)

.github/workflows/ci.yml

@@ -222,9 +222,10 @@ jobs:
           #     workers and the runner's 8 MiB funds one; the step "celeris#592
           #     settled-route retime rig" below runs them with memlock raised and
           #     CELERIS_REQUIRE_IOURING_WORKERS=1, where a skip is a failure.
-          #   * TestRouteAdaptive_SettleReopenCost and
-          #     TestWriteBufBackpressureClosesSlowConsumer: opt-in measurements
-          #     (CELERIS_592_COST=1, GOTEST_BACKPRESSURE=1), not checks.
+          #   * TestRouteAdaptive_SettleReopenCost,
+          #     TestWriteBufBackpressureClosesSlowConsumer and
+          #     TestProbeDrainReadFloodLatency885: opt-in measurements
+          #     (CELERIS_592_COST=1, GOTEST_BACKPRESSURE=1, CELERIS_PROBE_885=1), not checks.
           #   * the four alloc-count tests in middleware/sse: they skip
           #     under -race and every step of this job is -race (celeris#684);
           #     the step "celeris#890 allocation guards and alloc-count tests,
@@ -239,7 +240,7 @@ jobs:
           # too. That is safe only while this step excludes /engine/iouring$ (the
           # pkgs line above; the step after it runs that package with its own list).
           # A name added here allows that name in every package this step runs.
-          allowed='TestAdaptiveSettledRouteRetime592/iouring/settled|TestAdaptiveSettledRouteRetime592/iouring/negctrl_async|TestAdaptiveSettledRouteRetime592/iouring/negctrl_learning|TestRouteAdaptive_SettleReopenCost|TestWriteBufBackpressureClosesSlowConsumer|TestBrokerFanOutFormatsOnce|TestRingBufferAppendAllocs|TestSlowClientSendNonFullQueueAllocs|TestSlowClientDropPathAllocs888|TestPauseAcceptServesDeferredIdleConnectionsSynackRetriesZero|TestPauseAcceptNoGuardLosesAtSynackRetriesZero|TestProbeNonLinux|TestValidateEngineRequiresLinux'
+          allowed='TestAdaptiveSettledRouteRetime592/iouring/settled|TestAdaptiveSettledRouteRetime592/iouring/negctrl_async|TestAdaptiveSettledRouteRetime592/iouring/negctrl_learning|TestRouteAdaptive_SettleReopenCost|TestWriteBufBackpressureClosesSlowConsumer|TestProbeDrainReadFloodLatency885|TestBrokerFanOutFormatsOnce|TestRingBufferAppendAllocs|TestSlowClientSendNonFullQueueAllocs|TestSlowClientDropPathAllocs888|TestPauseAcceptServesDeferredIdleConnectionsSynackRetriesZero|TestPauseAcceptNoGuardLosesAtSynackRetriesZero|TestProbeNonLinux|TestValidateEngineRequiresLinux'
           required='TestSendfileEndToEndLargeFile|TestSendfileHEADNoBody|TestSendfileRangeRequest|TestSendfileSubThresholdFallback'
           want=$(( $(tr '|' '\n' <<<"$required" | wc -l) ))
           ran=$(grep -cE "^=== RUN   (${required})\$" /tmp/root.log || true)

internal/engine/epoll/adopt.go

@@ -150,6 +150,17 @@ func (l *Loop) refuseAdopt(fd int, carry engine.Carryover) {
 // clean HTTP/1 request boundary; any pending request bytes wait in the kernel
 // socket receive buffer and are picked up by the EPOLLIN edge after registration.
 func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryover, now int64) {
+	l.attachFD(ctx, fd, carry, now, false)
+}
+
+// attachFD is attachAdoptedFD with the replay's recv count settled by the
+// caller: recvCounted says that the carried bytes were read, and counted, by
+// this loop already, which is a reclaim (reclaimTransplant: drainRead counted
+// the recv that read them before tryTransplant moved the conn), so the replay
+// must not count them again (celeris#886 item 25). A conn adopted from another
+// engine has had its bytes counted there, and counts them here as its first
+// recv (celeris#886).
+func (l *Loop) attachFD(ctx context.Context, fd int, carry engine.Carryover, now int64, recvCounted bool) {
 	if fd < 0 {
 		l.refuseAdopt(fd, carry)
 		return
@@ -232,16 +243,19 @@ func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryov
 	// answer to a request no handler saw (celeris#543).
 	if len(carry.Buffered) > 0 {
 		if l.async {
-			l.replayCarriedAsync(cs, fd, carry.Buffered)
+			l.replayCarriedAsync(cs, fd, carry.Buffered, recvCounted)
 			return
 		}
 		perr := conn.ProcessH1(cs.ctx, carry.Buffered, cs.h1State, l.handler, cs.writeFn)
 		// The replay is the recv event of this conn's first bytes and counts
 		// one, whatever the verdict (celeris#886). drainRead counts a recv
 		// after ProcessH1 too, but returns before it on an h2c upgrade and
 		// never counts a direct-body recv: a carry that is an upgrade request
-		// counts one here and its recv would have counted none.
-		l.reqBatch++
+		// counts one here and its recv would have counted none. Not when this
+		// loop read and counted the bytes itself (a reclaim: recvCounted).
+		if !recvCounted {
+			l.reqBatch++
+		}
 		if perr != nil {
 			l.endCarriedReplay(cs, fd, perr)
 			return
@@ -259,15 +273,18 @@ func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryov
 // thread; the conn is fresh from attachAdoptedFD, so no dispatch goroutine
 // exists yet and the loop owns cs.h1State until the promotion hands it over.
 // The conn's EPOLLIN is already registered, so nothing is left to arm.
-func (l *Loop) replayCarriedAsync(cs *connState, fd int, data []byte) {
+func (l *Loop) replayCarriedAsync(cs *connState, fd int, data []byte, recvCounted bool) {
 	cs.h1State.InlineMode = true
 	perr := conn.ProcessH1(cs.ctx, data, cs.h1State, l.handler, cs.writeFn)
 	// One request counted for the replay, whichever way it ends below: it is
 	// the recv event of this conn's first bytes (the promotion included, as
 	// in drainRead, celeris#626; celeris#886). Unlike drainRead it also counts
 	// a carry that ends in an h2c upgrade, which that path returns before
-	// counting.
-	l.reqBatch++
+	// counting. Not when this loop counted the recv already (a reclaim,
+	// recvCounted).
+	if !recvCounted {
+		l.reqBatch++
+	}
 	// A handler that hijacked inline has had cs released inside the Hijack
 	// call, as in drainRead, so cs is not touched again after ErrHijacked.
 	if errors.Is(perr, conn.ErrHijacked) {
@@ -342,67 +359,16 @@ func (l *Loop) endCarriedReplay(cs *connState, fd int, perr error) {
 }
 
 // flushCarriedReplay sends the responses a replay's ProcessH1 queued, as
-// drainRead's inline flush does after a recv (celeris#908). The replay used
+// drainRead's inline flush does after a recv (celeris#908): it is that flush,
+// flushInline, which the two share since celeris#885 item 17. The replay used
 // to flush once and stop: a response larger than the socket buffers left its
 // rest in cs.writeBuf with no EPOLLOUT armed and the dirty list untouched
 // (EPOLLIN is edge-triggered, and nothing reads from the socket again until
 // the client sends), so the client got a truncated body and a stalled conn.
 // It also left pendingBytes at the sum of the replayed responses, which the
 // write hooks read as a backlog, and ignored a refused write.
 //
-// It stays a copy of drainRead's inline flush on purpose: one shared function
-// would put a call on the per-request path of every recv, and the copy runs
-// once per adoption (celeris#885 item 17 asks for the verdict and upgrade
-// blocks to be shared, which h1verdict.go does; this block is not one of them).
-//
-// Loop thread, on a conn fresh from attachAdoptedFD. cs.detachMu is released
-// before any close, which takes it again.
+// Loop thread, on a conn fresh from attachAdoptedFD.
 func (l *Loop) flushCarriedReplay(cs *connState, fd int) {
-	mu := cs.detachMu
-	if mu != nil {
-		mu.Lock()
-	}
-	if csWritePending(cs) {
-		queued := 0
-		if cs.writeSince != 0 {
-			queued = csPendingBytes(cs) // before this flush (celeris#885)
-		}
-		if fErr := l.flushWrites(cs, true); fErr != nil {
-			if cs.h1State != nil && cs.h1State.OnError != nil {
-				cs.h1State.OnError(fErr)
-			}
-			if mu != nil {
-				mu.Unlock()
-			}
-			if cs.dirty {
-				l.removeDirty(cs)
-			}
-			l.closeConn(fd)
-			return
-		}
-		if !csWritePending(cs) {
-			cs.pendingBytes = 0
-			cs.endWrite()
-			if cs.dirty {
-				l.removeDirty(cs)
-			}
-			l.disarmEpollOut(cs)
-		} else {
-			// The kernel send buffer is full: sync pendingBytes and arm
-			// EPOLLOUT, which handleWritable carries on from (a pending
-			// sendfile too), for a detached conn as for any other: the dirty
-			// list is a 0 ms epoll_wait for as long as the peer does not read
-			// (celeris#885 item 42).
-			cs.pendingBytes = csPendingBytes(cs)
-			cs.noteWriteProgress(queued, cs.pendingBytes)
-			l.armEpollOut(cs)
-		}
-	}
-	refused := cs.writeRefused
-	if mu != nil {
-		mu.Unlock()
-	}
-	if refused {
-		l.closeWhenFlushed(cs)
-	}
+	l.flushInline(cs, fd)
 }

internal/engine/epoll/async_protocol_edge_885_linux_test.go

@@ -0,0 +1,227 @@
+//go:build linux
+
+package epoll
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/hpack"
+
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
+	"github.com/goceleris/celeris/internal/resource"
+)
+
+// celeris#885 item 28. drainRead's async gate reads cs.protocol on the loop
+// thread, while the conn's dispatch goroutine writes it (switchToH2Local, on an
+// h2c upgrade it parses). A client that waits for the 101 before it sends
+// anything else is ordered after the write by the 101's flush (x/sys's
+// unix.Write and unix.Read are race-annotated as one synchronisation, which is
+// all that keeps the detector quiet for it). A client that sends while the
+// goroutine is still in the switch is not: this test holds the goroutine inside
+// the switch (the handler of an HTTP/2 stream that came in the same write runs
+// inline after the protocol is written) and sends a frame meanwhile. Run it
+// under -race.
+
+type edge885Handler struct {
+	started chan struct{}
+	release chan struct{}
+	eng     atomic.Pointer[Engine]
+}
+
+func (h *edge885Handler) HandleStream(_ context.Context, s *stream.Stream) error {
+	var path string
+	for _, hd := range s.GetHeaders() {
+		if hd[0] == ":path" {
+			path = hd[1]
+		}
+	}
+	if path == "/arm" {
+		// Runs inline on the loop thread of the conn it came on: the pool wait's
+		// gate (poolWaitGate) sets refuseRecv there, and so does this.
+		h.eng.Load().loops[s.WorkerID].refuseRecv = true
+	}
+	if path == "/sdp" {
+		// The pool wait gate's pass over the detached conns, which runs on the
+		// loop thread before closeIdleH1.
+		h.eng.Load().loops[s.WorkerID].stopDetachedProducers()
+	}
+	if path == "/who" {
+		path = strconv.Itoa(int(s.WorkerID))
+	}
+	if path == "/hold" {
+		select {
+		case h.started <- struct{}{}:
+		default:
+		}
+		<-h.release
+	}
+	if s.ResponseWriter == nil {
+		return nil
+	}
+	return s.ResponseWriter.WriteResponse(s, 200,
+		[][2]string{{"content-type", "text/plain"}, {"content-length", strconv.Itoa(len(path))}}, []byte(path))
+}
+
+// Only /a is async: it promotes the conn, and the upgrade request behind it is
+// parsed by the dispatch goroutine.
+func (*edge885Handler) RouteAsync(_, path string) bool { return path == "/a" }
+func (*edge885Handler) HasAsyncRoutes() bool           { return true }
+
+func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
+	runEdge885(t, "")
+}
+
+// TestRefusesRecvReadsProtocolInOrderWithTheUpgradeSwitch885 is the same
+// scenario with the pool wait's gate up (refuseRecv): drainRead then asks
+// refusesRecv about the conn before anything else, and refusesRecv read
+// cs.protocol, which the dispatch goroutine writes in the switch. The flag is
+// set by a request on a second conn of the same loop, handled inline on the
+// loop thread.
+func TestRefusesRecvReadsProtocolInOrderWithTheUpgradeSwitch885(t *testing.T) {
+	runEdge885(t, "/arm")
+}
+
+// TestStopDetachedProducersReadsH1StateInOrderWithTheUpgradeSwitch885: the same
+// scenario, the gate's pass over the detached conns (stopDetachedProducers) run
+// on the loop thread, from a request on a second conn, while the goroutine is
+// inside the switch. The pass read cs.h1State, which the switch sets to nil,
+// before it took detachMu.
+func TestStopDetachedProducersReadsH1StateInOrderWithTheUpgradeSwitch885(t *testing.T) {
+	runEdge885(t, "/sdp")
+}
+
+// runEdge885 runs the scenario; via names the request, on a second conn of the
+// same loop, whose handler runs a loop-thread step of the pool wait's gate while
+// the dispatch goroutine is inside the switch ("" for none). After the 101 the
+// conn must speak HTTP/2: a PING is answered, so a gate that refused every
+// promoted conn (including a switched one) fails it.
+func runEdge885(t *testing.T, via string) {
+	gate := via != ""
+	h := &edge885Handler{started: make(chan struct{}, 1), release: make(chan struct{})}
+	e, addr := startE2E885(t, true, h, func(c *resource.Config) { c.EnableH2Upgrade = true })
+	h.eng.Store(e)
+	c, err := net.DialTimeout("tcp", addr, 3*time.Second)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer func() { _ = c.Close() }()
+	br := bufio.NewReader(c)
+
+	var other net.Conn
+	if gate {
+		who := func(c net.Conn, br *bufio.Reader, path string) string {
+			_ = c.SetDeadline(time.Now().Add(10 * time.Second))
+			if _, err := c.Write([]byte("GET " + path + " HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+				t.Fatalf("write %s: %v", path, err)
+			}
+			resp, err := http.ReadResponse(br, nil)
+			if err != nil {
+				t.Fatalf("read %s: %v", path, err)
+			}
+			b, _ := io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			return string(b)
+		}
+		w := who(c, br, "/who")
+		for i := 0; i < 64 && other == nil; i++ {
+			oc, err := net.DialTimeout("tcp", addr, 3*time.Second)
+			if err != nil {
+				t.Fatalf("dial: %v", err)
+			}
+			if who(oc, bufio.NewReader(oc), "/who") == w {
+				other = oc
+			} else {
+				_ = oc.Close()
+			}
+		}
+		if other == nil {
+			t.Fatal("no second conn on the first conn's loop in 64 dials")
+		}
+		defer func() { _ = other.Close() }()
+	}
+
+	var hb bytes.Buffer
+	enc := hpack.NewEncoder(&hb)
+	for _, f := range [][2]string{{":method", "GET"}, {":scheme", "http"}, {":path", "/hold"}, {":authority", "x"}} {
+		_ = enc.WriteField(hpack.HeaderField{Name: f[0], Value: f[1]})
+	}
+	var frames bytes.Buffer
+	fr := http2.NewFramer(&frames, nil)
+	_ = fr.WriteSettings()
+	_ = fr.WriteHeaders(http2.HeadersFrameParam{StreamID: 3, BlockFragment: hb.Bytes(), EndStream: true, EndHeaders: true})
+	// One write: a request that makes the conn async, the h2c upgrade request
+	// behind it, and the client preface, SETTINGS and a request on stream 3.
+	all := "GET /a HTTP/1.1\r\nHost: x\r\n\r\n" + h2cUpgradeHead908() + http2.ClientPreface + frames.String()
+	if _, err := c.Write([]byte(all)); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	select {
+	case <-h.started:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the goroutine never reached the HTTP/2 stream's handler after the switch")
+	}
+	if gate {
+		// The gate goes up on the loop: this request runs inline on its thread.
+		_ = other.SetDeadline(time.Now().Add(10 * time.Second))
+		if _, err := other.Write([]byte("GET " + via + " HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+			t.Fatalf("write /arm: %v", err)
+		}
+		resp, err := http.ReadResponse(bufio.NewReader(other), nil)
+		if err != nil {
+			t.Fatalf("read /arm: %v", err)
+		}
+		_ = resp.Body.Close()
+	}
+	// The goroutine is inside the switch, the protocol written. The client
+	// sends a PING: the loop reads it and reads cs.protocol.
+	var ping bytes.Buffer
+	_ = http2.NewFramer(&ping, nil).WritePing(false, [8]byte{1})
+	if _, err := c.Write(ping.Bytes()); err != nil {
+		t.Fatalf("write the PING: %v", err)
+	}
+	time.Sleep(300 * time.Millisecond)
+	close(h.release)
+
+	// The conn goes on: the response to /a and the 101 arrive.
+	_ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
+	var got strings.Builder
+	for {
+		line, err := br.ReadString('\n')
+		got.WriteString(line)
+		if err != nil {
+			t.Fatalf("read the response to /a and the 101: %v (got %q)", err, got.String())
+		}
+		if strings.Contains(line, "HTTP/1.1 101") {
+			for {
+				l, err := br.ReadString('\n')
+				if err != nil {
+					t.Fatalf("the 101's headers: %v", err)
+				}
+				if l == "\r\n" {
+					break
+				}
+			}
+			fr := http2.NewFramer(nil, br)
+			for {
+				f, err := fr.ReadFrame()
+				if err != nil {
+					t.Fatalf("no PING ACK on the switched conn (via %q): %v", via, err)
+				}
+				if pf, ok := f.(*http2.PingFrame); ok && pf.IsAck() {
+					return
+				}
+			}
+		}
+	}
+}