Update latencies, log track stats (#379)

frostbyte73 · web-flow · commit c26bee6ae5e2 · 2023-06-04T23:57:09.000-07:00
* update latencies, log track stats

* update sdk ref

* fix track stats
diff --git a/go.mod b/go.mod
@@ -21,7 +21,7 @@ require (
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/protocol v1.5.8-0.20230601212100-a186ecb11a98
 	github.com/livekit/psrpc v0.3.0
-	github.com/livekit/server-sdk-go v1.0.11-0.20230601141132-d050553cfc5e
+	github.com/livekit/server-sdk-go v1.0.11-0.20230603013535-cd2ade940f03
 	github.com/pion/rtp v1.7.13
 	github.com/pion/webrtc/v3 v3.2.8
 	github.com/prometheus/client_golang v1.15.1
diff --git a/go.sum b/go.sum
@@ -179,8 +179,8 @@ github.com/livekit/protocol v1.5.8-0.20230601212100-a186ecb11a98 h1:OkQu5+sAgOzL
 github.com/livekit/protocol v1.5.8-0.20230601212100-a186ecb11a98/go.mod h1:ZaOnsvP+JS4s7vI1UO+JVdBagvvLp/lBXDAl2hkDS0I=
 github.com/livekit/psrpc v0.3.0 h1:giBZsfM3CWA0oIYXofsMITbVQtyW7u/ES9sQmVspHPM=
 github.com/livekit/psrpc v0.3.0/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U=
-github.com/livekit/server-sdk-go v1.0.11-0.20230601141132-d050553cfc5e h1:82QDZydt0ZR6zIQIPs8JmouDdjto5Dj9nzKjQ1PR3CU=
-github.com/livekit/server-sdk-go v1.0.11-0.20230601141132-d050553cfc5e/go.mod h1:8TeHbaUbm5ylUJ0OthbVNH4HIG2nTGFF9Ucr7zX99aM=
+github.com/livekit/server-sdk-go v1.0.11-0.20230603013535-cd2ade940f03 h1:/CJCedRin9wfM1gUjN7y6HbVREoW+DXB6qbmai5/bpo=
+github.com/livekit/server-sdk-go v1.0.11-0.20230603013535-cd2ade940f03/go.mod h1:8TeHbaUbm5ylUJ0OthbVNH4HIG2nTGFF9Ucr7zX99aM=
 github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs=
 github.com/mackerelio/go-osstat v0.2.4/go.mod h1:Zy+qzGdZs3A9cuIqmgbJvwbmLQH9dJvtio5ZjJTbdlQ=
 github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
diff --git a/pkg/config/pipeline.go b/pkg/config/pipeline.go
@@ -20,8 +20,8 @@ import (
 )
 
 const (
-	webLatency = uint64(15e8)
-	sdkLatency = uint64(43e8)
+	webLatency = uint64(2e9)
+	sdkLatency = uint64(3e9)
 )
 
 type PipelineConfig struct {
diff --git a/pkg/pipeline/input/audio.go b/pkg/pipeline/input/audio.go
@@ -12,7 +12,7 @@ import (
 	"github.com/livekit/egress/pkg/types"
 )
 
-const audioMixerLatency = uint64(41e8)
+const audioMixerLatency = uint64(2e9)
 
 type AudioInput struct {
 	decoder []*gst.Element
diff --git a/pkg/pipeline/output/stream.go b/pkg/pipeline/output/stream.go
@@ -33,7 +33,7 @@ func (b *Bin) buildStreamOutput(p *config.PipelineConfig) (*StreamOutput, error)
 		return nil, errors.ErrGstPipelineError(err)
 	}
 
-	mux, err := buildStreamMux(o)
+	mux, err := buildStreamMux(p, o)
 	if err != nil {
 		return nil, err
 	}
@@ -74,7 +74,7 @@ func (b *Bin) buildStreamOutput(p *config.PipelineConfig) (*StreamOutput, error)
 	}, nil
 }
 
-func buildStreamMux(o *config.StreamConfig) (*gst.Element, error) {
+func buildStreamMux(p *config.PipelineConfig, o *config.StreamConfig) (*gst.Element, error) {
 	switch o.OutputType {
 	case types.OutputTypeRTMP:
 		mux, err := gst.NewElement("flvmux")
@@ -85,7 +85,7 @@ func buildStreamMux(o *config.StreamConfig) (*gst.Element, error) {
 			return nil, errors.ErrGstPipelineError(err)
 		}
 		// add latency to give time for flvmux to receive and order packets from both streams
-		if err = mux.SetProperty("latency", uint64(2e8)); err != nil {
+		if err = mux.SetProperty("latency", p.Latency); err != nil {
 			return nil, errors.ErrGstPipelineError(err)
 		}
 
diff --git a/pkg/pipeline/source/sdk.go b/pkg/pipeline/source/sdk.go
@@ -106,15 +106,13 @@ func (s *SDKSource) CloseWriters() {
 		go func() {
 			defer wg.Done()
 			s.audioWriter.Drain(false)
-			logger.Debugw("audio writer finished")
 		}()
 	}
 	if s.videoWriter != nil {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
 			s.videoWriter.Drain(false)
-			logger.Debugw("video writer finished")
 		}()
 	}
 	wg.Wait()
diff --git a/pkg/pipeline/source/sdk/appwriter.go b/pkg/pipeline/source/sdk/appwriter.go
@@ -174,6 +174,12 @@ func (w *AppWriter) run() {
 			w.logger.Errorw("unexpected flow return", nil, "flowReturn", flow.String())
 		}
 	}
+	stats := w.TrackSynchronizer.GetTrackStats()
+	w.logger.Infow("writer finished",
+		"avg sample duration", stats.AvgSampleDuration,
+		"avg drift", stats.AvgDrift,
+		"max drift", stats.MaxDrift,
+	)
 	w.finished.Break()
 }
 

Original file line number	Diff line number	Diff line change
`@@ -20,8 +20,8 @@ import (`
`20`	`20`	`)`
`21`	`21`
`22`	`22`	`const (`
`23`		`- webLatency = uint64(15e8)`
`24`		`- sdkLatency = uint64(43e8)`
	`23`	`+ webLatency = uint64(2e9)`
	`24`	`+ sdkLatency = uint64(3e9)`
`25`	`25`	`)`
`26`	`26`
`27`	`27`	`type PipelineConfig struct {`
Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,7 @@ import (`
`12`	`12`	`"github.com/livekit/egress/pkg/types"`
`13`	`13`	`)`
`14`	`14`
`15`		`-const audioMixerLatency = uint64(41e8)`
	`15`	`+const audioMixerLatency = uint64(2e9)`
`16`	`16`
`17`	`17`	`type AudioInput struct {`
`18`	`18`	`decoder []*gst.Element`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ func (b Bin) buildStreamOutput(p config.PipelineConfig) (*StreamOutput, error)`
`33`	`33`	`return nil, errors.ErrGstPipelineError(err)`
`34`	`34`	`}`
`35`	`35`
`36`		`- mux, err := buildStreamMux(o)`
	`36`	`+ mux, err := buildStreamMux(p, o)`
`37`	`37`	`if err != nil {`
`38`	`38`	`return nil, err`
`39`	`39`	`}`
`@@ -74,7 +74,7 @@ func (b Bin) buildStreamOutput(p config.PipelineConfig) (*StreamOutput, error)`
`74`	`74`	`}, nil`
`75`	`75`	`}`
`76`	`76`
`77`		`-func buildStreamMux(o config.StreamConfig) (gst.Element, error) {`
	`77`	`+func buildStreamMux(p config.PipelineConfig, o config.StreamConfig) (*gst.Element, error) {`
`78`	`78`	`switch o.OutputType {`
`79`	`79`	`case types.OutputTypeRTMP:`
`80`	`80`	`mux, err := gst.NewElement("flvmux")`
`@@ -85,7 +85,7 @@ func buildStreamMux(o config.StreamConfig) (gst.Element, error) {`
`85`	`85`	`return nil, errors.ErrGstPipelineError(err)`
`86`	`86`	`}`
`87`	`87`	`// add latency to give time for flvmux to receive and order packets from both streams`
`88`		`- if err = mux.SetProperty("latency", uint64(2e8)); err != nil {`
	`88`	`+ if err = mux.SetProperty("latency", p.Latency); err != nil {`
`89`	`89`	`return nil, errors.ErrGstPipelineError(err)`
`90`	`90`	`}`
`91`	`91`
Original file line number	Diff line number	Diff line change
`@@ -174,6 +174,12 @@ func (w *AppWriter) run() {`
`174`	`174`	`w.logger.Errorw("unexpected flow return", nil, "flowReturn", flow.String())`
`175`	`175`	`}`
`176`	`176`	`}`
	`177`	`+ stats := w.TrackSynchronizer.GetTrackStats()`
	`178`	`+ w.logger.Infow("writer finished",`
	`179`	`+ "avg sample duration", stats.AvgSampleDuration,`
	`180`	`+ "avg drift", stats.AvgDrift,`
	`181`	`+ "max drift", stats.MaxDrift,`
	`182`	`+ )`
`177`	`183`	`w.finished.Break()`
`178`	`184`	`}`
`179`	`185`