forward.go RawHistoryFolder

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, ok := ev.(ports.PointerMotion); ok {
46		f.stats.Motions++
47		if n := len(f.queue); n > 0 {
48			if prev, tail := f.queue[n-1].(ports.PointerMotion); tail {
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				f.queue[n-1] = m
55				f.stats.Coalesced++
56				f.mu.Unlock()
57				return
58			}
59		}
60	}
61	f.queue = append(f.queue, ev)
62	if len(f.queue) > forwardWarn && !f.long {
63		f.long = true
64		f.log.Warn().Str("component", "input").Int("queued", len(f.queue)).Msg("core is not reading input; events queue up")
65	}
66	f.mu.Unlock()
67	select {
68	case f.wake <- struct{}{}:
69	default:
70	}
71}
72
73func (f *forwarder) pop() (ports.InputEvent, bool) {
74	f.mu.Lock()
75	defer f.mu.Unlock()
76	if len(f.queue) == 0 {
77		return nil, false
78	}
79	ev := f.queue[0]
80	f.queue[0] = nil
81	f.queue = f.queue[1:]
82	if f.long && len(f.queue) < forwardWarn {
83		f.long = false
84		f.log.Warn().Str("component", "input").Int("queued", len(f.queue)).Msg("input queue drained")
85	}
86	return ev, true
87}
88
89// take returns and resets the counters.
90func (f *forwarder) take() forwardStats {
91	f.mu.Lock()
92	defer f.mu.Unlock()
93	s := f.stats
94	f.stats = forwardStats{}
95	return s
96}
97
98// run delivers queued events to input until ctx ends.
99func (f *forwarder) run(ctx context.Context, input chan<- ports.InputEvent) {
100	for {
101		ev, ok := f.pop()
102		if !ok {
103			select {
104			case <-ctx.Done():
105				return
106			case <-f.wake:
107				continue
108			}
109		}
110		start := time.Now()
111		select {
112		case input <- ev:
113		case <-ctx.Done():
114			return
115		}
116		d := time.Since(start)
117		f.mu.Lock()
118		f.stats.Sent++
119		f.stats.MaxBlock = max(f.stats.MaxBlock, d)
120		f.mu.Unlock()
121	}
122}