diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index 3b41e512..0c3913ab 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -55,7 +55,8 @@ use std::time::Instant; use nesprotocol::{ CODEC_AV1, CODEC_H264, CODEC_H265, CODEC_KEEP, CODEC_PYROWAVE, ClientCaps, FLAG_KEYFRAME, - FLAG_RECONFIG, MSG_CLIENT_CAPS, MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, STREAM_VIDEO, + FLAG_RECONFIG, MSG_CLIENT_CAPS, MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, MSG_WINDOW_PAUSE, + MSG_WINDOW_RESUME, STREAM_VIDEO, decode_client_caps, decode_encode_settings_ext, encode_ipc_frame_from, }; use pixelforge::{ @@ -998,6 +999,10 @@ pub struct PipelineHandle { stall_probe: StallProbe, progress_epoch: Instant, idr_requested: Arc, + /// Set by a pause from the hub (ref d-0087): the present hook captures + /// nothing while it is, so nothing is blitted, encoded or sent. The game + /// is not touched. + paused: Arc, shutdown: Arc, pub capture_fps: Arc, pub encode_avg_ms: Arc, @@ -1089,6 +1094,8 @@ impl PipelineHandle { shutdown.clone(), ); let idr_requested = Arc::new(AtomicBool::new(false)); + let paused = Arc::new(AtomicBool::new(false)); + let paused_listener = paused.clone(); let capture_fps = Arc::new(AtomicU32::new(0)); let encode_avg_ms = Arc::new(AtomicU32::new(0)); let capture_ms = Arc::new(AtomicU32::new(0)); @@ -1229,6 +1236,9 @@ impl PipelineHandle { let mut buf = [0u8; 128]; loop { match sock.recv(&mut buf) { + Ok(1) if buf[0] == MSG_WINDOW_PAUSE || buf[0] == MSG_WINDOW_RESUME => { + apply_pause_message(buf[0], &paused_listener, &idr_thread); + } Ok(1) if buf[0] == MSG_IDR_REQUEST => { log::info!("IDR requested by client"); idr_thread.store(true, Ordering::Relaxed); @@ -1394,6 +1404,7 @@ impl PipelineHandle { stall_probe, progress_epoch, idr_requested, + paused, shutdown, capture_fps, encode_avg_ms, @@ -1468,6 +1479,17 @@ impl PipelineHandle { true } + /// Whether the window is paused: the hook captures nothing. + pub fn is_paused(&self) -> bool { + self.paused.load(Ordering::Relaxed) + } + + /// Say the capture thread is alive while paused, so the stall watchdog + /// does not report a stopped capture that was stopped on purpose. + pub fn note_paused(&self) { + self.capture_progress.note(self.progress_epoch, 7); + } + pub fn request_idr(&self) { self.idr_requested.store(true, Ordering::Relaxed); } @@ -1524,6 +1546,24 @@ pub struct EncodeSettingsChange { pub pyro: Option, } +/// Apply a pause or resume from the hub. Resume clears the flag and asks for a +/// 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) { + if msg == MSG_WINDOW_PAUSE { + if !paused.swap(true, Ordering::Relaxed) { + log::info!("paused by the client: capture and encode stopped"); + } + } else if msg == MSG_WINDOW_RESUME { + // The keyframe first: the flag lets the next present through, and the + // encoder must already know that frame is to be an IDR. + idr_requested.store(true, Ordering::Relaxed); + if paused.swap(false, Ordering::Relaxed) { + log::info!("resumed by the client: keyframe requested"); + } + } +} + fn encoder_thread( mut cfg: EncoderConfig, frame_rx: mpsc::Receiver, @@ -3848,6 +3888,22 @@ mod encoder_identity_tests { assert!(encoder_still_serves(HD, HD)); } + #[test] + fn pause_stops_and_resume_starts_with_a_keyframe() { + let paused = AtomicBool::new(false); + let idr = AtomicBool::new(false); + apply_pause_message(MSG_WINDOW_PAUSE, &paused, &idr); + assert!(paused.load(Ordering::Relaxed)); + assert!(!idr.load(Ordering::Relaxed), "a pause asks for nothing"); + apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); + assert!(!paused.load(Ordering::Relaxed)); + assert!(idr.load(Ordering::Relaxed)); + // A resume that was never paused is still a keyframe request, harmless. + idr.store(false, Ordering::Relaxed); + apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); + assert!(idr.load(Ordering::Relaxed)); + } + #[test] fn a_resolution_change_rebuilds_in_either_direction() { // The regression. Shrinking left the encoder sending the old geometry diff --git a/apps/nescapture/src/present.rs b/apps/nescapture/src/present.rs index c5650223..34b0ba5e 100644 --- a/apps/nescapture/src/present.rs +++ b/apps/nescapture/src/present.rs @@ -213,6 +213,17 @@ unsafe fn try_capture( (images[image_index], fmt, ext, images.len()) }; + // A paused window captures nothing (ref d-0087): no blit, no slot, no + // encode. Before the gate, so the gate's own bookkeeping does not run + // either, and the frame the client resumes on is a fresh one. + if let Ok(enc) = ds.encoder.lock() + && let Some(ref h) = *enc + && h.is_paused() + { + h.note_paused(); + return None; + } + // Gate before any GPU work is queued. A game presenting faster than the // target would otherwise pay a full blit for frames the // encoder throws away moments later. diff --git a/apps/neshub/src/capture_cmd.rs b/apps/neshub/src/capture_cmd.rs index 32cf9d55..66492112 100644 --- a/apps/neshub/src/capture_cmd.rs +++ b/apps/neshub/src/capture_cmd.rs @@ -44,6 +44,17 @@ pub fn keyframe_request(pid: Option) -> Vec { to_encoder(pid, &[MSG_IDR_REQUEST]) } +/// Stop `pid`'s encoder capturing and encoding; it keeps its game running. +/// Never replayed to a later encoder: it is about one window, not a setting. +pub fn window_pause(pid: Option) -> Vec { + to_encoder(pid, &[nesprotocol::MSG_WINDOW_PAUSE]) +} + +/// Start `pid`'s encoder again; it answers with a keyframe. +pub fn window_resume(pid: Option) -> Vec { + to_encoder(pid, &[nesprotocol::MSG_WINDOW_RESUME]) +} + /// Split a queued command into its target and the command itself. pub fn target_of(cmd: Vec) -> (Option, Vec) { if cmd.len() >= 5 && cmd[0] == TO_ENCODER { @@ -221,6 +232,25 @@ mod tests { assert_eq!(f.encoder(11), Some(vec![vec![MSG_IDR_REQUEST]])); } + #[test] + fn pause_and_resume_go_to_one_encoder_and_are_not_replayed() { + let mut f = Fanout::default(); + f.encoder(10); + f.encoder(20); + let (target, cmd) = target_of(window_pause(Some(20))); + assert_eq!(cmd, vec![nesprotocol::MSG_WINDOW_PAUSE]); + assert_eq!( + f.destinations_for(target), + vec![PathBuf::from("/tmp/nescapture-cmd-20.sock")] + ); + let (target, cmd) = target_of(window_resume(Some(20))); + assert_eq!(cmd, vec![nesprotocol::MSG_WINDOW_RESUME]); + assert_eq!(target, Some(20)); + // Targeted commands are not recorded, so a later encoder is not + // born paused. + assert_eq!(f.encoder(30).map(|r| r.len()), Some(1)); + } + #[test] fn pid_from_path() { assert_eq!( diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index c615548d..692b95cc 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -14,13 +14,13 @@ use nesprotocol::{ControlMode, ReceiverReport, decode_control_mode, decode_recei use nesprotocol::{FRAME_HDR_LEN, STREAM_VERSION, encode_frame}; use nesprotocol::{ MSG_CLIENT_CAPS, MSG_CONTROL_MODE, MSG_ENCODE_SETTINGS, MSG_GAMEPAD, MSG_GAMEPAD_FEEDBACK, - MSG_IDR_REQUEST, MSG_INPUT_BATCH, MSG_RECEIVER_REPORT, + MSG_IDR_REQUEST, MSG_INPUT_BATCH, MSG_RECEIVER_REPORT, MSG_WINDOW_PAUSE, MSG_WINDOW_RESUME, }; use crate::control::{Controller, PathView}; use crate::negotiation::{Negotiator, effective_caps}; -use crate::capture_cmd::keyframe_request; +use crate::capture_cmd::{keyframe_request, window_pause, window_resume}; use crate::dgram::{MediaItem, run_datagram_writer}; use crate::windows::{Source, Subscription, WindowTable, route, wants_cursor}; use nesprotocol::window::{MSG_WINDOW_SUBSCRIBE, STREAM_WINDOWS, WINDOW_DEFAULT, WindowEvent, WindowId}; @@ -843,6 +843,28 @@ async fn run_control_reader( // ref(d-0085): the other windows are fine. let _ = idr_cmd_tx.send(keyframe_request(pid_or_all(&subscribe.2))); } + MSG_WINDOW_PAUSE => { + // To the encoder of the window this client is shown, and + // to nobody else. Until the client has been sent a frame + // the hub does not know that encoder, and "every encoder" + // is the wrong answer for a pause: it would stop windows + // nobody asked about. + match pid_or_all(&subscribe.2) { + Some(pid) => { + info!("client pauses encoder {pid}"); + let _ = idr_cmd_tx.send(window_pause(Some(pid))); + } + None => debug!("pause from a client with no encoder yet; ignored"), + } + } + MSG_WINDOW_RESUME => { + // The same gate as a keyframe request: what arrives before + // the keyframe cannot be decoded and starves it. + awaiting_keyframe.store(true, Ordering::Relaxed); + let pid = pid_or_all(&subscribe.2); + info!("client resumes encoder {pid:?}"); + let _ = idr_cmd_tx.send(window_resume(pid)); + } MSG_CLIENT_CAPS => { // Passed through untouched, and in particular *without* // taking the controller off automatic the way encode diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs index 15180c9a..4650a570 100644 --- a/crates/nesprotocol/src/lib.rs +++ b/crates/nesprotocol/src/lib.rs @@ -103,6 +103,15 @@ pub const MSG_IDR_REQUEST: u8 = 0x10; // request a keyframe (desktop → hub → pub const MSG_ENCODE_SETTINGS: u8 = 0x12; // change encoder settings (desktop → hub → nescapture) pub const MSG_CLIENT_CAPS: u8 = 0x15; // what the client can decode (desktop → hub → nescapture) pub const MSG_SURFACE_COLOR: u8 = 0x16; // what the compositor was told a surface is (nescope → nescapture) +/// Stop capturing and encoding the window this client is shown (desktop → hub +/// → that window's nescapture), ref(d-0087). The game keeps running; nothing is +/// sent for the window until [`MSG_WINDOW_RESUME`]. No payload: the hub already +/// knows which window the client is subscribed to, as it does for +/// [`MSG_IDR_REQUEST`]. An old hub logs it as an unknown control message. +pub const MSG_WINDOW_PAUSE: u8 = 0x18; +/// Start again, with a keyframe within a second (desktop → hub → nescapture). +/// Resume is also what forces the keyframe, so the client need not ask for one. +pub const MSG_WINDOW_RESUME: u8 = 0x19; /// What the receiver actually got, once a second (desktop → hub). /// /// The hub cannot see this. Its own view of the path -- RTT, congestion window, @@ -1009,6 +1018,9 @@ mod media_control_tests { ("MSG_ENCODE_SETTINGS", MSG_ENCODE_SETTINGS), ("MSG_CLIENT_CAPS", MSG_CLIENT_CAPS), ("MSG_SURFACE_COLOR", MSG_SURFACE_COLOR), + ("MSG_WINDOW_PAUSE", MSG_WINDOW_PAUSE), + ("MSG_WINDOW_RESUME", MSG_WINDOW_RESUME), + ("MSG_WINDOW_SUBSCRIBE", window::MSG_WINDOW_SUBSCRIBE), ("MSG_RECEIVER_REPORT", MSG_RECEIVER_REPORT), ("MSG_CONTROL_MODE", MSG_CONTROL_MODE), ("MSG_INPUT_BATCH", MSG_INPUT_BATCH),