perf(server): wait for the event loop in the Go netpoller

commit: 634f823abd229e149fb194bbbeaa1a6a94a7e707
author: brice · 9d ago · 2026-10-13 19:16 UTC
committer: brice · 9d ago · 2026-10-13 20:11 UTC
parent: 64cf1d3
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +50 -10

M server/display.go +50 -9


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"errors"
6	6		"fmt"
	7	+	"os"
7	8		"runtime"
8	9		"strings"
9	10		"sync"

@@ -88,10 +89,6 @@ type Display struct {
88	89
89	90		// sources are the AddFD sources not yet removed (display goroutine only).
90	91		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
95	92		}
96	93
97	94		// maxDrainRounds bounds how many times one loop iteration re-drains calls and

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

@@ -219,21 +216,48 @@ func (d *Display) Run(ctx context.Context) error {
219	216		defer runtime.UnlockOSThread()
220	217		defer close(d.stopped)
221	218		defer d.destroy()
	219	+	// Wait for the event loop's epoll fd in Go's netpoller instead of blocking
	220	+	// in epoll_wait inside C: the goroutine parks, its P is released, and the
	221	+	// runtime's sysmon can sleep deeply while the server is idle. libwayland
	222	+	// is then dispatched without blocking.
	223	+	loopFile, err := d.openLoopFile()
	224	+	if err != nil {
	225	+	return err
	226	+	}
	227	+	defer loopFile.Close() // runs before destroy: the dup must not outlive the epoll fd
	228	+	rc, err := loopFile.SyscallConn()
	229	+	if err != nil {
	230	+	return fmt.Errorf("purego-libwayland: event loop fd: %w", err)
	231	+	}
	232	+	// Cancellation wakes the loop through the eventfd.
	233	+	stop := context.AfterFunc(ctx, d.wakeGuarded)
	234	+	defer stop()
	235	+	ready := func(fd uintptr) bool {
	236	+	// Level-triggered check: the netpoller is edge-triggered, and a
	237	+	// dispatch can leave events (libwayland takes 32 per epoll_wait).
	238	+	fds := [1]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	239	+	n, err := unix.Poll(fds[:], 0)
	240	+	return n > 0 \|\| (err != nil && err != unix.EINTR)
	241	+	}
222	242		consecutiveFailures := 0
223	243		for {
224	244		if ctx.Err() != nil {
225	245		return nil
226	246		}
227	247		// 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
	248	+	// is not written once per item; it only makes the wait below
229	249		// return at once when work arrives while the loop is asleep.
230	250		d.drain()
231	251		wlDisplayFlushClients(d.c)
232		-	// The timeout only bounds how late ctx cancellation is seen.
233		-	if wlEventLoopDispatch(d.loop, d.dispatchMS) < 0 {
	252	+	if err := rc.Read(ready); err != nil {
	253	+	return fmt.Errorf("purego-libwayland: wait for event loop: %w", err)
	254	+	}
	255	+	if ctx.Err() != nil {
	256	+	return nil
	257	+	}
	258	+	if wlEventLoopDispatch(d.loop, 0) < 0 {
234	259		errno := lastErrno()
235	260		consecutiveFailures++
236		-	// epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats.
237	261		if consecutiveFailures < 16 {
238	262		continue
239	263		}

@@ -244,6 +268,23 @@ func (d *Display) Run(ctx context.Context) error {
244	268		}
245	269		}
246	270
	271	+	// openLoopFile wraps a duplicate of the event loop's epoll fd for the netpoller.
	272	+	func (d Display) openLoopFile() (os.File, error) {
	273	+	fd := int(wlEventLoopGetFD(d.loop))
	274	+	if fd < 0 {
	275	+	return nil, errors.New("purego-libwayland: wl_event_loop_get_fd failed")
	276	+	}
	277	+	dup, err := unix.FcntlInt(uintptr(fd), unix.F_DUPFD_CLOEXEC, 0)
	278	+	if err != nil {
	279	+	return nil, fmt.Errorf("purego-libwayland: dup event loop fd: %w", err)
	280	+	}
	281	+	if err := unix.SetNonblock(dup, true); err != nil {
	282	+	unix.Close(dup)
	283	+	return nil, fmt.Errorf("purego-libwayland: nonblock event loop fd: %w", err)
	284	+	}
	285	+	return os.NewFile(uintptr(dup), "wl-event-loop"), nil
	286	+	}
	287	+
247	288		// drain runs queued posts and Do calls until none are left, at most
248	289		// maxDrainRounds rounds. Posts are taken again after each call is received and
249	290		// before it runs, so a Post that happened before a Do always runs before that

 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"runtime"
 	"strings"
 	"sync"
 	// sources are the AddFD sources not yet removed (display goroutine only).
 	sources map[*Source]struct{}
-	// dispatchMS bounds one event loop wait; it only bounds how late ctx
-	// cancellation is seen. Tests raise it so a lost wake hangs.
-	dispatchMS int32
 }
 // maxDrainRounds bounds how many times one loop iteration re-drains calls and
 		wlDisplayDestroy(c)
 		return nil, fmt.Errorf("purego-libwayland: eventfd: %w", err)
 	}
-	d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}, dispatchMS: 100}
+	d := &Display{c: c, loop: wlDisplayGetEventLoop(c), calls: make(chan func(), 64), wake: wake, stopped: make(chan struct{}), sources: map[*Source]struct{}{}}
 	// The callback only drains the counter; Run then executes queued calls.
 	if d.wakeSrc = wlEventLoopAddFD(d.loop, int32(wake), 1 /* WL_EVENT_READABLE */, cbWake, 0); d.wakeSrc == 0 {
 		unix.Close(wake)
 	defer runtime.UnlockOSThread()
 	defer close(d.stopped)
 	defer d.destroy()
+	// Wait for the event loop's epoll fd in Go's netpoller instead of blocking
+	// in epoll_wait inside C: the goroutine parks, its P is released, and the
+	// runtime's sysmon can sleep deeply while the server is idle. libwayland
+	// is then dispatched without blocking.
+	loopFile, err := d.openLoopFile()
+	if err != nil {
+		return err
+	}
+	defer loopFile.Close() // runs before destroy: the dup must not outlive the epoll fd
+	rc, err := loopFile.SyscallConn()
+	if err != nil {
+		return fmt.Errorf("purego-libwayland: event loop fd: %w", err)
+	}
+	// Cancellation wakes the loop through the eventfd.
+	stop := context.AfterFunc(ctx, d.wakeGuarded)
+	defer stop()
+	ready := func(fd uintptr) bool {
+		// Level-triggered check: the netpoller is edge-triggered, and a
+		// dispatch can leave events (libwayland takes 32 per epoll_wait).
+		fds := [1]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
+		n, err := unix.Poll(fds[:], 0)
+		return n > 0 || (err != nil && err != unix.EINTR)
+	}
 	consecutiveFailures := 0
 	for {
 		if ctx.Err() != nil {
 			return nil
 		}
 		// Run every queued post and call. Wakes are coalesced, so the eventfd
-		// is not written once per item; it only makes the dispatch below
+		// is not written once per item; it only makes the wait below
 		// return at once when work arrives while the loop is asleep.
 		d.drain()
 		wlDisplayFlushClients(d.c)
-		// The timeout only bounds how late ctx cancellation is seen.
+		if err := rc.Read(ready); err != nil {
-		if wlEventLoopDispatch(d.loop, d.dispatchMS) < 0 {
+			return fmt.Errorf("purego-libwayland: wait for event loop: %w", err)
+		}
+		if ctx.Err() != nil {
+			return nil
+		}
+		if wlEventLoopDispatch(d.loop, 0) < 0 {
 			errno := lastErrno()
 			consecutiveFailures++
-			// epoll_wait is interrupted by Go's preemption signals (EINTR); a persistent failure (EBADF, EINVAL) repeats.
 			if consecutiveFailures < 16 {
 				continue
 			}
 	}
 }
+// openLoopFile wraps a duplicate of the event loop's epoll fd for the netpoller.
+func (d *Display) openLoopFile() (*os.File, error) {
+	fd := int(wlEventLoopGetFD(d.loop))
+	if fd < 0 {
+		return nil, errors.New("purego-libwayland: wl_event_loop_get_fd failed")
+	}
+	dup, err := unix.FcntlInt(uintptr(fd), unix.F_DUPFD_CLOEXEC, 0)
+	if err != nil {
+		return nil, fmt.Errorf("purego-libwayland: dup event loop fd: %w", err)
+	}
+	if err := unix.SetNonblock(dup, true); err != nil {
+		unix.Close(dup)
+		return nil, fmt.Errorf("purego-libwayland: nonblock event loop fd: %w", err)
+	}
+	return os.NewFile(uintptr(dup), "wl-event-loop"), nil
+}
 // drain runs queued posts and Do calls until none are left, at most
 // maxDrainRounds rounds. Posts are taken again after each call is received and
 // before it runs, so a Post that happened before a Do always runs before that

M server/post_test.go +0 -1


@@ -395,7 +395,6 @@ func TestPostMultiProducerNoLostWake(t *testing.T) {
395	395		if err != nil {
396	396		t.Fatal(err)
397	397		}
398		-	d.dispatchMS = 60_000
399	398		ctx, cancel := context.WithCancel(context.Background())
400	399		runDone := make(chan error, 1)
401	400		go func() { runDone <- d.Run(ctx) }()


@@ -395,7 +395,6 @@ func TestPostMultiProducerNoLostWake(t *testing.T) {
395		if err != nil {	395	if err != nil {
396		t.Fatal(err)	396	t.Fatal(err)
397		}	397	}
398	-	d.dispatchMS = 60_000
399		ctx, cancel := context.WithCancel(context.Background())	398	ctx, cancel := context.WithCancel(context.Background())
400		runDone := make(chan error, 1)	399	runDone := make(chan error, 1)
401		go func() { runDone <- d.Run(ctx) }()	400	go func() { runDone <- d.Run(ctx) }()