Add the transcode session manager
Second step of ADR-008. A session spawns ffmpeg with the detected encoder's args, packages the file as HLS, and is reused while it is alive so two range requests do not start two ffmpegs. Three things it has to get right, and all three are about not burning the GPU: a session nobody is watching is reaped after a few minutes idle; concurrent sessions are capped rather than queued, because a GPU encodes one or two streams and a queue would just make everyone wait longer; and Shutdown stops everything so a restart does not leave ffmpeg running. Found a real bug in review, and it is the kind that only a real host catches. The hardware args were placed BEFORE -i. ffmpeg reads options before the input as INPUT options, so "-c:v h264_nvenc" was looked up as a decoder and failed with "Unknown decoder 'h264_nvenc'". Transcoding would have produced nothing on any NVENC host while looking correct in the code. Verified against the real hardware rather than reasoned about: the same ffmpeg command produces zero segments with the args before -i and a valid playlist and segments with them after. The test now asserts the ordering directly and names the reason, so a future reordering fails with an explanation instead of two arg lists. Red-green: moving them back fails with "the hardware args must follow the input". Verified end-to-end on this host: the Manager started a session against a real 20-second file and produced a 132-byte EVENT playlist and segments. Known limit: a stopped session leaves its segment directory on disk, so the cache directory needs pruning out of band.
This commit is contained in:
mithral
parent
549b503fdf
commit
365aac0cc6
3 changed files with 449 additions and 0 deletions
255
internal/transcode/session.go
Normal file
255
internal/transcode/session.go
Normal file
|
|
@ -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
|
||||
}
|
||||
190
internal/transcode/session_test.go
Normal file
190
internal/transcode/session_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
@ -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 (
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue