diff --git a/internal/engine/epoll/async_close_flush_1068_linux_test.go b/internal/engine/epoll/async_close_flush_1068_linux_test.go new file mode 100644 index 000000000..010bd4c4a --- /dev/null +++ b/internal/engine/epoll/async_close_flush_1068_linux_test.go @@ -0,0 +1,333 @@ +//go:build linux + +package epoll + +import ( + "bufio" + "context" + "errors" + "io" + "net" + "net/http" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/goceleris/celeris/internal/protocol/h2/stream" + "github.com/goceleris/celeris/internal/resource" +) + +// celeris#1068 (and the TestStreamedResponseIsDelivered/*/async-route/close +// item of #890): an async-route response that ends its connection while its +// remainder is still queued was cut short, at the same byte count each time +// (the bytes the socket buffers took at the first flush: 2634119 of a +// 3 MiB body, 2681685 of an 8 MiB one), and the client saw EOF. +// +// serveAsync, after a request that ends the conn (Connection: close, an error, +// a refused write), declared the conn closed (asyncClosed) and only then took +// asyncInMu to mark its goroutine gone (endDispatch). A drainDetachQueue that +// read the conn between the two found a conn declared closed whose goroutine +// was still counted as running, took that for a close requested from outside +// while a handler runs, and called closeConn: the handler's detachMu was free +// (the goroutine had released it after its flush), so closeConn went through +// and closed the descriptor with the rest of the response unsent. The entry +// that made the loop look was the one the goroutine enqueued for the remainder +// of its partial flush, before it declared the close. +// +// The test holds the goroutine in that gap, with the response's remainder +// queued, and waits for the loop to look. With the gap closed the loop cannot +// look until the goroutine has finished (the wait ends on its timeout), and +// sends the remainder before it closes. + +func TestAsyncRouteCloseSendsTheQueuedRemainder1068(t *testing.T) { + for _, tc := range []struct { + name string + // reqs requests are sent in one packet, header the lines after Host. + reqs int + header string + body int + // closesAtBoundary: the conn ends after whole responses only (the + // back-pressure refusal of celeris#761 drops the responses past the + // cap), so any number of them is right, but each one that is read + // must be whole, and then the stream ends (TestBackloggedPeerIsClosed + // takes an EOF in the next response's head as that end too). + closesAtBoundary bool + }{ + // TestStreamedResponseIsDelivered/*/async-route/close: one response + // larger than the socket buffers, then Connection: close. + {"connection-close", 1, "Connection: close\r\n", 8 << 20, false}, + // TestBackloggedPeerIsClosed/*/async-*: four pipelined 3 MiB responses + // for a client that reads nothing; the cap refuses one past 4 MiB. + {"refused-write", 4, "", 3 << 20, true}, + } { + t.Run(tc.name, func(t *testing.T) { + var ( + looked = make(chan struct{}) + lookOnce sync.Once + held = make(chan struct{}) + heldOnce sync.Once + ) + var queued atomic.Int64 + exitHook := func(cs *connState) { + queued.Store(int64(queued1068(cs))) + select { + case <-looked: + case <-time.After(500 * time.Millisecond): + } + heldOnce.Do(func() { close(held) }) + } + drainHook := func(*connState) { lookOnce.Do(func() { close(looked) }) } + testHookAsyncExitClosed.Store(&exitHook) + testHookDrainAsyncClosed.Store(&drainHook) + t.Cleanup(func() { + testHookAsyncExitClosed.Store(nil) + testHookDrainAsyncClosed.Store(nil) + }) + + _, addr := startE2E885(t, true, newPathHandler885(tc.body), func(*resource.Config) {}) + c, err := net.DialTimeout("tcp", addr, 3*time.Second) + if err != nil { + t.Fatal(err) + } + defer func() { _ = c.Close() }() + if _, err := io.WriteString(c, strings.Repeat("GET /big HTTP/1.1\r\nHost: x\r\n"+tc.header+"\r\n", tc.reqs)); err != nil { + t.Fatal(err) + } + // The client reads nothing until the goroutine has been held and + // let go, so the response cannot fit the socket buffers and its + // remainder is queued at every point of the hold. + select { + case <-held: + case <-time.After(10 * time.Second): + t.Fatal("the dispatch goroutine never reached the exit of a request that ends its conn") + } + // The arm proves nothing unless the remainder was queued while the + // goroutine was held (an 8 MiB response must outgrow the socket + // buffers of the host it runs on). + if q := queued.Load(); q <= 0 { + t.Fatalf("%d bytes were queued while the goroutine was held (-1: not readable): the arm would prove nothing", q) + } + _ = c.SetReadDeadline(time.Now().Add(20 * time.Second)) + br := bufio.NewReaderSize(c, 64<<10) + whole := 0 + for { + resp, err := http.ReadResponse(br, nil) + if err != nil { + if whole > 0 && tc.closesAtBoundary && (errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)) { + break + } + t.Fatalf("response %d: %v (looked=%v)", whole+1, err, isClosed1068(looked)) + } + n, err := io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + if err != nil || n != int64(tc.body) { + t.Fatalf("response %d: received %d of %d body bytes (%v): the loop closed the conn with the rest of the response unsent while the goroutine was still on its way out (looked=%v)", + whole+1, n, tc.body, err, isClosed1068(looked)) + } + whole++ + if !tc.closesAtBoundary || whole == tc.reqs { + break + } + } + if tc.closesAtBoundary && whole == tc.reqs { + t.Fatalf("all %d responses were sent to a client that read nothing: the cap did not refuse one", tc.reqs) + } + }) + } +} + +func isClosed1068(ch chan struct{}) bool { + select { + case <-ch: + return true + default: + return false + } +} + +// queued1068 is the bytes cs has queued for its socket, read under detachMu +// with TryLock only: the caller may hold asyncInMu, and the loop may hold +// detachMu for a flush. -1 if the lock stayed busy for 100 ms. +func queued1068(cs *connState) int { + for dl := time.Now().Add(100 * time.Millisecond); time.Now().Before(dl); time.Sleep(time.Millisecond) { + if cs.detachMu.TryLock() { + n := csPendingBytes(cs) + cs.detachMu.Unlock() + return n + } + } + return -1 +} + +// abortHandler1068: /big waits for proceed, then answers with a body larger +// than the socket buffers; /die panics, past every recovery (this handler is +// the stream.Handler itself). +type abortHandler1068 struct { + big []byte + entered chan struct{} + proceed chan struct{} + once sync.Once +} + +func (h *abortHandler1068) HandleStream(_ context.Context, s *stream.Stream) error { + if s.ResponseWriter == nil { + return nil + } + if s.Path == "/die" { + panic("celeris1068: handler panic") + } + h.once.Do(func() { close(h.entered) }) + <-h.proceed + return s.ResponseWriter.WriteResponse(s, 200, + [][2]string{{"content-type", "application/octet-stream"}, {"content-length", strconv.Itoa(len(h.big))}}, h.big) +} +func (*abortHandler1068) RouteAsync(_, _ string) bool { return true } +func (*abortHandler1068) HasAsyncRoutes() bool { return true } + +var _ stream.AsyncRouteResolver = (*abortHandler1068)(nil) + +// TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068: the same gap as above +// on the other way a dispatch goroutine ends a conn. abortAsyncHandler stores +// asyncClosed first, then lets go of detachMu and logs the panic with its +// stack, and only then calls endDispatch. A pipelining client's first response +// is partly sent and its remainder queued (the entry the goroutine enqueued +// for it is still in the detach queue) when the second request's handler +// panics; the loop that drains that entry in the window found a closed conn +// whose goroutine was running and closed it over the remainder. +// +// The second request is on the wire while the first handler still runs, so +// the goroutine takes it straight after the first flush without parking. The +// test holds the abort after its log and waits for the loop to look. +func TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068(t *testing.T) { + abortedGoroutine1068(t, false) +} + +// TestAbortedGoroutineFinishingBetweenTheDrainsReads1068: the loop decides +// whether the goroutine still runs, and whether to leave the close to it, in +// one asyncInMu section. Read in two (asyncRun, then dispatchBusy), a goroutine +// that reached endDispatch between the reads was taken for gone by the second +// and the conn was closed with closeConn, over the queued remainder, and the +// goroutine's own entry found it closed. The drain hook here holds the loop +// after its decision until the goroutine has finished. +func TestAbortedGoroutineFinishingBetweenTheDrainsReads1068(t *testing.T) { + abortedGoroutine1068(t, true) +} + +// abortedGoroutine1068 runs the scenario of the two tests above. The abort +// hook queues an entry for the loop itself, to force the window: the entry +// for the remainder is normally drained before the second request's handler +// has panicked. waitGone holds the loop's drain, after its decision, until the +// aborted goroutine has left (asyncRun false). +func abortedGoroutine1068(t *testing.T, waitGone bool) { + t.Helper() + // What stays queued (the body less what the socket takes, about 2.4 MB + // here) must be under the 4 MiB cap, or the second request is refused + // before its handler runs (ErrWriteBacklog) and the abort is never reached; + // and more than the kernel takes from the loop while it decides. + const body = 5 << 20 + var ( + looked = make(chan struct{}) + lookOnce sync.Once + held = make(chan struct{}) + heldOnce sync.Once + queued atomic.Int64 + ) + // The loop's look counts only once the goroutine is held: an earlier + // drain of the entry for the remainder finds the goroutine still holding + // detachMu (it has not released it before it logs), and leaves the close + // to it, as it always did. + var inWindow atomic.Bool + drainHook := func(cs *connState) { + if !inWindow.Load() { + return + } + lookOnce.Do(func() { close(looked) }) + if !waitGone { + return + } + for dl := time.Now().Add(2 * time.Second); time.Now().Before(dl); time.Sleep(time.Millisecond) { + cs.asyncInMu.Lock() + running := cs.asyncRun + cs.asyncInMu.Unlock() + if !running { + return + } + } + } + abortHook := func(l *Loop, cs *connState) { + queued.Store(int64(queued1068(cs))) + inWindow.Store(true) + // An entry naming cs reaches the loop while the goroutine is in the + // window. (The one enqueued for the remainder is normally drained + // before the handler of the second request has panicked; one that is + // drained late, or another, is all the loop needs to look.) + l.enqueueDetach(cs) + select { + case <-looked: + case <-time.After(500 * time.Millisecond): + } + heldOnce.Do(func() { close(held) }) + } + testHookAbortAsyncLogged.Store(&abortHook) + testHookDrainAsyncClosed.Store(&drainHook) + t.Cleanup(func() { + testHookAbortAsyncLogged.Store(nil) + testHookDrainAsyncClosed.Store(nil) + }) + + h := &abortHandler1068{big: newPathHandler885(body).big, entered: make(chan struct{}), proceed: make(chan struct{})} + e, addr := startE2E885(t, true, h, func(*resource.Config) {}) + // A small receive buffer keeps the kernel from taking the remainder on + // its own (the send buffer grows with the peer's window) while the loop + // is deciding. + c := dialRcvBuf885(t, addr, 16<<10) + req1 := "GET /big HTTP/1.1\r\nHost: x\r\n\r\n" + req2 := "GET /die HTTP/1.1\r\nHost: x\r\n\r\n" + if _, err := io.WriteString(c, req1); err != nil { + t.Fatal(err) + } + select { + case <-h.entered: + case <-time.After(10 * time.Second): + t.Fatal("the first handler never ran") + } + if _, err := io.WriteString(c, req2); err != nil { + t.Fatal(err) + } + // The loop has read the second request (into the goroutine's input) once + // the engine has counted both. + for dl := time.Now().Add(10 * time.Second); e.Metrics().BytesRead < uint64(len(req1)+len(req2)); time.Sleep(time.Millisecond) { + if time.Now().After(dl) { + t.Fatalf("the engine read %d of %d request bytes", e.Metrics().BytesRead, len(req1)+len(req2)) + } + } + close(h.proceed) + select { + case <-held: + case <-time.After(10 * time.Second): + t.Fatal("the dispatch goroutine never aborted") + } + // The abort hook waits only so long for the loop; a loop that never looked + // leaves nothing tested. + if !isClosed1068(looked) { + t.Fatal("the loop never drained the closed conn while the aborted goroutine was held: the test would prove nothing") + } + if q := queued.Load(); q <= 0 { + t.Fatalf("%d bytes were queued while the aborted goroutine was held (-1: not readable): the test would prove nothing", q) + } + _ = c.SetReadDeadline(time.Now().Add(20 * time.Second)) + br := bufio.NewReaderSize(c, 64<<10) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("response 1: %v (looked=%v)", err, isClosed1068(looked)) + } + n, err := io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + if err != nil || n != body { + t.Fatalf("response 1: received %d of %d body bytes (%v): the loop closed the conn over the remainder while the aborted goroutine was still on its way out (looked=%v)", + n, body, err, isClosed1068(looked)) + } +} diff --git a/internal/engine/epoll/listen_port0_895_linux_test.go b/internal/engine/epoll/listen_port0_895_linux_test.go index d91821432..1fb323055 100644 --- a/internal/engine/epoll/listen_port0_895_linux_test.go +++ b/internal/engine/epoll/listen_port0_895_linux_test.go @@ -5,6 +5,7 @@ package epoll import ( "context" "errors" + "fmt" "io" "log/slog" "net" @@ -37,6 +38,19 @@ import ( // local address has a port are counted. type sockets895 struct { bound, listening map[int]int + // owners says, for every port, what each socket counted in bound is: its + // descriptor, whether it listens, is connected (and to whom) or is only + // bound, and its SO_REUSEPORT. A census that finds a socket nobody expects + // (celeris#1104) then names it in the failure. + owners map[int][]string +} + +// owned895 is the owners of port, for a failure message. +func (s sockets895) owned895(port int) string { + if len(s.owners[port]) == 0 { + return "none" + } + return strings.Join(s.owners[port], "; ") } func census895(t *testing.T) sockets895 { @@ -45,7 +59,7 @@ func census895(t *testing.T) sockets895 { if err != nil { t.Fatalf("read /proc/self/fd: %v", err) } - s := sockets895{bound: map[int]int{}, listening: map[int]int{}} + s := sockets895{bound: map[int]int{}, listening: map[int]int{}, owners: map[int][]string{}} for _, ent := range ents { fd, err := strconv.Atoi(ent.Name()) if err != nil { @@ -73,13 +87,29 @@ func census895(t *testing.T) sockets895 { continue } s.bound[port]++ + state := "bound only" if v, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_ACCEPTCONN); err == nil && v == 1 { s.listening[port]++ + state = "listening" + } else if peer, err := unix.Getpeername(fd); err == nil { + state = fmt.Sprintf("connected to %s", sockaddrString895(peer)) } + reuse, _ := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT) + s.owners[port] = append(s.owners[port], fmt.Sprintf("fd %d %s (SO_REUSEPORT=%d)", fd, state, reuse)) } return s } +func sockaddrString895(sa unix.Sockaddr) string { + switch a := sa.(type) { + case *unix.SockaddrInet4: + return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port)) + case *unix.SockaddrInet6: + return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port)) + } + return fmt.Sprintf("%T", sa) +} + func ports895(m map[int]int) []int { var ps []int for p := range m { diff --git a/internal/engine/epoll/listen_port0_hold_895_linux_test.go b/internal/engine/epoll/listen_port0_hold_895_linux_test.go index 403bf7a5b..cc53f2914 100644 --- a/internal/engine/epoll/listen_port0_hold_895_linux_test.go +++ b/internal/engine/epoll/listen_port0_hold_895_linux_test.go @@ -89,8 +89,8 @@ func TestPort0IsHeldBetweenBeingChosenAndBeingBound895(t *testing.T) { if err := r.stop(t); err != nil { t.Fatalf("Listen: %v", err) } - if n := census895(t).bound[port]; n != 0 { - t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included)", n, port) + if c := census895(t); c.bound[port] != 0 { + t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included): %s", c.bound[port], port, c.owned895(port)) } } @@ -122,8 +122,8 @@ func TestPort0HoldIsReleasedWhenListenFails895(t *testing.T) { t.Fatal("the hook never ran") } port := portOfString895(t, addr) - if n := census895(t).bound[port]; n != 0 { - t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0", n, port) + if c := census895(t); c.bound[port] != 0 { + t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0: %s", c.bound[port], port, c.owned895(port)) } ln, err := net.Listen("tcp", addr) if err != nil { diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go index ec8b9774d..63078e554 100644 --- a/internal/engine/epoll/loop.go +++ b/internal/engine/epoll/loop.go @@ -2034,6 +2034,25 @@ func (l *Loop) drainH2Writes() { // and a nil test per close request, which is never a hot path. var testHookCloseWhenFlushed atomic.Pointer[func(*connState)] +// testHookAsyncExitClosed, when set, runs in serveAsync's exit for a request +// that ends its connection (Connection: close, an error, a refused write), +// right after the goroutine has declared the conn closed (asyncClosed) and +// before it has marked itself gone (endDispatch), with asyncInMu held for both: +// the loop cannot read the two apart. testHookDrainAsyncClosed runs on the +// loop's goroutine in drainDetachQueue when an entry names a conn declared +// closed whose goroutine is still counted as running, before the loop decides +// whether to close it. A test that holds the first and waits on the second +// sees the loop in the gap the lock closes (celeris#1068). Atomic pointers, as testHookCloseWhenFlushed is; a load and a +// nil test per close, never a hot path. +var testHookAsyncExitClosed atomic.Pointer[func(*connState)] +var testHookDrainAsyncClosed atomic.Pointer[func(*connState)] + +// testHookAbortAsyncLogged, when set, runs in abortAsyncHandler after its log +// record and before endDispatch, holding no lock: the one place an aborted +// goroutine is closed and still counted as running (celeris#1068). It gets the +// loop so that a test can queue an entry for the loop to drain in that place. +var testHookAbortAsyncLogged atomic.Pointer[func(*Loop, *connState)] + // discardWrite is the write function the run loop drains the H2 write queue of // a conn that is closing on a refused write into: the frames go nowhere. func discardWrite([]byte) {} @@ -2346,11 +2365,19 @@ func dispatchBusy(cs *connState, owe *bool) bool { return false // no async machinery: sync mode, no dispatch goroutine } cs.asyncInMu.Lock() + busy := dispatchBusyLocked(cs, owe) + cs.asyncInMu.Unlock() + return busy +} + +// dispatchBusyLocked is dispatchBusy's test, for a caller that holds +// cs.asyncInMu and needs it in the same section as another read of cs +// (drainDetachQueue's asyncClosed branch). +func dispatchBusyLocked(cs *connState, owe *bool) bool { busy := cs.asyncRun && !cs.asyncParked && !cs.asyncDetachUnlocked if busy && owe != nil { *owe = true } - cs.asyncInMu.Unlock() return busy } @@ -3245,8 +3272,14 @@ func (l *Loop) abortAsyncHandler(cs *connState, r any, holds *dispatchHolds) { ) } } - cs.asyncClosed.Store(true) + if h := testHookAbortAsyncLogged.Load(); h != nil { + (*h)(l, cs) + } + // Stored again, under the lock with endDispatch as at serveAsync's exits + // (the first store, above, is the one that matters here: it is early on + // purpose). cs.asyncInMu.Lock() + cs.asyncClosed.Store(true) cs.asyncInBuf = cs.asyncInBuf[:0] cs.endDispatch() // enqueued below: that is the hand-back cs.asyncInMu.Unlock() @@ -3411,8 +3444,10 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) { holds.detachMu = false cs.detachMu.Unlock() if promoteErr != nil { - cs.asyncClosed.Store(true) + // asyncClosed under asyncInMu, with endDispatch (see the exit + // at the end of this loop, celeris#1068). cs.asyncInMu.Lock() + cs.asyncClosed.Store(true) cs.asyncInBuf = cs.asyncInBuf[:0] cs.endDispatch() // enqueued below: that is the hand-back cs.asyncInMu.Unlock() @@ -3438,8 +3473,10 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) { // considers a hijacked conn "done with the request". Treat // it like nil here — the middleware now owns the conn. if processErr != nil && !errors.Is(processErr, conn.ErrHijacked) { - cs.asyncClosed.Store(true) + // asyncClosed under asyncInMu, with endDispatch (see the exit + // at the end of this loop, celeris#1068). cs.asyncInMu.Lock() + cs.asyncClosed.Store(true) cs.asyncInBuf = cs.asyncInBuf[:0] cs.endDispatch() // enqueued below: that is the hand-back cs.asyncInMu.Unlock() @@ -3506,8 +3543,21 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) { // on detachQueue, and write eventfd; the worker's // drainDetachQueue pass will observe asyncClosed and run // closeConn thread-safely. - cs.asyncClosed.Store(true) + // + // asyncClosed is stored under asyncInMu, with endDispatch: the loop + // reads asyncClosed and asyncRun together under that lock + // (drainDetachQueue), and a conn declared closed whose goroutine + // still counts as running is, to it, a close requested from outside + // while a handler runs, which closeConn carries out at once when + // detachMu is free, as it is here (the flush above is done). The + // entry enqueued for a partial flush's remainder could make it + // look in the gap between the two stores, and the conn was closed + // with that remainder unsent (celeris#1068). cs.asyncInMu.Lock() + cs.asyncClosed.Store(true) + if h := testHookAsyncExitClosed.Load(); h != nil { + (*h)(cs) + } cs.asyncInBuf = cs.asyncInBuf[:0] cs.endDispatch() // enqueued below: that is the hand-back cs.asyncInMu.Unlock() @@ -3624,10 +3674,28 @@ func (l *Loop) drainDetachQueue() { // than the socket buffers. While it still runs, closeConn leaves // the close to it (celeris#669), as before. if cs.asyncClosed.Load() { + // A goroutine that is still running a request is not done with + // the conn: the close is its to hand back (closeOwed), as + // closeConn does for a handler that holds detachMu. A goroutine + // that declared the conn closed itself and is on its way out + // (abortAsyncHandler logs between its store of asyncClosed and + // endDispatch) would otherwise be closed over, and the response + // it left queued cut short (celeris#1068). Its exit always + // enqueues cs again, and that entry finds it gone. Whether it + // runs, whether it is busy and the owed record are one section: + // read apart, a goroutine that finished between the reads was + // taken for gone by the second and closed over by closeConn. cs.asyncInMu.Lock() running := cs.asyncRun + busy := dispatchBusyLocked(cs, &cs.closeOwed) cs.asyncInMu.Unlock() if running { + if h := testHookDrainAsyncClosed.Load(); h != nil { + (*h)(cs) + } + if busy { + continue + } l.closeConn(cs.fd) } else { l.closeWhenFlushed(cs) diff --git a/internal/engine/epoll/pin_allowed_909_linux_test.go b/internal/engine/epoll/pin_allowed_909_linux_test.go index 11b4350c0..8ef77c338 100644 --- a/internal/engine/epoll/pin_allowed_909_linux_test.go +++ b/internal/engine/epoll/pin_allowed_909_linux_test.go @@ -5,6 +5,8 @@ package epoll import ( "bytes" "context" + "errors" + "fmt" "log/slog" "net" "strings" @@ -38,13 +40,33 @@ func (l *lockedBuf) String() string { return l.b.String() } +// logSink is where start909's engine logs and its caller reads back. +type logSink interface { + Write(p []byte) (int, error) + String() string +} + // start909 starts an epoll engine with the given worker count, logging into // log, and returns it with its stop function. -func start909(t *testing.T, workers int, log *lockedBuf) (*Engine, func()) { +func start909(t *testing.T, workers int, log logSink) (*Engine, func()) { t.Helper() + eng, stop, err := launch909(workers, log) + if err != nil { + t.Fatal(err) + } + return eng, func() { + if err := stop(); err != nil { + t.Error(err) + } + } +} + +// launch909 is start909 without a *testing.T, so that a goroutine other than +// the test's may start an engine and report why it could not. +func launch909(workers int, log logSink) (*Engine, func() error, error) { ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { - t.Fatalf("pick port: %v", err) + return nil, nil, fmt.Errorf("pick port: %w", err) } addr := ln.Addr().String() _ = ln.Close() @@ -55,7 +77,7 @@ func start909(t *testing.T, workers int, log *lockedBuf) (*Engine, func()) { Logger: slog.New(slog.NewTextHandler(log, nil)), }, respondingHandler{}) if err != nil { - t.Fatalf("New: %v", err) + return nil, nil, fmt.Errorf("New: %w", err) } ctx, cancel := context.WithCancel(context.Background()) done := make(chan error, 1) @@ -64,26 +86,36 @@ func start909(t *testing.T, workers int, log *lockedBuf) (*Engine, func()) { select { case err := <-done: cancel() - t.Fatalf("Listen returned before it bound: %v", err) + return nil, nil, fmt.Errorf("Listen returned before it bound: %w", err) default: } if time.Now().After(deadline) { cancel() - t.Fatal("the epoll engine did not bind within 10s") + return nil, nil, errors.New("the epoll engine did not bind within 10s") } time.Sleep(time.Millisecond) } - return eng, func() { + // Listen publishes the address and only then logs its listening line + // (celeris#890): a caller that reads the log on return must find it. + for deadline := time.Now().Add(10 * time.Second); !strings.Contains(log.String(), "epoll engine listening"); { + if time.Now().After(deadline) { + cancel() + return nil, nil, errors.New("the epoll engine did not log its listening line within 10s") + } + time.Sleep(time.Millisecond) + } + return eng, func() error { cancel() select { case err := <-done: if err != nil { - t.Errorf("Listen: %v", err) + return fmt.Errorf("Listen: %w", err) } + return nil case <-time.After(15 * time.Second): - t.Fatal("the epoll engine did not stop within 15s") + return errors.New("the epoll engine did not stop within 15s") } - } + }, nil } // celeris#909: with the process restricted to a mask that leaves out its lowest @@ -190,3 +222,64 @@ func TestFailedPinIsLoggedOnceAndNotPretended909(t *testing.T) { } _ = conn.Close() } + +// gatedLog holds the write of the engine's listening line until gate is +// closed. +type gatedLog struct { + lockedBuf + gate chan struct{} +} + +func (g *gatedLog) Write(p []byte) (int, error) { + if bytes.Contains(p, []byte("epoll engine listening")) { + <-g.gate + } + return g.lockedBuf.Write(p) +} + +// celeris#890 (TestNICIRQReportAddsNoLoop1036 failed in CI with a log that had +// the pinning lines and no "epoll engine listening" line): Listen publishes +// the address and only then logs that line, and start909 returned as soon as +// the address was there, so a test that read the log on return could read it +// before the line was written. start909 returns once the line is in the log. +func TestStart909ReturnsOnlyOnceTheListeningLineIsLogged(t *testing.T) { + g := &gatedLog{gate: make(chan struct{})} + type started struct { + eng *Engine + stop func() error + err error + } + ret := make(chan started, 1) + go func() { + eng, stop, err := launch909(2, g) + ret <- started{eng, stop, err} + }() + select { + case r := <-ret: + logged := g.String() + close(g.gate) + if r.err != nil { + t.Fatalf("launch909: %v", r.err) + } + _ = r.stop() + t.Fatalf("launch909 returned (address %v) before the listening line was in the log: %q", r.eng.Addr(), logged) + case <-time.After(300 * time.Millisecond): + } + close(g.gate) + select { + case r := <-ret: + if r.err != nil { + t.Fatalf("launch909: %v", r.err) + } + defer func() { + if err := r.stop(); err != nil { + t.Error(err) + } + }() + if !strings.Contains(g.String(), "epoll engine listening") { + t.Fatalf("launch909 returned without the listening line in the log: %q", g.String()) + } + case <-time.After(20 * time.Second): + t.Fatal("launch909 did not return once the listening line was logged") + } +} diff --git a/internal/engine/iouring/fd_lifetime_engine_test.go b/internal/engine/iouring/fd_lifetime_engine_test.go index 8df27b8e3..a4fdc2aae 100644 --- a/internal/engine/iouring/fd_lifetime_engine_test.go +++ b/internal/engine/iouring/fd_lifetime_engine_test.go @@ -163,6 +163,17 @@ type loadResult struct { // the server never answers surfaces as that client's read timeout; the client // then stops, so each lost request is exactly one error. func runKeepAliveLoad(t *testing.T, addr string, n int, stop <-chan struct{}) func() loadResult { + t.Helper() + return runKeepAliveLoadPaced(t, addr, n, stop, 0) +} + +// runKeepAliveLoadPaced is runKeepAliveLoad with think between a client's +// answered request and its next (0: none, back to back). A worker serving +// clients that never wait finds a completion ready at every pass of its loop +// and takes no wait at all, or only the one that has sends to wait for; with +// think the clients leave it idle between requests, and the wait it takes +// then is the timed one (celeris#884 item 49, the test of that wait). +func runKeepAliveLoadPaced(t *testing.T, addr string, n int, stop <-chan struct{}, think time.Duration) func() loadResult { t.Helper() var ok, errsN atomic.Int64 var mu sync.Mutex @@ -216,6 +227,13 @@ func runKeepAliveLoad(t *testing.T, addr string, n int, stop <-chan struct{}) fu return } ok.Add(1) + if think > 0 { + select { + case <-stop: + return + case <-time.After(think): + } + } } }() } diff --git a/internal/engine/iouring/listen_port0_895_linux_test.go b/internal/engine/iouring/listen_port0_895_linux_test.go index 31110c85b..38cce788f 100644 --- a/internal/engine/iouring/listen_port0_895_linux_test.go +++ b/internal/engine/iouring/listen_port0_895_linux_test.go @@ -5,6 +5,7 @@ package iouring import ( "context" "errors" + "fmt" "io" "log/slog" "net" @@ -43,6 +44,19 @@ import ( // local address has a port are counted. type sockets895 struct { bound, listening map[int]int + // owners says, for every port, what each socket counted in bound is: its + // descriptor, whether it listens, is connected (and to whom) or is only + // bound, and its SO_REUSEPORT. A census that finds a socket nobody expects + // (celeris#1104) then names it in the failure. + owners map[int][]string +} + +// owned895 is the owners of port, for a failure message. +func (s sockets895) owned895(port int) string { + if len(s.owners[port]) == 0 { + return "none" + } + return strings.Join(s.owners[port], "; ") } func census895(t *testing.T) sockets895 { @@ -51,7 +65,7 @@ func census895(t *testing.T) sockets895 { if err != nil { t.Fatalf("read /proc/self/fd: %v", err) } - s := sockets895{bound: map[int]int{}, listening: map[int]int{}} + s := sockets895{bound: map[int]int{}, listening: map[int]int{}, owners: map[int][]string{}} for _, ent := range ents { fd, err := strconv.Atoi(ent.Name()) if err != nil { @@ -79,13 +93,29 @@ func census895(t *testing.T) sockets895 { continue } s.bound[port]++ + state := "bound only" if v, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_ACCEPTCONN); err == nil && v == 1 { s.listening[port]++ + state = "listening" + } else if peer, err := unix.Getpeername(fd); err == nil { + state = fmt.Sprintf("connected to %s", sockaddrString895(peer)) } + reuse, _ := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT) + s.owners[port] = append(s.owners[port], fmt.Sprintf("fd %d %s (SO_REUSEPORT=%d)", fd, state, reuse)) } return s } +func sockaddrString895(sa unix.Sockaddr) string { + switch a := sa.(type) { + case *unix.SockaddrInet4: + return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port)) + case *unix.SockaddrInet6: + return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port)) + } + return fmt.Sprintf("%T", sa) +} + func ports895(m map[int]int) []int { var ps []int for p := range m { diff --git a/internal/engine/iouring/listen_port0_hold_895_linux_test.go b/internal/engine/iouring/listen_port0_hold_895_linux_test.go index 5c787181d..f38eb2d9a 100644 --- a/internal/engine/iouring/listen_port0_hold_895_linux_test.go +++ b/internal/engine/iouring/listen_port0_hold_895_linux_test.go @@ -107,8 +107,8 @@ func TestPort0IsHeldBetweenBeingChosenAndBeingBound895(t *testing.T) { if err := r.stop(t); err != nil { t.Fatalf("Listen: %v", err) } - if n := census895(t).bound[port]; n != 0 { - t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included)", n, port) + if c := census895(t); c.bound[port] != 0 { + t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included): %s", c.bound[port], port, c.owned895(port)) } } @@ -144,8 +144,8 @@ func TestPort0HoldIsReleasedWhenListenFails895(t *testing.T) { t.Fatal("the hook never ran") } port := portOfString895(t, addr) - if n := census895(t).bound[port]; n != 0 { - t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0", n, port) + if c := census895(t); c.bound[port] != 0 { + t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0: %s", c.bound[port], port, c.owned895(port)) } // bindWithin656, not an immediate bind: the sockets of a worker that // was cancelled are let go of asynchronously on a main without the diff --git a/internal/engine/iouring/run_ebusy_884_linux_test.go b/internal/engine/iouring/run_ebusy_884_linux_test.go index ae2486af4..801c2d573 100644 --- a/internal/engine/iouring/run_ebusy_884_linux_test.go +++ b/internal/engine/iouring/run_ebusy_884_linux_test.go @@ -134,10 +134,10 @@ func setRingWaitSeam884(t *testing.T, fn func(*Worker, ringWaitMode, error) erro // warmLoad884 dials n keep-alive connections, each answered once, so every // worker that will own one does; the returned stop ends the load the // connections then carry and reports what the clients saw. -func warmLoad884(t *testing.T, addr string, n int) (stop func() loadResult) { +func warmLoad884(t *testing.T, addr string, n int, think time.Duration) (stop func() loadResult) { t.Helper() ch := make(chan struct{}) - wait := runKeepAliveLoad(t, addr, n, ch) + wait := runKeepAliveLoadPaced(t, addr, n, ch, think) return func() loadResult { close(ch) return wait() @@ -151,10 +151,19 @@ func TestRunSurvivesARingWaitEBUSY884(t *testing.T) { for _, tc := range []struct { name string mode int32 // -1 any + // think is the clients' pause between requests. The timed wait is + // taken only when the loop finds no completion ready and no send to + // wait for, which 32 clients that send the next request at once keep + // from happening for the whole window: on a worker with a CPU to + // itself it did not happen once in 3 s of 79 000 requests, and the + // apparatus guard below failed 9 times in Coverage's -race + // -covermode=atomic build (celeris#890). The other two modes are the + // ones that load always reaches. + think time.Duration }{ - {"any-wait", -1}, - {"submit-and-wait-timeout", int32(waitModeTimed)}, - {"submit-and-wait-sends-pending", int32(waitModeSubmitSends)}, + {"any-wait", -1, 0}, + {"submit-and-wait-timeout", int32(waitModeTimed), 20 * time.Millisecond}, + {"submit-and-wait-sends-pending", int32(waitModeSubmitSends), 0}, } { t.Run(tc.name, func(t *testing.T) { inj := &ebusyInject884{err: ebusyErr884(), n: 40} @@ -162,7 +171,7 @@ func TestRunSurvivesARingWaitEBUSY884(t *testing.T) { setRingWaitSeam884(t, inj.seam) e, addr := startFDLEngine(t, transplantTestHandler{}, nil) - stop := warmLoad884(t, addr, 32) + stop := warmLoad884(t, addr, 32, tc.think) time.Sleep(200 * time.Millisecond) // every connection answers, un-injected inj.armed.Store(true) deadline := time.Now().Add(3 * time.Second) @@ -211,7 +220,7 @@ func TestRunStopsOnARingWaitFailure884(t *testing.T) { setRingWaitSeam884(t, inj.seam) e, addr := startFDLEngine(t, transplantTestHandler{}, nil) - stop := warmLoad884(t, addr, 32) + stop := warmLoad884(t, addr, 32, 0) time.Sleep(200 * time.Millisecond) inj.armed.Store(true) deadline := time.Now().Add(3 * time.Second)