feat(playback): balance transcode sessions across multiple GPUs (#425)

* feat(playback): balance transcode sessions across multiple GPUs

playback.hw_device now accepts a comma-separated render-device list (e.g.
"/dev/dri/renderD128,/dev/dri/renderD129"). Each transcode session resolves
the list to one concrete device at spawn — the present device with the
fewest active GPU sessions, ties keeping list order — and holds that device
for its whole lifetime (seek/audio restarts reuse it); the reservation
releases on session shutdown, idempotently, including early spawn-failure
paths. Software-accel sessions never reserve, so they cannot skew the
balance.

A single configured value keeps the historical pass-through contract and an
empty value still auto-detects, so existing deployments are unaffected.
PickRenderDevice is list-aware too, picking least-loaded without reserving,
which lets the non-session consumers (chapter thumbnails, download
artifacts, transcode nodes) spread load best-effort when given a list.

Motivation: hosts with two identical media GPUs (e.g. dual Arc A310)
previously pinned every session to one device while the second sat idle.

* feat(admin): GPU device picker for playback hw_device

The hw-accel detection endpoint now reports render_device_details — each
render device with a human label derived from its sysfs PCI vendor/device
ids ("Intel GPU (0x56a6)") — and the Playback settings page renders them as
per-device toggles instead of requiring a hand-typed device path. No
selection means auto (first available device); one selection pins every
session; multiple selections balance least-loaded. The stored
playback.hw_device value stays the comma-separated list, written in stable
detection order regardless of click order, and a configured-but-undetected
device stays visible so a temporarily missing GPU is not silently dropped
on save.

* fix(playback): make GPU selection and reservation atomic

Review follow-up: resolveSessionHWDevice previously selected the
least-loaded device and incremented its count in two separate critical
sections, so concurrent session starts could all pick the same device
before any reservation landed. Device presence checks now happen outside
the lock and selection + reservation share one critical section; a
concurrency test asserts an exact split across two devices for eight
simultaneous starts, which the two-step version cannot guarantee.

* refactor(playback): one typed GPU acquisition boundary, release on process exit

Replace the CSV-handling spread across resolveSessionHWDevice and
PickRenderDevice with HWDeviceSet + AcquireHWDevice in hwdevice.go: every
GPU workload resolves exactly one device immediately before spawn.
Balancing is explicitly QSV/VAAPI-only — NVENC addresses GPUs by CUDA
index/UUID, so a multi-entry list warns and uses the first entry instead
of collapsing through the path-presence filter. Sessions now release
their reservation only after ffmpeg has been reaped (shutdown waits on
done first), closing the window where a new start could pick a device
the old process still occupied. Render-device sysfs descriptions move to
gpudetect.go so the allocator file owns only selection/reservation.

* fix(downloads): prepared downloads acquire a GPU through the shared pool

PrepareFile resolves the configured hw_device list to one concrete
device via AcquireHWDevice and holds the reservation until ffmpeg exits
(Run is synchronous, so the deferred release is the process-exit
boundary). Download encodes now participate in the same active-load
accounting as streaming sessions instead of best-effort spreading.

* fix(chapterthumbs): resolve hw_device list per extraction via the shared pool

ExtractFrame acquires one concrete device from AcquireHWDevice for the
hardware attempt (released when the attempt finishes) instead of passing
the raw comma-separated value to ffmpeg as a single device path. The
service stops pre-resolving and caching a device at first use — the raw
configured value flows through and each extraction resolves it.

* fix(transcodenode): fresh starts use this node's configured hw_device

/transcode/start constructed TranscodeOpts with an empty HWDevice, so
fresh sessions auto-detected the first GPU and bypassed the configured
list while reconstructed sessions honored it. Both paths now feed the
node-local config value into StartTranscode's shared resolution.

* feat(admin): node-aware GPU inventory on /admin/system/hw-accel

playback.hw_device is one cluster-wide value consumed by every transcode
node, but the endpoint probed only whichever healthy node had the fewest
jobs — an admin could configure devices that don't exist on the other
nodes. The endpoint now probes every healthy node concurrently and
returns a nodes array (URL, name, resolved accel, devices, or probe
error) alongside the backward-compatible flat fields, and the config doc
states the homogeneous-path contract explicitly.

* feat(admin): GPU picker survives empty detection, warns on node divergence

The picker rows are now the union of detected devices and configured
entries, so configured-but-missing devices stay visible (and
deselectable) when detection returns nothing or an older node omits
render_device_details (plain render_devices paths fall back to a generic
label). Per-node inventories from the hw-accel endpoint drive two
warnings: a banner when responding nodes report different device sets,
and a per-row note listing nodes missing that device. The multi-select
is hidden for NVENC — balancing is QSV/VA-API only — with a notice when
a multi-device value is already stored.

* style: gofmt touched files

* fix(playback): release GPU reservations on process exit

---------

Co-authored-by: rxwatcher <rxwatcher@users.noreply.github.com>
Co-authored-by: Quick <31828688+Quick104@users.noreply.github.com>
This commit is contained in:
RXWatcher
2026-08-04 11:33:15 -04:00
committed by GitHub
co-authored by rxwatcher Quick
parent 3bdfc58512
commit 567cdd1a15
19 changed files with 1532 additions and 82 deletions
+73 -11
View File
@@ -4,6 +4,7 @@ import (
"context"
"log/slog"
"net/http"
"sync"
"github.com/Silo-Server/silo-server/internal/buildinfo"
"github.com/Silo-Server/silo-server/internal/nodepool"
@@ -28,24 +29,85 @@ func NewSystemHandler(transcodePool *nodepool.TranscodePool, jwtSecret string, f
}
}
// NodeHWAccel reports one transcode node's GPU inventory.
type NodeHWAccel struct {
NodeURL string `json:"node_url"`
NodeName string `json:"node_name,omitempty"`
Resolved string `json:"resolved,omitempty"`
RenderDevices []string `json:"render_devices,omitempty"`
RenderDeviceDetails []playback.RenderDeviceInfo `json:"render_device_details,omitempty"`
Error string `json:"error,omitempty"`
}
// HWAccelInventory is the admin hw-accel response: the primary probe (flat,
// backward compatible with the previous single-probe response) plus every
// healthy node's inventory. playback.hw_device is one cluster-wide value, so
// the UI needs all node inventories to warn when they diverge (devices must
// exist at identical paths on every node for the setting to be safe).
type HWAccelInventory struct {
playback.HWAccelInfo
Nodes []NodeHWAccel `json:"nodes,omitempty"`
}
// HandleHWAccel handles GET /admin/system/hw-accel.
// When transcode nodes are registered it delegates to the first healthy node.
// Otherwise it probes the local host.
// With transcode nodes registered it probes every healthy node (the primary
// flat fields come from the first that responds, preserving the historical
// shape); with none it probes the local host.
func (h *SystemHandler) HandleHWAccel(w http.ResponseWriter, r *http.Request) {
var healthy []*nodepool.Node
if h.transcodePool != nil {
if node := h.transcodePool.Acquire(); node != nil {
info, err := h.fetchRemoteHWAccel(r.Context(), node)
if err == nil {
writeJSON(w, http.StatusOK, info)
return
for _, node := range h.transcodePool.Nodes() {
if node != nil && node.Enabled && node.Healthy {
healthy = append(healthy, node)
}
slog.WarnContext(r.Context(), "hw-accel: remote node probe failed, falling back to local", "component", "api",
"node", node.URL, "error", err)
}
}
if len(healthy) == 0 {
writeJSON(w, http.StatusOK, HWAccelInventory{HWAccelInfo: playback.DetectHWAccelWithFFmpeg(h.ffmpegPath)})
return
}
info := playback.DetectHWAccelWithFFmpeg(h.ffmpegPath)
writeJSON(w, http.StatusOK, info)
inventory := HWAccelInventory{Nodes: make([]NodeHWAccel, len(healthy))}
infos := make([]playback.HWAccelInfo, len(healthy))
errs := make([]error, len(healthy))
var wg sync.WaitGroup
for i, node := range healthy {
wg.Add(1)
go func() {
defer wg.Done()
infos[i], errs[i] = h.fetchRemoteHWAccel(r.Context(), node)
}()
}
wg.Wait()
for i, node := range healthy {
entry := NodeHWAccel{NodeURL: node.URL, NodeName: node.Name}
if errs[i] != nil {
slog.WarnContext(r.Context(), "hw-accel: node probe failed", "component", "api",
"node", node.URL, "error", errs[i])
entry.Error = errs[i].Error()
} else {
entry.Resolved = infos[i].Resolved
entry.RenderDevices = infos[i].RenderDevices
entry.RenderDeviceDetails = infos[i].RenderDeviceDetails
}
inventory.Nodes[i] = entry
}
// Primary flat fields: the first node that answered; if none did, fall
// back to a local probe (matching the historical fallback behavior).
primaried := false
for i := range healthy {
if errs[i] == nil {
inventory.HWAccelInfo = infos[i]
primaried = true
break
}
}
if !primaried {
inventory.HWAccelInfo = playback.DetectHWAccelWithFFmpeg(h.ffmpegPath)
}
writeJSON(w, http.StatusOK, inventory)
}
// HandleBuildInfo handles GET /admin/system/build.
+110
View File
@@ -9,6 +9,8 @@ import (
"github.com/go-chi/chi/v5"
"github.com/Silo-Server/silo-server/internal/buildinfo"
"github.com/Silo-Server/silo-server/internal/nodepool"
"github.com/Silo-Server/silo-server/internal/playback"
)
func TestSystemBuildInfoResponse(t *testing.T) {
@@ -86,3 +88,111 @@ func TestSystemBuildInfoUnavailableResponseShape(t *testing.T) {
}
}
}
func TestHandleHWAccelAggregatesAllHealthyNodes(t *testing.T) {
t.Parallel()
nodeA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/hw-capabilities" {
http.NotFound(w, r)
return
}
json.NewEncoder(w).Encode(playback.HWAccelInfo{
Resolved: "qsv",
RenderDevices: []string{"/dev/dri/renderD128", "/dev/dri/renderD129"},
RenderDeviceDetails: []playback.RenderDeviceInfo{
{Path: "/dev/dri/renderD128", Description: "Intel GPU"},
{Path: "/dev/dri/renderD129", Description: "Intel GPU"},
},
})
}))
defer nodeA.Close()
nodeB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(playback.HWAccelInfo{
Resolved: "vaapi",
RenderDevices: []string{"/dev/dri/renderD128"},
RenderDeviceDetails: []playback.RenderDeviceInfo{
{Path: "/dev/dri/renderD128", Description: "AMD GPU"},
},
})
}))
defer nodeB.Close()
nodeDown := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "boom", http.StatusInternalServerError)
}))
defer nodeDown.Close()
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{
{ID: 1, Name: "node-a", URL: nodeA.URL, Enabled: true, Healthy: true},
{ID: 2, Name: "node-b", URL: nodeB.URL, Enabled: true, Healthy: true},
{ID: 3, Name: "node-down", URL: nodeDown.URL, Enabled: true, Healthy: true},
{ID: 4, Name: "node-unhealthy", URL: "http://unreachable.invalid", Enabled: true, Healthy: false},
})
handler := &SystemHandler{transcodePool: pool, jwtSecret: "secret"}
req := httptest.NewRequest(http.MethodGet, "/admin/system/hw-accel", nil)
rec := httptest.NewRecorder()
handler.HandleHWAccel(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", rec.Code, rec.Body.String())
}
var got HWAccelInventory
if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
t.Fatalf("decoding response: %v", err)
}
// Flat primary fields keep the historical single-probe shape, sourced
// from the first healthy node that answered.
if got.Resolved != "qsv" || got.Source != "transcode_node" || got.NodeURL != nodeA.URL {
t.Fatalf("primary = resolved %q source %q node %q, want first node's probe", got.Resolved, got.Source, got.NodeURL)
}
// Every healthy node appears, in pool order; the unhealthy one does not.
if len(got.Nodes) != 3 {
t.Fatalf("nodes len = %d, want 3: %+v", len(got.Nodes), got.Nodes)
}
if got.Nodes[0].NodeName != "node-a" || got.Nodes[0].Resolved != "qsv" || len(got.Nodes[0].RenderDeviceDetails) != 2 {
t.Fatalf("node-a entry = %+v", got.Nodes[0])
}
if got.Nodes[1].NodeName != "node-b" || got.Nodes[1].Resolved != "vaapi" || len(got.Nodes[1].RenderDevices) != 1 {
t.Fatalf("node-b entry = %+v", got.Nodes[1])
}
if got.Nodes[2].NodeName != "node-down" || got.Nodes[2].Error == "" {
t.Fatalf("failed node entry = %+v, want populated error", got.Nodes[2])
}
}
func TestHandleHWAccelAllNodeProbesFailFallsBackLocal(t *testing.T) {
t.Parallel()
nodeDown := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "boom", http.StatusInternalServerError)
}))
defer nodeDown.Close()
pool := nodepool.NewTranscodePool()
pool.SetNodes([]*nodepool.Node{
{ID: 1, Name: "node-down", URL: nodeDown.URL, Enabled: true, Healthy: true},
})
handler := &SystemHandler{transcodePool: pool, jwtSecret: "secret"}
req := httptest.NewRequest(http.MethodGet, "/admin/system/hw-accel", nil)
rec := httptest.NewRecorder()
handler.HandleHWAccel(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", rec.Code, rec.Body.String())
}
var got HWAccelInventory
if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
t.Fatalf("decoding response: %v", err)
}
if got.Source != "local" {
t.Fatalf("source = %q, want local fallback when every node probe fails", got.Source)
}
if len(got.Nodes) != 1 || got.Nodes[0].Error == "" {
t.Fatalf("nodes = %+v, want the failing node listed with its error", got.Nodes)
}
}
+5 -1
View File
@@ -33,7 +33,10 @@ func ExtractFrame(ctx context.Context, opts FrameExtractOptions) ([]byte, string
}
resolvedAccel := playback.ResolveHWAccelWithFFmpeg(opts.HWAccel, ffmpegPath)
resolvedDevice := opts.HWDevice
// Resolve a multi-device hw_device list to one concrete GPU for this
// extraction; the reservation spans only the hardware attempt below.
resolvedDevice, releaseHWDevice := playback.AcquireHWDevice(opts.HWDevice, resolvedAccel)
defer releaseHWDevice()
if resolvedDevice == "" && (resolvedAccel == "qsv" || resolvedAccel == "vaapi") {
resolvedDevice = playback.PickRenderDevice("")
}
@@ -48,6 +51,7 @@ func ExtractFrame(ctx context.Context, opts FrameExtractOptions) ([]byte, string
attemptCtx, cancel := context.WithTimeout(ctx, extractTimeoutForAttempt(true, opts.ToneMap))
data, err := runExtract(attemptCtx, ffmpegPath, args)
cancel()
releaseHWDevice()
if err == nil {
return data, "", nil
}
+14 -17
View File
@@ -89,18 +89,17 @@ type ChapterThumbnailRequest struct {
}
type Service struct {
fileRepo FileRepository
folderRepo FolderRepository
probeEnsurer ProbeEnsurer
settings SettingsReader
store ObjectStore
notifier ThumbnailNotifier
ffmpegPath string
hwAccel string
hwDevice string
hwResolveOnce sync.Once
resolvedHWAccel string
resolvedHWDevice string
fileRepo FileRepository
folderRepo FolderRepository
probeEnsurer ProbeEnsurer
settings SettingsReader
store ObjectStore
notifier ThumbnailNotifier
ffmpegPath string
hwAccel string
hwDevice string
hwResolveOnce sync.Once
resolvedHWAccel string
notifyNormal chan struct{}
notifyPriority chan struct{}
@@ -653,12 +652,10 @@ func (s *Service) extractFrameLocal(ctx context.Context, inputPath string, seekS
func (s *Service) resolveHWConfig() (string, string) {
s.hwResolveOnce.Do(func() {
s.resolvedHWAccel = playback.ResolveHWAccelWithFFmpeg(s.hwAccel, s.ffmpegPath)
s.resolvedHWDevice = s.hwDevice
if s.resolvedHWDevice == "" && (s.resolvedHWAccel == "qsv" || s.resolvedHWAccel == "vaapi") {
s.resolvedHWDevice = playback.PickRenderDevice("")
}
})
return s.resolvedHWAccel, s.resolvedHWDevice
// The configured device value passes through raw: ExtractFrame resolves it
// (multi-device balancing, empty-value auto-detection) per extraction.
return s.resolvedHWAccel, s.hwDevice
}
func (s *Service) chapterThumbnailExecutionMode(ctx context.Context) string {
+30
View File
@@ -674,3 +674,33 @@ func TestReserveRemoteNodeAccountsForReservations(t *testing.T) {
t.Fatalf("reserveRemoteNode() node = %#v, want node-2", node)
}
}
func TestExtractFrameResolvesMultiDeviceListToOneDevice(t *testing.T) {
service := &Service{
hwAccel: "vaapi",
hwDevice: "/dev/dri/renderD888,/dev/dri/renderD889",
}
file := &models.MediaFile{FilePath: "/media/movie.mkv"}
var hwArgs []string
service.runFFmpegFrameExtractFunc = func(_ context.Context, _ string, args []string) ([]byte, error) {
if hwArgs == nil {
hwArgs = append([]string(nil), args...)
}
return []byte("frame"), nil
}
if _, _, err := service.extractFrame(context.Background(), file, 5, chapterThumbnailHDRPolicyBestEffort); err != nil {
t.Fatalf("extractFrame() error = %v", err)
}
joined := strings.Join(hwArgs, " ")
// Neither test device exists, so the balancer deterministically falls back
// to the first entry — but never passes the raw list to ffmpeg.
if strings.Contains(joined, "/dev/dri/renderD888,/dev/dri/renderD889") {
t.Fatalf("ffmpeg args contain the raw device list:\n%s", joined)
}
if !strings.Contains(joined, "/dev/dri/renderD888") {
t.Fatalf("ffmpeg args missing a resolved device:\n%s", joined)
}
}
+15 -3
View File
@@ -157,9 +157,21 @@ func (c MatcherConfig) TVSeriesRootQueueEnabled() bool {
// PlaybackConfig holds transcoding and playback settings.
type PlaybackConfig struct {
FFmpegPath string `yaml:"ffmpeg_path"`
TranscodeDir string `yaml:"transcode_dir"`
HWAccel string `yaml:"hw_accel"`
FFmpegPath string `yaml:"ffmpeg_path"`
TranscodeDir string `yaml:"transcode_dir"`
HWAccel string `yaml:"hw_accel"`
// HWDevice is the GPU render device for hardware transcodes. A single
// path pins every GPU workload to that device; a comma-separated list
// (e.g. "/dev/dri/renderD128,/dev/dri/renderD129") balances workloads
// least-loaded across the listed devices. Empty auto-detects.
//
// Supported topology: balancing applies to QSV/VAAPI render devices
// only (NVENC identifies GPUs by CUDA index/UUID and always uses the
// first entry). The value is cluster-wide — every transcode node reads
// it — so multi-node deployments must expose the listed devices at
// identical paths on every node; devices absent on a node fall out of
// that node's rotation. The admin hw-accel endpoint reports each node's
// inventory so the UI can flag divergence.
HWDevice string `yaml:"hw_device"`
ChapterThumbnailWorkers int `yaml:"chapter_thumbnail_workers"`
ChapterThumbnailExecution string `yaml:"chapter_thumbnail_execution"`
+65 -11
View File
@@ -36,12 +36,13 @@ var nvencProbeCache = struct {
// HWAccelInfo describes the detected hardware acceleration capability.
type HWAccelInfo struct {
Resolved string `json:"resolved"`
RenderDevices []string `json:"render_devices"`
IntelDetected bool `json:"intel_detected"`
Source string `json:"source"`
NodeURL string `json:"node_url,omitempty"`
Transformations []TransformationV3 `json:"transformations,omitempty"`
Resolved string `json:"resolved"`
RenderDevices []string `json:"render_devices"`
RenderDeviceDetails []RenderDeviceInfo `json:"render_device_details"`
IntelDetected bool `json:"intel_detected"`
Source string `json:"source"`
NodeURL string `json:"node_url,omitempty"`
Transformations []TransformationV3 `json:"transformations,omitempty"`
}
// DetectHWAccel probes this host's GPU hardware and returns structured info.
@@ -60,15 +61,18 @@ func DetectHWAccelWithFFmpeg(ffmpegPath string) HWAccelInfo {
}
}
return HWAccelInfo{
Resolved: ResolveHWAccelWithFFmpeg("auto", ffmpegPath),
RenderDevices: devices,
IntelDetected: intel,
Source: "local",
Resolved: ResolveHWAccelWithFFmpeg("auto", ffmpegPath),
RenderDevices: devices,
RenderDeviceDetails: renderDeviceDetails(devices),
IntelDetected: intel,
Source: "local",
}
}
// PickRenderDevice returns the GPU render device path to use.
// If explicit is non-empty, it is returned as-is.
// If explicit is non-empty, it is returned as-is — multi-device lists are
// resolved to one device by AcquireHWDevice before args are built, so this
// never sees a list on a live path.
// Otherwise, it attempts to discover a render device under /dev/dri/.
// Returns empty string if no device is found (caller should fall back to CPU).
func PickRenderDevice(explicit string) string {
@@ -312,3 +316,53 @@ func detectRenderDevice(driDir string) string {
}
return ""
}
// RenderDeviceInfo describes one render device for operator-facing surfaces.
type RenderDeviceInfo struct {
Path string `json:"path"`
Description string `json:"description"`
}
// describeRenderDevice builds a short human label for a render device from
// its sysfs PCI vendor/device ids; best-effort, never fails.
func describeRenderDevice(renderDevPath string) string {
name := filepath.Base(renderDevPath)
vendor := readSysfsID(filepath.Join(sysClassDRMDir, name, "device", "vendor"))
label := ""
switch vendor {
case "0x8086":
label = "Intel GPU"
case "0x10de":
label = "NVIDIA GPU"
case "0x1002":
label = "AMD GPU"
case "":
return "GPU"
default:
label = "GPU (vendor " + vendor + ")"
}
if device := readSysfsID(filepath.Join(sysClassDRMDir, name, "device", "device")); device != "" && vendor != "0x1002" {
label += " (" + device + ")"
}
return label
}
func readSysfsID(path string) string {
data, err := os.ReadFile(path)
if err != nil {
return ""
}
return strings.TrimSpace(string(data))
}
// renderDeviceDetails describes every listed device.
func renderDeviceDetails(devices []string) []RenderDeviceInfo {
details := make([]RenderDeviceInfo, 0, len(devices))
for _, device := range devices {
details = append(details, RenderDeviceInfo{
Path: device,
Description: describeRenderDevice(device),
})
}
return details
}
+187
View File
@@ -0,0 +1,187 @@
package playback
import (
"log/slog"
"os"
"strings"
"sync"
)
// Multi-GPU balancing: playback.hw_device accepts a comma-separated list of
// render devices (e.g. "/dev/dri/renderD128,/dev/dri/renderD129"). Every GPU
// workload (streaming sessions, prepared downloads, chapter thumbnails)
// resolves the list to exactly one concrete device via AcquireHWDevice
// immediately before launching ffmpeg, and releases it when that process has
// exited. Balancing is supported for the render-device accelerators (QSV and
// VAAPI) on a homogeneous device list; NVENC identifies GPUs by CUDA
// index/UUID rather than render-node path, so a multi-entry list falls back
// to its first entry. A single configured value keeps the historical
// pass-through contract for every accelerator, so existing deployments are
// unaffected.
// HWDeviceSet is the parsed form of the playback.hw_device setting: an
// ordered list of device entries. Order is priority order — ties in load
// resolve to the earlier entry.
type HWDeviceSet struct {
devices []string
}
// ParseHWDeviceSet splits a configured hw_device value into its device set,
// trimming whitespace and dropping empty entries.
func ParseHWDeviceSet(configured string) HWDeviceSet {
if strings.TrimSpace(configured) == "" {
return HWDeviceSet{}
}
parts := strings.Split(configured, ",")
devices := make([]string, 0, len(parts))
for _, part := range parts {
if trimmed := strings.TrimSpace(part); trimmed != "" {
devices = append(devices, trimmed)
}
}
return HWDeviceSet{devices: devices}
}
// Empty reports whether no device is configured (auto-detection applies).
func (s HWDeviceSet) Empty() bool { return len(s.devices) == 0 }
// Multi reports whether more than one device is configured.
func (s HWDeviceSet) Multi() bool { return len(s.devices) > 1 }
// List returns the configured devices in priority order.
func (s HWDeviceSet) List() []string { return s.devices }
// First returns the first configured device, or "" when empty.
func (s HWDeviceSet) First() string {
if len(s.devices) == 0 {
return ""
}
return s.devices[0]
}
// hwDeviceStat reports whether a device path exists; overridable in tests.
var hwDeviceStat = func(path string) error {
_, err := os.Stat(path)
return err
}
// hwDeviceLoad tracks active GPU workloads per render device.
var hwDeviceLoad = struct {
mu sync.Mutex
counts map[string]int
}{counts: map[string]int{}}
// hwAccelBalancesRenderDevices reports whether the resolved acceleration mode
// selects GPUs by render-device path, which is what the balancer hands out.
// NVENC addresses GPUs by CUDA index/UUID and is deliberately excluded.
func hwAccelBalancesRenderDevices(hwAccel string) bool {
return hwAccel == "qsv" || hwAccel == "vaapi"
}
// presentHWDevices filters a device list to the entries that exist, falling
// back to the first entry when none do so the failure mode stays
// deterministic (ffmpeg reports the missing device, matching the historical
// wrong-path behavior of an explicit single value).
func presentHWDevices(devices []string) []string {
present := make([]string, 0, len(devices))
for _, device := range devices {
if hwDeviceStat(device) == nil {
present = append(present, device)
}
}
if len(present) == 0 {
return devices[:1]
}
return present
}
// leastLoadedHWDeviceLocked picks the device with the fewest active
// workloads, preserving list order on ties. Callers must hold hwDeviceLoad.mu.
func leastLoadedHWDeviceLocked(present []string) string {
best := present[0]
for _, device := range present[1:] {
if hwDeviceLoad.counts[device] < hwDeviceLoad.counts[best] {
best = device
}
}
return best
}
// newHWDeviceRelease returns an idempotent release for a reservation that has
// already been added to hwDeviceLoad.
func newHWDeviceRelease(device string) func() {
var once sync.Once
return func() {
once.Do(func() {
hwDeviceLoad.mu.Lock()
if hwDeviceLoad.counts[device] > 0 {
hwDeviceLoad.counts[device]--
}
hwDeviceLoad.mu.Unlock()
})
}
}
// reserveConcreteHWDevice reserves a device that was selected for an earlier
// process in the same transcode session. Restarts keep device affinity rather
// than running the least-loaded selection again.
func reserveConcreteHWDevice(device string) func() {
hwDeviceLoad.mu.Lock()
hwDeviceLoad.counts[device]++
count := hwDeviceLoad.counts[device]
hwDeviceLoad.mu.Unlock()
slog.Info("GPU workload device reserved", "device", device, "active_workloads", count)
return newHWDeviceRelease(device)
}
var nvencMultiDeviceWarnOnce sync.Once
// AcquireHWDevice resolves the configured hw_device value to exactly one
// device for one GPU workload. resolvedHWAccel must already be resolved (no
// "auto"). The returned release must be called exactly once when the ffmpeg
// process for this workload has exited; it is idempotent and a no-op for
// workloads that did not reserve (empty or single-device value, or an
// accelerator the balancer does not manage).
//
// - Empty value: returns "" so downstream auto-detection applies.
// - Single value: passes through unchanged for every accelerator.
// - Multi-device value with QSV/VAAPI: reserves the present device with the
// fewest active workloads (ties keep list order) until release.
// - Multi-device value with any other accelerator (including NVENC, which
// addresses GPUs by CUDA index/UUID, not render-node path): falls back to
// the first entry without reserving.
func AcquireHWDevice(configured, resolvedHWAccel string) (string, func()) {
noop := func() {}
set := ParseHWDeviceSet(configured)
if !set.Multi() {
return set.First(), noop
}
if !hwAccelBalancesRenderDevices(resolvedHWAccel) {
if resolvedHWAccel == "nvenc" {
nvencMultiDeviceWarnOnce.Do(func() {
slog.Warn("multi-device hw_device is not supported with NVENC (devices are CUDA index/UUID, not render paths); using the first entry",
"hw_device", configured, "using", set.First())
})
}
return set.First(), noop
}
// Select and reserve in one critical section so concurrent workload starts
// observe each other's reservations instead of piling onto one device.
present := presentHWDevices(set.List())
hwDeviceLoad.mu.Lock()
device := leastLoadedHWDeviceLocked(present)
hwDeviceLoad.counts[device]++
count := hwDeviceLoad.counts[device]
hwDeviceLoad.mu.Unlock()
slog.Info("GPU workload device selected", "device", device, "active_workloads", count)
return device, newHWDeviceRelease(device)
}
// hwDeviceActiveCount reports the active workload count for one device; test
// helper for asserting release boundaries.
func hwDeviceActiveCount(device string) int {
hwDeviceLoad.mu.Lock()
defer hwDeviceLoad.mu.Unlock()
return hwDeviceLoad.counts[device]
}
+284
View File
@@ -0,0 +1,284 @@
package playback
import (
"os"
"sync"
"testing"
)
// fakeDeviceStat installs a stat function that reports only the given paths as
// present, restoring the real one on cleanup.
func fakeDeviceStat(t *testing.T, present ...string) {
t.Helper()
set := make(map[string]bool, len(present))
for _, p := range present {
set[p] = true
}
orig := hwDeviceStat
hwDeviceStat = func(path string) error {
if set[path] {
return nil
}
return os.ErrNotExist
}
t.Cleanup(func() { hwDeviceStat = orig })
}
func resetDeviceLoad(t *testing.T) {
t.Helper()
hwDeviceLoad.mu.Lock()
hwDeviceLoad.counts = map[string]int{}
hwDeviceLoad.mu.Unlock()
}
func TestParseHWDeviceSet(t *testing.T) {
cases := []struct {
in string
want []string
}{
{"", nil},
{"/dev/dri/renderD128", []string{"/dev/dri/renderD128"}},
{"/dev/dri/renderD128,/dev/dri/renderD129", []string{"/dev/dri/renderD128", "/dev/dri/renderD129"}},
{" /dev/dri/renderD128 , /dev/dri/renderD129 ,", []string{"/dev/dri/renderD128", "/dev/dri/renderD129"}},
}
for _, tc := range cases {
got := ParseHWDeviceSet(tc.in)
if got.Empty() != (len(tc.want) == 0) || got.Multi() != (len(tc.want) > 1) {
t.Fatalf("ParseHWDeviceSet(%q) Empty/Multi mismatch for %v", tc.in, tc.want)
}
list := got.List()
if len(list) != len(tc.want) {
t.Fatalf("ParseHWDeviceSet(%q).List() = %v, want %v", tc.in, list, tc.want)
}
for i := range list {
if list[i] != tc.want[i] {
t.Fatalf("ParseHWDeviceSet(%q).List() = %v, want %v", tc.in, list, tc.want)
}
}
}
}
func TestAcquireHWDeviceEmptyValueStaysEmpty(t *testing.T) {
resetDeviceLoad(t)
device, release := AcquireHWDevice("", "qsv")
defer release()
if device != "" {
t.Fatalf("device = %q, want empty so auto-detection applies", device)
}
}
func TestAcquireHWDeviceSingleValuePassesThrough(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t) // nothing exists; single value must still pass through
for _, accel := range []string{"qsv", "vaapi", "nvenc", "none"} {
device, release := AcquireHWDevice("/dev/dri/renderD128", accel)
if device != "/dev/dri/renderD128" {
t.Fatalf("accel %s: device = %q, want explicit single value unchanged", accel, device)
}
release()
}
}
func TestAcquireHWDeviceBalancesAcrossList(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t, "/dev/dri/renderD128", "/dev/dri/renderD129")
configured := "/dev/dri/renderD128,/dev/dri/renderD129"
dev1, release1 := AcquireHWDevice(configured, "qsv")
if dev1 != "/dev/dri/renderD128" {
t.Fatalf("first workload device = %q, want first listed on tie", dev1)
}
dev2, release2 := AcquireHWDevice(configured, "vaapi")
if dev2 != "/dev/dri/renderD129" {
t.Fatalf("second workload device = %q, want least-loaded second device", dev2)
}
dev3, release3 := AcquireHWDevice(configured, "qsv")
if dev3 != "/dev/dri/renderD128" {
t.Fatalf("third workload device = %q, want round-back to first on tie", dev3)
}
// Releasing the first workload makes renderD128 least-loaded again.
release1()
release3()
dev4, release4 := AcquireHWDevice(configured, "qsv")
if dev4 != "/dev/dri/renderD128" {
t.Fatalf("post-release device = %q, want freed first device", dev4)
}
release2()
release4()
}
func TestAcquireHWDeviceSkipsMissingDevices(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t, "/dev/dri/renderD129")
device, release := AcquireHWDevice("/dev/dri/renderD128,/dev/dri/renderD129", "qsv")
defer release()
if device != "/dev/dri/renderD129" {
t.Fatalf("device = %q, want the only present device", device)
}
}
func TestAcquireHWDeviceAllMissingFallsBackToFirst(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t) // none exist
device, release := AcquireHWDevice("/dev/dri/renderD128,/dev/dri/renderD129", "qsv")
defer release()
if device != "/dev/dri/renderD128" {
t.Fatalf("device = %q, want deterministic first entry when none exist", device)
}
}
func TestAcquireHWDeviceNVENCMultiListUsesFirstWithoutReserving(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t) // NVENC entries are CUDA indexes/UUIDs, never present as paths
device, release := AcquireHWDevice("0,1", "nvenc")
defer release()
if device != "0" {
t.Fatalf("device = %q, want first NVENC entry", device)
}
if got := hwDeviceActiveCount("0"); got != 0 {
t.Fatalf("active count = %d, want no reservation for NVENC", got)
}
}
func TestAcquireHWDeviceSoftwareAccelDoesNotReserve(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t, "/dev/dri/renderD128", "/dev/dri/renderD129")
configured := "/dev/dri/renderD128,/dev/dri/renderD129"
_, releaseNone := AcquireHWDevice(configured, "none")
defer releaseNone()
// A software workload must not shift the balance: the next GPU workload
// still lands on the first device.
device, release := AcquireHWDevice(configured, "qsv")
defer release()
if device != "/dev/dri/renderD128" {
t.Fatalf("device = %q, want first device unaffected by software workload", device)
}
}
func TestAcquireHWDeviceReleaseIsIdempotent(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t, "/dev/dri/renderD128", "/dev/dri/renderD129")
configured := "/dev/dri/renderD128,/dev/dri/renderD129"
_, release1 := AcquireHWDevice(configured, "qsv")
release1()
release1() // double release must not underflow the count
dev, release2 := AcquireHWDevice(configured, "qsv")
defer release2()
if dev != "/dev/dri/renderD128" {
t.Fatalf("device = %q, want first device after idempotent release", dev)
}
hwDeviceLoad.mu.Lock()
defer hwDeviceLoad.mu.Unlock()
for device, count := range hwDeviceLoad.counts {
if count < 0 {
t.Fatalf("device %s count = %d, want never negative", device, count)
}
}
}
func TestPickRenderDeviceExplicitValuePassesThrough(t *testing.T) {
// PickRenderDevice is auto-detection only; list resolution happens in
// AcquireHWDevice before args are built, so an explicit value — even a
// stale CSV — passes through untouched.
if got := PickRenderDevice("/dev/dri/renderD42"); got != "/dev/dri/renderD42" {
t.Fatalf("PickRenderDevice(single) = %q, want unchanged explicit value", got)
}
}
func TestDetectHWAccelRenderDeviceDetails(t *testing.T) {
driDir := t.TempDir()
sysDir := t.TempDir()
for name, ids := range map[string][2]string{
"renderD128": {"0x8086", "0x56a6"},
"renderD129": {"0x10de", "0x2489"},
} {
if err := os.WriteFile(driDir+"/"+name, nil, 0o644); err != nil {
t.Fatal(err)
}
devDir := sysDir + "/" + name + "/device"
if err := os.MkdirAll(devDir, 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(devDir+"/vendor", []byte(ids[0]+"\n"), 0o644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(devDir+"/device", []byte(ids[1]+"\n"), 0o644); err != nil {
t.Fatal(err)
}
}
origDRI, origSys := defaultDRIDir, sysClassDRMDir
defaultDRIDir, sysClassDRMDir = driDir, sysDir
t.Cleanup(func() { defaultDRIDir, sysClassDRMDir = origDRI, origSys })
info := DetectHWAccel()
if len(info.RenderDeviceDetails) != 2 {
t.Fatalf("RenderDeviceDetails len = %d, want 2: %+v", len(info.RenderDeviceDetails), info.RenderDeviceDetails)
}
first, second := info.RenderDeviceDetails[0], info.RenderDeviceDetails[1]
if first.Path != driDir+"/renderD128" || first.Description != "Intel GPU (0x56a6)" {
t.Fatalf("first device = %+v, want Intel description", first)
}
if second.Path != driDir+"/renderD129" || second.Description != "NVIDIA GPU (0x2489)" {
t.Fatalf("second device = %+v, want NVIDIA description", second)
}
}
func TestDescribeRenderDeviceUnknownVendor(t *testing.T) {
sysDir := t.TempDir()
devDir := sysDir + "/renderD130/device"
if err := os.MkdirAll(devDir, 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(devDir+"/vendor", []byte("0x1002\n"), 0o644); err != nil {
t.Fatal(err)
}
origSys := sysClassDRMDir
sysClassDRMDir = sysDir
t.Cleanup(func() { sysClassDRMDir = origSys })
if got := describeRenderDevice("/dev/dri/renderD130"); got != "AMD GPU" {
t.Fatalf("describeRenderDevice() = %q, want AMD GPU without device id", got)
}
if got := describeRenderDevice("/dev/dri/renderD999"); got != "GPU" {
t.Fatalf("describeRenderDevice() = %q, want bare GPU for unreadable sysfs", got)
}
}
func TestAcquireHWDeviceConcurrentStartsBalanceExactly(t *testing.T) {
resetDeviceLoad(t)
fakeDeviceStat(t, "/dev/dri/renderD128", "/dev/dri/renderD129")
configured := "/dev/dri/renderD128,/dev/dri/renderD129"
const workloads = 8
var wg sync.WaitGroup
devices := make([]string, workloads)
releases := make([]func(), workloads)
for i := range workloads {
wg.Add(1)
go func() {
defer wg.Done()
devices[i], releases[i] = AcquireHWDevice(configured, "qsv")
}()
}
wg.Wait()
defer func() {
for _, release := range releases {
release()
}
}()
counts := map[string]int{}
for _, device := range devices {
counts[device]++
}
// Atomic select+reserve guarantees an exact split; a two-step selection
// could pile concurrent starts onto one device.
if counts["/dev/dri/renderD128"] != workloads/2 || counts["/dev/dri/renderD129"] != workloads/2 {
t.Fatalf("concurrent workload split = %v, want exact %d/%d", counts, workloads/2, workloads/2)
}
}
+8
View File
@@ -67,6 +67,14 @@ func PrepareFile(ctx context.Context, opts TranscodeOpts, outputPath string) err
// any stale partial first so a failed prior attempt can't be mistaken for output.
_ = os.Remove(partPath)
// Resolve a multi-device hw_device list to one concrete GPU for this
// encode. Run blocks until ffmpeg exits, so the deferred release fires at
// exactly the process-exit boundary.
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
hwDevice, releaseHWDevice := AcquireHWDevice(opts.HWDevice, opts.HWAccel)
opts.HWDevice = hwDevice
defer releaseHWDevice()
args := buildPrepareFileArgs(opts, partPath)
bin := opts.FFmpegPath
if bin == "" {
+45
View File
@@ -1,6 +1,9 @@
package playback
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
@@ -82,3 +85,45 @@ func TestResolvePrepareTarget(t *testing.T) {
t.Fatalf("transcode resolution = %q, want empty (source)", rt.Resolution)
}
}
func TestPrepareFileResolvesOneDeviceAndReleasesAfterExit(t *testing.T) {
resetDeviceLoad(t)
devA, devB := "/dev/dri/renderD888", "/dev/dri/renderD889"
fakeDeviceStat(t, devA, devB)
// Fake ffmpeg: record argv, create the output (last arg) so finalize works.
dir := t.TempDir()
argsFile := filepath.Join(dir, "args.txt")
script := filepath.Join(dir, "ffmpeg")
if err := os.WriteFile(script, []byte("#!/bin/sh\nprintf '%s\\n' \"$@\" > "+argsFile+"\neval \"touch \\${$#}\"\n"), 0o755); err != nil {
t.Fatal(err)
}
outputPath := filepath.Join(dir, "artifact.mp4")
err := PrepareFile(context.Background(), TranscodeOpts{
InputPath: "/nonexistent/input.mkv",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
FFmpegPath: script,
HWAccel: "vaapi",
HWDevice: devA + "," + devB,
}, outputPath)
if err != nil {
t.Fatalf("PrepareFile: %v", err)
}
argv, err := os.ReadFile(argsFile)
if err != nil {
t.Fatal(err)
}
got := string(argv)
if !strings.Contains(got, devA) {
t.Fatalf("ffmpeg args missing resolved device %s:\n%s", devA, got)
}
if strings.Contains(got, devA+","+devB) {
t.Fatalf("ffmpeg args contain the raw device list:\n%s", got)
}
if count := hwDeviceActiveCount(devA); count != 0 {
t.Fatalf("active count after PrepareFile returned = %d, want 0", count)
}
}
+50 -28
View File
@@ -101,6 +101,10 @@ type TranscodeSession struct {
stderrLineIndex int
stderrWriter *ffmpegStderrWriter
restartHook func(context.Context)
// reserveHWDeviceOnRestart is true when StartTranscode selected and reserved
// one device from a multi-device QSV/VAAPI setting. Each replacement ffmpeg
// process reacquires that same concrete device.
reserveHWDeviceOnRestart bool
}
// SetRestartHook registers a callback fired after every successful Restart.
@@ -171,21 +175,29 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
opts.SegmentDuration = defaultSegmentDuration
}
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
configuredHWDevices := ParseHWDeviceSet(opts.HWDevice)
reserveHWDeviceOnRestart := configuredHWDevices.Multi() && hwAccelBalancesRenderDevices(opts.HWAccel)
// Resolve a multi-device hw_device list to one concrete GPU. Restarts reuse
// the selected device, but each ffmpeg process owns its own reservation.
hwDevice, releaseHWDevice := AcquireHWDevice(opts.HWDevice, opts.HWAccel)
opts.HWDevice = hwDevice
// Ensure output directory exists.
if err := os.MkdirAll(opts.OutputDir, 0o755); err != nil {
releaseHWDevice()
return nil, fmt.Errorf("create output dir: %w", err)
}
ctx, cancel := context.WithCancel(ctx)
s := &TranscodeSession{
cancel: cancel,
opts: opts,
outputDir: opts.OutputDir,
running: true,
done: make(chan struct{}),
stderr: newBoundedTailBuffer(stderrTailMaxBytes),
lastRequestedSegment: opts.StartSegmentNumber,
cancel: cancel,
opts: opts,
outputDir: opts.OutputDir,
running: true,
done: make(chan struct{}),
stderr: newBoundedTailBuffer(stderrTailMaxBytes),
lastRequestedSegment: opts.StartSegmentNumber,
reserveHWDeviceOnRestart: reserveHWDeviceOnRestart,
}
args := buildFFmpegArgs(opts)
@@ -201,6 +213,7 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
stdinPipe, err := cmd.StdinPipe()
if err != nil {
cancel()
releaseHWDevice()
return nil, fmt.Errorf("create stdin pipe: %w", err)
}
cmd.Dir = opts.OutputDir
@@ -209,6 +222,7 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
if err := cmd.Start(); err != nil {
cancel()
releaseHWDevice()
s.logFFmpegEvent(ctx, "ffmpeg process exit error", err.Error())
return nil, fmt.Errorf("start ffmpeg: %w", err)
}
@@ -216,21 +230,28 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
s.stdinPipe = stdinPipe
s.logFFmpegEvent(ctx, "ffmpeg process started", "")
// Monitor ffmpeg in background.
go func() {
waitErr := cmd.Wait()
s.flushStderr(ctx)
s.mu.Lock()
s.running = false
s.waitErr = waitErr
s.mu.Unlock()
s.logWaitResult(ctx, waitErr)
close(s.done)
}()
// Monitor ffmpeg in background. The process-specific reservation is released
// before done closes, so waiters can safely launch a replacement process.
go s.monitorFFmpeg(ctx, cmd, s.done, releaseHWDevice)
return s, nil
}
func (s *TranscodeSession) monitorFFmpeg(ctx context.Context, cmd *exec.Cmd, done chan struct{}, releaseHWDevice func()) {
waitErr := cmd.Wait()
// The child no longer owns the device once Wait returns. Release before
// flushing/logging so slow diagnostics cannot make the allocator count a
// process that has already exited.
releaseHWDevice()
s.flushStderr(ctx)
s.mu.Lock()
s.running = false
s.waitErr = waitErr
s.mu.Unlock()
s.logWaitResult(ctx, waitErr)
close(done)
}
// IsMPEG2VideoCodec reports whether a probed video codec name identifies
// MPEG-2 video. It accepts common FFmpeg aliases because codec strings can
// come from scan metadata, direct probes, or client capability lists.
@@ -1611,6 +1632,7 @@ func (s *TranscodeSession) restart(
}
s.restartCount++
opts := s.opts
reserveHWDevice := s.reserveHWDeviceOnRestart
s.mu.Unlock()
// Copy-mode restarts must clean stale segments so ffmpeg writes fresh
@@ -1655,8 +1677,17 @@ func (s *TranscodeSession) restart(
cmd.Stderr = s.newStderrWriter(ctx)
cmd.WaitDelay = 3 * time.Second
// The previous process released its reservation before closing done. Keep
// this session on the same concrete GPU while accounting for the replacement
// process as a new active workload.
releaseHWDevice := func() {}
if reserveHWDevice {
releaseHWDevice = reserveConcreteHWDevice(opts.HWDevice)
}
if err := cmd.Start(); err != nil {
cancel()
releaseHWDevice()
s.mu.Lock()
s.restarting = false
s.waitErr = err
@@ -1680,16 +1711,7 @@ func (s *TranscodeSession) restart(
hook := s.restartHook
s.mu.Unlock()
go func() {
waitErr := cmd.Wait()
s.flushStderr(ctx)
s.mu.Lock()
s.running = false
s.waitErr = waitErr
s.mu.Unlock()
s.logWaitResult(ctx, waitErr)
close(s.done)
}()
go s.monitorFFmpeg(ctx, cmd, s.done, releaseHWDevice)
if hook != nil {
hook(ctx)
@@ -0,0 +1,232 @@
package playback
import (
"context"
"os"
"path/filepath"
"testing"
"time"
)
// fakeFFmpegScript writes a shell script that runs until killed, standing in
// for a long-lived ffmpeg process.
func fakeFFmpegScript(t *testing.T) string {
t.Helper()
path := filepath.Join(t.TempDir(), "ffmpeg")
script := "#!/bin/sh\nwhile :; do sleep 0.1; done\n"
if err := os.WriteFile(path, []byte(script), 0o755); err != nil {
t.Fatal(err)
}
return path
}
// restartableFakeFFmpegScript exits cleanly on its first invocation, then
// stays alive on later invocations so restart reservation behavior can be
// observed without running a real transcode.
func restartableFakeFFmpegScript(t *testing.T) string {
t.Helper()
path := filepath.Join(t.TempDir(), "ffmpeg")
script := "#!/bin/sh\n" +
"marker=\"$PWD/ffmpeg-started\"\n" +
"if [ ! -e \"$marker\" ]; then touch \"$marker\"; exit 0; fi\n" +
"while :; do sleep 0.1; done\n"
if err := os.WriteFile(path, []byte(script), 0o755); err != nil {
t.Fatal(err)
}
return path
}
func TestStartTranscodeHoldsGPUReservationUntilProcessExit(t *testing.T) {
resetDeviceLoad(t)
devA, devB := "/dev/dri/renderD888", "/dev/dri/renderD889"
fakeDeviceStat(t, devA, devB)
s, err := StartTranscode(context.Background(), TranscodeOpts{
InputPath: "/nonexistent/input.mkv",
OutputDir: t.TempDir(),
SessionID: "hwdevice-session",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
FFmpegPath: fakeFFmpegScript(t),
HWAccel: "qsv",
HWDevice: devA + "," + devB,
})
if err != nil {
t.Fatalf("StartTranscode: %v", err)
}
if got := s.Opts().HWDevice; got != devA {
t.Fatalf("session device = %q, want first listed device %q", got, devA)
}
if got := hwDeviceActiveCount(devA); got != 1 {
t.Fatalf("active count while running = %d, want 1", got)
}
// CloseProcess kills ffmpeg and waits for it to be reaped; the
// reservation must survive until that wait completes and be gone after.
if err := s.CloseProcess(); err != nil {
t.Fatalf("CloseProcess: %v", err)
}
select {
case <-s.Done():
case <-time.After(5 * time.Second):
t.Fatal("ffmpeg process did not exit")
}
if got := hwDeviceActiveCount(devA); got != 0 {
t.Fatalf("active count after close = %d, want 0", got)
}
}
func TestStartTranscodeReleasesReservationOnSpawnFailure(t *testing.T) {
resetDeviceLoad(t)
devA, devB := "/dev/dri/renderD888", "/dev/dri/renderD889"
fakeDeviceStat(t, devA, devB)
_, err := StartTranscode(context.Background(), TranscodeOpts{
InputPath: "/nonexistent/input.mkv",
OutputDir: t.TempDir(),
SessionID: "hwdevice-spawn-fail",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
FFmpegPath: filepath.Join(t.TempDir(), "missing-ffmpeg"),
HWAccel: "qsv",
HWDevice: devA + "," + devB,
})
if err == nil {
t.Fatal("StartTranscode succeeded with a missing ffmpeg binary")
}
if got := hwDeviceActiveCount(devA); got != 0 {
t.Fatalf("active count after failed spawn = %d, want 0", got)
}
}
func TestStartTranscodeReleasesReservationOnNaturalExit(t *testing.T) {
resetDeviceLoad(t)
devA, devB := "/dev/dri/renderD888", "/dev/dri/renderD889"
fakeDeviceStat(t, devA, devB)
s, err := StartTranscode(context.Background(), TranscodeOpts{
InputPath: "/nonexistent/input.mkv",
OutputDir: t.TempDir(),
SessionID: "hwdevice-natural-exit",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
FFmpegPath: restartableFakeFFmpegScript(t),
HWAccel: "qsv",
HWDevice: devA + "," + devB,
})
if err != nil {
t.Fatalf("StartTranscode: %v", err)
}
select {
case <-s.Done():
case <-time.After(5 * time.Second):
t.Fatal("ffmpeg process did not exit")
}
if got := hwDeviceActiveCount(devA); got != 0 {
t.Fatalf("active count after natural exit = %d, want 0", got)
}
if err := s.CloseProcess(); err != nil {
t.Fatalf("CloseProcess: %v", err)
}
}
func TestRestartReacquiresSameGPUAfterNaturalExit(t *testing.T) {
resetDeviceLoad(t)
devA, devB := "/dev/dri/renderD888", "/dev/dri/renderD889"
configured := devA + "," + devB
fakeDeviceStat(t, devA, devB)
s, err := StartTranscode(context.Background(), TranscodeOpts{
InputPath: "/nonexistent/input.mkv",
OutputDir: t.TempDir(),
SessionID: "hwdevice-restart",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
FFmpegPath: restartableFakeFFmpegScript(t),
HWAccel: "qsv",
HWDevice: configured,
})
if err != nil {
t.Fatalf("StartTranscode: %v", err)
}
t.Cleanup(func() { _ = s.CloseProcess() })
select {
case <-s.Done():
case <-time.After(5 * time.Second):
t.Fatal("initial ffmpeg process did not exit")
}
occupiedDevice, releaseOccupant := AcquireHWDevice(configured, "qsv")
defer releaseOccupant()
if occupiedDevice != devA {
t.Fatalf("device selected after natural exit = %q, want released device %q", occupiedDevice, devA)
}
if err := s.Restart(context.Background(), 10, 5); err != nil {
t.Fatalf("Restart: %v", err)
}
if got := s.Opts().HWDevice; got != devA {
t.Fatalf("restart device = %q, want original device %q", got, devA)
}
if got := hwDeviceActiveCount(devA); got != 2 {
t.Fatalf("active count while restarted process runs = %d, want 2", got)
}
if got := hwDeviceActiveCount(devB); got != 0 {
t.Fatalf("alternate device active count = %d, want 0", got)
}
if err := s.CloseProcess(); err != nil {
t.Fatalf("CloseProcess: %v", err)
}
if got := hwDeviceActiveCount(devA); got != 1 {
t.Fatalf("active count after restarted process exit = %d, want occupant only (1)", got)
}
}
func TestRestartReleasesReservationOnSpawnFailure(t *testing.T) {
resetDeviceLoad(t)
devA, devB := "/dev/dri/renderD888", "/dev/dri/renderD889"
fakeDeviceStat(t, devA, devB)
bin := restartableFakeFFmpegScript(t)
s, err := StartTranscode(context.Background(), TranscodeOpts{
InputPath: "/nonexistent/input.mkv",
OutputDir: t.TempDir(),
SessionID: "hwdevice-restart-spawn-fail",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
FFmpegPath: bin,
HWAccel: "qsv",
HWDevice: devA + "," + devB,
})
if err != nil {
t.Fatalf("StartTranscode: %v", err)
}
select {
case <-s.Done():
case <-time.After(5 * time.Second):
t.Fatal("initial ffmpeg process did not exit")
}
if err := os.Remove(bin); err != nil {
t.Fatalf("remove fake ffmpeg: %v", err)
}
if err := s.Restart(context.Background(), 10, 5); err == nil {
t.Fatal("Restart succeeded with a missing ffmpeg binary")
}
if got := hwDeviceActiveCount(devA); got != 0 {
t.Fatalf("active count after failed restart spawn = %d, want 0", got)
}
if err := s.CloseProcess(); err != nil {
t.Fatalf("CloseProcess: %v", err)
}
}
+13 -10
View File
@@ -493,16 +493,19 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
SegmentDuration: req.SegmentDuration,
FFmpegPath: cfg.Playback.FFmpegPath,
HWAccel: req.HWAccel,
HWDevice: "",
AudioTrackIndex: req.AudioTrackIndex,
SubtitleTrackIndex: req.SubtitleTrackIndex,
SubtitleBurnIn: req.SubtitleBurnIn,
SubtitleCodec: req.SubtitleCodec,
TotalDuration: req.TotalDuration,
FastStart: true,
NodeType: "transcode",
ExecutionMode: "transcode_node",
FFmpegLogSink: s.ffmpegSink,
// This node's configured device (or device list — StartTranscode
// resolves it to one GPU), matching what reconstruction uses so fresh
// and reconstructed sessions balance identically.
HWDevice: cfg.Playback.HWDevice,
AudioTrackIndex: req.AudioTrackIndex,
SubtitleTrackIndex: req.SubtitleTrackIndex,
SubtitleBurnIn: req.SubtitleBurnIn,
SubtitleCodec: req.SubtitleCodec,
TotalDuration: req.TotalDuration,
FastStart: true,
NodeType: "transcode",
ExecutionMode: "transcode_node",
FFmpegLogSink: s.ffmpegSink,
}
if opts.HWAccel == "" && cfg.Playback.HWAccel != "" {
+47
View File
@@ -402,3 +402,50 @@ func TestCopyModeReconstruct_SkipsFastSeek(t *testing.T) {
t.Fatalf("encoded card must apply the seg×dur fast seek")
}
}
// A fresh /transcode/start must resolve this node's configured hw_device list
// through the shared GPU pool — the same path reconstruction uses — rather
// than bypassing it with an empty device.
func TestHandleStartUsesConfiguredHWDeviceList(t *testing.T) {
server := newTestServer(t)
ffmpegPath := filepath.Join(t.TempDir(), "looping-ffmpeg.sh")
if err := os.WriteFile(ffmpegPath, []byte("#!/bin/sh\nwhile :; do sleep 0.1; done\n"), 0o755); err != nil {
t.Fatal(err)
}
// This test reaches the spawn/track path, so it needs a (no-op) tracker.
server.tracker = nodesessions.NewTracker(nil, "http://node", "node", "transcode")
cfg := server.watcher.Config()
cfg.Playback.FFmpegPath = ffmpegPath
// Neither device exists, so resolution deterministically lands on the
// first entry; the point is that the configured list reaches the session.
cfg.Playback.HWDevice = "/dev/dri/renderD888,/dev/dri/renderD889"
requestBody, err := json.Marshal(TranscodeStartRequest{
SessionID: "hwdevice-start-1",
InputPath: "/media/movie.mkv",
TargetCodecVideo: "h264",
TargetCodecAudio: "aac",
SegmentDuration: 2,
HWAccel: "vaapi",
})
if err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodPost, "/transcode/start", bytes.NewReader(requestBody))
rr := httptest.NewRecorder()
server.handleStart(rr, req)
if rr.Code != http.StatusAccepted {
t.Fatalf("status = %d, body = %s", rr.Code, rr.Body.String())
}
server.mu.RLock()
session := server.sessions["hwdevice-start-1"]
server.mu.RUnlock()
if session == nil {
t.Fatal("session was not registered")
}
defer session.CloseProcess()
if got := session.Opts().HWDevice; got != "/dev/dri/renderD888" {
t.Fatalf("session HWDevice = %q, want one concrete device from the configured list", got)
}
}
+17
View File
@@ -10,12 +10,29 @@ export interface BuildInfo {
available: boolean;
}
export interface RenderDeviceInfo {
path: string;
description: string;
}
export interface NodeHWAccel {
node_url: string;
node_name?: string;
resolved?: string;
render_devices?: string[];
render_device_details?: RenderDeviceInfo[];
error?: string;
}
export interface HWAccelInfo {
resolved: string;
render_devices: string[];
render_device_details?: RenderDeviceInfo[];
intel_detected: boolean;
source: "local" | "transcode_node";
node_url?: string;
/** Per-node inventories when transcode nodes are registered. */
nodes?: NodeHWAccel[];
}
export function useBuildInfo() {
@@ -1,14 +1,23 @@
import { useMemo } from "react";
import { useSettingsForm } from "@/hooks/useSettingsForm";
import { useHWAccelDetection } from "@/hooks/queries/admin/system";
import { Label } from "@/components/ui/label";
import { Switch } from "@/components/ui/switch";
import { SettingField } from "./SettingField";
import { SaveBar } from "./SaveBar";
import { FieldGroup } from "./FieldGroup";
import {
buildHWDeviceRows,
nodeInventoriesDiverge,
parseHWDeviceList,
toggleHWDevice,
} from "./playbackSettings.utils";
const KEYS = [
"playback.ffmpeg_path",
"playback.transcode_dir",
"playback.hw_accel",
"playback.hw_device",
"playback.transcode_enabled",
"playback.local_transcode_fallback",
"allow_4k_transcode",
@@ -24,7 +33,18 @@ const KEYS = [
export default function PlaybackSettings() {
const form = useSettingsForm({ keys: useMemo(() => KEYS, []) });
const hwDetection = useHWAccelDetection(form.getValue("playback.hw_accel") === "auto");
const hwAccel = form.getValue("playback.hw_accel");
const hwDetection = useHWAccelDetection(hwAccel !== "none");
const hwDevice = form.getValue("playback.hw_device");
const selectedDevices = parseHWDeviceList(hwDevice);
const deviceRows = buildHWDeviceRows(hwDetection.data, hwDevice);
const detectedPaths = deviceRows.filter((row) => row.detected).map((row) => row.path);
// Balancing is QSV/VAAPI-only: NVENC addresses GPUs by CUDA index/UUID, so
// the multi-select picker is hidden for it (the server uses the first
// configured entry).
const isNvenc =
hwAccel === "nvenc" || (hwAccel === "auto" && hwDetection.data?.resolved === "nvenc");
const inventoriesDiverge = nodeInventoriesDiverge(hwDetection.data);
if (form.isLoading) return <div>Loading...</div>;
@@ -79,6 +99,66 @@ export default function PlaybackSettings() {
{form.getValue("playback.hw_accel") === "auto" && hwDetection.isLoading && (
<p className="text-muted-foreground -mt-1 text-xs">Detecting hardware...</p>
)}
{hwAccel !== "none" && isNvenc && selectedDevices.length > 1 && (
<p className="-mt-1 text-xs text-amber-500">
Multi-GPU balancing supports QSV/VA-API only; with NVENC the server uses the first
configured device ({selectedDevices[0]}).
</p>
)}
{hwAccel !== "none" && !isNvenc && deviceRows.length > 0 && (
<div className="flex flex-col gap-2 py-3">
<div className="space-y-0.5">
<Label className="text-sm font-medium">GPU Devices</Label>
<p className="text-muted-foreground text-xs">
{selectedDevices.length === 0
? "Auto — the first available device handles every transcode. Select devices to pin or balance."
: selectedDevices.length === 1
? "All transcodes run on the selected device."
: "Transcode sessions balance across the selected devices (least loaded first)."}
{hwDetection.data?.source === "transcode_node" &&
" Devices reported by a transcode node."}
</p>
{inventoriesDiverge && (
<p className="text-xs text-amber-500">
This setting applies to every transcode node, but the nodes report different
devices. Only paths present on all nodes are safe to select.
</p>
)}
</div>
<div className="space-y-2">
{deviceRows.map((row) => (
<div key={row.path} className="flex items-center justify-between gap-3">
<div className="min-w-0">
<p
className={`truncate text-sm ${row.detected ? "" : "text-muted-foreground"}`}
>
{row.description}
</p>
<p className="text-muted-foreground truncate font-mono text-xs">{row.path}</p>
{row.missingOnNodes.length > 0 && (
<p className="truncate text-xs text-amber-500">
Not present on: {row.missingOnNodes.join(", ")}
</p>
)}
</div>
<Switch
checked={selectedDevices.includes(row.path)}
onCheckedChange={() =>
form.setValue(
"playback.hw_device",
toggleHWDevice(
form.getValue("playback.hw_device"),
row.path,
detectedPaths,
),
)
}
/>
</div>
))}
</div>
</div>
)}
<SettingField
label="Transcoding Enabled"
type="toggle"
@@ -0,0 +1,150 @@
import { describe, expect, it } from "vitest";
import {
buildHWDeviceRows,
nodeInventoriesDiverge,
parseHWDeviceList,
toggleHWDevice,
} from "./playbackSettings.utils";
const DETECTED = ["/dev/dri/renderD128", "/dev/dri/renderD129"];
describe("parseHWDeviceList", () => {
it("returns empty for unset or blank values", () => {
expect(parseHWDeviceList(undefined)).toEqual([]);
expect(parseHWDeviceList("")).toEqual([]);
expect(parseHWDeviceList(" , ")).toEqual([]);
});
it("splits and trims a comma list", () => {
expect(parseHWDeviceList(" /dev/dri/renderD128 ,/dev/dri/renderD129")).toEqual(DETECTED);
});
});
describe("toggleHWDevice", () => {
it("adds a device to an empty selection", () => {
expect(toggleHWDevice("", "/dev/dri/renderD129", DETECTED)).toBe("/dev/dri/renderD129");
});
it("keeps detection order regardless of click order", () => {
const afterSecond = toggleHWDevice("", "/dev/dri/renderD129", DETECTED);
const afterBoth = toggleHWDevice(afterSecond, "/dev/dri/renderD128", DETECTED);
expect(afterBoth).toBe("/dev/dri/renderD128,/dev/dri/renderD129");
});
it("removes an already-selected device", () => {
expect(
toggleHWDevice("/dev/dri/renderD128,/dev/dri/renderD129", "/dev/dri/renderD128", DETECTED),
).toBe("/dev/dri/renderD129");
});
it("preserves selected devices missing from the current detection pass", () => {
expect(toggleHWDevice("/dev/dri/renderD200", "/dev/dri/renderD128", DETECTED)).toBe(
"/dev/dri/renderD128,/dev/dri/renderD200",
);
});
});
describe("buildHWDeviceRows", () => {
const detection = (overrides: object) =>
({
resolved: "qsv",
render_devices: DETECTED,
render_device_details: DETECTED.map((path) => ({ path, description: "Intel GPU" })),
intel_detected: true,
source: "local",
...overrides,
}) as never;
it("keeps configured devices visible when detection returns nothing", () => {
const rows = buildHWDeviceRows(
detection({ render_devices: [], render_device_details: [] }),
"/dev/dri/renderD128,/dev/dri/renderD129",
);
expect(rows).toHaveLength(2);
expect(rows.every((row) => !row.detected)).toBe(true);
expect(rows.map((row) => row.path)).toEqual(DETECTED);
});
it("keeps configured devices visible with no detection data at all", () => {
const rows = buildHWDeviceRows(undefined, "/dev/dri/renderD128");
expect(rows).toEqual([
{
path: "/dev/dri/renderD128",
description: "Configured device not detected",
detected: false,
missingOnNodes: [],
},
]);
});
it("falls back to render_devices paths when an older node omits details", () => {
const rows = buildHWDeviceRows(detection({ render_device_details: undefined }), "");
expect(rows).toHaveLength(2);
expect(rows[0]).toMatchObject({ path: DETECTED[0], description: "GPU", detected: true });
});
it("merges detected and configured-but-missing devices", () => {
const rows = buildHWDeviceRows(detection({}), "/dev/dri/renderD200");
expect(rows).toHaveLength(3);
expect(rows[2]).toMatchObject({ path: "/dev/dri/renderD200", detected: false });
});
it("marks devices missing from responding nodes", () => {
const rows = buildHWDeviceRows(
detection({
nodes: [
{ node_url: "http://a", node_name: "node-a", render_devices: DETECTED },
{ node_url: "http://b", node_name: "node-b", render_devices: [DETECTED[0]] },
{ node_url: "http://c", node_name: "node-down", error: "boom" },
],
}),
"",
);
expect(rows.map((row) => row.missingOnNodes)).toEqual([[], ["node-b"]]);
});
});
describe("nodeInventoriesDiverge", () => {
const base = {
resolved: "qsv",
render_devices: [],
intel_detected: false,
source: "local",
} as never;
it("is false without nodes or with one node", () => {
expect(nodeInventoriesDiverge(undefined)).toBe(false);
expect(nodeInventoriesDiverge({ ...(base as object), nodes: [] } as never)).toBe(false);
expect(
nodeInventoriesDiverge({
...(base as object),
nodes: [{ node_url: "http://a", render_devices: DETECTED }],
} as never),
).toBe(false);
});
it("is false when responding nodes match (order-insensitive), ignoring failed nodes", () => {
expect(
nodeInventoriesDiverge({
...(base as object),
nodes: [
{ node_url: "http://a", render_devices: [DETECTED[0], DETECTED[1]] },
{ node_url: "http://b", render_devices: [DETECTED[1], DETECTED[0]] },
{ node_url: "http://c", error: "boom" },
],
} as never),
).toBe(false);
});
it("is true when responding nodes report different devices", () => {
expect(
nodeInventoriesDiverge({
...(base as object),
nodes: [
{ node_url: "http://a", render_devices: DETECTED },
{ node_url: "http://b", render_devices: [DETECTED[0]] },
],
} as never),
).toBe(true);
});
});
@@ -0,0 +1,106 @@
import type { HWAccelInfo } from "@/hooks/queries/admin/system";
// Helpers for the playback.hw_device GPU picker. The setting stores a
// comma-separated render-device list; the UI presents it as per-device
// toggles, with no selection meaning "auto" (server picks the first
// available device). The setting is cluster-wide, so rows carry per-node
// presence info when transcode nodes report their inventories.
export function parseHWDeviceList(value: string | undefined): string[] {
if (!value) return [];
return value
.split(",")
.map((part) => part.trim())
.filter((part) => part.length > 0);
}
/**
* Toggles one device in the stored list, preserving the order devices are
* detected in so the stored value stays stable regardless of click order.
*/
export function toggleHWDevice(
value: string | undefined,
device: string,
detectedOrder: string[],
): string {
const selected = new Set(parseHWDeviceList(value));
if (selected.has(device)) {
selected.delete(device);
} else {
selected.add(device);
}
const ordered = detectedOrder.filter((path) => selected.has(path));
// Preserve selected devices the current detection pass doesn't list (e.g.
// a temporarily unplugged GPU) rather than silently dropping them.
for (const path of selected) {
if (!detectedOrder.includes(path)) ordered.push(path);
}
return ordered.join(",");
}
export interface HWDeviceRow {
path: string;
description: string;
/** Present in the primary detection result. */
detected: boolean;
/** Names/URLs of responding nodes whose inventory lacks this device. */
missingOnNodes: string[];
}
/**
* Builds the picker rows: the union of detected devices and configured
* entries, so configured-but-missing devices stay visible (and deselectable)
* even when detection returns nothing or an older node omits
* render_device_details.
*/
export function buildHWDeviceRows(
detection: HWAccelInfo | undefined,
configured: string | undefined,
): HWDeviceRow[] {
const detected = detectedDevices(detection);
const respondingNodes = (detection?.nodes ?? []).filter((node) => !node.error);
const missingOn = (path: string) =>
respondingNodes
.filter((node) => !(node.render_devices ?? []).includes(path))
.map((node) => node.node_name || node.node_url);
const rows: HWDeviceRow[] = detected.map((device) => ({
path: device.path,
description: device.description,
detected: true,
missingOnNodes: missingOn(device.path),
}));
for (const path of parseHWDeviceList(configured)) {
if (rows.some((row) => row.path === path)) continue;
rows.push({
path,
description: "Configured device not detected",
detected: false,
missingOnNodes: missingOn(path),
});
}
return rows;
}
/**
* True when more than one transcode node responded and their render-device
* inventories differ — the cluster-wide hw_device value is only safe for
* paths present on every node, so the UI shows a warning.
*/
export function nodeInventoriesDiverge(detection: HWAccelInfo | undefined): boolean {
const inventories = (detection?.nodes ?? [])
.filter((node) => !node.error)
.map((node) => [...(node.render_devices ?? [])].sort().join(","));
return inventories.length > 1 && new Set(inventories).size > 1;
}
function detectedDevices(
detection: HWAccelInfo | undefined,
): { path: string; description: string }[] {
if (!detection) return [];
if (detection.render_device_details && detection.render_device_details.length > 0) {
return detection.render_device_details;
}
// Older nodes report only render_devices paths.
return (detection.render_devices ?? []).map((path) => ({ path, description: "GPU" }));
}