2022-11-10 12:09:35 +00:00
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
2022-11-10 14:54:32 +00:00
|
|
|
"bufio"
|
2022-11-10 12:09:35 +00:00
|
|
|
"fmt"
|
2022-11-10 14:54:32 +00:00
|
|
|
"io"
|
|
|
|
"log"
|
2022-11-10 12:09:35 +00:00
|
|
|
"net/http"
|
2022-11-10 14:54:32 +00:00
|
|
|
"os"
|
|
|
|
"os/exec"
|
|
|
|
"strconv"
|
|
|
|
"strings"
|
|
|
|
"sync"
|
2022-11-10 15:46:35 +00:00
|
|
|
"syscall"
|
2022-11-10 14:54:32 +00:00
|
|
|
"time"
|
2022-11-10 12:09:35 +00:00
|
|
|
)
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
type Chunk struct {
|
|
|
|
id int
|
|
|
|
done bool
|
|
|
|
notifs []chan bool
|
|
|
|
}
|
|
|
|
|
|
|
|
func NewChunk(id int) *Chunk {
|
|
|
|
return &Chunk{
|
|
|
|
id: id,
|
|
|
|
done: false,
|
|
|
|
notifs: make([]chan bool, 0),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-11-10 12:09:35 +00:00
|
|
|
type Stream struct {
|
2022-11-10 12:27:29 +00:00
|
|
|
c *Config
|
2022-11-10 12:09:35 +00:00
|
|
|
m *Manager
|
|
|
|
quality string
|
|
|
|
height int
|
|
|
|
width int
|
|
|
|
bitrate int
|
2022-11-10 14:54:32 +00:00
|
|
|
|
2022-11-10 15:46:35 +00:00
|
|
|
goal int
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
mutex sync.Mutex
|
|
|
|
chunks map[int]*Chunk
|
|
|
|
|
|
|
|
coder *exec.Cmd
|
2022-11-10 12:09:35 +00:00
|
|
|
}
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
func (s *Stream) ServeList(w http.ResponseWriter) error {
|
2022-11-10 12:09:35 +00:00
|
|
|
WriteM3U8ContentType(w)
|
|
|
|
w.Write([]byte("#EXTM3U\n"))
|
|
|
|
w.Write([]byte("#EXT-X-VERSION:4\n"))
|
|
|
|
w.Write([]byte("#EXT-X-MEDIA-SEQUENCE:0\n"))
|
|
|
|
w.Write([]byte("#EXT-X-PLAYLIST-TYPE:VOD\n"))
|
2022-11-10 17:39:09 +00:00
|
|
|
w.Write([]byte(fmt.Sprintf("#EXT-X-TARGETDURATION:%d\n", s.c.chunkSize)))
|
2022-11-10 12:09:35 +00:00
|
|
|
|
2022-11-10 12:45:10 +00:00
|
|
|
duration := s.m.probe.Duration.Seconds()
|
|
|
|
i := 0
|
|
|
|
for duration > 0 {
|
2022-11-10 17:39:09 +00:00
|
|
|
size := float64(s.c.chunkSize)
|
2022-11-10 12:45:10 +00:00
|
|
|
if duration < size {
|
|
|
|
size = duration
|
|
|
|
}
|
|
|
|
|
|
|
|
w.Write([]byte(fmt.Sprintf("#EXTINF:%.3f, nodesc\n", size)))
|
2022-11-10 12:09:35 +00:00
|
|
|
w.Write([]byte(fmt.Sprintf("%s-%06d.ts\n", s.quality, i)))
|
2022-11-10 12:45:10 +00:00
|
|
|
|
2022-11-10 17:39:09 +00:00
|
|
|
duration -= float64(s.c.chunkSize)
|
2022-11-10 12:45:10 +00:00
|
|
|
i++
|
2022-11-10 12:09:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
w.Write([]byte("#EXT-X-ENDLIST\n"))
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
2022-11-10 12:45:10 +00:00
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
func (s *Stream) ServeChunk(w http.ResponseWriter, id int) error {
|
|
|
|
s.mutex.Lock()
|
|
|
|
defer s.mutex.Unlock()
|
|
|
|
|
2022-11-10 15:46:35 +00:00
|
|
|
s.checkGoal(id)
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
// Already have this chunk
|
|
|
|
if chunk, ok := s.chunks[id]; ok {
|
|
|
|
// Chunk is finished, just return it
|
|
|
|
if chunk.done {
|
|
|
|
s.returnChunk(w, chunk)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Still waiting on transcoder
|
|
|
|
s.waitForChunk(w, chunk)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Will have this soon enough
|
|
|
|
foundBehind := false
|
2022-11-10 15:34:07 +00:00
|
|
|
for i := id - 1; i > id-4 && i >= 0; i-- {
|
2022-11-10 14:54:32 +00:00
|
|
|
if _, ok := s.chunks[i]; ok {
|
|
|
|
foundBehind = true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if foundBehind {
|
|
|
|
// Make sure the chunk exists
|
|
|
|
chunk := s.createChunk(id)
|
|
|
|
|
|
|
|
// Wait for it
|
|
|
|
s.waitForChunk(w, chunk)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Let's start over
|
|
|
|
s.restartAtChunk(w, id)
|
2022-11-10 12:45:10 +00:00
|
|
|
return nil
|
|
|
|
}
|
2022-11-10 14:54:32 +00:00
|
|
|
|
|
|
|
func (s *Stream) createChunk(id int) *Chunk {
|
|
|
|
if c, ok := s.chunks[id]; ok {
|
|
|
|
return c
|
|
|
|
} else {
|
|
|
|
s.chunks[id] = NewChunk(id)
|
|
|
|
return s.chunks[id]
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Stream) returnChunk(w http.ResponseWriter, chunk *Chunk) {
|
|
|
|
// Read file and write to response
|
|
|
|
filename := s.getTsPath(chunk.id)
|
|
|
|
f, err := os.Open(filename)
|
|
|
|
if err != nil {
|
|
|
|
log.Println(err)
|
|
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
defer f.Close()
|
|
|
|
log.Printf("Served chunk %d", chunk.id)
|
|
|
|
w.Header().Set("Content-Type", "video/MP2T")
|
|
|
|
io.Copy(w, f)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Stream) waitForChunk(w http.ResponseWriter, chunk *Chunk) {
|
|
|
|
if chunk.done {
|
|
|
|
s.returnChunk(w, chunk)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// Add our channel
|
|
|
|
notif := make(chan bool)
|
|
|
|
chunk.notifs = append(chunk.notifs, notif)
|
|
|
|
t := time.NewTimer(5 * time.Second)
|
|
|
|
|
|
|
|
s.mutex.Unlock()
|
|
|
|
|
|
|
|
select {
|
|
|
|
case <-notif:
|
|
|
|
t.Stop()
|
|
|
|
case <-t.C:
|
|
|
|
}
|
|
|
|
|
|
|
|
s.mutex.Lock()
|
|
|
|
|
|
|
|
// remove channel
|
|
|
|
for i, c := range chunk.notifs {
|
|
|
|
if c == notif {
|
|
|
|
chunk.notifs = append(chunk.notifs[:i], chunk.notifs[i+1:]...)
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// check for success
|
|
|
|
if chunk.done {
|
|
|
|
s.returnChunk(w, chunk)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Stream) restartAtChunk(w http.ResponseWriter, id int) {
|
2022-11-10 15:34:07 +00:00
|
|
|
// Stop current transcoder
|
|
|
|
if s.coder != nil {
|
|
|
|
s.coder.Process.Kill()
|
2022-11-10 15:46:35 +00:00
|
|
|
s.coder = nil
|
2022-11-10 15:34:07 +00:00
|
|
|
}
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
// Clear everything
|
|
|
|
s.chunks = make(map[int]*Chunk)
|
|
|
|
|
|
|
|
chunk := s.createChunk(id) // create first chunk
|
|
|
|
|
|
|
|
// Start the transcoder
|
2022-11-10 15:59:31 +00:00
|
|
|
s.goal = id + 5
|
2022-11-10 14:54:32 +00:00
|
|
|
s.transcode(id)
|
|
|
|
|
|
|
|
s.waitForChunk(w, chunk) // this is also a request
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Stream) transcode(startId int) {
|
2022-11-10 17:39:09 +00:00
|
|
|
startAt := float64(startId * s.c.chunkSize)
|
2022-11-10 14:54:32 +00:00
|
|
|
|
|
|
|
args := []string{
|
|
|
|
"-loglevel", "warning",
|
|
|
|
}
|
|
|
|
|
|
|
|
if startAt > 0 {
|
|
|
|
args = append(args, []string{
|
|
|
|
"-ss", fmt.Sprintf("%.6f", startAt),
|
|
|
|
}...)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Input specs
|
|
|
|
args = append(args, []string{
|
|
|
|
"-autorotate", "0", // consistent behavior
|
|
|
|
"-i", s.m.path, // Input file
|
|
|
|
"-copyts", // So the "-to" refers to the original TS
|
|
|
|
}...)
|
|
|
|
|
|
|
|
// QSV / encoder selection
|
|
|
|
VAAPI := os.Getenv("VAAPI") == "1"
|
|
|
|
CV := "libx264"
|
|
|
|
VF := ""
|
|
|
|
if VAAPI {
|
|
|
|
CV = "h264_vaapi"
|
|
|
|
VF = "scale_vaapi=w=SCALE_WIDTH:h=SCALE_HEIGHT:force_original_aspect_ratio=decrease"
|
|
|
|
extra := strings.Split("-hwaccel vaapi -hwaccel_device /dev/dri/renderD128 -hwaccel_output_format vaapi", " ")
|
|
|
|
args = append(args, extra...)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Scaling for output
|
|
|
|
var scale string
|
|
|
|
if VAAPI {
|
|
|
|
scale = strings.Replace(VF, "SCALE_WIDTH", fmt.Sprintf("%d", s.width), 1)
|
|
|
|
scale = strings.Replace(scale, "SCALE_HEIGHT", fmt.Sprintf("%d", s.height), 1)
|
|
|
|
} else if s.width >= s.height {
|
|
|
|
scale = fmt.Sprintf("scale=-2:%d", s.height)
|
|
|
|
} else {
|
|
|
|
scale = fmt.Sprintf("scale=%d:-2", s.width)
|
|
|
|
}
|
|
|
|
|
2022-11-11 01:56:38 +00:00
|
|
|
// not original (max)
|
|
|
|
if s.quality != "max" {
|
|
|
|
args = append(args, []string{
|
|
|
|
"-vf", scale,
|
|
|
|
"-maxrate", fmt.Sprintf("%dk", s.bitrate/1000),
|
|
|
|
"-bufsize", fmt.Sprintf("%dK", s.bitrate/3000),
|
|
|
|
}...)
|
|
|
|
}
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
// Output specs
|
|
|
|
args = append(args, []string{
|
|
|
|
"-c:v", CV,
|
|
|
|
"-profile:v", "high",
|
|
|
|
}...)
|
|
|
|
|
|
|
|
// Extra args only for x264
|
|
|
|
if !VAAPI {
|
|
|
|
args = append(args, []string{
|
2022-11-11 01:56:38 +00:00
|
|
|
"-preset", "faster",
|
2022-11-10 14:54:32 +00:00
|
|
|
"-level:v", "4.0",
|
|
|
|
}...)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Audio
|
|
|
|
args = append(args, []string{
|
|
|
|
"-c:a", "aac",
|
|
|
|
"-b:a", "192k",
|
|
|
|
}...)
|
|
|
|
|
|
|
|
// Segmenting specs
|
|
|
|
args = append(args, []string{
|
|
|
|
"-avoid_negative_ts", "disabled",
|
|
|
|
"-f", "hls",
|
2022-11-10 17:39:09 +00:00
|
|
|
"-hls_time", fmt.Sprintf("%d", s.c.chunkSize),
|
|
|
|
"-g", fmt.Sprintf("%d", s.c.chunkSize),
|
2022-11-10 14:54:32 +00:00
|
|
|
"-hls_segment_type", "mpegts",
|
|
|
|
"-start_number", fmt.Sprintf("%d", startId),
|
|
|
|
"-hls_segment_filename", s.getTsPath(-1),
|
|
|
|
"-",
|
|
|
|
}...)
|
|
|
|
|
|
|
|
s.coder = exec.Command(s.c.ffmpeg, args...)
|
2022-11-10 17:39:09 +00:00
|
|
|
log.Println("Starting FFmpeg process with args", strings.Join(s.coder.Args[:], " "))
|
2022-11-10 14:54:32 +00:00
|
|
|
|
|
|
|
cmdStdOut, err := s.coder.StdoutPipe()
|
|
|
|
if err != nil {
|
|
|
|
fmt.Printf("FATAL: ffmpeg command stdout failed with %s\n", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
cmdStdErr, err := s.coder.StderrPipe()
|
|
|
|
if err != nil {
|
|
|
|
fmt.Printf("FATAL: ffmpeg command stdout failed with %s\n", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
err = s.coder.Start()
|
|
|
|
if err != nil {
|
|
|
|
log.Printf("FATAL: ffmpeg command failed with %s\n", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
go s.monitorTranscodeOutput(cmdStdOut, startAt)
|
|
|
|
go s.monitorStderr(cmdStdErr)
|
|
|
|
}
|
|
|
|
|
2022-11-10 15:46:35 +00:00
|
|
|
func (s *Stream) checkGoal(id int) {
|
|
|
|
goal := id + 5
|
|
|
|
if goal > s.goal {
|
|
|
|
s.goal = goal
|
|
|
|
|
|
|
|
// resume encoding
|
|
|
|
if s.coder != nil {
|
2022-11-10 15:59:31 +00:00
|
|
|
log.Println("Resuming encoding")
|
2022-11-10 15:46:35 +00:00
|
|
|
s.coder.Process.Signal(syscall.SIGCONT)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
func (s *Stream) getTsPath(id int) string {
|
|
|
|
if id == -1 {
|
|
|
|
return fmt.Sprintf("/tmp/go-vod/%s/%s-%%06d.ts", s.m.id, s.quality)
|
|
|
|
}
|
|
|
|
return fmt.Sprintf("/tmp/go-vod/%s/%s-%06d.ts", s.m.id, s.quality, id)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Separate goroutine
|
|
|
|
func (s *Stream) monitorTranscodeOutput(cmdStdOut io.ReadCloser, startAt float64) {
|
2022-11-10 15:34:07 +00:00
|
|
|
s.mutex.Lock()
|
|
|
|
coder := s.coder
|
|
|
|
s.mutex.Unlock()
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
defer cmdStdOut.Close()
|
|
|
|
stdoutReader := bufio.NewReader(cmdStdOut)
|
|
|
|
|
|
|
|
for {
|
2022-11-10 15:34:07 +00:00
|
|
|
if s.coder != coder {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
2022-11-10 14:54:32 +00:00
|
|
|
line, err := stdoutReader.ReadBytes('\n')
|
|
|
|
if err == io.EOF {
|
|
|
|
if len(line) == 0 {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
if err != nil {
|
|
|
|
log.Fatal(err)
|
|
|
|
}
|
|
|
|
line = line[:(len(line) - 1)]
|
|
|
|
}
|
|
|
|
|
|
|
|
l := string(line)
|
|
|
|
|
|
|
|
if strings.Contains(l, ".ts") {
|
2022-11-10 15:46:35 +00:00
|
|
|
// 1080p-000003.ts
|
|
|
|
idx := strings.Split(strings.Split(l, "-")[1], ".")[0]
|
|
|
|
id, err := strconv.Atoi(idx)
|
|
|
|
if err != nil {
|
|
|
|
log.Println("Error parsing chunk id")
|
|
|
|
}
|
2022-11-10 15:34:07 +00:00
|
|
|
|
2022-11-10 15:46:35 +00:00
|
|
|
go func() {
|
2022-11-10 15:34:07 +00:00
|
|
|
s.mutex.Lock()
|
|
|
|
defer s.mutex.Unlock()
|
|
|
|
|
2022-11-10 15:46:35 +00:00
|
|
|
// The coder has changed; do nothing
|
2022-11-10 15:34:07 +00:00
|
|
|
if s.coder != coder {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-11-10 15:46:35 +00:00
|
|
|
// Notify everyone
|
2022-11-10 15:34:07 +00:00
|
|
|
chunk := s.createChunk(id)
|
2022-11-10 17:39:09 +00:00
|
|
|
if chunk.done {
|
|
|
|
return
|
|
|
|
}
|
2022-11-10 15:34:07 +00:00
|
|
|
chunk.done = true
|
|
|
|
for _, n := range chunk.notifs {
|
|
|
|
n <- true
|
|
|
|
}
|
2022-11-10 15:46:35 +00:00
|
|
|
|
|
|
|
// Check goal satisfied
|
|
|
|
if id >= s.goal {
|
|
|
|
log.Println("Goal satisfied, pausing encoding")
|
|
|
|
s.coder.Process.Signal(syscall.SIGSTOP)
|
|
|
|
}
|
2022-11-10 15:34:07 +00:00
|
|
|
}()
|
2022-11-10 14:54:32 +00:00
|
|
|
}
|
|
|
|
|
2022-11-10 15:34:07 +00:00
|
|
|
log.Println("ffmpeg:", l)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Join the process
|
2022-11-10 15:59:31 +00:00
|
|
|
coder.Wait()
|
2022-11-10 14:54:32 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (s *Stream) monitorStderr(cmdStdErr io.ReadCloser) {
|
|
|
|
stderrReader := bufio.NewReader(cmdStdErr)
|
|
|
|
|
|
|
|
for {
|
|
|
|
line, err := stderrReader.ReadBytes('\n')
|
|
|
|
if err == io.EOF {
|
|
|
|
if len(line) == 0 {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
if err != nil {
|
|
|
|
log.Fatal(err)
|
|
|
|
}
|
|
|
|
line = line[:(len(line) - 1)]
|
|
|
|
}
|
|
|
|
log.Println("ffmpeg-error:", string(line))
|
|
|
|
}
|
|
|
|
}
|