epoll: the write timeout of a parked response is a minimum rate on acknowledged bytes, no ceiling, inbound bytes are not progress (v1.6.0, Refs #885)

config.go

@@ -240,8 +240,14 @@ type Config struct {
 	//
 	// On the native engines (epoll, io_uring, adaptive) it is measured from the
 	// connection's last read (on io_uring also from the start and the completion
-	// of its sends, on epoll also from the end of a response the kernel did not
-	// take at once).
+	// of its sends). On epoll a response the kernel did not take whole is
+	// bounded by WriteTimeout's minimum rate alone, from the first sweep that
+	// finds it parked until it ends (see WriteTimeout): ReadTimeout and
+	// IdleTimeout run again from its end, and, where the socket reports no
+	// acknowledged bytes, from the last flush that took bytes. A request
+	// pipelined behind such a response is bounded by ReadHeaderTimeout for its
+	// headers and by ReadTimeout only once the response has been sent; the
+	// overrun check of an HTTP/1.1 async handler is suspended the same way.
 	//
 	// An HTTP/2 connection is not closed by it or by IdleTimeout while it has a
 	// stream open whose request is whole and no response is waiting for the
@@ -315,38 +321,68 @@ type Config struct {
 	// a large download or a chunked body written with [Context.StreamWriter]
 	// without Detach, stays under WriteTimeout as an absolute deadline for the
 	// whole response: raise it or set -1 for such routes. On the epoll engine
-	// (and adaptive while it runs on epoll) WriteTimeout bounds a response
-	// that is not detached differently: it is "no progress for WriteTimeout",
-	// the time a client may go without taking any of a response the kernel has
-	// not accepted whole, restarted each time it takes some. A download read
-	// steadily for longer than WriteTimeout is not cut there, and a client
-	// that reads nothing is closed WriteTimeout after the response stalled.
-	// Progress is seen when the engine flushes, which follows the socket
-	// becoming writable again, about when half of its send queue has drained,
-	// or the client sending something (a pipelined request, an HTTP/2 PING or
-	// WINDOW_UPDATE), whose read flushes what the room it made allows.
-	// A client that takes less than half a send queue per WriteTimeout and
-	// sends nothing (a deep send queue and a slow reader) is closed while it
-	// is still reading; one that takes
-	// about half a send queue per WriteTimeout is never closed, and holds the
-	// connection and the whole response queued for it in memory for as long as
-	// it keeps doing so (std and io_uring end such a response at WriteTimeout,
-	// as a whole). Nothing else bounds that on epoll: while a response is
-	// queued, ReadTimeout and IdleTimeout run from the last progress too, so
-	// they slide with it, and the earlier of ReadTimeout and WriteTimeout
-	// is the time a client may take no bytes at all (ReadTimeout 5s with
-	// WriteTimeout 60s closes a client that pauses for 6s). io_uring bounds the
-	// response as a whole, measured from the last read or completed flush
-	// of the connection (a client that sends another request restarts it), not
-	// from the request's headers as std does. On io_uring a response whose
+	// (and adaptive while it runs on epoll) WriteTimeout bounds a response that
+	// is not detached differently: it is a minimum rate on the bytes the client
+	// acknowledges, with no elapsed-time ceiling. A response the kernel did not
+	// accept whole is queued in the engine ("parked"); the engine's sweep reads
+	// how many bytes the client has acknowledged (tcp_info) and holds the
+	// connection another WriteTimeout if the client acknowledged at least 64 KiB
+	// in the last one (a floor of 64 KiB per WriteTimeout: 64 KiB/s at 1s,
+	// about 1 KiB/s at the 60s default), and closes it if it did not. A client
+	// that reads at or above the floor is never cut, however long the response
+	// takes, and one that reads steadily between two EPOLLOUT edges is not
+	// mistaken for a stalled one: a download completes whatever IdleTimeout is.
+	// A client that takes less, in whatever pieces (nothing, a byte at a time,
+	// a half send queue per WriteTimeout), is closed at the end of the first
+	// window, between WriteTimeout and WriteTimeout plus two sweeps after the
+	// response parked (the first sweep takes the baseline, the window ends at
+	// the first sweep a WriteTimeout later); one that stops after a window it
+	// passed, within two WriteTimeouts plus two sweeps after its last byte. A
+	// client that takes exactly 64 KiB per window holds the connection, and the
+	// part of the response not yet sent, for as many windows as the response has
+	// 64 KiB (a response of S bytes: S/64 KiB times WriteTimeout, 64 minutes
+	// for 4 MiB at 60s; a smaller WriteTimeout raises the floor), so size
+	// WriteTimeout with that in mind. WriteTimeout -1 is no write timeout at
+	// any rate. What the client sends is not progress: a pipelined request or
+	// an HTTP/2 PING does not hold a parked response. The times above hold
+	// with ReadHeaderTimeout set (the default): the engine's 25 ms sweep timer
+	// is armed only then, and without it a stalled response can outlast them by
+	// seconds (measured: WriteTimeout 1s, ReadHeaderTimeout -1, closed at 5 s,
+	// on main too).
+	//
+	// While the rule has a baseline for a parked response (from the first sweep
+	// that finds it parked), it is the only bound on the connection:
+	// ReadTimeout and IdleTimeout do not apply, so a client at or above the
+	// floor is not cut by either whatever they are next to WriteTimeout (the
+	// defaults are 60s, 600s and 60s), and one that stops is closed by the rule
+	// within the times above. They apply again from the end of the response. A
+	// socket that does not report acknowledged bytes (not TCP, or a TCP stack
+	// that leaves bytes_acked at 0) has the flush clock alone, "no flush took
+	// bytes for WriteTimeout", and ReadTimeout and IdleTimeout run from the
+	// last flush that took bytes there, as before the rule. The rule covers an
+	// HTTP/2 response parked on the socket too. A response that waits for the
+	// client's HTTP/2 flow-control window is not parked (nothing is queued on
+	// the socket) and WriteTimeout does not bound it on epoll: ReadTimeout and
+	// IdleTimeout do, from the client's last frame, so a client that sends a
+	// frame of any kind (a PING) at least once per ReadTimeout holds it, until
+	// per-stream deadlines are added. A request error read while a
+	// response is queued (a malformed request) puts the connection on the
+	// closing drain below, which counts each byte the client acknowledges and
+	// has no floor: a client that sends one and then takes a few bytes per
+	// WriteTimeout holds the connection, a known limit that is not changed here.
+	// io_uring is unchanged: it has no rate, and measures WriteTimeout from
+	// the last read, the start of a send from an idle connection or a
+	// completed send (a client that completes one send per WriteTimeout, or
+	// sends another request, restarts it), not from the request's headers as
+	// std does. On io_uring a response whose
 	// unsent remainder is under 64 KiB, to a client that stops reading, on an
 	// otherwise idle worker, can outlast WriteTimeout (measured with async
 	// responses of 96 to 144 KiB; the larger and the sync sizes are closed).
 	// On epoll, a response parked behind a handler that holds the connection's
 	// dispatch lock for longer than WriteTimeout can be closed at WriteTimeout
 	// even though nothing is queued for it (a narrow pipelined-handler case).
-	// A response that ends the connection (Connection: close, a refused write)
-	// has a drain of its own on both native engines: the connection is cut
+	// A response that ends the connection (Connection: close, a request error,
+	// a refused write) has a drain of its own on both native engines: the connection is cut
 	// once the client has taken no byte for WriteTimeout (epoll) or
 	// max(5s, WriteTimeout) (io_uring), and for 5 s on either when WriteTimeout
 	// is -1. A byte taken is a flush or send completion that moved the
@@ -371,7 +407,8 @@ type Config struct {
 	WriteTimeout time.Duration
 	// IdleTimeout is the max duration a keep-alive connection may be idle.
 	// Zero uses the default (600s). Set to -1 for no timeout. On an HTTP/2
-	// connection with a stream the server owes an answer for, see ReadTimeout.
+	// connection with a stream the server owes an answer for, see ReadTimeout;
+	// on epoll, for a response parked on the socket, see WriteTimeout.
 	IdleTimeout time.Duration
 	// ShutdownTimeout is the deadline of the graceful shutdown that cancelling
 	// the context of StartWithContext or StartWithListenerAndContext starts

internal/engine/epoll/conn.go

@@ -270,6 +270,14 @@ type connState struct {
 	// the 4-byte hole before closeSince: see TestConnStateStaysInItsSizeClass.)
 	gen uint32
 
+	// closeAcked, closeAckedKnown and closeSampled have a third user, a live
+	// conn's parked response (parkedBelowRate, write_rate_885.go, celeris#885
+	// item 51): the baseline of the rate rule's window, sampled by the
+	// sweep while writeSince is set and closeSince is 0, forgotten where the
+	// response's clock ends (endWriteSlow, the sweep's empty-queue case) and by
+	// markClosing, which hands them to the drain below. A conn is in one of the
+	// three states at a time.
+	//
 	// closeSince, closePending and closeAcked are the clock of that deferred
 	// close (celeris#876): closeSince is when the engine last saw the close
 	// move forward, in nanoseconds, from the request that deferred it

internal/engine/epoll/loop.go

@@ -370,6 +370,12 @@ type Loop struct {
 	// asking the kernel about a descriptor number that may now belong to
 	// something else. Written once before ready; read only after it.
 	listenAddr net.Addr
+
+	// ackedProbe replaces the getsockopt that reads a conn's acknowledged
+	// byte count when set (bytesAcked, write_rate_885.go): a test seam for a
+	// TCP stack whose TCP_INFO answers but does not fill it. nil in
+	// production. Loop thread (a test sets it before it drives the loop).
+	ackedProbe func(fd int) (uint64, error)
 }
 
 func newLoop(id, cpuID int, handler stream.Handler,
@@ -2279,6 +2285,7 @@ func (l *Loop) markClosing(cs *connState, pending int) {
 	}
 	cs.closeSince = time.Now().UnixNano()
 	cs.closePending = pending
+	cs.closeAckedKnown = false // a live conn's window baseline (parkedBelowRate) is not the drain's
 }
 
 // noteClosingProgress restarts the drain clock of a conn markClosing deferred
@@ -2373,6 +2380,14 @@ func (cs *connState) endWrite() {
 // inlines at its call sites (the drain paths of every response).
 func (cs *connState) endWriteSlow() {
 	cs.writeSince = 0
+	cs.closeAckedKnown = false // the baseline of the response's window (parkedBelowRate)
+	if cs.closeSince == 0 {
+		// A live conn's last sample is of the response that ended: the next
+		// one parks with no baseline and takes it at its first sweep, not
+		// after a quarter of WriteTimeout (closing and detached conns keep
+		// theirs, which they own).
+		cs.closeSampled = 0
+	}
 	cs.lastActivity = time.Now().UnixNano()
 }
 
@@ -4911,7 +4926,10 @@ func (cs *connState) endDetachedStall() {
 // write clock (writeSince), or is on the dirty list. WriteTimeout is "no
 // progress for WriteTimeout" there: the clock starts at the first partial flush
 // and restarts when a flush takes bytes (noteWriteProgress), where std's is an
-// absolute deadline for the whole response. It answers the instant IdleTimeout
+// absolute deadline for the whole response; where the peer's acknowledged bytes
+// can be read the minimum-rate rule decides instead (parkedBelowRate, celeris#885
+// item 51), and is then the only bound: the conn is exempt from the idle scan
+// for as long as the rule has a baseline. It answers the instant IdleTimeout
 // and ReadTimeout are measured from, the later of the last read and the clock,
 // and whether the conn is done for this sweep (closed, or exempt from the idle
 // scan). Kept out of line: the sweep of a conn without a clock does not run it.
@@ -4940,14 +4958,56 @@ func (l *Loop) sweepWriteClock(cs *connState, fd int, now, last int64) (from int
 		pending = csWritePending(cs)
 	}
 	if pending {
-		if wt := l.cfg.WriteTimeout; wt > 0 && time.Duration(now-ws) > wt {
-			l.closeConn(fd)
-			return last, true
+		if wt := l.cfg.WriteTimeout; wt > 0 {
+			// The minimum-rate rule (celeris#885 item 51) reaps a conn whose
+			// peer's acknowledged bytes it can see; the flush clock the rest.
+			if l.parkedBelowRate(cs, now, ws, int64(wt)) {
+				l.closeConn(fd)
+				return last, true
+			}
+			if cs.closeAckedKnown {
+				// The rule has a baseline for this response and is its only
+				// bound (celeris#885 item 51): a peer taking at least the floor
+				// per window is not stalled, whatever the idle clocks say. They
+				// run from the last flush that took bytes or the last window
+				// passed, and a client reading steadily between two EPOLLOUT
+				// edges moves neither (a ReadTimeout or IdleTimeout not above
+				// WriteTimeout, the defaults' shape, would cut it before the
+				// rule decides). The price: a request pipelined behind the
+				// response is bounded by ReadHeaderTimeout for its head (the
+				// hdrDL check above runs first and the loop keeps EPOLLIN armed
+				// while parked), but a pipelined request whose head is whole and
+				// whose body is pending has no header deadline, and ReadTimeout
+				// does not bound it until the response ends (measured: ReadTimeout
+				// 1 s cut it at 1.34 s on main, the head held it for the run;
+				// after the drain ReadTimeout applies again, gone at 6.94 s on
+				// both). The skip also bypasses h2ServerOwes' dispatchBusy answer
+				// for an async HTTP/1.1 conn (read, not measured): a pipelined
+				// handler that overruns ReadTimeout is not closed while an earlier
+				// response on the conn is rate-held. Bounded by the rule: the
+				// client must keep acknowledging the floor. A peer that stops is
+				// closed by the rule one window after it falls under the floor.
+				// Back to the idle scan when the response ends (endWrite) or the
+				// baseline is lost (a failed sample).
+				return last, true
+			}
+			if time.Duration(now-cs.writeSince) > wt {
+				l.closeConn(fd)
+				return last, true
+			}
 		}
 	} else {
 		// End the clock as endWrite does, from its last stamp, which is no
-		// later than the drain.
+		// later than the drain. A response the rule governed has no recent
+		// stamp (it moves once per window): the drain is dated by this
+		// sweep, so that ReadTimeout or IdleTimeout, shorter than the
+		// window, do not cut the conn on the instant its response ended.
+		if cs.closeAckedKnown {
+			ws = now
+		}
 		cs.writeSince = 0
+		cs.closeAckedKnown = false
+		cs.closeSampled = 0 // the next response's baseline is not delayed by this one's sample
 		if ws > last {
 			cs.lastActivity = ws
 		}

internal/engine/epoll/write_clock_885_linux_test.go

@@ -76,10 +76,16 @@ func TestWriteClockRestartsWhenTheFlushTakesBytes885(t *testing.T) {
 				"the clock must restart when a flush takes bytes", time.Since(start).Round(time.Millisecond), bound)
 		}
 	}
-	time.Sleep(bound + 100*time.Millisecond)
-	r.l.checkTimeouts()
+	// The peer took bytes in the window that is running (the rate rule holds a
+	// conn another window when the last passed, celeris#885 item 51), and none
+	// after: it goes when that window has passed, within two WriteTimeouts.
+	deadline := time.Now().Add(2*bound + 200*time.Millisecond)
+	for r.open() && time.Now().Before(deadline) {
+		time.Sleep(25 * time.Millisecond)
+		r.l.checkTimeouts()
+	}
 	if r.open() {
-		t.Fatalf("a peer that stopped reading was not reaped %v after its last byte", bound+100*time.Millisecond)
+		t.Fatalf("a peer that stopped reading was not reaped %v after its last byte", 2*bound+200*time.Millisecond)
 	}
 }