diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index ec967fcd..332abd9c 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -1550,6 +1550,11 @@ pub struct EncodeSettingsChange { /// keyframe in the same step, so the first frame after it is one the client /// can decode: everything it saw before the pause is stale. fn apply_pause_message(msg: u8, paused: &AtomicBool, idr_requested: &AtomicBool) { + log::info!( + "window {} received (was {})", + if msg == MSG_WINDOW_PAUSE { "pause" } else { "resume" }, + if paused.load(Ordering::Relaxed) { "paused" } else { "running" } + ); if msg == MSG_WINDOW_PAUSE { if !paused.swap(true, Ordering::Relaxed) { log::info!("paused by the client: capture and encode stopped"); diff --git a/apps/neshub/src/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index 8220cae7..82b5659c 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/apps/neshub/src/e2e_tests.rs @@ -578,6 +578,28 @@ mod windows { assert!(got.iter().all(|g| (g.0, g.1) == (0, 0))); } + /// The order a real box has: the client subscribes to the window when it + /// opens, and the game's encoder draws its first frame afterwards. The + /// client's encoder must follow the frames, or a pause has nowhere to go + /// (found on a real box, ref(d-0087)). + #[tokio::test(flavor = "multi_thread")] + async fn a_client_that_subscribed_before_the_first_frame_learns_its_encoder() { + let rig = rig().await; + tell(&rig, |m| encode_event_message(m, &opened(0, "launch_a"))).await; + tell(&rig, |m| encode_window_pid(m, 0, GAME_A)).await; + let me = rig._client_ep.addr().id; + let (idr, _rx) = tokio::sync::mpsc::unbounded_channel(); + rig.mgr.subscribe(me, 0, &idr).await; + assert_eq!(rig.mgr.encoder_pid_of(&me).await, Some(0), "nothing has drawn yet"); + draw(&rig, GAME_A, 0, true).await; + draw(&rig, GAME_A, 1, false).await; + let _ = received(&rig.client).await; + assert_eq!(rig.mgr.encoder_pid_of(&me).await, Some(GAME_A)); + let handle = rig.mgr.pause_handle_of(&me).await.expect("a session"); + handle.pause(GAME_A); + assert!(rig.mgr.encoder_paused(GAME_A)); + } + #[tokio::test(flavor = "multi_thread")] async fn a_session_that_goes_while_paused_releases_its_pause() { let rig = rig().await; diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 99eee293..e72fe0e8 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -305,7 +305,18 @@ async fn main() -> Result<()> { if destinations.first() != Some(&cmd_path) { for path in destinations { match sock.send_to(&bytes, &path) { - Ok(_) => {} + Ok(_) => { + if matches!( + bytes.first(), + Some(&nesprotocol::MSG_WINDOW_PAUSE | &nesprotocol::MSG_WINDOW_RESUME) + ) { + tracing::info!( + "sent {} to {}", + if bytes[0] == nesprotocol::MSG_WINDOW_PAUSE { "pause" } else { "resume" }, + path.display() + ); + } + } // Refused or absent: that process has gone. Err(e) if capture_cmd::encoder_is_gone(&e) => { tracing::info!("encoder at {} is gone: {e}", path.display()); diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 4bcfd0ab..e3108251 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -863,10 +863,13 @@ async fn run_control_reader( // does not know that encoder, and there is nothing to pause. match pid_or_all(&subscribe.2) { Some(pid) => { - info!("client pauses encoder {pid}"); + info!(client = %subscribe.0.fmt_short(), "pause: client pauses encoder {pid}"); pause.pause(pid); } - None => debug!("pause from a client with no encoder yet; ignored"), + None => info!( + client = %subscribe.0.fmt_short(), + "pause: no encoder known for this client yet; ignored" + ), } } MSG_WINDOW_RESUME => { @@ -874,6 +877,11 @@ async fn run_control_reader( // paused nothing starts nothing; the keyframe is asked for // either way, since the client restarted its picture. awaiting_keyframe.store(true, Ordering::Relaxed); + info!( + client = %subscribe.0.fmt_short(), + "resume: encoder {:?}", + pid_or_all(&subscribe.2) + ); pause.resume(); if let Some(pid) = pid_or_all(&subscribe.2) { let _ = idr_cmd_tx.send(keyframe_request(Some(pid))); @@ -1290,6 +1298,15 @@ impl SessionManager { self.sessions.lock().await.get(id).map(|s| s.pause.clone()) } + #[cfg(test)] + pub(crate) async fn encoder_pid_of(&self, id: &iroh::EndpointId) -> Option { + self.sessions + .lock() + .await + .get(id) + .map(|s| s.encoder_pid.load(Ordering::Relaxed)) + } + #[cfg(test)] pub(crate) fn encoder_paused(&self, pid: u32) -> bool { self.pauses.lock().unwrap().paused(pid) @@ -1493,7 +1510,20 @@ impl SessionManager { } } for session in sessions.values() { - match route(session.subscription(), window, note.primary) { + let routed = route(session.subscription(), window, note.primary); + // A client shown this frame is shown this encoder, whenever it + // subscribed. The pid used to be fixed at subscribe time, and a + // client that subscribed before the game drew its first frame (the + // usual order: the window opens, the client subscribes, the + // encoder draws) kept pid 0 for good, so a pause had no encoder + // to go to and was dropped, ref(d-0087). + if routed.is_some() + && let Some(pid) = source + && session.encoder_pid.load(Ordering::Relaxed) != pid + { + session.encoder_pid.store(pid, Ordering::Relaxed); + } + match routed { Some(WINDOW_DEFAULT) => session.send_video_frame(MediaItem::Coded(data.clone())), Some(w) => session.send_video_frame(MediaItem::Windowed(w, data.clone())), None => {}