Merge pull request #3 from bnema/perf/zero-alloc-callback

perf: reuse PipeWire callback buffer headers

commit: 7f8757719fb2f5c38b005247196d898573d66459
author: bnema · 3mo ago · 2026-07-15 05:14 UTC
committer: GitHub · 3mo ago · 2026-07-15 05:14 UTC
merge of: f33bfb5 + 6504b9c · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown
refs: tag: v0.1.5

4 files · +278 -38

M internal/core/player.go +5 -0


@@ -32,6 +32,10 @@ type player struct {
32	32		paused atomic.Bool
33	33		mu sync.Mutex
34	34
	35	+	// pwView reuses Go slice headers for PipeWire callback buffers. Its native
	36	+	// references are cleared when each callback completes.
	37	+	pwView pwBufferView
	38	+
35	39		// Stream cleanup fields — nil when no stream is active.
36	40		streamOps portout.StreamOps
37	41		streamPtr unsafe.Pointer

@@ -44,6 +48,7 @@ func newPlayer(config PlayerConfig, callbacks PlayerCallbacks) *player {
44	48		p := &player{
45	49		config: config,
46	50		callbacks: callbacks,
	51	+	pwView: newReusablePWBufferView(config.Channels, config.FramesPerBuffer),
47	52		}
48	53		p.state.Store(int32(PlayerStateIdle))
49	54		return p


@@ -32,6 +32,10 @@ type player struct {
32	paused atomic.Bool	32		paused atomic.Bool
33	mu sync.Mutex	33		mu sync.Mutex
34		34
		35	+	// pwView reuses Go slice headers for PipeWire callback buffers. Its native
		36	+	// references are cleared when each callback completes.
		37	+	pwView pwBufferView
		38	+
35	// Stream cleanup fields — nil when no stream is active.	39		// Stream cleanup fields — nil when no stream is active.
36	streamOps portout.StreamOps	40		streamOps portout.StreamOps
37	streamPtr unsafe.Pointer	41		streamPtr unsafe.Pointer

@@ -44,6 +48,7 @@ func newPlayer(config PlayerConfig, callbacks PlayerCallbacks) *player {
44	p := &player{	48		p := &player{
45	config: config,	49		config: config,
46	callbacks: callbacks,	50		callbacks: callbacks,
		51	+	pwView: newReusablePWBufferView(config.Channels, config.FramesPerBuffer),
47	}	52		}
48	p.state.Store(int32(PlayerStateIdle))	53		p.state.Store(int32(PlayerStateIdle))
49	return p	54		return p

A internal/core/player_callback_alloc_test.go +197 -0

@@ -0,0 +1,197 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"unsafe"
	7	+
	8	+	portout "github.com/bnema/purego-pipewire/internal/ports/out"
	9	+	)
	10	+
	11	+	// callbackOps is intentionally allocation-free so callback allocation tests
	12	+	// measure player code rather than a mocking framework.
	13	+	type callbackOps struct {
	14	+	buffers []unsafe.Pointer
	15	+	dequeueAt int
	16	+	queueErr error
	17	+	queued int
	18	+	}
	19	+
	20	+	func (o *callbackOps) CreatePlaybackStream(unsafe.Pointer, string, func()) (unsafe.Pointer, error) {
	21	+	return nil, errors.New("not implemented")
	22	+	}
	23	+	func (o *callbackOps) ConnectPlaybackStream(unsafe.Pointer, portout.PlaybackFormat) error {
	24	+	return errors.New("not implemented")
	25	+	}
	26	+	func (o *callbackOps) SetStreamActive(unsafe.Pointer, bool) error {
	27	+	return errors.New("not implemented")
	28	+	}
	29	+	func (o *callbackOps) DequeueBuffer(unsafe.Pointer) unsafe.Pointer {
	30	+	if o.dequeueAt >= len(o.buffers) {
	31	+	return nil
	32	+	}
	33	+	buf := o.buffers[o.dequeueAt]
	34	+	o.dequeueAt++
	35	+	return buf
	36	+	}
	37	+	func (o *callbackOps) QueueBuffer(unsafe.Pointer, unsafe.Pointer) error {
	38	+	o.queued++
	39	+	return o.queueErr
	40	+	}
	41	+	func (o *callbackOps) DisconnectStream(unsafe.Pointer) error { return nil }
	42	+	func (o *callbackOps) DestroyStream(unsafe.Pointer) {}
	43	+	func (o *callbackOps) CreateMainLoop() (unsafe.Pointer, error) {
	44	+	return nil, errors.New("not implemented")
	45	+	}
	46	+	func (o *callbackOps) RunMainLoop(unsafe.Pointer) error { return nil }
	47	+	func (o *callbackOps) QuitMainLoop(unsafe.Pointer) {}
	48	+	func (o *callbackOps) DestroyMainLoop(unsafe.Pointer) {}
	49	+
	50	+	func callbackBuffer(channels, frames, capacity int) (unsafe.Pointer, [][]float32, []spaData) {
	51	+	samples := make([][]float32, channels)
	52	+	data := make([]spaData, channels)
	53	+	chunks := make([]spaChunk, channels)
	54	+	for ch := range samples {
	55	+	samples[ch] = make([]float32, capacity)
	56	+	data[ch] = spaData{Maxsize: uint32(capacity * 4), Data: unsafe.Pointer(&samples[ch][0]), Chunk: &chunks[ch]}
	57	+	}
	58	+	spa := &spaBuffer{NDatas: uint32(channels), Datas: unsafe.Pointer(&data[0])}
	59	+	return unsafe.Pointer(&pwBuffer{Buffer: spa}), samples, data
	60	+	}
	61	+
	62	+	func newCallbackPlayer(ops callbackOps, fill func(PCMBuffer) (int, error)) *player {
	63	+	p := newPlayer(PlayerConfig{Channels: 2, FramesPerBuffer: 64}, PlayerCallbacks{Fill: fill})
	64	+	p.streamOps = ops
	65	+	p.streamPtr = opaqueTestPtr()
	66	+	p.setState(PlayerStatePlaying)
	67	+	return p
	68	+	}
	69	+
	70	+	// TestProcessCallbackSteadyStateAllocs is deliberately a RED-first regression
	71	+	// test: after its warm-up callback, processing must not allocate.
	72	+	func TestProcessCallbackSteadyStateAllocs(t *testing.T) {
	73	+	buffer, _, _ := callbackBuffer(2, 64, 64)
	74	+	ops := &callbackOps{buffers: make([]unsafe.Pointer, 1002)}
	75	+	for i := range ops.buffers {
	76	+	ops.buffers[i] = buffer
	77	+	}
	78	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
	79	+
	80	+	p.onProcess() // Allocate reusable headers before measuring steady state.
	81	+	allocs := testing.AllocsPerRun(1000, p.onProcess)
	82	+	if allocs != 0 {
	83	+	t.Fatalf("steady-state process callback allocations = %v, want 0", allocs)
	84	+	}
	85	+	}
	86	+
	87	+	func BenchmarkPlayerOnProcessStereo(b *testing.B) {
	88	+	buffer, _, _ := callbackBuffer(2, 64, 64)
	89	+	ops := &callbackOps{buffers: make([]unsafe.Pointer, b.N+1)}
	90	+	for i := range ops.buffers {
	91	+	ops.buffers[i] = buffer
	92	+	}
	93	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
	94	+
	95	+	p.onProcess() // Warm reusable callback metadata before reporting allocations.
	96	+	b.ReportAllocs()
	97	+	b.ResetTimer()
	98	+	for i := 0; i < b.N; i++ {
	99	+	p.onProcess()
	100	+	}
	101	+	}
	102	+
	103	+	func TestProcessCallbackRefreshesChangedPointersAndCapacities(t *testing.T) {
	104	+	first, firstSamples, _ := callbackBuffer(2, 64, 64)
	105	+	second, secondSamples, _ := callbackBuffer(2, 64, 96)
	106	+	ops := &callbackOps{buffers: []unsafe.Pointer{first, second}}
	107	+	calls := 0
	108	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) {
	109	+	calls++
	110	+	for ch := range buf.Samples {
	111	+	buf.Samples[ch][0] = float32(calls)
	112	+	if cap(buf.Samples[ch]) < buf.Frames {
	113	+	t.Fatalf("channel %d capacity = %d, want at least %d", ch, cap(buf.Samples[ch]), buf.Frames)
	114	+	}
	115	+	}
	116	+	return buf.Frames, nil
	117	+	})
	118	+
	119	+	p.onProcess()
	120	+	p.onProcess()
	121	+	if calls != 2 \|\| ops.queued != 2 {
	122	+	t.Fatalf("calls=%d queues=%d, want 2 each", calls, ops.queued)
	123	+	}
	124	+	for ch := range firstSamples {
	125	+	if firstSamples[ch][0] != 1 {
	126	+	t.Errorf("first buffer channel %d = %v, want 1", ch, firstSamples[ch][0])
	127	+	}
	128	+	if secondSamples[ch][0] != 2 {
	129	+	t.Errorf("second buffer channel %d = %v, want 2", ch, secondSamples[ch][0])
	130	+	}
	131	+	}
	132	+	}
	133	+
	134	+	func TestProcessCallbackInvalidMetadataClearsReusableViewAndRequeuesOnce(t *testing.T) {
	135	+	valid, _, _ := callbackBuffer(2, 64, 64)
	136	+	invalid, _, invalidData := callbackBuffer(2, 64, 64)
	137	+	// Fail after channel zero has already been refreshed to prove partial
	138	+	// validation cannot retain a native pointer from the rejected buffer.
	139	+	invalidData[1].Data = nil
	140	+	ops := &callbackOps{buffers: []unsafe.Pointer{valid, invalid}}
	141	+	fillCalls := 0
	142	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) {
	143	+	fillCalls++
	144	+	return buf.Frames, nil
	145	+	})
	146	+
	147	+	p.onProcess()
	148	+	p.setState(PlayerStatePlaying) // inspect the invalid-buffer behavior independently.
	149	+	p.onProcess()
	150	+
	151	+	if fillCalls != 1 {
	152	+	t.Fatalf("Fill calls = %d, want 1 after invalid metadata", fillCalls)
	153	+	}
	154	+	if ops.queued != 2 {
	155	+	t.Fatalf("QueueBuffer calls = %d, want exactly 2", ops.queued)
	156	+	}
	157	+	if p.pwView.buf != nil \|\| p.pwView.data != nil {
	158	+	t.Fatal("invalid metadata retained native buffer references")
	159	+	}
	160	+	for ch, samples := range p.pwView.samples {
	161	+	if samples != nil {
	162	+	t.Fatalf("invalid metadata retained channel %d samples", ch)
	163	+	}
	164	+	}
	165	+	}
	166	+
	167	+	func TestProcessCallbackClearsReusableViewAfterQueueAndQueueError(t *testing.T) {
	168	+	buffer, _, _ := callbackBuffer(2, 64, 64)
	169	+	ops := &callbackOps{buffers: []unsafe.Pointer{buffer, buffer}}
	170	+	clearedDuringError := false
	171	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
	172	+
	173	+	p.onProcess()
	174	+	if p.pwView.buf != nil \|\| p.pwView.data != nil {
	175	+	t.Fatal("successful callback retained reusable native buffer references")
	176	+	}
	177	+	for ch, samples := range p.pwView.samples {
	178	+	if samples != nil {
	179	+	t.Fatalf("successful callback retained channel %d samples", ch)
	180	+	}
	181	+	}
	182	+
	183	+	ops.queueErr = errors.New("queue failed")
	184	+	p.callbacks.OnError = func(error) {
	185	+	clearedDuringError = p.pwView.buf == nil && p.pwView.data == nil
	186	+	for _, samples := range p.pwView.samples {
	187	+	clearedDuringError = clearedDuringError && samples == nil
	188	+	}
	189	+	}
	190	+	p.setState(PlayerStatePlaying)
	191	+	p.onProcess()
	192	+	if !clearedDuringError {
	193	+	t.Fatal("queue error retained reusable native buffer references during OnError")
	194	+	}
	195	+	}
	196	+
	197	+	var _ portout.StreamOps = (*callbackOps)(nil)

@@ -0,0 +1,197 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"unsafe"
	7	+
	8	+	portout "github.com/bnema/purego-pipewire/internal/ports/out"
	9	+	)
	10	+
	11	+	// callbackOps is intentionally allocation-free so callback allocation tests
	12	+	// measure player code rather than a mocking framework.
	13	+	type callbackOps struct {
	14	+	buffers []unsafe.Pointer
	15	+	dequeueAt int
	16	+	queueErr error
	17	+	queued int
	18	+	}
	19	+
	20	+	func (o *callbackOps) CreatePlaybackStream(unsafe.Pointer, string, func()) (unsafe.Pointer, error) {
	21	+	return nil, errors.New("not implemented")
	22	+	}
	23	+	func (o *callbackOps) ConnectPlaybackStream(unsafe.Pointer, portout.PlaybackFormat) error {
	24	+	return errors.New("not implemented")
	25	+	}
	26	+	func (o *callbackOps) SetStreamActive(unsafe.Pointer, bool) error {
	27	+	return errors.New("not implemented")
	28	+	}
	29	+	func (o *callbackOps) DequeueBuffer(unsafe.Pointer) unsafe.Pointer {
	30	+	if o.dequeueAt >= len(o.buffers) {
	31	+	return nil
	32	+	}
	33	+	buf := o.buffers[o.dequeueAt]
	34	+	o.dequeueAt++
	35	+	return buf
	36	+	}
	37	+	func (o *callbackOps) QueueBuffer(unsafe.Pointer, unsafe.Pointer) error {
	38	+	o.queued++
	39	+	return o.queueErr
	40	+	}
	41	+	func (o *callbackOps) DisconnectStream(unsafe.Pointer) error { return nil }
	42	+	func (o *callbackOps) DestroyStream(unsafe.Pointer) {}
	43	+	func (o *callbackOps) CreateMainLoop() (unsafe.Pointer, error) {
	44	+	return nil, errors.New("not implemented")
	45	+	}
	46	+	func (o *callbackOps) RunMainLoop(unsafe.Pointer) error { return nil }
	47	+	func (o *callbackOps) QuitMainLoop(unsafe.Pointer) {}
	48	+	func (o *callbackOps) DestroyMainLoop(unsafe.Pointer) {}
	49	+
	50	+	func callbackBuffer(channels, frames, capacity int) (unsafe.Pointer, [][]float32, []spaData) {
	51	+	samples := make([][]float32, channels)
	52	+	data := make([]spaData, channels)
	53	+	chunks := make([]spaChunk, channels)
	54	+	for ch := range samples {
	55	+	samples[ch] = make([]float32, capacity)
	56	+	data[ch] = spaData{Maxsize: uint32(capacity * 4), Data: unsafe.Pointer(&samples[ch][0]), Chunk: &chunks[ch]}
	57	+	}
	58	+	spa := &spaBuffer{NDatas: uint32(channels), Datas: unsafe.Pointer(&data[0])}
	59	+	return unsafe.Pointer(&pwBuffer{Buffer: spa}), samples, data
	60	+	}
	61	+
	62	+	func newCallbackPlayer(ops callbackOps, fill func(PCMBuffer) (int, error)) *player {
	63	+	p := newPlayer(PlayerConfig{Channels: 2, FramesPerBuffer: 64}, PlayerCallbacks{Fill: fill})
	64	+	p.streamOps = ops
	65	+	p.streamPtr = opaqueTestPtr()
	66	+	p.setState(PlayerStatePlaying)
	67	+	return p
	68	+	}
	69	+
	70	+	// TestProcessCallbackSteadyStateAllocs is deliberately a RED-first regression
	71	+	// test: after its warm-up callback, processing must not allocate.
	72	+	func TestProcessCallbackSteadyStateAllocs(t *testing.T) {
	73	+	buffer, _, _ := callbackBuffer(2, 64, 64)
	74	+	ops := &callbackOps{buffers: make([]unsafe.Pointer, 1002)}
	75	+	for i := range ops.buffers {
	76	+	ops.buffers[i] = buffer
	77	+	}
	78	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
	79	+
	80	+	p.onProcess() // Allocate reusable headers before measuring steady state.
	81	+	allocs := testing.AllocsPerRun(1000, p.onProcess)
	82	+	if allocs != 0 {
	83	+	t.Fatalf("steady-state process callback allocations = %v, want 0", allocs)
	84	+	}
	85	+	}
	86	+
	87	+	func BenchmarkPlayerOnProcessStereo(b *testing.B) {
	88	+	buffer, _, _ := callbackBuffer(2, 64, 64)
	89	+	ops := &callbackOps{buffers: make([]unsafe.Pointer, b.N+1)}
	90	+	for i := range ops.buffers {
	91	+	ops.buffers[i] = buffer
	92	+	}
	93	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
	94	+
	95	+	p.onProcess() // Warm reusable callback metadata before reporting allocations.
	96	+	b.ReportAllocs()
	97	+	b.ResetTimer()
	98	+	for i := 0; i < b.N; i++ {
	99	+	p.onProcess()
	100	+	}
	101	+	}
	102	+
	103	+	func TestProcessCallbackRefreshesChangedPointersAndCapacities(t *testing.T) {
	104	+	first, firstSamples, _ := callbackBuffer(2, 64, 64)
	105	+	second, secondSamples, _ := callbackBuffer(2, 64, 96)
	106	+	ops := &callbackOps{buffers: []unsafe.Pointer{first, second}}
	107	+	calls := 0
	108	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) {
	109	+	calls++
	110	+	for ch := range buf.Samples {
	111	+	buf.Samples[ch][0] = float32(calls)
	112	+	if cap(buf.Samples[ch]) < buf.Frames {
	113	+	t.Fatalf("channel %d capacity = %d, want at least %d", ch, cap(buf.Samples[ch]), buf.Frames)
	114	+	}
	115	+	}
	116	+	return buf.Frames, nil
	117	+	})
	118	+
	119	+	p.onProcess()
	120	+	p.onProcess()
	121	+	if calls != 2 \|\| ops.queued != 2 {
	122	+	t.Fatalf("calls=%d queues=%d, want 2 each", calls, ops.queued)
	123	+	}
	124	+	for ch := range firstSamples {
	125	+	if firstSamples[ch][0] != 1 {
	126	+	t.Errorf("first buffer channel %d = %v, want 1", ch, firstSamples[ch][0])
	127	+	}
	128	+	if secondSamples[ch][0] != 2 {
	129	+	t.Errorf("second buffer channel %d = %v, want 2", ch, secondSamples[ch][0])
	130	+	}
	131	+	}
	132	+	}
	133	+
	134	+	func TestProcessCallbackInvalidMetadataClearsReusableViewAndRequeuesOnce(t *testing.T) {
	135	+	valid, _, _ := callbackBuffer(2, 64, 64)
	136	+	invalid, _, invalidData := callbackBuffer(2, 64, 64)
	137	+	// Fail after channel zero has already been refreshed to prove partial
	138	+	// validation cannot retain a native pointer from the rejected buffer.
	139	+	invalidData[1].Data = nil
	140	+	ops := &callbackOps{buffers: []unsafe.Pointer{valid, invalid}}
	141	+	fillCalls := 0
	142	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) {
	143	+	fillCalls++
	144	+	return buf.Frames, nil
	145	+	})
	146	+
	147	+	p.onProcess()
	148	+	p.setState(PlayerStatePlaying) // inspect the invalid-buffer behavior independently.
	149	+	p.onProcess()
	150	+
	151	+	if fillCalls != 1 {
	152	+	t.Fatalf("Fill calls = %d, want 1 after invalid metadata", fillCalls)
	153	+	}
	154	+	if ops.queued != 2 {
	155	+	t.Fatalf("QueueBuffer calls = %d, want exactly 2", ops.queued)
	156	+	}
	157	+	if p.pwView.buf != nil \|\| p.pwView.data != nil {
	158	+	t.Fatal("invalid metadata retained native buffer references")
	159	+	}
	160	+	for ch, samples := range p.pwView.samples {
	161	+	if samples != nil {
	162	+	t.Fatalf("invalid metadata retained channel %d samples", ch)
	163	+	}
	164	+	}
	165	+	}
	166	+
	167	+	func TestProcessCallbackClearsReusableViewAfterQueueAndQueueError(t *testing.T) {
	168	+	buffer, _, _ := callbackBuffer(2, 64, 64)
	169	+	ops := &callbackOps{buffers: []unsafe.Pointer{buffer, buffer}}
	170	+	clearedDuringError := false
	171	+	p := newCallbackPlayer(ops, func(buf *PCMBuffer) (int, error) { return buf.Frames, nil })
	172	+
	173	+	p.onProcess()
	174	+	if p.pwView.buf != nil \|\| p.pwView.data != nil {
	175	+	t.Fatal("successful callback retained reusable native buffer references")
	176	+	}
	177	+	for ch, samples := range p.pwView.samples {
	178	+	if samples != nil {
	179	+	t.Fatalf("successful callback retained channel %d samples", ch)
	180	+	}
	181	+	}
	182	+
	183	+	ops.queueErr = errors.New("queue failed")
	184	+	p.callbacks.OnError = func(error) {
	185	+	clearedDuringError = p.pwView.buf == nil && p.pwView.data == nil
	186	+	for _, samples := range p.pwView.samples {
	187	+	clearedDuringError = clearedDuringError && samples == nil
	188	+	}
	189	+	}
	190	+	p.setState(PlayerStatePlaying)
	191	+	p.onProcess()
	192	+	if !clearedDuringError {
	193	+	t.Fatal("queue error retained reusable native buffer references during OnError")
	194	+	}
	195	+	}
	196	+
	197	+	var _ portout.StreamOps = (*callbackOps)(nil)

M internal/core/player_rt.go +10 -6


@@ -163,6 +163,9 @@ func (p *player) fail(err error) {
163	163		// queues the buffer back. If any step fails, it routes the error
164	164		// through p.fail.
165	165		func (p *player) onProcess() {
	166	+	// Drop native references before returning from each callback.
	167	+	defer p.pwView.clear()
	168	+
166	169		ops := p.streamOps
167	170		if ops == nil {
168	171		return

@@ -180,24 +183,25 @@ func (p *player) onProcess() {
180	183
181	184		queueOnError := true
182	185		defer func() {
183		-	if err := ops.QueueBuffer(sp, bufPtr); err != nil && queueOnError {
	186	+	err := ops.QueueBuffer(sp, bufPtr)
	187	+	if err != nil && queueOnError {
	188	+	// Clear before OnError can reenter player methods.
	189	+	p.pwView.clear()
184	190		p.fail(fmt.Errorf("queue buffer: %w", err))
185	191		}
186	192		}()
187	193
188		-	frames := p.config.FramesPerBuffer
189		-	view, err := newPWBufferView(bufPtr, p.config.Channels, frames)
190		-	if err != nil {
	194	+	if err := p.pwView.refresh(bufPtr); err != nil {
191	195		queueOnError = false
192	196		p.fail(fmt.Errorf("buffer view: %w", err))
193	197		return
194	198		}
195	199
196		-	frames, err = p.processPCM(view.PCM())
	200	+	frames, err := p.processPCM(p.pwView.PCM())
197	201		if err != nil {
198	202		queueOnError = false
199	203		return
200	204		}
201	205
202		-	view.Commit(frames)
	206	+	p.pwView.Commit(frames)
203	207		}

 // queues the buffer back. If any step fails, it routes the error
 // through p.fail.
 func (p *player) onProcess() {
+	// Drop native references before returning from each callback.
+	defer p.pwView.clear()
 	ops := p.streamOps
 	if ops == nil {
 		return
 	queueOnError := true
 	defer func() {
-		if err := ops.QueueBuffer(sp, bufPtr); err != nil && queueOnError {
+		err := ops.QueueBuffer(sp, bufPtr)
+		if err != nil && queueOnError {
+			// Clear before OnError can reenter player methods.
+			p.pwView.clear()
 			p.fail(fmt.Errorf("queue buffer: %w", err))
 		}
 	}()
-	frames := p.config.FramesPerBuffer
+	if err := p.pwView.refresh(bufPtr); err != nil {
-	view, err := newPWBufferView(bufPtr, p.config.Channels, frames)
-	if err != nil {
 		queueOnError = false
 		p.fail(fmt.Errorf("buffer view: %w", err))
 		return
 	}
-	frames, err = p.processPCM(view.PCM())
+	frames, err := p.processPCM(p.pwView.PCM())
 	if err != nil {
 		queueOnError = false
 		return
 	}
-	view.Commit(frames)
+	p.pwView.Commit(frames)
 }

M internal/core/pwbuffer.go +66 -32


@@ -66,55 +66,89 @@ var (
66	66		ErrFrameOverflow = errors.New("requested frames exceed buffer capacity")
67	67		)
68	68
	69	+	// newReusablePWBufferView allocates the Go slice headers once per player.
	70	+	// refresh replaces their native backing pointers for each dequeued buffer.
	71	+	func newReusablePWBufferView(channels, frames int) pwBufferView {
	72	+	// Preserve constructor behavior for invalid channel counts: refresh will
	73	+	// reject them against the native metadata instead of panicking in make.
	74	+	var samples [][]float32
	75	+	if channels > 0 {
	76	+	samples = make([][]float32, channels)
	77	+	}
	78	+	return pwBufferView{
	79	+	channels: channels,
	80	+	frames: frames,
	81	+	pcm: PCMBuffer{
	82	+	Frames: frames,
	83	+	Channels: channels,
	84	+	Stride: 4,
	85	+	Samples: samples,
	86	+	},
	87	+	samples: samples,
	88	+	}
	89	+	}
	90	+
69	91		// newPWBufferView interprets bufPtr as a *pwBuffer and returns a view that
70		-	// exposes planar float32 samples for each channel. The caller must ensure
71		-	// bufPtr is a valid pw_buffer pointer returned by pw_stream_dequeue_buffer.
72		-	// The view holds a pointer directly into the original pwBuffer so that Commit
73		-	// writes back into the shared memory structure, not a Go copy.
	92	+	// exposes planar float32 samples for each channel. It is retained for callers
	93	+	// that need a standalone view; the realtime player instead reuses one view.
74	94		func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView, error) {
	95	+	view := newReusablePWBufferView(channels, frames)
	96	+	if err := view.refresh(bufPtr); err != nil {
	97	+	return nil, err
	98	+	}
	99	+	return &view, nil
	100	+	}
	101	+
	102	+	// refresh updates a reusable view from a dequeued native buffer. It clears all
	103	+	// native references before validation, so invalid metadata can never leave a
	104	+	// later callback writing through stale pointers.
	105	+	func (v *pwBufferView) refresh(bufPtr unsafe.Pointer) error {
	106	+	v.clear()
75	107		if bufPtr == nil {
76		-	return nil, ErrNilBufferPointer
	108	+	return ErrNilBufferPointer
77	109		}
78	110
79	111		pw := (*pwBuffer)(bufPtr)
80	112		if pw.Buffer == nil {
81		-	return nil, ErrNilBufferPointer
	113	+	return ErrNilBufferPointer
82	114		}
83		-
84		-	if pw.Buffer.NDatas != uint32(channels) {
85		-	return nil, ErrChannelMismatch
	115	+	if pw.Buffer.NDatas != uint32(v.channels) {
	116	+	return ErrChannelMismatch
	117	+	}
	118	+	if v.channels > 0 && pw.Buffer.Datas == nil {
	119	+	return ErrUnmappedData
86	120		}
87	121
88		-	// Datas is a pointer to a C array of spaData structs.
	122	+	// Datas is a pointer to a C array of spaData structs. unsafe.Slice creates
	123	+	// only a slice header; it does not copy or allocate the native metadata.
89	124		dataSlice := unsafe.Slice((*spaData)(pw.Buffer.Datas), int(pw.Buffer.NDatas))
90		-
91		-	samples := make([][]float32, channels)
92		-	for i := 0; i < channels; i++ {
93		-	data := dataSlice[i]
	125	+	for i := 0; i < v.channels; i++ {
	126	+	data := &dataSlice[i]
94	127		if data.Data == nil {
95		-	return nil, ErrUnmappedData
	128	+	v.clear()
	129	+	return ErrUnmappedData
96	130		}
97		-	// Reject if the buffer cannot hold the requested number of frames.
98	131		maxFrames := int(data.Maxsize) / 4
99		-	if frames > maxFrames {
100		-	return nil, fmt.Errorf("%w: requested %d frames but buffer holds %d", ErrFrameOverflow, frames, maxFrames)
	132	+	if v.frames > maxFrames {
	133	+	v.clear()
	134	+	return fmt.Errorf("%w: requested %d frames but buffer holds %d", ErrFrameOverflow, v.frames, maxFrames)
101	135		}
102		-	samples[i] = unsafe.Slice((*float32)(data.Data), frames)
	136	+	// The length and capacity intentionally match the configured callback
	137	+	// frame count, preserving the previous view contract even if PipeWire
	138	+	// supplied a larger mapped buffer.
	139	+	v.samples[i] = unsafe.Slice((*float32)(data.Data), v.frames)
103	140		}
	141	+	v.buf = pw
	142	+	v.data = dataSlice
	143	+	return nil
	144	+	}
104	145
105		-	return &pwBufferView{
106		-	buf: pw,
107		-	channels: channels,
108		-	frames: frames,
109		-	data: dataSlice,
110		-	pcm: PCMBuffer{
111		-	Frames: frames,
112		-	Channels: channels,
113		-	Stride: 4,
114		-	Samples: samples,
115		-	},
116		-	samples: samples,
117		-	}, nil
	146	+	func (v *pwBufferView) clear() {
	147	+	v.buf = nil
	148	+	v.data = nil
	149	+	for i := range v.samples {
	150	+	v.samples[i] = nil
	151	+	}
118	152		}
119	153
120	154		// PCM returns a PCMBuffer pointing at the buffer's planar float32 data.


@@ -66,55 +66,89 @@ var (
66		ErrFrameOverflow = errors.New("requested frames exceed buffer capacity")	66		ErrFrameOverflow = errors.New("requested frames exceed buffer capacity")
67		)	67		)
68			68
			69	+	// newReusablePWBufferView allocates the Go slice headers once per player.
			70	+	// refresh replaces their native backing pointers for each dequeued buffer.
			71	+	func newReusablePWBufferView(channels, frames int) pwBufferView {
			72	+	// Preserve constructor behavior for invalid channel counts: refresh will
			73	+	// reject them against the native metadata instead of panicking in make.
			74	+	var samples [][]float32
			75	+	if channels > 0 {
			76	+	samples = make([][]float32, channels)
			77	+	}
			78	+	return pwBufferView{
			79	+	channels: channels,
			80	+	frames: frames,
			81	+	pcm: PCMBuffer{
			82	+	Frames: frames,
			83	+	Channels: channels,
			84	+	Stride: 4,
			85	+	Samples: samples,
			86	+	},
			87	+	samples: samples,
			88	+	}
			89	+	}
			90	+
69		// newPWBufferView interprets bufPtr as a *pwBuffer and returns a view that	91		// newPWBufferView interprets bufPtr as a *pwBuffer and returns a view that
70	-	// exposes planar float32 samples for each channel. The caller must ensure	92	+	// exposes planar float32 samples for each channel. It is retained for callers
71	-	// bufPtr is a valid pw_buffer pointer returned by pw_stream_dequeue_buffer.	93	+	// that need a standalone view; the realtime player instead reuses one view.
72	-	// The view holds a pointer directly into the original pwBuffer so that Commit
73	-	// writes back into the shared memory structure, not a Go copy.
74		func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView, error) {	94		func newPWBufferView(bufPtr unsafe.Pointer, channels, frames int) (*pwBufferView, error) {
			95	+	view := newReusablePWBufferView(channels, frames)
			96	+	if err := view.refresh(bufPtr); err != nil {
			97	+	return nil, err
			98	+	}
			99	+	return &view, nil
			100	+	}
			101	+
			102	+	// refresh updates a reusable view from a dequeued native buffer. It clears all
			103	+	// native references before validation, so invalid metadata can never leave a
			104	+	// later callback writing through stale pointers.
			105	+	func (v *pwBufferView) refresh(bufPtr unsafe.Pointer) error {
			106	+	v.clear()
75		if bufPtr == nil {	107		if bufPtr == nil {
76	-	return nil, ErrNilBufferPointer	108	+	return ErrNilBufferPointer
77		}	109		}
78			110
79		pw := (*pwBuffer)(bufPtr)	111		pw := (*pwBuffer)(bufPtr)
80		if pw.Buffer == nil {	112		if pw.Buffer == nil {
81	-	return nil, ErrNilBufferPointer	113	+	return ErrNilBufferPointer
82		}	114		}
83	-		115	+	if pw.Buffer.NDatas != uint32(v.channels) {
84	-	if pw.Buffer.NDatas != uint32(channels) {	116	+	return ErrChannelMismatch
85	-	return nil, ErrChannelMismatch	117	+	}
			118	+	if v.channels > 0 && pw.Buffer.Datas == nil {
			119	+	return ErrUnmappedData
86		}	120		}
87			121
88	-	// Datas is a pointer to a C array of spaData structs.	122	+	// Datas is a pointer to a C array of spaData structs. unsafe.Slice creates
			123	+	// only a slice header; it does not copy or allocate the native metadata.
89		dataSlice := unsafe.Slice((*spaData)(pw.Buffer.Datas), int(pw.Buffer.NDatas))	124		dataSlice := unsafe.Slice((*spaData)(pw.Buffer.Datas), int(pw.Buffer.NDatas))
90	-		125	+	for i := 0; i < v.channels; i++ {
91	-	samples := make([][]float32, channels)	126	+	data := &dataSlice[i]
92	-	for i := 0; i < channels; i++ {
93	-	data := dataSlice[i]
94		if data.Data == nil {	127		if data.Data == nil {
95	-	return nil, ErrUnmappedData	128	+	v.clear()
			129	+	return ErrUnmappedData
96		}	130		}
97	-	// Reject if the buffer cannot hold the requested number of frames.
98		maxFrames := int(data.Maxsize) / 4	131		maxFrames := int(data.Maxsize) / 4
99	-	if frames > maxFrames {	132	+	if v.frames > maxFrames {
100	-	return nil, fmt.Errorf("%w: requested %d frames but buffer holds %d", ErrFrameOverflow, frames, maxFrames)	133	+	v.clear()
			134	+	return fmt.Errorf("%w: requested %d frames but buffer holds %d", ErrFrameOverflow, v.frames, maxFrames)
101		}	135		}
102	-	samples[i] = unsafe.Slice((*float32)(data.Data), frames)	136	+	// The length and capacity intentionally match the configured callback
			137	+	// frame count, preserving the previous view contract even if PipeWire
			138	+	// supplied a larger mapped buffer.
			139	+	v.samples[i] = unsafe.Slice((*float32)(data.Data), v.frames)
103		}	140		}
			141	+	v.buf = pw
			142	+	v.data = dataSlice
			143	+	return nil
			144	+	}
104			145
105	-	return &pwBufferView{	146	+	func (v *pwBufferView) clear() {
106	-	buf: pw,	147	+	v.buf = nil
107	-	channels: channels,	148	+	v.data = nil
108	-	frames: frames,	149	+	for i := range v.samples {
109	-	data: dataSlice,	150	+	v.samples[i] = nil
110	-	pcm: PCMBuffer{	151	+	}
111	-	Frames: frames,
112	-	Channels: channels,
113	-	Stride: 4,
114	-	Samples: samples,
115	-	},
116	-	samples: samples,
117	-	}, nil
118		}	152		}
119			153
120		// PCM returns a PCMBuffer pointing at the buffer's planar float32 data.	154		// PCM returns a PCMBuffer pointing at the buffer's planar float32 data.