mitchell/hamlab
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 14 Wiki Activity

files: seek anywhere in a re-encode; it restarts from there

Browse Source 
A re-encode's playlist is now the whole film on a fixed 4 s grid. A segment
that isn't on disk and isn't coming soon moves the encode there; runs stop
short of segments already made, and gaps are filled after. Every run is cut
the same way (-ss, -output_ts_offset start+10, frag_discont), so segments
from different runs join. Repackaging keeps its growing playlist.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Mitchell-Johnson
2026-10-20 13:08:52 -07:00
co-authored by Claude Opus 5.5
parent ff1373e838
commit 450a9d3327
8 changed files with 660 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/media/hls.go
+356 -31
View File
@@ -3,14 +3,26 @@ package media
// Playing what the browser cannot play as it is: an HLS stream, re-encoded on demand by
// ffmpeg on the machine's own GPU where it has one it can use, on its CPU otherwise. A
// stream is one file at one codec and size, in a folder of its own under
// <data dir>/transcode/: ffmpeg writes an event playlist and fragmented-MP4 segments there
// as fast as it can, and the player reads them as they land, so the range it can seek in
// grows until the whole file is through — on a GPU, several times faster than watching.
// Nothing is kept: a stream nobody has touched for a while is ended and its folder
// removed, and the folder is emptied when the daemon starts.
// <data dir>/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"
@@ -28,26 +40,46 @@ import (
// 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"`
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
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.
@@ -55,9 +87,10 @@ type StreamStatus struct {
Running bool `json:"running"`
Done bool `json:"done"`
Error string `json:"error,omitempty"`
Progress float64 `json:"progress"` // seconds of output written
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.
@@ -72,7 +105,15 @@ type Streamer struct {
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.
@@ -128,8 +169,15 @@ func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, qua
m.mu.Unlock()
return nil, fmt.Errorf("%d videos are already being re-encoded on this machine; one must finish or be closed first", running)
}
s := &Stream{ID: newID(), Path: path, Codec: codec, Quality: quality, Source: src, Started: time.Now(), Root: root, key: key, used: time.Now()}
s := &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()
@@ -149,18 +197,30 @@ func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, qua
return nil, err
}
_ = clearDir(s.dir)
args := hlsArgs(s.dir, path, src, r, codec == "copy")
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 = p, r.Engine
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 {
go s.watch()
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
}
@@ -176,12 +236,12 @@ func (m *Streamer) Open(ctx context.Context, c Caps, root bool, path, codec, qua
return nil, &DecodeError{Msg: last}
}
// waitFirstSegment watches for the playlist's first segment, or ffmpeg giving up.
// 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 hasSegments(playlist) {
if s.Seekable && exists(s.segPath(0)) || !s.Seekable && hasSegments(playlist) {
return "", true
}
if done, err := s.proc.exited(); done {
@@ -205,7 +265,7 @@ func (s *Stream) waitFirstSegment(ctx context.Context) (string, bool) {
}
}
// watch records how the encode ended.
// watch records how the encode ended (copy streams: one run, start to end).
func (s *Stream) watch() {
<-s.proc.done
s.mu.Lock()
@@ -222,6 +282,250 @@ func (s *Stream) watch() {
}
}
// ---- 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()
@@ -238,7 +542,11 @@ func (m *Streamer) Close(id string) {
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()
@@ -282,8 +590,8 @@ func (m *Streamer) Sweep(idle time.Duration) {
}
}
// File is the on-disk path of one of the stream's files — the playlist, the init segment
// or a media segment — and nothing else.
// 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
@@ -300,16 +608,33 @@ func (s *Stream) running() bool {
return s.proc != nil && !s.finished
}
// Status is read from ffmpeg's own progress file and the playlist.
// 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()
if b, err := os.ReadFile(filepath.Join(s.dir, "progress")); err == nil {
b, err := os.ReadFile(filepath.Join(s.dir, "progress"))
if err == nil {
st.Progress, st.Speed = parseProgress(b)
}
if b, err := os.ReadFile(filepath.Join(s.dir, "index.m3u8")); err == nil {
st.Segments = strings.Count(string(b), "#EXTINF")
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
}