perf(wayland): forward capture replies through an allocation-free post ring

commit: fa9d828b599b8d31d4403ea34764bcd48918e3f4
author: brice · 8d ago · 2026-10-14 18:44 UTC
committer: brice · 8d ago · 2026-10-14 21:39 UTC
parent: 844b5d1
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +282 -9

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 -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)$$' -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 -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)$$' -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 -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)$$' -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 -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)$$' -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

A internal/adapters/wayland/postring.go +98 -0

@@ -0,0 +1,98 @@
	1	+	package wayland
	2	+
	3	+	import (
	4	+	"sync"
	5	+
	6	+	"github.com/bnema/purego-libwayland/server"
	7	+	)
	8	+
	9	+	// postRing carries values from producer goroutines to the display goroutine
	10	+	// 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.
	13	+	//
	14	+	// 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),
	16	+	// so one pending drain serves every value pushed before it runs.
	17	+	type postRing[T any] struct {
	18	+	display *server.Display
	19	+	apply func(T)
	20	+	drainFn func()
	21	+	// space is signalled (never blocking) after each drain and after a push
	22	+	// that left room, to wake blocked pushers one after the other.
	23	+	space chan struct{}
	24	+
	25	+	mu sync.Mutex
	26	+	ring, spare []T
	27	+	limit int
	28	+	pending bool // a drain is posted and has not taken the ring yet
	29	+	}
	30	+
	31	+	// newPostRing returns a ring of limit values whose values apply runs on the
	32	+	// 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] {
	34	+	r := &postRing[T]{
	35	+	display: d,
	36	+	apply: apply,
	37	+	space: make(chan struct{}, 1),
	38	+	ring: make([]T, 0, limit),
	39	+	spare: make([]T, 0, limit),
	40	+	limit: limit,
	41	+	}
	42	+	r.drainFn = r.drain
	43	+	return r
	44	+	}
	45	+
	46	+	// Push queues v for the display goroutine and wakes it. A full ring blocks
	47	+	// the caller, which must not be the display goroutine, until a drain made
	48	+	// room; it reports false, with v dropped, once the display has stopped.
	49	+	func (r *postRing[T]) Push(v T) bool {
	50	+	for {
	51	+	r.mu.Lock()
	52	+	if len(r.ring) < r.limit {
	53	+	r.ring = append(r.ring, v)
	54	+	post := !r.pending
	55	+	r.pending = true
	56	+	room := len(r.ring) < r.limit
	57	+	r.mu.Unlock()
	58	+	if post {
	59	+	r.display.Post(r.drainFn)
	60	+	}
	61	+	if room {
	62	+	r.signal() // chain the wake to another blocked pusher
	63	+	}
	64	+	return true
	65	+	}
	66	+	r.mu.Unlock()
	67	+	select {
	68	+	case <-r.space:
	69	+	case <-r.display.Stopped():
	70	+	return false
	71	+	}
	72	+	}
	73	+	}
	74	+
	75	+	func (r *postRing[T]) signal() {
	76	+	select {
	77	+	case r.space <- struct{}{}:
	78	+	default:
	79	+	}
	80	+	}
	81	+
	82	+	// drain runs on the display goroutine and applies the batch in order.
	83	+	func (r *postRing[T]) drain() {
	84	+	r.mu.Lock()
	85	+	batch := r.ring
	86	+	r.ring, r.spare = r.spare[:0], nil
	87	+	r.pending = false
	88	+	r.mu.Unlock()
	89	+	r.signal()
	90	+	for _, v := range batch {
	91	+	r.apply(v)
	92	+	}
	93	+	// Do not retain what the values reference in the spare batch.
	94	+	clear(batch)
	95	+	r.mu.Lock()
	96	+	r.spare = batch[:0]
	97	+	r.mu.Unlock()
	98	+	}

@@ -0,0 +1,98 @@
	1	+	package wayland
	2	+
	3	+	import (
	4	+	"sync"
	5	+
	6	+	"github.com/bnema/purego-libwayland/server"
	7	+	)
	8	+
	9	+	// postRing carries values from producer goroutines to the display goroutine
	10	+	// 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.
	13	+	//
	14	+	// 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),
	16	+	// so one pending drain serves every value pushed before it runs.
	17	+	type postRing[T any] struct {
	18	+	display *server.Display
	19	+	apply func(T)
	20	+	drainFn func()
	21	+	// space is signalled (never blocking) after each drain and after a push
	22	+	// that left room, to wake blocked pushers one after the other.
	23	+	space chan struct{}
	24	+
	25	+	mu sync.Mutex
	26	+	ring, spare []T
	27	+	limit int
	28	+	pending bool // a drain is posted and has not taken the ring yet
	29	+	}
	30	+
	31	+	// newPostRing returns a ring of limit values whose values apply runs on the
	32	+	// 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] {
	34	+	r := &postRing[T]{
	35	+	display: d,
	36	+	apply: apply,
	37	+	space: make(chan struct{}, 1),
	38	+	ring: make([]T, 0, limit),
	39	+	spare: make([]T, 0, limit),
	40	+	limit: limit,
	41	+	}
	42	+	r.drainFn = r.drain
	43	+	return r
	44	+	}
	45	+
	46	+	// Push queues v for the display goroutine and wakes it. A full ring blocks
	47	+	// the caller, which must not be the display goroutine, until a drain made
	48	+	// room; it reports false, with v dropped, once the display has stopped.
	49	+	func (r *postRing[T]) Push(v T) bool {
	50	+	for {
	51	+	r.mu.Lock()
	52	+	if len(r.ring) < r.limit {
	53	+	r.ring = append(r.ring, v)
	54	+	post := !r.pending
	55	+	r.pending = true
	56	+	room := len(r.ring) < r.limit
	57	+	r.mu.Unlock()
	58	+	if post {
	59	+	r.display.Post(r.drainFn)
	60	+	}
	61	+	if room {
	62	+	r.signal() // chain the wake to another blocked pusher
	63	+	}
	64	+	return true
	65	+	}
	66	+	r.mu.Unlock()
	67	+	select {
	68	+	case <-r.space:
	69	+	case <-r.display.Stopped():
	70	+	return false
	71	+	}
	72	+	}
	73	+	}
	74	+
	75	+	func (r *postRing[T]) signal() {
	76	+	select {
	77	+	case r.space <- struct{}{}:
	78	+	default:
	79	+	}
	80	+	}
	81	+
	82	+	// drain runs on the display goroutine and applies the batch in order.
	83	+	func (r *postRing[T]) drain() {
	84	+	r.mu.Lock()
	85	+	batch := r.ring
	86	+	r.ring, r.spare = r.spare[:0], nil
	87	+	r.pending = false
	88	+	r.mu.Unlock()
	89	+	r.signal()
	90	+	for _, v := range batch {
	91	+	r.apply(v)
	92	+	}
	93	+	// Do not retain what the values reference in the spare batch.
	94	+	clear(batch)
	95	+	r.mu.Lock()
	96	+	r.spare = batch[:0]
	97	+	r.mu.Unlock()
	98	+	}

A internal/adapters/wayland/postring_test.go +160 -0

@@ -0,0 +1,160 @@
	1	+	package wayland
	2	+
	3	+	import (
	4	+	"context"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/neferwl/internal/adapters/logging"
	10	+	"github.com/bnema/neferwl/internal/ports"
	11	+	)
	12	+
	13	+	// Pushing a capture reply and applying it on the display goroutine allocates
	14	+	// nothing in steady state. The count includes the display goroutine:
	15	+	// AllocsPerRun reads the global malloc counter.
	16	+	func TestCapturedForwardAllocations(t *testing.T) {
	17	+	if raceEnabled {
	18	+	t.Skip("the race detector makes the purego wake callback allocate")
	19	+	}
	20	+	s, _, _, _, _, _ := presentServer(t)
	21	+	done := ports.CaptureDone{ID: 77, Output: "HEADLESS-1", Time: time.Unix(1, 2)}
	22	+	// Posts run in order: once the marker ran, the drain before it did. The
	23	+	// marker is bound once and the channel reused, so waiting allocates nothing.
	24	+	ran := make(chan struct{}, 1)
	25	+	marker := func() { ran <- struct{}{} }
	26	+	for range 4 {
	27	+	s.capturedRing.Push(done)
	28	+	s.display.Post(marker)
	29	+	<-ran
	30	+	}
	31	+	allocs := testing.AllocsPerRun(200, func() {
	32	+	s.capturedRing.Push(done)
	33	+	s.display.Post(marker)
	34	+	<-ran
	35	+	})
	36	+	if allocs != 0 {
	37	+	t.Errorf("push + drain: %.1f allocs, want 0", allocs)
	38	+	}
	39	+	}
	40	+
	41	+	// A full ring blocks the push until the display goroutine drains it, and no
	42	+	// value is lost or reordered.
	43	+	func TestPostRingFullBlocksWithoutLoss(t *testing.T) {
	44	+	s, _, _, _, _, _ := presentServer(t)
	45	+	const limit, total = 4, 3*4 + 1
	46	+	var got []int // display goroutine only
	47	+	r := newPostRing(s.display, limit, func(v int) { got = append(got, v) })
	48	+
	49	+	// Stall the display goroutine so nothing drains.
	50	+	release := make(chan struct{})
	51	+	stalled := make(chan struct{})
	52	+	s.display.Post(func() { close(stalled); <-release })
	53	+	<-stalled
	54	+
	55	+	pushed := make(chan int, total)
	56	+	var wg sync.WaitGroup
	57	+	wg.Add(1)
	58	+	go func() {
	59	+	defer wg.Done()
	60	+	for i := range total {
	61	+	if !r.Push(i) {
	62	+	t.Errorf("push %d refused", i)
	63	+	return
	64	+	}
	65	+	pushed <- i
	66	+	}
	67	+	}()
	68	+	for range limit {
	69	+	select {
	70	+	case <-pushed:
	71	+	case <-time.After(2 * time.Second):
	72	+	t.Fatal("pushes below the limit blocked")
	73	+	}
	74	+	}
	75	+	select {
	76	+	case i := <-pushed:
	77	+	t.Fatalf("push %d returned on a full ring", i)
	78	+	case <-time.After(50 * time.Millisecond):
	79	+	}
	80	+	close(release)
	81	+	wg.Wait()
	82	+	// The marker runs after every drain posted before it.
	83	+	var final []int
	84	+	ran := make(chan struct{})
	85	+	for {
	86	+	s.display.Post(func() {
	87	+	if len(got) == total {
	88	+	final = append(final, got...)
	89	+	}
	90	+	ran <- struct{}{}
	91	+	})
	92	+	<-ran
	93	+	if final != nil {
	94	+	break
	95	+	}
	96	+	time.Sleep(time.Millisecond)
	97	+	}
	98	+	for i, v := range final {
	99	+	if v != i {
	100	+	t.Fatalf("applied %v, want 0..%d in order", final, total-1)
	101	+	}
	102	+	}
	103	+	}
	104	+
	105	+	// A push blocked on a full ring resumes once the display drains it.
	106	+	func TestPostRingBlockedPushResumes(t *testing.T) {
	107	+	s, _, _, _, _, _ := presentServer(t)
	108	+	r := newPostRing(s.display, 1, func(int) {})
	109	+	release := make(chan struct{})
	110	+	stalled := make(chan struct{})
	111	+	s.display.Post(func() { close(stalled); <-release })
	112	+	<-stalled
	113	+	if !r.Push(1) {
	114	+	t.Fatal("first push refused")
	115	+	}
	116	+	result := make(chan bool, 1)
	117	+	go func() { result <- r.Push(2) }()
	118	+	select {
	119	+	case ok := <-result:
	120	+	t.Fatalf("push on a full ring returned %v", ok)
	121	+	case <-time.After(50 * time.Millisecond):
	122	+	}
	123	+	close(release)
	124	+	select {
	125	+	case ok := <-result:
	126	+	if !ok {
	127	+	t.Fatal("push refused while the display runs")
	128	+	}
	129	+	case <-time.After(2 * time.Second):
	130	+	t.Fatal("blocked push never resumed")
	131	+	}
	132	+	}
	133	+
	134	+	// A push blocked on a full ring returns false once the display has stopped,
	135	+	// instead of waiting for a drain that never comes.
	136	+	func TestPostRingPushStopsWithDisplay(t *testing.T) {
	137	+	s, err := New(Options{RuntimeDir: t.TempDir()}, Channels{}, logging.For(context.Background(), "wayland"))
	138	+	if err != nil {
	139	+	t.Fatal(err)
	140	+	}
	141	+	ctx, cancel := context.WithCancel(context.Background())
	142	+	done := make(chan error, 1)
	143	+	go func() { done <- s.Run(ctx) }()
	144	+	cancel()
	145	+	<-done
	146	+	r := newPostRing(s.display, 1, func(int) {})
	147	+	if !r.Push(1) { // room: queued, its drain is dropped by the stopped display
	148	+	t.Fatal("first push refused")
	149	+	}
	150	+	result := make(chan bool, 1)
	151	+	go func() { result <- r.Push(2) }()
	152	+	select {
	153	+	case ok := <-result:
	154	+	if ok {
	155	+	t.Fatal("push on a stopped display reported success")
	156	+	}
	157	+	case <-time.After(2 * time.Second):
	158	+	t.Fatal("push blocked after the display stopped")
	159	+	}
	160	+	}

@@ -0,0 +1,160 @@
	1	+	package wayland
	2	+
	3	+	import (
	4	+	"context"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/neferwl/internal/adapters/logging"
	10	+	"github.com/bnema/neferwl/internal/ports"
	11	+	)
	12	+
	13	+	// Pushing a capture reply and applying it on the display goroutine allocates
	14	+	// nothing in steady state. The count includes the display goroutine:
	15	+	// AllocsPerRun reads the global malloc counter.
	16	+	func TestCapturedForwardAllocations(t *testing.T) {
	17	+	if raceEnabled {
	18	+	t.Skip("the race detector makes the purego wake callback allocate")
	19	+	}
	20	+	s, _, _, _, _, _ := presentServer(t)
	21	+	done := ports.CaptureDone{ID: 77, Output: "HEADLESS-1", Time: time.Unix(1, 2)}
	22	+	// Posts run in order: once the marker ran, the drain before it did. The
	23	+	// marker is bound once and the channel reused, so waiting allocates nothing.
	24	+	ran := make(chan struct{}, 1)
	25	+	marker := func() { ran <- struct{}{} }
	26	+	for range 4 {
	27	+	s.capturedRing.Push(done)
	28	+	s.display.Post(marker)
	29	+	<-ran
	30	+	}
	31	+	allocs := testing.AllocsPerRun(200, func() {
	32	+	s.capturedRing.Push(done)
	33	+	s.display.Post(marker)
	34	+	<-ran
	35	+	})
	36	+	if allocs != 0 {
	37	+	t.Errorf("push + drain: %.1f allocs, want 0", allocs)
	38	+	}
	39	+	}
	40	+
	41	+	// A full ring blocks the push until the display goroutine drains it, and no
	42	+	// value is lost or reordered.
	43	+	func TestPostRingFullBlocksWithoutLoss(t *testing.T) {
	44	+	s, _, _, _, _, _ := presentServer(t)
	45	+	const limit, total = 4, 3*4 + 1
	46	+	var got []int // display goroutine only
	47	+	r := newPostRing(s.display, limit, func(v int) { got = append(got, v) })
	48	+
	49	+	// Stall the display goroutine so nothing drains.
	50	+	release := make(chan struct{})
	51	+	stalled := make(chan struct{})
	52	+	s.display.Post(func() { close(stalled); <-release })
	53	+	<-stalled
	54	+
	55	+	pushed := make(chan int, total)
	56	+	var wg sync.WaitGroup
	57	+	wg.Add(1)
	58	+	go func() {
	59	+	defer wg.Done()
	60	+	for i := range total {
	61	+	if !r.Push(i) {
	62	+	t.Errorf("push %d refused", i)
	63	+	return
	64	+	}
	65	+	pushed <- i
	66	+	}
	67	+	}()
	68	+	for range limit {
	69	+	select {
	70	+	case <-pushed:
	71	+	case <-time.After(2 * time.Second):
	72	+	t.Fatal("pushes below the limit blocked")
	73	+	}
	74	+	}
	75	+	select {
	76	+	case i := <-pushed:
	77	+	t.Fatalf("push %d returned on a full ring", i)
	78	+	case <-time.After(50 * time.Millisecond):
	79	+	}
	80	+	close(release)
	81	+	wg.Wait()
	82	+	// The marker runs after every drain posted before it.
	83	+	var final []int
	84	+	ran := make(chan struct{})
	85	+	for {
	86	+	s.display.Post(func() {
	87	+	if len(got) == total {
	88	+	final = append(final, got...)
	89	+	}
	90	+	ran <- struct{}{}
	91	+	})
	92	+	<-ran
	93	+	if final != nil {
	94	+	break
	95	+	}
	96	+	time.Sleep(time.Millisecond)
	97	+	}
	98	+	for i, v := range final {
	99	+	if v != i {
	100	+	t.Fatalf("applied %v, want 0..%d in order", final, total-1)
	101	+	}
	102	+	}
	103	+	}
	104	+
	105	+	// A push blocked on a full ring resumes once the display drains it.
	106	+	func TestPostRingBlockedPushResumes(t *testing.T) {
	107	+	s, _, _, _, _, _ := presentServer(t)
	108	+	r := newPostRing(s.display, 1, func(int) {})
	109	+	release := make(chan struct{})
	110	+	stalled := make(chan struct{})
	111	+	s.display.Post(func() { close(stalled); <-release })
	112	+	<-stalled
	113	+	if !r.Push(1) {
	114	+	t.Fatal("first push refused")
	115	+	}
	116	+	result := make(chan bool, 1)
	117	+	go func() { result <- r.Push(2) }()
	118	+	select {
	119	+	case ok := <-result:
	120	+	t.Fatalf("push on a full ring returned %v", ok)
	121	+	case <-time.After(50 * time.Millisecond):
	122	+	}
	123	+	close(release)
	124	+	select {
	125	+	case ok := <-result:
	126	+	if !ok {
	127	+	t.Fatal("push refused while the display runs")
	128	+	}
	129	+	case <-time.After(2 * time.Second):
	130	+	t.Fatal("blocked push never resumed")
	131	+	}
	132	+	}
	133	+
	134	+	// A push blocked on a full ring returns false once the display has stopped,
	135	+	// instead of waiting for a drain that never comes.
	136	+	func TestPostRingPushStopsWithDisplay(t *testing.T) {
	137	+	s, err := New(Options{RuntimeDir: t.TempDir()}, Channels{}, logging.For(context.Background(), "wayland"))
	138	+	if err != nil {
	139	+	t.Fatal(err)
	140	+	}
	141	+	ctx, cancel := context.WithCancel(context.Background())
	142	+	done := make(chan error, 1)
	143	+	go func() { done <- s.Run(ctx) }()
	144	+	cancel()
	145	+	<-done
	146	+	r := newPostRing(s.display, 1, func(int) {})
	147	+	if !r.Push(1) { // room: queued, its drain is dropped by the stopped display
	148	+	t.Fatal("first push refused")
	149	+	}
	150	+	result := make(chan bool, 1)
	151	+	go func() { result <- r.Push(2) }()
	152	+	select {
	153	+	case ok := <-result:
	154	+	if ok {
	155	+	t.Fatal("push on a stopped display reported success")
	156	+	}
	157	+	case <-time.After(2 * time.Second):
	158	+	t.Fatal("push blocked after the display stopped")
	159	+	}
	160	+	}

M internal/adapters/wayland/screencopy.go +19 -8


@@ -239,6 +239,24 @@ func (s Server) requestCapture(o output, rect image.Rectangle, cursor bool, b
239	239		return 0, false
240	240		}
241	241		}
	242	+
	243	+	// capturedRingSize bounds the replies waiting for the display goroutine: one
	244	+	// per admitted capture, and the cast frames in flight (8).
	245	+	const capturedRingSize = ports.MaxCaptureInflight + 8
	246	+
	247	+	// applyCaptured settles one capture reply on the display goroutine.
	248	+	func (s *Server) applyCaptured(done ports.CaptureDone) {
	249	+	delete(s.captureInflight, done.ID)
	250	+	if fn := s.captureReplies[done.ID]; fn != nil {
	251	+	delete(s.captureReplies, done.ID)
	252	+	fn(done)
	253	+	}
	254	+	s.maybeLocked()
	255	+	}
	256	+
	257	+	// forwardCaptured moves backend replies to the display goroutine through
	258	+	// capturedRing. Each reply releases an admission credit, so none is merged or
	259	+	// dropped; a full ring blocks this goroutine, never the display.
242	260		func (s *Server) forwardCaptured(ctxDone <-chan struct{}) {
243	261		for {
244	262		select {

@@ -250,14 +268,7 @@ func (s *Server) forwardCaptured(ctxDone <-chan struct{}) {
250	268		if !ok {
251	269		return
252	270		}
253		-	if !s.display.Do(func() {
254		-	delete(s.captureInflight, done.ID)
255		-	if fn := s.captureReplies[done.ID]; fn != nil {
256		-	delete(s.captureReplies, done.ID)
257		-	fn(done)
258		-	}
259		-	s.maybeLocked()
260		-	}) {
	271	+	if !s.capturedRing.Push(done) {
261	272		return
262	273		}
263	274		}

 		return 0, false
 	}
 }
+// capturedRingSize bounds the replies waiting for the display goroutine: one
+// per admitted capture, and the cast frames in flight (8).
+const capturedRingSize = ports.MaxCaptureInflight + 8
+// applyCaptured settles one capture reply on the display goroutine.
+func (s *Server) applyCaptured(done ports.CaptureDone) {
+	delete(s.captureInflight, done.ID)
+	if fn := s.captureReplies[done.ID]; fn != nil {
+		delete(s.captureReplies, done.ID)
+		fn(done)
+	}
+	s.maybeLocked()
+}
+// forwardCaptured moves backend replies to the display goroutine through
+// capturedRing. Each reply releases an admission credit, so none is merged or
+// dropped; a full ring blocks this goroutine, never the display.
 func (s *Server) forwardCaptured(ctxDone <-chan struct{}) {
 	for {
 		select {
 			if !ok {
 				return
 			}
-			if !s.display.Do(func() {
+			if !s.capturedRing.Push(done) {
-				delete(s.captureInflight, done.ID)
-				if fn := s.captureReplies[done.ID]; fn != nil {
-					delete(s.captureReplies, done.ID)
-					fn(done)
-				}
-				s.maybeLocked()
-			}) {
 				return
 			}
 		}

M internal/adapters/wayland/server.go +4 -0


@@ -111,6 +111,9 @@ type Server struct {
111	111		nextLeaseID uint64
112	112		nextCapture uint64
113	113		captureReplies map[uint64]func(ports.CaptureDone)
	114	+	// capturedRing hands backend replies to the display goroutine
	115	+	// (forwardCaptured); the display goroutine applies them in order.
	116	+	capturedRing *postRing[ports.CaptureDone]
114	117		// captureInflight counts accepted captures until the backend completes
115	118		// them, independent of client resources; only the display loop touches it.
116	119		captureInflight map[uint64]struct{}

@@ -358,6 +361,7 @@ func New(opts Options, ch Channels, log zerowrap.Logger) (*Server, error) {
358	361		s.paceSignalFn = s.paceSignalled
359	362		s.flushLinksFn = s.flushLinks
360	363		s.sink = newPresentSink(s)
	364	+	s.capturedRing = newPostRing(d, capturedRingSize, s.applyCaptured)
361	365		s.leaseDevices = map[string]*leaseDevice{}
362	366		s.pendingLeases = map[uint64]*leaseObject{}
363	367		s.pendingLeaseCards = map[uint64]string{}


@@ -111,6 +111,9 @@ type Server struct {
111	nextLeaseID uint64	111		nextLeaseID uint64
112	nextCapture uint64	112		nextCapture uint64
113	captureReplies map[uint64]func(ports.CaptureDone)	113		captureReplies map[uint64]func(ports.CaptureDone)
		114	+	// capturedRing hands backend replies to the display goroutine
		115	+	// (forwardCaptured); the display goroutine applies them in order.
		116	+	capturedRing *postRing[ports.CaptureDone]
114	// captureInflight counts accepted captures until the backend completes	117		// captureInflight counts accepted captures until the backend completes
115	// them, independent of client resources; only the display loop touches it.	118		// them, independent of client resources; only the display loop touches it.
116	captureInflight map[uint64]struct{}	119		captureInflight map[uint64]struct{}

@@ -358,6 +361,7 @@ func New(opts Options, ch Channels, log zerowrap.Logger) (*Server, error) {
358	s.paceSignalFn = s.paceSignalled	361		s.paceSignalFn = s.paceSignalled
359	s.flushLinksFn = s.flushLinks	362		s.flushLinksFn = s.flushLinks
360	s.sink = newPresentSink(s)	363		s.sink = newPresentSink(s)
		364	+	s.capturedRing = newPostRing(d, capturedRingSize, s.applyCaptured)
361	s.leaseDevices = map[string]*leaseDevice{}	365		s.leaseDevices = map[string]*leaseDevice{}
362	s.pendingLeases = map[uint64]*leaseObject{}	366		s.pendingLeases = map[uint64]*leaseObject{}
363	s.pendingLeaseCards = map[uint64]string{}	367		s.pendingLeaseCards = map[uint64]string{}