diff --git a/.github/root-test-shards.txt b/.github/root-test-shards.txt index d94166fcf..b24089817 100644 --- a/.github/root-test-shards.txt +++ b/.github/root-test-shards.txt @@ -60,3 +60,32 @@ b TestGlobalMiddlewareOfAnAsyncRouteKeepsTheLoop848 b TestH2MaxConcurrentStreamsHeadersContinuation981 b TestNoDeprecatedPublicAPI830 b TestHijackedConnNotCutByServerTimeoutsStd855 + +# The write deadline's cells (celeris#889, D2): 3 to 28 s each at the root, 70 s together, shared out so that the rest stays inside its timeout. +a TestH2SparseWindowUpdatesAreNotCut889 +a TestH2PausedAndDribbledStreamsOnAHealthyConnAreBounded889 +a TestH2StalledTailIsResetAtWriteTimeout889 +a TestH2WindowUpdateDribblerIsResetAtTheFirstJudgement889 +a TestH2SilentClientWithASmallRemainderIsLetGo889 +b TestH2ParallelDownloadsSharingASlowLinkAreNotCut889 +b TestH2WriteRateFloor889 +b TestH2ConnWindowStarvedStreamsAreJudgedPerConnection889 +b TestH2FlowingDownloadAndQuietStreamAreNotCut889 +b TestH2SilentClientWithAStalledResponseKeepsTheConnUntilTheStreamIsReset889 + +# The rest step of the Unit job (test/server, -race, -timeout=540s) was at 539.2 s on the merge of #1169 (main alone: 501 s), 0.8 s from its timeout: +# the 14 slowest tests of the root packages still in it (8 to 34 s each, 216 s of CI test time) move to the shards, which have 300 s of room. +a TestShutdownConnCloseContract884 +a TestStreamWriterOverCapH1_875 +a TestDefaultMaxRequestBodySize4MiB889 +a TestConcurrentResponsesOwnTheirBodies +a TestShutdownRunningAsyncHandlerAnswersCloseWithSlowDownloader884 +a TestShutdownPoolWaitStopsServingH1AndDetached884 +a TestSplitBodyResponsesKeepTheConnection +b TestIOUringShutdownGateConns884 +b TestSeveralCookieLinesAreOneFieldOnEveryEngine889 +b TestExpectContinueOverLimit889 +b TestEpollStaticFileDownloadAtAModerateRateCompletes885 +b TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885 +b TestShutdownIdleAndInProgressH1ConnsWithoutHTTP2_1165 +b TestShutdownPoolWaitGateClosesIdleH1AndEndsDetached885 diff --git a/config.go b/config.go index da2442af4..54a1c5c18 100644 --- a/config.go +++ b/config.go @@ -259,8 +259,9 @@ type Config struct { // read. A request without a body has no deadline. // // The connection's own clock (the last read) does not close an HTTP/2 - // connection that has a stream open whose request is whole and no response - // waiting for the client's flow-control window: a stream the server is + // connection that has a stream open whose request is whole (a response + // waiting for the client's flow-control window is bounded by WriteTimeout, + // see there, and no longer takes the exemption away): a stream the server is // writing to (an SSE stream) outlives a client that sends no frame. A stream // whose handler never answers is exempt for as long as it holds the stream, // where an HTTP/1.1 async handler that overruns ReadTimeout is closed. When @@ -310,7 +311,8 @@ type Config struct { // OnDisconnect runs), whatever the middleware's idle deadline or the last // read of the connection. A peer that reads, however slowly, and a // connection with nothing queued are not closed (there is no minimum - // rate); -1 sets no such bound. The close comes between about 3/4 + // rate; an HTTP/2 stream whose client lets its flow-control window run out + // is bounded by the minimum rate described below); -1 sets no such bound. The close comes between about 3/4 // and 5/4 of WriteTimeout after the peer last took a byte (its progress // is sampled every quarter of WriteTimeout; a SEND completion restarts the // clock at once), plus the engine's sweep interval (up to about 0.1 s on @@ -372,10 +374,54 @@ type Config struct { // 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 + // the socket); WriteTimeout bounds its stream in its own right, as a + // minimum rate, on the native engines: in each WriteTimeout window a stream + // that holds buffered bytes must have had at least 64 KiB of them (or all + // it held, if less) sent, released by the client's WINDOW_UPDATEs, or it is + // reset with RST_STREAM(INTERNAL_ERROR), as net/http resets a stream at its + // write deadline. There is no ceiling, where std's deadline is absolute: a + // stream that releases at least that on average, lagging the rate by no + // more than four windows' worth (surplus released in one update carries to + // the next windows, and the bytes in flight when the stream stalled cover its + // first), is not cut however long its response takes. A PING or a + // WINDOW_UPDATE of a byte is not progress. A client that updates in big steps + // (nghttp2 and curl, at half the window) is fine while the gap between its + // updates stays under about five WriteTimeouts: half the window divided by the + // rate it reads at. When the connection as a whole releases at least 64 KiB + // in a window, a stream that takes less (parallel downloads share the + // client's socket) is held while it takes half its fair share of it, and for + // three windows in a row when it takes less: a download queued behind its + // siblings' bytes in the client's socket is cut where the bytes ahead of it + // take longer than three WriteTimeouts to read (N streams of W bytes in flight + // on a link of L bytes/s: N x W / L), so a server whose clients are on slow + // links should set WriteTimeout to 5s or more. When the connection itself + // releases less than 64 KiB while it holds more, the first window is the + // verdict. The end of a response is given grace: a stream that holds less than + // 64 KiB, and has had half of it sent in the window, or any stream while the + // connection as a whole holds less than 64 KiB (the last bytes of the transfer + // are on the wire, and the grant for them can wait behind its siblings' bytes in + // the client's socket for more than a window), is held for three windows more + // and reset in the fourth if nothing moves it. A stream the client never + // grants is reset after 1+k windows alone, or 4+k with siblings that keep the + // connection moving, k being the number of whole 64 KiB in the bytes in flight + // when it stalled (about the peer's initial window), at most four: 1 and 4 + // WriteTimeouts for the default 64 KiB window, 5 and 8 for a window of 256 KiB or + // more. When what it has left to send is under 64 KiB (the response is a + // little longer than the window), the bytes in flight are spent a whole floor a + // window, and the tail grace above follows: 5 WriteTimeouts for the default + // window, 6 for 128 KiB, 8 for 256 KiB or more. A one-byte WINDOW_UPDATE + // dribbler is reset in window max(2, 1+k): the 2nd of the default window, the + // 3rd of 128 KiB, the 4th of 192 KiB, the 5th of 256 KiB or more. A stream + // whose own window has room but whose bytes wait for the connection's is + // credited with what the whole connection released. While such a stream + // lives, ReadTimeout and IdleTimeout do not close the connection + // (WriteTimeout is its bound), so a silent client's unread response, and the + // buffer it pins, lives for up to 8 WriteTimeouts (5, with the default window) + // where it used to be dropped with the whole connection at ReadTimeout. A + // handler blocked waiting for window keeps its own, absolute, bound of + // WriteTimeout from the start of the wait, whatever the client reads: parallel + // downloads on one connection whose responses exceed the connection's 4 MiB + // output budget can lose the later ones to it. 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 diff --git a/internal/conn/h2.go b/internal/conn/h2.go index 0a90c49fa..5aa70a71f 100644 --- a/internal/conn/h2.go +++ b/internal/conn/h2.go @@ -625,6 +625,7 @@ func NewH2State(handler stream.Handler, cfg H2Config, write func([]byte), wake * s.idleNs = int64(cfg.IdleTimeout) proc.GetManager().SetHints(&s.hints) proc.GetManager().SetRequestTimeout(cfg.ReadTimeout) + proc.GetManager().SetWriteTimeout(cfg.WriteTimeout) proc.SetHeaderBlockTimeout(cfg.ReadHeaderTimeout) if lg := cfg.Logger; lg != nil { proc.OnHandlerPanic = func(id uint32, r any, stack []byte) { @@ -1125,8 +1126,8 @@ func (s *H2State) Sweep(now int64, write func([]byte)) (res SweepResult) { res.CloseNow = true } } - dl, td, hd := h.ReqDL.Load(), h.TombDL.Load(), h.HdrDL.Load() - if (dl == 0 || now < dl) && (td == 0 || now < td) && (hd == 0 || now < hd) { + dl, td, hd, wd := h.ReqDL.Load(), h.TombDL.Load(), h.HdrDL.Load(), h.WrDL.Load() + if (dl == 0 || now < dl) && (td == 0 || now < td) && (hd == 0 || now < hd) && (wd == 0 || now < wd) { return res } if !s.mu.TryLock() { @@ -1153,6 +1154,24 @@ func (s *H2State) Sweep(now int64, write func([]byte)) (res SweepResult) { } } } + if wd != 0 && now >= wd { + // The write windows that have ended: the streams that released less than + // their floor are reset with RST_STREAM(INTERNAL_ERROR), in batches and + // under the same budget as the requests (a stalled client with many + // streams is one window's worth). + var ids [32]uint32 + began := time.Now() + for { + got := m.ExpiredWrites(now, ids[:0]) + for _, id := range got { + p.ExpireWrite(id) + res.Expired++ + } + if len(got) < len(ids) || time.Since(began) > expireBudget { + break + } + } + } if td != 0 && now >= td { p.DropTombstones(now) } diff --git a/internal/conn/h2_bench_cfg_889_test.go b/internal/conn/h2_bench_cfg_889_test.go index d2cc1ac0a..56f5a6663 100644 --- a/internal/conn/h2_bench_cfg_889_test.go +++ b/internal/conn/h2_bench_cfg_889_test.go @@ -3,9 +3,8 @@ package conn import "time" // benchCfg889 is the H2Config the request-deadline benchmarks run with: the -// defaults of ReadTimeout and ReadHeaderTimeout, so that a request with a body -// arms its deadline. (main has no such fields; the comparison builds main with a -// copy of this file that returns H2Config{}.) +// defaults of ReadTimeout, ReadHeaderTimeout and WriteTimeout, so that a request +// with a body arms its deadline and the write deadline is on. func benchCfg889() H2Config { - return H2Config{ReadTimeout: 60 * time.Second, ReadHeaderTimeout: 10 * time.Second} + return H2Config{ReadTimeout: 60 * time.Second, ReadHeaderTimeout: 10 * time.Second, WriteTimeout: 60 * time.Second} } diff --git a/internal/conn/h2_sweep_889_test.go b/internal/conn/h2_sweep_889_test.go index bd74b4cd4..0c1a297bb 100644 --- a/internal/conn/h2_sweep_889_test.go +++ b/internal/conn/h2_sweep_889_test.go @@ -9,6 +9,8 @@ import ( "golang.org/x/net/http2" "golang.org/x/net/http2/hpack" + + "github.com/goceleris/celeris/internal/protocol/h2/stream" ) // H2State.Sweep (celeris#889): the engines' per-connection look at the request @@ -537,3 +539,62 @@ func TestH2StateStaysInItsSizeClass889(t *testing.T) { t.Errorf("sizeof(H2State) + the 8-byte malloc header = %d, want at most 704 (its size class)", got) } } + +// TestSweepResetsAStalledStreamWithInternalError889: a stream that holds bytes for +// the client's window and releases less than its floor in WriteTimeout is reset +// by the sweep with RST_STREAM(INTERNAL_ERROR), written to the connection through +// the write queue; one that released its floor is not. +func TestSweepResetsAStalledStreamWithInternalError889(t *testing.T) { + r := newSweepRig889(t, H2Config{WriteTimeout: time.Second}) + r.written() + m := r.st.processor.GetManager() + m.UpdateConnectionWindow(1 << 20) + // A client with an initial window of 0: nothing is in flight when a stream stalls, + // so the first window is judged on what is released and not on the seed. + r.frames(func(fr *http2.Framer) { + _ = fr.WriteSettings(http2.Setting{ID: http2.SettingInitialWindowSize, Val: 0}) + }) + mk := func(id uint32) *stream.Stream { + s := m.CreateStream(id) + s.SetState(stream.StateHalfClosedRemote) + s.SetWindowSize(0) + s.BufferOutbound(make([]byte, 1<<20), true) + return s + } + a, b := mk(1), mk(3) + now := time.Now().UnixNano() + if res := r.st.Sweep(now+int64(500*time.Millisecond), r.write); res.Expired != 0 { + t.Fatalf("expired half a window in: %+v", res) + } + // Stream 3 releases its floor in every window; stream 1 nothing. The connection + // moves, so stream 1 is held for three windows (it may be queued behind its + // sibling) and the fourth is the verdict. + var res SweepResult + for w := 1; w <= 4; w++ { + b.SetWindowSize(65536) + m.FlushOutbound(b, func(uint32, bool, []byte) {}) + res = r.st.Sweep(now+int64(w)*int64(1100*time.Millisecond), r.write) + if (w < 4) != (res.Expired == 0) { + t.Fatalf("window %d: %+v", w, res) + } + } + if res.Expired != 1 || !res.Wrote { + t.Fatalf("sweep after the window: %+v, want one stream expired and the conn written", res) + } + var rsts []wframe889 + for _, f := range r.written() { + if f.typ == http2.FrameRSTStream { + rsts = append(rsts, f) + } + } + if len(rsts) != 1 || rsts[0].stream != 1 || rsts[0].code != http2.ErrCodeInternal { + t.Fatalf("frames written: %+v, want RST_STREAM(INTERNAL_ERROR) for stream 1 only", rsts) + } + if _, ok := m.GetStream(1); ok { + t.Fatal("stream 1 is still on the connection") + } + if _, ok := m.GetStream(3); !ok { + t.Fatal("stream 3, which released its floor, was taken") + } + _ = a +} diff --git a/internal/protocol/h2/stream/deadline.go b/internal/protocol/h2/stream/deadline.go index 6bbebdb0f..79788cdc3 100644 --- a/internal/protocol/h2/stream/deadline.go +++ b/internal/protocol/h2/stream/deadline.go @@ -163,7 +163,7 @@ func (p *Processor) TakeIgnoredActivity() bool { // ReadHeaderTimeout): ReadHeaderTimeout after the first HEADERS frame of a block // that does not end with it, the connection ends with GOAWAY(ENHANCE_YOUR_CALM). // 0 turns it off. Set once, before the first frame. -func (p *Processor) SetHeaderBlockTimeout(d time.Duration) { p.hdrTimeout = d } +func (p *Processor) SetHeaderBlockTimeout(d time.Duration) { p.hdrTimeout = clampTimeout(d) } // armHeaderBlock starts the deadline of the header block that has just begun // (a HEADERS frame without END_HEADERS). diff --git a/internal/protocol/h2/stream/flowcontrol.go b/internal/protocol/h2/stream/flowcontrol.go index 3061cbc23..444a1cde1 100644 --- a/internal/protocol/h2/stream/flowcontrol.go +++ b/internal/protocol/h2/stream/flowcontrol.go @@ -94,6 +94,7 @@ func (m *Manager) ReserveSendWindow(s *Stream, want int32) int32 { m.notifyStreamWindow(s.ID) } } + m.creditConn(connGrant) return connGrant } diff --git a/internal/protocol/h2/stream/manager.go b/internal/protocol/h2/stream/manager.go index a339f5ba4..58c0f3529 100644 --- a/internal/protocol/h2/stream/manager.go +++ b/internal/protocol/h2/stream/manager.go @@ -21,12 +21,14 @@ type Manager struct { maxFrameSize uint32 // peer's SETTINGS_MAX_FRAME_SIZE (caps what WE send) localMaxFrameSize uint32 // our advertised SETTINGS_MAX_FRAME_SIZE (caps what PEER sends us) initialWindowSize uint32 + initWin atomic.Uint32 // initialWindowSize for readers that hold no lock (write_deadline.go) activeStreams atomic.Uint32 requestOpen atomic.Uint32 hints *DeadlineHints ownHints DeadlineHints reqTimeout time.Duration - headerTableSize uint32 // peer's SETTINGS_HEADER_TABLE_SIZE (bound on our HPACK dynamic table) + wrTimeout time.Duration // the write deadline (SetWriteTimeout) + headerTableSize uint32 // peer's SETTINGS_HEADER_TABLE_SIZE (bound on our HPACK dynamic table) pendingConnWindowUpdate uint32 pendingStreamUpdates map[uint32]uint32 windowUpdateMu sync.Mutex @@ -45,6 +47,15 @@ type Manager struct { sendWindowMu sync.Mutex sendWaiters []*sendWaiter + // The write deadline's books (write_deadline.go): a record per stream that + // holds buffered bytes, in stalls, guarded by stallMu; stallN is how many, for + // a load on the paths that have none; connReleased is the bytes of DATA the + // connection released while a stream was stalled (ReserveSendWindow). + stallMu sync.Mutex + stalls map[uint32]*stall + stallN atomic.Int32 + connReleased atomic.Int64 + // retired holds the pool-handler streams whose handler has returned and // which are out of the map, waiting for the event loop to return them to // the stream pool (retire, drainRetired; celeris#951). hasRetired is @@ -89,6 +100,7 @@ func NewManager() *Manager { headerTableSize: 4096, // RFC 7540 §6.5.2 default } m.hints = &m.ownHints + m.initWin.Store(65535) return m } @@ -377,6 +389,7 @@ func (m *Manager) takeLocked(id uint32, s *Stream) (release bool) { prev := State(s.state.Swap(int32(StateClosed))) m.updateActiveCount(prev, StateClosed) m.dropPlaceholderLocked(s) + m.dropStall(id) f := s.flags.Load() if f&flagReqDeadline != 0 { m.disarmRequestLocked(s) @@ -428,6 +441,7 @@ func (m *Manager) RemoveStreamFromMap(id uint32) { delete(m.streams, id) m.priorityTree.RemoveStream(id) m.mu.Unlock() + m.dropStall(id) m.windowUpdateMu.Lock() delete(m.pendingStreamUpdates, id) @@ -478,6 +492,7 @@ func (m *Manager) retire(s *Stream) { s.OutboundBuffer = nil } id := s.ID + m.dropStall(id) // The request is dropped now, not at the loop's next batch: a detached // Context that reads the stream meanwhile sees nothing, as it did when // this goroutine reset the stream (celeris#904), and a connection that @@ -601,7 +616,7 @@ func (m *Manager) CountActiveStreams() int { func (m *Manager) ServerOwes() bool { return m.activeStreams.Load() > 0 && (m.reqTimeout > 0 || m.requestOpen.Load() == 0) && - m.outboundHeld.Load() == 0 && + (m.wrTimeout > 0 || m.outboundHeld.Load() == 0) && m.sendWindowWaiters.Load() == 0 } @@ -684,6 +699,7 @@ func (m *Manager) ApplySetting(id uint16, val uint32) { } m.mu.Lock() m.initialWindowSize = val + m.initWin.Store(val) m.mu.Unlock() case 0x5: // MAX_FRAME_SIZE if val < 16384 || val > 16777215 { @@ -791,6 +807,7 @@ func (m *Manager) Close() { // that returns from here on releases its stream itself (retire). m.closed = true m.placeholders = 0 + m.dropAllStalls() retired := m.retired m.retired = nil m.hasRetired.Store(false) @@ -809,7 +826,7 @@ func (m *Manager) Close() { // SetRequestTimeout sets the request deadline (the server's ReadTimeout): a // stream whose request is not whole that long after its HEADERS is expired by // ExpiredRequests. 0 turns it off. Set once, before the first frame. -func (m *Manager) SetRequestTimeout(d time.Duration) { m.reqTimeout = d } +func (m *Manager) SetRequestTimeout(d time.Duration) { m.reqTimeout = clampTimeout(d) } // RequestTimeout is the deadline SetRequestTimeout set. func (m *Manager) RequestTimeout() time.Duration { return m.reqTimeout } @@ -873,6 +890,7 @@ type DeadlineHints struct { ReqDL atomic.Int64 TombDL atomic.Int64 HdrDL atomic.Int64 + WrDL atomic.Int64 // the earliest end of a stalled stream's write window } // SetHints points the manager (and the processor that owns it) at h, which the diff --git a/internal/protocol/h2/stream/outbound.go b/internal/protocol/h2/stream/outbound.go index d93a27b14..2bc12c7f9 100644 --- a/internal/protocol/h2/stream/outbound.go +++ b/internal/protocol/h2/stream/outbound.go @@ -464,6 +464,7 @@ func (m *Manager) FlushOutbound(s *Stream, send OutboundSend) (finished bool) { finished = last && (s.OutboundEndStream || s.outboundAbandoned) send(s.ID, end, buf.Next(n)) m.refundOutbound(n) // celeris#893 + m.noteReleased(s.ID, n) if buf.Len() == 0 { buf.Reset() } @@ -538,6 +539,7 @@ func (m *Manager) sendOrBufferLocked(s *Stream, data []byte, send OutboundSend) } s.OutboundBuffer.Write(data[n:]) m.chargeOutbound(len(data) - n) + m.noteBuffered(s, len(data)-n) } } @@ -713,6 +715,7 @@ func (m *Manager) DiscardOutbound(s *Stream) { m.refundOutbound(buf.Len()) buf.Reset() } + m.dropStall(s.ID) s.OutboundEndStream = false s.outboundAbandoned = false } diff --git a/internal/protocol/h2/stream/processor.go b/internal/protocol/h2/stream/processor.go index def930646..7730a3a97 100644 --- a/internal/protocol/h2/stream/processor.go +++ b/internal/protocol/h2/stream/processor.go @@ -1184,6 +1184,7 @@ func (p *Processor) handleSettings(f *http2.SettingsFrame) error { newWindowSize := int32(s.Val) delta := newWindowSize - oldWindowSize p.manager.initialWindowSize = s.Val + p.manager.initWin.Store(s.Val) for sid, stream := range p.manager.streams { oldWin := stream.windowSize.Load() @@ -2364,6 +2365,7 @@ func (p *Processor) handleGoAway(f *http2.GoAwayFrame) error { prev := State(stream.state.Swap(int32(StateClosed))) p.manager.updateActiveCount(prev, StateClosed) p.manager.disarmRequestLocked(stream) + p.manager.dropStall(streamID) stream.Cancel() if stream.flags.Load()&flagAsyncRunning == 0 { stream.Release() diff --git a/internal/protocol/h2/stream/stream.go b/internal/protocol/h2/stream/stream.go index 16b74fc5f..4420d418b 100644 --- a/internal/protocol/h2/stream/stream.go +++ b/internal/protocol/h2/stream/stream.go @@ -1069,6 +1069,9 @@ func (s *Stream) bufferOutbound(data []byte, endStream bool) { } s.OutboundBuffer.Write(data) s.OutboundEndStream = endStream + if m := s.manager; m != nil { + m.noteBuffered(s, len(data)) + } s.mu.Unlock() } diff --git a/internal/protocol/h2/stream/write_deadline.go b/internal/protocol/h2/stream/write_deadline.go new file mode 100644 index 000000000..58ef90f6a --- /dev/null +++ b/internal/protocol/h2/stream/write_deadline.go @@ -0,0 +1,446 @@ +package stream + +import ( + "sync/atomic" + "time" + + "golang.org/x/net/http2" +) + +// The per-stream write deadline (celeris#889, D2). +// +// A response whose bytes wait for the peer's flow-control window (the tail of an +// inline handler's body, or a pool handler's, buffered in the stream's +// OutboundBuffer, Manager.outboundHeld) is bounded by WriteTimeout as a minimum +// rate, the rule #1119 gives a response parked on the socket: in a window of +// WriteTimeout the h2 layer must release at least min(64 KiB, what the stream +// held when the window began) of the stream's DATA, or the stream is reset with +// RST_STREAM(INTERNAL_ERROR), as net/http resets a stream at its write deadline +// and as #906 does for a pool handler that waits too long. There is no ceiling: +// a stream that takes its floor in every window is never cut, however long the +// response. The measure is bytes released (FlushOutbound), never frames: a PING, +// or a WINDOW_UPDATE that lets one byte out, moves nothing. +// +// What counts. A stream that released its floor in the window is held. The +// surplus beyond the floor carries to the next windows, up to writeBankWindows +// windows' worth (the bank), and a stream's first window is covered by the bytes +// it had in flight when it stalled (the seed, capped the same way): a client that +// sends few WINDOW_UPDATEs (nghttp2 and curl update when half of the window is +// consumed) releases in big steps, with windows between its updates that release +// nothing, and a rule that counted only the window it fell in reset it while it +// read far above the floor (review r7 F1). The guarantee is on the average: a stream +// that releases at least the floor per window on average, lagging that rate by no +// more than writeBankWindows windows' worth, survives however long its response. +// +// The connection's own rate matters too. A stream whose OWN send window has room +// but whose bytes wait for the CONNECTION window is credited with what the whole +// connection released in the window (a sibling takes it). And when the connection as +// a whole releases at least the floor, a stream that is below its own floor is not +// judged alone (review r7 F6): N parallel downloads that share a slow socket each +// get link/N, and the client reads the socket in the order the server sent, so a +// stream can wait behind its siblings' bytes for a window or more. It is held while +// it takes its share, at least half of floor/N (N the streams stalled), and held for +// up to writeIdleWindows consecutive windows in which it takes less (paused, +// dribbled or queued behind its siblings), and reset after that: the memory a +// stream that is not read pins is bounded by writeIdleWindows+1 windows, not by +// the siblings' life. When the connection itself releases less than the floor +// nothing is held (but a tail, below): the first window that does not move is the +// verdict, as for a single stream. A stream whose own window is shut is judged by its own releases +// and these allowances, even when the connection window is shut too, or a client +// could shut the conn window by dribbling to a sibling and shelter every stalled +// stream for ever. +// +// The end of a response is given a grace (tail grace, r8 F-A and r9 F-A2): the grant +// for the last bytes waits behind the siblings' bytes in the client's socket and can +// land after the window, and the backlog behind them can take more than a window to +// drain. A stream that holds under one floor and has had half of it released, and any +// stream while the whole connection holds under one floor, is held for up to +// writeIdleWindows windows without proof, reset in the next. The bank does not hide +// an unread stream: a window held on the bank alone pays a whole floor, however +// small the rest, so a response a little longer than the window is let go in the +// same number of windows as a long one (r9 F-H). +// +// State: a record per stalled stream, in a small map on the Manager, made on the +// first stall of the connection. Only a stream that holds buffered bytes has one, +// which is the rare case (a client that let its window run out), so no field +// costs every stream of every connection, and the common send touches nothing +// (noteBuffered is called where a buffer fills, not where DATA is sent). The map +// is guarded by stallMu, a leaf lock below a stream's mu (the buffering and +// flushing paths hold it) and below the manager's; the judgement takes it alone. +// The sweep reads one hint, DeadlineHints.WrDL, per connection. + +// writeMinRelease is the floor of the rate: the bytes a stalled stream must +// release per WriteTimeout window to be held another (celeris#885's +// parkedMinAcked, at the h2 layer): 64 KiB per window, about 64 KiB/s at a 1 s +// WriteTimeout and about 1 KiB/s at the 60 s default. +const writeMinRelease = 64 << 10 + +// writeBankWindows is how many windows' worth of surplus release a stream may carry +// to later windows (and how many its in-flight bytes may cover at the start): the +// bound on how far behind the floor's rate a client's WINDOW_UPDATEs may lag. +const writeBankWindows = 4 + +// writeIdleWindows is how many consecutive windows a stream that takes less than its +// share may be held, on a connection that moves above the floor, before it is reset. +const writeIdleWindows = 3 + +// maxTimeout bounds a deadline length so that now + timeout cannot overflow an +// int64 of nanoseconds (a timeout of math.MaxInt64, the common way to write "never", +// made every deadline due at once: review r7 F3): a hundred years. +const maxTimeout = 100 * 365 * 24 * time.Hour + +// clampTimeout is d, or maxTimeout if it is longer. +func clampTimeout(d time.Duration) time.Duration { + if d > maxTimeout { + return maxTimeout + } + return d +} + +// stall is the record of a stream that holds buffered bytes. +type stall struct { + armedAt int64 // when the current window began (wall clock ns) + held int // bytes the stream held when the window began + pending int // bytes it holds now + released int // bytes of its own released in the window + bank int // surplus released in earlier windows (and the seed), carried over + idle int // consecutive windows held below its share on a connection above the floor + connAt int64 // Manager.connReleased when the window began + connHeld int64 // Manager.outboundHeld when the window began +} + +// SetWriteTimeout sets the write deadline (the server's WriteTimeout): the length +// of the window a stalled stream is judged over. 0 or less turns it off. Set once, +// before the first frame. +func (m *Manager) SetWriteTimeout(d time.Duration) { m.wrTimeout = clampTimeout(d) } + +// WriteTimeout is the window SetWriteTimeout set. +func (m *Manager) WriteTimeout() time.Duration { return m.wrTimeout } + +// NextWriteDeadline is the earliest end of a stalled stream's window, or 0 if +// none is stalled (or the hint is stale after a stall ended: ExpiredWrites +// settles it). One atomic load; the engines' sweep reads it for every h2 conn. +func (m *Manager) NextWriteDeadline() int64 { return m.hints.WrDL.Load() } + +// noteBuffered records that n bytes were just added to the buffer of s, already +// charged to outboundHeld: the stream is stalled from here, and its window begins +// if it was not already, seeded with the bytes it had in flight (sent, not yet +// granted back: the peer's initial window less what is left of it), at most +// writeBankWindows floors, which cover the time the client needs to consume them +// before its first update. Called with the stream's lock held, on the event loop or +// on a pool goroutine. +func (m *Manager) noteBuffered(s *Stream, n int) { + if m.wrTimeout <= 0 || n <= 0 { + return + } + id := s.ID + m.stallMu.Lock() + st := m.stalls[id] + if st == nil { + now := time.Now().UnixNano() // read only when a window begins, not for every append to a stalled stream + if m.stalls == nil { + m.stalls = make(map[uint32]*stall) + } + seed := min(max(int(m.initWin.Load())-int(s.windowSize.Load()), 0), writeBankWindows*writeMinRelease) + st = &stall{armedAt: now, bank: seed, connAt: m.connReleased.Load(), connHeld: max(m.outboundHeld.Load(), int64(n))} + m.stalls[id] = st + m.stallN.Store(int32(len(m.stalls))) + m.lowerWrDL(now + int64(m.wrTimeout)) + } + st.pending += n + if st.held == 0 { // the window's first bytes + st.held = st.pending + } + m.stallMu.Unlock() +} + +// lowerWrDL lowers the hint to at, if at is earlier than what it holds. +func (m *Manager) lowerWrDL(at int64) { + for { + cur := m.hints.WrDL.Load() + if cur != 0 && cur <= at { + return + } + if m.hints.WrDL.CompareAndSwap(cur, at) { + return + } + } +} + +// noteReleased records that n bytes of the stream's buffer were just sent +// (FlushOutbound), with the stream's lock held: the stream's window is credited, +// and the record ends with the buffer. +func (m *Manager) noteReleased(id uint32, n int) { + if m.wrTimeout <= 0 || n <= 0 { + return + } + m.stallMu.Lock() + if st := m.stalls[id]; st != nil { + st.released += n + st.pending -= n + if st.pending <= 0 { + delete(m.stalls, id) + m.stallN.Store(int32(len(m.stalls))) + } + } + m.stallMu.Unlock() +} + +// dropStall ends the record of stream id, whose buffer was dropped or whose +// stream has left the connection. A load of stallN when no stream is stalled. +func (m *Manager) dropStall(id uint32) { + if m.stallN.Load() == 0 { + return + } + m.stallMu.Lock() + if _, ok := m.stalls[id]; ok { + delete(m.stalls, id) + m.stallN.Store(int32(len(m.stalls))) + } + m.stallMu.Unlock() +} + +// dropAllStalls ends every record (Close). +func (m *Manager) dropAllStalls() { + m.stallMu.Lock() + clear(m.stalls) + m.stallN.Store(0) + m.stallMu.Unlock() +} + +// stalledStreams is how many streams hold a record. For tests and the leak check. +func (m *Manager) stalledStreams() int { + m.stallMu.Lock() + defer m.stallMu.Unlock() + return len(m.stalls) +} + +// creditConn counts n bytes of DATA the connection released (ReserveSendWindow), +// for the judgement of a stream starved by the connection window. Only while a +// stream is stalled: one atomic load otherwise. +func (m *Manager) creditConn(n int32) { + if n > 0 && m.stallN.Load() != 0 { + m.connReleased.Add(int64(n)) + } +} + +// ExpiredWrites appends to out, up to its capacity, the IDs of the streams whose +// window has ended and who did not release their floor in it, and renews the +// window of those that did. It settles the hint to the earliest end of the windows +// that remain. Called by the loop thread under H2State.mu when NextWriteDeadline +// says one is due; it takes no lock another path holds while it takes stallMu: the +// records due are copied out under stallMu alone, the streams' windows are read +// under the manager's read lock alone, and the verdicts are applied under stallMu +// alone again. A record whose stream is gone from the connection ends here (the +// removal paths end theirs; this is the net). If out is full the hint is left due, +// so the next sweep takes the rest. +func (m *Manager) ExpiredWrites(now int64, out []uint32) []uint32 { + wt := int64(m.wrTimeout) + if wt <= 0 { + m.hints.WrDL.Store(0) + return out + } + var idBuf [32]uint32 + var stBuf [32]*stall + ids, sts := idBuf[:0], stBuf[:0] + m.stallMu.Lock() + for id, st := range m.stalls { + if st.armedAt+wt <= now { + ids, sts = append(ids, id), append(sts, st) + } + } + m.stallMu.Unlock() + + // The streams' windows, under the manager's lock alone. + var viewBuf [32]uint8 + views := viewBuf[:0] + const ( + viewGone = 1 << iota // the stream is not in the map + viewOwn // its own send window has room + ) + if len(ids) > 0 { + m.mu.RLock() + for _, id := range ids { + var v uint8 + if s, ok := m.streams[id]; !ok { + v = viewGone + } else if s.windowSize.Load() > 0 { + v = viewOwn + } + views = append(views, v) + } + m.mu.RUnlock() + } + connShut := atomic.LoadInt32(&m.connectionWindow) <= 0 + + full := false + m.stallMu.Lock() + for i, id := range ids { + st := m.stalls[id] + // The record judged is the one found due: one that ended and began again + // while the windows were read (a pool goroutine flushed the buffer whole and + // buffered a new tail) is another window, not due (review r7 F2). + if st == nil || st != sts[i] || st.armedAt+wt > now { + continue + } + if views[i]&viewGone != 0 { + delete(m.stalls, id) + continue + } + if !m.holdsLocked(st, views[i]&viewOwn != 0 && connShut, now) { + if len(out) == cap(out) { + full = true + continue + } + out = append(out, id) + } + } + var next int64 + for _, st := range m.stalls { + if at := st.armedAt + wt; next == 0 || at < next { + next = at + } + } + if full { + next = now // due again at the next sweep + } + m.stallN.Store(int32(len(m.stalls))) + m.hints.WrDL.Store(next) // under stallMu, as noteBuffered lowers it: an arm is not lost + m.stallMu.Unlock() + return out +} + +// holdsLocked is the verdict on the stalled stream st whose window has ended at +// now: it reports whether the stream is held, and if so begins its next window. +// starved says the stream's own window has room and the connection's is shut. +// m.stallMu must be held. The rule, tail grace included, is in the file's comment. +func (m *Manager) holdsLocked(st *stall, starved bool, now int64) bool { + connRel := m.connReleased.Load() - st.connAt + connFloor := min(int64(writeMinRelease), st.connHeld) + moving := connRel >= connFloor // the connection as a whole released its floor + + hold := false + switch { + case starved: + // Waiting for the connection window, which is no fault of the stream's own: + // its credit is the connection's. + hold = moving + st.idle = 0 + default: + total := st.released + st.bank + floor := min(writeMinRelease, st.held) + switch { + case total >= floor: + hold = true + charge := floor + if st.released < floor { + // Held on the bank alone: the window pays a whole floor, not the + // (smaller) rest, or a stream whose rest is under a floor and that is + // never read would spend a few bytes a window of its bank and live + // for tens of thousands of windows (r9 F-H). + charge = writeMinRelease + } + st.bank = min(max(total-charge, 0), writeBankWindows*writeMinRelease) + st.idle = 0 + case moving: + share := int(connFloor) / (2 * max(len(m.stalls), 1)) + if st.released > 0 && st.released >= share { + hold = true + st.idle = 0 + } else if st.idle++; st.idle <= writeIdleWindows { + hold = true + } + st.bank = 0 + case (st.held < writeMinRelease && st.released*2 >= floor) || st.connHeld < writeMinRelease: + // A tail. What is held (by the stream, or by the whole connection) is under + // one floor, and "all of it in one window" is brittle: the grant for the last + // bytes of a response waits behind its siblings' bytes in the socket and can + // land just after the window (r8 F-A: reset 8 to 16 bytes before the end of + // a download that took 21 s). For a stream whose own rest is small, half of + // it released is a grace window; when the whole connection holds under one + // floor, the end of the transfer is on the wire and the drain of its + // backlog can take more than one window (r9 F-A2: 9 of 20 parallel downloads + // reset 2 to 8 bytes from the end), so the connection's tail is held without + // that proof. Both are inside the idle bound (writeIdleWindows). + if st.idle++; st.idle <= writeIdleWindows { + hold = true + } + st.bank = 0 + } + } + if hold { + st.armedAt, st.held, st.released = now, st.pending, 0 + st.connAt, st.connHeld = m.connReleased.Load(), max(m.outboundHeld.Load(), int64(st.pending)) + } + return hold +} + +// expireWrite resets the stalled stream id: the bytes it holds are dropped +// (refunded to the budget), reset is called with the stream so that the +// RST_STREAM is queued behind the frames the stream already queued, and the +// stream leaves the connection, a pool handler still running on it retiring it as +// for a reset from the peer (takeLocked). Everything happens under m.mu, while the +// stream is in the map, as resetStream does (celeris#951). It reports whether the +// stream was there and was reset. A stream that is already cancelled (a pool +// handler's own reset at #906's bound, or the peer's) is taken out without a +// second RST_STREAM. +func (m *Manager) expireWrite(id uint32, reset func(*Stream)) bool { + m.mu.Lock() + s, ok := m.streams[id] + if !ok { + m.mu.Unlock() + m.dropStall(id) + return false + } + cancelled := s.IsCancelled() + s.mu.Lock() + buf := s.OutboundBuffer + if buf == nil || buf.Len() == 0 { + // Nothing is held any more: the response went out whole between the verdict + // and here (a pool goroutine flushed it). A reset now would follow its + // END_STREAM (review r7 F2). + s.mu.Unlock() + m.mu.Unlock() + m.dropStall(id) + return false + } + m.refundOutbound(buf.Len()) + buf.Reset() + s.OutboundEndStream = false + s.outboundAbandoned = false + s.mu.Unlock() + if !cancelled && reset != nil { + reset(s) + } + s.Cancel() + release := m.takeLocked(id, s) + m.mu.Unlock() + m.afterTake(id, s, release) + return !cancelled +} + +// ExpireWrite resets stream id, whose write window passed without its floor, with +// RST_STREAM(INTERNAL_ERROR) behind the frames it has queued, and remembers the +// identifier as a reset one, so that the client's late frames for it (a +// WINDOW_UPDATE, a RST_STREAM, a PRIORITY) are ignored as for any stream the +// server reset. No response is written: the headers and part of the body are out. +// Not charged to the refusal budgets: it is the server's decision. Called by the +// loop thread under H2State.mu. +func (p *Processor) ExpireWrite(id uint32) { + if p.manager.expireWrite(id, p.resetStalled) { + p.resetIDs.mark(id) + } +} + +// resetStalled queues the RST_STREAM of an expired write behind the stream's +// frames through the connection's writer when it can (Resetter), and writes it +// straight to the connection otherwise (a test recorder). +func (p *Processor) resetStalled(s *Stream) { + switch w := p.connWriter.(type) { + case Resetter: + w.ResetStream(s, http2.ErrCodeInternal) // cancels the stream as well + case nil: + default: + _ = w.WriteRSTStreamPriority(s.ID, http2.ErrCodeInternal) + } +} diff --git a/internal/protocol/h2/stream/write_deadline_889_test.go b/internal/protocol/h2/stream/write_deadline_889_test.go new file mode 100644 index 000000000..3d5a1bb6e --- /dev/null +++ b/internal/protocol/h2/stream/write_deadline_889_test.go @@ -0,0 +1,583 @@ +package stream + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "golang.org/x/net/http2" +) + +// The write deadline's books and verdicts (celeris#889, D2): the records, the +// floor, the conn credit, the removal paths, the reset. + +// rstRecorder889 is a connection writer that implements Resetter and keeps what it +// was asked to reset. +type rstRecorder889 struct { + *testResponseWriter + mu sync.Mutex + rst []rstRecord889 +} + +type rstRecord889 struct { + id uint32 + code http2.ErrCode +} + +func (r *rstRecorder889) ResetStream(s *Stream, code http2.ErrCode) { + r.mu.Lock() + r.rst = append(r.rst, rstRecord889{s.ID, code}) + r.mu.Unlock() + s.Cancel() +} + +func (r *rstRecorder889) count() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.rst) +} + +const wt889 = time.Second + +// wdSetup889 is a processor with a write deadline of wt889, an open connection +// window and a recorder for the resets. +func wdSetup889(t testing.TB) (*Processor, *Manager, *rstRecorder889) { + t.Helper() + rec := &rstRecorder889{testResponseWriter: newTestResponseWriter()} + p := NewProcessor(HandlerFunc(func(context.Context, *Stream) error { return nil }), newTestFrameWriter(), rec) + m := p.GetManager() + m.SetWriteTimeout(wt889) + m.UpdateConnectionWindow(1 << 30) + m.initWin.Store(0) // no seed: a stream's first window is judged on what it releases (the seed has its own tests) + return p, m, rec +} + +// stalled889 makes stream id with a zero window and n bytes buffered. +func stalled889(m *Manager, id uint32, n int) *Stream { + s := m.CreateStream(id) + s.SetState(StateOpen) + s.SetWindowSize(0) + s.BufferOutbound(make([]byte, n), true) + return s +} + +func discard889(uint32, bool, []byte) {} + +func dueAt889() int64 { return time.Now().UnixNano() + int64(wt889) + int64(time.Millisecond) } + +// A record is made when a buffer fills and only with a write deadline, and the hint +// is the end of the first window; sending that fits the window makes none. +func TestStallIsArmedByABufferOnly889(t *testing.T) { + _, m, _ := wdSetup889(t) + // Fits the window: sent, nothing buffered, no record, no hint. + s := m.CreateStream(1) + s.SetState(StateOpen) + if _, _, _, err := m.StreamWrite(s, s.Gen(), make([]byte, 1000), discard889); err != nil { + t.Fatal(err) + } + if m.stalledStreams() != 0 || m.NextWriteDeadline() != 0 { + t.Fatalf("a write that fit the window left %d records, hint %d", m.stalledStreams(), m.NextWriteDeadline()) + } + before := time.Now().UnixNano() + stalled889(m, 3, 100) + after := time.Now().UnixNano() + if m.stalledStreams() != 1 { + t.Fatalf("%d records after a buffer filled, want 1", m.stalledStreams()) + } + if h := m.NextWriteDeadline(); h < before+int64(wt889) || h > after+int64(wt889) { + t.Fatalf("hint %d, want the end of a window of %v that began between %d and %d", h, wt889, before, after) + } + // Without a write deadline there is no record. + _, m2, _ := wdSetup889(t) + m2.SetWriteTimeout(0) + stalled889(m2, 1, 100) + if m2.stalledStreams() != 0 || m2.NextWriteDeadline() != 0 { + t.Fatalf("WriteTimeout off: %d records, hint %d", m2.stalledStreams(), m2.NextWriteDeadline()) + } +} + +// The floor is min(64 KiB, what the stream held when the window began), and a +// window that passes it starts the next. +func TestExpiredWritesFloor889(t *testing.T) { + for _, tc := range []struct { + name string + held int + release int // bytes released in the window + expired bool // the stream is expired at the end of the window + recordIn bool // it still has a record afterwards (not expired, not drained) + }{ + {"nothing-released", 1 << 20, 0, true, true}, + {"floor-1", 1 << 20, 65535, true, true}, + {"floor", 1 << 20, 65536, false, true}, + {"above-floor", 1 << 20, 200000, false, true}, + {"small-tail-all-released", 10000, 10000, false, false}, // the buffer empties: the record ends + // A tail (held under one floor) that went out all but a little is not the + // verdict: half of it is a grace window (r8 F-A; it used to be reset, "all of it + // in one window" was the floor). The grace is bounded: TestWriteTailGraceIsBounded889. + {"small-tail-nearly-all", 10000, 9999, false, true}, + {"small-tail-half", 10000, 5000, false, true}, + // Less than half went out: this was the verdict in r8 and is held now. The + // connection holds under one floor, so the end of the transfer is on the wire and + // the drain of the backlog behind it can take more than a window (r9 F-A2: 9 of 20 + // parallel downloads were reset 2 to 8 bytes from the end). Still bounded by the + // idle windows; a client that never reads is let go (TestWriteUnreadSmallRemainderBounds889). + {"small-tail-under-half", 10000, 4999, false, true}, + } { + t.Run(tc.name, func(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, tc.held) + if tc.release > 0 { + s.SetWindowSize(int32(tc.release)) + m.FlushOutbound(s, discard889) + } + out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 8)) + if got := len(out) == 1; got != tc.expired { + t.Fatalf("expired = %v (%v), want %v", got, out, tc.expired) + } + if got := m.stalledStreams() == 1; got != tc.recordIn { + t.Fatalf("record kept = %v, want %v", got, tc.recordIn) + } + }) + } +} + +// Not before the window ends; a stream that passed a window is judged again a +// window later, and the verdict of every window is its own (no ceiling, no credit +// carried over). +func TestExpiredWritesWindows889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 1<<22) + start := time.Now().UnixNano() + if out := m.ExpiredWrites(start+int64(wt889)/2, make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("expired %v half a window in", out) + } + now := start + int64(wt889) + int64(time.Millisecond) + for w := 0; w < 3; w++ { // three windows, each with its floor + s.SetWindowSize(65536) + m.FlushOutbound(s, discard889) + if out := m.ExpiredWrites(now, make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("window %d: expired %v although it released its floor", w, out) + } + if h := m.NextWriteDeadline(); h != now+int64(wt889) { + t.Fatalf("window %d: hint %d, want the end of the next window %d", w, h, now+int64(wt889)) + } + now += int64(wt889) + int64(time.Millisecond) + } + // The fourth window releases nothing, and the credit of the third does not count. + if out := m.ExpiredWrites(now, make([]uint32, 0, 4)); len(out) != 1 || out[0] != 1 { + t.Fatalf("a window with no release: expired %v, want [1]", out) + } +} + +// A stream whose own window has room but whose bytes wait for the connection's is +// credited with what the connection released; one whose own window is shut is +// judged by its own releases, whatever the connection released. +func TestExpiredWritesConnStarved889(t *testing.T) { + setup := func(t *testing.T) (*Manager, *Stream, *Stream) { + _, m, _ := wdSetup889(t) + atomic.StoreInt32(&m.connectionWindow, 0) // shut + a := m.CreateStream(1) + a.SetState(StateOpen) + a.SetWindowSize(1 << 20) + a.BufferOutbound(make([]byte, 1<<20), true) + b := m.CreateStream(3) + b.SetState(StateOpen) + b.SetWindowSize(1 << 20) + b.BufferOutbound(make([]byte, 1<<20), true) + return m, a, b + } + t.Run("the-conn-released-the-floor", func(t *testing.T) { + m, a, _ := setup(t) + m.UpdateConnectionWindow(70000) + m.FlushOutbound(a, discard889) // a sibling takes all of it: b released nothing itself + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 8)); len(out) != 0 { + t.Fatalf("expired %v although the conn released 70000 bytes in the window", out) + } + }) + t.Run("starved-by-the-conn-window-is-held-as-long-as-the-conn-moves", func(t *testing.T) { + // Both streams have room in their own windows and the conn window is shut: + // the sibling takes every grant, the other gets nothing of its own for far + // more than writeIdleWindows windows, and is held all the time (it waits + // for the connection, which is no fault of its own). + m, a, _ := setup(t) + now := dueAt889() + for w := 0; w < 3*writeIdleWindows; w++ { + m.UpdateConnectionWindow(70000) + m.FlushOutbound(a, discard889) + if out := m.ExpiredWrites(now, make([]uint32, 0, 8)); len(out) != 0 { + t.Fatalf("window %d: expired %v while the conn released its floor", w, out) + } + now += int64(wt889) + int64(time.Millisecond) + } + }) + t.Run("the-conn-released-a-dribble", func(t *testing.T) { + m, a, _ := setup(t) + m.UpdateConnectionWindow(1) + m.FlushOutbound(a, discard889) + out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 8)) + if len(out) != 2 { + t.Fatalf("expired %v, want both starved streams", out) + } + }) + t.Run("own-window-shut-is-judged-alone", func(t *testing.T) { + // Stream 3's own window is shut while the conn released 70000 bytes, all to + // stream 1. It is not credited with them as a starved stream is: it takes + // nothing of its own, so it is held for writeIdleWindows windows (it may be + // queued behind its sibling) and reset at the next. + m, a, b := setup(t) + b.SetWindowSize(0) + now := dueAt889() + for w := 0; w <= writeIdleWindows+1; w++ { + m.UpdateConnectionWindow(70000) + m.FlushOutbound(a, discard889) + out := m.ExpiredWrites(now, make([]uint32, 0, 8)) + want := 0 + if w == writeIdleWindows { + want = 1 // the (writeIdleWindows+1)th window with nothing of its own + } + if len(out) != want || (want == 1 && out[0] != 3) { + t.Fatalf("window %d: expired %v, want %d (only stream 3, in window %d)", w, out, want, writeIdleWindows) + } + if want == 1 { + return + } + now += int64(wt889) + int64(time.Millisecond) + } + }) + t.Run("releases-outside-a-flush-count", func(t *testing.T) { + // A sibling that sends without buffering (a pool handler's direct send) + // also takes the conn window: ReserveSendWindow counts it. + m, _, _ := setup(t) + c := m.CreateStream(5) + c.SetState(StateOpen) + c.SetWindowSize(1 << 20) + m.UpdateConnectionWindow(70000) + if g := m.ReserveSendWindow(c, 70000); g != 70000 { + t.Fatalf("reserved %d", g) + } + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 8)); len(out) != 0 { + t.Fatalf("expired %v although a sibling's direct send took 70000 bytes of the conn window", out) + } + }) +} + +// A batch that fills leaves the hint due and the rest for the next sweep. +func TestExpiredWritesBatch889(t *testing.T) { + _, m, _ := wdSetup889(t) + for i := 0; i < 40; i++ { + stalled889(m, uint32(1+2*i), 100) + time.Sleep(2 * time.Millisecond) // a slow runner (the coverage build): the records are not armed in one instant + } + start := time.Now().UnixNano() // after the records: every window is judged for all of them (CI at 7e7f5ce: 2 of 40 armed after start+1ms missed window 1) + // The connection holds 4000 bytes, under one floor: its tail is held for the idle + // windows (r9 F-A2) rather than reset at the first one, and the batch limit is shown + // in the window after, when 40 are due. + for w := 1; w <= writeIdleWindows; w++ { + if len(windowEnd889(m, start, w)) != 0 { + t.Fatalf("window %d: expired, a tail is held for %d windows", w, writeIdleWindows) + } + } + now := start + int64(writeIdleWindows+1)*(int64(wt889)+int64(time.Millisecond)) + out := m.ExpiredWrites(now, make([]uint32, 0, 32)) + if len(out) != 32 { + t.Fatalf("first batch %d, want 32", len(out)) + } + if h := m.NextWriteDeadline(); h != now { + t.Fatalf("hint %d after a full batch, want due (%d)", h, now) + } + for _, id := range out { + m.DeleteStream(id) + } + if out2 := m.ExpiredWrites(now, make([]uint32, 0, 32)); len(out2) != 8 { + t.Fatalf("second batch %d, want the 8 left", len(out2)) + } +} + +// A stale hint (the stall ended) costs one look, which settles it. +func TestExpiredWritesStaleHint889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 100) + s.SetWindowSize(1000) + m.FlushOutbound(s, discard889) // drained: the record is gone, the hint stays + if m.NextWriteDeadline() == 0 { + t.Fatal("setup: no hint") + } + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("expired %v", out) + } + if m.NextWriteDeadline() != 0 { + t.Fatalf("hint %d after the look, want 0", m.NextWriteDeadline()) + } +} + +// Every path that ends a stalled stream ends its record. A record that outlived +// its stream would be a leak of the connection's life, and the net (a stream +// that is gone from the map) must not be what these rely on. +func TestStallRecordEndsWithEveryRemovalPath889(t *testing.T) { + for _, tc := range []struct { + name string + end func(t *testing.T, p *Processor, m *Manager, s *Stream) + }{ + {"DeleteStream", func(_ *testing.T, _ *Processor, m *Manager, s *Stream) { m.DeleteStream(s.ID) }}, + {"peer-RST_STREAM", func(t *testing.T, p *Processor, _ *Manager, s *Stream) { + if err := p.handleRSTStream(&http2.RSTStreamFrame{FrameHeader: http2.FrameHeader{StreamID: s.ID}}); err != nil { + t.Fatal(err) + } + }}, + {"DiscardOutbound", func(_ *testing.T, _ *Processor, m *Manager, s *Stream) { m.DiscardOutbound(s) }}, + {"flushed-whole", func(_ *testing.T, _ *Processor, m *Manager, s *Stream) { + s.SetWindowSize(1 << 20) + m.FlushOutbound(s, discard889) + }}, + {"retire", func(_ *testing.T, _ *Processor, m *Manager, s *Stream) { + s.flags.Or(flagAsyncRunning) + m.retire(s) + }}, + {"GOAWAY", func(t *testing.T, p *Processor, _ *Manager, _ *Stream) { + if err := p.handleGoAway(&http2.GoAwayFrame{LastStreamID: 0}); err != nil { + t.Fatal(err) + } + }}, + {"RemoveStreamFromMap", func(_ *testing.T, _ *Processor, m *Manager, s *Stream) { m.RemoveStreamFromMap(s.ID) }}, + {"Close", func(_ *testing.T, _ *Processor, m *Manager, _ *Stream) { m.Close() }}, + {"ExpireWrite", func(_ *testing.T, p *Processor, _ *Manager, s *Stream) { p.ExpireWrite(s.ID) }}, + } { + t.Run(tc.name, func(t *testing.T) { + p, m, _ := wdSetup889(t) + s := stalled889(m, 1, 5000) + stalled889(m, 3, 5000) // another stays only if the path is per stream + if m.stalledStreams() != 2 { + t.Fatalf("setup: %d records", m.stalledStreams()) + } + tc.end(t, p, m, s) + left := m.stalledStreams() + switch tc.name { + case "GOAWAY", "Close": + if left != 0 { + t.Fatalf("%d records left, want 0 (every stream above the last one is gone)", left) + } + default: + if left != 1 { + t.Fatalf("%d records left, want 1 (the other stream's)", left) + } + } + }) + } +} + +// ExpireWrite resets the stream once, with INTERNAL_ERROR, drops its bytes, takes +// it out of the connection and remembers the identifier; the second call, a stream +// the peer or a pool handler had already reset, writes nothing. +func TestExpireWriteResetsOnce889(t *testing.T) { + p, m, rec := wdSetup889(t) + s := stalled889(m, 1, 5000) + p.ExpireWrite(1) + if rec.count() != 1 || rec.rst[0] != (rstRecord889{1, http2.ErrCodeInternal}) { + t.Fatalf("resets %+v, want one RST_STREAM(INTERNAL_ERROR) for stream 1", rec.rst) + } + if _, ok := m.GetStream(1); ok { + t.Fatal("the stream is still in the connection") + } + if m.OutboundHeld() != 0 { + t.Fatalf("%d bytes still charged to the outbound budget", m.OutboundHeld()) + } + if !p.resetIDs.has(1) { + t.Fatal("the identifier is not remembered as reset: the client's late frames would be errors") + } + _ = s + p.ExpireWrite(1) + if rec.count() != 1 { + t.Fatalf("a second expiry wrote %d more resets", rec.count()-1) + } + // A stream reset by someone else first (a pool handler's own reset at #906's + // bound cancels it): taken out, no second RST_STREAM. + s3 := stalled889(m, 3, 5000) + s3.Cancel() + p.ExpireWrite(3) + if rec.count() != 1 { + t.Fatalf("a stream that was cancelled already got a second reset: %+v", rec.rst) + } + if _, ok := m.GetStream(3); ok { + t.Fatal("the cancelled stream stayed in the connection") + } + if m.stalledStreams() != 0 { + t.Fatalf("%d records left", m.stalledStreams()) + } +} + +// A pool handler blocked in AwaitSendWindow under #906's absolute bound, on a +// stream D2 expires, wakes with the stream's cancellation, not with the deadline: +// it does not send a second RST_STREAM. +func TestExpireWriteWakesAPoolHandlerAsReset889(t *testing.T) { + p, m, rec := wdSetup889(t) + s := stalled889(m, 1, 5000) + res := make(chan error, 1) + go func() { res <- s.AwaitSendWindow(time.Now().Add(time.Minute)) }() + for m.sendWindowWaiters.Load() == 0 { + time.Sleep(time.Millisecond) + } + p.ExpireWrite(1) + select { + case err := <-res: + if !errors.Is(err, context.Canceled) { + t.Fatalf("the waiter woke with %v, want context.Canceled (a reset), not the deadline", err) + } + case <-time.After(5 * time.Second): + t.Fatal("the waiter was not woken by the expiry") + } + if rec.count() != 1 { + t.Fatalf("%d resets", rec.count()) + } +} + +// With a write deadline a conn whose only business is a stalled stream is +// exempt from the conn clocks (D2 resets the stream); without one, or with a +// pool handler waiting in AwaitSendWindow, it is not. +func TestServerOwesWithAWriteDeadline889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 5000) + s.SetState(StateHalfClosedRemote) + if !m.ServerOwes() { + t.Fatal("a stalled stream does not exempt the conn although D2 bounds it") + } + m.sendWindowWaiters.Add(1) + if m.ServerOwes() { + t.Fatal("a pool handler waiting for the window exempts the conn: its bound is #906's, and the conn clock stays") + } + m.sendWindowWaiters.Add(-1) + m.SetWriteTimeout(0) + if m.ServerOwes() { + t.Fatal("without a write deadline a stalled stream exempts the conn (the old rule)") + } +} + +// The common path pays nothing: reserving window allocates nothing and touches no +// record while no stream is stalled. +func TestWriteDeadlineCommonPathIsFree889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := m.CreateStream(1) + s.SetState(StateOpen) + s.SetWindowSize(1 << 30) + if a := testing.AllocsPerRun(100, func() { m.ReserveSendWindow(s, 1000) }); a != 0 { + t.Fatalf("ReserveSendWindow allocated %.0f times", a) + } + if m.connReleased.Load() != 0 { + t.Fatal("the conn's releases are counted with no stream stalled") + } + stalled889(m, 3, 100) + m.ReserveSendWindow(s, 1000) + if m.connReleased.Load() != 1000 { + t.Fatalf("conn released %d, want 1000 while a stream is stalled", m.connReleased.Load()) + } +} + +// BenchmarkOneByteGrantWriteDeadline889 is BenchmarkOneByteGrant911 with the write +// deadline on (off, it is that benchmark): the price of a stalled stream's flush, +// which takes the books' lock once to credit what it released. +func BenchmarkOneByteGrantWriteDeadline889(b *testing.B) { + for _, on := range []bool{false, true} { + name := "off" + if on { + name = "on" + } + b.Run(name, func(b *testing.B) { + fw := newTestFrameWriter() + p := NewProcessor(HandlerFunc(func(context.Context, *Stream) error { return nil }), fw, newTestResponseWriter()) + m := p.GetManager() + if on { + m.SetWriteTimeout(time.Minute) + } + m.UpdateConnectionWindow(1 << 30) + s := m.CreateStream(1) + s.SetState(StateOpen) + s.SetWindowSize(0) + body := make([]byte, 1<<20) + s.BufferOutbound(body, true) + wu := makeWindowUpdateFrame(&testing.T{}, 1, 1) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if s.OutboundBuffer.Len() < 2 { + b.StopTimer() + s.BufferOutbound(body, true) + b.StartTimer() + } + if err := p.ProcessFrame(context.Background(), wu); err != nil { + b.Fatal(err) + } + } + }) + } +} + +// What a stalled stream's floor is made of, through the frames a client sends: a +// WINDOW_UPDATE that lets 70000 bytes out and a SETTINGS_INITIAL_WINDOW_SIZE raise +// that does are progress; a hundred WINDOW_UPDATEs of one byte are not. +func TestWriteProgressThroughFramesIsBytesReleased889(t *testing.T) { + ctx := context.Background() + mk := func(t *testing.T) (*Processor, *Manager) { + p, m, _ := wdSetup889(t) + stalled889(m, 1, 1<<20) + return p, m + } + t.Run("window-update-of-70000", func(t *testing.T) { + p, m := mk(t) + if err := p.ProcessFrame(ctx, makeWindowUpdateFrame(t, 1, 70000)); err != nil { + t.Fatal(err) + } + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("expired %v after a WINDOW_UPDATE that released 70000 bytes", out) + } + }) + t.Run("settings-initial-window-raise", func(t *testing.T) { + p, m := mk(t) + f := makeSettingsFrame(t, http2.Setting{ID: http2.SettingInitialWindowSize, Val: 65535 + 70000}) + if err := p.ProcessFrame(ctx, f); err != nil { + t.Fatal(err) + } + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("expired %v after a SETTINGS raise that released 70000 bytes", out) + } + }) + t.Run("a-hundred-one-byte-updates", func(t *testing.T) { + p, m := mk(t) + for i := 0; i < 100; i++ { + if err := p.ProcessFrame(ctx, makeWindowUpdateFrame(t, 1, 1)); err != nil { + t.Fatal(err) + } + } + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 4)); len(out) != 1 { + t.Fatalf("expired %v, want the dribbled stream", out) + } + }) + t.Run("a-ping-and-a-conn-window-update-with-no-stream-in-need", func(t *testing.T) { + p, m := mk(t) + for i := 0; i < 50; i++ { + if err := p.ProcessFrame(ctx, makeWindowUpdateFrame(t, 0, 1)); err != nil { + t.Fatal(err) + } + } + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 4)); len(out) != 1 { + t.Fatalf("expired %v: a conn WINDOW_UPDATE does not move a stream whose own window is shut", out) + } + }) +} + +// A pool handler's direct send (nothing buffered) makes no record: only a buffer does. +func TestPoolDirectSendLeavesNoRecord889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := m.CreateStream(1) + s.SetState(StateOpen) + s.flags.Or(flagAsyncRunning) + s.SetWindowSize(1 << 20) + n, _, onPool, err := m.StreamWrite(s, s.Gen(), make([]byte, 5000), discard889) + if err != nil || !onPool || n != 5000 { + t.Fatalf("StreamWrite = %d, onPool %v, %v", n, onPool, err) + } + if m.stalledStreams() != 0 || m.NextWriteDeadline() != 0 { + t.Fatalf("a direct send left %d records, hint %d", m.stalledStreams(), m.NextWriteDeadline()) + } +} diff --git a/internal/protocol/h2/stream/write_rate_889_test.go b/internal/protocol/h2/stream/write_rate_889_test.go new file mode 100644 index 000000000..9b26ebbee --- /dev/null +++ b/internal/protocol/h2/stream/write_rate_889_test.go @@ -0,0 +1,737 @@ +package stream + +import ( + "strconv" + "sync/atomic" + "testing" + "time" +) + +// The refinements of the minimum rate that keep it from cutting a slow reader that +// is moving (celeris#889, D2, review r7): the carry-over of surplus releases and the +// seed of a stream's first window (F1), the connection's rate and the bound on a +// stream that takes less than its share (F6), the record judged is the one found due +// (F2), deadlines that cannot overflow (F3), the floor that follows what the stream +// holds (F4). + +// window889 runs one window's end for the manager at the instant after n windows +// from the start, returning the streams expired. +func windowEnd889(m *Manager, start int64, n int) []uint32 { + return m.ExpiredWrites(start+int64(n)*(int64(wt889)+int64(time.Millisecond)), make([]uint32, 0, 32)) +} + +// Surplus released in one window carries to the next ones, up to writeBankWindows +// windows' worth: a client that updates every few windows survives, and one big +// release does not hold a stream longer than the bank. +func TestWriteSurplusCarriesOverUpToTheBank889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 8<<20) + start := time.Now().UnixNano() + s.SetWindowSize(3 << 20) // 3 MiB at once, in window 1 + m.FlushOutbound(s, discard889) + // Window 1 is held on the release, the bank is capped at writeBankWindows floors: + // it covers writeBankWindows more windows, and the one after is the verdict. + for w := 1; w <= 1+writeBankWindows; w++ { + if out := windowEnd889(m, start, w); len(out) != 0 { + t.Fatalf("window %d: expired %v, want it held by the carried surplus", w, out) + } + } + if out := windowEnd889(m, start, 2+writeBankWindows); len(out) != 1 { + t.Fatalf("window %d: expired %v, want the stream (the bank is spent)", 2+writeBankWindows, out) + } +} + +// A release of twice the floor every other window (or four times every fourth) is a +// rate of one floor per window: survives for ever; half of that is reset once the +// bank is spent. +func TestWriteAverageRateIsTheGuarantee889(t *testing.T) { + for _, tc := range []struct { + name string + every int // a release every `every` windows + release int + holds bool + }{ + {"floor-on-average", 2, 2 * 65536, true}, + {"floor-on-average-sparse", 4, 4 * 65536, true}, + {"half-the-floor", 2, 65536, false}, + } { + t.Run(tc.name, func(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(1 << 20) // the stream stalls with a megabyte in flight: a full seed + s := stalled889(m, 1, 64<<20) + start := time.Now().UnixNano() + expiredAt := 0 + for w := 1; w <= 200 && expiredAt == 0; w++ { + if (w-1)%tc.every == tc.every-1 { + s.SetWindowSize(int32(tc.release)) + m.FlushOutbound(s, discard889) + } + if out := windowEnd889(m, start, w); len(out) != 0 { + expiredAt = w + } + } + if tc.holds && expiredAt != 0 { + t.Fatalf("a stream that releases the floor on average was reset in window %d", expiredAt) + } + if !tc.holds && expiredAt == 0 { + t.Fatal("a stream that releases less than the floor on average was never reset") + } + }) + } +} + +// A stream's first window is covered by the bytes it had in flight when it stalled +// (the peer's initial window less what is left of it), up to the bank: a client with +// a large window and a slow first update is not reset at the first judgement. +func TestWriteSeedCoversTheFirstWindow889(t *testing.T) { + for _, tc := range []struct { + name string + initWin uint32 + windows int // windows held with nothing released + wantHeld int + }{ + {"1MiB-window-4-floors-of-cover", 1 << 20, 0, writeBankWindows}, + {"default-window-65535-is-not-a-floor", 65535, 0, 0}, + {"zero-in-flight", 0, 0, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(tc.initWin) + stalled889(m, 1, 8<<20) // window 0: the stream sent initWin bytes + start := time.Now().UnixNano() + held := 0 + for w := 1; w <= 10; w++ { + if out := windowEnd889(m, start, w); len(out) != 0 { + break + } + held++ + } + if held != tc.wantHeld { + t.Fatalf("held %d windows with nothing released, want %d", held, tc.wantHeld) + } + }) + } +} + +// A WINDOW_UPDATE(1) dribbler is reset at the first judgement that the seed does not +// cover: with a 64 KiB window the seed (65535) and a few bytes are over the floor +// once, and the second window is the verdict. +func TestWriteDribblerIsResetWithinTwoWindowsOfADefaultWindow889(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(65535) + s := stalled889(m, 1, 4<<20) + start := time.Now().UnixNano() + n := 0 + for w := 1; w <= 5 && n == 0; w++ { + for i := 0; i < 10; i++ { // 10 one-byte grants a window + s.SetWindowSize(1) + m.FlushOutbound(s, discard889) + } + if out := windowEnd889(m, start, w); len(out) != 0 { + n = w + } + } + if n == 0 || n > 2 { + t.Fatalf("the dribbler was reset in window %d, want 1 or 2", n) + } +} + +// On a connection that releases above the floor a stream that takes less than its +// share is held for writeIdleWindows windows and reset at the next; one that takes +// its share is held for as long as the connection moves; when the connection itself +// is below the floor the first window is the verdict. +func TestWriteIdleStreamOnAHealthyConnIsBounded889(t *testing.T) { + setup := func(t *testing.T) (*Manager, *Stream, *Stream) { + _, m, _ := wdSetup889(t) + a := stalled889(m, 1, 32<<20) + b := stalled889(m, 3, 32<<20) + return m, a, b + } + t.Run("zero-release-is-held-K-windows", func(t *testing.T) { + m, a, _ := setup(t) + start := time.Now().UnixNano() + for w := 1; w <= writeIdleWindows+1; w++ { + a.SetWindowSize(70000) + m.FlushOutbound(a, discard889) // the sibling moves the connection above the floor + out := windowEnd889(m, start, w) + if w <= writeIdleWindows && len(out) != 0 { + t.Fatalf("window %d: expired %v, want stream 3 held (it may be queued behind its sibling)", w, out) + } + if w == writeIdleWindows+1 && (len(out) != 1 || out[0] != 3) { + t.Fatalf("window %d: expired %v, want stream 3", w, out) + } + } + }) + t.Run("a-dribble-is-no-share", func(t *testing.T) { + m, a, b := setup(t) + start := time.Now().UnixNano() + for w := 1; w <= writeIdleWindows+1; w++ { + a.SetWindowSize(70000) + m.FlushOutbound(a, discard889) + for i := 0; i < 10; i++ { + b.SetWindowSize(1) + m.FlushOutbound(b, discard889) + } + out := windowEnd889(m, start, w) + if (w <= writeIdleWindows) != (len(out) == 0) { + t.Fatalf("window %d: expired %v", w, out) + } + } + }) + t.Run("its-share-holds-it-for-ever", func(t *testing.T) { + m, a, b := setup(t) + start := time.Now().UnixNano() + for w := 1; w <= 30; w++ { + a.SetWindowSize(40000) + m.FlushOutbound(a, discard889) + b.SetWindowSize(40000) // below the floor, above half of floor/2 + m.FlushOutbound(b, discard889) + if out := windowEnd889(m, start, w); len(out) != 0 { + t.Fatalf("window %d: expired %v although both take 40000 a window and the connection 80000", w, out) + } + } + }) + t.Run("a-slow-connection-is-the-verdict-at-once", func(t *testing.T) { + m, a, b := setup(t) + start := time.Now().UnixNano() + a.SetWindowSize(20000) + m.FlushOutbound(a, discard889) + b.SetWindowSize(20000) + m.FlushOutbound(b, discard889) // 40000 in all: below the floor + if out := windowEnd889(m, start, 1); len(out) != 2 { + t.Fatalf("expired %v, want both: the connection itself did not release its floor", out) + } + }) +} + +// The record a verdict is for is the one found due: one that ended and began again +// while the windows were read is a new window. +func TestExpiredWritesJudgesTheRecordItFoundDue889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 100000) + m.stallMu.Lock() + m.stalls[1].armedAt -= int64(2 * wt889) + m.stallMu.Unlock() + + m.mu.Lock() // hold the read of the streams' windows + done := make(chan []uint32) + go func() { done <- m.ExpiredWrites(time.Now().UnixNano(), make([]uint32, 0, 32)) }() + time.Sleep(50 * time.Millisecond) + // Meanwhile a pool goroutine's flush takes the whole buffer and it buffers a new tail. + s.SetWindowSize(1 << 20) + m.FlushOutbound(s, discard889) + s.SetWindowSize(0) + s.BufferOutbound(make([]byte, 100000), true) // not a tail: the tail grace would hold it + m.mu.Unlock() + if got := <-done; len(got) != 0 { + t.Fatalf("expired %v: the record judged was a new one whose window began microseconds ago", got) + } +} + +// A stream whose buffer went out whole between the verdict and the reset is not +// reset (its END_STREAM is out). +func TestExpireWriteSkipsAStreamWithNothingHeld889(t *testing.T) { + p, m, rec := wdSetup889(t) + s := stalled889(m, 1, 5000) + s.SetWindowSize(1 << 20) + m.FlushOutbound(s, discard889) + p.ExpireWrite(1) + if rec.count() != 0 { + t.Fatalf("reset a stream with nothing held: %+v", rec.rst) + } +} + +// A deadline of math.MaxInt64 (the common way to write "never") must not overflow: +// the window of a stalled stream and the request deadline are far in the future. +func TestHugeTimeoutsDoNotOverflow889(t *testing.T) { + const never = time.Duration(1<<63 - 1) + t.Run("write", func(t *testing.T) { + _, m, _ := wdSetup889(t) + m.SetWriteTimeout(never) + stalled889(m, 1, 5000) + if h := m.NextWriteDeadline(); h <= time.Now().UnixNano() { + t.Fatalf("hint %d is not in the future", h) + } + if out := m.ExpiredWrites(time.Now().UnixNano()+int64(time.Hour), make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("expired %v an hour in", out) + } + }) + t.Run("request", func(t *testing.T) { + m := NewManager() + m.SetRequestTimeout(never) + s := m.CreateStream(1) + s.SetState(StateOpen) + m.ArmRequest(s, time.Now().UnixNano()) + if h := m.NextRequestDeadline(); h <= time.Now().UnixNano() { + t.Fatalf("request deadline hint %d is not in the future", h) + } + if out := m.ExpiredRequests(time.Now().UnixNano()+int64(time.Hour), make([]uint32, 0, 4)); len(out) != 0 { + t.Fatalf("expired %v an hour in", out) + } + }) + t.Run("header-block", func(t *testing.T) { + p, _, _ := wdSetup889(t) + p.SetHeaderBlockTimeout(never) + if p.hdrTimeout <= 0 || p.hdrTimeout > maxTimeout { + t.Fatalf("header block timeout %v", p.hdrTimeout) + } + }) +} + +// The floor of a later window follows what the stream holds then, not what it held +// when its first window began. +func TestWriteFloorFollowsWhatTheStreamHolds889(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 100000) // held 100000: floor 65536 + start := time.Now().UnixNano() + s.SetWindowSize(70000) // 70000 released, 30000 left + m.FlushOutbound(s, discard889) + if out := windowEnd889(m, start, 1); len(out) != 0 { + t.Fatalf("window 1: expired %v", out) + } + s.BufferOutbound(make([]byte, 100000), true) // 130000 held now; the window began with 30000 + s.SetWindowSize(30000) // floor of this window is min(64 KiB, 30000) = 30000 + m.FlushOutbound(s, discard889) + if out := windowEnd889(m, start, 2); len(out) != 0 { + t.Fatalf("window 2: expired %v although it released the 30000 it held when the window began", out) + } +} + +// The tail of a response (what is held is under one floor) is not judged on +// "all of it in one window": the grant for the last bytes can land just after the +// window (r8 F-A). Half of it released is a grace window, inside the idle bound; +// nothing is the verdict, and a tail that goes on missing half of itself is reset +// after writeIdleWindows such windows. +func TestWriteTailGraceIsBounded889(t *testing.T) { + // Nothing released is not the verdict at the first window (r9 F-A2: a connection + // that holds under one floor is draining its backlog, and a window can pass + // with nothing going out); it is the verdict after writeIdleWindows of them. + t.Run("nothing-released-is-held-for-the-idle-windows", func(t *testing.T) { + _, m, _ := wdSetup889(t) + stalled889(m, 1, 1000) + if w := firstReset889(m, 1, time.Now().UnixNano(), 10, nil); w != writeIdleWindows+1 { + t.Fatalf("reset in window %d, want %d (held for %d)", w, writeIdleWindows+1, writeIdleWindows) + } + }) + t.Run("half-every-window-is-bounded", func(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 1000) + start := time.Now().UnixNano() + for w := 1; w <= writeIdleWindows+1; w++ { + s.SetWindowSize(500) // half of the 1000 held at the window's start... + m.FlushOutbound(s, discard889) + s.BufferOutbound(make([]byte, 500), true) // ...and the handler adds as much + out := windowEnd889(m, start, w) + if (w <= writeIdleWindows) != (len(out) == 0) { + t.Fatalf("window %d: expired %v (grace for %d windows)", w, out, writeIdleWindows) + } + } + }) +} + +// A tail held by the whole connection (several streams each waiting for a few +// bytes, the connection's floor made of those tails) is held without proof that it +// moves: the end of a transfer is on the wire, and the drain of its backlog can take +// more than a window (r9 F-A2; r8 judged by half of the connection's floor, which a +// slow drain does not reach). Bounded by the idle windows, whatever went out. +func TestWriteConnTailGrace889(t *testing.T) { + for _, tc := range []struct { + name string + release int + }{ + {"half-of-the-conn-floor-went-out", 100}, + {"under-half", 99}, + {"nothing-went-out", 0}, + } { + t.Run(tc.name, func(t *testing.T) { + _, m, _ := wdSetup889(t) + for i := 0; i < 3; i++ { + stalled889(m, uint32(1+2*i), 100) + time.Sleep(2 * time.Millisecond) // a slow runner (the coverage build): the records are not armed in one instant + } + start := time.Now().UnixNano() // after the records, so that window 1 is judged for all of them + c := m.CreateStream(99) + c.SetState(StateOpen) + c.SetWindowSize(1 << 20) + if tc.release > 0 { + m.ReserveSendWindow(c, int32(tc.release)) // a sibling's direct send: the conn releases + } + if w := firstReset889(m, 5, start, 10, nil); w != writeIdleWindows+1 { + t.Fatalf("stream 5 reset in window %d, want %d (held for the %d idle windows)", w, writeIdleWindows+1, writeIdleWindows) + } + }) + } +} + +// firstReset889 runs windows and reports the one in which stream id is expired +// (0 if it is not in limit windows). before(w) acts at the start of window w. +func firstReset889(m *Manager, id uint32, start int64, limit int, before func(w int)) int { + for w := 1; w <= limit; w++ { + if before != nil { + before(w) + } + for _, e := range windowEnd889(m, start, w) { + if e == id { + return w + } + } + } + return 0 +} + +// The true bounds of an unread stream (r8 F-C, derived from the code): the bytes it +// had in flight when it stalled (the peer's initial window less what was left of its +// own, up to four floors) are spent first, a floor a window; then, alone, the verdict; +// with a sibling that keeps the connection above the floor, writeIdleWindows more. +// So a stream the client never grants is reset in window 1+k alone and 1+k+3 with a +// moving sibling, k = min(4, floor(in flight / 64 KiB)): 5 and 8 for a window of +// 256 KiB or more (5 and 8 x WriteTimeout), against 1 and 4 for the default window. +func TestWriteUnreadStreamBounds889(t *testing.T) { + for _, tc := range []struct { + initWin uint32 + alone, sib int + }{ + {65535, 1, 4}, + {128 << 10, 3, 6}, + {192 << 10, 4, 7}, + {256 << 10, 5, 8}, + {1 << 20, 5, 8}, + {1<<31 - 1, 5, 8}, + } { + t.Run(strconv.Itoa(int(tc.initWin>>10))+"KiB", func(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(tc.initWin) + stalled889(m, 1, 8<<20) + if w := firstReset889(m, 1, time.Now().UnixNano(), 20, nil); w != tc.alone { + t.Fatalf("alone: reset in window %d, want %d", w, tc.alone) + } + _, m2, _ := wdSetup889(t) + m2.initWin.Store(tc.initWin) + b := stalled889(m2, 1, 8<<20) + a := m2.CreateStream(3) + a.SetState(StateOpen) + a.SetWindowSize(0) + a.BufferOutbound(make([]byte, 64<<20), true) + _ = b + w := firstReset889(m2, 1, time.Now().UnixNano(), 20, func(int) { a.SetWindowSize(70000); m2.FlushOutbound(a, discard889) }) + if w != tc.sib { + t.Fatalf("with a sibling that moves the connection: reset in window %d, want %d", w, tc.sib) + } + }) + } +} + +// A one-byte dribbler: reset in window 2 with a default window (the seed and a few +// bytes cover window 1), in window 5 with a window of 256 KiB or more (four floors +// of seed, spent a window each). +func TestWriteDribblerBounds889(t *testing.T) { + for _, tc := range []struct { + initWin uint32 + want int + }{{65535, 2}, {128 << 10, 3}, {256 << 10, 5}, {1 << 30, 5}} { + t.Run(strconv.Itoa(int(tc.initWin>>10))+"KiB", func(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(tc.initWin) + s := stalled889(m, 1, 8<<20) + w := firstReset889(m, 1, time.Now().UnixNano(), 20, func(int) { + for i := 0; i < 10; i++ { + s.SetWindowSize(1) + m.FlushOutbound(s, discard889) + } + }) + if w != tc.want { + t.Fatalf("reset in window %d, want %d", w, tc.want) + } + }) + } +} + +// The idle count restarts when the stream takes its share, its floor, or is +// starved by the connection window: a stream that alternates is not reset early. +func TestWriteIdleCountRestarts889(t *testing.T) { + setup := func(t *testing.T) (*Manager, *Stream, *Stream) { + _, m, _ := wdSetup889(t) + b := stalled889(m, 1, 64<<20) + a := stalled889(m, 3, 64<<20) + return m, a, b + } + move := func(m *Manager, a *Stream) { a.SetWindowSize(70000); m.FlushOutbound(a, discard889) } + t.Run("share", func(t *testing.T) { + m, a, b := setup(t) + got := firstReset889(m, 1, time.Now().UnixNano(), 20, func(w int) { + move(m, a) + if w == 2 { // one window in which it takes its share + b.SetWindowSize(20000) + m.FlushOutbound(b, discard889) + } + }) + if got != 2+writeIdleWindows+1 { + t.Fatalf("reset in window %d, want %d (idle 1, a share window, then %d idle windows held)", got, 2+writeIdleWindows+1, writeIdleWindows) + } + }) + t.Run("floor", func(t *testing.T) { + m, a, b := setup(t) + got := firstReset889(m, 1, time.Now().UnixNano(), 20, func(w int) { + move(m, a) + if w == 2 { + b.SetWindowSize(70000) + m.FlushOutbound(b, discard889) + } + }) + if got != 2+writeIdleWindows+1 { + t.Fatalf("reset in window %d, want %d", got, 2+writeIdleWindows+1) + } + }) + t.Run("starved", func(t *testing.T) { + m, a, b := setup(t) + got := firstReset889(m, 1, time.Now().UnixNano(), 20, func(w int) { + if w == 3 { // the conn window is shut and b's own has room: starved; a sibling takes the grant + atomic.StoreInt32(&m.connectionWindow, 0) + b.SetWindowSize(1 << 20) + a.SetWindowSize(70000) + m.UpdateConnectionWindow(70000) + m.FlushOutbound(a, discard889) + return + } + b.SetWindowSize(0) + m.UpdateConnectionWindow(70000) + move(m, a) + }) + if got != 3+writeIdleWindows+1 { + t.Fatalf("reset in window %d, want %d (a starved window restarts the count)", got, 3+writeIdleWindows+1) + } + }) +} + +// The share is half the connection's floor divided among the streams stalled. +func TestWriteShareScalesWithTheStreamsAndHalves889(t *testing.T) { + run := func(t *testing.T, n, release int) int { + _, m, _ := wdSetup889(t) + b := stalled889(m, 1, 64<<20) + a := stalled889(m, 3, 64<<20) + for i := 2; i < n; i++ { + stalled889(m, uint32(1+2*i), 64<<20) + } + return firstReset889(m, 1, time.Now().UnixNano(), 12, func(int) { + a.SetWindowSize(70000) + m.FlushOutbound(a, discard889) + b.SetWindowSize(int32(release)) + m.FlushOutbound(b, discard889) + }) + } + // N = 2: the share is 65536 / (2 x 2) = 16384. + if w := run(t, 2, 20000); w != 0 { + t.Fatalf("N=2, 20000 a window (above the share): reset in window %d", w) + } + if w := run(t, 2, 10000); w != 1+writeIdleWindows { + t.Fatalf("N=2, 10000 a window (under the share): reset in window %d, want %d", w, 1+writeIdleWindows) + } + // N = 10: the share is 65536 / 20 = 3276. + if w := run(t, 10, 4000); w != 0 { + t.Fatalf("N=10, 4000 a window (above the share): reset in window %d", w) + } + if w := run(t, 10, 2000); w != 1+writeIdleWindows { + t.Fatalf("N=10, 2000 a window (under the share): reset in window %d, want %d", w, 1+writeIdleWindows) + } +} + +// The seed is the bytes in flight, the initial window less what is left of it. +func TestWriteSeedIsWhatIsInFlight889(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(1 << 20) + s := m.CreateStream(1) + s.SetState(StateOpen) + s.SetWindowSize(1<<20 - 100000) // 100000 bytes sent, not granted back + s.BufferOutbound(make([]byte, 8<<20), true) + // 100000 covers window 1 (above the floor) and not window 2. + if w := firstReset889(m, 1, time.Now().UnixNano(), 10, nil); w != 2 { + t.Fatalf("reset in window %d, want 2: the seed is what is in flight, not the initial window", w) + } +} + +// A connection whose floor is its few held bytes is "moving" when it releases them. +func TestWriteConnFloorIsWhatTheConnHolds889(t *testing.T) { + _, m, _ := wdSetup889(t) + stalled889(m, 1, 100) + stalled889(m, 3, 100) + c := m.CreateStream(99) + c.SetState(StateOpen) + c.SetWindowSize(1 << 20) + m.ReserveSendWindow(c, 400) // a sibling releases 400 bytes: more than the connection held + if out := m.ExpiredWrites(dueAt889(), make([]uint32, 0, 8)); len(out) != 0 { + t.Fatalf("expired %v: the connection released more than it held (its floor is what it holds, not 64 KiB)", out) + } +} + +// The bank does not survive a window held on the share: a stream below its floor +// on a connection that does not move is the verdict. +func TestWriteBankIsSpentByAShareWindow889(t *testing.T) { + _, m, _ := wdSetup889(t) + b := stalled889(m, 1, 64<<20) + a := stalled889(m, 3, 64<<20) + start := time.Now().UnixNano() + // Window 1: b releases 95536 (above the floor: 30000 banked) and a moves the conn. + b.SetWindowSize(95536) + m.FlushOutbound(b, discard889) + a.SetWindowSize(70000) + m.FlushOutbound(a, discard889) + if out := windowEnd889(m, start, 1); len(out) != 0 { + t.Fatalf("window 1: %v", out) + } + // Window 2: b takes 20000 (under the floor, over its share) while a moves the conn. + b.SetWindowSize(20000) + m.FlushOutbound(b, discard889) + a.SetWindowSize(70000) + m.FlushOutbound(a, discard889) + if out := windowEnd889(m, start, 2); len(out) != 0 { + t.Fatalf("window 2: %v", out) + } + // Window 3: b takes 40000 and nothing else moves: 40000 is under the floor and the + // connection is under its floor. The 30000 banked in window 1 do not make 70000. + b.SetWindowSize(40000) + m.FlushOutbound(b, discard889) + reset := false + for _, id := range windowEnd889(m, start, 3) { + reset = reset || id == 1 + } + if !reset { + t.Fatal("window 3: stream 1 held; the bank of window 1 survived a window held on the share") + } +} + +// The stream's own tail is judged by half even when the connection holds much more +// (a sibling with a megabyte waiting) and does not move: the conn clause does not +// apply, the stream clause does. +func TestWriteStreamTailGraceBesideABigSibling889(t *testing.T) { + _, m, _ := wdSetup889(t) + stalled889(m, 1, 1<<20) // the sibling: nothing released, a verdict at once + tail := stalled889(m, 3, 10000) + tail.SetWindowSize(6000) // 6000 of its 10000, under the connection's floor + m.FlushOutbound(tail, discard889) + got := map[uint32]bool{} + for _, id := range m.ExpiredWrites(dueAt889(), make([]uint32, 0, 8)) { + got[id] = true + } + if got[3] { + t.Fatal("the tail that went out 60% was reset") + } + if !got[1] { + t.Fatal("the sibling that released nothing was held") + } +} + +// An unread stream whose unsent rest is under one floor is let go as fast as a large +// one (r9 F-H): the floor is what it holds, so the bytes in flight (the seed, up to 256 +// KiB) were spent at a few bytes per window, 65,536 windows for the default initial +// window and 262,145 for 256 KiB or more, and the connection was exempt from +// ReadTimeout and IdleTimeout all that time. A window that passes on the carried bank +// pays a whole floor: the bytes in flight are gone in 1+k windows (k as in +// TestWriteUnreadStreamBounds889), and the connection holding under one floor gets the +// tail's three idle windows on top: window 5 (default window), 6 (128 KiB), 8 (256 KiB +// or more). +func TestWriteUnreadSmallRemainderBounds889(t *testing.T) { + for _, held := range []int{1, 100, 10000, 60000} { + for _, tc := range []struct { + initWin uint32 + want int + }{{65535, 5}, {128 << 10, 6}, {1 << 20, 8}} { + t.Run(strconv.Itoa(held)+"B/"+strconv.Itoa(int(tc.initWin>>10))+"KiB", func(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(tc.initWin) + stalled889(m, 1, held) + if w := firstReset889(m, 1, time.Now().UnixNano(), 40, nil); w != tc.want { + t.Fatalf("reset in window %d, want %d (0 means not within 40)", w, tc.want) + } + }) + } + } +} + +// The carry-over of a streamed response survives a small window: a stream that +// released 312 KiB of 320 KiB in one window and then buffers a megabyte keeps its +// bank (r9, why the bank is not capped by the floor of the window judged). +func TestWriteBankSurvivesASmallRenewal889(t *testing.T) { + for _, order := range []string{"buffer-after-renewal", "buffer-before-renewal"} { + t.Run(order, func(t *testing.T) { + _, m, _ := wdSetup889(t) + s := stalled889(m, 1, 320<<10) + start := time.Now().UnixNano() + s.SetWindowSize(312 << 10) + m.FlushOutbound(s, discard889) + if order == "buffer-before-renewal" { + s.BufferOutbound(make([]byte, 1<<20), true) + } + got := firstReset889(m, 1, start, 50, func(w int) { + if w == 2 && order == "buffer-after-renewal" { + s.BufferOutbound(make([]byte, 1<<20), true) + } + }) + if got < 5 { + t.Fatalf("reset in window %d, want at least 5: the 256 KiB carried over must not be cut by a small window", got) + } + }) + } +} + +// A dribbler with a small rest does not live on its bank either (r9 F-H): one byte a +// window released is not a floor, so each window pays a whole one. +func TestWriteDribblerWithASmallRestSpendsItsBank889(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(1 << 20) // 256 KiB of bank + s := stalled889(m, 1, 1000) + got := firstReset889(m, 1, time.Now().UnixNano(), 40, func(int) { + s.SetWindowSize(1) + m.FlushOutbound(s, discard889) + s.BufferOutbound([]byte{0}, true) + }) + if got != 8 { + t.Fatalf("reset in window %d, want 8 (0 means not within 40: the bank was spent at the rest, not at a floor, a window)", got) + } +} + +// What a window pays beyond the bank is not owed later: a stream that spent a bank +// of less than a floor on a window it released nothing in (a whole floor is paid, the +// bank was 20000) starts the next from zero, not from a debt (the bank is clamped at 0). +// Then 135000 released in one window carry 134000 over (the floor of that window was the +// 1000 it held), which is one window for the floor and one for the rest. +func TestWriteBankIsNotADebt889(t *testing.T) { + _, m, _ := wdSetup889(t) + m.initWin.Store(20000) + s := stalled889(m, 1, 1000) + got := firstReset889(m, 1, time.Now().UnixNano(), 12, func(w int) { + if w == 2 { + s.BufferOutbound(make([]byte, 1<<20), true) + s.SetWindowSize(135000) + m.FlushOutbound(s, discard889) + } + }) + if got != 5 { + t.Fatalf("reset in window %d, want 5", got) + } +} + +// On a connection that holds less than a floor, a stream that takes its share while +// the connection moves is held and its idle count cleared (the share clause, not the +// tail's, which counts every such window): the connection's floor is what it holds +// (r8 p8, still true with the r9 tail grace). +func TestWriteSmallConnShareHoldClearsIdle889(t *testing.T) { + _, m, _ := wdSetup889(t) + a := stalled889(m, 1, 9000) + b := stalled889(m, 3, 1000) + c := m.CreateStream(99) + c.SetState(StateOpen) + c.SetWindowSize(1 << 30) + start := time.Now().UnixNano() + got := firstReset889(m, 1, start, 12, func(int) { + a.SetWindowSize(3000) // more than its share (10000 / (2 x 2)), less than half of what it holds + m.FlushOutbound(a, discard889) + a.BufferOutbound(make([]byte, 3000), true) + b.SetWindowSize(1000) // b takes all it holds: its floor + m.FlushOutbound(b, discard889) + b.BufferOutbound(make([]byte, 1000), true) + m.ReserveSendWindow(c, 8000) // the connection moves above what it holds + }) + if got != 0 { + t.Fatalf("stream 1 reset in window %d although it took its share of a moving connection every window", got) + } +} diff --git a/internal/resource/config.go b/internal/resource/config.go index 8e94efe47..de2d1fbb0 100644 --- a/internal/resource/config.go +++ b/internal/resource/config.go @@ -375,8 +375,10 @@ func (c Config) WithDefaults() Config { // A zero means "unset" only on the first pass: once normalised is true, 0 is // a disabled timeout this function must leave alone, otherwise the second // WithDefaults (in the engine constructor) reinstates the default over an -// explicitly disabled timeout — celeris#594. Any positive value is kept -// verbatim on every pass, so N stays N. +// explicitly disabled timeout — celeris#594. A positive value is kept on every +// pass up to MaxTimeout (a hundred years, which now + timeout cannot overflow +// with) and a larger one, math.MaxInt64 the usual way to write "never", is +// clamped to it, so N stays N for every N below that. func resolveTimeout(v, def time.Duration, normalised bool) time.Duration { switch { case v < 0: @@ -384,6 +386,12 @@ func resolveTimeout(v, def time.Duration, normalised bool) time.Duration { case v == 0 && !normalised: return def default: - return v + return min(v, MaxTimeout) } } + +// MaxTimeout is the longest timeout the engines keep: a hundred years. A timeout of +// math.MaxInt64 (the common way to write "never") overflowed every `now + timeout` +// the engines compute (a deadline in the past, due at once: celeris#889 review r8 +// F-F), and a hundred years is never in any practical sense. +const MaxTimeout = 100 * 365 * 24 * time.Hour diff --git a/internal/resource/timeout_clamp_889_test.go b/internal/resource/timeout_clamp_889_test.go new file mode 100644 index 000000000..868820fcb --- /dev/null +++ b/internal/resource/timeout_clamp_889_test.go @@ -0,0 +1,27 @@ +package resource + +import ( + "math" + "testing" + "time" +) + +// A timeout of math.MaxInt64 is kept as a hundred years, for every timeout, so that +// `now + timeout` in the engines (detectDeadline, the h2 deadlines) cannot overflow. +func TestHugeTimeoutsAreClamped889(t *testing.T) { + never := time.Duration(math.MaxInt64) + c := Config{ReadTimeout: never, ReadHeaderTimeout: never, WriteTimeout: never, IdleTimeout: never}.WithDefaults() + for name, got := range map[string]time.Duration{"read": c.ReadTimeout, "readHeader": c.ReadHeaderTimeout, "write": c.WriteTimeout, "idle": c.IdleTimeout} { + if got != MaxTimeout { + t.Errorf("%s timeout = %v, want %v", name, got, MaxTimeout) + } + if now := time.Now().UnixNano(); now+int64(got) < now { + t.Errorf("%s: now + timeout overflows", name) + } + } + // And the second pass (the engine constructor's) keeps it, and an ordinary value stays N. + c2 := Config{ReadTimeout: 5 * time.Second}.WithDefaults().WithDefaults() + if c2.ReadTimeout != 5*time.Second || c.WithDefaults().WriteTimeout != MaxTimeout { + t.Errorf("second pass: %v %v", c2.ReadTimeout, c.WithDefaults().WriteTimeout) + } +} diff --git a/test/server/h2_request_deadline_889_linux_test.go b/test/server/h2_request_deadline_889_linux_test.go index edc0cdf0e..3bda3347a 100644 --- a/test/server/h2_request_deadline_889_linux_test.go +++ b/test/server/h2_request_deadline_889_linux_test.go @@ -55,6 +55,7 @@ func rdServer889(t *testing.T, eng celeris.EngineType, mutate func(*celeris.Conf // rdEvent889 is what a stream (and its connection) did, as the client saw it. type rdEvent889 struct { + id uint32 // the stream awaitAny889 ended on status string // :status of the first HEADERS on the stream, "" if none reset bool // RST_STREAM on the stream rst http2.ErrCode // its code @@ -209,9 +210,9 @@ func (c *h2raw889) serves889(id uint32) { // TestH2PingingClientRequestIsAnsweredAndReset889 is the pinned ping cell of // the previous work FLIPPED: a client that never completes its POST and sends a // PING every 500 ms is no longer held: the request is answered 408 and reset at -// ReadTimeout, the conn stays and serves the next request. (The pinned cells for -// an unread RESPONSE stay pinned until the per-stream write deadline; see -// TestH2PingingClientHoldsAnUnfinishedStream889.) +// ReadTimeout, the conn stays and serves the next request. (The pinned cell for +// an unread RESPONSE was flipped the same way by the per-stream write deadline; +// see TestH2PingingClientHoldsAnUnfinishedStream889.) func TestH2PingingClientRequestIsAnsweredAndReset889(t *testing.T) { for _, e := range nativeEngines889() { t.Run(e.name, func(t *testing.T) { @@ -600,3 +601,46 @@ func TestH2BurstOfExpiriesIsAnsweredInOneSweep889(t *testing.T) { }) } } + +// awaitAny889 is await889 for several streams: it returns at the first reset or +// end of any of them (ev.id says which), the conn ending, or d since c.t0. +func (c *h2raw889) awaitAny889(ids []uint32, d time.Duration) rdEvent889 { + c.t.Helper() + var ev rdEvent889 + want := map[uint32]bool{} + for _, id := range ids { + want[id] = true + } + _ = c.conn.SetReadDeadline(c.t0.Add(d)) + for { + f, err := c.fr.ReadFrame() + ev.at = time.Since(c.t0) + if err != nil { + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + ev.timeout = true + return ev + } + ev.connEnd = "EOF" + return ev + } + switch f := f.(type) { + case *http2.DataFrame: + if want[f.StreamID] { + ev.data += len(f.Data()) + if f.StreamEnded() { + ev.id, ev.ended = f.StreamID, true + return ev + } + } + case *http2.RSTStreamFrame: + if want[f.StreamID] { + ev.id, ev.reset, ev.rst = f.StreamID, true, f.ErrCode + return ev + } + case *http2.GoAwayFrame: + ev.connEnd = "GOAWAY " + f.ErrCode.String() + return ev + } + } +} diff --git a/test/server/h2_silent_streams_889_linux_test.go b/test/server/h2_silent_streams_889_linux_test.go index e7fc35580..fe706ed4d 100644 --- a/test/server/h2_silent_streams_889_linux_test.go +++ b/test/server/h2_silent_streams_889_linux_test.go @@ -17,12 +17,14 @@ import ( // request, the WINDOW_UPDATE a response is waiting for) and then sends nothing // is a different case, and main reaps it at ReadTimeout. These cells hold the // engines to that: ReadTimeout 1 s, IdleTimeout off, WriteTimeout 2 s, routes -// sync and async, a client that sends no frame after its request. +// sync and async, a client that sends no frame after its request. (A response +// that waits for the window and is buffered in the stream is bounded by +// WriteTimeout now, and keeps the conn until its stream is reset: the conn is +// closed after that, still within the bound below.) // // A client that sends a frame of any kind at least once per ReadTimeout is not -// reaped, on main or here (it restarts the conn clock): only a per-stream -// deadline, which std has and the native engines do not, bounds one. The last -// test pins that. +// reaped by the conn clock (it restarts it): the per-stream deadlines bound its +// streams. The last test pins that for a response. // silentServer889 is the server of these cells: /upload889 (POST, answers ok // once its body is whole), /big889 (a 1 MiB or, with ?n=8388608, an 8 MiB @@ -84,9 +86,10 @@ var silentOwes889 = []struct { {"response-8MiB-untaken", func(c *h2raw889) { c.request(1, "GET", "/big889?n=8388608", true, true) }}, } -// pingCells889 are the cells a pinging client still holds open (see the last -// test). The request-open cell is gone from it: the per-stream request deadline -// flipped it (TestH2PingingClientRequestIsAnsweredAndReset889). +// pingCells889 are the cells of a client that pings and never takes a response +// (see the last test): the request-open cell was flipped by the per-stream request +// deadline (TestH2PingingClientRequestIsAnsweredAndReset889), the untaken response +// by the per-stream write deadline. var pingCells889 = []struct { name string req func(c *h2raw889) @@ -119,17 +122,16 @@ func TestH2SilentClientOwingTheServerIsReaped889(t *testing.T) { } } -// TestH2PingingClientHoldsAnUnfinishedStream889 pins what is still not bounded: -// a client that opens a stream, asks for a response it never takes, and sends -// one PING every 500 ms (ReadTimeout is 1 s), is not closed on the native -// engines (the PING restarts the conn clock, on main as well: measured on main -// f4ae3ad, all three engines). std ends such a stream at WriteTimeout -// (RST_STREAM INTERNAL_ERROR, 2.00 s). The request side of this was flipped by -// the per-stream request deadline (ReadTimeout, 408 and RST_STREAM(NO_ERROR)); -// the response side waits for the per-stream write deadline, the second stage -// of the per-stream deadlines work under celeris#889 (L19, D2). When it lands -// these cells are FLIPPED to "the stream is reset", not deleted, and this -// comment goes with them. +// TestH2PingingClientHoldsAnUnfinishedStream889 (name kept) was the pinned cell +// of a client that opens a stream, asks for a response it never takes, and sends +// one PING every 500 ms (ReadTimeout is 1 s): on main it was not closed and its +// stream was never reset on the native engines (measured on main f4ae3ad, all +// three engines), where std ends such a stream at WriteTimeout (RST_STREAM +// INTERNAL_ERROR, 2.00 s). The request side was flipped by the per-stream +// request deadline (ReadTimeout, 408 and RST_STREAM(NO_ERROR)); this is the +// response side, FLIPPED by the per-stream write deadline (L19, D2): the stream +// is reset with INTERNAL_ERROR at WriteTimeout (2 s), and the conn, whose client +// pings, stays and serves the next request. func TestH2PingingClientHoldsAnUnfinishedStream889(t *testing.T) { for _, e := range nativeEngines889() { for _, sc := range pingCells889 { @@ -138,32 +140,12 @@ func TestH2PingingClientHoldsAnUnfinishedStream889(t *testing.T) { addr := silentServer889(t, e.eng, false) c := dialH2raw889(t, addr, h2dial889{}) sc.req(c) - stop := make(chan struct{}) - done := make(chan struct{}) - go func() { - defer close(done) - tk := time.NewTicker(500 * time.Millisecond) - defer tk.Stop() - for { - select { - case <-stop: - return - case <-tk.C: - c.wmu.Lock() - _ = c.fr.WritePing(false, [8]byte{'8', '8', '9'}) - c.wmu.Unlock() - } - } - }() - o := c.watch(0, 6*time.Second) - close(stop) - <-done - t.Logf("H2PINGOWES889 engine=%s cell=%s: %s", e.name, sc.name, o) - if !o.open { - t.Fatalf("the connection was closed %v after the request although the client pinged every 500 ms (%s): %s; "+ - "this cell pins that a pinging client is not bounded by ReadTimeout (as on main), see the comment", - o.at.Round(time.Millisecond), sc.name, o.how) - } + stop := c.pingEvery889(500 * time.Millisecond) + defer stop() + ev := c.await889(1, 6*time.Second) + t.Logf("H2PINGOWES889 engine=%s cell=%s: %s", e.name, sc.name, ev) + wantWriteReset889(t, ev, 1800*time.Millisecond, 3500*time.Millisecond) + c.serves889(3) }) } } diff --git a/test/server/h2_write_deadline_889_linux_test.go b/test/server/h2_write_deadline_889_linux_test.go new file mode 100644 index 000000000..154292c78 --- /dev/null +++ b/test/server/h2_write_deadline_889_linux_test.go @@ -0,0 +1,293 @@ +//go:build linux + +package celeris_test + +import ( + "strconv" + "sync/atomic" + "testing" + "time" + + "golang.org/x/net/http2" + + "github.com/goceleris/celeris" +) + +// The per-stream write deadline of the native engines (celeris#889, L19 D2). +// +// A response whose bytes wait for the client's flow-control window (an inline +// handler's tail, buffered in the stream; a pool handler's) is bounded by +// WriteTimeout as a MINIMUM RATE, the rule #1119 gives a response parked on the +// socket: a stream that releases at least min(64 KiB, what it holds) of its DATA +// in a WriteTimeout window is held another window; one that does not is reset +// with RST_STREAM(INTERNAL_ERROR), as net/http resets a stream at its write +// deadline. No ceiling: a stream that takes its floor in every window is never +// cut. Progress is bytes released, never frames: a PING, or a WINDOW_UPDATE of one +// byte, is not progress. A stream whose own window has room but whose bytes wait +// for the CONNECTION window is credited with what the whole conn released. A +// conn with such a stream is exempt from ReadTimeout and IdleTimeout while the +// stream lives (the exemption of WriteTimeout > 0 replaces the old rule that +// closed the conn of a silent client, A2). +// +// Server: WriteTimeout 1 s unless a cell says so; ReadTimeout and IdleTimeout +// off. The client sends its request, takes the first window's DATA and then only +// what the cell says. + +const wd889 = time.Second + +// wdServer889 is the server of these cells: /inline889?n= an inline handler's +// body of n bytes (the tail beyond the windows is buffered on the loop), /pool889 +// the same from a pool handler, /tick889 an SSE-like stream of small chunks. +func wdServer889(t *testing.T, eng celeris.EngineType, read, idle, write time.Duration) string { + t.Helper() + cfg := celeris.Config{Engine: eng, ReadTimeout: read, IdleTimeout: idle, WriteTimeout: write} + var ended atomic.Bool + return startServerConfig761(t, cfg, func(s *celeris.Server) { + tickRoute889(s, "/tick889", 500*time.Millisecond, 4*time.Second, &ended) + s.GET("/inline889", func(c *celeris.Context) error { + n, _ := strconv.Atoi(c.Query("n")) + return c.Blob(200, "application/octet-stream", make([]byte, n)) + }) + s.GET("/pool889", func(c *celeris.Context) error { + n, _ := strconv.Atoi(c.Query("n")) + return c.Blob(200, "application/octet-stream", make([]byte, n)) + }).Async() + }) +} + +// wdPaths889 are the two ways a response comes to wait: from the loop and from +// the pool. +var wdPaths889 = []string{"inline889", "pool889"} + +func (c *h2raw889) window889(id, n uint32) { + c.wmu.Lock() + _ = c.fr.WriteWindowUpdate(id, n) + c.wmu.Unlock() +} + +// every889 runs f every d until the returned stop is called. +func every889(d time.Duration, f func()) (stop func()) { + quit, done := make(chan struct{}), make(chan struct{}) + go func() { + defer close(done) + tk := time.NewTicker(d) + defer tk.Stop() + for { + select { + case <-quit: + return + case <-tk.C: + f() + } + } + }() + return func() { close(quit); <-done } +} + +func wantWriteReset889(t *testing.T, ev rdEvent889, lo, hi time.Duration) { + t.Helper() + if ev.timeout || ev.connEnd != "" || !ev.reset { + t.Fatalf("want the stalled stream reset with RST_STREAM(INTERNAL_ERROR) in [%v, %v], conn open; got %s", lo, hi, ev) + } + if ev.rst != http2.ErrCodeInternal { + t.Fatalf("the stream was reset with %v, want INTERNAL_ERROR (what net/http and #906 send at a write deadline)", ev.rst) + } + if ev.at < lo || ev.at > hi { + t.Fatalf("the stream was reset at %v, want it in [%v, %v]", ev.at.Round(time.Millisecond), lo, hi) + } +} + +// TestH2StalledTailIsResetAtWriteTimeout889: a 4 MiB body, the client takes the +// first 65535 bytes (the initial window) and grants nothing, pinging every 300 ms: +// the stream is reset with INTERNAL_ERROR WriteTimeout after its tail was +// buffered, and the conn serves the next request, also after the client's late +// frames for the reset stream (WINDOW_UPDATE, RST_STREAM, PRIORITY). On main the +// stream lived for as long as the client pinged. +func TestH2StalledTailIsResetAtWriteTimeout889(t *testing.T) { + for _, e := range nativeEngines889() { + for _, path := range wdPaths889 { + t.Run(e.name+"/"+path, func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{}) + c.request(1, "GET", "/"+path+"?n=4194304", true, true) + stop := c.pingEvery889(300 * time.Millisecond) + defer stop() + ev := c.await889(1, 6*time.Second) + t.Logf("WD889 stalled engine=%s path=%s: %s", e.name, path, ev) + wantWriteReset889(t, ev, 900*time.Millisecond, 2500*time.Millisecond) + if ev.data < 60000 || ev.data > 70000 { + t.Fatalf("%d DATA bytes before the reset, want the first window (65535)", ev.data) + } + // Late frames for the stream the server reset keep the conn. + c.window889(1, 1000) + c.wmu.Lock() + _ = c.fr.WritePriority(1, http2.PriorityParam{Weight: 15, StreamDep: 0}) + _ = c.fr.WriteRSTStream(1, http2.ErrCodeCancel) + c.wmu.Unlock() + c.serves889(3) + }) + } + } +} + +// TestH2WindowUpdateDribblerIsResetAtTheFirstJudgement889: a WINDOW_UPDATE of one +// byte on the stream and on the connection every 100 ms releases a byte at a +// time, 600 bytes in 60 s at that rate: the stream is reset at the first +// judgement, WriteTimeout after the tail was buffered, as a client that grants +// nothing is. +func TestH2WindowUpdateDribblerIsResetAtTheFirstJudgement889(t *testing.T) { + for _, e := range nativeEngines889() { + t.Run(e.name, func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{}) + c.request(1, "GET", "/inline889?n=4194304", true, true) + stop := every889(100*time.Millisecond, func() { c.window889(1, 1); c.window889(0, 1) }) + defer stop() + ev := c.await889(1, 6*time.Second) + t.Logf("WD889 dribble engine=%s: %s", e.name, ev) + // The seed (the 65535 bytes in flight) and a few bytes cover the first + // window; the second is the verdict. + wantWriteReset889(t, ev, 900*time.Millisecond, 3500*time.Millisecond) + }) + } +} + +// TestH2WriteRateFloor889: the floor is min(64 KiB, what the stream holds) per +// window. A client that grants 65535 bytes (one less) once is reset at the first +// judgement; one that grants 65536 in each of three windows is held for three +// windows and reset in the first window it grants nothing in. +func TestH2WriteRateFloor889(t *testing.T) { + for _, e := range nativeEngines889() { + t.Run(e.name+"/floor-1", func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{}) + c.request(1, "GET", "/inline889?n=4194304", true, true) + time.AfterFunc(300*time.Millisecond, func() { c.window889(1, 65535); c.window889(0, 65535) }) + ev := c.await889(1, 6*time.Second) + t.Logf("WD889 floor-1 engine=%s: %s", e.name, ev) + // 65535 released and 65535 in flight at the start make two floors less + // 2 bytes: the first window is held, the second is the verdict. + wantWriteReset889(t, ev, 1800*time.Millisecond, 4000*time.Millisecond) + }) + t.Run(e.name+"/floor", func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{}) + c.request(1, "GET", "/inline889?n=4194304", true, true) + for k := range 3 { // in the middle of windows 1, 2 and 3 + time.AfterFunc(time.Duration(500+1000*k)*time.Millisecond, func() { c.window889(1, 65536); c.window889(0, 65536) }) + } + ev := c.await889(1, 9*time.Second) + t.Logf("WD889 floor engine=%s: %s", e.name, ev) + wantWriteReset889(t, ev, 3000*time.Millisecond, 7000*time.Millisecond) + }) + } +} + +// TestH2ConnWindowStarvedStreamsAreJudgedPerConnection889: two streams, each with +// room in its own window (the client's SETTINGS_INITIAL_WINDOW_SIZE is large) and +// the connection window shut. The conn window is the only thing they wait for, so +// a stream's credit is what the whole conn released. A dribble of one byte +// resets both at the first judgement; 128 KiB a second held for three windows +// holds both, though one sibling takes them all. +func TestH2ConnWindowStarvedStreamsAreJudgedPerConnection889(t *testing.T) { + for _, e := range nativeEngines889() { + t.Run(e.name+"/dribble", func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{initialWindow: 1 << 30}) + c.request(1, "GET", "/inline889?n=2097152", true, true) + c.request(3, "GET", "/inline889?n=2097152", true, true) + stop := every889(100*time.Millisecond, func() { c.window889(0, 1) }) + defer stop() + got := map[uint32]bool{} + var last rdEvent889 + for len(got) < 2 { + ev := c.awaitAny889([]uint32{1, 3}, 6*time.Second) + last = ev + if !ev.reset { + t.Fatalf("a starved stream was not reset (got %s, reset so far: %v)", ev, got) + } + got[ev.id] = true + } + t.Logf("WD889 conn-dribble engine=%s: both reset, the last at %v", e.name, last.at.Round(time.Millisecond)) + if last.at > 2500*time.Millisecond { + t.Fatalf("the second stream was reset at %v, want both at the first judgement (about 1 s)", last.at) + } + }) + t.Run(e.name+"/credited", func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{initialWindow: 1 << 30}) + c.request(1, "GET", "/inline889?n=2097152", true, true) + c.request(3, "GET", "/inline889?n=2097152", true, true) + var n atomic.Int32 + stop := every889(700*time.Millisecond, func() { + if n.Add(1) <= 9 { // 128 KiB to the conn, about 6.3 s of it: far more than the idle grace + c.window889(0, 131072) + } + }) + defer stop() + ev := c.awaitAny889([]uint32{1, 3}, 13*time.Second) + t.Logf("WD889 conn-credited engine=%s: stream %d %s", e.name, ev.id, ev) + if !ev.reset || ev.rst != http2.ErrCodeInternal { + t.Fatalf("want a reset with INTERNAL_ERROR once the grants stop, got %s", ev) + } + if ev.at < 6000*time.Millisecond { + t.Fatalf("a stream was reset at %v, while the conn released 128 KiB in every window (the starved sibling is credited with it)", ev.at.Round(time.Millisecond)) + } + }) + } +} + +// TestH2FlowingDownloadAndQuietStreamAreNotCut889: guards, green before and after. +// A body of 8 MiB taken at about 3 MB/s (longer than WriteTimeout, three windows: +// no ceiling) arrives whole; an SSE-like stream with window to spare, quiet +// between its chunks, is not touched. +func TestH2FlowingDownloadAndQuietStreamAreNotCut889(t *testing.T) { + for _, e := range nativeEngines889() { + for _, path := range wdPaths889 { + t.Run(e.name+"/"+path, func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{}) + const total = 8 << 20 + c.request(1, "GET", "/"+path+"?n="+strconv.Itoa(total), true, true) + got, err := c.takeAll889(total, func(n int) { + c.window889(0, uint32(n)) + c.window889(1, uint32(n)) + time.Sleep(time.Duration(n/3) * time.Microsecond) // about 3 MB/s + }) + t.Logf("WD889 flowing engine=%s path=%s: %d of %d bytes, err %v, at %v", e.name, path, got, total, err, time.Since(c.t0).Round(time.Millisecond)) + if err != nil || got != total { + t.Fatalf("a download that took its floor in every window was cut after %d of %d bytes: %v", got, total, err) + } + }) + } + t.Run(e.name+"/quiet-sse", func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, -1, -1, wd889), h2dial889{}) + c.request(1, "GET", "/tick889", true, true) + ev := c.await889(1, 6*time.Second) + t.Logf("WD889 quiet engine=%s: %s", e.name, ev) + if ev.reset || ev.connEnd != "" { + t.Fatalf("a quiet stream with window to spare was cut: %s", ev) + } + }) + } +} + +// TestH2SilentClientWithAStalledResponseKeepsTheConnUntilTheStreamIsReset889: +// ReadTimeout 1 s, WriteTimeout 3 s (the read side is the shorter, so the cell can +// tell whether the stalled stream exempts the conn), a client that sends nothing +// after its GET. The old rule (A2) closed the whole conn at ReadTimeout without a +// word; now the stream is reset at WriteTimeout, with INTERNAL_ERROR, before the +// conn goes. +func TestH2SilentClientWithAStalledResponseKeepsTheConnUntilTheStreamIsReset889(t *testing.T) { + for _, e := range nativeEngines889() { + t.Run(e.name, func(t *testing.T) { + t.Parallel() + c := dialH2raw889(t, wdServer889(t, e.eng, time.Second, -1, 3*time.Second), h2dial889{}) + c.request(1, "GET", "/inline889?n=4194304", true, true) + ev := c.await889(1, 8*time.Second) + t.Logf("WD889 silent engine=%s: %s", e.name, ev) + wantWriteReset889(t, ev, 2800*time.Millisecond, 5000*time.Millisecond) + }) + } +} diff --git a/test/server/h2_write_rate_889_linux_test.go b/test/server/h2_write_rate_889_linux_test.go new file mode 100644 index 000000000..f7dbda50a --- /dev/null +++ b/test/server/h2_write_rate_889_linux_test.go @@ -0,0 +1,333 @@ +//go:build linux + +package celeris_test + +import ( + "fmt" + "io" + "strconv" + "testing" + "time" + + "golang.org/x/net/http2" +) + +// The minimum rate of the write deadline must not cut a reader that is slow but +// moving (celeris#889, D2, review r7). Two ways it did: +// +// - a client that sends few WINDOW_UPDATEs (nghttp2/curl update when half of the +// window is consumed) is far above the floor on average but has no update in +// some windows: progress counted only when an update arrived and nothing +// carried over, so it was reset (curl --limit-rate 300k, WriteTimeout 1 s, was +// cut after 7.9 s where main completes); +// - parallel downloads that share a slow link are each judged on their share, +// link/N, and were reset while the connection as a whole moved far above the +// floor. +// +// The clients here mimic those policies with the raw framer; the connection's +// window is opened to 1 GiB, as Go's Transport and browsers do. + +// batched889 downloads path with stream window `window`, "consuming" at bps and +// sending a WINDOW_UPDATE (stream and conn) whenever step bytes are consumed and +// not yet granted back. It reports how the stream ended, the bytes received and when. +func batched889(t *testing.T, addr, path string, window uint32, step int, bps float64, limit time.Duration) (string, int, time.Duration) { + t.Helper() + c := dialH2raw889(t, addr, h2dial889{initialWindow: window}) + c.window889(0, 1<<30-65535) + _ = c.conn.SetReadDeadline(time.Time{}) + type ev struct { + n int + end string + } + ch := make(chan ev, 4096) + go func() { + for { + f, err := c.fr.ReadFrame() + if err != nil { + ch <- ev{end: "conn error: " + err.Error()} + return + } + switch f := f.(type) { + case *http2.DataFrame: + if f.StreamID == 1 { + e := ev{n: len(f.Data())} + if f.StreamEnded() { + e.end = "complete" + } + ch <- e + if e.end != "" { + return + } + } + case *http2.RSTStreamFrame: + if f.StreamID == 1 { + ch <- ev{end: "RST " + f.ErrCode.String()} + return + } + case *http2.GoAwayFrame: + ch <- ev{end: "GOAWAY " + f.ErrCode.String()} + return + case *http2.PingFrame: + if !f.IsAck() { + c.wmu.Lock() + _ = c.fr.WritePing(true, f.Data) + c.wmu.Unlock() + } + } + } + }() + c.request(1, "GET", path, true, true) + t0 := time.Now() + recv, acked := 0, 0 + tk := time.NewTicker(10 * time.Millisecond) + defer tk.Stop() + for { + select { + case e := <-ch: + recv += e.n + if e.end != "" { + return e.end, recv, time.Since(t0) + } + case <-tk.C: + el := time.Since(t0) + if el > limit { + return "still open at the limit", recv, el + } + consumed := min(int(bps*el.Seconds()), recv) + if consumed-acked >= step { + c.window889(1, uint32(consumed-acked)) + c.window889(0, uint32(consumed-acked)) + acked = consumed + } + } + } +} + +// TestH2SparseWindowUpdatesAreNotCut889: readers well above the floor (64 KiB/s at +// WriteTimeout 1 s) whose WINDOW_UPDATEs are further apart than WriteTimeout. The +// surplus released at one update carries over, up to a few windows' worth, and the +// bytes in flight when the stream stalls count for its first window, so a client +// that releases on average at least the floor survives. On the previous head each +// was reset at 1.0 s (or after the first gap). +func TestH2SparseWindowUpdatesAreNotCut889(t *testing.T) { + type cell struct { + name string + window uint32 + step int + bps float64 + size int + } + cells := []cell{ + // nghttp2's policy: update at half the window. 1 MiB window, 300 KB/s + // (4.6x the floor): the gap between updates is 1.7 s. + {"nghttp2-like/W1MiB/300KBps", 1 << 20, 1 << 19, 300e3, 2 << 20}, + // A larger window, a longer gap: half of 6 MiB at 1 MB/s is 3 s. + {"halfwindow/W6MiB/1MBps", 6 << 20, 3 << 20, 1e6, 8 << 20}, + // Go's Transport: an update every 4 KiB. Guard, green before and after. + {"go-like/W64KiB/100KBps", 65535, 4 << 10, 100e3, 512 << 10}, + } + for _, e := range nativeEngines889() { + for _, cl := range cells { + t.Run(e.name+"/"+cl.name, func(t *testing.T) { + t.Parallel() + addr := wdServer889(t, e.eng, -1, -1, wd889) + how, recv, at := batched889(t, addr, "/inline889?n="+strconv.Itoa(cl.size), cl.window, cl.step, cl.bps, 30*time.Second) + t.Logf("WR889 sparse %s/%s: %s after %v, %d of %d bytes", e.name, cl.name, how, at.Round(time.Millisecond), recv, cl.size) + if how != "complete" || recv != cl.size { + t.Fatalf("a reader above the floor was cut: %s after %v, %d of %d bytes", how, at.Round(time.Millisecond), recv, cl.size) + } + }) + } + } +} + +// throttled889 reads from r at most bps bytes per second. +type throttled889 struct { + r io.Reader + bps float64 + t0 time.Time + n int +} + +func (t *throttled889) Read(p []byte) (int, error) { + if len(p) > 4096 { + p = p[:4096] + } + if want := time.Duration(float64(t.n) / t.bps * float64(time.Second)); time.Since(t.t0) < want { + time.Sleep(want - time.Since(t.t0)) + } + n, err := t.r.Read(p) + t.n += n + return n, err +} + +// slowLink889 dials addr as a client that reads its socket at bps in total, opens +// the connection window to 1 GiB and grants a stream's window back 4 KiB at a time +// as it reads it (Go's Transport policy), with default 64 KiB stream windows. +// skip, if not nil, says which streams it does not grant at all (paused ones), and +// dribble which get a 1-byte update every 100 ms instead. +func slowLink889(t *testing.T, addr string, bps float64, ids []uint32, size int, skip, dribble map[uint32]bool) (res map[uint32]string, got map[uint32]int, at map[uint32]time.Duration) { + t.Helper() + c := dialH2raw889(t, addr, h2dial889{rcvBuf: 64 << 10}) + _ = c.conn.SetReadDeadline(time.Time{}) + c.window889(0, 1<<30-65535) + fr := http2.NewFramer(c.conn, &throttled889{r: c.conn, bps: bps, t0: time.Now()}) + c.fr = fr + for id := range dribble { + id := id + stop := every889(100*time.Millisecond, func() { c.window889(id, 1) }) + defer stop() + } + for _, id := range ids { + c.request(id, "GET", "/inline889?n="+strconv.Itoa(size), true, true) + } + t0 := time.Now() + got, res, at = map[uint32]int{}, map[uint32]string{}, map[uint32]time.Duration{} + unacked := map[uint32]int{} + want := len(ids) + for len(res) < want && time.Since(t0) < 40*time.Second { + f, err := fr.ReadFrame() + if err != nil { + t.Logf("WR889 slow link: conn error %v after %v", err, time.Since(t0).Round(time.Millisecond)) + break + } + switch f := f.(type) { + case *http2.DataFrame: + id := f.StreamID + got[id] += len(f.Data()) + unacked[id] += len(f.Data()) + if f.StreamEnded() { + res[id], at[id] = "complete", time.Since(t0) + } else if unacked[id] >= 4096 && !skip[id] && !dribble[id] { + c.window889(id, uint32(unacked[id])) + unacked[id] = 0 + } + case *http2.RSTStreamFrame: + res[f.StreamID], at[f.StreamID] = "RST "+f.ErrCode.String(), time.Since(t0) + case *http2.PingFrame: + if !f.IsAck() { + c.wmu.Lock() + _ = fr.WritePing(true, f.Data) + c.wmu.Unlock() + } + } + } + return res, got, at +} + +// TestH2ParallelDownloadsSharingASlowLinkAreNotCut889: N downloads on one +// connection whose client reads its socket slowly get link/N each, below the +// per-stream floor, while the connection as a whole moves above it. Each is held +// as long as it takes part (a stream that gets nothing is bounded: see the next +// cell). On the previous head each was reset at 2.3 s. +func TestH2ParallelDownloadsSharingASlowLinkAreNotCut889(t *testing.T) { + type cell struct { + n int + bps float64 + size int + } + // The last two end their responses while the link is still shared: the grant for + // a stream's last bytes waits behind its siblings' bytes in the socket and lands + // a moment after the window ends, when the stream (or the whole connection) holds + // less than one floor and "released all of it" is the verdict (r8 F-A: reset 8 to + // 16 bytes before the end of the file, after the whole transfer). The last shape + // has a backlog of 2.6 s at the end (12 x 64 KiB on 300 KB/s), longer than one + // window: a window with nothing released inside the drain was the verdict (r9 + // F-A2, the same class at WriteTimeout 2 s and 5 s with larger backlogs). + for _, e := range nativeEngines889() { + for _, cl := range []cell{{2, 100e3, 256 << 10}, {4, 200e3, 256 << 10}, {2, 100e3, 384 << 10}, {8, 200e3, 128 << 10}, {12, 300e3, 256 << 10}} { + t.Run(fmt.Sprintf("%s/%dstreams/%.0fKBps/%dKiB", e.name, cl.n, cl.bps/1e3, cl.size>>10), func(t *testing.T) { + t.Parallel() + addr := wdServer889(t, e.eng, -1, -1, wd889) + var ids []uint32 + for i := 0; i < cl.n; i++ { + ids = append(ids, uint32(2*i+1)) + } + res, got, at := slowLink889(t, addr, cl.bps, ids, cl.size, nil, nil) + t.Logf("WR889 shared %s/%dstreams/%.0fKBps/%dKiB: %v %v at %v", e.name, cl.n, cl.bps/1e3, cl.size>>10, res, got, at) + for _, id := range ids { + if res[id] != "complete" || got[id] != cl.size { + t.Fatalf("stream %d on a link shared by %d streams (the connection moved %.0f KB/s, above the floor): %q after %d of %d bytes at %v", id, cl.n, cl.bps/1e3, res[id], got[id], cl.size, at[id].Round(time.Millisecond)) + } + } + }) + } + } +} + +// TestH2PausedAndDribbledStreamsOnAHealthyConnAreBounded889: on a connection that +// moves well above the floor a stream the client does not read at all (no +// WINDOW_UPDATE), or feeds a byte every 100 ms, is held for a few windows (it may be +// waiting behind its siblings in the same socket) and reset after that: three +// windows with nothing of its own, so the fourth judgement, 4 s in. Its sibling, +// read at 250 KB/s, goes on. On the previous head it was reset at 1.0 s. +func TestH2PausedAndDribbledStreamsOnAHealthyConnAreBounded889(t *testing.T) { + for _, e := range nativeEngines889() { + for _, mode := range []string{"paused", "dribbled"} { + t.Run(e.name+"/"+mode, func(t *testing.T) { + t.Parallel() + addr := wdServer889(t, e.eng, -1, -1, wd889) + skip, dribble := map[uint32]bool{3: true}, map[uint32]bool(nil) + if mode == "dribbled" { + skip, dribble = nil, map[uint32]bool{3: true} + } + res, got, at := slowLink889(t, addr, 250e3, []uint32{1, 3}, 2<<20, skip, dribble) + t.Logf("WR889 paused %s/%s: %v %v at %v", e.name, mode, res, got, at) + if res[3] != "RST INTERNAL_ERROR" || at[3] < 3500*time.Millisecond || at[3] > 7*time.Second { + t.Fatalf("stream 3 (%s): %q at %v, want RST INTERNAL_ERROR in [3.5 s, 7 s] (three windows of grace, then the verdict)", mode, res[3], at[3].Round(time.Millisecond)) + } + if r := res[1]; r != "" && r != "complete" { + t.Fatalf("the healthy sibling, stream 1, ended with %q", r) + } + }) + } + } +} + +// TestH2SilentClientWithASmallRemainderIsLetGo889: a client that opens its +// connection window, asks for a response a little longer than its stream window and +// never grants the stream (it reads its socket, sends no stream WINDOW_UPDATE). The +// stream's unsent rest is under one floor, and its connection must not be held for +// as long as the bytes in flight last at a few bytes per window (r9 F-H: 65,536 +// windows for the default window, 262,145 for 256 KiB or more; ReadTimeout and +// IdleTimeout do not close a connection with such a stream, and main closed this one at +// 1 s). With the three timeouts at 1 s the stream is reset and the connection closed +// within 5 (default window) or 8 (>= 256 KiB) windows, 1.7x that on io_uring (measured 9.1 s and 13.7 s, +// in a container with 2 CPUs), against a bound of 16 s and 24 s that a loaded runner can take. +func TestH2SilentClientWithASmallRemainderIsLetGo889(t *testing.T) { + for _, e := range nativeEngines889() { + for _, cl := range []struct { + iw uint32 + extra int + limit time.Duration + }{{65535, 1000, 16 * time.Second}, {1 << 20, 1000, 24 * time.Second}} { + t.Run(fmt.Sprintf("%s/W%dKiB/extra%d", e.name, cl.iw>>10, cl.extra), func(t *testing.T) { + t.Parallel() + addr := wdServer889(t, e.eng, time.Second, time.Second, time.Second) + c := dialH2raw889(t, addr, h2dial889{initialWindow: cl.iw}) + c.window889(0, 1<<30-65535) + c.request(1, "GET", "/inline889?n="+strconv.Itoa(int(cl.iw)+cl.extra), true, true) + t0 := time.Now() + _ = c.conn.SetReadDeadline(t0.Add(cl.limit + 10*time.Second)) + reset, closedAt := false, time.Duration(0) + for closedAt == 0 { + f, err := c.fr.ReadFrame() + if err != nil { + closedAt = time.Since(t0) + if isTimeout889(err) { + t.Fatalf("the connection was still open %v after a request for %d bytes that was never granted (reset=%v): a client holds a connection with one GET", closedAt.Round(time.Millisecond), int(cl.iw)+cl.extra, reset) + } + break + } + if r, ok := f.(*http2.RSTStreamFrame); ok && r.StreamID == 1 && r.ErrCode == http2.ErrCodeInternal { + reset = true + } + } + t.Logf("WR889 small remainder %s/W%dKiB: reset=%v, connection closed at %v", e.name, cl.iw>>10, reset, closedAt.Round(10*time.Millisecond)) + if !reset || closedAt > cl.limit { + t.Fatalf("reset=%v, closed at %v, want the stream reset and the connection closed within %v", reset, closedAt.Round(time.Millisecond), cl.limit) + } + }) + } + } +}