397 lines
13 KiB
Rust
397 lines
13 KiB
Rust
//! Subprocess backend: spawn ffmpeg CLI for capture+encode.
|
|
//!
|
|
//! Spawns ffmpeg with the same hardware pipeline as `stream_av.sh`:
|
|
//! kmsgrab -vblank_source vsync → hwmap=derive_device=vaapi → scale_vaapi → h264_vaapi
|
|
//! + PulseAudio desktop audio + mic → amix → AAC
|
|
//!
|
|
//! -vblank_source vsync forces a frame grab on every display vblank, regardless
|
|
//! of page flips. Without it, kmsgrab only grabs when the compositor flips a
|
|
//! new buffer — a static/slow screen yields 1fps.
|
|
//!
|
|
//! ffmpeg outputs NUT format to stdout. We demux that pipe with ffmpeg-next
|
|
//! to get proper AVPackets (keyframe flags, timestamps) without parsing
|
|
//! bytestreams.
|
|
|
|
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, MediaType};
|
|
|
|
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.
|
|
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;
|
|
|
|
// Watch for stop flag on a separate thread and kill ffmpeg to unblock
|
|
// the packet iterator (which is a blocking read on the pipe fd).
|
|
let stop_watcher = stop.clone();
|
|
let child_pid = child.id();
|
|
std::thread::Builder::new()
|
|
.name("ffmpeg-stop-watcher".into())
|
|
.spawn(move || {
|
|
while !stop_watcher.load(Ordering::Relaxed) {
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
|
}
|
|
use nix::sys::signal::{kill, Signal};
|
|
use nix::unistd::Pid;
|
|
let _ = kill(Pid::from_raw(child_pid as i32), Signal::SIGINT);
|
|
info!("Stop watcher: sent SIGINT to ffmpeg pid={child_pid}");
|
|
})
|
|
.expect("spawn stop watcher");
|
|
|
|
let result = demux_and_send(fd, packet_tx, stop, &mut child);
|
|
|
|
// Clean up subprocess regardless of result.
|
|
kill_child(&mut child);
|
|
|
|
result
|
|
}
|
|
|
|
/// Detect PulseAudio audio sources for capture.
|
|
struct AudioSources {
|
|
monitor: Option<String>, // desktop audio (speaker tap)
|
|
mic: Option<String>, // microphone
|
|
pulse_server: String, // PULSE_SERVER env for root
|
|
}
|
|
|
|
fn detect_audio_sources() -> AudioSources {
|
|
// When running as root (sudo for kmsgrab), we need the real user's PulseAudio
|
|
let real_uid = std::env::var("SUDO_UID")
|
|
.unwrap_or_else(|_| unsafe { libc::getuid() }.to_string());
|
|
let pulse_server = format!("unix:/run/user/{real_uid}/pulse/native");
|
|
|
|
let monitor = detect_monitor_source(&pulse_server);
|
|
let mic = detect_default_source(&pulse_server);
|
|
|
|
// Don't use mic if it's the same as monitor (some systems set monitor as default)
|
|
let mic = match (&monitor, &mic) {
|
|
(Some(m), Some(d)) if m == d => None,
|
|
_ => mic,
|
|
};
|
|
|
|
info!("Audio sources — monitor: {:?}, mic: {:?}", monitor, mic);
|
|
AudioSources { monitor, mic, pulse_server }
|
|
}
|
|
|
|
fn detect_monitor_source(pulse_server: &str) -> Option<String> {
|
|
let output = Command::new("pactl")
|
|
.arg("info")
|
|
.env("PULSE_SERVER", pulse_server)
|
|
.output()
|
|
.ok()?;
|
|
let stdout = String::from_utf8_lossy(&output.stdout);
|
|
for line in stdout.lines() {
|
|
if line.contains("Default Sink:") {
|
|
let sink = line.split(':').nth(1)?.trim();
|
|
return Some(format!("{sink}.monitor"));
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
fn detect_default_source(pulse_server: &str) -> Option<String> {
|
|
let output = Command::new("pactl")
|
|
.args(["get-default-source"])
|
|
.env("PULSE_SERVER", pulse_server)
|
|
.output()
|
|
.ok()?;
|
|
let source = String::from_utf8_lossy(&output.stdout).trim().to_string();
|
|
if source.is_empty() { None } else { Some(source) }
|
|
}
|
|
|
|
fn spawn_ffmpeg(cfg: &SubprocessConfig) -> Result<Child> {
|
|
let audio = detect_audio_sources();
|
|
|
|
// fps filter after scale_vaapi pads/duplicates frames to fill gaps when
|
|
// kmsgrab captures fewer frames than the target rate (e.g. compositor
|
|
// skips flips on static content). Keeps the output stream at a stable fps.
|
|
let filter = format!(
|
|
"hwmap=derive_device=vaapi,scale_vaapi=w={}:h={}:format=nv12,fps={}",
|
|
cfg.width, cfg.height, cfg.fps,
|
|
);
|
|
|
|
let mut args: Vec<String> = vec![
|
|
// Hardware init
|
|
"-init_hw_device".into(), format!("drm=drm:{}", cfg.device),
|
|
"-init_hw_device".into(), "vaapi=va@drm".into(),
|
|
// Video input (kmsgrab)
|
|
// -vblank_source vsync: grab on every display vblank, not just page flips.
|
|
// Without this, a static screen (e.g. talking-head meeting) gives 1fps
|
|
// because the compositor rarely flips a new buffer.
|
|
"-thread_queue_size".into(), "512".into(),
|
|
"-device".into(), cfg.device.clone(),
|
|
"-f".into(), "kmsgrab".into(),
|
|
"-framerate".into(), cfg.fps.to_string(),
|
|
"-i".into(), "-".into(),
|
|
];
|
|
|
|
// Audio inputs
|
|
let has_monitor = audio.monitor.is_some();
|
|
let has_mic = audio.mic.is_some();
|
|
|
|
if let Some(ref monitor) = audio.monitor {
|
|
args.extend([
|
|
"-f".into(), "pulse".into(),
|
|
"-thread_queue_size".into(), "1024".into(),
|
|
"-i".into(), monitor.clone(),
|
|
]);
|
|
}
|
|
if let Some(ref mic) = audio.mic {
|
|
args.extend([
|
|
"-f".into(), "pulse".into(),
|
|
"-thread_queue_size".into(), "1024".into(),
|
|
"-i".into(), mic.clone(),
|
|
]);
|
|
}
|
|
|
|
// Audio filter: mix monitor + mic if both present
|
|
if has_monitor && has_mic {
|
|
args.extend([
|
|
"-filter_complex".into(),
|
|
"[1:a][2:a]amix=inputs=2:duration=longest[aout]".into(),
|
|
"-map".into(), "0:v".into(),
|
|
"-map".into(), "[aout]".into(),
|
|
]);
|
|
} else if has_monitor {
|
|
args.extend(["-map".into(), "0:v".into(), "-map".into(), "1:a".into()]);
|
|
}
|
|
// If no audio: no -map needed, only video output
|
|
|
|
// Video encoding
|
|
args.extend([
|
|
"-vf".into(), filter,
|
|
"-c:v".into(), "h264_vaapi".into(),
|
|
"-qp".into(), cfg.qp.to_string(),
|
|
"-g".into(), cfg.gop_size.to_string(),
|
|
"-bf".into(), "0".into(),
|
|
]);
|
|
|
|
// Audio encoding (if any audio source)
|
|
if has_monitor || has_mic {
|
|
args.extend([
|
|
"-c:a".into(), "aac".into(),
|
|
"-b:a".into(), "128k".into(),
|
|
]);
|
|
}
|
|
|
|
// Output
|
|
args.extend([
|
|
"-flush_packets".into(), "1".into(),
|
|
"-fflags".into(), "nobuffer".into(),
|
|
"-f".into(), "nut".into(),
|
|
"pipe:1".into(),
|
|
"-hide_banner".into(),
|
|
]);
|
|
|
|
info!("ffmpeg args: {:?}", args);
|
|
|
|
let child = Command::new("ffmpeg")
|
|
.args(&args)
|
|
.env("PULSE_SERVER", &audio.pulse_server)
|
|
.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")?;
|
|
|
|
// Find video stream
|
|
let video_stream = input_ctx
|
|
.streams()
|
|
.best(ffmpeg::media::Type::Video)
|
|
.context("no video stream in NUT output")?;
|
|
let video_idx = video_stream.index();
|
|
let video_tb = video_stream.time_base();
|
|
let video_tb_num = video_tb.numerator() as u32;
|
|
let video_tb_den = video_tb.denominator() as u32;
|
|
|
|
// Find audio stream (may not exist if no PulseAudio sources found)
|
|
let audio_info = input_ctx
|
|
.streams()
|
|
.best(ffmpeg::media::Type::Audio)
|
|
.map(|s| {
|
|
let tb = s.time_base();
|
|
(s.index(), tb.numerator() as u32, tb.denominator() as u32)
|
|
});
|
|
|
|
if let Some((idx, num, den)) = audio_info {
|
|
info!("Demux: video_idx={video_idx} tb={video_tb_num}/{video_tb_den}, \
|
|
audio_idx={idx} tb={num}/{den}");
|
|
} else {
|
|
info!("Demux: video_idx={video_idx} tb={video_tb_num}/{video_tb_den}, no audio");
|
|
}
|
|
|
|
let mut video_count = 0u64;
|
|
let mut audio_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;
|
|
}
|
|
|
|
let data = match packet.data() {
|
|
Some(d) => d.to_vec(),
|
|
None => continue,
|
|
};
|
|
|
|
let stream_idx = stream.index();
|
|
|
|
if stream_idx == video_idx {
|
|
let encoded = EncodedPacket {
|
|
media_type: MediaType::Video,
|
|
data,
|
|
pts: packet.pts().unwrap_or(0),
|
|
dts: packet.dts().unwrap_or(0),
|
|
keyframe: packet.is_key(),
|
|
time_base_num: video_tb_num,
|
|
time_base_den: video_tb_den,
|
|
};
|
|
video_count += 1;
|
|
if video_count % 300 == 1 {
|
|
info!("Subprocess: {video_count} video, {audio_count} audio packets");
|
|
}
|
|
if packet_tx.blocking_send(encoded).is_err() {
|
|
info!("Packet channel closed");
|
|
break;
|
|
}
|
|
} else if let Some((audio_idx, audio_tb_num, audio_tb_den)) = audio_info {
|
|
if stream_idx == audio_idx {
|
|
let encoded = EncodedPacket {
|
|
media_type: MediaType::Audio,
|
|
data,
|
|
pts: packet.pts().unwrap_or(0),
|
|
dts: packet.dts().unwrap_or(0),
|
|
keyframe: packet.is_key(),
|
|
time_base_num: audio_tb_num,
|
|
time_base_den: audio_tb_den,
|
|
};
|
|
audio_count += 1;
|
|
if packet_tx.blocking_send(encoded).is_err() {
|
|
info!("Packet channel closed");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
info!("Subprocess pipeline stopped ({video_count} video, {audio_count} audio 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();
|
|
}
|
|
|
|
// Wait up to 3 seconds, then SIGKILL.
|
|
for _ in 0..30 {
|
|
if child.try_wait().ok().flatten().is_some() {
|
|
info!("ffmpeg exited cleanly");
|
|
return;
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
|
}
|
|
warn!("ffmpeg didn't exit after SIGINT, killing");
|
|
child.kill().ok();
|
|
let _ = child.wait();
|
|
}
|