hamlab/internal/media/play.go

package media

// Playing a video in the browser at the quality the connection can take: HLS, made on the fly
// by ffmpeg on the device asked for (the Transcode device setting), YouTube's way. The player
// is told every quality is all there from the start, in Seg-second pieces, and each piece is
// made when it is first asked for.
//
// One ffmpeg runs per play at a time. It starts at the piece asked for and runs on ahead of the
// player, paused once it is far enough ahead; a piece asked for well past where it has got to
// (a seek), or at another quality, ends it and starts a new one there. Every run is made the
// same way (pieceArgs), so pieces from different runs and qualities follow on from one another.
//
// The file is a path on this machine or, when the device asked for is on another one, a URL
// the hub serves it at. Nothing is kept: a play nobody asks anything of is forgotten, and the
// folder is emptied when the daemon starts.

import (
	"bytes"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"io/fs"
	"log"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	runAhead = 10               // pieces a run may make past the last one asked for before it is paused
	jump     = 3                // a piece more than this far past what's made is a seek: start again there
	idleRun  = 45 * time.Second // a run nobody has asked anything of for this long is ended
	idlePlay = 30 * time.Minute // a play nobody has asked anything of for this long is forgotten
	budget   = 2 << 30          // bytes of pieces one play keeps; those asked for least lately go first
	// PieceWait is the longest a request waits for its piece: under the hub proxy's 60 s wait
	// for a node to answer.
	PieceWait = 45 * time.Second
)

var ErrNotFound = errors.New("no such piece")

// Plays is every play on this machine.
type Plays struct {
	dir     string
	self    string // this binary, for root runs through run-ffmpeg
	maxRuns int

	mu    sync.Mutex
	plays map[string]*Play
	stop  chan struct{}
}

// NewPlays keeps pieces under <dataDir>/transcode, which it empties first.
func NewPlays(dataDir, self string, maxRuns int) *Plays {
	dir := filepath.Join(dataDir, "transcode")
	_ = os.MkdirAll(dir, 0o700)
	_ = clearDir(dir)
	m := &Plays{dir: dir, self: self, maxRuns: max(maxRuns, 1), plays: map[string]*Play{}, stop: make(chan struct{})}
	go m.tend()
	return m
}

// Play is one video opened for watching.
type Play struct {
	ID     string `json:"id"`
	Name   string `json:"name"` // the file's name, for the record
	Info   Info   `json:"info"`
	Rungs  []Rung `json:"rungs"`  // best first
	Device string `json:"device"` // asked for: a render node, "cpu", or "" for the best here
	Remote bool   `json:"remote"` // the file is on another machine, read over the network

	caps Caps
	in   string
	root bool
	segs int
	dir  string
	m    *Plays

	mu    sync.Mutex
	seen  time.Time
	run   *playRun
	tried map[int]int // per rung: recipes that failed to make anything, skipped from now on
}

type playRun struct {
	rung   int
	recipe Recipe
	start  int // the first piece this run makes
	head   int // the last piece finished (start-1 before the first)
	want   int // the last piece asked for
	asked  time.Time
	proc   *ffproc
	paused bool
	dead   bool // ended, or exited
	killed bool // ended by us, not by itself
	exited bool
	err    error
	tick   chan struct{} // closed when head moves or the run ends
}

func newPlayID() string {
	b := make([]byte, 12)
	_, _ = rand.Read(b)
	return hex.EncodeToString(b)
}

// Open starts a play of in (a path, or an http URL for a file on another machine). Nothing
// runs until a piece is asked for.
func (m *Plays) Open(c Caps, root bool, in, name string, src Info, device string) (*Play, error) {
	rungs := Ladder(src)
	if len(rungs) == 0 || src.Duration <= 0 {
		return nil, errors.New("this video's length or picture size isn't known, so it can't be played")
	}
	if len(recipes(c, device, src, rungs[0])) == 0 {
		return nil, errors.New("nothing on this machine encodes H.264 (no GPU that does, and no libx264)")
	}
	segs := max(int(math.Ceil((src.Duration-0.25)/Seg)), 1) // the last piece takes in what's left
	p := &Play{ID: newPlayID(), Name: name, Info: src, Rungs: rungs, Device: device, Remote: strings.HasPrefix(in, "http"),
		caps: c, in: in, root: root, segs: segs, m: m, seen: time.Now(), tried: map[int]int{}}
	p.dir = filepath.Join(m.dir, p.ID)
	m.mu.Lock()
	m.plays[p.ID] = p
	m.mu.Unlock()
	return p, nil
}

func (m *Plays) Find(id string) (*Play, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	p, ok := m.plays[id]
	return p, ok
}

// Close ends a play and deletes its pieces.
func (m *Plays) Close(id string) {
	m.mu.Lock()
	p, ok := m.plays[id]
	delete(m.plays, id)
	m.mu.Unlock()
	if ok {
		p.mu.Lock()
		p.end()
		p.mu.Unlock()
		go os.RemoveAll(p.dir)
	}
}

func (p *Play) rung(name string) int {
	for i, r := range p.Rungs {
		if r.Name == name {
			return i
		}
	}
	return -1
}

// Master is the playlist of qualities, lowest first, so a player that starts on the first one
// listed starts on the one that surely plays.
func (p *Play) Master() []byte {
	var b bytes.Buffer
	b.WriteString("#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-INDEPENDENT-SEGMENTS\n")
	for i := len(p.Rungs) - 1; i >= 0; i-- {
		r := p.Rungs[i]
		fmt.Fprintf(&b, "#EXT-X-STREAM-INF:BANDWIDTH=%d,AVERAGE-BANDWIDTH=%d,RESOLUTION=%dx%d,FRAME-RATE=%.3f,CODECS=\"%s\"\n%s.m3u8\n",
			r.Peak, r.Bitrate+audioRate, r.Width, r.Height, r.FPS, r.codecs, r.Name)
	}
	return b.Bytes()
}

// Media is one quality's playlist: every piece, made or not.
func (p *Play) Media(name string) ([]byte, bool) {
	if p.rung(name) < 0 {
		return nil, false
	}
	var b bytes.Buffer
	last := p.Info.Duration - float64(p.segs-1)*Seg
	fmt.Fprintf(&b, "#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-TARGETDURATION:%d\n#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-INDEPENDENT-SEGMENTS\n",
		int(math.Ceil(math.Max(last, Seg))))
	for n := 0; n < p.segs; n++ {
		d := Seg
		if n == p.segs-1 {
			d = last
		}
		fmt.Fprintf(&b, "#EXTINF:%.6f,\n%s/%d.ts\n", d, name, n)
	}
	b.WriteString("#EXT-X-ENDLIST\n")
	return b.Bytes(), true
}

func (p *Play) piece(rung, n int) string {
	return filepath.Join(p.dir, p.Rungs[rung].Name, strconv.Itoa(n)+".ts")
}

// Segment answers the file of piece n at quality name, making it first if need be: it waits
// for the run already making it, or starts one there.
func (p *Play) Segment(done <-chan struct{}, name string, n int) (string, error) {
	ri := p.rung(name)
	if ri < 0 || n < 0 || n >= p.segs {
		return "", ErrNotFound
	}
	path := p.piece(ri, n)
	deadline := time.After(PieceWait)
	for {
		p.mu.Lock()
		now := time.Now()
		p.seen = now
		r := p.run
		if r != nil && r.rung == ri && !r.dead {
			r.want, r.asked = n, now
		}
		if exists(path) {
			p.mu.Unlock()
			_ = os.Chtimes(path, now, now) // the budget keeps what was asked for last
			return path, nil
		}
		if r != nil && r.exited && !r.killed && r.rung == ri && r.start == n {
			// This run started at this very piece and ended without it: that way of making
			// it doesn't work here. The next one is tried, if there is one.
			p.tried[ri]++
			log.Printf("files: play %s: %s made nothing at %s piece %d: %v: %s", p.Name, r.recipe.Engine, name, n, r.err, r.proc.lastWords())
			if p.tried[ri] >= len(recipes(p.caps, p.Device, p.Info, p.Rungs[ri])) {
				err := fmt.Errorf("couldn't make %s: %s", p.Rungs[ri].Label, r.proc.lastWords())
				p.mu.Unlock()
				return "", err
			}
			p.run = nil
			r = nil
		}
		if r == nil || r.dead || r.rung != ri || n < r.start || n <= r.head || n > r.head+jump {
			p.mu.Unlock()
			p.m.makeRoom(p)
			p.mu.Lock()
			if p.run == r { // nobody started one meanwhile
				if err := p.start(ri, n); err != nil {
					p.mu.Unlock()
					return "", err
				}
			}
			p.mu.Unlock()
			continue
		}
		tick := r.tick
		p.mu.Unlock()
		select {
		case <-tick:
		case <-time.After(200 * time.Millisecond):
		case <-done:
			return "", errors.New("gone")
		case <-deadline:
			return "", fmt.Errorf("%s piece %d took longer than %s", name, n, PieceWait)
		}
	}
}

// end stops this play's run. p.mu is held.
func (p *Play) end() {
	if r := p.run; r != nil && !r.dead {
		r.dead, r.killed = true, true
		r.proc.kill()
	}
}

// makeRoom ends the run asked of least lately, if every slot is taken by another play. It
// never holds two plays' locks at once.
func (m *Plays) makeRoom(except *Play) {
	m.mu.Lock()
	plays := make([]*Play, 0, len(m.plays))
	for _, p := range m.plays {
		if p != except {
			plays = append(plays, p)
		}
	}
	m.mu.Unlock()
	var oldest *Play
	var when time.Time
	running := 0
	for _, p := range plays {
		p.mu.Lock()
		if r := p.run; r != nil && !r.dead {
			running++
			if oldest == nil || r.asked.Before(when) {
				oldest, when = p, r.asked
			}
		}
		p.mu.Unlock()
	}
	if running >= m.maxRuns && oldest != nil {
		oldest.mu.Lock()
		oldest.end()
		oldest.mu.Unlock()
		log.Printf("files: %d plays encoding; ended %s's, asked of least lately", running, oldest.Name)
	}
}

// start begins a run making quality ri from piece n on, ending the one before. p.mu is held.
func (p *Play) start(ri, n int) error {
	p.end()
	rs := recipes(p.caps, p.Device, p.Info, p.Rungs[ri])
	if p.tried[ri] >= len(rs) {
		return errors.New("nothing here can make this quality")
	}
	rc := rs[p.tried[ri]]
	out := filepath.Join(p.dir, p.Rungs[ri].Name)
	if err := os.MkdirAll(out, 0o700); err != nil {
		return err
	}
	proc, err := startFF(p.caps, p.m.self, p.root, pieceArgs(out, p.in, p.Info, rc, p.Rungs[ri], n))
	if err != nil {
		return err
	}
	r := &playRun{rung: ri, recipe: rc, start: n, head: n - 1, want: n, asked: time.Now(), proc: proc, tick: make(chan struct{})}
	p.run = r
	go func() {
		<-proc.done
		p.mu.Lock()
		defer p.mu.Unlock()
		r.dead, r.exited, r.err = true, true, proc.err
		p.scan(r)
		close(r.tick)
		r.tick = make(chan struct{})
	}()
	return nil
}

// scan moves a run's head on past the pieces it has finished. p.mu is held.
func (p *Play) scan(r *playRun) bool {
	moved := false
	for r.head+1 < p.segs && exists(p.piece(r.rung, r.head+1)) {
		r.head++
		moved = true
	}
	return moved
}

// PlayStatus is what a play is doing, for the player's menu.
type PlayStatus struct {
	Engine  string `json:"engine,omitempty"` // what the run going now runs on
	Quality string `json:"quality,omitempty"`
	From    int    `json:"from"` // seconds into the film the run started at
	Ahead   int    `json:"ahead"`
	Paused  bool   `json:"paused,omitempty"`
}

func (p *Play) Status() PlayStatus {
	p.mu.Lock()
	defer p.mu.Unlock()
	r := p.run
	if r == nil || r.dead {
		return PlayStatus{}
	}
	return PlayStatus{Engine: r.recipe.Engine, Quality: p.Rungs[r.rung].Label, From: int(float64(r.start) * Seg), Ahead: r.head - r.want, Paused: r.paused}
}

// Engine is what this play would run on first, for the player to say before anything runs.
func (p *Play) Engine() string {
	rs := recipes(p.caps, p.Device, p.Info, p.Rungs[0])
	if len(rs) == 0 {
		return ""
	}
	return rs[0].Engine
}

// tend, ten times a second: notices finished pieces, pauses a run that is far enough ahead and
// wakes it when the player catches up, ends one nobody is asking anything of; and every ten
// seconds keeps each play inside its budget and forgets plays nobody watches.
func (m *Plays) tend() {
	t := time.NewTicker(100 * time.Millisecond)
	defer t.Stop()
	for i := 0; ; i++ {
		select {
		case <-m.stop:
			return
		case <-t.C:
		}
		m.mu.Lock()
		plays := make([]*Play, 0, len(m.plays))
		for _, p := range m.plays {
			plays = append(plays, p)
		}
		m.mu.Unlock()
		now := time.Now()
		for _, p := range plays {
			p.mu.Lock()
			if r := p.run; r != nil && !r.dead {
				if p.scan(r) {
					close(r.tick)
					r.tick = make(chan struct{})
				}
				ahead := r.head - r.want
				switch {
				case now.Sub(r.asked) > idleRun:
					p.end()
				case !r.paused && ahead >= runAhead:
					if r.proc.pause() == nil {
						r.paused = true
					}
				case r.paused && ahead < runAhead-3:
					if r.proc.resume() == nil {
						r.paused = false
					}
				}
			}
			idle := now.Sub(p.seen) > idlePlay
			p.mu.Unlock()
			if idle {
				m.Close(p.ID)
			} else if i%100 == 0 {
				p.trim()
			}
		}
	}
}

// trim deletes the pieces asked for least lately until the play is inside its budget, never
// the ones its run has made ahead of the player.
func (p *Play) trim() {
	type piece struct {
		path string
		size int64
		at   time.Time
	}
	var all []piece
	var total int64
	_ = filepath.WalkDir(p.dir, func(path string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() || !strings.HasSuffix(path, ".ts") {
			return nil
		}
		if fi, err := d.Info(); err == nil {
			all = append(all, piece{path, fi.Size(), fi.ModTime()})
			total += fi.Size()
		}
		return nil
	})
	if total <= budget {
		return
	}
	p.mu.Lock()
	keep := map[string]bool{}
	if r := p.run; r != nil && !r.dead {
		for n := r.want - 1; n <= r.head+1; n++ {
			if n >= 0 && n < p.segs {
				keep[p.piece(r.rung, n)] = true
			}
		}
	}
	p.mu.Unlock()
	sort.Slice(all, func(a, b int) bool { return all[a].at.Before(all[b].at) })
	for _, x := range all {
		if total <= budget*3/4 {
			break
		}
		if !keep[x.path] && os.Remove(x.path) == nil {
			total -= x.size
		}
	}
}