1package libinput
2
3import (
4 "context"
5 "sync"
6 "time"
7
8 "github.com/bnema/neferwl/internal/ports"
9 "github.com/bnema/zerowrap"
10)
11
12// forwarder decouples reading libinput from core. libinput keeps reading
13// devices at their full rate and never blocks on core; consecutive pointer
14// motions still waiting to be sent collapse into the latest one, while keys
15// and buttons keep their order relative to motion.
16type forwarder struct {
17 mu sync.Mutex
18 queue []ports.InputEvent
19 wake chan struct{}
20 stats forwardStats
21 // The queue is unbounded (no event is ever dropped); long is set
22 // while it is past forwardWarn entries, warned once each way.
23 log zerowrap.Logger
24 long bool
25}
26
27// forwardWarn is the queue length that means core has stalled.
28const forwardWarn = 4096
29
30// forwardStats is reset by take.
31type forwardStats struct {
32 Motions int // pointer motions read
33 Coalesced int // motions merged into a later one
34 Sent int // events delivered to core
35 MaxBlock time.Duration // longest wait for core to accept one event
36}
37
38func newForwarder(log zerowrap.Logger) *forwarder {
39 return &forwarder{wake: make(chan struct{}, 1), log: log}
40}
41
42// push queues ev without blocking.
43func (f *forwarder) push(ev ports.InputEvent) {
44 f.mu.Lock()
45 if m, state, wrapped, ok := queuedMotion(ev); ok {
46 f.stats.Motions++
47 if n := len(f.queue); n > 0 {
48 if prev, prevState, prevWrapped, tail := queuedMotion(f.queue[n-1]); tail && wrapped == prevWrapped && state == prevState {
49 // Relative deltas add up: games read them, not the position.
50 m.DX += prev.DX
51 m.DY += prev.DY
52 m.UnaccelDX += prev.UnaccelDX
53 m.UnaccelDY += prev.UnaccelDY
54 if wrapped {
55 f.queue[n-1] = ports.SecurityInput{State: state, Event: m}
56 } else {
57 f.queue[n-1] = m
58 }
59 f.stats.Coalesced++
60 f.mu.Unlock()
61 return
62 }
63 }
64 }
65 f.queue = append(f.queue, ev)
66 if len(f.queue) > forwardWarn && !f.long {
67 f.long = true
68 f.log.Warn().Str("component", "input").Int("queued", len(f.queue)).Msg("core is not reading input; events queue up")
69 }
70 f.mu.Unlock()
71 select {
72 case f.wake <- struct{}{}:
73 default:
74 }
75}
76
77// queuedMotion exposes only motion, retaining its exact production epoch.
78// Raw and stamped events never coalesce with each other or across transitions.
79func queuedMotion(ev ports.InputEvent) (ports.PointerMotion, ports.SecurityState, bool, bool) {
80 var state ports.SecurityState
81 wrapped := false
82 if secure, ok := ev.(ports.SecurityInput); ok {
83 state, ev, wrapped = secure.State, secure.Event, true
84 }
85 m, ok := ev.(ports.PointerMotion)
86 return m, state, wrapped, ok
87}
88
89func (f *forwarder) pop() (ports.InputEvent, bool) {
90 f.mu.Lock()
91 defer f.mu.Unlock()
92 if len(f.queue) == 0 {
93 return nil, false
94 }
95 ev := f.queue[0]
96 f.queue[0] = nil
97 f.queue = f.queue[1:]
98 if f.long && len(f.queue) < forwardWarn {
99 f.long = false
100 f.log.Warn().Str("component", "input").Int("queued", len(f.queue)).Msg("input queue drained")
101 }
102 return ev, true
103}
104
105// take returns and resets the counters.
106func (f *forwarder) take() forwardStats {
107 f.mu.Lock()
108 defer f.mu.Unlock()
109 s := f.stats
110 f.stats = forwardStats{}
111 return s
112}
113
114// run delivers queued events to input until ctx ends.
115func (f *forwarder) run(ctx context.Context, input chan<- ports.InputEvent) {
116 for {
117 ev, ok := f.pop()
118 if !ok {
119 select {
120 case <-ctx.Done():
121 return
122 case <-f.wake:
123 continue
124 }
125 }
126 start := time.Now()
127 select {
128 case input <- ev:
129 case <-ctx.Done():
130 return
131 }
132 d := time.Since(start)
133 f.mu.Lock()
134 f.stats.Sent++
135 f.stats.MaxBlock = max(f.stats.MaxBlock, d)
136 f.mu.Unlock()
137 }
138}