TEMP - ffmpeg output debugging

media/mediamtx.go

@@ -77,6 +77,7 @@
 	if err != nil {
 		return fmt.Errorf("failed to kick connection: %w", err)
 	}
+	defer resp.Body.Close()
 	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusBadRequest {
 		body, _ := io.ReadAll(resp.Body)
 		return fmt.Errorf("kick connection failed with status code: %d body: %s", resp.StatusCode, body)
@@ -100,6 +101,7 @@
 	if err != nil {
 		return false, fmt.Errorf("failed to get stream: %w", err)
 	}
+	defer resp.Body.Close()
 	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusBadRequest {
 		body, _ := io.ReadAll(resp.Body)
 		return false, fmt.Errorf("get stream failed with status code: %d body: %s", resp.StatusCode, body)

media/rtmp2segment.go

@@ -44,6 +44,9 @@
 
 	clog.Infof(ctx, "Starting segmentation for %s", outFilePattern)
 
+	// TODO processSegments needs to also be re-invoked after each retry;
+	//      processes that don't immediately fail are not fully retryable otherwise
+	//
 	// create first named pipe to preempt races between ffmpeg and processSegments
 	createNamedPipe(fmt.Sprintf(outFilePattern, 0))
 	go func() {
@@ -69,8 +72,12 @@
 			clog.Errorf(ctx, "Stopping segmentation in=%s err=%s", in, err)
 			break
 		}
+		if retryCount > 0 {
+			time.Sleep(5 * time.Second)
+		}
 		clog.Infof(ctx, "Starting segmentation. in=%s retryCount=%d", in, retryCount)
 		cmd := exec.CommandContext(procCtx, "ffmpeg",
+			"-analyzeduration", "2500000", // 2.5 seconds
 			"-i", in,
 			"-c:a", "copy",
 			"-c:v", "copy",
@@ -79,11 +86,10 @@
 		)
 		output, err := cmd.CombinedOutput()
 		if err != nil {
-			clog.Errorf(ctx, "Error receiving RTMP: %v", err)
+			clog.Errorf(ctx, "Error receiving RTMP: %v ffmpeg output: %s", err, output)
 			break
 		}
 		clog.Infof(ctx, "Segmentation stopped, will retry. retryCount=%d ffmpeg output: %s", retryCount, output)
-		time.Sleep(5 * time.Second)
 		retryCount++
 	}
 	completionSignal <- true

media/segment_reader.go

@@ -23,6 +23,7 @@
 type SwitchableSegmentReader struct {
 	mu     sync.RWMutex
 	reader SegmentHandler
+	seg    CloneableReader
 }
 
 func NewSwitchableSegmentReader() *SwitchableSegmentReader {
@@ -35,12 +36,18 @@
 	sr.mu.Lock()
 	defer sr.mu.Unlock()
 	sr.reader = newReader
+	if sr.seg != nil {
+		// immediately send the current segment instead of waiting for the next one
+		// clone since current segment may have already been partially consumed
+		sr.reader(sr.seg.Clone())
+	}
 }
 
 func (sr *SwitchableSegmentReader) Read(reader CloneableReader) {
-	sr.mu.RLock()
-	defer sr.mu.RUnlock()
+	sr.mu.Lock()
+	defer sr.mu.Unlock()
 	sr.reader(reader)
+	sr.seg = reader
 }
 
 func (sr *SwitchableSegmentReader) Close() {

server/ai_live_video.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"errors"
@@ -154,7 +155,7 @@
 func startTrickleSubscribe(ctx context.Context, url *url.URL, params aiRequestParams, onFistSegment func()) {
 	// subscribe to the outputs and send them into LPMS
 	subscriber := trickle.NewTrickleSubscriber(url.String())
-	r, w, err := os.Pipe()
+	_, w, err := os.Pipe()
 	if err != nil {
 		params.liveParams.stopPipeline(fmt.Errorf("error getting pipe for trickle-ffmpeg. url=%s %w", url, err))
 		return
@@ -168,7 +169,7 @@
 	ctx = clog.AddVal(ctx, "outputRTMPURL", params.liveParams.outputRTMPURL)
 	ctx = clog.AddVal(ctx, "mediaMTXOutputRTMPURL", params.liveParams.mediaMTXOutputRTMPURL)
 
-	multiWriter := &multiWriter{ctx: ctx, writers: []io.Writer{w, wMediaMTX}}
+	// multiWriter := &multiWriter{ctx: ctx, writers: []io.Writer{w, wMediaMTX}}
 
 	// read segments from trickle subscription
 	go func() {
@@ -215,7 +216,7 @@
 			seq := trickle.GetSeq(segment)
 			clog.V(8).Infof(ctx, "trickle subscribe read data received seq=%d", seq)
 
-			n, err := copySegment(segment, multiWriter)
+			n, err := copySegment(segment, wMediaMTX, seq)
 			if err != nil {
 				params.liveParams.stopPipeline(fmt.Errorf("trickle subscribe error copying: %w", err))
 				return
@@ -229,7 +230,7 @@
 	}()
 
 	// Studio Output ffmpeg process
-	go ffmpegOutput(ctx, params.liveParams.outputRTMPURL, r, params)
+	// go ffmpegOutput(ctx, params.liveParams.outputRTMPURL, r, params)
 
 	// MediaMTX Output ffmpeg process
 	go ffmpegOutput(ctx, params.liveParams.mediaMTXOutputRTMPURL, rMediaMTX, params)
@@ -250,22 +251,35 @@
 		}
 	}()
 	for {
-		clog.V(6).Infof(ctx, "Starting output rtmp")
+		clog.V(6).Infof(ctx, "Starting output rtmp to %s", outputUrl)
 		if !params.inputStreamExists() {
 			clog.Errorf(ctx, "Stopping output rtmp stream, input stream does not exist.")
 			break
 		}
 
 		cmd := exec.Command("ffmpeg",
+			"-loglevel", "debug",
+			"-analyzeduration", "2500000", // 2.5 seconds
+			"-skip_estimate_duration_from_pts", "true",
 			"-i", "pipe:0",
 			"-c:a", "copy",
 			"-c:v", "copy",
 			"-f", "flv",
 			outputUrl,
 		)
 		cmd.Stdin = r
-		output, err := cmd.CombinedOutput()
-		clog.Infof(ctx, "Process output: %s", output)
+		stdout, err := cmd.StderrPipe()
+		if err != nil {
+			clog.Errorf(ctx, "Error getting stdout pipe %v", err)
+		}
+		go func() {
+			scanner := bufio.NewScanner(stdout)
+			for scanner.Scan() {
+				clog.Infof(ctx, "ffmpeg output:%s", scanner.Text())
+			}
+		}()
+		cmd.Start()
+		err = cmd.Wait()
 		if err != nil {
 			clog.Errorf(ctx, "Error sending RTMP out: %v", err)
 			return
@@ -274,9 +288,18 @@
 	}
 }
 
-func copySegment(segment *http.Response, w io.Writer) (int64, error) {
+func copySegment(segment *http.Response, w io.Writer, nb int) (int64, error) {
 	defer segment.Body.Close()
-	return io.Copy(w, segment.Body)
+	var r io.Reader = segment.Body
+	if nb < 10 {
+		outputfile, err := os.Create(fmt.Sprintf("/tmp/seg-%d.ts", nb))
+		if err != nil {
+			clog.Errorf(context.Background(), "Error creating first segment file: %v", err)
+		} else {
+			r = io.TeeReader(segment.Body, outputfile)
+		}
+	}
+	return io.Copy(w, r)
 }
 
 func startControlPublish(control *url.URL, params aiRequestParams) {

server/ai_mediaserver.go

@@ -429,7 +429,13 @@
 		// If auth webhook is set and returns an output URL, this will be replaced
 		outputURL := qp.Get("rtmpOutput")
 
-		mediaMTXOutputURL := fmt.Sprintf("rtmp://%s/aiWebrtc/%s-out", remoteHost, streamName)
+		// Currently for webrtc we need to add a path prefix due to the ingress setup
+		mediaMTXStreamPrefix := r.PathValue("prefix")
+		if mediaMTXStreamPrefix != "" {
+			mediaMTXStreamPrefix = mediaMTXStreamPrefix + "/"
+		}
+
+		mediaMTXOutputURL := fmt.Sprintf("rtmp://%s/%s%s-out", remoteHost, mediaMTXStreamPrefix, streamName)
 		if outputURL == "" {
 			// re-publish to ourselves for now
 			// Not sure if we want this to be permanent
@@ -515,11 +521,6 @@
 		// Kick off the RTMP pull and segmentation as soon as possible
 		ssr := media.NewSwitchableSegmentReader()
 		go func() {
-			// Currently for webrtc we need to add a path prefix due to the ingress setup
-			mediaMTXStreamPrefix := r.PathValue("prefix")
-			if mediaMTXStreamPrefix != "" {
-				mediaMTXStreamPrefix = mediaMTXStreamPrefix + "/"
-			}
 			ms := media.MediaSegmenter{Workdir: ls.LivepeerNode.WorkDir, MediaMTXClient: mediaMTXClient}
 			ms.RunSegmentation(ctx, fmt.Sprintf("rtmp://%s/%s%s", remoteHost, mediaMTXStreamPrefix, streamName), ssr.Read)
 			ssr.Close()