fix: queue events handled during Roundtrip and Bind
When the pause ends, move the events parked meanwhile into the ring so the next Dispatch sees them; the overflow stays parked for the reader. Also skip wl_output v1 (no done event), stop re-arming a watch after the handler closed the connection, wrap errors with a neferclient prefix, and export NopHandler for handlers to embed.
- commit
- 77d28a9fd0496fa10799458e3276e7064fab0c01
- author
- brice · · 2026-10-01 15:41 UTC
- parent
- 0bae9b7
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +53 -19
M conn.go +23 -12
| @@ -29,7 +29,10 @@ var ( | |||
| 29 | 29 | ErrGlobalNotFound = wlturbo.ErrGlobalNotFound | |
| 30 | 30 | ) | |
| 31 | 31 | ||
| 32 | - | // Output describes one wl_output after its latest wl_output.done. | |
| 32 | + | // Output describes one wl_output after its latest wl_output.done. Only | |
| 33 | + | // wl_output globals of version 2 or later are bound (version 1 has no done | |
| 34 | + | // event, so its description is never known to be complete); older ones are | |
| 35 | + | // ignored. | |
| 33 | 36 | type Output struct { | |
| 34 | 37 | Global uint32 // registry name; identifies the output for its lifetime | |
| 35 | 38 | Name string // wl_output.name, empty before version 4 | |
| @@ -96,7 +99,7 @@ func Connect(ctx context.Context, display string) (*Conn, error) { | |||
| 96 | 99 | } | |
| 97 | 100 | sock, err := (&net.Dialer{Timeout: dialTimeout}).DialContext(ctx, "unix", path) | |
| 98 | 101 | if err != nil { | |
| 99 | - | return nil, fmt.Errorf("failed to connect to Wayland: %w", err) | |
| 102 | + | return nil, fmt.Errorf("neferclient: connect %s: %w", path, err) | |
| 100 | 103 | } | |
| 101 | 104 | return connectConn(ctx, sock) | |
| 102 | 105 | } | |
| @@ -124,7 +127,7 @@ func connectConn(ctx context.Context, sock net.Conn) (res *Conn, err error) { | |||
| 124 | 127 | display, err := wlturbo.ConnectFromConn(sock) | |
| 125 | 128 | if err != nil { | |
| 126 | 129 | _ = sock.Close() | |
| 127 | - | return nil, err | |
| 130 | + | return nil, fmt.Errorf("neferclient: %w", err) | |
| 128 | 131 | } | |
| 129 | 132 | c := &Conn{ | |
| 130 | 133 | sock: sock, | |
| @@ -169,13 +172,13 @@ func connectConn(ctx context.Context, sock net.Conn) (res *Conn, err error) { | |||
| 169 | 172 | // With no reader running, the owner's own roundtrips park every event in | |
| 170 | 173 | // the queue's held list, in order; applying it here is the startup drain. | |
| 171 | 174 | if err = display.Roundtrip(); err != nil { | |
| 172 | - | return nil, err | |
| 175 | + | return nil, fmt.Errorf("neferclient: registry roundtrip: %w", err) | |
| 173 | 176 | } | |
| 174 | 177 | if err = c.applyHeld(); err != nil { | |
| 175 | 178 | return nil, err | |
| 176 | 179 | } | |
| 177 | 180 | if err = display.Roundtrip(); err != nil { // wl_output properties follow the binds | |
| 178 | - | return nil, err | |
| 181 | + | return nil, fmt.Errorf("neferclient: output roundtrip: %w", err) | |
| 179 | 182 | } | |
| 180 | 183 | if err = c.applyHeld(); err != nil { | |
| 181 | 184 | return nil, err | |
| @@ -203,7 +206,7 @@ func (h removeHook) HandleRegistryGlobalRemove(e wl.RegistryGlobalRemoveEvent) { | |||
| 203 | 206 | // setupHandler applies setup events without notifying anyone and keeps the | |
| 204 | 207 | // first failure, which makes Connect fail. | |
| 205 | 208 | type setupHandler struct { | |
| 206 | - | nopHandler | |
| 209 | + | NopHandler | |
| 207 | 210 | err *error | |
| 208 | 211 | } | |
| 209 | 212 | ||
| @@ -238,7 +241,7 @@ func (c *Conn) readLoop() { | |||
| 238 | 241 | continue // a pause request interrupted the read | |
| 239 | 242 | } | |
| 240 | 243 | if !c.q.isStopped() { | |
| 241 | - | c.q.setFatal(err) | |
| 244 | + | c.q.setFatal(fmt.Errorf("neferclient: read: %w", err)) | |
| 242 | 245 | } | |
| 243 | 246 | return | |
| 244 | 247 | } | |
| @@ -291,7 +294,7 @@ func (c *Conn) resume() { | |||
| 291 | 294 | } | |
| 292 | 295 | c.pauseDepth-- | |
| 293 | 296 | if c.pauseDepth == 0 { | |
| 294 | - | c.q.setPausing(false) | |
| 297 | + | c.q.endPause() | |
| 295 | 298 | } | |
| 296 | 299 | } | |
| 297 | 300 | ||
| @@ -322,7 +325,7 @@ func (c *Conn) Dispatch(h Handler) error { | |||
| 322 | 325 | return errors.New("neferclient: Dispatch called from a Handler") | |
| 323 | 326 | } | |
| 324 | 327 | if h == nil { | |
| 325 | - | h = nopHandler{} | |
| 328 | + | h = NopHandler{} | |
| 326 | 329 | } | |
| 327 | 330 | c.dispatching = true | |
| 328 | 331 | defer func() { c.dispatching = false }() | |
| @@ -374,6 +377,9 @@ func (c *Conn) apply(ev *event, h Handler) { | |||
| 374 | 377 | return // unwatched while the event was queued | |
| 375 | 378 | } | |
| 376 | 379 | h.FDReady(id) | |
| 380 | + | if c.closed { | |
| 381 | + | return // the handler closed the connection | |
| 382 | + | } | |
| 377 | 383 | if _, ok := c.watched[ev.fd]; ok { // the handler may have unwatched it | |
| 378 | 384 | if err := c.armFD(int(ev.fd), unix.EPOLL_CTL_MOD); err != nil { | |
| 379 | 385 | h.Error(err) | |
| @@ -392,7 +398,7 @@ func (c *Conn) entry(global uint32) *outputEntry { | |||
| 392 | 398 | } | |
| 393 | 399 | ||
| 394 | 400 | func (c *Conn) bindOutput(global, version uint32) error { | |
| 395 | - | if version == 0 || c.entry(global) != nil { | |
| 401 | + | if version < 2 || c.entry(global) != nil { // v1 has no wl_output.done | |
| 396 | 402 | return nil | |
| 397 | 403 | } | |
| 398 | 404 | o := &outputEntry{pub: Output{Global: global, Scale: 1}, scale: 1} | |
| @@ -472,14 +478,19 @@ func (c *Conn) Outputs() []Output { | |||
| 472 | 478 | } | |
| 473 | 479 | ||
| 474 | 480 | // Roundtrip sends wl_display.sync and dispatches until it returns. The reader | |
| 475 | - | // is paused meanwhile; the events it handled are queued for [Conn.Dispatch]. | |
| 481 | + | // is paused meanwhile; when Roundtrip returns, every event handled before the | |
| 482 | + | // sync reply is queued for [Conn.Dispatch] (up to the ring capacity; the rest | |
| 483 | + | // follows as the reader resumes). | |
| 476 | 484 | func (c *Conn) Roundtrip() error { | |
| 477 | 485 | if c.closed { | |
| 478 | 486 | return ErrClosed | |
| 479 | 487 | } | |
| 480 | 488 | c.pause() | |
| 481 | 489 | defer c.resume() | |
| 482 | - | return c.display.Roundtrip() | |
| 490 | + | if err := c.display.Roundtrip(); err != nil { | |
| 491 | + | return fmt.Errorf("neferclient: roundtrip: %w", err) | |
| 492 | + | } | |
| 493 | + | return nil | |
| 483 | 494 | } | |
| 484 | 495 | ||
| 485 | 496 | // Bind binds the global named iface at min(announced, version) and returns the | |
| @@ -29,7 +29,10 @@ var ( | |||||
| 29 | ErrGlobalNotFound = wlturbo.ErrGlobalNotFound | 29 | ErrGlobalNotFound = wlturbo.ErrGlobalNotFound | ||
| 30 | ) | 30 | ) | ||
| 31 | 31 | ||||
| 32 | - | // Output describes one wl_output after its latest wl_output.done. | 32 | + | // Output describes one wl_output after its latest wl_output.done. Only |
| 33 | + | // wl_output globals of version 2 or later are bound (version 1 has no done | |||
| 34 | + | // event, so its description is never known to be complete); older ones are | |||
| 35 | + | // ignored. | |||
| 33 | type Output struct { | 36 | type Output struct { | ||
| 34 | Global uint32 // registry name; identifies the output for its lifetime | 37 | Global uint32 // registry name; identifies the output for its lifetime | ||
| 35 | Name string // wl_output.name, empty before version 4 | 38 | Name string // wl_output.name, empty before version 4 | ||
| @@ -96,7 +99,7 @@ func Connect(ctx context.Context, display string) (*Conn, error) { | |||||
| 96 | } | 99 | } | ||
| 97 | sock, err := (&net.Dialer{Timeout: dialTimeout}).DialContext(ctx, "unix", path) | 100 | sock, err := (&net.Dialer{Timeout: dialTimeout}).DialContext(ctx, "unix", path) | ||
| 98 | if err != nil { | 101 | if err != nil { | ||
| 99 | - | return nil, fmt.Errorf("failed to connect to Wayland: %w", err) | 102 | + | return nil, fmt.Errorf("neferclient: connect %s: %w", path, err) |
| 100 | } | 103 | } | ||
| 101 | return connectConn(ctx, sock) | 104 | return connectConn(ctx, sock) | ||
| 102 | } | 105 | } | ||
| @@ -124,7 +127,7 @@ func connectConn(ctx context.Context, sock net.Conn) (res *Conn, err error) { | |||||
| 124 | display, err := wlturbo.ConnectFromConn(sock) | 127 | display, err := wlturbo.ConnectFromConn(sock) | ||
| 125 | if err != nil { | 128 | if err != nil { | ||
| 126 | _ = sock.Close() | 129 | _ = sock.Close() | ||
| 127 | - | return nil, err | 130 | + | return nil, fmt.Errorf("neferclient: %w", err) |
| 128 | } | 131 | } | ||
| 129 | c := &Conn{ | 132 | c := &Conn{ | ||
| 130 | sock: sock, | 133 | sock: sock, | ||
| @@ -169,13 +172,13 @@ func connectConn(ctx context.Context, sock net.Conn) (res *Conn, err error) { | |||||
| 169 | // With no reader running, the owner's own roundtrips park every event in | 172 | // With no reader running, the owner's own roundtrips park every event in | ||
| 170 | // the queue's held list, in order; applying it here is the startup drain. | 173 | // the queue's held list, in order; applying it here is the startup drain. | ||
| 171 | if err = display.Roundtrip(); err != nil { | 174 | if err = display.Roundtrip(); err != nil { | ||
| 172 | - | return nil, err | 175 | + | return nil, fmt.Errorf("neferclient: registry roundtrip: %w", err) |
| 173 | } | 176 | } | ||
| 174 | if err = c.applyHeld(); err != nil { | 177 | if err = c.applyHeld(); err != nil { | ||
| 175 | return nil, err | 178 | return nil, err | ||
| 176 | } | 179 | } | ||
| 177 | if err = display.Roundtrip(); err != nil { // wl_output properties follow the binds | 180 | if err = display.Roundtrip(); err != nil { // wl_output properties follow the binds | ||
| 178 | - | return nil, err | 181 | + | return nil, fmt.Errorf("neferclient: output roundtrip: %w", err) |
| 179 | } | 182 | } | ||
| 180 | if err = c.applyHeld(); err != nil { | 183 | if err = c.applyHeld(); err != nil { | ||
| 181 | return nil, err | 184 | return nil, err | ||
| @@ -203,7 +206,7 @@ func (h removeHook) HandleRegistryGlobalRemove(e wl.RegistryGlobalRemoveEvent) { | |||||
| 203 | // setupHandler applies setup events without notifying anyone and keeps the | 206 | // setupHandler applies setup events without notifying anyone and keeps the | ||
| 204 | // first failure, which makes Connect fail. | 207 | // first failure, which makes Connect fail. | ||
| 205 | type setupHandler struct { | 208 | type setupHandler struct { | ||
| 206 | - | nopHandler | 209 | + | NopHandler |
| 207 | err *error | 210 | err *error | ||
| 208 | } | 211 | } | ||
| 209 | 212 | ||||
| @@ -238,7 +241,7 @@ func (c *Conn) readLoop() { | |||||
| 238 | continue // a pause request interrupted the read | 241 | continue // a pause request interrupted the read | ||
| 239 | } | 242 | } | ||
| 240 | if !c.q.isStopped() { | 243 | if !c.q.isStopped() { | ||
| 241 | - | c.q.setFatal(err) | 244 | + | c.q.setFatal(fmt.Errorf("neferclient: read: %w", err)) |
| 242 | } | 245 | } | ||
| 243 | return | 246 | return | ||
| 244 | } | 247 | } | ||
| @@ -291,7 +294,7 @@ func (c *Conn) resume() { | |||||
| 291 | } | 294 | } | ||
| 292 | c.pauseDepth-- | 295 | c.pauseDepth-- | ||
| 293 | if c.pauseDepth == 0 { | 296 | if c.pauseDepth == 0 { | ||
| 294 | - | c.q.setPausing(false) | 297 | + | c.q.endPause() |
| 295 | } | 298 | } | ||
| 296 | } | 299 | } | ||
| 297 | 300 | ||||
| @@ -322,7 +325,7 @@ func (c *Conn) Dispatch(h Handler) error { | |||||
| 322 | return errors.New("neferclient: Dispatch called from a Handler") | 325 | return errors.New("neferclient: Dispatch called from a Handler") | ||
| 323 | } | 326 | } | ||
| 324 | if h == nil { | 327 | if h == nil { | ||
| 325 | - | h = nopHandler{} | 328 | + | h = NopHandler{} |
| 326 | } | 329 | } | ||
| 327 | c.dispatching = true | 330 | c.dispatching = true | ||
| 328 | defer func() { c.dispatching = false }() | 331 | defer func() { c.dispatching = false }() | ||
| @@ -374,6 +377,9 @@ func (c *Conn) apply(ev *event, h Handler) { | |||||
| 374 | return // unwatched while the event was queued | 377 | return // unwatched while the event was queued | ||
| 375 | } | 378 | } | ||
| 376 | h.FDReady(id) | 379 | h.FDReady(id) | ||
| 380 | + | if c.closed { | |||
| 381 | + | return // the handler closed the connection | |||
| 382 | + | } | |||
| 377 | if _, ok := c.watched[ev.fd]; ok { // the handler may have unwatched it | 383 | if _, ok := c.watched[ev.fd]; ok { // the handler may have unwatched it | ||
| 378 | if err := c.armFD(int(ev.fd), unix.EPOLL_CTL_MOD); err != nil { | 384 | if err := c.armFD(int(ev.fd), unix.EPOLL_CTL_MOD); err != nil { | ||
| 379 | h.Error(err) | 385 | h.Error(err) | ||
| @@ -392,7 +398,7 @@ func (c *Conn) entry(global uint32) *outputEntry { | |||||
| 392 | } | 398 | } | ||
| 393 | 399 | ||||
| 394 | func (c *Conn) bindOutput(global, version uint32) error { | 400 | func (c *Conn) bindOutput(global, version uint32) error { | ||
| 395 | - | if version == 0 || c.entry(global) != nil { | 401 | + | if version < 2 || c.entry(global) != nil { // v1 has no wl_output.done |
| 396 | return nil | 402 | return nil | ||
| 397 | } | 403 | } | ||
| 398 | o := &outputEntry{pub: Output{Global: global, Scale: 1}, scale: 1} | 404 | o := &outputEntry{pub: Output{Global: global, Scale: 1}, scale: 1} | ||
| @@ -472,14 +478,19 @@ func (c *Conn) Outputs() []Output { | |||||
| 472 | } | 478 | } | ||
| 473 | 479 | ||||
| 474 | // Roundtrip sends wl_display.sync and dispatches until it returns. The reader | 480 | // Roundtrip sends wl_display.sync and dispatches until it returns. The reader | ||
| 475 | - | // is paused meanwhile; the events it handled are queued for [Conn.Dispatch]. | 481 | + | // is paused meanwhile; when Roundtrip returns, every event handled before the |
| 482 | + | // sync reply is queued for [Conn.Dispatch] (up to the ring capacity; the rest | |||
| 483 | + | // follows as the reader resumes). | |||
| 476 | func (c *Conn) Roundtrip() error { | 484 | func (c *Conn) Roundtrip() error { | ||
| 477 | if c.closed { | 485 | if c.closed { | ||
| 478 | return ErrClosed | 486 | return ErrClosed | ||
| 479 | } | 487 | } | ||
| 480 | c.pause() | 488 | c.pause() | ||
| 481 | defer c.resume() | 489 | defer c.resume() | ||
| 482 | - | return c.display.Roundtrip() | 490 | + | if err := c.display.Roundtrip(); err != nil { |
| 491 | + | return fmt.Errorf("neferclient: roundtrip: %w", err) | |||
| 492 | + | } | |||
| 493 | + | return nil | |||
| 483 | } | 494 | } | ||
| 484 | 495 | ||||
| 485 | // Bind binds the global named iface at min(announced, version) and returns the | 496 | // Bind binds the global named iface at min(announced, version) and returns the | ||
M handler.go +8 -7
| @@ -25,11 +25,12 @@ type Handler interface { | |||
| 25 | 25 | Error(err error) | |
| 26 | 26 | } | |
| 27 | 27 | ||
| 28 | - | // nopHandler discards every event. Connect uses it to apply the events queued | |
| 29 | - | // during setup, and Dispatch uses it for a nil Handler. | |
| 30 | - | type nopHandler struct{} | |
| 28 | + | // NopHandler implements every Handler method as a no-op. Embed it in a handler | |
| 29 | + | // and override only the methods you need; methods added to Handler in later | |
| 30 | + | // versions get their no-op here, so embedding keeps the type compiling. | |
| 31 | + | type NopHandler struct{} | |
| 31 | 32 | ||
| 32 | - | func (nopHandler) OutputAdded(*Output) {} | |
| 33 | - | func (nopHandler) OutputRemoved(uint32) {} | |
| 34 | - | func (nopHandler) FDReady(uint64) {} | |
| 35 | - | func (nopHandler) Error(error) {} | |
| 33 | + | func (NopHandler) OutputAdded(*Output) {} | |
| 34 | + | func (NopHandler) OutputRemoved(uint32) {} | |
| 35 | + | func (NopHandler) FDReady(uint64) {} | |
| 36 | + | func (NopHandler) Error(error) {} | |
| @@ -25,11 +25,12 @@ type Handler interface { | |||||
| 25 | Error(err error) | 25 | Error(err error) | ||
| 26 | } | 26 | } | ||
| 27 | 27 | ||||
| 28 | - | // nopHandler discards every event. Connect uses it to apply the events queued | 28 | + | // NopHandler implements every Handler method as a no-op. Embed it in a handler |
| 29 | - | // during setup, and Dispatch uses it for a nil Handler. | 29 | + | // and override only the methods you need; methods added to Handler in later |
| 30 | - | type nopHandler struct{} | 30 | + | // versions get their no-op here, so embedding keeps the type compiling. |
| 31 | + | type NopHandler struct{} | |||
| 31 | 32 | ||||
| 32 | - | func (nopHandler) OutputAdded(*Output) {} | 33 | + | func (NopHandler) OutputAdded(*Output) {} |
| 33 | - | func (nopHandler) OutputRemoved(uint32) {} | 34 | + | func (NopHandler) OutputRemoved(uint32) {} |
| 34 | - | func (nopHandler) FDReady(uint64) {} | 35 | + | func (NopHandler) FDReady(uint64) {} |
| 35 | - | func (nopHandler) Error(error) {} | 36 | + | func (NopHandler) Error(error) {} |
M queue.go +22 -0
| @@ -161,6 +161,28 @@ func (q *queue) setPausing(v bool) { | |||
| 161 | 161 | q.mu.Unlock() | |
| 162 | 162 | } | |
| 163 | 163 | ||
| 164 | + | // endPause ends a pause started by setPausing(true). Events producers parked | |
| 165 | + | // meanwhile are moved into the ring, in order, as far as it has room, so that | |
| 166 | + | // everything handled during the pause is visible to the next Dispatch. The | |
| 167 | + | // remainder stays at the front of held; the reader re-posts it when it resumes. | |
| 168 | + | func (q *queue) endPause() { | |
| 169 | + | q.mu.Lock() | |
| 170 | + | m := min(len(q.held), ringSize-q.n) | |
| 171 | + | for i := range m { | |
| 172 | + | q.ring[(q.head+q.n)%ringSize] = q.held[i] | |
| 173 | + | q.n++ | |
| 174 | + | } | |
| 175 | + | if m > 0 { | |
| 176 | + | q.held = q.held[:copy(q.held, q.held[m:])] | |
| 177 | + | } | |
| 178 | + | q.pausing = false | |
| 179 | + | q.cond.Broadcast() | |
| 180 | + | q.mu.Unlock() | |
| 181 | + | if m > 0 { | |
| 182 | + | q.signal() | |
| 183 | + | } | |
| 184 | + | } | |
| 185 | + | ||
| 164 | 186 | // awaitParked blocks until the reader is parked or gone. | |
| 165 | 187 | func (q *queue) awaitParked() { | |
| 166 | 188 | q.mu.Lock() | |
| @@ -161,6 +161,28 @@ func (q *queue) setPausing(v bool) { | |||||
| 161 | q.mu.Unlock() | 161 | q.mu.Unlock() | ||
| 162 | } | 162 | } | ||
| 163 | 163 | ||||
| 164 | + | // endPause ends a pause started by setPausing(true). Events producers parked | |||
| 165 | + | // meanwhile are moved into the ring, in order, as far as it has room, so that | |||
| 166 | + | // everything handled during the pause is visible to the next Dispatch. The | |||
| 167 | + | // remainder stays at the front of held; the reader re-posts it when it resumes. | |||
| 168 | + | func (q *queue) endPause() { | |||
| 169 | + | q.mu.Lock() | |||
| 170 | + | m := min(len(q.held), ringSize-q.n) | |||
| 171 | + | for i := range m { | |||
| 172 | + | q.ring[(q.head+q.n)%ringSize] = q.held[i] | |||
| 173 | + | q.n++ | |||
| 174 | + | } | |||
| 175 | + | if m > 0 { | |||
| 176 | + | q.held = q.held[:copy(q.held, q.held[m:])] | |||
| 177 | + | } | |||
| 178 | + | q.pausing = false | |||
| 179 | + | q.cond.Broadcast() | |||
| 180 | + | q.mu.Unlock() | |||
| 181 | + | if m > 0 { | |||
| 182 | + | q.signal() | |||
| 183 | + | } | |||
| 184 | + | } | |||
| 185 | + | ||||
| 164 | // awaitParked blocks until the reader is parked or gone. | 186 | // awaitParked blocks until the reader is parked or gone. | ||
| 165 | func (q *queue) awaitParked() { | 187 | func (q *queue) awaitParked() { | ||
| 166 | q.mu.Lock() | 188 | q.mu.Lock() | ||