Allow enabling jitter buffer for a percent of the calls. (#368)

Author: dennwc

pkg/config/config.go

@@ -93,8 +93,9 @@ type Config struct {
 	HideInboundPort bool `yaml:"hide_inbound_port"`
 
 	// AudioDTMF forces SIP to generate audio DTMF tones in addition to digital.
-	AudioDTMF          bool `yaml:"audio_dtmf"`
-	EnableJitterBuffer bool `yaml:"enable_jitter_buffer"`
+	AudioDTMF              bool    `yaml:"audio_dtmf"`
+	EnableJitterBuffer     bool    `yaml:"enable_jitter_buffer"`
+	EnableJitterBufferProb float64 `yaml:"enable_jitter_buffer_prob"`
 
 	// internal
 	ServiceName string `yaml:"-"`

pkg/sip/inbound.go

@@ -353,6 +353,7 @@ type inboundCall struct {
 	forwardDTMF atomic.Bool
 	done        atomic.Bool
 	started     core.Fuse
+	jitterBuf   bool
 }
 
 func (s *Server) newInboundCall(
@@ -375,7 +376,9 @@ func (s *Server) newInboundCall(
 		extraAttrs: extra,
 		dtmf:       make(chan dtmf.Event, 10),
 		lkRoom:     NewRoom(log), // we need it created earlier so that the audio mixer is available for pin prompts
+		jitterBuf:  SelectValueBool(s.conf.EnableJitterBuffer, s.conf.EnableJitterBufferProb),
 	}
+	c.log = c.log.WithValues("jitterBuf", c.jitterBuf)
 	c.ctx, c.cancel = context.WithCancel(context.Background())
 	s.cmu.Lock()
 	s.activeCalls[cc.Tag()] = c
@@ -520,6 +523,7 @@ func (c *inboundCall) handleInvite(ctx context.Context, req *sip.Request, trunkI
 	if disp.RingingTimeout <= 0 {
 		disp.RingingTimeout = defaultRingingTimeout
 	}
+	disp.Room.JitterBuf = c.jitterBuf
 	ctx, cancel := context.WithTimeout(ctx, disp.MaxCallDuration)
 	defer cancel()
 	status := CallRinging
@@ -591,7 +595,7 @@ func (c *inboundCall) runMediaConn(offerData []byte, enc livekit.SIPMediaEncrypt
 		Ports:               conf.RTPPort,
 		MediaTimeoutInitial: c.s.conf.MediaTimeoutInitial,
 		MediaTimeout:        c.s.conf.MediaTimeout,
-		EnableJitterBuffer:  c.s.conf.EnableJitterBuffer,
+		EnableJitterBuffer:  c.jitterBuf,
 	}, RoomSampleRate)
 	if err != nil {
 		return nil, err

pkg/sip/outbound.go

@@ -24,10 +24,11 @@ import (
 
 	"github.com/frostbyte73/core"
 	"github.com/icholy/digest"
-	msdk "github.com/livekit/media-sdk"
 	"github.com/pkg/errors"
 	"golang.org/x/exp/maps"
 
+	msdk "github.com/livekit/media-sdk"
+
 	"github.com/livekit/media-sdk/dtmf"
 	"github.com/livekit/media-sdk/sdp"
 	"github.com/livekit/media-sdk/tones"
@@ -64,14 +65,15 @@ type sipOutboundConfig struct {
 }
 
 type outboundCall struct {
-	c       *Client
-	log     logger.Logger
-	state   *CallState
-	cc      *sipOutbound
-	media   *MediaPort
-	started core.Fuse
-	stopped core.Fuse
-	closing core.Fuse
+	c         *Client
+	log       logger.Logger
+	state     *CallState
+	cc        *sipOutbound
+	media     *MediaPort
+	started   core.Fuse
+	stopped   core.Fuse
+	closing   core.Fuse
+	jitterBuf bool
 
 	mu       sync.RWMutex
 	mon      *stats.CallMonitor
@@ -87,18 +89,22 @@ func (c *Client) newCall(ctx context.Context, conf *config.Config, log logger.Lo
 	if sipConf.ringingTimeout <= 0 {
 		sipConf.ringingTimeout = defaultRingingTimeout
 	}
+	jitterBuf := SelectValueBool(conf.EnableJitterBuffer, conf.EnableJitterBufferProb)
+	room.JitterBuf = jitterBuf
 
 	tr := TransportFrom(sipConf.transport)
 	contact := c.ContactURI(tr)
 	if sipConf.host == "" {
 		sipConf.host = contact.GetHost()
 	}
 	call := &outboundCall{
-		c:       c,
-		log:     log,
-		sipConf: sipConf,
-		state:   state,
+		c:         c,
+		log:       log,
+		sipConf:   sipConf,
+		state:     state,
+		jitterBuf: jitterBuf,
 	}
+	c.log = c.log.WithValues("jitterBuf", call.jitterBuf)
 	call.cc = c.newOutbound(log, id, URI{
 		User:      sipConf.from,
 		Host:      sipConf.host,
@@ -124,7 +130,7 @@ func (c *Client) newCall(ctx context.Context, conf *config.Config, log logger.Lo
 		Ports:               conf.RTPPort,
 		MediaTimeoutInitial: c.conf.MediaTimeoutInitial,
 		MediaTimeout:        c.conf.MediaTimeout,
-		EnableJitterBuffer:  c.conf.EnableJitterBuffer,
+		EnableJitterBuffer:  call.jitterBuf,
 	}, RoomSampleRate)
 	if err != nil {
 		call.close(errors.Wrap(err, "media failed"), callDropped, "media-failed", livekit.DisconnectReason_UNKNOWN_REASON)

pkg/sip/room.go

@@ -21,9 +21,9 @@ import (
 	"sync/atomic"
 
 	"github.com/frostbyte73/core"
-	msdk "github.com/livekit/media-sdk"
 	"github.com/pion/webrtc/v4"
 
+	msdk "github.com/livekit/media-sdk"
 	"github.com/livekit/media-sdk/dtmf"
 	"github.com/livekit/media-sdk/rtp"
 	"github.com/livekit/protocol/livekit"
@@ -73,6 +73,7 @@ type RoomConfig struct {
 	Participant ParticipantConfig
 	RoomPreset  string
 	RoomConfig  *livekit.RoomConfiguration
+	JitterBuf   bool
 }
 
 func NewRoom(log logger.Logger) *Room {

pkg/sip/types.go

@@ -16,6 +16,7 @@ package sip
 
 import (
 	"errors"
+	"math/rand/v2"
 	"net"
 	"net/netip"
 	"strconv"
@@ -324,3 +325,17 @@ func sdpEncryption(e livekit.SIPMediaEncryption) (sdp.Encryption, error) {
 	}
 	return sdp.EncryptionAllow, errors.New("invalid SIP media encryption type")
 }
+
+func SelectValue[T any](then, els T, probElse float64) T {
+	if probElse <= 0 {
+		return then
+	}
+	if rand.Float64() < probElse {
+		return then
+	}
+	return els
+}
+
+func SelectValueBool(then bool, probElse float64) bool {
+	return SelectValue(then, !then, probElse)
+}