Log media call stats. (#376)

Author: dennwc

pkg/mixer/mixer.go

@@ -16,7 +16,9 @@ package mixer
 
 import (
 	"fmt"
+	"slices"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/frostbyte73/core"
@@ -35,6 +37,26 @@ const (
 	inputBufferMin = inputBufferFrames/2 + 1
 )
 
+type Stats struct {
+	Tracks      atomic.Int64
+	TracksTotal atomic.Uint64
+	Restarts    atomic.Uint64
+
+	Mixes      atomic.Uint64
+	TimedMixes atomic.Uint64
+	JumpMixes  atomic.Uint64
+	ZeroMixes  atomic.Uint64
+
+	InputSamples atomic.Uint64
+	InputFrames  atomic.Uint64
+
+	MixedSamples atomic.Uint64
+	MixedFrames  atomic.Uint64
+
+	OutputSamples atomic.Uint64
+	OutputFrames  atomic.Uint64
+}
+
 type Input struct {
 	m          *Mixer
 	sampleRate int
@@ -58,11 +80,13 @@ type Mixer struct {
 	lastMixEndTs time.Time
 	stopped      core.Fuse
 	mixCnt       uint
+
+	stats *Stats
 }
 
-func NewMixer(out msdk.Writer[msdk.PCM16Sample], bufferDur time.Duration) *Mixer {
+func NewMixer(out msdk.Writer[msdk.PCM16Sample], bufferDur time.Duration, st *Stats) *Mixer {
 	mixSize := int(time.Duration(out.SampleRate()) * bufferDur / time.Second)
-	m := newMixer(out, mixSize)
+	m := newMixer(out, mixSize, st)
 	m.tickerDur = bufferDur
 	m.ticker = time.NewTicker(bufferDur)
 
@@ -71,12 +95,16 @@ func NewMixer(out msdk.Writer[msdk.PCM16Sample], bufferDur time.Duration) *Mixer
 	return m
 }
 
-func newMixer(out msdk.Writer[msdk.PCM16Sample], mixSize int) *Mixer {
+func newMixer(out msdk.Writer[msdk.PCM16Sample], mixSize int, st *Stats) *Mixer {
+	if st == nil {
+		st = new(Stats)
+	}
 	return &Mixer{
 		out:        out,
 		sampleRate: out.SampleRate(),
 		mixBuf:     make([]int32, mixSize),
 		mixTmp:     make(msdk.PCM16Sample, mixSize),
+		stats:      st,
 	}
 }
 
@@ -90,6 +118,10 @@ func (m *Mixer) mixInputs() {
 		if n == 0 {
 			continue
 		}
+
+		m.stats.MixedFrames.Add(1)
+		m.stats.MixedSamples.Add(uint64(n))
+
 		m.mixTmp = m.mixTmp[:n]
 		for j, v := range m.mixTmp {
 			// Add the samples. This can potentially lead to overflow, but is unlikely and dividing by the source
@@ -106,6 +138,7 @@ func (m *Mixer) reset() {
 }
 
 func (m *Mixer) mixOnce() {
+	m.stats.Mixes.Add(1)
 	m.mixCnt++
 	m.reset()
 	m.mixInputs()
@@ -121,6 +154,10 @@ func (m *Mixer) mixOnce() {
 		}
 		out[i] = int16(v)
 	}
+
+	m.stats.OutputFrames.Add(1)
+	m.stats.OutputSamples.Add(uint64(len(out)))
+
 	_ = m.out.WriteSample(out)
 }
 
@@ -129,6 +166,7 @@ func (m *Mixer) mixUpdate() {
 	now := time.Now()
 
 	if m.lastMixEndTs.IsZero() {
+		m.stats.TimedMixes.Add(1)
 		m.lastMixEndTs = now
 		n = 1
 	} else {
@@ -137,8 +175,12 @@ func (m *Mixer) mixUpdate() {
 		if dt := now.Sub(m.lastMixEndTs); dt > 0 {
 			n = int(dt / m.tickerDur)
 			m.lastMixEndTs = m.lastMixEndTs.Add(time.Duration(n) * m.tickerDur)
+			m.stats.JumpMixes.Add(uint64(n))
 		}
 	}
+	if n == 0 {
+		m.stats.ZeroMixes.Add(1)
+	}
 	if n > inputBufferFrames {
 		n = inputBufferFrames
 		// reset
@@ -178,6 +220,9 @@ func (m *Mixer) NewInput() *Input {
 		return nil
 	}
 
+	m.stats.Tracks.Add(1)
+	m.stats.TracksTotal.Add(1)
+
 	inp := &Input{
 		m:          m,
 		sampleRate: m.sampleRate,
@@ -194,12 +239,12 @@ func (m *Mixer) RemoveInput(inp *Input) {
 	}
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	for i, cur := range m.inputs {
-		if cur == inp {
-			m.inputs = append(m.inputs[:i], m.inputs[i+1:]...)
-			break
-		}
+	i := slices.Index(m.inputs, inp)
+	if i < 0 {
+		return
 	}
+	m.inputs = slices.Delete(m.inputs, i, i+1)
+	m.stats.Tracks.Add(-1)
 }
 
 func (m *Mixer) String() string {
@@ -223,6 +268,7 @@ func (i *Input) readSample(bufMin int, out msdk.PCM16Sample) (int, error) {
 	n, err := i.buf.Read(out)
 	if n == 0 {
 		i.buffering = true // starving; pause the input and start buffering again
+		i.m.stats.Restarts.Add(1)
 	}
 	return n, err
 }
@@ -246,6 +292,10 @@ func (i *Input) Close() error {
 func (i *Input) WriteSample(sample msdk.PCM16Sample) error {
 	i.mu.Lock()
 	defer i.mu.Unlock()
+
+	i.m.stats.InputFrames.Add(1)
+	i.m.stats.InputSamples.Add(uint64(len(sample)))
+
 	_, err := i.buf.Write(sample)
 	return err
 }

pkg/mixer/mixer_test.go

@@ -61,7 +61,7 @@ type testMixer struct {
 func newTestMixer(t testing.TB) *testMixer {
 	m := &testMixer{t: t}
 
-	m.Mixer = newMixer(newTestWriter(&m.sample, 8000), 5)
+	m.Mixer = newMixer(newTestWriter(&m.sample, 8000), 5, nil)
 	return m
 }
 

pkg/sip/inbound.go

@@ -353,6 +353,7 @@ type inboundCall struct {
 	forwardDTMF atomic.Bool
 	done        atomic.Bool
 	started     core.Fuse
+	stats       Stats
 	jitterBuf   bool
 }
 
@@ -375,9 +376,10 @@ func (s *Server) newInboundCall(
 		state:      state,
 		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),
 	}
+	// we need it created earlier so that the audio mixer is available for pin prompts
+	c.lkRoom = NewRoom(log, &c.stats.Room)
 	c.log = c.log.WithValues("jitterBuf", c.jitterBuf)
 	c.ctx, c.cancel = context.WithCancel(context.Background())
 	s.cmu.Lock()
@@ -596,6 +598,7 @@ func (c *inboundCall) runMediaConn(offerData []byte, enc livekit.SIPMediaEncrypt
 		MediaTimeoutInitial: c.s.conf.MediaTimeoutInitial,
 		MediaTimeout:        c.s.conf.MediaTimeout,
 		EnableJitterBuffer:  c.jitterBuf,
+		Stats:               &c.stats.Port,
 	}, RoomSampleRate)
 	if err != nil {
 		return nil, err
@@ -770,6 +773,9 @@ func (c *inboundCall) close(error bool, status CallStatus, reason string) {
 	c.mon.CallTerminate(reason)
 	sipCode, sipStatus := status.SIPStatus()
 	log := c.log.WithValues("status", sipCode, "reason", reason)
+	defer func() {
+		log.Infow("call statistics", "stats", c.stats.Load())
+	}()
 	if error {
 		log.Warnw("Closing inbound call with error", nil)
 	} else {