diff --git a/internal/transcode/session.go b/internal/transcode/session.go new file mode 100644 index 0000000..87d5f0b --- /dev/null +++ b/internal/transcode/session.go @@ -0,0 +1,255 @@ +package transcode + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "sync" + "time" +) + +// Errors Start returns. They are sentinel values so callers can tell a missing +// capability from a full server from a preference to retry later. +var ( + // ErrUnavailable means no hardware encoder passed detection's probe. + ErrUnavailable = errors.New("transcode: no hardware encoder available") + // ErrAtCapacity means the concurrent-session cap is reached. The request + // is refused rather than queued: a queued transcode looks like a hang. + ErrAtCapacity = errors.New("transcode: concurrent session cap reached") + // ErrShutdown means the manager has been shut down. + ErrShutdown = errors.New("transcode: manager shut down") +) + +const ( + // maxSessions caps concurrent transcodes. A GPU encodes one or two + // streams, not ten. + maxSessions = 2 + // idleTimeout is how long a session may go untouched before its encoder + // is stopped, so an abandoned stream does not pin the GPU forever. + idleTimeout = 3 * time.Minute + // reapEvery is how often the idle reaper runs. + reapEvery = 30 * time.Second +) + +// Session is one running transcode: ffmpeg writing an HLS rendition of a single +// item into its own directory. It is reused across requests for the same item, +// so a returned Session may have other holders. Close releases the caller's +// hold; the encoder is stopped once no holder remains, or when the session goes +// idle. +type Session interface { + // PlaylistPath is the HLS playlist ffmpeg writes and a player fetches. + PlaylistPath() string + // Dir is the directory holding the playlist and its segments. + Dir() string + // Close releases this caller's hold on the session. Returns nil if it was + // already released. + Close() error +} + +// Manager starts, reuses, reaps and caps transcode sessions. +// +// The same item gets one session while it is alive, so the two range requests a +// player makes for one file do not start two encoders. A session nobody has +// touched for idleTimeout is stopped. At most max sessions run at once; a +// request past the cap is an error, never a queue. +type Manager struct { + // hw, newCmd and now are the impure edges, injected so a test can run a + // session with no hardware and no ffmpeg, the same way detect injects its + // probe. + hw func() Hardware + newCmd func(ctx context.Context, args []string) *exec.Cmd + now func() time.Time + + // max caps concurrent sessions; idle is how long an untouched session is + // kept before it is reaped. + max int + idle time.Duration + + mu sync.Mutex + sessions map[int64]*session + seq int64 + closed bool + + stop chan struct{} + stopOnce sync.Once + wg sync.WaitGroup +} + +// New builds a Manager that uses the host's detected capability and the real +// ffmpeg binary, and starts its idle reaper. +func New() *Manager { + m := &Manager{ + hw: Detect, + newCmd: newFFmpeg, + now: time.Now, + max: maxSessions, + idle: idleTimeout, + sessions: map[int64]*session{}, + stop: make(chan struct{}), + } + m.wg.Add(1) + go m.reaper() + return m +} + +// newFFmpeg is the production command builder. +func newFFmpeg(ctx context.Context, args []string) *exec.Cmd { + return exec.CommandContext(ctx, "ffmpeg", args...) +} + +// Start spawns ffmpeg to package inputPath as HLS under cacheDir/transcode/ and +// returns the session. Asking for the same item again while its session is alive +// reuses it. It returns ErrUnavailable, ErrAtCapacity or ErrShutdown instead of +// queueing. +func (m *Manager) Start(itemID int64, inputPath, cacheDir string) (Session, error) { + hw := m.hw() + if !hw.Available() { + return nil, fmt.Errorf("%w: no VAAPI, NVENC or QSV encoder passed its probe", ErrUnavailable) + } + + m.mu.Lock() + defer m.mu.Unlock() + + if m.closed { + return nil, ErrShutdown + } + if s, ok := m.sessions[itemID]; ok { + s.refs++ + s.last = m.now() + return s, nil + } + if len(m.sessions) >= m.max { + return nil, fmt.Errorf("%w (%d running)", ErrAtCapacity, m.max) + } + + m.seq++ + dir := filepath.Join(cacheDir, "transcode", strconv.FormatInt(itemID, 10)+"-"+strconv.FormatInt(m.seq, 10)) + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("transcode: create session dir: %w", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cmd := m.newCmd(ctx, sessionArgs(hw, inputPath, dir)) + if err := cmd.Start(); err != nil { + cancel() + _ = os.RemoveAll(dir) + return nil, fmt.Errorf("transcode: start ffmpeg: %w", err) + } + // Reap the child when it exits; without a Wait it would linger as a zombie. + go func() { _ = cmd.Wait() }() + + s := &session{m: m, itemID: itemID, dir: dir, cancel: cancel, refs: 1, last: m.now()} + m.sessions[itemID] = s + return s, nil +} + +// sessionArgs is the ffmpeg command line: the encoder's hardware arguments, +// then the input, then AAC audio and event HLS packaging into dir. +func sessionArgs(hw Hardware, inputPath, dir string) []string { + // The hardware arguments go AFTER -i. Placed before it, ffmpeg reads them as + // INPUT options and "-c:v h264_nvenc" is looked up as a decoder, failing + // with "Unknown decoder 'h264_nvenc'". Verified against a real NVENC host: + // before -i produces no segments at all, after -i produces them. + args := make([]string, 0, len(hw.Args)+14) + args = append(args, "-i", inputPath) + args = append(args, hw.Args...) + return append(args, + "-c:a", "aac", "-b:a", "128k", + "-f", "hls", + "-hls_time", "6", + "-hls_playlist_type", "event", + "-hls_segment_filename", filepath.Join(dir, "seg%05d.ts"), + filepath.Join(dir, "playlist.m3u8"), + ) +} + +// Shutdown stops every session and the idle reaper. Call it once at process +// exit; it is safe to call more than once. +func (m *Manager) Shutdown() { + m.stopOnce.Do(func() { close(m.stop) }) + m.wg.Wait() + + m.mu.Lock() + defer m.mu.Unlock() + m.closed = true + for _, s := range m.sessions { + m.stopLocked(s) + } +} + +// reap stops every session untouched for at least idle. It is separate from the +// ticker so a test can drive it directly. +func (m *Manager) reap(now time.Time) { + m.mu.Lock() + defer m.mu.Unlock() + for _, s := range m.sessions { + if now.Sub(s.last) >= m.idle { + m.stopLocked(s) + } + } +} + +func (m *Manager) reaper() { + defer m.wg.Done() + t := time.NewTicker(reapEvery) + defer t.Stop() + for { + select { + case <-m.stop: + return + case <-t.C: + m.reap(m.now()) + } + } +} + +// stopLocked kills a session's encoder and drops it from the manager. Callers +// must hold m.mu. +func (m *Manager) stopLocked(s *session) { + if s.stopped { + return + } + s.stopped = true + s.cancel() + if cur, ok := m.sessions[s.itemID]; ok && cur == s { + delete(m.sessions, s.itemID) + } +} + +// session is Manager's concrete Session. +// +// shortcut: a stopped session leaves its segment directory on disk; the cache +// directory is expected to be pruned out of band. Add eviction here if that +// cache is not otherwise bounded. +type session struct { + m *Manager + itemID int64 + dir string + cancel context.CancelFunc + refs int + last time.Time + // stopped guards against double Close and against Close racing the reaper. + stopped bool +} + +func (s *session) Dir() string { return s.dir } +func (s *session) PlaylistPath() string { return filepath.Join(s.dir, "playlist.m3u8") } + +// Close releases one hold. The encoder is stopped once the last hold is gone. +func (s *session) Close() error { + s.m.mu.Lock() + defer s.m.mu.Unlock() + if s.stopped { + return nil + } + s.refs-- + if s.refs > 0 { + return nil + } + s.m.stopLocked(s) + return nil +} diff --git a/internal/transcode/session_test.go b/internal/transcode/session_test.go new file mode 100644 index 0000000..a1022c3 --- /dev/null +++ b/internal/transcode/session_test.go @@ -0,0 +1,190 @@ +package transcode + +import ( + "context" + "errors" + "os/exec" + "path/filepath" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" +) + +// testManager builds a Manager that never touches real hardware or ffmpeg: the +// capability and the command are injected, the same way detect injects its +// probe. The stub command blocks until its context is cancelled, standing in +// for ffmpeg. +func testManager(t *testing.T) (*Manager, func() int) { + t.Helper() + var spawns atomic.Int32 + m := New() + m.hw = func() Hardware { + return Hardware{Encoder: EncoderVAAPI, Args: []string{"-vaapi_device", "/dev/dri/renderD128", "-c:v", "h264_vaapi"}} + } + m.newCmd = func(ctx context.Context, args []string) *exec.Cmd { + spawns.Add(1) + return exec.CommandContext(ctx, "sleep", "3600") + } + t.Cleanup(m.Shutdown) + return m, func() int { return int(spawns.Load()) } +} + +// TestStartAndReuse is the reason Start exists: two requests for the same item +// must not start two encoders. +func TestStartAndReuse(t *testing.T) { + m, spawns := testManager(t) + cacheDir := t.TempDir() + + a, err := m.Start(7, "/media/movie.mkv", cacheDir) + if err != nil { + t.Fatalf("Start: %v", err) + } + b, err := m.Start(7, "/media/movie.mkv", cacheDir) + if err != nil { + t.Fatalf("Start (reuse): %v", err) + } + if a != b { + t.Errorf("second Start returned a different session: %v then %v", a.Dir(), b.Dir()) + } + if got := spawns(); got != 1 { + t.Errorf("spawned %d ffmpegs, want 1", got) + } + if want := filepath.Join(a.Dir(), "playlist.m3u8"); a.PlaylistPath() != want { + t.Errorf("PlaylistPath = %q, want %q", a.PlaylistPath(), want) + } + if !strings.HasPrefix(a.Dir(), filepath.Join(cacheDir, "transcode")) { + t.Errorf("Dir = %q, want it under %q", a.Dir(), filepath.Join(cacheDir, "transcode")) + } +} + +// TestStartArgs pins the ffmpeg command line the task requires. +func TestStartArgs(t *testing.T) { + m := New() + t.Cleanup(m.Shutdown) + m.hw = func() Hardware { + return Hardware{Encoder: EncoderNVENC, Args: []string{"-c:v", "h264_nvenc"}} + } + var got []string + m.newCmd = func(ctx context.Context, args []string) *exec.Cmd { + got = append([]string(nil), args...) + return exec.CommandContext(ctx, "sleep", "3600") + } + + s, err := m.Start(9, "/m/in.mkv", t.TempDir()) + if err != nil { + t.Fatalf("Start: %v", err) + } + // The hardware args must come AFTER -i. Before it, ffmpeg treats them as + // INPUT options and "-c:v h264_nvenc" is looked up as a decoder: on a real + // NVENC host that fails with "Unknown decoder 'h264_nvenc'" and produces no + // segments at all. This ordering is load-bearing, not cosmetic. + want := []string{ + "-i", "/m/in.mkv", + "-c:v", "h264_nvenc", + "-c:a", "aac", "-b:a", "128k", + "-f", "hls", + "-hls_time", "6", + "-hls_playlist_type", "event", + "-hls_segment_filename", filepath.Join(s.Dir(), "seg%05d.ts"), + filepath.Join(s.Dir(), "playlist.m3u8"), + } + if !reflect.DeepEqual(got, want) { + t.Errorf("args = %q, want %q", got, want) + } + + // State the invariant directly as well, so a future reordering fails with a + // message that explains why rather than only showing two arg lists. + in, enc := indexOf(got, "-i"), indexOf(got, "-c:v") + if in < 0 || enc < 0 || enc < in { + t.Errorf("encoder at %d, -i at %d: the hardware args must follow the input", enc, in) + } +} + +func indexOf(args []string, want string) int { + for i, a := range args { + if a == want { + return i + } + } + return -1 +} + +// TestIdleReap proves a session nobody touches is stopped, so an abandoned +// stream does not burn the GPU forever. +func TestIdleReap(t *testing.T) { + m, spawns := testManager(t) + m.idle = time.Minute + cacheDir := t.TempDir() + + a, err := m.Start(1, "/m/in.mkv", cacheDir) + if err != nil { + t.Fatalf("Start: %v", err) + } + + // Not yet idle: the session survives and is still reused. + m.reap(m.now()) + if b, _ := m.Start(1, "/m/in.mkv", cacheDir); b != a { + t.Fatal("session reaped before it was idle") + } + + // Idle: reaped, so the next request starts a fresh encoder. + m.reap(m.now().Add(time.Hour)) + c, err := m.Start(1, "/m/in.mkv", cacheDir) + if err != nil { + t.Fatalf("Start after reap: %v", err) + } + if c == a { + t.Error("idle session was reused after being reaped") + } + if got := spawns(); got != 2 { + t.Errorf("spawned %d ffmpegs, want 2", got) + } +} + +// TestAtCapacity proves a request past the cap is refused, not queued. +func TestAtCapacity(t *testing.T) { + m, _ := testManager(t) + m.max = 1 + cacheDir := t.TempDir() + + if _, err := m.Start(1, "/m/a.mkv", cacheDir); err != nil { + t.Fatalf("Start: %v", err) + } + _, err := m.Start(2, "/m/b.mkv", cacheDir) + if !errors.Is(err, ErrAtCapacity) { + t.Errorf("Start past cap: got %v, want ErrAtCapacity", err) + } +} + +// TestStartNoCapability proves a GPU-less host gets a clear refusal. +func TestStartNoCapability(t *testing.T) { + m := New() + t.Cleanup(m.Shutdown) + m.hw = func() Hardware { return Hardware{} } + + _, err := m.Start(1, "/m/in.mkv", t.TempDir()) + if !errors.Is(err, ErrUnavailable) { + t.Errorf("Start with no capability: got %v, want ErrUnavailable", err) + } +} + +// TestShutdownStopsSessions proves shutdown tears down running encoders and +// refuses new work afterwards. +func TestShutdownStopsSessions(t *testing.T) { + m := New() + m.hw = func() Hardware { return Hardware{Encoder: EncoderNVENC, Args: []string{"-c:v", "h264_nvenc"}} } + m.newCmd = func(ctx context.Context, args []string) *exec.Cmd { + return exec.CommandContext(ctx, "sleep", "3600") + } + if _, err := m.Start(1, "/m/a.mkv", t.TempDir()); err != nil { + t.Fatalf("Start: %v", err) + } + + m.Shutdown() + + if _, err := m.Start(2, "/m/b.mkv", t.TempDir()); !errors.Is(err, ErrShutdown) { + t.Errorf("Start after Shutdown: got %v, want ErrShutdown", err) + } +} diff --git a/internal/transcode/transcode.go b/internal/transcode/transcode.go index ced6cdc..4ef9977 100644 --- a/internal/transcode/transcode.go +++ b/internal/transcode/transcode.go @@ -10,6 +10,10 @@ // // Detection is cached for the process lifetime. It spawns ffmpeg, so it must // not run per request; callers ask Detect and the spawn happens at most once. +// +// Manager, in session.go, turns that capability into running HLS sessions: +// it spawns ffmpeg with the detected encoder's args, reuses a session for the +// same item, reaps idle ones and caps how many run at once. package transcode import (