detached 2106223

Merge pull request 'perf(adapters): park descriptor waits in the Go netpoller' (#46) from perf/netpoll-waits into main

commit
2106223884ac1c6280f3efceef20b0cbbfab28f0
author
bnema · · 2026-10-14 04:38 UTC
merge of
0d28346 + 5f1215d · diff against the first parent
signature
SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown
19 files · +983 -67
Unified
UnifiedSplit
M Makefile +1 -1
@@ -61,7 +61,7 @@ arch: adapter-imports
6161 # Adapters stay independent: production code may import only the shared
6262 # adapter libraries below. outputkit holds what both output backends (drm,
6363 # 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
6565 adapter-imports:
6666 @edges=$$(CGO_ENABLED=0 go list -f '{{$$p := .ImportPath}}{{range .Imports}}{{$$p}} {{.}}{{"\n"}}{{end}}' ./internal/adapters/...) || exit 1; \
6767 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 shared61 # 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|xkb64+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 go.mod +1 -1
@@ -6,7 +6,7 @@ require (
66 github.com/bnema/go-wayland-bindings v0.1.0
77 github.com/bnema/kvconf v0.1.2
88 github.com/bnema/purego v0.13.0-bnema.1
9- github.com/bnema/purego-libwayland v0.10.0
9+ github.com/bnema/purego-libwayland v0.11.0
1010 github.com/bnema/purego-vulkan v0.6.0
1111 github.com/bnema/wlturbo v0.5.0
1212 github.com/bnema/zerowrap v1.4.1
@@ -6,7 +6,7 @@ require (
6 github.com/bnema/go-wayland-bindings v0.1.06 github.com/bnema/go-wayland-bindings v0.1.0
7 github.com/bnema/kvconf v0.1.27 github.com/bnema/kvconf v0.1.2
8 github.com/bnema/purego v0.13.0-bnema.18 github.com/bnema/purego v0.13.0-bnema.1
9- github.com/bnema/purego-libwayland v0.10.09+ github.com/bnema/purego-libwayland v0.11.0
10 github.com/bnema/purego-vulkan v0.6.010 github.com/bnema/purego-vulkan v0.6.0
11 github.com/bnema/wlturbo v0.5.011 github.com/bnema/wlturbo v0.5.0
12 github.com/bnema/zerowrap v1.4.112 github.com/bnema/zerowrap v1.4.1
M go.sum +2 -2
@@ -4,8 +4,8 @@ github.com/bnema/kvconf v0.1.2 h1:1hDKL9jkQSBdR9dSKbwTONhUUBfJgAdh5vu+yFHOb70=
44 github.com/bnema/kvconf v0.1.2/go.mod h1:+P2X9I/sLx+Lfir45s9a7WA0GnC22WgKbZWE9LK6lcU=
55 github.com/bnema/purego v0.13.0-bnema.1 h1:fKRdztnvTmRqAjANMoN+utEIH73dvPqVPJh2iZCCAO8=
66 github.com/bnema/purego v0.13.0-bnema.1/go.mod h1:xS1iX3noypxnbOUEUAqOvkfnUzgHuu19VtL4+dfUMJM=
7-github.com/bnema/purego-libwayland v0.10.0 h1:N/NTfj1BGdYiabCJ8W0tbBwyYt+2FNtrweoAODe521c=
8-github.com/bnema/purego-libwayland v0.10.0/go.mod h1:ZGc8Hz68KwJEHztCBU68o0pRFmDZgGFe3dJ+U1MHgwA=
7+github.com/bnema/purego-libwayland v0.11.0 h1:9DHAgg1M9+blG5rAC9EWlJt2GFRnA/pxuXE0S0S2Cl8=
8+github.com/bnema/purego-libwayland v0.11.0/go.mod h1:ZGc8Hz68KwJEHztCBU68o0pRFmDZgGFe3dJ+U1MHgwA=
99 github.com/bnema/purego-vulkan v0.6.0 h1:SJgu0smuDC2vtBv0dJ6dmRKewJW5EUXQwTXyb/p9JSE=
1010 github.com/bnema/purego-vulkan v0.6.0/go.mod h1:7kFoR1p01JbknuZlfA20oeImryBe8Fx+Zd9hm2uSUM4=
1111 github.com/bnema/wlturbo v0.5.0 h1:AUP3IAd4ZXM3ZrJ8zu65lAejKtYMwSoSDE2TS6k3mn4=
@@ -4,8 +4,8 @@ github.com/bnema/kvconf v0.1.2 h1:1hDKL9jkQSBdR9dSKbwTONhUUBfJgAdh5vu+yFHOb70=
4 github.com/bnema/kvconf v0.1.2/go.mod h1:+P2X9I/sLx+Lfir45s9a7WA0GnC22WgKbZWE9LK6lcU=4 github.com/bnema/kvconf v0.1.2/go.mod h1:+P2X9I/sLx+Lfir45s9a7WA0GnC22WgKbZWE9LK6lcU=
5 github.com/bnema/purego v0.13.0-bnema.1 h1:fKRdztnvTmRqAjANMoN+utEIH73dvPqVPJh2iZCCAO8=5 github.com/bnema/purego v0.13.0-bnema.1 h1:fKRdztnvTmRqAjANMoN+utEIH73dvPqVPJh2iZCCAO8=
6 github.com/bnema/purego v0.13.0-bnema.1/go.mod h1:xS1iX3noypxnbOUEUAqOvkfnUzgHuu19VtL4+dfUMJM=6 github.com/bnema/purego v0.13.0-bnema.1/go.mod h1:xS1iX3noypxnbOUEUAqOvkfnUzgHuu19VtL4+dfUMJM=
7-github.com/bnema/purego-libwayland v0.10.0 h1:N/NTfj1BGdYiabCJ8W0tbBwyYt+2FNtrweoAODe521c=7+github.com/bnema/purego-libwayland v0.11.0 h1:9DHAgg1M9+blG5rAC9EWlJt2GFRnA/pxuXE0S0S2Cl8=
8-github.com/bnema/purego-libwayland v0.10.0/go.mod h1:ZGc8Hz68KwJEHztCBU68o0pRFmDZgGFe3dJ+U1MHgwA=8+github.com/bnema/purego-libwayland v0.11.0/go.mod h1:ZGc8Hz68KwJEHztCBU68o0pRFmDZgGFe3dJ+U1MHgwA=
9 github.com/bnema/purego-vulkan v0.6.0 h1:SJgu0smuDC2vtBv0dJ6dmRKewJW5EUXQwTXyb/p9JSE=9 github.com/bnema/purego-vulkan v0.6.0 h1:SJgu0smuDC2vtBv0dJ6dmRKewJW5EUXQwTXyb/p9JSE=
10 github.com/bnema/purego-vulkan v0.6.0/go.mod h1:7kFoR1p01JbknuZlfA20oeImryBe8Fx+Zd9hm2uSUM4=10 github.com/bnema/purego-vulkan v0.6.0/go.mod h1:7kFoR1p01JbknuZlfA20oeImryBe8Fx+Zd9hm2uSUM4=
11 github.com/bnema/wlturbo v0.5.0 h1:AUP3IAd4ZXM3ZrJ8zu65lAejKtYMwSoSDE2TS6k3mn4=11 github.com/bnema/wlturbo v0.5.0 h1:AUP3IAd4ZXM3ZrJ8zu65lAejKtYMwSoSDE2TS6k3mn4=
M internal/adapters/captureallow/store.go +19 -5
@@ -13,6 +13,7 @@ import (
1313 "time"
1414 "unsafe"
1515
16+ "github.com/bnema/neferwl/internal/adapters/fdwait"
1617 "github.com/bnema/zerowrap"
1718 "golang.org/x/sys/unix"
1819 )
@@ -186,6 +187,16 @@ func (s *Store) Run(ctx context.Context) error {
186187 return err
187188 }
188189 defer unix.Close(fd)
190+ waiter, err := fdwait.New()
191+ if err != nil {
192+ return err
193+ }
194+ defer waiter.Close()
195+ waiter.StopOn(ctx)
196+ if err := waiter.Pin(int32(fd)); err != nil {
197+ return err
198+ }
199+ fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
189200 dir, name := filepath.Dir(s.path), filepath.Base(s.path)
190201 wd := -1
191202 var retry, pending time.Time
@@ -209,9 +220,10 @@ func (s *Store) Run(ctx context.Context) error {
209220 return err
210221 }
211222 }
212- timeout := 200
223+ // Sleep until an event, the shutdown, or the next deadline.
224+ timeout := -1
213225 if !retry.IsZero() {
214- timeout = min(timeout, int(time.Until(retry).Milliseconds())+1)
226+ timeout = max(0, int(time.Until(retry).Milliseconds())+1)
215227 }
216228 if !pending.IsZero() {
217229 remaining := time.Until(pending)
@@ -221,10 +233,12 @@ func (s *Store) Run(ctx context.Context) error {
221233 quiet = false
222234 continue
223235 }
224- timeout = min(timeout, int(remaining.Milliseconds())+1)
236+ if ms := int(remaining.Milliseconds()) + 1; timeout < 0 || ms < timeout {
237+ timeout = ms
238+ }
225239 }
226- fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
227- if _, err := unix.Poll(fds, timeout); err != nil {
240+ fds[0].Revents = 0
241+ if _, err := waiter.Poll(fds, timeout); err != nil {
228242 if err == unix.EINTR {
229243 continue
230244 }
@@ -13,6 +13,7 @@ import (
13 "time"13 "time"
14 "unsafe"14 "unsafe"
15 15
16+ "github.com/bnema/neferwl/internal/adapters/fdwait"
16 "github.com/bnema/zerowrap"17 "github.com/bnema/zerowrap"
17 "golang.org/x/sys/unix"18 "golang.org/x/sys/unix"
18 )19 )
@@ -186,6 +187,16 @@ func (s *Store) Run(ctx context.Context) error {
186 return err187 return err
187 }188 }
188 defer unix.Close(fd)189 defer unix.Close(fd)
190+ waiter, err := fdwait.New()
191+ if err != nil {
192+ return err
193+ }
194+ defer waiter.Close()
195+ waiter.StopOn(ctx)
196+ if err := waiter.Pin(int32(fd)); err != nil {
197+ return err
198+ }
199+ fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
189 dir, name := filepath.Dir(s.path), filepath.Base(s.path)200 dir, name := filepath.Dir(s.path), filepath.Base(s.path)
190 wd := -1201 wd := -1
191 var retry, pending time.Time202 var retry, pending time.Time
@@ -209,9 +220,10 @@ func (s *Store) Run(ctx context.Context) error {
209 return err220 return err
210 }221 }
211 }222 }
212- timeout := 200223+ // Sleep until an event, the shutdown, or the next deadline.
224+ timeout := -1
213 if !retry.IsZero() {225 if !retry.IsZero() {
214- timeout = min(timeout, int(time.Until(retry).Milliseconds())+1)226+ timeout = max(0, int(time.Until(retry).Milliseconds())+1)
215 }227 }
216 if !pending.IsZero() {228 if !pending.IsZero() {
217 remaining := time.Until(pending)229 remaining := time.Until(pending)
@@ -221,10 +233,12 @@ func (s *Store) Run(ctx context.Context) error {
221 quiet = false233 quiet = false
222 continue234 continue
223 }235 }
224- timeout = min(timeout, int(remaining.Milliseconds())+1)236+ if ms := int(remaining.Milliseconds()) + 1; timeout < 0 || ms < timeout {
237+ timeout = ms
238+ }
225 }239 }
226- fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}240+ fds[0].Revents = 0
227- if _, err := unix.Poll(fds, timeout); err != nil {241+ if _, err := waiter.Poll(fds, timeout); err != nil {
228 if err == unix.EINTR {242 if err == unix.EINTR {
229 continue243 continue
230 }244 }
M internal/adapters/drm/card.go +12 -2
@@ -8,6 +8,7 @@ import (
88 "sync"
99 "sync/atomic"
1010
11+ "github.com/bnema/neferwl/internal/adapters/fdwait"
1112 "github.com/bnema/neferwl/internal/ports"
1213 "github.com/bnema/zerowrap"
1314 "golang.org/x/sys/unix"
@@ -247,11 +248,20 @@ func (c *Card) open(conn connector, mode modeInfo) (*Output, error) {
247248 // output of their CRTC. Run it once per card. An event is never dropped:
248249 // the output owes it a pending commit.
249250 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()
256+ waiter.StopOn(ctx)
257+ if err := waiter.Pin(int32(c.fd)); err != nil {
258+ return fmt.Errorf("drm event waiter: %w", err)
259+ }
250260 buf := make([]byte, 1024)
251261 var events []flipEvent // reused by every read
262+ fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
252263 for ctx.Err() == nil {
253- fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
254- n, err := unix.Poll(fds, 100)
264+ n, err := waiter.Poll(fds, -1)
255265 if err != nil && !errors.Is(err, unix.EINTR) {
256266 return fmt.Errorf("drm poll: %w", err)
257267 }
@@ -8,6 +8,7 @@ import (
8 "sync"8 "sync"
9 "sync/atomic"9 "sync/atomic"
10 10
11+ "github.com/bnema/neferwl/internal/adapters/fdwait"
11 "github.com/bnema/neferwl/internal/ports"12 "github.com/bnema/neferwl/internal/ports"
12 "github.com/bnema/zerowrap"13 "github.com/bnema/zerowrap"
13 "golang.org/x/sys/unix"14 "golang.org/x/sys/unix"
@@ -247,11 +248,20 @@ func (c *Card) open(conn connector, mode modeInfo) (*Output, error) {
247 // output of their CRTC. Run it once per card. An event is never dropped:248 // output of their CRTC. Run it once per card. An event is never dropped:
248 // the output owes it a pending commit.249 // the output owes it a pending commit.
249 func (c *Card) ReadEvents(ctx context.Context) error {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()
256+ waiter.StopOn(ctx)
257+ if err := waiter.Pin(int32(c.fd)); err != nil {
258+ return fmt.Errorf("drm event waiter: %w", err)
259+ }
250 buf := make([]byte, 1024)260 buf := make([]byte, 1024)
251 var events []flipEvent // reused by every read261 var events []flipEvent // reused by every read
262+ fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
252 for ctx.Err() == nil {263 for ctx.Err() == nil {
253- fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}264+ n, err := waiter.Poll(fds, -1)
254- n, err := unix.Poll(fds, 100)
255 if err != nil && !errors.Is(err, unix.EINTR) {265 if err != nil && !errors.Is(err, unix.EINTR) {
256 return fmt.Errorf("drm poll: %w", err)266 return fmt.Errorf("drm poll: %w", err)
257 }267 }
M internal/adapters/drm/hotplug.go +12 -2
@@ -6,6 +6,7 @@ import (
66 "fmt"
77 "sync"
88
9+ "github.com/bnema/neferwl/internal/adapters/fdwait"
910 "github.com/bnema/purego"
1011 "golang.org/x/sys/unix"
1112 )
@@ -77,9 +78,18 @@ func WatchHotplug(ctx context.Context, changed chan<- struct{}) error {
7778 return errors.New("udev monitor setup failed")
7879 }
7980 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()
86+ waiter.StopOn(ctx)
87+ if err := waiter.Pin(fd); err != nil {
88+ return fmt.Errorf("udev waiter: %w", err)
89+ }
90+ fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
8091 for ctx.Err() == nil {
81- fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
82- n, err := unix.Poll(fds, 100)
92+ n, err := waiter.Poll(fds, -1)
8393 if err != nil && !errors.Is(err, unix.EINTR) {
8494 return fmt.Errorf("udev poll: %w", err)
8595 }
@@ -6,6 +6,7 @@ import (
6 "fmt"6 "fmt"
7 "sync"7 "sync"
8 8
9+ "github.com/bnema/neferwl/internal/adapters/fdwait"
9 "github.com/bnema/purego"10 "github.com/bnema/purego"
10 "golang.org/x/sys/unix"11 "golang.org/x/sys/unix"
11 )12 )
@@ -77,9 +78,18 @@ func WatchHotplug(ctx context.Context, changed chan<- struct{}) error {
77 return errors.New("udev monitor setup failed")78 return errors.New("udev monitor setup failed")
78 }79 }
79 fd := monitorFD(mon)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()
86+ waiter.StopOn(ctx)
87+ if err := waiter.Pin(fd); err != nil {
88+ return fmt.Errorf("udev waiter: %w", err)
89+ }
90+ fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
80 for ctx.Err() == nil {91 for ctx.Err() == nil {
81- fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}92+ n, err := waiter.Poll(fds, -1)
82- n, err := unix.Poll(fds, 100)
83 if err != nil && !errors.Is(err, unix.EINTR) {93 if err != nil && !errors.Is(err, unix.EINTR) {
84 return fmt.Errorf("udev poll: %w", err)94 return fmt.Errorf("udev poll: %w", err)
85 }95 }
A internal/adapters/fdwait/duplicate_test.go +60 -0
@@ -0,0 +1,60 @@
1+package fdwait
2+
3+import (
4+ "golang.org/x/sys/unix"
5+ "testing"
6+ "time"
7+)
8+
9+// Epoll retains a registration after its fd closes if another descriptor
10+// still refers to the same open file description.
11+func TestRemovedDuplicatedDescriptorDoesNotKeepPollReadable(t *testing.T) {
12+ p, err := New()
13+ if err != nil {
14+ t.Fatal(err)
15+ }
16+ defer p.Close()
17+ old, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
18+ if err != nil {
19+ t.Fatal(err)
20+ }
21+ dup, err := unix.Dup(old)
22+ if err != nil {
23+ unix.Close(old)
24+ t.Fatal(err)
25+ }
26+ defer unix.Close(dup)
27+ if _, err := p.Poll([]unix.PollFd{{Fd: int32(old), Events: unix.POLLIN}}, 1); err != nil {
28+ t.Fatal(err)
29+ }
30+ // Create the next descriptor before closing old so its number is not
31+ // reused. This leaves an old epoll registration with no usable fd.
32+ fresh, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
33+ if err != nil {
34+ t.Fatal(err)
35+ }
36+ defer unix.Close(fresh)
37+ unix.Close(old)
38+ if _, err := unix.Write(dup, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
39+ t.Fatal(err)
40+ }
41+ if err := p.raw.Control(func(ep uintptr) {
42+ var events [1]unix.EpollEvent
43+ if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil {
44+ t.Errorf("closed watched descriptor still makes epoll readable: %d %v", n, err)
45+ }
46+ }); err != nil {
47+ t.Fatal(err)
48+ }
49+ done := make(chan error, 1)
50+ go func() { _, err := p.Poll([]unix.PollFd{{Fd: int32(fresh), Events: unix.POLLIN}}, 20); done <- err }()
51+ select {
52+ case err := <-done:
53+ if err != nil {
54+ t.Fatal(err)
55+ }
56+ case <-time.After(time.Second):
57+ p.Close()
58+ t.Fatal("stale epoll readiness prevents timeout")
59+ }
60+}
@@ -0,0 +1,60 @@
1+package fdwait
2+
3+import (
4+ "golang.org/x/sys/unix"
5+ "testing"
6+ "time"
7+)
8+
9+// Epoll retains a registration after its fd closes if another descriptor
10+// still refers to the same open file description.
11+func TestRemovedDuplicatedDescriptorDoesNotKeepPollReadable(t *testing.T) {
12+ p, err := New()
13+ if err != nil {
14+ t.Fatal(err)
15+ }
16+ defer p.Close()
17+ old, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
18+ if err != nil {
19+ t.Fatal(err)
20+ }
21+ dup, err := unix.Dup(old)
22+ if err != nil {
23+ unix.Close(old)
24+ t.Fatal(err)
25+ }
26+ defer unix.Close(dup)
27+ if _, err := p.Poll([]unix.PollFd{{Fd: int32(old), Events: unix.POLLIN}}, 1); err != nil {
28+ t.Fatal(err)
29+ }
30+ // Create the next descriptor before closing old so its number is not
31+ // reused. This leaves an old epoll registration with no usable fd.
32+ fresh, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
33+ if err != nil {
34+ t.Fatal(err)
35+ }
36+ defer unix.Close(fresh)
37+ unix.Close(old)
38+ if _, err := unix.Write(dup, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
39+ t.Fatal(err)
40+ }
41+ if err := p.raw.Control(func(ep uintptr) {
42+ var events [1]unix.EpollEvent
43+ if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil {
44+ t.Errorf("closed watched descriptor still makes epoll readable: %d %v", n, err)
45+ }
46+ }); err != nil {
47+ t.Fatal(err)
48+ }
49+ done := make(chan error, 1)
50+ go func() { _, err := p.Poll([]unix.PollFd{{Fd: int32(fresh), Events: unix.POLLIN}}, 20); done <- err }()
51+ select {
52+ case err := <-done:
53+ if err != nil {
54+ t.Fatal(err)
55+ }
56+ case <-time.After(time.Second):
57+ p.Close()
58+ t.Fatal("stale epoll readiness prevents timeout")
59+ }
60+}
A internal/adapters/fdwait/edges_test.go +253 -0
@@ -0,0 +1,253 @@
1+package fdwait
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+ "time"
8+
9+ "golang.org/x/sys/unix"
10+)
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+
131+func eventfd(t *testing.T) int {
132+ t.Helper()
133+ fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
134+ if err != nil {
135+ t.Fatal(err)
136+ }
137+ t.Cleanup(func() { unix.Close(fd) })
138+ return fd
139+}
140+
141+func signal(t *testing.T, fd int) {
142+ t.Helper()
143+ if _, err := unix.Write(fd, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
144+ t.Fatal(err)
145+ }
146+}
147+
148+// assertNoWatches fails if Poll left a registration behind, by making every
149+// test descriptor readable and checking epoll reports nothing.
150+func assertNoWatches(t *testing.T, p *Poller, fds ...int) {
151+ t.Helper()
152+ for _, fd := range fds {
153+ signal(t, fd)
154+ }
155+ if err := p.raw.Control(func(ep uintptr) {
156+ var events [4]unix.EpollEvent
157+ if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil {
158+ t.Errorf("registrations left after Poll returned: %d %v", n, err)
159+ }
160+ }); err != nil {
161+ t.Fatal(err)
162+ }
163+}
164+
165+// Poll(fds, 0) reports existing readiness like poll(2): an expired netpoll
166+// deadline never runs the readiness check.
167+func TestPollZeroTimeoutReportsReady(t *testing.T) {
168+ p, err := New()
169+ if err != nil {
170+ t.Fatal(err)
171+ }
172+ defer p.Close()
173+ fd := eventfd(t)
174+ fds := []unix.PollFd{{Fd: -1}, {Fd: int32(fd), Events: unix.POLLIN}}
175+ if n, err := p.Poll(fds, 0); n != 0 || err != nil {
176+ t.Fatalf("idle: %d %v", n, err)
177+ }
178+ signal(t, fd)
179+ if n, err := p.Poll(fds, 0); n != 1 || err != nil || fds[1].Revents&unix.POLLIN == 0 {
180+ t.Fatalf("ready: %d %v %+v", n, err, fds)
181+ }
182+}
183+
184+// A descriptor listed twice is watched once and removed once.
185+func TestPollDuplicateDescriptor(t *testing.T) {
186+ p, err := New()
187+ if err != nil {
188+ t.Fatal(err)
189+ }
190+ defer p.Close()
191+ fd := eventfd(t)
192+ fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: int32(fd), Events: unix.POLLIN}}
193+ if n, err := p.Poll(fds, 1); n != 0 || err != nil {
194+ t.Fatalf("duplicate: %d %v", n, err)
195+ }
196+ assertNoWatches(t, p, fd)
197+}
198+
199+// A failed registration removes the ones already added in the same call.
200+func TestPollFailedWatchRemovesAdded(t *testing.T) {
201+ p, err := New()
202+ if err != nil {
203+ t.Fatal(err)
204+ }
205+ defer p.Close()
206+ fd := eventfd(t)
207+ var self int32
208+ if err := p.raw.Control(func(ep uintptr) { self = int32(ep) }); err != nil {
209+ t.Fatal(err)
210+ }
211+ // epoll refuses to watch itself (EINVAL) and is not ready.
212+ fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: self, Events: unix.POLLIN}}
213+ if _, err := p.Poll(fds, 1); err == nil {
214+ t.Fatal("watching the epoll fd itself succeeded")
215+ }
216+ assertNoWatches(t, p, fd)
217+}
218+
219+// A registration whose fd was closed while another reference keeps its file
220+// open cannot be removed: Poll fails from then on instead of spinning.
221+func TestPollFailsAfterLostRegistration(t *testing.T) {
222+ p, err := New()
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ defer p.Close()
227+ fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
228+ if err != nil {
229+ t.Fatal(err)
230+ }
231+ dup, err := unix.Dup(fd)
232+ if err != nil {
233+ unix.Close(fd)
234+ t.Fatal(err)
235+ }
236+ defer unix.Close(dup)
237+ if err := p.raw.Control(func(ep uintptr) {
238+ ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)}
239+ if err := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev); err != nil {
240+ t.Fatal(err)
241+ }
242+ }); err != nil {
243+ t.Fatal(err)
244+ }
245+ unix.Close(fd) // what a caller closing fd during a wait does
246+ p.fds, p.added = []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, 1
247+ p.unwatch()
248+ p.fds = nil
249+ other := eventfd(t)
250+ if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {
251+ t.Fatal("Poll succeeded with a registration it could not remove")
252+ }
253+}
@@ -0,0 +1,253 @@
1+package fdwait
2+
3+import (
4+ "context"
5+ "errors"
6+ "testing"
7+ "time"
8+
9+ "golang.org/x/sys/unix"
10+)
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+
131+func eventfd(t *testing.T) int {
132+ t.Helper()
133+ fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
134+ if err != nil {
135+ t.Fatal(err)
136+ }
137+ t.Cleanup(func() { unix.Close(fd) })
138+ return fd
139+}
140+
141+func signal(t *testing.T, fd int) {
142+ t.Helper()
143+ if _, err := unix.Write(fd, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
144+ t.Fatal(err)
145+ }
146+}
147+
148+// assertNoWatches fails if Poll left a registration behind, by making every
149+// test descriptor readable and checking epoll reports nothing.
150+func assertNoWatches(t *testing.T, p *Poller, fds ...int) {
151+ t.Helper()
152+ for _, fd := range fds {
153+ signal(t, fd)
154+ }
155+ if err := p.raw.Control(func(ep uintptr) {
156+ var events [4]unix.EpollEvent
157+ if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil {
158+ t.Errorf("registrations left after Poll returned: %d %v", n, err)
159+ }
160+ }); err != nil {
161+ t.Fatal(err)
162+ }
163+}
164+
165+// Poll(fds, 0) reports existing readiness like poll(2): an expired netpoll
166+// deadline never runs the readiness check.
167+func TestPollZeroTimeoutReportsReady(t *testing.T) {
168+ p, err := New()
169+ if err != nil {
170+ t.Fatal(err)
171+ }
172+ defer p.Close()
173+ fd := eventfd(t)
174+ fds := []unix.PollFd{{Fd: -1}, {Fd: int32(fd), Events: unix.POLLIN}}
175+ if n, err := p.Poll(fds, 0); n != 0 || err != nil {
176+ t.Fatalf("idle: %d %v", n, err)
177+ }
178+ signal(t, fd)
179+ if n, err := p.Poll(fds, 0); n != 1 || err != nil || fds[1].Revents&unix.POLLIN == 0 {
180+ t.Fatalf("ready: %d %v %+v", n, err, fds)
181+ }
182+}
183+
184+// A descriptor listed twice is watched once and removed once.
185+func TestPollDuplicateDescriptor(t *testing.T) {
186+ p, err := New()
187+ if err != nil {
188+ t.Fatal(err)
189+ }
190+ defer p.Close()
191+ fd := eventfd(t)
192+ fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: int32(fd), Events: unix.POLLIN}}
193+ if n, err := p.Poll(fds, 1); n != 0 || err != nil {
194+ t.Fatalf("duplicate: %d %v", n, err)
195+ }
196+ assertNoWatches(t, p, fd)
197+}
198+
199+// A failed registration removes the ones already added in the same call.
200+func TestPollFailedWatchRemovesAdded(t *testing.T) {
201+ p, err := New()
202+ if err != nil {
203+ t.Fatal(err)
204+ }
205+ defer p.Close()
206+ fd := eventfd(t)
207+ var self int32
208+ if err := p.raw.Control(func(ep uintptr) { self = int32(ep) }); err != nil {
209+ t.Fatal(err)
210+ }
211+ // epoll refuses to watch itself (EINVAL) and is not ready.
212+ fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: self, Events: unix.POLLIN}}
213+ if _, err := p.Poll(fds, 1); err == nil {
214+ t.Fatal("watching the epoll fd itself succeeded")
215+ }
216+ assertNoWatches(t, p, fd)
217+}
218+
219+// A registration whose fd was closed while another reference keeps its file
220+// open cannot be removed: Poll fails from then on instead of spinning.
221+func TestPollFailsAfterLostRegistration(t *testing.T) {
222+ p, err := New()
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ defer p.Close()
227+ fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
228+ if err != nil {
229+ t.Fatal(err)
230+ }
231+ dup, err := unix.Dup(fd)
232+ if err != nil {
233+ unix.Close(fd)
234+ t.Fatal(err)
235+ }
236+ defer unix.Close(dup)
237+ if err := p.raw.Control(func(ep uintptr) {
238+ ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)}
239+ if err := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev); err != nil {
240+ t.Fatal(err)
241+ }
242+ }); err != nil {
243+ t.Fatal(err)
244+ }
245+ unix.Close(fd) // what a caller closing fd during a wait does
246+ p.fds, p.added = []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, 1
247+ p.unwatch()
248+ p.fds = nil
249+ other := eventfd(t)
250+ if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {
251+ t.Fatal("Poll succeeded with a registration it could not remove")
252+ }
253+}
A internal/adapters/fdwait/poller.go +292 -0
@@ -0,0 +1,292 @@
1+// Package fdwait parks descriptor waits in Go's netpoller.
2+package fdwait
3+
4+import (
5+ "context"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "slices"
10+ "syscall"
11+ "time"
12+
13+ "golang.org/x/sys/unix"
14+)
15+
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.
24+type Poller struct {
25+ file *os.File
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
34+ // broken is set when a registration could not be removed: it may keep
35+ // the epoll descriptor readable forever, so every later Poll fails
36+ // instead of spinning.
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)
56+}
57+
58+func New() (*Poller, error) {
59+ fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
60+ if err != nil {
61+ return nil, fmt.Errorf("fdwait epoll create: %w", err)
62+ }
63+ if err = unix.SetNonblock(fd, true); err != nil {
64+ unix.Close(fd)
65+ return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)
66+ }
67+ p := &Poller{file: os.NewFile(uintptr(fd), "fdwait")}
68+ if p.raw, err = p.file.SyscallConn(); err != nil {
69+ p.file.Close()
70+ return nil, fmt.Errorf("fdwait epoll conn: %w", err)
71+ }
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
88+}
89+
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 }
141+
142+// Poll waits like poll(2) for readable/error events on fds and fills their
143+// Revents. It never consumes caller events. Negative fds are ignored and a
144+// descriptor listed twice is watched once. timeout is in milliseconds; a
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.
147+//
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.
153+func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
154+ if p.broken != nil {
155+ return 0, p.broken
156+ }
157+ p.fds = fds
158+ defer func() { p.fds = nil }()
159+ transient := false
160+ for i := range fds {
161+ transient = transient || !p.skip(i)
162+ }
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
167+ }
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
185+ }
186+ if err != nil {
187+ return 0, err
188+ }
189+ err = p.raw.Read(p.readFn)
190+ if errors.Is(err, os.ErrDeadlineExceeded) {
191+ return 0, nil
192+ }
193+ if err != nil {
194+ return 0, err
195+ }
196+ return p.n, p.pollErr
197+}
198+
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()
235+ }
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
263+ }
264+ }
265+ p.added++
266+ }
267+}
268+
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+ }
275+}
276+
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)
284+ }
285+ }
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+}
@@ -0,0 +1,292 @@
1+// Package fdwait parks descriptor waits in Go's netpoller.
2+package fdwait
3+
4+import (
5+ "context"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "slices"
10+ "syscall"
11+ "time"
12+
13+ "golang.org/x/sys/unix"
14+)
15+
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.
24+type Poller struct {
25+ file *os.File
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
34+ // broken is set when a registration could not be removed: it may keep
35+ // the epoll descriptor readable forever, so every later Poll fails
36+ // instead of spinning.
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)
56+}
57+
58+func New() (*Poller, error) {
59+ fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
60+ if err != nil {
61+ return nil, fmt.Errorf("fdwait epoll create: %w", err)
62+ }
63+ if err = unix.SetNonblock(fd, true); err != nil {
64+ unix.Close(fd)
65+ return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)
66+ }
67+ p := &Poller{file: os.NewFile(uintptr(fd), "fdwait")}
68+ if p.raw, err = p.file.SyscallConn(); err != nil {
69+ p.file.Close()
70+ return nil, fmt.Errorf("fdwait epoll conn: %w", err)
71+ }
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
88+}
89+
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 }
141+
142+// Poll waits like poll(2) for readable/error events on fds and fills their
143+// Revents. It never consumes caller events. Negative fds are ignored and a
144+// descriptor listed twice is watched once. timeout is in milliseconds; a
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.
147+//
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.
153+func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
154+ if p.broken != nil {
155+ return 0, p.broken
156+ }
157+ p.fds = fds
158+ defer func() { p.fds = nil }()
159+ transient := false
160+ for i := range fds {
161+ transient = transient || !p.skip(i)
162+ }
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
167+ }
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
185+ }
186+ if err != nil {
187+ return 0, err
188+ }
189+ err = p.raw.Read(p.readFn)
190+ if errors.Is(err, os.ErrDeadlineExceeded) {
191+ return 0, nil
192+ }
193+ if err != nil {
194+ return 0, err
195+ }
196+ return p.n, p.pollErr
197+}
198+
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()
235+ }
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
263+ }
264+ }
265+ p.added++
266+ }
267+}
268+
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+ }
275+}
276+
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)
284+ }
285+ }
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+}
A internal/adapters/fdwait/poller_test.go +90 -0
@@ -0,0 +1,90 @@
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 err := p.raw.Control(func(ep uintptr) {
56+ var events [1]unix.EpollEvent
57+ if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil {
58+ t.Errorf("registration retained after Poll returned: %d %v", n, err)
59+ }
60+ }); err != nil {
61+ t.Fatal(err)
62+ }
63+}
64+
65+func TestCloseUnblocksPoll(t *testing.T) {
66+ p, err := New()
67+ if err != nil {
68+ t.Fatal(err)
69+ }
70+ defer p.Close()
71+ fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
72+ if err != nil {
73+ t.Fatal(err)
74+ }
75+ defer unix.Close(fd)
76+ done := make(chan error, 1)
77+ go func() { _, e := p.Poll([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, -1); done <- e }()
78+ // Close is safe both before and during the netpoll wait.
79+ if err := p.Close(); err != nil {
80+ t.Fatal(err)
81+ }
82+ select {
83+ case err := <-done:
84+ if err == nil {
85+ t.Fatal("closed wait succeeded")
86+ }
87+ case <-time.After(time.Second):
88+ t.Fatal("close did not unblock wait")
89+ }
90+}
@@ -0,0 +1,90 @@
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 err := p.raw.Control(func(ep uintptr) {
56+ var events [1]unix.EpollEvent
57+ if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 || err != nil {
58+ t.Errorf("registration retained after Poll returned: %d %v", n, err)
59+ }
60+ }); err != nil {
61+ t.Fatal(err)
62+ }
63+}
64+
65+func TestCloseUnblocksPoll(t *testing.T) {
66+ p, err := New()
67+ if err != nil {
68+ t.Fatal(err)
69+ }
70+ defer p.Close()
71+ fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
72+ if err != nil {
73+ t.Fatal(err)
74+ }
75+ defer unix.Close(fd)
76+ done := make(chan error, 1)
77+ go func() { _, e := p.Poll([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}, -1); done <- e }()
78+ // Close is safe both before and during the netpoll wait.
79+ if err := p.Close(); err != nil {
80+ t.Fatal(err)
81+ }
82+ select {
83+ case err := <-done:
84+ if err == nil {
85+ t.Fatal("closed wait succeeded")
86+ }
87+ case <-time.After(time.Second):
88+ t.Fatal("close did not unblock wait")
89+ }
90+}
M internal/adapters/libinput/libinput.go +110 -35
@@ -10,6 +10,7 @@ import (
1010 "time"
1111 "unsafe"
1212
13+ "github.com/bnema/neferwl/internal/adapters/fdwait"
1314 "github.com/bnema/neferwl/internal/adapters/xkb"
1415 "github.com/bnema/neferwl/internal/ports"
1516 "github.com/bnema/purego"
@@ -236,6 +237,82 @@ type Options struct {
236237 LogKeys bool
237238 }
238239
240+// control is one control message of Options for Run to apply. keymap is the
241+// replacement keymap it carries, if any: a control dropped at shutdown
242+// closes it instead.
243+type control struct {
244+ apply func()
245+ keymap *xkb.Keymap
246+}
247+
248+func (c control) discard() {
249+ if c.keymap != nil {
250+ c.keymap.Close()
251+ }
252+}
253+
254+// relayControls hands the control messages of opts to Run on controls,
255+// waking its poll for each, until ctx ends. It only reads the channels of
256+// opts; the controls run on Run's goroutine, the only one to touch p, in, li
257+// and opts.Keymap.
258+func relayControls(ctx context.Context, opts *Options, p *pointer, in *inputState, li uintptr, controls chan<- control, wake func()) {
259+ for {
260+ var c control
261+ var f func()
262+ select {
263+ case <-ctx.Done():
264+ return
265+ case on := <-opts.Active:
266+ f = func() {
267+ if on {
268+ resume(li)
269+ } else {
270+ suspend(li)
271+ }
272+ opts.Log.Info().Bool("active", on).Msg("input")
273+ }
274+ case km := <-opts.Keymaps:
275+ c.keymap = km
276+ f = func() {
277+ if opts.Security != nil {
278+ km.QuarantineFrom(opts.Keymap)
279+ }
280+ opts.Keymap.Close()
281+ opts.Keymap = km
282+ opts.Log.Info().Msg("keymap replaced")
283+ }
284+ case l := <-opts.Layouts:
285+ f = func() {
286+ p.setLayout(l)
287+ p.moved(opts.MoveCursor, false)
288+ }
289+ case pc := <-opts.Constraints:
290+ f = func() {
291+ p.constrain(pc)
292+ p.moved(opts.MoveCursor, false)
293+ }
294+ case dc := <-opts.DeviceConfigs:
295+ f = func() {
296+ in.cfg = dc
297+ for dev := range in.devices {
298+ in.configure(dev, opts.Log)
299+ }
300+ t, m := dc.Touchpad, dc.Mouse
301+ opts.Log.Info().Bool("natural_scroll", t.NaturalScroll).Bool("tap", t.Tap).Float64("accel_speed", t.AccelSpeed).Str("accel_profile", t.AccelProfile).Bool("left_handed", t.LeftHanded).Float64("scroll_factor", t.ScrollFactor).
302+ Bool("mouse_natural_scroll", m.NaturalScroll).Float64("mouse_accel_speed", m.AccelSpeed).Str("mouse_accel_profile", m.AccelProfile).Bool("mouse_left_handed", m.LeftHanded).Msg("input devices reloaded")
303+ }
304+ }
305+ c.apply = f
306+ select {
307+ case controls <- c:
308+ wake()
309+ case <-ctx.Done():
310+ c.discard()
311+ return
312+ }
313+ }
314+}
315+
239316 // statsEvery is how often Run logs input throughput.
240317 const statsEvery = 10 * time.Second
241318
@@ -267,53 +344,51 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error
267344 p.moved(opts.MoveCursor, false)
268345 in := &inputState{cfg: opts.DeviceConfig, dev: libinputDevices{}, devices: map[uintptr]bool{}}
269346 defer in.release()
347+ waiter, err := fdwait.New()
348+ if err != nil {
349+ return fmt.Errorf("libinput waiter: %w", err)
350+ }
351+ defer waiter.Close()
352+ waiter.StopOn(ctx)
270353 fd := getFD(li)
354+ if err := waiter.Pin(fd); err != nil {
355+ return fmt.Errorf("libinput waiter: %w", err)
356+ }
271357 fwd := newForwarder(opts.Log)
272358 fwdCtx, stopFwd := context.WithCancel(ctx)
273359 var fwdDone sync.WaitGroup
274360 fwdDone.Go(func() { fwd.run(fwdCtx, input) })
275- defer func() { stopFwd(); fwdDone.Wait() }()
276- lastStats := time.Now()
361+ // Control messages wake the poll instead of a polling timeout; they are
362+ // applied on this goroutine, which owns libinput.
363+ controls := make(chan control, 8)
364+ fwdDone.Go(func() { relayControls(fwdCtx, &opts, p, in, li, controls, waiter.Wake) })
365+ defer func() {
366+ stopFwd()
367+ fwdDone.Wait()
368+ // Controls never applied: free what they carry.
369+ for len(controls) > 0 {
370+ (<-controls).discard()
371+ }
372+ }()
373+ fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
374+ nextStats := time.Now().Add(statsEvery)
277375 for ctx.Err() == nil {
278- if time.Since(lastStats) >= statsEvery {
279- lastStats = time.Now()
376+ if !time.Now().Before(nextStats) {
377+ nextStats = time.Now().Add(statsEvery)
280378 if s := fwd.take(); s.Motions > 0 || s.Sent > 0 {
281379 opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats")
282380 }
283381 }
284- select {
285- case on := <-opts.Active:
286- if on {
287- resume(li)
288- } else {
289- suspend(li)
290- }
291- opts.Log.Info().Bool("active", on).Msg("input")
292- case km := <-opts.Keymaps:
293- if opts.Security != nil {
294- km.QuarantineFrom(opts.Keymap)
295- }
296- opts.Keymap.Close()
297- opts.Keymap = km
298- opts.Log.Info().Msg("keymap replaced")
299- case l := <-opts.Layouts:
300- p.setLayout(l)
301- p.moved(opts.MoveCursor, false)
302- case c := <-opts.Constraints:
303- p.constrain(c)
304- p.moved(opts.MoveCursor, false)
305- case c := <-opts.DeviceConfigs:
306- in.cfg = c
307- for dev := range in.devices {
308- in.configure(dev, opts.Log)
382+ apply:
383+ for {
384+ select {
385+ case c := <-controls:
386+ c.apply()
387+ default:
388+ break apply
309389 }
310- t, m := c.Touchpad, c.Mouse
311- opts.Log.Info().Bool("natural_scroll", t.NaturalScroll).Bool("tap", t.Tap).Float64("accel_speed", t.AccelSpeed).Str("accel_profile", t.AccelProfile).Bool("left_handed", t.LeftHanded).Float64("scroll_factor", t.ScrollFactor).
312- Bool("mouse_natural_scroll", m.NaturalScroll).Float64("mouse_accel_speed", m.AccelSpeed).Str("mouse_accel_profile", m.AccelProfile).Bool("mouse_left_handed", m.LeftHanded).Msg("input devices reloaded")
313- default:
314390 }
315- fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
316- n, err := unix.Poll(fds, 100)
391+ n, err := waiter.Poll(fds, max(0, int(time.Until(nextStats).Milliseconds())+1))
317392 if err != nil && !errors.Is(err, unix.EINTR) {
318393 return fmt.Errorf("libinput poll: %w", err)
319394 }
@@ -10,6 +10,7 @@ import (
10 "time"10 "time"
11 "unsafe"11 "unsafe"
12 12
13+ "github.com/bnema/neferwl/internal/adapters/fdwait"
13 "github.com/bnema/neferwl/internal/adapters/xkb"14 "github.com/bnema/neferwl/internal/adapters/xkb"
14 "github.com/bnema/neferwl/internal/ports"15 "github.com/bnema/neferwl/internal/ports"
15 "github.com/bnema/purego"16 "github.com/bnema/purego"
@@ -236,6 +237,82 @@ type Options struct {
236 LogKeys bool237 LogKeys bool
237 }238 }
238 239
240+// control is one control message of Options for Run to apply. keymap is the
241+// replacement keymap it carries, if any: a control dropped at shutdown
242+// closes it instead.
243+type control struct {
244+ apply func()
245+ keymap *xkb.Keymap
246+}
247+
248+func (c control) discard() {
249+ if c.keymap != nil {
250+ c.keymap.Close()
251+ }
252+}
253+
254+// relayControls hands the control messages of opts to Run on controls,
255+// waking its poll for each, until ctx ends. It only reads the channels of
256+// opts; the controls run on Run's goroutine, the only one to touch p, in, li
257+// and opts.Keymap.
258+func relayControls(ctx context.Context, opts *Options, p *pointer, in *inputState, li uintptr, controls chan<- control, wake func()) {
259+ for {
260+ var c control
261+ var f func()
262+ select {
263+ case <-ctx.Done():
264+ return
265+ case on := <-opts.Active:
266+ f = func() {
267+ if on {
268+ resume(li)
269+ } else {
270+ suspend(li)
271+ }
272+ opts.Log.Info().Bool("active", on).Msg("input")
273+ }
274+ case km := <-opts.Keymaps:
275+ c.keymap = km
276+ f = func() {
277+ if opts.Security != nil {
278+ km.QuarantineFrom(opts.Keymap)
279+ }
280+ opts.Keymap.Close()
281+ opts.Keymap = km
282+ opts.Log.Info().Msg("keymap replaced")
283+ }
284+ case l := <-opts.Layouts:
285+ f = func() {
286+ p.setLayout(l)
287+ p.moved(opts.MoveCursor, false)
288+ }
289+ case pc := <-opts.Constraints:
290+ f = func() {
291+ p.constrain(pc)
292+ p.moved(opts.MoveCursor, false)
293+ }
294+ case dc := <-opts.DeviceConfigs:
295+ f = func() {
296+ in.cfg = dc
297+ for dev := range in.devices {
298+ in.configure(dev, opts.Log)
299+ }
300+ t, m := dc.Touchpad, dc.Mouse
301+ opts.Log.Info().Bool("natural_scroll", t.NaturalScroll).Bool("tap", t.Tap).Float64("accel_speed", t.AccelSpeed).Str("accel_profile", t.AccelProfile).Bool("left_handed", t.LeftHanded).Float64("scroll_factor", t.ScrollFactor).
302+ Bool("mouse_natural_scroll", m.NaturalScroll).Float64("mouse_accel_speed", m.AccelSpeed).Str("mouse_accel_profile", m.AccelProfile).Bool("mouse_left_handed", m.LeftHanded).Msg("input devices reloaded")
303+ }
304+ }
305+ c.apply = f
306+ select {
307+ case controls <- c:
308+ wake()
309+ case <-ctx.Done():
310+ c.discard()
311+ return
312+ }
313+ }
314+}
315+
239 // statsEvery is how often Run logs input throughput.316 // statsEvery is how often Run logs input throughput.
240 const statsEvery = 10 * time.Second317 const statsEvery = 10 * time.Second
241 318
@@ -267,53 +344,51 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error
267 p.moved(opts.MoveCursor, false)344 p.moved(opts.MoveCursor, false)
268 in := &inputState{cfg: opts.DeviceConfig, dev: libinputDevices{}, devices: map[uintptr]bool{}}345 in := &inputState{cfg: opts.DeviceConfig, dev: libinputDevices{}, devices: map[uintptr]bool{}}
269 defer in.release()346 defer in.release()
347+ waiter, err := fdwait.New()
348+ if err != nil {
349+ return fmt.Errorf("libinput waiter: %w", err)
350+ }
351+ defer waiter.Close()
352+ waiter.StopOn(ctx)
270 fd := getFD(li)353 fd := getFD(li)
354+ if err := waiter.Pin(fd); err != nil {
355+ return fmt.Errorf("libinput waiter: %w", err)
356+ }
271 fwd := newForwarder(opts.Log)357 fwd := newForwarder(opts.Log)
272 fwdCtx, stopFwd := context.WithCancel(ctx)358 fwdCtx, stopFwd := context.WithCancel(ctx)
273 var fwdDone sync.WaitGroup359 var fwdDone sync.WaitGroup
274 fwdDone.Go(func() { fwd.run(fwdCtx, input) })360 fwdDone.Go(func() { fwd.run(fwdCtx, input) })
275- defer func() { stopFwd(); fwdDone.Wait() }()361+ // Control messages wake the poll instead of a polling timeout; they are
276- lastStats := time.Now()362+ // applied on this goroutine, which owns libinput.
363+ controls := make(chan control, 8)
364+ fwdDone.Go(func() { relayControls(fwdCtx, &opts, p, in, li, controls, waiter.Wake) })
365+ defer func() {
366+ stopFwd()
367+ fwdDone.Wait()
368+ // Controls never applied: free what they carry.
369+ for len(controls) > 0 {
370+ (<-controls).discard()
371+ }
372+ }()
373+ fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
374+ nextStats := time.Now().Add(statsEvery)
277 for ctx.Err() == nil {375 for ctx.Err() == nil {
278- if time.Since(lastStats) >= statsEvery {376+ if !time.Now().Before(nextStats) {
279- lastStats = time.Now()377+ nextStats = time.Now().Add(statsEvery)
280 if s := fwd.take(); s.Motions > 0 || s.Sent > 0 {378 if s := fwd.take(); s.Motions > 0 || s.Sent > 0 {
281 opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats")379 opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats")
282 }380 }
283 }381 }
284- select {382+ apply:
285- case on := <-opts.Active:383+ for {
286- if on {384+ select {
287- resume(li)385+ case c := <-controls:
288- } else {386+ c.apply()
289- suspend(li)387+ default:
290- }388+ break apply
291- opts.Log.Info().Bool("active", on).Msg("input")
292- case km := <-opts.Keymaps:
293- if opts.Security != nil {
294- km.QuarantineFrom(opts.Keymap)
295- }
296- opts.Keymap.Close()
297- opts.Keymap = km
298- opts.Log.Info().Msg("keymap replaced")
299- case l := <-opts.Layouts:
300- p.setLayout(l)
301- p.moved(opts.MoveCursor, false)
302- case c := <-opts.Constraints:
303- p.constrain(c)
304- p.moved(opts.MoveCursor, false)
305- case c := <-opts.DeviceConfigs:
306- in.cfg = c
307- for dev := range in.devices {
308- in.configure(dev, opts.Log)
309 }389 }
310- t, m := c.Touchpad, c.Mouse
311- opts.Log.Info().Bool("natural_scroll", t.NaturalScroll).Bool("tap", t.Tap).Float64("accel_speed", t.AccelSpeed).Str("accel_profile", t.AccelProfile).Bool("left_handed", t.LeftHanded).Float64("scroll_factor", t.ScrollFactor).
312- Bool("mouse_natural_scroll", m.NaturalScroll).Float64("mouse_accel_speed", m.AccelSpeed).Str("mouse_accel_profile", m.AccelProfile).Bool("mouse_left_handed", m.LeftHanded).Msg("input devices reloaded")
313- default:
314 }390 }
315- fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}391+ n, err := waiter.Poll(fds, max(0, int(time.Until(nextStats).Milliseconds())+1))
316- n, err := unix.Poll(fds, 100)
317 if err != nil && !errors.Is(err, unix.EINTR) {392 if err != nil && !errors.Is(err, unix.EINTR) {
318 return fmt.Errorf("libinput poll: %w", err)393 return fmt.Errorf("libinput poll: %w", err)
319 }394 }
M internal/adapters/seat/seat.go +40 -5
@@ -10,6 +10,7 @@ import (
1010 "sync"
1111 "unsafe"
1212
13+ "github.com/bnema/neferwl/internal/adapters/fdwait"
1314 "github.com/bnema/purego"
1415 "github.com/bnema/zerowrap"
1516 "golang.org/x/sys/unix"
@@ -75,6 +76,9 @@ type Seat struct {
7576 // manager: the last moment this process is still DRM master.
7677 beforeDisable func()
7778 disabler disabler
79+ // waiter is Run's poller: calls that make libseat queue events wake it
80+ // to dispatch them.
81+ waiter *fdwait.Poller
7882 }
7983
8084 // disabler acks a seat disable to the seat manager (libseat_disable_seat),
@@ -189,29 +193,58 @@ func (s *Seat) dispatchLocked(ctx context.Context, timeoutMs int) error {
189193
190194 // Run dispatches seat events until ctx ends. The owner calls Close afterwards,
191195 // once outputs have restored their CRTC state.
196+// It sleeps until the seat fd is readable or a seat call wakes it.
192197 func (s *Seat) Run(ctx context.Context) error {
198+ waiter, err := fdwait.New()
199+ if err != nil {
200+ return fmt.Errorf("seat waiter: %w", err)
201+ }
202+ defer waiter.Close()
203+ waiter.StopOn(ctx)
204+ s.mu.Lock()
205+ fd := getFD(s.handle)
206+ s.waiter = waiter
207+ s.mu.Unlock()
208+ defer func() {
209+ s.mu.Lock()
210+ s.waiter = nil
211+ s.mu.Unlock()
212+ }()
213+ if err := waiter.Pin(fd); err != nil {
214+ return fmt.Errorf("seat waiter: %w", err)
215+ }
216+ fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
193217 for ctx.Err() == nil {
194- fds := []unix.PollFd{{Fd: getFD(s.handle), Events: unix.POLLIN}}
195- if _, err := unix.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) {
196- return fmt.Errorf("seat poll: %w", err)
197- }
198- // Dispatch every tick: libseat may queue events without the fd becoming readable.
218+ // Dispatch on every return, a wake included: libseat may queue
219+ // events while answering a request, without the fd becoming
220+ // readable (its backends ping themselves, the wake covers it too).
199221 s.mu.Lock()
200222 r := dispatch(s.handle, 0)
201223 s.mu.Unlock()
202224 if r < 0 {
203225 return errors.New("libseat_dispatch failed")
204226 }
227+ if _, err := waiter.Poll(fds, -1); err != nil && !errors.Is(err, unix.EINTR) {
228+ return fmt.Errorf("seat poll: %w", err)
229+ }
205230 }
206231 return nil
207232 }
208233
234+// wakeLocked has Run dispatch the events a seat request may have queued.
235+func (s *Seat) wakeLocked() {
236+ if s.waiter != nil {
237+ s.waiter.Wake()
238+ }
239+}
240+
209241 // OpenDevice opens path through the seat manager and returns its fd.
210242 func (s *Seat) OpenDevice(path string) (int, error) {
211243 s.mu.Lock()
212244 defer s.mu.Unlock()
213245 var fd int32 = -1
214246 id := openDevice(s.handle, path, &fd)
247+ s.wakeLocked()
215248 if id < 0 {
216249 s.log.Warn().Str("path", path).Msg("open device failed")
217250 return -1, fmt.Errorf("libseat_open_device %s failed", path)
@@ -231,6 +264,7 @@ func (s *Seat) CloseDevice(fd int) {
231264 }
232265 delete(s.devices, int32(fd))
233266 closeDevice(s.handle, id)
267+ s.wakeLocked()
234268 _ = unix.Close(fd)
235269 s.log.Info().Int("fd", fd).Msg("device closed")
236270 }
@@ -243,6 +277,7 @@ func (s *Seat) SwitchVT(vt int) {
243277 if switchSession(s.handle, int32(vt)) < 0 {
244278 s.log.Warn().Int("vt", vt).Msg("switch vt failed")
245279 }
280+ s.wakeLocked()
246281 }
247282
248283 // Close releases all devices and the seat; the VT returns to text mode. Safe to call twice.
@@ -10,6 +10,7 @@ import (
10 "sync"10 "sync"
11 "unsafe"11 "unsafe"
12 12
13+ "github.com/bnema/neferwl/internal/adapters/fdwait"
13 "github.com/bnema/purego"14 "github.com/bnema/purego"
14 "github.com/bnema/zerowrap"15 "github.com/bnema/zerowrap"
15 "golang.org/x/sys/unix"16 "golang.org/x/sys/unix"
@@ -75,6 +76,9 @@ type Seat struct {
75 // manager: the last moment this process is still DRM master.76 // manager: the last moment this process is still DRM master.
76 beforeDisable func()77 beforeDisable func()
77 disabler disabler78 disabler disabler
79+ // waiter is Run's poller: calls that make libseat queue events wake it
80+ // to dispatch them.
81+ waiter *fdwait.Poller
78 }82 }
79 83
80 // disabler acks a seat disable to the seat manager (libseat_disable_seat),84 // disabler acks a seat disable to the seat manager (libseat_disable_seat),
@@ -189,29 +193,58 @@ func (s *Seat) dispatchLocked(ctx context.Context, timeoutMs int) error {
189 193
190 // Run dispatches seat events until ctx ends. The owner calls Close afterwards,194 // Run dispatches seat events until ctx ends. The owner calls Close afterwards,
191 // once outputs have restored their CRTC state.195 // once outputs have restored their CRTC state.
196+// It sleeps until the seat fd is readable or a seat call wakes it.
192 func (s *Seat) Run(ctx context.Context) error {197 func (s *Seat) Run(ctx context.Context) error {
198+ waiter, err := fdwait.New()
199+ if err != nil {
200+ return fmt.Errorf("seat waiter: %w", err)
201+ }
202+ defer waiter.Close()
203+ waiter.StopOn(ctx)
204+ s.mu.Lock()
205+ fd := getFD(s.handle)
206+ s.waiter = waiter
207+ s.mu.Unlock()
208+ defer func() {
209+ s.mu.Lock()
210+ s.waiter = nil
211+ s.mu.Unlock()
212+ }()
213+ if err := waiter.Pin(fd); err != nil {
214+ return fmt.Errorf("seat waiter: %w", err)
215+ }
216+ fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}}
193 for ctx.Err() == nil {217 for ctx.Err() == nil {
194- fds := []unix.PollFd{{Fd: getFD(s.handle), Events: unix.POLLIN}}218+ // Dispatch on every return, a wake included: libseat may queue
195- if _, err := unix.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) {219+ // events while answering a request, without the fd becoming
196- return fmt.Errorf("seat poll: %w", err)220+ // readable (its backends ping themselves, the wake covers it too).
197- }
198- // Dispatch every tick: libseat may queue events without the fd becoming readable.
199 s.mu.Lock()221 s.mu.Lock()
200 r := dispatch(s.handle, 0)222 r := dispatch(s.handle, 0)
201 s.mu.Unlock()223 s.mu.Unlock()
202 if r < 0 {224 if r < 0 {
203 return errors.New("libseat_dispatch failed")225 return errors.New("libseat_dispatch failed")
204 }226 }
227+ if _, err := waiter.Poll(fds, -1); err != nil && !errors.Is(err, unix.EINTR) {
228+ return fmt.Errorf("seat poll: %w", err)
229+ }
205 }230 }
206 return nil231 return nil
207 }232 }
208 233
234+// wakeLocked has Run dispatch the events a seat request may have queued.
235+func (s *Seat) wakeLocked() {
236+ if s.waiter != nil {
237+ s.waiter.Wake()
238+ }
239+}
240+
209 // OpenDevice opens path through the seat manager and returns its fd.241 // OpenDevice opens path through the seat manager and returns its fd.
210 func (s *Seat) OpenDevice(path string) (int, error) {242 func (s *Seat) OpenDevice(path string) (int, error) {
211 s.mu.Lock()243 s.mu.Lock()
212 defer s.mu.Unlock()244 defer s.mu.Unlock()
213 var fd int32 = -1245 var fd int32 = -1
214 id := openDevice(s.handle, path, &fd)246 id := openDevice(s.handle, path, &fd)
247+ s.wakeLocked()
215 if id < 0 {248 if id < 0 {
216 s.log.Warn().Str("path", path).Msg("open device failed")249 s.log.Warn().Str("path", path).Msg("open device failed")
217 return -1, fmt.Errorf("libseat_open_device %s failed", path)250 return -1, fmt.Errorf("libseat_open_device %s failed", path)
@@ -231,6 +264,7 @@ func (s *Seat) CloseDevice(fd int) {
231 }264 }
232 delete(s.devices, int32(fd))265 delete(s.devices, int32(fd))
233 closeDevice(s.handle, id)266 closeDevice(s.handle, id)
267+ s.wakeLocked()
234 _ = unix.Close(fd)268 _ = unix.Close(fd)
235 s.log.Info().Int("fd", fd).Msg("device closed")269 s.log.Info().Int("fd", fd).Msg("device closed")
236 }270 }
@@ -243,6 +277,7 @@ func (s *Seat) SwitchVT(vt int) {
243 if switchSession(s.handle, int32(vt)) < 0 {277 if switchSession(s.handle, int32(vt)) < 0 {
244 s.log.Warn().Int("vt", vt).Msg("switch vt failed")278 s.log.Warn().Int("vt", vt).Msg("switch vt failed")
245 }279 }
280+ s.wakeLocked()
246 }281 }
247 282
248 // Close releases all devices and the seat; the VT returns to text mode. Safe to call twice.283 // Close releases all devices and the seat; the VT returns to text mode. Safe to call twice.
M internal/adapters/wayland/acquire_failure_test.go +40 -2
@@ -120,6 +120,44 @@ func TestImplicitAcquireWatchFailureDiscardsCommit(t *testing.T) {
120120 }
121121 }
122122
123+// An idle waiter sleeps in one poll with no timeout; cancelling ctx ends it.
124+func TestSyncWaiterIdleNeverTicks(t *testing.T) {
125+ sw, err := newSyncWaiter(func() {})
126+ if err != nil {
127+ t.Fatal(err)
128+ }
129+ defer sw.close()
130+ poller := newMockacquirePoller(t)
131+ real := sw.poller
132+ sw.poller = poller
133+ polls := make(chan int, 16)
134+ poller.EXPECT().poll(mock.Anything, mock.Anything).RunAndReturn(func(fds []unix.PollFd, timeout int) (int, error) {
135+ polls <- timeout
136+ return real.poll(fds, timeout)
137+ })
138+ ctx, cancel := context.WithCancel(context.Background())
139+ defer cancel()
140+ done := make(chan struct{})
141+ go func() { sw.run(ctx); close(done) }()
142+ select {
143+ case timeout := <-polls:
144+ if timeout != -1 {
145+ t.Fatalf("idle poll timeout %d, want -1", timeout)
146+ }
147+ case <-time.After(5 * time.Second):
148+ t.Fatal("the waiter never polled")
149+ }
150+ cancel()
151+ select {
152+ case <-done:
153+ case <-time.After(5 * time.Second):
154+ t.Fatal("cancel did not stop the waiter")
155+ }
156+ if n := len(polls); n != 0 {
157+ t.Fatalf("%d extra polls while idle", n)
158+ }
159+}
160+
123161 func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) {
124162 poller := newMockacquirePoller(t)
125163 wake := make(chan struct{}, 1)
@@ -135,8 +173,8 @@ func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) {
135173 }
136174 active := sw.add(fd)
137175 var added *syncWait
138- poller.EXPECT().poll(mock.Anything, 100).RunAndReturn(func(fds []unix.PollFd, _ int) (int, error) { fds[0].Revents = unix.POLLIN; return 1, nil }).Once()
139- poller.EXPECT().poll(mock.Anything, 100).RunAndReturn(func([]unix.PollFd, int) (int, error) {
176+ poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func(fds []unix.PollFd, _ int) (int, error) { fds[0].Revents = unix.POLLIN; return 1, nil }).Once()
177+ poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func([]unix.PollFd, int) (int, error) {
140178 fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC)
141179 if err != nil {
142180 t.Error(err)
@@ -120,6 +120,44 @@ func TestImplicitAcquireWatchFailureDiscardsCommit(t *testing.T) {
120 }120 }
121 }121 }
122 122
123+// An idle waiter sleeps in one poll with no timeout; cancelling ctx ends it.
124+func TestSyncWaiterIdleNeverTicks(t *testing.T) {
125+ sw, err := newSyncWaiter(func() {})
126+ if err != nil {
127+ t.Fatal(err)
128+ }
129+ defer sw.close()
130+ poller := newMockacquirePoller(t)
131+ real := sw.poller
132+ sw.poller = poller
133+ polls := make(chan int, 16)
134+ poller.EXPECT().poll(mock.Anything, mock.Anything).RunAndReturn(func(fds []unix.PollFd, timeout int) (int, error) {
135+ polls <- timeout
136+ return real.poll(fds, timeout)
137+ })
138+ ctx, cancel := context.WithCancel(context.Background())
139+ defer cancel()
140+ done := make(chan struct{})
141+ go func() { sw.run(ctx); close(done) }()
142+ select {
143+ case timeout := <-polls:
144+ if timeout != -1 {
145+ t.Fatalf("idle poll timeout %d, want -1", timeout)
146+ }
147+ case <-time.After(5 * time.Second):
148+ t.Fatal("the waiter never polled")
149+ }
150+ cancel()
151+ select {
152+ case <-done:
153+ case <-time.After(5 * time.Second):
154+ t.Fatal("cancel did not stop the waiter")
155+ }
156+ if n := len(polls); n != 0 {
157+ t.Fatalf("%d extra polls while idle", n)
158+ }
159+}
160+
123 func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) {161 func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) {
124 poller := newMockacquirePoller(t)162 poller := newMockacquirePoller(t)
125 wake := make(chan struct{}, 1)163 wake := make(chan struct{}, 1)
@@ -135,8 +173,8 @@ func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) {
135 }173 }
136 active := sw.add(fd)174 active := sw.add(fd)
137 var added *syncWait175 var added *syncWait
138- poller.EXPECT().poll(mock.Anything, 100).RunAndReturn(func(fds []unix.PollFd, _ int) (int, error) { fds[0].Revents = unix.POLLIN; return 1, nil }).Once()176+ poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func(fds []unix.PollFd, _ int) (int, error) { fds[0].Revents = unix.POLLIN; return 1, nil }).Once()
139- poller.EXPECT().poll(mock.Anything, 100).RunAndReturn(func([]unix.PollFd, int) (int, error) {177+ poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func([]unix.PollFd, int) (int, error) {
140 fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC)178 fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC)
141 if err != nil {179 if err != nil {
142 t.Error(err)180 t.Error(err)
M internal/adapters/wayland/alloc_hotpath_test.go +1 -0
@@ -65,6 +65,7 @@ func TestCommitSyncStorageIsPerUpdate(t *testing.T) {
6565 t.Fatal(err)
6666 }
6767 defer sw.pipeR.Close()
68+ defer sw.waiter.Close()
6869 defer sw.close()
6970 srv.syncWait = sw
7071 tl := &timeline{dev: dev, handle: 7, alive: true}
@@ -65,6 +65,7 @@ func TestCommitSyncStorageIsPerUpdate(t *testing.T) {
65 t.Fatal(err)65 t.Fatal(err)
66 }66 }
67 defer sw.pipeR.Close()67 defer sw.pipeR.Close()
68+ defer sw.waiter.Close()
68 defer sw.close()69 defer sw.close()
69 srv.syncWait = sw70 srv.syncWait = sw
70 tl := &timeline{dev: dev, handle: 7, alive: true}71 tl := &timeline{dev: dev, handle: 7, alive: true}
M internal/adapters/wayland/syncobj.go +25 -6
@@ -9,6 +9,7 @@ import (
99
1010 "github.com/bnema/go-wayland-bindings/server/linuxdrmsyncobj"
1111 "github.com/bnema/go-wayland-bindings/server/wayland"
12+ "github.com/bnema/neferwl/internal/adapters/fdwait"
1213 "github.com/bnema/neferwl/internal/ports"
1314 "github.com/bnema/purego-libwayland/server"
1415 "github.com/bnema/zerowrap"
@@ -197,14 +198,15 @@ type acquirePoller interface {
197198 poll([]unix.PollFd, int) (int, error)
198199 }
199200
200-type unixAcquirePoller struct{}
201+type unixAcquirePoller struct{ waiter *fdwait.Poller }
201202
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)
204205 }
205206
206207 type syncWaiter struct {
207208 poller acquirePoller
209+ waiter *fdwait.Poller
208210 terminal bool // polling ended in failure; protected by mu
209211 mu sync.Mutex
210212 ready map[*syncWait]bool
@@ -229,7 +231,13 @@ func newSyncWaiter(wake func()) (*syncWaiter, error) {
229231 if err != nil {
230232 return nil, err
231233 }
232- return &syncWaiter{poller: unixAcquirePoller{}, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
234+ p, err := fdwait.New()
235+ if err != nil {
236+ r.Close()
237+ w.Close()
238+ return nil, err
239+ }
240+ return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
233241 }
234242
235243 // watch registers an eventfd for p; the wait is ready once it fires.
@@ -375,6 +383,17 @@ func (sw *syncWaiter) fail(waits []*syncWait) {
375383 // run polls every watched eventfd; it owns and closes them.
376384 func (sw *syncWaiter) run(ctx context.Context) {
377385 defer sw.pipeR.Close()
386+ defer sw.waiter.Close()
387+ pipeFD := int(sw.pipeR.Fd())
388+ // The pipe lives as long as the poller; the eventfds come and go. The
389+ // poll sleeps until a fence, a new or cancelled wait (the pipe), or the
390+ // end of ctx.
391+ sw.waiter.StopOn(ctx)
392+ if err := sw.waiter.Pin(int32(pipeFD)); err != nil {
393+ sw.log.Warn().Err(err).Msg("sync waiter")
394+ sw.fail(nil)
395+ return
396+ }
378397 var waits []*syncWait
379398 defer func() {
380399 sw.mu.Lock()
@@ -395,11 +414,11 @@ func (sw *syncWaiter) run(ctx context.Context) {
395414 live = append(live, w)
396415 }
397416 waits = live
398- fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}
417+ fds := []unix.PollFd{{Fd: int32(pipeFD), Events: unix.POLLIN}}
399418 for _, w := range waits {
400419 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
401420 }
402- n, err := sw.poller.poll(fds, 100)
421+ n, err := sw.poller.poll(fds, -1)
403422 if err != nil && !errors.Is(err, unix.EINTR) {
404423 sw.fail(waits)
405424 return
@@ -9,6 +9,7 @@ import (
9 9
10 "github.com/bnema/go-wayland-bindings/server/linuxdrmsyncobj"10 "github.com/bnema/go-wayland-bindings/server/linuxdrmsyncobj"
11 "github.com/bnema/go-wayland-bindings/server/wayland"11 "github.com/bnema/go-wayland-bindings/server/wayland"
12+ "github.com/bnema/neferwl/internal/adapters/fdwait"
12 "github.com/bnema/neferwl/internal/ports"13 "github.com/bnema/neferwl/internal/ports"
13 "github.com/bnema/purego-libwayland/server"14 "github.com/bnema/purego-libwayland/server"
14 "github.com/bnema/zerowrap"15 "github.com/bnema/zerowrap"
@@ -197,14 +198,15 @@ type acquirePoller interface {
197 poll([]unix.PollFd, int) (int, error)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+func (p unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
203- return unix.Poll(fds, timeout)204+ return p.waiter.Poll(fds, timeout)
204 }205 }
205 206
206 type syncWaiter struct {207 type syncWaiter struct {
207 poller acquirePoller208 poller acquirePoller
209+ waiter *fdwait.Poller
208 terminal bool // polling ended in failure; protected by mu210 terminal bool // polling ended in failure; protected by mu
209 mu sync.Mutex211 mu sync.Mutex
210 ready map[*syncWait]bool212 ready map[*syncWait]bool
@@ -229,7 +231,13 @@ func newSyncWaiter(wake func()) (*syncWaiter, error) {
229 if err != nil {231 if err != nil {
230 return nil, err232 return nil, err
231 }233 }
232- return &syncWaiter{poller: unixAcquirePoller{}, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil234+ p, err := fdwait.New()
235+ if err != nil {
236+ r.Close()
237+ w.Close()
238+ return nil, err
239+ }
240+ return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
233 }241 }
234 242
235 // watch registers an eventfd for p; the wait is ready once it fires.243 // watch registers an eventfd for p; the wait is ready once it fires.
@@ -375,6 +383,17 @@ func (sw *syncWaiter) fail(waits []*syncWait) {
375 // run polls every watched eventfd; it owns and closes them.383 // run polls every watched eventfd; it owns and closes them.
376 func (sw *syncWaiter) run(ctx context.Context) {384 func (sw *syncWaiter) run(ctx context.Context) {
377 defer sw.pipeR.Close()385 defer sw.pipeR.Close()
386+ defer sw.waiter.Close()
387+ pipeFD := int(sw.pipeR.Fd())
388+ // The pipe lives as long as the poller; the eventfds come and go. The
389+ // poll sleeps until a fence, a new or cancelled wait (the pipe), or the
390+ // end of ctx.
391+ sw.waiter.StopOn(ctx)
392+ if err := sw.waiter.Pin(int32(pipeFD)); err != nil {
393+ sw.log.Warn().Err(err).Msg("sync waiter")
394+ sw.fail(nil)
395+ return
396+ }
378 var waits []*syncWait397 var waits []*syncWait
379 defer func() {398 defer func() {
380 sw.mu.Lock()399 sw.mu.Lock()
@@ -395,11 +414,11 @@ func (sw *syncWaiter) run(ctx context.Context) {
395 live = append(live, w)414 live = append(live, w)
396 }415 }
397 waits = live416 waits = live
398- fds := []unix.PollFd{{Fd: int32(sw.pipeR.Fd()), Events: unix.POLLIN}}417+ fds := []unix.PollFd{{Fd: int32(pipeFD), Events: unix.POLLIN}}
399 for _, w := range waits {418 for _, w := range waits {
400 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})419 fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN})
401 }420 }
402- n, err := sw.poller.poll(fds, 100)421+ n, err := sw.poller.poll(fds, -1)
403 if err != nil && !errors.Is(err, unix.EINTR) {422 if err != nil && !errors.Is(err, unix.EINTR) {
404 sw.fail(waits)423 sw.fail(waits)
405 return424 return
M internal/adapters/wayland/syncwait_test.go +2 -0
@@ -20,6 +20,7 @@ func TestSyncWaitAlreadySignalled(t *testing.T) {
2020 t.Fatal(err)
2121 }
2222 defer sw.pipeR.Close()
23+ defer sw.waiter.Close()
2324 defer sw.close()
2425 w, err := sw.watch(syncPoint{tl: &timeline{dev: dev, handle: 7}, point: 4})
2526 if err != nil || !sw.fired(w) {
@@ -34,6 +35,7 @@ func TestImplicitPollErrorRejectedAndWarnOnce(t *testing.T) {
3435 t.Fatal(err)
3536 }
3637 defer sw.pipeR.Close()
38+ defer sw.waiter.Close()
3739 defer sw.close()
3840 var output bytes.Buffer
3941 sw.log = zerowrap.New(zerowrap.Config{Output: &output})
@@ -20,6 +20,7 @@ func TestSyncWaitAlreadySignalled(t *testing.T) {
20 t.Fatal(err)20 t.Fatal(err)
21 }21 }
22 defer sw.pipeR.Close()22 defer sw.pipeR.Close()
23+ defer sw.waiter.Close()
23 defer sw.close()24 defer sw.close()
24 w, err := sw.watch(syncPoint{tl: &timeline{dev: dev, handle: 7}, point: 4})25 w, err := sw.watch(syncPoint{tl: &timeline{dev: dev, handle: 7}, point: 4})
25 if err != nil || !sw.fired(w) {26 if err != nil || !sw.fired(w) {
@@ -34,6 +35,7 @@ func TestImplicitPollErrorRejectedAndWarnOnce(t *testing.T) {
34 t.Fatal(err)35 t.Fatal(err)
35 }36 }
36 defer sw.pipeR.Close()37 defer sw.pipeR.Close()
38+ defer sw.waiter.Close()
37 defer sw.close()39 defer sw.close()
38 var output bytes.Buffer40 var output bytes.Buffer
39 sw.log = zerowrap.New(zerowrap.Config{Output: &output})41 sw.log = zerowrap.New(zerowrap.Config{Output: &output})
M tools/bench/README.md +2 -0
@@ -8,3 +8,5 @@ Requires `go`, `chromium` and `python3`. Frames are counted from the renderer's
88 tools/bench/bench.sh gif.html 15 # 32x32 GIF, 20 fps
99
1010 Environment: `BENCH_X11=1` runs chromium through XWayland, `BENCH_DEBUG=<categories>` adds `--debug` categories, `BENCH_KEEP=1` leaves the processes running, `BENCH_PPROF=<addr>` sets the pprof address (default `localhost:6061`).
11+
12+`idle.html` is a static page: run it with `BENCH_IDLE=1 tools/bench/bench.sh idle.html 15` (no frame is required). The output also reports `cs/s` and, for the thread with the most voluntary switches (usually the Go runtime's sysmon), its scheduler wakes per second (`top_wakes/s`, schedstat field 3). `BENCH_ROOT=<worktree>` builds neferwl from another worktree for A/B runs.
@@ -8,3 +8,5 @@ Requires `go`, `chromium` and `python3`. Frames are counted from the renderer's
8 tools/bench/bench.sh gif.html 15 # 32x32 GIF, 20 fps8 tools/bench/bench.sh gif.html 15 # 32x32 GIF, 20 fps
9 9
10 Environment: `BENCH_X11=1` runs chromium through XWayland, `BENCH_DEBUG=<categories>` adds `--debug` categories, `BENCH_KEEP=1` leaves the processes running, `BENCH_PPROF=<addr>` sets the pprof address (default `localhost:6061`).10 Environment: `BENCH_X11=1` runs chromium through XWayland, `BENCH_DEBUG=<categories>` adds `--debug` categories, `BENCH_KEEP=1` leaves the processes running, `BENCH_PPROF=<addr>` sets the pprof address (default `localhost:6061`).
11+
12+`idle.html` is a static page: run it with `BENCH_IDLE=1 tools/bench/bench.sh idle.html 15` (no frame is required). The output also reports `cs/s` and, for the thread with the most voluntary switches (usually the Go runtime's sysmon), its scheduler wakes per second (`top_wakes/s`, schedstat field 3). `BENCH_ROOT=<worktree>` builds neferwl from another worktree for A/B runs.
M tools/bench/bench.sh +18 -6
@@ -7,12 +7,14 @@
77 # BENCH_DEBUG=cat extra --debug categories (comma separated)
88 # BENCH_KEEP=1 leave neferwl and chromium running (prints PIDs)
99 # BENCH_PPROF=addr pprof listen address (default localhost:6061)
10+# BENCH_ROOT=dir build neferwl from another worktree (A/B runs)
11+# BENCH_IDLE=1 static page: no frames expected, per-second numbers only
1012 set -euo pipefail
1113
1214 page=${1:?usage: bench.sh <page.html> <seconds>}
1315 secs=${2:?usage: bench.sh <page.html> <seconds>}
1416 here=$(cd "$(dirname "$0")" && pwd)
15-root=$(cd "$here/../.." && pwd)
17+root=${BENCH_ROOT:-$(cd "$here/../.." && pwd)}
1618 tmp=$(mktemp -d)
1719 pids=()
1820
@@ -67,22 +69,32 @@ cs() { cat /proc/"$nwl"/task/*/status | awk '/^voluntary_ctxt/{v+=$2} END{print
6769 cpu_ns() { cat /proc/"$nwl"/task/*/schedstat | awk '{v+=$1} END{printf "%d\n", v}'; }
6870 frames() { grep -c '"message":"frame"' "$log" || true; }
6971
70-t0=$(cpu_ns); f0=$(frames); c0=$(cs)
72+# Per-thread "tid sched_in_count voluntary_switches" (schedstat field 3 = times scheduled in).
73+threads() { for d in /proc/"$nwl"/task/*; do
74+ echo "${d##*/} $(awk '{print $3}' "$d/schedstat" 2>/dev/null) $(awk '/^voluntary_ctxt/{print $2}' "$d/status" 2>/dev/null)"
75+done; }
76+
77+th0=$(threads); t0=$(cpu_ns); f0=$(frames); c0=$(cs)
7178 sleep "$secs"
72-t1=$(cpu_ns); f1=$(frames); c1=$(cs)
79+th1=$(threads); t1=$(cpu_ns); f1=$(frames); c1=$(cs)
7380
74-if [ "$((f1 - f0))" -le 0 ]; then
81+if [ -z "${BENCH_IDLE:-}" ] && [ "$((f1 - f0))" -le 0 ]; then
7582 echo "no frame presented in ${secs}s: the page did not animate or neferwl stalled" >&2
7683 tail -n 20 "$tmp/nwl.out" >&2
7784 exit 1
7885 fi
7986
80-python3 - "$((t1 - t0))" "$((f1 - f0))" "$((c1 - c0))" "$secs" "$page" <<'EOF'
87+python3 - "$((t1 - t0))" "$((f1 - f0))" "$((c1 - c0))" "$secs" "$page" "$th0" "$th1" <<'EOF'
8188 import sys
8289 t, f, c, s, page = int(sys.argv[1]), int(sys.argv[2]), int(sys.argv[3]), float(sys.argv[4]), sys.argv[5]
90+def parse(x):
91+ return {l.split()[0]: tuple(int(v) for v in l.split()[1:]) for l in x.splitlines() if len(l.split()) == 3}
92+a, b = parse(sys.argv[6]), parse(sys.argv[7])
93+d = {k: (b[k][0] - a.get(k, (0, 0))[0], b[k][1] - a.get(k, (0, 0))[1]) for k in b}
94+top = max(d, key=lambda k: d[k][1])
8395 cpu = t / 1e9 / s * 100
8496 fr = max(f, 1)
85-print(f"{page} cpu%={cpu:.2f} fps={f / s:.1f} ms/frame={t / 1e6 / fr:.3f} cs/frame={c / fr:.0f}")
97+print(f"{page} cpu%={cpu:.2f} fps={f / s:.1f} ms/frame={t / 1e6 / fr:.3f} cs/frame={c / fr:.0f} cs/s={c / s:.0f} top_tid={top} top_wakes/s={d[top][0] / s:.0f} top_vol/s={d[top][1] / s:.0f}")
8698 EOF
8799
88100 if [ -n "${BENCH_KEEP:-}" ]; then
@@ -7,12 +7,14 @@
7 # BENCH_DEBUG=cat extra --debug categories (comma separated)7 # BENCH_DEBUG=cat extra --debug categories (comma separated)
8 # BENCH_KEEP=1 leave neferwl and chromium running (prints PIDs)8 # BENCH_KEEP=1 leave neferwl and chromium running (prints PIDs)
9 # BENCH_PPROF=addr pprof listen address (default localhost:6061)9 # BENCH_PPROF=addr pprof listen address (default localhost:6061)
10+# BENCH_ROOT=dir build neferwl from another worktree (A/B runs)
11+# BENCH_IDLE=1 static page: no frames expected, per-second numbers only
10 set -euo pipefail12 set -euo pipefail
11 13
12 page=${1:?usage: bench.sh <page.html> <seconds>}14 page=${1:?usage: bench.sh <page.html> <seconds>}
13 secs=${2:?usage: bench.sh <page.html> <seconds>}15 secs=${2:?usage: bench.sh <page.html> <seconds>}
14 here=$(cd "$(dirname "$0")" && pwd)16 here=$(cd "$(dirname "$0")" && pwd)
15-root=$(cd "$here/../.." && pwd)17+root=${BENCH_ROOT:-$(cd "$here/../.." && pwd)}
16 tmp=$(mktemp -d)18 tmp=$(mktemp -d)
17 pids=()19 pids=()
18 20
@@ -67,22 +69,32 @@ cs() { cat /proc/"$nwl"/task/*/status | awk '/^voluntary_ctxt/{v+=$2} END{print
67 cpu_ns() { cat /proc/"$nwl"/task/*/schedstat | awk '{v+=$1} END{printf "%d\n", v}'; }69 cpu_ns() { cat /proc/"$nwl"/task/*/schedstat | awk '{v+=$1} END{printf "%d\n", v}'; }
68 frames() { grep -c '"message":"frame"' "$log" || true; }70 frames() { grep -c '"message":"frame"' "$log" || true; }
69 71
70-t0=$(cpu_ns); f0=$(frames); c0=$(cs)72+# Per-thread "tid sched_in_count voluntary_switches" (schedstat field 3 = times scheduled in).
73+threads() { for d in /proc/"$nwl"/task/*; do
74+ echo "${d##*/} $(awk '{print $3}' "$d/schedstat" 2>/dev/null) $(awk '/^voluntary_ctxt/{print $2}' "$d/status" 2>/dev/null)"
75+done; }
76+
77+th0=$(threads); t0=$(cpu_ns); f0=$(frames); c0=$(cs)
71 sleep "$secs"78 sleep "$secs"
72-t1=$(cpu_ns); f1=$(frames); c1=$(cs)79+th1=$(threads); t1=$(cpu_ns); f1=$(frames); c1=$(cs)
73 80
74-if [ "$((f1 - f0))" -le 0 ]; then81+if [ -z "${BENCH_IDLE:-}" ] && [ "$((f1 - f0))" -le 0 ]; then
75 echo "no frame presented in ${secs}s: the page did not animate or neferwl stalled" >&282 echo "no frame presented in ${secs}s: the page did not animate or neferwl stalled" >&2
76 tail -n 20 "$tmp/nwl.out" >&283 tail -n 20 "$tmp/nwl.out" >&2
77 exit 184 exit 1
78 fi85 fi
79 86
80-python3 - "$((t1 - t0))" "$((f1 - f0))" "$((c1 - c0))" "$secs" "$page" <<'EOF'87+python3 - "$((t1 - t0))" "$((f1 - f0))" "$((c1 - c0))" "$secs" "$page" "$th0" "$th1" <<'EOF'
81 import sys88 import sys
82 t, f, c, s, page = int(sys.argv[1]), int(sys.argv[2]), int(sys.argv[3]), float(sys.argv[4]), sys.argv[5]89 t, f, c, s, page = int(sys.argv[1]), int(sys.argv[2]), int(sys.argv[3]), float(sys.argv[4]), sys.argv[5]
90+def parse(x):
91+ return {l.split()[0]: tuple(int(v) for v in l.split()[1:]) for l in x.splitlines() if len(l.split()) == 3}
92+a, b = parse(sys.argv[6]), parse(sys.argv[7])
93+d = {k: (b[k][0] - a.get(k, (0, 0))[0], b[k][1] - a.get(k, (0, 0))[1]) for k in b}
94+top = max(d, key=lambda k: d[k][1])
83 cpu = t / 1e9 / s * 10095 cpu = t / 1e9 / s * 100
84 fr = max(f, 1)96 fr = max(f, 1)
85-print(f"{page} cpu%={cpu:.2f} fps={f / s:.1f} ms/frame={t / 1e6 / fr:.3f} cs/frame={c / fr:.0f}")97+print(f"{page} cpu%={cpu:.2f} fps={f / s:.1f} ms/frame={t / 1e6 / fr:.3f} cs/frame={c / fr:.0f} cs/s={c / s:.0f} top_tid={top} top_wakes/s={d[top][0] / s:.0f} top_vol/s={d[top][1] / s:.0f}")
86 EOF98 EOF
87 99
88 if [ -n "${BENCH_KEEP:-}" ]; then100 if [ -n "${BENCH_KEEP:-}" ]; then
A tools/bench/idle.html +3 -0
@@ -0,0 +1,3 @@
1+<!doctype html>
2+<html><head><meta charset="utf-8"><title>idle</title></head>
3+<body style="margin:0;background:#202030;color:#ddd;font:24px sans-serif"><p>NeferWL idle benchmark: static page, nothing animates.</p></body></html>
@@ -0,0 +1,3 @@
1+<!doctype html>
2+<html><head><meta charset="utf-8"><title>idle</title></head>
3+<body style="margin:0;background:#202030;color:#ddd;font:24px sans-serif"><p>NeferWL idle benchmark: static page, nothing animates.</p></body></html>

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 23

Tags 10

↑↓ moveenter openesc close