From 4399b86581712fedd9959d2266273a8773b4bf8b Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:34:30 +0100 Subject: [PATCH 1/2] test/server: an io_uring h2 stream whose chunks are further apart than WriteTimeout is not cut at a chunk (red on main; v1.6.0, Refs #889) 16 h2c streams, WriteTimeout 100 ms, a chunk every 250 ms (ticks 13 ms apart), ReadTimeout and IdleTimeout off, clients that read and send nothing: on main 4 to 10 of 16 are cut within 6 s on io_uring in 4 of 4 runs (a run in which the ticks fell on the same phase of the sweep cut none: hence the spread); epoll and adaptive stay open. The control: a client that requests 48 MiB and reads none of it is still cut on all three native engines. --- ...ouring_write_clock_quiet_889_linux_test.go | 250 ++++++++++++++++++ 1 file changed, 250 insertions(+) create mode 100644 test/server/iouring_write_clock_quiet_889_linux_test.go diff --git a/test/server/iouring_write_clock_quiet_889_linux_test.go b/test/server/iouring_write_clock_quiet_889_linux_test.go new file mode 100644 index 000000000..9d76b7fc4 --- /dev/null +++ b/test/server/iouring_write_clock_quiet_889_linux_test.go @@ -0,0 +1,250 @@ +//go:build linux + +package celeris_test + +import ( + "bytes" + "errors" + "fmt" + "io" + "net" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" + + "github.com/goceleris/celeris" +) + +// The write clock of an h2 stream on io_uring (celeris#889, L19). The sweep +// measures WriteTimeout, for a conn that has a send queued or in flight, from +// lastActivity: the last read or completed send. A stream that is quiet for +// longer than WriteTimeout between two chunks has a clock as old as the gap, so +// when a sweep (every 50 to 100 ms) finds the next chunk's SEND in flight, the +// conn is closed, mid-stream, with the chunk: measured with WriteTimeout 100 ms +// and a chunk every 250 ms, one chunk in 16 to 44 (3.75, 6.26 and 10.75 s in +// three runs), and once in CI with WriteTimeout 1 s and a gap of 2 s. epoll and +// adaptive have their own write clock (writeSince) and stay open. + +const ( + wtq889Timeout = 100 * time.Millisecond + wtq889Tick = 250 * time.Millisecond + wtq889Hold = 6 * time.Second + // wtq889Streams are the conns of one server: each chunk of each is a + // chance for a sweep to find its send in flight, so that the defect does + // not depend on one conn's luck (one stream is cut within 6 s in about + // half of the runs on main). Their ticks differ by 13 ms each, so that the + // chunks do not all fall on the same phase of the sweep. + wtq889Streams = 16 +) + +// wtqDial is a raw h2c client (prior knowledge) with one GET open. +func wtqDial(addr, path string, window uint32) (net.Conn, *http2.Framer, error) { + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err != nil { + return nil, nil, err + } + if _, err := conn.Write([]byte(http2.ClientPreface)); err != nil { + _ = conn.Close() + return nil, nil, err + } + fr := http2.NewFramer(conn, conn) + var settings []http2.Setting + if window != 0 { + settings = append(settings, http2.Setting{ID: http2.SettingInitialWindowSize, Val: window}) + } + if err := fr.WriteSettings(settings...); err != nil { + _ = conn.Close() + return nil, nil, err + } + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + for { + f, err := fr.ReadFrame() + if err != nil { + _ = conn.Close() + return nil, nil, fmt.Errorf("no SETTINGS from the server: %w", err) + } + if sf, ok := f.(*http2.SettingsFrame); ok && !sf.IsAck() { + if err := fr.WriteSettingsAck(); err != nil { + _ = conn.Close() + return nil, nil, err + } + break + } + } + var hb bytes.Buffer + enc := hpack.NewEncoder(&hb) + for _, f := range [][2]string{{":method", "GET"}, {":scheme", "http"}, {":authority", addr}, {":path", path}} { + _ = enc.WriteField(hpack.HeaderField{Name: f[0], Value: f[1]}) + } + if err := fr.WriteHeaders(http2.HeadersFrameParam{StreamID: 1, BlockFragment: hb.Bytes(), EndStream: true, EndHeaders: true}); err != nil { + _ = conn.Close() + return nil, nil, err + } + return conn, fr, nil +} + +// TestQuietH2StreamIsNotCutByWriteTimeout889: wtq889Streams h2c streams, each +// with a chunk every 250 ms and a client that reads them and sends nothing, +// WriteTimeout 100 ms (ReadTimeout and IdleTimeout off): every stream is +// still open after 6 s on every native engine. Red on main on io_uring. +func TestQuietH2StreamIsNotCutByWriteTimeout889(t *testing.T) { + for _, e := range engines761 { + if e.name == "std" { + continue // std's WriteTimeout is an absolute deadline per response (celeris#855) + } + t.Run(e.name, func(t *testing.T) { + t.Parallel() + cfg := celeris.Config{Engine: e.eng, ReadTimeout: -1, IdleTimeout: -1, WriteTimeout: wtq889Timeout} + addr := startServerConfig761(t, cfg, func(s *celeris.Server) { + s.GET("/tick889", func(c *celeris.Context) error { + sw := c.StreamWriter() + if sw == nil { + return errors.New("no StreamWriter") + } + if err := sw.WriteHeader(200, [][2]string{{"content-type", "text/event-stream"}}); err != nil { + return err + } + tick := wtq889Tick + if ms, err := strconv.Atoi(c.Query("ms")); err == nil && ms > 0 { + tick = time.Duration(ms) * time.Millisecond + } + tk := time.NewTicker(tick) + defer tk.Stop() + stop := time.After(2 * wtq889Hold) + done := c.Context().Done() + for { + if _, err := sw.Write([]byte("data: tick\n\n")); err != nil { + return nil + } + if err := sw.Flush(); err != nil { + return nil + } + select { + case <-done: + return nil + case <-stop: + return sw.Close() + case <-tk.C: + } + } + }).Async() + }) + var ( + wg sync.WaitGroup + cut atomic.Int32 + mu sync.Mutex + first string + chunks atomic.Int64 + ) + for i := 0; i < wtq889Streams; i++ { + wg.Add(1) + go func() { + defer wg.Done() + conn, fr, err := wtqDial(addr, "/tick889?ms="+strconv.Itoa(int(wtq889Tick/time.Millisecond)+13*i), 0) + if err != nil { + cut.Add(1) + mu.Lock() + if first == "" { + first = err.Error() + } + mu.Unlock() + return + } + defer func() { _ = conn.Close() }() + t0 := time.Now() + _ = conn.SetReadDeadline(t0.Add(wtq889Hold)) + for { + f, err := fr.ReadFrame() + if err != nil { + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + return // still open at the end of the hold + } + cut.Add(1) + mu.Lock() + if first == "" { + first = fmt.Sprintf("%v after %v", err, time.Since(t0).Round(time.Millisecond)) + if errors.Is(err, io.EOF) { + first = fmt.Sprintf("EOF after %v", time.Since(t0).Round(time.Millisecond)) + } + } + mu.Unlock() + return + } + if df, ok := f.(*http2.DataFrame); ok && df.StreamID == 1 { + chunks.Add(1) + } + } + }() + } + wg.Wait() + t.Logf("WTQ889 engine=%s: %d of %d streams cut within %v (WriteTimeout %v, a chunk every %v; %d chunks seen)%s", + e.name, cut.Load(), wtq889Streams, wtq889Hold, wtq889Timeout, wtq889Tick, chunks.Load(), map[bool]string{true: ", first: " + first}[first != ""]) + if n := cut.Load(); n > 0 { + t.Fatalf("%d of %d streams were cut within %v by WriteTimeout %v with a chunk every %v (first: %s)", n, wtq889Streams, wtq889Hold, wtq889Timeout, wtq889Tick, first) + } + }) + } +} + +// TestNonReadingH2ClientIsStillCutByWriteTimeout889: the other side of it: a +// client that requests a response far over what the sockets buffer and reads +// none of it is cut on the native engines: WriteTimeout (1 s) ends the conn, on +// io_uring after the 5 s its close waits for the queued SEND, so the client has +// seen the end of the conn by 12 s. (After 6 s io_uring is still draining: the +// response is delivered whole if the client reads then.) +func TestNonReadingH2ClientIsStillCutByWriteTimeout889(t *testing.T) { + for _, e := range engines761 { + if e.name == "std" { + continue + } + t.Run(e.name, func(t *testing.T) { + t.Parallel() + cfg := celeris.Config{Engine: e.eng, ReadTimeout: -1, IdleTimeout: -1, WriteTimeout: time.Second} + body := make([]byte, 48<<20) + addr := startServerConfig761(t, cfg, func(s *celeris.Server) { + s.GET("/big889", func(c *celeris.Context) error { return c.Blob(200, "application/octet-stream", body) }).Async() + }) + conn, fr, err := wtqDial(addr, "/big889", 1<<31-1) + if err != nil { + t.Fatal(err) + } + defer func() { _ = conn.Close() }() + // The stream window is 2^31-1 (SETTINGS) and so is the connection's + // (WINDOW_UPDATE), so that the server writes all it has, and then + // nothing is read: the socket fills, the SEND stalls, and the conn + // is cut at WriteTimeout. Reading after 6 s finds the bytes the + // kernel took and then the end of the conn; a conn that was not cut + // has the client's read time out (or the whole body). + if err := fr.WriteWindowUpdate(0, 1<<31-1-65535); err != nil { + t.Fatal(err) + } + t0 := time.Now() + time.Sleep(12 * time.Second) + _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + var got int64 + for { + f, err := fr.ReadFrame() + if err != nil { + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + t.Fatalf("the conn was still open 16 s after the request, with %.1f MiB of the 48 MiB taken and the client reading nothing (WriteTimeout 1 s)", float64(got)/(1<<20)) + } + t.Logf("WTSTALL889 engine=%s: the conn was cut (%v) after %.1f MiB of the 48 MiB were taken, the client having read nothing for 12 s", e.name, err, float64(got)/(1<<20)) + return + } + if df, ok := f.(*http2.DataFrame); ok { + got += int64(len(df.Data())) + if df.StreamEnded() { + t.Fatalf("the whole 48 MiB response was delivered to a client that read nothing for 12 s (WriteTimeout 1 s): not cut (%v)", time.Since(t0)) + } + } + } + }) + } +} From 58b6554c8763bb3540fa79886fb52b27d3d40062 Mon Sep 17 00:00:00 2001 From: Albert Bausili Date: Tue, 27 Oct 2026 09:34:30 +0100 Subject: [PATCH 2/2] iouring: the write clock of a conn starts when its send starts (v1.6.0, Refs #889) checkTimeouts measures WriteTimeout for a conn with a send queued or in flight from lastActivity, the last read or completed send. For a stream that was quiet for longer than WriteTimeout between two chunks that clock is as old as the gap, so a sweep that found the next chunk's SEND in flight closed the conn with the chunk. flushSend stamps lastActivity (cachedNow, a store of a clock the iteration already read) when the conn goes from idle to having output. A completed send already restamped the same clock, so what ReadTimeout and IdleTimeout mean is unchanged; a client that takes nothing is still cut. --- internal/engine/iouring/worker.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go index 23dfa8da7..27ca72f51 100644 --- a/internal/engine/iouring/worker.go +++ b/internal/engine/iouring/worker.go @@ -7200,6 +7200,15 @@ func (w *Worker) flushSend(cs *connState) bool { return false } + // The conn goes from idle to having output: the write clock starts here. + // lastActivity is the last read or completed send, and the sweep measures + // WriteTimeout for a conn with a send in flight from it, so for a stream + // that was quiet for longer than WriteTimeout the whole gap counted, and a + // sweep that found the next chunk's SEND in flight closed the conn with it + // (celeris#889). A store of the clock the iteration already read: the + // recv, the wake of a handler goroutine's write, or the sweep refreshed + // cachedNow before this ran. + cs.lastActivity = w.cachedNow cs.sendBuf, cs.writeBuf = cs.writeBuf, cs.sendBuf[:0] sqe := w.ring.GetSQE()