memories/go-vod/transcoder/manager.go

376 lines
9.0 KiB
Go
Raw Normal View History

2023-11-01 08:03:28 +00:00
package transcoder
2022-11-10 11:24:33 +00:00
import (
2022-11-10 12:27:29 +00:00
"bytes"
"context"
"encoding/json"
2022-11-12 12:35:05 +00:00
"errors"
2022-11-10 11:24:33 +00:00
"fmt"
2022-11-11 03:40:53 +00:00
"hash/fnv"
2022-11-10 12:27:29 +00:00
"log"
2022-11-10 12:09:35 +00:00
"math"
2022-11-10 11:24:33 +00:00
"net/http"
2022-11-11 03:40:53 +00:00
"os"
2022-11-10 12:27:29 +00:00
"os/exec"
2022-11-10 12:09:35 +00:00
"sort"
2022-11-10 12:45:10 +00:00
"strconv"
"strings"
2022-11-10 12:27:29 +00:00
"time"
2022-11-10 11:24:33 +00:00
)
type Manager struct {
2022-11-10 12:27:29 +00:00
c *Config
2022-11-11 03:23:28 +00:00
path string
2022-11-11 03:40:53 +00:00
tempDir string
2022-11-11 03:23:28 +00:00
id string
close chan string
inactive int
2022-11-10 12:09:35 +00:00
2022-11-10 12:27:29 +00:00
probe *ProbeVideoData
2022-11-10 12:09:35 +00:00
numChunks int
streams map[string]*Stream
2022-11-10 11:24:33 +00:00
}
2022-11-10 12:27:29 +00:00
type ProbeVideoData struct {
2022-11-12 12:35:05 +00:00
Width int
Height int
Duration time.Duration
FrameRate int
2022-11-12 17:50:16 +00:00
CodecName string
2022-11-12 18:51:31 +00:00
BitRate int
2023-05-15 01:10:17 +00:00
Rotation int
2022-11-10 12:27:29 +00:00
}
func NewManager(c *Config, path string, id string, close chan string) (*Manager, error) {
m := &Manager{c: c, path: path, id: id, close: close}
2022-11-10 12:09:35 +00:00
m.streams = make(map[string]*Stream)
2022-11-11 03:40:53 +00:00
h := fnv.New32a()
h.Write([]byte(path))
ph := fmt.Sprint(h.Sum32())
2023-03-09 19:57:15 +00:00
m.tempDir = fmt.Sprintf("%s/%s-%s", m.c.TempDir, id, ph)
2022-11-11 03:40:53 +00:00
// Delete temp dir if exists
os.RemoveAll(m.tempDir)
os.MkdirAll(m.tempDir, 0755)
2022-11-10 12:27:29 +00:00
if err := m.ffprobe(); err != nil {
return nil, err
}
2023-03-09 19:57:15 +00:00
m.numChunks = int(math.Ceil(m.probe.Duration.Seconds() / float64(c.ChunkSize)))
2022-11-10 12:27:29 +00:00
// Possible streams
2023-10-25 01:45:40 +00:00
m.streams["480p"] = &Stream{c: c, m: m, quality: "480p", height: 480, width: 854, bitrate: 400}
m.streams["720p"] = &Stream{c: c, m: m, quality: "720p", height: 720, width: 1280, bitrate: 700}
m.streams["1080p"] = &Stream{c: c, m: m, quality: "1080p", height: 1080, width: 1920, bitrate: 1000}
m.streams["1440p"] = &Stream{c: c, m: m, quality: "1440p", height: 1440, width: 2560, bitrate: 1400}
m.streams["2160p"] = &Stream{c: c, m: m, quality: "2160p", height: 2160, width: 3840, bitrate: 3000}
2022-11-10 12:09:35 +00:00
2023-08-05 18:24:07 +00:00
// height is our primary dimension for scaling
// using the probed size, we adjust the width of the stream
// the smaller dimemension of the output should match the height here
smDim, lgDim := m.probe.Height, m.probe.Width
if m.probe.Height > m.probe.Width {
smDim, lgDim = lgDim, smDim
}
2023-10-25 01:45:40 +00:00
// Get the reference bitrate. This is the same as the current bitrate
// if the video is H.264, otherwise use double the current bitrate.
2023-10-27 04:56:31 +00:00
refBitrate := int(float64(m.probe.BitRate) / 2.0)
2023-10-25 01:45:40 +00:00
if m.probe.CodecName != CODEC_H264 {
refBitrate *= 2
}
// If bitrate could not be read, use 10Mbps
if refBitrate == 0 {
refBitrate = 10000000
}
// Get the multiplier for the reference bitrate.
// For this get the nearest stream size to the original.
origPixels := float64(m.probe.Height * m.probe.Width)
nearestPixels := float64(0)
nearestStream := ""
for key, stream := range m.streams {
streamPixels := float64(stream.height * stream.width)
if nearestPixels == 0 || math.Abs(origPixels-streamPixels) < math.Abs(origPixels-nearestPixels) {
nearestPixels = streamPixels
nearestStream = key
}
}
// Get the bitrate multiplier. This is the ratio of the reference
// bitrate to the nearest stream bitrate, so we can scale all streams.
bitrateMultiplier := 1.0
if nearestStream != "" {
bitrateMultiplier = float64(refBitrate) / float64(m.streams[nearestStream].bitrate)
}
2022-11-10 12:27:29 +00:00
// Only keep streams that are smaller than the video
for k, stream := range m.streams {
2022-12-01 21:05:19 +00:00
stream.order = 0
2023-10-25 01:45:40 +00:00
// scale bitrate using the multiplier
stream.bitrate = int(math.Ceil(float64(stream.bitrate) * bitrateMultiplier))
2022-11-12 12:35:05 +00:00
2023-08-05 18:24:07 +00:00
// now store the width of the stream as the larger dimension
stream.width = int(math.Ceil(float64(lgDim) * float64(stream.height) / float64(smDim)))
// remove invalid streams
2023-10-25 01:01:57 +00:00
if (stream.height >= smDim || stream.width >= lgDim) || // no upscaling; we're not AI
2023-08-05 18:24:07 +00:00
(float64(stream.bitrate) > float64(m.probe.BitRate)*0.8) || // no more than 80% of original bitrate
(stream.height%2 != 0 || stream.width%2 != 0) { // no odd dimensions
// remove stream
2022-11-10 12:27:29 +00:00
delete(m.streams, k)
2023-08-05 18:24:07 +00:00
continue
2022-11-12 18:51:31 +00:00
}
2022-11-10 12:27:29 +00:00
}
2022-11-10 12:09:35 +00:00
2022-11-11 01:56:38 +00:00
// Original stream
2023-04-04 02:37:10 +00:00
m.streams[QUALITY_MAX] = &Stream{
2022-12-01 21:05:19 +00:00
c: c, m: m,
2023-04-04 02:37:10 +00:00
quality: QUALITY_MAX,
2022-12-03 06:06:03 +00:00
height: m.probe.Height,
width: m.probe.Width,
2023-10-25 01:45:40 +00:00
bitrate: refBitrate,
2022-12-03 06:06:03 +00:00
order: 1,
2022-11-11 01:56:38 +00:00
}
2022-11-11 02:49:55 +00:00
// Start all streams
for _, stream := range m.streams {
go stream.Run()
}
2022-11-11 02:20:47 +00:00
log.Printf("%s: new manager for %s", m.id, m.path)
2022-11-10 12:45:10 +00:00
2022-11-11 03:23:28 +00:00
// Check for inactivity
go func() {
t := time.NewTicker(5 * time.Second)
defer t.Stop()
for {
<-t.C
2022-11-11 04:14:38 +00:00
if m.inactive == -1 {
t.Stop()
return
}
2022-11-11 03:23:28 +00:00
m.inactive++
// Check if any stream is active
for _, stream := range m.streams {
if stream.coder != nil {
m.inactive = 0
break
}
}
// Nothing done for 5 minutes
2023-03-09 19:57:15 +00:00
if m.inactive >= m.c.ManagerIdleTime/5 {
2022-11-11 04:14:38 +00:00
t.Stop()
2022-11-11 03:23:28 +00:00
m.Destroy()
m.close <- m.id
return
}
}
}()
2022-11-10 12:27:29 +00:00
return m, nil
2022-11-10 11:24:33 +00:00
}
2022-11-11 03:23:28 +00:00
// Destroys streams. DOES NOT emit on the close channel.
func (m *Manager) Destroy() {
2022-11-11 04:14:38 +00:00
log.Printf("%s: destroying manager", m.id)
m.inactive = -1
2022-11-11 03:23:28 +00:00
for _, stream := range m.streams {
stream.Stop()
}
2022-11-11 03:40:53 +00:00
// Delete temp dir
os.RemoveAll(m.tempDir)
2023-03-17 21:06:54 +00:00
// Delete file if temp
freeIfTemp(m.path)
2022-11-11 03:23:28 +00:00
}
2022-11-10 12:09:35 +00:00
func (m *Manager) ServeHTTP(w http.ResponseWriter, r *http.Request, chunk string) error {
2022-11-10 12:45:10 +00:00
// Master list
2022-11-10 12:09:35 +00:00
if chunk == "index.m3u8" {
2022-12-03 06:06:03 +00:00
return m.ServeIndex(w, r)
2022-11-10 12:09:35 +00:00
}
2022-11-10 12:45:10 +00:00
// Stream list
m3u8Sfx := ".m3u8"
if strings.HasSuffix(chunk, m3u8Sfx) {
quality := strings.TrimSuffix(chunk, m3u8Sfx)
if stream, ok := m.streams[quality]; ok {
2022-12-03 06:06:03 +00:00
return stream.ServeList(w, r)
2022-11-10 12:45:10 +00:00
}
}
// Stream chunk
tsSfx := ".ts"
if strings.HasSuffix(chunk, tsSfx) {
parts := strings.Split(chunk, "-")
if len(parts) != 2 {
w.WriteHeader(http.StatusBadRequest)
return nil
}
quality := parts[0]
chunkIdStr := strings.TrimSuffix(parts[1], tsSfx)
chunkId, err := strconv.Atoi(chunkIdStr)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
return nil
}
if stream, ok := m.streams[quality]; ok {
2022-11-10 14:54:32 +00:00
return stream.ServeChunk(w, chunkId)
2022-11-10 12:45:10 +00:00
}
2022-11-10 12:09:35 +00:00
}
2022-11-29 21:00:36 +00:00
// Stream full video
2023-09-30 18:34:18 +00:00
mp4Sfx := ".mp4"
if strings.HasSuffix(chunk, mp4Sfx) {
quality := strings.TrimSuffix(chunk, mp4Sfx)
2022-11-29 21:00:36 +00:00
if stream, ok := m.streams[quality]; ok {
2022-12-03 15:24:24 +00:00
return stream.ServeFullVideo(w, r)
2022-11-29 21:00:36 +00:00
}
2023-09-30 18:34:18 +00:00
// Fall back to original
return m.streams[QUALITY_MAX].ServeFullVideo(w, r)
2022-11-29 21:00:36 +00:00
}
2022-11-10 12:09:35 +00:00
w.WriteHeader(http.StatusNotFound)
return nil
}
2022-12-03 06:06:03 +00:00
func (m *Manager) ServeIndex(w http.ResponseWriter, r *http.Request) error {
2022-11-10 12:09:35 +00:00
WriteM3U8ContentType(w)
w.Write([]byte("#EXTM3U\n"))
// get sorted streams by bitrate
streams := make([]*Stream, 0)
for _, stream := range m.streams {
streams = append(streams, stream)
}
sort.Slice(streams, func(i, j int) bool {
2022-12-01 21:05:19 +00:00
return streams[i].order < streams[j].order ||
2022-12-03 06:06:03 +00:00
(streams[i].order == streams[j].order && streams[i].bitrate < streams[j].bitrate)
2022-11-10 12:09:35 +00:00
})
// Write all streams
2022-12-03 06:06:03 +00:00
query := GetQueryString(r)
2022-11-10 12:09:35 +00:00
for _, stream := range streams {
2022-12-03 06:06:03 +00:00
s := fmt.Sprintf("#EXT-X-STREAM-INF:BANDWIDTH=%d,RESOLUTION=%dx%d,FRAME-RATE=%d\n%s.m3u8%s\n", stream.bitrate, stream.width, stream.height, m.probe.FrameRate, stream.quality, query)
2022-11-10 12:09:35 +00:00
w.Write([]byte(s))
}
return nil
}
2022-11-10 11:24:33 +00:00
2022-11-10 12:27:29 +00:00
func (m *Manager) ffprobe() error {
args := []string{
// Hide debug information
"-v", "error",
2023-05-15 01:38:50 +00:00
// Show everything
2023-08-25 05:12:35 +00:00
"-show_entries", "format:stream",
2022-11-10 12:27:29 +00:00
"-select_streams", "v", // Video stream only, we're not interested in audio
"-of", "json",
m.path,
}
2023-10-21 18:36:22 +00:00
ctx, cancel := context.WithDeadline(context.TODO(), time.Now().Add(5*time.Second))
defer cancel()
2023-03-09 19:57:15 +00:00
cmd := exec.CommandContext(ctx, m.c.FFprobe, args...)
2022-11-10 12:27:29 +00:00
var stdout, stderr bytes.Buffer
cmd.Stdout = &stdout
cmd.Stderr = &stderr
err := cmd.Run()
if err != nil {
log.Println(stderr.String())
return err
}
out := struct {
Streams []struct {
2023-05-15 01:38:50 +00:00
Width int `json:"width"`
Height int `json:"height"`
Duration string `json:"duration"`
FrameRate string `json:"avg_frame_rate"`
CodecName string `json:"codec_name"`
BitRate string `json:"bit_rate"`
SideDataList []struct {
SideDataType string `json:"side_data_type"`
Rotation int `json:"rotation"`
} `json:"side_data_list"`
2022-11-10 12:27:29 +00:00
} `json:"streams"`
Format struct {
Duration string `json:"duration"`
} `json:"format"`
}{}
2023-05-15 01:39:30 +00:00
if err := json.Unmarshal(stdout.Bytes(), &out); err != nil {
2022-11-10 12:27:29 +00:00
return err
}
2022-11-12 12:35:05 +00:00
if len(out.Streams) == 0 {
return errors.New("no video streams found")
}
2022-11-10 12:27:29 +00:00
var duration time.Duration
if out.Streams[0].Duration != "" {
2023-08-25 05:12:35 +00:00
duration, _ = time.ParseDuration(out.Streams[0].Duration + "s")
} else if out.Format.Duration != "" {
duration, _ = time.ParseDuration(out.Format.Duration + "s")
2022-11-10 12:27:29 +00:00
}
2022-11-12 12:35:05 +00:00
// FrameRate is a fraction string
frac := strings.Split(out.Streams[0].FrameRate, "/")
if len(frac) != 2 {
frac = []string{"30", "1"}
}
num, e1 := strconv.Atoi(frac[0])
den, e2 := strconv.Atoi(frac[1])
if e1 != nil || e2 != nil {
num = 30
den = 1
}
frameRate := float64(num) / float64(den)
2022-11-12 18:51:31 +00:00
// BitRate is a string
bitRate, err := strconv.Atoi(out.Streams[0].BitRate)
if err != nil {
bitRate = 5000000
}
2023-05-15 01:38:50 +00:00
// Get rotation from side data
rotation := 0
for _, sideData := range out.Streams[0].SideDataList {
if sideData.SideDataType == "Display Matrix" {
rotation = sideData.Rotation
}
2023-05-15 01:10:17 +00:00
}
2022-11-10 12:27:29 +00:00
m.probe = &ProbeVideoData{
2022-11-12 12:35:05 +00:00
Width: out.Streams[0].Width,
Height: out.Streams[0].Height,
Duration: duration,
FrameRate: int(frameRate),
2022-11-12 17:50:16 +00:00
CodecName: out.Streams[0].CodecName,
2022-11-12 18:51:31 +00:00
BitRate: bitRate,
2023-05-15 01:10:17 +00:00
Rotation: rotation,
2022-11-10 12:27:29 +00:00
}
return nil
}