- commit
- cd3aee492d5f6ca82a1cac56dd421a2a0a724f64
- parent
- 146879330b40d81715503ef66b3e6217a36c8bdc
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 1172 | +----------------------------- |
| A | crates/ | 1195 | ++++++++++++++++++++++++++++++ |
@@ -1,7 +1,6 @@
11 use std::borrow::Cow;22 use std::process::{ExitStatus, Stdio};33 use std::sync::Arc;4-use std::sync::atomic::{AtomicU64, Ordering};54 use std::time::{Duration, Instant};65 76 use anyhow::{Context, Result, bail, ensure};@@ -10,8 +9,6 @@ use sigil_protocol::PointerSurfaceDimensions;
109 use tokio::io::{AsyncRead, AsyncReadExt};1110 use tokio::process::Command;1211 use tokio::sync::watch;13-14-use tracing::debug;1512 use tracing::warn;1613 1714 use crate::clock::SessionClock;@@ -21,333 +18,25 @@ use crate::config::{
2118 2219 mod annexb;2320 mod external;21+mod in_process;2422 mod preflight;2523 2624 2725 use external::gamescope_pipeline_args;2826 use external::{forward_annex_b_stream, spawn_gamescope_pipewire_with_target};27+28+pub(crate) use in_process::EncoderControlTestHarness;29+30+use in_process::spawn_gamescope_pipewire_in_process_with_target;31+pub use in_process::{EncoderControl, EncoderControlStatus};2932 use preflight::preflight_gamescope_static;3033 pub(crate) use preflight::{preflight_gstreamer_element, validate_executable};3134 32-33-use gstreamer::prelude::ObjectExt;34-3535 const MAX_ENCODE_BUFFER: usize = 16 * 1024 * 1024;3636 const MAX_CURRENT_GOP_BYTES: usize = 32 * 1024 * 1024;3737 const MAX_PW_DUMP_OUTPUT: usize = 8 * 1024 * 1024;3838 const MAX_DIAGNOSTIC_OUTPUT: usize = 512 * 1024;3939 const PREFLIGHT_TIMEOUT: Duration = Duration::from_secs(5);40-const MAX_ENCODER_BITRATE_KBPS: u32 = 100_000;41-42-const RESOLUTION_APPLY_TIMEOUT: Duration = Duration::from_secs(2);43-44-45-struct RevisedBitrate {46- revision: u64,47- kbps: u32,48-}49-50-51-struct RevisedResolution {52- revision: u64,53- width: u16,54- height: u16,55-}56-57-58-struct EncoderControlDesired {59- revision: u64,60- bitrate: Option<RevisedBitrate>,61- resolution: Option<RevisedResolution>,62- force_keyframe_revision: Option<u64>,63-}64-65-66-pub struct EncoderControlStatus {67- pub latest_revision: u64,68- pub applied_bitrate_revision: Option<u64>,69- pub applied_bitrate_kbps: Option<u32>,70- pub requested_resolution_revision: Option<u64>,71- pub applied_resolution_revision: Option<u64>,72- pub applied_width: Option<u16>,73- pub applied_height: Option<u16>,74- pub requested_force_keyframe_revision: Option<u64>,75- pub acknowledged_force_keyframe_revision: Option<u64>,76-}77-78-/// Bounded latest-state control for an in-process encoder.79-///80-/// Both directions are Tokio watch channels: callers can update control state81-/// arbitrarily often without building a queue, while the encoder applies only82-/// the latest revision. Revisions are process-monotonic and never wrap.83-84-pub struct EncoderControl {85- initial_width: u16,86- initial_height: u16,87- next_revision: Arc<AtomicU64>,88- desired: watch::Sender<EncoderControlDesired>,89- status: watch::Receiver<EncoderControlStatus>,90-}91-92-93-94-95-96-impl EncoderControl {97- fn new(98- initial_width: u16,99- initial_height: u16,100- ) -> (101- Self,102- watch::Receiver<EncoderControlDesired>,103- watch::Sender<EncoderControlStatus>,104- ) {105- let (desired, desired_rx) = watch::channel(EncoderControlDesired::default());106- let (status_tx, status) = watch::channel(EncoderControlStatus::default());107- (108- Self {109- initial_width,110- initial_height,111- next_revision: Arc::new(AtomicU64::new(0)),112- desired,113- status,114- },115- desired_rx,116- status_tx,117- )118- }119-120- pub(crate) fn initial_dimensions(&self) -> (u16, u16) {121- (self.initial_width, self.initial_height)122- }123-124- fn next_revision(&self) -> Result<u64> {125- self.next_revision126- .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |revision| {127- revision.checked_add(1)128- })129- .map(|revision| revision + 1)130- .map_err(|_| anyhow::anyhow!("encoder control revision exhausted"))131- }132-133- pub fn request_bitrate_kbps(&self, kbps: u32) -> Result<u64> {134- ensure!(135- (1..=MAX_ENCODER_BITRATE_KBPS).contains(&kbps),136- "encoder bitrate must be between 1 and {MAX_ENCODER_BITRATE_KBPS} kbps"137- );138- let revision = self.next_revision()?;139- self.desired.send_modify(|desired| {140- desired.revision = revision;141- desired.bitrate = Some(RevisedBitrate { revision, kbps });142- });143- Ok(revision)144- }145-146- pub fn request_force_keyframe(&self) -> Result<u64> {147- let revision = self.next_revision()?;148- self.desired.send_modify(|desired| {149- desired.revision = revision;150- desired.force_keyframe_revision = Some(revision);151- });152- Ok(revision)153- }154-155- pub fn request_resolution(&self, width: u16, height: u16) -> Result<u64> {156- ensure!(157- width >= 64 && height >= 64,158- "encoder dimensions must be at least 64x64"159- );160- ensure!(161- width.is_multiple_of(2) && height.is_multiple_of(2),162- "H.264 encoder dimensions must be even"163- );164- let revision = self.next_revision()?;165- self.desired.send_modify(|desired| {166- desired.revision = revision;167- desired.resolution = Some(RevisedResolution {168- revision,169- width,170- height,171- });172- });173- Ok(revision)174- }175-176- pub fn status(&self) -> EncoderControlStatus {177- *self.status.borrow()178- }179-180- /// Wait until the encoder confirms the exact bitrate revision and exposes181- /// its property readback. A newer applied revision means this request was182- /// coalesced before application and therefore fails rather than pretending183- /// the requested revision committed.184- pub async fn wait_for_bitrate_applied(185- &self,186- revision: u64,187- timeout: Duration,188- ) -> Result<EncoderControlStatus> {189- let mut status = self.status.clone();190- let wait = async {191- loop {192- let snapshot = *status.borrow_and_update();193- match snapshot.applied_bitrate_revision {194- Some(applied) if applied == revision => return Ok(snapshot),195- Some(applied) if applied > revision => bail!(196- "encoder bitrate revision {revision} was superseded by revision {applied}"197- ),198- _ => {}199- }200- status201- .changed()202- .await203- .context("encoder control status closed before bitrate was applied")?;204- }205- };206- tokio::time::timeout(timeout, wait)207- .await208- .with_context(|| format!("timed out waiting for encoder bitrate revision {revision}"))?209- }210-211- /// Wait for a forced-keyframe request to be committed by observing a212- /// keyframe that also carries codec configuration.213- pub async fn wait_for_force_keyframe_acknowledged(214- &self,215- revision: u64,216- timeout: Duration,217- ) -> Result<EncoderControlStatus> {218- let mut status = self.status.clone();219- let wait = async {220- loop {221- let snapshot = *status.borrow_and_update();222- match snapshot.acknowledged_force_keyframe_revision {223- Some(acknowledged) if acknowledged == revision => return Ok(snapshot),224- Some(acknowledged) if acknowledged > revision => bail!(225- "force-keyframe revision {revision} was superseded by revision {acknowledged}"226- ),227- _ => {}228- }229- if snapshot230- .requested_force_keyframe_revision231- .is_some_and(|requested| requested > revision)232- {233- bail!("force-keyframe revision {revision} was coalesced before application");234- }235- status.changed().await.context(236- "encoder control status closed before forced keyframe was acknowledged",237- )?;238- }239- };240- tokio::time::timeout(timeout, wait)241- .await242- .with_context(|| format!("timed out waiting for force-keyframe revision {revision}"))?243- }244-245- /// Wait for recovery to reach any configured IDR at or after `revision`.246- ///247- /// Recovery requests are deliberately coalesced with later encoder248- /// controls. Unlike transactional adaptive updates, a newer configured IDR249- /// still satisfies the decoder barrier established by an older request.250- pub async fn wait_for_recovery_keyframe_acknowledged(251- &self,252- revision: u64,253- timeout: Duration,254- ) -> Result<EncoderControlStatus> {255- let mut status = self.status.clone();256- let wait = async {257- loop {258- let snapshot = *status.borrow_and_update();259- if snapshot260- .acknowledged_force_keyframe_revision261- .is_some_and(|acknowledged| acknowledged >= revision)262- {263- return Ok(snapshot);264- }265- status.changed().await.context(266- "encoder control status closed before recovery keyframe was acknowledged",267- )?;268- }269- };270- tokio::time::timeout(timeout, wait).await.with_context(|| {271- format!("timed out waiting for recovery keyframe revision {revision}")272- })?273- }274-275- /// Wait until the exact target dimensions emerge from the encoder on an276- /// independently decodable access unit. A property write or caps event is277- /// not an application acknowledgement.278- pub async fn wait_for_resolution_applied(279- &self,280- revision: u64,281- width: u16,282- height: u16,283- timeout: Duration,284- ) -> Result<EncoderControlStatus> {285- let mut status = self.status.clone();286- let wait = async {287- loop {288- let snapshot = *status.borrow_and_update();289- match snapshot.applied_resolution_revision {290- Some(applied) if applied == revision => {291- ensure!(292- snapshot.applied_width == Some(width)293- && snapshot.applied_height == Some(height),294- "encoder resolution revision {revision} acknowledged {:?}x{:?}, expected {width}x{height}",295- snapshot.applied_width,296- snapshot.applied_height297- );298- return Ok(snapshot);299- }300- Some(applied) if applied > revision => bail!(301- "encoder resolution revision {revision} was superseded by revision {applied}"302- ),303- _ => {}304- }305- if snapshot306- .requested_resolution_revision307- .is_some_and(|requested| requested > revision)308- {309- bail!(310- "encoder resolution revision {revision} was coalesced before application"311- );312- }313- status314- .changed()315- .await316- .context("encoder control status closed before resolution was applied")?;317- }318- };319- tokio::time::timeout(timeout, wait).await.with_context(|| {320- format!("timed out waiting for encoder resolution revision {revision}")321- })?322- }323-}324-325-326-pub(crate) struct EncoderControlTestHarness {327- pub control: EncoderControl,328- pub status: watch::Sender<EncoderControlStatus>,329- _desired: watch::Receiver<EncoderControlDesired>,330-}331-332-333-impl EncoderControlTestHarness {334- pub(crate) fn new() -> Self {335- Self::with_dimensions(1_280, 800)336- }337-338- pub(crate) fn with_dimensions(width: u16, height: u16) -> Self {339- let (control, desired, status) = EncoderControl::new(width, height);340- Self {341- control,342- status,343- _desired: desired,344- }345- }346-347- pub(crate) fn requested_force_keyframe_revision(&self) -> Option<u64> {348- self._desired.borrow().force_keyframe_revision349- }350-}35140 35241 35342 pub struct EncodedFrame {@@ -514,121 +203,6 @@ fn spawn_gamescope_pipewire_with_preflight(
514203 }515204 }516205 517-518-fn spawn_gamescope_pipewire_in_process_with_target(519- config: HostConfig,520- session_clock: SessionClock,521- preflight: GamescopePreflight,522-) -> Result<EncodedSource> {523- let description = gamescope_in_process_pipeline_description(524- &config,525- &preflight.target_object,526- preflight.video_mode,527- )?;528- let max_gop_frames = interactive_gop_frames(preflight.video_mode.framerate) as usize;529- let expected_device_path = Some(530- gamescope_config(&config)?531- .vaapi_render_node532- .to_string_lossy()533- .into_owned(),534- );535- spawn_in_process_pipeline(536- description,537- session_clock,538- max_gop_frames,539- Some(preflight.pointer_surface_dimensions),540- expected_device_path,541- preflight.video_mode.width,542- preflight.video_mode.height,543- )544-}545-546-547-fn spawn_in_process_pipeline(548- description: String,549- session_clock: SessionClock,550- max_gop_frames: usize,551- pointer_surface_dimensions: Option<PointerSurfaceDimensions>,552- expected_device_path: Option<String>,553- initial_width: u16,554- initial_height: u16,555-) -> Result<EncodedSource> {556- let (sender, receiver) = watch::channel(None);557- let (current_gop_sender, current_gop) = watch::channel(None);558- let (control, desired, status) = EncoderControl::new(initial_width, initial_height);559- let task = tokio::spawn(async move {560- tokio::task::spawn_blocking(move || {561- run_in_process_pipeline(562- &description,563- session_clock,564- sender,565- current_gop_sender,566- max_gop_frames,567- desired,568- status,569- expected_device_path.as_deref(),570- initial_width,571- initial_height,572- )573- })574- .await575- .context("in-process GStreamer worker panicked")?576- });577- Ok(EncodedSource {578- frames: receiver,579- current_gop,580- task,581- pointer_surface_dimensions,582- encoder_control: Some(control),583- })584-}585-586-587-fn gamescope_in_process_pipeline_description(588- config: &HostConfig,589- target_object: &str,590- video_mode: CaptureVideoMode,591-) -> Result<String> {592- let pipewire = gamescope_config(config)?;593- validate_pipewire_target_object(target_object)?;594- ensure!(595- pipewire.rate_control == VaapiRateControl::Cbr,596- "in-process GStreamer requires CBR so bitrate can be changed while playing"597- );598- let bitrate = pipewire599- .bitrate_kbps600- .context("CBR bitrate is missing after validation")?;601- let raw_caps = format!(602- "video/x-raw,format=NV12,width={},height={},framerate={}/1",603- video_mode.width, video_mode.height, video_mode.framerate604- );605- Ok(format!(606- "pipewiresrc do-timestamp=true min-buffers=1 max-buffers=4 use-bufferpool=false target-object={target_object} \607- ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream \608- ! videorate drop-only=true max-rate={} \609- ! videoconvert ! videoscale \610- ! capsfilter name=sigil_scale_caps caps={raw_caps} caps-change-mode=delayed \611- ! {} name=sigil_encoder rate-control=cbr bitrate={bitrate} target-usage=7 key-int-max={} b-frames=0 ref-frames=1 aud=true \612- ! h264parse config-interval=-1 \613- ! video/x-h264,stream-format=byte-stream,alignment=au \614- ! appsink name=sigil_sink max-buffers=1 drop=false sync=false",615- video_mode.framerate,616- pipewire.vaapi_encoder,617- interactive_gop_frames(video_mode.framerate),618- ))619-}620-621-622-fn validate_pipewire_target_object(target_object: &str) -> Result<()> {623- ensure!(624- !target_object.is_empty()625- && target_object.len() <= 32626- && target_object.bytes().all(|byte| byte.is_ascii_digit()),627- "resolved PipeWire object.serial is invalid"628- );629- Ok(())630-}631-632206 async fn run_test_pattern(633207 config: HostConfig,634208 session_clock: SessionClock,@@ -717,37 +291,6 @@ struct AccessUnitPublisher<'a> {
717291 sequence: u64,718292 }719293 720-721-722-struct PendingResolutionTransition {723- target: RevisedResolution,724- deadline: Instant,725-}726-727-728-fn classify_resolution_sample(729- current: (u16, u16),730- pending: Option<PendingResolutionTransition>,731- sample: (u16, u16),732- configured_idr: bool,733-) -> Result<Option<bool>> {734- if let Some(pending) = pending {735- if sample != (pending.target.width, pending.target.height) || !configured_idr {736- return Ok(None);737- }738- return Ok(Some(true));739- }740- ensure!(741- sample == current,742- "encoder changed resolution from {}x{} to {}x{} without a pending control revision",743- current.0,744- current.1,745- sample.0,746- sample.1747- );748- Ok(Some(false))749-}750-751294 impl<'a> AccessUnitPublisher<'a> {752295 fn new(753296 session_clock: SessionClock,@@ -823,395 +366,6 @@ impl<'a> AccessUnitPublisher<'a> {
823366 }824367 }825368 826-827-fn run_in_process_pipeline(828- description: &str,829- session_clock: SessionClock,830- sender: watch::Sender<Option<EncodedFrame>>,831- current_gop_sender: watch::Sender<Option<EncodedGop>>,832- max_gop_frames: usize,833- mut desired: watch::Receiver<EncoderControlDesired>,834- status: watch::Sender<EncoderControlStatus>,835- expected_device_path: Option<&str>,836- initial_width: u16,837- initial_height: u16,838-) -> Result<()> {839- use gstreamer as gst;840- use gstreamer::prelude::*;841- use gstreamer_app as gst_app;842- gst::init().context("initializing in-process GStreamer")?;843- let pipeline = gst::parse::launch(description)844- .context("parsing in-process GStreamer pipeline")?845- .downcast::<gst::Pipeline>()846- .map_err(|_| {847- anyhow::anyhow!("in-process GStreamer description did not create a pipeline")848- })?;849- let encoder = pipeline850- .by_name("sigil_encoder")851- .context("in-process GStreamer pipeline has no named encoder")?;852- let scale_caps = pipeline853- .by_name("sigil_scale_caps")854- .context("in-process GStreamer pipeline has no named scale capsfilter")?;855- let sink = pipeline856- .by_name("sigil_sink")857- .context("in-process GStreamer pipeline has no named appsink")?858- .downcast::<gst_app::AppSink>()859- .map_err(|_| anyhow::anyhow!("named in-process sink is not an appsink"))?;860- sink.set_max_buffers(1);861- // Encoded dependent AUs must not be dropped here: the raw queue before the862- // encoder is the only latest-frame boundary. This one-buffer sink applies863- // bounded backpressure until the publishing loop consumes the AU.864- sink.set_drop(false);865- sink.set_sync(false);866- validate_mutable_playing_bitrate(&encoder)?;867- validate_resolution_capsfilter(&scale_caps)?;868-869- pipeline870- .set_state(gst::State::Playing)871- .context("starting in-process GStreamer pipeline")?;872- let result = (|| {873- let (state_change, state, pending) = pipeline.state(gst::ClockTime::from_seconds(5));874- state_change.context("in-process GStreamer pipeline failed while waiting for PLAYING")?;875- ensure!(876- state == gst::State::Playing,877- "in-process GStreamer pipeline did not reach PLAYING within 5 seconds (state {state:?}, pending {pending:?})"878- );879- if let Some(expected_device_path) = expected_device_path {880- validate_active_encoder_device_path(&encoder, expected_device_path)?;881- }882- let bus = pipeline883- .bus()884- .context("in-process GStreamer pipeline has no bus")?;885- let mut publisher = AccessUnitPublisher::new(886- session_clock,887- &sender,888- ¤t_gop_sender,889- max_gop_frames,890- initial_width,891- initial_height,892- );893- let mut applied_desired_revision = 0_u64;894- let mut pending_force_keyframe_revision = None;895- let mut pending_resolution = None;896- status.send_modify(|status| {897- status.applied_width = Some(initial_width);898- status.applied_height = Some(initial_height);899- });900-901- loop {902- if let Some(message) = bus.timed_pop(gst::ClockTime::ZERO) {903- match message.view() {904- gst::MessageView::Error(error) => {905- bail!(906- "in-process GStreamer pipeline error from {:?}: {} ({:?})",907- error.src().map(|source| source.path_string()),908- error.error(),909- error.debug()910- );911- }912- gst::MessageView::Eos(_) => {913- bail!("in-process GStreamer pipeline ended unexpectedly");914- }915- _ => {}916- }917- }918-919- let desired_state = *desired.borrow_and_update();920- if desired_state.revision > applied_desired_revision {921- apply_encoder_control(922- &encoder,923- &scale_caps,924- &sink,925- desired_state,926- &status,927- ¤t_gop_sender,928- &mut pending_force_keyframe_revision,929- &mut pending_resolution,930- )?;931- applied_desired_revision = desired_state.revision;932- }933-934- if pending_resolution.is_some_and(|pending: PendingResolutionTransition| {935- Instant::now() >= pending.deadline936- }) {937- let pending = pending_resolution.expect("checked as present");938- bail!(939- "encoder resolution revision {} did not produce a configured {}x{} IDR within {:?}",940- pending.target.revision,941- pending.target.width,942- pending.target.height,943- RESOLUTION_APPLY_TIMEOUT944- );945- }946-947- if let Some(sample) = sink.try_pull_sample(gst::ClockTime::from_mseconds(50)) {948- let (width, height) = encoded_sample_dimensions(&sample)?;949- let buffer = sample950- .buffer()951- .context("in-process appsink sample has no buffer")?;952- let map = buffer953- .map_readable()954- .context("mapping in-process encoded access unit")?;955- ensure!(956- map.as_slice().len() <= MAX_ENCODE_BUFFER,957- "in-process encoded access unit exceeds {MAX_ENCODE_BUFFER} bytes"958- );959- let access_unit = map.as_slice();960- let configured_idr =961- is_h264_keyframe(access_unit) && has_h264_codec_config(access_unit);962- let Some(discontinuity) = classify_resolution_sample(963- (publisher.width, publisher.height),964- pending_resolution,965- (width, height),966- configured_idr,967- )?968- else {969- continue;970- };971- let configured_idr = publisher.publish_with_metadata(972- access_unit.to_vec(),973- width,974- height,975- discontinuity,976- )?;977- if discontinuity {978- let applied = pending_resolution979- .take()980- .expect("discontinuity requires a pending resolution");981- status.send_modify(|status| {982- status.applied_resolution_revision = Some(applied.target.revision);983- status.applied_width = Some(width);984- status.applied_height = Some(height);985- });986- }987- acknowledge_force_keyframe_if_configured_idr(988- configured_idr,989- &mut pending_force_keyframe_revision,990- &status,991- );992- } else if sink.is_eos() {993- bail!("in-process GStreamer appsink reached EOS unexpectedly");994- }995-996- if publisher.receivers_closed() {997- debug!("in-process encoded source has no receivers");998- return Ok(());999- }1000- }1001- })();1002- if let Err(error) = pipeline.set_state(gst::State::Null) {1003- warn!(%error, "failed to stop in-process GStreamer pipeline cleanly");1004- }1005- result1006-}1007-1008-1009-fn validate_active_encoder_device_path(1010- encoder: &gstreamer::Element,1011- expected_device_path: &str,1012-) -> Result<()> {1013- use gstreamer as gst;1014- use gstreamer::prelude::*;1015-1016- let property = encoder1017- .find_property("device-path")1018- .context("configured in-process VA encoder has no device-path property")?;1019- ensure!(1020- property.flags().contains(gst::glib::ParamFlags::READABLE),1021- "configured in-process VA encoder device-path property is not readable"1022- );1023- let observed_device_path = encoder.property::<Option<String>>("device-path");1024- ensure!(1025- observed_device_path.as_deref() == Some(expected_device_path),1026- "configured in-process VA encoder uses device-path {observed_device_path:?}, expected {expected_device_path:?}"1027- );1028- Ok(())1029-}1030-1031-1032-fn validate_mutable_playing_bitrate(encoder: &gstreamer::Element) -> Result<()> {1033- use gstreamer as gst;1034- use gstreamer::prelude::*;1035-1036- let property = encoder1037- .find_property("bitrate")1038- .context("configured in-process encoder has no bitrate property")?;1039- ensure!(1040- property.flags().contains(gst::glib::ParamFlags::WRITABLE),1041- "configured in-process encoder bitrate property is not writable"1042- );1043- ensure!(1044- property.flags().contains(gst::PARAM_FLAG_MUTABLE_PLAYING),1045- "configured in-process encoder bitrate property is not mutable while playing"1046- );1047- Ok(())1048-}1049-1050-1051-fn validate_resolution_capsfilter(capsfilter: &gstreamer::Element) -> Result<()> {1052- use gstreamer::prelude::*;1053-1054- let property = capsfilter1055- .find_property("caps")1056- .context("configured scale capsfilter has no caps property")?;1057- ensure!(1058- property1059- .flags()1060- .contains(gstreamer::glib::ParamFlags::WRITABLE),1061- "configured scale capsfilter caps property is not writable"1062- );1063- let caps = capsfilter.property::<gstreamer::Caps>("caps");1064- let _ = dimensions_from_caps(&caps)?;1065- Ok(())1066-}1067-1068-1069-fn dimensions_from_caps(caps: &gstreamer::CapsRef) -> Result<(u16, u16)> {1070- let structure = caps1071- .structure(0)1072- .context("configured video caps have no structure")?;1073- let width = structure1074- .get::<i32>("width")1075- .context("configured video caps have no integer width")?;1076- let height = structure1077- .get::<i32>("height")1078- .context("configured video caps have no integer height")?;1079- Ok((1080- u16::try_from(width).context("configured video caps width is outside u16")?,1081- u16::try_from(height).context("configured video caps height is outside u16")?,1082- ))1083-}1084-1085-1086-fn encoded_sample_dimensions(sample: &gstreamer::Sample) -> Result<(u16, u16)> {1087- let caps = sample1088- .caps()1089- .context("in-process encoded sample has no negotiated caps")?;1090- dimensions_from_caps(caps)1091-}1092-1093-1094-fn apply_resolution_caps(1095- capsfilter: &gstreamer::Element,1096- resolution: RevisedResolution,1097-) -> Result<()> {1098- use gstreamer::prelude::*;1099-1100- let mut caps = capsfilter.property::<gstreamer::Caps>("caps");1101- let caps_mut = caps.make_mut();1102- let structure = caps_mut1103- .structure_mut(0)1104- .context("configured scale capsfilter has no caps structure")?;1105- structure.set("width", i32::from(resolution.width));1106- structure.set("height", i32::from(resolution.height));1107- capsfilter.set_property("caps", caps);1108- let readback = capsfilter.property::<gstreamer::Caps>("caps");1109- ensure!(1110- dimensions_from_caps(&readback)? == (resolution.width, resolution.height),1111- "scale capsfilter resolution readback does not match requested {}x{}",1112- resolution.width,1113- resolution.height1114- );1115- Ok(())1116-}1117-1118-1119-fn push_configured_keyframe_request(sink: &gstreamer_app::AppSink) -> Result<()> {1120- use gstreamer::prelude::*;1121-1122- let event = gstreamer_video::UpstreamForceKeyUnitEvent::builder()1123- .all_headers(true)1124- .build();1125- let sink_pad = sink1126- .static_pad("sink")1127- .context("configured in-process appsink has no sink pad")?;1128- ensure!(1129- sink_pad.push_event(event),1130- "in-process encoder rejected upstream ForceKeyUnit"1131- );1132- Ok(())1133-}1134-1135-1136-fn apply_encoder_control(1137- encoder: &gstreamer::Element,1138- scale_caps: &gstreamer::Element,1139- sink: &gstreamer_app::AppSink,1140- desired: EncoderControlDesired,1141- status: &watch::Sender<EncoderControlStatus>,1142- current_gop_sender: &watch::Sender<Option<EncodedGop>>,1143- pending_force_keyframe_revision: &mut Option<u64>,1144- pending_resolution: &mut Option<PendingResolutionTransition>,1145-) -> Result<()> {1146- let current = *status.borrow();1147- if let Some(bitrate) = desired.bitrate1148- && current1149- .applied_bitrate_revision1150- .is_none_or(|revision| bitrate.revision > revision)1151- {1152- encoder.set_property("bitrate", bitrate.kbps);1153- let readback = encoder.property::<u32>("bitrate");1154- ensure!(1155- readback == bitrate.kbps,1156- "encoder bitrate readback mismatch: requested {} kbps, observed {readback} kbps",1157- bitrate.kbps1158- );1159- status.send_modify(|status| {1160- status.applied_bitrate_revision = Some(bitrate.revision);1161- status.applied_bitrate_kbps = Some(readback);1162- });1163- }1164-1165- let mut keyframe_requested = false;1166- if let Some(resolution) = desired.resolution1167- && current1168- .requested_resolution_revision1169- .is_none_or(|requested| resolution.revision > requested)1170- {1171- apply_resolution_caps(scale_caps, resolution)?;1172- current_gop_sender.send_replace(None);1173- push_configured_keyframe_request(sink)?;1174- keyframe_requested = true;1175- *pending_resolution = Some(PendingResolutionTransition {1176- target: resolution,1177- deadline: Instant::now() + RESOLUTION_APPLY_TIMEOUT,1178- });1179- status.send_modify(|status| {1180- status.requested_resolution_revision = Some(resolution.revision);1181- });1182- }1183-1184- if let Some(revision) = desired.force_keyframe_revision1185- && current1186- .requested_force_keyframe_revision1187- .is_none_or(|requested| revision > requested)1188- {1189- if !keyframe_requested {1190- push_configured_keyframe_request(sink)?;1191- }1192- *pending_force_keyframe_revision = Some(revision);1193- status.send_modify(|status| {1194- status.requested_force_keyframe_revision = Some(revision);1195- });1196- }1197-1198- status.send_modify(|status| status.latest_revision = desired.revision);1199- Ok(())1200-}1201-1202-1203-fn acknowledge_force_keyframe_if_configured_idr(1204- configured_idr: bool,1205- pending_revision: &mut Option<u64>,1206- status: &watch::Sender<EncoderControlStatus>,1207-) {1208- if configured_idr && let Some(revision) = pending_revision.take() {1209- status.send_modify(|status| {1210- status.acknowledged_force_keyframe_revision = Some(revision);1211- });1212- }1213-}1214-1215369 fn publish_encoded_frame(1216370 sender: &watch::Sender<Option<EncodedFrame>>,1217371 current_gop_sender: &watch::Sender<Option<EncodedGop>>,@@ -1676,290 +830,6 @@ mod tests {
1676830 .unwrap()1677831 }1678832 1679- 1680- fn encoder_control_retains_initial_session_dimensions() {1681- let (native, _desired, _status) = EncoderControl::new(2_560, 1_600);1682- let (other, _desired, _status) = EncoderControl::new(1_366, 768);1683-1684- assert_eq!(native.initial_dimensions(), (2_560, 1_600));1685- assert_eq!(other.initial_dimensions(), (1_366, 768));1686- native.request_resolution(1_280, 800).unwrap();1687- assert_eq!(native.initial_dimensions(), (2_560, 1_600));1688- assert_eq!(other.initial_dimensions(), (1_366, 768));1689- }1690-1691- 1692- async fn encoder_control_coalesces_latest_state_and_acknowledges_only_configured_idr() {1693- let (control, mut desired, status) = EncoderControl::new(1_280, 800);1694-1695- let superseded_bitrate_revision = control.request_bitrate_kbps(4_000).unwrap();1696- let bitrate_revision = control.request_bitrate_kbps(8_000).unwrap();1697- let superseded_resolution_revision = control.request_resolution(960, 600).unwrap();1698- let resolution_revision = control.request_resolution(640, 400).unwrap();1699- let force_keyframe_revision = control.request_force_keyframe().unwrap();1700-1701- assert!(superseded_bitrate_revision < bitrate_revision);1702- assert!(bitrate_revision < superseded_resolution_revision);1703- assert!(superseded_resolution_revision < resolution_revision);1704- assert!(resolution_revision < force_keyframe_revision);1705- let latest = *desired.borrow_and_update();1706- assert_eq!(latest.revision, force_keyframe_revision);1707- assert_eq!(1708- latest.bitrate,1709- Some(RevisedBitrate {1710- revision: bitrate_revision,1711- kbps: 8_000,1712- })1713- );1714- assert_eq!(1715- latest.resolution,1716- Some(RevisedResolution {1717- revision: resolution_revision,1718- width: 640,1719- height: 400,1720- })1721- );1722- assert_eq!(1723- latest.force_keyframe_revision,1724- Some(force_keyframe_revision)1725- );1726-1727- status.send_modify(|status| {1728- status.latest_revision = latest.revision;1729- status.applied_bitrate_revision = Some(bitrate_revision);1730- status.applied_bitrate_kbps = Some(8_000);1731- status.requested_resolution_revision = Some(resolution_revision);1732- status.applied_resolution_revision = Some(resolution_revision);1733- status.applied_width = Some(640);1734- status.applied_height = Some(400);1735- status.requested_force_keyframe_revision = Some(force_keyframe_revision);1736- });1737- let mut pending = Some(force_keyframe_revision);1738- acknowledge_force_keyframe_if_configured_idr(false, &mut pending, &status);1739- assert_eq!(pending, Some(force_keyframe_revision));1740- assert_eq!(control.status().acknowledged_force_keyframe_revision, None);1741-1742- acknowledge_force_keyframe_if_configured_idr(true, &mut pending, &status);1743- assert_eq!(pending, None);1744- let applied = control1745- .wait_for_bitrate_applied(bitrate_revision, Duration::from_millis(10))1746- .await1747- .unwrap();1748- assert_eq!(applied.applied_bitrate_kbps, Some(8_000));1749- let applied_resolution = control1750- .wait_for_resolution_applied(resolution_revision, 640, 400, Duration::from_millis(10))1751- .await1752- .unwrap();1753- assert_eq!(applied_resolution.applied_width, Some(640));1754- assert_eq!(applied_resolution.applied_height, Some(400));1755- assert!(1756- control1757- .wait_for_resolution_applied(1758- superseded_resolution_revision,1759- 960,1760- 600,1761- Duration::from_millis(10),1762- )1763- .await1764- .unwrap_err()1765- .to_string()1766- .contains("superseded")1767- );1768- let acknowledged = control1769- .wait_for_force_keyframe_acknowledged(1770- force_keyframe_revision,1771- Duration::from_millis(10),1772- )1773- .await1774- .unwrap();1775- assert_eq!(1776- acknowledged.acknowledged_force_keyframe_revision,1777- Some(force_keyframe_revision)1778- );1779- assert!(control.request_bitrate_kbps(0).is_err());1780- assert!(control.request_resolution(0, 400).is_err());1781- assert!(control.request_resolution(641, 400).is_err());1782- assert!(1783- control1784- .request_bitrate_kbps(MAX_ENCODER_BITRATE_KBPS + 1)1785- .is_err()1786- );1787-1788- let (closed_control, _desired, closed_status) = EncoderControl::new(1_280, 800);1789- let closed_revision = closed_control.request_force_keyframe().unwrap();1790- drop(closed_status);1791- assert!(1792- closed_control1793- .wait_for_force_keyframe_acknowledged(closed_revision, Duration::from_millis(10))1794- .await1795- .unwrap_err()1796- .to_string()1797- .contains("status closed")1798- );1799- }1800-1801- 1802- async fn recovery_waiter_accepts_a_newer_configured_idr_revision() {1803- let (control, _desired, status) = EncoderControl::new(1_280, 800);1804- let recovery_revision = control.request_force_keyframe().unwrap();1805- let newer_revision = control.request_force_keyframe().unwrap();1806- assert!(newer_revision > recovery_revision);1807- status.send_modify(|status| {1808- status.requested_force_keyframe_revision = Some(newer_revision);1809- status.acknowledged_force_keyframe_revision = Some(newer_revision);1810- });1811-1812- let acknowledged = control1813- .wait_for_recovery_keyframe_acknowledged(recovery_revision, Duration::from_millis(10))1814- .await1815- .unwrap();1816- assert_eq!(1817- acknowledged.acknowledged_force_keyframe_revision,1818- Some(newer_revision)1819- );1820- assert!(1821- control1822- .wait_for_force_keyframe_acknowledged(recovery_revision, Duration::from_millis(10),)1823- .await1824- .unwrap_err()1825- .to_string()1826- .contains("superseded")1827- );1828- }1829-1830- 1831- 1832- 1833- async fn in_process_gstreamer_x264_smoke() {1834- let description = "videotestsrc is-live=true pattern=ball \1835- ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream \1836- ! videoconvert ! videoscale \1837- ! capsfilter name=sigil_scale_caps caps=video/x-raw,width=320,height=180,framerate=30/1 caps-change-mode=delayed \1838- ! x264enc name=sigil_encoder bitrate=2000 tune=zerolatency speed-preset=ultrafast key-int-max=15 bframes=0 byte-stream=true aud=true \1839- ! h264parse config-interval=-1 \1840- ! video/x-h264,stream-format=byte-stream,alignment=au \1841- ! appsink name=sigil_sink max-buffers=1 drop=false sync=false"1842- .to_owned();1843- let mut source =1844- spawn_in_process_pipeline(description, SessionClock::start(), 15, None, None, 320, 180)1845- .unwrap();1846- let control = source1847- .encoder_control1848- .clone()1849- .expect("x264 source has no encoder control");1850-1851- tokio::time::timeout(Duration::from_secs(10), async {1852- loop {1853- source.frames.changed().await.unwrap();1854- if source1855- .frames1856- .borrow()1857- .as_ref()1858- .is_some_and(|frame| frame.keyframe && frame.codec_config)1859- {1860- break;1861- }1862- }1863- })1864- .await1865- .expect("x264 did not publish an initial configured IDR");1866-1867- let bitrate_revision = control.request_bitrate_kbps(1_500).unwrap();1868- let force_keyframe_revision = control.request_force_keyframe().unwrap();1869- let bitrate_status = control1870- .wait_for_bitrate_applied(bitrate_revision, Duration::from_secs(10))1871- .await1872- .expect("x264 did not apply the bitrate revision");1873- assert_eq!(bitrate_status.applied_bitrate_kbps, Some(1_500));1874- control1875- .wait_for_force_keyframe_acknowledged(force_keyframe_revision, Duration::from_secs(10))1876- .await1877- .expect("x264 did not acknowledge a configured forced IDR");1878- let before_resolution_sequence = source1879- .current_gop1880- .borrow()1881- .as_ref()1882- .and_then(|gop| gop.frames.first())1883- .map(|frame| frame.sequence)1884- .expect("x264 has no configured GOP before resolution transition");1885-1886- let reduced_revision = control.request_resolution(256, 144).unwrap();1887- control1888- .wait_for_resolution_applied(reduced_revision, 256, 144, Duration::from_secs(10))1889- .await1890- .expect("x264 did not apply the reduced resolution");1891- let reduced = source1892- .current_gop1893- .borrow()1894- .as_ref()1895- .and_then(|gop| gop.frames.first())1896- .cloned()1897- .expect("x264 has no reduced configured GOP");1898- assert_eq!((reduced.width, reduced.height), (256, 144));1899- assert!(reduced.keyframe && reduced.codec_config && reduced.discontinuity);1900- assert!(reduced.sequence > before_resolution_sequence);1901-1902- let restored_revision = control.request_resolution(320, 180).unwrap();1903- control1904- .wait_for_resolution_applied(restored_revision, 320, 180, Duration::from_secs(10))1905- .await1906- .expect("x264 did not restore the native resolution");1907- let restored = source1908- .current_gop1909- .borrow()1910- .as_ref()1911- .and_then(|gop| gop.frames.first())1912- .cloned()1913- .expect("x264 has no restored configured GOP");1914- assert_eq!((restored.width, restored.height), (320, 180));1915- assert!(restored.keyframe && restored.codec_config && restored.discontinuity);1916- assert!(restored.sequence > reduced.sequence);1917-1918- {1919- let current_gop = source.current_gop.borrow();1920- let gop = current_gop1921- .as_ref()1922- .expect("x264 did not retain a current GOP");1923- assert!(gop.frames[0].keyframe && gop.frames[0].codec_config);1924- assert!(gop.frames.len() <= 15);1925- }1926- let EncodedSource {1927- frames,1928- current_gop,1929- task,1930- ..1931- } = source;1932- drop(frames);1933- drop(current_gop);1934- tokio::time::timeout(Duration::from_secs(2), task)1935- .await1936- .expect("x264 pipeline did not stop after its receivers closed")1937- .expect("x264 pipeline task panicked")1938- .expect("x264 pipeline returned an error");1939- }1940-1941- 1942- 1943- fn in_process_pipeline_drops_only_before_encoding() {1944- let mut config = gamescope_config();1945- config.gamescope_pipewire.as_mut().unwrap().encoder_backend =1946- GamescopeEncoderBackend::InProcessGstreamer;1947- let description =1948- gamescope_in_process_pipeline_description(&config, "1234", configured_video_mode())1949- .unwrap();1950-1951- assert!(description.contains(1952- "queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream"1953- ));1954- assert!(description.contains(1955- "capsfilter name=sigil_scale_caps caps=video/x-raw,format=NV12,width=1280,height=800,framerate=60/1 caps-change-mode=delayed"1956- ));1957- assert!(1958- description.contains("appsink name=sigil_sink max-buffers=1 drop=false sync=false")1959- );1960- assert!(!description.contains("appsink name=sigil_sink max-buffers=1 drop=true"));1961- }1962-1963833 1964834 fn detects_h264_keyframes() {1965835 assert!(is_h264_keyframe(&[0, 0, 0, 1, 5, 1, 2]));@@ -1976,36 +846,6 @@ mod tests {
1976846 assert!(!has_h264_codec_config(&[0, 0, 1, 8, 1, 2]));1977847 }1978848 1979- 1980- fn resolution_transition_suppresses_until_exact_configured_target_idr() {1981- let transition = PendingResolutionTransition {1982- target: RevisedResolution {1983- revision: 7,1984- width: 960,1985- height: 600,1986- },1987- deadline: Instant::now() + Duration::from_secs(1),1988- };1989- assert!(transition.deadline > Instant::now());1990- assert_eq!(1991- classify_resolution_sample((1280, 800), Some(transition), (1280, 800), true).unwrap(),1992- None1993- );1994- assert_eq!(1995- classify_resolution_sample((1280, 800), Some(transition), (960, 600), false).unwrap(),1996- None1997- );1998- assert_eq!(1999- classify_resolution_sample((1280, 800), Some(transition), (960, 600), true).unwrap(),2000- Some(true)2001- );2002- assert_eq!(2003- classify_resolution_sample((960, 600), None, (960, 600), false).unwrap(),2004- Some(false)2005- );2006- assert!(classify_resolution_sample((960, 600), None, (1280, 800), true).is_err());2007- }2008-2009849 2010850 fn publisher_carries_exact_resolution_and_discontinuity_into_new_gop() {2011851 let (frame_sender, frame_receiver) = watch::channel(None);@@ -0,0 +1,1195 @@
1+use std::sync::Arc;2+use std::sync::atomic::{AtomicU64, Ordering};3+use std::time::{Duration, Instant};4+5+use anyhow::{Context, Result, bail, ensure};6+use tokio::sync::watch;7+8+9+use crate::clock::SessionClock;10+11+use crate::config::{HostConfig, VaapiRateControl};12+13+use sigil_protocol::PointerSurfaceDimensions;14+15+16+use super::{17+ AccessUnitPublisher, CaptureVideoMode, EncodedFrame, EncodedGop, EncodedSource,18+ GamescopePreflight, MAX_ENCODE_BUFFER, gamescope_config, has_h264_codec_config,19+ interactive_gop_frames, is_h264_keyframe,20+};21+22+23+use gstreamer::prelude::ObjectExt;24+25+use tracing::{debug, warn};26+27+28+const RESOLUTION_APPLY_TIMEOUT: Duration = Duration::from_secs(2);29+30+const MAX_ENCODER_BITRATE_KBPS: u32 = 100_000;31+32+33+struct RevisedBitrate {34+ revision: u64,35+ kbps: u32,36+}37+38+39+struct RevisedResolution {40+ revision: u64,41+ width: u16,42+ height: u16,43+}44+45+46+struct EncoderControlDesired {47+ revision: u64,48+ bitrate: Option<RevisedBitrate>,49+ resolution: Option<RevisedResolution>,50+ force_keyframe_revision: Option<u64>,51+}52+53+54+pub struct EncoderControlStatus {55+ pub latest_revision: u64,56+ pub applied_bitrate_revision: Option<u64>,57+ pub applied_bitrate_kbps: Option<u32>,58+ pub requested_resolution_revision: Option<u64>,59+ pub applied_resolution_revision: Option<u64>,60+ pub applied_width: Option<u16>,61+ pub applied_height: Option<u16>,62+ pub requested_force_keyframe_revision: Option<u64>,63+ pub acknowledged_force_keyframe_revision: Option<u64>,64+}65+66+/// Bounded latest-state control for an in-process encoder.67+///68+/// Both directions are Tokio watch channels: callers can update control state69+/// arbitrarily often without building a queue, while the encoder applies only70+/// the latest revision. Revisions are process-monotonic and never wrap.71+72+pub struct EncoderControl {73+ initial_width: u16,74+ initial_height: u16,75+ next_revision: Arc<AtomicU64>,76+ desired: watch::Sender<EncoderControlDesired>,77+ status: watch::Receiver<EncoderControlStatus>,78+}79+80+81+82+83+84+impl EncoderControl {85+ fn new(86+ initial_width: u16,87+ initial_height: u16,88+ ) -> (89+ Self,90+ watch::Receiver<EncoderControlDesired>,91+ watch::Sender<EncoderControlStatus>,92+ ) {93+ let (desired, desired_rx) = watch::channel(EncoderControlDesired::default());94+ let (status_tx, status) = watch::channel(EncoderControlStatus::default());95+ (96+ Self {97+ initial_width,98+ initial_height,99+ next_revision: Arc::new(AtomicU64::new(0)),100+ desired,101+ status,102+ },103+ desired_rx,104+ status_tx,105+ )106+ }107+108+ pub(crate) fn initial_dimensions(&self) -> (u16, u16) {109+ (self.initial_width, self.initial_height)110+ }111+112+ fn next_revision(&self) -> Result<u64> {113+ self.next_revision114+ .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |revision| {115+ revision.checked_add(1)116+ })117+ .map(|revision| revision + 1)118+ .map_err(|_| anyhow::anyhow!("encoder control revision exhausted"))119+ }120+121+ pub fn request_bitrate_kbps(&self, kbps: u32) -> Result<u64> {122+ ensure!(123+ (1..=MAX_ENCODER_BITRATE_KBPS).contains(&kbps),124+ "encoder bitrate must be between 1 and {MAX_ENCODER_BITRATE_KBPS} kbps"125+ );126+ let revision = self.next_revision()?;127+ self.desired.send_modify(|desired| {128+ desired.revision = revision;129+ desired.bitrate = Some(RevisedBitrate { revision, kbps });130+ });131+ Ok(revision)132+ }133+134+ pub fn request_force_keyframe(&self) -> Result<u64> {135+ let revision = self.next_revision()?;136+ self.desired.send_modify(|desired| {137+ desired.revision = revision;138+ desired.force_keyframe_revision = Some(revision);139+ });140+ Ok(revision)141+ }142+143+ pub fn request_resolution(&self, width: u16, height: u16) -> Result<u64> {144+ ensure!(145+ width >= 64 && height >= 64,146+ "encoder dimensions must be at least 64x64"147+ );148+ ensure!(149+ width.is_multiple_of(2) && height.is_multiple_of(2),150+ "H.264 encoder dimensions must be even"151+ );152+ let revision = self.next_revision()?;153+ self.desired.send_modify(|desired| {154+ desired.revision = revision;155+ desired.resolution = Some(RevisedResolution {156+ revision,157+ width,158+ height,159+ });160+ });161+ Ok(revision)162+ }163+164+ pub fn status(&self) -> EncoderControlStatus {165+ *self.status.borrow()166+ }167+168+ /// Wait until the encoder confirms the exact bitrate revision and exposes169+ /// its property readback. A newer applied revision means this request was170+ /// coalesced before application and therefore fails rather than pretending171+ /// the requested revision committed.172+ pub async fn wait_for_bitrate_applied(173+ &self,174+ revision: u64,175+ timeout: Duration,176+ ) -> Result<EncoderControlStatus> {177+ let mut status = self.status.clone();178+ let wait = async {179+ loop {180+ let snapshot = *status.borrow_and_update();181+ match snapshot.applied_bitrate_revision {182+ Some(applied) if applied == revision => return Ok(snapshot),183+ Some(applied) if applied > revision => bail!(184+ "encoder bitrate revision {revision} was superseded by revision {applied}"185+ ),186+ _ => {}187+ }188+ status189+ .changed()190+ .await191+ .context("encoder control status closed before bitrate was applied")?;192+ }193+ };194+ tokio::time::timeout(timeout, wait)195+ .await196+ .with_context(|| format!("timed out waiting for encoder bitrate revision {revision}"))?197+ }198+199+ /// Wait for a forced-keyframe request to be committed by observing a200+ /// keyframe that also carries codec configuration.201+ pub async fn wait_for_force_keyframe_acknowledged(202+ &self,203+ revision: u64,204+ timeout: Duration,205+ ) -> Result<EncoderControlStatus> {206+ let mut status = self.status.clone();207+ let wait = async {208+ loop {209+ let snapshot = *status.borrow_and_update();210+ match snapshot.acknowledged_force_keyframe_revision {211+ Some(acknowledged) if acknowledged == revision => return Ok(snapshot),212+ Some(acknowledged) if acknowledged > revision => bail!(213+ "force-keyframe revision {revision} was superseded by revision {acknowledged}"214+ ),215+ _ => {}216+ }217+ if snapshot218+ .requested_force_keyframe_revision219+ .is_some_and(|requested| requested > revision)220+ {221+ bail!("force-keyframe revision {revision} was coalesced before application");222+ }223+ status.changed().await.context(224+ "encoder control status closed before forced keyframe was acknowledged",225+ )?;226+ }227+ };228+ tokio::time::timeout(timeout, wait)229+ .await230+ .with_context(|| format!("timed out waiting for force-keyframe revision {revision}"))?231+ }232+233+ /// Wait for recovery to reach any configured IDR at or after `revision`.234+ ///235+ /// Recovery requests are deliberately coalesced with later encoder236+ /// controls. Unlike transactional adaptive updates, a newer configured IDR237+ /// still satisfies the decoder barrier established by an older request.238+ pub async fn wait_for_recovery_keyframe_acknowledged(239+ &self,240+ revision: u64,241+ timeout: Duration,242+ ) -> Result<EncoderControlStatus> {243+ let mut status = self.status.clone();244+ let wait = async {245+ loop {246+ let snapshot = *status.borrow_and_update();247+ if snapshot248+ .acknowledged_force_keyframe_revision249+ .is_some_and(|acknowledged| acknowledged >= revision)250+ {251+ return Ok(snapshot);252+ }253+ status.changed().await.context(254+ "encoder control status closed before recovery keyframe was acknowledged",255+ )?;256+ }257+ };258+ tokio::time::timeout(timeout, wait).await.with_context(|| {259+ format!("timed out waiting for recovery keyframe revision {revision}")260+ })?261+ }262+263+ /// Wait until the exact target dimensions emerge from the encoder on an264+ /// independently decodable access unit. A property write or caps event is265+ /// not an application acknowledgement.266+ pub async fn wait_for_resolution_applied(267+ &self,268+ revision: u64,269+ width: u16,270+ height: u16,271+ timeout: Duration,272+ ) -> Result<EncoderControlStatus> {273+ let mut status = self.status.clone();274+ let wait = async {275+ loop {276+ let snapshot = *status.borrow_and_update();277+ match snapshot.applied_resolution_revision {278+ Some(applied) if applied == revision => {279+ ensure!(280+ snapshot.applied_width == Some(width)281+ && snapshot.applied_height == Some(height),282+ "encoder resolution revision {revision} acknowledged {:?}x{:?}, expected {width}x{height}",283+ snapshot.applied_width,284+ snapshot.applied_height285+ );286+ return Ok(snapshot);287+ }288+ Some(applied) if applied > revision => bail!(289+ "encoder resolution revision {revision} was superseded by revision {applied}"290+ ),291+ _ => {}292+ }293+ if snapshot294+ .requested_resolution_revision295+ .is_some_and(|requested| requested > revision)296+ {297+ bail!(298+ "encoder resolution revision {revision} was coalesced before application"299+ );300+ }301+ status302+ .changed()303+ .await304+ .context("encoder control status closed before resolution was applied")?;305+ }306+ };307+ tokio::time::timeout(timeout, wait).await.with_context(|| {308+ format!("timed out waiting for encoder resolution revision {revision}")309+ })?310+ }311+}312+313+314+pub(crate) struct EncoderControlTestHarness {315+ pub control: EncoderControl,316+ pub status: watch::Sender<EncoderControlStatus>,317+ _desired: watch::Receiver<EncoderControlDesired>,318+}319+320+321+impl EncoderControlTestHarness {322+ pub(crate) fn new() -> Self {323+ Self::with_dimensions(1_280, 800)324+ }325+326+ pub(crate) fn with_dimensions(width: u16, height: u16) -> Self {327+ let (control, desired, status) = EncoderControl::new(width, height);328+ Self {329+ control,330+ status,331+ _desired: desired,332+ }333+ }334+335+ pub(crate) fn requested_force_keyframe_revision(&self) -> Option<u64> {336+ self._desired.borrow().force_keyframe_revision337+ }338+}339+340+341+pub(super) fn spawn_gamescope_pipewire_in_process_with_target(342+ config: HostConfig,343+ session_clock: SessionClock,344+ preflight: GamescopePreflight,345+) -> Result<EncodedSource> {346+ let description = gamescope_in_process_pipeline_description(347+ &config,348+ &preflight.target_object,349+ preflight.video_mode,350+ )?;351+ let max_gop_frames = interactive_gop_frames(preflight.video_mode.framerate) as usize;352+ let expected_device_path = Some(353+ gamescope_config(&config)?354+ .vaapi_render_node355+ .to_string_lossy()356+ .into_owned(),357+ );358+ spawn_in_process_pipeline(359+ description,360+ session_clock,361+ max_gop_frames,362+ Some(preflight.pointer_surface_dimensions),363+ expected_device_path,364+ preflight.video_mode.width,365+ preflight.video_mode.height,366+ )367+}368+369+370+fn spawn_in_process_pipeline(371+ description: String,372+ session_clock: SessionClock,373+ max_gop_frames: usize,374+ pointer_surface_dimensions: Option<PointerSurfaceDimensions>,375+ expected_device_path: Option<String>,376+ initial_width: u16,377+ initial_height: u16,378+) -> Result<EncodedSource> {379+ let (sender, receiver) = watch::channel(None);380+ let (current_gop_sender, current_gop) = watch::channel(None);381+ let (control, desired, status) = EncoderControl::new(initial_width, initial_height);382+ let task = tokio::spawn(async move {383+ tokio::task::spawn_blocking(move || {384+ run_in_process_pipeline(385+ &description,386+ session_clock,387+ sender,388+ current_gop_sender,389+ max_gop_frames,390+ desired,391+ status,392+ expected_device_path.as_deref(),393+ initial_width,394+ initial_height,395+ )396+ })397+ .await398+ .context("in-process GStreamer worker panicked")?399+ });400+ Ok(EncodedSource {401+ frames: receiver,402+ current_gop,403+ task,404+ pointer_surface_dimensions,405+ encoder_control: Some(control),406+ })407+}408+409+410+fn gamescope_in_process_pipeline_description(411+ config: &HostConfig,412+ target_object: &str,413+ video_mode: CaptureVideoMode,414+) -> Result<String> {415+ let pipewire = gamescope_config(config)?;416+ validate_pipewire_target_object(target_object)?;417+ ensure!(418+ pipewire.rate_control == VaapiRateControl::Cbr,419+ "in-process GStreamer requires CBR so bitrate can be changed while playing"420+ );421+ let bitrate = pipewire422+ .bitrate_kbps423+ .context("CBR bitrate is missing after validation")?;424+ let raw_caps = format!(425+ "video/x-raw,format=NV12,width={},height={},framerate={}/1",426+ video_mode.width, video_mode.height, video_mode.framerate427+ );428+ Ok(format!(429+ "pipewiresrc do-timestamp=true min-buffers=1 max-buffers=4 use-bufferpool=false target-object={target_object} \430+ ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream \431+ ! videorate drop-only=true max-rate={} \432+ ! videoconvert ! videoscale \433+ ! capsfilter name=sigil_scale_caps caps={raw_caps} caps-change-mode=delayed \434+ ! {} name=sigil_encoder rate-control=cbr bitrate={bitrate} target-usage=7 key-int-max={} b-frames=0 ref-frames=1 aud=true \435+ ! h264parse config-interval=-1 \436+ ! video/x-h264,stream-format=byte-stream,alignment=au \437+ ! appsink name=sigil_sink max-buffers=1 drop=false sync=false",438+ video_mode.framerate,439+ pipewire.vaapi_encoder,440+ interactive_gop_frames(video_mode.framerate),441+ ))442+}443+444+445+fn validate_pipewire_target_object(target_object: &str) -> Result<()> {446+ ensure!(447+ !target_object.is_empty()448+ && target_object.len() <= 32449+ && target_object.bytes().all(|byte| byte.is_ascii_digit()),450+ "resolved PipeWire object.serial is invalid"451+ );452+ Ok(())453+}454+455+456+457+struct PendingResolutionTransition {458+ target: RevisedResolution,459+ deadline: Instant,460+}461+462+463+fn classify_resolution_sample(464+ current: (u16, u16),465+ pending: Option<PendingResolutionTransition>,466+ sample: (u16, u16),467+ configured_idr: bool,468+) -> Result<Option<bool>> {469+ if let Some(pending) = pending {470+ if sample != (pending.target.width, pending.target.height) || !configured_idr {471+ return Ok(None);472+ }473+ return Ok(Some(true));474+ }475+ ensure!(476+ sample == current,477+ "encoder changed resolution from {}x{} to {}x{} without a pending control revision",478+ current.0,479+ current.1,480+ sample.0,481+ sample.1482+ );483+ Ok(Some(false))484+}485+486+487+fn run_in_process_pipeline(488+ description: &str,489+ session_clock: SessionClock,490+ sender: watch::Sender<Option<EncodedFrame>>,491+ current_gop_sender: watch::Sender<Option<EncodedGop>>,492+ max_gop_frames: usize,493+ mut desired: watch::Receiver<EncoderControlDesired>,494+ status: watch::Sender<EncoderControlStatus>,495+ expected_device_path: Option<&str>,496+ initial_width: u16,497+ initial_height: u16,498+) -> Result<()> {499+ use gstreamer as gst;500+ use gstreamer::prelude::*;501+ use gstreamer_app as gst_app;502+ gst::init().context("initializing in-process GStreamer")?;503+ let pipeline = gst::parse::launch(description)504+ .context("parsing in-process GStreamer pipeline")?505+ .downcast::<gst::Pipeline>()506+ .map_err(|_| {507+ anyhow::anyhow!("in-process GStreamer description did not create a pipeline")508+ })?;509+ let encoder = pipeline510+ .by_name("sigil_encoder")511+ .context("in-process GStreamer pipeline has no named encoder")?;512+ let scale_caps = pipeline513+ .by_name("sigil_scale_caps")514+ .context("in-process GStreamer pipeline has no named scale capsfilter")?;515+ let sink = pipeline516+ .by_name("sigil_sink")517+ .context("in-process GStreamer pipeline has no named appsink")?518+ .downcast::<gst_app::AppSink>()519+ .map_err(|_| anyhow::anyhow!("named in-process sink is not an appsink"))?;520+ sink.set_max_buffers(1);521+ // Encoded dependent AUs must not be dropped here: the raw queue before the522+ // encoder is the only latest-frame boundary. This one-buffer sink applies523+ // bounded backpressure until the publishing loop consumes the AU.524+ sink.set_drop(false);525+ sink.set_sync(false);526+ validate_mutable_playing_bitrate(&encoder)?;527+ validate_resolution_capsfilter(&scale_caps)?;528+529+ pipeline530+ .set_state(gst::State::Playing)531+ .context("starting in-process GStreamer pipeline")?;532+ let result = (|| {533+ let (state_change, state, pending) = pipeline.state(gst::ClockTime::from_seconds(5));534+ state_change.context("in-process GStreamer pipeline failed while waiting for PLAYING")?;535+ ensure!(536+ state == gst::State::Playing,537+ "in-process GStreamer pipeline did not reach PLAYING within 5 seconds (state {state:?}, pending {pending:?})"538+ );539+ if let Some(expected_device_path) = expected_device_path {540+ validate_active_encoder_device_path(&encoder, expected_device_path)?;541+ }542+ let bus = pipeline543+ .bus()544+ .context("in-process GStreamer pipeline has no bus")?;545+ let mut publisher = AccessUnitPublisher::new(546+ session_clock,547+ &sender,548+ ¤t_gop_sender,549+ max_gop_frames,550+ initial_width,551+ initial_height,552+ );553+ let mut applied_desired_revision = 0_u64;554+ let mut pending_force_keyframe_revision = None;555+ let mut pending_resolution = None;556+ status.send_modify(|status| {557+ status.applied_width = Some(initial_width);558+ status.applied_height = Some(initial_height);559+ });560+561+ loop {562+ if let Some(message) = bus.timed_pop(gst::ClockTime::ZERO) {563+ match message.view() {564+ gst::MessageView::Error(error) => {565+ bail!(566+ "in-process GStreamer pipeline error from {:?}: {} ({:?})",567+ error.src().map(|source| source.path_string()),568+ error.error(),569+ error.debug()570+ );571+ }572+ gst::MessageView::Eos(_) => {573+ bail!("in-process GStreamer pipeline ended unexpectedly");574+ }575+ _ => {}576+ }577+ }578+579+ let desired_state = *desired.borrow_and_update();580+ if desired_state.revision > applied_desired_revision {581+ apply_encoder_control(582+ &encoder,583+ &scale_caps,584+ &sink,585+ desired_state,586+ &status,587+ ¤t_gop_sender,588+ &mut pending_force_keyframe_revision,589+ &mut pending_resolution,590+ )?;591+ applied_desired_revision = desired_state.revision;592+ }593+594+ if pending_resolution.is_some_and(|pending: PendingResolutionTransition| {595+ Instant::now() >= pending.deadline596+ }) {597+ let pending = pending_resolution.expect("checked as present");598+ bail!(599+ "encoder resolution revision {} did not produce a configured {}x{} IDR within {:?}",600+ pending.target.revision,601+ pending.target.width,602+ pending.target.height,603+ RESOLUTION_APPLY_TIMEOUT604+ );605+ }606+607+ if let Some(sample) = sink.try_pull_sample(gst::ClockTime::from_mseconds(50)) {608+ let (width, height) = encoded_sample_dimensions(&sample)?;609+ let buffer = sample610+ .buffer()611+ .context("in-process appsink sample has no buffer")?;612+ let map = buffer613+ .map_readable()614+ .context("mapping in-process encoded access unit")?;615+ ensure!(616+ map.as_slice().len() <= MAX_ENCODE_BUFFER,617+ "in-process encoded access unit exceeds {MAX_ENCODE_BUFFER} bytes"618+ );619+ let access_unit = map.as_slice();620+ let configured_idr =621+ is_h264_keyframe(access_unit) && has_h264_codec_config(access_unit);622+ let Some(discontinuity) = classify_resolution_sample(623+ (publisher.width, publisher.height),624+ pending_resolution,625+ (width, height),626+ configured_idr,627+ )?628+ else {629+ continue;630+ };631+ let configured_idr = publisher.publish_with_metadata(632+ access_unit.to_vec(),633+ width,634+ height,635+ discontinuity,636+ )?;637+ if discontinuity {638+ let applied = pending_resolution639+ .take()640+ .expect("discontinuity requires a pending resolution");641+ status.send_modify(|status| {642+ status.applied_resolution_revision = Some(applied.target.revision);643+ status.applied_width = Some(width);644+ status.applied_height = Some(height);645+ });646+ }647+ acknowledge_force_keyframe_if_configured_idr(648+ configured_idr,649+ &mut pending_force_keyframe_revision,650+ &status,651+ );652+ } else if sink.is_eos() {653+ bail!("in-process GStreamer appsink reached EOS unexpectedly");654+ }655+656+ if publisher.receivers_closed() {657+ debug!("in-process encoded source has no receivers");658+ return Ok(());659+ }660+ }661+ })();662+ if let Err(error) = pipeline.set_state(gst::State::Null) {663+ warn!(%error, "failed to stop in-process GStreamer pipeline cleanly");664+ }665+ result666+}667+668+669+fn validate_active_encoder_device_path(670+ encoder: &gstreamer::Element,671+ expected_device_path: &str,672+) -> Result<()> {673+ use gstreamer as gst;674+ use gstreamer::prelude::*;675+676+ let property = encoder677+ .find_property("device-path")678+ .context("configured in-process VA encoder has no device-path property")?;679+ ensure!(680+ property.flags().contains(gst::glib::ParamFlags::READABLE),681+ "configured in-process VA encoder device-path property is not readable"682+ );683+ let observed_device_path = encoder.property::<Option<String>>("device-path");684+ ensure!(685+ observed_device_path.as_deref() == Some(expected_device_path),686+ "configured in-process VA encoder uses device-path {observed_device_path:?}, expected {expected_device_path:?}"687+ );688+ Ok(())689+}690+691+692+fn validate_mutable_playing_bitrate(encoder: &gstreamer::Element) -> Result<()> {693+ use gstreamer as gst;694+ use gstreamer::prelude::*;695+696+ let property = encoder697+ .find_property("bitrate")698+ .context("configured in-process encoder has no bitrate property")?;699+ ensure!(700+ property.flags().contains(gst::glib::ParamFlags::WRITABLE),701+ "configured in-process encoder bitrate property is not writable"702+ );703+ ensure!(704+ property.flags().contains(gst::PARAM_FLAG_MUTABLE_PLAYING),705+ "configured in-process encoder bitrate property is not mutable while playing"706+ );707+ Ok(())708+}709+710+711+fn validate_resolution_capsfilter(capsfilter: &gstreamer::Element) -> Result<()> {712+ use gstreamer::prelude::*;713+714+ let property = capsfilter715+ .find_property("caps")716+ .context("configured scale capsfilter has no caps property")?;717+ ensure!(718+ property719+ .flags()720+ .contains(gstreamer::glib::ParamFlags::WRITABLE),721+ "configured scale capsfilter caps property is not writable"722+ );723+ let caps = capsfilter.property::<gstreamer::Caps>("caps");724+ let _ = dimensions_from_caps(&caps)?;725+ Ok(())726+}727+728+729+fn dimensions_from_caps(caps: &gstreamer::CapsRef) -> Result<(u16, u16)> {730+ let structure = caps731+ .structure(0)732+ .context("configured video caps have no structure")?;733+ let width = structure734+ .get::<i32>("width")735+ .context("configured video caps have no integer width")?;736+ let height = structure737+ .get::<i32>("height")738+ .context("configured video caps have no integer height")?;739+ Ok((740+ u16::try_from(width).context("configured video caps width is outside u16")?,741+ u16::try_from(height).context("configured video caps height is outside u16")?,742+ ))743+}744+745+746+fn encoded_sample_dimensions(sample: &gstreamer::Sample) -> Result<(u16, u16)> {747+ let caps = sample748+ .caps()749+ .context("in-process encoded sample has no negotiated caps")?;750+ dimensions_from_caps(caps)751+}752+753+754+fn apply_resolution_caps(755+ capsfilter: &gstreamer::Element,756+ resolution: RevisedResolution,757+) -> Result<()> {758+ use gstreamer::prelude::*;759+760+ let mut caps = capsfilter.property::<gstreamer::Caps>("caps");761+ let caps_mut = caps.make_mut();762+ let structure = caps_mut763+ .structure_mut(0)764+ .context("configured scale capsfilter has no caps structure")?;765+ structure.set("width", i32::from(resolution.width));766+ structure.set("height", i32::from(resolution.height));767+ capsfilter.set_property("caps", caps);768+ let readback = capsfilter.property::<gstreamer::Caps>("caps");769+ ensure!(770+ dimensions_from_caps(&readback)? == (resolution.width, resolution.height),771+ "scale capsfilter resolution readback does not match requested {}x{}",772+ resolution.width,773+ resolution.height774+ );775+ Ok(())776+}777+778+779+fn push_configured_keyframe_request(sink: &gstreamer_app::AppSink) -> Result<()> {780+ use gstreamer::prelude::*;781+782+ let event = gstreamer_video::UpstreamForceKeyUnitEvent::builder()783+ .all_headers(true)784+ .build();785+ let sink_pad = sink786+ .static_pad("sink")787+ .context("configured in-process appsink has no sink pad")?;788+ ensure!(789+ sink_pad.push_event(event),790+ "in-process encoder rejected upstream ForceKeyUnit"791+ );792+ Ok(())793+}794+795+796+fn apply_encoder_control(797+ encoder: &gstreamer::Element,798+ scale_caps: &gstreamer::Element,799+ sink: &gstreamer_app::AppSink,800+ desired: EncoderControlDesired,801+ status: &watch::Sender<EncoderControlStatus>,802+ current_gop_sender: &watch::Sender<Option<EncodedGop>>,803+ pending_force_keyframe_revision: &mut Option<u64>,804+ pending_resolution: &mut Option<PendingResolutionTransition>,805+) -> Result<()> {806+ let current = *status.borrow();807+ if let Some(bitrate) = desired.bitrate808+ && current809+ .applied_bitrate_revision810+ .is_none_or(|revision| bitrate.revision > revision)811+ {812+ encoder.set_property("bitrate", bitrate.kbps);813+ let readback = encoder.property::<u32>("bitrate");814+ ensure!(815+ readback == bitrate.kbps,816+ "encoder bitrate readback mismatch: requested {} kbps, observed {readback} kbps",817+ bitrate.kbps818+ );819+ status.send_modify(|status| {820+ status.applied_bitrate_revision = Some(bitrate.revision);821+ status.applied_bitrate_kbps = Some(readback);822+ });823+ }824+825+ let mut keyframe_requested = false;826+ if let Some(resolution) = desired.resolution827+ && current828+ .requested_resolution_revision829+ .is_none_or(|requested| resolution.revision > requested)830+ {831+ apply_resolution_caps(scale_caps, resolution)?;832+ current_gop_sender.send_replace(None);833+ push_configured_keyframe_request(sink)?;834+ keyframe_requested = true;835+ *pending_resolution = Some(PendingResolutionTransition {836+ target: resolution,837+ deadline: Instant::now() + RESOLUTION_APPLY_TIMEOUT,838+ });839+ status.send_modify(|status| {840+ status.requested_resolution_revision = Some(resolution.revision);841+ });842+ }843+844+ if let Some(revision) = desired.force_keyframe_revision845+ && current846+ .requested_force_keyframe_revision847+ .is_none_or(|requested| revision > requested)848+ {849+ if !keyframe_requested {850+ push_configured_keyframe_request(sink)?;851+ }852+ *pending_force_keyframe_revision = Some(revision);853+ status.send_modify(|status| {854+ status.requested_force_keyframe_revision = Some(revision);855+ });856+ }857+858+ status.send_modify(|status| status.latest_revision = desired.revision);859+ Ok(())860+}861+862+863+fn acknowledge_force_keyframe_if_configured_idr(864+ configured_idr: bool,865+ pending_revision: &mut Option<u64>,866+ status: &watch::Sender<EncoderControlStatus>,867+) {868+ if configured_idr && let Some(revision) = pending_revision.take() {869+ status.send_modify(|status| {870+ status.acknowledged_force_keyframe_revision = Some(revision);871+ });872+ }873+}874+875+876+mod tests {877+ use super::*;878+879+ 880+ fn encoder_control_retains_initial_session_dimensions() {881+ let (native, _desired, _status) = EncoderControl::new(2_560, 1_600);882+ let (other, _desired, _status) = EncoderControl::new(1_366, 768);883+884+ assert_eq!(native.initial_dimensions(), (2_560, 1_600));885+ assert_eq!(other.initial_dimensions(), (1_366, 768));886+ native.request_resolution(1_280, 800).unwrap();887+ assert_eq!(native.initial_dimensions(), (2_560, 1_600));888+ assert_eq!(other.initial_dimensions(), (1_366, 768));889+ }890+891+ 892+ async fn encoder_control_coalesces_latest_state_and_acknowledges_only_configured_idr() {893+ let (control, mut desired, status) = EncoderControl::new(1_280, 800);894+895+ let superseded_bitrate_revision = control.request_bitrate_kbps(4_000).unwrap();896+ let bitrate_revision = control.request_bitrate_kbps(8_000).unwrap();897+ let superseded_resolution_revision = control.request_resolution(960, 600).unwrap();898+ let resolution_revision = control.request_resolution(640, 400).unwrap();899+ let force_keyframe_revision = control.request_force_keyframe().unwrap();900+901+ assert!(superseded_bitrate_revision < bitrate_revision);902+ assert!(bitrate_revision < superseded_resolution_revision);903+ assert!(superseded_resolution_revision < resolution_revision);904+ assert!(resolution_revision < force_keyframe_revision);905+ let latest = *desired.borrow_and_update();906+ assert_eq!(latest.revision, force_keyframe_revision);907+ assert_eq!(908+ latest.bitrate,909+ Some(RevisedBitrate {910+ revision: bitrate_revision,911+ kbps: 8_000,912+ })913+ );914+ assert_eq!(915+ latest.resolution,916+ Some(RevisedResolution {917+ revision: resolution_revision,918+ width: 640,919+ height: 400,920+ })921+ );922+ assert_eq!(923+ latest.force_keyframe_revision,924+ Some(force_keyframe_revision)925+ );926+927+ status.send_modify(|status| {928+ status.latest_revision = latest.revision;929+ status.applied_bitrate_revision = Some(bitrate_revision);930+ status.applied_bitrate_kbps = Some(8_000);931+ status.requested_resolution_revision = Some(resolution_revision);932+ status.applied_resolution_revision = Some(resolution_revision);933+ status.applied_width = Some(640);934+ status.applied_height = Some(400);935+ status.requested_force_keyframe_revision = Some(force_keyframe_revision);936+ });937+ let mut pending = Some(force_keyframe_revision);938+ acknowledge_force_keyframe_if_configured_idr(false, &mut pending, &status);939+ assert_eq!(pending, Some(force_keyframe_revision));940+ assert_eq!(control.status().acknowledged_force_keyframe_revision, None);941+942+ acknowledge_force_keyframe_if_configured_idr(true, &mut pending, &status);943+ assert_eq!(pending, None);944+ let applied = control945+ .wait_for_bitrate_applied(bitrate_revision, Duration::from_millis(10))946+ .await947+ .unwrap();948+ assert_eq!(applied.applied_bitrate_kbps, Some(8_000));949+ let applied_resolution = control950+ .wait_for_resolution_applied(resolution_revision, 640, 400, Duration::from_millis(10))951+ .await952+ .unwrap();953+ assert_eq!(applied_resolution.applied_width, Some(640));954+ assert_eq!(applied_resolution.applied_height, Some(400));955+ assert!(956+ control957+ .wait_for_resolution_applied(958+ superseded_resolution_revision,959+ 960,960+ 600,961+ Duration::from_millis(10),962+ )963+ .await964+ .unwrap_err()965+ .to_string()966+ .contains("superseded")967+ );968+ let acknowledged = control969+ .wait_for_force_keyframe_acknowledged(970+ force_keyframe_revision,971+ Duration::from_millis(10),972+ )973+ .await974+ .unwrap();975+ assert_eq!(976+ acknowledged.acknowledged_force_keyframe_revision,977+ Some(force_keyframe_revision)978+ );979+ assert!(control.request_bitrate_kbps(0).is_err());980+ assert!(control.request_resolution(0, 400).is_err());981+ assert!(control.request_resolution(641, 400).is_err());982+ assert!(983+ control984+ .request_bitrate_kbps(MAX_ENCODER_BITRATE_KBPS + 1)985+ .is_err()986+ );987+988+ let (closed_control, _desired, closed_status) = EncoderControl::new(1_280, 800);989+ let closed_revision = closed_control.request_force_keyframe().unwrap();990+ drop(closed_status);991+ assert!(992+ closed_control993+ .wait_for_force_keyframe_acknowledged(closed_revision, Duration::from_millis(10))994+ .await995+ .unwrap_err()996+ .to_string()997+ .contains("status closed")998+ );999+ }1000+1001+ 1002+ async fn recovery_waiter_accepts_a_newer_configured_idr_revision() {1003+ let (control, _desired, status) = EncoderControl::new(1_280, 800);1004+ let recovery_revision = control.request_force_keyframe().unwrap();1005+ let newer_revision = control.request_force_keyframe().unwrap();1006+ assert!(newer_revision > recovery_revision);1007+ status.send_modify(|status| {1008+ status.requested_force_keyframe_revision = Some(newer_revision);1009+ status.acknowledged_force_keyframe_revision = Some(newer_revision);1010+ });1011+1012+ let acknowledged = control1013+ .wait_for_recovery_keyframe_acknowledged(recovery_revision, Duration::from_millis(10))1014+ .await1015+ .unwrap();1016+ assert_eq!(1017+ acknowledged.acknowledged_force_keyframe_revision,1018+ Some(newer_revision)1019+ );1020+ assert!(1021+ control1022+ .wait_for_force_keyframe_acknowledged(recovery_revision, Duration::from_millis(10),)1023+ .await1024+ .unwrap_err()1025+ .to_string()1026+ .contains("superseded")1027+ );1028+ }1029+1030+ 1031+ fn resolution_transition_suppresses_until_exact_configured_target_idr() {1032+ let transition = PendingResolutionTransition {1033+ target: RevisedResolution {1034+ revision: 7,1035+ width: 960,1036+ height: 600,1037+ },1038+ deadline: Instant::now() + Duration::from_secs(1),1039+ };1040+ assert!(transition.deadline > Instant::now());1041+ assert_eq!(1042+ classify_resolution_sample((1280, 800), Some(transition), (1280, 800), true).unwrap(),1043+ None1044+ );1045+ assert_eq!(1046+ classify_resolution_sample((1280, 800), Some(transition), (960, 600), false).unwrap(),1047+ None1048+ );1049+ assert_eq!(1050+ classify_resolution_sample((1280, 800), Some(transition), (960, 600), true).unwrap(),1051+ Some(true)1052+ );1053+ assert_eq!(1054+ classify_resolution_sample((960, 600), None, (960, 600), false).unwrap(),1055+ Some(false)1056+ );1057+ assert!(classify_resolution_sample((960, 600), None, (1280, 800), true).is_err());1058+ }1059+1060+ 1061+ 1062+ fn in_process_pipeline_drops_only_before_encoding() {1063+ use super::super::tests::{configured_video_mode, gamescope_config};1064+ use crate::config::GamescopeEncoderBackend;1065+1066+ let mut config = gamescope_config();1067+ config.gamescope_pipewire.as_mut().unwrap().encoder_backend =1068+ GamescopeEncoderBackend::InProcessGstreamer;1069+ let description =1070+ gamescope_in_process_pipeline_description(&config, "1234", configured_video_mode())1071+ .unwrap();1072+1073+ assert!(description.contains(1074+ "queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream"1075+ ));1076+ assert!(description.contains(1077+ "capsfilter name=sigil_scale_caps caps=video/x-raw,format=NV12,width=1280,height=800,framerate=60/1 caps-change-mode=delayed"1078+ ));1079+ assert!(1080+ description.contains("appsink name=sigil_sink max-buffers=1 drop=false sync=false")1081+ );1082+ assert!(!description.contains("appsink name=sigil_sink max-buffers=1 drop=true"));1083+ }1084+1085+ 1086+ 1087+ 1088+ async fn in_process_gstreamer_x264_smoke() {1089+ let description = "videotestsrc is-live=true pattern=ball \1090+ ! queue max-size-buffers=1 max-size-bytes=0 max-size-time=0 leaky=downstream \1091+ ! videoconvert ! videoscale \1092+ ! capsfilter name=sigil_scale_caps caps=video/x-raw,width=320,height=180,framerate=30/1 caps-change-mode=delayed \1093+ ! x264enc name=sigil_encoder bitrate=2000 tune=zerolatency speed-preset=ultrafast key-int-max=15 bframes=0 byte-stream=true aud=true \1094+ ! h264parse config-interval=-1 \1095+ ! video/x-h264,stream-format=byte-stream,alignment=au \1096+ ! appsink name=sigil_sink max-buffers=1 drop=false sync=false"1097+ .to_owned();1098+ let mut source =1099+ spawn_in_process_pipeline(description, SessionClock::start(), 15, None, None, 320, 180)1100+ .unwrap();1101+ let control = source1102+ .encoder_control1103+ .clone()1104+ .expect("x264 source has no encoder control");1105+1106+ tokio::time::timeout(Duration::from_secs(10), async {1107+ loop {1108+ source.frames.changed().await.unwrap();1109+ if source1110+ .frames1111+ .borrow()1112+ .as_ref()1113+ .is_some_and(|frame| frame.keyframe && frame.codec_config)1114+ {1115+ break;1116+ }1117+ }1118+ })1119+ .await1120+ .expect("x264 did not publish an initial configured IDR");1121+1122+ let bitrate_revision = control.request_bitrate_kbps(1_500).unwrap();1123+ let force_keyframe_revision = control.request_force_keyframe().unwrap();1124+ let bitrate_status = control1125+ .wait_for_bitrate_applied(bitrate_revision, Duration::from_secs(10))1126+ .await1127+ .expect("x264 did not apply the bitrate revision");1128+ assert_eq!(bitrate_status.applied_bitrate_kbps, Some(1_500));1129+ control1130+ .wait_for_force_keyframe_acknowledged(force_keyframe_revision, Duration::from_secs(10))1131+ .await1132+ .expect("x264 did not acknowledge a configured forced IDR");1133+ let before_resolution_sequence = source1134+ .current_gop1135+ .borrow()1136+ .as_ref()1137+ .and_then(|gop| gop.frames.first())1138+ .map(|frame| frame.sequence)1139+ .expect("x264 has no configured GOP before resolution transition");1140+1141+ let reduced_revision = control.request_resolution(256, 144).unwrap();1142+ control1143+ .wait_for_resolution_applied(reduced_revision, 256, 144, Duration::from_secs(10))1144+ .await1145+ .expect("x264 did not apply the reduced resolution");1146+ let reduced = source1147+ .current_gop1148+ .borrow()1149+ .as_ref()1150+ .and_then(|gop| gop.frames.first())1151+ .cloned()1152+ .expect("x264 has no reduced configured GOP");1153+ assert_eq!((reduced.width, reduced.height), (256, 144));1154+ assert!(reduced.keyframe && reduced.codec_config && reduced.discontinuity);1155+ assert!(reduced.sequence > before_resolution_sequence);1156+1157+ let restored_revision = control.request_resolution(320, 180).unwrap();1158+ control1159+ .wait_for_resolution_applied(restored_revision, 320, 180, Duration::from_secs(10))1160+ .await1161+ .expect("x264 did not restore the native resolution");1162+ let restored = source1163+ .current_gop1164+ .borrow()1165+ .as_ref()1166+ .and_then(|gop| gop.frames.first())1167+ .cloned()1168+ .expect("x264 has no restored configured GOP");1169+ assert_eq!((restored.width, restored.height), (320, 180));1170+ assert!(restored.keyframe && restored.codec_config && restored.discontinuity);1171+ assert!(restored.sequence > reduced.sequence);1172+1173+ {1174+ let current_gop = source.current_gop.borrow();1175+ let gop = current_gop1176+ .as_ref()1177+ .expect("x264 did not retain a current GOP");1178+ assert!(gop.frames[0].keyframe && gop.frames[0].codec_config);1179+ assert!(gop.frames.len() <= 15);1180+ }1181+ let EncodedSource {1182+ frames,1183+ current_gop,1184+ task,1185+ ..1186+ } = source;1187+ drop(frames);1188+ drop(current_gop);1189+ tokio::time::timeout(Duration::from_secs(2), task)1190+ .await1191+ .expect("x264 pipeline did not stop after its receivers closed")1192+ .expect("x264 pipeline task panicked")1193+ .expect("x264 pipeline returned an error");1194+ }1195+}