adaptive: error-rate revert floor of 3 faults, ioUringViable counts the started loops, slotsMu/Workers/bind-wait tests, the four 2-CPU test failures (v1.6.0)

adaptive_settled_retime_bar_886_linux_test.go

@@ -3,7 +3,9 @@
 package celeris
 
 import (
+	"runtime"
 	"strings"
+	"sync"
 	"testing"
 )
 
@@ -175,3 +177,70 @@ created by github.com/goceleris/celeris/internal/engine/iouring.(*Engine).Listen
 		})
 	}
 }
+
+// TestClassifyLoopStackReadsEveryMutexWait589 (celeris#886 item 27): the
+// classifier counted only a sync.(*Mutex).Lock frame. A worker parked on an
+// RWMutex, or one whose Lock frame is not in the dump but whose header carries
+// the wait reason, is as parked as the verbatim celeris#593 one. The stacks are
+// synthetic, built on the shape of that verbatim dump.
+func TestClassifyLoopStackReadsEveryMutexWait589(t *testing.T) {
+	loop := func(header, frames string) string {
+		return header + "\n" + frames +
+			"github.com/goceleris/celeris/internal/engine/epoll.(*Loop).run(0xc000123456, {0xdd3928, 0xc000628870})\n" +
+			"\t/src/internal/engine/epoll/loop.go:100 +0x1688"
+	}
+	for _, tc := range []struct {
+		name, stack, want string
+	}{
+		{"Mutex frame", loop("goroutine 7 [sync.Mutex.Lock]:", "sync.(*Mutex).Lock(...)\n\t/usr/local/go/src/sync/mutex.go:46\n"), "other_Mutex_Lock"},
+		{"RWMutex write lock frame", loop("goroutine 7 [sync.RWMutex.Lock]:", "sync.(*RWMutex).Lock(0xc0001)\n\t/usr/local/go/src/sync/rwmutex.go:150 +0x40\n"), "other_Mutex_Lock"},
+		{"RWMutex read lock frame", loop("goroutine 7 [sync.RWMutex.RLock]:", "sync.(*RWMutex).RLock(...)\n\t/usr/local/go/src/sync/rwmutex.go:70\n"), "other_Mutex_Lock"},
+		{"wait reason in the header only", loop("goroutine 7 [sync.RWMutex.RLock, locked to thread]:", ""), "other_Mutex_Lock"},
+		{"Mutex wait reason in the header only", loop("goroutine 7 [sync.Mutex.Lock, 3 minutes]:", ""), "other_Mutex_Lock"},
+		{"checkTimeouts on an RWMutex", loop("goroutine 7 [sync.RWMutex.RLock]:", "sync.(*RWMutex).RLock(...)\ngithub.com/goceleris/celeris/internal/engine/epoll.(*Loop).checkTimeouts(0xc0001)\n"), "checkTimeouts_detachMu_Lock"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if !isLoopGoroutine589(tc.stack) {
+				t.Fatalf("the synthetic stack is not an event loop to isLoopGoroutine589:\n%s", tc.stack)
+			}
+			got := classifyLoopStack589(tc.stack)
+			if got != tc.want || !mutexParked589(got) {
+				t.Errorf("classifyLoopStack589 = %q (parked %v), want %q (parked)", got, mutexParked589(got), tc.want)
+			}
+		})
+	}
+	running := loop("goroutine 7 [running]:", "")
+	if got := classifyLoopStack589(running); mutexParked589(got) {
+		t.Errorf("a running loop is classified %q, parked", got)
+	}
+}
+
+// TestFullStackIsNotCutAtItsFirstBuffer589: capture used one fixed 1 MiB buffer
+// and read what fitted. A process with enough goroutines to overflow it lost
+// the tail of the dump. Park enough goroutines to need more than 1 MiB, and
+// require every one of them in the dump.
+func TestFullStackIsNotCutAtItsFirstBuffer589(t *testing.T) {
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	const n = 20000
+	for range n {
+		wg.Add(1)
+		go func() { defer wg.Done(); <-stop }()
+	}
+	defer func() { close(stop); wg.Wait() }()
+
+	before := runtime.NumGoroutine()
+	dump := fullStack589()
+	after := runtime.NumGoroutine()
+	if len(dump) <= 1<<20 {
+		t.Fatalf("%d parked goroutines produced a %d byte dump, not over the 1 MiB first buffer: the test does not reach the growth", n, len(dump))
+	}
+	headers := strings.Count(string(dump), "\ngoroutine ") + 1
+	// The lower bound is what the test is for. The upper one only says the
+	// dump is a dump of this process, and goroutines left by an earlier test
+	// can exit while it is taken, so it is read on both sides of the dump.
+	if hi := max(before, after) + 50; headers < n || headers > hi {
+		t.Errorf("the dump holds %d goroutines, want at least the %d parked ones and at most %d (%d before the dump, %d after)",
+			headers, n, hi, before, after)
+	}
+}

adaptive_settled_retime_linux_test.go

@@ -52,11 +52,28 @@ func mutexParked589(class string) bool {
 	return class == "checkTimeouts_detachMu_Lock" || class == "other_Mutex_Lock"
 }
 
+// parkedOnMutex589 reports whether a goroutine dump is a goroutine waiting for
+// a sync mutex: a frame in a Lock/RLock of sync.Mutex or sync.RWMutex, or the
+// wait reason in the goroutine header ("[sync.Mutex.Lock", "[sync.RWMutex.Lock",
+// "[sync.RWMutex.RLock"), which survives the inlining that can hide the Lock
+// frame (celeris#886 item 27).
+func parkedOnMutex589(g string) bool {
+	for _, m := range [...]string{
+		"sync.(*Mutex).Lock", "sync.(*RWMutex).Lock", "sync.(*RWMutex).RLock",
+		"[sync.Mutex.Lock", "[sync.RWMutex.Lock", "[sync.RWMutex.RLock",
+	} {
+		if strings.Contains(g, m) {
+			return true
+		}
+	}
+	return false
+}
+
 func classifyLoopStack589(g string) string {
 	switch {
-	case strings.Contains(g, "checkTimeouts") && strings.Contains(g, "sync.(*Mutex).Lock"):
+	case strings.Contains(g, "checkTimeouts") && parkedOnMutex589(g):
 		return "checkTimeouts_detachMu_Lock"
-	case strings.Contains(g, "sync.(*Mutex).Lock"):
+	case parkedOnMutex589(g):
 		return "other_Mutex_Lock"
 	case strings.Contains(g, "SubmitAndWait") || strings.Contains(g, "EpollWait"):
 		return "kernel_wait" // io_uring_enter / epoll_wait: the loop is idle, not pinned
@@ -86,13 +103,26 @@ func isLoopGoroutine589(g string) bool {
 	return strings.Contains(g, "internal/engine/iouring.(*Worker).run(") || strings.Contains(g, "internal/engine/epoll.(*Loop).run(")
 }
 
+// fullStack589 is the stack of every goroutine. runtime.Stack truncates
+// silently when the buffer is full, and a dump cut short could lose the stack of
+// the pinned worker and read as not parked, so the buffer grows until the dump
+// fits (celeris#886 item 27).
+func fullStack589() []byte {
+	buf := make([]byte, 1<<20)
+	for {
+		if n := runtime.Stack(buf, true); n < len(buf) {
+			return buf[:n]
+		}
+		buf = make([]byte, 2*len(buf))
+	}
+}
+
 func (st *stackTally589) capture(t *testing.T) {
 	t.Helper()
-	buf := make([]byte, 1<<20)
-	n := runtime.Stack(buf, true)
+	dump := fullStack589()
 	st.captures++
 	parkedHere := false
-	for _, g := range strings.Split(string(buf[:n]), "\n\n") {
+	for _, g := range strings.Split(string(dump), "\n\n") {
 		if !isLoopGoroutine589(g) {
 			continue
 		}
@@ -610,11 +640,15 @@ func assertSettledStall589(o stall589Obs) error {
 // re-timed, so a settled route whose store turns slow is promoted and stops
 // running on the engine worker. State is asserted on both engines. The latency
 // observable differs by engine: epoll is judged on the ratio below, io_uring on
-// the fraction of post-promotion samples that waited for the blocked Set,
-// because io_uring separately pinned its worker in the timeout sweep even when
-// every conn is async-dispatched (celeris#593; fixed on main by PR #604). A
-// pinned worker shows that fraction at 0.3 and the fixed tree at 0, so it
-// needs no reference window and is not subject to the VOID case.
+// the worker's own stack (ioUringPin592): no event-loop goroutine parked on a
+// mutex in a quarter or more of at least stall592PinMinCaptures captures taken
+// while a /ping sat unanswered for 100 ms. io_uring needs that because it
+// separately pinned its worker in the timeout sweep even when every conn is
+// async-dispatched (celeris#593; fixed on main by PR #604). The stack read
+// needs no reference window and is not subject to the VOID case. It replaced a
+// bar on the fraction of post-promotion samples that waited for the blocked
+// Set (0.3 pinned, 0 fixed), which a starved box moved as much as the pin did
+// (celeris#886 item 1).
 //
 // The latency observable is RELATIVE (celeris#622). What the settled arm
 // claims is that a re-timed route stops costing the worker — not that a /ping
@@ -1378,6 +1412,39 @@ func runStall589(t *testing.T, engName string, engType EngineType, mode stall589
 		return time.Since(t0), nil
 	}
 
+	// pingWatched is pingOnce on the shape the stack watch needs: the request
+	// runs on its own goroutine and a 100 ms timer waits for it, and a probe
+	// still unanswered when the timer fires calls onStall (nil: nothing to do)
+	// before waiting out the answer. BOTH windows probe through it, with the
+	// same goroutine and the same timer per probe, so that the two medians
+	// differ only in the engine state they observed (celeris#622, #886 item 27);
+	// only the post-promotion window passes an onStall that takes the stacks.
+	// With CELERIS_589_FRESH the probe is pingOnce's fresh-conn one in both.
+	pingWatched := func(onStall func()) (time.Duration, error) {
+		if fresh {
+			return pingOnce()
+		}
+		t0 := time.Now()
+		done := make(chan error, 1)
+		go func() { done <- get589(pc, pbr, "/ping") }()
+		stalled := time.NewTimer(100 * time.Millisecond)
+		defer stalled.Stop()
+		select {
+		case err := <-done:
+			if err != nil {
+				return 0, err
+			}
+		case <-stalled.C:
+			if onStall != nil {
+				onStall()
+			}
+			if err := <-done; err != nil {
+				return 0, err
+			}
+		}
+		return time.Since(t0), nil
+	}
+
 	// celeris#592: the settled classification is re-opened every
 	// adaptiveSettleTTL, the re-timed run is ~D (far over
 	// adaptiveBlockingThreshold) and the route promotes. Wait for that —
@@ -1408,7 +1475,7 @@ func runStall589(t *testing.T, engName string, engType EngineType, mode stall589
 					return
 				default:
 				}
-				d, err := pingOnce()
+				d, err := pingWatched(nil)
 				if err != nil {
 					select {
 					case refErr <- err:
@@ -1469,38 +1536,22 @@ func runStall589(t *testing.T, engName string, engType EngineType, mode stall589
 	// (stackTally589, celeris#886 item 1): 100 ms into any /ping that has not
 	// been answered, the event-loop goroutines' stacks are classified.
 	stackDiag := !fresh
+	if stackDiag {
+		// 100 ms into EVERY stalled /ping, classify the event-loop goroutines'
+		// blocking site (tallied on the STACKTALLY589 line; CELERIS_589_STACK=1
+		// also dumps the first one in full).
+		o.stacks = &stackTally589{}
+	}
 	for time.Since(winStart) < stall589Window {
+		var onStall func()
 		if stackDiag {
-			// 100 ms into EVERY stalled /ping, classify the event-loop
-			// goroutines' blocking site (tallied on the STACKTALLY589 line;
-			// CELERIS_589_STACK=1 also dumps the first one in full).
-			if o.stacks == nil {
-				o.stacks = &stackTally589{}
-			}
-			t0 := time.Now()
-			done := make(chan error, 1)
-			go func() { done <- get589(pc, pbr, "/ping") }()
-			select {
-			case err := <-done:
-				if err != nil {
-					t.Fatalf("GET /ping during window: %v", err)
-				}
-			case <-time.After(100 * time.Millisecond):
-				o.stacks.capture(t)
-				if err := <-done; err != nil {
-					t.Fatalf("GET /ping during window: %v", err)
-				}
-			}
-			lat = append(lat, time.Since(t0))
-		} else {
-			// The same call the reference window made (celeris#622), so the
-			// two medians differ only in the engine state they observed.
-			d, err := pingOnce()
-			if err != nil {
-				t.Fatalf("GET /ping during window: %v", err)
-			}
-			lat = append(lat, d)
+			onStall = func() { o.stacks.capture(t) }
+		}
+		d, err := pingWatched(onStall)
+		if err != nil {
+			t.Fatalf("GET /ping during window: %v", err)
 		}
+		lat = append(lat, d)
 		time.Sleep(stall589Spacing)
 	}
 	winEnd := time.Now()

internal/adaptive/bind_wait_890_linux_test.go

@@ -0,0 +1,82 @@
+//go:build linux
+
+package adaptive
+
+import (
+	"net"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// celeris#890 item 98: s0Bind (and newBoundAdaptiveH) gave Listen 3 s to
+// publish its address and failed "adaptive engine never bound"; under CPU
+// starvation that deadline was missed 1 time in 400 runs. awaitBind890 is the
+// wait that replaced it; these tests drive it with a fake address source.
+
+func fakeAddr890() net.Addr { return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 4242} }
+
+// TestAwaitBindOutwaitsTheOldThreeSeconds890: the limit the tests use is well
+// past the 3 s that was missed. It pins the constant; that the wait waits is
+// what the four tests below show.
+func TestAwaitBindOutwaitsTheOldThreeSeconds890(t *testing.T) {
+	if bindLimit890 < 15*time.Second {
+		t.Errorf("bindLimit890 = %v, want at least 15 s (it was 3 s when the flake hit)", bindLimit890)
+	}
+}
+
+func TestAwaitBindReturnsTheAddressAndHowLongItTook890(t *testing.T) {
+	var published atomic.Pointer[net.Addr]
+	go func() {
+		time.Sleep(120 * time.Millisecond)
+		a := fakeAddr890()
+		published.Store(&a)
+	}()
+	addr := func() net.Addr {
+		if p := published.Load(); p != nil {
+			return *p
+		}
+		return nil
+	}
+	a, took, err := awaitBind890(addr, make(chan struct{}), 10*time.Second)
+	if err != nil || a == nil {
+		t.Fatalf("awaitBind890 = %v, %v, %v for an address published after 120 ms", a, took, err)
+	}
+	if took < 100*time.Millisecond || took > 5*time.Second {
+		t.Errorf("took = %v, want about 120 ms", took)
+	}
+}
+
+func TestAwaitBindStopsWhenListenReturns890(t *testing.T) {
+	ended := make(chan struct{})
+	close(ended)
+	start := time.Now()
+	a, _, err := awaitBind890(func() net.Addr { return nil }, ended, 10*time.Second)
+	if err == nil || a != nil {
+		t.Fatalf("awaitBind890 = %v, %v after Listen returned, want an error", a, err)
+	}
+	if time.Since(start) > 2*time.Second {
+		t.Errorf("took %v to notice that Listen had returned", time.Since(start))
+	}
+}
+
+func TestAwaitBindGivesUpAtTheLimit890(t *testing.T) {
+	a, took, err := awaitBind890(func() net.Addr { return nil }, make(chan struct{}), 150*time.Millisecond)
+	if err == nil || a != nil {
+		t.Fatalf("awaitBind890 = %v, %v with no address, want an error at the limit", a, err)
+	}
+	if took < 150*time.Millisecond || took > 5*time.Second {
+		t.Errorf("took = %v, want about the 150 ms limit", took)
+	}
+}
+
+// TestAwaitBindSeesAnAddressPublishedAsListenReturns890: an address that is
+// there when Listen has returned is an address.
+func TestAwaitBindSeesAnAddressPublishedAsListenReturns890(t *testing.T) {
+	ended := make(chan struct{})
+	close(ended)
+	a, _, err := awaitBind890(fakeAddr890, ended, 10*time.Second)
+	if err != nil || a == nil {
+		t.Fatalf("awaitBind890 = %v, %v, want the address", a, err)
+	}
+}

internal/adaptive/controller.go

@@ -33,7 +33,9 @@ import (
 //     sends to peers that had already gone away (celeris#856, see
 //     revertErrors), per unit of work: the larger of the interval's requests
 //     and its connection-seconds, so a streaming workload with a handful of
-//     requests is not read as faulty (celeris#886 item 9, see faultRate).
+//     requests is not read as faulty (celeris#886 item 9, see faultRate). An
+//     interval with fewer than minRevertFaults faults reads rate 0, so one
+//     stray accept error does not revert an engine (celeris#886 item 29).
 //
 // The oscillation lock (3 switches in 5 min → 5 min lock) and the post-switch
 // cooldown bound any residual thrash and hold io_uring after the fast snap.