Reap degraded PipeWire consumers before spawning replacements, require the lone replacement to meet startup cadence health, and bound graceful source shutdown before aborting.
- commit
- c69d7a85368847d84d4391c419bd4a6b461d40d5
- parent
- b7809e9e3fd3cadf62f4ce7b80ce66968ae0d98f
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 194 | ++++++++++++++++-------------- |
@@ -44,7 +44,7 @@ const GAMESCOPE_STARTUP_SAMPLE_WINDOW: Duration = Duration::from_millis(500);
4444 const GAMESCOPE_STARTUP_MIN_OBSERVATION_SPAN: Duration = Duration::from_millis(100);4545 const GAMESCOPE_STARTUP_HEALTHY_MIN_FRAMES: u64 = 8;4646 const GAMESCOPE_STARTUP_HEALTHY_MIN_FPS: f64 = 45.0;47-const GAMESCOPE_STARTUP_MATERIAL_FPS_GAIN: f64 = 15.0;47+const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);4848 4949 5050 pub struct SessionRegistry {@@ -240,6 +240,19 @@ impl SourceTaskGuard {
240240 let _ = task.await;241241 }242242 }243+244+ async fn wait_or_abort(mut self, grace_timeout: Duration) {245+ let Some(mut task) = self.0.take() else {246+ return;247+ };248+ if tokio::time::timeout(grace_timeout, &mut task)249+ .await250+ .is_err()251+ {252+ task.abort();253+ let _ = task.await;254+ }255+ }243256 }244257 245258 impl Drop for SourceTaskGuard {@@ -309,6 +322,10 @@ impl StartupCadenceSample {
309322 }310323 }311324 325+fn startup_source_needs_restart(sample: StartupCadenceSample) -> bool {326+ !sample.is_healthy()327+}328+312329 async fn sample_startup_cadence(313330 receiver: &mut tokio::sync::watch::Receiver<Option<EncodedFrame>>,314331 window: Duration,@@ -338,40 +355,31 @@ async fn sample_startup_cadence(
338355 sample339356 }340357 341-fn prefer_startup_challenger(342- primary: StartupCadenceSample,343- challenger: StartupCadenceSample,344-) -> bool {345- if !challenger.receiver_open {346- return false;347- }348- if !primary.receiver_open {349- return true;350- }351- let primary_frames = primary.frame_progress();352- let challenger_frames = challenger.frame_progress();353- if primary_frames == 0 {354- return challenger_frames > 0;355- }356- if challenger.is_healthy() && !primary.is_healthy() {357- return true;358- }359- primary.has_representative_span()360- && challenger.has_representative_span()361- && challenger_frames >= primary_frames.saturating_add(2)362- && challenger.fps() >= primary.fps() + GAMESCOPE_STARTUP_MATERIAL_FPS_GAIN358+async fn reap_encoded_source(source: EncodedSource) {359+ reap_encoded_source_with_timeout(source, SOURCE_REAP_GRACE_TIMEOUT).await;363360 }364361 365-async fn reap_encoded_source(source: EncodedSource) {366- let EncodedSource { task, .. } = source;367- SourceTaskGuard::new(task).abort_and_wait().await;362+async fn reap_encoded_source_with_timeout(source: EncodedSource, grace_timeout: Duration) {363+ let EncodedSource {364+ frames,365+ current_gop,366+ task,367+ pointer_surface_dimensions: _,368+ } = source;369+ // Closing both bounded outputs gives the capture task a chance to kill and370+ // wait for its GStreamer child itself. Abort is only the bounded fallback.371+ drop(frames);372+ drop(current_gop);373+ SourceTaskGuard::new(task)374+ .wait_or_abort(grace_timeout)375+ .await;368376 }369377 370378 async fn select_gamescope_startup_source(371379 config: HostConfig,372380 session_clock: SessionClock,373381 mut primary: EncodedSource,374-) -> EncodedSource {382+) -> Result<EncodedSource> {375383 let primary_sample =376384 sample_startup_cadence(&mut primary.frames, GAMESCOPE_STARTUP_SAMPLE_WINDOW).await;377385 info!(@@ -380,33 +388,37 @@ async fn select_gamescope_startup_source(
380388 receiver_open = primary_sample.receiver_open,381389 "sampled primary Gamescope capture startup"382390 );383- if primary_sample.is_healthy() {384- return primary;391+ if !startup_source_needs_restart(primary_sample) {392+ return Ok(primary);385393 }386394 387- let Ok(mut challenger) = spawn_gamescope_pipewire_after_static_preflight(config, session_clock)395+ // Gamescope's PipeWire export does not reliably fan out full cadence to396+ // two simultaneous consumers. Reap an unhealthy startup pipeline before397+ // opening its replacement; overlapping them can divide vblank deliveries398+ // and permanently strand the selected source at half rate.399+ reap_encoded_source(primary).await;400+ let mut replacement = spawn_gamescope_pipewire_after_static_preflight(config, session_clock)388401 .await389- .inspect_err(|error| warn!(%error, "Gamescope startup challenger was unavailable"))390- else {391- return primary;392- };393- let challenger_sample =394- sample_startup_cadence(&mut challenger.frames, GAMESCOPE_STARTUP_SAMPLE_WINDOW).await;395- let use_challenger = prefer_startup_challenger(primary_sample, challenger_sample);402+ .context("restarting unhealthy Gamescope capture pipeline")?;403+ let replacement_sample =404+ sample_startup_cadence(&mut replacement.frames, GAMESCOPE_STARTUP_SAMPLE_WINDOW).await;396405 info!(397- frames = challenger_sample.frame_progress(),398- fps = challenger_sample.fps(),399- receiver_open = challenger_sample.receiver_open,400- selected = use_challenger,401- "sampled bounded Gamescope capture startup challenger"406+ frames = replacement_sample.frame_progress(),407+ fps = replacement_sample.fps(),408+ receiver_open = replacement_sample.receiver_open,409+ "sampled sequential Gamescope capture startup replacement"402410 );403- if use_challenger {404- reap_encoded_source(primary).await;405- challenger406- } else {407- reap_encoded_source(challenger).await;408- primary411+ if !replacement_sample.is_healthy() {412+ let frames = replacement_sample.frame_progress();413+ let fps = replacement_sample.fps();414+ let receiver_open = replacement_sample.receiver_open;415+ reap_encoded_source(replacement).await;416+ bail!(417+ "replacement Gamescope capture pipeline remained unhealthy: \418+ frames={frames}, fps={fps:.2}, receiver_open={receiver_open}"419+ );409420 }421+ Ok(replacement)410422 }411423 412424 @@ -762,11 +774,11 @@ async fn serve_media(
762774 lease.session_clock,763775 )764776 .await?;765- Ok(select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await)777+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await766778 }767779 };768780 let EncodedSource {769- frames: _frame_receiver,781+ frames: frame_receiver,770782 current_gop: mut current_gop_receiver,771783 task: source_task,772784 pointer_surface_dimensions,@@ -875,7 +887,9 @@ async fn serve_media(
875887 }876888 .await;877889 878- source_task.abort_and_wait().await;890+ drop(current_gop_receiver);891+ drop(frame_receiver);892+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;879893 drop(lease);880894 info!(%remote, "media client released");881895 session_result@@ -915,11 +929,11 @@ async fn serve_media_v2(
915929 lease.session_clock,916930 )917931 .await?;918- Ok(select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await)932+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await919933 }920934 };921935 let EncodedSource {922- frames: _frame_receiver,936+ frames: frame_receiver,923937 current_gop: mut current_gop_receiver,924938 task: source_task,925939 pointer_surface_dimensions,@@ -949,7 +963,9 @@ async fn serve_media_v2(
949963 let session_result =950964 run_media_v2_session(&connection, &config, &mut current_gop_receiver, remote).await;951965 952- source_task.abort_and_wait().await;966+ drop(current_gop_receiver);967+ drop(frame_receiver);968+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;953969 drop(lease);954970 info!(%remote, "media v2 client released");955971 session_result@@ -1706,48 +1722,22 @@ mod tests {
17061722 }17071723 17081724 1709- fn startup_challenger_requires_materially_healthier_progress() {1725+ fn startup_restart_requires_sustained_healthy_progress() {17101726 let slow = startup_sample(12.0, 6, true);17111727 let healthy = startup_sample(60.0, 8, true);1712- assert!(prefer_startup_challenger(slow, healthy));1713-1714- assert!(!prefer_startup_challenger(1715- startup_sample(12.0, 6, true),1716- startup_sample(12.0, 6, true)1717- ));1718- assert!(!prefer_startup_challenger(1719- startup_sample(0.0, 0, true),1720- startup_sample(0.0, 0, true)1721- ));1722- assert!(prefer_startup_challenger(1723- startup_sample(0.0, 0, true),1724- startup_sample(0.0, 1, true)1725- ));1726- assert!(prefer_startup_challenger(1727- startup_sample(0.0, 0, false),1728- startup_sample(12.0, 6, true)1729- ));1730- assert!(!prefer_startup_challenger(1731- startup_sample(12.0, 6, true),1732- startup_sample(60.0, 8, false)1733- ));1728+ assert!(startup_source_needs_restart(slow));1729+ assert!(!startup_source_needs_restart(healthy));1730+ assert!(startup_source_needs_restart(startup_sample(60.0, 8, false)));1731+ assert!(startup_source_needs_restart(startup_sample(0.0, 0, true)));17341732 }17351733 17361734 1737- fn startup_bursts_are_not_sustained_health_or_material_gain() {1735+ fn startup_bursts_are_not_sustained_health() {17381736 let burst = startup_sample(10_000.0, 8, true);17391737 assert!(burst.fps() > GAMESCOPE_STARTUP_HEALTHY_MIN_FPS);17401738 assert!(!burst.has_representative_span());17411739 assert!(!burst.is_healthy());1742-1743- assert!(!prefer_startup_challenger(1744- startup_sample(12.0, 6, true),1745- burst1746- ));1747- assert!(prefer_startup_challenger(1748- startup_sample(0.0, 0, true),1749- burst1750- ));1740+ assert!(startup_source_needs_restart(burst));17511741 }17521742 17531743 @@ -1776,7 +1766,31 @@ mod tests {
17761766 }17771767 17781768 1779- async fn reaping_unselected_source_always_terminates_its_task() {1769+ async fn reaping_source_closes_outputs_before_waiting_for_task_cleanup() {1770+ let (frame_sender, frame_receiver) = tokio::sync::watch::channel(None);1771+ let (gop_sender, gop_receiver) = tokio::sync::watch::channel(None);1772+ let (reaped_tx, reaped_rx) = tokio::sync::oneshot::channel();1773+ let task = tokio::spawn(async move {1774+ frame_sender.closed().await;1775+ gop_sender.closed().await;1776+ let _ = reaped_tx.send(());1777+ Ok(())1778+ });1779+ let source = EncodedSource {1780+ frames: frame_receiver,1781+ current_gop: gop_receiver,1782+ task,1783+ pointer_surface_dimensions: None,1784+ };1785+1786+ reap_encoded_source_with_timeout(source, Duration::from_millis(100)).await;1787+ reaped_rx1788+ .await1789+ .expect("source task did not observe both closed outputs");1790+ }1791+1792+ 1793+ async fn reaping_stalled_source_aborts_after_bounded_grace() {17801794 let (_frame_sender, frame_receiver) = tokio::sync::watch::channel(None);17811795 let (_gop_sender, gop_receiver) = tokio::sync::watch::channel(None);17821796 let (started_tx, started_rx) = tokio::sync::oneshot::channel();@@ -1794,10 +1808,10 @@ mod tests {
17941808 };17951809 17961810 started_rx.await.unwrap();1797- reap_encoded_source(source).await;1811+ reap_encoded_source_with_timeout(source, Duration::from_millis(10)).await;17981812 tokio::time::timeout(Duration::from_millis(100), reaped_rx)17991813 .await1800- .expect("unselected source task was not reaped")1814+ .expect("stalled source task was not aborted and reaped")18011815 .unwrap();18021816 }18031817