phase 3: subprocess backend, dual-backend pipeline, packets flowing

media/client/src/backends/subprocess.rs

@@ -0,0 +1,231 @@
+//! Subprocess backend: spawn ffmpeg CLI for capture+encode.
+//!
+//! Spawns ffmpeg with the same hardware pipeline as `stream_av.py`:
+//!   kmsgrab → hwmap=derive_device=vaapi → scale_vaapi → h264_vaapi
+//!
+//! ffmpeg outputs NUT format to stdout. We demux that pipe with ffmpeg-next
+//! to get proper AVPackets (keyframe flags, timestamps) without parsing
+//! bytestreams. NUT is lighter than mpegts — no TS overhead, exact packet
+//! metadata in the container layer.
+//!
+//! This approach works where the direct VAAPI API path fails: hwmap uses
+//! fftools' internal AVFilterGraph.hw_device_ctx (removed from public API
+//! in ffmpeg 7+), so X2RGB10LE format negotiation succeeds.
+
+use std::os::fd::AsRawFd;
+use std::os::unix::io::RawFd;
+use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use tracing::{error, info, warn};
+
+use crate::encoder::EncodedPacket;
+
+pub struct SubprocessConfig {
+    pub device: String,
+    pub fps: u32,
+    pub width: u32,
+    pub height: u32,
+    pub qp: u32,
+    pub gop_size: u32,
+}
+
+impl Default for SubprocessConfig {
+    fn default() -> Self {
+        Self {
+            device: "/dev/dri/card0".into(),
+            fps: 30,
+            width: 1920,
+            height: 1080,
+            qp: 20,
+            gop_size: 30,
+        }
+    }
+}
+
+/// Run the subprocess pipeline. Blocks until stop is set or ffmpeg exits.
+pub fn run(
+    config: SubprocessConfig,
+    packet_tx: tokio::sync::mpsc::Sender<EncodedPacket>,
+    stop: Arc<AtomicBool>,
+) -> Result<()> {
+    let mut child = spawn_ffmpeg(&config).context("spawn ffmpeg")?;
+    info!("ffmpeg subprocess pid={}", child.id());
+
+    // Drain stderr on a separate thread so ffmpeg doesn't block on a full pipe.
+    let stderr = child.stderr.take().expect("stderr piped");
+    let stop_on_fatal = stop.clone();
+    std::thread::Builder::new()
+        .name("ffmpeg-stderr".into())
+        .spawn(move || watch_stderr(stderr, stop_on_fatal))
+        .expect("spawn stderr thread");
+
+    // Get the raw fd from stdout before handing it to ffmpeg-next.
+    // ffmpeg-next takes ownership of the input context but we keep the Child
+    // alive so the fd stays valid.
+    let stdout = child.stdout.take().expect("stdout piped");
+    let fd: RawFd = stdout.as_raw_fd();
+
+    // Keep stdout alive for the duration of demuxing.
+    let _stdout_guard = stdout;
+
+    let result = demux_and_send(fd, packet_tx, stop, &mut child);
+
+    // Clean up subprocess regardless of result.
+    kill_child(&mut child);
+
+    result
+}
+
+fn spawn_ffmpeg(cfg: &SubprocessConfig) -> Result<Child> {
+    let filter = format!(
+        "hwmap=derive_device=vaapi,scale_vaapi=w={}:h={}:format=nv12,fps={}",
+        cfg.width, cfg.height, cfg.fps,
+    );
+
+    let child = Command::new("ffmpeg")
+        .args([
+            "-init_hw_device", &format!("drm=drm:{}", cfg.device),
+            "-init_hw_device", "vaapi=va@drm",
+            "-thread_queue_size", "64",
+            "-device", &cfg.device,
+            "-f", "kmsgrab",
+            "-framerate", &cfg.fps.to_string(),
+            "-i", "-",
+            "-vf", &filter,
+            "-c:v", "h264_vaapi",
+            "-qp", &cfg.qp.to_string(),
+            "-g", &cfg.gop_size.to_string(),
+            "-bf", "0",
+            "-flush_packets", "1",
+            "-fflags", "nobuffer",
+            "-f", "nut",
+            "pipe:1",
+            "-hide_banner",
+        ])
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .context("failed to spawn ffmpeg — is it in PATH?")?;
+
+    Ok(child)
+}
+
+fn demux_and_send(
+    fd: RawFd,
+    packet_tx: tokio::sync::mpsc::Sender<EncodedPacket>,
+    stop: Arc<AtomicBool>,
+    child: &mut Child,
+) -> Result<()> {
+    ffmpeg::init().context("ffmpeg init")?;
+
+    // Open the NUT stream from the pipe fd.
+    let pipe_url = format!("pipe:{fd}");
+    let mut input_ctx = ffmpeg::format::input(&pipe_url)
+        .context("open ffmpeg input from pipe")?;
+
+    let video_stream = input_ctx
+        .streams()
+        .best(ffmpeg::media::Type::Video)
+        .context("no video stream in NUT output")?;
+
+    let stream_idx = video_stream.index();
+    let time_base = video_stream.time_base();
+    let tb_num = time_base.numerator() as u32;
+    let tb_den = time_base.denominator() as u32;
+
+    info!(
+        "Subprocess demux ready: stream_idx={}, time_base={}/{}",
+        stream_idx, tb_num, tb_den
+    );
+
+    let mut packet_count = 0u64;
+
+    for (stream, packet) in input_ctx.packets() {
+        if stop.load(Ordering::Relaxed) {
+            break;
+        }
+
+        // ffmpeg process died
+        if let Some(status) = child.try_wait().ok().flatten() {
+            warn!("ffmpeg exited with {status}");
+            break;
+        }
+
+        if stream.index() != stream_idx {
+            continue;
+        }
+
+        let data = match packet.data() {
+            Some(d) => d.to_vec(),
+            None => continue,
+        };
+
+        let encoded = EncodedPacket {
+            data,
+            pts: packet.pts().unwrap_or(0),
+            dts: packet.dts().unwrap_or(0),
+            keyframe: packet.is_key(),
+            time_base_num: tb_num,
+            time_base_den: tb_den,
+        };
+
+        packet_count += 1;
+        if packet_count % 300 == 1 {
+            info!("Subprocess: {packet_count} packets encoded");
+        }
+
+        if packet_tx.blocking_send(encoded).is_err() {
+            info!("Packet channel closed, stopping subprocess pipeline");
+            break;
+        }
+    }
+
+    info!("Subprocess pipeline stopped ({packet_count} packets)");
+    Ok(())
+}
+
+fn watch_stderr(stderr: std::process::ChildStderr, stop: Arc<AtomicBool>) {
+    use std::io::{BufRead, BufReader};
+
+    const FATAL: &[&str] = &["framebuffer format changed", "Error during demuxing"];
+
+    for line in BufReader::new(stderr).lines() {
+        let line = match line {
+            Ok(l) => l,
+            Err(_) => break,
+        };
+        if !line.is_empty() {
+            info!("[ffmpeg] {line}");
+        }
+        if FATAL.iter().any(|p| line.contains(p)) {
+            error!("Fatal ffmpeg error — stopping: {line}");
+            stop.store(true, Ordering::Relaxed);
+        }
+    }
+    info!("[ffmpeg] stderr closed");
+}
+
+fn kill_child(child: &mut Child) {
+    use nix::sys::signal::{killpg, Signal};
+    use nix::unistd::Pid;
+
+    if child.try_wait().ok().flatten().is_some() {
+        return; // already exited
+    }
+
+    // Send SIGINT to the process group so ffmpeg can flush cleanly.
+    if let Ok(pgid) = nix::unistd::getpgid(Some(Pid::from_raw(child.id() as i32))) {
+        let _ = killpg(pgid, Signal::SIGINT);
+    } else {
+        child.kill().ok();
+    }
+
+    match child.wait() {
+        Ok(s) => info!("ffmpeg exited: {s}"),
+        Err(e) => warn!("ffmpeg wait error: {e}"),
+    }
+}