Adapt WebRTC streaming to latest frame

Author: DjDeveloperr

cli/native/XCWNativeBridge.h

@@ -70,6 +70,7 @@ void * _Nullable xcw_native_session_create(const char * _Nonnull udid, char * _N
 void xcw_native_session_destroy(void * _Nullable handle);
 bool xcw_native_session_start(void * _Nonnull handle, char * _Nullable * _Nullable error_message);
 void xcw_native_session_request_refresh(void * _Nonnull handle);
+void xcw_native_session_request_keyframe(void * _Nonnull handle);
 bool xcw_native_session_send_touch(void * _Nonnull handle, double x, double y, const char * _Nonnull phase, char * _Nullable * _Nullable error_message);
 bool xcw_native_session_send_multitouch(void * _Nonnull handle, double x1, double y1, double x2, double y2, const char * _Nonnull phase, char * _Nullable * _Nullable error_message);
 bool xcw_native_session_send_key(void * _Nonnull handle, uint16_t key_code, uint32_t modifiers, char * _Nullable * _Nullable error_message);

cli/native/XCWNativeBridge.m

@@ -668,6 +668,12 @@ void xcw_native_session_request_refresh(void *handle) {
     }
 }
 
+void xcw_native_session_request_keyframe(void *handle) {
+    @autoreleasepool {
+        [XCWNativeSessionFromHandle(handle) requestKeyFrame];
+    }
+}
+
 bool xcw_native_session_send_touch(void *handle, double x, double y, const char *phase, char **error_message) {
     @autoreleasepool {
         NSError *error = nil;

cli/native/XCWNativeSession.h

@@ -13,6 +13,7 @@ NS_ASSUME_NONNULL_BEGIN
 
 - (BOOL)start:(NSError * _Nullable * _Nullable)error;
 - (void)requestRefresh;
+- (void)requestKeyFrame;
 - (BOOL)sendTouchAtX:(double)x
                    y:(double)y
                phase:(NSString *)phase

cli/native/XCWNativeSession.m

@@ -94,6 +94,10 @@ - (void)requestRefresh {
     [self.session requestFrameRefresh];
 }
 
+- (void)requestKeyFrame {
+    [self.session requestKeyFrameRefresh];
+}
+
 - (BOOL)sendTouchAtX:(double)x
                    y:(double)y
                phase:(NSString *)phase

client/src/features/stream/streamWorkerClient.ts

@@ -565,7 +565,7 @@ function configureLowLatencyReceiver(receiver: RTCRtpReceiver) {
     jitterBufferTarget?: number;
   };
   if ("jitterBufferTarget" in lowLatencyReceiver) {
-    lowLatencyReceiver.jitterBufferTarget = 0.03;
+    lowLatencyReceiver.jitterBufferTarget = 0.001;
   }
 }
 

server/src/native/bridge.rs

@@ -624,6 +624,12 @@ impl NativeSession {
         }
     }
 
+    pub fn request_keyframe(&self) {
+        unsafe {
+            ffi::xcw_native_session_request_keyframe(self.handle);
+        }
+    }
+
     pub unsafe fn set_frame_callback(
         &self,
         callback: Option<ffi::xcw_native_frame_callback>,

server/src/native/ffi.rs

@@ -174,6 +174,7 @@ unsafe extern "C" {
     pub fn xcw_native_session_destroy(handle: *mut c_void);
     pub fn xcw_native_session_start(handle: *mut c_void, error_message: *mut *mut c_char) -> bool;
     pub fn xcw_native_session_request_refresh(handle: *mut c_void);
+    pub fn xcw_native_session_request_keyframe(handle: *mut c_void);
     pub fn xcw_native_session_set_frame_callback(
         handle: *mut c_void,
         callback: Option<xcw_native_frame_callback>,

server/src/simulators/session.rs

@@ -14,8 +14,9 @@ use tokio::task;
 use tokio::time::{timeout, Instant};
 use tracing::debug;
 
-const FRAME_BROADCAST_CAPACITY: usize = 240;
-const MIN_REFRESH_INTERVAL_MS: u64 = 66;
+const FRAME_BROADCAST_CAPACITY: usize = 32;
+const MIN_REFRESH_INTERVAL_MS: u64 = 16;
+const MIN_KEYFRAME_INTERVAL_MS: u64 = 250;
 
 pub struct SimulatorSession {
     inner: Arc<SimulatorSessionInner>,
@@ -34,6 +35,7 @@ struct SimulatorSessionInner {
     display_height: AtomicU64,
     frame_sequence: AtomicU64,
     last_refresh_ms: AtomicU64,
+    last_keyframe_ms: AtomicU64,
 }
 
 impl SimulatorSession {
@@ -56,6 +58,7 @@ impl SimulatorSession {
             display_height: AtomicU64::new(0),
             frame_sequence: AtomicU64::new(0),
             last_refresh_ms: AtomicU64::new(0),
+            last_keyframe_ms: AtomicU64::new(0),
         });
 
         let user_data = Weak::into_raw(Arc::downgrade(&inner)) as *mut c_void;
@@ -149,6 +152,22 @@ impl SimulatorSession {
         self.inner.native.request_refresh();
     }
 
+    pub fn request_keyframe(&self) {
+        let now = now_ms();
+        let previous = self.inner.last_keyframe_ms.load(Ordering::Relaxed);
+        if now.saturating_sub(previous) < MIN_KEYFRAME_INTERVAL_MS {
+            self.request_refresh();
+            return;
+        }
+        self.inner.last_keyframe_ms.store(now, Ordering::Relaxed);
+        self.inner.last_refresh_ms.store(now, Ordering::Relaxed);
+        self.inner
+            .metrics
+            .keyframe_requests
+            .fetch_add(1, Ordering::Relaxed);
+        self.inner.native.request_keyframe();
+    }
+
     pub fn send_touch(&self, x: f64, y: f64, phase: &str) -> Result<(), AppError> {
         self.inner.native.send_touch(x, y, phase)
     }

server/src/transport/webrtc.rs

@@ -26,8 +26,10 @@ const ANNEX_B_START_CODE: &[u8] = &[0, 0, 0, 1];
 const DEFAULT_STUN_URL: &str = "stun:stun.l.google.com:19302";
 const WEBRTC_CONTROL_CHANNEL_LABEL: &str = "simdeck-control";
 const WEBRTC_BOOTSTRAP_KEYFRAME_INTERVAL: Duration = Duration::from_millis(250);
-const WEBRTC_BOOTSTRAP_KEYFRAME_REPEATS: u8 = 12;
-const WEBRTC_REFRESH_INTERVAL: Duration = Duration::from_millis(66);
+const WEBRTC_BOOTSTRAP_KEYFRAME_REPEATS: u8 = 4;
+const WEBRTC_MIN_REFRESH_INTERVAL: Duration = Duration::from_millis(16);
+const WEBRTC_MAX_REFRESH_INTERVAL: Duration = Duration::from_millis(100);
+const WEBRTC_WRITE_TIMEOUT: Duration = Duration::from_millis(120);
 
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -61,7 +63,7 @@ pub async fn create_answer(
         state.registry.remove(&udid);
         return Err(error);
     }
-    session.request_refresh();
+    session.request_keyframe();
     info!(
         "WebRTC offer for {udid}: remote_candidates={} remote_candidate_types={} ice_servers={}",
         count_sdp_candidates(&payload.sdp),
@@ -350,15 +352,19 @@ async fn stream_h264_frames(
     let mut last_sequence = 0u64;
     let mut send_timing = WebRtcSendTiming::new();
     let mut bootstrap_interval = time::interval(WEBRTC_BOOTSTRAP_KEYFRAME_INTERVAL);
-    let mut refresh_interval = time::interval(WEBRTC_REFRESH_INTERVAL);
-    refresh_interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
+    let mut refresh_sleep = Box::pin(time::sleep(WEBRTC_MIN_REFRESH_INTERVAL));
     let mut bootstrap_frames_remaining = WEBRTC_BOOTSTRAP_KEYFRAME_REPEATS;
+    let mut adaptive_refresh_interval = WEBRTC_MIN_REFRESH_INTERVAL;
+    let mut waiting_for_keyframe = false;
     let _guard = WebRtcMetricsGuard::new(state.metrics.clone());
 
     loop {
         tokio::select! {
-            _ = refresh_interval.tick() => {
+            _ = &mut refresh_sleep => {
                 session.request_refresh();
+                refresh_sleep
+                    .as_mut()
+                    .reset(time::Instant::now() + adaptive_refresh_interval);
             }
             _ = bootstrap_interval.tick(), if bootstrap_frames_remaining > 0 => {
                 if let Err(error) = write_frame_sample(
@@ -380,7 +386,8 @@ async fn stream_h264_frames(
                             .metrics
                             .frames_dropped_server
                             .fetch_add(skipped, Ordering::Relaxed);
-                        session.request_refresh();
+                        waiting_for_keyframe = true;
+                        session.request_keyframe();
                         continue;
                     }
                     Err(broadcast::error::RecvError::Closed) => break,
@@ -392,7 +399,8 @@ async fn stream_h264_frames(
                         .frames_dropped_server
                         .fetch_add(skipped, Ordering::Relaxed);
                     if !frame.is_keyframe {
-                        session.request_refresh();
+                        waiting_for_keyframe = true;
+                        session.request_keyframe();
                         continue;
                     }
                 }
@@ -401,14 +409,32 @@ async fn stream_h264_frames(
                         .metrics
                         .frames_dropped_server
                         .fetch_add(frame.frame_sequence - last_sequence - 1, Ordering::Relaxed);
-                    session.request_refresh();
+                    waiting_for_keyframe = true;
+                    session.request_keyframe();
+                    continue;
+                }
+                if waiting_for_keyframe && !frame.is_keyframe {
+                    state.metrics.frames_dropped_server.fetch_add(1, Ordering::Relaxed);
                     continue;
                 }
                 if frame.is_keyframe {
                     latest_keyframe = frame.clone();
+                    waiting_for_keyframe = false;
                 }
                 let duration = send_timing.duration_for(&frame);
-                if let Err(error) = write_frame_sample(&video_track, &frame, duration).await {
+                let started_at = time::Instant::now();
+                let write_result = time::timeout(
+                    WEBRTC_WRITE_TIMEOUT,
+                    write_frame_sample(&video_track, &frame, duration),
+                ).await;
+                adaptive_refresh_interval = adaptive_interval_for_write(started_at.elapsed());
+                if let Err(error) = write_result
+                    .map_err(|_| anyhow::anyhow!(
+                        "timed out writing WebRTC frame after {}ms",
+                        WEBRTC_WRITE_TIMEOUT.as_millis()
+                    ))
+                    .and_then(|result| result)
+                {
                     warn!("WebRTC frame write failed for {udid}: {error}");
                     break;
                 }
@@ -421,6 +447,14 @@ async fn stream_h264_frames(
     let _ = peer_connection.close().await;
 }
 
+fn adaptive_interval_for_write(write_elapsed: Duration) -> Duration {
+    let target_ms = (write_elapsed.as_millis() as u64).saturating_mul(2).clamp(
+        WEBRTC_MIN_REFRESH_INTERVAL.as_millis() as u64,
+        WEBRTC_MAX_REFRESH_INTERVAL.as_millis() as u64,
+    );
+    Duration::from_millis(target_ms)
+}
+
 async fn write_frame_sample(
     video_track: &TrackLocalStaticSample,
     frame: &crate::transport::packet::SharedFrame,