Rework transcoding session loop cancel

livepeer · May 2, 2022 · d0d908b · d0d908b
1 parent ac5cc85
commit d0d908b
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 5 deletions.
diff --git a/core/lb.go b/core/lb.go
@@ -40,9 +40,11 @@ type LoadBalancingTranscoder struct {
 }
 
 func (lb *LoadBalancingTranscoder) EndTranscodingSession(sessionId string) {
+	lb.mu.RLock()
+	defer lb.mu.RUnlock()
 	if session, exists := lb.sessions[sessionId]; exists {
-		session.transcoder.Stop()
-		close(session.done)
+		// signal transcode loop finish for this session
+		close(session.stop)
 		clog.V(common.DEBUG).Infof(context.TODO(), "LB: Transcode session id=%s teared down", session.key)
 	} else {
 		clog.V(common.DEBUG).Infof(context.TODO(), "LB: Transcode session id=%s already finished", session.key)
@@ -112,6 +114,7 @@ func (lb *LoadBalancingTranscoder) createSession(ctx context.Context, md *SegTra
 		transcoder:  lpmsSession,
 		key:         key,
 		done:        make(chan struct{}),
+		stop:        make(chan struct{}),
 		sender:      make(chan *transcoderParams, maxSegmentChannels),
 		makeContext: transcodeLoopContext,
 	}
@@ -169,7 +172,10 @@ type transcoderSession struct {
 	key        string
 
 	sender      chan *transcoderParams
+	// channel to handle Orchestrator error or shutdown during transcoding
 	done        chan struct{}
+	// channel to signal transcoding loop stop, done channel is not used when not transcoding
+	stop		chan struct{}
 	makeContext func() (context.Context, context.CancelFunc)
 }
 
@@ -186,6 +192,10 @@ func (sess *transcoderSession) loop(logCtx context.Context) {
 	for {
 		ctx, cancel := sess.makeContext()
 		select {
+		case <-sess.stop:
+			// Terminate the session after a period of inactivity
+			clog.V(common.DEBUG).Infof(logCtx, "LB: Transcode loop stopped for key=%s", sess.key)
+			return
 		case <-ctx.Done():
 			// Terminate the session after a period of inactivity
 			clog.V(common.DEBUG).Infof(logCtx, "LB: Transcode loop timed out for key=%s", sess.key)

diff --git a/core/transcoder.go b/core/transcoder.go
@@ -23,7 +23,7 @@ import (
 
 type Transcoder interface {
 	Transcode(ctx context.Context, md *SegTranscodingMetadata) (*TranscodeData, error)
-	EndSession(sessionId string)
+	EndTranscodingSession(sessionId string)
 }
 
 type LocalTranscoder struct {
@@ -74,7 +74,7 @@ func (lt *LocalTranscoder) Transcode(ctx context.Context, md *SegTranscodingMeta
 	return resToTranscodeData(ctx, res, opts)
 }
 
-func (lt* LocalTranscoder) EndSession(sessionId string) {
+func (lt* LocalTranscoder) EndTranscodingSession(sessionId string) {
 	// no-op for software transcoder
 }
 
@@ -126,6 +126,10 @@ func (nv *NetintTranscoder) Transcode(ctx context.Context, md *SegTranscodingMet
 	return resToTranscodeData(ctx, res, out)
 }
 
+func (lt *LocalTranscoder) Stop() {
+	//no-op for software transcoder
+}
+
 func (nv *NvidiaTranscoder) Transcode(ctx context.Context, md *SegTranscodingMetadata) (td *TranscodeData, retErr error) {
 	// Returns UnrecoverableError instead of panicking to gracefully notify orchestrator about transcoder's failure
 	defer recoverFromPanic(&retErr)
@@ -160,10 +164,14 @@ func (nv *NvidiaTranscoder) Transcode(ctx context.Context, md *SegTranscodingMet
 	return resToTranscodeData(ctx, res, out)
 }
 
-func (nv* NvidiaTranscoder) EndSession(sessionId string) {
+func (nv* NvidiaTranscoder) EndTranscodingSession(sessionId string) {
 	nv.Stop()
 }
 
+func (nt* NetintTranscoder) EndTranscodingSession(sessionId string) {
+	nt.Stop()
+}
+
 type transcodeTestParams struct {
 	TestAvailable bool
 	Cap           Capability