Merge pull request 'feat(server): add non-blocking Display.Post and Display.AddFD' (#1) from feat/display-post into main

Reviewed-on: https://git.bnema.dev/bnema/purego-libwayland/pulls/1

commit: 64cf1d3650c218576fcba9cf7970fa9c1dac5f09
author: bnema · 8d ago · 2026-10-13 20:03 UTC
merge of: 2e85f57 + 8f0bec4 · diff against the first parent
signature: SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown
refs: tag: v0.10.0

4 files · +823 -27

M .github/workflows/ci.yml +17 -8


@@ -7,20 +7,28 @@ on:
7	7		jobs:
8	8		test:
9	9		runs-on: ubuntu-latest
	10	+	container:
	11	+	image: archlinux:latest
	12	+	options: --init
10	13		steps:
	14	+	- run: pacman -Syu --noconfirm --needed git go make gcc wayland
11	15		- uses: actions/checkout@v4
12		-	- uses: actions/setup-go@v5
13		-	with:
14		-	go-version: '1.27'
15		-	- run: sudo apt-get update && sudo apt-get install -y --no-install-recommends libwayland-server0
	16	+	- run: \|
	17	+	git config --global --add safe.directory "$GITHUB_WORKSPACE"
	18	+	go version
	19	+	test "$(go env GOVERSION \| cut -d. -f1-2)" = go1.27
16	20		- run: make check
17	21		- run: make race
18	22
19	23		bindings:
20	24		runs-on: ubuntu-latest
	25	+	container:
	26	+	image: archlinux:latest
	27	+	options: --init
21	28		permissions:
22	29		contents: read
23	30		steps:
	31	+	- run: pacman -Syu --noconfirm --needed git go gcc wayland
24	32		- uses: actions/checkout@v4
25	33		with:
26	34		path: purego-libwayland

@@ -31,10 +39,11 @@ jobs:
31	39		ref: main
32	40		path: bindings
33	41		persist-credentials: false
34		-	- uses: actions/setup-go@v5
35		-	with:
36		-	go-version-file: bindings/go.mod
37		-	- run: sudo apt-get update && sudo apt-get install -y --no-install-recommends libwayland-server0
	42	+	- run: \|
	43	+	git config --global --add safe.directory "$GITHUB_WORKSPACE/purego-libwayland"
	44	+	git config --global --add safe.directory "$GITHUB_WORKSPACE/bindings"
	45	+	go version
	46	+	test "$(go env GOVERSION \| cut -d. -f1-2)" = go1.27
38	47		- name: Build and test go-wayland-bindings against this checkout
39	48		working-directory: bindings
40	49		run: \|

 jobs:
   test:
     runs-on: ubuntu-latest
+    container:
+      image: archlinux:latest
+      options: --init
     steps:
+      - run: pacman -Syu --noconfirm --needed git go make gcc wayland
       - uses: actions/checkout@v4
-      - uses: actions/setup-go@v5
+      - run: |
-        with:
+          git config --global --add safe.directory "$GITHUB_WORKSPACE"
-          go-version: '1.27'
+          go version
-      - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends libwayland-server0
+          test "$(go env GOVERSION | cut -d. -f1-2)" = go1.27
       - run: make check
       - run: make race
   bindings:
     runs-on: ubuntu-latest
+    container:
+      image: archlinux:latest
+      options: --init
     permissions:
       contents: read
     steps:
+      - run: pacman -Syu --noconfirm --needed git go gcc wayland
       - uses: actions/checkout@v4
         with:
           path: purego-libwayland
           ref: main
           path: bindings
           persist-credentials: false
-      - uses: actions/setup-go@v5
+      - run: |
-        with:
+          git config --global --add safe.directory "$GITHUB_WORKSPACE/purego-libwayland"
-          go-version-file: bindings/go.mod
+          git config --global --add safe.directory "$GITHUB_WORKSPACE/bindings"
-      - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends libwayland-server0
+          go version
+          test "$(go env GOVERSION | cut -d. -f1-2)" = go1.27
       - name: Build and test go-wayland-bindings against this checkout
         working-directory: bindings
         run: |

M server/display.go +225 -19


@@ -6,6 +6,7 @@ import (
6	6		"fmt"
7	7		"runtime"
8	8		"strings"
	9	+	"sync"
9	10		"sync/atomic"
10	11		"unsafe"
11	12

@@ -48,15 +49,19 @@ var live = struct {
48	49		nextBind uintptr
49	50		// clientGone maps wl_listener addresses to OnDestroy functions.
50	51		clientGone map[uintptr]func()
51		-	}{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}}
	52	+	// fds maps AddFD tokens to their readable callbacks.
	53	+	fds map[uintptr]func()
	54	+	nextFD uintptr
	55	+	}{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}, fds: map[uintptr]func(){}}
52	56
53		-	// Display owns a wl_display. All methods except Do must run on the goroutine
54		-	// that calls Run, or before Run starts.
	57	+	// Display owns a wl_display. All methods except Do and Post must run on the
	58	+	// goroutine that calls Run, or before Run starts.
55	59		//
56	60		// Resource, bind and client-listener state is shared by the whole process, so
57	61		// only one Display may be running at a time per process.
58		-	// Do wakes the loop through an eventfd so calls run at once instead of after
59		-	// the dispatch timeout.
	62	+	//
	63	+	// Do and Post wake the loop through an eventfd so calls run at once instead of
	64	+	// after the dispatch timeout.
60	65		type Display struct {
61	66		c uintptr
62	67		loop uintptr

@@ -67,8 +72,33 @@ type Display struct {
67	72		wakeSrc uintptr
68	73		stopped chan struct{}
69	74		state atomic.Uint32
	75	+
	76	+	// postMu guards posts and closed. spare is only used by the display
	77	+	// goroutine: it is the emptied batch swapped back in on the next drain, so
	78	+	// steady-state Post does not allocate.
	79	+	postMu sync.Mutex
	80	+	posts []func()
	81	+	spare []func()
	82	+	// armed is true while an eventfd wake for posts is pending, so a burst of
	83	+	// Posts writes the eventfd once.
	84	+	armed atomic.Bool
	85	+	closed atomic.Bool // set under postMu before the wake fd is closed
	86	+	// wakeWrites counts eventfd writes made for posts (diagnostics and tests).
	87	+	wakeWrites atomic.Uint64
	88	+
	89	+	// sources are the AddFD sources not yet removed (display goroutine only).
	90	+	sources map[*Source]struct{}
	91	+
	92	+	// dispatchMS bounds one event loop wait; it only bounds how late ctx
	93	+	// cancellation is seen. Tests raise it so a lost wake hangs.
	94	+	dispatchMS int32
70	95		}
71	96
	97	+	// maxDrainRounds bounds how many times one loop iteration re-drains calls and
	98	+	// posts that were queued by callbacks, so a self-posting callback cannot starve
	99	+	// libwayland's dispatch and client flushing.
	100	+	const maxDrainRounds = 8
	101	+
72	102		func NewDisplay() (*Display, error) {
73	103		if err := load(); err != nil {
74	104		return nil, err

@@ -82,7 +112,7 @@ func NewDisplay() (*Display, error) {
82	112		wlDisplayDestroy(c)
83	113		return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err)
84	114		}
85		-	d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{})}
	115	+	d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}, dispatchMS: 100}
86	116		// The callback only drains the counter; Run then executes queued calls.
87	117		if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 {
88	118		unix.Close(wake)

@@ -163,6 +193,17 @@ func (d *Display) Close() {
163	193
164	194		// destroy frees the display and the wake-up fds.
165	195		func (d *Display) destroy() {
	196	+	// Stop accepting posts before the wake fd is closed (its number could be
	197	+	// reused), and drop what never ran.
	198	+	d.postMu.Lock()
	199	+	d.closed.Store(true)
	200	+	clear(d.posts)
	201	+	d.posts, d.spare = nil, nil
	202	+	d.postMu.Unlock()
	203	+	for s := range d.sources {
	204	+	s.remove()
	205	+	}
	206	+	clear(d.sources)
166	207		wlDisplayDestroyClients(d.c)
167	208		wlEventSourceRemove(d.wakeSrc)
168	209		wlDisplayDestroy(d.c)

@@ -183,19 +224,13 @@ func (d *Display) Run(ctx context.Context) error {
183	224		if ctx.Err() != nil {
184	225		return nil
185	226		}
186		-	// Run every queued call; each Do also wrote the eventfd.
187		-	drain:
188		-	for {
189		-	select {
190		-	case fn := <-d.calls:
191		-	fn()
192		-	default:
193		-	break drain
194		-	}
195		-	}
	227	+	// Run every queued post and call. Wakes are coalesced, so the eventfd
	228	+	// is not written once per item; it only makes the dispatch below
	229	+	// return at once when work arrives while the loop is asleep.
	230	+	d.drain()
196	231		wlDisplayFlushClients(d.c)
197	232		// The timeout only bounds how late ctx cancellation is seen.
198		-	if wlEventLoopDispatch(d.loop, 100) < 0 {
	233	+	if wlEventLoopDispatch(d.loop, d.dispatchMS) < 0 {
199	234		errno := lastErrno()
200	235		consecutiveFailures++
201	236		// epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats.

@@ -209,6 +244,170 @@ func (d *Display) Run(ctx context.Context) error {
209	244		}
210	245		}
211	246
	247	+	// drain runs queued posts and Do calls until none are left, at most
	248	+	// maxDrainRounds rounds. Posts are taken again after each call is received and
	249	+	// before it runs, so a Post that happened before a Do always runs before that
	250	+	// Do's function. When the cap is hit the loop is woken again so the next
	251	+	// dispatch returns at once.
	252	+	func (d *Display) drain() {
	253	+	for range maxDrainRounds {
	254	+	ran := d.runPosts()
	255	+	calls:
	256	+	for {
	257	+	select {
	258	+	case fn := <-d.calls:
	259	+	d.runPosts()
	260	+	fn()
	261	+	ran = true
	262	+	default:
	263	+	break calls
	264	+	}
	265	+	}
	266	+	if !ran {
	267	+	return
	268	+	}
	269	+	}
	270	+	d.armed.Store(true)
	271	+	d.wakeLoop()
	272	+	}
	273	+
	274	+	// runPosts runs the posts queued so far and reports whether there were any.
	275	+	func (d *Display) runPosts() bool {
	276	+	// Clear armed before taking the batch: a Post that lands after the swap
	277	+	// sees armed false and writes the eventfd again.
	278	+	d.armed.Store(false)
	279	+	d.postMu.Lock()
	280	+	batch := d.posts
	281	+	d.posts, d.spare = d.spare, nil
	282	+	d.postMu.Unlock()
	283	+	for _, fn := range batch {
	284	+	fn()
	285	+	}
	286	+	clear(batch)
	287	+	d.spare = batch[:0]
	288	+	return len(batch) > 0
	289	+	}
	290	+
	291	+	// wakeOne is the eventfd increment. It is read-only, so sharing it between
	292	+	// goroutines is safe and avoids a per-call buffer (which escapes under -race).
	293	+	var wakeOne = [8]byte{1}
	294	+
	295	+	// wakeLoop writes the eventfd on behalf of posts. The caller must hold postMu
	296	+	// (or be the display goroutine, which destroy cannot overlap) so the fd cannot
	297	+	// be closed, and its number reused, underneath the write.
	298	+	func (d *Display) wakeLoop() {
	299	+	d.wakeWrites.Add(1)
	300	+	_, _ = unix.Write(d.wake, wakeOne[:])
	301	+	}
	302	+
	303	+	// wakeGuarded wakes the loop from any goroutine: it takes postMu and skips the
	304	+	// write once the display is closed, so a late caller never writes to a reused
	305	+	// fd number.
	306	+	func (d *Display) wakeGuarded() {
	307	+	d.postMu.Lock()
	308	+	defer d.postMu.Unlock()
	309	+	if d.closed.Load() {
	310	+	return
	311	+	}
	312	+	_, _ = unix.Write(d.wake, wakeOne[:])
	313	+	}
	314	+
	315	+	// Post queues fn to run on the display goroutine and returns at once. Unlike
	316	+	// Do it never waits, and unlike Do it may be called from any goroutine,
	317	+	// including the display goroutine itself.
	318	+	//
	319	+	// Posts run in the order they were made, and a Post made before a Do (by the
	320	+	// same goroutine, or otherwise known to happen before it) runs before that Do's
	321	+	// function. A Post made by a posted function or a Do function runs in a later
	322	+	// drain round of the same loop iteration (at most 8 rounds; past that the loop
	323	+	// is woken again and continues after one libwayland dispatch). Wakes are coalesced:
	324	+	// any number of Posts while one wake is pending write the eventfd once.
	325	+	//
	326	+	// fn must not block: it runs on the goroutine that serves every client. After
	327	+	// the display has stopped or been closed, Post silently drops fn, and posts
	328	+	// that had not run when the display stopped are dropped too. In steady state
	329	+	// Post does not allocate.
	330	+	//
	331	+	// The queue is unbounded: Post applies no backpressure, so a producer that
	332	+	// outpaces the display goroutine grows memory without limit. Callers that can
	333	+	// produce faster than the loop runs must bound or coalesce their own posts.
	334	+	func (d *Display) Post(fn func()) {
	335	+	if fn == nil {
	336	+	return
	337	+	}
	338	+	d.postMu.Lock()
	339	+	defer d.postMu.Unlock()
	340	+	if d.closed.Load() {
	341	+	return
	342	+	}
	343	+	d.posts = append(d.posts, fn)
	344	+	// Written under postMu so destroy cannot close the fd underneath.
	345	+	if d.armed.CompareAndSwap(false, true) {
	346	+	d.wakeLoop()
	347	+	}
	348	+	}
	349	+
	350	+	// Source is an fd registered with the display's event loop by AddFD.
	351	+	type Source struct {
	352	+	d *Display
	353	+	c uintptr
	354	+	token uintptr
	355	+	}
	356	+
	357	+	// AddFD calls onReadable on the display goroutine whenever fd is readable. The
	358	+	// registration is level-triggered: onReadable must consume the readiness (read
	359	+	// the eventfd or timerfd, accept, ...) or it is called again at once. It
	360	+	// should not block.
	361	+	//
	362	+	// onReadable is also called when the fd hangs up or errors (EPOLLHUP,
	363	+	// EPOLLERR), which level-triggered polling keeps reporting. The callback should
	364	+	// read the fd, and on EOF or error call Remove (it may do so from within the
	365	+	// callback), or it is called again at once.
	366	+	//
	367	+	// libwayland duplicates fd; the caller still owns and must close its own
	368	+	// descriptor, and should do so after Remove. Call AddFD only on the display
	369	+	// goroutine or before Run starts. Sources not removed are released when the
	370	+	// display is destroyed.
	371	+	func (d Display) AddFD(fd int, onReadable func()) (Source, error) {
	372	+	if fd < 0 {
	373	+	return nil, fmt.Errorf("purego-libwayland: AddFD: invalid fd %d", fd)
	374	+	}
	375	+	if onReadable == nil {
	376	+	return nil, errors.New("purego-libwayland: AddFD: nil callback")
	377	+	}
	378	+	if d.closed.Load() {
	379	+	return nil, errors.New("purego-libwayland: AddFD: display closed")
	380	+	}
	381	+	live.nextFD++
	382	+	token := live.nextFD
	383	+	live.fds[token] = onReadable
	384	+	c := wlEventLoopAddFD(d.loop, int32(fd), 1 /* WL_EVENT_READABLE */, cbFD, token)
	385	+	if c == 0 {
	386	+	delete(live.fds, token)
	387	+	return nil, fmt.Errorf("purego-libwayland: wl_event_loop_add_fd(%d) failed", fd)
	388	+	}
	389	+	s := &Source{d: d, c: c, token: token}
	390	+	d.sources[s] = struct{}{}
	391	+	return s, nil
	392	+	}
	393	+
	394	+	// Remove stops the callbacks and releases libwayland's duplicate of the fd.
	395	+	// It may be called from the source's own callback. Display goroutine only;
	396	+	// idempotent, and a no-op after the display was destroyed.
	397	+	func (s *Source) Remove() {
	398	+	if s == nil \|\| s.c == 0 {
	399	+	return
	400	+	}
	401	+	delete(s.d.sources, s)
	402	+	s.remove()
	403	+	}
	404	+
	405	+	func (s *Source) remove() {
	406	+	delete(live.fds, s.token)
	407	+	wlEventSourceRemove(s.c)
	408	+	s.c = 0
	409	+	}
	410	+
212	411		// Do runs fn on the display goroutine and waits for it. It returns false if
213	412		// the display loop has stopped (fn was not run, or its run was not confirmed).
214	413		// Do must not be called from the display goroutine: it would deadlock.

@@ -219,8 +418,7 @@ func (d *Display) Do(fn func()) bool {
219	418		case <-d.stopped:
220	419		return false
221	420		}
222		-	var one = [8]byte{1}
223		-	_, _ = unix.Write(d.wake, one[:])
	421	+	d.wakeGuarded()
224	422		select {
225	423		case <-done:
226	424		return true

@@ -418,6 +616,14 @@ func wake(fd int32, mask uint32, data uintptr) int32 {
418	616		return 0
419	617		}
420	618
	619	+	// fdReadable dispatches an AddFD source (wl_event_loop_fd_func_t).
	620	+	func fdReadable(_ int32, _ uint32, data uintptr) int32 {
	621	+	if fn := live.fds[data]; fn != nil {
	622	+	fn()
	623	+	}
	624	+	return 0
	625	+	}
	626	+
421	627		func bind(client uintptr, data uintptr, version uint32, id uint32) {
422	628		if fn := live.binds[data]; fn != nil {
423	629		fn(Client{client}, version, id)


@@ -6,6 +6,7 @@ import (
6		"fmt"	6		"fmt"
7		"runtime"	7		"runtime"
8		"strings"	8		"strings"
			9	+	"sync"
9		"sync/atomic"	10		"sync/atomic"
10		"unsafe"	11		"unsafe"
11			12

@@ -48,15 +49,19 @@ var live = struct {
48		nextBind uintptr	49		nextBind uintptr
49		// clientGone maps wl_listener addresses to OnDestroy functions.	50		// clientGone maps wl_listener addresses to OnDestroy functions.
50		clientGone map[uintptr]func()	51		clientGone map[uintptr]func()
51	-	}{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}}	52	+	// fds maps AddFD tokens to their readable callbacks.
			53	+	fds map[uintptr]func()
			54	+	nextFD uintptr
			55	+	}{resources: map[uintptr]*Resource{}, binds: map[uintptr]BindFunc{}, clientGone: map[uintptr]func(){}, fds: map[uintptr]func(){}}
52			56
53	-	// Display owns a wl_display. All methods except Do must run on the goroutine	57	+	// Display owns a wl_display. All methods except Do and Post must run on the
54	-	// that calls Run, or before Run starts.	58	+	// goroutine that calls Run, or before Run starts.
55		//	59		//
56		// Resource, bind and client-listener state is shared by the whole process, so	60		// Resource, bind and client-listener state is shared by the whole process, so
57		// only one Display may be running at a time per process.	61		// only one Display may be running at a time per process.
58	-	// Do wakes the loop through an eventfd so calls run at once instead of after	62	+	//
59	-	// the dispatch timeout.	63	+	// Do and Post wake the loop through an eventfd so calls run at once instead of
			64	+	// after the dispatch timeout.
60		type Display struct {	65		type Display struct {
61		c uintptr	66		c uintptr
62		loop uintptr	67		loop uintptr

@@ -67,8 +72,33 @@ type Display struct {
67		wakeSrc uintptr	72		wakeSrc uintptr
68		stopped chan struct{}	73		stopped chan struct{}
69		state atomic.Uint32	74		state atomic.Uint32
			75	+
			76	+	// postMu guards posts and closed. spare is only used by the display
			77	+	// goroutine: it is the emptied batch swapped back in on the next drain, so
			78	+	// steady-state Post does not allocate.
			79	+	postMu sync.Mutex
			80	+	posts []func()
			81	+	spare []func()
			82	+	// armed is true while an eventfd wake for posts is pending, so a burst of
			83	+	// Posts writes the eventfd once.
			84	+	armed atomic.Bool
			85	+	closed atomic.Bool // set under postMu before the wake fd is closed
			86	+	// wakeWrites counts eventfd writes made for posts (diagnostics and tests).
			87	+	wakeWrites atomic.Uint64
			88	+
			89	+	// sources are the AddFD sources not yet removed (display goroutine only).
			90	+	sources map[*Source]struct{}
			91	+
			92	+	// dispatchMS bounds one event loop wait; it only bounds how late ctx
			93	+	// cancellation is seen. Tests raise it so a lost wake hangs.
			94	+	dispatchMS int32
70		}	95		}
71			96
			97	+	// maxDrainRounds bounds how many times one loop iteration re-drains calls and
			98	+	// posts that were queued by callbacks, so a self-posting callback cannot starve
			99	+	// libwayland's dispatch and client flushing.
			100	+	const maxDrainRounds = 8
			101	+
72		func NewDisplay() (*Display, error) {	102		func NewDisplay() (*Display, error) {
73		if err := load(); err != nil {	103		if err := load(); err != nil {
74		return nil, err	104		return nil, err

@@ -82,7 +112,7 @@ func NewDisplay() (*Display, error) {
82		wlDisplayDestroy(c)	112		wlDisplayDestroy(c)
83		return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err)	113		return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err)
84		}	114		}
85	-	d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{})}	115	+	d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}, dispatchMS: 100}
86		// The callback only drains the counter; Run then executes queued calls.	116		// The callback only drains the counter; Run then executes queued calls.
87		if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 {	117		if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 {
88		unix.Close(wake)	118		unix.Close(wake)

@@ -163,6 +193,17 @@ func (d *Display) Close() {
163			193
164		// destroy frees the display and the wake-up fds.	194		// destroy frees the display and the wake-up fds.
165		func (d *Display) destroy() {	195		func (d *Display) destroy() {
			196	+	// Stop accepting posts before the wake fd is closed (its number could be
			197	+	// reused), and drop what never ran.
			198	+	d.postMu.Lock()
			199	+	d.closed.Store(true)
			200	+	clear(d.posts)
			201	+	d.posts, d.spare = nil, nil
			202	+	d.postMu.Unlock()
			203	+	for s := range d.sources {
			204	+	s.remove()
			205	+	}
			206	+	clear(d.sources)
166		wlDisplayDestroyClients(d.c)	207		wlDisplayDestroyClients(d.c)
167		wlEventSourceRemove(d.wakeSrc)	208		wlEventSourceRemove(d.wakeSrc)
168		wlDisplayDestroy(d.c)	209		wlDisplayDestroy(d.c)

@@ -183,19 +224,13 @@ func (d *Display) Run(ctx context.Context) error {
183		if ctx.Err() != nil {	224		if ctx.Err() != nil {
184		return nil	225		return nil
185		}	226		}
186	-	// Run every queued call; each Do also wrote the eventfd.	227	+	// Run every queued post and call. Wakes are coalesced, so the eventfd
187	-	drain:	228	+	// is not written once per item; it only makes the dispatch below
188	-	for {	229	+	// return at once when work arrives while the loop is asleep.
189	-	select {	230	+	d.drain()
190	-	case fn := <-d.calls:
191	-	fn()
192	-	default:
193	-	break drain
194	-	}
195	-	}
196		wlDisplayFlushClients(d.c)	231		wlDisplayFlushClients(d.c)
197		// The timeout only bounds how late ctx cancellation is seen.	232		// The timeout only bounds how late ctx cancellation is seen.
198	-	if wlEventLoopDispatch(d.loop, 100) < 0 {	233	+	if wlEventLoopDispatch(d.loop, d.dispatchMS) < 0 {
199		errno := lastErrno()	234		errno := lastErrno()
200		consecutiveFailures++	235		consecutiveFailures++
201		// epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats.	236		// epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats.

@@ -209,6 +244,170 @@ func (d *Display) Run(ctx context.Context) error {
209		}	244		}
210		}	245		}
211			246
			247	+	// drain runs queued posts and Do calls until none are left, at most
			248	+	// maxDrainRounds rounds. Posts are taken again after each call is received and
			249	+	// before it runs, so a Post that happened before a Do always runs before that
			250	+	// Do's function. When the cap is hit the loop is woken again so the next
			251	+	// dispatch returns at once.
			252	+	func (d *Display) drain() {
			253	+	for range maxDrainRounds {
			254	+	ran := d.runPosts()
			255	+	calls:
			256	+	for {
			257	+	select {
			258	+	case fn := <-d.calls:
			259	+	d.runPosts()
			260	+	fn()
			261	+	ran = true
			262	+	default:
			263	+	break calls
			264	+	}
			265	+	}
			266	+	if !ran {
			267	+	return
			268	+	}
			269	+	}
			270	+	d.armed.Store(true)
			271	+	d.wakeLoop()
			272	+	}
			273	+
			274	+	// runPosts runs the posts queued so far and reports whether there were any.
			275	+	func (d *Display) runPosts() bool {
			276	+	// Clear armed before taking the batch: a Post that lands after the swap
			277	+	// sees armed false and writes the eventfd again.
			278	+	d.armed.Store(false)
			279	+	d.postMu.Lock()
			280	+	batch := d.posts
			281	+	d.posts, d.spare = d.spare, nil
			282	+	d.postMu.Unlock()
			283	+	for _, fn := range batch {
			284	+	fn()
			285	+	}
			286	+	clear(batch)
			287	+	d.spare = batch[:0]
			288	+	return len(batch) > 0
			289	+	}
			290	+
			291	+	// wakeOne is the eventfd increment. It is read-only, so sharing it between
			292	+	// goroutines is safe and avoids a per-call buffer (which escapes under -race).
			293	+	var wakeOne = [8]byte{1}
			294	+
			295	+	// wakeLoop writes the eventfd on behalf of posts. The caller must hold postMu
			296	+	// (or be the display goroutine, which destroy cannot overlap) so the fd cannot
			297	+	// be closed, and its number reused, underneath the write.
			298	+	func (d *Display) wakeLoop() {
			299	+	d.wakeWrites.Add(1)
			300	+	_, _ = unix.Write(d.wake, wakeOne[:])
			301	+	}
			302	+
			303	+	// wakeGuarded wakes the loop from any goroutine: it takes postMu and skips the
			304	+	// write once the display is closed, so a late caller never writes to a reused
			305	+	// fd number.
			306	+	func (d *Display) wakeGuarded() {
			307	+	d.postMu.Lock()
			308	+	defer d.postMu.Unlock()
			309	+	if d.closed.Load() {
			310	+	return
			311	+	}
			312	+	_, _ = unix.Write(d.wake, wakeOne[:])
			313	+	}
			314	+
			315	+	// Post queues fn to run on the display goroutine and returns at once. Unlike
			316	+	// Do it never waits, and unlike Do it may be called from any goroutine,
			317	+	// including the display goroutine itself.
			318	+	//
			319	+	// Posts run in the order they were made, and a Post made before a Do (by the
			320	+	// same goroutine, or otherwise known to happen before it) runs before that Do's
			321	+	// function. A Post made by a posted function or a Do function runs in a later
			322	+	// drain round of the same loop iteration (at most 8 rounds; past that the loop
			323	+	// is woken again and continues after one libwayland dispatch). Wakes are coalesced:
			324	+	// any number of Posts while one wake is pending write the eventfd once.
			325	+	//
			326	+	// fn must not block: it runs on the goroutine that serves every client. After
			327	+	// the display has stopped or been closed, Post silently drops fn, and posts
			328	+	// that had not run when the display stopped are dropped too. In steady state
			329	+	// Post does not allocate.
			330	+	//
			331	+	// The queue is unbounded: Post applies no backpressure, so a producer that
			332	+	// outpaces the display goroutine grows memory without limit. Callers that can
			333	+	// produce faster than the loop runs must bound or coalesce their own posts.
			334	+	func (d *Display) Post(fn func()) {
			335	+	if fn == nil {
			336	+	return
			337	+	}
			338	+	d.postMu.Lock()
			339	+	defer d.postMu.Unlock()
			340	+	if d.closed.Load() {
			341	+	return
			342	+	}
			343	+	d.posts = append(d.posts, fn)
			344	+	// Written under postMu so destroy cannot close the fd underneath.
			345	+	if d.armed.CompareAndSwap(false, true) {
			346	+	d.wakeLoop()
			347	+	}
			348	+	}
			349	+
			350	+	// Source is an fd registered with the display's event loop by AddFD.
			351	+	type Source struct {
			352	+	d *Display
			353	+	c uintptr
			354	+	token uintptr
			355	+	}
			356	+
			357	+	// AddFD calls onReadable on the display goroutine whenever fd is readable. The
			358	+	// registration is level-triggered: onReadable must consume the readiness (read
			359	+	// the eventfd or timerfd, accept, ...) or it is called again at once. It
			360	+	// should not block.
			361	+	//
			362	+	// onReadable is also called when the fd hangs up or errors (EPOLLHUP,
			363	+	// EPOLLERR), which level-triggered polling keeps reporting. The callback should
			364	+	// read the fd, and on EOF or error call Remove (it may do so from within the
			365	+	// callback), or it is called again at once.
			366	+	//
			367	+	// libwayland duplicates fd; the caller still owns and must close its own
			368	+	// descriptor, and should do so after Remove. Call AddFD only on the display
			369	+	// goroutine or before Run starts. Sources not removed are released when the
			370	+	// display is destroyed.
			371	+	func (d Display) AddFD(fd int, onReadable func()) (Source, error) {
			372	+	if fd < 0 {
			373	+	return nil, fmt.Errorf("purego-libwayland: AddFD: invalid fd %d", fd)
			374	+	}
			375	+	if onReadable == nil {
			376	+	return nil, errors.New("purego-libwayland: AddFD: nil callback")
			377	+	}
			378	+	if d.closed.Load() {
			379	+	return nil, errors.New("purego-libwayland: AddFD: display closed")
			380	+	}
			381	+	live.nextFD++
			382	+	token := live.nextFD
			383	+	live.fds[token] = onReadable
			384	+	c := wlEventLoopAddFD(d.loop, int32(fd), 1 /* WL_EVENT_READABLE */, cbFD, token)
			385	+	if c == 0 {
			386	+	delete(live.fds, token)
			387	+	return nil, fmt.Errorf("purego-libwayland: wl_event_loop_add_fd(%d) failed", fd)
			388	+	}
			389	+	s := &Source{d: d, c: c, token: token}
			390	+	d.sources[s] = struct{}{}
			391	+	return s, nil
			392	+	}
			393	+
			394	+	// Remove stops the callbacks and releases libwayland's duplicate of the fd.
			395	+	// It may be called from the source's own callback. Display goroutine only;
			396	+	// idempotent, and a no-op after the display was destroyed.
			397	+	func (s *Source) Remove() {
			398	+	if s == nil \|\| s.c == 0 {
			399	+	return
			400	+	}
			401	+	delete(s.d.sources, s)
			402	+	s.remove()
			403	+	}
			404	+
			405	+	func (s *Source) remove() {
			406	+	delete(live.fds, s.token)
			407	+	wlEventSourceRemove(s.c)
			408	+	s.c = 0
			409	+	}
			410	+
212		// Do runs fn on the display goroutine and waits for it. It returns false if	411		// Do runs fn on the display goroutine and waits for it. It returns false if
213		// the display loop has stopped (fn was not run, or its run was not confirmed).	412		// the display loop has stopped (fn was not run, or its run was not confirmed).
214		// Do must not be called from the display goroutine: it would deadlock.	413		// Do must not be called from the display goroutine: it would deadlock.

@@ -219,8 +418,7 @@ func (d *Display) Do(fn func()) bool {
219		case <-d.stopped:	418		case <-d.stopped:
220		return false	419		return false
221		}	420		}
222	-	var one = [8]byte{1}	421	+	d.wakeGuarded()
223	-	_, _ = unix.Write(d.wake, one[:])
224		select {	422		select {
225		case <-done:	423		case <-done:
226		return true	424		return true

@@ -418,6 +616,14 @@ func wake(fd int32, mask uint32, data uintptr) int32 {
418		return 0	616		return 0
419		}	617		}
420			618
			619	+	// fdReadable dispatches an AddFD source (wl_event_loop_fd_func_t).
			620	+	func fdReadable(_ int32, _ uint32, data uintptr) int32 {
			621	+	if fn := live.fds[data]; fn != nil {
			622	+	fn()
			623	+	}
			624	+	return 0
			625	+	}
			626	+
421		func bind(client uintptr, data uintptr, version uint32, id uint32) {	627		func bind(client uintptr, data uintptr, version uint32, id uint32) {
422		if fn := live.binds[data]; fn != nil {	628		if fn := live.binds[data]; fn != nil {
423		fn(Client{client}, version, id)	629		fn(Client{client}, version, id)

M server/lib.go +4 -0


@@ -43,6 +43,7 @@ var (
43	43		cbClientGone uintptr
44	44		cbBind uintptr
45	45		cbWake uintptr
	46	+	cbFD uintptr
46	47		)
47	48
48	49		func load() error {

@@ -127,6 +128,9 @@ func load() error {
127	128		cbWake = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
128	129		return uintptr(uint32(wake(int32(a.Int(0)), uint32(a.Int(1)), a.Int(2))))
129	130		})
	131	+	cbFD = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
	132	+	return uintptr(uint32(fdReadable(int32(a.Int(0)), uint32(a.Int(1)), a.Int(2))))
	133	+	})
130	134		})
131	135		return loadErr
132	136		}


@@ -43,6 +43,7 @@ var (
43	cbClientGone uintptr	43		cbClientGone uintptr
44	cbBind uintptr	44		cbBind uintptr
45	cbWake uintptr	45		cbWake uintptr
		46	+	cbFD uintptr
46	)	47		)
47		48
48	func load() error {	49		func load() error {

@@ -127,6 +128,9 @@ func load() error {
127	cbWake = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {	128		cbWake = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
128	return uintptr(uint32(wake(int32(a.Int(0)), uint32(a.Int(1)), a.Int(2))))	129		return uintptr(uint32(wake(int32(a.Int(0)), uint32(a.Int(1)), a.Int(2))))
129	})	130		})
		131	+	cbFD = purego.NewCallbackInts(func(a *purego.CallbackArgs) uintptr {
		132	+	return uintptr(uint32(fdReadable(int32(a.Int(0)), uint32(a.Int(1)), a.Int(2))))
		133	+	})
130	})	134		})
131	return loadErr	135		return loadErr
132	}	136		}

A server/post_test.go +577 -0

@@ -0,0 +1,577 @@
	1	+	package server
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+	"sync/atomic"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"golang.org/x/sys/unix"
	12	+	)
	13	+
	14	+	// runDisplay starts a display and stops it at test cleanup.
	15	+	func runDisplay(t testing.T) Display {
	16	+	t.Helper()
	17	+	d, err := NewDisplay()
	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 <- d.Run(ctx) }()
	24	+	t.Cleanup(func() {
	25	+	cancel()
	26	+	if err := <-done; err != nil {
	27	+	t.Error(err)
	28	+	}
	29	+	})
	30	+	return d
	31	+	}
	32	+
	33	+	func waitFor(t *testing.T, ch <-chan struct{}, what string) {
	34	+	t.Helper()
	35	+	select {
	36	+	case <-ch:
	37	+	case <-time.After(3 * time.Second):
	38	+	t.Fatalf("timeout waiting for %s", what)
	39	+	}
	40	+	}
	41	+
	42	+	// displayTID reports the thread the display goroutine is locked to.
	43	+	func displayTID(t testing.T, d Display) int {
	44	+	t.Helper()
	45	+	tid := 0
	46	+	if !d.Do(func() { tid = unix.Gettid() }) {
	47	+	t.Fatal("display stopped")
	48	+	}
	49	+	return tid
	50	+	}
	51	+
	52	+	func TestPostRunsOnDisplayGoroutine(t *testing.T) {
	53	+	d := runDisplay(t)
	54	+	want := displayTID(t, d)
	55	+	got := make(chan int, 1)
	56	+	go d.Post(func() { got <- unix.Gettid() })
	57	+	select {
	58	+	case tid := <-got:
	59	+	if tid != want {
	60	+	t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
	61	+	}
	62	+	case <-time.After(3 * time.Second):
	63	+	t.Fatal("post did not run")
	64	+	}
	65	+	}
	66	+
	67	+	func TestPostFIFOWithDo(t *testing.T) {
	68	+	d := runDisplay(t)
	69	+	var order []int
	70	+	for i := range 200 {
	71	+	d.Post(func() { order = append(order, 2*i) })
	72	+	if !d.Do(func() { order = append(order, 2*i+1) }) {
	73	+	t.Fatal("display stopped")
	74	+	}
	75	+	}
	76	+	// Do returned, so the display goroutine is done with order.
	77	+	for i, v := range order {
	78	+	if v != i {
	79	+	t.Fatalf("order[%d] = %d, want %d (%v)", i, v, i, order[:i+1])
	80	+	}
	81	+	}
	82	+	if len(order) != 400 {
	83	+	t.Fatalf("%d callbacks ran, want 400", len(order))
	84	+	}
	85	+	}
	86	+
	87	+	func TestPostFIFOOrder(t *testing.T) {
	88	+	d := runDisplay(t)
	89	+	var order []int
	90	+	for i := range 1000 {
	91	+	d.Post(func() { order = append(order, i) })
	92	+	}
	93	+	d.Do(func() {})
	94	+	for i, v := range order {
	95	+	if v != i {
	96	+	t.Fatalf("order[%d] = %d", i, v)
	97	+	}
	98	+	}
	99	+	if len(order) != 1000 {
	100	+	t.Fatalf("%d posts ran, want 1000", len(order))
	101	+	}
	102	+	}
	103	+
	104	+	func TestPostCoalescesWakes(t *testing.T) {
	105	+	d := runDisplay(t)
	106	+	entered, release := make(chan struct{}), make(chan struct{})
	107	+	doDone := make(chan struct{})
	108	+	// Hold the display goroutine so every Post lands while one wake is pending.
	109	+	go func() { d.Do(func() { close(entered); <-release }); close(doDone) }()
	110	+	waitFor(t, entered, "display busy")
	111	+	before := d.wakeWrites.Load()
	112	+	var mu sync.Mutex
	113	+	ran := 0
	114	+	for range 1000 {
	115	+	d.Post(func() { mu.Lock(); ran++; mu.Unlock() })
	116	+	}
	117	+	if w := d.wakeWrites.Load() - before; w > 2 {
	118	+	t.Fatalf("1000 posts wrote the eventfd %d times", w)
	119	+	}
	120	+	close(release)
	121	+	waitFor(t, doDone, "Do")
	122	+	d.Do(func() {})
	123	+	mu.Lock()
	124	+	defer mu.Unlock()
	125	+	if ran != 1000 {
	126	+	t.Fatalf("%d posts ran, want 1000", ran)
	127	+	}
	128	+	}
	129	+
	130	+	func TestPostAfterStopDropped(t *testing.T) {
	131	+	d, err := NewDisplay()
	132	+	if err != nil {
	133	+	t.Fatal(err)
	134	+	}
	135	+	ctx, cancel := context.WithCancel(context.Background())
	136	+	done := make(chan error, 1)
	137	+	go func() { done <- d.Run(ctx) }()
	138	+	d.Do(func() {})
	139	+	cancel()
	140	+	if err := <-done; err != nil {
	141	+	t.Fatal(err)
	142	+	}
	143	+	<-d.Stopped()
	144	+	d.Post(func() { t.Error("post ran after stop") })
	145	+	time.Sleep(20 * time.Millisecond)
	146	+
	147	+	// A display closed before Run drops posts too.
	148	+	c, err := NewDisplay()
	149	+	if err != nil {
	150	+	t.Fatal(err)
	151	+	}
	152	+	c.Close()
	153	+	c.Post(func() { t.Error("post ran after close") })
	154	+	c.Post(nil)
	155	+	}
	156	+
	157	+	func TestPostFromPostedFunc(t *testing.T) {
	158	+	d := runDisplay(t)
	159	+	var order []string
	160	+	done := make(chan struct{})
	161	+	d.Post(func() {
	162	+	order = append(order, "outer")
	163	+	d.Post(func() {
	164	+	order = append(order, "inner")
	165	+	close(done)
	166	+	})
	167	+	order = append(order, "outer-end")
	168	+	})
	169	+	waitFor(t, done, "nested post")
	170	+	d.Do(func() {})
	171	+	want := []string{"outer", "outer-end", "inner"}
	172	+	if len(order) != len(want) {
	173	+	t.Fatalf("order %v", order)
	174	+	}
	175	+	for i := range want {
	176	+	if order[i] != want[i] {
	177	+	t.Fatalf("order %v, want %v", order, want)
	178	+	}
	179	+	}
	180	+	}
	181	+
	182	+	// A callback that keeps posting must not stop libwayland from dispatching, and
	183	+	// the capped rounds must be resumed by a wake instead of waiting for the
	184	+	// dispatch timeout.
	185	+	func TestPostSelfRepostingIsCappedAndResumes(t *testing.T) {
	186	+	d := runDisplay(t)
	187	+	const total = 100 // > maxDrainRounds, so the cap is hit
	188	+	var n int
	189	+	done := make(chan struct{})
	190	+	var step func()
	191	+	step = func() {
	192	+	n++
	193	+	if n == total {
	194	+	close(done)
	195	+	return
	196	+	}
	197	+	d.Post(step)
	198	+	}
	199	+	start := time.Now()
	200	+	d.Post(step)
	201	+	waitFor(t, done, "self-reposting chain")
	202	+	if e := time.Since(start); e > 500*time.Millisecond {
	203	+	t.Fatalf("chain of %d posts took %v; wake not re-armed at the round cap", total, e)
	204	+	}
	205	+	// Do still gets served.
	206	+	if !d.Do(func() {}) {
	207	+	t.Fatal("Do failed")
	208	+	}
	209	+	}
	210	+
	211	+	func TestPostAllocations(t *testing.T) {
	212	+	d, err := NewDisplay()
	213	+	if err != nil {
	214	+	t.Fatal(err)
	215	+	}
	216	+	defer d.Close()
	217	+	n := 0
	218	+	fn := func() { n++ }
	219	+	// Warm up both batch buffers, then drain on this goroutine (not running).
	220	+	for range 4 {
	221	+	d.Post(fn)
	222	+	d.Post(fn)
	223	+	d.drain()
	224	+	}
	225	+	if allocs := testing.AllocsPerRun(1000, func() {
	226	+	d.Post(fn)
	227	+	d.Post(fn)
	228	+	d.drain()
	229	+	}); allocs > 0 {
	230	+	t.Fatalf("Post+drain allocates %v per run", allocs)
	231	+	}
	232	+	if n == 0 {
	233	+	t.Fatal("posts did not run")
	234	+	}
	235	+	}
	236	+
	237	+	func TestAddFDEventfdOnDisplayGoroutine(t *testing.T) {
	238	+	d := runDisplay(t)
	239	+	want := displayTID(t, d)
	240	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	241	+	if err != nil {
	242	+	t.Fatal(err)
	243	+	}
	244	+	defer unix.Close(efd)
	245	+	got := make(chan int, 4)
	246	+	var src *Source
	247	+	if !d.Do(func() {
	248	+	src, err = d.AddFD(efd, func() {
	249	+	var b [8]byte
	250	+	_, _ = unix.Read(efd, b[:])
	251	+	got <- unix.Gettid()
	252	+	})
	253	+	}) \|\| err != nil {
	254	+	t.Fatalf("AddFD: %v", err)
	255	+	}
	256	+	one := [8]byte{1}
	257	+	if _, err := unix.Write(efd, one[:]); err != nil {
	258	+	t.Fatal(err)
	259	+	}
	260	+	select {
	261	+	case tid := <-got:
	262	+	if tid != want {
	263	+	t.Fatalf("callback on thread %d, display thread is %d", tid, want)
	264	+	}
	265	+	case <-time.After(3 * time.Second):
	266	+	t.Fatal("callback did not fire")
	267	+	}
	268	+
	269	+	// Remove stops callbacks; it is idempotent.
	270	+	d.Do(func() { src.Remove(); src.Remove() })
	271	+	if _, err := unix.Write(efd, one[:]); err != nil {
	272	+	t.Fatal(err)
	273	+	}
	274	+	d.Do(func() {})
	275	+	time.Sleep(30 * time.Millisecond)
	276	+	select {
	277	+	case <-got:
	278	+	t.Fatal("callback fired after Remove")
	279	+	default:
	280	+	}
	281	+	d.Do(func() {
	282	+	if len(live.fds) != 0 {
	283	+	t.Errorf("%d fd callbacks leaked", len(live.fds))
	284	+	}
	285	+	})
	286	+	}
	287	+
	288	+	func TestAddFDTimerfdFiresOnce(t *testing.T) {
	289	+	d := runDisplay(t)
	290	+	tfd, err := unix.TimerfdCreate(unix.CLOCK_MONOTONIC, unix.TFD_NONBLOCK\|unix.TFD_CLOEXEC)
	291	+	if err != nil {
	292	+	t.Skip(err)
	293	+	}
	294	+	defer unix.Close(tfd)
	295	+	fired := make(chan struct{}, 8)
	296	+	if !d.Do(func() {
	297	+	_, err = d.AddFD(tfd, func() {
	298	+	var b [8]byte
	299	+	_, _ = unix.Read(tfd, b[:])
	300	+	fired <- struct{}{}
	301	+	})
	302	+	}) \|\| err != nil {
	303	+	t.Fatalf("AddFD: %v", err)
	304	+	}
	305	+	start := time.Now()
	306	+	if err := unix.TimerfdSettime(tfd, 0, &unix.ItimerSpec{Value: unix.NsecToTimespec(int64(time.Millisecond))}, nil); err != nil {
	307	+	t.Fatal(err)
	308	+	}
	309	+	waitFor(t, fired, "timerfd")
	310	+	if e := time.Since(start); e > 500*time.Millisecond {
	311	+	t.Fatalf("1 ms timer fired after %v", e)
	312	+	}
	313	+	time.Sleep(50 * time.Millisecond)
	314	+	if len(fired) != 0 {
	315	+	t.Fatalf("one-shot timer fired %d more times", len(fired))
	316	+	}
	317	+	}
	318	+
	319	+	func TestAddFDBeforeRunAndRemoveInCallback(t *testing.T) {
	320	+	d, err := NewDisplay()
	321	+	if err != nil {
	322	+	t.Fatal(err)
	323	+	}
	324	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	325	+	if err != nil {
	326	+	t.Fatal(err)
	327	+	}
	328	+	defer unix.Close(efd)
	329	+	fired := make(chan struct{}, 4)
	330	+	var src *Source
	331	+	src, err = d.AddFD(efd, func() {
	332	+	var b [8]byte
	333	+	_, _ = unix.Read(efd, b[:])
	334	+	src.Remove() // from its own callback
	335	+	fired <- struct{}{}
	336	+	})
	337	+	if err != nil {
	338	+	t.Fatal(err)
	339	+	}
	340	+	ctx, cancel := context.WithCancel(context.Background())
	341	+	done := make(chan error, 1)
	342	+	go func() { done <- d.Run(ctx) }()
	343	+	defer func() { cancel(); <-done }()
	344	+	one := [8]byte{1}
	345	+	for range 2 {
	346	+	if _, err := unix.Write(efd, one[:]); err != nil {
	347	+	t.Fatal(err)
	348	+	}
	349	+	time.Sleep(20 * time.Millisecond)
	350	+	}
	351	+	waitFor(t, fired, "callback")
	352	+	if len(fired) != 0 {
	353	+	t.Fatal("callback fired again after removing itself")
	354	+	}
	355	+	}
	356	+
	357	+	func TestAddFDErrors(t *testing.T) {
	358	+	d, err := NewDisplay()
	359	+	if err != nil {
	360	+	t.Fatal(err)
	361	+	}
	362	+	defer d.Close()
	363	+	if _, err := d.AddFD(-1, func() {}); err == nil {
	364	+	t.Error("negative fd accepted")
	365	+	}
	366	+	if _, err := d.AddFD(0, nil); err == nil {
	367	+	t.Error("nil callback accepted")
	368	+	}
	369	+	// A source left registered is freed with the display; Remove is then a no-op.
	370	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	371	+	if err != nil {
	372	+	t.Fatal(err)
	373	+	}
	374	+	defer unix.Close(efd)
	375	+	src, err := d.AddFD(efd, func() {})
	376	+	if err != nil {
	377	+	t.Fatal(err)
	378	+	}
	379	+	d.Close()
	380	+	src.Remove()
	381	+	(*Source)(nil).Remove()
	382	+	if _, err := d.AddFD(efd, func() {}); err == nil {
	383	+	t.Error("AddFD accepted on a closed display")
	384	+	}
	385	+	if len(live.fds) != 0 {
	386	+	t.Errorf("%d fd callbacks leaked", len(live.fds))
	387	+	}
	388	+	}
	389	+
	390	+	// Posts from many goroutines must wake the loop without any Do to rescue a
	391	+	// lost wake. The dispatch timeout is raised far past waitFor's limit, so a
	392	+	// lost wake hangs the round instead of only slowing it.
	393	+	func TestPostMultiProducerNoLostWake(t *testing.T) {
	394	+	d, err := NewDisplay()
	395	+	if err != nil {
	396	+	t.Fatal(err)
	397	+	}
	398	+	d.dispatchMS = 60_000
	399	+	ctx, cancel := context.WithCancel(context.Background())
	400	+	runDone := make(chan error, 1)
	401	+	go func() { runDone <- d.Run(ctx) }()
	402	+	t.Cleanup(func() {
	403	+	cancel()
	404	+	// Wake the loop so it sees the cancellation before the long timeout.
	405	+	d.Post(func() {})
	406	+	if err := <-runDone; err != nil {
	407	+	t.Error(err)
	408	+	}
	409	+	})
	410	+	const producers, perProducer, rounds = 8, 250, 10
	411	+	for range rounds {
	412	+	// Let the loop go idle in epoll_wait.
	413	+	time.Sleep(5 * time.Millisecond)
	414	+	var left atomic.Int32
	415	+	left.Store(producers * perProducer)
	416	+	done := make(chan struct{})
	417	+	var wg sync.WaitGroup
	418	+	for range producers {
	419	+	wg.Add(1)
	420	+	go func() {
	421	+	defer wg.Done()
	422	+	for range perProducer {
	423	+	d.Post(func() {
	424	+	if left.Add(-1) == 0 {
	425	+	close(done)
	426	+	}
	427	+	})
	428	+	}
	429	+	}()
	430	+	}
	431	+	waitFor(t, done, "all posts (a lost wake hangs here)")
	432	+	wg.Wait()
	433	+	}
	434	+	}
	435	+
	436	+	// Do and Post racing with shutdown must neither panic nor run anything after
	437	+	// Stopped, and Do's wake must not write to the closed eventfd's number once it
	438	+	// is reused.
	439	+	func TestDoPostRaceWithCancel(t *testing.T) {
	440	+	for range 20 {
	441	+	d, err := NewDisplay()
	442	+	if err != nil {
	443	+	t.Fatal(err)
	444	+	}
	445	+	ctx, cancel := context.WithCancel(context.Background())
	446	+	done := make(chan error, 1)
	447	+	go func() { done <- d.Run(ctx) }()
	448	+
	449	+	var stopped atomic.Bool
	450	+	var wg sync.WaitGroup
	451	+	for range 4 {
	452	+	wg.Add(2)
	453	+	go func() {
	454	+	defer wg.Done()
	455	+	for range 200 {
	456	+	d.Do(func() {
	457	+	if stopped.Load() {
	458	+	t.Error("Do function ran after Stopped")
	459	+	}
	460	+	})
	461	+	}
	462	+	}()
	463	+	go func() {
	464	+	defer wg.Done()
	465	+	for range 200 {
	466	+	d.Post(func() {
	467	+	if stopped.Load() {
	468	+	t.Error("post ran after Stopped")
	469	+	}
	470	+	})
	471	+	}
	472	+	}()
	473	+	}
	474	+	time.Sleep(time.Millisecond)
	475	+	cancel()
	476	+	if err := <-done; err != nil {
	477	+	t.Fatal(err)
	478	+	}
	479	+	<-d.Stopped()
	480	+	stopped.Store(true)
	481	+
	482	+	// The closed wake fd's number is free for reuse now. Late Do and Post
	483	+	// calls must not write to whoever gets it.
	484	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	485	+	if err != nil {
	486	+	t.Fatal(err)
	487	+	}
	488	+	// Force the reuse: make the old wake fd number refer to efd, unless
	489	+	// efd already got that number. Never clobber an fd in use.
	490	+	if efd != d.wake {
	491	+	if _, err := unix.FcntlInt(uintptr(d.wake), unix.F_GETFD, 0); !errors.Is(err, unix.EBADF) {
	492	+	t.Fatalf("old wake fd %d is still in use: %v", d.wake, err)
	493	+	}
	494	+	if err := unix.Dup3(efd, d.wake, unix.O_CLOEXEC); err != nil {
	495	+	t.Fatal(err)
	496	+	}
	497	+	}
	498	+	for range 50 {
	499	+	d.Do(func() { t.Error("Do function ran after Stopped") })
	500	+	d.Post(func() { t.Error("post ran after Stopped") })
	501	+	}
	502	+	wg.Wait()
	503	+	var b [8]byte
	504	+	if _, err := unix.Read(efd, b[:]); !errors.Is(err, unix.EAGAIN) {
	505	+	t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err)
	506	+	}
	507	+	unix.Close(efd)
	508	+	if efd != d.wake {
	509	+	unix.Close(d.wake)
	510	+	}
	511	+	}
	512	+	}
	513	+
	514	+	// Sources still registered when Run exits are released with the display.
	515	+	func TestRunExitReleasesSources(t *testing.T) {
	516	+	d, err := NewDisplay()
	517	+	if err != nil {
	518	+	t.Fatal(err)
	519	+	}
	520	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	521	+	if err != nil {
	522	+	t.Fatal(err)
	523	+	}
	524	+	defer unix.Close(efd)
	525	+	ctx, cancel := context.WithCancel(context.Background())
	526	+	done := make(chan error, 1)
	527	+	go func() { done <- d.Run(ctx) }()
	528	+	var addErr error
	529	+	if !d.Do(func() { _, addErr = d.AddFD(efd, func() {}) }) \|\| addErr != nil {
	530	+	t.Fatalf("AddFD: %v", addErr)
	531	+	}
	532	+	cancel()
	533	+	if err := <-done; err != nil {
	534	+	t.Fatal(err)
	535	+	}
	536	+	<-d.Stopped()
	537	+	if len(live.fds) != 0 {
	538	+	t.Errorf("%d fd callbacks leaked after Run exited", len(live.fds))
	539	+	}
	540	+	if len(d.sources) != 0 {
	541	+	t.Errorf("%d sources still tracked after Run exited", len(d.sources))
	542	+	}
	543	+	}
	544	+
	545	+	func TestPostFromAddFDCallback(t *testing.T) {
	546	+	d := runDisplay(t)
	547	+	want := displayTID(t, d)
	548	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	549	+	if err != nil {
	550	+	t.Fatal(err)
	551	+	}
	552	+	defer unix.Close(efd)
	553	+	got := make(chan int, 1)
	554	+	var src *Source
	555	+	if !d.Do(func() {
	556	+	src, err = d.AddFD(efd, func() {
	557	+	var b [8]byte
	558	+	_, _ = unix.Read(efd, b[:])
	559	+	d.Post(func() { got <- unix.Gettid() })
	560	+	})
	561	+	}) \|\| err != nil {
	562	+	t.Fatalf("AddFD: %v", err)
	563	+	}
	564	+	one := [8]byte{1}
	565	+	if _, err := unix.Write(efd, one[:]); err != nil {
	566	+	t.Fatal(err)
	567	+	}
	568	+	select {
	569	+	case tid := <-got:
	570	+	if tid != want {
	571	+	t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
	572	+	}
	573	+	case <-time.After(3 * time.Second):
	574	+	t.Fatal("post from AddFD callback did not run")
	575	+	}
	576	+	d.Do(func() { src.Remove() })
	577	+	}

@@ -0,0 +1,577 @@
	1	+	package server
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+	"sync/atomic"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"golang.org/x/sys/unix"
	12	+	)
	13	+
	14	+	// runDisplay starts a display and stops it at test cleanup.
	15	+	func runDisplay(t testing.T) Display {
	16	+	t.Helper()
	17	+	d, err := NewDisplay()
	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 <- d.Run(ctx) }()
	24	+	t.Cleanup(func() {
	25	+	cancel()
	26	+	if err := <-done; err != nil {
	27	+	t.Error(err)
	28	+	}
	29	+	})
	30	+	return d
	31	+	}
	32	+
	33	+	func waitFor(t *testing.T, ch <-chan struct{}, what string) {
	34	+	t.Helper()
	35	+	select {
	36	+	case <-ch:
	37	+	case <-time.After(3 * time.Second):
	38	+	t.Fatalf("timeout waiting for %s", what)
	39	+	}
	40	+	}
	41	+
	42	+	// displayTID reports the thread the display goroutine is locked to.
	43	+	func displayTID(t testing.T, d Display) int {
	44	+	t.Helper()
	45	+	tid := 0
	46	+	if !d.Do(func() { tid = unix.Gettid() }) {
	47	+	t.Fatal("display stopped")
	48	+	}
	49	+	return tid
	50	+	}
	51	+
	52	+	func TestPostRunsOnDisplayGoroutine(t *testing.T) {
	53	+	d := runDisplay(t)
	54	+	want := displayTID(t, d)
	55	+	got := make(chan int, 1)
	56	+	go d.Post(func() { got <- unix.Gettid() })
	57	+	select {
	58	+	case tid := <-got:
	59	+	if tid != want {
	60	+	t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
	61	+	}
	62	+	case <-time.After(3 * time.Second):
	63	+	t.Fatal("post did not run")
	64	+	}
	65	+	}
	66	+
	67	+	func TestPostFIFOWithDo(t *testing.T) {
	68	+	d := runDisplay(t)
	69	+	var order []int
	70	+	for i := range 200 {
	71	+	d.Post(func() { order = append(order, 2*i) })
	72	+	if !d.Do(func() { order = append(order, 2*i+1) }) {
	73	+	t.Fatal("display stopped")
	74	+	}
	75	+	}
	76	+	// Do returned, so the display goroutine is done with order.
	77	+	for i, v := range order {
	78	+	if v != i {
	79	+	t.Fatalf("order[%d] = %d, want %d (%v)", i, v, i, order[:i+1])
	80	+	}
	81	+	}
	82	+	if len(order) != 400 {
	83	+	t.Fatalf("%d callbacks ran, want 400", len(order))
	84	+	}
	85	+	}
	86	+
	87	+	func TestPostFIFOOrder(t *testing.T) {
	88	+	d := runDisplay(t)
	89	+	var order []int
	90	+	for i := range 1000 {
	91	+	d.Post(func() { order = append(order, i) })
	92	+	}
	93	+	d.Do(func() {})
	94	+	for i, v := range order {
	95	+	if v != i {
	96	+	t.Fatalf("order[%d] = %d", i, v)
	97	+	}
	98	+	}
	99	+	if len(order) != 1000 {
	100	+	t.Fatalf("%d posts ran, want 1000", len(order))
	101	+	}
	102	+	}
	103	+
	104	+	func TestPostCoalescesWakes(t *testing.T) {
	105	+	d := runDisplay(t)
	106	+	entered, release := make(chan struct{}), make(chan struct{})
	107	+	doDone := make(chan struct{})
	108	+	// Hold the display goroutine so every Post lands while one wake is pending.
	109	+	go func() { d.Do(func() { close(entered); <-release }); close(doDone) }()
	110	+	waitFor(t, entered, "display busy")
	111	+	before := d.wakeWrites.Load()
	112	+	var mu sync.Mutex
	113	+	ran := 0
	114	+	for range 1000 {
	115	+	d.Post(func() { mu.Lock(); ran++; mu.Unlock() })
	116	+	}
	117	+	if w := d.wakeWrites.Load() - before; w > 2 {
	118	+	t.Fatalf("1000 posts wrote the eventfd %d times", w)
	119	+	}
	120	+	close(release)
	121	+	waitFor(t, doDone, "Do")
	122	+	d.Do(func() {})
	123	+	mu.Lock()
	124	+	defer mu.Unlock()
	125	+	if ran != 1000 {
	126	+	t.Fatalf("%d posts ran, want 1000", ran)
	127	+	}
	128	+	}
	129	+
	130	+	func TestPostAfterStopDropped(t *testing.T) {
	131	+	d, err := NewDisplay()
	132	+	if err != nil {
	133	+	t.Fatal(err)
	134	+	}
	135	+	ctx, cancel := context.WithCancel(context.Background())
	136	+	done := make(chan error, 1)
	137	+	go func() { done <- d.Run(ctx) }()
	138	+	d.Do(func() {})
	139	+	cancel()
	140	+	if err := <-done; err != nil {
	141	+	t.Fatal(err)
	142	+	}
	143	+	<-d.Stopped()
	144	+	d.Post(func() { t.Error("post ran after stop") })
	145	+	time.Sleep(20 * time.Millisecond)
	146	+
	147	+	// A display closed before Run drops posts too.
	148	+	c, err := NewDisplay()
	149	+	if err != nil {
	150	+	t.Fatal(err)
	151	+	}
	152	+	c.Close()
	153	+	c.Post(func() { t.Error("post ran after close") })
	154	+	c.Post(nil)
	155	+	}
	156	+
	157	+	func TestPostFromPostedFunc(t *testing.T) {
	158	+	d := runDisplay(t)
	159	+	var order []string
	160	+	done := make(chan struct{})
	161	+	d.Post(func() {
	162	+	order = append(order, "outer")
	163	+	d.Post(func() {
	164	+	order = append(order, "inner")
	165	+	close(done)
	166	+	})
	167	+	order = append(order, "outer-end")
	168	+	})
	169	+	waitFor(t, done, "nested post")
	170	+	d.Do(func() {})
	171	+	want := []string{"outer", "outer-end", "inner"}
	172	+	if len(order) != len(want) {
	173	+	t.Fatalf("order %v", order)
	174	+	}
	175	+	for i := range want {
	176	+	if order[i] != want[i] {
	177	+	t.Fatalf("order %v, want %v", order, want)
	178	+	}
	179	+	}
	180	+	}
	181	+
	182	+	// A callback that keeps posting must not stop libwayland from dispatching, and
	183	+	// the capped rounds must be resumed by a wake instead of waiting for the
	184	+	// dispatch timeout.
	185	+	func TestPostSelfRepostingIsCappedAndResumes(t *testing.T) {
	186	+	d := runDisplay(t)
	187	+	const total = 100 // > maxDrainRounds, so the cap is hit
	188	+	var n int
	189	+	done := make(chan struct{})
	190	+	var step func()
	191	+	step = func() {
	192	+	n++
	193	+	if n == total {
	194	+	close(done)
	195	+	return
	196	+	}
	197	+	d.Post(step)
	198	+	}
	199	+	start := time.Now()
	200	+	d.Post(step)
	201	+	waitFor(t, done, "self-reposting chain")
	202	+	if e := time.Since(start); e > 500*time.Millisecond {
	203	+	t.Fatalf("chain of %d posts took %v; wake not re-armed at the round cap", total, e)
	204	+	}
	205	+	// Do still gets served.
	206	+	if !d.Do(func() {}) {
	207	+	t.Fatal("Do failed")
	208	+	}
	209	+	}
	210	+
	211	+	func TestPostAllocations(t *testing.T) {
	212	+	d, err := NewDisplay()
	213	+	if err != nil {
	214	+	t.Fatal(err)
	215	+	}
	216	+	defer d.Close()
	217	+	n := 0
	218	+	fn := func() { n++ }
	219	+	// Warm up both batch buffers, then drain on this goroutine (not running).
	220	+	for range 4 {
	221	+	d.Post(fn)
	222	+	d.Post(fn)
	223	+	d.drain()
	224	+	}
	225	+	if allocs := testing.AllocsPerRun(1000, func() {
	226	+	d.Post(fn)
	227	+	d.Post(fn)
	228	+	d.drain()
	229	+	}); allocs > 0 {
	230	+	t.Fatalf("Post+drain allocates %v per run", allocs)
	231	+	}
	232	+	if n == 0 {
	233	+	t.Fatal("posts did not run")
	234	+	}
	235	+	}
	236	+
	237	+	func TestAddFDEventfdOnDisplayGoroutine(t *testing.T) {
	238	+	d := runDisplay(t)
	239	+	want := displayTID(t, d)
	240	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	241	+	if err != nil {
	242	+	t.Fatal(err)
	243	+	}
	244	+	defer unix.Close(efd)
	245	+	got := make(chan int, 4)
	246	+	var src *Source
	247	+	if !d.Do(func() {
	248	+	src, err = d.AddFD(efd, func() {
	249	+	var b [8]byte
	250	+	_, _ = unix.Read(efd, b[:])
	251	+	got <- unix.Gettid()
	252	+	})
	253	+	}) \|\| err != nil {
	254	+	t.Fatalf("AddFD: %v", err)
	255	+	}
	256	+	one := [8]byte{1}
	257	+	if _, err := unix.Write(efd, one[:]); err != nil {
	258	+	t.Fatal(err)
	259	+	}
	260	+	select {
	261	+	case tid := <-got:
	262	+	if tid != want {
	263	+	t.Fatalf("callback on thread %d, display thread is %d", tid, want)
	264	+	}
	265	+	case <-time.After(3 * time.Second):
	266	+	t.Fatal("callback did not fire")
	267	+	}
	268	+
	269	+	// Remove stops callbacks; it is idempotent.
	270	+	d.Do(func() { src.Remove(); src.Remove() })
	271	+	if _, err := unix.Write(efd, one[:]); err != nil {
	272	+	t.Fatal(err)
	273	+	}
	274	+	d.Do(func() {})
	275	+	time.Sleep(30 * time.Millisecond)
	276	+	select {
	277	+	case <-got:
	278	+	t.Fatal("callback fired after Remove")
	279	+	default:
	280	+	}
	281	+	d.Do(func() {
	282	+	if len(live.fds) != 0 {
	283	+	t.Errorf("%d fd callbacks leaked", len(live.fds))
	284	+	}
	285	+	})
	286	+	}
	287	+
	288	+	func TestAddFDTimerfdFiresOnce(t *testing.T) {
	289	+	d := runDisplay(t)
	290	+	tfd, err := unix.TimerfdCreate(unix.CLOCK_MONOTONIC, unix.TFD_NONBLOCK\|unix.TFD_CLOEXEC)
	291	+	if err != nil {
	292	+	t.Skip(err)
	293	+	}
	294	+	defer unix.Close(tfd)
	295	+	fired := make(chan struct{}, 8)
	296	+	if !d.Do(func() {
	297	+	_, err = d.AddFD(tfd, func() {
	298	+	var b [8]byte
	299	+	_, _ = unix.Read(tfd, b[:])
	300	+	fired <- struct{}{}
	301	+	})
	302	+	}) \|\| err != nil {
	303	+	t.Fatalf("AddFD: %v", err)
	304	+	}
	305	+	start := time.Now()
	306	+	if err := unix.TimerfdSettime(tfd, 0, &unix.ItimerSpec{Value: unix.NsecToTimespec(int64(time.Millisecond))}, nil); err != nil {
	307	+	t.Fatal(err)
	308	+	}
	309	+	waitFor(t, fired, "timerfd")
	310	+	if e := time.Since(start); e > 500*time.Millisecond {
	311	+	t.Fatalf("1 ms timer fired after %v", e)
	312	+	}
	313	+	time.Sleep(50 * time.Millisecond)
	314	+	if len(fired) != 0 {
	315	+	t.Fatalf("one-shot timer fired %d more times", len(fired))
	316	+	}
	317	+	}
	318	+
	319	+	func TestAddFDBeforeRunAndRemoveInCallback(t *testing.T) {
	320	+	d, err := NewDisplay()
	321	+	if err != nil {
	322	+	t.Fatal(err)
	323	+	}
	324	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	325	+	if err != nil {
	326	+	t.Fatal(err)
	327	+	}
	328	+	defer unix.Close(efd)
	329	+	fired := make(chan struct{}, 4)
	330	+	var src *Source
	331	+	src, err = d.AddFD(efd, func() {
	332	+	var b [8]byte
	333	+	_, _ = unix.Read(efd, b[:])
	334	+	src.Remove() // from its own callback
	335	+	fired <- struct{}{}
	336	+	})
	337	+	if err != nil {
	338	+	t.Fatal(err)
	339	+	}
	340	+	ctx, cancel := context.WithCancel(context.Background())
	341	+	done := make(chan error, 1)
	342	+	go func() { done <- d.Run(ctx) }()
	343	+	defer func() { cancel(); <-done }()
	344	+	one := [8]byte{1}
	345	+	for range 2 {
	346	+	if _, err := unix.Write(efd, one[:]); err != nil {
	347	+	t.Fatal(err)
	348	+	}
	349	+	time.Sleep(20 * time.Millisecond)
	350	+	}
	351	+	waitFor(t, fired, "callback")
	352	+	if len(fired) != 0 {
	353	+	t.Fatal("callback fired again after removing itself")
	354	+	}
	355	+	}
	356	+
	357	+	func TestAddFDErrors(t *testing.T) {
	358	+	d, err := NewDisplay()
	359	+	if err != nil {
	360	+	t.Fatal(err)
	361	+	}
	362	+	defer d.Close()
	363	+	if _, err := d.AddFD(-1, func() {}); err == nil {
	364	+	t.Error("negative fd accepted")
	365	+	}
	366	+	if _, err := d.AddFD(0, nil); err == nil {
	367	+	t.Error("nil callback accepted")
	368	+	}
	369	+	// A source left registered is freed with the display; Remove is then a no-op.
	370	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	371	+	if err != nil {
	372	+	t.Fatal(err)
	373	+	}
	374	+	defer unix.Close(efd)
	375	+	src, err := d.AddFD(efd, func() {})
	376	+	if err != nil {
	377	+	t.Fatal(err)
	378	+	}
	379	+	d.Close()
	380	+	src.Remove()
	381	+	(*Source)(nil).Remove()
	382	+	if _, err := d.AddFD(efd, func() {}); err == nil {
	383	+	t.Error("AddFD accepted on a closed display")
	384	+	}
	385	+	if len(live.fds) != 0 {
	386	+	t.Errorf("%d fd callbacks leaked", len(live.fds))
	387	+	}
	388	+	}
	389	+
	390	+	// Posts from many goroutines must wake the loop without any Do to rescue a
	391	+	// lost wake. The dispatch timeout is raised far past waitFor's limit, so a
	392	+	// lost wake hangs the round instead of only slowing it.
	393	+	func TestPostMultiProducerNoLostWake(t *testing.T) {
	394	+	d, err := NewDisplay()
	395	+	if err != nil {
	396	+	t.Fatal(err)
	397	+	}
	398	+	d.dispatchMS = 60_000
	399	+	ctx, cancel := context.WithCancel(context.Background())
	400	+	runDone := make(chan error, 1)
	401	+	go func() { runDone <- d.Run(ctx) }()
	402	+	t.Cleanup(func() {
	403	+	cancel()
	404	+	// Wake the loop so it sees the cancellation before the long timeout.
	405	+	d.Post(func() {})
	406	+	if err := <-runDone; err != nil {
	407	+	t.Error(err)
	408	+	}
	409	+	})
	410	+	const producers, perProducer, rounds = 8, 250, 10
	411	+	for range rounds {
	412	+	// Let the loop go idle in epoll_wait.
	413	+	time.Sleep(5 * time.Millisecond)
	414	+	var left atomic.Int32
	415	+	left.Store(producers * perProducer)
	416	+	done := make(chan struct{})
	417	+	var wg sync.WaitGroup
	418	+	for range producers {
	419	+	wg.Add(1)
	420	+	go func() {
	421	+	defer wg.Done()
	422	+	for range perProducer {
	423	+	d.Post(func() {
	424	+	if left.Add(-1) == 0 {
	425	+	close(done)
	426	+	}
	427	+	})
	428	+	}
	429	+	}()
	430	+	}
	431	+	waitFor(t, done, "all posts (a lost wake hangs here)")
	432	+	wg.Wait()
	433	+	}
	434	+	}
	435	+
	436	+	// Do and Post racing with shutdown must neither panic nor run anything after
	437	+	// Stopped, and Do's wake must not write to the closed eventfd's number once it
	438	+	// is reused.
	439	+	func TestDoPostRaceWithCancel(t *testing.T) {
	440	+	for range 20 {
	441	+	d, err := NewDisplay()
	442	+	if err != nil {
	443	+	t.Fatal(err)
	444	+	}
	445	+	ctx, cancel := context.WithCancel(context.Background())
	446	+	done := make(chan error, 1)
	447	+	go func() { done <- d.Run(ctx) }()
	448	+
	449	+	var stopped atomic.Bool
	450	+	var wg sync.WaitGroup
	451	+	for range 4 {
	452	+	wg.Add(2)
	453	+	go func() {
	454	+	defer wg.Done()
	455	+	for range 200 {
	456	+	d.Do(func() {
	457	+	if stopped.Load() {
	458	+	t.Error("Do function ran after Stopped")
	459	+	}
	460	+	})
	461	+	}
	462	+	}()
	463	+	go func() {
	464	+	defer wg.Done()
	465	+	for range 200 {
	466	+	d.Post(func() {
	467	+	if stopped.Load() {
	468	+	t.Error("post ran after Stopped")
	469	+	}
	470	+	})
	471	+	}
	472	+	}()
	473	+	}
	474	+	time.Sleep(time.Millisecond)
	475	+	cancel()
	476	+	if err := <-done; err != nil {
	477	+	t.Fatal(err)
	478	+	}
	479	+	<-d.Stopped()
	480	+	stopped.Store(true)
	481	+
	482	+	// The closed wake fd's number is free for reuse now. Late Do and Post
	483	+	// calls must not write to whoever gets it.
	484	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	485	+	if err != nil {
	486	+	t.Fatal(err)
	487	+	}
	488	+	// Force the reuse: make the old wake fd number refer to efd, unless
	489	+	// efd already got that number. Never clobber an fd in use.
	490	+	if efd != d.wake {
	491	+	if _, err := unix.FcntlInt(uintptr(d.wake), unix.F_GETFD, 0); !errors.Is(err, unix.EBADF) {
	492	+	t.Fatalf("old wake fd %d is still in use: %v", d.wake, err)
	493	+	}
	494	+	if err := unix.Dup3(efd, d.wake, unix.O_CLOEXEC); err != nil {
	495	+	t.Fatal(err)
	496	+	}
	497	+	}
	498	+	for range 50 {
	499	+	d.Do(func() { t.Error("Do function ran after Stopped") })
	500	+	d.Post(func() { t.Error("post ran after Stopped") })
	501	+	}
	502	+	wg.Wait()
	503	+	var b [8]byte
	504	+	if _, err := unix.Read(efd, b[:]); !errors.Is(err, unix.EAGAIN) {
	505	+	t.Fatalf("eventfd %d (display wake fd was %d) was written after stop: %v", efd, d.wake, err)
	506	+	}
	507	+	unix.Close(efd)
	508	+	if efd != d.wake {
	509	+	unix.Close(d.wake)
	510	+	}
	511	+	}
	512	+	}
	513	+
	514	+	// Sources still registered when Run exits are released with the display.
	515	+	func TestRunExitReleasesSources(t *testing.T) {
	516	+	d, err := NewDisplay()
	517	+	if err != nil {
	518	+	t.Fatal(err)
	519	+	}
	520	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	521	+	if err != nil {
	522	+	t.Fatal(err)
	523	+	}
	524	+	defer unix.Close(efd)
	525	+	ctx, cancel := context.WithCancel(context.Background())
	526	+	done := make(chan error, 1)
	527	+	go func() { done <- d.Run(ctx) }()
	528	+	var addErr error
	529	+	if !d.Do(func() { _, addErr = d.AddFD(efd, func() {}) }) \|\| addErr != nil {
	530	+	t.Fatalf("AddFD: %v", addErr)
	531	+	}
	532	+	cancel()
	533	+	if err := <-done; err != nil {
	534	+	t.Fatal(err)
	535	+	}
	536	+	<-d.Stopped()
	537	+	if len(live.fds) != 0 {
	538	+	t.Errorf("%d fd callbacks leaked after Run exited", len(live.fds))
	539	+	}
	540	+	if len(d.sources) != 0 {
	541	+	t.Errorf("%d sources still tracked after Run exited", len(d.sources))
	542	+	}
	543	+	}
	544	+
	545	+	func TestPostFromAddFDCallback(t *testing.T) {
	546	+	d := runDisplay(t)
	547	+	want := displayTID(t, d)
	548	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	549	+	if err != nil {
	550	+	t.Fatal(err)
	551	+	}
	552	+	defer unix.Close(efd)
	553	+	got := make(chan int, 1)
	554	+	var src *Source
	555	+	if !d.Do(func() {
	556	+	src, err = d.AddFD(efd, func() {
	557	+	var b [8]byte
	558	+	_, _ = unix.Read(efd, b[:])
	559	+	d.Post(func() { got <- unix.Gettid() })
	560	+	})
	561	+	}) \|\| err != nil {
	562	+	t.Fatalf("AddFD: %v", err)
	563	+	}
	564	+	one := [8]byte{1}
	565	+	if _, err := unix.Write(efd, one[:]); err != nil {
	566	+	t.Fatal(err)
	567	+	}
	568	+	select {
	569	+	case tid := <-got:
	570	+	if tid != want {
	571	+	t.Fatalf("post ran on thread %d, display thread is %d", tid, want)
	572	+	}
	573	+	case <-time.After(3 * time.Second):
	574	+	t.Fatal("post from AddFD callback did not run")
	575	+	}
	576	+	d.Do(func() { src.Remove() })
	577	+	}