Files: re-encode a video the browser can't play, on the machine's own GPU, and play that
The viewer tries the file as it is first. When the browser gives up — Safari and an MKV, HEVC without a hardware decoder — or on a tap, a picker: codec (H.264, HEVC, AV1, or the original codec repackaged) and size (original, 1080p, 720p, 480p), each codec labelled with the engine this machine would use. POST /api/files/play runs ffmpeg into an HLS event playlist of fragmented-MP4 segments under <data dir>/transcode/<id>/; Safari plays HLS itself and everything else goes through hls.js, loaded only then. The recipes are tried best first: a card on the bus before the integrated GPU, decoding on the same card before decoding on the CPU and uploading, the software encoders last; a recipe that produces no first segment is dropped for the next. The seekable range grows as ffmpeg gets ahead; status says how far. Streams are reused for the same request, closed when the player goes, swept after ten idle minutes, and the folder is emptied at start. Root mode runs ffmpeg as root through a helper that ends when the daemon lets go of it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Mitchell-Johnson
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
aa4ae57c22
commit
f3c547adfa
@@ -0,0 +1,162 @@
|
||||
package api
|
||||
|
||||
// Playing a video the browser cannot play as it is: POST /api/files/play starts an HLS
|
||||
// re-encode on this machine (internal/media), GET …/play/{id}/index.m3u8 and the segments
|
||||
// feed the player, GET …/play/{id} is its progress, DELETE ends it. /api/files/play-caps
|
||||
// says which codecs this machine can make and on what. Root mode (?sudo=1) reads the file
|
||||
// and runs ffmpeg as root, through the same sudo the file reads use.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"hamlab/internal/files"
|
||||
"hamlab/internal/media"
|
||||
)
|
||||
|
||||
const playLimit = 3 // re-encodes at once on one machine; a GPU does two or three, a CPU one
|
||||
|
||||
type playCaps struct {
|
||||
Codecs []playCodec `json:"codecs"`
|
||||
Qualities []media.Quality `json:"qualities"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
type playCodec struct {
|
||||
media.Codec
|
||||
Engine string `json:"engine,omitempty"` // "" when nothing here encodes it
|
||||
}
|
||||
|
||||
// playCapsHandler: GET /api/files/play-caps[?sudo=1]
|
||||
func (s *Server) playCapsHandler(w http.ResponseWriter, r *http.Request) {
|
||||
root := r.URL.Query().Get("sudo") == "1"
|
||||
if root && s.fsSudo == nil {
|
||||
jsonErr(w, 403, "root access needs passwordless sudo on the host")
|
||||
return
|
||||
}
|
||||
caps, err := s.thumbs.capsFor(r.Context(), root, false)
|
||||
pc := playCaps{Codecs: []playCodec{}, Qualities: media.Qualities}
|
||||
if err != nil {
|
||||
pc.Error = err.Error()
|
||||
} else if caps.Error != "" {
|
||||
pc.Error = caps.Error
|
||||
}
|
||||
for _, c := range media.Codecs {
|
||||
x := playCodec{Codec: c}
|
||||
if pc.Error == "" {
|
||||
x.Engine = media.EngineFor(caps, c.ID)
|
||||
}
|
||||
pc.Codecs = append(pc.Codecs, x)
|
||||
}
|
||||
jsonOK(w, 200, pc)
|
||||
}
|
||||
|
||||
// playStart: POST /api/files/play[?sudo=1] {path, codec, quality}
|
||||
func (s *Server) playStart(w http.ResponseWriter, r *http.Request) {
|
||||
fsys, ok := s.fsFor(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
root := r.URL.Query().Get("sudo") == "1"
|
||||
var body struct{ Path, Codec, Quality string }
|
||||
if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body); err != nil {
|
||||
jsonErr(w, 400, "bad request")
|
||||
return
|
||||
}
|
||||
p, err := files.Clean(body.Path)
|
||||
if err != nil {
|
||||
jsonErr(w, 400, err.Error())
|
||||
return
|
||||
}
|
||||
if media.Kind(p) != "video" {
|
||||
jsonErr(w, 415, "not a video")
|
||||
return
|
||||
}
|
||||
if _, err := fsys.Stat(r.Context(), p); err != nil {
|
||||
fsErr(w, err)
|
||||
return
|
||||
}
|
||||
caps, err := s.thumbs.capsFor(r.Context(), root, false)
|
||||
if err != nil {
|
||||
jsonErr(w, 503, err.Error())
|
||||
return
|
||||
}
|
||||
if caps.Error != "" {
|
||||
jsonErr(w, 503, caps.Error)
|
||||
return
|
||||
}
|
||||
ictx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
|
||||
src, err := media.Inspect(ictx, caps, p, root)
|
||||
cancel()
|
||||
if err != nil {
|
||||
playErr(w, err)
|
||||
return
|
||||
}
|
||||
st, err := s.plays.Open(r.Context(), caps, root, p, body.Codec, body.Quality, src)
|
||||
if err != nil {
|
||||
playErr(w, err)
|
||||
return
|
||||
}
|
||||
log.Printf("files: play %s as %s %s on %s (from %s)", p, st.Codec, st.Quality, st.Engine, clientIP(r))
|
||||
jsonOK(w, 200, map[string]any{"stream": st, "url": "/api/files/play/" + st.ID + "/index.m3u8"})
|
||||
}
|
||||
|
||||
func playErr(w http.ResponseWriter, err error) {
|
||||
var de *media.DecodeError
|
||||
switch {
|
||||
case errors.As(err, &de):
|
||||
jsonErr(w, 422, de.Msg)
|
||||
case strings.Contains(err.Error(), "already being re-encoded"):
|
||||
jsonErr(w, 429, err.Error())
|
||||
case errors.Is(err, context.DeadlineExceeded):
|
||||
jsonErr(w, 504, "the file took too long to read")
|
||||
default:
|
||||
jsonErr(w, 500, err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// playStatus: GET /api/files/play/{id}
|
||||
func (s *Server) playStatus(w http.ResponseWriter, r *http.Request) {
|
||||
st, ok := s.plays.Find(r.PathValue("id"))
|
||||
if !ok {
|
||||
jsonErr(w, 404, "no such stream")
|
||||
return
|
||||
}
|
||||
jsonOK(w, 200, st.Status())
|
||||
}
|
||||
|
||||
// playStop: DELETE /api/files/play/{id}
|
||||
func (s *Server) playStop(w http.ResponseWriter, r *http.Request) {
|
||||
s.plays.Close(r.PathValue("id"))
|
||||
w.WriteHeader(204)
|
||||
}
|
||||
|
||||
// playFile: GET /api/files/play/{id}/{file} — the playlist, the init segment, a segment.
|
||||
func (s *Server) playFile(w http.ResponseWriter, r *http.Request) {
|
||||
st, ok := s.plays.Find(r.PathValue("id"))
|
||||
if !ok {
|
||||
jsonErr(w, 404, "no such stream")
|
||||
return
|
||||
}
|
||||
p, ok := st.File(r.PathValue("file"))
|
||||
if !ok {
|
||||
jsonErr(w, 404, "not a stream file")
|
||||
return
|
||||
}
|
||||
switch filepath.Ext(p) {
|
||||
case ".m3u8":
|
||||
w.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
|
||||
default:
|
||||
w.Header().Set("Content-Type", "video/mp4")
|
||||
}
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
w.Header().Set("X-Content-Type-Options", "nosniff")
|
||||
http.ServeFile(w, r, p)
|
||||
}
|
||||
@@ -35,6 +35,7 @@ import (
|
||||
"hamlab/internal/files"
|
||||
"hamlab/internal/hardware"
|
||||
"hamlab/internal/incus"
|
||||
"hamlab/internal/media"
|
||||
"hamlab/internal/metrics"
|
||||
"hamlab/internal/network"
|
||||
"hamlab/internal/nodes"
|
||||
@@ -80,6 +81,7 @@ type Server struct {
|
||||
fsLocal files.FS
|
||||
fsSudo files.FS // nil without passwordless sudo
|
||||
thumbs *thumbSvc
|
||||
plays *media.Streamer
|
||||
self string // this binary, for helpers run as tasks
|
||||
nodes *nodes.Manager
|
||||
fingerprint string
|
||||
@@ -103,6 +105,7 @@ func New(cfg *config.Config, sessions *auth.Sessions, mgr *term.Manager, col *me
|
||||
s.self = self
|
||||
s.space = space.New(self, filepath.Join(cfg.DataDir(), "space"))
|
||||
s.thumbs = newThumbSvc(cfg.DataDir(), self)
|
||||
s.plays = media.NewStreamer(cfg.DataDir(), self, playLimit)
|
||||
s.hostName() // read it once now so the first request doesn't have to
|
||||
if u, err := user.Current(); err == nil && u.Username != "" {
|
||||
s.user = u.Username
|
||||
@@ -310,6 +313,11 @@ func (s *Server) routes() *http.ServeMux {
|
||||
m.HandleFunc("GET /api/files/thumbs", s.auth(s.thumbsStatus))
|
||||
m.HandleFunc("POST /api/files/thumbs/install", s.auth(s.thumbsInstall))
|
||||
m.HandleFunc("POST /api/files/thumbs/gpu-access", s.auth(s.thumbsGPUAccess))
|
||||
m.HandleFunc("GET /api/files/play-caps", s.auth(s.playCapsHandler))
|
||||
m.HandleFunc("POST /api/files/play", s.auth(s.playStart))
|
||||
m.HandleFunc("GET /api/files/play/{id}", s.auth(s.playStatus))
|
||||
m.HandleFunc("DELETE /api/files/play/{id}", s.auth(s.playStop))
|
||||
m.HandleFunc("GET /api/files/play/{id}/{file}", s.auth(s.playFile))
|
||||
m.HandleFunc("GET /api/hardware", s.auth(s.hardwareReport))
|
||||
m.HandleFunc("GET /api/network", s.auth(s.networkOverview))
|
||||
m.HandleFunc("GET /api/vpn", s.auth(s.vpnReport))
|
||||
|
||||
@@ -0,0 +1,211 @@
|
||||
package media
|
||||
|
||||
// What to run when a file is to be re-encoded for a browser that cannot play it as it is:
|
||||
// the codecs and sizes on offer, the output size and bit rate a choice comes to for a
|
||||
// given source, the ways this machine could do it in the order worth trying, and the
|
||||
// ffmpeg argument list for one of them. All decisions, no processes, so it can be tested.
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"sort"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// Codec is one target a stream may be re-encoded to. "copy" keeps the codec and only
|
||||
// repackages: for a browser that lacks the container, not the decoder.
|
||||
type Codec struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
}
|
||||
|
||||
var Codecs = []Codec{{"h264", "H.264"}, {"hevc", "HEVC"}, {"av1", "AV1"}, {"copy", "Original codec, repackaged"}}
|
||||
|
||||
// Quality is a picture height; the bit rate follows from the size the output ends up at.
|
||||
type Quality struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
Height int `json:"height"` // 0: the source's own
|
||||
}
|
||||
|
||||
var Qualities = []Quality{{"original", "Original size", 0}, {"1080p", "1080p", 1080}, {"720p", "720p", 720}, {"480p", "480p", 480}}
|
||||
|
||||
func qualityByID(id string) (Quality, bool) {
|
||||
for _, q := range Qualities {
|
||||
if q.ID == id {
|
||||
return q, true
|
||||
}
|
||||
}
|
||||
return Quality{}, false
|
||||
}
|
||||
|
||||
func codecKnown(id string) bool {
|
||||
for _, c := range Codecs {
|
||||
if c.ID == id {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// fit is the output: never larger than the source, even dimensions, and a bit rate by the
|
||||
// size it comes to — a quarter more at 50 fps and up, less for the codecs that need less.
|
||||
func fit(q Quality, src Info, codec string) (w, h, kbps int) {
|
||||
w, h = src.Video.Width, src.Video.Height
|
||||
if q.Height > 0 && h > q.Height {
|
||||
w = int(math.Round(float64(w) * float64(q.Height) / float64(h)))
|
||||
h = q.Height
|
||||
}
|
||||
w, h = w&^1, h&^1
|
||||
switch {
|
||||
case h >= 2160:
|
||||
kbps = 20000
|
||||
case h >= 1440:
|
||||
kbps = 12000
|
||||
case h >= 1080:
|
||||
kbps = 8000
|
||||
case h >= 720:
|
||||
kbps = 4000
|
||||
default:
|
||||
kbps = 2000
|
||||
}
|
||||
if src.Video.FPS >= 50 {
|
||||
kbps = kbps * 5 / 4
|
||||
}
|
||||
if codec == "hevc" || codec == "av1" {
|
||||
kbps = kbps * 6 / 10
|
||||
}
|
||||
return w, h, kbps
|
||||
}
|
||||
|
||||
// Recipe is one way to make the stream: the engine's name for the record, and the ffmpeg
|
||||
// arguments that go before the input, the video filter, and the video codec.
|
||||
type Recipe struct {
|
||||
Engine string
|
||||
Pre []string
|
||||
VF string
|
||||
Video []string
|
||||
}
|
||||
|
||||
// recipes are the ways to try, best first: a card on the bus before the integrated GPU;
|
||||
// on a GPU, decoding on the same card (frames never leave it) before decoding in software
|
||||
// and uploading; the software encoders last. An empty list means nothing here can.
|
||||
func recipes(c Caps, codec string, src Info, w, h, kbps int) []Recipe {
|
||||
if codec == "copy" {
|
||||
v := []string{"-c:v", "copy"}
|
||||
if src.Video.Codec == "hevc" {
|
||||
v = append(v, "-tag:v", "hvc1") // Safari reads hvc1, not hev1
|
||||
}
|
||||
return []Recipe{{Engine: "Original codec", Video: v}}
|
||||
}
|
||||
var out []Recipe
|
||||
gpus := append([]GPU(nil), c.GPUs...)
|
||||
sort.SliceStable(gpus, func(i, j int) bool { return !gpus[i].Integrated && gpus[j].Integrated })
|
||||
rate := []string{"-b:v", strconv.Itoa(kbps) + "k", "-maxrate", strconv.Itoa(kbps*5/4) + "k", "-bufsize", strconv.Itoa(kbps*5/2) + "k"}
|
||||
for i := range gpus {
|
||||
g := &gpus[i]
|
||||
if g.API != "vaapi" || !has(g.Encodes, codec) {
|
||||
continue
|
||||
}
|
||||
video := append([]string{"-c:v", codec + "_vaapi"}, rate...)
|
||||
switch codec {
|
||||
case "h264":
|
||||
video = append(video, "-profile:v", "high")
|
||||
case "hevc":
|
||||
video = append(video, "-tag:v", "hvc1")
|
||||
}
|
||||
name := g.Name
|
||||
if name == "" {
|
||||
name = g.Device
|
||||
}
|
||||
scale := fmt.Sprintf("scale_vaapi=w=%d:h=%d:format=nv12", w, h)
|
||||
if has(g.Codecs, src.Video.Codec) {
|
||||
vf := scale
|
||||
if src.HDR() {
|
||||
vf = "tonemap_vaapi=format=nv12:t=bt709:p=bt709:m=bt709," + vf
|
||||
}
|
||||
out = append(out, Recipe{Engine: name + " (vaapi)", Pre: []string{"-hwaccel", "vaapi", "-hwaccel_device", g.Device, "-hwaccel_output_format", "vaapi"}, VF: vf, Video: video})
|
||||
}
|
||||
out = append(out, Recipe{Engine: name + " (vaapi, decoded on the CPU)", Pre: []string{"-init_hw_device", "vaapi=va:" + g.Device, "-filter_hw_device", "va"},
|
||||
VF: fmt.Sprintf("format=nv12,scale=%d:%d,hwupload", w, h), Video: video})
|
||||
}
|
||||
swVF := fmt.Sprintf("scale=%d:%d,format=yuv420p", w, h)
|
||||
if src.HDR() {
|
||||
swVF = fmt.Sprintf("zscale=t=linear:npl=100,format=gbrpf32le,zscale=p=bt709,tonemap=hable,zscale=t=bt709:m=bt709,scale=%d:%d,format=yuv420p", w, h)
|
||||
}
|
||||
cap := []string{"-maxrate", strconv.Itoa(kbps*5/4) + "k", "-bufsize", strconv.Itoa(kbps*5/2) + "k"}
|
||||
switch {
|
||||
case codec == "h264" && has(c.Encoders, "libx264"):
|
||||
out = append(out, Recipe{Engine: "CPU (libx264)", VF: swVF, Video: append([]string{"-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-profile:v", "high"}, cap...)})
|
||||
case codec == "hevc" && has(c.Encoders, "libx265"):
|
||||
out = append(out, Recipe{Engine: "CPU (libx265)", VF: swVF, Video: append([]string{"-c:v", "libx265", "-preset", "fast", "-crf", "26", "-tag:v", "hvc1"}, cap...)})
|
||||
case codec == "av1" && has(c.Encoders, "libsvtav1"):
|
||||
out = append(out, Recipe{Engine: "CPU (SVT-AV1)", VF: swVF, Video: append([]string{"-c:v", "libsvtav1", "-preset", "8", "-crf", "35"}, cap...)})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// EngineFor is the engine the picker will say a codec runs on, "" when none can.
|
||||
func EngineFor(c Caps, codec string) string {
|
||||
if c.Error != "" {
|
||||
return ""
|
||||
}
|
||||
if codec == "copy" {
|
||||
return "Original codec"
|
||||
}
|
||||
r := recipes(c, codec, Info{Video: &Track{Codec: "h264", Width: 1920, Height: 1080}}, 1920, 1080, 8000)
|
||||
if len(r) == 0 {
|
||||
return ""
|
||||
}
|
||||
return r[0].Engine
|
||||
}
|
||||
|
||||
// hlsArgs is the whole ffmpeg argument list for one recipe: the input, the first video and
|
||||
// audio streams, the picture through the recipe, sound as AAC unless it already is, a
|
||||
// keyframe every segment, and an event playlist of fragmented-MP4 segments in dir.
|
||||
func hlsArgs(dir, in string, src Info, r Recipe, copyCodec bool) []string {
|
||||
a := []string{"-nostdin", "-hide_banner", "-loglevel", "error", "-y"}
|
||||
a = append(a, r.Pre...)
|
||||
a = append(a, "-i", in, "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn")
|
||||
if r.VF != "" {
|
||||
a = append(a, "-vf", r.VF)
|
||||
}
|
||||
a = append(a, r.Video...)
|
||||
switch {
|
||||
case src.Audio == nil:
|
||||
a = append(a, "-an")
|
||||
case src.Audio.Codec == "aac":
|
||||
a = append(a, "-c:a", "copy")
|
||||
default:
|
||||
a = append(a, "-c:a", "aac", "-b:a", "160k", "-ac", "2")
|
||||
}
|
||||
if !copyCodec {
|
||||
fps := src.Video.FPS
|
||||
if fps <= 0 {
|
||||
fps = 30
|
||||
}
|
||||
gop := int(math.Round(fps * hlsSeconds))
|
||||
if gop < 24 {
|
||||
gop = 24
|
||||
}
|
||||
if gop > 300 {
|
||||
gop = 300
|
||||
}
|
||||
a = append(a, "-g", strconv.Itoa(gop), "-force_key_frames", fmt.Sprintf("expr:gte(t,n_forced*%d)", hlsSeconds))
|
||||
}
|
||||
a = append(a, "-f", "hls", "-hls_time", strconv.Itoa(hlsSeconds), "-hls_playlist_type", "event", "-hls_segment_type", "fmp4",
|
||||
"-hls_fmp4_init_filename", "init.mp4", "-hls_segment_filename", dir+"/seg%05d.m4s", "-hls_flags", "independent_segments+temp_file",
|
||||
"-progress", dir+"/progress", "-stats_period", "1", dir+"/index.m3u8")
|
||||
return a
|
||||
}
|
||||
|
||||
const hlsSeconds = 4
|
||||
|
||||
func has(list []string, x string) bool {
|
||||
for _, v := range list {
|
||||
if v == x {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
package media
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func src(codec string, w, h int, fps float64) Info {
|
||||
return Info{Duration: 100, Video: &Track{Codec: codec, Width: w, Height: h, FPS: fps}, Audio: &Track{Codec: "opus", Channels: 2}}
|
||||
}
|
||||
|
||||
func TestFit(t *testing.T) {
|
||||
q := func(id string) Quality { x, _ := qualityByID(id); return x }
|
||||
for _, c := range []struct {
|
||||
q string
|
||||
in Info
|
||||
codec string
|
||||
w, h, kbs int
|
||||
}{
|
||||
{"1080p", src("hevc", 3840, 2160, 60), "h264", 1920, 1080, 10000},
|
||||
{"original", src("hevc", 3840, 2160, 30), "h264", 3840, 2160, 20000},
|
||||
{"720p", src("h264", 2292, 960, 60), "h264", 1718, 720, 5000}, // odd width made even
|
||||
{"1080p", src("h264", 1280, 720, 30), "h264", 1280, 720, 4000}, // never upscaled
|
||||
{"480p", src("h264", 1920, 1080, 30), "hevc", 852, 480, 1200}, // 853.3 rounds down, then even
|
||||
{"original", src("av1", 640, 360, 24), "av1", 640, 360, 1200},
|
||||
} {
|
||||
w, h, k := fit(q(c.q), c.in, c.codec)
|
||||
if w != c.w || h != c.h || k != c.kbs {
|
||||
t.Errorf("%s %s %dx%d@%.0f → %dx%d %dk, want %dx%d %dk", c.q, c.codec, c.in.Video.Width, c.in.Video.Height, c.in.Video.FPS, w, h, k, c.w, c.h, c.kbs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func gridCaps() Caps {
|
||||
return Caps{FFmpeg: "ffmpeg", FFprobe: "ffprobe", Encoders: []string{"aac", "libsvtav1", "libx264", "libx265"},
|
||||
GPUs: []GPU{
|
||||
{API: "vaapi", Device: "/dev/dri/renderD128", Name: "Intel UHD Graphics 750", Integrated: true, Codecs: []string{"h264", "hevc", "vp9"}, Encodes: []string{"h264", "hevc"}},
|
||||
{API: "vaapi", Device: "/dev/dri/renderD129", Name: "Intel Arc A380", Codecs: []string{"h264", "hevc", "vp9"}, Encodes: []string{"av1", "h264", "hevc"}},
|
||||
}}
|
||||
}
|
||||
|
||||
func engineNames(r []Recipe) []string {
|
||||
out := make([]string, len(r))
|
||||
for i := range r {
|
||||
out[i] = r[i].Engine
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestRecipesOrder(t *testing.T) {
|
||||
c := gridCaps()
|
||||
// h264 from hevc: the card first, decoding on it; then the iGPU the same way; then the CPU
|
||||
got := engineNames(recipes(c, "h264", src("hevc", 3840, 2160, 60), 1920, 1080, 10000))
|
||||
want := []string{"Intel Arc A380 (vaapi)", "Intel Arc A380 (vaapi, decoded on the CPU)", "Intel UHD Graphics 750 (vaapi)", "Intel UHD Graphics 750 (vaapi, decoded on the CPU)", "CPU (libx264)"}
|
||||
if strings.Join(got, "|") != strings.Join(want, "|") {
|
||||
t.Fatalf("got %v", got)
|
||||
}
|
||||
// av1 source: neither card decodes it, so no same-card recipe, and only the Arc encodes av1
|
||||
got = engineNames(recipes(c, "av1", src("av1", 1920, 1080, 30), 1920, 1080, 4800))
|
||||
want = []string{"Intel Arc A380 (vaapi, decoded on the CPU)", "CPU (SVT-AV1)"}
|
||||
if strings.Join(got, "|") != strings.Join(want, "|") {
|
||||
t.Fatalf("av1: got %v", got)
|
||||
}
|
||||
// copy is one recipe, no filter, hvc1 for hevc
|
||||
r := recipes(c, "copy", src("hevc", 1920, 1080, 30), 1920, 1080, 0)
|
||||
if len(r) != 1 || r[0].VF != "" || !has(r[0].Video, "hvc1") || !has(r[0].Video, "copy") {
|
||||
t.Fatalf("copy: %+v", r)
|
||||
}
|
||||
// a machine with no GPU and no x264: nothing for h264, something for av1
|
||||
bare := Caps{FFmpeg: "ffmpeg", FFprobe: "ffprobe", Encoders: []string{"aac", "libsvtav1"}}
|
||||
if len(recipes(bare, "h264", src("h264", 1280, 720, 30), 1280, 720, 4000)) != 0 {
|
||||
t.Fatal("no x264 means no h264")
|
||||
}
|
||||
if EngineFor(bare, "av1") != "CPU (SVT-AV1)" || EngineFor(bare, "h264") != "" || EngineFor(bare, "copy") != "Original codec" {
|
||||
t.Fatal("engineFor")
|
||||
}
|
||||
if EngineFor(c, "hevc") != "Intel Arc A380 (vaapi)" {
|
||||
t.Fatalf("engineFor on grid: %q", EngineFor(c, "hevc"))
|
||||
}
|
||||
// HDR gets tone-mapped on both paths
|
||||
hdr := src("hevc", 3840, 2160, 60)
|
||||
hdr.Video.ColorTransfer = "smpte2084"
|
||||
r = recipes(c, "h264", hdr, 1920, 1080, 10000)
|
||||
if !strings.HasPrefix(r[0].VF, "tonemap_vaapi") || !strings.Contains(r[len(r)-1].VF, "tonemap=hable") {
|
||||
t.Fatalf("hdr: %q … %q", r[0].VF, r[len(r)-1].VF)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHLSArgs(t *testing.T) {
|
||||
in := src("h264", 1920, 1080, 60)
|
||||
r := recipes(gridCaps(), "h264", in, 1920, 1080, 10000)[0]
|
||||
a := strings.Join(hlsArgs("/d", "/v.mkv", in, r, false), " ")
|
||||
for _, want := range []string{"-hwaccel vaapi -hwaccel_device /dev/dri/renderD129", "-i /v.mkv", "-map 0:v:0 -map 0:a:0?", "-vf scale_vaapi=w=1920:h=1080:format=nv12", "-c:v h264_vaapi -b:v 10000k -maxrate 12500k", "-c:a aac -b:a 160k -ac 2", "-g 240 -force_key_frames expr:gte(t,n_forced*4)", "-hls_segment_type fmp4", "-hls_segment_filename /d/seg%05d.m4s", "/d/index.m3u8"} {
|
||||
if !strings.Contains(a, want) {
|
||||
t.Errorf("missing %q in\n%s", want, a)
|
||||
}
|
||||
}
|
||||
// aac stays as it is; no audio means none; copy sets no keyframes
|
||||
in.Audio.Codec = "aac"
|
||||
if a := strings.Join(hlsArgs("/d", "/v", in, r, false), " "); !strings.Contains(a, "-c:a copy") {
|
||||
t.Error("aac should be copied")
|
||||
}
|
||||
in.Audio = nil
|
||||
if a := strings.Join(hlsArgs("/d", "/v", in, r, false), " "); !strings.Contains(a, " -an ") {
|
||||
t.Error("no audio stream → -an")
|
||||
}
|
||||
cp := recipes(gridCaps(), "copy", in, 1920, 1080, 0)[0]
|
||||
if a := strings.Join(hlsArgs("/d", "/v", in, cp, true), " "); strings.Contains(a, "force_key_frames") || strings.Contains(a, "-vf") {
|
||||
t.Error("copy must not ask for keyframes or filter")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseProgress(t *testing.T) {
|
||||
b := []byte("frame=10\nout_time_us=1000000\nspeed=2.5x\nprogress=continue\nframe=20\nout_time_us=2500000\nspeed=3.1x\nprogress=continue\n")
|
||||
s, sp := parseProgress(b)
|
||||
if s != 2.5 || sp != 3.1 {
|
||||
t.Fatalf("got %v %v", s, sp)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package media
|
||||
|
||||
// Running one long ffmpeg: as the daemon's user directly, or as root through this binary's
|
||||
// run-ffmpeg helper when the file needs root to read. The helper is told to finish by its
|
||||
// standard input closing, which also happens by itself if the daemon goes away — so a
|
||||
// transcode never outlives the console that asked for it.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"hamlab/internal/priv"
|
||||
)
|
||||
|
||||
type ffproc struct {
|
||||
cmd *exec.Cmd
|
||||
in io.WriteCloser // the helper's stdin, root mode only
|
||||
done chan struct{}
|
||||
err error
|
||||
tail *tailBuf
|
||||
}
|
||||
|
||||
// startFF launches ffmpeg with args, under nice where there is one.
|
||||
func startFF(c Caps, self string, root bool, args []string) (*ffproc, error) {
|
||||
var argv []string
|
||||
if root {
|
||||
argv = priv.Wrap(append([]string{self, "run-ffmpeg"}, args...)...)
|
||||
} else {
|
||||
argv = append([]string{c.FFmpeg}, args...)
|
||||
if c.Nice != "" {
|
||||
argv = append([]string{c.Nice, "-n", "10"}, argv...)
|
||||
}
|
||||
}
|
||||
cmd := exec.Command(argv[0], argv[1:]...)
|
||||
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
|
||||
p := &ffproc{cmd: cmd, done: make(chan struct{}), tail: &tailBuf{}}
|
||||
cmd.Stderr = p.tail
|
||||
cmd.Stdout = io.Discard
|
||||
if root {
|
||||
in, err := cmd.StdinPipe()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
p.in = in
|
||||
}
|
||||
if err := cmd.Start(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
go func() {
|
||||
p.err = cmd.Wait()
|
||||
close(p.done)
|
||||
}()
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// exited reports whether ffmpeg has finished, and how.
|
||||
func (p *ffproc) exited() (bool, error) {
|
||||
select {
|
||||
case <-p.done:
|
||||
return true, p.err
|
||||
default:
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
// end asks ffmpeg to finish and waits a little for it.
|
||||
func (p *ffproc) end() {
|
||||
if done, _ := p.exited(); done {
|
||||
return
|
||||
}
|
||||
if p.in != nil {
|
||||
p.in.Close()
|
||||
} else {
|
||||
_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
|
||||
}
|
||||
select {
|
||||
case <-p.done:
|
||||
case <-time.After(5 * time.Second):
|
||||
_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL)
|
||||
<-p.done
|
||||
}
|
||||
}
|
||||
|
||||
// lastWords is what ffmpeg said last, for the record.
|
||||
func (p *ffproc) lastWords() string { return p.tail.last() }
|
||||
|
||||
// tailBuf keeps the last few KB written to it.
|
||||
type tailBuf struct {
|
||||
mu sync.Mutex
|
||||
b []byte
|
||||
}
|
||||
|
||||
const tailKeep = 4096
|
||||
|
||||
func (t *tailBuf) Write(b []byte) (int, error) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.b = append(t.b, b...)
|
||||
if len(t.b) > tailKeep {
|
||||
t.b = t.b[len(t.b)-tailKeep:]
|
||||
}
|
||||
return len(b), nil
|
||||
}
|
||||
|
||||
func (t *tailBuf) last() string {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return lastLine(string(bytes.TrimSpace(t.b)))
|
||||
}
|
||||
|
||||
// RunFFmpegHelper implements `hamlab run-ffmpeg <args…>`: ffmpeg as this user (root, when
|
||||
// the daemon asked through sudo), finishing when stdin closes. Returns ffmpeg's exit code.
|
||||
func RunFFmpegHelper(args []string, stdin io.Reader, stderr io.Writer) int {
|
||||
ffmpeg, err := exec.LookPath("ffmpeg")
|
||||
if err != nil {
|
||||
io.WriteString(stderr, "ffmpeg is not installed\n")
|
||||
return 1
|
||||
}
|
||||
argv := []string{ffmpeg}
|
||||
if nice, err := exec.LookPath("nice"); err == nil {
|
||||
argv = []string{nice, "-n", "10", ffmpeg}
|
||||
}
|
||||
cmd := exec.Command(argv[0], append(argv[1:], args...)...)
|
||||
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
|
||||
cmd.Stderr = stderr
|
||||
cmd.Stdout = io.Discard
|
||||
if err := cmd.Start(); err != nil {
|
||||
io.WriteString(stderr, err.Error()+"\n")
|
||||
return 1
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
_, _ = io.Copy(io.Discard, stdin) // returns when the daemon lets go
|
||||
cancel()
|
||||
}()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- cmd.Wait() }()
|
||||
var werr error
|
||||
select {
|
||||
case werr = <-done:
|
||||
case <-ctx.Done():
|
||||
_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM)
|
||||
select {
|
||||
case werr = <-done:
|
||||
case <-time.After(5 * time.Second):
|
||||
_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
|
||||
werr = <-done
|
||||
}
|
||||
}
|
||||
var ee *exec.ExitError
|
||||
if errors.As(werr, &ee) {
|
||||
return ee.ExitCode()
|
||||
}
|
||||
if werr != nil {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// clearDir empties a folder without removing it.
|
||||
func clearDir(dir string) error {
|
||||
ents, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, e := range ents {
|
||||
if err := os.RemoveAll(dir + "/" + e.Name()); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func hasSegments(playlist string) bool {
|
||||
b, err := os.ReadFile(playlist)
|
||||
return err == nil && strings.Contains(string(b), "#EXTINF")
|
||||
}
|
||||
@@ -0,0 +1,323 @@
|
||||
package media
|
||||
|
||||
// Playing what the browser cannot play as it is: an HLS stream, re-encoded on demand by
|
||||
// ffmpeg on the machine's own GPU where it has one it can use, on its CPU otherwise. A
|
||||
// stream is one file at one codec and size, in a folder of its own under
|
||||
// <data dir>/transcode/: ffmpeg writes an event playlist and fragmented-MP4 segments there
|
||||
// as fast as it can, and the player reads them as they land, so the range it can seek in
|
||||
// grows until the whole file is through — on a GPU, several times faster than watching.
|
||||
// Nothing is kept: a stream nobody has touched for a while is ended and its folder
|
||||
// removed, and the folder is emptied when the daemon starts.
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Stream is one re-encode, in progress or finished.
|
||||
type Stream struct {
|
||||
ID string `json:"id"`
|
||||
Path string `json:"path"`
|
||||
Codec string `json:"codec"`
|
||||
Quality string `json:"quality"`
|
||||
Engine string `json:"engine"`
|
||||
Width int `json:"width"`
|
||||
Height int `json:"height"`
|
||||
Kbps int `json:"kbps"`
|
||||
Source Info `json:"source"`
|
||||
Started time.Time `json:"started"`
|
||||
Root bool `json:"root"`
|
||||
|
||||
key string
|
||||
dir string
|
||||
mu sync.Mutex
|
||||
used time.Time
|
||||
proc *ffproc
|
||||
finished bool
|
||||
failure string
|
||||
}
|
||||
|
||||
// StreamStatus is what the player asks for while the file is still going through.
|
||||
type StreamStatus struct {
|
||||
Running bool `json:"running"`
|
||||
Done bool `json:"done"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Progress float64 `json:"progress"` // seconds of output written
|
||||
Speed float64 `json:"speed"` // × realtime
|
||||
Segments int `json:"segments"`
|
||||
}
|
||||
|
||||
// Streamer keeps one machine's streams.
|
||||
type Streamer struct {
|
||||
dir string
|
||||
self string
|
||||
limit int
|
||||
mu sync.Mutex
|
||||
streams map[string]*Stream
|
||||
}
|
||||
|
||||
const streamIdle = 10 * time.Minute
|
||||
const firstSegmentWait = 40 * time.Second
|
||||
|
||||
var streamFile = regexp.MustCompile(`^(index\.m3u8|init\.mp4|seg\d{5}\.m4s)$`)
|
||||
|
||||
// NewStreamer empties the folder — a daemon restart orphans every stream — and starts the
|
||||
// sweeper.
|
||||
func NewStreamer(dataDir, self string, limit int) *Streamer {
|
||||
m := &Streamer{dir: filepath.Join(dataDir, "transcode"), self: self, limit: limit, streams: map[string]*Stream{}}
|
||||
_ = os.MkdirAll(m.dir, 0o755)
|
||||
_ = clearDir(m.dir)
|
||||
go func() {
|
||||
for range time.Tick(30 * time.Second) {
|
||||
m.Sweep(streamIdle)
|
||||
}
|
||||
}()
|
||||
return m
|
||||
}
|
||||
|
||||
// Open starts a stream, or hands back the one already serving the same request. It
|
||||
// returns once the first segment is on disk, so the player's first fetch is answered.
|
||||
func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, quality string, src Info) (*Stream, error) {
|
||||
if c.Error != "" {
|
||||
return nil, errors.New(c.Error)
|
||||
}
|
||||
if src.Video == nil {
|
||||
return nil, &DecodeError{Msg: "no video stream"}
|
||||
}
|
||||
q, ok := qualityByID(quality)
|
||||
if !ok {
|
||||
return nil, errors.New("unknown quality " + quality)
|
||||
}
|
||||
if !codecKnown(codec) {
|
||||
return nil, errors.New("unknown codec " + codec)
|
||||
}
|
||||
key := fmt.Sprintf("%s|%d|%.0f|%s|%s|%v", path, src.Size, src.Duration, codec, quality, root)
|
||||
m.mu.Lock()
|
||||
for _, s := range m.streams {
|
||||
if s.key == key {
|
||||
m.mu.Unlock()
|
||||
s.touch()
|
||||
return s, nil
|
||||
}
|
||||
}
|
||||
running := 0
|
||||
for _, s := range m.streams {
|
||||
if s.running() {
|
||||
running++
|
||||
}
|
||||
}
|
||||
if running >= m.limit {
|
||||
m.mu.Unlock()
|
||||
return nil, fmt.Errorf("%d videos are already being re-encoded on this machine; one must finish or be closed first", running)
|
||||
}
|
||||
s := &Stream{ID: newID(), Path: path, Codec: codec, Quality: quality, Source: src, Started: time.Now(), Root: root, key: key, used: time.Now()}
|
||||
s.dir = filepath.Join(m.dir, s.ID)
|
||||
m.streams[s.ID] = s
|
||||
m.mu.Unlock()
|
||||
|
||||
s.Width, s.Height, s.Kbps = fit(q, src, codec)
|
||||
if codec == "copy" {
|
||||
s.Width, s.Height, s.Kbps = src.Video.Width, src.Video.Height, int(src.BitRate/1000)
|
||||
}
|
||||
rs := recipes(c, codec, src, s.Width, s.Height, s.Kbps)
|
||||
if len(rs) == 0 {
|
||||
m.forget(s)
|
||||
return nil, errors.New("nothing on this machine encodes " + codec)
|
||||
}
|
||||
var last string
|
||||
for _, r := range rs {
|
||||
if err := os.MkdirAll(s.dir, 0o755); err != nil { // the root helper's ffmpeg writes here too
|
||||
m.forget(s)
|
||||
return nil, err
|
||||
}
|
||||
_ = clearDir(s.dir)
|
||||
args := hlsArgs(s.dir, path, src, r, codec == "copy")
|
||||
p, err := startFF(c, m.self, root, args)
|
||||
if err != nil {
|
||||
last = r.Engine + ": " + err.Error()
|
||||
continue
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.proc, s.Engine = p, r.Engine
|
||||
s.mu.Unlock()
|
||||
if msg, ok := s.waitFirstSegment(ctx); ok {
|
||||
go s.watch()
|
||||
log.Printf("media: stream %s: %s → %s %s %dx%d %dk on %s", s.ID, filepath.Base(path), codec, quality, s.Width, s.Height, s.Kbps, r.Engine)
|
||||
return s, nil
|
||||
} else {
|
||||
last = r.Engine + ": " + msg
|
||||
log.Printf("media: stream %s on %s: %s", filepath.Base(path), r.Engine, msg)
|
||||
}
|
||||
p.end()
|
||||
}
|
||||
m.forget(s)
|
||||
return nil, &DecodeError{Msg: last}
|
||||
}
|
||||
|
||||
// waitFirstSegment watches for the playlist's first segment, or ffmpeg giving up.
|
||||
func (s *Stream) waitFirstSegment(ctx context.Context) (string, bool) {
|
||||
deadline := time.Now().Add(firstSegmentWait)
|
||||
playlist := filepath.Join(s.dir, "index.m3u8")
|
||||
for {
|
||||
if hasSegments(playlist) {
|
||||
return "", true
|
||||
}
|
||||
if done, err := s.proc.exited(); done {
|
||||
msg := s.proc.lastWords()
|
||||
if msg == "" && err != nil {
|
||||
msg = err.Error()
|
||||
}
|
||||
if msg == "" {
|
||||
msg = "ffmpeg ended before writing a segment"
|
||||
}
|
||||
return msg, false
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return "no segment after " + firstSegmentWait.String(), false
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err().Error(), false
|
||||
case <-time.After(250 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// watch records how the encode ended.
|
||||
func (s *Stream) watch() {
|
||||
<-s.proc.done
|
||||
s.mu.Lock()
|
||||
s.finished = true
|
||||
if s.proc.err != nil {
|
||||
s.failure = s.proc.lastWords()
|
||||
if s.failure == "" {
|
||||
s.failure = s.proc.err.Error()
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
if s.failure != "" {
|
||||
log.Printf("media: stream %s ended early: %s", s.ID, s.failure)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Streamer) Find(id string) (*Stream, bool) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
s, ok := m.streams[id]
|
||||
return s, ok
|
||||
}
|
||||
|
||||
// Close ends a stream and removes its folder.
|
||||
func (m *Streamer) Close(id string) {
|
||||
m.mu.Lock()
|
||||
s, ok := m.streams[id]
|
||||
m.mu.Unlock()
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
p := s.proc
|
||||
s.mu.Unlock()
|
||||
if p != nil {
|
||||
p.end()
|
||||
}
|
||||
m.forget(s)
|
||||
}
|
||||
|
||||
func (m *Streamer) forget(s *Stream) {
|
||||
m.mu.Lock()
|
||||
delete(m.streams, s.ID)
|
||||
m.mu.Unlock()
|
||||
_ = os.RemoveAll(s.dir)
|
||||
}
|
||||
|
||||
// Sweep closes what nobody has touched for a while.
|
||||
func (m *Streamer) Sweep(idle time.Duration) {
|
||||
m.mu.Lock()
|
||||
var old []string
|
||||
for id, s := range m.streams {
|
||||
s.mu.Lock()
|
||||
if time.Since(s.used) > idle {
|
||||
old = append(old, id)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
}
|
||||
m.mu.Unlock()
|
||||
for _, id := range old {
|
||||
log.Printf("media: stream %s idle, closed", id)
|
||||
m.Close(id)
|
||||
}
|
||||
}
|
||||
|
||||
// File is the on-disk path of one of the stream's files — the playlist, the init segment
|
||||
// or a media segment — and nothing else.
|
||||
func (s *Stream) File(name string) (string, bool) {
|
||||
if !streamFile.MatchString(name) {
|
||||
return "", false
|
||||
}
|
||||
s.touch()
|
||||
return filepath.Join(s.dir, name), true
|
||||
}
|
||||
|
||||
func (s *Stream) touch() { s.mu.Lock(); s.used = time.Now(); s.mu.Unlock() }
|
||||
|
||||
func (s *Stream) running() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.proc != nil && !s.finished
|
||||
}
|
||||
|
||||
// Status is read from ffmpeg's own progress file and the playlist.
|
||||
func (s *Stream) Status() StreamStatus {
|
||||
s.mu.Lock()
|
||||
st := StreamStatus{Running: s.proc != nil && !s.finished, Done: s.finished && s.failure == "", Error: s.failure}
|
||||
s.mu.Unlock()
|
||||
if b, err := os.ReadFile(filepath.Join(s.dir, "progress")); err == nil {
|
||||
st.Progress, st.Speed = parseProgress(b)
|
||||
}
|
||||
if b, err := os.ReadFile(filepath.Join(s.dir, "index.m3u8")); err == nil {
|
||||
st.Segments = strings.Count(string(b), "#EXTINF")
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// parseProgress takes the last out_time_us and speed from ffmpeg's -progress output.
|
||||
func parseProgress(b []byte) (seconds, speed float64) {
|
||||
sc := bufio.NewScanner(strings.NewReader(string(b)))
|
||||
for sc.Scan() {
|
||||
k, v, ok := strings.Cut(sc.Text(), "=")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
v = strings.TrimSpace(v)
|
||||
switch k {
|
||||
case "out_time_us":
|
||||
if us, err := strconv.ParseInt(v, 10, 64); err == nil && us >= 0 {
|
||||
seconds = float64(us) / 1e6
|
||||
}
|
||||
case "speed":
|
||||
if x, err := strconv.ParseFloat(strings.TrimSuffix(v, "x"), 64); err == nil {
|
||||
speed = x
|
||||
}
|
||||
}
|
||||
}
|
||||
return seconds, speed
|
||||
}
|
||||
|
||||
func newID() string {
|
||||
b := make([]byte, 8)
|
||||
_, _ = rand.Read(b)
|
||||
return hex.EncodeToString(b)
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package media
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The real thing, when ffmpeg is here: a generated clip goes through as H.264 and comes out
|
||||
// as a playlist of segments, the status follows it to the end, and closing removes it all.
|
||||
func TestStreamWithFFmpeg(t *testing.T) {
|
||||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||||
t.Skip("no ffmpeg")
|
||||
}
|
||||
c := Probe(context.Background())
|
||||
if c.Error != "" {
|
||||
t.Skip(c.Error)
|
||||
}
|
||||
if EngineFor(c, "h264") == "" {
|
||||
t.Skip("nothing here encodes h264")
|
||||
}
|
||||
dir := t.TempDir()
|
||||
clip := filepath.Join(dir, "clip.mp4")
|
||||
out, err := exec.Command("ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "testsrc=size=640x360:rate=25:duration=9", "-f", "lavfi", "-i", "sine=frequency=440:duration=9",
|
||||
"-c:v", "mpeg4", "-q:v", "5", "-c:a", "aac", "-shortest", clip).CombinedOutput()
|
||||
if err != nil {
|
||||
t.Skipf("could not make a clip: %s", out)
|
||||
}
|
||||
src, err := Inspect(context.Background(), c, clip, false)
|
||||
if err != nil || src.Video == nil || src.Audio == nil || src.Video.Codec != "mpeg4" || src.Duration < 8 {
|
||||
t.Fatalf("inspect: %+v %v", src, err)
|
||||
}
|
||||
m := NewStreamer(dir, "hamlab", 2)
|
||||
s, err := m.Open(context.Background(), c, false, clip, "h264", "480p", src)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if s.Width != 640 || s.Height != 360 || s.Engine == "" {
|
||||
t.Fatalf("stream: %+v", s)
|
||||
}
|
||||
if p, ok := s.File("index.m3u8"); !ok {
|
||||
t.Fatal("no playlist")
|
||||
} else if !hasSegments(p) {
|
||||
t.Fatal("Open returned before the first segment")
|
||||
}
|
||||
for _, bad := range []string{"../x", "progress", "seg1.m4s", "index.m3u8/", "init.mp4.tmp"} {
|
||||
if _, ok := s.File(bad); ok {
|
||||
t.Fatalf("%q must not be served", bad)
|
||||
}
|
||||
}
|
||||
// the same request is the same stream
|
||||
if again, err := m.Open(context.Background(), c, false, clip, "h264", "480p", src); err != nil || again.ID != s.ID {
|
||||
t.Fatal("not reused")
|
||||
}
|
||||
var st StreamStatus
|
||||
for i := 0; i < 120; i++ {
|
||||
st = s.Status()
|
||||
if st.Done || st.Error != "" {
|
||||
break
|
||||
}
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
}
|
||||
if !st.Done || st.Error != "" || st.Segments < 2 || st.Progress < 8 {
|
||||
t.Fatalf("status at the end: %+v", st)
|
||||
}
|
||||
if init, _ := s.File("init.mp4"); !exists(init) {
|
||||
t.Fatal("no init segment")
|
||||
}
|
||||
m.Close(s.ID)
|
||||
if _, ok := m.Find(s.ID); ok || exists(s.dir) {
|
||||
t.Fatal("close must forget and remove")
|
||||
}
|
||||
// an unplayable source is a DecodeError, and leaves nothing behind
|
||||
bad := filepath.Join(dir, "bad.mkv")
|
||||
_ = os.WriteFile(bad, []byte("not a video"), 0o644)
|
||||
if _, err := Inspect(context.Background(), c, bad, false); err == nil {
|
||||
t.Fatal("inspect of garbage")
|
||||
}
|
||||
if left, _ := os.ReadDir(m.dir); len(left) != 0 {
|
||||
t.Fatalf("left behind: %v", left)
|
||||
}
|
||||
}
|
||||
@@ -12,8 +12,8 @@ import (
|
||||
"hamlab/internal/priv"
|
||||
)
|
||||
|
||||
// Stream is one stream of a media file, as ffprobe reports it.
|
||||
type Stream struct {
|
||||
// Track is one stream of a media file, as ffprobe reports it.
|
||||
type Track struct {
|
||||
Index int `json:"index"`
|
||||
Type string `json:"type"` // video | audio
|
||||
Codec string `json:"codec"`
|
||||
@@ -33,8 +33,8 @@ type Info struct {
|
||||
Duration float64 `json:"duration"`
|
||||
BitRate int64 `json:"bit_rate,omitempty"` // bits per second, whole file
|
||||
Size int64 `json:"size,omitempty"`
|
||||
Video *Stream `json:"video,omitempty"` // the first real picture stream; cover art is not it
|
||||
Audio *Stream `json:"audio,omitempty"` // the first sound stream
|
||||
Video *Track `json:"video,omitempty"` // the first real picture stream; cover art is not it
|
||||
Audio *Track `json:"audio,omitempty"` // the first sound stream
|
||||
}
|
||||
|
||||
// HDR reports a picture that needs tone mapping to look right on an SDR screen.
|
||||
@@ -101,9 +101,9 @@ func Inspect(ctx context.Context, c Caps, path string, root bool) (Info, error)
|
||||
in.Duration, _ = strconv.ParseFloat(pj.Format.Duration, 64)
|
||||
in.BitRate, _ = strconv.ParseInt(pj.Format.BitRate, 10, 64)
|
||||
in.Size, _ = strconv.ParseInt(pj.Format.Size, 10, 64)
|
||||
var art *Stream
|
||||
var art *Track
|
||||
for _, s := range pj.Streams {
|
||||
st := &Stream{Index: s.Index, Type: s.CodecType, Codec: s.CodecName, Profile: s.Profile, PixFmt: s.PixFmt, Width: s.Width, Height: s.Height,
|
||||
st := &Track{Index: s.Index, Type: s.CodecType, Codec: s.CodecName, Profile: s.Profile, PixFmt: s.PixFmt, Width: s.Width, Height: s.Height,
|
||||
ColorTransfer: s.ColorTransfer, Channels: s.Channels, AttachedPic: s.Disposition.AttachedPic != 0}
|
||||
st.FPS = fps(s.AvgFrameRate)
|
||||
if st.FPS == 0 {
|
||||
|
||||
+49
-12
@@ -39,6 +39,8 @@ type GPU struct {
|
||||
Device string `json:"device"` // /dev/dri/renderD128; "0" for cuda (its device index)
|
||||
Name string `json:"name,omitempty"` // "Intel Arc A380"
|
||||
Codecs []string `json:"codecs,omitempty"` // ffmpeg codec names it decodes; nil when nothing could say
|
||||
Encodes []string `json:"encodes,omitempty"` // ffmpeg codec names it encodes, from vainfo
|
||||
Integrated bool `json:"integrated,omitempty"` // on the CPU die; a card on the bus is preferred for encoding
|
||||
Reason string `json:"reason,omitempty"` // Unusable only: why
|
||||
Permission bool `json:"permission,omitempty"` // Unusable only: the node exists but this user may not open it
|
||||
}
|
||||
@@ -49,7 +51,8 @@ type Caps struct {
|
||||
FFprobe string `json:"ffprobe,omitempty"`
|
||||
Nice string `json:"nice,omitempty"`
|
||||
Version string `json:"version,omitempty"`
|
||||
GPUs []GPU `json:"gpus"` // usable, in preference order
|
||||
GPUs []GPU `json:"gpus"` // usable, in node order
|
||||
Encoders []string `json:"encoders,omitempty"` // the encoders ffmpeg itself was built with that matter here
|
||||
Unusable []GPU `json:"unusable,omitempty"` // render nodes that exist but cannot be used, with the reason
|
||||
Error string `json:"error,omitempty"` // why nothing can be made at all
|
||||
ProbedAt time.Time `json:"probed_at"`
|
||||
@@ -139,13 +142,14 @@ func Probe(ctx context.Context) Caps {
|
||||
return c
|
||||
}
|
||||
c.Version = ffmpegVersion(ctx, c.FFmpeg)
|
||||
c.Encoders = encoders(ctx, c.FFmpeg)
|
||||
accels := hwaccels(ctx, c.FFmpeg)
|
||||
if accels["vaapi"] {
|
||||
nodes, _ := filepath.Glob("/dev/dri/renderD*")
|
||||
sort.Strings(nodes)
|
||||
vainfo, _ := exec.LookPath("vainfo")
|
||||
for _, dev := range nodes {
|
||||
g := GPU{API: "vaapi", Device: dev, Name: metrics.RenderNodeName(dev)}
|
||||
g := GPU{API: "vaapi", Device: dev, Name: metrics.RenderNodeName(dev), Integrated: metrics.RenderNodeIntegrated(dev)}
|
||||
if err := openable(dev); err != nil {
|
||||
g.Reason, g.Permission = err.Error(), errors.Is(err, os.ErrPermission)
|
||||
c.Unusable = append(c.Unusable, g)
|
||||
@@ -157,7 +161,7 @@ func Probe(ctx context.Context) Caps {
|
||||
continue
|
||||
}
|
||||
if vainfo != "" {
|
||||
g.Codecs = vainfoCodecs(ctx, vainfo, dev)
|
||||
g.Codecs, g.Encodes = vainfoCodecs(ctx, vainfo, dev)
|
||||
}
|
||||
c.GPUs = append(c.GPUs, g)
|
||||
}
|
||||
@@ -246,28 +250,61 @@ func hwaccels(ctx context.Context, ffmpeg string) map[string]bool {
|
||||
return m
|
||||
}
|
||||
|
||||
var vaProfileRe = regexp.MustCompile(`VAProfile([A-Za-z0-9]+?)(?:Simple|Main|High|Profile|Baseline|Constrained|Scc|Advanced|Version|Stereo|Multiview)\w*\s*:\s*VAEntrypointVLD`)
|
||||
var vaProfileRe = regexp.MustCompile(`VAProfile([A-Za-z0-9]+?)(?:Simple|Main|High|Profile|Baseline|Constrained|Scc|Advanced|Version|Stereo|Multiview)\w*\s*:\s*VAEntrypoint(VLD|EncSlice|EncSliceLP)\b`)
|
||||
|
||||
// wantedEncoders are the encoders this package may ask ffmpeg for.
|
||||
var wantedEncoders = set("libx264 libx265 libsvtav1 libaom-av1 libopenh264 h264_vaapi hevc_vaapi av1_vaapi aac")
|
||||
|
||||
// encoders lists the wanted encoders this ffmpeg was built with.
|
||||
func encoders(ctx context.Context, ffmpeg string) []string {
|
||||
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
out, _ := exec.CommandContext(ctx, ffmpeg, "-hide_banner", "-encoders").Output()
|
||||
var found []string
|
||||
for _, l := range strings.Split(string(out), "\n") {
|
||||
f := strings.Fields(l)
|
||||
if len(f) >= 2 && wantedEncoders[f[1]] {
|
||||
found = append(found, f[1])
|
||||
}
|
||||
}
|
||||
sort.Strings(found)
|
||||
return found
|
||||
}
|
||||
|
||||
// vaCodec maps a VAProfile family to the ffmpeg codec it decodes.
|
||||
var vaCodec = map[string]string{"H264": "h264", "HEVC": "hevc", "VP9": "vp9", "AV1": "av1", "MPEG2": "mpeg2video", "VC1": "vc1", "VP8": "vp8", "JPEG": "mjpeg", "MPEG4": "mpeg4", "H263": "h263"}
|
||||
|
||||
func vainfoCodecs(ctx context.Context, vainfo, dev string) []string {
|
||||
func vainfoCodecs(ctx context.Context, vainfo, dev string) (decodes, encodes []string) {
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
out, _ := exec.CommandContext(ctx, vainfo, "--display", "drm", "--device", dev).Output()
|
||||
return parseVainfo(string(out))
|
||||
}
|
||||
|
||||
// parseVainfo keeps the decode (VLD) profiles and names their codecs the way ffmpeg does.
|
||||
func parseVainfo(text string) []string {
|
||||
seen := map[string]bool{}
|
||||
// parseVainfo names the codecs the decode (VLD) and encode (EncSlice) profiles cover,
|
||||
// the way ffmpeg names them.
|
||||
func parseVainfo(text string) (decodes, encodes []string) {
|
||||
dec, enc := map[string]bool{}, map[string]bool{}
|
||||
for _, m := range vaProfileRe.FindAllStringSubmatch(text, -1) {
|
||||
if c, ok := vaCodec[m[1]]; ok {
|
||||
seen[c] = true
|
||||
c, ok := vaCodec[m[1]]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if m[2] == "VLD" {
|
||||
dec[c] = true
|
||||
} else {
|
||||
enc[c] = true
|
||||
}
|
||||
}
|
||||
out := make([]string, 0, len(seen))
|
||||
for c := range seen {
|
||||
return keys(dec), keys(enc)
|
||||
}
|
||||
|
||||
func keys(m map[string]bool) []string {
|
||||
if len(m) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(m))
|
||||
for c := range m {
|
||||
out = append(out, c)
|
||||
}
|
||||
sort.Strings(out)
|
||||
|
||||
@@ -41,19 +41,31 @@ func TestParseVainfo(t *testing.T) {
|
||||
VAProfileH264High : VAEntrypointEncSliceLP
|
||||
VAProfileJPEGBaseline : VAEntrypointVLD
|
||||
VAProfileHEVCMain10 : VAEntrypointVLD
|
||||
VAProfileHEVCMain10 : VAEntrypointEncSlice
|
||||
VAProfileHEVCMain444_12 : VAEntrypointVLD
|
||||
VAProfileVP9Profile2 : VAEntrypointVLD
|
||||
VAProfileAV1Profile0 : VAEntrypointEncSliceLP
|
||||
`
|
||||
got := parseVainfo(text)
|
||||
want := []string{"h264", "hevc", "mjpeg", "mpeg2video", "vp9"}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("got %v, want %v", got, want)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("got %v, want %v", got, want)
|
||||
dec, enc := parseVainfo(text)
|
||||
same := func(got, want []string) bool {
|
||||
if len(got) != len(want) {
|
||||
return false
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
if !same(dec, []string{"h264", "hevc", "mjpeg", "mpeg2video", "vp9"}) {
|
||||
t.Fatalf("decodes = %v", dec)
|
||||
}
|
||||
if !same(enc, []string{"av1", "h264", "hevc"}) {
|
||||
t.Fatalf("encodes = %v", enc)
|
||||
}
|
||||
if d, e := parseVainfo(""); d != nil || e != nil {
|
||||
t.Fatal("nothing from nothing")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -264,6 +264,17 @@ func RenderNodeName(node string) string {
|
||||
return gpuName(dev, ueventValue(dev+"/uevent", "PCI_SLOT_NAME"), driver)
|
||||
}
|
||||
|
||||
// RenderNodeIntegrated reports a render node whose GPU sits on the CPU die (PCI root bus 0,
|
||||
// or no PCI bus at all), as opposed to a card on the bus.
|
||||
func RenderNodeIntegrated(node string) bool {
|
||||
dev, err := filepath.EvalSymlinks("/sys/class/drm/" + filepath.Base(node) + "/device")
|
||||
if err != nil {
|
||||
return true
|
||||
}
|
||||
pci := ueventValue(dev+"/uevent", "PCI_SLOT_NAME")
|
||||
return pci == "" || strings.HasPrefix(pci, "0000:00:")
|
||||
}
|
||||
|
||||
var gpuVendors = map[string]string{"0x8086": "Intel", "0x1002": "AMD", "0x10de": "NVIDIA", "0x1af4": "Virtio", "0x106b": "Apple"}
|
||||
|
||||
var (
|
||||
|
||||
Reference in New Issue
Block a user