move latencies to config (#915)

* move latencies to config

* fix imageQueueLatency

* another miss

* fix mux latency datatype

Author: frostbyte73

pkg/config/base.go

@@ -55,6 +55,14 @@ type BaseConfig struct {
 	Insecure    bool                   `yaml:"insecure"`     // allow chrome to connect to an insecure websocket
 	Debug       DebugConfig            `yaml:"debug"`        // create dot file on internal error
 	ChromeFlags map[string]interface{} `yaml:"chrome_flags"` // additional flags to pass to Chrome
+	Latency     LatencyConfig          `yaml:"latency"`      // gstreamer latencies, modifying this may break the service
+}
+
+type SessionLimits struct {
+	FileOutputMaxDuration    time.Duration `yaml:"file_output_max_duration"`
+	StreamOutputMaxDuration  time.Duration `yaml:"stream_output_max_duration"`
+	SegmentOutputMaxDuration time.Duration `yaml:"segment_output_max_duration"`
+	ImageOutputMaxDuration   time.Duration `yaml:"image_output_max_duration"`
 }
 
 type DebugConfig struct {
@@ -65,17 +73,11 @@ type DebugConfig struct {
 	StorageConfig       `yaml:",inline"` // upload config (S3, Azure, GCP, or AliOSS)
 }
 
-type ProxyConfig struct {
-	Url      string `yaml:"url"`
-	Username string `yaml:"username"`
-	Password string `yaml:"password"`
-}
-
-type SessionLimits struct {
-	FileOutputMaxDuration    time.Duration `yaml:"file_output_max_duration"`
-	StreamOutputMaxDuration  time.Duration `yaml:"stream_output_max_duration"`
-	SegmentOutputMaxDuration time.Duration `yaml:"segment_output_max_duration"`
-	ImageOutputMaxDuration   time.Duration `yaml:"image_output_max_duration"`
+type LatencyConfig struct {
+	JitterBufferLatency time.Duration
+	AudioMixerLatency   time.Duration
+	PipelineLatency     time.Duration
+	AppSrcDrainTimeout  time.Duration
 }
 
 func (c *BaseConfig) initLogger(values ...interface{}) error {

pkg/config/pipeline.go

@@ -37,13 +37,6 @@ import (
 	lksdk "github.com/livekit/server-sdk-go/v2"
 )
 
-const (
-	JitterBufferLatency = time.Second * 2
-	AudioMixerLatency   = uint64(25e8)
-	PipelineLatency     = uint64(3e9)
-	AppSrcDrainTimeout  = time.Millisecond * 3500
-)
-
 type PipelineConfig struct {
 	BaseConfig `yaml:",inline"`
 

pkg/config/service.go

@@ -43,6 +43,11 @@ const (
 
 	defaultIOCreateTimeout = time.Second * 15
 	defaultIOUpdateTimeout = time.Second * 30
+
+	defaultJitterBufferLatency = time.Second * 2
+	defaultAudioMixerLatency   = time.Millisecond * 2500
+	defaultPipelineLatency     = time.Second * 3
+	defaultAppSrcDrainTimeout  = time.Millisecond * 3500
 )
 
 type ServiceConfig struct {
@@ -149,4 +154,17 @@ func (c *ServiceConfig) InitDefaults() {
 	if c.MaxUploadQueue <= 0 {
 		c.MaxUploadQueue = maxUploadQueue
 	}
+
+	if c.Latency.JitterBufferLatency == 0 {
+		c.Latency.JitterBufferLatency = defaultJitterBufferLatency
+	}
+	if c.Latency.AudioMixerLatency == 0 {
+		c.Latency.AudioMixerLatency = defaultAudioMixerLatency
+	}
+	if c.Latency.PipelineLatency == 0 {
+		c.Latency.PipelineLatency = defaultPipelineLatency
+	}
+	if c.Latency.AppSrcDrainTimeout == 0 {
+		c.Latency.AppSrcDrainTimeout = defaultAppSrcDrainTimeout
+	}
 }

pkg/config/storage.go

@@ -62,6 +62,12 @@ type GCPConfig struct {
 	ProxyConfig     *ProxyConfig `yaml:"proxy_config"`
 }
 
+type ProxyConfig struct {
+	Url      string `yaml:"url"`
+	Username string `yaml:"username"`
+	Password string `yaml:"password"`
+}
+
 func (p *PipelineConfig) getStorageConfig(req egress.UploadRequest) (*StorageConfig, error) {
 	sc := &StorageConfig{}
 	if p.StorageConfig != nil {

pkg/gstreamer/bin.go

@@ -35,7 +35,7 @@ type Bin struct {
 	pipeline *gst.Pipeline
 	mu       sync.Mutex
 	bin      *gst.Bin
-	latency  uint64
+	latency  time.Duration
 
 	linkFunc   func() error
 	shouldLink func(string) bool

pkg/gstreamer/builder.go

@@ -15,18 +15,20 @@
 package gstreamer
 
 import (
+	"time"
+
 	"github.com/go-gst/go-gst/gst"
 
 	"github.com/livekit/egress/pkg/errors"
 )
 
-func BuildQueue(name string, latency uint64, leaky bool) (*gst.Element, error) {
+func BuildQueue(name string, latency time.Duration, leaky bool) (*gst.Element, error) {
 	queue, err := gst.NewElementWithName("queue", name)
 	if err != nil {
 		return nil, errors.ErrGstPipelineError(err)
 	}
 	if latency > 0 {
-		if err = queue.SetProperty("max-size-time", latency); err != nil {
+		if err = queue.SetProperty("max-size-time", uint64(latency)); err != nil {
 			return nil, errors.ErrGstPipelineError(err)
 		}
 		if err = queue.SetProperty("max-size-bytes", uint(0)); err != nil {

pkg/gstreamer/pipeline.go

@@ -38,7 +38,7 @@ type Pipeline struct {
 
 // A pipeline can have either elements or src and sink bins. If you add both you will get a wrong hierarchy error
 // Bins can contain both elements and src and sink bins
-func NewPipeline(name string, latency uint64, callbacks *Callbacks) (*Pipeline, error) {
+func NewPipeline(name string, latency time.Duration, callbacks *Callbacks) (*Pipeline, error) {
 	pipeline, err := gst.NewPipeline(name)
 	if err != nil {
 		return nil, err

pkg/pipeline/builder/audio.go

@@ -75,7 +75,7 @@ func BuildAudioBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) error
 			return err
 		}
 	} else {
-		queue, err := gstreamer.BuildQueue("audio_queue", config.PipelineLatency, true)
+		queue, err := gstreamer.BuildQueue("audio_queue", p.Latency.PipelineLatency, true)
 		if err != nil {
 			return errors.ErrGstPipelineError(err)
 		}
@@ -275,7 +275,7 @@ func (b *AudioBin) addMixer() error {
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
-	if err = audioMixer.SetProperty("latency", config.AudioMixerLatency); err != nil {
+	if err = audioMixer.SetProperty("latency", uint64(b.conf.Latency.AudioMixerLatency)); err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
 
@@ -318,7 +318,7 @@ func (b *AudioBin) addEncoder() error {
 }
 
 func addAudioConverter(b *gstreamer.Bin, p *config.PipelineConfig, channel int) error {
-	audioQueue, err := gstreamer.BuildQueue("audio_input_queue", config.PipelineLatency, true)
+	audioQueue, err := gstreamer.BuildQueue("audio_input_queue", p.Latency.PipelineLatency, true)
 	if err != nil {
 		return err
 	}

pkg/pipeline/builder/image.go

@@ -28,7 +28,7 @@ import (
 )
 
 const (
-	imageQueueLatency = uint64(200 * time.Millisecond)
+	imageQueueLatency = 200 * time.Millisecond
 )
 
 func BuildImageBin(c *config.ImageConfig, pipeline *gstreamer.Pipeline, p *config.PipelineConfig) (*gstreamer.Bin, error) {

pkg/pipeline/builder/stream.go

@@ -35,6 +35,7 @@ type StreamBin struct {
 	Bin        *gstreamer.Bin
 	OutputType types.OutputType
 
+	latency  time.Duration
 	mu       sync.RWMutex
 	pipeline *gstreamer.Pipeline
 }
@@ -50,7 +51,7 @@ type Stream struct {
 	failed         atomic.Bool
 }
 
-func BuildStreamBin(pipeline *gstreamer.Pipeline, o *config.StreamConfig) (*StreamBin, error) {
+func BuildStreamBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig, o *config.StreamConfig) (*StreamBin, error) {
 	b := pipeline.NewBin("stream")
 
 	var mux *gst.Element
@@ -68,7 +69,7 @@ func BuildStreamBin(pipeline *gstreamer.Pipeline, o *config.StreamConfig) (*Stre
 			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", config.PipelineLatency); err != nil {
+		if err = mux.SetProperty("latency", uint64(p.Latency.PipelineLatency)); err != nil {
 			return nil, errors.ErrGstPipelineError(err)
 		}
 
@@ -104,6 +105,7 @@ func BuildStreamBin(pipeline *gstreamer.Pipeline, o *config.StreamConfig) (*Stre
 	sb := &StreamBin{
 		Bin:        b,
 		OutputType: o.OutputType,
+		latency:    p.Latency.PipelineLatency,
 	}
 
 	return sb, nil
@@ -113,7 +115,7 @@ func (sb *StreamBin) BuildStream(stream *config.Stream, framerate int32) (*Strea
 	stream.Name = utils.NewGuid("")
 	b := sb.Bin.NewBin(stream.Name)
 
-	queue, err := gstreamer.BuildQueue(fmt.Sprintf("queue_%s", stream.Name), config.PipelineLatency, true)
+	queue, err := gstreamer.BuildQueue(fmt.Sprintf("queue_%s", stream.Name), sb.latency, true)
 	if err != nil {
 		return nil, errors.ErrGstPipelineError(err)
 	}