SDK timing (#910)

* fix eos timing, appsrc EOF

* fix onEOSSent

Author: frostbyte73

pkg/config/pipeline.go

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

pkg/pipeline/builder/audio.go

@@ -29,8 +29,9 @@ import (
 )
 
 const (
-	audioMixerLatency = uint64(25e8)
-	audioChannelNone  = 0
+	audioChannelStereo = 0
+	audioChannelLeft   = 1
+	audioChannelRight  = 2
 )
 
 type AudioBin struct {
@@ -74,7 +75,7 @@ func BuildAudioBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) error
 			return err
 		}
 	} else {
-		queue, err := gstreamer.BuildQueue("audio_queue", config.Latency, true)
+		queue, err := gstreamer.BuildQueue("audio_queue", config.PipelineLatency, true)
 		if err != nil {
 			return errors.ErrGstPipelineError(err)
 		}
@@ -129,7 +130,7 @@ func (b *AudioBin) buildWebInput() error {
 		return err
 	}
 
-	if err = addAudioConverter(b.bin, b.conf, audioChannelNone); err != nil {
+	if err = addAudioConverter(b.bin, b.conf, audioChannelStereo); err != nil {
 		return err
 	}
 	if b.conf.AudioTranscoding {
@@ -223,13 +224,13 @@ func (b *AudioBin) addAudioAppSrcBin(ts *config.TrackSource) error {
 func (b *AudioBin) getChannel(ts *config.TrackSource) int {
 	switch b.conf.AudioMixing {
 	case livekit.AudioMixing_DEFAULT_MIXING:
-		return 0
+		return audioChannelStereo
 
 	case livekit.AudioMixing_DUAL_CHANNEL_AGENT:
 		if ts.ParticipantKind == lksdk.ParticipantAgent {
-			return 1
+			return audioChannelLeft
 		} else {
-			return 2
+			return audioChannelRight
 		}
 
 	case livekit.AudioMixing_DUAL_CHANNEL_ALTERNATE:
@@ -238,7 +239,7 @@ func (b *AudioBin) getChannel(ts *config.TrackSource) int {
 		return next%2 + 1
 	}
 
-	return 0
+	return audioChannelStereo
 }
 
 func (b *AudioBin) addAudioTestSrcBin() error {
@@ -261,7 +262,7 @@ func (b *AudioBin) addAudioTestSrcBin() error {
 		return errors.ErrGstPipelineError(err)
 	}
 
-	audioCaps, err := newAudioCapsFilter(b.conf, audioChannelNone)
+	audioCaps, err := newAudioCapsFilter(b.conf, audioChannelStereo)
 	if err != nil {
 		return err
 	}
@@ -274,11 +275,11 @@ func (b *AudioBin) addMixer() error {
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
-	if err = audioMixer.SetProperty("latency", audioMixerLatency); err != nil {
+	if err = audioMixer.SetProperty("latency", config.AudioMixerLatency); err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
 
-	mixedCaps, err := newAudioCapsFilter(b.conf, audioChannelNone)
+	mixedCaps, err := newAudioCapsFilter(b.conf, audioChannelStereo)
 	if err != nil {
 		return err
 	}
@@ -317,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.Latency, true)
+	audioQueue, err := gstreamer.BuildQueue("audio_input_queue", config.PipelineLatency, true)
 	if err != nil {
 		return err
 	}
@@ -342,7 +343,7 @@ func addAudioConverter(b *gstreamer.Bin, p *config.PipelineConfig, channel int)
 
 func newAudioCapsFilter(p *config.PipelineConfig, channel int) (*gst.Element, error) {
 	var channelCaps string
-	if channel == audioChannelNone {
+	if channel == audioChannelStereo {
 		channelCaps = "channels=2"
 	} else {
 		channelCaps = fmt.Sprintf("channels=1,channel-mask=(bitmask)0x%d", channel)

pkg/pipeline/builder/stream.go

@@ -68,7 +68,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.Latency); err != nil {
+		if err = mux.SetProperty("latency", config.PipelineLatency); err != nil {
 			return nil, errors.ErrGstPipelineError(err)
 		}
 
@@ -113,7 +113,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.Latency, true)
+	queue, err := gstreamer.BuildQueue(fmt.Sprintf("queue_%s", stream.Name), config.PipelineLatency, true)
 	if err != nil {
 		return nil, errors.ErrGstPipelineError(err)
 	}

pkg/pipeline/builder/video.go

@@ -86,7 +86,7 @@ func BuildVideoBin(pipeline *gstreamer.Pipeline, p *config.PipelineConfig) error
 			return tee.GetRequestPad("src_%u")
 		}
 	} else if len(p.GetEncodedOutputs()) > 0 {
-		queue, err := gstreamer.BuildQueue("video_queue", config.Latency, true)
+		queue, err := gstreamer.BuildQueue("video_queue", config.PipelineLatency, true)
 		if err != nil {
 			return errors.ErrGstPipelineError(err)
 		}
@@ -199,7 +199,7 @@ func (b *VideoBin) buildWebInput() error {
 		return errors.ErrGstPipelineError(err)
 	}
 
-	videoQueue, err := gstreamer.BuildQueue("video_input_queue", config.Latency, true)
+	videoQueue, err := gstreamer.BuildQueue("video_input_queue", config.PipelineLatency, true)
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
@@ -259,17 +259,6 @@ func (b *VideoBin) buildSDKInput() error {
 
 	if b.conf.VideoDecoding {
 		b.bin.SetGetSrcPad(b.getSrcPad)
-		b.bin.SetEOSFunc(func() bool {
-			b.mu.Lock()
-			selected := b.selectedPad
-			pad := b.pads[videoTestSrcName]
-			b.mu.Unlock()
-
-			if selected == videoTestSrcName {
-				pad.SendEvent(gst.NewEOSEvent())
-			}
-			return false
-		})
 
 		if err := b.addVideoTestSrcBin(); err != nil {
 			return err
@@ -313,6 +302,9 @@ func (b *VideoBin) addAppSrcBin(ts *config.TrackSource) error {
 
 func (b *VideoBin) buildAppSrcBin(ts *config.TrackSource, name string) (*gstreamer.Bin, error) {
 	appSrcBin := b.bin.NewBin(name)
+	appSrcBin.SetEOSFunc(func() bool {
+		return false
+	})
 	ts.AppSrc.Element.SetArg("format", "time")
 	if err := ts.AppSrc.Element.SetProperty("is-live", true); err != nil {
 		return nil, errors.ErrGstPipelineError(err)
@@ -472,7 +464,7 @@ func (b *VideoBin) addVideoTestSrcBin() error {
 	}
 	videoTestSrc.SetArg("pattern", "black")
 
-	queue, err := gstreamer.BuildQueue("video_test_src_queue", config.Latency, false)
+	queue, err := gstreamer.BuildQueue("video_test_src_queue", config.PipelineLatency, false)
 	if err != nil {
 		return err
 	}
@@ -521,7 +513,7 @@ func (b *VideoBin) addSelector() error {
 }
 
 func (b *VideoBin) addEncoder() error {
-	videoQueue, err := gstreamer.BuildQueue("video_encoder_queue", config.Latency, false)
+	videoQueue, err := gstreamer.BuildQueue("video_encoder_queue", config.PipelineLatency, false)
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
@@ -648,7 +640,7 @@ func (b *VideoBin) addDecodedVideoSink() error {
 }
 
 func (b *VideoBin) addVideoConverter(bin *gstreamer.Bin) error {
-	videoQueue, err := gstreamer.BuildQueue("video_input_queue", config.Latency, true)
+	videoQueue, err := gstreamer.BuildQueue("video_input_queue", config.PipelineLatency, true)
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
 	}

pkg/pipeline/controller.go

@@ -113,7 +113,7 @@ func New(ctx context.Context, conf *config.PipelineConfig, ipcServiceClient ipc.
 }
 
 func (c *Controller) BuildPipeline() error {
-	p, err := gstreamer.NewPipeline(pipelineName, config.Latency, c.callbacks)
+	p, err := gstreamer.NewPipeline(pipelineName, config.PipelineLatency, c.callbacks)
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
 	}
@@ -334,7 +334,7 @@ func (c *Controller) streamFailed(ctx context.Context, stream *config.Stream, st
 func (c *Controller) onEOSSent() {
 	// for video-only track/track composite, EOS might have already
 	// made it through the pipeline by the time endRecording is closed
-	if c.SourceType == types.SourceTypeSDK && !c.AudioEnabled {
+	if (c.RequestType == types.RequestTypeTrack || c.RequestType == types.RequestTypeTrackComposite) && !c.AudioEnabled {
 		// this will not actually send a second EOS, but will make sure everything is in the correct state
 		c.SendEOS(context.Background(), livekit.EndReasonSrcClosed)
 	}

pkg/pipeline/source/sdk.go

@@ -122,17 +122,13 @@ func (s *SDKSource) CloseWriters() {
 	s.closed.Once(func() {
 		s.sync.End()
 
-		var wg sync.WaitGroup
 		s.mu.Lock()
-		wg.Add(len(s.writers))
 		for _, w := range s.writers {
 			go func(writer *sdk.AppWriter) {
-				defer wg.Done()
 				writer.Drain(false)
 			}(w)
 		}
 		s.mu.Unlock()
-		wg.Wait()
 	})
 }
 

pkg/pipeline/source/sdk/appwriter.go

@@ -39,8 +39,6 @@ import (
 )
 
 const (
-	latency           = time.Second * 2
-	drainTimeout      = time.Millisecond * 3500
 	errBufferTooSmall = "buffer too small"
 )
 
@@ -139,7 +137,7 @@ func NewAppWriter(
 
 	w.buffer = jitter.NewBuffer(
 		depacketizer,
-		latency,
+		config.JitterBufferLatency,
 		w.samples,
 		jitter.WithLogger(w.logger),
 		jitter.WithPacketLossHandler(w.sendPLI),
@@ -178,8 +176,6 @@ func (w *AppWriter) start() {
 		}
 	}
 
-	w.draining.Break()
-
 	w.logger.Infow("writer finished")
 	if w.csvLogger != nil {
 		w.csvLogger.Close()
@@ -231,7 +227,7 @@ func (w *AppWriter) handleReadError(err error) {
 		if lastRecv.IsZero() {
 			lastRecv = w.startTime
 		}
-		if w.pub.IsMuted() || time.Since(lastRecv) > latency {
+		if w.pub.IsMuted() || time.Since(lastRecv) > config.JitterBufferLatency {
 			// set track inactive
 			w.logger.Debugw("track inactive", "timestamp", time.Since(w.startTime))
 			w.active.Store(false)
@@ -247,6 +243,7 @@ func (w *AppWriter) handleReadError(err error) {
 		if !errors.Is(err, io.EOF) {
 			w.logger.Errorw("could not read packet", err)
 		}
+		w.draining.Break()
 		w.endStream.Break()
 	}
 }
@@ -267,7 +264,8 @@ func (w *AppWriter) pushSamples() {
 		case sample := <-w.samples:
 			for _, pkt := range sample {
 				if err := w.pushPacket(pkt); err != nil {
-					w.draining.Once(func() { w.endStream.Break() })
+					w.draining.Break()
+					w.endStream.Break()
 				}
 			}
 		}
@@ -310,7 +308,7 @@ func (w *AppWriter) Drain(force bool) {
 		if force || !w.active.Load() {
 			w.endStream.Break()
 		} else {
-			time.AfterFunc(drainTimeout, func() { w.endStream.Break() })
+			time.AfterFunc(config.AppSrcDrainTimeout, func() { w.endStream.Break() })
 		}
 	})