1use std::ffi::OsString;2use std::process::Stdio;34use anyhow::{Context, Result, bail, ensure};5use tokio::io::{AsyncRead, AsyncReadExt};6use tokio::process::Command;7use tokio::sync::{oneshot, watch};8use tracing::debug;910use crate::clock::SessionClock;11use crate::config::{HostConfig, VaapiRateControl};1213use super::annexb::AnnexBAccessUnitParser;14use super::{15 AccessUnitPublisher, CaptureVideoMode, EncodedFrame, EncodedGop, EncodedSource,16 GamescopePreflight, gamescope_config, interactive_gop_frames, log_stderr_chunks,17};1819pub(super) fn spawn_gamescope_pipewire_with_target(20 config: HostConfig,21 session_clock: SessionClock,22 preflight: GamescopePreflight,23) -> Result<EncodedSource> {24 spawn_gamescope_pipewire_with_target_and_shutdown(config, session_clock, preflight, None)25}2627pub(super) fn spawn_gamescope_pipewire_with_target_and_shutdown(28 config: HostConfig,29 session_clock: SessionClock,30 preflight: GamescopePreflight,31 shutdown: Option<oneshot::Receiver<()>>,32) -> Result<EncodedSource> {33 let (sender, receiver) = watch::channel(None);34 let (current_gop_sender, current_gop) = watch::channel(None);35 let target_object = preflight.target_object;36 let video_mode = preflight.video_mode;37 let task = tokio::spawn(async move {38 run_gamescope_pipewire(39 config,40 session_clock,41 target_object,42 video_mode,43 sender,44 current_gop_sender,45 shutdown,46 )47 .await48 });49 Ok(EncodedSource {50 frames: receiver,51 current_gop,52 task,53 pointer_surface_dimensions: Some(preflight.pointer_surface_dimensions),54 encoder_control: None,55 })56}5758async fn run_gamescope_pipewire(59 config: HostConfig,60 session_clock: SessionClock,61 target_object: String,62 video_mode: CaptureVideoMode,63 sender: watch::Sender<Option<EncodedFrame>>,64 current_gop_sender: watch::Sender<Option<EncodedGop>>,65 shutdown: Option<oneshot::Receiver<()>>,66) -> Result<()> {67 let pipewire = gamescope_config(&config)?;68 let args = gamescope_pipeline_args(&config, &target_object, video_mode)?;69 let mut command = Command::new(&pipewire.gst_launch_path);70 command71 .args(args)72 .env("LC_ALL", "C")73 // This is honored by the legacy VAAPI plugin and harmless for GstVA.74 // GstVA selection is independently checked against device-path above.75 .env("GST_VAAPI_DRM_DEVICE", &pipewire.vaapi_render_node)76 .stdin(Stdio::null())77 .stdout(Stdio::piped())78 .stderr(Stdio::piped())79 .kill_on_drop(true);8081 let mut child = command.spawn().with_context(|| {82 format!(83 "starting Gamescope pipeline with {}",84 pipewire.gst_launch_path.display()85 )86 })?;87 let mut stdout = child88 .stdout89 .take()90 .context("GStreamer stdout was not piped")?;91 let stderr = child92 .stderr93 .take()94 .context("GStreamer stderr was not piped")?;95 let stderr_task = tokio::spawn(log_stderr_chunks(stderr, "gstreamer"));9697 let result = {98 let forward = forward_annex_b_stream(99 session_clock,100 &mut stdout,101 &sender,102 ¤t_gop_sender,103 interactive_gop_frames(video_mode.framerate) as usize,104 video_mode.width,105 video_mode.height,106 );107 tokio::pin!(forward);108 match shutdown {109 Some(mut shutdown) => tokio::select! {110 _ = &mut shutdown => Ok(()),111 result = &mut forward => result,112 },113 None => forward.await,114 }115 };116 let _ = child.kill().await;117 let exit_status = child.wait().await.ok();118 stderr_task.abort();119120 match result {121 Ok(()) => Ok(()),122 Err(error) => {123 if let Some(status) = exit_status {124 Err(error).with_context(|| format!("GStreamer pipeline exited with {status}"))125 } else {126 Err(error)127 }128 }129 }130}131132pub(super) async fn forward_annex_b_stream<R>(133 session_clock: SessionClock,134 stdout: &mut R,135 sender: &watch::Sender<Option<EncodedFrame>>,136 current_gop_sender: &watch::Sender<Option<EncodedGop>>,137 max_gop_frames: usize,138 width: u16,139 height: u16,140) -> Result<()>141where142 R: AsyncRead + Unpin,143{144 let mut publisher = AccessUnitPublisher::new(145 session_clock,146 sender,147 current_gop_sender,148 max_gop_frames,149 width,150 height,151 );152 let mut parser = AnnexBAccessUnitParser::default();153 let mut chunk = [0_u8; 16 * 1024];154155 loop {156 let count = stdout157 .read(&mut chunk)158 .await159 .context("reading encoded source output")?;160 if count == 0 {161 bail!("encoded source ended unexpectedly");162 }163164 for access_unit in parser.push(&chunk[..count])? {165 publisher.publish(access_unit)?;166 if publisher.receivers_closed() {167 debug!("encoded source has no receivers");168 return Ok(());169 }170 }171 }172}173174pub(super) fn gamescope_pipeline_args(175 config: &HostConfig,176 target_object: &str,177 video_mode: CaptureVideoMode,178) -> Result<Vec<OsString>> {179 let pipewire = gamescope_config(config)?;180 ensure!(181 !target_object.is_empty()182 && target_object.len() <= 32183 && target_object.bytes().all(|byte| byte.is_ascii_digit()),184 "resolved PipeWire object.serial is invalid"185 );186 let caps = format!(187 "video/x-raw,format=NV12,width={},height={},framerate={}/1",188 video_mode.width, video_mode.height, video_mode.framerate189 );190 let encoded_caps = "video/x-h264,stream-format=byte-stream,alignment=au";191 let mut args = vec![192 "--quiet".to_owned(),193 "pipewiresrc".to_owned(),194 "do-timestamp=true".to_owned(),195 // Keep Gamescope's negotiated producer pool bounded at its preferred196 // four buffers, but copy each delivered frame into GStreamer's own197 // buffer before downstream conversion/encoding can retain it. This198 // returns producer buffers promptly instead of starving Gamescope at199 // vblank. `use-bufferpool=false` is the supported replacement for the200 // deprecated `always-copy=true` spelling.201 "min-buffers=1".to_owned(),202 "max-buffers=4".to_owned(),203 "use-bufferpool=false".to_owned(),204 format!("target-object={target_object}"),205 "!".to_owned(),206 "queue".to_owned(),207 "max-size-buffers=1".to_owned(),208 "max-size-bytes=0".to_owned(),209 "max-size-time=0".to_owned(),210 "leaky=downstream".to_owned(),211 "!".to_owned(),212 "videorate".to_owned(),213 // Gamescope is damage-driven. Never manufacture replacement frames:214 // doing so can turn an irregular live source into stale catch-up work.215 "drop-only=true".to_owned(),216 format!("max-rate={}", video_mode.framerate),217 "!".to_owned(),218 "videoconvert".to_owned(),219 "!".to_owned(),220 "videoscale".to_owned(),221 "!".to_owned(),222 caps,223 "!".to_owned(),224 pipewire.vaapi_encoder.clone(),225 ];226 args.push(format!(227 "rate-control={}",228 match pipewire.rate_control {229 VaapiRateControl::Cbr => "cbr",230 VaapiRateControl::Cqp => "cqp",231 }232 ));233 match pipewire.rate_control {234 VaapiRateControl::Cbr => args.push(format!(235 "bitrate={}",236 pipewire237 .bitrate_kbps238 .context("CBR bitrate is missing after validation")?239 )),240 VaapiRateControl::Cqp => {241 let quantizer = pipewire242 .quantizer243 .context("CQP quantizer is missing after validation")?;244 args.push(format!("qpi={quantizer}"));245 args.push(format!("qpp={quantizer}"));246 }247 }248 args.extend([249 "target-usage=7".to_owned(),250 format!(251 "key-int-max={}",252 interactive_gop_frames(video_mode.framerate)253 ),254 "b-frames=0".to_owned(),255 "ref-frames=1".to_owned(),256 "aud=true".to_owned(),257 "!".to_owned(),258 "h264parse".to_owned(),259 "config-interval=-1".to_owned(),260 "!".to_owned(),261 encoded_caps.to_owned(),262 "!".to_owned(),263 "fdsink".to_owned(),264 "fd=1".to_owned(),265 // Gamescope's PipeWire timestamps can advance in larger jumps than266 // the negotiated frame duration. Clocking this terminal sink makes it267 // wait on those producer timestamps while the one-frame leaky queue268 // drops fresh motion. Videorate already caps the live stream at the269 // configured maximum, so drain encoded access units immediately.270 "sync=false".to_owned(),271 ]);272 Ok(args.into_iter().map(OsString::from).collect())273}274275276mod tests {277 use super::super::tests::{configured_video_mode, gamescope_config};278 use super::*;279280 281 fn builds_bounded_low_latency_gamescope_pipeline() {282 let config = gamescope_config();283 let args: Vec<String> = gamescope_pipeline_args(&config, "1234", configured_video_mode())284 .unwrap()285 .into_iter()286 .map(|arg| arg.into_string().unwrap())287 .collect();288289 assert_eq!(args[0], "--quiet");290 assert!(291 args.windows(2)292 .any(|args| { args == ["pipewiresrc".to_owned(), "do-timestamp=true".to_owned()] })293 );294 assert!(args.contains(&"target-object=1234".to_owned()));295 assert!(args.contains(&"min-buffers=1".to_owned()));296 assert!(args.contains(&"max-buffers=4".to_owned()));297 assert!(args.contains(&"use-bufferpool=false".to_owned()));298 assert!(args.contains(&"max-size-buffers=1".to_owned()));299 assert!(args.contains(&"leaky=downstream".to_owned()));300 assert!(301 args.windows(2)302 .any(|args| { args == ["videorate".to_owned(), "drop-only=true".to_owned()] })303 );304 assert!(305 args.contains(306 &"video/x-raw,format=NV12,width=1280,height=800,framerate=60/1".to_owned()307 )308 );309 assert!(args.contains(&"vah264enc".to_owned()));310 assert!(args.contains(&"bitrate=12000".to_owned()));311 assert!(args.contains(&"rate-control=cbr".to_owned()));312 assert!(args.contains(&"target-usage=7".to_owned()));313 assert!(args.contains(&"key-int-max=30".to_owned()));314 assert!(args.contains(&"b-frames=0".to_owned()));315 assert!(args.contains(&"aud=true".to_owned()));316 assert!(args.contains(&"config-interval=-1".to_owned()));317 assert!(args.contains(&"video/x-h264,stream-format=byte-stream,alignment=au".to_owned()));318 assert!(args.ends_with(&[319 "fdsink".to_owned(),320 "fd=1".to_owned(),321 "sync=false".to_owned()322 ]));323 let videorate = args.iter().position(|arg| arg == "videorate").unwrap();324 let queue = args.iter().position(|arg| arg == "queue").unwrap();325 let videoconvert = args.iter().position(|arg| arg == "videoconvert").unwrap();326 let videoscale = args.iter().position(|arg| arg == "videoscale").unwrap();327 assert!(queue < videorate && videorate < videoconvert && videoconvert < videoscale);328 assert!(gamescope_pipeline_args(&config, "gamescope", configured_video_mode()).is_err());329 }330331 332 fn builds_pipeline_for_resolved_non_fixture_mode() {333 let config = gamescope_config();334 let mode = CaptureVideoMode {335 width: 1_920,336 height: 1_080,337 framerate: 72,338 };339 let args: Vec<String> = gamescope_pipeline_args(&config, "1234", mode)340 .unwrap()341 .into_iter()342 .map(|arg| arg.into_string().unwrap())343 .collect();344345 assert!(346 args.contains(347 &"video/x-raw,format=NV12,width=1920,height=1080,framerate=72/1".to_owned()348 )349 );350 assert!(args.contains(&"max-rate=72".to_owned()));351 assert!(args.contains(&"key-int-max=36".to_owned()));352 }353354 355 fn builds_explicit_low_power_cqp_pipeline() {356 let mut config = gamescope_config();357 let pipewire = config.gamescope_pipewire.as_mut().unwrap();358 pipewire.vaapi_encoder = "vah264lpenc".into();359 pipewire.rate_control = VaapiRateControl::Cqp;360 pipewire.bitrate_kbps = None;361 pipewire.quantizer = Some(24);362 config.validate().unwrap();363364 let args: Vec<String> = gamescope_pipeline_args(&config, "1234", configured_video_mode())365 .unwrap()366 .into_iter()367 .map(|arg| arg.into_string().unwrap())368 .collect();369 assert!(args.contains(&"vah264lpenc".to_owned()));370 assert!(args.contains(&"rate-control=cqp".to_owned()));371 assert!(args.contains(&"qpi=24".to_owned()));372 assert!(args.contains(&"qpp=24".to_owned()));373 assert!(!args.iter().any(|arg| arg.starts_with("bitrate=")));374 }375}