perf(fdwait): pin long-lived fds and wake waits without timeouts

Pin registers a descriptor once for the poller's lifetime, so a Poll
on pinned fds makes no epoll_ctl call and allocates nothing. A pinned
eventfd lets Wake end a wait from any goroutine, and StopOn ends every
Poll once a context is done, so loops need no polling timeout to see
work or shutdown. The readiness check runs under the wake file's
reference, so a concurrent Close cannot hand its number to a new file.
A ready pinned fd that a Poll does not list fails it instead of spinning.

commit: 182dc8b0c1ec67561041e70be4c76a7539aae29f
author: brice · 9d ago · 2026-10-14 04:06 UTC
committer: brice · 8d ago · 2026-10-14 04:35 UTC
parent: bda62c7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +348 -72

M internal/adapters/fdwait/edges_test.go +125 -1

@@ -1,11 +1,133 @@
1	1		package fdwait
2	2
3	3		import (
	4	+	"context"
	5	+	"errors"
4	6		"testing"
	7	+	"time"
5	8
6	9		"golang.org/x/sys/unix"
7	10		)
8	11
	12	+	// A pinned fd is reported like a transient one, costs no epoll call per
	13	+	// Poll, and a ready pinned fd a Poll does not list fails it instead of
	14	+	// spinning.
	15	+	func TestPollPinned(t *testing.T) {
	16	+	p, err := New()
	17	+	if err != nil {
	18	+	t.Fatal(err)
	19	+	}
	20	+	defer p.Close()
	21	+	pinned, other := eventfd(t), eventfd(t)
	22	+	if err := p.Pin(int32(pinned)); err != nil {
	23	+	t.Fatal(err)
	24	+	}
	25	+	fds := []unix.PollFd{{Fd: int32(pinned), Events: unix.POLLIN}}
	26	+	if n, err := p.Poll(fds, 1); n != 0 \|\| err != nil {
	27	+	t.Fatalf("idle: %d %v", n, err)
	28	+	}
	29	+	go unix.Write(pinned, []byte{1, 0, 0, 0, 0, 0, 0, 0})
	30	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLIN == 0 {
	31	+	t.Fatalf("pinned ready: %d %v %+v", n, err, fds)
	32	+	}
	33	+	// Still registered after Poll returns: it is pinned.
	34	+	if err := p.raw.Control(func(ep uintptr) {
	35	+	var events [2]unix.EpollEvent
	36	+	if n, err := unix.EpollWait(int(ep), events[:], 0); n != 1 \|\| err != nil \|\| events[0].Fd != int32(pinned) {
	37	+	t.Errorf("pinned registration: %d %v %+v", n, err, events[:n])
	38	+	}
	39	+	}); err != nil {
	40	+	t.Fatal(err)
	41	+	}
	42	+	if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1000); err == nil {
	43	+	t.Fatal("Poll without a ready pinned fd succeeded")
	44	+	}
	45	+	}
	46	+
	47	+	// Polling only pinned fds allocates nothing and makes no epoll_ctl call.
	48	+	func TestPollPinnedAllocations(t *testing.T) {
	49	+	p, err := New()
	50	+	if err != nil {
	51	+	t.Fatal(err)
	52	+	}
	53	+	defer p.Close()
	54	+	fd := eventfd(t)
	55	+	if err := p.Pin(int32(fd)); err != nil {
	56	+	t.Fatal(err)
	57	+	}
	58	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	59	+	signal(t, fd)
	60	+	p.Poll(fds, -1)
	61	+	if allocs := testing.AllocsPerRun(100, func() { p.Poll(fds, -1) }); allocs != 0 {
	62	+	t.Errorf("pinned ready Poll: %.1f allocs, want 0", allocs)
	63	+	}
	64	+	}
	65	+
	66	+	// Wake ends a wait without a timeout and returns no events; wakes before a
	67	+	// Poll coalesce into one early return.
	68	+	func TestPollWake(t *testing.T) {
	69	+	p, err := New()
	70	+	if err != nil {
	71	+	t.Fatal(err)
	72	+	}
	73	+	defer p.Close()
	74	+	fds := []unix.PollFd{{Fd: int32(eventfd(t)), Events: unix.POLLIN}}
	75	+	done := make(chan error, 1)
	76	+	go func() {
	77	+	n, err := p.Poll(fds, -1)
	78	+	if err == nil && n != 0 {
	79	+	err = errors.New("woken poll reported events")
	80	+	}
	81	+	done <- err
	82	+	}()
	83	+	p.Wake()
	84	+	select {
	85	+	case err := <-done:
	86	+	if err != nil {
	87	+	t.Fatal(err)
	88	+	}
	89	+	case <-time.After(5 * time.Second):
	90	+	t.Fatal("Wake did not end the wait")
	91	+	}
	92	+	p.Wake()
	93	+	p.Wake()
	94	+	if n, err := p.Poll(fds, 1000); n != 0 \|\| err != nil {
	95	+	t.Fatalf("coalesced wake: %d %v", n, err)
	96	+	}
	97	+	if n, err := p.Poll(fds, 1); n != 0 \|\| err != nil {
	98	+	t.Fatalf("second wake was not coalesced: %d %v", n, err)
	99	+	}
	100	+	}
	101	+
	102	+	// After the StopOn context ends, a running and every later Poll return at
	103	+	// once, also with no timeout.
	104	+	func TestPollStopOn(t *testing.T) {
	105	+	p, err := New()
	106	+	if err != nil {
	107	+	t.Fatal(err)
	108	+	}
	109	+	defer p.Close()
	110	+	ctx, cancel := context.WithCancel(context.Background())
	111	+	p.StopOn(ctx)
	112	+	fds := []unix.PollFd{{Fd: int32(eventfd(t)), Events: unix.POLLIN}}
	113	+	done := make(chan error, 1)
	114	+	go func() { _, err := p.Poll(fds, -1); done <- err }()
	115	+	cancel()
	116	+	select {
	117	+	case err := <-done:
	118	+	if err != nil {
	119	+	t.Fatal(err)
	120	+	}
	121	+	case <-time.After(5 * time.Second):
	122	+	t.Fatal("cancel did not end the wait")
	123	+	}
	124	+	for range 3 {
	125	+	if n, err := p.Poll(fds, -1); n != 0 \|\| err != nil {
	126	+	t.Fatalf("poll after stop: %d %v", n, err)
	127	+	}
	128	+	}
	129	+	}
	130	+
9	131		func eventfd(t *testing.T) int {
10	132		t.Helper()
11	133		fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)

@@ -121,7 +243,9 @@ func TestPollFailsAfterLostRegistration(t *testing.T) {
121	243		t.Fatal(err)
122	244		}
123	245		unix.Close(fd) // what a caller closing fd during a wait does
124		-	p.unwatch([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}})
	246	+	p.fds, p.added = []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, 1
	247	+	p.unwatch()
	248	+	p.fds = nil
125	249		other := eventfd(t)
126	250		if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {
127	251		t.Fatal("Poll succeeded with a registration it could not remove")

@@ -1,11 +1,133 @@
1		package fdwait	1		package fdwait
2			2
3		import (	3		import (
			4	+	"context"
			5	+	"errors"
4		"testing"	6		"testing"
			7	+	"time"
5			8
6		"golang.org/x/sys/unix"	9		"golang.org/x/sys/unix"
7		)	10		)
8			11
			12	+	// A pinned fd is reported like a transient one, costs no epoll call per
			13	+	// Poll, and a ready pinned fd a Poll does not list fails it instead of
			14	+	// spinning.
			15	+	func TestPollPinned(t *testing.T) {
			16	+	p, err := New()
			17	+	if err != nil {
			18	+	t.Fatal(err)
			19	+	}
			20	+	defer p.Close()
			21	+	pinned, other := eventfd(t), eventfd(t)
			22	+	if err := p.Pin(int32(pinned)); err != nil {
			23	+	t.Fatal(err)
			24	+	}
			25	+	fds := []unix.PollFd{{Fd: int32(pinned), Events: unix.POLLIN}}
			26	+	if n, err := p.Poll(fds, 1); n != 0 \|\| err != nil {
			27	+	t.Fatalf("idle: %d %v", n, err)
			28	+	}
			29	+	go unix.Write(pinned, []byte{1, 0, 0, 0, 0, 0, 0, 0})
			30	+	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLIN == 0 {
			31	+	t.Fatalf("pinned ready: %d %v %+v", n, err, fds)
			32	+	}
			33	+	// Still registered after Poll returns: it is pinned.
			34	+	if err := p.raw.Control(func(ep uintptr) {
			35	+	var events [2]unix.EpollEvent
			36	+	if n, err := unix.EpollWait(int(ep), events[:], 0); n != 1 \|\| err != nil \|\| events[0].Fd != int32(pinned) {
			37	+	t.Errorf("pinned registration: %d %v %+v", n, err, events[:n])
			38	+	}
			39	+	}); err != nil {
			40	+	t.Fatal(err)
			41	+	}
			42	+	if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1000); err == nil {
			43	+	t.Fatal("Poll without a ready pinned fd succeeded")
			44	+	}
			45	+	}
			46	+
			47	+	// Polling only pinned fds allocates nothing and makes no epoll_ctl call.
			48	+	func TestPollPinnedAllocations(t *testing.T) {
			49	+	p, err := New()
			50	+	if err != nil {
			51	+	t.Fatal(err)
			52	+	}
			53	+	defer p.Close()
			54	+	fd := eventfd(t)
			55	+	if err := p.Pin(int32(fd)); err != nil {
			56	+	t.Fatal(err)
			57	+	}
			58	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
			59	+	signal(t, fd)
			60	+	p.Poll(fds, -1)
			61	+	if allocs := testing.AllocsPerRun(100, func() { p.Poll(fds, -1) }); allocs != 0 {
			62	+	t.Errorf("pinned ready Poll: %.1f allocs, want 0", allocs)
			63	+	}
			64	+	}
			65	+
			66	+	// Wake ends a wait without a timeout and returns no events; wakes before a
			67	+	// Poll coalesce into one early return.
			68	+	func TestPollWake(t *testing.T) {
			69	+	p, err := New()
			70	+	if err != nil {
			71	+	t.Fatal(err)
			72	+	}
			73	+	defer p.Close()
			74	+	fds := []unix.PollFd{{Fd: int32(eventfd(t)), Events: unix.POLLIN}}
			75	+	done := make(chan error, 1)
			76	+	go func() {
			77	+	n, err := p.Poll(fds, -1)
			78	+	if err == nil && n != 0 {
			79	+	err = errors.New("woken poll reported events")
			80	+	}
			81	+	done <- err
			82	+	}()
			83	+	p.Wake()
			84	+	select {
			85	+	case err := <-done:
			86	+	if err != nil {
			87	+	t.Fatal(err)
			88	+	}
			89	+	case <-time.After(5 * time.Second):
			90	+	t.Fatal("Wake did not end the wait")
			91	+	}
			92	+	p.Wake()
			93	+	p.Wake()
			94	+	if n, err := p.Poll(fds, 1000); n != 0 \|\| err != nil {
			95	+	t.Fatalf("coalesced wake: %d %v", n, err)
			96	+	}
			97	+	if n, err := p.Poll(fds, 1); n != 0 \|\| err != nil {
			98	+	t.Fatalf("second wake was not coalesced: %d %v", n, err)
			99	+	}
			100	+	}
			101	+
			102	+	// After the StopOn context ends, a running and every later Poll return at
			103	+	// once, also with no timeout.
			104	+	func TestPollStopOn(t *testing.T) {
			105	+	p, err := New()
			106	+	if err != nil {
			107	+	t.Fatal(err)
			108	+	}
			109	+	defer p.Close()
			110	+	ctx, cancel := context.WithCancel(context.Background())
			111	+	p.StopOn(ctx)
			112	+	fds := []unix.PollFd{{Fd: int32(eventfd(t)), Events: unix.POLLIN}}
			113	+	done := make(chan error, 1)
			114	+	go func() { _, err := p.Poll(fds, -1); done <- err }()
			115	+	cancel()
			116	+	select {
			117	+	case err := <-done:
			118	+	if err != nil {
			119	+	t.Fatal(err)
			120	+	}
			121	+	case <-time.After(5 * time.Second):
			122	+	t.Fatal("cancel did not end the wait")
			123	+	}
			124	+	for range 3 {
			125	+	if n, err := p.Poll(fds, -1); n != 0 \|\| err != nil {
			126	+	t.Fatalf("poll after stop: %d %v", n, err)
			127	+	}
			128	+	}
			129	+	}
			130	+
9		func eventfd(t *testing.T) int {	131		func eventfd(t *testing.T) int {
10		t.Helper()	132		t.Helper()
11		fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)	133		fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)

@@ -121,7 +243,9 @@ func TestPollFailsAfterLostRegistration(t *testing.T) {
121		t.Fatal(err)	243		t.Fatal(err)
122		}	244		}
123		unix.Close(fd) // what a caller closing fd during a wait does	245		unix.Close(fd) // what a caller closing fd during a wait does
124	-	p.unwatch([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}})	246	+	p.fds, p.added = []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, 1
			247	+	p.unwatch()
			248	+	p.fds = nil
125		other := eventfd(t)	249		other := eventfd(t)
126		if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {	250		if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {
127		t.Fatal("Poll succeeded with a registration it could not remove")	251		t.Fatal("Poll succeeded with a registration it could not remove")

M internal/adapters/fdwait/poller.go +223 -71


@@ -2,6 +2,7 @@
2	2		package fdwait
3	3
4	4		import (
	5	+	"context"
5	6		"errors"
6	7		"fmt"
7	8		"os"

@@ -12,16 +13,46 @@ import (
12	13		"golang.org/x/sys/unix"
13	14		)
14	15
15		-	// Poller owns an epoll descriptor and is used by one goroutine; Close may
16		-	// also run from another goroutine to end a wait. Watched descriptors remain
17		-	// owned by the caller, who must not close them while a Poll waits.
	16	+	// Poller owns an epoll descriptor and is used by one goroutine; Wake and
	17	+	// Close may run from any goroutine. Watched descriptors remain owned by the
	18	+	// caller, who must not close them while a Poll waits.
	19	+	//
	20	+	// Descriptors that live as long as the poller are registered once with Pin;
	21	+	// others are registered for one Poll call. A Poll returns on a watched fd,
	22	+	// on Wake, once the context given to StopOn ends, or on Close: a loop needs
	23	+	// no polling timeout to notice work or shutdown.
18	24		type Poller struct {
19	25		file *os.File
20	26		raw syscall.RawConn
	27	+	// wake is an eventfd pinned in the epoll set: Wake writes it.
	28	+	wake *os.File
	29	+	wakeRaw syscall.RawConn
	30	+	wakeFD int32
	31	+	ctx context.Context
	32	+	stop func() bool
	33	+	pinned []int32 // besides wakeFD
21	34		// broken is set when a registration could not be removed: it may keep
22	35		// the epoll descriptor readable forever, so every later Poll fails
23	36		// instead of spinning.
24	37		broken error
	38	+	// noDeadline: the read deadline is already cleared.
	39	+	noDeadline bool
	40	+
	41	+	// State of the running Poll, used by the callbacks below. They are
	42	+	// bound once in New, so a Poll allocates nothing.
	43	+	fds []unix.PollFd
	44	+	check []unix.PollFd
	45	+	added int
	46	+	ctlErr error
	47	+	n int
	48	+	pollErr error
	49	+	buf [8]byte
	50	+	addFn func(uintptr)
	51	+	delFn func(uintptr)
	52	+	checkFn func(uintptr)
	53	+	readFn func(uintptr) bool
	54	+	drainFn func(uintptr)
	55	+	kickFn func(uintptr)
25	56		}
26	57
27	58		func New() (*Poller, error) {

@@ -33,108 +64,229 @@ func New() (*Poller, error) {
33	64		unix.Close(fd)
34	65		return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)
35	66		}
36		-	f := os.NewFile(uintptr(fd), "fdwait")
37		-	raw, err := f.SyscallConn()
38		-	if err != nil {
39		-	f.Close()
	67	+	p := &Poller{file: os.NewFile(uintptr(fd), "fdwait")}
	68	+	if p.raw, err = p.file.SyscallConn(); err != nil {
	69	+	p.file.Close()
40	70		return nil, fmt.Errorf("fdwait epoll conn: %w", err)
41	71		}
42		-	return &Poller{file: f, raw: raw}, nil
	72	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
	73	+	if err != nil {
	74	+	p.file.Close()
	75	+	return nil, fmt.Errorf("fdwait wake eventfd: %w", err)
	76	+	}
	77	+	p.wake, p.wakeFD = os.NewFile(uintptr(efd), "fdwait-wake"), int32(efd)
	78	+	p.addFn, p.delFn, p.readFn, p.checkFn = p.watch, p.unwatchAll, p.ready, p.pollNow
	79	+	p.drainFn, p.kickFn = p.drain, p.kick
	80	+	if p.wakeRaw, err = p.wake.SyscallConn(); err == nil {
	81	+	err = p.control(unix.EPOLL_CTL_ADD, p.wakeFD)
	82	+	}
	83	+	if err != nil {
	84	+	p.Close()
	85	+	return nil, fmt.Errorf("fdwait wake: %w", err)
	86	+	}
	87	+	return p, nil
43	88		}
44	89
45		-	func (p *Poller) Close() error { return p.file.Close() }
	90	+	// Close ends a running Poll and releases the epoll and wake descriptors.
	91	+	func (p *Poller) Close() error {
	92	+	if p.stop != nil {
	93	+	p.stop()
	94	+	}
	95	+	err := p.file.Close()
	96	+	p.wake.Close()
	97	+	return err
	98	+	}
	99	+
	100	+	func (p *Poller) control(op int, fd int32) error {
	101	+	var err error
	102	+	if cerr := p.raw.Control(func(ep uintptr) {
	103	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: fd}
	104	+	err = unix.EpollCtl(int(ep), op, int(fd), &ev)
	105	+	}); cerr != nil {
	106	+	return cerr
	107	+	}
	108	+	return err
	109	+	}
	110	+
	111	+	// Pin registers fd for every later Poll, which then costs it no epoll call.
	112	+	// fd must stay open until Close, and every Poll must list it: a ready pinned
	113	+	// fd that a Poll does not list fails that Poll.
	114	+	func (p *Poller) Pin(fd int32) error {
	115	+	if err := p.control(unix.EPOLL_CTL_ADD, fd); err != nil {
	116	+	return fmt.Errorf("fdwait pin fd %d: %w", fd, err)
	117	+	}
	118	+	p.pinned = append(p.pinned, fd)
	119	+	return nil
	120	+	}
	121	+
	122	+	// StopOn makes every Poll return (0, nil) at once after ctx ends. Call it
	123	+	// once, before polling.
	124	+	func (p *Poller) StopOn(ctx context.Context) {
	125	+	p.ctx = ctx
	126	+	p.stop = context.AfterFunc(ctx, p.Wake)
	127	+	}
	128	+
	129	+	// Wake makes the running or next Poll return. Wakes before that Poll
	130	+	// coalesce. Safe from any goroutine, also after Close.
	131	+	func (p *Poller) Wake() { _ = p.wakeRaw.Control(p.kickFn) }
	132	+
	133	+	func (p *Poller) kick(fd uintptr) {
	134	+	one := [8]byte{1}
	135	+	_, _ = unix.Write(int(fd), one[:])
	136	+	}
	137	+
	138	+	func (p *Poller) drain(fd uintptr) { _, _ = unix.Read(int(fd), p.buf[:]) }
	139	+
	140	+	func (p *Poller) stopped() bool { return p.ctx != nil && p.ctx.Err() != nil }
46	141
47	142		// Poll waits like poll(2) for readable/error events on fds and fills their
48	143		// Revents. It never consumes caller events. Negative fds are ignored and a
49	144		// descriptor listed twice is watched once. timeout is in milliseconds; a
50		-	// negative one waits until an event or Close.
	145	+	// negative one waits until an event, Wake, the end of the StopOn context or
	146	+	// Close. A Wake or stop returns the events seen, possibly none.
51	147		//
52		-	// Each call explicitly removes its epoll registrations before returning.
53		-	// Relying on close to remove them is unsafe: another fd or a kernel fence
54		-	// may retain the open file description and keep the epoll fd readable.
55		-	// When nothing is ready yet, a descriptor epoll cannot watch (EPERM,
56		-	// EINVAL, ELOOP for epoll nested too deep) fails the call.
	148	+	// Each call removes the registrations it made before returning. Relying on
	149	+	// close to remove them is unsafe: another fd or a kernel fence may retain
	150	+	// the open file description and keep the epoll fd readable. When nothing is
	151	+	// ready yet, a descriptor epoll cannot watch (EPERM, EINVAL, ELOOP for epoll
	152	+	// nested too deep) fails the call.
57	153		func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
58	154		if p.broken != nil {
59	155		return 0, p.broken
60	156		}
61		-	// Ready descriptors need no registration, and an expired deadline
62		-	// would skip the readiness check in RawConn.Read.
63		-	if n, err := pollNow(fds); n != 0 \|\| err != nil \|\| timeout == 0 {
64		-	return n, err
65		-	}
66		-	var ctlErr error
67		-	added := 0
68		-	err := p.raw.Control(func(ep uintptr) {
69		-	for _, f := range fds {
70		-	if f.Fd >= 0 {
71		-	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
72		-	e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev)
73		-	if e != nil && !(errors.Is(e, unix.EEXIST) && listedBefore(fds, added)) {
74		-	ctlErr = fmt.Errorf("fdwait watch fd %d: %w", f.Fd, e)
75		-	return
76		-	}
77		-	}
78		-	added++
79		-	}
80		-	})
81		-	defer p.unwatch(fds[:added])
82		-	if err != nil {
83		-	return 0, err
	157	+	p.fds = fds
	158	+	defer func() { p.fds = nil }()
	159	+	transient := false
	160	+	for i := range fds {
	161	+	transient = transient \|\| !p.skip(i)
84	162		}
85		-	if ctlErr != nil {
86		-	return 0, ctlErr
	163	+	// An expired deadline skips RawConn.Read's own check, and a ready
	164	+	// transient fd needs no registration. Pinned fds are checked by Read.
	165	+	if (timeout == 0 \|\| transient) && p.ready(0) \|\| timeout == 0 {
	166	+	return p.n, p.pollErr
87	167		}
88		-	if timeout > 0 {
89		-	err = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
90		-	} else {
91		-	err = p.file.SetReadDeadline(time.Time{})
	168	+	p.added, p.ctlErr = 0, nil
	169	+	if transient {
	170	+	err := p.raw.Control(p.addFn)
	171	+	defer p.unwatch()
	172	+	if err != nil {
	173	+	return 0, err
	174	+	}
	175	+	if p.ctlErr != nil {
	176	+	return 0, p.ctlErr
	177	+	}
	178	+	}
	179	+	var err error
	180	+	switch {
	181	+	case timeout > 0:
	182	+	err, p.noDeadline = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout)*time.Millisecond)), false
	183	+	case !p.noDeadline:
	184	+	err, p.noDeadline = p.file.SetReadDeadline(time.Time{}), true
92	185		}
93	186		if err != nil {
94	187		return 0, err
95	188		}
96		-	var n int
97		-	var pollErr error
98		-	err = p.raw.Read(func(_ uintptr) bool {
99		-	n, pollErr = pollNow(fds)
100		-	return n != 0 \|\| pollErr != nil
101		-	})
	189	+	err = p.raw.Read(p.readFn)
102	190		if errors.Is(err, os.ErrDeadlineExceeded) {
103	191		return 0, nil
104	192		}
105	193		if err != nil {
106	194		return 0, err
107	195		}
108		-	return n, pollErr
	196	+	return p.n, p.pollErr
109	197		}
110	198
111		-	// unwatch removes the registrations Poll added for fds. A failure means a
112		-	// watched fd was closed during the wait: its registration may outlive it.
113		-	func (p *Poller) unwatch(fds []unix.PollFd) {
114		-	p.raw.Control(func(ep uintptr) {
115		-	for i, f := range fds {
116		-	if f.Fd < 0 \|\| listedBefore(fds, i) {
117		-	continue
	199	+	// ready checks the caller's fds, the wake fd and unlisted pinned fds without
	200	+	// blocking, retrying an interrupted check. It reports whether Poll returns.
	201	+	func (p *Poller) ready(uintptr) bool {
	202	+	p.check = append(append(p.check[:0], p.fds...), unix.PollFd{Fd: p.wakeFD, Events: unix.POLLIN})
	203	+	for _, fd := range p.pinned {
	204	+	if !slices.ContainsFunc(p.fds, func(f unix.PollFd) bool { return f.Fd == fd }) {
	205	+	p.check = append(p.check, unix.PollFd{Fd: fd, Events: unix.POLLIN})
	206	+	}
	207	+	}
	208	+	// Under the wake file's reference, so a concurrent Close cannot free
	209	+	// its number for reuse during the check.
	210	+	if p.wakeRaw.Control(p.checkFn) != nil {
	211	+	p.n, p.pollErr = 0, os.ErrClosed
	212	+	}
	213	+	if p.pollErr != nil {
	214	+	p.n = 0
	215	+	return true
	216	+	}
	217	+	for i := range p.fds {
	218	+	p.fds[i].Revents = p.check[i].Revents
	219	+	}
	220	+	for _, f := range p.check[len(p.fds)+1:] {
	221	+	if f.Revents != 0 {
	222	+	// Its readiness keeps the epoll fd readable: waiting would spin.
	223	+	p.n, p.pollErr = 0, fmt.Errorf("fdwait: pinned fd %d is ready but not polled", f.Fd)
	224	+	return true
	225	+	}
	226	+	}
	227	+	woke := p.check[len(p.fds)].Revents != 0
	228	+	if woke {
	229	+	p.n--
	230	+	// A stopped poller stays awake: every later Poll returns at once.
	231	+	if !p.stopped() {
	232	+	_ = p.wakeRaw.Control(p.drainFn)
	233	+	if p.stopped() { // the stop's write may have been drained
	234	+	p.Wake()
118	235		}
119		-	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil); e != nil && p.broken == nil {
120		-	p.broken = fmt.Errorf("fdwait unwatch fd %d (closed during a wait?): %w", f.Fd, e)
	236	+	}
	237	+	}
	238	+	return p.n != 0 \|\| woke
	239	+	}
	240	+
	241	+	func (p *Poller) pollNow(uintptr) {
	242	+	for {
	243	+	p.n, p.pollErr = unix.Poll(p.check, 0)
	244	+	if !errors.Is(p.pollErr, unix.EINTR) {
	245	+	return
	246	+	}
	247	+	}
	248	+	}
	249	+
	250	+	// skip reports whether fds[i] needs no registration of its own.
	251	+	func (p *Poller) skip(i int) bool {
	252	+	fd := p.fds[i].Fd
	253	+	return fd < 0 \|\| slices.Contains(p.pinned, fd) \|\| listedBefore(p.fds, i)
	254	+	}
	255	+
	256	+	func (p *Poller) watch(ep uintptr) {
	257	+	for i, f := range p.fds {
	258	+	if !p.skip(i) {
	259	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
	260	+	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev); e != nil {
	261	+	p.ctlErr = fmt.Errorf("fdwait watch fd %d: %w", f.Fd, e)
	262	+	return
121	263		}
122	264		}
123		-	})
	265	+	p.added++
	266	+	}
124	267		}
125	268
126		-	// listedBefore reports whether fds[i].Fd also appears earlier in fds.
127		-	func listedBefore(fds []unix.PollFd, i int) bool {
128		-	fd := fds[i].Fd
129		-	return slices.ContainsFunc(fds[:i], func(f unix.PollFd) bool { return f.Fd == fd })
	269	+	// unwatch removes the registrations Poll added. A failure means a watched
	270	+	// fd was closed during the wait: its registration may outlive it.
	271	+	func (p *Poller) unwatch() {
	272	+	if p.added > 0 {
	273	+	_ = p.raw.Control(p.delFn)
	274	+	}
130	275		}
131	276
132		-	// pollNow checks fds without blocking, retrying an interrupted check.
133		-	func pollNow(fds []unix.PollFd) (int, error) {
134		-	for {
135		-	n, err := unix.Poll(fds, 0)
136		-	if !errors.Is(err, unix.EINTR) {
137		-	return n, err
	277	+	func (p *Poller) unwatchAll(ep uintptr) {
	278	+	for i, f := range p.fds[:p.added] {
	279	+	if p.skip(i) {
	280	+	continue
	281	+	}
	282	+	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil); e != nil && p.broken == nil {
	283	+	p.broken = fmt.Errorf("fdwait unwatch fd %d (closed during a wait?): %w", f.Fd, e)
138	284		}
139	285		}
140	286		}
	287	+
	288	+	// listedBefore reports whether fds[i].Fd also appears earlier in fds.
	289	+	func listedBefore(fds []unix.PollFd, i int) bool {
	290	+	fd := fds[i].Fd
	291	+	return slices.ContainsFunc(fds[:i], func(f unix.PollFd) bool { return f.Fd == fd })
	292	+	}


@@ -2,6 +2,7 @@
2		package fdwait	2		package fdwait
3			3
4		import (	4		import (
			5	+	"context"
5		"errors"	6		"errors"
6		"fmt"	7		"fmt"
7		"os"	8		"os"

@@ -12,16 +13,46 @@ import (
12		"golang.org/x/sys/unix"	13		"golang.org/x/sys/unix"
13		)	14		)
14			15
15	-	// Poller owns an epoll descriptor and is used by one goroutine; Close may	16	+	// Poller owns an epoll descriptor and is used by one goroutine; Wake and
16	-	// also run from another goroutine to end a wait. Watched descriptors remain	17	+	// Close may run from any goroutine. Watched descriptors remain owned by the
17	-	// owned by the caller, who must not close them while a Poll waits.	18	+	// caller, who must not close them while a Poll waits.
			19	+	//
			20	+	// Descriptors that live as long as the poller are registered once with Pin;
			21	+	// others are registered for one Poll call. A Poll returns on a watched fd,
			22	+	// on Wake, once the context given to StopOn ends, or on Close: a loop needs
			23	+	// no polling timeout to notice work or shutdown.
18		type Poller struct {	24		type Poller struct {
19		file *os.File	25		file *os.File
20		raw syscall.RawConn	26		raw syscall.RawConn
			27	+	// wake is an eventfd pinned in the epoll set: Wake writes it.
			28	+	wake *os.File
			29	+	wakeRaw syscall.RawConn
			30	+	wakeFD int32
			31	+	ctx context.Context
			32	+	stop func() bool
			33	+	pinned []int32 // besides wakeFD
21		// broken is set when a registration could not be removed: it may keep	34		// broken is set when a registration could not be removed: it may keep
22		// the epoll descriptor readable forever, so every later Poll fails	35		// the epoll descriptor readable forever, so every later Poll fails
23		// instead of spinning.	36		// instead of spinning.
24		broken error	37		broken error
			38	+	// noDeadline: the read deadline is already cleared.
			39	+	noDeadline bool
			40	+
			41	+	// State of the running Poll, used by the callbacks below. They are
			42	+	// bound once in New, so a Poll allocates nothing.
			43	+	fds []unix.PollFd
			44	+	check []unix.PollFd
			45	+	added int
			46	+	ctlErr error
			47	+	n int
			48	+	pollErr error
			49	+	buf [8]byte
			50	+	addFn func(uintptr)
			51	+	delFn func(uintptr)
			52	+	checkFn func(uintptr)
			53	+	readFn func(uintptr) bool
			54	+	drainFn func(uintptr)
			55	+	kickFn func(uintptr)
25		}	56		}
26			57
27		func New() (*Poller, error) {	58		func New() (*Poller, error) {

@@ -33,108 +64,229 @@ func New() (*Poller, error) {
33		unix.Close(fd)	64		unix.Close(fd)
34		return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)	65		return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)
35		}	66		}
36	-	f := os.NewFile(uintptr(fd), "fdwait")	67	+	p := &Poller{file: os.NewFile(uintptr(fd), "fdwait")}
37	-	raw, err := f.SyscallConn()	68	+	if p.raw, err = p.file.SyscallConn(); err != nil {
38	-	if err != nil {	69	+	p.file.Close()
39	-	f.Close()
40		return nil, fmt.Errorf("fdwait epoll conn: %w", err)	70		return nil, fmt.Errorf("fdwait epoll conn: %w", err)
41		}	71		}
42	-	return &Poller{file: f, raw: raw}, nil	72	+	efd, err := unix.Eventfd(0, unix.EFD_CLOEXEC\|unix.EFD_NONBLOCK)
			73	+	if err != nil {
			74	+	p.file.Close()
			75	+	return nil, fmt.Errorf("fdwait wake eventfd: %w", err)
			76	+	}
			77	+	p.wake, p.wakeFD = os.NewFile(uintptr(efd), "fdwait-wake"), int32(efd)
			78	+	p.addFn, p.delFn, p.readFn, p.checkFn = p.watch, p.unwatchAll, p.ready, p.pollNow
			79	+	p.drainFn, p.kickFn = p.drain, p.kick
			80	+	if p.wakeRaw, err = p.wake.SyscallConn(); err == nil {
			81	+	err = p.control(unix.EPOLL_CTL_ADD, p.wakeFD)
			82	+	}
			83	+	if err != nil {
			84	+	p.Close()
			85	+	return nil, fmt.Errorf("fdwait wake: %w", err)
			86	+	}
			87	+	return p, nil
43		}	88		}
44			89
45	-	func (p *Poller) Close() error { return p.file.Close() }	90	+	// Close ends a running Poll and releases the epoll and wake descriptors.
			91	+	func (p *Poller) Close() error {
			92	+	if p.stop != nil {
			93	+	p.stop()
			94	+	}
			95	+	err := p.file.Close()
			96	+	p.wake.Close()
			97	+	return err
			98	+	}
			99	+
			100	+	func (p *Poller) control(op int, fd int32) error {
			101	+	var err error
			102	+	if cerr := p.raw.Control(func(ep uintptr) {
			103	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: fd}
			104	+	err = unix.EpollCtl(int(ep), op, int(fd), &ev)
			105	+	}); cerr != nil {
			106	+	return cerr
			107	+	}
			108	+	return err
			109	+	}
			110	+
			111	+	// Pin registers fd for every later Poll, which then costs it no epoll call.
			112	+	// fd must stay open until Close, and every Poll must list it: a ready pinned
			113	+	// fd that a Poll does not list fails that Poll.
			114	+	func (p *Poller) Pin(fd int32) error {
			115	+	if err := p.control(unix.EPOLL_CTL_ADD, fd); err != nil {
			116	+	return fmt.Errorf("fdwait pin fd %d: %w", fd, err)
			117	+	}
			118	+	p.pinned = append(p.pinned, fd)
			119	+	return nil
			120	+	}
			121	+
			122	+	// StopOn makes every Poll return (0, nil) at once after ctx ends. Call it
			123	+	// once, before polling.
			124	+	func (p *Poller) StopOn(ctx context.Context) {
			125	+	p.ctx = ctx
			126	+	p.stop = context.AfterFunc(ctx, p.Wake)
			127	+	}
			128	+
			129	+	// Wake makes the running or next Poll return. Wakes before that Poll
			130	+	// coalesce. Safe from any goroutine, also after Close.
			131	+	func (p *Poller) Wake() { _ = p.wakeRaw.Control(p.kickFn) }
			132	+
			133	+	func (p *Poller) kick(fd uintptr) {
			134	+	one := [8]byte{1}
			135	+	_, _ = unix.Write(int(fd), one[:])
			136	+	}
			137	+
			138	+	func (p *Poller) drain(fd uintptr) { _, _ = unix.Read(int(fd), p.buf[:]) }
			139	+
			140	+	func (p *Poller) stopped() bool { return p.ctx != nil && p.ctx.Err() != nil }
46			141
47		// Poll waits like poll(2) for readable/error events on fds and fills their	142		// Poll waits like poll(2) for readable/error events on fds and fills their
48		// Revents. It never consumes caller events. Negative fds are ignored and a	143		// Revents. It never consumes caller events. Negative fds are ignored and a
49		// descriptor listed twice is watched once. timeout is in milliseconds; a	144		// descriptor listed twice is watched once. timeout is in milliseconds; a
50	-	// negative one waits until an event or Close.	145	+	// negative one waits until an event, Wake, the end of the StopOn context or
			146	+	// Close. A Wake or stop returns the events seen, possibly none.
51		//	147		//
52	-	// Each call explicitly removes its epoll registrations before returning.	148	+	// Each call removes the registrations it made before returning. Relying on
53	-	// Relying on close to remove them is unsafe: another fd or a kernel fence	149	+	// close to remove them is unsafe: another fd or a kernel fence may retain
54	-	// may retain the open file description and keep the epoll fd readable.	150	+	// the open file description and keep the epoll fd readable. When nothing is
55	-	// When nothing is ready yet, a descriptor epoll cannot watch (EPERM,	151	+	// ready yet, a descriptor epoll cannot watch (EPERM, EINVAL, ELOOP for epoll
56	-	// EINVAL, ELOOP for epoll nested too deep) fails the call.	152	+	// nested too deep) fails the call.
57		func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {	153		func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
58		if p.broken != nil {	154		if p.broken != nil {
59		return 0, p.broken	155		return 0, p.broken
60		}	156		}
61	-	// Ready descriptors need no registration, and an expired deadline	157	+	p.fds = fds
62	-	// would skip the readiness check in RawConn.Read.	158	+	defer func() { p.fds = nil }()
63	-	if n, err := pollNow(fds); n != 0 \|\| err != nil \|\| timeout == 0 {	159	+	transient := false
64	-	return n, err	160	+	for i := range fds {
65	-	}	161	+	transient = transient \|\| !p.skip(i)
66	-	var ctlErr error
67	-	added := 0
68	-	err := p.raw.Control(func(ep uintptr) {
69	-	for _, f := range fds {
70	-	if f.Fd >= 0 {
71	-	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
72	-	e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev)
73	-	if e != nil && !(errors.Is(e, unix.EEXIST) && listedBefore(fds, added)) {
74	-	ctlErr = fmt.Errorf("fdwait watch fd %d: %w", f.Fd, e)
75	-	return
76	-	}
77	-	}
78	-	added++
79	-	}
80	-	})
81	-	defer p.unwatch(fds[:added])
82	-	if err != nil {
83	-	return 0, err
84		}	162		}
85	-	if ctlErr != nil {	163	+	// An expired deadline skips RawConn.Read's own check, and a ready
86	-	return 0, ctlErr	164	+	// transient fd needs no registration. Pinned fds are checked by Read.
			165	+	if (timeout == 0 \|\| transient) && p.ready(0) \|\| timeout == 0 {
			166	+	return p.n, p.pollErr
87		}	167		}
88	-	if timeout > 0 {	168	+	p.added, p.ctlErr = 0, nil
89	-	err = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))	169	+	if transient {
90	-	} else {	170	+	err := p.raw.Control(p.addFn)
91	-	err = p.file.SetReadDeadline(time.Time{})	171	+	defer p.unwatch()
			172	+	if err != nil {
			173	+	return 0, err
			174	+	}
			175	+	if p.ctlErr != nil {
			176	+	return 0, p.ctlErr
			177	+	}
			178	+	}
			179	+	var err error
			180	+	switch {
			181	+	case timeout > 0:
			182	+	err, p.noDeadline = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout)*time.Millisecond)), false
			183	+	case !p.noDeadline:
			184	+	err, p.noDeadline = p.file.SetReadDeadline(time.Time{}), true
92		}	185		}
93		if err != nil {	186		if err != nil {
94		return 0, err	187		return 0, err
95		}	188		}
96	-	var n int	189	+	err = p.raw.Read(p.readFn)
97	-	var pollErr error
98	-	err = p.raw.Read(func(_ uintptr) bool {
99	-	n, pollErr = pollNow(fds)
100	-	return n != 0 \|\| pollErr != nil
101	-	})
102		if errors.Is(err, os.ErrDeadlineExceeded) {	190		if errors.Is(err, os.ErrDeadlineExceeded) {
103		return 0, nil	191		return 0, nil
104		}	192		}
105		if err != nil {	193		if err != nil {
106		return 0, err	194		return 0, err
107		}	195		}
108	-	return n, pollErr	196	+	return p.n, p.pollErr
109		}	197		}
110			198
111	-	// unwatch removes the registrations Poll added for fds. A failure means a	199	+	// ready checks the caller's fds, the wake fd and unlisted pinned fds without
112	-	// watched fd was closed during the wait: its registration may outlive it.	200	+	// blocking, retrying an interrupted check. It reports whether Poll returns.
113	-	func (p *Poller) unwatch(fds []unix.PollFd) {	201	+	func (p *Poller) ready(uintptr) bool {
114	-	p.raw.Control(func(ep uintptr) {	202	+	p.check = append(append(p.check[:0], p.fds...), unix.PollFd{Fd: p.wakeFD, Events: unix.POLLIN})
115	-	for i, f := range fds {	203	+	for _, fd := range p.pinned {
116	-	if f.Fd < 0 \|\| listedBefore(fds, i) {	204	+	if !slices.ContainsFunc(p.fds, func(f unix.PollFd) bool { return f.Fd == fd }) {
117	-	continue	205	+	p.check = append(p.check, unix.PollFd{Fd: fd, Events: unix.POLLIN})
			206	+	}
			207	+	}
			208	+	// Under the wake file's reference, so a concurrent Close cannot free
			209	+	// its number for reuse during the check.
			210	+	if p.wakeRaw.Control(p.checkFn) != nil {
			211	+	p.n, p.pollErr = 0, os.ErrClosed
			212	+	}
			213	+	if p.pollErr != nil {
			214	+	p.n = 0
			215	+	return true
			216	+	}
			217	+	for i := range p.fds {
			218	+	p.fds[i].Revents = p.check[i].Revents
			219	+	}
			220	+	for _, f := range p.check[len(p.fds)+1:] {
			221	+	if f.Revents != 0 {
			222	+	// Its readiness keeps the epoll fd readable: waiting would spin.
			223	+	p.n, p.pollErr = 0, fmt.Errorf("fdwait: pinned fd %d is ready but not polled", f.Fd)
			224	+	return true
			225	+	}
			226	+	}
			227	+	woke := p.check[len(p.fds)].Revents != 0
			228	+	if woke {
			229	+	p.n--
			230	+	// A stopped poller stays awake: every later Poll returns at once.
			231	+	if !p.stopped() {
			232	+	_ = p.wakeRaw.Control(p.drainFn)
			233	+	if p.stopped() { // the stop's write may have been drained
			234	+	p.Wake()
118		}	235		}
119	-	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil); e != nil && p.broken == nil {	236	+	}
120	-	p.broken = fmt.Errorf("fdwait unwatch fd %d (closed during a wait?): %w", f.Fd, e)	237	+	}
			238	+	return p.n != 0 \|\| woke
			239	+	}
			240	+
			241	+	func (p *Poller) pollNow(uintptr) {
			242	+	for {
			243	+	p.n, p.pollErr = unix.Poll(p.check, 0)
			244	+	if !errors.Is(p.pollErr, unix.EINTR) {
			245	+	return
			246	+	}
			247	+	}
			248	+	}
			249	+
			250	+	// skip reports whether fds[i] needs no registration of its own.
			251	+	func (p *Poller) skip(i int) bool {
			252	+	fd := p.fds[i].Fd
			253	+	return fd < 0 \|\| slices.Contains(p.pinned, fd) \|\| listedBefore(p.fds, i)
			254	+	}
			255	+
			256	+	func (p *Poller) watch(ep uintptr) {
			257	+	for i, f := range p.fds {
			258	+	if !p.skip(i) {
			259	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
			260	+	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev); e != nil {
			261	+	p.ctlErr = fmt.Errorf("fdwait watch fd %d: %w", f.Fd, e)
			262	+	return
121		}	263		}
122		}	264		}
123	-	})	265	+	p.added++
			266	+	}
124		}	267		}
125			268
126	-	// listedBefore reports whether fds[i].Fd also appears earlier in fds.	269	+	// unwatch removes the registrations Poll added. A failure means a watched
127	-	func listedBefore(fds []unix.PollFd, i int) bool {	270	+	// fd was closed during the wait: its registration may outlive it.
128	-	fd := fds[i].Fd	271	+	func (p *Poller) unwatch() {
129	-	return slices.ContainsFunc(fds[:i], func(f unix.PollFd) bool { return f.Fd == fd })	272	+	if p.added > 0 {
			273	+	_ = p.raw.Control(p.delFn)
			274	+	}
130		}	275		}
131			276
132	-	// pollNow checks fds without blocking, retrying an interrupted check.	277	+	func (p *Poller) unwatchAll(ep uintptr) {
133	-	func pollNow(fds []unix.PollFd) (int, error) {	278	+	for i, f := range p.fds[:p.added] {
134	-	for {	279	+	if p.skip(i) {
135	-	n, err := unix.Poll(fds, 0)	280	+	continue
136	-	if !errors.Is(err, unix.EINTR) {	281	+	}
137	-	return n, err	282	+	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil); e != nil && p.broken == nil {
			283	+	p.broken = fmt.Errorf("fdwait unwatch fd %d (closed during a wait?): %w", f.Fd, e)
138		}	284		}
139		}	285		}
140		}	286		}
			287	+
			288	+	// listedBefore reports whether fds[i].Fd also appears earlier in fds.
			289	+	func listedBefore(fds []unix.PollFd, i int) bool {
			290	+	fd := fds[i].Fd
			291	+	return slices.ContainsFunc(fds[:i], func(f unix.PollFd) bool { return f.Fd == fd })
			292	+	}