1package wayland
2
3import (
4 "context"
5 "math"
6 "os"
7 "testing"
8 "time"
9
10 "github.com/bnema/go-wayland-bindings/server/wayland"
11 "github.com/bnema/neferwl/internal/ports"
12 "github.com/stretchr/testify/mock"
13 "golang.org/x/sys/unix"
14)
15
16func TestExplicitAcquireWatchFailureDiscardsCommit(t *testing.T) {
17 for _, failure := range []string{"signalled", "eventfd"} {
18 t.Run(failure, func(t *testing.T) {
19 h := newSyncHarness(t)
20 h.releases = map[uint32]*syncReleaseProxy{}
21 // Order setup after import and reset on the display owner so no
22 // concurrent device call can read the generated expectations.
23 if err := h.c.Roundtrip(); err != nil {
24 t.Fatal(err)
25 }
26 h.s.display.Do(func() { h.dev.ExpectedCalls = nil })
27 h.dev.EXPECT().fdToHandle(mock.Anything).Return(7, nil).Maybe()
28 h.dev.EXPECT().destroy(mock.Anything).Return(nil).Maybe()
29 if failure == "signalled" {
30 h.dev.EXPECT().signalled(uint32(7), uint64(101)).Return(false, unix.EIO).Once()
31 } else {
32 h.dev.EXPECT().signalled(uint32(7), uint64(101)).Return(false, nil).Once()
33 h.dev.EXPECT().eventfd(uint32(7), uint64(101), mock.Anything).Return(unix.EIO).Once()
34 }
35 drainContents(h.contents)
36 buffer := h.dmabuf()
37 h.commit(buffer, 101, 102)
38 if got, ok := h.content(50 * time.Millisecond); ok {
39 t.Fatalf("untracked acquire published %+v", got)
40 }
41 h.s.display.Do(func() {
42 surf := h.s.windows[h.win].xdg.surface
43 if len(surf.queue) != 0 || surf.hold.acquire != nil {
44 t.Error("failed acquire retained queued or rendered fence")
45 }
46 for _, tl := range h.s.timelines {
47 if tl.uses != 0 {
48 t.Errorf("timeline uses=%d", tl.uses)
49 }
50 }
51 })
52 if len(h.signalled()) != 0 {
53 t.Fatal("unsafe release point signalled")
54 }
55 select {
56 case <-h.releases[buffer].released:
57 t.Fatal("explicit buffer release event")
58 default:
59 }
60 })
61 }
62}
63
64func TestImplicitAcquireWatchFailureDiscardsCommit(t *testing.T) {
65 for _, failure := range []string{"poll", "hangup"} {
66 t.Run(failure, func(t *testing.T) {
67 h := newSyncHarness(t)
68 h.releases = map[uint32]*syncReleaseProxy{}
69 drainContents(h.contents)
70 buffer := h.dmabuf()
71 var plane *os.File
72 var writer *os.File
73 if failure == "poll" {
74 plane = os.NewFile(uintptr(math.MaxInt32), "invalid")
75 } else {
76 var err error
77 plane, writer, err = os.Pipe()
78 if err != nil {
79 t.Fatal(err)
80 }
81 defer plane.Close()
82 }
83 // Real pollable descriptors exercise observation errors without replacing
84 // watch/poll functions or constructing a handwritten device double.
85 if err := h.c.Roundtrip(); err != nil {
86 t.Fatal(err)
87 }
88 h.s.display.Do(func() {
89 var buf *dmabufBuffer
90 var resource *wayland.Buffer
91 for r, b := range h.s.buffers {
92 if r.ID() == buffer {
93 buf = b.(*dmabufBuffer)
94 resource = wayland.WrapBuffer(r)
95 }
96 }
97 if buf == nil {
98 t.Error("buffer missing")
99 return
100 }
101 original := buf.buf.Planes[0].File
102 defer func() { buf.buf.Planes[0].File = original }()
103 buf.buf.Planes[0].File = plane
104 surf := h.s.windows[h.win].xdg.surface
105 surf.Attach(surf.wl, resource, 0, 0)
106 surf.Commit(surf.wl)
107 })
108 if writer != nil {
109 writer.Close()
110 }
111 if got, ok := h.content(100 * time.Millisecond); ok {
112 t.Fatalf("failed implicit acquire published %+v", got)
113 }
114 h.s.display.Do(func() {
115 if len(h.s.windows[h.win].xdg.surface.queue) != 0 {
116 t.Error("failed implicit acquire left queue blocked")
117 }
118 })
119 })
120 }
121}
122
123// An idle waiter sleeps in one poll with no timeout; cancelling ctx ends it.
124func TestSyncWaiterIdleNeverTicks(t *testing.T) {
125 sw, err := newSyncWaiter(func() {})
126 if err != nil {
127 t.Fatal(err)
128 }
129 defer sw.close()
130 poller := newMockacquirePoller(t)
131 real := sw.poller
132 sw.poller = poller
133 polls := make(chan int, 16)
134 poller.EXPECT().poll(mock.Anything, mock.Anything).RunAndReturn(func(fds []unix.PollFd, timeout int) (int, error) {
135 polls <- timeout
136 return real.poll(fds, timeout)
137 })
138 ctx, cancel := context.WithCancel(context.Background())
139 defer cancel()
140 done := make(chan struct{})
141 go func() { sw.run(ctx); close(done) }()
142 select {
143 case timeout := <-polls:
144 if timeout != -1 {
145 t.Fatalf("idle poll timeout %d, want -1", timeout)
146 }
147 case <-time.After(5 * time.Second):
148 t.Fatal("the waiter never polled")
149 }
150 cancel()
151 select {
152 case <-done:
153 case <-time.After(5 * time.Second):
154 t.Fatal("cancel did not stop the waiter")
155 }
156 if n := len(polls); n != 0 {
157 t.Fatalf("%d extra polls while idle", n)
158 }
159}
160
161func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) {
162 poller := newMockacquirePoller(t)
163 wake := make(chan struct{}, 1)
164 sw, err := newSyncWaiter(func() { wake <- struct{}{} })
165 if err != nil {
166 t.Fatal(err)
167 }
168 sw.poller = poller
169 defer sw.close()
170 fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC)
171 if err != nil {
172 t.Fatal(err)
173 }
174 active := sw.add(fd)
175 var added *syncWait
176 poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func(fds []unix.PollFd, _ int) (int, error) { fds[0].Revents = unix.POLLIN; return 1, nil }).Once()
177 poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func([]unix.PollFd, int) (int, error) {
178 fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC)
179 if err != nil {
180 t.Error(err)
181 return -1, err
182 }
183 added = sw.add(fd)
184 return -1, unix.EIO
185 }).Once()
186 sw.run(context.Background())
187 select {
188 case <-wake:
189 default:
190 t.Fatal("terminal failure did not wake owner")
191 }
192 for _, w := range []*syncWait{active, added} {
193 if w == nil || !sw.fired(w) || !w.failed {
194 t.Fatal("wait stranded after terminal poll failure")
195 }
196 }
197 if !sw.stopped() {
198 t.Fatal("waiter not terminal")
199 }
200 var waits [4]*syncWait
201 if err := sw.watchImplicit(&ports.DMABuf{}, &waits); err == nil {
202 t.Fatal("new implicit watch accepted")
203 }
204}
205
206// A waiter whose pipe cannot be pinned fails its waits and still closes
207// their descriptors.
208func TestSyncWaiterPinFailureClosesWaits(t *testing.T) {
209 sw, err := newSyncWaiter(func() {})
210 if err != nil {
211 t.Fatal(err)
212 }
213 defer sw.close()
214 // Pinned already: run's own Pin fails with EEXIST.
215 if err := sw.waiter.Pin(int32(sw.pipeR.Fd())); err != nil {
216 t.Fatal(err)
217 }
218 // A pipe stands in for the eventfd: once run closes the read end, a
219 // write fails with EPIPE, whatever reuses the descriptor number.
220 var p [2]int
221 if err := unix.Pipe2(p[:], unix.O_CLOEXEC|unix.O_NONBLOCK); err != nil {
222 t.Fatal(err)
223 }
224 defer unix.Close(p[1])
225 w := sw.add(p[0])
226 sw.run(context.Background())
227 if !sw.fired(w) || !w.failed {
228 t.Fatal("wait not failed after pin failure")
229 }
230 if _, err := unix.Write(p[1], []byte{0}); err != unix.EPIPE {
231 t.Fatalf("wait descriptor still open: write %v", err)
232 }
233}
234
235func TestFailedRootRetiresEveryChildPredecessorGraph(t *testing.T) {
236 h := newSyncHarness(t)
237 sw := h.s.syncWait
238 root := &surface{server: h.s}
239 child := &surface{server: h.s}
240 a, b := &surface{server: h.s}, &surface{server: h.s}
241 tl := &timeline{dev: h.dev, handle: 7, alive: true, uses: 4}
242 var waits [2]*syncWait
243 var graphs [2]*update
244 h.s.display.Do(func() {
245 for i, surf := range []*surface{a, b} {
246 fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC|unix.EFD_NONBLOCK)
247 if err != nil {
248 t.Error(err)
249 return
250 }
251 waits[i] = sw.add(fd)
252 surf.next.sync = &commitSync{acquire: syncPoint{tl: tl, point: uint64(21 + i*2)}, release: syncPoint{tl: tl, point: uint64(22 + i*2)}, wait: waits[i]}
253 surf.queueUpdate()
254 graphs[i] = surf.queue[len(surf.queue)-1]
255 graphs[i].refs++ // pin retired state for assertions
256 }
257 child.queueUpdate()
258 child.queueUpdate()
259 one, two := child.queue[0], child.queue[1]
260 one.deps = []*update{graphs[0]}
261 graphs[0].refs++
262 two.deps = []*update{graphs[1]}
263 graphs[1].refs++
264 root.queueUpdate()
265 u := root.queue[0]
266 u.deps = []*update{two}
267 two.refs++
268 u.failed = true
269 child.queueUpdate()
270 later := child.queue[2]
271 u.applyGraph()
272 if len(child.queue) != 1 || child.queue[0] != later || len(a.queue) != 0 || len(b.queue) != 0 || tl.uses != 0 {
273 t.Errorf("retirement stranded prefix graphs: uses=%d child=%d a=%d b=%d", tl.uses, len(child.queue), len(a.queue), len(b.queue))
274 }
275 for i, g := range graphs {
276 if !g.retired || g.refs != 1 || !waits[i].cancelled.Load() {
277 t.Error("grandchild not safely retired")
278 }
279 g.releaseRef()
280 }
281 child.dropQueue()
282 })
283 if len(h.signalled()) != 0 {
284 t.Fatal("release signalled before acquire")
285 }
286}
287
288func TestAcquireExportFailurePreservesCurrentContent(t *testing.T) {
289 h := newSyncHarness(t)
290 h.releases = map[uint32]*syncReleaseProxy{}
291 if err := h.c.Roundtrip(); err != nil {
292 t.Fatal(err)
293 }
294 h.s.display.Do(func() { h.dev.ExpectedCalls = nil })
295 h.dev.EXPECT().destroy(mock.Anything).Return(nil).Maybe()
296 h.dev.EXPECT().signalled(uint32(7), uint64(111)).Return(true, nil).Once()
297 h.dev.EXPECT().exportSyncFile(uint32(7), uint64(111)).Return(nil, unix.EIO).Once()
298 h.dev.EXPECT().signal(uint32(7), uint64(112)).Return(nil).Once() // observed acquire: safe discarded release
299 drainContents(h.contents)
300 buffer := h.dmabuf()
301 var before ports.SurfaceContent
302 h.s.display.Do(func() { before = h.s.windows[h.win].xdg.surface.content })
303 h.commit(buffer, 111, 112)
304 if got, ok := h.content(50 * time.Millisecond); ok {
305 t.Fatalf("export failure published %+v", got)
306 }
307 h.s.display.Do(func() {
308 surf := h.s.windows[h.win].xdg.surface
309 if surf.content.Seq != before.Seq || surf.content.DMABuf != before.DMABuf || surf.hold.acquire != nil || len(surf.queue) != 0 {
310 t.Error("export failure changed current content")
311 }
312 for _, tl := range h.s.timelines {
313 if tl.uses != 0 {
314 t.Error("export failure pinned timeline")
315 }
316 }
317 })
318}