fix(input): never block libinput on core, batch pointer commands per display round trip
Each PointerMotionTo cost one display.Do round trip, which waited for the wl_event_loop_dispatch 5 ms timeout, so core accepted ~150 commands/s. Core then blocked libinput on the 32-slot input channel and the pointer trailed the mouse by a growing backlog of stale motion. libinput now hands events to a forwarder that collapses queued motion into the latest position (keys and buttons keep their order), so reading devices never waits for core. The wayland command loop drains queued commands and applies them in one round trip, collapsing consecutive motion per window. Adds 10 s info stats: input (motions, coalesced, sent, max_block_ms) and drm cursor (cursor_moves, cursor_ioctls, cursor_max_ioctl_ms).
- commit
- 1bf994bb331d4e1bb00dc1fc7ad1ff0ee4688789
- author
- brice · · 2026-09-24 16:04 UTC
- parent
- ff79c0c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
8 files · +339 -13
M internal/adapters/drm/cursor.go +27 -2
| @@ -3,6 +3,7 @@ package drm | |||
| 3 | 3 | import ( | |
| 4 | 4 | "fmt" | |
| 5 | 5 | "sync" | |
| 6 | + | "time" | |
| 6 | 7 | "unsafe" | |
| 7 | 8 | ) | |
| 8 | 9 | ||
| @@ -44,14 +45,31 @@ type Cursor struct { | |||
| 44 | 45 | shown bool | |
| 45 | 46 | ioctl func(*modeCursor2) error | |
| 46 | 47 | ||
| 47 | - | mu sync.Mutex // guards x, y; never held across an ioctl | |
| 48 | - | x, y int // physical position of the hotspot | |
| 48 | + | mu sync.Mutex // guards x, y and stats; never held across an ioctl | |
| 49 | + | x, y int // physical position of the hotspot | |
| 50 | + | stats CursorStats | |
| 49 | 51 | ||
| 50 | 52 | wake chan struct{} | |
| 51 | 53 | stop chan struct{} | |
| 52 | 54 | done chan struct{} | |
| 53 | 55 | } | |
| 54 | 56 | ||
| 57 | + | // CursorStats counts cursor work since the last TakeStats. | |
| 58 | + | type CursorStats struct { | |
| 59 | + | Moves int // Move calls | |
| 60 | + | Ioctls int // move ioctls issued by the worker | |
| 61 | + | MaxIoctl time.Duration // slowest move ioctl | |
| 62 | + | } | |
| 63 | + | ||
| 64 | + | // TakeStats returns and resets the counters. | |
| 65 | + | func (c *Cursor) TakeStats() CursorStats { | |
| 66 | + | c.mu.Lock() | |
| 67 | + | defer c.mu.Unlock() | |
| 68 | + | s := c.stats | |
| 69 | + | c.stats = CursorStats{} | |
| 70 | + | return s | |
| 71 | + | } | |
| 72 | + | ||
| 55 | 73 | // newCursor allocates a cursor buffer at the driver's preferred size (64 when | |
| 56 | 74 | // the driver does not say). | |
| 57 | 75 | func newCursor(fd int, crtc uint32) (*Cursor, error) { | |
| @@ -101,7 +119,13 @@ func (c *Cursor) start() { | |||
| 101 | 119 | v := modeCursor2{flags: cursorMove, crtcID: c.crtc, x: int32(x - c.hotX), y: int32(y - c.hotY)} | |
| 102 | 120 | // Errors while another session owns the card (VT switched | |
| 103 | 121 | // away) are ignored; Reapply restores the cursor. | |
| 122 | + | start := time.Now() | |
| 104 | 123 | _ = c.ioctl(&v) | |
| 124 | + | d := time.Since(start) | |
| 125 | + | c.mu.Lock() | |
| 126 | + | c.stats.Ioctls++ | |
| 127 | + | c.stats.MaxIoctl = max(c.stats.MaxIoctl, d) | |
| 128 | + | c.mu.Unlock() | |
| 105 | 129 | } | |
| 106 | 130 | c.io.Unlock() | |
| 107 | 131 | } | |
| @@ -151,6 +175,7 @@ func (c *Cursor) apply() error { | |||
| 151 | 175 | func (c *Cursor) Move(x, y float64) { | |
| 152 | 176 | c.mu.Lock() | |
| 153 | 177 | c.x, c.y = int(x), int(y) | |
| 178 | + | c.stats.Moves++ | |
| 154 | 179 | c.mu.Unlock() | |
| 155 | 180 | select { | |
| 156 | 181 | case c.wake <- struct{}{}: | |
| @@ -3,6 +3,7 @@ package drm | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "fmt" | 4 | "fmt" | ||
| 5 | "sync" | 5 | "sync" | ||
| 6 | + | "time" | |||
| 6 | "unsafe" | 7 | "unsafe" | ||
| 7 | ) | 8 | ) | ||
| 8 | 9 | ||||
| @@ -44,14 +45,31 @@ type Cursor struct { | |||||
| 44 | shown bool | 45 | shown bool | ||
| 45 | ioctl func(*modeCursor2) error | 46 | ioctl func(*modeCursor2) error | ||
| 46 | 47 | ||||
| 47 | - | mu sync.Mutex // guards x, y; never held across an ioctl | 48 | + | mu sync.Mutex // guards x, y and stats; never held across an ioctl |
| 48 | - | x, y int // physical position of the hotspot | 49 | + | x, y int // physical position of the hotspot |
| 50 | + | stats CursorStats | |||
| 49 | 51 | ||||
| 50 | wake chan struct{} | 52 | wake chan struct{} | ||
| 51 | stop chan struct{} | 53 | stop chan struct{} | ||
| 52 | done chan struct{} | 54 | done chan struct{} | ||
| 53 | } | 55 | } | ||
| 54 | 56 | ||||
| 57 | + | // CursorStats counts cursor work since the last TakeStats. | |||
| 58 | + | type CursorStats struct { | |||
| 59 | + | Moves int // Move calls | |||
| 60 | + | Ioctls int // move ioctls issued by the worker | |||
| 61 | + | MaxIoctl time.Duration // slowest move ioctl | |||
| 62 | + | } | |||
| 63 | + | ||||
| 64 | + | // TakeStats returns and resets the counters. | |||
| 65 | + | func (c *Cursor) TakeStats() CursorStats { | |||
| 66 | + | c.mu.Lock() | |||
| 67 | + | defer c.mu.Unlock() | |||
| 68 | + | s := c.stats | |||
| 69 | + | c.stats = CursorStats{} | |||
| 70 | + | return s | |||
| 71 | + | } | |||
| 72 | + | ||||
| 55 | // newCursor allocates a cursor buffer at the driver's preferred size (64 when | 73 | // newCursor allocates a cursor buffer at the driver's preferred size (64 when | ||
| 56 | // the driver does not say). | 74 | // the driver does not say). | ||
| 57 | func newCursor(fd int, crtc uint32) (*Cursor, error) { | 75 | func newCursor(fd int, crtc uint32) (*Cursor, error) { | ||
| @@ -101,7 +119,13 @@ func (c *Cursor) start() { | |||||
| 101 | v := modeCursor2{flags: cursorMove, crtcID: c.crtc, x: int32(x - c.hotX), y: int32(y - c.hotY)} | 119 | v := modeCursor2{flags: cursorMove, crtcID: c.crtc, x: int32(x - c.hotX), y: int32(y - c.hotY)} | ||
| 102 | // Errors while another session owns the card (VT switched | 120 | // Errors while another session owns the card (VT switched | ||
| 103 | // away) are ignored; Reapply restores the cursor. | 121 | // away) are ignored; Reapply restores the cursor. | ||
| 122 | + | start := time.Now() | |||
| 104 | _ = c.ioctl(&v) | 123 | _ = c.ioctl(&v) | ||
| 124 | + | d := time.Since(start) | |||
| 125 | + | c.mu.Lock() | |||
| 126 | + | c.stats.Ioctls++ | |||
| 127 | + | c.stats.MaxIoctl = max(c.stats.MaxIoctl, d) | |||
| 128 | + | c.mu.Unlock() | |||
| 105 | } | 129 | } | ||
| 106 | c.io.Unlock() | 130 | c.io.Unlock() | ||
| 107 | } | 131 | } | ||
| @@ -151,6 +175,7 @@ func (c *Cursor) apply() error { | |||||
| 151 | func (c *Cursor) Move(x, y float64) { | 175 | func (c *Cursor) Move(x, y float64) { | ||
| 152 | c.mu.Lock() | 176 | c.mu.Lock() | ||
| 153 | c.x, c.y = int(x), int(y) | 177 | c.x, c.y = int(x), int(y) | ||
| 178 | + | c.stats.Moves++ | |||
| 154 | c.mu.Unlock() | 179 | c.mu.Unlock() | ||
| 155 | select { | 180 | select { | ||
| 156 | case c.wake <- struct{}{}: | 181 | case c.wake <- struct{}{}: | ||
M internal/adapters/drm/cursor_test.go +19 -0
| @@ -68,3 +68,22 @@ func TestCursorCloseStopsWorker(t *testing.T) { | |||
| 68 | 68 | c.close() // idempotent | |
| 69 | 69 | c.Move(3, 4) // after close: no worker, must not block or panic | |
| 70 | 70 | } | |
| 71 | + | ||
| 72 | + | func TestCursorStats(t *testing.T) { | |
| 73 | + | applied := make(chan struct{}, 16) | |
| 74 | + | c := &Cursor{fd: -1, buf: &dumbBuffer{}, size: 64, shown: true} | |
| 75 | + | c.ioctl = func(*modeCursor2) error { applied <- struct{}{}; return nil } | |
| 76 | + | c.start() | |
| 77 | + | defer c.close() | |
| 78 | + | c.Move(1, 1) | |
| 79 | + | <-applied | |
| 80 | + | c.Move(2, 2) | |
| 81 | + | <-applied | |
| 82 | + | s := c.TakeStats() | |
| 83 | + | if s.Moves != 2 || s.Ioctls < 1 || s.Ioctls > 2 { | |
| 84 | + | t.Fatalf("stats %+v", s) | |
| 85 | + | } | |
| 86 | + | if s := c.TakeStats(); s.Moves != 0 { | |
| 87 | + | t.Fatalf("not reset: %+v", s) | |
| 88 | + | } | |
| 89 | + | } | |
| @@ -68,3 +68,22 @@ func TestCursorCloseStopsWorker(t *testing.T) { | |||||
| 68 | c.close() // idempotent | 68 | c.close() // idempotent | ||
| 69 | c.Move(3, 4) // after close: no worker, must not block or panic | 69 | c.Move(3, 4) // after close: no worker, must not block or panic | ||
| 70 | } | 70 | } | ||
| 71 | + | ||||
| 72 | + | func TestCursorStats(t *testing.T) { | |||
| 73 | + | applied := make(chan struct{}, 16) | |||
| 74 | + | c := &Cursor{fd: -1, buf: &dumbBuffer{}, size: 64, shown: true} | |||
| 75 | + | c.ioctl = func(*modeCursor2) error { applied <- struct{}{}; return nil } | |||
| 76 | + | c.start() | |||
| 77 | + | defer c.close() | |||
| 78 | + | c.Move(1, 1) | |||
| 79 | + | <-applied | |||
| 80 | + | c.Move(2, 2) | |||
| 81 | + | <-applied | |||
| 82 | + | s := c.TakeStats() | |||
| 83 | + | if s.Moves != 2 || s.Ioctls < 1 || s.Ioctls > 2 { | |||
| 84 | + | t.Fatalf("stats %+v", s) | |||
| 85 | + | } | |||
| 86 | + | if s := c.TakeStats(); s.Moves != 0 { | |||
| 87 | + | t.Fatalf("not reset: %+v", s) | |||
| 88 | + | } | |||
| 89 | + | } | |||
M internal/adapters/drm/output.go +6 -1
| @@ -225,7 +225,12 @@ func (o *Output) Run(ctx context.Context, newRenderer func(w, h int) (Renderer, | |||
| 225 | 225 | } | |
| 226 | 226 | dirty = true | |
| 227 | 227 | case <-stats.C: | |
| 228 | - | o.log.Info().Int("frames", frame).Int("flips", o.flips).Msg("stats") | |
| 228 | + | ev := o.log.Info().Int("frames", frame).Int("flips", o.flips) | |
| 229 | + | if o.cursor != nil { | |
| 230 | + | cs := o.cursor.TakeStats() | |
| 231 | + | ev = ev.Int("cursor_moves", cs.Moves).Int("cursor_ioctls", cs.Ioctls).Dur("cursor_max_ioctl_ms", cs.MaxIoctl) | |
| 232 | + | } | |
| 233 | + | ev.Msg("stats") | |
| 229 | 234 | } | |
| 230 | 235 | if !dirty || !haveScene || !enabled || o.pending { | |
| 231 | 236 | continue | |
| @@ -225,7 +225,12 @@ func (o *Output) Run(ctx context.Context, newRenderer func(w, h int) (Renderer, | |||||
| 225 | } | 225 | } | ||
| 226 | dirty = true | 226 | dirty = true | ||
| 227 | case <-stats.C: | 227 | case <-stats.C: | ||
| 228 | - | o.log.Info().Int("frames", frame).Int("flips", o.flips).Msg("stats") | 228 | + | ev := o.log.Info().Int("frames", frame).Int("flips", o.flips) |
| 229 | + | if o.cursor != nil { | |||
| 230 | + | cs := o.cursor.TakeStats() | |||
| 231 | + | ev = ev.Int("cursor_moves", cs.Moves).Int("cursor_ioctls", cs.Ioctls).Dur("cursor_max_ioctl_ms", cs.MaxIoctl) | |||
| 232 | + | } | |||
| 233 | + | ev.Msg("stats") | |||
| 229 | } | 234 | } | ||
| 230 | if !dirty || !haveScene || !enabled || o.pending { | 235 | if !dirty || !haveScene || !enabled || o.pending { | ||
| 231 | continue | 236 | continue | ||
A internal/adapters/libinput/forward.go +103 -0
| @@ -0,0 +1,103 @@ | |||
| 1 | + | package libinput | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "sync" | |
| 6 | + | "time" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/nefertty/internal/ports" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | // forwarder decouples reading libinput from core. libinput keeps reading | |
| 12 | + | // devices at their full rate and never blocks on core; consecutive pointer | |
| 13 | + | // motions still waiting to be sent collapse into the latest one, while keys | |
| 14 | + | // and buttons keep their order relative to motion. | |
| 15 | + | type forwarder struct { | |
| 16 | + | mu sync.Mutex | |
| 17 | + | queue []ports.InputEvent | |
| 18 | + | wake chan struct{} | |
| 19 | + | stats forwardStats | |
| 20 | + | notify func() // test hook: called after each send | |
| 21 | + | } | |
| 22 | + | ||
| 23 | + | // forwardStats is reset by take. | |
| 24 | + | type forwardStats struct { | |
| 25 | + | Motions int // pointer motions read | |
| 26 | + | Coalesced int // motions merged into a later one | |
| 27 | + | Sent int // events delivered to core | |
| 28 | + | MaxBlock time.Duration // longest wait for core to accept one event | |
| 29 | + | } | |
| 30 | + | ||
| 31 | + | func newForwarder() *forwarder { return &forwarder{wake: make(chan struct{}, 1)} } | |
| 32 | + | ||
| 33 | + | // push queues ev without blocking. | |
| 34 | + | func (f *forwarder) push(ev ports.InputEvent) { | |
| 35 | + | f.mu.Lock() | |
| 36 | + | if m, ok := ev.(ports.PointerMotion); ok { | |
| 37 | + | f.stats.Motions++ | |
| 38 | + | if n := len(f.queue); n > 0 { | |
| 39 | + | if _, tail := f.queue[n-1].(ports.PointerMotion); tail { | |
| 40 | + | f.queue[n-1] = m | |
| 41 | + | f.stats.Coalesced++ | |
| 42 | + | f.mu.Unlock() | |
| 43 | + | return | |
| 44 | + | } | |
| 45 | + | } | |
| 46 | + | } | |
| 47 | + | f.queue = append(f.queue, ev) | |
| 48 | + | f.mu.Unlock() | |
| 49 | + | select { | |
| 50 | + | case f.wake <- struct{}{}: | |
| 51 | + | default: | |
| 52 | + | } | |
| 53 | + | } | |
| 54 | + | ||
| 55 | + | func (f *forwarder) pop() (ports.InputEvent, bool) { | |
| 56 | + | f.mu.Lock() | |
| 57 | + | defer f.mu.Unlock() | |
| 58 | + | if len(f.queue) == 0 { | |
| 59 | + | return nil, false | |
| 60 | + | } | |
| 61 | + | ev := f.queue[0] | |
| 62 | + | f.queue[0] = nil | |
| 63 | + | f.queue = f.queue[1:] | |
| 64 | + | return ev, true | |
| 65 | + | } | |
| 66 | + | ||
| 67 | + | // take returns and resets the counters. | |
| 68 | + | func (f *forwarder) take() forwardStats { | |
| 69 | + | f.mu.Lock() | |
| 70 | + | defer f.mu.Unlock() | |
| 71 | + | s := f.stats | |
| 72 | + | f.stats = forwardStats{} | |
| 73 | + | return s | |
| 74 | + | } | |
| 75 | + | ||
| 76 | + | // run delivers queued events to input until ctx ends. | |
| 77 | + | func (f *forwarder) run(ctx context.Context, input chan<- ports.InputEvent) { | |
| 78 | + | for { | |
| 79 | + | ev, ok := f.pop() | |
| 80 | + | if !ok { | |
| 81 | + | select { | |
| 82 | + | case <-ctx.Done(): | |
| 83 | + | return | |
| 84 | + | case <-f.wake: | |
| 85 | + | continue | |
| 86 | + | } | |
| 87 | + | } | |
| 88 | + | start := time.Now() | |
| 89 | + | select { | |
| 90 | + | case input <- ev: | |
| 91 | + | case <-ctx.Done(): | |
| 92 | + | return | |
| 93 | + | } | |
| 94 | + | d := time.Since(start) | |
| 95 | + | f.mu.Lock() | |
| 96 | + | f.stats.Sent++ | |
| 97 | + | f.stats.MaxBlock = max(f.stats.MaxBlock, d) | |
| 98 | + | f.mu.Unlock() | |
| 99 | + | if f.notify != nil { | |
| 100 | + | f.notify() | |
| 101 | + | } | |
| 102 | + | } | |
| 103 | + | } | |
| @@ -0,0 +1,103 @@ | |||||
| 1 | + | package libinput | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "sync" | |||
| 6 | + | "time" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/nefertty/internal/ports" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | // forwarder decouples reading libinput from core. libinput keeps reading | |||
| 12 | + | // devices at their full rate and never blocks on core; consecutive pointer | |||
| 13 | + | // motions still waiting to be sent collapse into the latest one, while keys | |||
| 14 | + | // and buttons keep their order relative to motion. | |||
| 15 | + | type forwarder struct { | |||
| 16 | + | mu sync.Mutex | |||
| 17 | + | queue []ports.InputEvent | |||
| 18 | + | wake chan struct{} | |||
| 19 | + | stats forwardStats | |||
| 20 | + | notify func() // test hook: called after each send | |||
| 21 | + | } | |||
| 22 | + | ||||
| 23 | + | // forwardStats is reset by take. | |||
| 24 | + | type forwardStats struct { | |||
| 25 | + | Motions int // pointer motions read | |||
| 26 | + | Coalesced int // motions merged into a later one | |||
| 27 | + | Sent int // events delivered to core | |||
| 28 | + | MaxBlock time.Duration // longest wait for core to accept one event | |||
| 29 | + | } | |||
| 30 | + | ||||
| 31 | + | func newForwarder() *forwarder { return &forwarder{wake: make(chan struct{}, 1)} } | |||
| 32 | + | ||||
| 33 | + | // push queues ev without blocking. | |||
| 34 | + | func (f *forwarder) push(ev ports.InputEvent) { | |||
| 35 | + | f.mu.Lock() | |||
| 36 | + | if m, ok := ev.(ports.PointerMotion); ok { | |||
| 37 | + | f.stats.Motions++ | |||
| 38 | + | if n := len(f.queue); n > 0 { | |||
| 39 | + | if _, tail := f.queue[n-1].(ports.PointerMotion); tail { | |||
| 40 | + | f.queue[n-1] = m | |||
| 41 | + | f.stats.Coalesced++ | |||
| 42 | + | f.mu.Unlock() | |||
| 43 | + | return | |||
| 44 | + | } | |||
| 45 | + | } | |||
| 46 | + | } | |||
| 47 | + | f.queue = append(f.queue, ev) | |||
| 48 | + | f.mu.Unlock() | |||
| 49 | + | select { | |||
| 50 | + | case f.wake <- struct{}{}: | |||
| 51 | + | default: | |||
| 52 | + | } | |||
| 53 | + | } | |||
| 54 | + | ||||
| 55 | + | func (f *forwarder) pop() (ports.InputEvent, bool) { | |||
| 56 | + | f.mu.Lock() | |||
| 57 | + | defer f.mu.Unlock() | |||
| 58 | + | if len(f.queue) == 0 { | |||
| 59 | + | return nil, false | |||
| 60 | + | } | |||
| 61 | + | ev := f.queue[0] | |||
| 62 | + | f.queue[0] = nil | |||
| 63 | + | f.queue = f.queue[1:] | |||
| 64 | + | return ev, true | |||
| 65 | + | } | |||
| 66 | + | ||||
| 67 | + | // take returns and resets the counters. | |||
| 68 | + | func (f *forwarder) take() forwardStats { | |||
| 69 | + | f.mu.Lock() | |||
| 70 | + | defer f.mu.Unlock() | |||
| 71 | + | s := f.stats | |||
| 72 | + | f.stats = forwardStats{} | |||
| 73 | + | return s | |||
| 74 | + | } | |||
| 75 | + | ||||
| 76 | + | // run delivers queued events to input until ctx ends. | |||
| 77 | + | func (f *forwarder) run(ctx context.Context, input chan<- ports.InputEvent) { | |||
| 78 | + | for { | |||
| 79 | + | ev, ok := f.pop() | |||
| 80 | + | if !ok { | |||
| 81 | + | select { | |||
| 82 | + | case <-ctx.Done(): | |||
| 83 | + | return | |||
| 84 | + | case <-f.wake: | |||
| 85 | + | continue | |||
| 86 | + | } | |||
| 87 | + | } | |||
| 88 | + | start := time.Now() | |||
| 89 | + | select { | |||
| 90 | + | case input <- ev: | |||
| 91 | + | case <-ctx.Done(): | |||
| 92 | + | return | |||
| 93 | + | } | |||
| 94 | + | d := time.Since(start) | |||
| 95 | + | f.mu.Lock() | |||
| 96 | + | f.stats.Sent++ | |||
| 97 | + | f.stats.MaxBlock = max(f.stats.MaxBlock, d) | |||
| 98 | + | f.mu.Unlock() | |||
| 99 | + | if f.notify != nil { | |||
| 100 | + | f.notify() | |||
| 101 | + | } | |||
| 102 | + | } | |||
| 103 | + | } | |||
A internal/adapters/libinput/forward_test.go +70 -0
| @@ -0,0 +1,70 @@ | |||
| 1 | + | package libinput | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "testing" | |
| 6 | + | "time" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/nefertty/internal/ports" | |
| 9 | + | ) | |
| 10 | + | ||
| 11 | + | // A slow consumer must not block push, and stale motion must collapse into | |
| 12 | + | // the latest position while keys and buttons keep their order. | |
| 13 | + | func TestForwarderCoalescesMotionWithoutBlocking(t *testing.T) { | |
| 14 | + | f := newForwarder() | |
| 15 | + | input := make(chan ports.InputEvent) // unbuffered: core is busy | |
| 16 | + | start := time.Now() | |
| 17 | + | for i := range 1000 { | |
| 18 | + | f.push(ports.PointerMotion{X: float64(i)}) | |
| 19 | + | } | |
| 20 | + | f.push(ports.PointerButton{Button: 272, Pressed: true}) | |
| 21 | + | f.push(ports.PointerMotion{X: 2000}) | |
| 22 | + | f.push(ports.PointerMotion{X: 2001}) | |
| 23 | + | if d := time.Since(start); d > 100*time.Millisecond { | |
| 24 | + | t.Fatalf("push blocked for %v", d) | |
| 25 | + | } | |
| 26 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 27 | + | defer cancel() | |
| 28 | + | go f.run(ctx, input) | |
| 29 | + | want := []ports.InputEvent{ | |
| 30 | + | ports.PointerMotion{X: 999}, | |
| 31 | + | ports.PointerButton{Button: 272, Pressed: true}, | |
| 32 | + | ports.PointerMotion{X: 2001}, | |
| 33 | + | } | |
| 34 | + | for i, w := range want { | |
| 35 | + | select { | |
| 36 | + | case got := <-input: | |
| 37 | + | if got != w { | |
| 38 | + | t.Fatalf("event %d: got %#v, want %#v", i, got, w) | |
| 39 | + | } | |
| 40 | + | case <-time.After(2 * time.Second): | |
| 41 | + | t.Fatalf("event %d not delivered", i) | |
| 42 | + | } | |
| 43 | + | } | |
| 44 | + | select { | |
| 45 | + | case got := <-input: | |
| 46 | + | t.Fatalf("unexpected extra event %#v", got) | |
| 47 | + | case <-time.After(20 * time.Millisecond): | |
| 48 | + | } | |
| 49 | + | s := f.take() | |
| 50 | + | if s.Motions != 1002 || s.Coalesced != 1000 || s.Sent != 3 { | |
| 51 | + | t.Fatalf("stats %+v", s) | |
| 52 | + | } | |
| 53 | + | if s := f.take(); s != (forwardStats{}) { | |
| 54 | + | t.Fatalf("take did not reset: %+v", s) | |
| 55 | + | } | |
| 56 | + | } | |
| 57 | + | ||
| 58 | + | func TestForwarderStopsOnCancel(t *testing.T) { | |
| 59 | + | f := newForwarder() | |
| 60 | + | f.push(ports.PointerMotion{X: 1}) | |
| 61 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 62 | + | done := make(chan struct{}) | |
| 63 | + | go func() { f.run(ctx, make(chan ports.InputEvent)); close(done) }() | |
| 64 | + | cancel() | |
| 65 | + | select { | |
| 66 | + | case <-done: | |
| 67 | + | case <-time.After(2 * time.Second): | |
| 68 | + | t.Fatal("run did not stop while blocked on send") | |
| 69 | + | } | |
| 70 | + | } | |
| @@ -0,0 +1,70 @@ | |||||
| 1 | + | package libinput | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "testing" | |||
| 6 | + | "time" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/nefertty/internal/ports" | |||
| 9 | + | ) | |||
| 10 | + | ||||
| 11 | + | // A slow consumer must not block push, and stale motion must collapse into | |||
| 12 | + | // the latest position while keys and buttons keep their order. | |||
| 13 | + | func TestForwarderCoalescesMotionWithoutBlocking(t *testing.T) { | |||
| 14 | + | f := newForwarder() | |||
| 15 | + | input := make(chan ports.InputEvent) // unbuffered: core is busy | |||
| 16 | + | start := time.Now() | |||
| 17 | + | for i := range 1000 { | |||
| 18 | + | f.push(ports.PointerMotion{X: float64(i)}) | |||
| 19 | + | } | |||
| 20 | + | f.push(ports.PointerButton{Button: 272, Pressed: true}) | |||
| 21 | + | f.push(ports.PointerMotion{X: 2000}) | |||
| 22 | + | f.push(ports.PointerMotion{X: 2001}) | |||
| 23 | + | if d := time.Since(start); d > 100*time.Millisecond { | |||
| 24 | + | t.Fatalf("push blocked for %v", d) | |||
| 25 | + | } | |||
| 26 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 27 | + | defer cancel() | |||
| 28 | + | go f.run(ctx, input) | |||
| 29 | + | want := []ports.InputEvent{ | |||
| 30 | + | ports.PointerMotion{X: 999}, | |||
| 31 | + | ports.PointerButton{Button: 272, Pressed: true}, | |||
| 32 | + | ports.PointerMotion{X: 2001}, | |||
| 33 | + | } | |||
| 34 | + | for i, w := range want { | |||
| 35 | + | select { | |||
| 36 | + | case got := <-input: | |||
| 37 | + | if got != w { | |||
| 38 | + | t.Fatalf("event %d: got %#v, want %#v", i, got, w) | |||
| 39 | + | } | |||
| 40 | + | case <-time.After(2 * time.Second): | |||
| 41 | + | t.Fatalf("event %d not delivered", i) | |||
| 42 | + | } | |||
| 43 | + | } | |||
| 44 | + | select { | |||
| 45 | + | case got := <-input: | |||
| 46 | + | t.Fatalf("unexpected extra event %#v", got) | |||
| 47 | + | case <-time.After(20 * time.Millisecond): | |||
| 48 | + | } | |||
| 49 | + | s := f.take() | |||
| 50 | + | if s.Motions != 1002 || s.Coalesced != 1000 || s.Sent != 3 { | |||
| 51 | + | t.Fatalf("stats %+v", s) | |||
| 52 | + | } | |||
| 53 | + | if s := f.take(); s != (forwardStats{}) { | |||
| 54 | + | t.Fatalf("take did not reset: %+v", s) | |||
| 55 | + | } | |||
| 56 | + | } | |||
| 57 | + | ||||
| 58 | + | func TestForwarderStopsOnCancel(t *testing.T) { | |||
| 59 | + | f := newForwarder() | |||
| 60 | + | f.push(ports.PointerMotion{X: 1}) | |||
| 61 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 62 | + | done := make(chan struct{}) | |||
| 63 | + | go func() { f.run(ctx, make(chan ports.InputEvent)); close(done) }() | |||
| 64 | + | cancel() | |||
| 65 | + | select { | |||
| 66 | + | case <-done: | |||
| 67 | + | case <-time.After(2 * time.Second): | |||
| 68 | + | t.Fatal("run did not stop while blocked on send") | |||
| 69 | + | } | |||
| 70 | + | } | |||
M internal/adapters/libinput/libinput.go +17 -7
| @@ -148,6 +148,9 @@ type Options struct { | |||
| 148 | 148 | Log zerowrap.Logger | |
| 149 | 149 | } | |
| 150 | 150 | ||
| 151 | + | // statsEvery is how often Run logs input throughput. | |
| 152 | + | const statsEvery = 10 * time.Second | |
| 153 | + | ||
| 151 | 154 | // Run owns libinput and the keymap, converting device events into input events. | |
| 152 | 155 | func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error { | |
| 153 | 156 | defer func() { opts.Keymap.Close() }() | |
| @@ -178,7 +181,19 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error | |||
| 178 | 181 | opts.MoveCursor(p.x, p.y) | |
| 179 | 182 | } | |
| 180 | 183 | fd := getFD(li) | |
| 184 | + | fwd := newForwarder() | |
| 185 | + | fwdCtx, stopFwd := context.WithCancel(ctx) | |
| 186 | + | var fwdDone sync.WaitGroup | |
| 187 | + | fwdDone.Go(func() { fwd.run(fwdCtx, input) }) | |
| 188 | + | defer func() { stopFwd(); fwdDone.Wait() }() | |
| 189 | + | lastStats := time.Now() | |
| 181 | 190 | for ctx.Err() == nil { | |
| 191 | + | if time.Since(lastStats) >= statsEvery { | |
| 192 | + | lastStats = time.Now() | |
| 193 | + | if s := fwd.take(); s.Motions > 0 || s.Sent > 0 { | |
| 194 | + | opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats") | |
| 195 | + | } | |
| 196 | + | } | |
| 182 | 197 | select { | |
| 183 | 198 | case on := <-opts.Active: | |
| 184 | 199 | if on { | |
| @@ -210,13 +225,8 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error | |||
| 210 | 225 | if err != nil { | |
| 211 | 226 | return err | |
| 212 | 227 | } | |
| 213 | - | if out == nil { | |
| 214 | - | continue | |
| 215 | - | } | |
| 216 | - | select { | |
| 217 | - | case input <- out: | |
| 218 | - | case <-ctx.Done(): | |
| 219 | - | return nil | |
| 228 | + | if out != nil { | |
| 229 | + | fwd.push(out) | |
| 220 | 230 | } | |
| 221 | 231 | } | |
| 222 | 232 | } | |
| @@ -148,6 +148,9 @@ type Options struct { | |||||
| 148 | Log zerowrap.Logger | 148 | Log zerowrap.Logger | ||
| 149 | } | 149 | } | ||
| 150 | 150 | ||||
| 151 | + | // statsEvery is how often Run logs input throughput. | |||
| 152 | + | const statsEvery = 10 * time.Second | |||
| 153 | + | ||||
| 151 | // Run owns libinput and the keymap, converting device events into input events. | 154 | // Run owns libinput and the keymap, converting device events into input events. | ||
| 152 | func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error { | 155 | func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error { | ||
| 153 | defer func() { opts.Keymap.Close() }() | 156 | defer func() { opts.Keymap.Close() }() | ||
| @@ -178,7 +181,19 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error | |||||
| 178 | opts.MoveCursor(p.x, p.y) | 181 | opts.MoveCursor(p.x, p.y) | ||
| 179 | } | 182 | } | ||
| 180 | fd := getFD(li) | 183 | fd := getFD(li) | ||
| 184 | + | fwd := newForwarder() | |||
| 185 | + | fwdCtx, stopFwd := context.WithCancel(ctx) | |||
| 186 | + | var fwdDone sync.WaitGroup | |||
| 187 | + | fwdDone.Go(func() { fwd.run(fwdCtx, input) }) | |||
| 188 | + | defer func() { stopFwd(); fwdDone.Wait() }() | |||
| 189 | + | lastStats := time.Now() | |||
| 181 | for ctx.Err() == nil { | 190 | for ctx.Err() == nil { | ||
| 191 | + | if time.Since(lastStats) >= statsEvery { | |||
| 192 | + | lastStats = time.Now() | |||
| 193 | + | if s := fwd.take(); s.Motions > 0 || s.Sent > 0 { | |||
| 194 | + | opts.Log.Info().Int("motions", s.Motions).Int("coalesced", s.Coalesced).Int("sent", s.Sent).Dur("max_block_ms", s.MaxBlock).Msg("input stats") | |||
| 195 | + | } | |||
| 196 | + | } | |||
| 182 | select { | 197 | select { | ||
| 183 | case on := <-opts.Active: | 198 | case on := <-opts.Active: | ||
| 184 | if on { | 199 | if on { | ||
| @@ -210,13 +225,8 @@ func Run(ctx context.Context, opts Options, input chan<- ports.InputEvent) error | |||||
| 210 | if err != nil { | 225 | if err != nil { | ||
| 211 | return err | 226 | return err | ||
| 212 | } | 227 | } | ||
| 213 | - | if out == nil { | 228 | + | if out != nil { |
| 214 | - | continue | 229 | + | fwd.push(out) |
| 215 | - | } | |||
| 216 | - | select { | |||
| 217 | - | case input <- out: | |||
| 218 | - | case <-ctx.Done(): | |||
| 219 | - | return nil | |||
| 220 | } | 230 | } | ||
| 221 | } | 231 | } | ||
| 222 | } | 232 | } | ||
A internal/adapters/wayland/commands_test.go +59 -0
| @@ -0,0 +1,59 @@ | |||
| 1 | + | package wayland | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "reflect" | |
| 6 | + | "testing" | |
| 7 | + | "time" | |
| 8 | + | ||
| 9 | + | "github.com/bnema/nefertty/internal/logging" | |
| 10 | + | "github.com/bnema/nefertty/internal/ports" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | func TestDrainCommandsCollapsesMotion(t *testing.T) { | |
| 14 | + | cmds := make(chan ports.ClientCommand, 16) | |
| 15 | + | cmds <- ports.PointerMotionTo{ID: 1, X: 2} | |
| 16 | + | cmds <- ports.PointerMotionTo{ID: 1, X: 3} | |
| 17 | + | cmds <- ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true} | |
| 18 | + | cmds <- ports.PointerMotionTo{ID: 1, X: 4} | |
| 19 | + | cmds <- ports.PointerFocus{ID: 2} | |
| 20 | + | cmds <- ports.PointerMotionTo{ID: 2, X: 5} | |
| 21 | + | cmds <- ports.PointerMotionTo{ID: 2, X: 6} | |
| 22 | + | batch, open := drainCommands(ports.PointerMotionTo{ID: 1, X: 1}, cmds) | |
| 23 | + | want := []ports.ClientCommand{ | |
| 24 | + | ports.PointerMotionTo{ID: 1, X: 3}, | |
| 25 | + | ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true}, | |
| 26 | + | ports.PointerMotionTo{ID: 1, X: 4}, | |
| 27 | + | ports.PointerFocus{ID: 2}, | |
| 28 | + | ports.PointerMotionTo{ID: 2, X: 6}, | |
| 29 | + | } | |
| 30 | + | if !open || !reflect.DeepEqual(batch, want) { | |
| 31 | + | t.Fatalf("open=%v batch=%#v", open, batch) | |
| 32 | + | } | |
| 33 | + | close(cmds) | |
| 34 | + | if _, open := drainCommands(ports.PointerFocus{}, cmds); open { | |
| 35 | + | t.Fatal("closed channel reported open") | |
| 36 | + | } | |
| 37 | + | } | |
| 38 | + | ||
| 39 | + | // Each command used to cost one display.Do round trip (up to one 5 ms | |
| 40 | + | // event-loop dispatch), capping pointer motion at ~150/s. A 1 kHz mouse | |
| 41 | + | // must be absorbed without core blocking. | |
| 42 | + | func TestCommandThroughput(t *testing.T) { | |
| 43 | + | cmds := make(chan ports.ClientCommand, 32) | |
| 44 | + | s, err := New(Options{RuntimeDir: t.TempDir(), OutputWidth: 1920, OutputHeight: 1080}, Channels{Commands: cmds}, logging.For(context.Background(), "wayland")) | |
| 45 | + | if err != nil { | |
| 46 | + | t.Fatal(err) | |
| 47 | + | } | |
| 48 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 49 | + | done := make(chan error, 1) | |
| 50 | + | go func() { done <- s.Run(ctx) }() | |
| 51 | + | defer func() { cancel(); <-done }() | |
| 52 | + | start := time.Now() | |
| 53 | + | for i := range 2000 { | |
| 54 | + | cmds <- ports.PointerMotionTo{ID: 1, X: float64(i), Y: 1} | |
| 55 | + | } | |
| 56 | + | if d := time.Since(start); d > time.Second { | |
| 57 | + | t.Fatalf("2000 motion commands took %v (%.0f/s); want well above 1000/s", d, 2000/d.Seconds()) | |
| 58 | + | } | |
| 59 | + | } | |
| @@ -0,0 +1,59 @@ | |||||
| 1 | + | package wayland | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "reflect" | |||
| 6 | + | "testing" | |||
| 7 | + | "time" | |||
| 8 | + | ||||
| 9 | + | "github.com/bnema/nefertty/internal/logging" | |||
| 10 | + | "github.com/bnema/nefertty/internal/ports" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | func TestDrainCommandsCollapsesMotion(t *testing.T) { | |||
| 14 | + | cmds := make(chan ports.ClientCommand, 16) | |||
| 15 | + | cmds <- ports.PointerMotionTo{ID: 1, X: 2} | |||
| 16 | + | cmds <- ports.PointerMotionTo{ID: 1, X: 3} | |||
| 17 | + | cmds <- ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true} | |||
| 18 | + | cmds <- ports.PointerMotionTo{ID: 1, X: 4} | |||
| 19 | + | cmds <- ports.PointerFocus{ID: 2} | |||
| 20 | + | cmds <- ports.PointerMotionTo{ID: 2, X: 5} | |||
| 21 | + | cmds <- ports.PointerMotionTo{ID: 2, X: 6} | |||
| 22 | + | batch, open := drainCommands(ports.PointerMotionTo{ID: 1, X: 1}, cmds) | |||
| 23 | + | want := []ports.ClientCommand{ | |||
| 24 | + | ports.PointerMotionTo{ID: 1, X: 3}, | |||
| 25 | + | ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true}, | |||
| 26 | + | ports.PointerMotionTo{ID: 1, X: 4}, | |||
| 27 | + | ports.PointerFocus{ID: 2}, | |||
| 28 | + | ports.PointerMotionTo{ID: 2, X: 6}, | |||
| 29 | + | } | |||
| 30 | + | if !open || !reflect.DeepEqual(batch, want) { | |||
| 31 | + | t.Fatalf("open=%v batch=%#v", open, batch) | |||
| 32 | + | } | |||
| 33 | + | close(cmds) | |||
| 34 | + | if _, open := drainCommands(ports.PointerFocus{}, cmds); open { | |||
| 35 | + | t.Fatal("closed channel reported open") | |||
| 36 | + | } | |||
| 37 | + | } | |||
| 38 | + | ||||
| 39 | + | // Each command used to cost one display.Do round trip (up to one 5 ms | |||
| 40 | + | // event-loop dispatch), capping pointer motion at ~150/s. A 1 kHz mouse | |||
| 41 | + | // must be absorbed without core blocking. | |||
| 42 | + | func TestCommandThroughput(t *testing.T) { | |||
| 43 | + | cmds := make(chan ports.ClientCommand, 32) | |||
| 44 | + | s, err := New(Options{RuntimeDir: t.TempDir(), OutputWidth: 1920, OutputHeight: 1080}, Channels{Commands: cmds}, logging.For(context.Background(), "wayland")) | |||
| 45 | + | if err != nil { | |||
| 46 | + | t.Fatal(err) | |||
| 47 | + | } | |||
| 48 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 49 | + | done := make(chan error, 1) | |||
| 50 | + | go func() { done <- s.Run(ctx) }() | |||
| 51 | + | defer func() { cancel(); <-done }() | |||
| 52 | + | start := time.Now() | |||
| 53 | + | for i := range 2000 { | |||
| 54 | + | cmds <- ports.PointerMotionTo{ID: 1, X: float64(i), Y: 1} | |||
| 55 | + | } | |||
| 56 | + | if d := time.Since(start); d > time.Second { | |||
| 57 | + | t.Fatalf("2000 motion commands took %v (%.0f/s); want well above 1000/s", d, 2000/d.Seconds()) | |||
| 58 | + | } | |||
| 59 | + | } | |||
M internal/adapters/wayland/server.go +38 -3
| @@ -174,11 +174,18 @@ func (s *Server) Run(ctx context.Context) error { | |||
| 174 | 174 | if ctx.Err() != nil { | |
| 175 | 175 | return | |
| 176 | 176 | } | |
| 177 | + | // One display round trip per batch: Do waits for the event | |
| 178 | + | // loop, so a round trip per motion capped pointer events at | |
| 179 | + | // ~150/s and let a backlog of stale motions build up. | |
| 180 | + | batch, open := drainCommands(cmd, s.channels.Commands) | |
| 177 | 181 | if !s.display.Do(func() { | |
| 178 | - | if ctx.Err() == nil { | |
| 179 | - | s.apply(cmd) | |
| 182 | + | for _, c := range batch { | |
| 183 | + | if ctx.Err() != nil { | |
| 184 | + | return | |
| 185 | + | } | |
| 186 | + | s.apply(c) | |
| 180 | 187 | } | |
| 181 | - | }) { | |
| 188 | + | }) || !open { | |
| 182 | 189 | return | |
| 183 | 190 | } | |
| 184 | 191 | } | |
| @@ -315,6 +322,34 @@ func (s *Server) forwardContents(ctx context.Context) { | |||
| 315 | 322 | } | |
| 316 | 323 | } | |
| 317 | 324 | ||
| 325 | + | // maxCommandBatch bounds how many queued commands one display round trip applies. | |
| 326 | + | const maxCommandBatch = 256 | |
| 327 | + | ||
| 328 | + | // drainCommands returns first plus the commands already queued, with runs of | |
| 329 | + | // pointer motion for the same window collapsed into the latest one. open is | |
| 330 | + | // false once the channel is closed. | |
| 331 | + | func drainCommands(first ports.ClientCommand, cmds <-chan ports.ClientCommand) (batch []ports.ClientCommand, open bool) { | |
| 332 | + | batch = append(batch, first) | |
| 333 | + | for len(batch) < maxCommandBatch { | |
| 334 | + | select { | |
| 335 | + | case c, ok := <-cmds: | |
| 336 | + | if !ok { | |
| 337 | + | return batch, false | |
| 338 | + | } | |
| 339 | + | if m, isMotion := c.(ports.PointerMotionTo); isMotion { | |
| 340 | + | if last, lastMotion := batch[len(batch)-1].(ports.PointerMotionTo); lastMotion && last.ID == m.ID { | |
| 341 | + | batch[len(batch)-1] = m | |
| 342 | + | continue | |
| 343 | + | } | |
| 344 | + | } | |
| 345 | + | batch = append(batch, c) | |
| 346 | + | default: | |
| 347 | + | return batch, true | |
| 348 | + | } | |
| 349 | + | } | |
| 350 | + | return batch, true | |
| 351 | + | } | |
| 352 | + | ||
| 318 | 353 | func (s *Server) apply(cmd ports.ClientCommand) { | |
| 319 | 354 | switch c := cmd.(type) { | |
| 320 | 355 | case ports.PointerFocus: | |
| @@ -174,11 +174,18 @@ func (s *Server) Run(ctx context.Context) error { | |||||
| 174 | if ctx.Err() != nil { | 174 | if ctx.Err() != nil { | ||
| 175 | return | 175 | return | ||
| 176 | } | 176 | } | ||
| 177 | + | // One display round trip per batch: Do waits for the event | |||
| 178 | + | // loop, so a round trip per motion capped pointer events at | |||
| 179 | + | // ~150/s and let a backlog of stale motions build up. | |||
| 180 | + | batch, open := drainCommands(cmd, s.channels.Commands) | |||
| 177 | if !s.display.Do(func() { | 181 | if !s.display.Do(func() { | ||
| 178 | - | if ctx.Err() == nil { | 182 | + | for _, c := range batch { |
| 179 | - | s.apply(cmd) | 183 | + | if ctx.Err() != nil { |
| 184 | + | return | |||
| 185 | + | } | |||
| 186 | + | s.apply(c) | |||
| 180 | } | 187 | } | ||
| 181 | - | }) { | 188 | + | }) || !open { |
| 182 | return | 189 | return | ||
| 183 | } | 190 | } | ||
| 184 | } | 191 | } | ||
| @@ -315,6 +322,34 @@ func (s *Server) forwardContents(ctx context.Context) { | |||||
| 315 | } | 322 | } | ||
| 316 | } | 323 | } | ||
| 317 | 324 | ||||
| 325 | + | // maxCommandBatch bounds how many queued commands one display round trip applies. | |||
| 326 | + | const maxCommandBatch = 256 | |||
| 327 | + | ||||
| 328 | + | // drainCommands returns first plus the commands already queued, with runs of | |||
| 329 | + | // pointer motion for the same window collapsed into the latest one. open is | |||
| 330 | + | // false once the channel is closed. | |||
| 331 | + | func drainCommands(first ports.ClientCommand, cmds <-chan ports.ClientCommand) (batch []ports.ClientCommand, open bool) { | |||
| 332 | + | batch = append(batch, first) | |||
| 333 | + | for len(batch) < maxCommandBatch { | |||
| 334 | + | select { | |||
| 335 | + | case c, ok := <-cmds: | |||
| 336 | + | if !ok { | |||
| 337 | + | return batch, false | |||
| 338 | + | } | |||
| 339 | + | if m, isMotion := c.(ports.PointerMotionTo); isMotion { | |||
| 340 | + | if last, lastMotion := batch[len(batch)-1].(ports.PointerMotionTo); lastMotion && last.ID == m.ID { | |||
| 341 | + | batch[len(batch)-1] = m | |||
| 342 | + | continue | |||
| 343 | + | } | |||
| 344 | + | } | |||
| 345 | + | batch = append(batch, c) | |||
| 346 | + | default: | |||
| 347 | + | return batch, true | |||
| 348 | + | } | |||
| 349 | + | } | |||
| 350 | + | return batch, true | |||
| 351 | + | } | |||
| 352 | + | ||||
| 318 | func (s *Server) apply(cmd ports.ClientCommand) { | 353 | func (s *Server) apply(cmd ports.ClientCommand) { | ||
| 319 | switch c := cmd.(type) { | 354 | switch c := cmd.(type) { | ||
| 320 | case ports.PointerFocus: | 355 | case ports.PointerFocus: | ||