perf(adapters): sleep until work instead of 100 ms polls
DRM events, udev hotplug, libinput, the seat, the capture allowlist watcher and the sync waiter pin their long-lived fd and wait with no timeout, ending on ctx through StopOn. libinput control messages are relayed to its loop with a wake; its next wake is the 10 s stats deadline. The seat loop dispatches after each request that may queue libseat events (open, close, VT switch), woken by the request itself. The allowlist watcher sleeps until its retry or debounce deadline.
- commit
- 0c5c351d0c0f2e4eea6be8144cfebfb36c903602
- author
- brice · · 2026-10-14 04:06 UTC
- committer
- brice · · 2026-10-14 04:35 UTC
- parent
- 182dc8b
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
7 files · +176 -50
M internal/adapters/captureallow/store.go +12 -4
| @@ -192,6 +192,11 @@ func (s *Store) Run(ctx context.Context) error { | |||
| 192 | 192 | return err | |
| 193 | 193 | } | |
| 194 | 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}} | |
| 195 | 200 | dir, name := filepath.Dir(s.path), filepath.Base(s.path) | |
| 196 | 201 | wd := -1 | |
| 197 | 202 | var retry, pending time.Time | |
| @@ -215,9 +220,10 @@ func (s *Store) Run(ctx context.Context) error { | |||
| 215 | 220 | return err | |
| 216 | 221 | } | |
| 217 | 222 | } | |
| 218 | - | timeout := 200 | |
| 223 | + | // Sleep until an event, the shutdown, or the next deadline. | |
| 224 | + | timeout := -1 | |
| 219 | 225 | if !retry.IsZero() { | |
| 220 | - | timeout = min(timeout, int(time.Until(retry).Milliseconds())+1) | |
| 226 | + | timeout = max(0, int(time.Until(retry).Milliseconds())+1) | |
| 221 | 227 | } | |
| 222 | 228 | if !pending.IsZero() { | |
| 223 | 229 | remaining := time.Until(pending) | |
| @@ -227,9 +233,11 @@ func (s *Store) Run(ctx context.Context) error { | |||
| 227 | 233 | quiet = false | |
| 228 | 234 | continue | |
| 229 | 235 | } | |
| 230 | - | timeout = min(timeout, int(remaining.Milliseconds())+1) | |
| 236 | + | if ms := int(remaining.Milliseconds()) + 1; timeout < 0 || ms < timeout { | |
| 237 | + | timeout = ms | |
| 238 | + | } | |
| 231 | 239 | } | |
| 232 | - | fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}} | |
| 240 | + | fds[0].Revents = 0 | |
| 233 | 241 | if _, err := waiter.Poll(fds, timeout); err != nil { | |
| 234 | 242 | if err == unix.EINTR { | |
| 235 | 243 | continue | |
| @@ -192,6 +192,11 @@ func (s *Store) Run(ctx context.Context) error { | |||||
| 192 | return err | 192 | return err | ||
| 193 | } | 193 | } | ||
| 194 | defer waiter.Close() | 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}} | |||
| 195 | dir, name := filepath.Dir(s.path), filepath.Base(s.path) | 200 | dir, name := filepath.Dir(s.path), filepath.Base(s.path) | ||
| 196 | wd := -1 | 201 | wd := -1 | ||
| 197 | var retry, pending time.Time | 202 | var retry, pending time.Time | ||
| @@ -215,9 +220,10 @@ func (s *Store) Run(ctx context.Context) error { | |||||
| 215 | return err | 220 | return err | ||
| 216 | } | 221 | } | ||
| 217 | } | 222 | } | ||
| 218 | - | timeout := 200 | 223 | + | // Sleep until an event, the shutdown, or the next deadline. |
| 224 | + | timeout := -1 | |||
| 219 | if !retry.IsZero() { | 225 | if !retry.IsZero() { | ||
| 220 | - | timeout = min(timeout, int(time.Until(retry).Milliseconds())+1) | 226 | + | timeout = max(0, int(time.Until(retry).Milliseconds())+1) |
| 221 | } | 227 | } | ||
| 222 | if !pending.IsZero() { | 228 | if !pending.IsZero() { | ||
| 223 | remaining := time.Until(pending) | 229 | remaining := time.Until(pending) | ||
| @@ -227,9 +233,11 @@ func (s *Store) Run(ctx context.Context) error { | |||||
| 227 | quiet = false | 233 | quiet = false | ||
| 228 | continue | 234 | continue | ||
| 229 | } | 235 | } | ||
| 230 | - | timeout = min(timeout, int(remaining.Milliseconds())+1) | 236 | + | if ms := int(remaining.Milliseconds()) + 1; timeout < 0 || ms < timeout { |
| 237 | + | timeout = ms | |||
| 238 | + | } | |||
| 231 | } | 239 | } | ||
| 232 | - | fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}} | 240 | + | fds[0].Revents = 0 |
| 233 | if _, err := waiter.Poll(fds, timeout); err != nil { | 241 | if _, err := waiter.Poll(fds, timeout); err != nil { | ||
| 234 | if err == unix.EINTR { | 242 | if err == unix.EINTR { | ||
| 235 | continue | 243 | continue | ||
M internal/adapters/drm/card.go +6 -2
| @@ -253,11 +253,15 @@ func (c *Card) ReadEvents(ctx context.Context) error { | |||
| 253 | 253 | return fmt.Errorf("drm event waiter: %w", err) | |
| 254 | 254 | } | |
| 255 | 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 | + | } | |
| 256 | 260 | buf := make([]byte, 1024) | |
| 257 | 261 | var events []flipEvent // reused by every read | |
| 262 | + | fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}} | |
| 258 | 263 | for ctx.Err() == nil { | |
| 259 | - | fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}} | |
| 260 | - | n, err := waiter.Poll(fds, 100) | |
| 264 | + | n, err := waiter.Poll(fds, -1) | |
| 261 | 265 | if err != nil && !errors.Is(err, unix.EINTR) { | |
| 262 | 266 | return fmt.Errorf("drm poll: %w", err) | |
| 263 | 267 | } | |
| @@ -253,11 +253,15 @@ func (c *Card) ReadEvents(ctx context.Context) error { | |||||
| 253 | return fmt.Errorf("drm event waiter: %w", err) | 253 | return fmt.Errorf("drm event waiter: %w", err) | ||
| 254 | } | 254 | } | ||
| 255 | defer waiter.Close() | 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 | + | } | |||
| 256 | buf := make([]byte, 1024) | 260 | buf := make([]byte, 1024) | ||
| 257 | var events []flipEvent // reused by every read | 261 | var events []flipEvent // reused by every read | ||
| 262 | + | fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}} | |||
| 258 | for ctx.Err() == nil { | 263 | for ctx.Err() == nil { | ||
| 259 | - | fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}} | 264 | + | n, err := waiter.Poll(fds, -1) |
| 260 | - | n, err := waiter.Poll(fds, 100) | |||
| 261 | if err != nil && !errors.Is(err, unix.EINTR) { | 265 | if err != nil && !errors.Is(err, unix.EINTR) { | ||
| 262 | return fmt.Errorf("drm poll: %w", err) | 266 | return fmt.Errorf("drm poll: %w", err) | ||
| 263 | } | 267 | } | ||
M internal/adapters/drm/hotplug.go +6 -2
| @@ -83,9 +83,13 @@ func WatchHotplug(ctx context.Context, changed chan<- struct{}) error { | |||
| 83 | 83 | return fmt.Errorf("udev waiter: %w", err) | |
| 84 | 84 | } | |
| 85 | 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}} | |
| 86 | 91 | for ctx.Err() == nil { | |
| 87 | - | fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}} | |
| 88 | - | n, err := waiter.Poll(fds, 100) | |
| 92 | + | n, err := waiter.Poll(fds, -1) | |
| 89 | 93 | if err != nil && !errors.Is(err, unix.EINTR) { | |
| 90 | 94 | return fmt.Errorf("udev poll: %w", err) | |
| 91 | 95 | } | |
| @@ -83,9 +83,13 @@ func WatchHotplug(ctx context.Context, changed chan<- struct{}) error { | |||||
| 83 | return fmt.Errorf("udev waiter: %w", err) | 83 | return fmt.Errorf("udev waiter: %w", err) | ||
| 84 | } | 84 | } | ||
| 85 | defer waiter.Close() | 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}} | |||
| 86 | for ctx.Err() == nil { | 91 | for ctx.Err() == nil { | ||
| 87 | - | fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}} | 92 | + | n, err := waiter.Poll(fds, -1) |
| 88 | - | n, err := waiter.Poll(fds, 100) | |||
| 89 | if err != nil && !errors.Is(err, unix.EINTR) { | 93 | if err != nil && !errors.Is(err, unix.EINTR) { | ||
| 90 | return fmt.Errorf("udev poll: %w", err) | 94 | return fmt.Errorf("udev poll: %w", err) | ||
| 91 | } | 95 | } | ||
M internal/adapters/libinput/libinput.go +78 -34
| @@ -237,6 +237,64 @@ type Options struct { | |||
| 237 | 237 | LogKeys bool | |
| 238 | 238 | } | |
| 239 | 239 | ||
| 240 | + | // relayControls hands the control messages of opts to Run as closures on | |
| 241 | + | // controls, waking its poll for each, until ctx ends. It only reads the | |
| 242 | + | // channels of opts; the closures run on Run's goroutine, the only one to | |
| 243 | + | // touch p, in, li and opts.Keymap. | |
| 244 | + | func relayControls(ctx context.Context, opts *Options, p *pointer, in *inputState, li uintptr, controls chan<- func(), wake func()) { | |
| 245 | + | for { | |
| 246 | + | var f func() | |
| 247 | + | select { | |
| 248 | + | case <-ctx.Done(): | |
| 249 | + | return | |
| 250 | + | case on := <-opts.Active: | |
| 251 | + | f = func() { | |
| 252 | + | if on { | |
| 253 | + | resume(li) | |
| 254 | + | } else { | |
| 255 | + | suspend(li) | |
| 256 | + | } | |
| 257 | + | opts.Log.Info().Bool("active", on).Msg("input") | |
| 258 | + | } | |
| 259 | + | case km := <-opts.Keymaps: | |
| 260 | + | f = func() { | |
| 261 | + | if opts.Security != nil { | |
| 262 | + | km.QuarantineFrom(opts.Keymap) | |
| 263 | + | } | |
| 264 | + | opts.Keymap.Close() | |
| 265 | + | opts.Keymap = km | |
| 266 | + | opts.Log.Info().Msg("keymap replaced") | |
| 267 | + | } | |
| 268 | + | case l := <-opts.Layouts: | |
| 269 | + | f = func() { | |
| 270 | + | p.setLayout(l) | |
| 271 | + | p.moved(opts.MoveCursor, false) | |
| 272 | + | } | |
| 273 | + | case c := <-opts.Constraints: | |
| 274 | + | f = func() { | |
| 275 | + | p.constrain(c) | |
| 276 | + | p.moved(opts.MoveCursor, false) | |
| 277 | + | } | |
| 278 | + | case c := <-opts.DeviceConfigs: | |
| 279 | + | f = func() { | |
| 280 | + | in.cfg = c | |
| 281 | + | for dev := range in.devices { | |
| 282 | + | in.configure(dev, opts.Log) | |
| 283 | + | } | |
| 284 | + | t, m := c.Touchpad, c.Mouse | |
| 285 | + | 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). | |
| 286 | + | 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") | |
| 287 | + | } | |
| 288 | + | } | |
| 289 | + | select { | |
| 290 | + | case controls <- f: | |
| 291 | + | wake() | |
| 292 | + | case <-ctx.Done(): | |
| 293 | + | return | |
| 294 | + | } | |
| 295 | + | } | |
| 296 | + | } | |
| 297 | + | ||
| 240 | 298 | // statsEvery is how often Run logs input throughput. | |
| 241 | 299 | const statsEvery = 10 * time.Second | |
| 242 | 300 | ||
| @@ -273,53 +331,39 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error | |||
| 273 | 331 | return fmt.Errorf("libinput waiter: %w", err) | |
| 274 | 332 | } | |
| 275 | 333 | defer waiter.Close() | |
| 334 | + | waiter.StopOn(ctx) | |
| 276 | 335 | fd := getFD(li) | |
| 336 | + | if err := waiter.Pin(fd); err != nil { | |
| 337 | + | return fmt.Errorf("libinput waiter: %w", err) | |
| 338 | + | } | |
| 277 | 339 | fwd := newForwarder(opts.Log) | |
| 278 | 340 | fwdCtx, stopFwd := context.WithCancel(ctx) | |
| 279 | 341 | var fwdDone sync.WaitGroup | |
| 280 | 342 | fwdDone.Go(func() { fwd.run(fwdCtx, input) }) | |
| 343 | + | // Control messages wake the poll instead of a polling timeout; they are | |
| 344 | + | // applied on this goroutine, which owns libinput. | |
| 345 | + | controls := make(chan func(), 8) | |
| 346 | + | fwdDone.Go(func() { relayControls(fwdCtx, &opts, p, in, li, controls, waiter.Wake) }) | |
| 281 | 347 | defer func() { stopFwd(); fwdDone.Wait() }() | |
| 282 | - | lastStats := time.Now() | |
| 348 | + | fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}} | |
| 349 | + | nextStats := time.Now().Add(statsEvery) | |
| 283 | 350 | for ctx.Err() == nil { | |
| 284 | - | if time.Since(lastStats) >= statsEvery { | |
| 285 | - | lastStats = time.Now() | |
| 351 | + | if !time.Now().Before(nextStats) { | |
| 352 | + | nextStats = time.Now().Add(statsEvery) | |
| 286 | 353 | if s := fwd.take(); s.Motions > 0 || s.Sent > 0 { | |
| 287 | 354 | opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats") | |
| 288 | 355 | } | |
| 289 | 356 | } | |
| 290 | - | select { | |
| 291 | - | case on := <-opts.Active: | |
| 292 | - | if on { | |
| 293 | - | resume(li) | |
| 294 | - | } else { | |
| 295 | - | suspend(li) | |
| 296 | - | } | |
| 297 | - | opts.Log.Info().Bool("active", on).Msg("input") | |
| 298 | - | case km := <-opts.Keymaps: | |
| 299 | - | if opts.Security != nil { | |
| 300 | - | km.QuarantineFrom(opts.Keymap) | |
| 301 | - | } | |
| 302 | - | opts.Keymap.Close() | |
| 303 | - | opts.Keymap = km | |
| 304 | - | opts.Log.Info().Msg("keymap replaced") | |
| 305 | - | case l := <-opts.Layouts: | |
| 306 | - | p.setLayout(l) | |
| 307 | - | p.moved(opts.MoveCursor, false) | |
| 308 | - | case c := <-opts.Constraints: | |
| 309 | - | p.constrain(c) | |
| 310 | - | p.moved(opts.MoveCursor, false) | |
| 311 | - | case c := <-opts.DeviceConfigs: | |
| 312 | - | in.cfg = c | |
| 313 | - | for dev := range in.devices { | |
| 314 | - | in.configure(dev, opts.Log) | |
| 357 | + | apply: | |
| 358 | + | for { | |
| 359 | + | select { | |
| 360 | + | case f := <-controls: | |
| 361 | + | f() | |
| 362 | + | default: | |
| 363 | + | break apply | |
| 315 | 364 | } | |
| 316 | - | t, m := c.Touchpad, c.Mouse | |
| 317 | - | 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). | |
| 318 | - | 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") | |
| 319 | - | default: | |
| 320 | 365 | } | |
| 321 | - | fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}} | |
| 322 | - | n, err := waiter.Poll(fds, 100) | |
| 366 | + | n, err := waiter.Poll(fds, int(time.Until(nextStats).Milliseconds())+1) | |
| 323 | 367 | if err != nil && !errors.Is(err, unix.EINTR) { | |
| 324 | 368 | return fmt.Errorf("libinput poll: %w", err) | |
| 325 | 369 | } | |
| @@ -237,6 +237,64 @@ type Options struct { | |||||
| 237 | LogKeys bool | 237 | LogKeys bool | ||
| 238 | } | 238 | } | ||
| 239 | 239 | ||||
| 240 | + | // relayControls hands the control messages of opts to Run as closures on | |||
| 241 | + | // controls, waking its poll for each, until ctx ends. It only reads the | |||
| 242 | + | // channels of opts; the closures run on Run's goroutine, the only one to | |||
| 243 | + | // touch p, in, li and opts.Keymap. | |||
| 244 | + | func relayControls(ctx context.Context, opts *Options, p *pointer, in *inputState, li uintptr, controls chan<- func(), wake func()) { | |||
| 245 | + | for { | |||
| 246 | + | var f func() | |||
| 247 | + | select { | |||
| 248 | + | case <-ctx.Done(): | |||
| 249 | + | return | |||
| 250 | + | case on := <-opts.Active: | |||
| 251 | + | f = func() { | |||
| 252 | + | if on { | |||
| 253 | + | resume(li) | |||
| 254 | + | } else { | |||
| 255 | + | suspend(li) | |||
| 256 | + | } | |||
| 257 | + | opts.Log.Info().Bool("active", on).Msg("input") | |||
| 258 | + | } | |||
| 259 | + | case km := <-opts.Keymaps: | |||
| 260 | + | f = func() { | |||
| 261 | + | if opts.Security != nil { | |||
| 262 | + | km.QuarantineFrom(opts.Keymap) | |||
| 263 | + | } | |||
| 264 | + | opts.Keymap.Close() | |||
| 265 | + | opts.Keymap = km | |||
| 266 | + | opts.Log.Info().Msg("keymap replaced") | |||
| 267 | + | } | |||
| 268 | + | case l := <-opts.Layouts: | |||
| 269 | + | f = func() { | |||
| 270 | + | p.setLayout(l) | |||
| 271 | + | p.moved(opts.MoveCursor, false) | |||
| 272 | + | } | |||
| 273 | + | case c := <-opts.Constraints: | |||
| 274 | + | f = func() { | |||
| 275 | + | p.constrain(c) | |||
| 276 | + | p.moved(opts.MoveCursor, false) | |||
| 277 | + | } | |||
| 278 | + | case c := <-opts.DeviceConfigs: | |||
| 279 | + | f = func() { | |||
| 280 | + | in.cfg = c | |||
| 281 | + | for dev := range in.devices { | |||
| 282 | + | in.configure(dev, opts.Log) | |||
| 283 | + | } | |||
| 284 | + | t, m := c.Touchpad, c.Mouse | |||
| 285 | + | 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). | |||
| 286 | + | 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") | |||
| 287 | + | } | |||
| 288 | + | } | |||
| 289 | + | select { | |||
| 290 | + | case controls <- f: | |||
| 291 | + | wake() | |||
| 292 | + | case <-ctx.Done(): | |||
| 293 | + | return | |||
| 294 | + | } | |||
| 295 | + | } | |||
| 296 | + | } | |||
| 297 | + | ||||
| 240 | // statsEvery is how often Run logs input throughput. | 298 | // statsEvery is how often Run logs input throughput. | ||
| 241 | const statsEvery = 10 * time.Second | 299 | const statsEvery = 10 * time.Second | ||
| 242 | 300 | ||||
| @@ -273,53 +331,39 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error | |||||
| 273 | return fmt.Errorf("libinput waiter: %w", err) | 331 | return fmt.Errorf("libinput waiter: %w", err) | ||
| 274 | } | 332 | } | ||
| 275 | defer waiter.Close() | 333 | defer waiter.Close() | ||
| 334 | + | waiter.StopOn(ctx) | |||
| 276 | fd := getFD(li) | 335 | fd := getFD(li) | ||
| 336 | + | if err := waiter.Pin(fd); err != nil { | |||
| 337 | + | return fmt.Errorf("libinput waiter: %w", err) | |||
| 338 | + | } | |||
| 277 | fwd := newForwarder(opts.Log) | 339 | fwd := newForwarder(opts.Log) | ||
| 278 | fwdCtx, stopFwd := context.WithCancel(ctx) | 340 | fwdCtx, stopFwd := context.WithCancel(ctx) | ||
| 279 | var fwdDone sync.WaitGroup | 341 | var fwdDone sync.WaitGroup | ||
| 280 | fwdDone.Go(func() { fwd.run(fwdCtx, input) }) | 342 | fwdDone.Go(func() { fwd.run(fwdCtx, input) }) | ||
| 343 | + | // Control messages wake the poll instead of a polling timeout; they are | |||
| 344 | + | // applied on this goroutine, which owns libinput. | |||
| 345 | + | controls := make(chan func(), 8) | |||
| 346 | + | fwdDone.Go(func() { relayControls(fwdCtx, &opts, p, in, li, controls, waiter.Wake) }) | |||
| 281 | defer func() { stopFwd(); fwdDone.Wait() }() | 347 | defer func() { stopFwd(); fwdDone.Wait() }() | ||
| 282 | - | lastStats := time.Now() | 348 | + | fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}} |
| 349 | + | nextStats := time.Now().Add(statsEvery) | |||
| 283 | for ctx.Err() == nil { | 350 | for ctx.Err() == nil { | ||
| 284 | - | if time.Since(lastStats) >= statsEvery { | 351 | + | if !time.Now().Before(nextStats) { |
| 285 | - | lastStats = time.Now() | 352 | + | nextStats = time.Now().Add(statsEvery) |
| 286 | if s := fwd.take(); s.Motions > 0 || s.Sent > 0 { | 353 | if s := fwd.take(); s.Motions > 0 || s.Sent > 0 { | ||
| 287 | opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats") | 354 | opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats") | ||
| 288 | } | 355 | } | ||
| 289 | } | 356 | } | ||
| 290 | - | select { | 357 | + | apply: |
| 291 | - | case on := <-opts.Active: | 358 | + | for { |
| 292 | - | if on { | 359 | + | select { |
| 293 | - | resume(li) | 360 | + | case f := <-controls: |
| 294 | - | } else { | 361 | + | f() |
| 295 | - | suspend(li) | 362 | + | default: |
| 296 | - | } | 363 | + | break apply |
| 297 | - | opts.Log.Info().Bool("active", on).Msg("input") | |||
| 298 | - | case km := <-opts.Keymaps: | |||
| 299 | - | if opts.Security != nil { | |||
| 300 | - | km.QuarantineFrom(opts.Keymap) | |||
| 301 | - | } | |||
| 302 | - | opts.Keymap.Close() | |||
| 303 | - | opts.Keymap = km | |||
| 304 | - | opts.Log.Info().Msg("keymap replaced") | |||
| 305 | - | case l := <-opts.Layouts: | |||
| 306 | - | p.setLayout(l) | |||
| 307 | - | p.moved(opts.MoveCursor, false) | |||
| 308 | - | case c := <-opts.Constraints: | |||
| 309 | - | p.constrain(c) | |||
| 310 | - | p.moved(opts.MoveCursor, false) | |||
| 311 | - | case c := <-opts.DeviceConfigs: | |||
| 312 | - | in.cfg = c | |||
| 313 | - | for dev := range in.devices { | |||
| 314 | - | in.configure(dev, opts.Log) | |||
| 315 | } | 364 | } | ||
| 316 | - | t, m := c.Touchpad, c.Mouse | |||
| 317 | - | 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). | |||
| 318 | - | 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") | |||
| 319 | - | default: | |||
| 320 | } | 365 | } | ||
| 321 | - | fds := []unix.PollFd{{Fd: fd, Events: unix.POLLIN}} | 366 | + | n, err := waiter.Poll(fds, int(time.Until(nextStats).Milliseconds())+1) |
| 322 | - | n, err := waiter.Poll(fds, 100) | |||
| 323 | if err != nil && !errors.Is(err, unix.EINTR) { | 367 | if err != nil && !errors.Is(err, unix.EINTR) { | ||
| 324 | return fmt.Errorf("libinput poll: %w", err) | 368 | return fmt.Errorf("libinput poll: %w", err) | ||
| 325 | } | 369 | } | ||
M internal/adapters/seat/seat.go +34 -5
| @@ -76,6 +76,9 @@ type Seat struct { | |||
| 76 | 76 | // manager: the last moment this process is still DRM master. | |
| 77 | 77 | beforeDisable func() | |
| 78 | 78 | disabler disabler | |
| 79 | + | // waiter is Run's poller: calls that make libseat queue events wake it | |
| 80 | + | // to dispatch them. | |
| 81 | + | waiter *fdwait.Poller | |
| 79 | 82 | } | |
| 80 | 83 | ||
| 81 | 84 | // disabler acks a seat disable to the seat manager (libseat_disable_seat), | |
| @@ -190,34 +193,58 @@ func (s *Seat) dispatchLocked(ctx context.Context, timeoutMs int) error { | |||
| 190 | 193 | ||
| 191 | 194 | // Run dispatches seat events until ctx ends. The owner calls Close afterwards, | |
| 192 | 195 | // once outputs have restored their CRTC state. | |
| 196 | + | // It sleeps until the seat fd is readable or a seat call wakes it. | |
| 193 | 197 | func (s *Seat) Run(ctx context.Context) error { | |
| 194 | 198 | waiter, err := fdwait.New() | |
| 195 | 199 | if err != nil { | |
| 196 | 200 | return fmt.Errorf("seat waiter: %w", err) | |
| 197 | 201 | } | |
| 198 | 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}} | |
| 199 | 217 | for ctx.Err() == nil { | |
| 200 | - | fds := []unix.PollFd{{Fd: getFD(s.handle), Events: unix.POLLIN}} | |
| 201 | - | if _, err := waiter.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) { | |
| 202 | - | return fmt.Errorf("seat poll: %w", err) | |
| 203 | - | } | |
| 204 | - | // 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). | |
| 205 | 221 | s.mu.Lock() | |
| 206 | 222 | r := dispatch(s.handle, 0) | |
| 207 | 223 | s.mu.Unlock() | |
| 208 | 224 | if r < 0 { | |
| 209 | 225 | return errors.New("libseat_dispatch failed") | |
| 210 | 226 | } | |
| 227 | + | if _, err := waiter.Poll(fds, -1); err != nil && !errors.Is(err, unix.EINTR) { | |
| 228 | + | return fmt.Errorf("seat poll: %w", err) | |
| 229 | + | } | |
| 211 | 230 | } | |
| 212 | 231 | return nil | |
| 213 | 232 | } | |
| 214 | 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 | + | ||
| 215 | 241 | // OpenDevice opens path through the seat manager and returns its fd. | |
| 216 | 242 | func (s *Seat) OpenDevice(path string) (int, error) { | |
| 217 | 243 | s.mu.Lock() | |
| 218 | 244 | defer s.mu.Unlock() | |
| 219 | 245 | var fd int32 = -1 | |
| 220 | 246 | id := openDevice(s.handle, path, &fd) | |
| 247 | + | s.wakeLocked() | |
| 221 | 248 | if id < 0 { | |
| 222 | 249 | s.log.Warn().Str("path", path).Msg("open device failed") | |
| 223 | 250 | return -1, fmt.Errorf("libseat_open_device %s failed", path) | |
| @@ -237,6 +264,7 @@ func (s *Seat) CloseDevice(fd int) { | |||
| 237 | 264 | } | |
| 238 | 265 | delete(s.devices, int32(fd)) | |
| 239 | 266 | closeDevice(s.handle, id) | |
| 267 | + | s.wakeLocked() | |
| 240 | 268 | _ = unix.Close(fd) | |
| 241 | 269 | s.log.Info().Int("fd", fd).Msg("device closed") | |
| 242 | 270 | } | |
| @@ -249,6 +277,7 @@ func (s *Seat) SwitchVT(vt int) { | |||
| 249 | 277 | if switchSession(s.handle, int32(vt)) < 0 { | |
| 250 | 278 | s.log.Warn().Int("vt", vt).Msg("switch vt failed") | |
| 251 | 279 | } | |
| 280 | + | s.wakeLocked() | |
| 252 | 281 | } | |
| 253 | 282 | ||
| 254 | 283 | // Close releases all devices and the seat; the VT returns to text mode. Safe to call twice. | |
| @@ -76,6 +76,9 @@ type Seat struct { | |||||
| 76 | // manager: the last moment this process is still DRM master. | 76 | // manager: the last moment this process is still DRM master. | ||
| 77 | beforeDisable func() | 77 | beforeDisable func() | ||
| 78 | disabler disabler | 78 | disabler disabler | ||
| 79 | + | // waiter is Run's poller: calls that make libseat queue events wake it | |||
| 80 | + | // to dispatch them. | |||
| 81 | + | waiter *fdwait.Poller | |||
| 79 | } | 82 | } | ||
| 80 | 83 | ||||
| 81 | // 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), | ||
| @@ -190,34 +193,58 @@ func (s *Seat) dispatchLocked(ctx context.Context, timeoutMs int) error { | |||||
| 190 | 193 | ||||
| 191 | // 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, | ||
| 192 | // 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. | |||
| 193 | func (s *Seat) Run(ctx context.Context) error { | 197 | func (s *Seat) Run(ctx context.Context) error { | ||
| 194 | waiter, err := fdwait.New() | 198 | waiter, err := fdwait.New() | ||
| 195 | if err != nil { | 199 | if err != nil { | ||
| 196 | return fmt.Errorf("seat waiter: %w", err) | 200 | return fmt.Errorf("seat waiter: %w", err) | ||
| 197 | } | 201 | } | ||
| 198 | defer waiter.Close() | 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}} | |||
| 199 | for ctx.Err() == nil { | 217 | for ctx.Err() == nil { | ||
| 200 | - | fds := []unix.PollFd{{Fd: getFD(s.handle), Events: unix.POLLIN}} | 218 | + | // Dispatch on every return, a wake included: libseat may queue |
| 201 | - | if _, err := waiter.Poll(fds, 100); err != nil && !errors.Is(err, unix.EINTR) { | 219 | + | // events while answering a request, without the fd becoming |
| 202 | - | return fmt.Errorf("seat poll: %w", err) | 220 | + | // readable (its backends ping themselves, the wake covers it too). |
| 203 | - | } | |||
| 204 | - | // Dispatch every tick: libseat may queue events without the fd becoming readable. | |||
| 205 | s.mu.Lock() | 221 | s.mu.Lock() | ||
| 206 | r := dispatch(s.handle, 0) | 222 | r := dispatch(s.handle, 0) | ||
| 207 | s.mu.Unlock() | 223 | s.mu.Unlock() | ||
| 208 | if r < 0 { | 224 | if r < 0 { | ||
| 209 | return errors.New("libseat_dispatch failed") | 225 | return errors.New("libseat_dispatch failed") | ||
| 210 | } | 226 | } | ||
| 227 | + | if _, err := waiter.Poll(fds, -1); err != nil && !errors.Is(err, unix.EINTR) { | |||
| 228 | + | return fmt.Errorf("seat poll: %w", err) | |||
| 229 | + | } | |||
| 211 | } | 230 | } | ||
| 212 | return nil | 231 | return nil | ||
| 213 | } | 232 | } | ||
| 214 | 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 | + | ||||
| 215 | // OpenDevice opens path through the seat manager and returns its fd. | 241 | // OpenDevice opens path through the seat manager and returns its fd. | ||
| 216 | func (s *Seat) OpenDevice(path string) (int, error) { | 242 | func (s *Seat) OpenDevice(path string) (int, error) { | ||
| 217 | s.mu.Lock() | 243 | s.mu.Lock() | ||
| 218 | defer s.mu.Unlock() | 244 | defer s.mu.Unlock() | ||
| 219 | var fd int32 = -1 | 245 | var fd int32 = -1 | ||
| 220 | id := openDevice(s.handle, path, &fd) | 246 | id := openDevice(s.handle, path, &fd) | ||
| 247 | + | s.wakeLocked() | |||
| 221 | if id < 0 { | 248 | if id < 0 { | ||
| 222 | s.log.Warn().Str("path", path).Msg("open device failed") | 249 | s.log.Warn().Str("path", path).Msg("open device failed") | ||
| 223 | return -1, fmt.Errorf("libseat_open_device %s failed", path) | 250 | return -1, fmt.Errorf("libseat_open_device %s failed", path) | ||
| @@ -237,6 +264,7 @@ func (s *Seat) CloseDevice(fd int) { | |||||
| 237 | } | 264 | } | ||
| 238 | delete(s.devices, int32(fd)) | 265 | delete(s.devices, int32(fd)) | ||
| 239 | closeDevice(s.handle, id) | 266 | closeDevice(s.handle, id) | ||
| 267 | + | s.wakeLocked() | |||
| 240 | _ = unix.Close(fd) | 268 | _ = unix.Close(fd) | ||
| 241 | s.log.Info().Int("fd", fd).Msg("device closed") | 269 | s.log.Info().Int("fd", fd).Msg("device closed") | ||
| 242 | } | 270 | } | ||
| @@ -249,6 +277,7 @@ func (s *Seat) SwitchVT(vt int) { | |||||
| 249 | if switchSession(s.handle, int32(vt)) < 0 { | 277 | if switchSession(s.handle, int32(vt)) < 0 { | ||
| 250 | s.log.Warn().Int("vt", vt).Msg("switch vt failed") | 278 | s.log.Warn().Int("vt", vt).Msg("switch vt failed") | ||
| 251 | } | 279 | } | ||
| 280 | + | s.wakeLocked() | |||
| 252 | } | 281 | } | ||
| 253 | 282 | ||||
| 254 | // 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 +30 -2
| @@ -120,6 +120,34 @@ 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 | + | done := make(chan struct{}) | |
| 140 | + | go func() { sw.run(ctx); close(done) }() | |
| 141 | + | if timeout := <-polls; timeout != -1 { | |
| 142 | + | t.Fatalf("idle poll timeout %d, want -1", timeout) | |
| 143 | + | } | |
| 144 | + | cancel() | |
| 145 | + | <-done | |
| 146 | + | if n := len(polls); n != 0 { | |
| 147 | + | t.Fatalf("%d extra polls while idle", n) | |
| 148 | + | } | |
| 149 | + | } | |
| 150 | + | ||
| 123 | 151 | func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) { | |
| 124 | 152 | poller := newMockacquirePoller(t) | |
| 125 | 153 | wake := make(chan struct{}, 1) | |
| @@ -135,8 +163,8 @@ func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) { | |||
| 135 | 163 | } | |
| 136 | 164 | active := sw.add(fd) | |
| 137 | 165 | 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) { | |
| 166 | + | poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func(fds []unix.PollFd, _ int) (int, error) { fds[0].Revents = unix.POLLIN; return 1, nil }).Once() | |
| 167 | + | poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func([]unix.PollFd, int) (int, error) { | |
| 140 | 168 | fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC) | |
| 141 | 169 | if err != nil { | |
| 142 | 170 | t.Error(err) | |
| @@ -120,6 +120,34 @@ 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 | + | done := make(chan struct{}) | |||
| 140 | + | go func() { sw.run(ctx); close(done) }() | |||
| 141 | + | if timeout := <-polls; timeout != -1 { | |||
| 142 | + | t.Fatalf("idle poll timeout %d, want -1", timeout) | |||
| 143 | + | } | |||
| 144 | + | cancel() | |||
| 145 | + | <-done | |||
| 146 | + | if n := len(polls); n != 0 { | |||
| 147 | + | t.Fatalf("%d extra polls while idle", n) | |||
| 148 | + | } | |||
| 149 | + | } | |||
| 150 | + | ||||
| 123 | func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) { | 151 | func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) { | ||
| 124 | poller := newMockacquirePoller(t) | 152 | poller := newMockacquirePoller(t) | ||
| 125 | wake := make(chan struct{}, 1) | 153 | wake := make(chan struct{}, 1) | ||
| @@ -135,8 +163,8 @@ func TestAcquirePollTerminalFailureRetiresActiveAndAdded(t *testing.T) { | |||||
| 135 | } | 163 | } | ||
| 136 | active := sw.add(fd) | 164 | active := sw.add(fd) | ||
| 137 | var added *syncWait | 165 | 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() | 166 | + | 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) { | 167 | + | poller.EXPECT().poll(mock.Anything, -1).RunAndReturn(func([]unix.PollFd, int) (int, error) { |
| 140 | fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC) | 168 | fd, err := unix.Eventfd(0, unix.EFD_CLOEXEC) | ||
| 141 | if err != nil { | 169 | if err != nil { | ||
| 142 | t.Error(err) | 170 | t.Error(err) | ||
M internal/adapters/wayland/syncobj.go +10 -1
| @@ -385,6 +385,15 @@ func (sw *syncWaiter) run(ctx context.Context) { | |||
| 385 | 385 | defer sw.pipeR.Close() | |
| 386 | 386 | defer sw.waiter.Close() | |
| 387 | 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 | + | } | |
| 388 | 397 | var waits []*syncWait | |
| 389 | 398 | defer func() { | |
| 390 | 399 | sw.mu.Lock() | |
| @@ -409,7 +418,7 @@ func (sw *syncWaiter) run(ctx context.Context) { | |||
| 409 | 418 | for _, w := range waits { | |
| 410 | 419 | fds = append(fds, unix.PollFd{Fd: int32(w.efd), Events: unix.POLLIN}) | |
| 411 | 420 | } | |
| 412 | - | n, err := sw.poller.poll(fds, 100) | |
| 421 | + | n, err := sw.poller.poll(fds, -1) | |
| 413 | 422 | if err != nil && !errors.Is(err, unix.EINTR) { | |
| 414 | 423 | sw.fail(waits) | |
| 415 | 424 | return | |
| @@ -385,6 +385,15 @@ func (sw *syncWaiter) run(ctx context.Context) { | |||||
| 385 | defer sw.pipeR.Close() | 385 | defer sw.pipeR.Close() | ||
| 386 | defer sw.waiter.Close() | 386 | defer sw.waiter.Close() | ||
| 387 | pipeFD := int(sw.pipeR.Fd()) | 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 | + | } | |||
| 388 | var waits []*syncWait | 397 | var waits []*syncWait | ||
| 389 | defer func() { | 398 | defer func() { | ||
| 390 | sw.mu.Lock() | 399 | sw.mu.Lock() | ||
| @@ -409,7 +418,7 @@ func (sw *syncWaiter) run(ctx context.Context) { | |||||
| 409 | for _, w := range waits { | 418 | for _, w := range waits { | ||
| 410 | 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}) | ||
| 411 | } | 420 | } | ||
| 412 | - | n, err := sw.poller.poll(fds, 100) | 421 | + | n, err := sw.poller.poll(fds, -1) |
| 413 | if err != nil && !errors.Is(err, unix.EINTR) { | 422 | if err != nil && !errors.Is(err, unix.EINTR) { | ||
| 414 | sw.fail(waits) | 423 | sw.fail(waits) | ||
| 415 | return | 424 | return | ||