1package wayland
2
3import (
4 "context"
5 "sync/atomic"
6 "testing"
7 "time"
8
9 "github.com/bnema/neferwl/internal/adapters/logging"
10 "github.com/bnema/neferwl/internal/ports"
11)
12
13// linkServer is a running server with no client, to publish contents on its
14// display goroutine.
15func linkServer(t *testing.T) *Server {
16 t.Helper()
17 s, err := New(Options{RuntimeDir: t.TempDir(), Outputs: testOutputs}, Channels{}, logging.For(context.Background(), "wayland"))
18 if err != nil {
19 t.Fatal(err)
20 }
21 ctx, cancel := context.WithCancel(context.Background())
22 done := make(chan error, 1)
23 go func() { done <- s.Run(ctx) }()
24 t.Cleanup(func() { cancel(); <-done })
25 return s
26}
27
28func solid(id ports.WindowID) ports.SurfaceContent {
29 return ports.SurfaceContent{ID: id, Solid: &ports.SolidColor{}}
30}
31
32func (s *Server) publish(t *testing.T, contents ...ports.SurfaceContent) {
33 t.Helper()
34 if !s.display.Do(func() {
35 for _, c := range contents {
36 s.emitContent(c, damage{full: true})
37 }
38 }) {
39 t.Fatal("display stopped")
40 }
41}
42
43func nextBatch(t *testing.T, in <-chan *ports.ContentBatch) *ports.ContentBatch {
44 t.Helper()
45 select {
46 case b := <-in:
47 return b
48 case <-time.After(2 * time.Second):
49 t.Fatal("no batch")
50 return nil
51 }
52}
53
54// noBatch asserts that no batch is on the channel once the display goroutine
55// went through everything posted so far (a Return's flush included): sends
56// happen only there, so there is nothing left to wait for.
57func noBatch(t *testing.T, s *Server, in <-chan *ports.ContentBatch) {
58 t.Helper()
59 s.display.Do(func() {})
60 select {
61 case b := <-in:
62 t.Fatalf("unexpected batch %+v", b.Items)
63 default:
64 }
65}
66
67// Contents published while the output has not taken the previous batch
68// replace each other per window: the output gets the first batch, then one
69// with only the newest content, once it returned the first. (Ordering per
70// window holds because a batch holds one entry per window and batches are
71// FIFO; the release accounting needs only the newest Seq.)
72func TestContentBatchKeepsLatestPerWindow(t *testing.T) {
73 s := linkServer(t)
74 in, ret := s.AttachOutput("A")
75 s.publish(t, solid(1)) // sent at once
76 s.publish(t, solid(1), solid(2), solid(1)) // wait for the output
77 first := nextBatch(t, in)
78 if len(first.Items) != 1 || first.Items[0].ID != 1 || first.Items[0].Seq != 1 {
79 t.Fatalf("first batch %+v", first.Items)
80 }
81 noBatch(t, s, in) // nothing moves before the output returns a batch...
82 first.Reset()
83 ret.Return(first) // ...which sends the waiting contents
84 second := nextBatch(t, in)
85 if len(second.Items) != 2 {
86 t.Fatalf("second batch %+v", second.Items)
87 }
88 for _, c := range second.Items {
89 switch c.ID {
90 case 1:
91 if c.Seq != 3 {
92 t.Errorf("window 1 seq %d, want the newest (3)", c.Seq)
93 }
94 case 2:
95 if c.Seq != 1 {
96 t.Errorf("window 2 seq %d, want 1", c.Seq)
97 }
98 default:
99 t.Errorf("unexpected window %d", c.ID)
100 }
101 }
102}
103
104// An output attached later starts with the latest content of every live
105// window and nothing for windows that were emptied.
106func TestContentBatchAttachSeedsLiveWindows(t *testing.T) {
107 s := linkServer(t)
108 s.publish(t, solid(1), solid(2), solid(3), ports.SurfaceContent{ID: 2}, solid(1))
109 in, _ := s.AttachOutput("A")
110 b := nextBatch(t, in)
111 got := map[ports.WindowID]uint64{}
112 for _, c := range b.Items {
113 got[c.ID] = c.Seq
114 }
115 if len(got) != 2 || len(b.Items) != 2 || got[1] != 2 || got[3] != 1 {
116 t.Fatalf("seed batch %+v", b.Items)
117 }
118 // An attach with nothing to show sends nothing.
119 s.display.Do(func() { clear(s.latest) })
120 in2, _ := s.AttachOutput("B")
121 noBatch(t, s, in2)
122}
123
124// An empty content reaches the attached outputs (it releases the window's
125// buffers) and removes the window from what a later output starts with.
126func TestContentBatchDeliversEmpty(t *testing.T) {
127 s := linkServer(t)
128 in, ret := s.AttachOutput("A")
129 s.publish(t, solid(1))
130 b := nextBatch(t, in)
131 b.Reset()
132 ret.Return(b)
133 s.publish(t, ports.SurfaceContent{ID: 1})
134 b = nextBatch(t, in)
135 if len(b.Items) != 1 || b.Items[0].ID != 1 || !b.Items[0].Empty() || b.Items[0].Seq != 2 {
136 t.Fatalf("empty not delivered: %+v", b.Items)
137 }
138 var left int
139 s.display.Do(func() { left = len(s.latest) })
140 if left != 0 {
141 t.Fatalf("latest keeps %d emptied windows", left)
142 }
143 // Superseded by an empty before the output looked: the empty wins.
144 b.Reset()
145 ret.Return(b)
146 s.publish(t, solid(1)) // sent
147 s.publish(t, solid(1), ports.SurfaceContent{ID: 1}) // pending: last one is empty
148 first := nextBatch(t, in)
149 first.Reset()
150 ret.Return(first)
151 b = nextBatch(t, in)
152 if len(b.Items) != 1 || !b.Items[0].Empty() || b.Items[0].Seq != 5 {
153 t.Fatalf("emptied window: %+v", b.Items)
154 }
155}
156
157// Every attached output gets its own batches; detaching one stops its feed
158// only, and its output may still Return what it holds.
159func TestContentBatchDetach(t *testing.T) {
160 s := linkServer(t)
161 inA, retA := s.AttachOutput("A")
162 inB, _ := s.AttachOutput("B")
163 s.publish(t, solid(1))
164 a, b := nextBatch(t, inA), nextBatch(t, inB)
165 if len(a.Items) != 1 || len(b.Items) != 1 || a == b {
166 t.Fatalf("batches %+v %+v", a.Items, b.Items)
167 }
168 s.DetachOutput(retA)
169 // Waiting contents of a detached link are never sent; Return is safe.
170 a.Reset()
171 retA.Return(a)
172 s.publish(t, solid(1))
173 noBatch(t, s, inA)
174 if got := nextBatch(t, inB); len(got.Items) != 1 {
175 t.Fatalf("B batch %+v", got.Items)
176 }
177 // Attaching under a known name replaces the link.
178 inA2, _ := s.AttachOutput("A")
179 if got := nextBatch(t, inA2); len(got.Items) != 1 {
180 t.Fatalf("re-attached batch %+v", got.Items)
181 }
182 var links int
183 s.display.Do(func() { links = len(s.links) })
184 if links != 2 {
185 t.Fatalf("%d links, want 2", links)
186 }
187}
188
189// Detaching a feed by its handle never removes a newer feed of the same
190// name: A attached again replaced the first link, and detaching the old
191// handle leaves the new one fed.
192func TestDetachOldHandleKeepsReplacedLink(t *testing.T) {
193 s := linkServer(t)
194 _, oldRet := s.AttachOutput("A")
195 in, _ := s.AttachOutput("A")
196 s.DetachOutput(oldRet)
197 var links int
198 s.display.Do(func() { links = len(s.links) })
199 if links != 1 {
200 t.Fatalf("%d links, want the new one only", links)
201 }
202 s.publish(t, solid(1))
203 if b := nextBatch(t, in); len(b.Items) != 1 {
204 t.Fatalf("new link batch %+v", b.Items)
205 }
206}
207
208// Emit, send, consume and return a batch without allocating once the
209// batches have their capacity. The consumer runs inline, like an output that
210// keeps up.
211func TestContentBatchCycleAllocations(t *testing.T) {
212 s := &Server{
213 contentSeq: map[ports.WindowID]uint64{},
214 damage: map[ports.WindowID][]ports.SeqDamage{},
215 latest: map[ports.WindowID]ports.SurfaceContent{},
216 }
217 in, ret := s.attachOutput("A")
218 c := solid(1)
219 other := solid(2)
220 cycle := func() {
221 for _, c := range []ports.SurfaceContent{c, other} {
222 s.emitContent(c, damage{full: true})
223 select {
224 case b := <-in:
225 for i := range b.Items {
226 _ = b.Items[i].Seq
227 }
228 b.Reset()
229 ret.Return(b)
230 default:
231 panic("no batch sent")
232 }
233 }
234 }
235 for range 3 * damageHistoryCap {
236 cycle()
237 }
238 if allocs := testing.AllocsPerRun(200, cycle); allocs != 0 {
239 t.Errorf("content batch cycle: %.2f allocs, want 0", allocs)
240 }
241}
242
243// A late-attached output gets all the live windows in one batch, however
244// many there are, and AttachOutput returns without the output reading.
245func TestContentBatchAttachReplaysLargeSet(t *testing.T) {
246 s := linkServer(t)
247 const windows = 100
248 all := make([]ports.SurfaceContent, 0, windows)
249 for id := 1; id <= windows; id++ {
250 all = append(all, solid(ports.WindowID(id)))
251 }
252 s.publish(t, all...)
253 attached := make(chan (<-chan *ports.ContentBatch), 1)
254 go func() { in, _ := s.AttachOutput("A"); attached <- in }()
255 var in <-chan *ports.ContentBatch
256 select {
257 case in = <-attached:
258 case <-time.After(2 * time.Second):
259 t.Fatal("AttachOutput waits for the output to read")
260 }
261 b := nextBatch(t, in)
262 ids := map[ports.WindowID]struct{}{}
263 for _, c := range b.Items {
264 ids[c.ID] = struct{}{}
265 }
266 if len(b.Items) != windows || len(ids) != windows {
267 t.Fatalf("replay batch holds %d items, %d distinct windows, want %d", len(b.Items), len(ids), windows)
268 }
269}
270
271// The output returns batches in bursts and holds some while content keeps
272// coming, so the channel is full and no batch is free when emitContent runs;
273// Returns race with it from another goroutine. Whatever the interleaving, the
274// output ends up with the newest Seq of every window (the pending/flushPosted
275// handshake never strands content). Run with -race -count=20.
276func TestContentBatchHandshakeNeverStrandsContent(t *testing.T) {
277 s := linkServer(t)
278 in, ret := s.AttachOutput("A")
279 const windows, rounds = 4, 3000
280 var last [windows + 1]atomic.Uint64
281 stop, ended := make(chan struct{}), make(chan struct{})
282 t.Cleanup(func() { close(stop); <-ended })
283 go func() {
284 defer close(ended)
285 var held []*ports.ContentBatch
286 giveBack := func(n int) {
287 for ; n > 0 && len(held) > 0; n-- {
288 b := held[0]
289 held = held[1:]
290 b.Reset()
291 ret.Return(b)
292 }
293 }
294 for n := 0; ; n++ {
295 var b *ports.ContentBatch
296 select {
297 case b = <-in:
298 default:
299 // Nothing waiting: hand everything back, as an idle
300 // output does, then block.
301 giveBack(len(held))
302 select {
303 case b = <-in:
304 case <-stop:
305 return
306 }
307 }
308 for _, c := range b.Items {
309 if c.Seq > last[c.ID].Load() {
310 last[c.ID].Store(c.Seq)
311 }
312 }
313 held = append(held, b)
314 if len(held) >= 2 {
315 giveBack(1 + n%2) // a burst of one or two
316 }
317 }
318 }()
319 for range rounds {
320 s.publishRound(windows)
321 }
322 var want [windows + 1]uint64
323 s.display.Do(func() {
324 for id := 1; id <= windows; id++ {
325 want[id] = s.contentSeq[ports.WindowID(id)]
326 }
327 })
328 for deadline := time.Now().Add(5 * time.Second); ; {
329 missing := 0
330 for id := 1; id <= windows; id++ {
331 if last[id].Load() != want[id] {
332 missing++
333 }
334 }
335 if missing == 0 {
336 return
337 }
338 if time.Now().After(deadline) {
339 for id := 1; id <= windows; id++ {
340 t.Errorf("window %d: output saw seq %d, newest is %d", id, last[id].Load(), want[id])
341 }
342 t.Fatal("content stranded")
343 }
344 time.Sleep(time.Millisecond)
345 }
346}
347
348// publishRound publishes one content per window in one display turn.
349func (s *Server) publishRound(windows int) {
350 s.display.Do(func() {
351 for id := 1; id <= windows; id++ {
352 s.emitContent(solid(ports.WindowID(id)), damage{full: true})
353 }
354 })
355}