Disable resampler on the input audio. (#487)

Author: dennwc

pkg/sip/inbound.go

@@ -913,6 +913,7 @@ func (c *inboundCall) runMediaConn(tid traceid.ID, offerData []byte, enc livekit
 		MediaTimeout:        c.s.conf.MediaTimeout,
 		EnableJitterBuffer:  c.jitterBuf,
 		Stats:               &c.stats.Port,
+		NoInputResample:     !RoomResample,
 	}, RoomSampleRate)
 	if err != nil {
 		return nil, err

pkg/sip/media.go

@@ -151,6 +151,7 @@ func (s *Stats) MarshalJSON() ([]byte, error) {
 const (
 	channels       = 1
 	RoomSampleRate = 48000
+	RoomResample   = false
 )
 
 func newRTPStatsHandler(mon *stats.CallMonitor, typ string, r rtp.Handler) rtp.Handler {

pkg/sip/media_port.go

@@ -131,6 +131,7 @@ type MediaOptions struct {
 	MediaTimeout        time.Duration
 	Stats               *PortStats
 	EnableJitterBuffer  bool
+	NoInputResample     bool
 }
 
 func NewMediaPort(tid traceid.ID, log logger.Logger, mon *stats.CallMonitor, opts *MediaOptions, sampleRate int) (*MediaPort, error) {
@@ -158,6 +159,10 @@ func NewMediaPortWith(tid traceid.ID, log logger.Logger, mon *stats.CallMonitor,
 		conn = c
 	}
 	mediaTimeout := make(chan struct{})
+	inSampleRate := sampleRate
+	if opts.NoInputResample {
+		inSampleRate = -1 // set only after SDP is accepted
+	}
 	p := &MediaPort{
 		tid:              tid,
 		log:              log,
@@ -169,7 +174,7 @@ func NewMediaPortWith(tid traceid.ID, log logger.Logger, mon *stats.CallMonitor,
 		jitterEnabled:    opts.EnableJitterBuffer,
 		port:             newUDPConn(log, conn),
 		audioOut:         msdk.NewSwitchWriter(sampleRate),
-		audioIn:          msdk.NewSwitchWriter(sampleRate),
+		audioIn:          msdk.NewSwitchWriter(inSampleRate),
 		stats:            opts.Stats,
 	}
 	p.timeoutInitial.Store(&opts.MediaTimeoutInitial)
@@ -628,14 +633,19 @@ func (p *MediaPort) setupOutput(tid traceid.ID) error {
 
 func (p *MediaPort) setupInput() {
 	// Decoding pipeline (SIP RTP -> LK PCM)
-	audioHandler := p.conf.Audio.Codec.DecodeRTP(p.audioIn, p.conf.Audio.Type)
+	codec := p.conf.Audio.Codec
+	codecInfo := codec.Info()
+	if p.opts.NoInputResample {
+		p.audioIn.SetSampleRate(codecInfo.SampleRate)
+	}
+	audioHandler := codec.DecodeRTP(p.audioIn, p.conf.Audio.Type)
 	p.audioInHandler = audioHandler
 
 	mux := rtp.NewMux(nil)
 	mux.SetDefault(newRTPStatsHandler(p.mon, "", nil))
 	mux.Register(
 		p.conf.Audio.Type, newRTPHandlerCount(
-			newRTPStatsHandler(p.mon, p.conf.Audio.Codec.Info().SDPName, audioHandler),
+			newRTPStatsHandler(p.mon, codecInfo.SDPName, audioHandler),
 			&p.stats.AudioPackets, &p.stats.AudioBytes,
 		),
 	)

pkg/sip/media_port_test.go

@@ -199,8 +199,9 @@ func TestMediaPort(t *testing.T) {
 					log := logger.GetLogger()
 
 					m1, err := NewMediaPortWith(1, log.WithName("one"), nil, c1, &MediaOptions{
-						IP:    newIP("1.1.1.1"),
-						Ports: rtcconfig.PortRange{Start: 10000},
+						IP:              newIP("1.1.1.1"),
+						Ports:           rtcconfig.PortRange{Start: 10000},
+						NoInputResample: true,
 					}, tconf.Rate)
 					require.NoError(t, err)
 					defer m1.Close()
@@ -235,11 +236,13 @@ func TestMediaPort(t *testing.T) {
 					err = m2.SetConfig(conf)
 					require.NoError(t, err)
 
-					require.Equal(t, info.SDPName, m1.Config().Audio.Codec.Info().SDPName)
-					require.Equal(t, info.SDPName, m2.Config().Audio.Codec.Info().SDPName)
+					codec1 := m1.Config().Audio.Codec
+					codec2 := m2.Config().Audio.Codec
+					require.Equal(t, info.SDPName, codec1.Info().SDPName)
+					require.Equal(t, info.SDPName, codec2.Info().SDPName)
 
 					var buf1 msdk.PCM16Sample
-					bw1 := msdk.NewPCM16BufferWriter(&buf1, tconf.Rate)
+					bw1 := msdk.NewPCM16BufferWriter(&buf1, codec1.Info().SampleRate)
 					m1.WriteAudioTo(bw1)
 
 					var buf2 msdk.PCM16Sample
@@ -257,7 +260,8 @@ func TestMediaPort(t *testing.T) {
 						sample2[i] = -5116
 					}
 
-					writes := 1
+					writes1 := 1
+					writes2 := 1
 					if tconf.Rate == nativeRate {
 						expChain := fmt.Sprintf("Switch(%d) -> %s(encode) -> RTP(%d)", nativeRate, name, nativeRate)
 						require.Equal(t, expChain, w1.String())
@@ -271,20 +275,31 @@ func TestMediaPort(t *testing.T) {
 						require.Equal(t, expChain, w1.String())
 						require.Equal(t, expChain, w2.String())
 
-						expChain = fmt.Sprintf("RTP(%d) -> %s(decode) -> Resample(%d->48000) -> Switch(48000) -> Buffer(48000)", nativeRate, name, nativeRate)
-						require.Equal(t, expChain, PrintAudioInWriter(m1))
-						require.Equal(t, expChain, PrintAudioInWriter(m2))
-
-						writes += 2 // resampler will buffer a few frames
+						// This side does not resample the received audio, it uses sample rate of the RTP source.
+						expChain1 := fmt.Sprintf("RTP(%d) -> %s(decode) -> Switch(%d) -> Buffer(%d)", nativeRate, name, nativeRate, nativeRate)
+						// This side resamples the received audio to the expected sample rate.
+						expChain2 := fmt.Sprintf("RTP(%d) -> %s(decode) -> Resample(%d->48000) -> Switch(48000) -> Buffer(48000)", nativeRate, name, nativeRate)
+						require.Equal(t, expChain1, PrintAudioInWriter(m1))
+						require.Equal(t, expChain2, PrintAudioInWriter(m2))
+
+						// resampler will buffer a few frames
+						writes1 += 2
+						writes2 += 2
+						// a few more because of higher resample quality required
 						if nativeRate == 8000 {
-							writes += 3 // a few more because of higher resample quality required
+							writes1 += 3
+							writes2 += 5
+						}
+						if strings.HasPrefix(info.SDPName, "G722/") {
+							writes2 += 1
 						}
 					}
 
-					for range writes {
+					for range writes1 {
 						err = w1.WriteSample(sample1)
 						require.NoError(t, err)
-
+					}
+					for range writes2 {
 						err = w2.WriteSample(sample2)
 						require.NoError(t, err)
 					}
@@ -309,6 +324,7 @@ func TestMediaPort(t *testing.T) {
 }
 
 func checkPCM(t testing.TB, exp, got msdk.PCM16Sample) {
+	t.Helper()
 	require.Equal(t, len(exp), len(got))
 	expSamples := slices.Clone(exp)
 	slices.Sort(expSamples)
@@ -338,6 +354,7 @@ func newMediaPair(t testing.TB, opt1, opt2 *MediaOptions) (m1, m2 *MediaPort) {
 
 	opt1.IP = newIP("1.1.1.1")
 	opt1.Ports = rtcconfig.PortRange{Start: 10000}
+	opt1.NoInputResample = true
 
 	opt2.IP = newIP("2.2.2.2")
 	opt2.Ports = rtcconfig.PortRange{Start: 20000}

pkg/sip/outbound.go

@@ -140,6 +140,7 @@ func (c *Client) newCall(ctx context.Context, tid traceid.ID, conf *config.Confi
 		MediaTimeout:        c.conf.MediaTimeout,
 		EnableJitterBuffer:  call.jitterBuf,
 		Stats:               &call.stats.Port,
+		NoInputResample:     !RoomResample,
 	}, RoomSampleRate)
 	if err != nil {
 		call.close(errors.Wrap(err, "media failed"), callDropped, "media-failed", livekit.DisconnectReason_UNKNOWN_REASON)