remote trackpad and keyboard server
git clone https://git.lucas.co/cce-remote.git
src/stream.rs (16K)
1 //! Live-view delivery: latest-frame-wins, clocked by client acks.
2 //!
3 //! The failure this module exists to prevent: the original MJPEG path pushed
4 //! every frame, in order, into a blocking TCP write. Nothing between the
5 //! compositor and the phone ever dropped a frame, so the kernel's send buffer
6 //! (hundreds of KB, ~10-30 frames) became a queue — the moment wifi throughput
7 //! dipped below the frame rate, the queue filled, and every frame the phone
8 //! showed was queue-depth old. Latency accumulated and never drained: "fine at
9 //! first, unusable after a short time".
10 //!
11 //! The fix is sender-side flow control, the same shape VNC/RDP use:
12 //!
13 //! - A `Slot` holds only the NEWEST frame from the source (winstream →
14 //! screencopy → grim, tried in that order by `spawn_producer`). Overwriting
15 //! is the drop point: stale frames cease to exist before they cost anything.
16 //! - `run_ws_sender` sends one frame, then waits for the page's `n` ack before
17 //! sending the newest frame available. In-flight is capped at ONE frame, so
18 //! degraded wifi costs frame RATE, never growing latency.
19 //! - Encoding happens at send time, only for frames actually sent, at a
20 //! (max_edge, jpeg_quality) picked by `adapt()` from the measured send→ack
21 //! time — readable resolution when the link allows, graceful degradation
22 //! when it doesn't.
23 //!
24 //! `/stream` (MJPEG over HTTP) remains as a curl-debuggable endpoint via
25 //! `run_mjpeg_sender`, thin over the same slot; the page no longer uses it.
26
27 use std::io::Write;
28 use std::sync::atomic::{AtomicBool, Ordering};
29 use std::sync::{Arc, Condvar, Mutex};
30 use std::time::{Duration, Instant};
31
32 /// One captured frame. `Raw` is encoded at send time; `Jpeg` (grim fallback)
33 /// is passed through as-is, so adaptation does not apply to it.
34 #[derive(Clone)]
35 pub enum Payload {
36 Raw { data: Vec<u8>, w: u32, h: u32, stride: u32, rgb: (usize, usize, usize) },
37 Jpeg(Vec<u8>),
38 }
39
40 /// The latest-wins seam between the frame source and however many senders are
41 /// consuming it. `push` overwrites; senders track the last seq they delivered.
42 pub struct Slot {
43 inner: Mutex<Inner>,
44 cv: Condvar,
45 }
46
47 struct Inner {
48 seq: u64,
49 frame: Option<Payload>,
50 alive: bool,
51 }
52
53 impl Slot {
54 pub fn new() -> Arc<Self> {
55 Arc::new(Self {
56 inner: Mutex::new(Inner { seq: 0, frame: None, alive: true }),
57 cv: Condvar::new(),
58 })
59 }
60
61 pub fn push(&self, p: Payload) {
62 let mut g = self.inner.lock().unwrap();
63 g.seq += 1;
64 g.frame = Some(p);
65 self.cv.notify_all();
66 }
67
68 /// The producer died (source unavailable/broke). Senders return, the
69 /// client reconnects, and the fresh producer re-picks a source.
70 pub fn close(&self) {
71 self.inner.lock().unwrap().alive = false;
72 self.cv.notify_all();
73 }
74
75 /// Newest frame with seq > `last`: Ok(Some) on a new frame, Ok(None) on
76 /// timeout (window idle — sources force keepalive frames ≤20s), Err when
77 /// the producer is gone.
78 pub fn wait_newer(&self, last: u64, timeout: Duration) -> Result<Option<(u64, Payload)>, ()> {
79 let deadline = Instant::now() + timeout;
80 let mut g = self.inner.lock().unwrap();
81 loop {
82 if g.seq > last {
83 return Ok(Some((g.seq, g.frame.clone().unwrap())));
84 }
85 if !g.alive {
86 return Err(());
87 }
88 let now = Instant::now();
89 if now >= deadline {
90 return Ok(None);
91 }
92 let (ng, _) = self.cv.wait_timeout(g, deadline - now).unwrap();
93 g = ng;
94 }
95 }
96 }
97
98 fn encode(p: &Payload, max_edge: u32, quality: u8) -> Result<Vec<u8>, String> {
99 match p {
100 Payload::Raw { data, w, h, stride, rgb } => {
101 crate::screencopy::downscale_encode(data, *w, *h, *stride, *rgb, max_edge, quality)
102 }
103 Payload::Jpeg(b) => Ok(b.clone()),
104 }
105 }
106
107 // ---- adaptation ------------------------------------------------------------
108
109 /// (max_edge px, jpeg quality), best first. The point of the live view is
110 /// READING the window, so the ladder gives up size grudgingly: it alternates
111 /// size and quality steps and lets quality RISE where size falls (840,58 after
112 /// 1120,55), rather than exhausting one axis before touching the other.
113 pub const LADDER: &[(u32, u8)] = &[
114 (1400, 68),
115 (1120, 68),
116 (1120, 55),
117 (840, 58),
118 (840, 46),
119 (560, 48),
120 ];
121 pub const START_LEVEL: usize = 1;
122
123 /// Pick the next ladder level from the smoothed send→ack time. Downgrades are
124 /// immediate (lag is being felt NOW); upgrades need `acks_since_change` of
125 /// stability so the level does not oscillate at a threshold. The dead band
126 /// between the two thresholds is the hysteresis.
127 pub fn adapt(level: usize, ewma_ms: f64, acks_since_change: u32) -> usize {
128 if ewma_ms > 220.0 {
129 (level + 1).min(LADDER.len() - 1)
130 } else if ewma_ms < 90.0 && acks_since_change >= 10 {
131 level.saturating_sub(1)
132 } else {
133 level
134 }
135 }
136
137 // ---- the frame source ------------------------------------------------------
138
139 /// Source preference is unchanged from the MJPEG design: compositor window
140 /// stream (per-window damage, follows focus) → screencopy (output damage) →
141 /// grim. The producer owns the source; on source death it closes the slot.
142 pub fn spawn_producer(slot: Arc<Slot>, stop: Arc<AtomicBool>) {
143 std::thread::spawn(move || {
144 let result = match crate::winstream::Reader::connect() {
145 Ok(mut r) => produce_winstream(&mut r, &slot, &stop),
146 Err(e) => {
147 eprintln!("[cce-remote] window stream unavailable ({e}), trying screencopy");
148 match crate::screencopy::CaptureSession::new() {
149 Ok(mut s) => produce_screencopy(&mut s, &slot, &stop),
150 Err(e2) => {
151 eprintln!("[cce-remote] screencopy unavailable ({e2}), falling back to grim");
152 produce_grim(&slot, &stop)
153 }
154 }
155 }
156 };
157 if let Err(e) = result {
158 if !stop.load(Ordering::Relaxed) {
159 eprintln!("[cce-remote] frame source ended: {e}");
160 }
161 }
162 slot.close();
163 });
164 }
165
166 fn produce_winstream(
167 r: &mut crate::winstream::Reader,
168 slot: &Slot,
169 stop: &AtomicBool,
170 ) -> Result<(), String> {
171 loop {
172 if stop.load(Ordering::Relaxed) {
173 return Ok(());
174 }
175 // Blocks ≤20s: the compositor keepalives every ≤15s, so a stopped
176 // sender's producer lingers at most one keepalive interval.
177 slot.push(r.next()?);
178 }
179 }
180
181 fn produce_screencopy(
182 sess: &mut crate::screencopy::CaptureSession,
183 slot: &Slot,
184 stop: &AtomicBool,
185 ) -> Result<(), String> {
186 let mut rect: Option<(i32, i32, i32, i32)> = None;
187 let mut rect_at: Option<Instant> = None;
188 let mut force_full = true; // first frame immediately; also after timeouts
189 loop {
190 if stop.load(Ordering::Relaxed) {
191 return Ok(());
192 }
193 if rect_at.is_none_or(|t| t.elapsed() > Duration::from_millis(500)) {
194 if let Some((_, x, y, w, h)) = crate::focused_window() {
195 let r = (x as i32, y as i32, w as i32, h as i32);
196 if Some(r) != rect {
197 force_full = true; // focus moved: don't wait for damage
198 }
199 rect = Some(r);
200 }
201 rect_at = Some(Instant::now());
202 }
203 let Some(r) = rect else {
204 std::thread::sleep(Duration::from_millis(300));
205 continue;
206 };
207 // The 20s damage timeout doubles as the keepalive cadence.
208 let timeout = if force_full { Duration::from_secs(5) } else { Duration::from_secs(20) };
209 match sess.next_frame(r, !force_full, timeout) {
210 Ok(Some(frame)) => {
211 force_full = false;
212 slot.push(frame);
213 // cap runaway damage bursts (~30 fps)
214 std::thread::sleep(Duration::from_millis(33));
215 }
216 Ok(None) => force_full = true,
217 Err(e) => return Err(e),
218 }
219 }
220 }
221
222 fn produce_grim(slot: &Slot, stop: &AtomicBool) -> Result<(), String> {
223 loop {
224 if stop.load(Ordering::Relaxed) {
225 return Ok(());
226 }
227 let Some((_, x, y, w, h)) = crate::focused_window() else {
228 std::thread::sleep(Duration::from_millis(400));
229 continue;
230 };
231 let out = std::process::Command::new("grim")
232 .args([
233 "-g",
234 &format!("{},{} {}x{}", x as i32, y as i32, w as i32, h as i32),
235 "-t", "jpeg", "-q", "65", "-s", "0.5", "-",
236 ])
237 .output();
238 match out {
239 Ok(o) if o.status.success() && o.stdout.starts_with(&[0xff, 0xd8]) => {
240 slot.push(Payload::Jpeg(o.stdout));
241 std::thread::sleep(Duration::from_millis(350));
242 }
243 _ => std::thread::sleep(Duration::from_millis(400)),
244 }
245 }
246 }
247
248 // ---- senders ---------------------------------------------------------------
249
250 /// Ack-clocked delivery over the page's stream WebSocket. The page sends `n`
251 /// after RENDERING each frame — so the measured send→ack time covers network,
252 /// decode and paint, i.e. what the user actually experiences — and only then
253 /// does the newest frame go out.
254 pub fn run_ws_sender(ws: &mut tungstenite::WebSocket<std::net::TcpStream>, slot: &Slot) {
255 let mut last_seq = 0u64;
256 let mut level = START_LEVEL;
257 let mut ewma_ms = 120.0f64;
258 let mut acks_since_change = 0u32;
259 let mut sent_at: Option<Instant> = None;
260 // Acks can legitimately stop for a long time (iOS suspends the page when
261 // backgrounded); pings distinguish suspended-but-alive from gone. The
262 // browser answers pings in its network stack, JS not required.
263 let _ = ws.get_ref().set_read_timeout(Some(Duration::from_secs(75)));
264 let mut silent = 0u32;
265 loop {
266 // 1. wait for the ack of the previous frame
267 loop {
268 match ws.read() {
269 Ok(tungstenite::Message::Text(t)) if t == "n" => break,
270 Ok(tungstenite::Message::Close(_)) => return,
271 Ok(_) => {
272 silent = 0; // pong: peer alive, keep waiting
273 continue;
274 }
275 Err(tungstenite::Error::Io(e))
276 if e.kind() == std::io::ErrorKind::WouldBlock
277 || e.kind() == std::io::ErrorKind::TimedOut =>
278 {
279 silent += 1;
280 if silent >= 2 {
281 return; // two silent windows with no pong: gone
282 }
283 if ws.send(tungstenite::Message::Ping(Vec::new())).is_err() {
284 return;
285 }
286 continue;
287 }
288 Err(_) => return,
289 }
290 }
291 silent = 0;
292 if let Some(t0) = sent_at.take() {
293 let ms = t0.elapsed().as_secs_f64() * 1000.0;
294 ewma_ms = 0.7 * ewma_ms + 0.3 * ms;
295 acks_since_change += 1;
296 let next = adapt(level, ewma_ms, acks_since_change);
297 if next != level {
298 level = next;
299 acks_since_change = 0;
300 }
301 }
302 // 2. newest frame (blocks while the window is idle; sources force
303 // keepalive frames ≤20s, so this wakes regularly)
304 let (seq, payload) = loop {
305 match slot.wait_newer(last_seq, Duration::from_secs(30)) {
306 Ok(Some(x)) => break x,
307 Ok(None) => {
308 if ws.send(tungstenite::Message::Ping(Vec::new())).is_err() {
309 return;
310 }
311 }
312 Err(()) => return,
313 }
314 };
315 last_seq = seq;
316 let (edge, quality) = LADDER[level];
317 let Ok(jpeg) = encode(&payload, edge, quality) else { continue };
318 sent_at = Some(Instant::now());
319 if ws.send(tungstenite::Message::Binary(jpeg)).is_err() {
320 return;
321 }
322 }
323 }
324
325 /// MJPEG over the slot, fixed 560/q60 — kept as the curl-debuggable endpoint.
326 /// Latest-wins still applies (each iteration encodes only the newest frame),
327 /// but without acks the TCP buffer can still queue a few frames; the page no
328 /// longer uses this path.
329 pub fn run_mjpeg_sender(tcp: &mut std::net::TcpStream, slot: &Slot) {
330 let mut last = 0u64;
331 loop {
332 match slot.wait_newer(last, Duration::from_secs(25)) {
333 Ok(Some((seq, p))) => {
334 last = seq;
335 let Ok(jpeg) = encode(&p, 560, 60) else { continue };
336 let head = format!(
337 "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: {}\r\n\r\n",
338 jpeg.len()
339 );
340 if tcp.write_all(head.as_bytes()).is_err()
341 || tcp.write_all(&jpeg).is_err()
342 || tcp.write_all(b"\r\n").is_err()
343 {
344 return; // client gone
345 }
346 }
347 Ok(None) => continue, // idle; dead clients surface on the next write
348 Err(()) => return,
349 }
350 }
351 }
352
353 #[cfg(test)]
354 mod tests {
355 use super::*;
356
357 // The slot IS the fix: if it ever queues instead of overwriting, the
358 // unbounded-lag failure mode comes back silently.
359
360 fn raw(tag: u8) -> Payload {
361 Payload::Raw { data: vec![tag; 4], w: 1, h: 1, stride: 4, rgb: (0, 1, 2) }
362 }
363 fn tag_of(p: &Payload) -> u8 {
364 match p {
365 Payload::Raw { data, .. } => data[0],
366 Payload::Jpeg(b) => b[0],
367 }
368 }
369
370 #[test]
371 fn slot_overwrites_never_queues() {
372 let s = Slot::new();
373 s.push(raw(1));
374 s.push(raw(2));
375 s.push(raw(3));
376 // A consumer that fell behind gets the NEWEST frame, once — frames 1
377 // and 2 are gone, not waiting their turn.
378 let (seq, p) = s.wait_newer(0, Duration::from_millis(10)).unwrap().unwrap();
379 assert_eq!(seq, 3);
380 assert_eq!(tag_of(&p), 3);
381 // Nothing newer: times out rather than re-delivering.
382 assert!(s.wait_newer(seq, Duration::from_millis(10)).unwrap().is_none());
383 }
384
385 #[test]
386 fn slot_close_wakes_and_errs() {
387 let s = Slot::new();
388 s.push(raw(9));
389 let _ = s.wait_newer(0, Duration::from_millis(10)).unwrap().unwrap();
390 s.close();
391 assert!(s.wait_newer(99, Duration::from_secs(5)).is_err(), "close must wake, not time out");
392 }
393
394 #[test]
395 fn slot_wakes_a_blocked_waiter() {
396 let s = Slot::new();
397 let s2 = Arc::clone(&s);
398 let t = std::thread::spawn(move || s2.wait_newer(0, Duration::from_secs(5)));
399 std::thread::sleep(Duration::from_millis(30));
400 s.push(raw(7));
401 let got = t.join().unwrap().unwrap().unwrap();
402 assert_eq!(got.0, 1);
403 assert_eq!(tag_of(&got.1), 7);
404 }
405
406 #[test]
407 fn ladder_prefers_resolution_over_quality() {
408 // The point of the live view is reading the window: stepping down the
409 // ladder must drop quality before it drops size.
410 for pair in LADDER.windows(2) {
411 let ((e1, _), (e2, _)) = (pair[0], pair[1]);
412 assert!(e2 <= e1, "ladder edge must be non-increasing: {pair:?}");
413 }
414 assert!(START_LEVEL < LADDER.len());
415 }
416
417 #[test]
418 fn adapt_downgrades_immediately_upgrades_cautiously() {
419 // Lag is felt now: no stability requirement to step down.
420 assert_eq!(adapt(1, 300.0, 0), 2);
421 // Upgrades need sustained headroom, or the level oscillates at the
422 // threshold.
423 assert_eq!(adapt(2, 50.0, 3), 2);
424 assert_eq!(adapt(2, 50.0, 10), 1);
425 // The dead band holds steady in both directions.
426 assert_eq!(adapt(2, 150.0, 100), 2);
427 }
428
429 #[test]
430 fn adapt_clamps_at_both_ends() {
431 let worst = LADDER.len() - 1;
432 assert_eq!(adapt(worst, 10_000.0, 0), worst);
433 assert_eq!(adapt(0, 1.0, 1000), 0);
434 }
435 }