5eb9885 test(xwm): make sinkPeer deterministic · neferwl

M internal/adapters/wayland/xwm/configure_test.go +16 -8


@@ -2,7 +2,9 @@ package xwm
2	2
3	3		import (
4	4		"encoding/binary"
	5	+	"runtime"
5	6		"testing"
	7	+	"time"
6	8
7	9		"bnema.dev/xturbo/xproto"
8	10		"github.com/bnema/neferwl/internal/ports"

@@ -506,16 +508,22 @@ func TestCloseWindow(t *testing.T) {
506	508
507	509		func sinkPeer(t testing.T, x XWM, p *xpeer) {
508	510		t.Helper()
	511	+	// Everything written after sink.Store(true) is discarded, so one marker
	512	+	// request proves the peer loop is in sink mode and idle again.
	513	+	// A round trip first: the peer has consumed every earlier request, so the
	514	+	// marker is the only one it can see in sink mode.
	515	+	require.NoError(t, x.conn.Sync(&x.h))
	516	+	before := p.discarded.Load()
509	517		p.sink.Store(true)
510		-	n := len(p.requests())
511		-	xproto.GetInputFocus(x.conn.Buf()) // served by the old loop iteration
	518	+	xproto.GetInputFocus(x.conn.Buf())
512	519		require.NoError(t, x.conn.Flush())
513		-	p.waitReq(n, func(r xpeerReq) bool { return r.major == opGetInputFocus })
514		-	// The peer records a request and answers it under mu: once the lock is
515		-	// free again it is back in its sink loop, which does not allocate.
516		-	p.mu.Lock()
517		-	_ = p.seq
518		-	p.mu.Unlock()
	520	+	deadline := time.Now().Add(30 * time.Second) // failure guard only: the wait is event driven
	521	+	for p.discarded.Load() == before {
	522	+	if time.Now().After(deadline) {
	523	+	t.Fatal("xpeer: sink marker never discarded")
	524	+	}
	525	+	runtime.Gosched()
	526	+	}
519	527		x.src = nil // no source: write backpressure would not allocate either
520	528		x.post = func(func()) {}
521	529		x.warnedRing = true // the sink never answers: Expect slots stay used

 import (
 	"encoding/binary"
+	"runtime"
 	"testing"
+	"time"
 	"bnema.dev/xturbo/xproto"
 	"github.com/bnema/neferwl/internal/ports"
 func sinkPeer(t *testing.T, x *XWM, p *xpeer) {
 	t.Helper()
+	// Everything written after sink.Store(true) is discarded, so one marker
+	// request proves the peer loop is in sink mode and idle again.
+	// A round trip first: the peer has consumed every earlier request, so the
+	// marker is the only one it can see in sink mode.
+	require.NoError(t, x.conn.Sync(&x.h))
+	before := p.discarded.Load()
 	p.sink.Store(true)
-	n := len(p.requests())
+	xproto.GetInputFocus(x.conn.Buf())
-	xproto.GetInputFocus(x.conn.Buf()) // served by the old loop iteration
 	require.NoError(t, x.conn.Flush())
-	p.waitReq(n, func(r xpeerReq) bool { return r.major == opGetInputFocus })
+	deadline := time.Now().Add(30 * time.Second) // failure guard only: the wait is event driven
-	// The peer records a request and answers it under mu: once the lock is
+	for p.discarded.Load() == before {
-	// free again it is back in its sink loop, which does not allocate.
+		if time.Now().After(deadline) {
-	p.mu.Lock()
+			t.Fatal("xpeer: sink marker never discarded")
-	_ = p.seq
+		}
-	p.mu.Unlock()
+		runtime.Gosched()
+	}
 	x.src = nil // no source: write backpressure would not allocate either
 	x.post = func(func()) {}
 	x.warnedRing = true // the sink never answers: Expect slots stay used

M internal/adapters/wayland/xwm/streams_test.go +15 -1


@@ -5,6 +5,7 @@ import (
5	5		"os"
6	6		"path/filepath"
7	7		"runtime"
	8	+	"runtime/debug"
8	9		"strings"
9	10		"testing"
10	11		"time"

@@ -139,11 +140,18 @@ func replay(t testing.T, k streamKind) (mallocs uint64, hs harness) {
139	140		stream := loadStream(t, k)
140	141		hs = streamScenario(t)
141	142		hs.p.pause()
142		-	poke(hs)
	143	+	// A peer blocked in its read consumes one more request before it parks
	144	+	// (this one, or an earlier one still in flight); a parked peer leaves it
	145	+	// unread. Either way the peer ends up parked, with nothing half read.
	146	+	xproto.GetInputFocus(hs.x.conn.Buf())
	147	+	hs.x.flush()
	148	+	hs.p.waitParked()
143	149		_, err := hs.p.f.Write(stream)
144	150		require.NoError(t, err)
145	151
146	152		x := hs.x
	153	+	// The harness post queue grows by append, at batching-dependent moments.
	154	+	x.post = func(func()) {}
147	155		base := x.Stats().Events + x.Stats().Replies
148	156		drive := func(until uint64) {
149	157		deadline := time.Now().Add(10 * time.Second)

@@ -162,6 +170,12 @@ func replay(t testing.T, k streamKind) (mallocs uint64, hs harness) {
162	170		target := base + uint64(streamCycles*k.perCycle)
163	171
164	172		var m0, m1 runtime.MemStats
	173	+	// A GC cycle finishing inside the window allocates runtime-internal
	174	+	// objects (sudogs), which Mallocs counts: keep the collector off.
	175	+	defer debug.SetGCPercent(debug.SetGCPercent(-1))
	176	+	// Like testing.AllocsPerRun: one P, so no other goroutine's runtime
	177	+	// bookkeeping (timers, netpoll) lands in the window.
	178	+	defer runtime.GOMAXPROCS(runtime.GOMAXPROCS(1))
165	179		runtime.GC()
166	180		runtime.ReadMemStats(&m0)
167	181		drive(target)

 	"os"
 	"path/filepath"
 	"runtime"
+	"runtime/debug"
 	"strings"
 	"testing"
 	"time"
 	stream := loadStream(t, k)
 	hs = streamScenario(t)
 	hs.p.pause()
-	poke(hs)
+	// A peer blocked in its read consumes one more request before it parks
+	// (this one, or an earlier one still in flight); a parked peer leaves it
+	// unread. Either way the peer ends up parked, with nothing half read.
+	xproto.GetInputFocus(hs.x.conn.Buf())
+	hs.x.flush()
+	hs.p.waitParked()
 	_, err := hs.p.f.Write(stream)
 	require.NoError(t, err)
 	x := hs.x
+	// The harness post queue grows by append, at batching-dependent moments.
+	x.post = func(func()) {}
 	base := x.Stats().Events + x.Stats().Replies
 	drive := func(until uint64) {
 		deadline := time.Now().Add(10 * time.Second)
 	target := base + uint64(streamCycles*k.perCycle)
 	var m0, m1 runtime.MemStats
+	// A GC cycle finishing inside the window allocates runtime-internal
+	// objects (sudogs), which Mallocs counts: keep the collector off.
+	defer debug.SetGCPercent(debug.SetGCPercent(-1))
+	// Like testing.AllocsPerRun: one P, so no other goroutine's runtime
+	// bookkeeping (timers, netpoll) lands in the window.
+	defer runtime.GOMAXPROCS(runtime.GOMAXPROCS(1))
 	runtime.GC()
 	runtime.ReadMemStats(&m0)
 	drive(target)

M internal/adapters/wayland/xwm/xpeer_test.go +40 -16


@@ -4,6 +4,7 @@ import (
4	4		"encoding/binary"
5	5		"io"
6	6		"os"
	7	+	"runtime"
7	8		"sync"
8	9		"sync/atomic"
9	10		"testing"

@@ -92,10 +93,13 @@ type xpeer struct {
92	93		tap []byte // everything written to the client, when recording
93	94		record bool
94	95
95		-	gate atomic.Pointer[chan struct{}] // closed = reading allowed
96		-	sbuf [4096]byte
97		-	sink atomic.Bool // discard requests without recording or allocating
98		-	done chan struct{}
	96	+	gate atomic.Pointer[chan struct{}] // closed = reading allowed
	97	+	sbuf [4096]byte
	98	+	parked, reading atomic.Bool // peer state: at the closed gate / blocked in the header read
	99	+	sink atomic.Bool // discard requests without recording or allocating
	100	+	// discarded counts requests fully dropped in sink mode.
	101	+	discarded atomic.Uint64
	102	+	done chan struct{}
99	103		}
100	104
101	105		func fullPeerExts() map[string]xpeerExt {

@@ -163,17 +167,30 @@ func (p *xpeer) serve() {
163	167		return
164	168		}
165	169		for {
166		-	<-*p.gate.Load()
	170	+	select {
	171	+	case <-*p.gate.Load():
	172	+	default:
	173	+	p.parked.Store(true)
	174	+	<-*p.gate.Load()
	175	+	p.parked.Store(false)
	176	+	}
	177	+	// The sink decision follows the header read, so a request written
	178	+	// after sink.Store(true) is always discarded, never recorded.
	179	+	p.reading.Store(true)
	180	+	_, err := io.ReadFull(p.f, p.sbuf[:4])
	181	+	p.reading.Store(false)
	182	+	if err != nil {
	183	+	return
	184	+	}
167	185		if p.sink.Load() {
168		-	if !p.discard() {
	186	+	if !p.discardRest() {
169	187		return
170	188		}
	189	+	p.discarded.Add(1)
171	190		continue
172	191		}
173	192		var h [4]byte
174		-	if _, err := io.ReadFull(p.f, h[:]); err != nil {
175		-	return
176		-	}
	193	+	copy(h[:], p.sbuf[:4])
177	194		raw := h[:]
178	195		n := int(binary.LittleEndian.Uint16(h[2:])) * 4
179	196		if n == 0 { // BIG-REQUESTS header

@@ -198,12 +215,9 @@ func (p *xpeer) serve() {
198	215		}
199	216		}
200	217
201		-	// discard reads and drops one request without allocating. It reports false
202		-	// when the connection ended.
203		-	func (p *xpeer) discard() bool {
204		-	if _, err := io.ReadFull(p.f, p.sbuf[:4]); err != nil {
205		-	return false
206		-	}
	218	+	// discardRest drops the rest of the request whose 4-byte header is in
	219	+	// p.sbuf, without allocating. It reports false when the connection ended.
	220	+	func (p *xpeer) discardRest() bool {
207	221		n := int(binary.LittleEndian.Uint16(p.sbuf[2:])) * 4
208	222		left := n - 4
209	223		if n == 0 {

@@ -519,12 +533,22 @@ func (p *xpeer) destroyNotify(win uint32) {
519	533		})
520	534		}
521	535
522		-	// pause makes the peer stop reading after the request it is serving.
	536	+	// pause closes the gate: the peer parks before reading its next request.
523	537		func (p *xpeer) pause() {
524	538		g := make(chan struct{})
525	539		p.gate.Store(&g)
526	540		}
527	541
	542	+	// waitParked waits until the peer goroutine is parked at the closed gate. A
	543	+	// peer blocked in its header read only gets there after one more request, so
	544	+	// the caller must have one in flight (or be sure it is parked already).
	545	+	func (p *xpeer) waitParked() {
	546	+	p.t.Helper()
	547	+	for !p.parked.Load() {
	548	+	runtime.Gosched()
	549	+	}
	550	+	}
	551	+
528	552		// resume lets the peer read again.
529	553		func (p *xpeer) resume() {
530	554		g := *p.gate.Load()

 	"encoding/binary"
 	"io"
 	"os"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"testing"
 	tap      []byte // everything written to the client, when recording
 	record   bool
-	gate atomic.Pointer[chan struct{}] // closed = reading allowed
+	gate            atomic.Pointer[chan struct{}] // closed = reading allowed
-	sbuf [4096]byte
+	sbuf            [4096]byte
-	sink atomic.Bool // discard requests without recording or allocating
+	parked, reading atomic.Bool // peer state: at the closed gate / blocked in the header read
-	done chan struct{}
+	sink            atomic.Bool // discard requests without recording or allocating
+	// discarded counts requests fully dropped in sink mode.
+	discarded atomic.Uint64
+	done      chan struct{}
 }
 func fullPeerExts() map[string]xpeerExt {
 		return
 	}
 	for {
-		<-*p.gate.Load()
+		select {
+		case <-*p.gate.Load():
+		default:
+			p.parked.Store(true)
+			<-*p.gate.Load()
+			p.parked.Store(false)
+		}
+		// The sink decision follows the header read, so a request written
+		// after sink.Store(true) is always discarded, never recorded.
+		p.reading.Store(true)
+		_, err := io.ReadFull(p.f, p.sbuf[:4])
+		p.reading.Store(false)
+		if err != nil {
+			return
+		}
 		if p.sink.Load() {
-			if !p.discard() {
+			if !p.discardRest() {
 				return
 			}
+			p.discarded.Add(1)
 			continue
 		}
 		var h [4]byte
-		if _, err := io.ReadFull(p.f, h[:]); err != nil {
+		copy(h[:], p.sbuf[:4])
-			return
-		}
 		raw := h[:]
 		n := int(binary.LittleEndian.Uint16(h[2:])) * 4
 		if n == 0 { // BIG-REQUESTS header
 	}
 }
-// discard reads and drops one request without allocating. It reports false
+// discardRest drops the rest of the request whose 4-byte header is in
-// when the connection ended.
+// p.sbuf, without allocating. It reports false when the connection ended.
-func (p *xpeer) discard() bool {
+func (p *xpeer) discardRest() bool {
-	if _, err := io.ReadFull(p.f, p.sbuf[:4]); err != nil {
-		return false
-	}
 	n := int(binary.LittleEndian.Uint16(p.sbuf[2:])) * 4
 	left := n - 4
 	if n == 0 {
 	})
 }
-// pause makes the peer stop reading after the request it is serving.
+// pause closes the gate: the peer parks before reading its next request.
 func (p *xpeer) pause() {
 	g := make(chan struct{})
 	p.gate.Store(&g)
 }
+// waitParked waits until the peer goroutine is parked at the closed gate. A
+// peer blocked in its header read only gets there after one more request, so
+// the caller must have one in flight (or be sure it is parked already).
+func (p *xpeer) waitParked() {
+	p.t.Helper()
+	for !p.parked.Load() {
+		runtime.Gosched()
+	}
+}
 // resume lets the peer read again.
 func (p *xpeer) resume() {
 	g := *p.gate.Load()