Share play: the picture follows the rate ladder during a share (T38)

crates/net/src/play.rs

@@ -24,7 +24,7 @@ use anyplay_control::{Link, ShareStats};
 use anyplay_share::control::{self as wire, Role};
 use anyplay_share::link::{FromWorker, RingReady, Start, ToWorker, WorkerStats};
 use anyplay_share::pad::{PadSender, TYPE_PAD};
-use anyplay_share::rate::{PathKind, RateConfig, RateController, RateInput};
+use anyplay_share::rate::{LADDER, PathKind, RateConfig, RateController, RateInput, Target};
 use bytes::Bytes;
 use futures_util::StreamExt;
 use iroh::endpoint::Connection;
@@ -384,6 +384,44 @@ fn path_kind(link: Link) -> PathKind {
     }
 }
 
+/// The lowest rate the encoder is given while the ladder pauses the video: the picture does
+/// not stop, it only gets as small as it can.
+const PAUSED_KBPS: u32 = 300;
+
+/// What a new target asks of the host's worker.
+#[derive(Debug, PartialEq, Eq)]
+enum Retune {
+    Keep,
+    /// Only the bitrate or the loss protection changed: the running encoder follows.
+    Rate,
+    /// The size or the frame rate changed: a new epoch, a new encoder, a key frame.
+    Shape,
+}
+
+/// Moves the worker's setup to `target` and says what the worker has to be told.
+fn retune(start: &mut Start, target: &Target) -> Retune {
+    let kbps = if target.paused {
+        target.kbps.max(PAUSED_KBPS)
+    } else {
+        target.kbps
+    };
+    let shape = (target.width, target.height, target.fps);
+    let result = if shape != (start.width, start.height, start.fps) {
+        start.epoch = start.epoch.wrapping_add(1);
+        start.width = target.width;
+        start.height = target.height;
+        start.fps = target.fps;
+        Retune::Shape
+    } else if (kbps, target.parity_percent) != (start.kbps, start.parity_percent) {
+        Retune::Rate
+    } else {
+        return Retune::Keep;
+    };
+    start.kbps = kbps;
+    start.parity_percent = target.parity_percent;
+    result
+}
+
 /// Whether a mode gives the guest a pad.
 fn has_pad(mode: ShareMode) -> bool {
     matches!(mode, ShareMode::HandOver | ShareMode::Together)
@@ -887,11 +925,13 @@ impl Task {
         let launch = PlayLaunch {
             role: Role::Host,
             title: active.bind.title.clone(),
+            // The ring has the size of the ladder's top rung, so that the picture can go up the
+            // ladder as well as down; the worker scales each picture to the rung it streams.
             ring: Some(RingLaunch {
                 name: anyplay_share::ring::name_from_random(random),
-                width: target.width,
-                height: target.height,
-                fps: target.fps,
+                width: LADDER[0].width,
+                height: LADDER[0].height,
+                fps: LADDER[0].fps,
             }),
         };
         let port = match self.workers.launch_play(&launch) {
@@ -1053,8 +1093,8 @@ impl Task {
             .max(anyplay_share::packetiser::MIN_PAYLOAD)
     }
 
-    /// The guest's report feeds the controller; its target goes to the worker. The picture
-    /// keeps the size of the frame ring the game fills, so only the rate follows.
+    /// The guest's report feeds the controller; its target goes to the worker: a new rate at
+    /// once, a new size or frame rate as a new epoch that the guest hears of in a `config`.
     fn guest_reports(&mut self, report: wire::Report) {
         let now_us = self.now_us();
         let Some(active) = &mut self.active else {
@@ -1078,13 +1118,18 @@ impl Task {
         let Some(start) = &mut worker.start else {
             return;
         };
-        if (target.kbps, target.parity_percent) != (start.kbps, start.parity_percent) {
-            start.kbps = target.kbps;
-            start.parity_percent = target.parity_percent;
-            let _ = worker.to.send(ToWorker::Rate {
-                kbps: target.kbps,
-                parity_percent: target.parity_percent,
-            });
+        match retune(start, &target) {
+            Retune::Keep => {}
+            Retune::Rate => {
+                let _ = worker.to.send(ToWorker::Rate {
+                    kbps: start.kbps,
+                    parity_percent: start.parity_percent,
+                });
+            }
+            Retune::Shape => {
+                let _ = worker.to.send(ToWorker::Start(start.clone()));
+                self.send_config();
+            }
         }
     }
 
@@ -1454,3 +1499,81 @@ async fn dial(endpoint: Endpoint, id: EndpointId, relay: Option<RelayUrl>, net:
         pause = (pause * 2).min(LONGEST_PAUSE);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn start() -> Start {
+        Start {
+            epoch: 0,
+            codec: "h264".into(),
+            width: 1280,
+            height: 720,
+            fps: 30,
+            kbps: 4_000,
+            parity_percent: 0,
+            max_payload: 1180,
+        }
+    }
+
+    fn target(width: u32, height: u32, fps: u32, kbps: u32) -> Target {
+        Target {
+            kbps,
+            width,
+            height,
+            fps,
+            parity_percent: 0,
+            paused: false,
+            generation: 0,
+        }
+    }
+
+    #[test]
+    fn the_same_target_changes_nothing() {
+        let mut s = start();
+        assert_eq!(retune(&mut s, &target(1280, 720, 30, 4_000)), Retune::Keep);
+        assert_eq!(s, start());
+    }
+
+    #[test]
+    fn a_new_rate_alone_keeps_the_epoch() {
+        let mut s = start();
+        let mut t = target(1280, 720, 30, 3_200);
+        t.parity_percent = 10;
+        assert_eq!(retune(&mut s, &t), Retune::Rate);
+        assert_eq!((s.epoch, s.kbps, s.parity_percent), (0, 3_200, 10));
+    }
+
+    #[test]
+    fn a_new_size_or_frame_rate_starts_a_new_epoch() {
+        let mut s = start();
+        assert_eq!(retune(&mut s, &target(960, 540, 30, 2_000)), Retune::Shape);
+        assert_eq!(
+            (s.epoch, s.width, s.height, s.fps, s.kbps),
+            (1, 960, 540, 30, 2_000)
+        );
+        assert_eq!(retune(&mut s, &target(1280, 720, 60, 6_000)), Retune::Shape);
+        assert_eq!((s.epoch, s.height, s.fps), (2, 720, 60));
+        // the codec and the datagram size stay
+        assert_eq!((s.codec.as_str(), s.max_payload), ("h264", 1180));
+    }
+
+    #[test]
+    fn the_epoch_wraps() {
+        let mut s = start();
+        s.epoch = u8::MAX;
+        assert_eq!(retune(&mut s, &target(960, 540, 30, 2_000)), Retune::Shape);
+        assert_eq!(s.epoch, 0);
+    }
+
+    #[test]
+    fn a_paused_ladder_keeps_a_small_picture_going() {
+        let mut s = start();
+        let mut t = target(960, 540, 30, 0);
+        t.paused = true;
+        assert_eq!(retune(&mut s, &t), Retune::Shape);
+        assert_eq!(s.kbps, PAUSED_KBPS);
+        assert_eq!(retune(&mut s, &t), Retune::Keep, "it does not flap");
+    }
+}

crates/net/tests/play.rs

@@ -23,8 +23,8 @@ use anyplay_net::screen::{PlayLaunch, WorkerPort, Workers};
 use anyplay_net::traffic::Traffic;
 use anyplay_net::tunnel::{Options, RelayAccess, Tunnels};
 use anyplay_share::codec::{Frame as Picture, VideoDecoder, VideoEncoder};
-use anyplay_share::control::Role;
-use anyplay_share::link::{Capabilities, FromWorker};
+use anyplay_share::control::{Report, Role};
+use anyplay_share::link::{Capabilities, FromWorker, Start, ToWorker};
 use anyplay_share::pad::{PadState, button};
 use anyplay_share::ring::RingWriter;
 use anyplay_share::testcodec::{TestDecoder, TestEncoder, TestPattern};
@@ -69,9 +69,40 @@ struct TestWorkers {
     launches: Mutex<Vec<PlayLaunch>>,
     /// Where the guest's viewer would put the pad it read.
     pad: Mutex<Option<std::sync::mpsc::Sender<FromWorker>>>,
+    /// Every setup the net service gave a worker, with the role of the worker.
+    setups: Mutex<Vec<(Role, Start)>>,
 }
 
 impl TestWorkers {
+    /// The setups a role's worker was given, in order.
+    fn setups(&self, role: Role) -> Vec<Start> {
+        self.setups
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|(r, _)| *r == role)
+            .map(|(_, start)| start.clone())
+            .collect()
+    }
+
+    /// The guest's viewer reports: `lost` of every `received + lost` packets went missing.
+    fn report(&self, received: u32, lost: u32) {
+        let pad = self.pad.lock().unwrap();
+        pad.as_ref()
+            .expect("a guest worker runs")
+            .send(FromWorker::Report(Report {
+                frame: 0,
+                received,
+                lost,
+                recovered: 0,
+                delay_trend_us: 0,
+                queue_ms: 0,
+                decoded_fps: 30,
+                shown_fps: 30,
+            }))
+            .unwrap();
+    }
+
     fn press(&self, buttons: u32) {
         let pad = self.pad.lock().unwrap();
         pad.as_ref()
@@ -117,8 +148,12 @@ impl Workers for Shared {
         if role == Role::Guest {
             *self.0.pad.lock().unwrap() = Some(worker_out.clone());
         }
+        let workers = self.0.clone();
         std::thread::spawn(move || {
             while let Some(message) = to_rx.blocking_recv() {
+                if let ToWorker::Start(start) = &message {
+                    workers.setups.lock().unwrap().push((role, start.clone()));
+                }
                 if worker_in_tx.send(message).is_err() {
                     return;
                 }
@@ -649,3 +684,51 @@ async fn a_game_that_cannot_open_the_ring_ends_the_share() {
         host.saw(|s| matches!(s, Seen::Notice(message) if message.contains("would not share")))
     );
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn loss_steps_the_picture_down_the_ladder_and_the_guest_follows() {
+    let (_relay, host, guest, _game) = playing(ShareMode::Watch).await;
+    // The game fills a ring of the ladder's top rung, whatever the stream starts at.
+    let ring = host.workers.launches.lock().unwrap()[0]
+        .ring
+        .clone()
+        .unwrap();
+    assert_eq!((ring.width, ring.height, ring.fps), (1920, 1080, 60));
+    let first = host.workers.setups(Role::Host)[0].clone();
+    assert_eq!(first.epoch, 0);
+    assert!(first.height >= 720, "the stream starts high: {first:?}");
+    let shown_before = guest.workers.shown.lock().unwrap().numbers.len();
+
+    // The guest reports heavy loss for a few seconds.
+    for _ in 0..50 {
+        guest.workers.report(50, 8);
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    eventually("the host's worker to be given the smaller picture", || {
+        host.workers
+            .setups(Role::Host)
+            .last()
+            .is_some_and(|s| s.height == 540 && s.epoch > 0)
+    })
+    .await;
+    let host_setup = host.workers.setups(Role::Host).last().cloned().unwrap();
+    assert_eq!((host_setup.width, host_setup.fps), (960, 30));
+
+    // The guest hears it in a config and gives its viewer the same epoch and size.
+    eventually("the guest's viewer to follow", || {
+        guest
+            .workers
+            .setups(Role::Guest)
+            .last()
+            .is_some_and(|s| (s.epoch, s.width, s.height) == (host_setup.epoch, 960, 540))
+    })
+    .await;
+
+    // And the pictures go on, whole.
+    eventually("pictures after the step", || {
+        guest.workers.shown.lock().unwrap().numbers.len() > shown_before + 20
+    })
+    .await;
+    let shown = guest.workers.shown.lock().unwrap();
+    assert_eq!(shown.damaged, 0, "a damaged picture was shown");
+}

crates/share/src/rate.rs

@@ -179,6 +179,8 @@ pub struct RateController {
     window: VecDeque<(u64, u32, u32)>,
     parity: u8,
     generation: u32,
+    /// The first rung the player's picture-size limit allows (`share_quality`).
+    top_rung: usize,
 }
 
 fn rung_for(budget: f64) -> usize {
@@ -205,12 +207,33 @@ impl RateController {
             window: VecDeque::new(),
             parity: 0,
             generation: 0,
+            top_rung: 0,
         };
         c.budget = c.budget.min(c.cap());
         c.rung = rung_for(c.budget);
         c
     }
 
+    /// The tallest picture the player allows (`share_quality`); 0 lifts the limit. The ladder
+    /// never goes above the first rung that fits, and this limit alone never pauses the video:
+    /// a limit under 540 still gives 540p.
+    pub fn set_max_height(&mut self, max_height: u32) {
+        self.top_rung = if max_height == 0 {
+            0
+        } else {
+            LADDER
+                .iter()
+                .position(|r| r.height <= max_height)
+                .unwrap_or(LADDER.len() - 1)
+        };
+        if self.rung < self.top_rung {
+            self.rung = self.top_rung;
+            self.below_since = None;
+            self.above_since = None;
+            self.generation += 1;
+        }
+    }
+
     /// The path changed (direct to relayed or back).
     pub fn set_path(&mut self, path: PathKind) {
         self.path = path;
@@ -349,7 +372,7 @@ impl RateController {
     }
 
     fn update_rung(&mut self, now: u64) {
-        let want = rung_for(self.budget);
+        let want = rung_for(self.budget).max(self.top_rung);
         let before = self.rung;
         if want > self.rung {
             // The budget is below the current rung.
@@ -791,4 +814,35 @@ mod tests {
         assert!(ts.iter().all(|t| t.height <= 720 && t.kbps <= 6_000));
         assert_eq!(ts.last().unwrap().fps, 60);
     }
+
+    #[test]
+    fn the_players_height_limit_holds_the_ladder_down_and_lifts_again() {
+        let mut c = RateController::new(RateConfig::default(), PathKind::Direct, 0);
+        let mut link = Link {
+            capacity_kbps: 50_000.0,
+            queue_kbit: 0.0,
+            buffer_ms: 100.0,
+        };
+        let a = simulate(&mut c, &mut link, 0, 60);
+        assert_eq!(a.last().unwrap().height, 1080);
+        let before = a.last().unwrap().generation;
+        c.set_max_height(720);
+        let t = c.target();
+        assert_eq!(t.height, 720, "the limit applies at once");
+        assert!(t.generation > before, "a new epoch starts");
+        let b = simulate(&mut c, &mut link, 60, 120);
+        assert!(b.iter().all(|t| t.height <= 720), "stays under the limit");
+        assert_eq!(b.last().unwrap().fps, 60);
+        c.set_max_height(0);
+        let d = simulate(&mut c, &mut link, 120, 200);
+        assert_eq!(d.last().unwrap().height, 1080, "the limit is lifted");
+    }
+
+    #[test]
+    fn a_height_limit_under_the_last_rung_gives_the_last_rung_not_a_pause() {
+        let mut c = RateController::new(RateConfig::default(), PathKind::Direct, 0);
+        c.set_max_height(240);
+        let t = c.target();
+        assert_eq!((t.height, t.paused), (540, false));
+    }
 }