1use std::sync::Arc;2use std::sync::atomic::{AtomicU64, Ordering};3use std::time::Duration;45use std::time::Instant;67use anyhow::{Context, Result, bail, ensure};8use tokio::sync::watch;91011use crate::clock::SessionClock;1213use crate::config::{HostConfig, VaapiRateControl};1415use sigil_protocol::PointerSurfaceDimensions;161718use super::{19 AccessUnitPublisher, CaptureVideoMode, EncodedFrame, EncodedGop, EncodedSource,20 GamescopePreflight, MAX_ENCODE_BUFFER, gamescope_config, has_h264_codec_config,21 interactive_gop_frames, is_h264_keyframe,22};232425use gstreamer::prelude::ObjectExt;2627use tracing::{debug, warn};282930const RESOLUTION_APPLY_TIMEOUT: Duration = Duration::from_secs(2);3132const MAX_ENCODER_BITRATE_KBPS: u32 = 100_000;333435struct RevisedBitrate {36 revision: u64,37 kbps: u32,38}394041struct RevisedResolution {42 revision: u64,43 width: u16,44 height: u16,45}464748struct EncoderControlDesired {49 revision: u64,50 bitrate: Option<RevisedBitrate>,51 resolution: Option<RevisedResolution>,52 force_keyframe_revision: Option<u64>,53}545556pub struct EncoderControlStatus {57 pub latest_revision: u64,58 pub applied_bitrate_revision: Option<u64>,59 pub applied_bitrate_kbps: Option<u32>,60 pub requested_resolution_revision: Option<u64>,61 pub applied_resolution_revision: Option<u64>,62 pub applied_width: Option<u16>,63 pub applied_height: Option<u16>,64 pub requested_force_keyframe_revision: Option<u64>,65 pub acknowledged_force_keyframe_revision: Option<u64>,66}6768/// Bounded latest-state control for an in-process encoder.69///70/// Both directions are Tokio watch channels: callers can update control state71/// arbitrarily often without building a queue, while the encoder applies only72/// the latest revision. Revisions are process-monotonic and never wrap.7374pub struct EncoderControl {75 initial_width: u16,76 initial_height: u16,77 next_revision: Arc<AtomicU64>,78 desired: watch::Sender<EncoderControlDesired>,79 status: watch::Receiver<EncoderControlStatus>,80}818283848586impl EncoderControl {87 fn new(88 initial_width: u16,89 initial_height: u16,90 ) -> (91 Self,92 watch::Receiver<EncoderControlDesired>,93 watch::Sender<EncoderControlStatus>,94 ) {95 let (desired, desired_rx) = watch::channel(EncoderControlDesired::default());96 let (status_tx, status) = watch::channel(EncoderControlStatus::default());97 (98 Self {99 initial_width,100 initial_height,101 next_revision: Arc::new(AtomicU64::new(0)),102 desired,103 status,104 },105 desired_rx,106 status_tx,107 )108 }109110 pub(crate) fn initial_dimensions(&self) -> (u16, u16) {111 (self.initial_width, self.initial_height)112 }113114 fn next_revision(&self) -> Result<u64> {115 self.next_revision116 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |revision| {117 revision.checked_add(1)118 })119 .map(|revision| revision + 1)120 .map_err(|_| anyhow::anyhow!("encoder control revision exhausted"))121 }122123 pub fn request_bitrate_kbps(&self, kbps: u32) -> Result<u64> {124 ensure!(125 (1..=MAX_ENCODER_BITRATE_KBPS).contains(&kbps),126 "encoder bitrate must be between 1 and {MAX_ENCODER_BITRATE_KBPS} kbps"127 );128 let revision = self.next_revision()?;129 self.desired.send_modify(|desired| {130 desired.revision = revision;131 desired.bitrate = Some(RevisedBitrate { revision, kbps });132 });133 Ok(revision)134 }135136 pub fn request_force_keyframe(&self) -> Result<u64> {137 let revision = self.next_revision()?;138 self.desired.send_modify(|desired| {139 desired.revision = revision;140 desired.force_keyframe_revision = Some(revision);141 });142 Ok(revision)143 }144145 pub fn request_resolution(&self, width: u16, height: u16) -> Result<u64> {146 ensure!(147 width >= 64 && height >= 64,148 "encoder dimensions must be at least 64x64"149 );150 ensure!(151 width.is_multiple_of(2) && height.is_multiple_of(2),152 "H.264 encoder dimensions must be even"153 );154 let revision = self.next_revision()?;155 self.desired.send_modify(|desired| {156 desired.revision = revision;157 desired.resolution = Some(RevisedResolution {158 revision,159 width,160 height,161 });162 });163 Ok(revision)164 }165166 pub fn status(&self) -> EncoderControlStatus {167 *self.status.borrow()168 }169170 /// Wait until the encoder confirms the exact bitrate revision and exposes171 /// its property readback. A newer applied revision means this request was172 /// coalesced before application and therefore fails rather than pretending173 /// the requested revision committed.174 pub async fn wait_for_bitrate_applied(175 &self,176 revision: u64,177 timeout: Duration,178 ) -> Result<EncoderControlStatus> {179 let mut status = self.status.clone();180 let wait = async {181 loop {182 let snapshot = *status.borrow_and_update();183 match snapshot.applied_bitrate_revision {184 Some(applied) if applied == revision => return Ok(snapshot),185 Some(applied) if applied > revision => bail!(186 "encoder bitrate revision {revision} was superseded by revision {applied}"187 ),188 _ => {}189 }190 status191 .changed()192 .await193 .context("encoder control status closed before bitrate was applied")?;194 }195 };196 tokio::time::timeout(timeout, wait)197 .await198 .with_context(|| format!("timed out waiting for encoder bitrate revision {revision}"))?199 }200201 /// Wait for a forced-keyframe request to be committed by observing a202 /// keyframe that also carries codec configuration.203 pub async fn wait_for_force_keyframe_acknowledged(204 &self,205 revision: u64,206 timeout: Duration,207 ) -> Result<EncoderControlStatus> {208 let mut status = self.status.clone();209 let wait = async {210 loop {211 let snapshot = *status.borrow_and_update();212 match snapshot.acknowledged_force_keyframe_revision {213 Some(acknowledged) if acknowledged == revision => return Ok(snapshot),214 Some(acknowledged) if acknowledged > revision => bail!(215 "force-keyframe revision {revision} was superseded by revision {acknowledged}"216 ),217 _ => {}218 }219 if snapshot220 .requested_force_keyframe_revision221 .is_some_and(|requested| requested > revision)222 {223 bail!("force-keyframe revision {revision} was coalesced before application");224 }225 status.changed().await.context(226 "encoder control status closed before forced keyframe was acknowledged",227 )?;228 }229 };230 tokio::time::timeout(timeout, wait)231 .await232 .with_context(|| format!("timed out waiting for force-keyframe revision {revision}"))?233 }234235 /// Wait for recovery to reach any configured IDR at or after `revision`.236 ///237 /// Recovery requests are deliberately coalesced with later encoder238 /// controls. Unlike transactional adaptive updates, a newer configured IDR239 /// still satisfies the decoder barrier established by an older request.240 pub async fn wait_for_recovery_keyframe_acknowledged(241 &self,242 revision: u64,243 timeout: Duration,244 ) -> Result<EncoderControlStatus> {245 let mut status = self.status.clone();246 let wait = async {247 loop {248 let snapshot = *status.borrow_and_update();249 if snapshot250 .acknowledged_force_keyframe_revision251 .is_some_and(|acknowledged| acknowledged >= revision)252 {253 return Ok(snapshot);254 }255 status.changed().await.context(256 "encoder control status closed before recovery keyframe was acknowledged",257 )?;258 }259 };260 tokio::time::timeout(timeout, wait).await.with_context(|| {261 format!("timed out waiting for recovery keyframe revision {revision}")262 })?263 }264265 /// Wait until the exact target dimensions emerge from the encoder on an266 /// independently decodable access unit. A property write or caps event is267 /// not an application acknowledgement.268 pub async fn wait_for_resolution_applied(269 &self,270 revision: u64,271 width: u16,272 height: u16,273 timeout: Duration,274 ) -> Result<EncoderControlStatus> {275 let mut status = self.status.clone();276 let wait = async {277 loop {278 let snapshot = *status.borrow_and_update();279 match snapshot.applied_resolution_revision {280 Some(applied) if applied == revision => {281 ensure!(282 snapshot.applied_width == Some(width)283 && snapshot.applied_height == Some(height),284 "encoder resolution revision {revision} acknowledged {:?}x{:?}, expected {width}x{height}",285 snapshot.applied_width,286 snapshot.applied_height287 );288 return Ok(snapshot);289 }290 Some(applied) if applied > revision => bail!(291 "encoder resolution revision {revision} was superseded by revision {applied}"292 ),293 _ => {}294 }295 if snapshot296 .requested_resolution_revision297 .is_some_and(|requested| requested > revision)298 {299 bail!(300 "encoder resolution revision {revision} was coalesced before application"301 );302 }303 status304 .changed()305 .await306 .context("encoder control status closed before resolution was applied")?;307 }308 };309 tokio::time::timeout(timeout, wait).await.with_context(|| {310 format!("timed out waiting for encoder resolution revision {revision}")311 })?312 }313}314315316pub(crate) struct EncoderControlTestHarness {317 pub control: EncoderControl,318 pub status: watch::Sender<EncoderControlStatus>,319 _desired: watch::Receiver<EncoderControlDesired>,320}321322323impl EncoderControlTestHarness {324 pub(crate) fn new() -> Self {325 Self::with_dimensions(1_280, 800)326 }327328 pub(crate) fn with_dimensions(width: u16, height: u16) -> Self {329 let (control, desired, status) = EncoderControl::new(width, height);330 Self {331 control,332 status,333 _desired: desired,334 }335 }336337 pub(crate) fn requested_force_keyframe_revision(&self) -> Option<u64> {338 self._desired.borrow().force_keyframe_revision339 }340341 pub(crate) fn requested_bitrate(&self) -> Option<(u64, u32)> {342 self._desired343 .borrow()344 .bitrate345 .map(|bitrate| (bitrate.revision, bitrate.kbps))346 }347348 pub(crate) fn requested_resolution(&self) -> Option<(u64, u16, u16)> {349 self._desired350 .borrow()351 .resolution352 .map(|resolution| (resolution.revision, resolution.width, resolution.height))353 }354}355356357pub(super) fn spawn_gamescope_pipewire_in_process_with_target(358 config: HostConfig,359 session_clock: SessionClock,360 preflight: GamescopePreflight,361) -> Result<EncodedSource> {362 let description = gamescope_in_process_pipeline_description(363 &config,364 &preflight.target_object,365 preflight.video_mode,366 )?;367 let max_gop_frames = interactive_gop_frames(preflight.video_mode.framerate) as usize;368 let expected_device_path = Some(369 gamescope_config(&config)?370 .vaapi_render_node371 .to_string_lossy()372 .into_owned(),373 );374 spawn_in_process_pipeline(375 description,376 session_clock,377 max_gop_frames,378 Some(preflight.pointer_surface_dimensions),379 expected_device_path,380 preflight.video_mode.width,381 preflight.video_mode.height,382 )383}384385386fn spawn_in_process_pipeline(387 description: String,388 session_clock: SessionClock,389 max_gop_frames: usize,390 pointer_surface_dimensions: Option<PointerSurfaceDimensions>,391 expected_device_path: Option<String>,392 initial_width: u16,393 initial_height: u16,394) -> Result<EncodedSource> {395 let (sender, receiver) = watch::channel(None);396 let (current_gop_sender, current_gop) = watch::channel(None);397 let (control, desired, status) = EncoderControl::new(initial_width, initial_height);398 let task = tokio::spawn(async move {399 tokio::task::spawn_blocking(move || {400 run_in_process_pipeline(401 &description,402 session_clock,403 sender,404 current_gop_sender,405 max_gop_frames,406 desired,407 status,408 expected_device_path.as_deref(),409 initial_width,410 initial_height,411 )412 })413 .await414 .context("in-process GStreamer worker panicked")?415 });416 Ok(EncodedSource {417 frames: receiver,418 current_gop,419 task,420 pointer_surface_dimensions,421 encoder_control: Some(control),422 })423}424425426fn gamescope_in_process_pipeline_description(427 config: &HostConfig,428 target_object: &str,429 video_mode: CaptureVideoMode,430) -> Result<String> {431 let pipewire = gamescope_config(config)?;432 validate_pipewire_target_object(target_object)?;433 ensure!(434 pipewire.rate_control == VaapiRateControl::Cbr,435 "in-process GStreamer requires CBR so bitrate can be changed while playing"436 );437 let bitrate = pipewire438 .bitrate_kbps439 .context("CBR bitrate is missing after validation")?;440 let raw_caps = format!(441 "video/x-raw,format=NV12,width={},height={},framerate={}/1",442 video_mode.width, video_mode.height, video_mode.framerate443 );444 Ok(format!(445 "pipewiresrc do-timestamp=true min-buffers=1 max-buffers=4 use-bufferpool=false target-object={target_object} \446 ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream \447 ! videorate drop-only=true max-rate={} \448 ! videoconvert ! videoscale \449 ! capsfilter name=sigil_scale_caps caps={raw_caps} caps-change-mode=delayed \450 ! {} name=sigil_encoder rate-control=cbr bitrate={bitrate} target-usage=7 key-int-max={} b-frames=0 ref-frames=1 aud=true \451 ! h264parse config-interval=-1 \452 ! video/x-h264,stream-format=byte-stream,alignment=au \453 ! appsink name=sigil_sink max-buffers=1 drop=false sync=false",454 video_mode.framerate,455 pipewire.vaapi_encoder,456 interactive_gop_frames(video_mode.framerate),457 ))458}459460461fn validate_pipewire_target_object(target_object: &str) -> Result<()> {462 ensure!(463 !target_object.is_empty()464 && target_object.len() <= 32465 && target_object.bytes().all(|byte| byte.is_ascii_digit()),466 "resolved PipeWire object.serial is invalid"467 );468 Ok(())469}470471472473struct PendingResolutionTransition {474 target: RevisedResolution,475 deadline: Instant,476}477478479fn classify_resolution_sample(480 current: (u16, u16),481 pending: Option<PendingResolutionTransition>,482 sample: (u16, u16),483 configured_idr: bool,484) -> Result<Option<bool>> {485 if let Some(pending) = pending {486 if sample != (pending.target.width, pending.target.height) || !configured_idr {487 return Ok(None);488 }489 return Ok(Some(true));490 }491 ensure!(492 sample == current,493 "encoder changed resolution from {}x{} to {}x{} without a pending control revision",494 current.0,495 current.1,496 sample.0,497 sample.1498 );499 Ok(Some(false))500}501502503fn run_in_process_pipeline(504 description: &str,505 session_clock: SessionClock,506 sender: watch::Sender<Option<EncodedFrame>>,507 current_gop_sender: watch::Sender<Option<EncodedGop>>,508 max_gop_frames: usize,509 mut desired: watch::Receiver<EncoderControlDesired>,510 status: watch::Sender<EncoderControlStatus>,511 expected_device_path: Option<&str>,512 initial_width: u16,513 initial_height: u16,514) -> Result<()> {515 use gstreamer as gst;516 use gstreamer::prelude::*;517 use gstreamer_app as gst_app;518 gst::init().context("initializing in-process GStreamer")?;519 let pipeline = gst::parse::launch(description)520 .context("parsing in-process GStreamer pipeline")?521 .downcast::<gst::Pipeline>()522 .map_err(|_| {523 anyhow::anyhow!("in-process GStreamer description did not create a pipeline")524 })?;525 let encoder = pipeline526 .by_name("sigil_encoder")527 .context("in-process GStreamer pipeline has no named encoder")?;528 let scale_caps = pipeline529 .by_name("sigil_scale_caps")530 .context("in-process GStreamer pipeline has no named scale capsfilter")?;531 let sink = pipeline532 .by_name("sigil_sink")533 .context("in-process GStreamer pipeline has no named appsink")?534 .downcast::<gst_app::AppSink>()535 .map_err(|_| anyhow::anyhow!("named in-process sink is not an appsink"))?;536 sink.set_max_buffers(1);537 // Encoded dependent AUs must not be dropped here: the raw queue before the538 // encoder is the only latest-frame boundary. This one-buffer sink applies539 // bounded backpressure until the publishing loop consumes the AU.540 sink.set_drop(false);541 sink.set_sync(false);542 validate_mutable_playing_bitrate(&encoder)?;543 validate_resolution_capsfilter(&scale_caps)?;544545 pipeline546 .set_state(gst::State::Playing)547 .context("starting in-process GStreamer pipeline")?;548 let result = (|| {549 let (state_change, state, pending) = pipeline.state(gst::ClockTime::from_seconds(5));550 state_change.context("in-process GStreamer pipeline failed while waiting for PLAYING")?;551 ensure!(552 state == gst::State::Playing,553 "in-process GStreamer pipeline did not reach PLAYING within 5 seconds (state {state:?}, pending {pending:?})"554 );555 if let Some(expected_device_path) = expected_device_path {556 validate_active_encoder_device_path(&encoder, expected_device_path)?;557 }558 let bus = pipeline559 .bus()560 .context("in-process GStreamer pipeline has no bus")?;561 let mut publisher = AccessUnitPublisher::new(562 session_clock,563 &sender,564 ¤t_gop_sender,565 max_gop_frames,566 initial_width,567 initial_height,568 );569 let mut applied_desired_revision = 0_u64;570 let mut pending_force_keyframe_revision = None;571 let mut pending_resolution = None;572 status.send_modify(|status| {573 status.applied_width = Some(initial_width);574 status.applied_height = Some(initial_height);575 });576577 loop {578 if let Some(message) = bus.timed_pop(gst::ClockTime::ZERO) {579 match message.view() {580 gst::MessageView::Error(error) => {581 bail!(582 "in-process GStreamer pipeline error from {:?}: {} ({:?})",583 error.src().map(|source| source.path_string()),584 error.error(),585 error.debug()586 );587 }588 gst::MessageView::Eos(_) => {589 bail!("in-process GStreamer pipeline ended unexpectedly");590 }591 _ => {}592 }593 }594595 let desired_state = *desired.borrow_and_update();596 if desired_state.revision > applied_desired_revision {597 apply_encoder_control(598 &encoder,599 &scale_caps,600 &sink,601 desired_state,602 &status,603 ¤t_gop_sender,604 &mut pending_force_keyframe_revision,605 &mut pending_resolution,606 )?;607 applied_desired_revision = desired_state.revision;608 }609610 if pending_resolution.is_some_and(|pending: PendingResolutionTransition| {611 Instant::now() >= pending.deadline612 }) {613 let pending = pending_resolution.expect("checked as present");614 bail!(615 "encoder resolution revision {} did not produce a configured {}x{} IDR within {:?}",616 pending.target.revision,617 pending.target.width,618 pending.target.height,619 RESOLUTION_APPLY_TIMEOUT620 );621 }622623 if let Some(sample) = sink.try_pull_sample(gst::ClockTime::from_mseconds(50)) {624 let (width, height) = encoded_sample_dimensions(&sample)?;625 let buffer = sample626 .buffer()627 .context("in-process appsink sample has no buffer")?;628 let map = buffer629 .map_readable()630 .context("mapping in-process encoded access unit")?;631 ensure!(632 map.as_slice().len() <= MAX_ENCODE_BUFFER,633 "in-process encoded access unit exceeds {MAX_ENCODE_BUFFER} bytes"634 );635 let access_unit = map.as_slice();636 let configured_idr =637 is_h264_keyframe(access_unit) && has_h264_codec_config(access_unit);638 let Some(discontinuity) = classify_resolution_sample(639 (publisher.width, publisher.height),640 pending_resolution,641 (width, height),642 configured_idr,643 )?644 else {645 continue;646 };647 let configured_idr = publisher.publish_with_metadata(648 access_unit.to_vec(),649 width,650 height,651 discontinuity,652 )?;653 if discontinuity {654 let applied = pending_resolution655 .take()656 .expect("discontinuity requires a pending resolution");657 status.send_modify(|status| {658 status.applied_resolution_revision = Some(applied.target.revision);659 status.applied_width = Some(width);660 status.applied_height = Some(height);661 });662 }663 acknowledge_force_keyframe_if_configured_idr(664 configured_idr,665 &mut pending_force_keyframe_revision,666 &status,667 );668 } else if sink.is_eos() {669 bail!("in-process GStreamer appsink reached EOS unexpectedly");670 }671672 if publisher.receivers_closed() {673 debug!("in-process encoded source has no receivers");674 return Ok(());675 }676 }677 })();678 if let Err(error) = pipeline.set_state(gst::State::Null) {679 warn!(%error, "failed to stop in-process GStreamer pipeline cleanly");680 }681 result682}683684685fn validate_active_encoder_device_path(686 encoder: &gstreamer::Element,687 expected_device_path: &str,688) -> Result<()> {689 use gstreamer as gst;690 use gstreamer::prelude::*;691692 let property = encoder693 .find_property("device-path")694 .context("configured in-process VA encoder has no device-path property")?;695 ensure!(696 property.flags().contains(gst::glib::ParamFlags::READABLE),697 "configured in-process VA encoder device-path property is not readable"698 );699 let observed_device_path = encoder.property::<Option<String>>("device-path");700 ensure!(701 observed_device_path.as_deref() == Some(expected_device_path),702 "configured in-process VA encoder uses device-path {observed_device_path:?}, expected {expected_device_path:?}"703 );704 Ok(())705}706707708fn validate_mutable_playing_bitrate(encoder: &gstreamer::Element) -> Result<()> {709 use gstreamer as gst;710 use gstreamer::prelude::*;711712 let property = encoder713 .find_property("bitrate")714 .context("configured in-process encoder has no bitrate property")?;715 ensure!(716 property.flags().contains(gst::glib::ParamFlags::WRITABLE),717 "configured in-process encoder bitrate property is not writable"718 );719 ensure!(720 property.flags().contains(gst::PARAM_FLAG_MUTABLE_PLAYING),721 "configured in-process encoder bitrate property is not mutable while playing"722 );723 Ok(())724}725726727fn validate_resolution_capsfilter(capsfilter: &gstreamer::Element) -> Result<()> {728 use gstreamer::prelude::*;729730 let property = capsfilter731 .find_property("caps")732 .context("configured scale capsfilter has no caps property")?;733 ensure!(734 property735 .flags()736 .contains(gstreamer::glib::ParamFlags::WRITABLE),737 "configured scale capsfilter caps property is not writable"738 );739 let caps = capsfilter.property::<gstreamer::Caps>("caps");740 let _ = dimensions_from_caps(&caps)?;741 Ok(())742}743744745fn dimensions_from_caps(caps: &gstreamer::CapsRef) -> Result<(u16, u16)> {746 let structure = caps747 .structure(0)748 .context("configured video caps have no structure")?;749 let width = structure750 .get::<i32>("width")751 .context("configured video caps have no integer width")?;752 let height = structure753 .get::<i32>("height")754 .context("configured video caps have no integer height")?;755 Ok((756 u16::try_from(width).context("configured video caps width is outside u16")?,757 u16::try_from(height).context("configured video caps height is outside u16")?,758 ))759}760761762fn encoded_sample_dimensions(sample: &gstreamer::Sample) -> Result<(u16, u16)> {763 let caps = sample764 .caps()765 .context("in-process encoded sample has no negotiated caps")?;766 dimensions_from_caps(caps)767}768769770fn apply_resolution_caps(771 capsfilter: &gstreamer::Element,772 resolution: RevisedResolution,773) -> Result<()> {774 use gstreamer::prelude::*;775776 let mut caps = capsfilter.property::<gstreamer::Caps>("caps");777 let caps_mut = caps.make_mut();778 let structure = caps_mut779 .structure_mut(0)780 .context("configured scale capsfilter has no caps structure")?;781 structure.set("width", i32::from(resolution.width));782 structure.set("height", i32::from(resolution.height));783 capsfilter.set_property("caps", caps);784 let readback = capsfilter.property::<gstreamer::Caps>("caps");785 ensure!(786 dimensions_from_caps(&readback)? == (resolution.width, resolution.height),787 "scale capsfilter resolution readback does not match requested {}x{}",788 resolution.width,789 resolution.height790 );791 Ok(())792}793794795fn push_configured_keyframe_request(sink: &gstreamer_app::AppSink) -> Result<()> {796 use gstreamer::prelude::*;797798 let event = gstreamer_video::UpstreamForceKeyUnitEvent::builder()799 .all_headers(true)800 .build();801 let sink_pad = sink802 .static_pad("sink")803 .context("configured in-process appsink has no sink pad")?;804 ensure!(805 sink_pad.push_event(event),806 "in-process encoder rejected upstream ForceKeyUnit"807 );808 Ok(())809}810811812fn apply_encoder_control(813 encoder: &gstreamer::Element,814 scale_caps: &gstreamer::Element,815 sink: &gstreamer_app::AppSink,816 desired: EncoderControlDesired,817 status: &watch::Sender<EncoderControlStatus>,818 current_gop_sender: &watch::Sender<Option<EncodedGop>>,819 pending_force_keyframe_revision: &mut Option<u64>,820 pending_resolution: &mut Option<PendingResolutionTransition>,821) -> Result<()> {822 let current = *status.borrow();823 if let Some(bitrate) = desired.bitrate824 && current825 .applied_bitrate_revision826 .is_none_or(|revision| bitrate.revision > revision)827 {828 encoder.set_property("bitrate", bitrate.kbps);829 let readback = encoder.property::<u32>("bitrate");830 ensure!(831 readback == bitrate.kbps,832 "encoder bitrate readback mismatch: requested {} kbps, observed {readback} kbps",833 bitrate.kbps834 );835 status.send_modify(|status| {836 status.applied_bitrate_revision = Some(bitrate.revision);837 status.applied_bitrate_kbps = Some(readback);838 });839 }840841 let mut keyframe_requested = false;842 if let Some(resolution) = desired.resolution843 && current844 .requested_resolution_revision845 .is_none_or(|requested| resolution.revision > requested)846 {847 apply_resolution_caps(scale_caps, resolution)?;848 current_gop_sender.send_replace(None);849 push_configured_keyframe_request(sink)?;850 keyframe_requested = true;851 *pending_resolution = Some(PendingResolutionTransition {852 target: resolution,853 deadline: Instant::now() + RESOLUTION_APPLY_TIMEOUT,854 });855 status.send_modify(|status| {856 status.requested_resolution_revision = Some(resolution.revision);857 });858 }859860 if let Some(revision) = desired.force_keyframe_revision861 && current862 .requested_force_keyframe_revision863 .is_none_or(|requested| revision > requested)864 {865 if !keyframe_requested {866 push_configured_keyframe_request(sink)?;867 }868 *pending_force_keyframe_revision = Some(revision);869 status.send_modify(|status| {870 status.requested_force_keyframe_revision = Some(revision);871 });872 }873874 status.send_modify(|status| status.latest_revision = desired.revision);875 Ok(())876}877878879fn acknowledge_force_keyframe_if_configured_idr(880 configured_idr: bool,881 pending_revision: &mut Option<u64>,882 status: &watch::Sender<EncoderControlStatus>,883) {884 if configured_idr && let Some(revision) = pending_revision.take() {885 status.send_modify(|status| {886 status.acknowledged_force_keyframe_revision = Some(revision);887 });888 }889}890891892mod tests {893 use super::*;894895 896 fn encoder_control_retains_initial_session_dimensions() {897 let (native, _desired, _status) = EncoderControl::new(2_560, 1_600);898 let (other, _desired, _status) = EncoderControl::new(1_366, 768);899900 assert_eq!(native.initial_dimensions(), (2_560, 1_600));901 assert_eq!(other.initial_dimensions(), (1_366, 768));902 native.request_resolution(1_280, 800).unwrap();903 assert_eq!(native.initial_dimensions(), (2_560, 1_600));904 assert_eq!(other.initial_dimensions(), (1_366, 768));905 }906907 908 async fn encoder_control_coalesces_latest_state_and_acknowledges_only_configured_idr() {909 let (control, mut desired, status) = EncoderControl::new(1_280, 800);910911 let superseded_bitrate_revision = control.request_bitrate_kbps(4_000).unwrap();912 let bitrate_revision = control.request_bitrate_kbps(8_000).unwrap();913 let superseded_resolution_revision = control.request_resolution(960, 600).unwrap();914 let resolution_revision = control.request_resolution(640, 400).unwrap();915 let force_keyframe_revision = control.request_force_keyframe().unwrap();916917 assert!(superseded_bitrate_revision < bitrate_revision);918 assert!(bitrate_revision < superseded_resolution_revision);919 assert!(superseded_resolution_revision < resolution_revision);920 assert!(resolution_revision < force_keyframe_revision);921 let latest = *desired.borrow_and_update();922 assert_eq!(latest.revision, force_keyframe_revision);923 assert_eq!(924 latest.bitrate,925 Some(RevisedBitrate {926 revision: bitrate_revision,927 kbps: 8_000,928 })929 );930 assert_eq!(931 latest.resolution,932 Some(RevisedResolution {933 revision: resolution_revision,934 width: 640,935 height: 400,936 })937 );938 assert_eq!(939 latest.force_keyframe_revision,940 Some(force_keyframe_revision)941 );942943 status.send_modify(|status| {944 status.latest_revision = latest.revision;945 status.applied_bitrate_revision = Some(bitrate_revision);946 status.applied_bitrate_kbps = Some(8_000);947 status.requested_resolution_revision = Some(resolution_revision);948 status.applied_resolution_revision = Some(resolution_revision);949 status.applied_width = Some(640);950 status.applied_height = Some(400);951 status.requested_force_keyframe_revision = Some(force_keyframe_revision);952 });953 let mut pending = Some(force_keyframe_revision);954 acknowledge_force_keyframe_if_configured_idr(false, &mut pending, &status);955 assert_eq!(pending, Some(force_keyframe_revision));956 assert_eq!(control.status().acknowledged_force_keyframe_revision, None);957958 acknowledge_force_keyframe_if_configured_idr(true, &mut pending, &status);959 assert_eq!(pending, None);960 let applied = control961 .wait_for_bitrate_applied(bitrate_revision, Duration::from_millis(10))962 .await963 .unwrap();964 assert_eq!(applied.applied_bitrate_kbps, Some(8_000));965 let applied_resolution = control966 .wait_for_resolution_applied(resolution_revision, 640, 400, Duration::from_millis(10))967 .await968 .unwrap();969 assert_eq!(applied_resolution.applied_width, Some(640));970 assert_eq!(applied_resolution.applied_height, Some(400));971 assert!(972 control973 .wait_for_resolution_applied(974 superseded_resolution_revision,975 960,976 600,977 Duration::from_millis(10),978 )979 .await980 .unwrap_err()981 .to_string()982 .contains("superseded")983 );984 let acknowledged = control985 .wait_for_force_keyframe_acknowledged(986 force_keyframe_revision,987 Duration::from_millis(10),988 )989 .await990 .unwrap();991 assert_eq!(992 acknowledged.acknowledged_force_keyframe_revision,993 Some(force_keyframe_revision)994 );995 assert!(control.request_bitrate_kbps(0).is_err());996 assert!(control.request_resolution(0, 400).is_err());997 assert!(control.request_resolution(641, 400).is_err());998 assert!(999 control1000 .request_bitrate_kbps(MAX_ENCODER_BITRATE_KBPS + 1)1001 .is_err()1002 );10031004 let (closed_control, _desired, closed_status) = EncoderControl::new(1_280, 800);1005 let closed_revision = closed_control.request_force_keyframe().unwrap();1006 drop(closed_status);1007 assert!(1008 closed_control1009 .wait_for_force_keyframe_acknowledged(closed_revision, Duration::from_millis(10))1010 .await1011 .unwrap_err()1012 .to_string()1013 .contains("status closed")1014 );1015 }10161017 1018 async fn recovery_waiter_accepts_a_newer_configured_idr_revision() {1019 let (control, _desired, status) = EncoderControl::new(1_280, 800);1020 let recovery_revision = control.request_force_keyframe().unwrap();1021 let newer_revision = control.request_force_keyframe().unwrap();1022 assert!(newer_revision > recovery_revision);1023 status.send_modify(|status| {1024 status.requested_force_keyframe_revision = Some(newer_revision);1025 status.acknowledged_force_keyframe_revision = Some(newer_revision);1026 });10271028 let acknowledged = control1029 .wait_for_recovery_keyframe_acknowledged(recovery_revision, Duration::from_millis(10))1030 .await1031 .unwrap();1032 assert_eq!(1033 acknowledged.acknowledged_force_keyframe_revision,1034 Some(newer_revision)1035 );1036 assert!(1037 control1038 .wait_for_force_keyframe_acknowledged(recovery_revision, Duration::from_millis(10),)1039 .await1040 .unwrap_err()1041 .to_string()1042 .contains("superseded")1043 );1044 }10451046 1047 fn resolution_transition_suppresses_until_exact_configured_target_idr() {1048 let transition = PendingResolutionTransition {1049 target: RevisedResolution {1050 revision: 7,1051 width: 960,1052 height: 600,1053 },1054 deadline: Instant::now() + Duration::from_secs(1),1055 };1056 assert!(transition.deadline > Instant::now());1057 assert_eq!(1058 classify_resolution_sample((1280, 800), Some(transition), (1280, 800), true).unwrap(),1059 None1060 );1061 assert_eq!(1062 classify_resolution_sample((1280, 800), Some(transition), (960, 600), false).unwrap(),1063 None1064 );1065 assert_eq!(1066 classify_resolution_sample((1280, 800), Some(transition), (960, 600), true).unwrap(),1067 Some(true)1068 );1069 assert_eq!(1070 classify_resolution_sample((960, 600), None, (960, 600), false).unwrap(),1071 Some(false)1072 );1073 assert!(classify_resolution_sample((960, 600), None, (1280, 800), true).is_err());1074 }10751076 1077 1078 fn in_process_pipeline_drops_only_before_encoding() {1079 use super::super::tests::{configured_video_mode, gamescope_config};1080 use crate::config::GamescopeEncoderBackend;10811082 let mut config = gamescope_config();1083 config.gamescope_pipewire.as_mut().unwrap().encoder_backend =1084 GamescopeEncoderBackend::InProcessGstreamer;1085 let description =1086 gamescope_in_process_pipeline_description(&config, "1234", configured_video_mode())1087 .unwrap();10881089 assert!(description.contains(1090 "queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream"1091 ));1092 assert!(description.contains(1093 "capsfilter name=sigil_scale_caps caps=video/x-raw,format=NV12,width=1280,height=800,framerate=60/1 caps-change-mode=delayed"1094 ));1095 assert!(1096 description.contains("appsink name=sigil_sink max-buffers=1 drop=false sync=false")1097 );1098 assert!(!description.contains("appsink name=sigil_sink max-buffers=1 drop=true"));1099 }11001101 1102 1103 1104 async fn in_process_gstreamer_x264_smoke() {1105 let description = "videotestsrc is-live=true pattern=ball \1106 ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream \1107 ! videoconvert ! videoscale \1108 ! capsfilter name=sigil_scale_caps caps=video/x-raw,width=320,height=180,framerate=30/1 caps-change-mode=delayed \1109 ! x264enc name=sigil_encoder bitrate=2000 tune=zerolatency speed-preset=ultrafast key-int-max=15 bframes=0 byte-stream=true aud=true \1110 ! h264parse config-interval=-1 \1111 ! video/x-h264,stream-format=byte-stream,alignment=au \1112 ! appsink name=sigil_sink max-buffers=1 drop=false sync=false"1113 .to_owned();1114 let mut source =1115 spawn_in_process_pipeline(description, SessionClock::start(), 15, None, None, 320, 180)1116 .unwrap();1117 let control = source1118 .encoder_control1119 .clone()1120 .expect("x264 source has no encoder control");11211122 tokio::time::timeout(Duration::from_secs(10), async {1123 loop {1124 source.frames.changed().await.unwrap();1125 if source1126 .frames1127 .borrow()1128 .as_ref()1129 .is_some_and(|frame| frame.keyframe && frame.codec_config)1130 {1131 break;1132 }1133 }1134 })1135 .await1136 .expect("x264 did not publish an initial configured IDR");11371138 let bitrate_revision = control.request_bitrate_kbps(1_500).unwrap();1139 let force_keyframe_revision = control.request_force_keyframe().unwrap();1140 let bitrate_status = control1141 .wait_for_bitrate_applied(bitrate_revision, Duration::from_secs(10))1142 .await1143 .expect("x264 did not apply the bitrate revision");1144 assert_eq!(bitrate_status.applied_bitrate_kbps, Some(1_500));1145 control1146 .wait_for_force_keyframe_acknowledged(force_keyframe_revision, Duration::from_secs(10))1147 .await1148 .expect("x264 did not acknowledge a configured forced IDR");1149 let before_resolution_sequence = source1150 .current_gop1151 .borrow()1152 .as_ref()1153 .and_then(|gop| gop.frames.first())1154 .map(|frame| frame.sequence)1155 .expect("x264 has no configured GOP before resolution transition");11561157 let reduced_revision = control.request_resolution(256, 144).unwrap();1158 control1159 .wait_for_resolution_applied(reduced_revision, 256, 144, Duration::from_secs(10))1160 .await1161 .expect("x264 did not apply the reduced resolution");1162 let reduced = source1163 .current_gop1164 .borrow()1165 .as_ref()1166 .and_then(|gop| gop.frames.first())1167 .cloned()1168 .expect("x264 has no reduced configured GOP");1169 assert_eq!((reduced.width, reduced.height), (256, 144));1170 assert!(reduced.keyframe && reduced.codec_config && reduced.discontinuity);1171 assert!(reduced.sequence > before_resolution_sequence);11721173 let restored_revision = control.request_resolution(320, 180).unwrap();1174 control1175 .wait_for_resolution_applied(restored_revision, 320, 180, Duration::from_secs(10))1176 .await1177 .expect("x264 did not restore the native resolution");1178 let restored = source1179 .current_gop1180 .borrow()1181 .as_ref()1182 .and_then(|gop| gop.frames.first())1183 .cloned()1184 .expect("x264 has no restored configured GOP");1185 assert_eq!((restored.width, restored.height), (320, 180));1186 assert!(restored.keyframe && restored.codec_config && restored.discontinuity);1187 assert!(restored.sequence > reduced.sequence);11881189 {1190 let current_gop = source.current_gop.borrow();1191 let gop = current_gop1192 .as_ref()1193 .expect("x264 did not retain a current GOP");1194 assert!(gop.frames[0].keyframe && gop.frames[0].codec_config);1195 assert!(gop.frames.len() <= 15);1196 }1197 let EncodedSource {1198 frames,1199 current_gop,1200 task,1201 ..1202 } = source;1203 drop(frames);1204 drop(current_gop);1205 tokio::time::timeout(Duration::from_secs(2), task)1206 .await1207 .expect("x264 pipeline did not stop after its receivers closed")1208 .expect("x264 pipeline task panicked")1209 .expect("x264 pipeline returned an error");1210 }1211}