perf: chunk and process playback segments concurrently

bluenviron · Aug 16, 2024 · ed0caa0 · ed0caa0
1 parent c5059fa
commit ed0caa0
Showing 1 changed file with 110 additions and 45 deletions.
diff --git a/internal/playback/on_list.go b/internal/playback/on_list.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"sort"
 	"time"
 
 	"github.com/bluenviron/mediacommon/pkg/formats/fmp4"
@@ -23,58 +24,122 @@ func (d listEntryDuration) MarshalJSON() ([]byte, error) {
 type listEntry struct {
 	Start    time.Time         `json:"start"`
 	Duration listEntryDuration `json:"duration"`
+	init     *fmp4.Init
+}
+
+func computeDurationAndConcatenateFMP4(segments []*Segment) ([]listEntry, error) {
+	out := []listEntry{}
+	var prevInit *fmp4.Init
+
+	for _, seg := range segments {
+		err := func() error {
+			f, err := os.Open(seg.Fpath)
+			if err != nil {
+				return err
+			}
+			defer f.Close()
+
+			init, err := segmentFMP4ReadInit(f)
+			if err != nil {
+				return err
+			}
+
+			_, err = f.Seek(0, io.SeekStart)
+			if err != nil {
+				return err
+			}
+
+			maxDuration, err := segmentFMP4ReadMaxDuration(f, init)
+			if err != nil {
+				return err
+			}
+
+			if len(out) != 0 && segmentFMP4CanBeConcatenated(
+				prevInit,
+				out[len(out)-1].Start.Add(time.Duration(out[len(out)-1].Duration)),
+				init,
+				seg.Start) {
+				prevStart := out[len(out)-1].Start
+				curEnd := seg.Start.Add(maxDuration)
+				out[len(out)-1].Duration = listEntryDuration(curEnd.Sub(prevStart))
+			} else {
+				out = append(out, listEntry{
+					Start:    seg.Start,
+					Duration: listEntryDuration(maxDuration),
+					init:     init,
+				})
+			}
+
+			prevInit = init
+
+			return nil
+		}()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return out, nil
+}
+
+type concatEntryRes struct {
+	entry []listEntry
+	err   error
 }
 
 func computeDurationAndConcatenate(recordFormat conf.RecordFormat, segments []*Segment) ([]listEntry, error) {
 	if recordFormat == conf.RecordFormatFMP4 {
+		const numWorkers = 4
+		chunkSize := (len(segments) + numWorkers - 1) / numWorkers
+		if chunkSize < 1 {
+			chunkSize = 1
+		}
+
+		ch := make(chan (*concatEntryRes), numWorkers)
+		defer close(ch)
+
+		numChunks := 0
+		for i := 0; i < len(segments); i += chunkSize {
+			end := i + chunkSize
+			if end > len(segments) {
+				end = len(segments)
+			}
+
+			numChunks++
+			go func(segments []*Segment) {
+				entry, err := computeDurationAndConcatenateFMP4(segments)
+				ch <- &concatEntryRes{entry: entry, err: err}
+			}(segments[i:end])
+		}
+
+		entries := []listEntry{}
+		for i := 0; i < numChunks; i++ {
+			res := <-ch
+			if res.err != nil {
+				return nil, res.err
+			}
+			entries = append(entries, res.entry...)
+		}
+		sort.Slice(entries, func(i, j int) bool {
+			return entries[i].Start.Before(entries[j].Start)
+		})
+
 		out := []listEntry{}
 		var prevInit *fmp4.Init
-
-		for _, seg := range segments {
-			err := func() error {
-				f, err := os.Open(seg.Fpath)
-				if err != nil {
-					return err
-				}
-				defer f.Close()
-
-				init, err := segmentFMP4ReadInit(f)
-				if err != nil {
-					return err
-				}
-
-				_, err = f.Seek(0, io.SeekStart)
-				if err != nil {
-					return err
-				}
-
-				maxDuration, err := segmentFMP4ReadMaxDuration(f, init)
-				if err != nil {
-					return err
-				}
-
-				if len(out) != 0 && segmentFMP4CanBeConcatenated(
-					prevInit,
-					out[len(out)-1].Start.Add(time.Duration(out[len(out)-1].Duration)),
-					init,
-					seg.Start) {
-					prevStart := out[len(out)-1].Start
-					curEnd := seg.Start.Add(maxDuration)
-					out[len(out)-1].Duration = listEntryDuration(curEnd.Sub(prevStart))
-				} else {
-					out = append(out, listEntry{
-						Start:    seg.Start,
-						Duration: listEntryDuration(maxDuration),
-					})
-				}
-
-				prevInit = init
-
-				return nil
-			}()
-			if err != nil {
-				return nil, err
+		for _, entry := range entries {
+			if len(out) != 0 && segmentFMP4CanBeConcatenated(
+				prevInit,
+				out[len(out)-1].Start.Add(time.Duration(out[len(out)-1].Duration)),
+				entry.init,
+				entry.Start) {
+				prevStart := out[len(out)-1].Start
+				curEnd := entry.Start.Add(time.Duration(entry.Duration))
+				out[len(out)-1].Duration = listEntryDuration(curEnd.Sub(prevStart))
+			} else {
+				out = append(out, entry)
 			}
+
+			prevInit = entry.init
 		}
 
 		return out, nil