1use super::*;23const GAMESCOPE_STARTUP_SAMPLE_WINDOW: Duration = Duration::from_millis(500);4const GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN: Duration = Duration::from_millis(100);5const GAMESCOPE_STARTUP_TARGET_MIN_FRAMES: u64 = 8;6const GAMESCOPE_STARTUP_TARGET_MIN_FPS: f64 = 45.0;789struct StartupCadenceSample {10 first_sequence: Option<u64>,11 last_sequence: Option<u64>,12 first_observed_at: Option<Instant>,13 last_observed_at: Option<Instant>,14 receiver_open: bool,15 decodable_gop_ready: bool,16}1718impl StartupCadenceSample {19 fn observe(&mut self, frame: &EncodedFrame) {20 if self.first_sequence.is_none() {21 self.first_sequence = Some(frame.sequence);22 self.first_observed_at = Some(frame.observed_at);23 }24 if self25 .last_sequence26 .is_none_or(|sequence| frame.sequence >= sequence)27 {28 self.last_sequence = Some(frame.sequence);29 self.last_observed_at = Some(frame.observed_at);30 }31 }3233 fn frame_progress(self) -> u64 {34 self.first_sequence35 .zip(self.last_sequence)36 .map_or(0, |(first, last)| last.saturating_sub(first) + 1)37 }3839 fn fps(self) -> f64 {40 let frames = self.frame_progress();41 let elapsed = self.observation_span();42 if frames < 2 || elapsed.is_zero() {43 return 0.0;44 }45 (frames - 1) as f64 / elapsed.as_secs_f64()46 }4748 fn observation_span(self) -> Duration {49 self.first_observed_at50 .zip(self.last_observed_at)51 .map_or(Duration::ZERO, |(first, last)| {52 last.saturating_duration_since(first)53 })54 }5556 fn has_representative_span(self) -> bool {57 self.observation_span() >= GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN58 }5960 fn is_usable(self) -> bool {61 self.receiver_open && self.frame_progress() > 0 && self.decodable_gop_ready62 }6364 fn meets_target_cadence(self) -> bool {65 self.receiver_open66 && self.has_representative_span()67 && self.frame_progress() >= GAMESCOPE_STARTUP_TARGET_MIN_FRAMES68 && self.fps() >= GAMESCOPE_STARTUP_TARGET_MIN_FPS69 }70}7172fn startup_source_needs_restart(sample: StartupCadenceSample) -> bool {73 // An open producer with no frames is ambiguous by design: damage-driven74 // Gamescope capture may be healthy and simply have nothing to repaint.75 // Restart only on a closed producer or observed encoded progress that76 // never yielded a decodable GOP.77 !sample.receiver_open || (sample.frame_progress() > 0 && !sample.decodable_gop_ready)78}7980fn replacement_source_has_failed(sample: StartupCadenceSample) -> bool {81 // After the one sequential retry, open output channels mean the capture82 // task is still alive. A damage-driven Gamescope source may legitimately83 // remain frameless until the next repaint, so do not turn the sampling84 // deadline into a session-establishment deadline.85 !sample.receiver_open86}8788async fn sample_startup_cadence(89 receiver: &mut tokio::sync::watch::Receiver<Option<EncodedFrame>>,90 current_gop: &tokio::sync::watch::Receiver<Option<EncodedGop>>,91 window: Duration,92) -> StartupCadenceSample {93 let mut sample = StartupCadenceSample {94 receiver_open: true,95 ..StartupCadenceSample::default()96 };97 if let Some(frame) = receiver.borrow_and_update().as_ref() {98 sample.observe(frame);99 }100 sample.decodable_gop_ready = current_gop.borrow().is_some();101 let deadline = tokio::time::Instant::now() + window;102 while !sample.is_usable() || !sample.has_representative_span() {103 match tokio::time::timeout_at(deadline, receiver.changed()).await {104 Ok(Ok(())) => {105 if let Some(frame) = receiver.borrow_and_update().as_ref() {106 sample.observe(frame);107 }108 sample.decodable_gop_ready = current_gop.borrow().is_some();109 }110 Ok(Err(_)) => {111 sample.receiver_open = false;112 break;113 }114 Err(_) => break,115 }116 }117 sample118}119120async fn reap_encoded_source(source: EncodedSource) {121 reap_encoded_source_with_timeout(source, SOURCE_REAP_GRACE_TIMEOUT).await;122}123124async fn reap_encoded_source_with_timeout(source: EncodedSource, grace_timeout: Duration) {125 let EncodedSource {126 frames,127 current_gop,128 task,129 pointer_surface_dimensions: _,130 encoder_control: _,131 } = source;132 // Closing both bounded outputs gives the capture task a chance to kill and133 // wait for its GStreamer child itself. Abort is only the bounded fallback.134 drop(frames);135 drop(current_gop);136 SourceTaskGuard::new(task)137 .wait_or_abort(grace_timeout)138 .await;139}140141pub(super) async fn select_gamescope_startup_source(142 config: HostConfig,143 session_clock: SessionClock,144 mut primary: EncodedSource,145) -> Result<EncodedSource> {146 let primary_sample = sample_startup_cadence(147 &mut primary.frames,148 &primary.current_gop,149 GAMESCOPE_STARTUP_SAMPLE_WINDOW,150 )151 .await;152 info!(153 frames = primary_sample.frame_progress(),154 fps = primary_sample.fps(),155 receiver_open = primary_sample.receiver_open,156 decodable_gop_ready = primary_sample.decodable_gop_ready,157 target_cadence = primary_sample.meets_target_cadence(),158 "sampled primary Gamescope capture startup"159 );160 if !startup_source_needs_restart(primary_sample) {161 if !primary_sample.is_usable() {162 warn!(163 "primary Gamescope capture is live but awaiting a damage-driven decodable repaint"164 );165 }166 return Ok(primary);167 }168169 // Gamescope's PipeWire export does not reliably fan out full cadence to170 // two simultaneous consumers. Reap an unhealthy startup pipeline before171 // opening its replacement; overlapping them can divide vblank deliveries172 // and permanently strand the selected source at half rate.173 reap_encoded_source(primary).await;174 let mut replacement = spawn_gamescope_pipewire_after_static_preflight(config, session_clock)175 .await176 .context("restarting unhealthy Gamescope capture pipeline")?;177 let replacement_sample = sample_startup_cadence(178 &mut replacement.frames,179 &replacement.current_gop,180 GAMESCOPE_STARTUP_SAMPLE_WINDOW,181 )182 .await;183 info!(184 frames = replacement_sample.frame_progress(),185 fps = replacement_sample.fps(),186 receiver_open = replacement_sample.receiver_open,187 decodable_gop_ready = replacement_sample.decodable_gop_ready,188 target_cadence = replacement_sample.meets_target_cadence(),189 "sampled sequential Gamescope capture startup replacement"190 );191 if replacement_source_has_failed(replacement_sample) {192 let frames = replacement_sample.frame_progress();193 let fps = replacement_sample.fps();194 let receiver_open = replacement_sample.receiver_open;195 let decodable_gop_ready = replacement_sample.decodable_gop_ready;196 reap_encoded_source(replacement).await;197 bail!(198 "replacement Gamescope capture pipeline remained unhealthy: \199 frames={frames}, fps={fps:.2}, receiver_open={receiver_open}, \200 decodable_gop_ready={decodable_gop_ready}"201 );202 }203 if !replacement_sample.is_usable() {204 warn!(205 frames = replacement_sample.frame_progress(),206 fps = replacement_sample.fps(),207 decodable_gop_ready = replacement_sample.decodable_gop_ready,208 "sequential Gamescope capture is live but awaiting a damage-driven decodable repaint"209 );210 }211 Ok(replacement)212}213214215mod tests {216 use super::super::DropNotify;217 use super::*;218219 fn startup_sample(fps: f64, frames: u64, receiver_open: bool) -> StartupCadenceSample {220 if frames == 0 {221 return StartupCadenceSample {222 receiver_open,223 ..StartupCadenceSample::default()224 };225 }226 let first = Instant::now();227 let span = if frames > 1 && fps > 0.0 {228 Duration::from_secs_f64((frames - 1) as f64 / fps)229 } else {230 Duration::ZERO231 };232 StartupCadenceSample {233 first_sequence: Some(0),234 last_sequence: Some(frames - 1),235 first_observed_at: Some(first),236 last_observed_at: Some(first + span),237 receiver_open,238 decodable_gop_ready: true,239 }240 }241242 243 fn startup_restart_requires_closed_or_active_undecodable_source() {244 let slow = startup_sample(12.0, 6, true);245 let target = startup_sample(60.0, 8, true);246 assert!(!startup_source_needs_restart(slow));247 assert!(!slow.meets_target_cadence());248 assert!(!startup_source_needs_restart(target));249 assert!(target.meets_target_cadence());250 assert!(startup_source_needs_restart(startup_sample(60.0, 8, false)));251 assert!(!startup_source_needs_restart(startup_sample(0.0, 0, true)));252 let mut undecodable = startup_sample(60.0, 8, true);253 undecodable.decodable_gop_ready = false;254 assert!(startup_source_needs_restart(undecodable));255 }256257 258 fn startup_bursts_are_not_sustained_health() {259 let burst = startup_sample(10_000.0, 8, true);260 assert!(burst.fps() > GAMESCOPE_STARTUP_TARGET_MIN_FPS);261 assert!(!burst.has_representative_span());262 assert!(!burst.meets_target_cadence());263 assert!(!startup_source_needs_restart(burst));264 }265266 267 fn live_replacement_can_wait_for_damage_driven_repaint() {268 let frameless = startup_sample(0.0, 0, true);269 assert!(!startup_source_needs_restart(frameless));270 assert!(!replacement_source_has_failed(frameless));271272 let closed = startup_sample(0.0, 0, false);273 assert!(replacement_source_has_failed(closed));274 }275276 277 async fn startup_sampling_preserves_live_frameless_source() {278 let (frame_sender, mut frame_receiver) = tokio::sync::watch::channel(None);279 let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(None);280281 let sample =282 sample_startup_cadence(&mut frame_receiver, &gop_receiver, Duration::ZERO).await;283284 assert_eq!(sample.frame_progress(), 0);285 assert!(sample.receiver_open);286 assert!(!replacement_source_has_failed(sample));287 assert!(!frame_sender.is_closed());288 }289290 291 async fn startup_sampling_preserves_current_decodable_gop() {292 let now = Instant::now();293 let frame = EncodedFrame {294 sequence: 0,295 width: 1_280,296 height: 800,297 capture_timestamp_micros: 0,298 presentation_timestamp_micros: 0,299 observed_at: now,300 keyframe: true,301 codec_config: true,302 discontinuity: false,303 data: Arc::from([1_u8, 2, 3]),304 };305 let (_frame_sender, mut frame_receiver) = tokio::sync::watch::channel(Some(frame.clone()));306 let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(Some(EncodedGop {307 frames: vec![frame],308 payload_bytes: 3,309 }));310311 let sample =312 sample_startup_cadence(&mut frame_receiver, &gop_receiver, Duration::ZERO).await;313 assert_eq!(sample.frame_progress(), 1);314 let current_gop = gop_receiver.borrow().clone().unwrap();315 assert_eq!(current_gop.frames.len(), 1);316 assert!(current_gop.frames[0].keyframe && current_gop.frames[0].codec_config);317 }318319 320 async fn reaping_source_closes_outputs_before_waiting_for_task_cleanup() {321 let (frame_sender, frame_receiver) = tokio::sync::watch::channel(None);322 let (gop_sender, gop_receiver) = tokio::sync::watch::channel(None);323 let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();324 let task = tokio::spawn(async move {325 frame_sender.closed().await;326 gop_sender.closed().await;327 let _ = reaped_tx.send(());328 Ok(())329 });330 let source = EncodedSource {331 frames: frame_receiver,332 current_gop: gop_receiver,333 task,334 pointer_surface_dimensions: None,335 encoder_control: None,336 };337338 reap_encoded_source_with_timeout(source, Duration::from_millis(100)).await;339 reaped_rx340 .await341 .expect("source task did not observe both closed outputs");342 }343344 345 async fn reaping_stalled_source_aborts_after_bounded_grace() {346 let (_frame_sender, frame_receiver) = tokio::sync::watch::channel(None);347 let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(None);348 let (started_tx, started_rx) = tokio::sync::oneshot::channel();349 let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();350 let task = tokio::spawn(async move {351 let _notify = DropNotify(Some(reaped_tx));352 let _ = started_tx.send(());353 std::future::pending::<Result<()>>().await354 });355 let source = EncodedSource {356 frames: frame_receiver,357 current_gop: gop_receiver,358 task,359 pointer_surface_dimensions: None,360 encoder_control: None,361 };362363 started_rx.await.unwrap();364 reap_encoded_source_with_timeout(source, Duration::from_millis(10)).await;365 tokio::time::timeout(Duration::from_millis(100), reaped_rx)366 .await367 .expect("stalled source task was not aborted and reaped")368 .unwrap();369 }370}