#!/usr/bin/python3 """Pinned PeerTube runner 0.2.0 FFmpeg adapter. Never executes a CPU video encoder.""" import json import os import re import subprocess import sys import threading import time from pathlib import Path REAL = '/usr/bin/ffmpeg' PROBE = '/usr/bin/ffprobe' DEVICE = os.environ.get('PT_RENDER_DEVICE', '/dev/dri/renderD128') STATE = Path(os.environ.get('PT_STATE_DIR', '/var/lib/pt-worker/state')) def event(kind, **fields): STATE.mkdir(parents=True, exist_ok=True) data = (json.dumps({'time': time.time(), 'kind': kind, **fields}) + '\n').encode() fd = os.open(STATE / 'events.jsonl', os.O_CREAT | os.O_APPEND | os.O_WRONLY, 0o600) try: os.write(fd, data) finally: os.close(fd) def probe(path): p = subprocess.run([PROBE, '-v', 'error', '-show_streams', '-show_format', '-of', 'json', path], capture_output=True, text=True, timeout=30) if p.returncode: raise ValueError('input probe failed') return json.loads(p.stdout) def engines(pid): """i915 DRM fdinfo: cumulative hardware engine time, deduplicated by client id.""" result, clients = {}, set() for f in Path(f'/proc/{pid}/fdinfo').glob('*'): try: text = f.read_text() except OSError: continue cid = re.search(r'^drm-client-id:\s*(\d+)', text, re.M) if not cid or cid[1] in clients: continue clients.add(cid[1]) for key, value in re.findall(r'^drm-engine-([\w-]+):\s*(\d+)\s+ns', text, re.M): result[key] = result.get(key, 0) + int(value) return result def rewrite(args, input_probe=None): # fluent-ffmpeg queries capabilities before building commands. These do no encode. if args in (['-version'], ['-formats'], ['-encoders'], ['-filters'], ['-codecs'], ['-hwaccels']): return args, None pairs = list(zip(args, args[1:])) video_options = [(k, v) for k, v in pairs if k in ('-vcodec', '-c:v', '-codec:v')] if len(video_options) > 1: raise ValueError('multiple video codec options refused') if '-vn' in args and not video_options: return args, None # audio-only VOD renditions have no video encode if not video_options or video_options[0][1] != 'libx264': raise ValueError('unknown runner command or video encoder refused') if '-filter_complex' in args or '-hwaccel' in args or '-init_hw_device' in args: raise ValueError('unreviewed filter/device override refused') inputs = [v for k, v in pairs if k == '-i'] if not inputs: raise ValueError('missing video input') data = input_probe if input_probe is not None else probe(inputs[0]) stream = next((s for s in data['streams'] if s['codec_type'] == 'video'), None) if stream is None: raise ValueError('missing video stream') codec = stream['codec_name'] still = codec in ('png', 'mjpeg', 'webp') and '-loop' in args decoder = {'h264': 'h264_qsv', 'hevc': 'hevc_qsv', 'vp9': 'vp9_qsv', 'av1': 'av1_qsv'}.get(codec) if not decoder and not still: raise ValueError('source codec has no reviewed QSV decode path') if stream.get('pix_fmt') not in ('yuv420p', 'yuv420p10le', 'nv12', 'p010le') and not still: raise ValueError('unsupported pixel format; CPU decode fallback refused') if stream.get('tags', {}).get('rotate') or any(s.get('rotation', 0) for s in stream.get('side_data_list', [])): raise ValueError('rotated source requires reviewed GPU transpose path') width, height = stream['width'], stream['height'] duration = float(data.get('format', {}).get('duration', 0) or 0) w, h = width // 2 * 2, height // 2 * 2 output = ['-init_hw_device', f'vaapi=va:{DEVICE}', '-init_hw_device', 'qsv=qs@va', '-filter_hw_device', 'qs'] i, first_input = 0, True while i < len(args): key = args[i] val = args[i + 1] if i + 1 < len(args) else None if key == '-i' and first_input: if not still: output += ['-hwaccel', 'qsv', '-hwaccel_output_format', 'qsv', '-c:v', decoder, '-noautorotate'] output += [key, val] first_input = False i += 2 elif key in ('-vcodec', '-c:v', '-codec:v'): output += [key, 'h264_qsv'] i += 2 elif key in ('-pix_fmt', '-preset', '-preset:v', '-b_strategy', '-bf', '-tune', '-crf'): i += 2 elif key in ('-vf', '-filter:v'): if val == 'scale=trunc(iw/2)*2:trunc(ih/2)*2' and still: w, h = width // 2 * 2, height // 2 * 2 else: match = re.fullmatch(r'scale=w=(-2|\d+):h=(-2|\d+)', val or '') if not match: raise ValueError('unreviewed software filter refused') w, h = map(int, match.groups()) if w == -2 and h > 0: w = max(2, round(width * h / height / 2) * 2) elif h == -2 and w > 0: h = max(2, round(height * w / width / 2) * 2) if w <= 0 or h <= 0 or w % 2 or h % 2: raise ValueError('invalid output dimensions') prefix = 'format=nv12,hwupload=extra_hw_frames=64,' if still else '' output += [key, prefix + f'scale_qsv=w={w}:h={h}:format=nv12'] i += 2 else: output.append(key) i += 1 if '-vf' not in args and '-filter:v' not in args: prefix = 'format=nv12,hwupload=extra_hw_frames=64,' if still else '' output[-1:-1] = ['-vf', prefix + f'scale_qsv=w={width // 2 * 2}:h={height // 2 * 2}:format=nv12'] # Keep PeerTube's rate cap and GOP; QSV has no x264 CRF mode. rate = next((v for k, v in pairs if k in ('-maxrate:v', '-maxrate')), None) extra = ['-preset', 'medium', '-bf', '2'] if rate and not any(k in ('-b:v', '-vb') for k, _ in pairs): extra += ['-b:v', rate] output[-1:-1] = extra job_uuid = None try: job_uuid = json.loads((STATE / 'inputs' / (Path(inputs[0]).name + '.json')).read_text())['uuid'] except (OSError, ValueError, KeyError): pass return output, {'job_uuid': job_uuid, 'source_codec': codec, 'target_codec': 'h264', 'encoder': 'h264_qsv', 'source_width': width, 'source_height': height, 'duration': duration, 'target_width': w, 'target_height': h, 'static_image_upload': still} def main(): try: args, info = rewrite(sys.argv[1:]) if info is None: os.execv(REAL, [REAL, *args]) if not os.access(DEVICE, os.R_OK | os.W_OK): raise ValueError('render device unavailable') started = time.monotonic() p = subprocess.Popen([REAL, *args], stderr=subprocess.PIPE, text=True) progress = {} def stderr_progress(): for line in p.stderr: sys.stderr.write(line) sys.stderr.flush() for key, value in re.findall(r'(frame|fps|speed)=\s*([\d.]+)', line): progress[key] = float(value) reader = threading.Thread(target=stderr_progress, daemon=True) reader.start() maximum = {} while p.poll() is None: for key, value in engines(p.pid).items(): maximum[key] = max(maximum.get(key, 0), value) time.sleep(0.1) elapsed = time.monotonic() - started reader.join(timeout=2) event('encode', **info, seconds=elapsed, returncode=p.returncode, engine_ns=maximum, ffmpeg_fps=progress.get('fps'), frames=progress.get('frame'), ffmpeg_speed=progress.get('speed'), realtime_factor=info['duration'] / elapsed if info['duration'] else None) if p.returncode: event('gpu_failure', reason='qsv_transcode_failed') sys.exit(p.returncode) except (ValueError, OSError, subprocess.SubprocessError, KeyError) as e: # Messages are fixed strings; never expose source URLs or runner credentials. event('gpu_failure', reason=type(e).__name__) print('PT-WORKER: QSV adapter refused command; worker must drain', file=sys.stderr) sys.exit(78) if __name__ == '__main__': main()