Make sure EOS reaches sinks (livekit#408)

milos-lk · web-flow · commit e42b67acf7d2 · 2025-12-11T23:57:14.000+01:00
* Make sure EOS reaches sinks
New sample callback (app sink callback) is doing push to a blocking queue and in case we are e.g blocked inside output writeSample - inside WaitForMediaTime , there is nothing to pump the data out and the app sink thread gets stuck on the blocking push, not being able to process EOS on time

The change singnals EOS seen on the source out of band - so the output gets a chance to decide to wait a bit for remaining data to go through but also to cancel and close if it takes too much. This should prevent pipeline frozen scenario (where EOS doesn't reach the sink). Eventually we can get rid of the existing workaround - but leaving it just as a safty net for now.

* tidy logging
diff --git a/pkg/media/eos_dispatcher.go b/pkg/media/eos_dispatcher.go
@@ -0,0 +1,43 @@
+package media
+
+import "sync"
+
+// eosDispatcher broadcasts a single EOS signal to any listener that registers,
+// replaying the signal to late subscribers if it already fired.
+type eosDispatcher struct {
+	mu        sync.Mutex
+	fired     bool
+	listeners []func()
+}
+
+func newEOSDispatcher() *eosDispatcher {
+	return &eosDispatcher{}
+}
+
+func (d *eosDispatcher) Fire() {
+	d.mu.Lock()
+	if d.fired {
+		d.mu.Unlock()
+		return
+	}
+	d.fired = true
+	listeners := append([]func(){}, d.listeners...)
+	d.listeners = nil
+	d.mu.Unlock()
+
+	for _, l := range listeners {
+		go l()
+	}
+}
+
+func (d *eosDispatcher) AddListener(f func()) {
+	d.mu.Lock()
+	if d.fired {
+		d.mu.Unlock()
+		// Replay immediately if EOS already fired
+		go f()
+		return
+	}
+	d.listeners = append(d.listeners, f)
+	d.mu.Unlock()
+}
diff --git a/pkg/media/input.go b/pkg/media/input.go
@@ -71,6 +71,8 @@ type Input struct {
 	gateOffset     time.Duration
 	gateTerminator sync.Once
 	latencyCaps    *gst.Caps
+
+	onEOS func()
 }
 
 type OutputReadyFunc func(pad *gst.Pad, kind types.StreamKind)
@@ -143,8 +145,16 @@ func (i *Input) OnOutputReady(f OutputReadyFunc) {
 	i.onOutputReady = f
 }
 
+func (i *Input) SetOnEOS(f func()) {
+	i.onEOS = f
+}
+
 func (i *Input) Start(ctx context.Context, onCloseTimeout func(ctx context.Context)) error {
 	return i.source.Start(ctx, func() {
+		if i.onEOS != nil {
+			i.onEOS()
+		}
+
 		go func() {
 			t := time.NewTimer(5 * time.Second)
 			select {
diff --git a/pkg/media/output.go b/pkg/media/output.go
@@ -41,6 +41,7 @@ const (
 
 	queueCapacity         = 5
 	latencySampleInterval = 500 * time.Millisecond
+	eosQueueDrainTimeout  = 2 * time.Second
 )
 
 // Output manages GStreamer elements that converts & encodes video to the specification that's
@@ -56,6 +57,7 @@ type Output struct {
 	isPlayingTooSlow   func() bool
 	trackStatsGatherer *stats.MediaTrackStatGatherer
 	queue              *utils.BlockingQueue[*sample]
+	eos                *eosDispatcher
 
 	localTrack   atomic.Pointer[lksdk_output.LocalTrack]
 	stopDropping func()
@@ -84,8 +86,8 @@ type AudioOutput struct {
 	codec livekit.AudioCodec
 }
 
-func NewVideoOutput(codec livekit.VideoCodec, layer *livekit.VideoLayer, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, statsGatherer *stats.LocalMediaStatsGatherer) (*VideoOutput, error) {
-	e, err := newVideoOutput(codec, outputSync, isPlayingTooSlow)
+func NewVideoOutput(codec livekit.VideoCodec, layer *livekit.VideoLayer, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, statsGatherer *stats.LocalMediaStatsGatherer, eos *eosDispatcher) (*VideoOutput, error) {
+	e, err := newVideoOutput(codec, outputSync, isPlayingTooSlow, eos)
 	if err != nil {
 		return nil, err
 	}
@@ -248,8 +250,8 @@ func NewVideoOutput(codec livekit.VideoCodec, layer *livekit.VideoLayer, outputS
 	return e, nil
 }
 
-func NewAudioOutput(options *livekit.IngressAudioEncodingOptions, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, statsGatherer *stats.LocalMediaStatsGatherer) (*AudioOutput, error) {
-	e, err := newAudioOutput(options.AudioCodec, outputSync, isPlayingTooSlow)
+func NewAudioOutput(options *livekit.IngressAudioEncodingOptions, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, statsGatherer *stats.LocalMediaStatsGatherer, eos *eosDispatcher) (*AudioOutput, error) {
+	e, err := newAudioOutput(options.AudioCodec, outputSync, isPlayingTooSlow, eos)
 	if err != nil {
 		return nil, err
 	}
@@ -354,8 +356,8 @@ func NewAudioOutput(options *livekit.IngressAudioEncodingOptions, outputSync *ut
 	return e, nil
 }
 
-func newVideoOutput(codec livekit.VideoCodec, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool) (*VideoOutput, error) {
-	e, err := newOutput(outputSync, isPlayingTooSlow)
+func newVideoOutput(codec livekit.VideoCodec, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, eos *eosDispatcher) (*VideoOutput, error) {
+	e, err := newOutput(outputSync, isPlayingTooSlow, eos)
 	if err != nil {
 		return nil, err
 	}
@@ -373,8 +375,8 @@ func newVideoOutput(codec livekit.VideoCodec, outputSync *utils.TrackOutputSynch
 	return o, nil
 }
 
-func newAudioOutput(codec livekit.AudioCodec, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool) (*AudioOutput, error) {
-	e, err := newOutput(outputSync, isPlayingTooSlow)
+func newAudioOutput(codec livekit.AudioCodec, outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, eos *eosDispatcher) (*AudioOutput, error) {
+	e, err := newOutput(outputSync, isPlayingTooSlow, eos)
 	if err != nil {
 		return nil, err
 	}
@@ -392,7 +394,7 @@ func newAudioOutput(codec livekit.AudioCodec, outputSync *utils.TrackOutputSynch
 	return o, nil
 }
 
-func newOutput(outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool) (*Output, error) {
+func newOutput(outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func() bool, eos *eosDispatcher) (*Output, error) {
 	sink, err := app.NewAppSink()
 	if err != nil {
 		return nil, err
@@ -403,6 +405,11 @@ func newOutput(outputSync *utils.TrackOutputSynchronizer, isPlayingTooSlow func(
 		sink:             sink,
 		outputSync:       outputSync,
 		isPlayingTooSlow: isPlayingTooSlow,
+		eos:              eos,
+	}
+
+	if e.eos != nil {
+		e.eos.AddListener(e.onSourceEOS)
 	}
 
 	e.start()
@@ -512,6 +519,7 @@ func (e *Output) QueueLength() int {
 }
 
 func (e *Output) Close() error {
+	e.logger.Debugw("closing output")
 
 	e.closed.Break()
 	e.outputSync.Close()
@@ -520,6 +528,22 @@ func (e *Output) Close() error {
 	return nil
 }
 
+func (e *Output) onSourceEOS() {
+	e.logger.Debugw("eos received, eventually closing queue after timeout")
+	go func() {
+		timer := time.NewTimer(eosQueueDrainTimeout)
+		defer timer.Stop()
+
+		select {
+		case <-timer.C:
+			e.Close()
+			e.logger.Debugw("output closed on eos timeout")
+		case <-e.closed.Watch():
+			// already closed as a result of handling EOS in-band
+		}
+	}()
+}
+
 func (e *VideoOutput) handleSample(sink *app.Sink) gst.FlowReturn {
 	// Return an error if the last write failed
 	if errPtr := e.pipelineErr.Load(); errPtr != nil {
diff --git a/pkg/media/pipeline.go b/pkg/media/pipeline.go
@@ -52,6 +52,8 @@ type Pipeline struct {
 	cancel atomic.Pointer[context.CancelFunc]
 
 	pipelineErr chan error
+
+	eos *eosDispatcher
 }
 
 func New(ctx context.Context, conf *config.Config, params *params.Params, g *stats.LocalMediaStatsGatherer) (*Pipeline, error) {
@@ -83,8 +85,11 @@ func New(ctx context.Context, conf *config.Config, params *params.Params, g *sta
 		pipeline:    pipeline,
 		input:       input,
 		pipelineErr: make(chan error, 1),
+		eos:         newEOSDispatcher(),
 	}
 
+	input.SetOnEOS(p.eos.Fire)
+
 	sink, err := NewWebRTCSink(ctx, params, func() {
 		if cancel := p.cancel.Load(); cancel != nil {
 			(*cancel)()
@@ -93,7 +98,7 @@ func New(ctx context.Context, conf *config.Config, params *params.Params, g *sta
 		if p.loop != nil {
 			p.loop.Quit()
 		}
-	}, g)
+	}, g, p.eos)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/media/webrtc_sink.go b/pkg/media/webrtc_sink.go
@@ -54,13 +54,14 @@ type WebRTCSink struct {
 	outputSync      *utils.OutputSynchronizer
 	spliceProcessor *SpliceProcessor
 	statsGatherer   *stats.LocalMediaStatsGatherer
+	eos             *eosDispatcher
 
 	// logging
 	tooSlowThrottle  core.Throttle
 	tooSlowLogEvents atomic.Int32
 }
 
-func NewWebRTCSink(ctx context.Context, p *params.Params, onFailure func(), statsGatherer *stats.LocalMediaStatsGatherer) (*WebRTCSink, error) {
+func NewWebRTCSink(ctx context.Context, p *params.Params, onFailure func(), statsGatherer *stats.LocalMediaStatsGatherer, eos *eosDispatcher) (*WebRTCSink, error) {
 	ctx, span := tracer.Start(ctx, "media.NewWebRTCSink")
 	defer span.End()
 
@@ -70,6 +71,7 @@ func NewWebRTCSink(ctx context.Context, p *params.Params, onFailure func(), stat
 		errChan:         make(chan error),
 		outputSync:      utils.NewOutputSynchronizer(),
 		statsGatherer:   statsGatherer,
+		eos:             eos,
 		tooSlowThrottle: core.NewThrottle(5 * time.Second),
 	}
 
@@ -104,7 +106,7 @@ func NewWebRTCSink(ctx context.Context, p *params.Params, onFailure func(), stat
 }
 
 func (s *WebRTCSink) addAudioTrack() (*Output, error) {
-	output, err := NewAudioOutput(s.params.AudioEncodingOptions, s.outputSync.AddTrack(), s.isPlayingTooSlow, s.statsGatherer)
+	output, err := NewAudioOutput(s.params.AudioEncodingOptions, s.outputSync.AddTrack(), s.isPlayingTooSlow, s.statsGatherer, s.eos)
 	if err != nil {
 		logger.Errorw("could not create output", err)
 		return nil, err
@@ -190,7 +192,7 @@ func (s *WebRTCSink) addVideoTrack(w, h int) ([]*Output, error) {
 	sortedLayers := filterAndSortLayersByQuality(s.params.VideoEncodingOptions.Layers, w, h)
 
 	for _, layer := range sortedLayers {
-		output, err := NewVideoOutput(s.params.VideoEncodingOptions.VideoCodec, layer, s.outputSync.AddTrack(), s.isPlayingTooSlow, s.statsGatherer)
+		output, err := NewVideoOutput(s.params.VideoEncodingOptions.VideoCodec, layer, s.outputSync.AddTrack(), s.isPlayingTooSlow, s.statsGatherer, s.eos)
 		if err != nil {
 			return nil, err
 		}
