fix(wayland): keep flips when the presentation ring overflows

A full ring now folds the oldest report into the next report of the same
output (the incoming one included): a flip-less successor carries the flip,
a successor with a flip merges it. Nothing is dropped when no successor
exists. Merges are counted and warned about at most every 10 s. The
test-only drain counter goes; tests observe frame callbacks instead.

commit: c9850c5042b6a949df630b66a80927505e065c96
author: brice · 9d ago · 2026-10-13 17:58 UTC
parent: b76d5bc
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +439 -74

M internal/adapters/wayland/presentation_test.go +7 -1


@@ -47,11 +47,17 @@ func (p feedbackEvents) Dispatch(e wlturbo.Event) {
47	47
48	48		// presentServer is a server whose flips the test sends.
49	49		func presentServer(t testing.T) (Server, chan ports.ClientEvent, chan ports.ClientCommand, <-chan ports.SurfaceContent, ports.PresentationSink, string) {
	50	+	t.Helper()
	51	+	return presentServerOn(t, testOutputs)
	52	+	}
	53	+
	54	+	// presentServerOn is presentServer with its own output layout.
	55	+	func presentServerOn(t testing.T, layout ports.Layout) (Server, chan ports.ClientEvent, chan ports.ClientCommand, <-chan ports.SurfaceContent, ports.PresentationSink, string) {
50	56		t.Helper()
51	57		dir := t.TempDir()
52	58		events := make(chan ports.ClientEvent, 16)
53	59		commands := make(chan ports.ClientCommand, 16)
54		-	s, err := New(Options{RuntimeDir: dir, Outputs: testOutputs}, Channels{Events: events, Commands: commands}, logging.For(context.Background(), "wayland"))
	60	+	s, err := New(Options{RuntimeDir: dir, Outputs: layout}, Channels{Events: events, Commands: commands}, logging.For(context.Background(), "wayland"))
55	61		if err != nil {
56	62		t.Fatal(err)
57	63		}

 // presentServer is a server whose flips the test sends.
 func presentServer(t *testing.T) (*Server, chan ports.ClientEvent, chan ports.ClientCommand, <-chan ports.SurfaceContent, ports.PresentationSink, string) {
+	t.Helper()
+	return presentServerOn(t, testOutputs)
+}
+// presentServerOn is presentServer with its own output layout.
+func presentServerOn(t *testing.T, layout ports.Layout) (*Server, chan ports.ClientEvent, chan ports.ClientCommand, <-chan ports.SurfaceContent, ports.PresentationSink, string) {
 	t.Helper()
 	dir := t.TempDir()
 	events := make(chan ports.ClientEvent, 16)
 	commands := make(chan ports.ClientCommand, 16)
-	s, err := New(Options{RuntimeDir: dir, Outputs: testOutputs}, Channels{Events: events, Commands: commands}, logging.For(context.Background(), "wayland"))
+	s, err := New(Options{RuntimeDir: dir, Outputs: layout}, Channels{Events: events, Commands: commands}, logging.For(context.Background(), "wayland"))
 	if err != nil {
 		t.Fatal(err)
 	}

M internal/adapters/wayland/presented.go +74 -31


@@ -3,29 +3,36 @@ package wayland
3	3		import (
4	4		"slices"
5	5		"sync"
6		-	"sync/atomic"
	6	+	"time"
7	7
8	8		"github.com/bnema/neferwl/internal/adapters/outputkit/presented"
9	9		"github.com/bnema/neferwl/internal/ports"
10	10		)
11	11
12		-	// maxPresentRing bounds the reports waiting for the display goroutine. Each
13		-	// output already bounds its own queue (presented.MaxUnsent), so the ring only
14		-	// fills when the display loop is stalled.
	12	+	// maxPresentRing bounds the reports waiting for the display goroutine. Output
	13	+	// queues are flushed every loop turn and hold no backlog, so the ring only
	14	+	// fills when the display loop is stalled; this sink alone then merges flips.
15	15		const maxPresentRing = 64
16	16
	17	+	// mergeWarnEvery spaces the warnings about merged flips.
	18	+	const mergeWarnEvery = 10 * time.Second
	19	+
17	20		// presentSink is the ports.PresentationSink wayland offers to the outputs:
18	21		// output owner goroutines Push reports, the display goroutine drains them and
19	22		// paces frames. The ring is a transport queue between those goroutines, not
20		-	// window state: the mutex guards only the two slices and the pending flag.
	23	+	// window state: the mutex guards only the slices, the pending flag and the
	24	+	// merge counters.
21	25		type presentSink struct {
22	26		s *Server
23	27
24		-	mu sync.Mutex
25		-	ring, spare []ports.OutputPresented
26		-	pending bool // a drain is posted and has not taken the ring yet
	28	+	mu sync.Mutex
	29	+	ring, spare []ports.OutputPresented
	30	+	pending bool // a drain is posted and has not taken the ring yet
	31	+	// merged counts flips merged since the last warning;
	32	+	// lastWarn is when that warning was logged.
	33	+	merged int
	34	+	lastWarn time.Time
27	35		drainFn func()
28		-	drains atomic.Uint64 // finished drains
29	36		scratchFlips map[string]bool
30	37		}
31	38

@@ -44,46 +51,83 @@ func newPresentSink(s Server) presentSink {
44	51		func (s *Server) Presentation() ports.PresentationSink { return s.sink }
45	52
46	53		// Push queues a report for the display goroutine and wakes it; one wake
47		-	// serves every report pushed before the drain runs.
	54	+	// serves every report pushed before the drain runs. When the ring is full the
	55	+	// flip of the oldest report is carried or merged forward; merges are counted
	56	+	// and warned about at most every mergeWarnEvery.
48	57		func (p *presentSink) Push(r ports.OutputPresented) {
49	58		p.mu.Lock()
50		-	if len(p.ring) >= maxPresentRing {
51		-	p.dropOldestLocked()
52		-	}
53		-	p.ring = append(p.ring, r)
	59	+	warn, merged := p.enqueueLocked(r)
54	60		post := !p.pending
55	61		p.pending = true
56	62		p.mu.Unlock()
	63	+	if warn {
	64	+	p.s.log.Warn().Int("merged", merged).Msg("outputs report faster than the display loop drains; flips merged")
	65	+	}
57	66		if post {
58	67		p.s.display.Post(p.drainFn)
59	68		}
60	69		}
61	70
62		-	// dropOldestLocked makes room: the oldest report that a later report of the
63		-	// same output follows is folded into it (flips merge as in presented.Queue).
64		-	func (p *presentSink) dropOldestLocked() {
65		-	victim := 0
	71	+	// enqueueLocked appends r, making room first when the ring is full. It
	72	+	// reports whether to warn now and how many flips were merged since the last
	73	+	// warning.
	74	+	func (p *presentSink) enqueueLocked(r ports.OutputPresented) (warn bool, merged int) {
	75	+	if len(p.ring) >= maxPresentRing {
	76	+	if p.makeRoomLocked(&r) {
	77	+	p.merged++
	78	+	}
	79	+	if now := p.s.clock.Now(); p.merged > 0 && now.Sub(p.lastWarn) >= mergeWarnEvery {
	80	+	warn, merged = true, p.merged
	81	+	p.merged, p.lastWarn = 0, now
	82	+	}
	83	+	}
	84	+	p.ring = append(p.ring, r)
	85	+	return warn, merged
	86	+	}
	87	+
	88	+	// makeRoomLocked frees a ring slot for incoming without losing a flip: the
	89	+	// oldest report that a later report of the same output follows (in the ring,
	90	+	// or incoming itself) is removed and its flip carried into that report. A
	91	+	// flip-less successor takes the flip over; a successor with a flip merges it
	92	+	// (presented.Merge: the older feedback is discarded), which is what the
	93	+	// result counts.
	94	+	//
	95	+	// When no report has a successor, every ring entry and incoming belong to
	96	+	// distinct outputs: only possible with more than maxPresentRing outputs.
	97	+	// Nothing is dropped then; the ring grows by one, bounded by the number of
	98	+	// outputs, because the next report of any of them finds a successor.
	99	+	func (p presentSink) makeRoomLocked(incoming ports.OutputPresented) (merged bool) {
66	100		for i := range p.ring {
67		-	if next := p.nextOfOutputLocked(i); next > 0 {
68		-	if old, nf := p.ring[i].Flip, p.ring[next].Flip; old != nil && nf != nil {
69		-	presented.Merge(old, nf)
70		-	}
71		-	victim = i
72		-	break
	101	+	next := p.successorLocked(i, incoming)
	102	+	if next == nil {
	103	+	continue
	104	+	}
	105	+	switch old := p.ring[i].Flip; {
	106	+	case old == nil:
	107	+	case next.Flip == nil:
	108	+	next.Flip = old
	109	+	default:
	110	+	presented.Merge(old, next.Flip)
	111	+	merged = true
73	112		}
	113	+	p.ring = slices.Delete(p.ring, i, i+1)
	114	+	return merged
74	115		}
75		-	p.ring = slices.Delete(p.ring, victim, victim+1)
	116	+	return false
76	117		}
77	118
78		-	// nextOfOutputLocked returns the index of the next report of i's output, 0
79		-	// when none follows.
80		-	func (p *presentSink) nextOfOutputLocked(i int) int {
	119	+	// successorLocked returns the next report of ring[i]'s output: a later ring
	120	+	// entry, else incoming; nil when none follows.
	121	+	func (p presentSink) successorLocked(i int, incoming ports.OutputPresented) *ports.OutputPresented {
81	122		for j := i + 1; j < len(p.ring); j++ {
82	123		if p.ring[j].Output == p.ring[i].Output {
83		-	return j
	124	+	return &p.ring[j]
84	125		}
85	126		}
86		-	return 0
	127	+	if incoming.Output == p.ring[i].Output {
	128	+	return incoming
	129	+	}
	130	+	return nil
87	131		}
88	132
89	133		// drain runs on the display goroutine: it paces frames for every report that

@@ -108,5 +152,4 @@ func (p *presentSink) drain() {
108	152		p.mu.Lock()
109	153		p.spare = batch[:0]
110	154		p.mu.Unlock()
111		-	p.drains.Add(1)
112	155		}

 import (
 	"slices"
 	"sync"
-	"sync/atomic"
+	"time"
 	"github.com/bnema/neferwl/internal/adapters/outputkit/presented"
 	"github.com/bnema/neferwl/internal/ports"
 )
-// maxPresentRing bounds the reports waiting for the display goroutine. Each
+// maxPresentRing bounds the reports waiting for the display goroutine. Output
-// output already bounds its own queue (presented.MaxUnsent), so the ring only
+// queues are flushed every loop turn and hold no backlog, so the ring only
-// fills when the display loop is stalled.
+// fills when the display loop is stalled; this sink alone then merges flips.
 const maxPresentRing = 64
+// mergeWarnEvery spaces the warnings about merged flips.
+const mergeWarnEvery = 10 * time.Second
 // presentSink is the ports.PresentationSink wayland offers to the outputs:
 // output owner goroutines Push reports, the display goroutine drains them and
 // paces frames. The ring is a transport queue between those goroutines, not
-// window state: the mutex guards only the two slices and the pending flag.
+// window state: the mutex guards only the slices, the pending flag and the
+// merge counters.
 type presentSink struct {
 	s *Server
-	mu           sync.Mutex
+	mu          sync.Mutex
-	ring, spare  []ports.OutputPresented
+	ring, spare []ports.OutputPresented
-	pending      bool // a drain is posted and has not taken the ring yet
+	pending     bool // a drain is posted and has not taken the ring yet
+	// merged counts flips merged since the last warning;
+	// lastWarn is when that warning was logged.
+	merged       int
+	lastWarn     time.Time
 	drainFn      func()
-	drains       atomic.Uint64 // finished drains
 	scratchFlips map[string]bool
 }
 func (s *Server) Presentation() ports.PresentationSink { return s.sink }
 // Push queues a report for the display goroutine and wakes it; one wake
-// serves every report pushed before the drain runs.
+// serves every report pushed before the drain runs. When the ring is full the
+// flip of the oldest report is carried or merged forward; merges are counted
+// and warned about at most every mergeWarnEvery.
 func (p *presentSink) Push(r ports.OutputPresented) {
 	p.mu.Lock()
-	if len(p.ring) >= maxPresentRing {
+	warn, merged := p.enqueueLocked(r)
-		p.dropOldestLocked()
-	}
-	p.ring = append(p.ring, r)
 	post := !p.pending
 	p.pending = true
 	p.mu.Unlock()
+	if warn {
+		p.s.log.Warn().Int("merged", merged).Msg("outputs report faster than the display loop drains; flips merged")
+	}
 	if post {
 		p.s.display.Post(p.drainFn)
 	}
 }
-// dropOldestLocked makes room: the oldest report that a later report of the
+// enqueueLocked appends r, making room first when the ring is full. It
-// same output follows is folded into it (flips merge as in presented.Queue).
+// reports whether to warn now and how many flips were merged since the last
-func (p *presentSink) dropOldestLocked() {
+// warning.
-	victim := 0
+func (p *presentSink) enqueueLocked(r ports.OutputPresented) (warn bool, merged int) {
+	if len(p.ring) >= maxPresentRing {
+		if p.makeRoomLocked(&r) {
+			p.merged++
+		}
+		if now := p.s.clock.Now(); p.merged > 0 && now.Sub(p.lastWarn) >= mergeWarnEvery {
+			warn, merged = true, p.merged
+			p.merged, p.lastWarn = 0, now
+		}
+	}
+	p.ring = append(p.ring, r)
+	return warn, merged
+}
+// makeRoomLocked frees a ring slot for incoming without losing a flip: the
+// oldest report that a later report of the same output follows (in the ring,
+// or incoming itself) is removed and its flip carried into that report. A
+// flip-less successor takes the flip over; a successor with a flip merges it
+// (presented.Merge: the older feedback is discarded), which is what the
+// result counts.
+//
+// When no report has a successor, every ring entry and incoming belong to
+// distinct outputs: only possible with more than maxPresentRing outputs.
+// Nothing is dropped then; the ring grows by one, bounded by the number of
+// outputs, because the next report of any of them finds a successor.
+func (p *presentSink) makeRoomLocked(incoming *ports.OutputPresented) (merged bool) {
 	for i := range p.ring {
-		if next := p.nextOfOutputLocked(i); next > 0 {
+		next := p.successorLocked(i, incoming)
-			if old, nf := p.ring[i].Flip, p.ring[next].Flip; old != nil && nf != nil {
+		if next == nil {
-				presented.Merge(old, nf)
+			continue
-			}
+		}
-			victim = i
+		switch old := p.ring[i].Flip; {
-			break
+		case old == nil:
+		case next.Flip == nil:
+			next.Flip = old
+		default:
+			presented.Merge(old, next.Flip)
+			merged = true
 		}
+		p.ring = slices.Delete(p.ring, i, i+1)
+		return merged
 	}
-	p.ring = slices.Delete(p.ring, victim, victim+1)
+	return false
 }
-// nextOfOutputLocked returns the index of the next report of i's output, 0
+// successorLocked returns the next report of ring[i]'s output: a later ring
-// when none follows.
+// entry, else incoming; nil when none follows.
-func (p *presentSink) nextOfOutputLocked(i int) int {
+func (p *presentSink) successorLocked(i int, incoming *ports.OutputPresented) *ports.OutputPresented {
 	for j := i + 1; j < len(p.ring); j++ {
 		if p.ring[j].Output == p.ring[i].Output {
-			return j
+			return &p.ring[j]
 		}
 	}
-	return 0
+	if incoming.Output == p.ring[i].Output {
+		return incoming
+	}
+	return nil
 }
 // drain runs on the display goroutine: it paces frames for every report that
 	p.mu.Lock()
 	p.spare = batch[:0]
 	p.mu.Unlock()
-	p.drains.Add(1)
 }

M internal/adapters/wayland/presented_test.go +354 -41

@@ -1,60 +1,314 @@
1	1		package wayland
2	2
3	3		import (
4		-	"runtime"
	4	+	"context"
	5	+	"encoding/binary"
	6	+	"strings"
5	7		"testing"
6	8		"time"
7	9
8	10		"github.com/bnema/go-wayland-bindings/server/fifo"
9	11		"github.com/bnema/go-wayland-bindings/server/wayland"
	12	+	"github.com/bnema/neferwl/internal/adapters/logging"
	13	+	portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
10	14		"github.com/bnema/neferwl/internal/ports"
	15	+	"github.com/bnema/wlturbo"
	16	+	"github.com/bnema/zerowrap"
	17	+	"golang.org/x/sys/unix"
11	18		)
12	19
13		-	// Flips pushed while the display goroutine is busy are paced by one drain:
14		-	// one tick, one burst of frame callbacks per output, not one per stale flip.
	20	+	// slowOutputs has two outputs refreshing once a second: their deadlines never
	21	+	// fire inside a test, so only flips pace their frame callbacks.
	22	+	var slowOutputs = ports.Layout{
	23	+	{Info: ports.OutputInfo{Name: "HEADLESS-1", Width: 1920, Height: 1080, RefreshMilli: 1000}, Width: 1920, Height: 1080, Scale: 1},
	24	+	{Info: ports.OutputInfo{Name: "HEADLESS-2", Width: 1920, Height: 1080, RefreshMilli: 1000}, X: 1920, Width: 1920, Height: 1080, Scale: 1},
	25	+	}
	26	+
	27	+	// frameDoneProxy records the time of a wl_callback.done.
	28	+	type frameDoneProxy struct {
	29	+	wlturbo.BaseProxy
	30	+	done chan uint32
	31	+	}
	32	+
	33	+	func (p frameDoneProxy) Dispatch(e wlturbo.Event) {
	34	+	if e.Opcode == uint16(wayland.CallbackEventDone) {
	35	+	p.done <- binary.NativeEndian.Uint32(e.Data())
	36	+	}
	37	+	}
	38	+
	39	+	// requestFrame commits the surface with a new frame callback.
	40	+	func requestFrame(t testing.T, c wlturbo.Display, surf uint32) *frameDoneProxy {
	41	+	t.Helper()
	42	+	cb := c.AllocateID()
	43	+	p := &frameDoneProxy{done: make(chan uint32, 2)}
	44	+	p.SetID(cb)
	45	+	registerWireProxy(c, p)
	46	+	requestProtocol(t, c, surf, wayland.SurfaceRequestFrame, cb)
	47	+	requestProtocol(t, c, surf, wayland.SurfaceRequestCommit)
	48	+	if err := c.Roundtrip(); err != nil {
	49	+	t.Fatal(err)
	50	+	}
	51	+	return p
	52	+	}
	53	+
	54	+	// awaitFrame waits for the callback to be answered and returns its time.
	55	+	func awaitFrame(t testing.T, c wlturbo.Display, p *frameDoneProxy) uint32 {
	56	+	t.Helper()
	57	+	for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
	58	+	if err := c.Roundtrip(); err != nil {
	59	+	t.Fatal(err)
	60	+	}
	61	+	select {
	62	+	case ms := <-p.done:
	63	+	return ms
	64	+	case <-time.After(2 * time.Millisecond):
	65	+	}
	66	+	}
	67	+	t.Fatal("frame callback not answered")
	68	+	return 0
	69	+	}
	70	+
	71	+	// placeWindow puts the window on an output and waits until wayland applied it.
	72	+	func placeWindow(t testing.T, s Server, commands chan<- ports.ClientCommand, w ports.WindowMapped, output string) {
	73	+	t.Helper()
	74	+	commands <- ports.ConfigureWindow{ID: w.ID, Width: 100, Height: 100, Output: output, Visible: true}
	75	+	for deadline := time.Now().Add(2 * time.Second); ; {
	76	+	var got string
	77	+	s.display.Do(func() { got = s.frameOutput(s.windows[w.ID].xdg.surface) })
	78	+	if got == output {
	79	+	return
	80	+	}
	81	+	if time.Now().After(deadline) {
	82	+	t.Fatalf("window not placed on %s", output)
	83	+	}
	84	+	time.Sleep(time.Millisecond)
	85	+	}
	86	+	}
	87	+
	88	+	// Flips pushed while the display goroutine is busy are paced by one tick per
	89	+	// output, not one per stale flip. The frame callback of each output is
	90	+	// answered once, with the time of that output's last flip: a tick per flip
	91	+	// would have answered it with the first flip's time.
15	92		func TestPresentationPushCoalescesWhileDisplayBusy(t *testing.T) {
16		-	s, _, _, _, sink, _ := presentServer(t)
17		-	release, started := make(chan struct{}), make(chan struct{})
18		-	finished := make(chan struct{})
	93	+	s, events, commands, _, sink, dir := presentServerOn(t, slowOutputs)
	94	+	c := protocolClient(t, s, dir)
	95	+	mk := surfaceMapper(t, c, events)
	96	+	outputs := []string{"HEADLESS-1", "HEADLESS-2"}
	97	+	surfs := make([]uint32, len(outputs))
	98	+	for i, name := range outputs {
	99	+	w, surf, _ := mk()
	100	+	placeWindow(t, s, commands, w, name)
	101	+	surfs[i] = surf
	102	+	// The first callback of an idle output is answered at once and
	103	+	// starts its one-second period: the next one waits for a flip.
	104	+	awaitFrame(t, c, requestFrame(t, c, surf))
	105	+	}
	106	+	waiting := make([]*frameDoneProxy, len(outputs))
	107	+	for i := range outputs {
	108	+	waiting[i] = requestFrame(t, c, surfs[i])
	109	+	select {
	110	+	case ms := <-waiting[i].done:
	111	+	t.Fatalf("%s answered before any flip (at %d ms)", outputs[i], ms)
	112	+	default:
	113	+	}
	114	+	}
	115	+	var base time.Duration
	116	+	release, started, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
19	117		go func() {
20		-	s.display.Do(func() { close(started); <-release })
	118	+	s.display.Do(func() { base = monotonic(s.started); close(started); <-release })
21	119		close(finished)
22	120		}()
23	121		<-started
24		-	for i := range 5 {
25		-	sink.Push(ports.OutputPresented{Output: "HEADLESS-1", Flip: &ports.FlipInfo{Seq: uint64(i + 1), HardwareClock: true}})
	122	+	const flips = 5
	123	+	const step = 100 * time.Millisecond
	124	+	for i := range flips {
	125	+	for _, name := range outputs {
	126	+	sink.Push(ports.OutputPresented{Output: name, Flip: &ports.FlipInfo{When: base + time.Duration(i+1)*step, Seq: uint64(i + 1), HardwareClock: true}})
	127	+	}
26	128		}
27	129		close(release)
28	130		<-finished
29		-	var drains, seq uint64
	131	+	last := uint32((flips * step).Milliseconds())
	132	+	for i, name := range outputs {
	133	+	ms := awaitFrame(t, c, waiting[i])
	134	+	// base is read on the display goroutine a little before the server
	135	+	// answers: a few ms of slack, far below one step.
	136	+	if ms < last-5 \|\| ms > last+5 {
	137	+	t.Fatalf("%s answered at %d ms, want the last flip's %d ms (the first flip is at %d)", name, ms, last, step.Milliseconds())
	138	+	}
	139	+	}
	140	+	var seq uint64
30	141		var reported bool
31	142		s.display.Do(func() {
32		-	drains = s.sink.drains.Load()
33	143		seq = s.flips["HEADLESS-1"].Seq
34	144		_, reported = s.reports["HEADLESS-1"]
35	145		})
36		-	if drains != 1 {
37		-	t.Fatalf("%d drains for 5 flips pushed while busy, want 1", drains)
38		-	}
39		-	if !reported \|\| seq != 5 {
	146	+	if !reported \|\| seq != flips {
40	147		t.Fatalf("last flip %d, reported %v: every flip must still be answered in order", seq, reported)
41	148		}
42	149		}
43	150
44		-	// A full ring merges the oldest flip of an output into its next one.
	151	+	func ringOf(outputs ...string) *presentSink {
	152	+	p := &presentSink{ring: make([]ports.OutputPresented, 0, maxPresentRing+1)}
	153	+	for _, o := range outputs {
	154	+	p.ring = append(p.ring, ports.OutputPresented{Output: o})
	155	+	}
	156	+	return p
	157	+	}
	158	+
	159	+	// fill pads the ring with reports of distinct outputs up to the bound.
	160	+	func (p *presentSink) fill() {
	161	+	for i := 0; len(p.ring) < maxPresentRing; i++ {
	162	+	p.ring = append(p.ring, ports.OutputPresented{Output: "pad" + string(rune('A'+i/26)) + string(rune('a'+i%26))})
	163	+	}
	164	+	}
	165	+
	166	+	func flipOf(seq uint64, shows ...ports.WindowID) *ports.FlipInfo {
	167	+	f := &ports.FlipInfo{Seq: seq}
	168	+	if len(shows) > 0 {
	169	+	f.Shows = map[ports.WindowID]uint64{}
	170	+	for _, id := range shows {
	171	+	f.Shows[id] = seq
	172	+	}
	173	+	}
	174	+	return f
	175	+	}
	176	+
	177	+	// A full ring merges the oldest flip of an output into its next flip.
45	178		func TestPresentationRingMergesPastBound(t *testing.T) {
46		-	p := &presentSink{ring: make([]ports.OutputPresented, 0, maxPresentRing)}
	179	+	p := ringOf()
47	180		for i := range maxPresentRing {
48		-	p.ring = append(p.ring, ports.OutputPresented{Output: "A", Flip: &ports.FlipInfo{Seq: uint64(i + 1)}})
	181	+	p.ring = append(p.ring, ports.OutputPresented{Output: "A", Flip: flipOf(uint64(i + 1))})
	182	+	}
	183	+	in := ports.OutputPresented{Output: "A", Flip: flipOf(maxPresentRing + 1)}
	184	+	if !p.makeRoomLocked(&in) {
	185	+	t.Fatal("two flips merged but not reported")
49	186		}
50		-	p.mu.Lock()
51		-	p.dropOldestLocked()
52		-	p.mu.Unlock()
53	187		if len(p.ring) != maxPresentRing-1 \|\| p.ring[0].Flip.Seq != 2 \|\| p.ring[0].Flip.Merged != 1 {
54	188		t.Fatalf("ring %d, first seq %d merged %d", len(p.ring), p.ring[0].Flip.Seq, p.ring[0].Flip.Merged)
55	189		}
56	190		}
57	191
	192	+	// A flip followed by a flip-less report of the same output is not lost: the
	193	+	// later report carries it.
	194	+	func TestPresentationRingKeepsFlipBeforeFliplessReport(t *testing.T) {
	195	+	p := ringOf("A", "A")
	196	+	p.ring[0].Flip = flipOf(7, 3)
	197	+	p.fill()
	198	+	p.ring = p.ring[:maxPresentRing]
	199	+	in := ports.OutputPresented{Output: "B", Flip: flipOf(9)}
	200	+	if p.makeRoomLocked(&in) {
	201	+	t.Fatal("no flips merged, yet reported")
	202	+	}
	203	+	if len(p.ring) != maxPresentRing-1 \|\| p.ring[0].Output != "A" \|\| p.ring[0].Flip == nil \|\| p.ring[0].Flip.Seq != 7 \|\| p.ring[0].Flip.Merged != 0 {
	204	+	t.Fatalf("flip lost: ring %d, first %+v", len(p.ring), p.ring[0])
	205	+	}
	206	+	}
	207	+
	208	+	// With several outputs interleaved, the victim is the oldest report that has
	209	+	// a successor of its own output; other outputs' reports are untouched.
	210	+	func TestPresentationRingInterleavedOutputs(t *testing.T) {
	211	+	p := ringOf("A", "B", "A", "B")
	212	+	p.ring[0].Flip, p.ring[1].Flip = flipOf(1, 10), flipOf(2, 20)
	213	+	p.ring[2].Flip, p.ring[3].Flip = flipOf(3, 11), nil
	214	+	p.fill()
	215	+	in := ports.OutputPresented{Output: "C"}
	216	+	if !p.makeRoomLocked(&in) {
	217	+	t.Fatal("merge not reported")
	218	+	}
	219	+	if p.ring[0].Output != "B" \|\| p.ring[0].Flip.Seq != 2 \|\| p.ring[0].Flip.Merged != 0 {
	220	+	t.Fatalf("B's flip touched: %+v", p.ring[0])
	221	+	}
	222	+	a := p.ring[1]
	223	+	if a.Output != "A" \|\| a.Flip.Seq != 3 \|\| a.Flip.Merged != 1 \|\| a.Flip.Shows[10] != 1 \|\| a.Flip.Shows[11] != 3 {
	224	+	t.Fatalf("A's flips not merged: %+v", a.Flip)
	225	+	}
	226	+	if p.ring[2].Output != "B" \|\| p.ring[2].Flip != nil {
	227	+	t.Fatalf("B's flip-less report moved: %+v", p.ring[2])
	228	+	}
	229	+	}
	230	+
	231	+	// The incoming report counts as the successor of a ring report.
	232	+	func TestPresentationRingIncomingTakesOverFlip(t *testing.T) {
	233	+	p := ringOf("A")
	234	+	p.ring[0].Flip = flipOf(4)
	235	+	p.fill()
	236	+	in := ports.OutputPresented{Output: "A"}
	237	+	if p.makeRoomLocked(&in) {
	238	+	t.Fatal("no merge expected")
	239	+	}
	240	+	if in.Flip == nil \|\| in.Flip.Seq != 4 \|\| len(p.ring) != maxPresentRing-1 {
	241	+	t.Fatalf("incoming %+v, ring %d", in.Flip, len(p.ring))
	242	+	}
	243	+	}
	244	+
	245	+	// With no successor for any report (more outputs than ring slots) nothing is
	246	+	// dropped: the ring grows by one and every flip is still delivered.
	247	+	func TestPresentationRingWithoutSuccessorKeepsEverything(t *testing.T) {
	248	+	p := ringOf()
	249	+	for i := range maxPresentRing {
	250	+	p.ring = append(p.ring, ports.OutputPresented{Output: "o" + string(rune('A'+i/26)) + string(rune('a'+i%26)), Flip: flipOf(uint64(i + 1))})
	251	+	}
	252	+	in := ports.OutputPresented{Output: "other", Flip: flipOf(99)}
	253	+	if p.makeRoomLocked(&in) \|\| len(p.ring) != maxPresentRing {
	254	+	t.Fatalf("merged or dropped: ring %d", len(p.ring))
	255	+	}
	256	+	}
	257	+
	258	+	// Merges are counted, and warned about at most once per mergeWarnEvery.
	259	+	func TestPresentationRingWarnsRateLimited(t *testing.T) {
	260	+	now := time.Unix(1000, 0)
	261	+	clk := portsmocks.NewMockClock(t)
	262	+	clk.EXPECT().Now().RunAndReturn(func() time.Time { return now })
	263	+	p := ringOf()
	264	+	p.s = &Server{clock: clk}
	265	+	for i := range maxPresentRing {
	266	+	p.ring = append(p.ring, ports.OutputPresented{Output: "A", Flip: flipOf(uint64(i + 1))})
	267	+	}
	268	+	push := func(seq uint64) (bool, int) {
	269	+	return p.enqueueLocked(ports.OutputPresented{Output: "A", Flip: flipOf(seq)})
	270	+	}
	271	+	if warn, n := push(100); !warn \|\| n != 1 {
	272	+	t.Fatalf("first merge: warn %v merged %d", warn, n)
	273	+	}
	274	+	for i := range 3 {
	275	+	now = now.Add(time.Second)
	276	+	if warn, _ := push(uint64(101 + i)); warn {
	277	+	t.Fatalf("warned again after %v", time.Duration(i+1)*time.Second)
	278	+	}
	279	+	}
	280	+	now = now.Add(mergeWarnEvery)
	281	+	if warn, n := push(200); !warn \|\| n != 4 {
	282	+	t.Fatalf("after the interval: warn %v merged %d, want the 4 merges since", warn, n)
	283	+	}
	284	+	}
	285	+
	286	+	// A stalled display loop makes outputs merge flips; wayland says so once.
	287	+	func TestPresentationMergeWarnsInLog(t *testing.T) {
	288	+	var out lockedBuffer
	289	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out}).WithField("component", "wayland")
	290	+	s, _, _, _, _ := contentServerWith(t, Options{}, log)
	291	+	release, started, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
	292	+	go func() {
	293	+	s.display.Do(func() { close(started); <-release })
	294	+	close(finished)
	295	+	}()
	296	+	<-started
	297	+	for i := range 3 * maxPresentRing {
	298	+	s.Presentation().Push(ports.OutputPresented{Output: "HEADLESS-1", Flip: &ports.FlipInfo{Seq: uint64(i + 1)}})
	299	+	}
	300	+	close(release)
	301	+	<-finished
	302	+	s.display.Do(func() {})
	303	+	got := out.String()
	304	+	if n := strings.Count(got, "outputs report faster than the display loop drains; flips merged"); n != 1 {
	305	+	t.Fatalf("%d warnings: %s", n, got)
	306	+	}
	307	+	if !strings.Contains(got, `"component":"wayland"`) \|\| !strings.Contains(got, `"merged":`) {
	308	+	t.Fatalf("warning lacks fields: %s", got)
	309	+	}
	310	+	}
	311	+
58	312		// Pushing a report and draining it on the display goroutine allocates nothing
59	313		// in steady state. The count includes the display goroutine: AllocsPerRun
60	314		// reads the global malloc counter.

@@ -64,16 +318,20 @@ func TestPresentationPushAllocations(t *testing.T) {
64	318		}
65	319		s, _, _, _, sink, _ := presentServer(t)
66	320		r := ports.OutputPresented{Output: "HEADLESS-1"}
	321	+	// Posts run in order: once the marker ran, the drain before it did.
	322	+	// The marker is bound once and the channel is reused, so waiting for it
	323	+	// allocates nothing.
	324	+	ran := make(chan struct{}, 1)
	325	+	marker := func() { ran <- struct{}{} }
67	326		for range 4 {
68	327		sink.Push(r)
69		-	s.display.Do(func() {})
	328	+	s.display.Post(marker)
	329	+	<-ran
70	330		}
71	331		allocs := testing.AllocsPerRun(200, func() {
72		-	want := s.sink.drains.Load() + 1
73	332		sink.Push(r)
74		-	for s.sink.drains.Load() < want { // wait for the drain without allocating
75		-	runtime.Gosched()
76		-	}
	333	+	s.display.Post(marker)
	334	+	<-ran
77	335		})
78	336		if allocs != 0 {
79	337		t.Errorf("push + drain: %.1f allocs, want 0", allocs)

@@ -81,18 +339,84 @@ func TestPresentationPushAllocations(t *testing.T) {
81	339		}
82	340
83	341		// A fifo barrier on an output that does not flip clears at barrierAt+period,
84		-	// driven by the pace timerfd: the next frame is not late by a timer tick.
	342	+	// driven by the pace timerfd. On a one-second output (so nothing expires during
	343	+	// the test) the timerfd must be armed, no later than that deadline (other waits,
	344	+	// like held buffers, may be earlier); and its callback, run once the deadline
	345	+	// has passed, applies the queued commit.
	346	+	func TestPaceTimerfdArmedForBarrierDeadline(t *testing.T) {
	347	+	s, events, commands, _, dir := contentServerWith(t, Options{Outputs: slowOutputs}, testLog())
	348	+	c := protocolClient(t, s, dir)
	349	+	w, surf, _ := surfaceMapper(t, c, events)()
	350	+	placeWindow(t, s, commands, w, "HEADLESS-1")
	351	+	fifoQueue(t, c, surf)
	352	+	var remaining, want, period time.Duration
	353	+	var queued, left int
	354	+	var barrier bool
	355	+	var err error
	356	+	s.display.Do(func() {
	357	+	root := s.windows[w.ID].xdg.surface
	358	+	period, queued = s.outputPeriod(s.fifoOutput(root)), len(root.queue)
	359	+	var spec unix.ItimerSpec
	360	+	err = unix.TimerfdGettime(s.paceFD, &spec)
	361	+	remaining = time.Duration(spec.Value.Nano())
	362	+	want = time.Until(root.barrierAt.Add(period))
	363	+	// The deadline passes: the timerfd callback clears the barrier.
	364	+	root.barrierAt = time.Now().Add(-period)
	365	+	s.onPaceTimer()
	366	+	barrier, left = root.barrier, len(root.queue)
	367	+	})
	368	+	if err != nil {
	369	+	t.Fatal(err)
	370	+	}
	371	+	if queued != 1 \|\| period != time.Second {
	372	+	t.Fatalf("queued %d commits, period %v: the second commit must wait for the barrier", queued, period)
	373	+	}
	374	+	if remaining <= 0 \|\| remaining > want+50*time.Millisecond {
	375	+	t.Fatalf("pace timerfd fires in %v, want armed at or before barrierAt+period (%v)", remaining, want)
	376	+	}
	377	+	if barrier \|\| left != 0 {
	378	+	t.Fatalf("timer callback after the deadline: barrier %v, %d queued", barrier, left)
	379	+	}
	380	+	}
	381	+
	382	+	// End to end on a 20 Hz output: the queued commit is applied at barrierAt+50 ms
	383	+	// (the timerfd), not at the next 100 ms held-buffer tick; the bound is generous
	384	+	// (half a period) against scheduler noise.
85	385		func TestPaceTimerfdClearsBarrierOnDeadline(t *testing.T) {
86		-	s, events, _, contents, dir := contentServer(t)
	386	+	out := ports.Layout{{Info: ports.OutputInfo{Name: "HEADLESS-1", Width: 1920, Height: 1080, RefreshMilli: 20000}, Width: 1920, Height: 1080, Scale: 1}}
	387	+	s, events, commands, contents, dir := contentServerWith(t, Options{Outputs: out}, testLog())
87	388		c := protocolClient(t, s, dir)
88	389		w, surf, _ := surfaceMapper(t, c, events)()
	390	+	placeWindow(t, s, commands, w, "HEADLESS-1")
	391	+	nextContent(t, contents, w.ID) // the mapping commit
	392	+	fifoQueue(t, c, surf)
	393	+	var barrierAt time.Time
	394	+	var period time.Duration
	395	+	s.display.Do(func() {
	396	+	root := s.windows[w.ID].xdg.surface
	397	+	barrierAt, period = root.barrierAt, s.outputPeriod(s.fifoOutput(root))
	398	+	})
	399	+	nextContent(t, contents, w.ID) // the commit that set the barrier
	400	+	late := nextContent(t, contents, w.ID).Sub(barrierAt.Add(period))
	401	+	// Never early; late by far less than the period (not by a polling tick).
	402	+	if late < -time.Millisecond \|\| late > period/2 {
	403	+	t.Fatalf("queued commit applied %v after barrierAt+%v", late, period)
	404	+	}
	405	+	}
	406	+
	407	+	func testLog() zerowrap.Logger {
	408	+	return logging.For(context.Background(), "wayland")
	409	+	}
	410	+
	411	+	// fifoQueue sets a barrier with one commit and queues a second commit behind
	412	+	// wait_barrier.
	413	+	func fifoQueue(t testing.T, c wlturbo.Display, surf uint32) {
	414	+	t.Helper()
89	415		manager := bindProtocol(t, c, "wp_fifo_manager_v1")
90	416		f := c.AllocateID()
91	417		registerProtocol(t, c, f)
92	418		requestProtocol(t, c, manager, fifo.WpFifoManagerV1RequestGetFifo, f, surf)
93	419		buf := shmBuffer(t, c)
94		-	drainContents(contents)
95		-
96	420		requestProtocol(t, c, f, fifo.WpFifoV1RequestSetBarrier)
97	421		requestProtocol(t, c, surf, wayland.SurfaceRequestAttach, buf, int32(0), int32(0))
98	422		requestProtocol(t, c, surf, wayland.SurfaceRequestCommit)

@@ -102,15 +426,4 @@ func TestPaceTimerfdClearsBarrierOnDeadline(t *testing.T) {
102	426		if err := c.Roundtrip(); err != nil {
103	427		t.Fatal(err)
104	428		}
105		-	var barrierAt time.Time
106		-	var period time.Duration
107		-	s.display.Do(func() {
108		-	root := s.windows[w.ID].xdg.surface
109		-	barrierAt, period = root.barrierAt, s.outputPeriod(s.fifoOutput(root))
110		-	})
111		-	nextContent(t, contents, w.ID) // the first commit
112		-	late := nextContent(t, contents, w.ID).Sub(barrierAt.Add(period))
113		-	if late < -time.Millisecond \|\| late > 2*time.Millisecond {
114		-	t.Fatalf("queued commit applied %v after barrierAt+%v", late, period)
115		-	}
116	429		}

@@ -1,60 +1,314 @@
1		package wayland	1		package wayland
2			2
3		import (	3		import (
4	-	"runtime"	4	+	"context"
			5	+	"encoding/binary"
			6	+	"strings"
5		"testing"	7		"testing"
6		"time"	8		"time"
7			9
8		"github.com/bnema/go-wayland-bindings/server/fifo"	10		"github.com/bnema/go-wayland-bindings/server/fifo"
9		"github.com/bnema/go-wayland-bindings/server/wayland"	11		"github.com/bnema/go-wayland-bindings/server/wayland"
			12	+	"github.com/bnema/neferwl/internal/adapters/logging"
			13	+	portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
10		"github.com/bnema/neferwl/internal/ports"	14		"github.com/bnema/neferwl/internal/ports"
			15	+	"github.com/bnema/wlturbo"
			16	+	"github.com/bnema/zerowrap"
			17	+	"golang.org/x/sys/unix"
11		)	18		)
12			19
13	-	// Flips pushed while the display goroutine is busy are paced by one drain:	20	+	// slowOutputs has two outputs refreshing once a second: their deadlines never
14	-	// one tick, one burst of frame callbacks per output, not one per stale flip.	21	+	// fire inside a test, so only flips pace their frame callbacks.
			22	+	var slowOutputs = ports.Layout{
			23	+	{Info: ports.OutputInfo{Name: "HEADLESS-1", Width: 1920, Height: 1080, RefreshMilli: 1000}, Width: 1920, Height: 1080, Scale: 1},
			24	+	{Info: ports.OutputInfo{Name: "HEADLESS-2", Width: 1920, Height: 1080, RefreshMilli: 1000}, X: 1920, Width: 1920, Height: 1080, Scale: 1},
			25	+	}
			26	+
			27	+	// frameDoneProxy records the time of a wl_callback.done.
			28	+	type frameDoneProxy struct {
			29	+	wlturbo.BaseProxy
			30	+	done chan uint32
			31	+	}
			32	+
			33	+	func (p frameDoneProxy) Dispatch(e wlturbo.Event) {
			34	+	if e.Opcode == uint16(wayland.CallbackEventDone) {
			35	+	p.done <- binary.NativeEndian.Uint32(e.Data())
			36	+	}
			37	+	}
			38	+
			39	+	// requestFrame commits the surface with a new frame callback.
			40	+	func requestFrame(t testing.T, c wlturbo.Display, surf uint32) *frameDoneProxy {
			41	+	t.Helper()
			42	+	cb := c.AllocateID()
			43	+	p := &frameDoneProxy{done: make(chan uint32, 2)}
			44	+	p.SetID(cb)
			45	+	registerWireProxy(c, p)
			46	+	requestProtocol(t, c, surf, wayland.SurfaceRequestFrame, cb)
			47	+	requestProtocol(t, c, surf, wayland.SurfaceRequestCommit)
			48	+	if err := c.Roundtrip(); err != nil {
			49	+	t.Fatal(err)
			50	+	}
			51	+	return p
			52	+	}
			53	+
			54	+	// awaitFrame waits for the callback to be answered and returns its time.
			55	+	func awaitFrame(t testing.T, c wlturbo.Display, p *frameDoneProxy) uint32 {
			56	+	t.Helper()
			57	+	for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
			58	+	if err := c.Roundtrip(); err != nil {
			59	+	t.Fatal(err)
			60	+	}
			61	+	select {
			62	+	case ms := <-p.done:
			63	+	return ms
			64	+	case <-time.After(2 * time.Millisecond):
			65	+	}
			66	+	}
			67	+	t.Fatal("frame callback not answered")
			68	+	return 0
			69	+	}
			70	+
			71	+	// placeWindow puts the window on an output and waits until wayland applied it.
			72	+	func placeWindow(t testing.T, s Server, commands chan<- ports.ClientCommand, w ports.WindowMapped, output string) {
			73	+	t.Helper()
			74	+	commands <- ports.ConfigureWindow{ID: w.ID, Width: 100, Height: 100, Output: output, Visible: true}
			75	+	for deadline := time.Now().Add(2 * time.Second); ; {
			76	+	var got string
			77	+	s.display.Do(func() { got = s.frameOutput(s.windows[w.ID].xdg.surface) })
			78	+	if got == output {
			79	+	return
			80	+	}
			81	+	if time.Now().After(deadline) {
			82	+	t.Fatalf("window not placed on %s", output)
			83	+	}
			84	+	time.Sleep(time.Millisecond)
			85	+	}
			86	+	}
			87	+
			88	+	// Flips pushed while the display goroutine is busy are paced by one tick per
			89	+	// output, not one per stale flip. The frame callback of each output is
			90	+	// answered once, with the time of that output's last flip: a tick per flip
			91	+	// would have answered it with the first flip's time.
15		func TestPresentationPushCoalescesWhileDisplayBusy(t *testing.T) {	92		func TestPresentationPushCoalescesWhileDisplayBusy(t *testing.T) {
16	-	s, _, _, _, sink, _ := presentServer(t)	93	+	s, events, commands, _, sink, dir := presentServerOn(t, slowOutputs)
17	-	release, started := make(chan struct{}), make(chan struct{})	94	+	c := protocolClient(t, s, dir)
18	-	finished := make(chan struct{})	95	+	mk := surfaceMapper(t, c, events)
			96	+	outputs := []string{"HEADLESS-1", "HEADLESS-2"}
			97	+	surfs := make([]uint32, len(outputs))
			98	+	for i, name := range outputs {
			99	+	w, surf, _ := mk()
			100	+	placeWindow(t, s, commands, w, name)
			101	+	surfs[i] = surf
			102	+	// The first callback of an idle output is answered at once and
			103	+	// starts its one-second period: the next one waits for a flip.
			104	+	awaitFrame(t, c, requestFrame(t, c, surf))
			105	+	}
			106	+	waiting := make([]*frameDoneProxy, len(outputs))
			107	+	for i := range outputs {
			108	+	waiting[i] = requestFrame(t, c, surfs[i])
			109	+	select {
			110	+	case ms := <-waiting[i].done:
			111	+	t.Fatalf("%s answered before any flip (at %d ms)", outputs[i], ms)
			112	+	default:
			113	+	}
			114	+	}
			115	+	var base time.Duration
			116	+	release, started, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
19		go func() {	117		go func() {
20	-	s.display.Do(func() { close(started); <-release })	118	+	s.display.Do(func() { base = monotonic(s.started); close(started); <-release })
21		close(finished)	119		close(finished)
22		}()	120		}()
23		<-started	121		<-started
24	-	for i := range 5 {	122	+	const flips = 5
25	-	sink.Push(ports.OutputPresented{Output: "HEADLESS-1", Flip: &ports.FlipInfo{Seq: uint64(i + 1), HardwareClock: true}})	123	+	const step = 100 * time.Millisecond
			124	+	for i := range flips {
			125	+	for _, name := range outputs {
			126	+	sink.Push(ports.OutputPresented{Output: name, Flip: &ports.FlipInfo{When: base + time.Duration(i+1)*step, Seq: uint64(i + 1), HardwareClock: true}})
			127	+	}
26		}	128		}
27		close(release)	129		close(release)
28		<-finished	130		<-finished
29	-	var drains, seq uint64	131	+	last := uint32((flips * step).Milliseconds())
			132	+	for i, name := range outputs {
			133	+	ms := awaitFrame(t, c, waiting[i])
			134	+	// base is read on the display goroutine a little before the server
			135	+	// answers: a few ms of slack, far below one step.
			136	+	if ms < last-5 \|\| ms > last+5 {
			137	+	t.Fatalf("%s answered at %d ms, want the last flip's %d ms (the first flip is at %d)", name, ms, last, step.Milliseconds())
			138	+	}
			139	+	}
			140	+	var seq uint64
30		var reported bool	141		var reported bool
31		s.display.Do(func() {	142		s.display.Do(func() {
32	-	drains = s.sink.drains.Load()
33		seq = s.flips["HEADLESS-1"].Seq	143		seq = s.flips["HEADLESS-1"].Seq
34		_, reported = s.reports["HEADLESS-1"]	144		_, reported = s.reports["HEADLESS-1"]
35		})	145		})
36	-	if drains != 1 {	146	+	if !reported \|\| seq != flips {
37	-	t.Fatalf("%d drains for 5 flips pushed while busy, want 1", drains)
38	-	}
39	-	if !reported \|\| seq != 5 {
40		t.Fatalf("last flip %d, reported %v: every flip must still be answered in order", seq, reported)	147		t.Fatalf("last flip %d, reported %v: every flip must still be answered in order", seq, reported)
41		}	148		}
42		}	149		}
43			150
44	-	// A full ring merges the oldest flip of an output into its next one.	151	+	func ringOf(outputs ...string) *presentSink {
			152	+	p := &presentSink{ring: make([]ports.OutputPresented, 0, maxPresentRing+1)}
			153	+	for _, o := range outputs {
			154	+	p.ring = append(p.ring, ports.OutputPresented{Output: o})
			155	+	}
			156	+	return p
			157	+	}
			158	+
			159	+	// fill pads the ring with reports of distinct outputs up to the bound.
			160	+	func (p *presentSink) fill() {
			161	+	for i := 0; len(p.ring) < maxPresentRing; i++ {
			162	+	p.ring = append(p.ring, ports.OutputPresented{Output: "pad" + string(rune('A'+i/26)) + string(rune('a'+i%26))})
			163	+	}
			164	+	}
			165	+
			166	+	func flipOf(seq uint64, shows ...ports.WindowID) *ports.FlipInfo {
			167	+	f := &ports.FlipInfo{Seq: seq}
			168	+	if len(shows) > 0 {
			169	+	f.Shows = map[ports.WindowID]uint64{}
			170	+	for _, id := range shows {
			171	+	f.Shows[id] = seq
			172	+	}
			173	+	}
			174	+	return f
			175	+	}
			176	+
			177	+	// A full ring merges the oldest flip of an output into its next flip.
45		func TestPresentationRingMergesPastBound(t *testing.T) {	178		func TestPresentationRingMergesPastBound(t *testing.T) {
46	-	p := &presentSink{ring: make([]ports.OutputPresented, 0, maxPresentRing)}	179	+	p := ringOf()
47		for i := range maxPresentRing {	180		for i := range maxPresentRing {
48	-	p.ring = append(p.ring, ports.OutputPresented{Output: "A", Flip: &ports.FlipInfo{Seq: uint64(i + 1)}})	181	+	p.ring = append(p.ring, ports.OutputPresented{Output: "A", Flip: flipOf(uint64(i + 1))})
			182	+	}
			183	+	in := ports.OutputPresented{Output: "A", Flip: flipOf(maxPresentRing + 1)}
			184	+	if !p.makeRoomLocked(&in) {
			185	+	t.Fatal("two flips merged but not reported")
49		}	186		}
50	-	p.mu.Lock()
51	-	p.dropOldestLocked()
52	-	p.mu.Unlock()
53		if len(p.ring) != maxPresentRing-1 \|\| p.ring[0].Flip.Seq != 2 \|\| p.ring[0].Flip.Merged != 1 {	187		if len(p.ring) != maxPresentRing-1 \|\| p.ring[0].Flip.Seq != 2 \|\| p.ring[0].Flip.Merged != 1 {
54		t.Fatalf("ring %d, first seq %d merged %d", len(p.ring), p.ring[0].Flip.Seq, p.ring[0].Flip.Merged)	188		t.Fatalf("ring %d, first seq %d merged %d", len(p.ring), p.ring[0].Flip.Seq, p.ring[0].Flip.Merged)
55		}	189		}
56		}	190		}
57			191
			192	+	// A flip followed by a flip-less report of the same output is not lost: the
			193	+	// later report carries it.
			194	+	func TestPresentationRingKeepsFlipBeforeFliplessReport(t *testing.T) {
			195	+	p := ringOf("A", "A")
			196	+	p.ring[0].Flip = flipOf(7, 3)
			197	+	p.fill()
			198	+	p.ring = p.ring[:maxPresentRing]
			199	+	in := ports.OutputPresented{Output: "B", Flip: flipOf(9)}
			200	+	if p.makeRoomLocked(&in) {
			201	+	t.Fatal("no flips merged, yet reported")
			202	+	}
			203	+	if len(p.ring) != maxPresentRing-1 \|\| p.ring[0].Output != "A" \|\| p.ring[0].Flip == nil \|\| p.ring[0].Flip.Seq != 7 \|\| p.ring[0].Flip.Merged != 0 {
			204	+	t.Fatalf("flip lost: ring %d, first %+v", len(p.ring), p.ring[0])
			205	+	}
			206	+	}
			207	+
			208	+	// With several outputs interleaved, the victim is the oldest report that has
			209	+	// a successor of its own output; other outputs' reports are untouched.
			210	+	func TestPresentationRingInterleavedOutputs(t *testing.T) {
			211	+	p := ringOf("A", "B", "A", "B")
			212	+	p.ring[0].Flip, p.ring[1].Flip = flipOf(1, 10), flipOf(2, 20)
			213	+	p.ring[2].Flip, p.ring[3].Flip = flipOf(3, 11), nil
			214	+	p.fill()
			215	+	in := ports.OutputPresented{Output: "C"}
			216	+	if !p.makeRoomLocked(&in) {
			217	+	t.Fatal("merge not reported")
			218	+	}
			219	+	if p.ring[0].Output != "B" \|\| p.ring[0].Flip.Seq != 2 \|\| p.ring[0].Flip.Merged != 0 {
			220	+	t.Fatalf("B's flip touched: %+v", p.ring[0])
			221	+	}
			222	+	a := p.ring[1]
			223	+	if a.Output != "A" \|\| a.Flip.Seq != 3 \|\| a.Flip.Merged != 1 \|\| a.Flip.Shows[10] != 1 \|\| a.Flip.Shows[11] != 3 {
			224	+	t.Fatalf("A's flips not merged: %+v", a.Flip)
			225	+	}
			226	+	if p.ring[2].Output != "B" \|\| p.ring[2].Flip != nil {
			227	+	t.Fatalf("B's flip-less report moved: %+v", p.ring[2])
			228	+	}
			229	+	}
			230	+
			231	+	// The incoming report counts as the successor of a ring report.
			232	+	func TestPresentationRingIncomingTakesOverFlip(t *testing.T) {
			233	+	p := ringOf("A")
			234	+	p.ring[0].Flip = flipOf(4)
			235	+	p.fill()
			236	+	in := ports.OutputPresented{Output: "A"}
			237	+	if p.makeRoomLocked(&in) {
			238	+	t.Fatal("no merge expected")
			239	+	}
			240	+	if in.Flip == nil \|\| in.Flip.Seq != 4 \|\| len(p.ring) != maxPresentRing-1 {
			241	+	t.Fatalf("incoming %+v, ring %d", in.Flip, len(p.ring))
			242	+	}
			243	+	}
			244	+
			245	+	// With no successor for any report (more outputs than ring slots) nothing is
			246	+	// dropped: the ring grows by one and every flip is still delivered.
			247	+	func TestPresentationRingWithoutSuccessorKeepsEverything(t *testing.T) {
			248	+	p := ringOf()
			249	+	for i := range maxPresentRing {
			250	+	p.ring = append(p.ring, ports.OutputPresented{Output: "o" + string(rune('A'+i/26)) + string(rune('a'+i%26)), Flip: flipOf(uint64(i + 1))})
			251	+	}
			252	+	in := ports.OutputPresented{Output: "other", Flip: flipOf(99)}
			253	+	if p.makeRoomLocked(&in) \|\| len(p.ring) != maxPresentRing {
			254	+	t.Fatalf("merged or dropped: ring %d", len(p.ring))
			255	+	}
			256	+	}
			257	+
			258	+	// Merges are counted, and warned about at most once per mergeWarnEvery.
			259	+	func TestPresentationRingWarnsRateLimited(t *testing.T) {
			260	+	now := time.Unix(1000, 0)
			261	+	clk := portsmocks.NewMockClock(t)
			262	+	clk.EXPECT().Now().RunAndReturn(func() time.Time { return now })
			263	+	p := ringOf()
			264	+	p.s = &Server{clock: clk}
			265	+	for i := range maxPresentRing {
			266	+	p.ring = append(p.ring, ports.OutputPresented{Output: "A", Flip: flipOf(uint64(i + 1))})
			267	+	}
			268	+	push := func(seq uint64) (bool, int) {
			269	+	return p.enqueueLocked(ports.OutputPresented{Output: "A", Flip: flipOf(seq)})
			270	+	}
			271	+	if warn, n := push(100); !warn \|\| n != 1 {
			272	+	t.Fatalf("first merge: warn %v merged %d", warn, n)
			273	+	}
			274	+	for i := range 3 {
			275	+	now = now.Add(time.Second)
			276	+	if warn, _ := push(uint64(101 + i)); warn {
			277	+	t.Fatalf("warned again after %v", time.Duration(i+1)*time.Second)
			278	+	}
			279	+	}
			280	+	now = now.Add(mergeWarnEvery)
			281	+	if warn, n := push(200); !warn \|\| n != 4 {
			282	+	t.Fatalf("after the interval: warn %v merged %d, want the 4 merges since", warn, n)
			283	+	}
			284	+	}
			285	+
			286	+	// A stalled display loop makes outputs merge flips; wayland says so once.
			287	+	func TestPresentationMergeWarnsInLog(t *testing.T) {
			288	+	var out lockedBuffer
			289	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out}).WithField("component", "wayland")
			290	+	s, _, _, _, _ := contentServerWith(t, Options{}, log)
			291	+	release, started, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
			292	+	go func() {
			293	+	s.display.Do(func() { close(started); <-release })
			294	+	close(finished)
			295	+	}()
			296	+	<-started
			297	+	for i := range 3 * maxPresentRing {
			298	+	s.Presentation().Push(ports.OutputPresented{Output: "HEADLESS-1", Flip: &ports.FlipInfo{Seq: uint64(i + 1)}})
			299	+	}
			300	+	close(release)
			301	+	<-finished
			302	+	s.display.Do(func() {})
			303	+	got := out.String()
			304	+	if n := strings.Count(got, "outputs report faster than the display loop drains; flips merged"); n != 1 {
			305	+	t.Fatalf("%d warnings: %s", n, got)
			306	+	}
			307	+	if !strings.Contains(got, `"component":"wayland"`) \|\| !strings.Contains(got, `"merged":`) {
			308	+	t.Fatalf("warning lacks fields: %s", got)
			309	+	}
			310	+	}
			311	+
58		// Pushing a report and draining it on the display goroutine allocates nothing	312		// Pushing a report and draining it on the display goroutine allocates nothing
59		// in steady state. The count includes the display goroutine: AllocsPerRun	313		// in steady state. The count includes the display goroutine: AllocsPerRun
60		// reads the global malloc counter.	314		// reads the global malloc counter.

@@ -64,16 +318,20 @@ func TestPresentationPushAllocations(t *testing.T) {
64		}	318		}
65		s, _, _, _, sink, _ := presentServer(t)	319		s, _, _, _, sink, _ := presentServer(t)
66		r := ports.OutputPresented{Output: "HEADLESS-1"}	320		r := ports.OutputPresented{Output: "HEADLESS-1"}
			321	+	// Posts run in order: once the marker ran, the drain before it did.
			322	+	// The marker is bound once and the channel is reused, so waiting for it
			323	+	// allocates nothing.
			324	+	ran := make(chan struct{}, 1)
			325	+	marker := func() { ran <- struct{}{} }
67		for range 4 {	326		for range 4 {
68		sink.Push(r)	327		sink.Push(r)
69	-	s.display.Do(func() {})	328	+	s.display.Post(marker)
			329	+	<-ran
70		}	330		}
71		allocs := testing.AllocsPerRun(200, func() {	331		allocs := testing.AllocsPerRun(200, func() {
72	-	want := s.sink.drains.Load() + 1
73		sink.Push(r)	332		sink.Push(r)
74	-	for s.sink.drains.Load() < want { // wait for the drain without allocating	333	+	s.display.Post(marker)
75	-	runtime.Gosched()	334	+	<-ran
76	-	}
77		})	335		})
78		if allocs != 0 {	336		if allocs != 0 {
79		t.Errorf("push + drain: %.1f allocs, want 0", allocs)	337		t.Errorf("push + drain: %.1f allocs, want 0", allocs)

@@ -81,18 +339,84 @@ func TestPresentationPushAllocations(t *testing.T) {
81		}	339		}
82			340
83		// A fifo barrier on an output that does not flip clears at barrierAt+period,	341		// A fifo barrier on an output that does not flip clears at barrierAt+period,
84	-	// driven by the pace timerfd: the next frame is not late by a timer tick.	342	+	// driven by the pace timerfd. On a one-second output (so nothing expires during
			343	+	// the test) the timerfd must be armed, no later than that deadline (other waits,
			344	+	// like held buffers, may be earlier); and its callback, run once the deadline
			345	+	// has passed, applies the queued commit.
			346	+	func TestPaceTimerfdArmedForBarrierDeadline(t *testing.T) {
			347	+	s, events, commands, _, dir := contentServerWith(t, Options{Outputs: slowOutputs}, testLog())
			348	+	c := protocolClient(t, s, dir)
			349	+	w, surf, _ := surfaceMapper(t, c, events)()
			350	+	placeWindow(t, s, commands, w, "HEADLESS-1")
			351	+	fifoQueue(t, c, surf)
			352	+	var remaining, want, period time.Duration
			353	+	var queued, left int
			354	+	var barrier bool
			355	+	var err error
			356	+	s.display.Do(func() {
			357	+	root := s.windows[w.ID].xdg.surface
			358	+	period, queued = s.outputPeriod(s.fifoOutput(root)), len(root.queue)
			359	+	var spec unix.ItimerSpec
			360	+	err = unix.TimerfdGettime(s.paceFD, &spec)
			361	+	remaining = time.Duration(spec.Value.Nano())
			362	+	want = time.Until(root.barrierAt.Add(period))
			363	+	// The deadline passes: the timerfd callback clears the barrier.
			364	+	root.barrierAt = time.Now().Add(-period)
			365	+	s.onPaceTimer()
			366	+	barrier, left = root.barrier, len(root.queue)
			367	+	})
			368	+	if err != nil {
			369	+	t.Fatal(err)
			370	+	}
			371	+	if queued != 1 \|\| period != time.Second {
			372	+	t.Fatalf("queued %d commits, period %v: the second commit must wait for the barrier", queued, period)
			373	+	}
			374	+	if remaining <= 0 \|\| remaining > want+50*time.Millisecond {
			375	+	t.Fatalf("pace timerfd fires in %v, want armed at or before barrierAt+period (%v)", remaining, want)
			376	+	}
			377	+	if barrier \|\| left != 0 {
			378	+	t.Fatalf("timer callback after the deadline: barrier %v, %d queued", barrier, left)
			379	+	}
			380	+	}
			381	+
			382	+	// End to end on a 20 Hz output: the queued commit is applied at barrierAt+50 ms
			383	+	// (the timerfd), not at the next 100 ms held-buffer tick; the bound is generous
			384	+	// (half a period) against scheduler noise.
85		func TestPaceTimerfdClearsBarrierOnDeadline(t *testing.T) {	385		func TestPaceTimerfdClearsBarrierOnDeadline(t *testing.T) {
86	-	s, events, _, contents, dir := contentServer(t)	386	+	out := ports.Layout{{Info: ports.OutputInfo{Name: "HEADLESS-1", Width: 1920, Height: 1080, RefreshMilli: 20000}, Width: 1920, Height: 1080, Scale: 1}}
			387	+	s, events, commands, contents, dir := contentServerWith(t, Options{Outputs: out}, testLog())
87		c := protocolClient(t, s, dir)	388		c := protocolClient(t, s, dir)
88		w, surf, _ := surfaceMapper(t, c, events)()	389		w, surf, _ := surfaceMapper(t, c, events)()
			390	+	placeWindow(t, s, commands, w, "HEADLESS-1")
			391	+	nextContent(t, contents, w.ID) // the mapping commit
			392	+	fifoQueue(t, c, surf)
			393	+	var barrierAt time.Time
			394	+	var period time.Duration
			395	+	s.display.Do(func() {
			396	+	root := s.windows[w.ID].xdg.surface
			397	+	barrierAt, period = root.barrierAt, s.outputPeriod(s.fifoOutput(root))
			398	+	})
			399	+	nextContent(t, contents, w.ID) // the commit that set the barrier
			400	+	late := nextContent(t, contents, w.ID).Sub(barrierAt.Add(period))
			401	+	// Never early; late by far less than the period (not by a polling tick).
			402	+	if late < -time.Millisecond \|\| late > period/2 {
			403	+	t.Fatalf("queued commit applied %v after barrierAt+%v", late, period)
			404	+	}
			405	+	}
			406	+
			407	+	func testLog() zerowrap.Logger {
			408	+	return logging.For(context.Background(), "wayland")
			409	+	}
			410	+
			411	+	// fifoQueue sets a barrier with one commit and queues a second commit behind
			412	+	// wait_barrier.
			413	+	func fifoQueue(t testing.T, c wlturbo.Display, surf uint32) {
			414	+	t.Helper()
89		manager := bindProtocol(t, c, "wp_fifo_manager_v1")	415		manager := bindProtocol(t, c, "wp_fifo_manager_v1")
90		f := c.AllocateID()	416		f := c.AllocateID()
91		registerProtocol(t, c, f)	417		registerProtocol(t, c, f)
92		requestProtocol(t, c, manager, fifo.WpFifoManagerV1RequestGetFifo, f, surf)	418		requestProtocol(t, c, manager, fifo.WpFifoManagerV1RequestGetFifo, f, surf)
93		buf := shmBuffer(t, c)	419		buf := shmBuffer(t, c)
94	-	drainContents(contents)
95	-
96		requestProtocol(t, c, f, fifo.WpFifoV1RequestSetBarrier)	420		requestProtocol(t, c, f, fifo.WpFifoV1RequestSetBarrier)
97		requestProtocol(t, c, surf, wayland.SurfaceRequestAttach, buf, int32(0), int32(0))	421		requestProtocol(t, c, surf, wayland.SurfaceRequestAttach, buf, int32(0), int32(0))
98		requestProtocol(t, c, surf, wayland.SurfaceRequestCommit)	422		requestProtocol(t, c, surf, wayland.SurfaceRequestCommit)

@@ -102,15 +426,4 @@ func TestPaceTimerfdClearsBarrierOnDeadline(t *testing.T) {
102		if err := c.Roundtrip(); err != nil {	426		if err := c.Roundtrip(); err != nil {
103		t.Fatal(err)	427		t.Fatal(err)
104		}	428		}
105	-	var barrierAt time.Time
106	-	var period time.Duration
107	-	s.display.Do(func() {
108	-	root := s.windows[w.ID].xdg.surface
109	-	barrierAt, period = root.barrierAt, s.outputPeriod(s.fifoOutput(root))
110	-	})
111	-	nextContent(t, contents, w.ID) // the first commit
112	-	late := nextContent(t, contents, w.ID).Sub(barrierAt.Add(period))
113	-	if late < -time.Millisecond \|\| late > 2*time.Millisecond {
114	-	t.Fatalf("queued commit applied %v after barrierAt+%v", late, period)
115	-	}
116		}	429		}

M internal/adapters/wayland/scale_test.go +4 -1


@@ -63,7 +63,10 @@ func contentServerWith(t testing.T, opts Options, log zerowrap.Logger) (Server
63	63		dir := t.TempDir()
64	64		events := make(chan ports.ClientEvent, 16)
65	65		commands := make(chan ports.ClientCommand, 16)
66		-	opts.RuntimeDir, opts.Outputs = dir, testOutputs
	66	+	opts.RuntimeDir = dir
	67	+	if opts.Outputs == nil {
	68	+	opts.Outputs = testOutputs
	69	+	}
67	70		s, err := New(opts, Channels{Events: events, Commands: commands}, log)
68	71		if err != nil {
69	72		t.Fatal(err)

 	dir := t.TempDir()
 	events := make(chan ports.ClientEvent, 16)
 	commands := make(chan ports.ClientCommand, 16)
-	opts.RuntimeDir, opts.Outputs = dir, testOutputs
+	opts.RuntimeDir = dir
+	if opts.Outputs == nil {
+		opts.Outputs = testOutputs
+	}
 	s, err := New(opts, Channels{Events: events, Commands: commands}, log)
 	if err != nil {
 		t.Fatal(err)