fix: pin PipeWire connect params for stream lifetime

commit: 715769bb29df437cb8d77120e69f83306913a218
author: brice · 6mo ago · 2026-04-10 13:42 UTC
parent: af7a9d7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +140 -3

M internal/capi/stream_ops.go +24 -3


@@ -49,9 +49,18 @@ type streamCallbackStorage struct {
49	49
50	50		// streamOpsImpl implements portout.StreamOps using the real PipeWire C API
51	51		// bindings registered via purego.
	52	+	//
	53	+	// Ownership/lifetime contract:
	54	+	// - pinned holds streamCallbackStorage for each stream created via CreatePlaybackStream.
	55	+	// It keeps Go callback closures alive so PipeWire can call them safely.
	56	+	// - pinnedParams holds connectParams for each stream connected via ConnectPlaybackStream.
	57	+	// PipeWire reads the SPA POD params asynchronously during format negotiation, so the
	58	+	// backing byte storage must remain live until DestroyStream releases it.
	59	+	// - Both maps are keyed by the stream pointer and must be cleaned up in DestroyStream.
52	60		type streamOpsImpl struct {
53		-	mu sync.Mutex
54		-	pinned map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
	61	+	mu sync.Mutex
	62	+	pinned map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
	63	+	pinnedParams map[unsafe.Pointer]*connectParams // keyed by stream ptr
55	64		}
56	65
57	66		// Verify interface compliance at compile time.

@@ -120,6 +129,17 @@ func (s *streamOpsImpl) ConnectPlaybackStream(streamPtr unsafe.Pointer, format p
120	129		if ret < 0 {
121	130		return &PWError{Func: "pw_stream_connect", Code: ret}
122	131		}
	132	+
	133	+	// Pin the connectParams for the lifetime of the stream. PipeWire reads
	134	+	// the SPA POD params asynchronously during format negotiation, so the
	135	+	// backing storage must remain alive until the stream is destroyed.
	136	+	s.mu.Lock()
	137	+	if s.pinnedParams == nil {
	138	+	s.pinnedParams = make(map[unsafe.Pointer]*connectParams)
	139	+	}
	140	+	s.pinnedParams[streamPtr] = cp
	141	+	s.mu.Unlock()
	142	+
123	143		return nil
124	144		}
125	145

@@ -160,8 +180,9 @@ func (s *streamOpsImpl) DestroyStream(streamPtr unsafe.Pointer) {
160	180		s.mu.Unlock()
161	181		return
162	182		}
163		-	// Unpin callback storage — safe now that the stream is being destroyed.
	183	+	// Unpin callback storage and connect params — safe now that the stream is being destroyed.
164	184		delete(s.pinned, streamPtr)
	185	+	delete(s.pinnedParams, streamPtr)
165	186		s.mu.Unlock()
166	187
167	188		// Use the generated pw_stream_destroy binding for proper cleanup.

 // streamOpsImpl implements portout.StreamOps using the real PipeWire C API
 // bindings registered via purego.
+//
+// Ownership/lifetime contract:
+//   - pinned holds streamCallbackStorage for each stream created via CreatePlaybackStream.
+//     It keeps Go callback closures alive so PipeWire can call them safely.
+//   - pinnedParams holds connectParams for each stream connected via ConnectPlaybackStream.
+//     PipeWire reads the SPA POD params asynchronously during format negotiation, so the
+//     backing byte storage must remain live until DestroyStream releases it.
+//   - Both maps are keyed by the stream pointer and must be cleaned up in DestroyStream.
 type streamOpsImpl struct {
-	mu     sync.Mutex
+	mu           sync.Mutex
-	pinned map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
+	pinned       map[unsafe.Pointer]*streamCallbackStorage // keyed by stream ptr
+	pinnedParams map[unsafe.Pointer]*connectParams         // keyed by stream ptr
 }
 // Verify interface compliance at compile time.
 	if ret < 0 {
 		return &PWError{Func: "pw_stream_connect", Code: ret}
 	}
+	// Pin the connectParams for the lifetime of the stream. PipeWire reads
+	// the SPA POD params asynchronously during format negotiation, so the
+	// backing storage must remain alive until the stream is destroyed.
+	s.mu.Lock()
+	if s.pinnedParams == nil {
+		s.pinnedParams = make(map[unsafe.Pointer]*connectParams)
+	}
+	s.pinnedParams[streamPtr] = cp
+	s.mu.Unlock()
 	return nil
 }
 		s.mu.Unlock()
 		return
 	}
-	// Unpin callback storage — safe now that the stream is being destroyed.
+	// Unpin callback storage and connect params — safe now that the stream is being destroyed.
 	delete(s.pinned, streamPtr)
+	delete(s.pinnedParams, streamPtr)
 	s.mu.Unlock()
 	// Use the generated pw_stream_destroy binding for proper cleanup.

M internal/capi/stream_ops_test.go +116 -0


@@ -496,3 +496,119 @@ func TestConnectPlaybackStreamPassesParamsFromHelper(t *testing.T) {
496	496		t.Fatal("expected nParams > 0 from buildRawAudioParams, got 0")
497	497		}
498	498		}
	499	+
	500	+	// TestConnectPlaybackStreamPinsParamsAfterSuccess verifies that ConnectPlaybackStream
	501	+	// stores the connectParams in pinnedParams so the SPA POD backing storage outlives
	502	+	// the function call.
	503	+	func TestConnectPlaybackStreamPinsParamsAfterSuccess(t *testing.T) {
	504	+	origConnect := pw_stream_connect
	505	+	t.Cleanup(func() { pw_stream_connect = origConnect })
	506	+
	507	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
	508	+	return 0
	509	+	}
	510	+
	511	+	ops := &streamOpsImpl{}
	512	+	fakePtr := unsafe.Pointer(uintptr(0x5678))
	513	+
	514	+	validFmt := portout.PlaybackFormat{
	515	+	SampleRate: 48000,
	516	+	Channels: 2,
	517	+	FramesPerBuffer: 1024,
	518	+	}
	519	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
	520	+	if err != nil {
	521	+	t.Fatalf("unexpected error: %v", err)
	522	+	}
	523	+
	524	+	cp, ok := ops.pinnedParams[fakePtr]
	525	+	if !ok {
	526	+	t.Fatal("expected connectParams to be pinned for stream after ConnectPlaybackStream")
	527	+	}
	528	+	if cp == nil {
	529	+	t.Fatal("pinned connectParams should not be nil")
	530	+	}
	531	+	// Verify the pinned params point to real SPA POD data.
	532	+	if cp.Count() != 1 {
	533	+	t.Fatalf("expected pinned params Count()=1, got %d", cp.Count())
	534	+	}
	535	+	if cp.Pointer() == nil {
	536	+	t.Fatal("expected pinned params Pointer() to be non-nil")
	537	+	}
	538	+	}
	539	+
	540	+	// TestConnectPlaybackStreamDoesNotPinParamsOnFailure verifies that when pw_stream_connect
	541	+	// returns a negative result, connectParams are NOT pinned.
	542	+	func TestConnectPlaybackStreamDoesNotPinParamsOnFailure(t *testing.T) {
	543	+	origConnect := pw_stream_connect
	544	+	t.Cleanup(func() { pw_stream_connect = origConnect })
	545	+
	546	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
	547	+	return -22 // EINVAL
	548	+	}
	549	+
	550	+	ops := &streamOpsImpl{}
	551	+	fakePtr := unsafe.Pointer(uintptr(0x5678))
	552	+
	553	+	validFmt := portout.PlaybackFormat{
	554	+	SampleRate: 48000,
	555	+	Channels: 2,
	556	+	FramesPerBuffer: 1024,
	557	+	}
	558	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
	559	+	if err == nil {
	560	+	t.Fatal("expected error for negative return code")
	561	+	}
	562	+
	563	+	if _, ok := ops.pinnedParams[fakePtr]; ok {
	564	+	t.Fatal("connectParams should NOT be pinned when pw_stream_connect fails")
	565	+	}
	566	+	}
	567	+
	568	+	// TestDestroyStreamReleasesPinnedParams verifies that DestroyStream removes both the
	569	+	// callback storage and the connect params from the pinned maps.
	570	+	func TestDestroyStreamReleasesPinnedParams(t *testing.T) {
	571	+	origConnect := pw_stream_connect
	572	+	origDestroy := pw_stream_destroy
	573	+	t.Cleanup(func() {
	574	+	pw_stream_connect = origConnect
	575	+	pw_stream_destroy = origDestroy
	576	+	})
	577	+
	578	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
	579	+	return 0
	580	+	}
	581	+	pw_stream_destroy = func(_ unsafe.Pointer) {}
	582	+
	583	+	ops := &streamOpsImpl{
	584	+	pinned: make(map[unsafe.Pointer]*streamCallbackStorage),
	585	+	}
	586	+
	587	+	fakePtr := unsafe.Pointer(uintptr(0xDEAD))
	588	+	ops.pinned[fakePtr] = &streamCallbackStorage{}
	589	+
	590	+	validFmt := portout.PlaybackFormat{
	591	+	SampleRate: 48000,
	592	+	Channels: 2,
	593	+	FramesPerBuffer: 1024,
	594	+	}
	595	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
	596	+	if err != nil {
	597	+	t.Fatalf("unexpected error: %v", err)
	598	+	}
	599	+
	600	+	// Verify params are pinned.
	601	+	if _, ok := ops.pinnedParams[fakePtr]; !ok {
	602	+	t.Fatal("expected connectParams to be pinned before DestroyStream")
	603	+	}
	604	+
	605	+	ops.DestroyStream(fakePtr)
	606	+
	607	+	// Both maps should no longer contain the stream.
	608	+	if _, ok := ops.pinned[fakePtr]; ok {
	609	+	t.Error("expected callback storage to be removed after DestroyStream")
	610	+	}
	611	+	if _, ok := ops.pinnedParams[fakePtr]; ok {
	612	+	t.Error("expected connectParams to be removed after DestroyStream")
	613	+	}
	614	+	}


@@ -496,3 +496,119 @@ func TestConnectPlaybackStreamPassesParamsFromHelper(t *testing.T) {
496	t.Fatal("expected nParams > 0 from buildRawAudioParams, got 0")	496		t.Fatal("expected nParams > 0 from buildRawAudioParams, got 0")
497	}	497		}
498	}	498		}
		499	+
		500	+	// TestConnectPlaybackStreamPinsParamsAfterSuccess verifies that ConnectPlaybackStream
		501	+	// stores the connectParams in pinnedParams so the SPA POD backing storage outlives
		502	+	// the function call.
		503	+	func TestConnectPlaybackStreamPinsParamsAfterSuccess(t *testing.T) {
		504	+	origConnect := pw_stream_connect
		505	+	t.Cleanup(func() { pw_stream_connect = origConnect })
		506	+
		507	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
		508	+	return 0
		509	+	}
		510	+
		511	+	ops := &streamOpsImpl{}
		512	+	fakePtr := unsafe.Pointer(uintptr(0x5678))
		513	+
		514	+	validFmt := portout.PlaybackFormat{
		515	+	SampleRate: 48000,
		516	+	Channels: 2,
		517	+	FramesPerBuffer: 1024,
		518	+	}
		519	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
		520	+	if err != nil {
		521	+	t.Fatalf("unexpected error: %v", err)
		522	+	}
		523	+
		524	+	cp, ok := ops.pinnedParams[fakePtr]
		525	+	if !ok {
		526	+	t.Fatal("expected connectParams to be pinned for stream after ConnectPlaybackStream")
		527	+	}
		528	+	if cp == nil {
		529	+	t.Fatal("pinned connectParams should not be nil")
		530	+	}
		531	+	// Verify the pinned params point to real SPA POD data.
		532	+	if cp.Count() != 1 {
		533	+	t.Fatalf("expected pinned params Count()=1, got %d", cp.Count())
		534	+	}
		535	+	if cp.Pointer() == nil {
		536	+	t.Fatal("expected pinned params Pointer() to be non-nil")
		537	+	}
		538	+	}
		539	+
		540	+	// TestConnectPlaybackStreamDoesNotPinParamsOnFailure verifies that when pw_stream_connect
		541	+	// returns a negative result, connectParams are NOT pinned.
		542	+	func TestConnectPlaybackStreamDoesNotPinParamsOnFailure(t *testing.T) {
		543	+	origConnect := pw_stream_connect
		544	+	t.Cleanup(func() { pw_stream_connect = origConnect })
		545	+
		546	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
		547	+	return -22 // EINVAL
		548	+	}
		549	+
		550	+	ops := &streamOpsImpl{}
		551	+	fakePtr := unsafe.Pointer(uintptr(0x5678))
		552	+
		553	+	validFmt := portout.PlaybackFormat{
		554	+	SampleRate: 48000,
		555	+	Channels: 2,
		556	+	FramesPerBuffer: 1024,
		557	+	}
		558	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
		559	+	if err == nil {
		560	+	t.Fatal("expected error for negative return code")
		561	+	}
		562	+
		563	+	if _, ok := ops.pinnedParams[fakePtr]; ok {
		564	+	t.Fatal("connectParams should NOT be pinned when pw_stream_connect fails")
		565	+	}
		566	+	}
		567	+
		568	+	// TestDestroyStreamReleasesPinnedParams verifies that DestroyStream removes both the
		569	+	// callback storage and the connect params from the pinned maps.
		570	+	func TestDestroyStreamReleasesPinnedParams(t *testing.T) {
		571	+	origConnect := pw_stream_connect
		572	+	origDestroy := pw_stream_destroy
		573	+	t.Cleanup(func() {
		574	+	pw_stream_connect = origConnect
		575	+	pw_stream_destroy = origDestroy
		576	+	})
		577	+
		578	+	pw_stream_connect = func(stream unsafe.Pointer, direction int32, id uint32, flags uint32, ports unsafe.Pointer, n_ports uint32) int32 {
		579	+	return 0
		580	+	}
		581	+	pw_stream_destroy = func(_ unsafe.Pointer) {}
		582	+
		583	+	ops := &streamOpsImpl{
		584	+	pinned: make(map[unsafe.Pointer]*streamCallbackStorage),
		585	+	}
		586	+
		587	+	fakePtr := unsafe.Pointer(uintptr(0xDEAD))
		588	+	ops.pinned[fakePtr] = &streamCallbackStorage{}
		589	+
		590	+	validFmt := portout.PlaybackFormat{
		591	+	SampleRate: 48000,
		592	+	Channels: 2,
		593	+	FramesPerBuffer: 1024,
		594	+	}
		595	+	err := ops.ConnectPlaybackStream(fakePtr, validFmt)
		596	+	if err != nil {
		597	+	t.Fatalf("unexpected error: %v", err)
		598	+	}
		599	+
		600	+	// Verify params are pinned.
		601	+	if _, ok := ops.pinnedParams[fakePtr]; !ok {
		602	+	t.Fatal("expected connectParams to be pinned before DestroyStream")
		603	+	}
		604	+
		605	+	ops.DestroyStream(fakePtr)
		606	+
		607	+	// Both maps should no longer contain the stream.
		608	+	if _, ok := ops.pinned[fakePtr]; ok {
		609	+	t.Error("expected callback storage to be removed after DestroyStream")
		610	+	}
		611	+	if _, ok := ops.pinnedParams[fakePtr]; ok {
		612	+	t.Error("expected connectParams to be removed after DestroyStream")
		613	+	}
		614	+	}