diff --git a/internal/api/play.go b/internal/api/play.go index c5730bf..83fbfa7 100644 --- a/internal/api/play.go +++ b/internal/api/play.go @@ -1,141 +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. +// Playing a video in the Files viewer, YouTube's way: Auto (the player reads the connection), +// any quality by hand, or the file itself. The qualities are made on the fly as HLS by +// internal/media, on the device the Transcode device setting names — which may be on another +// machine than the file. +// +// On every machine: +// POST /api/files/play[?sudo=1] {path | url, name, device} open a play +// GET /api/files/play/{id}/master.m3u8, /{rung}.m3u8, /{rung}/{n}.ts +// GET /api/files/play/{id} what it is doing +// DELETE /api/files/play/{id} (or POST …/close, for a beacon) end it +// GET /api/files/play-devices what it can encode on +// On the hub: +// POST /api/play {node, path, sudo} open a play of a file on node, wherever the setting +// says, answering which machine the play is on +// GET|PUT /api/transcode the setting, and every machine's devices +// GET /api/play-src/{token} a file, for a machine transcoding it, by a token +// that names that one file; ranges pass through import ( "context" + "crypto/rand" + "encoding/hex" "encoding/json" "errors" + "fmt" "io" "log" + "net" "net/http" + "net/url" "path/filepath" + "strconv" "strings" + "sync" "time" "hamlab/internal/files" "hamlab/internal/media" + "hamlab/internal/store" ) -const playLimit = 3 // re-encodes at once on one machine; a GPU does two or three, a CPU one +// playRuns is how many plays may be encoding at once on one machine; a fourth ends the one +// asked of least lately. The A380 makes about two 8K-sourced streams in real time. +const playRuns = 3 -type playCaps struct { - Codecs []playCodec `json:"codecs"` - Qualities []media.Quality `json:"qualities"` - Error string `json:"error,omitempty"` +type playReq struct { + Path string `json:"path,omitempty"` + URL string `json:"url,omitempty"` // a file on another machine, as the hub serves it + Name string `json:"name,omitempty"` + Device string `json:"device,omitempty"` } -type playCodec struct { - media.Codec - Engine string `json:"engine,omitempty"` // "" when nothing here encodes it +type playOut struct { + Node string `json:"node"` // the machine the play is on: "" for the hub; filled in by the hub + Play *media.Play `json:"play"` + Master string `json:"master"` // the playlist, under that machine's /api + Engine string `json:"engine"` + Host string `json:"host"` // that machine's name, for the menu + Note string `json:"note,omitempty"` + Status media.PlayStatus `json:"status"` } -// playCapsHandler: GET /api/files/play-caps[?sudo=1] -func (s *Server) playCapsHandler(w http.ResponseWriter, r *http.Request) { +// playDevices: GET /api/files/play-devices — what this machine could make the player's H.264 +// on, probed as the daemon's own user (a play from another machine never runs as root). +func (s *Server) playDevices(w http.ResponseWriter, r *http.Request) { + caps, err := s.thumbs.capsFor(r.Context(), false, false) + if err != nil { + caps.Error = err.Error() + } + jsonOK(w, 200, map[string]any{"host": s.hostName(), "devices": media.Devices(caps)}) +} + +// playStart: POST /api/files/play[?sudo=1] — open a play of a file here, or of a URL the hub +// serves a file on another machine at. +func (s *Server) playStart(w http.ResponseWriter, r *http.Request) { + var body playReq + if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body); err != nil { + jsonErr(w, 400, "bad request") + return + } 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} + out, code, err := s.openPlay(r.Context(), body, root) if err != nil { - pc.Error = err.Error() - } else if caps.Error != "" { - pc.Error = caps.Error + jsonErr(w, code, err.Error()) + return } - 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) + jsonOK(w, 200, out) } -// 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 +func (s *Server) openPlay(ctx context.Context, body playReq, root bool) (playOut, int, error) { + if root && s.fsSudo == nil && body.URL == "" { + return playOut{}, 403, errors.New("root access needs passwordless sudo on the host") } - 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 + in, name := body.Path, body.Name + if body.URL != "" { + u, err := url.Parse(body.URL) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") || body.Path != "" { + return playOut{}, 400, errors.New("a play is of a path or of an http URL") + } + in, root = body.URL, false + } else { + p, err := files.Clean(body.Path) + if err != nil { + return playOut{}, 400, err + } + if media.Kind(p) != "video" { + return playOut{}, 415, errors.New("not a video") + } + fsys := s.fsLocal + if root { + fsys = s.fsSudo + } + if _, err := fsys.Stat(ctx, p); err != nil { + return playOut{}, 404, err + } + in = p + if name == "" { + name = filepath.Base(p) + } + } + caps, err := s.thumbs.capsFor(ctx, root, false) + if err == nil && caps.Error != "" { + err = errors.New(caps.Error) } - p, err := files.Clean(body.Path) if err != nil { - jsonErr(w, 400, err.Error()) - return + return playOut{}, 503, err } - if media.Kind(p) != "video" { - jsonErr(w, 415, "not a video") - return - } - if err := media.ValidChoice(body.Codec, body.Quality); err != nil { - jsonErr(w, 400, err.Error()) - 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) + ictx, cancel := context.WithTimeout(ctx, 30*time.Second) + src, err := media.Inspect(ictx, caps, in, root) cancel() if err != nil { - playErr(w, err) - return + var de *media.DecodeError + if errors.As(err, &de) { + return playOut{}, 422, errors.New(de.Msg) + } + return playOut{}, 504, err } - st, err := s.plays.Open(r.Context(), caps, root, p, body.Codec, body.Quality, src) + p, err := s.plays.Open(caps, root, in, name, src, body.Device) 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 strings.HasPrefix(err.Error(), "unknown ") || strings.HasPrefix(err.Error(), "nothing on this machine encodes"): - jsonErr(w, 400, err.Error()) - case errors.Is(err, context.DeadlineExceeded): - jsonErr(w, 504, "the file took too long to read") - default: - jsonErr(w, 500, err.Error()) + return playOut{}, 422, err } + log.Printf("files: play %s on %s (%d qualities, %s)", name, p.Engine(), len(p.Rungs), map[bool]string{true: "from another machine", false: "here"}[p.Remote]) + return playOut{Play: p, Master: "/api/files/play/" + p.ID + "/master.m3u8", Engine: p.Engine(), Host: s.hostName(), Status: p.Status()}, 200, nil } // playStatus: GET /api/files/play/{id} func (s *Server) playStatus(w http.ResponseWriter, r *http.Request) { - st, ok := s.plays.Find(r.PathValue("id")) + p, ok := s.plays.Find(r.PathValue("id")) if !ok { - jsonErr(w, 404, "no such stream") + jsonErr(w, 404, "no such play") return } - jsonOK(w, 200, st.Status()) + jsonOK(w, 200, p.Status()) } // playStop: DELETE /api/files/play/{id}, or POST …/close — the form a page can send on its @@ -145,49 +166,365 @@ func (s *Server) playStop(w http.ResponseWriter, r *http.Request) { w.WriteHeader(204) } -// playFile: GET /api/files/play/{id}/{file} — the playlist, the init segment, a segment. -// A seekable stream's playlist is written whole, up front; a segment of it that isn't on disk -// yet is waited for, or re-encoded from there if the encode isn't heading that way (a seek). -func (s *Server) playFile(w http.ResponseWriter, r *http.Request) { - st, ok := s.plays.Find(r.PathValue("id")) +// playPlaylist: GET /api/files/play/{id}/{file} — master.m3u8, or one quality's .m3u8. +func (s *Server) playPlaylist(w http.ResponseWriter, r *http.Request) { + p, ok := s.plays.Find(r.PathValue("id")) if !ok { - jsonErr(w, 404, "no such stream") + jsonErr(w, 404, "no such play") return } - name := r.PathValue("file") - p, ok := st.File(name) - if !ok { - jsonErr(w, 404, "not a stream file") - return - } - w.Header().Set("Cache-Control", "no-store") - w.Header().Set("X-Content-Type-Options", "nosniff") - if st.Seekable { - if name == "index.m3u8" { - w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") - _, _ = w.Write(st.Playlist()) + f := r.PathValue("file") + var b []byte + if f == "master.m3u8" { + b = p.Master() + } else if name, isList := strings.CutSuffix(f, ".m3u8"); isList { + if b, ok = p.Media(name); !ok { + jsonErr(w, 404, "no such quality") return } - if i, isSeg := media.SegmentIndex(name); isSeg { - sp, err := st.Segment(r.Context(), i) - switch { - case errors.Is(err, media.ErrNotInFilm): - jsonErr(w, 404, err.Error()) - return - case r.Context().Err() != nil: - return // the player moved on - case err != nil: - jsonErr(w, 503, err.Error()) - return + } else { + jsonErr(w, 404, "not a play file") + return + } + w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") + w.Header().Set("Cache-Control", "no-store") + _, _ = w.Write(b) +} + +// playPiece: GET /api/files/play/{id}/{rung}/{seg} — a piece, made first if need be. +func (s *Server) playPiece(w http.ResponseWriter, r *http.Request) { + p, ok := s.plays.Find(r.PathValue("id")) + if !ok { + jsonErr(w, 404, "no such play") + return + } + ns, isTS := strings.CutSuffix(r.PathValue("seg"), ".ts") + n, err := strconv.Atoi(ns) + if !isTS || err != nil || strconv.Itoa(n) != ns { + jsonErr(w, 404, "not a piece") + return + } + file, err := p.Segment(r.Context().Done(), r.PathValue("rung"), n) + switch { + case errors.Is(err, media.ErrNotFound): + jsonErr(w, 404, err.Error()) + return + case r.Context().Err() != nil: + return // the player moved on + case err != nil: + jsonErr(w, 503, err.Error()) + return + } + w.Header().Set("Content-Type", "video/mp2t") + w.Header().Set("Cache-Control", "private, max-age=3600") + http.ServeFile(w, r, file) +} + +// ---- on the hub: where plays are made ---- + +// transcodeChoice is the setting; nil is the machine the file is on. +func (s *Server) transcodeChoice() *store.Transcode { + var t *store.Transcode + s.store.View(func(st *store.State) { + if st.Transcode != nil { + c := *st.Transcode + t = &c + } + }) + return t +} + +type transcodeMachine struct { + Node string `json:"node"` // "" is the hub + Host string `json:"host"` + Devices []media.Device `json:"devices"` + Error string `json:"error,omitempty"` // the machine didn't answer +} + +// getTranscode: GET /api/transcode — the setting, and what every machine can encode on, +// asked of each now. +func (s *Server) getTranscode(w http.ResponseWriter, r *http.Request) { + names := []string{""} + if s.nodes != nil { + names = append(names, s.nodes.Names()...) + } + out := make([]transcodeMachine, len(names)) + var wg sync.WaitGroup + for i, n := range names { + wg.Add(1) + go func() { + defer wg.Done() + m := transcodeMachine{Node: n, Host: n} + ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second) + defer cancel() + if n == "" { + caps, err := s.thumbs.capsFor(ctx, false, false) + if err != nil { + caps.Error = err.Error() + } + m.Host, m.Devices = s.hostName(), media.Devices(caps) + } else { + code, b, err := s.nodes.Do(ctx, n, "GET", "/api/files/play-devices", nil, nil) + var d struct { + Host string `json:"host"` + Devices []media.Device `json:"devices"` + } + switch { + case err != nil: + m.Error = err.Error() + case code == 404: + m.Error = "runs an older Ham Lab" + case code != 200 || json.Unmarshal(b, &d) != nil: + m.Error = fmt.Sprintf("answered %d", code) + default: + m.Devices = d.Devices + } } - p = sp + out[i] = m + }() + } + wg.Wait() + jsonOK(w, 200, map[string]any{"selected": s.transcodeChoice(), "machines": out}) +} + +// putTranscode: PUT /api/transcode {node, device}, or {} for the machine the file is on. +func (s *Server) putTranscode(w http.ResponseWriter, r *http.Request) { + var body *store.Transcode + if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil { + jsonErr(w, 400, "expected {\"node\": …, \"device\": …}") + return + } + if body != nil && body.Device == "" { + body = nil + } + if body != nil { + known := body.Node == "" + if s.nodes != nil { + for _, n := range s.nodes.Names() { + known = known || n == body.Node + } + } + if !known { + jsonErr(w, 400, "no such machine") + return + } + if body.Device != media.CPU && !strings.HasPrefix(body.Device, "/dev/dri/") { + jsonErr(w, 400, "a device is a render node or \"cpu\"") + return } } - switch filepath.Ext(p) { - case ".m3u8": - w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") - default: - w.Header().Set("Content-Type", "video/mp4") + if err := s.store.Update(func(st *store.State) { st.Transcode = body }); err != nil { + jsonErr(w, 500, err.Error()) + return } - http.ServeFile(w, r, p) + s.getTranscode(w, r) +} + +// hubPlay: POST /api/play {node, path, sudo} — open a play of a file on node ("" the hub) on +// the device the setting names. When that is on another machine, that machine reads the file +// through this hub, by a token good for that one file. If it can't be reached, the file's own +// machine makes it instead, and the answer says so. +func (s *Server) hubPlay(w http.ResponseWriter, r *http.Request) { + var body struct { + Node string `json:"node"` + Path string `json:"path"` + Sudo bool `json:"sudo"` + } + 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 body.Node != "" && (s.nodes == nil || !has(s.nodes.Names(), body.Node)) { + jsonErr(w, 404, "no such machine") + return + } + choice := s.transcodeChoice() + var note string + if choice != nil && choice.Node != body.Node { + if s.nodes != nil && choice.Node != "" && !has(s.nodes.Names(), choice.Node) { + note = "The transcode machine " + choice.Node + " is no longer managed here; made where the file is." + } else { + src, err := s.srcURL(choice.Node, body.Node, p, body.Sudo) + if err == nil { + req := playReq{URL: src, Name: filepath.Base(p), Device: choice.Device} + out, code, err := s.playOn(r.Context(), choice.Node, req, false) + if err == nil { + jsonOK(w, 200, out) + return + } + if code == 422 || code == 415 { + jsonErr(w, code, err.Error()) + return + } + } + log.Printf("files: play on %s failed (%v); making it where the file is", nodeLabel(choice.Node), err) + note = "Couldn't make it on " + nodeLabel(choice.Node) + " (" + err.Error() + "), so it is made where the file is." + } + } + device := "" + if choice != nil && choice.Node == body.Node { + device = choice.Device + } + out, code, err := s.playOn(r.Context(), body.Node, playReq{Path: p, Device: device}, body.Sudo) + if err != nil { + jsonErr(w, code, err.Error()) + return + } + out.Note = note + jsonOK(w, 200, out) +} + +func nodeLabel(n string) string { + if n == "" { + return "the hub" + } + return n +} + +func has(list []string, x string) bool { + for _, v := range list { + if v == x { + return true + } + } + return false +} + +// playOn opens a play on a machine: here, or through that node's API. +func (s *Server) playOn(ctx context.Context, node string, req playReq, sudo bool) (playOut, int, error) { + if node == "" { + out, code, err := s.openPlay(ctx, req, sudo) + return out, code, err + } + b, _ := json.Marshal(req) + path := "/api/files/play" + if sudo { + path += "?sudo=1" + } + code, resp, err := s.nodes.Do(ctx, node, "POST", path, b, http.Header{"Content-Type": []string{"application/json"}}) + if err != nil { + return playOut{}, 502, err + } + if code != 200 { + var e struct{ Error string } + _ = json.Unmarshal(resp, &e) + if e.Error == "" { + e.Error = fmt.Sprintf("%s answered %d", node, code) + } + return playOut{}, code, errors.New(e.Error) + } + var out playOut + if err := json.Unmarshal(resp, &out); err != nil { + return playOut{}, 502, err + } + out.Node = node + return out, 200, nil +} + +// ---- the file, for the machine transcoding it ---- + +type srcGrant struct { + node, path string + sudo bool + until time.Time +} + +type srcGrants struct { + mu sync.Mutex + m map[string]*srcGrant +} + +const srcLife = 12 * time.Hour // from its last use: a film paused overnight still plays + +var grants = &srcGrants{m: map[string]*srcGrant{}} + +func (g *srcGrants) mint(node, path string, sudo bool) string { + b := make([]byte, 24) + _, _ = rand.Read(b) + tok := hex.EncodeToString(b) + now := time.Now() + g.mu.Lock() + defer g.mu.Unlock() + for k, v := range g.m { + if now.After(v.until) { + delete(g.m, k) + } + } + g.m[tok] = &srcGrant{node: node, path: path, sudo: sudo, until: now.Add(srcLife)} + return tok +} + +func (g *srcGrants) use(tok string) (srcGrant, bool) { + g.mu.Lock() + defer g.mu.Unlock() + v, ok := g.m[tok] + if !ok || time.Now().After(v.until) { + return srcGrant{}, false + } + v.until = time.Now().Add(srcLife) + return *v, true +} + +// srcURL is where the machine `to` can read a file on machine `from` through this hub: this +// hub's own address as `to` reaches it, and a token for that one file. +func (s *Server) srcURL(to, from, path string, sudo bool) (string, error) { + host, port, err := net.SplitHostPort(s.cfg.Listen) + if err != nil { + return "", err + } + ip := net.ParseIP(host) + if host == "" || ip == nil || ip.IsUnspecified() || ip.IsLoopback() { + // Listening everywhere: the address this machine reaches `to` from is one it answers on. + target := "127.0.0.1:1" + if to != "" { + if u, err := url.Parse(s.nodes.URL(to)); err == nil && u.Host != "" { + target = u.Host + } + } + c, err := net.Dial("udp", target) + if err != nil { + return "", err + } + host = c.LocalAddr().(*net.UDPAddr).IP.String() + c.Close() + } + scheme := "http" + if s.fingerprint != "" { + scheme = "https" + } + return scheme + "://" + net.JoinHostPort(host, port) + "/api/play-src/" + grants.mint(from, path, sudo), nil +} + +// playSrc: GET /api/play-src/{token} — the one file a token names, ranges and all, read from +// its machine. The token is the only key: it is minted for a machine transcoding that file. +func (s *Server) playSrc(w http.ResponseWriter, r *http.Request) { + g, ok := grants.use(r.PathValue("token")) + if !ok { + jsonErr(w, 404, "no such file") + return + } + q := url.Values{"path": {g.path}} + if g.sudo { + q.Set("sudo", "1") + } + r2 := r.Clone(r.Context()) + r2.URL.RawQuery = q.Encode() + r2.Header.Del("Cookie") + if g.node == "" { + r2.URL.Path = "/api/files/read" + s.filesRead(w, r2) + return + } + p, ok := s.nodes.Proxy(g.node) + if !ok { + jsonErr(w, 404, "no such machine") + return + } + r2.URL.Path = "/api/nodes/" + g.node + "/files/read" + p.ServeHTTP(w, r2) } diff --git a/internal/api/play_test.go b/internal/api/play_test.go new file mode 100644 index 0000000..ad5e740 --- /dev/null +++ b/internal/api/play_test.go @@ -0,0 +1,51 @@ +package api + +import ( + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "hamlab/internal/config" + "hamlab/internal/files" +) + +// The file a machine transcodes from another is read through the hub by a token naming that +// one file: ranges pass through, the query can't point it anywhere else, and an unknown token +// is nothing. +func TestPlaySrc(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "film.mp4") + if err := os.WriteFile(p, []byte("0123456789"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "other.mp4"), []byte("secret"), 0o644); err != nil { + t.Fatal(err) + } + s := &Server{fsLocal: files.Local{}, cfg: &config.Config{Listen: "10.0.0.234:8100"}} + u, err := s.srcURL("", "", p, false) + if err != nil || !strings.HasPrefix(u, "http://10.0.0.234:8100/api/play-src/") { + t.Fatalf("url %q %v", u, err) + } + tok := strings.TrimPrefix(u, "http://10.0.0.234:8100/api/play-src/") + get := func(tok, query, rng string) *httptest.ResponseRecorder { + r := httptest.NewRequest("GET", "/api/play-src/"+tok+query, nil) + r.SetPathValue("token", tok) + if rng != "" { + r.Header.Set("Range", rng) + } + w := httptest.NewRecorder() + s.playSrc(w, r) + return w + } + if w := get(tok, "", "bytes=2-4"); w.Code != 206 || w.Body.String() != "234" { + t.Fatalf("range: %d %q", w.Code, w.Body) + } + if w := get(tok, "?path="+filepath.Join(dir, "other.mp4"), ""); w.Body.String() != "0123456789" { + t.Fatalf("the query moved it: %q", w.Body) + } + if w := get("nope", "", ""); w.Code != 404 { + t.Fatalf("unknown token: %d", w.Code) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 6176060..dab1a94 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -79,7 +79,7 @@ type Server struct { fsLocal files.FS fsSudo files.FS // nil without passwordless sudo thumbs *thumbSvc - plays *media.Streamer + plays *media.Plays self string // this binary, for helpers run as tasks nodes *nodes.Manager fingerprint string @@ -103,7 +103,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.plays = media.NewPlays(cfg.DataDir(), self, playRuns) 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 @@ -279,12 +279,17 @@ 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("GET /api/files/play-devices", s.auth(s.playDevices)) 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("POST /api/files/play/{id}/close", s.auth(s.playStop)) - m.HandleFunc("GET /api/files/play/{id}/{file}", s.auth(s.playFile)) + m.HandleFunc("GET /api/files/play/{id}/{file}", s.auth(s.playPlaylist)) + m.HandleFunc("GET /api/files/play/{id}/{rung}/{seg}", s.auth(s.playPiece)) + m.HandleFunc("POST /api/play", s.auth(s.hubPlay)) + m.HandleFunc("GET /api/transcode", s.auth(s.getTranscode)) + m.HandleFunc("PUT /api/transcode", s.auth(s.putTranscode)) + m.HandleFunc("GET /api/play-src/{token}", s.playSrc) // the token is the key 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)) diff --git a/internal/media/encode.go b/internal/media/encode.go index 5883abd..886f5c9 100644 --- a/internal/media/encode.go +++ b/internal/media/encode.go @@ -1,280 +1,305 @@ 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. +// What to run when a video is played in the browser: the qualities it is offered in (the +// ladder), the ways a machine could make one of them on the device asked for, in the order +// worth trying, and the ffmpeg argument list for one piece-making run. All decisions, no +// processes, so it can be tested. Everything is H.264 High and AAC: what every browser plays. import ( "fmt" "math" "sort" "strconv" + "strings" ) -// 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"` +// Rung is one quality the player can pick: a size, a frame rate and a bit rate. +type Rung struct { + Name string `json:"name"` // in URLs: the width, "1920" + Label string `json:"label"` // what the menu says: "1080s", "720p" + Width int `json:"width"` + Height int `json:"height"` + FPS float64 `json:"fps"` + Bitrate int `json:"bitrate"` // video, bits a second, on average + Peak int `json:"peak"` // what the playlist promises, sound included + level string // H.264 level, "5.2" + codecs string // for the master playlist } -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 +// One step of the ladder: the long side for a 360 video (its width), the short side for a flat +// one, and the bit rate (Mb/s ×10) at more than 30 frames a second and at 30 or fewer. +type step struct { + size int + label string + hi, lo int + cap30 bool } -var Qualities = []Quality{{"original", "Original size", 0}, {"1080p", "1080p", 1080}, {"720p", "720p", 720}, {"480p", "480p", 480}} +// A 360 picture is seen a quarter of its width at a time, so it needs more pixels than a flat +// one for the same sharpness; YouTube names its sizes by the flat picture they look like +// ("1080s"), and so does this. +var vrSteps = []step{ + {3840, "2160s", 220, 160, false}, + {2880, "1440s", 120, 90, false}, + {1920, "1080s", 60, 45, false}, + {1280, "720s", 25, 25, true}, +} -func qualityByID(id string) (Quality, bool) { - for _, q := range Qualities { - if q.ID == id { - return q, true +var flatSteps = []step{ + {2160, "2160p", 160, 120, false}, + {1440, "1440p", 90, 65, false}, + {1080, "1080p", 50, 35, false}, + {720, "720p", 25, 25, true}, + {480, "480p", 12, 12, true}, +} + +const audioRate = 160_000 + +func even(x float64) int { return int(math.Round(x/2)) * 2 } + +// Ladder is the qualities a video is offered in, best first: every step no bigger than the +// picture itself, or the picture's own size if it is smaller than all of them. +func Ladder(i Info) []Rung { + v := i.Video + if v == nil || v.Width <= 0 || v.Height <= 0 { + return nil + } + vr := v.Spherical + steps := flatSteps + src := min(v.Width, v.Height) // a flat picture is named by its short side + if vr { + steps, src = vrSteps, v.Width + } + // A picture between two steps (a 2292×960 capture) gets its own size on top, at the bit + // rate of the step above, so the best on offer is never less than the file. + var out []Rung + for k, s := range steps { + if s.size > src { + continue } - } - return Quality{}, false -} - -// ValidChoice says whether a codec and quality are ones on offer — checked before any work. -func ValidChoice(codec, quality string) error { - if !codecKnown(codec) { - return fmt.Errorf("unknown codec %q", codec) - } - if _, ok := qualityByID(quality); !ok { - return fmt.Errorf("unknown quality %q", quality) - } - return nil -} - -func codecKnown(id string) bool { - for _, c := range Codecs { - if c.ID == id { - return true + if len(out) == 0 && s.size < src && k > 0 { // never above the top step: H.264 stops there + own := steps[k-1] + own.size, own.label, own.cap30 = src, strconv.Itoa(src)+map[bool]string{true: "s", false: "p"}[vr], false + out = append(out, rung(i, own)) } + out = append(out, rung(i, s)) } - return false + if len(out) == 0 { + last := steps[len(steps)-1] + last.size, last.label = src, strconv.Itoa(src)+map[bool]string{true: "s", false: "p"}[vr] + out = append(out, rung(i, last)) + } + return out } -// 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 +func rung(i Info, s step) Rung { + v := i.Video + var w, h int switch { - case h >= 2160: - kbps = 20000 - case h >= 1440: - kbps = 12000 - case h >= 1080: - kbps = 8000 - case h >= 720: - kbps = 4000 - default: - kbps = 2000 + case v.Spherical: + w, h = s.size, even(float64(s.size)*float64(v.Height)/float64(v.Width)) + case v.Width >= v.Height: + h, w = s.size, even(float64(s.size)*float64(v.Width)/float64(v.Height)) + default: // standing up + w, h = s.size, even(float64(s.size)*float64(v.Height)/float64(v.Width)) } - if src.Video.FPS >= 50 { - kbps = kbps * 5 / 4 + fps := v.FPS + if fps <= 0 { + fps = 30 } - if codec == "hevc" || codec == "av1" { - kbps = kbps * 6 / 10 + for fps > 60.5 || (s.cap30 && fps > 30.5) { + fps /= 2 } - return w, h, kbps + fps = math.Round(fps*1000) / 1000 + br := s.lo + if fps > 30.5 { + br = s.hi + } + r := Rung{Name: strconv.Itoa(w), Label: s.label, Width: w, Height: h, FPS: fps, Bitrate: br * 100_000} + r.Peak = r.Bitrate*3/2 + audioRate + r.level = level(w, h, fps) + lv, _ := strconv.ParseFloat(r.level, 64) + r.codecs = fmt.Sprintf("avc1.6400%02x", int(math.Round(lv*10))) + if i.Audio != nil { + r.codecs += ",mp4a.40.2" + } + return r } -// 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. +// level is the lowest H.264 level that takes this many macroblocks a frame and a second. +func level(w, h int, fps float64) string { + mbs := ((w + 15) / 16) * ((h + 15) / 16) + rate := float64(mbs) * fps + for _, l := range []struct { + name string + mbps float64 + frame int + }{ + {"3.1", 108000, 3600}, {"3.2", 216000, 5120}, {"4.0", 245760, 8192}, {"4.2", 522240, 8704}, + {"5.0", 589824, 22080}, {"5.1", 983040, 36864}, {"5.2", 2073600, 36864}, + } { + if rate <= l.mbps && mbs <= l.frame { + return l.name + } + } + return "6.2" +} + +// fpsArg is a frame rate as ffmpeg's fps filter takes it, exactly for the NTSC rates. +func fpsArg(f float64) string { + for _, n := range []float64{24, 30, 60, 120} { + if math.Abs(f-n*1000/1001) < 0.01 { + return fmt.Sprintf("%d/1001", int(n*1000)) + } + } + return strings.TrimRight(strings.TrimRight(strconv.FormatFloat(f, 'f', 3, 64), "0"), ".") +} + +// Recipe is one way to make a quality: the engine's name for the record, the device it runs +// on ("cpu" for the software encoder), and the ffmpeg arguments that go before the input, the +// video filter, and the video codec. type Recipe struct { Engine string + Device 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}} +// CPU is the device name for software encoding. +const CPU = "cpu" + +// Device is one thing a machine can make H.264 on, for the Transcode device setting. +type Device struct { + ID string `json:"id"` // a render node, or "cpu" + Name string `json:"name"` // "Intel Arc A380", "CPU (libx264)" + Integrated bool `json:"integrated,omitempty"` + OK bool `json:"ok"` + Reason string `json:"reason,omitempty"` // when not OK: why +} + +// Devices lists what this machine could make the player's H.264 on: every GPU, usable or not +// (and why not), then the CPU. +func Devices(c Caps) []Device { + var out []Device + if c.Error != "" { + return []Device{{ID: CPU, Name: "CPU", Reason: c.Error}} } - 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 + for _, g := range c.GPUs { + d := Device{ID: g.Device, Name: g.Name, Integrated: g.Integrated} + if d.Name == "" { + d.Name = g.Device } - 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") + switch { + case g.API != "vaapi": + d.Reason = "only VAAPI devices encode here" + case !has(g.Encodes, "h264"): + d.Reason = "doesn't encode H.264" + default: + d.OK = true } + out = append(out, d) + } + for _, g := range c.Unusable { 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, Device{ID: g.Device, Name: name, Integrated: g.Integrated, Reason: g.Reason}) + } + d := Device{ID: CPU, Name: "CPU (libx264)", OK: has(c.Encoders, "libx264")} + if !d.OK { + d.Reason = "this ffmpeg has no libx264" + } + return append(out, d) +} + +// recipes are the ways to make quality r of src on device, best first: on a GPU, decoding on +// the same card (frames never leave it) before decoding in software and uploading; the CPU +// last, as the fallback whatever was asked. device "" is any: a card on the bus before the +// integrated GPU. An empty list means nothing here can. +func recipes(c Caps, device string, src Info, r Rung) []Recipe { + var out []Recipe + gpus := append([]GPU(nil), c.GPUs...) + sort.SliceStable(gpus, func(i, j int) bool { return !gpus[i].Integrated && gpus[j].Integrated }) + kbps := r.Bitrate / 1000 + rate := []string{"-b:v", strconv.Itoa(kbps) + "k", "-maxrate", strconv.Itoa(kbps*3/2) + "k", "-bufsize", strconv.Itoa(kbps*2) + "k", + "-profile:v", "high", "-level", r.level} + fps := "" + if src.Video == nil || src.Video.FPS <= 0 || math.Abs(src.Video.FPS-r.FPS) > 0.01 { + fps = "fps=" + fpsArg(r.FPS) + "," + } + if device != CPU { + for i := range gpus { + g := &gpus[i] + if g.API != "vaapi" || !has(g.Encodes, "h264") || (device != "" && g.Device != device) { + continue } - out = append(out, Recipe{Engine: name + " (vaapi)", Pre: []string{"-hwaccel", "vaapi", "-hwaccel_device", g.Device, "-hwaccel_output_format", "vaapi"}, VF: vf, Video: video}) + video := append(append([]string{"-c:v", "h264_vaapi"}, rate...), "-bf", "0") + name := g.Name + if name == "" { + name = g.Device + } + scale := fmt.Sprintf("scale_vaapi=w=%d:h=%d:format=nv12", r.Width, r.Height) + if src.Video != nil && has(g.Codecs, src.Video.Codec) { + vf := fps + scale + if src.HDR() { + vf = fps + "tonemap_vaapi=format=nv12:t=bt709:p=bt709:m=bt709," + scale + } + out = append(out, Recipe{Engine: name, Device: g.Device, Pre: []string{"-hwaccel", "vaapi", "-hwaccel_device", g.Device, "-hwaccel_output_format", "vaapi"}, VF: vf, Video: video}) + } + out = append(out, Recipe{Engine: name + ", decoded on the CPU", Device: g.Device, Pre: []string{"-init_hw_device", "vaapi=va:" + g.Device, "-filter_hw_device", "va"}, + VF: fmt.Sprintf("%sformat=nv12,scale=%d:%d,hwupload", fps, r.Width, r.Height), 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...)}) + if has(c.Encoders, "libx264") { + vf := fmt.Sprintf("%sscale=%d:%d,format=yuv420p", fps, r.Width, r.Height) + if src.HDR() { + vf = fmt.Sprintf("%szscale=t=linear:npl=100,format=gbrpf32le,zscale=p=bt709,tonemap=hable,zscale=t=bt709:m=bt709,scale=%d:%d,format=yuv420p", fps, r.Width, r.Height) + } + out = append(out, Recipe{Engine: "CPU (libx264)", Device: CPU, VF: vf, Video: append([]string{"-c:v", "libx264", "-preset", "veryfast", "-sc_threshold", "0"}, rate...)}) } 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 -} +// Seg is the length of a piece, in seconds. +const Seg = 3.0 -// 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 - -// tsBase is added to every run's timestamps. A run from the top would otherwise start just -// below zero (the B-frame delay) and the muxer would shift it to make room, while a run from a -// seek point would not: with the base, no run is ever shifted, and a segment means the same -// stretch of the film whichever run made it. +// tsBase is added to every run's times, so pieces from different runs follow on: without it +// a run from the top would start just below zero (the B-frame and AAC delay) and the muxer +// would shift that run alone. const tsBase = 10 -// segArgs is one run of a seekable re-encode: segments from..to-1 of the fixed 4 s grid. -// Every run is made the same way, the first included, so their segments are interchangeable: -// an accurate seek to from×4 s, timestamps put back where that is in the film (+ tsBase), -// kept as they are in each fragment (frag_discont), numbered from `from`, and an end just -// short of segment `to`, which is already on disk, so the run never writes over it. Segments -// land in dir as seg%05d.m4s; init is the init segment's name (only the first run's is used). -func segArgs(dir, in string, src Info, r Recipe, from, to, n int, init string) []string { - fps := 30.0 - if src.Video != nil && src.Video.FPS > 0 { - fps = src.Video.FPS - } - start := from * hlsSeconds +// pieceArgs is one run making quality r from piece n on, into dir as .ts: an accurate seek +// to n×Seg, a keyframe on every Seg mark and nowhere else (so every piece starts on one, and +// pieces from any run and quality line up), times put back where that is in the film. in is a +// path or, for a file on another machine, an http URL. +func pieceArgs(dir, in string, src Info, rc Recipe, r Rung, n int) []string { + at := strconv.FormatFloat(float64(n)*Seg, 'f', 3, 64) a := []string{"-nostdin", "-hide_banner", "-loglevel", "error", "-y"} - a = append(a, r.Pre...) - a = append(a, "-ss", strconv.Itoa(start), "-i", in, "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn") - if r.VF != "" { - a = append(a, "-vf", r.VF) + a = append(a, rc.Pre...) + if strings.HasPrefix(in, "http://") || strings.HasPrefix(in, "https://") { + a = append(a, "-reconnect", "1", "-reconnect_on_network_error", "1", "-reconnect_delay_max", "4") } - a = append(a, r.Video...) - switch { - case src.Audio == nil: + a = append(a, "-ss", at, "-i", in, "-map", "0:v:0", "-map", "0:a:0?", "-sn", "-dn") + if rc.VF != "" { + a = append(a, "-vf", rc.VF) + } + a = append(a, rc.Video...) + if src.Audio == nil { a = append(a, "-an") - case src.Audio.Codec == "aac": - a = append(a, "-c:a", "copy") - default: + } else { a = append(a, "-c:a", "aac", "-b:a", "160k", "-ac", "2") } - 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)) - if to < n { - a = append(a, "-t", strconv.FormatFloat(float64((to-from)*hlsSeconds)-0.5/fps, 'f', 3, 64)) - } - a = append(a, "-output_ts_offset", strconv.Itoa(start+tsBase), - "-f", "hls", "-hls_time", strconv.Itoa(hlsSeconds), "-hls_playlist_type", "event", "-hls_segment_type", "fmp4", - "-hls_segment_options", "movflags=+frag_discont", "-start_number", strconv.Itoa(from), - "-hls_fmp4_init_filename", init, "-hls_segment_filename", dir+"/seg%05d.m4s", "-hls_flags", "independent_segments+temp_file", - "-progress", dir+"/progress", "-stats_period", "1", dir+"/run.m3u8") + gop := int(math.Ceil(r.FPS*Seg)) * 4 + a = append(a, "-g", strconv.Itoa(gop), "-force_key_frames", fmt.Sprintf("expr:gte(t,n_forced*%g)", Seg), + "-output_ts_offset", strconv.FormatFloat(float64(n)*Seg+tsBase, 'f', 3, 64), + "-f", "hls", "-hls_time", strconv.FormatFloat(Seg-0.5, 'f', 2, 64), "-hls_list_size", "0", + "-hls_flags", "temp_file", "-hls_segment_type", "mpegts", "-start_number", strconv.Itoa(n), + "-hls_segment_filename", dir+"/%d.ts", dir+"/ffmpeg.m3u8") return a } -// segCount is how many 4 s segments a seekable re-encode of a file this long lists. The last -// 0.3 s are left out of the count, so a container whose sound runs on past the picture never -// lists a segment no frame starts. -func segCount(duration float64) int { - d := duration - 0.3 - if d < 0 { - d = 0 - } - return int(d/hlsSeconds) + 1 -} - func has(list []string, x string) bool { for _, v := range list { if v == x { diff --git a/internal/media/encode_test.go b/internal/media/encode_test.go index 408ecd4..901e4c7 100644 --- a/internal/media/encode_test.go +++ b/internal/media/encode_test.go @@ -9,25 +9,40 @@ 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 TestLadder(t *testing.T) { + vr := src("hevc", 7680, 3840, 50) + vr.Video.Spherical = true + r := Ladder(vr) + if len(r) != 4 || r[0].Name != "3840" || r[0].Height != 1920 || r[0].level != "5.2" || r[3].FPS != 25 || r[0].Label != "2160s" { + t.Fatalf("8K 360: %+v", r) + } + if r[0].codecs != "avc1.640034,mp4a.40.2" { + t.Fatalf("codecs %q", r[0].codecs) + } + r = Ladder(src("h264", 2292, 960, 60)) // a capture: its own size on top, then the steps under it + if len(r) != 3 || r[0].Label != "960p" || r[0].Width != 2292 || r[0].FPS != 60 || r[0].Bitrate != 5_000_000 || r[1].Label != "720p" || r[1].Width != 1720 || r[1].FPS != 30 { + t.Fatalf("2292×960: %+v", r) + } + r = Ladder(src("h264", 1920, 1080, 59.94)) + if len(r) != 3 || r[0].Label != "1080p" || r[0].FPS != 59.94 || r[1].FPS != 29.97 || r[2].Width != 854 { + t.Fatalf("1080p: %+v", r) + } + if r = Ladder(src("h264", 1080, 1920, 30)); r[0].Width != 1080 || r[0].Height != 1920 { // standing up + t.Fatalf("portrait: %+v", r) + } + if r = Ladder(src("h264", 320, 240, 30)); len(r) != 1 || r[0].Width != 320 || r[0].Label != "240p" { + t.Fatalf("tiny: %+v", r) + } + silent := src("h264", 1280, 720, 30) + silent.Audio = nil + if r = Ladder(silent); strings.Contains(r[0].codecs, "mp4a") { + t.Fatalf("no sound, no AAC: %q", r[0].codecs) + } + if Ladder(Info{Duration: 10}) != nil { + t.Fatal("no picture, no ladder") + } + if fpsArg(29.97) != "30000/1001" || fpsArg(25) != "25" { + t.Fatal(fpsArg(29.97), fpsArg(25)) } } @@ -47,111 +62,76 @@ func engineNames(r []Recipe) []string { return out } -func TestRecipesOrder(t *testing.T) { +func TestRecipes(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) + in := src("hevc", 3840, 2160, 60) + r1080 := Ladder(in)[2] + // any device: the card first, decoding on it; then the iGPU the same way; then the CPU + got := strings.Join(engineNames(recipes(c, "", in, r1080)), "|") + if got != "Intel Arc A380|Intel Arc A380, decoded on the CPU|Intel UHD Graphics 750|Intel UHD Graphics 750, decoded on the CPU|CPU (libx264)" { + t.Fatalf("any: %s", 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) + // the iGPU asked for: only it, then the CPU as the fallback + if got = strings.Join(engineNames(recipes(c, "/dev/dri/renderD128", in, r1080)), "|"); got != "Intel UHD Graphics 750|Intel UHD Graphics 750, decoded on the CPU|CPU (libx264)" { + t.Fatalf("iGPU: %s", 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) + if got = strings.Join(engineNames(recipes(c, CPU, in, r1080)), "|"); got != "CPU (libx264)" { + t.Fatalf("cpu: %s", got) } - // 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") + // an AV1 source: neither card decodes it, so no same-card recipe + if got = strings.Join(engineNames(recipes(c, "/dev/dri/renderD129", src("av1", 1920, 1080, 30), r1080)), "|"); got != "Intel Arc A380, decoded on the CPU|CPU (libx264)" { + t.Fatalf("av1: %s", got) } - if EngineFor(bare, "av1") != "CPU (SVT-AV1)" || EngineFor(bare, "h264") != "" || EngineFor(bare, "copy") != "Original codec" { - t.Fatal("engineFor") + // a machine with no GPU and no x264: nothing + bare := Caps{FFmpeg: "ffmpeg", FFprobe: "ffprobe", Encoders: []string{"aac"}} + if len(recipes(bare, "", in, r1080)) != 0 { + t.Fatal("no x264 means nothing") } - if EngineFor(c, "hevc") != "Intel Arc A380 (vaapi)" { - t.Fatalf("engineFor on grid: %q", EngineFor(c, "hevc")) - } - // HDR gets tone-mapped on both paths + // 30 fps from 60 for the lower rungs; HDR tone-mapped on both paths + r720 := Ladder(in)[3] 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) + rs := recipes(c, "", hdr, r720) + if !strings.HasPrefix(rs[0].VF, "fps=30,tonemap_vaapi") || !strings.Contains(rs[len(rs)-1].VF, "tonemap=hable") { + t.Fatalf("hdr: %q … %q", rs[0].VF, rs[len(rs)-1].VF) + } + if rs := recipes(c, "", in, r1080); strings.Contains(rs[0].VF, "fps=") { + t.Fatalf("same rate needs no fps filter: %q", rs[0].VF) } } -func TestHLSArgs(t *testing.T) { +func TestDevices(t *testing.T) { + c := gridCaps() + c.Unusable = []GPU{{Device: "/dev/dri/renderD130", Name: "Something", Reason: "permission denied"}} + d := Devices(c) + if len(d) != 4 || !d[0].OK || !d[1].OK || d[2].OK || d[2].Reason == "" || d[3].ID != CPU || !d[3].OK { + t.Fatalf("%+v", d) + } + if d := Devices(Caps{Error: "ffmpeg is not installed"}); len(d) != 1 || d[0].OK { + t.Fatalf("no ffmpeg: %+v", d) + } +} + +func TestPieceArgs(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"} { + r := Ladder(in)[0] + rc := recipes(gridCaps(), "/dev/dri/renderD129", in, r)[0] + a := strings.Join(pieceArgs("/d", "/v.mkv", in, rc, r, 12), " ") + for _, want := range []string{"-hwaccel vaapi -hwaccel_device /dev/dri/renderD129", "-ss 36.000 -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 5000k -maxrate 7500k", "-level 4.2", "-c:a aac -b:a 160k -ac 2", + "-force_key_frames expr:gte(t,n_forced*3)", "-output_ts_offset 46.000", "-hls_time 2.50", "-start_number 12", "-hls_segment_filename /d/%d.ts"} { 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") + if strings.Contains(a, "-reconnect") { + t.Error("a path needs no reconnect") + } + if a := strings.Join(pieceArgs("/d", "http://10.0.0.234:8100/api/play-src/x", in, rc, r, 0), " "); !strings.Contains(a, "-reconnect 1") { + t.Error("a URL reconnects") } 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 TestSegArgs(t *testing.T) { - in := src("h264", 1920, 1080, 25) - r := recipes(gridCaps(), "h264", in, 1920, 1080, 10000)[0] - first := strings.Join(segArgs("/d", "/v.mkv", in, r, 0, 900, 900, "init.mp4"), " ") - for _, want := range []string{"-ss 0 -i /v.mkv", "-output_ts_offset 10", "-start_number 0", "movflags=+frag_discont", "-hls_fmp4_init_filename init.mp4", "-hls_segment_filename /d/seg%05d.m4s", "/d/run.m3u8", "-force_key_frames expr:gte(t,n_forced*4)"} { - if !strings.Contains(first, want) { - t.Errorf("first run: missing %q in\n%s", want, first) - } - } - if strings.Contains(first, " -t ") { - t.Error("a run to the end has no -t") - } - // a seek: segments 30..44, stopping half a frame short of 45, which is on disk - seek := strings.Join(segArgs("/d", "/v.mkv", in, r, 30, 45, 900, "init-run.mp4"), " ") - for _, want := range []string{"-ss 120 -i /v.mkv", "-t 59.980", "-output_ts_offset 130", "-start_number 30", "-hls_fmp4_init_filename init-run.mp4"} { - if !strings.Contains(seek, want) { - t.Errorf("seek run: missing %q in\n%s", want, seek) - } - } - if strings.Index(seek, "-ss 120") > strings.Index(seek, "-i /v.mkv") || strings.Index(seek, "-hwaccel") > strings.Index(seek, "-i /v.mkv") { - t.Error("-ss and the decoder's options go before -i") - } - for d, n := range map[float64]int{9: 3, 8.2: 2, 8.31: 3, 64: 16, 0.1: 1, 3600: 900} { - if got := segCount(d); got != n { - t.Errorf("segCount(%v) = %d, want %d", d, got, n) - } - } -} - -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) - } -} - -func TestValidChoice(t *testing.T) { - if err := ValidChoice("h264", "720p"); err != nil { - t.Fatal(err) - } - if ValidChoice("vp8", "720p") == nil || ValidChoice("h264", "9p") == nil || ValidChoice("", "") == nil { - t.Fatal("bad choices must be refused") + if a := strings.Join(pieceArgs("/d", "/v", in, rc, r, 0), " "); !strings.Contains(a, "-an") { + t.Error("no sound means -an") } } diff --git a/internal/media/ffrun.go b/internal/media/ffrun.go index 754a31b..6564ae7 100644 --- a/internal/media/ffrun.go +++ b/internal/media/ffrun.go @@ -6,13 +6,13 @@ package media // transcode never outlives the console that asked for it. import ( + "bufio" "bytes" "context" "errors" "io" "os" "os/exec" - "strings" "sync" "syscall" "time" @@ -92,6 +92,35 @@ func (p *ffproc) end() { } } +// pause stops ffmpeg where it is (SIGSTOP), and resume lets it go on: a stream far enough +// ahead of its player waits rather than making the rest of the film. A root run is paused by +// its helper, which is told on its standard input. +func (p *ffproc) pause() error { return p.signal(syscall.SIGSTOP, "stop\n") } +func (p *ffproc) resume() error { return p.signal(syscall.SIGCONT, "cont\n") } + +func (p *ffproc) signal(sig syscall.Signal, word string) error { + if done, _ := p.exited(); done { + return errors.New("ended") + } + if p.in != nil { + _, err := io.WriteString(p.in, word) + return err + } + return syscall.Kill(-p.cmd.Process.Pid, sig) +} + +// kill ends ffmpeg now, without waiting: a paused one too. +func (p *ffproc) kill() { + if done, _ := p.exited(); done { + return + } + if p.in != nil { + p.in.Close() // the helper ends it, waking it first if it is paused + return + } + _ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL) +} + // lastWords is what ffmpeg said last, for the record. func (p *ffproc) lastWords() string { return p.tail.last() } @@ -141,7 +170,17 @@ func RunFFmpegHelper(args []string, stdin io.Reader, stderr io.Writer) int { } ctx, cancel := context.WithCancel(context.Background()) go func() { - _, _ = io.Copy(io.Discard, stdin) // returns when the daemon lets go + // "stop" and "cont" pause and resume it; the end of input (the daemon letting go) + // ends it. + sc := bufio.NewScanner(stdin) + for sc.Scan() { + switch sc.Text() { + case "stop": + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGSTOP) + case "cont": + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGCONT) + } + } cancel() }() done := make(chan error, 1) @@ -151,6 +190,7 @@ func RunFFmpegHelper(args []string, stdin io.Reader, stderr io.Writer) int { case werr = <-done: case <-ctx.Done(): _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM) + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGCONT) // a paused one can't act on TERM select { case werr = <-done: case <-time.After(5 * time.Second): @@ -181,8 +221,3 @@ func clearDir(dir string) error { } return nil } - -func hasSegments(playlist string) bool { - b, err := os.ReadFile(playlist) - return err == nil && strings.Contains(string(b), "#EXTINF") -} diff --git a/internal/media/hls.go b/internal/media/hls.go deleted file mode 100644 index b88651c..0000000 --- a/internal/media/hls.go +++ /dev/null @@ -1,669 +0,0 @@ -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 -// /transcode/. -// -// A re-encode is seekable anywhere from the start. Its playlist is written up front, for -// the whole film: a fixed grid of 4 s segments, so the player's timeline is the film's. -// ffmpeg runs ahead from the top as fast as it can; a request for a segment it hasn't -// reached and won't reach soon — a seek ahead, or back into a stretch never made — ends that -// run and starts another at the segment asked for, which stops on its own where segments -// already exist. When a run finishes, the next gap is filled, until the whole file is on -// disk. Every run is made the same way (segArgs), so segments from different runs join. -// -// "Repackaged" (copy) can't cut on a fixed grid — its keyframes are the source's — so it is -// an event playlist that grows as ffmpeg writes it, seekable within what's done. Repackaging -// runs at disk speed, so that is all of it within seconds. -// -// 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" - "bytes" - "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"` - Seekable bool `json:"seekable"` // seeking anywhere restarts the re-encode there - - key string - dir string - mu sync.Mutex - used time.Time - proc *ffproc // the run going now (or the only one, for copy) - finished bool - failure string - gone bool // forgotten by the Streamer - - // seekable streams only - caps Caps - self string - recipe Recipe - n int // segments in the playlist - end int // segments past this are not in the file after all (n until a run finds out) - run *segRun - runMu sync.Mutex // one run change at a time; held across ending one run and starting the next - asked time.Time // when the request that last moved the run arrived: an older one never moves it back - initOK bool // a later run's init segment was checked against the first's - initBad bool -} - -// segRun is one ffmpeg over segments from..to-1. -type segRun struct { - from, to int - proc *ffproc - stopped bool // ended by us: what it wrote after that is not to be trusted -} - -// 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 (copy), or on disk (seekable) - Speed float64 `json:"speed"` // × realtime - Segments int `json:"segments"` - From float64 `json:"from"` // seekable: where the run going now started, seconds -} - -// 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 - -// segmentWait is how long a request for a segment waits for a run to write it. -const segmentWait = 45 * time.Second // under the hub proxy's 60 s wait for a node to answer - -// lookahead: a segment this many past what the run has written is waited for, not -// restarted at. A few seconds of film, which any engine here writes faster than that. -const lookahead = 3 - -var streamFile = regexp.MustCompile(`^(index\.m3u8|init\.mp4|seg\d{5}\.m4s)$`) -var segName = regexp.MustCompile(`^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 err := ValidChoice(codec, quality); err != nil { - return nil, err - } - q, _ := qualityByID(quality) - if src.Video == nil { - return nil, &DecodeError{Msg: "no video stream"} - } - if src.Video.Width <= 0 || src.Video.Height <= 0 { - return nil, &DecodeError{Msg: "the picture's size is unknown"} - } - 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 { - if s.failed() { // not the stream to hand back: let it go and make a fresh one - delete(m.streams, s.ID) - _ = os.RemoveAll(s.dir) - break - } - 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(), - caps: c, self: m.self} - s.dir = filepath.Join(m.dir, s.ID) - // A fixed grid needs the film's length; without one it can only grow as it goes. - s.Seekable = codec != "copy" && src.Duration > 0 - if s.Seekable { - s.n = segCount(src.Duration) - s.end = s.n - } - 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) - var args []string - if s.Seekable { - args = segArgs(s.dir, path, src, r, 0, s.n, s.n, "init.mp4") - } else { - 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, s.recipe = p, r.Engine, r - if s.Seekable { - s.run = &segRun{from: 0, to: s.n, proc: p} - } - s.mu.Unlock() - msg, ok := s.waitFirstSegment(ctx) - if ok { - if s.Seekable { - go s.watchRun(s.run) - } else { - 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 - } - p.end() - if ctx.Err() != nil || s.closed() { // the caller left, or somebody closed it meanwhile: not the recipe's fault - m.forget(s) - return nil, errors.New("cancelled") - } - last = r.Engine + ": " + msg - log.Printf("media: stream %s on %s: %s", filepath.Base(path), r.Engine, msg) - } - m.forget(s) - return nil, &DecodeError{Msg: last} -} - -// waitFirstSegment watches for the 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 s.Seekable && exists(s.segPath(0)) || !s.Seekable && 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 (copy streams: one run, start to end). -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) - } -} - -// ---- seekable streams ---- - -func (s *Stream) segPath(i int) string { return filepath.Join(s.dir, fmt.Sprintf("seg%05d.m4s", i)) } - -// written is the last segment of a run on disk, counting on from where it started. -func (s *Stream) written(r *segRun) int { - i := r.from - for i < r.to && exists(s.segPath(i)) { - i++ - } - return i - 1 -} - -// watchRun follows one run to its end. A run that ends by itself — the end of the film, or -// the segments it was to stop short of — is followed by the next gap. One that failed fails -// the stream; one we ended is the business of whoever ended it. -func (s *Stream) watchRun(r *segRun) { - <-r.proc.done - s.mu.Lock() - if s.gone || r.stopped || s.run != r { - s.mu.Unlock() - return - } - if r.proc.err != nil { - s.failure = r.proc.lastWords() - if s.failure == "" { - s.failure = r.proc.err.Error() - } - s.finished = true - s.mu.Unlock() - log.Printf("media: stream %s: run from segment %d failed: %s", s.ID, r.from, s.failure) - return - } - if w := s.written(r); w+1 < r.to && r.to == s.n { - // ran to the end of the file before the end of the grid: the picture is shorter than - // the container says, and those last segments will never exist - s.end = w + 1 - log.Printf("media: stream %s: the film ends at segment %d of %d", s.ID, s.end, s.n) - } - s.checkInit(r) - s.mu.Unlock() - s.fillNext(r.to) -} - -// checkInit compares a later run's init segment with the first run's, once. They come out -// identical (same encoder, same settings); if one day they don't, the log says so. -func (s *Stream) checkInit(r *segRun) { - if r.from == 0 || s.initOK { - return - } - a, err1 := os.ReadFile(filepath.Join(s.dir, "init.mp4")) - b, err2 := os.ReadFile(filepath.Join(s.dir, "init-run.mp4")) - if err1 != nil || err2 != nil { - return - } - s.initOK = true - if !bytes.Equal(a, b) && !s.initBad { - s.initBad = true - log.Printf("media: stream %s: a restarted run's init segment differs from the first (%s); its segments may not join", s.ID, s.Engine) - } -} - -// missingFrom is the first segment not on disk at or after i (wrapping round to the start), -// or -1 when the whole film is there. -func (s *Stream) missingFrom(i int) int { - end := s.end - for k := 0; k < end; k++ { - j := (i + k) % end - if !exists(s.segPath(j)) { - return j - } - } - return -1 -} - -// fillNext starts a run at the next gap after `after`; when there is none, the stream is done. -func (s *Stream) fillNext(after int) { - s.runMu.Lock() - defer s.runMu.Unlock() - s.mu.Lock() - if s.gone || s.failure != "" || (s.run != nil && !s.run.done()) { - s.mu.Unlock() - return - } - end := s.end - s.mu.Unlock() - if after >= end { - after = 0 - } - next := s.missingFrom(after) - if next < 0 { - s.mu.Lock() - s.finished, s.proc = true, nil - s.mu.Unlock() - log.Printf("media: stream %s: all %d segments on disk", s.ID, end) - return - } - s.startRun(next) -} - -func (r *segRun) done() bool { d, _ := r.proc.exited(); return d } - -// startRun ends the run going now and starts one at segment from, stopping short of the next -// segment already on disk. The run it ends is asked to stop and waited for, and whatever it -// finished writing after being asked is removed: ffmpeg closes the segment it was in the -// middle of on its way out, short. Called with runMu held, mu not. -func (s *Stream) startRun(from int) { - s.mu.Lock() - old := s.run - end := s.end - s.mu.Unlock() - if old != nil && !old.done() { - had := map[int]bool{} - for i := old.from; i < old.to; i++ { - had[i] = exists(s.segPath(i)) - } - s.mu.Lock() - old.stopped = true - s.mu.Unlock() - old.proc.end() - for i := old.from; i < old.to; i++ { - if !had[i] && exists(s.segPath(i)) { - _ = os.Remove(s.segPath(i)) - } - } - } - to := from + 1 - for to < end && !exists(s.segPath(to)) { - to++ - } - s.mu.Lock() - if s.gone { - s.mu.Unlock() - return - } - s.mu.Unlock() - init := "init-run.mp4" - if !exists(filepath.Join(s.dir, "init.mp4")) { - init = "init.mp4" - } - p, err := startFF(s.caps, s.self, s.Root, segArgs(s.dir, s.Path, s.Source, s.recipe, from, to, s.n, init)) - s.mu.Lock() - defer s.mu.Unlock() - if err != nil { - s.failure, s.finished = "restarting at "+strconv.Itoa(from*hlsSeconds)+" s: "+err.Error(), true - return - } - r := &segRun{from: from, to: to, proc: p} - s.run, s.proc, s.finished = r, p, false - go s.watchRun(r) - log.Printf("media: stream %s: run from %d s to %d s", s.ID, from*hlsSeconds, to*hlsSeconds) -} - -// Playlist is the whole film's, written before most of it exists. -func (s *Stream) Playlist() []byte { - var b strings.Builder - b.WriteString("#EXTM3U\n#EXT-X-VERSION:7\n") - fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", hlsSeconds) - b.WriteString("#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-INDEPENDENT-SEGMENTS\n#EXT-X-MAP:URI=\"init.mp4\"\n") - for i := 0; i < s.n; i++ { - d := float64(hlsSeconds) - if i == s.n-1 { - d = s.Source.Duration - float64(i*hlsSeconds) - } - fmt.Fprintf(&b, "#EXTINF:%.6f,\nseg%05d.m4s\n", d, i) - } - b.WriteString("#EXT-X-ENDLIST\n") - return []byte(b.String()) -} - -// ErrNotInFilm is a segment the playlist lists but the film turned out not to reach. -var ErrNotInFilm = errors.New("past the end of the film") - -// Segment is the path of segment i once it is on disk. When the run going now will write it -// soon, it waits; otherwise this is a seek, and the run moves there — unless a newer request -// has moved it since this one arrived. -func (s *Stream) Segment(ctx context.Context, i int) (string, error) { - asked := time.Now() - deadline := asked.Add(segmentWait) - p := s.segPath(i) - for { - s.touch() - if exists(p) { - return p, nil - } - s.mu.Lock() - switch { - case s.gone: - s.mu.Unlock() - return "", errors.New("the stream was closed") - case i >= s.end: - s.mu.Unlock() - return "", ErrNotInFilm - case s.failure != "": - f := s.failure - s.mu.Unlock() - return "", errors.New(f) - } - r := s.run - coming := r != nil && !r.done() && r.from <= i && i < r.to && i <= s.written(r)+lookahead - move := !coming && !asked.Before(s.asked) - if move { - s.asked = asked - } - s.mu.Unlock() - if move { - s.runMu.Lock() - // still the newest, and still not on its way? (a run may have moved for a - // newer request while this one waited for the lock) - s.mu.Lock() - r = s.run - coming = r != nil && !r.done() && r.from <= i && i < r.to && i <= s.written(r)+lookahead - newest := !asked.Before(s.asked) - s.mu.Unlock() - if !coming && newest && !exists(p) { - log.Printf("media: stream %s: seek to %d s", s.ID, i*hlsSeconds) - s.startRun(i) - } - s.runMu.Unlock() - continue - } - if time.Now().After(deadline) { - return "", fmt.Errorf("segment %d didn't arrive in %s", i, segmentWait) - } - select { - case <-ctx.Done(): - return "", ctx.Err() - case <-time.After(150 * time.Millisecond): - } - } -} - -// SegmentIndex reads the number out of a segment's name. -func SegmentIndex(name string) (int, bool) { - m := segName.FindStringSubmatch(name) - if m == nil { - return 0, false - } - i, err := strconv.Atoi(m[1]) - return i, err == nil -} - -// ---- both kinds ---- - -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() - s.gone = true // a run ending now is ours, not a gap to fill - p := s.proc - if s.run != nil { - s.run.stopped = true - } - 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() - s.mu.Lock() - s.gone = true - s.mu.Unlock() - _ = os.RemoveAll(s.dir) -} - -func (s *Stream) closed() bool { s.mu.Lock(); defer s.mu.Unlock(); return s.gone } - -func (s *Stream) failed() bool { - s.mu.Lock() - defer s.mu.Unlock() - return s.finished && s.failure != "" -} - -// 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 (copy streams; a -// seekable one's is 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 what is on disk. -func (s *Stream) Status() StreamStatus { - s.mu.Lock() - st := StreamStatus{Running: s.proc != nil && !s.finished, Done: s.finished && s.failure == "", Error: s.failure} - seekable, end, run := s.Seekable, s.end, s.run - s.mu.Unlock() - b, err := os.ReadFile(filepath.Join(s.dir, "progress")) - if err == nil { - st.Progress, st.Speed = parseProgress(b) - } - if !seekable { - if b, err := os.ReadFile(filepath.Join(s.dir, "index.m3u8")); err == nil { - st.Segments = strings.Count(string(b), "#EXTINF") - } - return st - } - for i := 0; i < end; i++ { - if exists(s.segPath(i)) { - st.Segments++ - } - } - st.Progress = float64(st.Segments * hlsSeconds) - if st.Progress > s.Source.Duration { - st.Progress = s.Source.Duration - } - if run != nil && st.Running { - st.From = float64(run.from * hlsSeconds) - } - 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) -} diff --git a/internal/media/hls_test.go b/internal/media/hls_test.go deleted file mode 100644 index d2c803f..0000000 --- a/internal/media/hls_test.go +++ /dev/null @@ -1,220 +0,0 @@ -package media - -import ( - "context" - "errors" - "math" - "os" - "os/exec" - "path/filepath" - "sort" - "strconv" - "strings" - "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 == "" || !s.Seekable { - t.Fatalf("stream: %+v", s) - } - if _, ok := s.File("index.m3u8"); !ok { - t.Fatal("no playlist") - } - if pl := string(s.Playlist()); !strings.Contains(pl, "#EXT-X-PLAYLIST-TYPE:VOD") || !strings.HasSuffix(pl, "#EXT-X-ENDLIST\n") || strings.Count(pl, "#EXTINF") != 3 { - t.Fatalf("playlist for 9 s:\n%s", pl) - } - if !exists(s.segPath(0)) { - 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 != 3 || 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) - } -} - -// A seek ahead of the encode: the run moves there, the segment comes from a run that started -// at it, the gaps are filled afterwards, and the segments of the different runs, joined, are -// one film with no hole and nothing twice beyond a frame at a join. -func TestSeekRestartsTheEncode(t *testing.T) { - if _, err := exec.LookPath("ffmpeg"); err != nil { - t.Skip("no ffmpeg") - } - c := Probe(context.Background()) - if c.Error != "" || EngineFor(c, "h264") == "" { - t.Skip("nothing here encodes h264") - } - dir := t.TempDir() - clip := filepath.Join(dir, "clip.mkv") - // big and busy enough that the first run is still near the start when the seek comes - out, err := exec.Command("ffmpeg", "-nostdin", "-v", "error", "-f", "lavfi", "-i", "testsrc2=size=1920x1080:rate=30000/1001:duration=64", "-f", "lavfi", "-i", "sine=frequency=440:duration=64", - "-c:v", "libx264", "-preset", "ultrafast", "-g", "250", "-c:a", "libopus", "-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 { - t.Fatal(err) - } - m := NewStreamer(dir, "hamlab", 2) - s, err := m.Open(context.Background(), c, false, clip, "h264", "1080p", src) - if err != nil { - t.Fatal(err) - } - defer m.Close(s.ID) - if s.n != 16 { - t.Fatalf("%d segments for 64 s", s.n) - } - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - p, err := s.Segment(ctx, 12) - if err != nil { - t.Fatal(err) - } - s.mu.Lock() - from := s.run.from - s.mu.Unlock() - if from != 12 { - t.Fatalf("the run after the seek starts at segment %d, not 12 (the first run was already there?)", from) - } - if got := firstVideoPTS(t, filepath.Join(s.dir, "init.mp4"), p); math.Abs(got-(48+tsBase)) > 0.05 { - t.Fatalf("segment 12 starts at %.3f, want %d", got, 48+tsBase) - } - if _, err := s.Segment(ctx, 16); !errors.Is(err, ErrNotInFilm) { - t.Fatalf("past the end: %v", err) - } - var st StreamStatus - for i := 0; i < 480; i++ { - if st = s.Status(); st.Done || st.Error != "" { - break - } - time.Sleep(250 * time.Millisecond) - } - if !st.Done || st.Segments != 16 { - t.Fatalf("not filled in: %+v", st) - } - // the joined film: every segment in order behind the one init segment - joined := filepath.Join(dir, "joined.mp4") - f, _ := os.Create(joined) - for _, n := range append([]string{"init.mp4"}, segNames(16)...) { - b, err := os.ReadFile(filepath.Join(s.dir, n)) - if err != nil { - t.Fatal(err) - } - _, _ = f.Write(b) - } - f.Close() - if out, err := exec.Command("ffmpeg", "-v", "error", "-i", joined, "-f", "null", "-").CombinedOutput(); err != nil || strings.Contains(string(out), "error") { - t.Fatalf("the joined film doesn't decode cleanly: %v %s", err, out) - } - pts := videoPTS(t, joined) - if len(pts) < 1900 || pts[0] > tsBase+0.1 || pts[len(pts)-1] < tsBase+63.5 { - t.Fatalf("%d frames, %.3f .. %.3f", len(pts), pts[0], pts[len(pts)-1]) - } - for i := 1; i < len(pts); i++ { - if d := pts[i] - pts[i-1]; d > 0.07 { // a frame is 0.033; a join may lose up to one - t.Fatalf("a hole of %.3f s at %.3f", d, pts[i-1]) - } - } -} - -func segNames(n int) []string { - var out []string - for i := 0; i < n; i++ { - out = append(out, "seg"+strconv.FormatInt(int64(100000+i), 10)[1:]+".m4s") - } - return out -} - -func videoPTS(t *testing.T, file string) []float64 { - out, err := exec.Command("ffprobe", "-v", "error", "-select_streams", "v", "-show_entries", "packet=pts_time", "-of", "csv=p=0", file).Output() - if err != nil { - t.Fatal(err) - } - var pts []float64 - seen := map[string]bool{} - for _, l := range strings.Fields(string(out)) { - if seen[l] { // the same frame from both runs at a join - continue - } - seen[l] = true - if v, err := strconv.ParseFloat(l, 64); err == nil { - pts = append(pts, v) - } - } - sort.Float64s(pts) - return pts -} - -func firstVideoPTS(t *testing.T, init, seg string) float64 { - dir := t.TempDir() - j := filepath.Join(dir, "one.mp4") - a, _ := os.ReadFile(init) - b, _ := os.ReadFile(seg) - _ = os.WriteFile(j, append(a, b...), 0o644) - pts := videoPTS(t, j) - if len(pts) == 0 { - t.Fatal("no frames in the segment") - } - return pts[0] -} diff --git a/internal/media/inspect.go b/internal/media/inspect.go index 1e601b0..ab722f5 100644 --- a/internal/media/inspect.go +++ b/internal/media/inspect.go @@ -25,6 +25,7 @@ type Track struct { ColorTransfer string `json:"color_transfer,omitempty"` // smpte2084 / arib-std-b67 mean HDR Channels int `json:"channels,omitempty"` AttachedPic bool `json:"attached_pic,omitempty"` + Spherical bool `json:"spherical,omitempty"` // an equirectangular 360 picture } // Info is what a file holds: the picture to take, the sound to go with it, and the whole. @@ -58,6 +59,10 @@ type probeJSON struct { Disposition struct { AttachedPic int `json:"attached_pic"` } `json:"disposition"` + SideData []struct { + Type string `json:"side_data_type"` + Projection string `json:"projection"` + } `json:"side_data_list"` } `json:"streams"` Format struct { FormatName string `json:"format_name"` @@ -74,7 +79,7 @@ func Inspect(ctx context.Context, c Caps, path string, root bool) (Info, error) return Info{}, errors.New("ffprobe is not installed") } argv := []string{c.FFprobe, "-v", "error", "-show_entries", - "stream=index,codec_type,codec_name,profile,pix_fmt,width,height,r_frame_rate,avg_frame_rate,color_transfer,channels,disposition:format=format_name,duration,bit_rate,size", + "stream=index,codec_type,codec_name,profile,pix_fmt,width,height,r_frame_rate,avg_frame_rate,color_transfer,channels,disposition:stream_side_data=side_data_type,projection:format=format_name,duration,bit_rate,size", "-of", "json", path} if root { argv = priv.Wrap(argv...) @@ -105,6 +110,15 @@ func Inspect(ctx context.Context, c Caps, path string, root bool) (Info, error) for _, s := range pj.Streams { 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} + // A 360 video says so in its spherical tag (ffmpeg reads the old XML one and the sv3d + // box); an untagged one is still taken for 360 when it is 2:1 and too big for anything + // else. + for _, sd := range s.SideData { + if sd.Type == "Spherical Mapping" && strings.HasPrefix(sd.Projection, "equirectangular") { + st.Spherical = true + } + } + st.Spherical = st.Spherical || (s.CodecType == "video" && s.Width == 2*s.Height && s.Width >= 3840) st.FPS = fps(s.AvgFrameRate) if st.FPS == 0 { st.FPS = fps(s.RFrameRate) diff --git a/internal/media/play.go b/internal/media/play.go new file mode 100644 index 0000000..afa29a5 --- /dev/null +++ b/internal/media/play.go @@ -0,0 +1,462 @@ +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 /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 + } + } +} diff --git a/internal/media/play_test.go b/internal/media/play_test.go new file mode 100644 index 0000000..33c662a --- /dev/null +++ b/internal/media/play_test.go @@ -0,0 +1,129 @@ +package media + +import ( + "context" + "encoding/json" + "math" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "testing" + "time" +) + +// inspectPiece answers when a piece's first video packet is shown, whether it is a keyframe, +// how many keyframes the piece has, and when its last packet is. +func inspectPiece(t *testing.T, path string) (first float64, key bool, keys int, last float64) { + t.Helper() + b, err := exec.Command("ffprobe", "-v", "error", "-show_entries", "packet=pts_time,flags,codec_type", "-of", "json", path).Output() + if err != nil { + t.Fatalf("ffprobe %s: %v", path, err) + } + var p struct { + Packets []struct { + PTS string `json:"pts_time"` + Flags string `json:"flags"` + Type string `json:"codec_type"` + } `json:"packets"` + } + _ = json.Unmarshal(b, &p) + first = -1 + for _, k := range p.Packets { + if k.Type != "video" { + continue + } + x, _ := strconv.ParseFloat(k.PTS, 64) + if first < 0 { + first, key = x, strings.HasPrefix(k.Flags, "K") + } + first = math.Min(first, x) + last = math.Max(last, x) + if strings.HasPrefix(k.Flags, "K") { + keys++ + } + } + return +} + +// TestPieces makes pieces of a made-up video (or SRC, on DEVICE, to try the real thing on a +// machine with a GPU) in order and after a seek, at two qualities, and checks they line up: +// each starts on a keyframe at its own place in the film, with no other keyframe inside it. +func TestPieces(t *testing.T) { + if _, err := exec.LookPath("ffmpeg"); err != nil { + t.Skip("no ffmpeg") + } + dir := t.TempDir() + in := os.Getenv("SRC") + if in == "" { + in = filepath.Join(dir, "in.mkv") + if out, err := exec.Command("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc2=size=1920x1080:rate=50:duration=40", + "-f", "lavfi", "-i", "sine=frequency=440:duration=40", "-c:v", "libx264", "-preset", "ultrafast", "-g", "500", + "-c:a", "libopus", "-shortest", in).CombinedOutput(); err != nil { + t.Fatalf("making the test video: %v: %s", err, out) + } + } + caps := Probe(context.Background()) + if caps.Error != "" { + t.Skip(caps.Error) + } + info, err := Inspect(context.Background(), caps, in, false) + if err != nil { + t.Fatal(err) + } + m := NewPlays(dir, "", 2) + device := os.Getenv("DEVICE") + if device == "" { + device = CPU + } + p, err := m.Open(caps, false, in, "in.mkv", info, device) + if err != nil { + t.Fatal(err) + } + defer m.Close(p.ID) + if !strings.Contains(string(p.Master()), "RESOLUTION=") { + t.Fatal(string(p.Master())) + } + low := p.Rungs[len(p.Rungs)-1].Name + var offset float64 + far := min(p.segs-2, 60) + for i, n := range []int{0, 1, 2, far, far + 1, 3} { + t0 := time.Now() + path, err := p.Segment(nil, low, n) + if err != nil { + t.Fatalf("piece %d: %v", n, err) + } + first, key, keys, last := inspectPiece(t, path) + if i == 0 { + offset = first + } + at := first - offset + t.Logf("piece %d: %d ms on %s, starts %.3f, %d keyframe(s), %.2f s long", n, time.Since(t0).Milliseconds(), p.Status().Engine, at, keys, last-first) + if !key || keys != 1 { + t.Errorf("piece %d: first packet keyframe %v, %d keyframes", n, key, keys) + } + if math.Abs(at-float64(n)*Seg) > 0.05 { + t.Errorf("piece %d starts at %.3f, want %.3f", n, at, float64(n)*Seg) + } + if n < p.segs-1 && math.Abs(last-first-Seg) > 0.1 { + t.Errorf("piece %d is %.3f s long", n, last-first) + } + } + // another quality, at a piece already made at the first: its own run, lined up too + if len(p.Rungs) > 1 { + path, err := p.Segment(nil, p.Rungs[0].Name, 2) + if err != nil { + t.Fatal(err) + } + if first, key, _, _ := inspectPiece(t, path); !key || math.Abs(first-offset-2*Seg) > 0.05 { + t.Errorf("%s piece 2 starts %.3f (keyframe %v)", p.Rungs[0].Name, first-offset, key) + } + } + if _, err := p.Segment(nil, low, p.segs); err != ErrNotFound { + t.Errorf("past the end: %v", err) + } + if _, err := p.Segment(nil, "999", 0); err != ErrNotFound { + t.Errorf("no such quality: %v", err) + } +} diff --git a/internal/nodes/nodes.go b/internal/nodes/nodes.go index 2944129..c054b86 100644 --- a/internal/nodes/nodes.go +++ b/internal/nodes/nodes.go @@ -404,6 +404,12 @@ func (m *Manager) Do(ctx context.Context, name, method, path string, body []byte return resp.StatusCode, b, err } +// URL is a managed machine's address, "" for one not managed here. +func (m *Manager) URL(name string) string { + n, _ := m.node(name) + return n.URL +} + // Names lists the managed machines. func (m *Manager) Names() []string { var out []string diff --git a/internal/store/store.go b/internal/store/store.go index 8ee55c5..c283f53 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -75,6 +75,16 @@ type State struct { KVMCookies map[string]map[string]string `json:"kvm_cookies,omitempty"` SamplerMs int `json:"sampler_ms,omitempty"` // how often this machine samples; the hub keeps every machine on the console's refresh rate HistorySec int `json:"history_sec,omitempty"` // how far back the graphs reach; the hub keeps every machine on the same one + // Transcode is where videos played in the Files viewer are made (internal/api/play.go): + // a machine and a device on it. nil is the machine the file is on, its best device. + Transcode *Transcode `json:"transcode,omitempty"` +} + +// Transcode names a device on a machine: Node "" is the hub itself; Device is a render node +// or "cpu". +type Transcode struct { + Node string `json:"node"` + Device string `json:"device"` } // UserState returns the prefs of an account, making an empty one if needed. diff --git a/web/src/Files.tsx b/web/src/Files.tsx index 1f358de..80f0027 100644 --- a/web/src/Files.tsx +++ b/web/src/Files.tsx @@ -267,7 +267,7 @@ function Viewer({ path, sudo, onClose, onChanged }: { path: string; sudo: boolea