detached 9784d35

refactor(wayland): forward client commands through a ring, not a display round trip

The command forwarder waited in display.Do for each batch, so a display
goroutine busy with client requests stopped it draining core's commands.
Commands now go to a postRing applied on the display goroutine in order;
only the forwarder waits when the ring is full. Adjacent pointer motions of
one window, bare or in the same security envelope, merge in the ring with
their deltas summed, also while the display is busy.
commit
9784d35386ad8b7c79c885eda790e5f069dafcc5
author
brice · · 2026-10-18 04:59 UTC
parent
89b78d7
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
6 files · +292 -69
Unified
UnifiedSplit
M Makefile +1 -1
@@ -69,7 +69,7 @@ adapter-imports:
6969 [ -z "$$bad" ] || { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }
7070 # Fast, mandatory guards for tiled content publications and SHM copies.
7171 perf-check:
72- CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture ./internal/adapters/screencast -run '^(TestAnimateAllocations|TestPublishAllocations|TestPublishSubscribersAllocations|TestPublishStashSlideAllocations|TestPublishOverviewAllocations|TestPublishLeavingAllocations|TestOutputRoutingAllocations|TestReportSeenAllocations|TestFrameLifecycleTransitionsAllocations|TestFullscreenShownAllocations|TestFrameDecisionAllocations|TestCommitFrameAllocations|TestColorPipelineAllocations|TestShownBySnapshotAllocations|TestFlipShowsAllocations|TestMRUNoteFocusDoesNotAllocateWhenSettled|TestDueFramesAllocations|TestRenderSteadyStateAllocations|TestRenderSteadyStateDMABufAllocations|TestSceneOpaqueRegionAllocations|TestSceneDeltaAllocations|TestAccountFlipAllocations|TestFlipDoneAllocations|TestSceneWalkUnchangedTiledSHM|TestTiledCommitPublishAllocations|TestCapturedCommitApplyAllocations|TestCapturedViewportApplyAllocations|TestEffectiveInputEmptyTreeAllocations|TestHeadlessTiledSHMCallbackAndCopyBudget|TestReleaseHeldAllocations|TestCommitDamageAllocations|TestEmitContentAllocations|TestPresentationPushAllocations|TestAnimatingFrameSendAllocations|TestContentBatchCycleAllocations|TestWriteMappedAllocations|TestCapturedForwardAllocations|TestCastFrameAllocations|TestCastProducerAllocations|TestCaptureCycleAllocations|TestLogScenesAllocations|TestOverviewNamedRowsAllocations|TestBufferContentAllocations|TestDueFramesFireScratch|TestRetireSemaphoreAllocatesNothing)$$' -count=1
72+ CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture ./internal/adapters/screencast -run '^(TestAnimateAllocations|TestPublishAllocations|TestPublishSubscribersAllocations|TestPublishStashSlideAllocations|TestPublishOverviewAllocations|TestPublishLeavingAllocations|TestOutputRoutingAllocations|TestReportSeenAllocations|TestFrameLifecycleTransitionsAllocations|TestFullscreenShownAllocations|TestFrameDecisionAllocations|TestCommitFrameAllocations|TestColorPipelineAllocations|TestShownBySnapshotAllocations|TestFlipShowsAllocations|TestMRUNoteFocusDoesNotAllocateWhenSettled|TestDueFramesAllocations|TestRenderSteadyStateAllocations|TestRenderSteadyStateDMABufAllocations|TestSceneOpaqueRegionAllocations|TestSceneDeltaAllocations|TestAccountFlipAllocations|TestFlipDoneAllocations|TestSceneWalkUnchangedTiledSHM|TestTiledCommitPublishAllocations|TestCapturedCommitApplyAllocations|TestCapturedViewportApplyAllocations|TestEffectiveInputEmptyTreeAllocations|TestHeadlessTiledSHMCallbackAndCopyBudget|TestReleaseHeldAllocations|TestCommitDamageAllocations|TestEmitContentAllocations|TestPresentationPushAllocations|TestAnimatingFrameSendAllocations|TestContentBatchCycleAllocations|TestWriteMappedAllocations|TestCapturedForwardAllocations|TestCommandForwardAllocations|TestCastFrameAllocations|TestCastProducerAllocations|TestCaptureCycleAllocations|TestLogScenesAllocations|TestOverviewNamedRowsAllocations|TestBufferContentAllocations|TestDueFramesFireScratch|TestRetireSemaphoreAllocatesNothing)$$' -count=1
7373 check: vet test arch fakes-check log-check perf-check
7474
7575 # On-demand SDR/HDR colour check with examples/testpattern on a headless
@@ -69,7 +69,7 @@ adapter-imports:
69 [ -z "$$bad" ] || { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }69 [ -z "$$bad" ] || { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }
70 # Fast, mandatory guards for tiled content publications and SHM copies.70 # Fast, mandatory guards for tiled content publications and SHM copies.
71 perf-check:71 perf-check:
72- CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture ./internal/adapters/screencast -run '^(TestAnimateAllocations|TestPublishAllocations|TestPublishSubscribersAllocations|TestPublishStashSlideAllocations|TestPublishOverviewAllocations|TestPublishLeavingAllocations|TestOutputRoutingAllocations|TestReportSeenAllocations|TestFrameLifecycleTransitionsAllocations|TestFullscreenShownAllocations|TestFrameDecisionAllocations|TestCommitFrameAllocations|TestColorPipelineAllocations|TestShownBySnapshotAllocations|TestFlipShowsAllocations|TestMRUNoteFocusDoesNotAllocateWhenSettled|TestDueFramesAllocations|TestRenderSteadyStateAllocations|TestRenderSteadyStateDMABufAllocations|TestSceneOpaqueRegionAllocations|TestSceneDeltaAllocations|TestAccountFlipAllocations|TestFlipDoneAllocations|TestSceneWalkUnchangedTiledSHM|TestTiledCommitPublishAllocations|TestCapturedCommitApplyAllocations|TestCapturedViewportApplyAllocations|TestEffectiveInputEmptyTreeAllocations|TestHeadlessTiledSHMCallbackAndCopyBudget|TestReleaseHeldAllocations|TestCommitDamageAllocations|TestEmitContentAllocations|TestPresentationPushAllocations|TestAnimatingFrameSendAllocations|TestContentBatchCycleAllocations|TestWriteMappedAllocations|TestCapturedForwardAllocations|TestCastFrameAllocations|TestCastProducerAllocations|TestCaptureCycleAllocations|TestLogScenesAllocations|TestOverviewNamedRowsAllocations|TestBufferContentAllocations|TestDueFramesFireScratch|TestRetireSemaphoreAllocatesNothing)$$' -count=172+ CGO_ENABLED=0 go test ./internal/core ./internal/app ./internal/adapters/drm ./internal/adapters/headless ./internal/adapters/vulkan ./internal/adapters/wayland ./internal/adapters/outputkit/capture ./internal/adapters/screencast -run '^(TestAnimateAllocations|TestPublishAllocations|TestPublishSubscribersAllocations|TestPublishStashSlideAllocations|TestPublishOverviewAllocations|TestPublishLeavingAllocations|TestOutputRoutingAllocations|TestReportSeenAllocations|TestFrameLifecycleTransitionsAllocations|TestFullscreenShownAllocations|TestFrameDecisionAllocations|TestCommitFrameAllocations|TestColorPipelineAllocations|TestShownBySnapshotAllocations|TestFlipShowsAllocations|TestMRUNoteFocusDoesNotAllocateWhenSettled|TestDueFramesAllocations|TestRenderSteadyStateAllocations|TestRenderSteadyStateDMABufAllocations|TestSceneOpaqueRegionAllocations|TestSceneDeltaAllocations|TestAccountFlipAllocations|TestFlipDoneAllocations|TestSceneWalkUnchangedTiledSHM|TestTiledCommitPublishAllocations|TestCapturedCommitApplyAllocations|TestCapturedViewportApplyAllocations|TestEffectiveInputEmptyTreeAllocations|TestHeadlessTiledSHMCallbackAndCopyBudget|TestReleaseHeldAllocations|TestCommitDamageAllocations|TestEmitContentAllocations|TestPresentationPushAllocations|TestAnimatingFrameSendAllocations|TestContentBatchCycleAllocations|TestWriteMappedAllocations|TestCapturedForwardAllocations|TestCommandForwardAllocations|TestCastFrameAllocations|TestCastProducerAllocations|TestCaptureCycleAllocations|TestLogScenesAllocations|TestOverviewNamedRowsAllocations|TestBufferContentAllocations|TestDueFramesFireScratch|TestRetireSemaphoreAllocatesNothing)$$' -count=1
73 check: vet test arch fakes-check log-check perf-check73 check: vet test arch fakes-check log-check perf-check
74 74
75 # On-demand SDR/HDR colour check with examples/testpattern on a headless75 # On-demand SDR/HDR colour check with examples/testpattern on a headless
M internal/adapters/wayland/alloc_hotpath_test.go +33 -0
@@ -208,3 +208,36 @@ func TestBufferContentAllocations(t *testing.T) {
208208 t.Fatal("content from a truncated pool")
209209 }
210210 }
211+
212+// Forwarding a command allocates nothing on the push path: queueing a
213+// command, folding a motion into the queued one and draining the ring. (Only
214+// applying a merged motion boxes the summed value again, once per merged run,
215+// as the old batch coalescing did once per merge; that is excluded here.)
216+func TestCommandForwardAllocations(t *testing.T) {
217+ if raceEnabled {
218+ t.Skip("the race detector makes the purego wake callback allocate")
219+ }
220+ s, _, _, _, _, _ := presentServer(t)
221+ r := newCommandRing(s.display, func(queuedCommand) {})
222+ var button ports.ClientCommand = ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true}
223+ var motion ports.ClientCommand = ports.PointerMotionTo{ID: 1, DX: 1}
224+ state := ports.SecurityState{Generation: 2, Protected: true}
225+ var wrapped ports.ClientCommand = ports.SecurityCommand{State: state, Command: ports.PointerMotionTo{ID: 1, DX: 1}}
226+ ran := make(chan struct{}, 1)
227+ marker := func() { ran <- struct{}{} }
228+ cycle := func() {
229+ r.Push(queue(button))
230+ r.Push(queue(motion))
231+ r.Push(queue(motion)) // folded into the previous one
232+ r.Push(queue(wrapped))
233+ r.Push(queue(wrapped)) // enveloped, same state: folded as well
234+ s.display.Post(marker)
235+ <-ran
236+ }
237+ for range 4 {
238+ cycle()
239+ }
240+ if allocs := testing.AllocsPerRun(200, cycle); allocs != 0 {
241+ t.Errorf("command push + merge + drain: %.1f allocs, want 0", allocs)
242+ }
243+}
@@ -208,3 +208,36 @@ func TestBufferContentAllocations(t *testing.T) {
208 t.Fatal("content from a truncated pool")208 t.Fatal("content from a truncated pool")
209 }209 }
210 }210 }
211+
212+// Forwarding a command allocates nothing on the push path: queueing a
213+// command, folding a motion into the queued one and draining the ring. (Only
214+// applying a merged motion boxes the summed value again, once per merged run,
215+// as the old batch coalescing did once per merge; that is excluded here.)
216+func TestCommandForwardAllocations(t *testing.T) {
217+ if raceEnabled {
218+ t.Skip("the race detector makes the purego wake callback allocate")
219+ }
220+ s, _, _, _, _, _ := presentServer(t)
221+ r := newCommandRing(s.display, func(queuedCommand) {})
222+ var button ports.ClientCommand = ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true}
223+ var motion ports.ClientCommand = ports.PointerMotionTo{ID: 1, DX: 1}
224+ state := ports.SecurityState{Generation: 2, Protected: true}
225+ var wrapped ports.ClientCommand = ports.SecurityCommand{State: state, Command: ports.PointerMotionTo{ID: 1, DX: 1}}
226+ ran := make(chan struct{}, 1)
227+ marker := func() { ran <- struct{}{} }
228+ cycle := func() {
229+ r.Push(queue(button))
230+ r.Push(queue(motion))
231+ r.Push(queue(motion)) // folded into the previous one
232+ r.Push(queue(wrapped))
233+ r.Push(queue(wrapped)) // enveloped, same state: folded as well
234+ s.display.Post(marker)
235+ <-ran
236+ }
237+ for range 4 {
238+ cycle()
239+ }
240+ if allocs := testing.AllocsPerRun(200, cycle); allocs != 0 {
241+ t.Errorf("command push + merge + drain: %.1f allocs, want 0", allocs)
242+ }
243+}
A internal/adapters/wayland/commandring.go +87 -0
@@ -0,0 +1,87 @@
1+package wayland
2+
3+import (
4+ "github.com/bnema/neferwl/internal/ports"
5+ "github.com/bnema/purego-libwayland/server"
6+)
7+
8+// commandRingSize bounds the client commands waiting for the display
9+// goroutine. It absorbs a few display frames of a 1 kHz mouse and a burst of
10+// key/button commands while the display parses client requests; only the
11+// forwarder goroutine waits when it is full, so core never blocks on the
12+// display directly, only on the Commands channel (cap 32) behind the forwarder.
13+const commandRingSize = 256
14+
15+// queuedCommand is a ClientCommand in the command ring. Pointer motion, bare
16+// or wrapped in a SecurityCommand (core wraps every one in production), keeps
17+// its value and envelope state inline so folding a motion into a queued one
18+// allocates nothing; the command is boxed again only if a merge happened, at
19+// apply time, once per merged run.
20+type queuedCommand struct {
21+ cmd ports.ClientCommand
22+ motion ports.PointerMotionTo // valid if isMotion
23+ state ports.SecurityState // envelope state, valid if enveloped
24+ // isMotion: cmd is a PointerMotionTo, bare or directly inside a
25+ // SecurityCommand (then enveloped is set).
26+ isMotion, enveloped bool
27+ // merged: motion holds the sum of several motions and cmd is stale.
28+ merged bool
29+}
30+
31+func queue(cmd ports.ClientCommand) queuedCommand {
32+ q := queuedCommand{cmd: cmd}
33+ if env, ok := cmd.(ports.SecurityCommand); ok {
34+ q.state, q.enveloped = env.State, true
35+ cmd = env.Command
36+ }
37+ q.motion, q.isMotion = cmd.(ports.PointerMotionTo)
38+ return q
39+}
40+
41+func (q *queuedCommand) command() ports.ClientCommand {
42+ switch {
43+ case !q.merged:
44+ return q.cmd
45+ case q.enveloped:
46+ return ports.SecurityCommand{State: q.state, Command: q.motion}
47+ }
48+ return q.motion
49+}
50+
51+// mergeMotion folds v into last when both are pointer motion for the same
52+// window, both bare or both enveloped with an identical security state, and
53+// last is the most recent queued command, so no other command is ever
54+// overtaken and a bare motion never takes the place of an enveloped one. The
55+// position is the latest; relative deltas add up so locked pointers (games)
56+// lose no movement.
57+func mergeMotion(last *queuedCommand, v queuedCommand) bool {
58+ if !last.isMotion || !v.isMotion || last.motion.ID != v.motion.ID ||
59+ last.enveloped != v.enveloped || last.state != v.state {
60+ return false
61+ }
62+ m := v.motion
63+ m.DX += last.motion.DX
64+ m.DY += last.motion.DY
65+ m.UnaccelDX += last.motion.UnaccelDX
66+ m.UnaccelDY += last.motion.UnaccelDY
67+ last.motion, last.merged = m, true
68+ return true
69+}
70+
71+// newCommandRing returns the ring whose commands apply runs on the display
72+// goroutine, in arrival order. Adjacent motions for one window are coalesced
73+// at push time, against the last queued command only, under the ring mutex.
74+// That also coalesces while the display is busy (a drain-side batch would
75+// only see what the channel held at that moment), and a merge needs no room.
76+func newCommandRing(d *server.Display, apply func(queuedCommand)) *postRing[queuedCommand] {
77+ return newMergingPostRing(d, commandRingSize, apply, mergeMotion)
78+}
79+
80+// applyQueued runs on the display goroutine; it stops applying once the
81+// server context is done, like the per-command check of the old batch.
82+func (s *Server) applyQueued(q queuedCommand) {
83+ if s.ctx != nil && s.ctx.Err() != nil {
84+ return
85+ }
86+ s.apply(q.command())
87+}
@@ -0,0 +1,87 @@
1+package wayland
2+
3+import (
4+ "github.com/bnema/neferwl/internal/ports"
5+ "github.com/bnema/purego-libwayland/server"
6+)
7+
8+// commandRingSize bounds the client commands waiting for the display
9+// goroutine. It absorbs a few display frames of a 1 kHz mouse and a burst of
10+// key/button commands while the display parses client requests; only the
11+// forwarder goroutine waits when it is full, so core never blocks on the
12+// display directly, only on the Commands channel (cap 32) behind the forwarder.
13+const commandRingSize = 256
14+
15+// queuedCommand is a ClientCommand in the command ring. Pointer motion, bare
16+// or wrapped in a SecurityCommand (core wraps every one in production), keeps
17+// its value and envelope state inline so folding a motion into a queued one
18+// allocates nothing; the command is boxed again only if a merge happened, at
19+// apply time, once per merged run.
20+type queuedCommand struct {
21+ cmd ports.ClientCommand
22+ motion ports.PointerMotionTo // valid if isMotion
23+ state ports.SecurityState // envelope state, valid if enveloped
24+ // isMotion: cmd is a PointerMotionTo, bare or directly inside a
25+ // SecurityCommand (then enveloped is set).
26+ isMotion, enveloped bool
27+ // merged: motion holds the sum of several motions and cmd is stale.
28+ merged bool
29+}
30+
31+func queue(cmd ports.ClientCommand) queuedCommand {
32+ q := queuedCommand{cmd: cmd}
33+ if env, ok := cmd.(ports.SecurityCommand); ok {
34+ q.state, q.enveloped = env.State, true
35+ cmd = env.Command
36+ }
37+ q.motion, q.isMotion = cmd.(ports.PointerMotionTo)
38+ return q
39+}
40+
41+func (q *queuedCommand) command() ports.ClientCommand {
42+ switch {
43+ case !q.merged:
44+ return q.cmd
45+ case q.enveloped:
46+ return ports.SecurityCommand{State: q.state, Command: q.motion}
47+ }
48+ return q.motion
49+}
50+
51+// mergeMotion folds v into last when both are pointer motion for the same
52+// window, both bare or both enveloped with an identical security state, and
53+// last is the most recent queued command, so no other command is ever
54+// overtaken and a bare motion never takes the place of an enveloped one. The
55+// position is the latest; relative deltas add up so locked pointers (games)
56+// lose no movement.
57+func mergeMotion(last *queuedCommand, v queuedCommand) bool {
58+ if !last.isMotion || !v.isMotion || last.motion.ID != v.motion.ID ||
59+ last.enveloped != v.enveloped || last.state != v.state {
60+ return false
61+ }
62+ m := v.motion
63+ m.DX += last.motion.DX
64+ m.DY += last.motion.DY
65+ m.UnaccelDX += last.motion.UnaccelDX
66+ m.UnaccelDY += last.motion.UnaccelDY
67+ last.motion, last.merged = m, true
68+ return true
69+}
70+
71+// newCommandRing returns the ring whose commands apply runs on the display
72+// goroutine, in arrival order. Adjacent motions for one window are coalesced
73+// at push time, against the last queued command only, under the ring mutex.
74+// That also coalesces while the display is busy (a drain-side batch would
75+// only see what the channel held at that moment), and a merge needs no room.
76+func newCommandRing(d *server.Display, apply func(queuedCommand)) *postRing[queuedCommand] {
77+ return newMergingPostRing(d, commandRingSize, apply, mergeMotion)
78+}
79+
80+// applyQueued runs on the display goroutine; it stops applying once the
81+// server context is done, like the per-command check of the old batch.
82+func (s *Server) applyQueued(q queuedCommand) {
83+ if s.ctx != nil && s.ctx.Err() != nil {
84+ return
85+ }
86+ s.apply(q.command())
87+}
M internal/adapters/wayland/commands_test.go +141 -15
@@ -10,16 +10,46 @@ import (
1010 "github.com/bnema/neferwl/internal/ports"
1111 )
1212
13-func TestDrainCommandsCollapsesMotion(t *testing.T) {
14- cmds := make(chan ports.ClientCommand, 16)
15- cmds <- ports.PointerMotionTo{ID: 1, X: 2, DX: 0.25, UnaccelDX: 0.5}
16- cmds <- ports.PointerMotionTo{ID: 1, X: 3, DX: 0.25, DY: -0.5, UnaccelDX: 0.5, UnaccelDY: -1}
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, DX: 0.25, UnaccelDX: 0.5}, cmds)
13+// Adjacent motions for one window fold into the last queued command, summing
14+// the relative deltas; nothing is merged across another command, and the
15+// order of arrival is kept.
16+func TestCommandRingCollapsesMotionInOrder(t *testing.T) {
17+ s, _, _, _, _, _ := presentServer(t)
18+ locked := ports.SecurityState{Generation: 2, Protected: true}
19+ unlocked := ports.SecurityState{Generation: 3}
20+ var got []ports.ClientCommand // display goroutine only
21+ r := newCommandRing(s.display, func(q queuedCommand) { got = append(got, q.command()) })
22+ release := make(chan struct{})
23+ stalled := make(chan struct{})
24+ s.display.Post(func() { close(stalled); <-release })
25+ <-stalled
26+ for _, c := range []ports.ClientCommand{
27+ ports.PointerMotionTo{ID: 1, X: 1, DX: 0.25, UnaccelDX: 0.5},
28+ ports.PointerMotionTo{ID: 1, X: 2, DX: 0.25, UnaccelDX: 0.5},
29+ ports.PointerMotionTo{ID: 1, X: 3, DX: 0.25, DY: -0.5, UnaccelDX: 0.5, UnaccelDY: -1},
30+ ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true},
31+ ports.PointerMotionTo{ID: 1, X: 4},
32+ ports.PointerFocus{ID: 2},
33+ ports.PointerMotionTo{ID: 2, X: 5},
34+ ports.PointerMotionTo{ID: 2, X: 6},
35+ ports.PointerMotionTo{ID: 1, X: 7}, // other window: not adjacent-same
36+ // Enveloped, as core sends them: same state merges, a different
37+ // state, a bare motion, or another window does not.
38+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 8, DX: 1, UnaccelDY: 2}},
39+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 9, DX: 2, UnaccelDY: 3}},
40+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 10, DY: 4}},
41+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 11, DX: 1}},
42+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 12, DX: 1}},
43+ ports.PointerMotionTo{ID: 1, X: 13},
44+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 14}},
45+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 2, X: 15}},
46+ ports.SecurityCommand{State: unlocked, Command: ports.PointerButtonTo{ID: 2}},
47+ } {
48+ if !r.Push(queue(c)) {
49+ t.Fatal("push refused")
50+ }
51+ }
52+ close(release)
2353 want := []ports.ClientCommand{
2454 // Slow motions add up: none of the sub-pixel deltas is lost.
2555 ports.PointerMotionTo{ID: 1, X: 3, DX: 0.75, DY: -0.5, UnaccelDX: 1.5, UnaccelDY: -1},
@@ -27,13 +57,109 @@ func TestDrainCommandsCollapsesMotion(t *testing.T) {
2757 ports.PointerMotionTo{ID: 1, X: 4},
2858 ports.PointerFocus{ID: 2},
2959 ports.PointerMotionTo{ID: 2, X: 6},
60+ ports.PointerMotionTo{ID: 1, X: 7},
61+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 10, DX: 3, DY: 4, UnaccelDY: 5}},
62+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 12, DX: 2}},
63+ ports.PointerMotionTo{ID: 1, X: 13},
64+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 14}},
65+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 2, X: 15}},
66+ ports.SecurityCommand{State: unlocked, Command: ports.PointerButtonTo{ID: 2}},
67+ }
68+ var final []ports.ClientCommand
69+ s.display.Do(func() { final = got }) // runs after the drain posted before it
70+ if !reflect.DeepEqual(final, want) {
71+ t.Fatalf("applied %#v, want %#v", final, want)
72+ }
73+}
74+
75+// A merge needs no room: motion keeps folding into the last queued command
76+// while the ring is full of other commands, and the push does not block.
77+func TestCommandRingMergesWhenFull(t *testing.T) {
78+ s, _, _, _, _, _ := presentServer(t)
79+ var got []ports.ClientCommand
80+ r := newCommandRing(s.display, func(q queuedCommand) { got = append(got, q.command()) })
81+ release := make(chan struct{})
82+ stalled := make(chan struct{})
83+ s.display.Post(func() { close(stalled); <-release })
84+ <-stalled
85+ for i := range commandRingSize - 1 {
86+ r.Push(queue(ports.PointerFocus{ID: ports.WindowID(i)}))
87+ }
88+ done := make(chan struct{})
89+ go func() {
90+ defer close(done)
91+ for range 1000 { // fills the ring with one motion, then folds the rest
92+ r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1}))
93+ }
94+ }()
95+ select {
96+ case <-done:
97+ case <-time.After(2 * time.Second):
98+ t.Fatal("motion push blocked on a full ring")
99+ }
100+ close(release)
101+ var last ports.ClientCommand
102+ s.display.Do(func() { last = got[len(got)-1] })
103+ if m, ok := last.(ports.PointerMotionTo); !ok || m.DX != 1000 {
104+ t.Fatalf("last applied %#v, want one motion with DX 1000", last)
105+ }
106+}
107+
108+// A busy display goroutine must not stall the forwarder: it keeps draining
109+// the Commands channel into the ring up to the ring capacity, and only then
110+// does it wait (and core with it, through the full channel).
111+func TestForwarderDrainsWhileDisplayBusy(t *testing.T) {
112+ cmds := make(chan ports.ClientCommand, 32)
113+ s, err := New(Options{RuntimeDir: t.TempDir(), Outputs: testOutputs}, Channels{Commands: cmds}, logging.For(context.Background(), "wayland"))
114+ if err != nil {
115+ t.Fatal(err)
116+ }
117+ ctx, cancel := context.WithCancel(context.Background())
118+ done := make(chan error, 1)
119+ go func() { done <- s.Run(ctx) }()
120+ defer func() { cancel(); <-done }()
121+
122+ release := make(chan struct{})
123+ stalled := make(chan struct{})
124+ s.display.Post(func() { close(stalled); <-release })
125+ <-stalled
126+
127+ // Not mergeable: every command takes a slot.
128+ const total = commandRingSize + 100
129+ sent := make(chan struct{})
130+ go func() {
131+ defer close(sent)
132+ for i := range total {
133+ select {
134+ case cmds <- ports.PointerFocus{ID: ports.WindowID(i + 1)}:
135+ case <-ctx.Done():
136+ return
137+ }
138+ }
139+ }()
140+ queued := func() int {
141+ s.commands.mu.Lock()
142+ defer s.commands.mu.Unlock()
143+ return len(s.commands.ring)
144+ }
145+ // Ring full, one command in the blocked forwarder, a full channel.
146+ deadline := time.Now().Add(5 * time.Second)
147+ for queued() != commandRingSize || len(cmds) != cap(cmds) {
148+ if time.Now().After(deadline) {
149+ t.Fatalf("ring holds %d (want %d), channel %d (want %d)", queued(), commandRingSize, len(cmds), cap(cmds))
150+ }
151+ time.Sleep(time.Millisecond)
30152 }
31- if !open || !reflect.DeepEqual(batch, want) {
32- t.Fatalf("open=%v batch=%#v", open, batch)
153+ select {
154+ case <-sent:
155+ t.Fatal("sender finished although the ring is full")
156+ default:
33157 }
34- close(cmds)
35- if _, open := drainCommands(ports.PointerFocus{}, cmds); open {
36- t.Fatal("closed channel reported open")
158+ close(release)
159+ select {
160+ case <-sent:
161+ case <-time.After(5 * time.Second):
162+ t.Fatal("forwarder did not resume after the display was released")
37163 }
38164 }
39165
@@ -10,16 +10,46 @@ import (
10 "github.com/bnema/neferwl/internal/ports"10 "github.com/bnema/neferwl/internal/ports"
11 )11 )
12 12
13-func TestDrainCommandsCollapsesMotion(t *testing.T) {13+// Adjacent motions for one window fold into the last queued command, summing
14- cmds := make(chan ports.ClientCommand, 16)14+// the relative deltas; nothing is merged across another command, and the
15- cmds <- ports.PointerMotionTo{ID: 1, X: 2, DX: 0.25, UnaccelDX: 0.5}15+// order of arrival is kept.
16- cmds <- ports.PointerMotionTo{ID: 1, X: 3, DX: 0.25, DY: -0.5, UnaccelDX: 0.5, UnaccelDY: -1}16+func TestCommandRingCollapsesMotionInOrder(t *testing.T) {
17- cmds <- ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true}17+ s, _, _, _, _, _ := presentServer(t)
18- cmds <- ports.PointerMotionTo{ID: 1, X: 4}18+ locked := ports.SecurityState{Generation: 2, Protected: true}
19- cmds <- ports.PointerFocus{ID: 2}19+ unlocked := ports.SecurityState{Generation: 3}
20- cmds <- ports.PointerMotionTo{ID: 2, X: 5}20+ var got []ports.ClientCommand // display goroutine only
21- cmds <- ports.PointerMotionTo{ID: 2, X: 6}21+ r := newCommandRing(s.display, func(q queuedCommand) { got = append(got, q.command()) })
22- batch, open := drainCommands(ports.PointerMotionTo{ID: 1, X: 1, DX: 0.25, UnaccelDX: 0.5}, cmds)22+ release := make(chan struct{})
23+ stalled := make(chan struct{})
24+ s.display.Post(func() { close(stalled); <-release })
25+ <-stalled
26+ for _, c := range []ports.ClientCommand{
27+ ports.PointerMotionTo{ID: 1, X: 1, DX: 0.25, UnaccelDX: 0.5},
28+ ports.PointerMotionTo{ID: 1, X: 2, DX: 0.25, UnaccelDX: 0.5},
29+ ports.PointerMotionTo{ID: 1, X: 3, DX: 0.25, DY: -0.5, UnaccelDX: 0.5, UnaccelDY: -1},
30+ ports.PointerButtonTo{ID: 1, Button: 272, Pressed: true},
31+ ports.PointerMotionTo{ID: 1, X: 4},
32+ ports.PointerFocus{ID: 2},
33+ ports.PointerMotionTo{ID: 2, X: 5},
34+ ports.PointerMotionTo{ID: 2, X: 6},
35+ ports.PointerMotionTo{ID: 1, X: 7}, // other window: not adjacent-same
36+ // Enveloped, as core sends them: same state merges, a different
37+ // state, a bare motion, or another window does not.
38+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 8, DX: 1, UnaccelDY: 2}},
39+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 9, DX: 2, UnaccelDY: 3}},
40+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 10, DY: 4}},
41+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 11, DX: 1}},
42+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 12, DX: 1}},
43+ ports.PointerMotionTo{ID: 1, X: 13},
44+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 14}},
45+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 2, X: 15}},
46+ ports.SecurityCommand{State: unlocked, Command: ports.PointerButtonTo{ID: 2}},
47+ } {
48+ if !r.Push(queue(c)) {
49+ t.Fatal("push refused")
50+ }
51+ }
52+ close(release)
23 want := []ports.ClientCommand{53 want := []ports.ClientCommand{
24 // Slow motions add up: none of the sub-pixel deltas is lost.54 // Slow motions add up: none of the sub-pixel deltas is lost.
25 ports.PointerMotionTo{ID: 1, X: 3, DX: 0.75, DY: -0.5, UnaccelDX: 1.5, UnaccelDY: -1},55 ports.PointerMotionTo{ID: 1, X: 3, DX: 0.75, DY: -0.5, UnaccelDX: 1.5, UnaccelDY: -1},
@@ -27,13 +57,109 @@ func TestDrainCommandsCollapsesMotion(t *testing.T) {
27 ports.PointerMotionTo{ID: 1, X: 4},57 ports.PointerMotionTo{ID: 1, X: 4},
28 ports.PointerFocus{ID: 2},58 ports.PointerFocus{ID: 2},
29 ports.PointerMotionTo{ID: 2, X: 6},59 ports.PointerMotionTo{ID: 2, X: 6},
60+ ports.PointerMotionTo{ID: 1, X: 7},
61+ ports.SecurityCommand{State: locked, Command: ports.PointerMotionTo{ID: 1, X: 10, DX: 3, DY: 4, UnaccelDY: 5}},
62+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 12, DX: 2}},
63+ ports.PointerMotionTo{ID: 1, X: 13},
64+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 1, X: 14}},
65+ ports.SecurityCommand{State: unlocked, Command: ports.PointerMotionTo{ID: 2, X: 15}},
66+ ports.SecurityCommand{State: unlocked, Command: ports.PointerButtonTo{ID: 2}},
67+ }
68+ var final []ports.ClientCommand
69+ s.display.Do(func() { final = got }) // runs after the drain posted before it
70+ if !reflect.DeepEqual(final, want) {
71+ t.Fatalf("applied %#v, want %#v", final, want)
72+ }
73+}
74+
75+// A merge needs no room: motion keeps folding into the last queued command
76+// while the ring is full of other commands, and the push does not block.
77+func TestCommandRingMergesWhenFull(t *testing.T) {
78+ s, _, _, _, _, _ := presentServer(t)
79+ var got []ports.ClientCommand
80+ r := newCommandRing(s.display, func(q queuedCommand) { got = append(got, q.command()) })
81+ release := make(chan struct{})
82+ stalled := make(chan struct{})
83+ s.display.Post(func() { close(stalled); <-release })
84+ <-stalled
85+ for i := range commandRingSize - 1 {
86+ r.Push(queue(ports.PointerFocus{ID: ports.WindowID(i)}))
87+ }
88+ done := make(chan struct{})
89+ go func() {
90+ defer close(done)
91+ for range 1000 { // fills the ring with one motion, then folds the rest
92+ r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1}))
93+ }
94+ }()
95+ select {
96+ case <-done:
97+ case <-time.After(2 * time.Second):
98+ t.Fatal("motion push blocked on a full ring")
99+ }
100+ close(release)
101+ var last ports.ClientCommand
102+ s.display.Do(func() { last = got[len(got)-1] })
103+ if m, ok := last.(ports.PointerMotionTo); !ok || m.DX != 1000 {
104+ t.Fatalf("last applied %#v, want one motion with DX 1000", last)
105+ }
106+}
107+
108+// A busy display goroutine must not stall the forwarder: it keeps draining
109+// the Commands channel into the ring up to the ring capacity, and only then
110+// does it wait (and core with it, through the full channel).
111+func TestForwarderDrainsWhileDisplayBusy(t *testing.T) {
112+ cmds := make(chan ports.ClientCommand, 32)
113+ s, err := New(Options{RuntimeDir: t.TempDir(), Outputs: testOutputs}, Channels{Commands: cmds}, logging.For(context.Background(), "wayland"))
114+ if err != nil {
115+ t.Fatal(err)
116+ }
117+ ctx, cancel := context.WithCancel(context.Background())
118+ done := make(chan error, 1)
119+ go func() { done <- s.Run(ctx) }()
120+ defer func() { cancel(); <-done }()
121+
122+ release := make(chan struct{})
123+ stalled := make(chan struct{})
124+ s.display.Post(func() { close(stalled); <-release })
125+ <-stalled
126+
127+ // Not mergeable: every command takes a slot.
128+ const total = commandRingSize + 100
129+ sent := make(chan struct{})
130+ go func() {
131+ defer close(sent)
132+ for i := range total {
133+ select {
134+ case cmds <- ports.PointerFocus{ID: ports.WindowID(i + 1)}:
135+ case <-ctx.Done():
136+ return
137+ }
138+ }
139+ }()
140+ queued := func() int {
141+ s.commands.mu.Lock()
142+ defer s.commands.mu.Unlock()
143+ return len(s.commands.ring)
144+ }
145+ // Ring full, one command in the blocked forwarder, a full channel.
146+ deadline := time.Now().Add(5 * time.Second)
147+ for queued() != commandRingSize || len(cmds) != cap(cmds) {
148+ if time.Now().After(deadline) {
149+ t.Fatalf("ring holds %d (want %d), channel %d (want %d)", queued(), commandRingSize, len(cmds), cap(cmds))
150+ }
151+ time.Sleep(time.Millisecond)
30 }152 }
31- if !open || !reflect.DeepEqual(batch, want) {153+ select {
32- t.Fatalf("open=%v batch=%#v", open, batch)154+ case <-sent:
155+ t.Fatal("sender finished although the ring is full")
156+ default:
33 }157 }
34- close(cmds)158+ close(release)
35- if _, open := drainCommands(ports.PointerFocus{}, cmds); open {159+ select {
36- t.Fatal("closed channel reported open")160+ case <-sent:
161+ case <-time.After(5 * time.Second):
162+ t.Fatal("forwarder did not resume after the display was released")
37 }163 }
38 }164 }
39 165
M internal/adapters/wayland/postring.go +20 -2
@@ -8,8 +8,10 @@ import (
88
99 // postRing carries values from producer goroutines to the display goroutine
1010 // without allocating. It is a transport queue, not window state: the mutex
11-// guards only the slices and the pending flag. Values are never merged or
12-// dropped, in order of arrival.
11+// guards only the slices and the pending flag. Values are applied in order of
12+// arrival and never dropped. A ring built with a merge function may fold a
13+// pushed value into the last queued one (and only that one, under the mutex),
14+// so nothing is ever reordered.
1315 //
1416 // It binds its drain once and posts it at most once per batch: Display.Post
1517 // is unbounded (a Post per value would let a fast producer grow the queue),
@@ -18,6 +20,9 @@ type postRing[T any] struct {
1820 display *server.Display
1921 apply func(T)
2022 drainFn func()
23+ // merge, when set, tries to fold v into the last queued value, which has
24+ // not been drained yet. It runs under mu and must not block or allocate.
25+ merge func(last *T, v T) bool
2126 // space is signalled (never blocking) after each drain and after a push
2227 // that left room, to wake blocked pushers one after the other.
2328 space chan struct{}
@@ -31,6 +36,12 @@ type postRing[T any] struct {
3136 // newPostRing returns a ring of limit values whose values apply runs on the
3237 // display goroutine, in order. limit must be at least 1.
3338 func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T] {
39+ return newMergingPostRing(d, limit, apply, nil)
40+}
41+
42+// newMergingPostRing is newPostRing with a merge function (see postRing). A
43+// merged push needs no room: it succeeds on a full ring.
44+func newMergingPostRing[T any](d *server.Display, limit int, apply func(T), merge func(last *T, v T) bool) *postRing[T] {
3445 r := &postRing[T]{
3546 display: d,
3647 apply: apply,
@@ -38,6 +49,7 @@ func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T
3849 ring: make([]T, 0, limit),
3950 spare: make([]T, 0, limit),
4051 limit: limit,
52+ merge: merge,
4153 }
4254 r.drainFn = r.drain
4355 return r
@@ -49,6 +61,12 @@ func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T
4961 func (r *postRing[T]) Push(v T) bool {
5062 for {
5163 r.mu.Lock()
64+ // A non-empty ring always has its drain posted, so the last value is
65+ // still queued: merging into it needs neither room nor a new post.
66+ if n := len(r.ring); r.merge != nil && n > 0 && r.merge(&r.ring[n-1], v) {
67+ r.mu.Unlock()
68+ return true
69+ }
5270 if len(r.ring) < r.limit {
5371 r.ring = append(r.ring, v)
5472 post := !r.pending
@@ -8,8 +8,10 @@ import (
8 8
9 // postRing carries values from producer goroutines to the display goroutine9 // postRing carries values from producer goroutines to the display goroutine
10 // without allocating. It is a transport queue, not window state: the mutex10 // without allocating. It is a transport queue, not window state: the mutex
11-// guards only the slices and the pending flag. Values are never merged or11+// guards only the slices and the pending flag. Values are applied in order of
12-// dropped, in order of arrival.12+// arrival and never dropped. A ring built with a merge function may fold a
13+// pushed value into the last queued one (and only that one, under the mutex),
14+// so nothing is ever reordered.
13 //15 //
14 // It binds its drain once and posts it at most once per batch: Display.Post16 // It binds its drain once and posts it at most once per batch: Display.Post
15 // is unbounded (a Post per value would let a fast producer grow the queue),17 // is unbounded (a Post per value would let a fast producer grow the queue),
@@ -18,6 +20,9 @@ type postRing[T any] struct {
18 display *server.Display20 display *server.Display
19 apply func(T)21 apply func(T)
20 drainFn func()22 drainFn func()
23+ // merge, when set, tries to fold v into the last queued value, which has
24+ // not been drained yet. It runs under mu and must not block or allocate.
25+ merge func(last *T, v T) bool
21 // space is signalled (never blocking) after each drain and after a push26 // space is signalled (never blocking) after each drain and after a push
22 // that left room, to wake blocked pushers one after the other.27 // that left room, to wake blocked pushers one after the other.
23 space chan struct{}28 space chan struct{}
@@ -31,6 +36,12 @@ type postRing[T any] struct {
31 // newPostRing returns a ring of limit values whose values apply runs on the36 // newPostRing returns a ring of limit values whose values apply runs on the
32 // display goroutine, in order. limit must be at least 1.37 // display goroutine, in order. limit must be at least 1.
33 func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T] {38 func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T] {
39+ return newMergingPostRing(d, limit, apply, nil)
40+}
41+
42+// newMergingPostRing is newPostRing with a merge function (see postRing). A
43+// merged push needs no room: it succeeds on a full ring.
44+func newMergingPostRing[T any](d *server.Display, limit int, apply func(T), merge func(last *T, v T) bool) *postRing[T] {
34 r := &postRing[T]{45 r := &postRing[T]{
35 display: d,46 display: d,
36 apply: apply,47 apply: apply,
@@ -38,6 +49,7 @@ func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T
38 ring: make([]T, 0, limit),49 ring: make([]T, 0, limit),
39 spare: make([]T, 0, limit),50 spare: make([]T, 0, limit),
40 limit: limit,51 limit: limit,
52+ merge: merge,
41 }53 }
42 r.drainFn = r.drain54 r.drainFn = r.drain
43 return r55 return r
@@ -49,6 +61,12 @@ func newPostRing[T any](d *server.Display, limit int, apply func(T)) *postRing[T
49 func (r *postRing[T]) Push(v T) bool {61 func (r *postRing[T]) Push(v T) bool {
50 for {62 for {
51 r.mu.Lock()63 r.mu.Lock()
64+ // A non-empty ring always has its drain posted, so the last value is
65+ // still queued: merging into it needs neither room nor a new post.
66+ if n := len(r.ring); r.merge != nil && n > 0 && r.merge(&r.ring[n-1], v) {
67+ r.mu.Unlock()
68+ return true
69+ }
52 if len(r.ring) < r.limit {70 if len(r.ring) < r.limit {
53 r.ring = append(r.ring, v)71 r.ring = append(r.ring, v)
54 post := !r.pending72 post := !r.pending
M internal/adapters/wayland/server.go +10 -51
@@ -131,10 +131,12 @@ type Server struct {
131131 // Screencast (screencast.go): casts by the adapter's ID and by the ID
132132 // core knows them by; castRing carries the adapter's commands to the
133133 // display goroutine; the castInv fields coalesce inventory publication.
134- casts ports.CastSink
135- castSessions map[uint64]*castSession
136- castByCore map[uint64]*castSession
137- castRing *postRing[ports.CastCommand]
134+ casts ports.CastSink
135+ castSessions map[uint64]*castSession
136+ castByCore map[uint64]*castSession
137+ castRing *postRing[ports.CastCommand]
138+ // commands carries core's client commands to the display goroutine.
139+ commands *postRing[queuedCommand]
138140 castInvFn func()
139141 castInvPosted bool
140142 castInvSent bool
@@ -389,6 +391,7 @@ func New(opts Options, ch Channels, log zerowrap.Logger) (*Server, error) {
389391 s.casts = opts.Casts
390392 s.castSessions, s.castByCore = map[uint64]*castSession{}, map[uint64]*castSession{}
391393 s.castRing = newPostRing(d, castRingSize, s.applyCast)
394+ s.commands = newCommandRing(d, s.applyQueued)
392395 s.castInvFn = s.publishCastInventory
393396 s.leaseDevices = map[string]*leaseDevice{}
394397 s.pendingLeases = map[uint64]*leaseObject{}
@@ -555,18 +558,9 @@ func (s *Server) Run(ctx context.Context) error {
555558 if ctx.Err() != nil {
556559 return
557560 }
558- // One display round trip per batch: Do waits for the event
559- // loop, so a round trip per motion capped pointer events at
560- // ~150/s and let a backlog of stale motions build up.
561- batch, open := drainCommands(cmd, s.channels.Commands)
562- if !s.display.Do(func() {
563- for _, c := range batch {
564- if ctx.Err() != nil {
565- return
566- }
567- s.apply(c)
568- }
569- }) || !open {
561+ // Never wait for the display goroutine: a busy display only
562+ // fills the ring, and only then does this goroutine wait.
563+ if !s.commands.Push(queue(cmd)) {
570564 return
571565 }
572566 }
@@ -746,41 +740,6 @@ func (s *Server) emitContent(c ports.SurfaceContent, d damage) {
746740 s.castCommitted(c.ID)
747741 }
748742
749-// maxCommandBatch bounds how many queued commands one display round trip applies.
750-const maxCommandBatch = 256
751-
752-// drainCommands returns first plus the commands already queued, with runs of
753-// pointer motion for the same window collapsed into the latest one, their
754-// relative deltas summed. open is
755-// false once the channel is closed.
756-func drainCommands(first ports.ClientCommand, cmds <-chan ports.ClientCommand) (batch []ports.ClientCommand, open bool) {
757- batch = append(batch, first)
758- for len(batch) < maxCommandBatch {
759- select {
760- case c, ok := <-cmds:
761- if !ok {
762- return batch, false
763- }
764- if m, isMotion := c.(ports.PointerMotionTo); isMotion {
765- if last, lastMotion := batch[len(batch)-1].(ports.PointerMotionTo); lastMotion && last.ID == m.ID {
766- // The position is the latest; relative deltas add up so
767- // locked pointers (games) lose no movement.
768- m.DX += last.DX
769- m.DY += last.DY
770- m.UnaccelDX += last.UnaccelDX
771- m.UnaccelDY += last.UnaccelDY
772- batch[len(batch)-1] = m
773- continue
774- }
775- }
776- batch = append(batch, c)
777- default:
778- return batch, true
779- }
780- }
781- return batch, true
782-}
783-
784743 func (s *Server) apply(cmd ports.ClientCommand) {
785744 if envelope, ok := cmd.(ports.SecurityCommand); ok {
786745 if s.security == nil || envelope.State != s.security.Snapshot() {
@@ -131,10 +131,12 @@ type Server struct {
131 // Screencast (screencast.go): casts by the adapter's ID and by the ID131 // Screencast (screencast.go): casts by the adapter's ID and by the ID
132 // core knows them by; castRing carries the adapter's commands to the132 // core knows them by; castRing carries the adapter's commands to the
133 // display goroutine; the castInv fields coalesce inventory publication.133 // display goroutine; the castInv fields coalesce inventory publication.
134- casts ports.CastSink134+ casts ports.CastSink
135- castSessions map[uint64]*castSession135+ castSessions map[uint64]*castSession
136- castByCore map[uint64]*castSession136+ castByCore map[uint64]*castSession
137- castRing *postRing[ports.CastCommand]137+ castRing *postRing[ports.CastCommand]
138+ // commands carries core's client commands to the display goroutine.
139+ commands *postRing[queuedCommand]
138 castInvFn func()140 castInvFn func()
139 castInvPosted bool141 castInvPosted bool
140 castInvSent bool142 castInvSent bool
@@ -389,6 +391,7 @@ func New(opts Options, ch Channels, log zerowrap.Logger) (*Server, error) {
389 s.casts = opts.Casts391 s.casts = opts.Casts
390 s.castSessions, s.castByCore = map[uint64]*castSession{}, map[uint64]*castSession{}392 s.castSessions, s.castByCore = map[uint64]*castSession{}, map[uint64]*castSession{}
391 s.castRing = newPostRing(d, castRingSize, s.applyCast)393 s.castRing = newPostRing(d, castRingSize, s.applyCast)
394+ s.commands = newCommandRing(d, s.applyQueued)
392 s.castInvFn = s.publishCastInventory395 s.castInvFn = s.publishCastInventory
393 s.leaseDevices = map[string]*leaseDevice{}396 s.leaseDevices = map[string]*leaseDevice{}
394 s.pendingLeases = map[uint64]*leaseObject{}397 s.pendingLeases = map[uint64]*leaseObject{}
@@ -555,18 +558,9 @@ func (s *Server) Run(ctx context.Context) error {
555 if ctx.Err() != nil {558 if ctx.Err() != nil {
556 return559 return
557 }560 }
558- // One display round trip per batch: Do waits for the event561+ // Never wait for the display goroutine: a busy display only
559- // loop, so a round trip per motion capped pointer events at562+ // fills the ring, and only then does this goroutine wait.
560- // ~150/s and let a backlog of stale motions build up.563+ if !s.commands.Push(queue(cmd)) {
561- batch, open := drainCommands(cmd, s.channels.Commands)
562- if !s.display.Do(func() {
563- for _, c := range batch {
564- if ctx.Err() != nil {
565- return
566- }
567- s.apply(c)
568- }
569- }) || !open {
570 return564 return
571 }565 }
572 }566 }
@@ -746,41 +740,6 @@ func (s *Server) emitContent(c ports.SurfaceContent, d damage) {
746 s.castCommitted(c.ID)740 s.castCommitted(c.ID)
747 }741 }
748 742
749-// maxCommandBatch bounds how many queued commands one display round trip applies.
750-const maxCommandBatch = 256
751-
752-// drainCommands returns first plus the commands already queued, with runs of
753-// pointer motion for the same window collapsed into the latest one, their
754-// relative deltas summed. open is
755-// false once the channel is closed.
756-func drainCommands(first ports.ClientCommand, cmds <-chan ports.ClientCommand) (batch []ports.ClientCommand, open bool) {
757- batch = append(batch, first)
758- for len(batch) < maxCommandBatch {
759- select {
760- case c, ok := <-cmds:
761- if !ok {
762- return batch, false
763- }
764- if m, isMotion := c.(ports.PointerMotionTo); isMotion {
765- if last, lastMotion := batch[len(batch)-1].(ports.PointerMotionTo); lastMotion && last.ID == m.ID {
766- // The position is the latest; relative deltas add up so
767- // locked pointers (games) lose no movement.
768- m.DX += last.DX
769- m.DY += last.DY
770- m.UnaccelDX += last.UnaccelDX
771- m.UnaccelDY += last.UnaccelDY
772- batch[len(batch)-1] = m
773- continue
774- }
775- }
776- batch = append(batch, c)
777- default:
778- return batch, true
779- }
780- }
781- return batch, true
782-}
783-
784 func (s *Server) apply(cmd ports.ClientCommand) {743 func (s *Server) apply(cmd ports.ClientCommand) {
785 if envelope, ok := cmd.(ports.SecurityCommand); ok {744 if envelope, ok := cmd.(ports.SecurityCommand); ok {
786 if s.security == nil || envelope.State != s.security.Snapshot() {745 if s.security == nil || envelope.State != s.security.Snapshot() {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 25

Tags 10

↑↓ moveenter openesc close