android/session/
controller.rs

1//! The emulator's controller in a session: its touches, and its frames on
2//! the event stream with hashes that leave the status bar out (design D7).
3
4use std::collections::VecDeque;
5use std::future::Future;
6use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
7use std::sync::mpsc::Sender;
8use std::sync::{Arc, Mutex, PoisonError};
9use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
10
11use jpeg_encoder::{ColorType, Encoder};
12use mirror::backend::{BackendError, EncodeRequest, FrameMeta, FrameSource, StreamItem, Timed};
13use protocol::code;
14use protocol::message::Rotation;
15
16use crate::emulator::{
17    self, BYTES_PER_PIXEL, Controller, Endpoint, Frame, FrameHasher, Options, Unavailable,
18};
19
20/// How long after input the agent did not inject a read waits, as the
21/// agent's idle wait counts a quiet window from its own input (design D8's
22/// 200 ms).
23pub const INPUT_QUIET: Duration = Duration::from_millis(200);
24
25/// How many recent frames are kept for encoding.
26pub(crate) const KEPT_FRAMES: usize = 4;
27
28/// The longest lag between the emulator's timestamp and a frame's arrival
29/// that is believed; anything longer is a clock difference.
30const MAX_LAG: Duration = Duration::from_secs(1);
31
32/// How long the frame pump waits for a frame before checking whether it
33/// should stop.
34const FRAME_WAIT: Duration = Duration::from_millis(250);
35
36/// How long the frame pump waits before subscribing again after the stream
37/// failed.
38const RESUBSCRIBE_AFTER: Duration = Duration::from_millis(500);
39
40/// A connected controller and the runtime its calls run on.
41pub struct ControllerLink {
42    runtime: tokio::runtime::Runtime,
43    controller: Controller,
44    endpoint: String,
45}
46
47impl std::fmt::Debug for ControllerLink {
48    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
49        f.debug_struct("ControllerLink")
50            .field("controller", &self.controller)
51            .finish_non_exhaustive()
52    }
53}
54
55impl ControllerLink {
56    /// Connects to the controller at `endpoint` on a runtime of its own.
57    pub fn connect(endpoint: &Endpoint, options: Options) -> Result<ControllerLink, Unavailable> {
58        let runtime = tokio::runtime::Builder::new_multi_thread()
59            .worker_threads(2)
60            .thread_name("android-controller")
61            .enable_all()
62            .build()
63            .map_err(|e| {
64                Unavailable::new(
65                    &endpoint.serial,
66                    endpoint.console_port,
67                    format!("cannot start the controller's runtime: {e}"),
68                )
69            })?;
70        let controller = runtime.block_on(Controller::connect(endpoint, options))?;
71        Ok(ControllerLink {
72            runtime,
73            controller,
74            endpoint: endpoint.uri.clone(),
75        })
76    }
77
78    pub fn controller(&self) -> &Controller {
79        &self.controller
80    }
81
82    /// The controller's address.
83    pub fn endpoint(&self) -> &str {
84        &self.endpoint
85    }
86
87    /// Whether the AVD declares a keyboard, so keys could go this way.
88    pub fn keyboard(&self) -> bool {
89        self.controller.keys_supported()
90    }
91
92    /// Runs `future` on the link's runtime from a thread outside it.
93    pub(crate) fn block_on<F: Future>(&self, future: F) -> F::Output {
94        self.runtime.block_on(future)
95    }
96
97    /// Streams frames of `width` into the pump's store and onto its channel
98    /// until it is stopped or nothing reads the channel any more.
99    pub(crate) fn start_frames(&self, width: u32, pump: FramePump) {
100        let controller = self.controller.clone();
101        self.runtime.spawn(pump.run(controller, width));
102    }
103}
104
105/// The metadata of `frame`, the `seq`-th of the stream: its hash leaves the
106/// status bar out; its capture time is its arrival minus the lag the
107/// emulator's timestamp shows, at most a second.
108pub fn frame_meta(
109    frame: &Frame,
110    seq: u64,
111    hasher: &FrameHasher,
112    arrived: Instant,
113    now_epoch_us: u64,
114) -> FrameMeta {
115    let lag = match frame.timestamp_us() {
116        0 => Duration::ZERO,
117        produced => Duration::from_micros(now_epoch_us.saturating_sub(produced)).min(MAX_LAG),
118    };
119    FrameMeta {
120        seq,
121        at: arrived.checked_sub(lag).unwrap_or(arrived),
122        hash: hasher.key(frame).hash,
123        width: frame.width(),
124        height: frame.height(),
125        rotation: Rotation::R0,
126    }
127}
128
129/// Microseconds since the Unix epoch, the emulator's timestamp clock.
130fn epoch_us() -> u64 {
131    SystemTime::now()
132        .duration_since(UNIX_EPOCH)
133        .map_or(0, |d| u64::try_from(d.as_micros()).unwrap_or(u64::MAX))
134}
135
136/// RGB888 `pixels` of `width` by `height` turned clockwise by `rotation`;
137/// returns the pixels and their new size.
138pub fn rotate_rgb(
139    pixels: &[u8],
140    width: u32,
141    height: u32,
142    rotation: Rotation,
143) -> (Vec<u8>, u32, u32) {
144    if !matches!(rotation, Rotation::R90 | Rotation::R180 | Rotation::R270) {
145        return (pixels.to_vec(), width, height);
146    }
147    let (w, h) = (width as usize, height as usize);
148    let (out_w, out_h) = if rotation == Rotation::R180 {
149        (w, h)
150    } else {
151        (h, w)
152    };
153    let mut out = vec![0u8; out_w * out_h * BYTES_PER_PIXEL];
154    for y in 0..out_h {
155        for x in 0..out_w {
156            let (sx, sy) = match rotation {
157                Rotation::R90 => (y, h - 1 - x),
158                Rotation::R180 => (w - 1 - x, h - 1 - y),
159                _ => (w - 1 - y, x),
160            };
161            let from = (sy * w + sx) * BYTES_PER_PIXEL;
162            let to = (y * out_w + x) * BYTES_PER_PIXEL;
163            out[to..to + BYTES_PER_PIXEL].copy_from_slice(&pixels[from..from + BYTES_PER_PIXEL]);
164        }
165    }
166    (out, out_w as u32, out_h as u32)
167}
168
169/// The input pixels that output pixel `o` of `to` covers along an axis of
170/// `from`: at least one.
171fn span(o: u32, from: u32, to: u32) -> (usize, usize) {
172    let (o, from, to) = (u64::from(o), u64::from(from), u64::from(to));
173    let start = o * from / to;
174    let end = ((o + 1) * from).div_ceil(to).max(start + 1).min(from);
175    (start as usize, end as usize)
176}
177
178/// RGB888 `pixels` of `width` by `height` scaled to `to_width` by
179/// `to_height`: each output pixel averages the input area it covers, or
180/// repeats the nearest one when enlarging.
181pub fn resize_rgb(
182    pixels: &[u8],
183    width: u32,
184    height: u32,
185    to_width: u32,
186    to_height: u32,
187) -> Vec<u8> {
188    let (w, to_w) = (width as usize, to_width as usize);
189    let mut out = vec![0u8; to_w * to_height as usize * BYTES_PER_PIXEL];
190    let columns: Vec<(usize, usize)> = (0..to_width).map(|x| span(x, width, to_width)).collect();
191    for oy in 0..to_height {
192        let (y0, y1) = span(oy, height, to_height);
193        for (ox, &(x0, x1)) in columns.iter().enumerate() {
194            let mut sum = [0u64; BYTES_PER_PIXEL];
195            for y in y0..y1 {
196                let row = &pixels[(y * w + x0) * BYTES_PER_PIXEL..(y * w + x1) * BYTES_PER_PIXEL];
197                for px in row.as_chunks::<BYTES_PER_PIXEL>().0 {
198                    for (total, &channel) in sum.iter_mut().zip(px) {
199                        *total += u64::from(channel);
200                    }
201                }
202            }
203            let count = ((y1 - y0) * (x1 - x0)) as u64;
204            let at = (oy as usize * to_w + ox) * BYTES_PER_PIXEL;
205            for (c, total) in sum.iter().enumerate() {
206                out[at + c] = ((total + count / 2) / count) as u8;
207            }
208        }
209    }
210    out
211}
212
213/// `frame` as a JPEG: turned by `request.rotation`, then scaled to exactly
214/// `request.width` by `request.height`, at `request.quality`.
215pub fn encode_jpeg(frame: &Frame, request: &EncodeRequest) -> Result<Vec<u8>, BackendError> {
216    let edge = |pixels: u32| u16::try_from(pixels).ok().filter(|&e| e > 0);
217    let (Some(width), Some(height)) = (edge(request.width), edge(request.height)) else {
218        return Err(BackendError::new(
219            code::INVALID_PARAMS,
220            format!(
221                "a JPEG of {}x{} pixels cannot be encoded",
222                request.width, request.height
223            ),
224        ));
225    };
226    let (turned, w, h) = rotate_rgb(
227        frame.pixels(),
228        frame.width(),
229        frame.height(),
230        request.rotation,
231    );
232    let pixels = if (w, h) == (request.width, request.height) {
233        turned
234    } else {
235        resize_rgb(&turned, w, h, request.width, request.height)
236    };
237    let mut out = Vec::new();
238    Encoder::new(&mut out, request.quality.clamp(1, 100))
239        .encode(&pixels, width, height, ColorType::Rgb)
240        .map_err(|e| BackendError::new(code::INTERNAL, format!("JPEG encoding failed: {e}")))?;
241    Ok(out)
242}
243
244/// The most recent frames with their metadata, newest last.
245#[derive(Default)]
246pub(crate) struct FrameStore {
247    frames: VecDeque<(FrameMeta, Frame)>,
248}
249
250impl FrameStore {
251    pub(crate) fn push(&mut self, meta: FrameMeta, frame: Frame) {
252        self.frames.push_back((meta, frame));
253        while self.frames.len() > KEPT_FRAMES {
254            self.frames.pop_front();
255        }
256    }
257
258    pub(crate) fn latest(&self) -> Option<FrameMeta> {
259        self.frames.back().map(|(meta, _)| *meta)
260    }
261
262    pub(crate) fn get(&self, seq: u64) -> Option<&Frame> {
263        self.frames
264            .iter()
265            .find(|(meta, _)| meta.seq == seq)
266            .map(|(_, frame)| frame)
267    }
268}
269
270/// What the frame pump shares with the session.
271#[derive(Clone)]
272pub(crate) struct FramePump {
273    pub(crate) store: Arc<Mutex<FrameStore>>,
274    pub(crate) items: Sender<Timed<StreamItem>>,
275    pub(crate) received: Arc<AtomicU64>,
276    pub(crate) stop: Arc<AtomicBool>,
277    /// The last failure of the frame stream, for the session's status.
278    pub(crate) error: Arc<Mutex<Option<String>>>,
279}
280
281impl FramePump {
282    async fn run(self, controller: Controller, width: u32) {
283        let hasher = FrameHasher::default();
284        let mut seq = 0u64;
285        while !self.stop.load(Ordering::SeqCst) {
286            let mut stream = match controller.frames(width).await {
287                Ok(stream) => stream,
288                Err(error) => {
289                    self.fail(&error);
290                    tokio::time::sleep(RESUBSCRIBE_AFTER).await;
291                    continue;
292                }
293            };
294            while !self.stop.load(Ordering::SeqCst) {
295                match stream.next(FRAME_WAIT).await {
296                    Ok(Some(frame)) => {
297                        seq += 1;
298                        let arrived = Instant::now();
299                        let meta = frame_meta(&frame, seq, &hasher, arrived, epoch_us());
300                        self.store
301                            .lock()
302                            .unwrap_or_else(PoisonError::into_inner)
303                            .push(meta, frame);
304                        self.received.fetch_add(1, Ordering::SeqCst);
305                        let item = Timed {
306                            at: arrived,
307                            item: StreamItem::Frame(meta),
308                        };
309                        if self.items.send(item).is_err() {
310                            return;
311                        }
312                    }
313                    Ok(None) => {}
314                    // A frame whose bytes do not fit its size; the stream goes on.
315                    Err(emulator::Error::Frame { .. }) => {}
316                    Err(error) => {
317                        self.fail(&error);
318                        break;
319                    }
320                }
321            }
322            tokio::time::sleep(RESUBSCRIBE_AFTER).await;
323        }
324    }
325
326    fn fail(&self, error: &emulator::Error) {
327        *self.error.lock().unwrap_or_else(PoisonError::into_inner) = Some(error.to_string());
328    }
329}
330
331/// The session's frame source: the latest controller frames.
332pub(crate) struct ControllerFrames {
333    pub(crate) store: Arc<Mutex<FrameStore>>,
334}
335
336impl FrameSource for ControllerFrames {
337    fn latest(&self) -> Option<FrameMeta> {
338        self.store
339            .lock()
340            .unwrap_or_else(PoisonError::into_inner)
341            .latest()
342    }
343
344    fn encode(&mut self, request: &EncodeRequest) -> Result<Vec<u8>, BackendError> {
345        let frame = self
346            .store
347            .lock()
348            .unwrap_or_else(PoisonError::into_inner)
349            .get(request.seq)
350            .cloned()
351            .ok_or_else(|| BackendError {
352                code: "no_frame".into(),
353                message: format!(
354                    "frame {} is no longer held; ask for the latest frame again",
355                    request.seq
356                ),
357                retryable: true,
358            })?;
359        encode_jpeg(&frame, request)
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use super::*;
366    use prost::bytes::Bytes;
367
368    fn frame(width: u32, height: u32, fill: impl Fn(u32, u32) -> [u8; 3]) -> Frame {
369        let mut pixels = Vec::with_capacity((width * height * 3) as usize);
370        for y in 0..height {
371            for x in 0..width {
372                pixels.extend_from_slice(&fill(x, y));
373            }
374        }
375        Frame::new(width, height, Bytes::from(pixels)).unwrap()
376    }
377
378    fn pixel(pixels: &[u8], width: u32, x: u32, y: u32) -> [u8; 3] {
379        let at = ((y * width + x) * 3) as usize;
380        [pixels[at], pixels[at + 1], pixels[at + 2]]
381    }
382
383    #[test]
384    fn frame_hashes_leave_the_status_bar_out() {
385        let hasher = FrameHasher::default();
386        let now = Instant::now();
387        let base = frame(10, 100, |_, _| [9, 9, 9]);
388        let clock = frame(10, 100, |x, y| {
389            if y < 3 && x == 1 {
390                [200, 0, 0]
391            } else {
392                [9, 9, 9]
393            }
394        });
395        let content = frame(10, 100, |x, y| {
396            if y == 50 && x == 1 {
397                [200, 0, 0]
398            } else {
399                [9, 9, 9]
400            }
401        });
402        let a = frame_meta(&base, 1, &hasher, now, 0);
403        let b = frame_meta(&clock, 2, &hasher, now, 0);
404        let c = frame_meta(&content, 3, &hasher, now, 0);
405        assert_eq!(a.hash, b.hash, "a status bar change is no change");
406        assert_ne!(a.hash, c.hash);
407        assert_eq!((c.seq, c.width, c.height), (3, 10, 100));
408        assert_eq!(c.rotation, Rotation::R0);
409    }
410
411    #[test]
412    fn the_capture_time_is_the_arrival_minus_the_emulators_lag() {
413        let hasher = FrameHasher::default();
414        let arrived = Instant::now();
415        let produced = frame(2, 2, |_, _| [0, 0, 0]).with_timing(7, 1_000_000);
416        let meta = frame_meta(&produced, 1, &hasher, arrived, 1_015_000);
417        assert_eq!(arrived - meta.at, Duration::from_millis(15));
418        // A clock that runs ahead or a missing timestamp gives the arrival;
419        // a lag over a second is capped.
420        assert_eq!(
421            frame_meta(&produced, 1, &hasher, arrived, 900_000).at,
422            arrived
423        );
424        let untimed = frame(2, 2, |_, _| [0, 0, 0]);
425        assert_eq!(
426            frame_meta(&untimed, 1, &hasher, arrived, 5_000_000).at,
427            arrived
428        );
429        let late = frame_meta(&produced, 1, &hasher, arrived, 9_000_000);
430        assert_eq!(arrived - late.at, Duration::from_secs(1));
431    }
432
433    /// A 3 by 2 image: A B C over D E F.
434    fn letters() -> Vec<u8> {
435        b"ABCDEF".iter().flat_map(|&c| [c, c, c]).collect()
436    }
437
438    fn rows(pixels: &[u8], width: u32, height: u32) -> Vec<String> {
439        (0..height)
440            .map(|y| {
441                (0..width)
442                    .map(|x| pixel(pixels, width, x, y)[0] as char)
443                    .collect()
444            })
445            .collect()
446    }
447
448    #[test]
449    fn rotation_turns_the_pixels_clockwise() {
450        let (r0, w, h) = rotate_rgb(&letters(), 3, 2, Rotation::R0);
451        assert_eq!(rows(&r0, w, h), ["ABC", "DEF"]);
452        let (r90, w, h) = rotate_rgb(&letters(), 3, 2, Rotation::R90);
453        assert_eq!((w, h), (2, 3));
454        assert_eq!(rows(&r90, w, h), ["DA", "EB", "FC"]);
455        let (r180, w, h) = rotate_rgb(&letters(), 3, 2, Rotation::R180);
456        assert_eq!(rows(&r180, w, h), ["FED", "CBA"]);
457        let (r270, w, h) = rotate_rgb(&letters(), 3, 2, Rotation::R270);
458        assert_eq!(rows(&r270, w, h), ["CF", "BE", "AD"]);
459    }
460
461    #[test]
462    fn shrinking_averages_and_enlarging_repeats() {
463        // Left half red, right half blue, 4 by 2.
464        let halves = frame(4, 2, |x, _| if x < 2 { [200, 0, 0] } else { [0, 0, 100] });
465        let small = resize_rgb(halves.pixels(), 4, 2, 2, 1);
466        assert_eq!(small, [200, 0, 0, 0, 0, 100]);
467        // A 2 by 1 checker averages to its mean.
468        let checker = [0, 0, 0, 255, 255, 255];
469        assert_eq!(resize_rgb(&checker, 2, 1, 1, 1), [128, 128, 128]);
470        let one = [10, 20, 30];
471        assert_eq!(resize_rgb(&one, 1, 1, 2, 2), [10, 20, 30].repeat(4));
472    }
473
474    /// The width and height in a JPEG's start-of-frame segment.
475    fn jpeg_size(data: &[u8]) -> (u32, u32) {
476        assert_eq!(&data[..2], [0xFF, 0xD8], "no start of image");
477        assert_eq!(&data[data.len() - 2..], [0xFF, 0xD9], "no end of image");
478        let mut at = 2;
479        while at + 9 < data.len() {
480            assert_eq!(data[at], 0xFF);
481            let marker = data[at + 1];
482            let len = u16::from_be_bytes([data[at + 2], data[at + 3]]) as usize;
483            if (0xC0..=0xC2).contains(&marker) {
484                let h = u16::from_be_bytes([data[at + 5], data[at + 6]]);
485                let w = u16::from_be_bytes([data[at + 7], data[at + 8]]);
486                return (u32::from(w), u32::from(h));
487            }
488            at += 2 + len;
489        }
490        panic!("no start of frame");
491    }
492
493    #[test]
494    fn frames_encode_as_jpeg_turned_and_scaled_to_the_request() {
495        let wide = frame(160, 80, |x, y| [(x % 256) as u8, (y % 256) as u8, 7]);
496        let request = EncodeRequest {
497            seq: 1,
498            width: 40,
499            height: 80,
500            quality: 80,
501            rotation: Rotation::R90,
502        };
503        let data = encode_jpeg(&wide, &request).unwrap();
504        assert_eq!(jpeg_size(&data), (40, 80));
505    }
506
507    #[test]
508    fn the_store_keeps_the_latest_frames_by_sequence() {
509        let hasher = FrameHasher::default();
510        let now = Instant::now();
511        let mut store = FrameStore::default();
512        assert_eq!(store.latest(), None);
513        for seq in 1..=6u64 {
514            let f = frame(2, 2, |_, _| [seq as u8, 0, 0]);
515            store.push(frame_meta(&f, seq, &hasher, now, 0), f);
516        }
517        assert_eq!(store.latest().map(|m| m.seq), Some(6));
518        assert!(store.get(6).is_some());
519        assert!(store.get(3).is_some());
520        assert!(store.get(2).is_none(), "only {KEPT_FRAMES} frames are kept");
521        assert_eq!(store.get(5).unwrap().pixels()[0], 5);
522    }
523
524    /// Times `resize_rgb` from a textured frame the size of the emulator's
525    /// 1080x2400 display to the 1000-pixel long edge the mirror asks for by
526    /// default. The checksum shows whether two builds computed the same
527    /// pixels.
528    #[test]
529    #[ignore = "a measurement; run with --release --ignored --nocapture on a quiet machine"]
530    fn measure_the_rgb_resize() {
531        use std::hint::black_box;
532        const WARMUP: usize = 10;
533        const RUNS: usize = 200;
534        let (width, height, to_width, to_height) = (1080, 2400, 450, 1000);
535        let source = frame(width, height, |x, y| {
536            let v = (x * 31 + y * 17 + (x * y) % 97) as u8;
537            [v, v.wrapping_mul(3), v ^ 0x5A]
538        });
539        let mut samples = Vec::with_capacity(RUNS);
540        let mut small = Vec::new();
541        for run in 0..WARMUP + RUNS {
542            let started = Instant::now();
543            small = resize_rgb(
544                black_box(source.pixels()),
545                width,
546                height,
547                to_width,
548                to_height,
549            );
550            let elapsed = started.elapsed();
551            if run >= WARMUP {
552                samples.push(elapsed.as_secs_f64() * 1e3);
553            }
554        }
555        samples.sort_by(f64::total_cmp);
556        let percentile = |percent: usize| samples[(samples.len() * percent).div_ceil(100) - 1];
557        let checksum = small.iter().fold(0xcbf2_9ce4_8422_2325_u64, |hash, &byte| {
558            (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
559        });
560        eprintln!(
561            "resize_rgb {width}x{height} to {to_width}x{to_height}, {RUNS} runs: p50 {:.3} ms, p95 {:.3} ms, checksum {checksum:016x}",
562            percentile(50),
563            percentile(95)
564        );
565    }
566}
android/session/controller.rs

android/session/
controller.rs
