f4c0e07 perf(adapters): park descriptor waits in the Go netpoller · neferwl

M Makefile +1 -1


@@ -61,7 +61,7 @@ arch: adapter-imports
61	61		# Adapters stay independent: production code may import only the shared
62	62		# adapter libraries below. outputkit holds what both output backends (drm,
63	63		# headless) run; the others are leaf helpers.
64		-	ADAPTER_LIBS := busretry\|outputkit/capture\|outputkit/corewake\|outputkit/presented\|outputkit/surfaces\|captureallow\|clock\|logging\|sessionsecurity\|syncfile\|workspaceid\|xkb
	64	+	ADAPTER_LIBS := fdwait\|busretry\|outputkit/capture\|outputkit/corewake\|outputkit/presented\|outputkit/surfaces\|captureallow\|clock\|logging\|sessionsecurity\|syncfile\|workspaceid\|xkb
65	65		adapter-imports:
66	66		@edges=$$(CGO_ENABLED=0 go list -f '{{$$p := .ImportPath}}{{range .Imports}}{{$$p}} {{.}}{{"\n"}}{{end}}' ./internal/adapters/...) \|\| exit 1; \
67	67		bad=$$(echo "$$edges" \| grep -E ' \S+/internal/adapters/' \| grep -vE '/internal/adapters/($(ADAPTER_LIBS))$$' \


@@ -61,7 +61,7 @@ arch: adapter-imports
61		# Adapters stay independent: production code may import only the shared	61		# Adapters stay independent: production code may import only the shared
62		# adapter libraries below. outputkit holds what both output backends (drm,	62		# adapter libraries below. outputkit holds what both output backends (drm,
63		# headless) run; the others are leaf helpers.	63		# headless) run; the others are leaf helpers.
64	-	ADAPTER_LIBS := busretry\|outputkit/capture\|outputkit/corewake\|outputkit/presented\|outputkit/surfaces\|captureallow\|clock\|logging\|sessionsecurity\|syncfile\|workspaceid\|xkb	64	+	ADAPTER_LIBS := fdwait\|busretry\|outputkit/capture\|outputkit/corewake\|outputkit/presented\|outputkit/surfaces\|captureallow\|clock\|logging\|sessionsecurity\|syncfile\|workspaceid\|xkb
65		adapter-imports:	65		adapter-imports:
66		@edges=$$(CGO_ENABLED=0 go list -f '{{$$p := .ImportPath}}{{range .Imports}}{{$$p}} {{.}}{{"\n"}}{{end}}' ./internal/adapters/...) \|\| exit 1; \	66		@edges=$$(CGO_ENABLED=0 go list -f '{{$$p := .ImportPath}}{{range .Imports}}{{$$p}} {{.}}{{"\n"}}{{end}}' ./internal/adapters/...) \|\| exit 1; \
67		bad=$$(echo "$$edges" \| grep -E ' \S+/internal/adapters/' \| grep -vE '/internal/adapters/($(ADAPTER_LIBS))$$' \	67		bad=$$(echo "$$edges" \| grep -E ' \S+/internal/adapters/' \| grep -vE '/internal/adapters/($(ADAPTER_LIBS))$$' \

M internal/adapters/captureallow/store.go +7 -1


@@ -13,6 +13,7 @@ import (
13	13		"time"
14	14		"unsafe"
15	15
	16	+	"github.com/bnema/neferwl/internal/adapters/fdwait"
16	17		"github.com/bnema/zerowrap"
17	18		"golang.org/x/sys/unix"
18	19		)

@@ -186,6 +187,11 @@ func (s *Store) Run(ctx context.Context) error {
186	187		return err
187	188		}
188	189		defer unix.Close(fd)
	190	+	waiter, err := fdwait.New()
	191	+	if err != nil {
	192	+	return err
	193	+	}
	194	+	defer waiter.Close()
189	195		dir, name := filepath.Dir(s.path), filepath.Base(s.path)
190	196		wd := -1
191	197		var retry, pending time.Time

@@ -224,7 +230,7 @@ func (s *Store) Run(ctx context.Context) error {
224	230		timeout = min(timeout, int(remaining.Milliseconds())+1)
225	231		}
226	232		fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
227		-	if _, err := unix.Poll(fds, timeout); err != nil {
	233	+	if _, err := waiter.Poll(fds, timeout); err != nil {
228	234		if err == unix.EINTR {
229	235		continue
230	236		}

 	"time"
 	"unsafe"
+	"github.com/bnema/neferwl/internal/adapters/fdwait"
 	"github.com/bnema/zerowrap"
 	"golang.org/x/sys/unix"
 )
 		return err
 	}
 	defer unix.Close(fd)
+	waiter, err := fdwait.New()
+	if err != nil {
+		return err
+	}
+	defer waiter.Close()
 	dir, name := filepath.Dir(s.path), filepath.Base(s.path)
 	wd := -1
 	var retry, pending time.Time
 			timeout = min(timeout, int(remaining.Milliseconds())+1)
 		}
 		fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
-		if _, err := unix.Poll(fds, timeout); err != nil {
+		if _, err := waiter.Poll(fds, timeout); err != nil {
 			if err == unix.EINTR {
 				continue
 			}

M internal/adapters/drm/card.go +7 -1


@@ -8,6 +8,7 @@ import (
8	8		"sync"
9	9		"sync/atomic"
10	10
	11	+	"github.com/bnema/neferwl/internal/adapters/fdwait"
11	12		"github.com/bnema/neferwl/internal/ports"
12	13		"github.com/bnema/zerowrap"
13	14		"golang.org/x/sys/unix"

@@ -247,11 +248,16 @@ func (c Card) open(conn connector, mode modeInfo) (Output, error) {
247	248		// output of their CRTC. Run it once per card. An event is never dropped:
248	249		// the output owes it a pending commit.
249	250		func (c *Card) ReadEvents(ctx context.Context) error {
	251	+	waiter, err := fdwait.New()
	252	+	if err != nil {
	253	+	return fmt.Errorf("drm event waiter: %w", err)
	254	+	}
	255	+	defer waiter.Close()
250	256		buf := make([]byte, 1024)
251	257		var events []flipEvent // reused by every read
252	258		for ctx.Err() == nil {
253	259		fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
254		-	n, err := unix.Poll(fds, 100)
	260	+	n, err := waiter.Poll(fds, 100)
255	261		if err != nil && !errors.Is(err, unix.EINTR) {
256	262		return fmt.Errorf("drm poll: %w", err)
257	263		}

 	"sync"
 	"sync/atomic"
+	"github.com/bnema/neferwl/internal/adapters/fdwait"
 	"github.com/bnema/neferwl/internal/ports"
 	"github.com/bnema/zerowrap"
 	"golang.org/x/sys/unix"
 // output of their CRTC. Run it once per card. An event is never dropped:
 // the output owes it a pending commit.
 func (c *Card) ReadEvents(ctx context.Context) error {
+	waiter, err := fdwait.New()
+	if err != nil {
+		return fmt.Errorf("drm event waiter: %w", err)
+	}
+	defer waiter.Close()
 	buf := make([]byte, 1024)
 	var events []flipEvent // reused by every read
 	for ctx.Err() == nil {
 		fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
-		n, err := unix.Poll(fds, 100)
+		n, err := waiter.Poll(fds, 100)
 		if err != nil && !errors.Is(err, unix.EINTR) {
 			return fmt.Errorf("drm poll: %w", err)
 		}

M internal/adapters/drm/hotplug.go +7 -1


@@ -6,6 +6,7 @@ import (
6	6		"fmt"
7	7		"sync"
8	8
	9	+	"github.com/bnema/neferwl/internal/adapters/fdwait"
9	10		"github.com/bnema/purego"
10	11		"golang.org/x/sys/unix"
11	12		)

@@ -77,9 +78,14 @@ func WatchHotplug(ctx context.Context, changed chan<- struct{}) error {
77	78		return errors.New("udev monitor setup failed")
78	79		}
79	80		fd := monitorFD(mon)
	81	+	waiter, err := fdwait.New()
	82	+	if err != nil {
	83	+	return fmt.Errorf("udev waiter: %w", err)
	84	+	}
	85	+	defer waiter.Close()
80	86		for ctx.Err() == nil {
81	87		fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
82		-	n, err := unix.Poll(fds, 100)
	88	+	n, err := waiter.Poll(fds, 100)
83	89		if err != nil && !errors.Is(err, unix.EINTR) {
84	90		return fmt.Errorf("udev poll: %w", err)
85	91		}

 	"fmt"
 	"sync"
+	"github.com/bnema/neferwl/internal/adapters/fdwait"
 	"github.com/bnema/purego"
 	"golang.org/x/sys/unix"
 )
 		return errors.New("udev monitor setup failed")
 	}
 	fd := monitorFD(mon)
+	waiter, err := fdwait.New()
+	if err != nil {
+		return fmt.Errorf("udev waiter: %w", err)
+	}
+	defer waiter.Close()
 	for ctx.Err() == nil {
 		fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
-		n, err := unix.Poll(fds, 100)
+		n, err := waiter.Poll(fds, 100)
 		if err != nil && !errors.Is(err, unix.EINTR) {
 			return fmt.Errorf("udev poll: %w", err)
 		}

A internal/adapters/fdwait/poller.go +111 -0

@@ -0,0 +1,111 @@
	1	+	// Package fdwait parks descriptor waits in Go's netpoller.
	2	+	package fdwait
	3	+
	4	+	import (
	5	+	"errors"
	6	+	"os"
	7	+	"syscall"
	8	+	"time"
	9	+
	10	+	"golang.org/x/sys/unix"
	11	+	)
	12	+
	13	+	// Poller owns an epoll descriptor and is used by one goroutine.
	14	+	// Watched descriptors remain owned by the caller.
	15	+	type Poller struct {
	16	+	file *os.File
	17	+	raw syscall.RawConn
	18	+	registered map[int]struct{}
	19	+	}
	20	+
	21	+	func New() (*Poller, error) {
	22	+	fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
	23	+	if err != nil {
	24	+	return nil, err
	25	+	}
	26	+	if err = unix.SetNonblock(fd, true); err != nil {
	27	+	unix.Close(fd)
	28	+	return nil, err
	29	+	}
	30	+	f := os.NewFile(uintptr(fd), "fdwait")
	31	+	raw, err := f.SyscallConn()
	32	+	if err != nil {
	33	+	f.Close()
	34	+	return nil, err
	35	+	}
	36	+	return &Poller{file: f, raw: raw, registered: make(map[int]struct{})}, nil
	37	+	}
	38	+
	39	+	func (p *Poller) Close() error { return p.file.Close() }
	40	+
	41	+	// Poll has the readable/error semantics of unix.Poll. A negative timeout
	42	+	// waits indefinitely. Timeout expiration returns zero, not an error.
	43	+	func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
	44	+	// Remove registrations no longer requested. Closing a watched fd already
	45	+	// removes it from epoll, so ENOENT and EBADF are harmless here.
	46	+	err := p.raw.Control(func(ep uintptr) {
	47	+	for fd := range p.registered {
	48	+	found := false
	49	+	for _, f := range fds {
	50	+	if int(f.Fd) == fd {
	51	+	found = true
	52	+	break
	53	+	}
	54	+	}
	55	+	if !found {
	56	+	unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, fd, nil)
	57	+	delete(p.registered, fd)
	58	+	}
	59	+	}
	60	+	})
	61	+	if err != nil {
	62	+	return 0, err
	63	+	}
	64	+	var ctlErr error
	65	+	err = p.raw.Control(func(ep uintptr) {
	66	+	for _, f := range fds {
	67	+	fd := int(f.Fd)
	68	+	if fd < 0 {
	69	+	continue
	70	+	}
	71	+	// MOD detects a descriptor number recycled after its old file closed.
	72	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)}
	73	+	e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_MOD, fd, &ev)
	74	+	if errors.Is(e, unix.ENOENT) {
	75	+	e = unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev)
	76	+	}
	77	+	if e != nil {
	78	+	ctlErr = e
	79	+	return
	80	+	}
	81	+	p.registered[fd] = struct{}{}
	82	+	}
	83	+	})
	84	+	if err != nil {
	85	+	return 0, err
	86	+	}
	87	+	if ctlErr != nil {
	88	+	return 0, ctlErr
	89	+	}
	90	+	if timeout >= 0 {
	91	+	err = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
	92	+	} else {
	93	+	err = p.file.SetReadDeadline(time.Time{})
	94	+	}
	95	+	if err != nil {
	96	+	return 0, err
	97	+	}
	98	+	var n int
	99	+	var pollErr error
	100	+	err = p.raw.Read(func(_ uintptr) bool {
	101	+	n, pollErr = unix.Poll(fds, 0)
	102	+	return n > 0 \|\| (pollErr != nil && !errors.Is(pollErr, unix.EINTR))
	103	+	})
	104	+	if errors.Is(err, os.ErrDeadlineExceeded) {
	105	+	return 0, nil
	106	+	}
	107	+	if err != nil {
	108	+	return 0, err
	109	+	}
	110	+	return n, pollErr
	111	+	}

@@ -0,0 +1,111 @@
	1	+	// Package fdwait parks descriptor waits in Go's netpoller.
	2	+	package fdwait
	3	+
	4	+	import (
	5	+	"errors"
	6	+	"os"
	7	+	"syscall"
	8	+	"time"
	9	+
	10	+	"golang.org/x/sys/unix"
	11	+	)
	12	+
	13	+	// Poller owns an epoll descriptor and is used by one goroutine.
	14	+	// Watched descriptors remain owned by the caller.
	15	+	type Poller struct {
	16	+	file *os.File
	17	+	raw syscall.RawConn
	18	+	registered map[int]struct{}
	19	+	}
	20	+
	21	+	func New() (*Poller, error) {
	22	+	fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
	23	+	if err != nil {
	24	+	return nil, err
	25	+	}
	26	+	if err = unix.SetNonblock(fd, true); err != nil {
	27	+	unix.Close(fd)
	28	+	return nil, err
	29	+	}
	30	+	f := os.NewFile(uintptr(fd), "fdwait")
	31	+	raw, err := f.SyscallConn()
	32	+	if err != nil {
	33	+	f.Close()
	34	+	return nil, err
	35	+	}
	36	+	return &Poller{file: f, raw: raw, registered: make(map[int]struct{})}, nil
	37	+	}
	38	+
	39	+	func (p *Poller) Close() error { return p.file.Close() }
	40	+
	41	+	// Poll has the readable/error semantics of unix.Poll. A negative timeout
	42	+	// waits indefinitely. Timeout expiration returns zero, not an error.
	43	+	func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
	44	+	// Remove registrations no longer requested. Closing a watched fd already
	45	+	// removes it from epoll, so ENOENT and EBADF are harmless here.
	46	+	err := p.raw.Control(func(ep uintptr) {
	47	+	for fd := range p.registered {
	48	+	found := false
	49	+	for _, f := range fds {
	50	+	if int(f.Fd) == fd {
	51	+	found = true
	52	+	break
	53	+	}
	54	+	}
	55	+	if !found {
	56	+	unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, fd, nil)
	57	+	delete(p.registered, fd)
	58	+	}
	59	+	}
	60	+	})
	61	+	if err != nil {
	62	+	return 0, err
	63	+	}
	64	+	var ctlErr error
	65	+	err = p.raw.Control(func(ep uintptr) {
	66	+	for _, f := range fds {
	67	+	fd := int(f.Fd)
	68	+	if fd < 0 {
	69	+	continue
	70	+	}
	71	+	// MOD detects a descriptor number recycled after its old file closed.
	72	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)}
	73	+	e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_MOD, fd, &ev)
	74	+	if errors.Is(e, unix.ENOENT) {
	75	+	e = unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev)
	76	+	}
	77	+	if e != nil {
	78	+	ctlErr = e
	79	+	return
	80	+	}
	81	+	p.registered[fd] = struct{}{}
	82	+	}
	83	+	})
	84	+	if err != nil {
	85	+	return 0, err
	86	+	}
	87	+	if ctlErr != nil {
	88	+	return 0, ctlErr
	89	+	}
	90	+	if timeout >= 0 {
	91	+	err = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
	92	+	} else {
	93	+	err = p.file.SetReadDeadline(time.Time{})
	94	+	}
	95	+	if err != nil {
	96	+	return 0, err
	97	+	}
	98	+	var n int
	99	+	var pollErr error
	100	+	err = p.raw.Read(func(_ uintptr) bool {
	101	+	n, pollErr = unix.Poll(fds, 0)
	102	+	return n > 0 \|\| (pollErr != nil && !errors.Is(pollErr, unix.EINTR))
	103	+	})
	104	+	if errors.Is(err, os.ErrDeadlineExceeded) {
	105	+	return 0, nil
	106	+	}
	107	+	if err != nil {
	108	+	return 0, err
	109	+	}
	110	+	return n, pollErr
	111	+	}

A internal/adapters/fdwait/poller_test.go +85 -0

@@ -0,0 +1,85 @@
	1	+	package fdwait
	2	+
	3	+	import (
	4	+	"golang.org/x/sys/unix"
	5	+	"testing"
	6	+	"time"
	7	+	)
	8	+
	9	+	func TestPollReadinessTimeoutAndReplacement(t *testing.T) {
	10	+	p, err := New()
	11	+	if err != nil {
	12	+	t.Fatal(err)
	13	+	}
	14	+	defer p.Close()
	15	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+	defer unix.Close(fd)
	20	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	21	+	if n, err := p.Poll(fds, 5); n != 0 \|\| err != nil {
	22	+	t.Fatalf("timeout: %d %v", n, err)
	23	+	}
	24	+	done := make(chan error, 1)
	25	+	go func() { _, e := unix.Write(fd, []byte{1, 0, 0, 0, 0, 0, 0, 0}); done <- e }()
	26	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLIN == 0 {
	27	+	t.Fatalf("readiness: %d %v %+v", n, err, fds)
	28	+	}
	29	+	if err := <-done; err != nil {
	30	+	t.Fatal(err)
	31	+	}
	32	+	// Level readiness must remain visible before the caller consumes it.
	33	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
	34	+	t.Fatalf("level: %d %v", n, err)
	35	+	}
	36	+	var b [8]byte
	37	+	if _, err := unix.Read(fd, b[:]); err != nil {
	38	+	t.Fatal(err)
	39	+	}
	40	+	if n, err := p.Poll(fds, 5); n != 0 \|\| err != nil {
	41	+	t.Fatalf("consumed: %d %v", n, err)
	42	+	}
	43	+	other, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	44	+	if err != nil {
	45	+	t.Fatal(err)
	46	+	}
	47	+	defer unix.Close(other)
	48	+	fds[0].Fd = int32(other)
	49	+	if _, err := unix.Write(other, b[:]); err != nil {
	50	+	t.Fatal(err)
	51	+	}
	52	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
	53	+	t.Fatalf("replacement: %d %v", n, err)
	54	+	}
	55	+	if _, ok := p.registered[fd]; ok {
	56	+	t.Fatal("old registration retained")
	57	+	}
	58	+	}
	59	+
	60	+	func TestCloseUnblocksPoll(t *testing.T) {
	61	+	p, err := New()
	62	+	if err != nil {
	63	+	t.Fatal(err)
	64	+	}
	65	+	defer p.Close()
	66	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	67	+	if err != nil {
	68	+	t.Fatal(err)
	69	+	}
	70	+	defer unix.Close(fd)
	71	+	done := make(chan error, 1)
	72	+	go func() { _, e := p.Poll([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, -1); done <- e }()
	73	+	// Close is safe both before and during the netpoll wait.
	74	+	if err := p.Close(); err != nil {
	75	+	t.Fatal(err)
	76	+	}
	77	+	select {
	78	+	case err := <-done:
	79	+	if err == nil {
	80	+	t.Fatal("closed wait succeeded")
	81	+	}
	82	+	case <-time.After(time.Second):
	83	+	t.Fatal("close did not unblock wait")
	84	+	}
	85	+	}

@@ -0,0 +1,85 @@
	1	+	package fdwait
	2	+
	3	+	import (
	4	+	"golang.org/x/sys/unix"
	5	+	"testing"
	6	+	"time"
	7	+	)
	8	+
	9	+	func TestPollReadinessTimeoutAndReplacement(t *testing.T) {
	10	+	p, err := New()
	11	+	if err != nil {
	12	+	t.Fatal(err)
	13	+	}
	14	+	defer p.Close()
	15	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+	defer unix.Close(fd)
	20	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	21	+	if n, err := p.Poll(fds, 5); n != 0 \|\| err != nil {
	22	+	t.Fatalf("timeout: %d %v", n, err)
	23	+	}
	24	+	done := make(chan error, 1)
	25	+	go func() { _, e := unix.Write(fd, []byte{1, 0, 0, 0, 0, 0, 0, 0}); done <- e }()
	26	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLIN == 0 {
	27	+	t.Fatalf("readiness: %d %v %+v", n, err, fds)
	28	+	}
	29	+	if err := <-done; err != nil {
	30	+	t.Fatal(err)
	31	+	}
	32	+	// Level readiness must remain visible before the caller consumes it.
	33	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
	34	+	t.Fatalf("level: %d %v", n, err)
	35	+	}
	36	+	var b [8]byte
	37	+	if _, err := unix.Read(fd, b[:]); err != nil {
	38	+	t.Fatal(err)
	39	+	}
	40	+	if n, err := p.Poll(fds, 5); n != 0 \|\| err != nil {
	41	+	t.Fatalf("consumed: %d %v", n, err)
	42	+	}
	43	+	other, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	44	+	if err != nil {
	45	+	t.Fatal(err)
	46	+	}
	47	+	defer unix.Close(other)
	48	+	fds[0].Fd = int32(other)
	49	+	if _, err := unix.Write(other, b[:]); err != nil {
	50	+	t.Fatal(err)
	51	+	}
	52	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
	53	+	t.Fatalf("replacement: %d %v", n, err)
	54	+	}
	55	+	if _, ok := p.registered[fd]; ok {
	56	+	t.Fatal("old registration retained")
	57	+	}
	58	+	}
	59	+
	60	+	func TestCloseUnblocksPoll(t *testing.T) {
	61	+	p, err := New()
	62	+	if err != nil {
	63	+	t.Fatal(err)
	64	+	}
	65	+	defer p.Close()
	66	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	67	+	if err != nil {
	68	+	t.Fatal(err)
	69	+	}
	70	+	defer unix.Close(fd)
	71	+	done := make(chan error, 1)
	72	+	go func() { _, e := p.Poll([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, -1); done <- e }()
	73	+	// Close is safe both before and during the netpoll wait.
	74	+	if err := p.Close(); err != nil {
	75	+	t.Fatal(err)
	76	+	}
	77	+	select {
	78	+	case err := <-done:
	79	+	if err == nil {
	80	+	t.Fatal("closed wait succeeded")
	81	+	}
	82	+	case <-time.After(time.Second):
	83	+	t.Fatal("close did not unblock wait")
	84	+	}
	85	+	}

A internal/adapters/fdwait/reuse_test.go +58 -0

@@ -0,0 +1,58 @@
	1	+	package fdwait
	2	+
	3	+	import (
	4	+	"golang.org/x/sys/unix"
	5	+	"testing"
	6	+	)
	7	+
	8	+	// Closing a watched file removes it from epoll. Reusing its number must
	9	+	// register the new file, not leave it waiting on the old registration.
	10	+	func TestPollRecycledDescriptor(t *testing.T) {
	11	+	p, err := New()
	12	+	if err != nil {
	13	+	t.Fatal(err)
	14	+	}
	15	+	defer p.Close()
	16	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	17	+	if err != nil {
	18	+	t.Fatal(err)
	19	+	}
	20	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	21	+	if _, err := p.Poll(fds, 1); err != nil {
	22	+	unix.Close(fd)
	23	+	t.Fatal(err)
	24	+	}
	25	+	unix.Close(fd)
	26	+	replacement, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	27	+	if err != nil {
	28	+	t.Fatal(err)
	29	+	}
	30	+	defer unix.Close(replacement)
	31	+	if replacement != fd {
	32	+	t.Skip("descriptor number not reused")
	33	+	}
	34	+	if _, err := unix.Write(replacement, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
	35	+	t.Fatal(err)
	36	+	}
	37	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
	38	+	t.Fatalf("reused descriptor: %d %v", n, err)
	39	+	}
	40	+	}
	41	+
	42	+	func TestPollHangup(t *testing.T) {
	43	+	p, err := New()
	44	+	if err != nil {
	45	+	t.Fatal(err)
	46	+	}
	47	+	defer p.Close()
	48	+	var pipe [2]int
	49	+	if err := unix.Pipe2(pipe[:], unix.O_NONBLOCK\|unix.O_CLOEXEC); err != nil {
	50	+	t.Fatal(err)
	51	+	}
	52	+	defer unix.Close(pipe[0])
	53	+	unix.Close(pipe[1])
	54	+	fds := []unix.PollFd{{Fd: int32(pipe[0]), Events: unix.POLLIN}}
	55	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLHUP == 0 {
	56	+	t.Fatalf("hangup: %d %v %+v", n, err, fds)
	57	+	}
	58	+	}

@@ -0,0 +1,58 @@
	1	+	package fdwait
	2	+
	3	+	import (
	4	+	"golang.org/x/sys/unix"
	5	+	"testing"
	6	+	)
	7	+
	8	+	// Closing a watched file removes it from epoll. Reusing its number must
	9	+	// register the new file, not leave it waiting on the old registration.
	10	+	func TestPollRecycledDescriptor(t *testing.T) {
	11	+	p, err := New()
	12	+	if err != nil {
	13	+	t.Fatal(err)
	14	+	}
	15	+	defer p.Close()
	16	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	17	+	if err != nil {
	18	+	t.Fatal(err)
	19	+	}
	20	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	21	+	if _, err := p.Poll(fds, 1); err != nil {
	22	+	unix.Close(fd)
	23	+	t.Fatal(err)
	24	+	}
	25	+	unix.Close(fd)
	26	+	replacement, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	27	+	if err != nil {
	28	+	t.Fatal(err)
	29	+	}
	30	+	defer unix.Close(replacement)
	31	+	if replacement != fd {
	32	+	t.Skip("descriptor number not reused")
	33	+	}
	34	+	if _, err := unix.Write(replacement, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
	35	+	t.Fatal(err)
	36	+	}
	37	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
	38	+	t.Fatalf("reused descriptor: %d %v", n, err)
	39	+	}
	40	+	}
	41	+
	42	+	func TestPollHangup(t *testing.T) {
	43	+	p, err := New()
	44	+	if err != nil {
	45	+	t.Fatal(err)
	46	+	}
	47	+	defer p.Close()
	48	+	var pipe [2]int
	49	+	if err := unix.Pipe2(pipe[:], unix.O_NONBLOCK\|unix.O_CLOEXEC); err != nil {
	50	+	t.Fatal(err)
	51	+	}
	52	+	defer unix.Close(pipe[0])
	53	+	unix.Close(pipe[1])
	54	+	fds := []unix.PollFd{{Fd: int32(pipe[0]), Events: unix.POLLIN}}
	55	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLHUP == 0 {
	56	+	t.Fatalf("hangup: %d %v %+v", n, err, fds)
	57	+	}
	58	+	}

M internal/adapters/libinput/libinput.go +7 -1


@@ -10,6 +10,7 @@ import (
10	10		"time"
11	11		"unsafe"
12	12
	13	+	"github.com/bnema/neferwl/internal/adapters/fdwait"
13	14		"github.com/bnema/neferwl/internal/adapters/xkb"
14	15		"github.com/bnema/neferwl/internal/ports"
15	16		"github.com/bnema/purego"

@@ -267,6 +268,11 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error
267	268		p.moved(opts.MoveCursor, false)
268	269		in := &inputState{cfg: opts.DeviceConfig, dev: libinputDevices{}, devices: map[uintptr]bool{}}
269	270		defer in.release()
	271	+	waiter, err := fdwait.New()
	272	+	if err != nil {
	273	+	return fmt.Errorf("libinput waiter: %w", err)
	274	+	}
	275	+	defer waiter.Close()
270	276		fd := getFD(li)
271	277		fwd := newForwarder(opts.Log)
272	278		fwdCtx, stopFwd := context.WithCancel(ctx)

@@ -313,7 +319,7 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error
313	319		default:
314	320		}
315	321		fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
316		-	n, err := unix.Poll(fds, 100)
	322	+	n, err := waiter.Poll(fds, 100)
317	323		if err != nil && !errors.Is(err, unix.EINTR) {
318	324		return fmt.Errorf("libinput poll: %w", err)
319	325		}

 	"time"
 	"unsafe"
+	"github.com/bnema/neferwl/internal/adapters/fdwait"
 	"github.com/bnema/neferwl/internal/adapters/xkb"
 	"github.com/bnema/neferwl/internal/ports"
 	"github.com/bnema/purego"
 	p.moved(opts.MoveCursor, false)
 	in := &inputState{cfg: opts.DeviceConfig, dev: libinputDevices{}, devices: map[uintptr]bool{}}
 	defer in.release()
+	waiter, err := fdwait.New()
+	if err != nil {
+		return fmt.Errorf("libinput waiter: %w", err)
+	}
+	defer waiter.Close()
 	fd := getFD(li)
 	fwd := newForwarder(opts.Log)
 	fwdCtx, stopFwd := context.WithCancel(ctx)
 		default:
 		}
 		fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
-		n, err := unix.Poll(fds, 100)
+		n, err := waiter.Poll(fds, 100)
 		if err != nil && !errors.Is(err, unix.EINTR) {
 			return fmt.Errorf("libinput poll: %w", err)
 		}

M internal/adapters/seat/seat.go +7 -1


@@ -10,6 +10,7 @@ import (
10	10		"sync"
11	11		"unsafe"
12	12
	13	+	"github.com/bnema/neferwl/internal/adapters/fdwait"
13	14		"github.com/bnema/purego"
14	15		"github.com/bnema/zerowrap"
15	16		"golang.org/x/sys/unix"

@@ -190,9 +191,14 @@ func (s *Seat) dispatchLocked(ctx context.Context, timeoutMs int) error {
190	191		// Run dispatches seat events until ctx ends. The owner calls Close afterwards,
191	192		// once outputs have restored their CRTC state.
192	193		func (s *Seat) Run(ctx context.Context) error {
	194	+	waiter, err := fdwait.New()
	195	+	if err != nil {
	196	+	return fmt.Errorf("seat waiter: %w", err)
	197	+	}
	198	+	defer waiter.Close()
193	199		for ctx.Err() == nil {
194	200		fds := []unix.PollFd{{Fd: getFD(s.handle), Events: unix.POLLIN}}
195		-	if _, err := unix.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) {
	201	+	if _, err := waiter.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) {
196	202		return fmt.Errorf("seat poll: %w", err)
197	203		}
198	204		// Dispatch every tick: libseat may queue events without the fd becoming readable.

 	"sync"
 	"unsafe"
+	"github.com/bnema/neferwl/internal/adapters/fdwait"
 	"github.com/bnema/purego"
 	"github.com/bnema/zerowrap"
 	"golang.org/x/sys/unix"
 // Run dispatches seat events until ctx ends. The owner calls Close afterwards,
 // once outputs have restored their CRTC state.
 func (s *Seat) Run(ctx context.Context) error {
+	waiter, err := fdwait.New()
+	if err != nil {
+		return fmt.Errorf("seat waiter: %w", err)
+	}
+	defer waiter.Close()
 	for ctx.Err() == nil {
 		fds := []unix.PollFd{{Fd: getFD(s.handle), Events: unix.POLLIN}}
-		if _, err := unix.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) {
+		if _, err := waiter.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) {
 			return fmt.Errorf("seat poll: %w", err)
 		}
 		// Dispatch every tick: libseat may queue events without the fd becoming readable.

M internal/adapters/wayland/syncobj.go +21 -5


@@ -9,6 +9,7 @@ import (
9	9
10	10		"github.com/bnema/go-wayland-bindings/server/linuxdrmsyncobj"
11	11		"github.com/bnema/go-wayland-bindings/server/wayland"
	12	+	"github.com/bnema/neferwl/internal/adapters/fdwait"
12	13		"github.com/bnema/neferwl/internal/ports"
13	14		"github.com/bnema/purego-libwayland/server"
14	15		"github.com/bnema/zerowrap"

@@ -197,14 +198,16 @@ type acquirePoller interface {
197	198		poll([]unix.PollFd, int) (int, error)
198	199		}
199	200
200		-	type unixAcquirePoller struct{}
	201	+	type unixAcquirePoller struct{ waiter *fdwait.Poller }
201	202
202		-	func (unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
203		-	return unix.Poll(fds, timeout)
	203	+	func (p unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
	204	+	return p.waiter.Poll(fds, timeout)
204	205		}
205	206
206	207		type syncWaiter struct {
207	208		poller acquirePoller
	209	+	waiter *fdwait.Poller
	210	+	pipeFD int
208	211		terminal bool // polling ended in failure; protected by mu
209	212		mu sync.Mutex
210	213		ready map[*syncWait]bool

@@ -229,7 +232,13 @@ func newSyncWaiter(wake func()) (*syncWaiter, error) {
229	232		if err != nil {
230	233		return nil, err
231	234		}
232		-	return &syncWaiter{poller: unixAcquirePoller{}, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
	235	+	p, err := fdwait.New()
	236	+	if err != nil {
	237	+	r.Close()
	238	+	w.Close()
	239	+	return nil, err
	240	+	}
	241	+	return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, pipeFD: int(r.Fd()), ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
233	242		}
234	243
235	244		// watch registers an eventfd for p; the wait is ready once it fires.

@@ -375,6 +384,13 @@ func (sw syncWaiter) fail(waits []syncWait) {
375	384		// run polls every watched eventfd; it owns and closes them.
376	385		func (sw *syncWaiter) run(ctx context.Context) {
377	386		defer sw.pipeR.Close()
	387	+	if sw.waiter != nil {
	388	+	defer sw.waiter.Close()
	389	+	}
	390	+	pipeFD := sw.pipeFD
	391	+	if sw.waiter == nil { // bare test fixtures own their pipe
	392	+	pipeFD = int(sw.pipeR.Fd())
	393	+	}
378	394		var waits []*syncWait
379	395		defer func() {
380	396		sw.mu.Lock()

@@ -395,7 +411,7 @@ func (sw *syncWaiter) run(ctx context.Context) {
395	411		live = append(live, w)
396	412		}
397	413		waits = live
398		-	fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}
	414	+	fds := []unix.PollFd{{Fd: int32(pipeFD), Events: unix.POLLIN}}
399	415		for _, w := range waits {
400	416		fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
401	417		}

 	"github.com/bnema/go-wayland-bindings/server/linuxdrmsyncobj"
 	"github.com/bnema/go-wayland-bindings/server/wayland"
+	"github.com/bnema/neferwl/internal/adapters/fdwait"
 	"github.com/bnema/neferwl/internal/ports"
 	"github.com/bnema/purego-libwayland/server"
 	"github.com/bnema/zerowrap"
 	poll([]unix.PollFd, int) (int, error)
 }
-type unixAcquirePoller struct{}
+type unixAcquirePoller struct{ waiter *fdwait.Poller }
-func (unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
+func (p unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
-	return unix.Poll(fds, timeout)
+	return p.waiter.Poll(fds, timeout)
 }
 type syncWaiter struct {
 	poller    acquirePoller
+	waiter    *fdwait.Poller
+	pipeFD    int
 	terminal  bool // polling ended in failure; protected by mu
 	mu        sync.Mutex
 	ready     map[*syncWait]bool
 	if err != nil {
 		return nil, err
 	}
-	return &syncWaiter{poller: unixAcquirePoller{}, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
+	p, err := fdwait.New()
+	if err != nil {
+		r.Close()
+		w.Close()
+		return nil, err
+	}
+	return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, pipeFD: int(r.Fd()), ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
 }
 // watch registers an eventfd for p; the wait is ready once it fires.
 // run polls every watched eventfd; it owns and closes them.
 func (sw *syncWaiter) run(ctx context.Context) {
 	defer sw.pipeR.Close()
+	if sw.waiter != nil {
+		defer sw.waiter.Close()
+	}
+	pipeFD := sw.pipeFD
+	if sw.waiter == nil { // bare test fixtures own their pipe
+		pipeFD = int(sw.pipeR.Fd())
+	}
 	var waits []*syncWait
 	defer func() {
 		sw.mu.Lock()
 			live = append(live, w)
 		}
 		waits = live
-		fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}
+		fds := []unix.PollFd{{Fd: int32(pipeFD), Events: unix.POLLIN}}
 		for _, w := range waits {
 			fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
 		}