diff --git a/pkg/sip/media_pipeline.go b/pkg/sip/media_pipeline.go index 19578f488..57a7b780d 100644 --- a/pkg/sip/media_pipeline.go +++ b/pkg/sip/media_pipeline.go @@ -176,6 +176,8 @@ func (p *mediaPortPipeline) setupInput(mc *sdp.MediaConfig, audioToRoom msdk.PCM // And these are only available before decoding, hence it wraps both audioHandler & sink audioHandler = newSilenceFiller(audioHandler, sink, codecInfo.RTPClockRate, codecInfo.SampleRate, p.conf.log) + audioHandler = NewSerializedRTPHandler(audioHandler) // SilenceFiller/Codecs are not thread-safe + mux := rtp.NewMux(nil) mux.SetDefault(newRTPStatsHandler(p.conf.mon, "", nil)) @@ -405,6 +407,31 @@ func (p *mediaPortPipeline) Close() error { return errors.Join(errs...) } +func NewSerializedRTPHandler(w rtp.HandlerCloser) rtp.HandlerCloser { + return &serializedRTPHandler{w: w} +} + +type serializedRTPHandler struct { + mu sync.Mutex + w rtp.HandlerCloser +} + +func (s *serializedRTPHandler) String() string { + return s.w.String() +} + +func (s *serializedRTPHandler) HandleRTP(h *rtp.Header, payload []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + return s.w.HandleRTP(h, payload) +} + +func (s *serializedRTPHandler) Close() { + s.mu.Lock() + defer s.mu.Unlock() + s.w.Close() +} + // dtmfOutWriter sends SipDTMF as RFC 4733 telephone-events (optional in-band audio). type dtmfOutWriter struct { log logger.Logger diff --git a/pkg/sip/media_pipeline_test.go b/pkg/sip/media_pipeline_test.go index 7067ffd57..4a7530fcb 100644 --- a/pkg/sip/media_pipeline_test.go +++ b/pkg/sip/media_pipeline_test.go @@ -528,9 +528,10 @@ func TestMediaPipelineTeardownMultiSSRC(t *testing.T) { h.injectAudio(0x22222222, 1, 160, sample) require.Eventually(t, func() bool { - return h.ssrcCount.Load() >= 2 - }, time.Second, 5*time.Millisecond, "expected AcceptStream for two SSRCs") - assert.Equal(t, h.packetCount.Load(), uint64(2)) + return h.ssrcCount.Load() >= 2 && h.packetCount.Load() >= 2 + }, time.Second, 5*time.Millisecond, "expected AcceptStream and HandleRTP for two SSRCs") + assert.Equal(t, uint64(2), h.ssrcCount.Load()) + assert.Equal(t, uint64(2), h.packetCount.Load()) done := make(chan error, 1) go func() { @@ -544,6 +545,57 @@ func TestMediaPipelineTeardownMultiSSRC(t *testing.T) { } } +func TestMediaPipelineConcurrentSSRCPump(t *testing.T) { + const ( + ssrcCount = 3 + packets = 30 // Currently the built-in limit of media-sdk's ssrc mux + ) + codec := audioCodecByName(t, g711.ULawSDPNameAndRate) + h := newPipelineHarness(t, RoomSampleRate) + h.configure(codec, testAudioPT(codec), testDTMFPT, false) + + silence := make(msdk.PCM16Sample, codec.Info().SampleRate/int(time.Second/msrtp.DefFrameDur)) + var encoded msrtp.Buffer + clock := codec.Info().RTPClockRate + if clock == 0 { + clock = codec.Info().SampleRate + } + enc := msrtp.EncodePCM(msrtp.NewSeqWriter(&encoded).NewStream(h.audioPT, clock), h.codec) + require.NoError(t, enc.WriteSample(silence)) + require.NoError(t, enc.Close()) + require.NotEmpty(t, encoded, "codec produced no RTP") + payload := slices.Clone(encoded[0].Payload) + + pkt := &rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: h.audioPT, + }, + Payload: payload, + } + for i := range packets { + pkt.SequenceNumber = uint16(i) + pkt.SSRC = uint32(i % ssrcCount) + h.injectRTP(pkt) + } + + require.Eventually(t, func() bool { return h.ssrcCount.Load() == ssrcCount }, time.Second, time.Millisecond, "expected %d SSRCs", ssrcCount) + assert.Eventually(t, func() bool { return h.packetCount.Load() == packets }, time.Second, time.Millisecond, "expected %d packets", packets) + assert.Equal(t, uint64(ssrcCount), h.ssrcCount.Load()) + assert.Equal(t, uint64(packets), h.packetCount.Load()) + + done := make(chan error, 1) + go func() { + done <- h.pipeline.Close() + }() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("pipeline.Close hung under concurrent SSRC pumps") + } +} + func TestMediaPipelineReuseUDPConn(t *testing.T) { const rate = 48000 d := pipelineTestDTMF[1] // event-only