c0f82f3 perf(core): reuse the outbox progress signal and skip doomed direct sends · neferwl

M Makefile +1 -1


@@ -69,7 +69,7 @@ adapter-imports:
69	69		[ -z "$$bad" ] \|\| { echo "$$bad"; echo 'adapter imports another adapter: use ports or a shared library (Makefile ADAPTER_LIBS)' >&2; exit 1; }
70	70		# Fast, mandatory guards for tiled content publications and SHM copies.
71	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\|TestCommandForwardAllocations\|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\|TestOutboxQueuedDeliveryAllocations\|TestCastFrameAllocations\|TestCastProducerAllocations\|TestCaptureCycleAllocations\|TestLogScenesAllocations\|TestOverviewNamedRowsAllocations\|TestBufferContentAllocations\|TestDueFramesFireScratch\|TestRetireSemaphoreAllocatesNothing)$$' -count=1
73	73		check: vet test arch fakes-check log-check perf-check
74	74
75	75		# 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\|TestCommandForwardAllocations\|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\|TestOutboxQueuedDeliveryAllocations\|TestCastFrameAllocations\|TestCastProducerAllocations\|TestCaptureCycleAllocations\|TestLogScenesAllocations\|TestOverviewNamedRowsAllocations\|TestBufferContentAllocations\|TestDueFramesFireScratch\|TestRetireSemaphoreAllocatesNothing)$$' -count=1
73		check: vet test arch fakes-check log-check perf-check	73		check: vet test arch fakes-check log-check perf-check
74			74
75		# On-demand SDR/HDR colour check with examples/testpattern on a headless	75		# On-demand SDR/HDR colour check with examples/testpattern on a headless

M internal/app/outputrouting_test.go +18 -4


@@ -3,6 +3,7 @@ package app
3	3		import (
4	4		"context"
5	5		"os"
	6	+	"sync"
6	7		"testing"
7	8		"time"
8	9

@@ -164,8 +165,21 @@ func TestBlockedStartDoesNotDelayRouting(t *testing.T) {
164	165		return nil, ret
165	166		}).Once()
166	167		attacher.EXPECT().DetachOutput(mock.Anything).Maybe()
167		-	started := make(chan bool)
168		-	go func() { started <- set.start(ctx, "B", idleRun) }()
	168	+	releaseB := sync.OnceFunc(func() { close(release) })
	169	+	wait := sync.OnceValue(set.wait)
	170	+	started := make(chan bool, 1)
	171	+	startDone := make(chan struct{})
	172	+	go func() {
	173	+	defer close(startDone)
	174	+	started <- set.start(ctx, "B", idleRun)
	175	+	}()
	176	+	// A failed assertion still unblocks and joins B's start.
	177	+	t.Cleanup(func() {
	178	+	releaseB()
	179	+	cancel()
	180	+	<-startDone
	181	+	_ = wait()
	182	+	})
169	183		<-attaching
170	184		a := set.outs["A"]
171	185		deadline := time.After(2 * time.Second)

@@ -192,11 +206,11 @@ func TestBlockedStartDoesNotDelayRouting(t *testing.T) {
192	206		if got := <-a.cursor; got.Shape != "text" {
193	207		t.Fatalf("A cursor %+v", got)
194	208		}
195		-	close(release)
	209	+	releaseB()
196	210		if !<-started {
197	211		t.Fatal("B did not start")
198	212		}
199		-	if err := set.wait(); err != nil {
	213	+	if err := wait(); err != nil {
200	214		t.Fatal(err)
201	215		}
202	216		}

 import (
 	"context"
 	"os"
+	"sync"
 	"testing"
 	"time"
 		return nil, ret
 	}).Once()
 	attacher.EXPECT().DetachOutput(mock.Anything).Maybe()
-	started := make(chan bool)
+	releaseB := sync.OnceFunc(func() { close(release) })
-	go func() { started <- set.start(ctx, "B", idleRun) }()
+	wait := sync.OnceValue(set.wait)
+	started := make(chan bool, 1)
+	startDone := make(chan struct{})
+	go func() {
+		defer close(startDone)
+		started <- set.start(ctx, "B", idleRun)
+	}()
+	// A failed assertion still unblocks and joins B's start.
+	t.Cleanup(func() {
+		releaseB()
+		cancel()
+		<-startDone
+		_ = wait()
+	})
 	<-attaching
 	a := set.outs["A"]
 	deadline := time.After(2 * time.Second)
 	if got := <-a.cursor; got.Shape != "text" {
 		t.Fatalf("A cursor %+v", got)
 	}
-	close(release)
+	releaseB()
 	if !<-started {
 		t.Fatal("B did not start")
 	}
-	if err := set.wait(); err != nil {
+	if err := wait(); err != nil {
 		t.Fatal(err)
 	}
 }

M internal/core/outbox.go +16 -7


@@ -25,7 +25,8 @@ type outbox struct {
25	25		// sending: a command pushed then must queue behind it.
26	26		busy bool
27	27		// pushed and sent count the entries queued and delivered (a merge adds
28		-	// none); progress is closed and replaced at each delivery, for flush.
	28	+	// none); progress is signalled (never blocking) at each delivery, for
	29	+	// flush. The owner is its only waiter, so one pending signal is enough.
29	30		pushed, sent uint64
30	31		progress chan struct{}
31	32		// running is set while a sender (run) owns delivery.

@@ -63,7 +64,7 @@ func (o *outbox) backlog() uint64 {
63	64		}
64	65
65	66		func newOutbox() outbox {
66		-	return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{})}
	67	+	return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{}, 1)}
67	68		}
68	69
69	70		// command returns what the sender delivers.

@@ -91,7 +92,10 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
91	92		e := outCmd{cmd: v, state: state, secured: secured}
92	93		e.motion, e.isMotion = v.(ports.PointerMotionTo)
93	94		o.mu.Lock()
94		-	if !o.busy && len(o.queue) == 0 {
	95	+	// A full channel would make the attempt fail after building the
	96	+	// envelope; queue at once instead. The length is only a hint: queued
	97	+	// delivery keeps the order either way.
	98	+	if !o.busy && len(o.queue) == 0 && (cap(out) == 0 \|\| len(out) < cap(out)) {
95	99		// A send that cannot block is safe under the lock.
96	100		select {
97	101		case out <- e.command():

@@ -135,7 +139,7 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
135	139		func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
136	140		for {
137	141		o.mu.Lock()
138		-	done, progress, running := o.sent >= seq, o.progress, o.running
	142	+	done, running := o.sent >= seq, o.running
139	143		o.mu.Unlock()
140	144		if done {
141	145		return nil

@@ -149,8 +153,9 @@ func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq
149	153		}
150	154		continue
151	155		}
	156	+	// A signal left from an earlier delivery only costs a recheck.
152	157		select {
153		-	case <-progress:
	158	+	case <-o.progress:
154	159		case <-ctx.Done():
155	160		return ctx.Err()
156	161		}

@@ -178,10 +183,14 @@ func (o *outbox) recycle(batch []outCmd) {
178	183		o.busy = false
179	184		if n > 0 {
180	185		o.sent += n
181		-	close(o.progress)
182		-	o.progress = make(chan struct{})
183	186		}
184	187		o.mu.Unlock()
	188	+	if n > 0 {
	189	+	select {
	190	+	case o.progress <- struct{}{}:
	191	+	default:
	192	+	}
	193	+	}
185	194		}
186	195
187	196		// start marks a sender as running; call it before starting run, so that a

 	// sending: a command pushed then must queue behind it.
 	busy bool
 	// pushed and sent count the entries queued and delivered (a merge adds
-	// none); progress is closed and replaced at each delivery, for flush.
+	// none); progress is signalled (never blocking) at each delivery, for
+	// flush. The owner is its only waiter, so one pending signal is enough.
 	pushed, sent uint64
 	progress     chan struct{}
 	// running is set while a sender (run) owns delivery.
 }
 func newOutbox() outbox {
-	return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{})}
+	return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{}, 1)}
 }
 // command returns what the sender delivers.
 	e := outCmd{cmd: v, state: state, secured: secured}
 	e.motion, e.isMotion = v.(ports.PointerMotionTo)
 	o.mu.Lock()
-	if !o.busy && len(o.queue) == 0 {
+	// A full channel would make the attempt fail after building the
+	// envelope; queue at once instead. The length is only a hint: queued
+	// delivery keeps the order either way.
+	if !o.busy && len(o.queue) == 0 && (cap(out) == 0 || len(out) < cap(out)) {
 		// A send that cannot block is safe under the lock.
 		select {
 		case out <- e.command():
 func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
 	for {
 		o.mu.Lock()
-		done, progress, running := o.sent >= seq, o.progress, o.running
+		done, running := o.sent >= seq, o.running
 		o.mu.Unlock()
 		if done {
 			return nil
 			}
 			continue
 		}
+		// A signal left from an earlier delivery only costs a recheck.
 		select {
-		case <-progress:
+		case <-o.progress:
 		case <-ctx.Done():
 			return ctx.Err()
 		}
 	o.busy = false
 	if n > 0 {
 		o.sent += n
-		close(o.progress)
-		o.progress = make(chan struct{})
 	}
 	o.mu.Unlock()
+	if n > 0 {
+		select {
+		case o.progress <- struct{}{}:
+		default:
+		}
+	}
 }
 // start marks a sender as running; call it before starting run, so that a

M internal/core/outbox_test.go +31 -0


@@ -122,3 +122,34 @@ func TestOutboxIsBounded(t *testing.T) {
122	122		}
123	123		})
124	124		}
	125	+
	126	+	// Queuing a command and delivering it allocates nothing once warm: the
	127	+	// batches are recycled and flush's progress signal is reused.
	128	+	func TestOutboxQueuedDeliveryAllocations(t *testing.T) {
	129	+	o := newOutbox()
	130	+	full := make(chan ports.ClientCommand, 1)
	131	+	full <- ports.FocusWindow{}
	132	+	sink := make(chan ports.ClientCommand, 8)
	133	+	var cmd ports.ClientCommand = ports.FocusWindow{ID: 1}
	134	+	var state ports.SecurityState
	135	+	ctx := context.Background()
	136	+	cycle := func() {
	137	+	// full has no room: push queues, even for a secured command.
	138	+	o.push(full, cmd, state, false)
	139	+	o.push(full, cmd, state, true)
	140	+	o.deliver(ctx, sink, o.take())
	141	+	<-sink
	142	+	<-sink
	143	+	select {
	144	+	case <-o.progress:
	145	+	default:
	146	+	t.Fatal("delivery left no progress signal")
	147	+	}
	148	+	}
	149	+	cycle()
	150	+	// The secured delivery boxes its envelope once; the failed direct send
	151	+	// must not box another.
	152	+	if n := testing.AllocsPerRun(100, cycle); n > 1 {
	153	+	t.Fatalf("%v allocations per queued delivery, want at most 1", n)
	154	+	}
	155	+	}


@@ -122,3 +122,34 @@ func TestOutboxIsBounded(t *testing.T) {
122	}	122		}
123	})	123		})
124	}	124		}
		125	+
		126	+	// Queuing a command and delivering it allocates nothing once warm: the
		127	+	// batches are recycled and flush's progress signal is reused.
		128	+	func TestOutboxQueuedDeliveryAllocations(t *testing.T) {
		129	+	o := newOutbox()
		130	+	full := make(chan ports.ClientCommand, 1)
		131	+	full <- ports.FocusWindow{}
		132	+	sink := make(chan ports.ClientCommand, 8)
		133	+	var cmd ports.ClientCommand = ports.FocusWindow{ID: 1}
		134	+	var state ports.SecurityState
		135	+	ctx := context.Background()
		136	+	cycle := func() {
		137	+	// full has no room: push queues, even for a secured command.
		138	+	o.push(full, cmd, state, false)
		139	+	o.push(full, cmd, state, true)
		140	+	o.deliver(ctx, sink, o.take())
		141	+	<-sink
		142	+	<-sink
		143	+	select {
		144	+	case <-o.progress:
		145	+	default:
		146	+	t.Fatal("delivery left no progress signal")
		147	+	}
		148	+	}
		149	+	cycle()
		150	+	// The secured delivery boxes its envelope once; the failed direct send
		151	+	// must not box another.
		152	+	if n := testing.AllocsPerRun(100, cycle); n > 1 {
		153	+	t.Fatalf("%v allocations per queued delivery, want at most 1", n)
		154	+	}
		155	+	}