Merge pull request #3953 from 21pages/fps

client side fps control for reduce delay

flutter/lib/desktop/pages/desktop_setting_page.dart

@@ -1228,9 +1228,9 @@ class _DisplayState extends State<_Display> {
                   children: [
                     Slider(
                       value: fpsValue.value,
-                      min: 10.0,
+                      min: 5.0,
                       max: 120.0,
-                      divisions: 22,
+                      divisions: 23,
                       onChanged: (double value) async {
                         fpsValue.value = value;
                         await bind.mainSetUserDefaultOption(

flutter/lib/desktop/widgets/remote_toolbar.dart

@@ -1237,7 +1237,7 @@ class _DisplayMenuState extends State<_DisplayMenu> {
     final fpsOption =
         await bind.sessionGetOption(id: widget.id, arg: 'custom-fps');
     fpsInitValue = fpsOption == null ? 30 : double.tryParse(fpsOption) ?? 30;
-    if (fpsInitValue < 10 || fpsInitValue > 120) {
+    if (fpsInitValue < 5 || fpsInitValue > 120) {
       fpsInitValue = 30;
     }
     final RxDouble fpsSliderValue = RxDouble(fpsInitValue);
@@ -1260,9 +1260,9 @@ class _DisplayMenuState extends State<_DisplayMenu> {
         children: [
           Obx((() => Slider(
                 value: fpsSliderValue.value,
-                min: 10,
+                min: 5,
                 max: 120,
-                divisions: 22,
+                divisions: 23,
                 onChanged: (double value) {
                   fpsSliderValue.value = value;
                   debouncerFps.value = value;

libs/hbb_common/src/config.rs

@@ -1363,7 +1363,7 @@ impl UserDefaultConfig {
             "image_quality" => self.get_string(key, "balanced", vec!["best", "low", "custom"]),
             "codec-preference" => self.get_string(key, "auto", vec!["vp8", "vp9", "h264", "h265"]),
             "custom_image_quality" => self.get_double_string(key, 50.0, 10.0, 100.0),
-            "custom-fps" => self.get_double_string(key, 30.0, 10.0, 120.0),
+            "custom-fps" => self.get_double_string(key, 30.0, 5.0, 120.0),
             _ => self
                 .options
                 .get(key)

src/client.rs

@@ -3,7 +3,10 @@ use std::{
     net::SocketAddr,
     ops::Deref,
     str::FromStr,
-    sync::{mpsc, Arc, Mutex, RwLock},
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        mpsc, Arc, Mutex, RwLock,
+    },
 };
 
 pub use async_trait::async_trait;
@@ -1291,11 +1294,12 @@ impl LoginConfigHandler {
                 config.custom_image_quality[0]
             };
             msg.custom_image_quality = quality << 8;
+            #[cfg(feature = "flutter")]
+            if let Some(custom_fps) = self.options.get("custom-fps") {
+                msg.custom_fps = custom_fps.parse().unwrap_or(30);
+            }
             n += 1;
         }
-        if let Some(custom_fps) = self.options.get("custom-fps") {
-            msg.custom_fps = custom_fps.parse().unwrap_or(30);
-        }
         let view_only = self.get_toggle_option("view-only");
         if view_only {
             msg.disable_keyboard = BoolOption::Yes.into();
@@ -1677,7 +1681,12 @@ pub type MediaSender = mpsc::Sender<MediaData>;
 /// * `video_callback` - The callback for video frame. Being called when a video frame is ready.
 pub fn start_video_audio_threads<F>(
     video_callback: F,
-) -> (MediaSender, MediaSender, Arc<ArrayQueue<VideoFrame>>)
+) -> (
+    MediaSender,
+    MediaSender,
+    Arc<ArrayQueue<VideoFrame>>,
+    Arc<AtomicUsize>,
+)
 where
     F: 'static + FnMut(&mut Vec<u8>) + Send,
 {
@@ -1685,21 +1694,48 @@ where
     let video_queue = Arc::new(ArrayQueue::<VideoFrame>::new(VIDEO_QUEUE_SIZE));
     let video_queue_cloned = video_queue.clone();
     let mut video_callback = video_callback;
+    let mut duration = std::time::Duration::ZERO;
+    let mut count = 0;
+    let fps = Arc::new(AtomicUsize::new(0));
+    let decode_fps = fps.clone();
+    let mut skip_beginning = 0;
 
     std::thread::spawn(move || {
         let mut video_handler = VideoHandler::new();
         loop {
             if let Ok(data) = video_receiver.recv() {
                 match data {
-                    MediaData::VideoFrame(vf) => {
+                    MediaData::VideoFrame(_) | MediaData::VideoQueue => {
-                        if let Ok(true) = video_handler.handle_frame(*vf) {
+                        let vf = if let MediaData::VideoFrame(vf) = data {
+                            *vf
+                        } else {
+                            if let Some(vf) = video_queue.pop() {
+                                vf
+                            } else {
+                                continue;
+                            }
+                        };
+                        let start = std::time::Instant::now();
+                        if let Ok(true) = video_handler.handle_frame(vf) {
                             video_callback(&mut video_handler.rgb);
-                        }
+                            // fps calculation
-                    }
+                            // The first frame will be very slow
-                    MediaData::VideoQueue => {
+                            if skip_beginning < 5 {
-                        if let Some(vf) = video_queue.pop() {
+                                skip_beginning += 1;
-                            if let Ok(true) = video_handler.handle_frame(vf) {
+                                continue;
-                                video_callback(&mut video_handler.rgb);
+                            }
+                            duration += start.elapsed();
+                            count += 1;
+                            if count % 10 == 0 {
+                                fps.store(
+                                    (count * 1000 / duration.as_millis()) as usize,
+                                    Ordering::Relaxed,
+                                );
+                            }
+                            // Clear to get real-time fps
+                            if count > 300 {
+                                count = 0;
+                                duration = Duration::ZERO;
                             }
                         }
                     }
@@ -1718,7 +1754,7 @@ where
         log::info!("Video decoder loop exits");
     });
     let audio_sender = start_audio_thread();
-    return (video_sender, audio_sender, video_queue_cloned);
+    return (video_sender, audio_sender, video_queue_cloned, decode_fps);
 }
 
 /// Start an audio thread

src/client/io_loop.rs

@@ -65,6 +65,8 @@ pub struct Remote<T: InvokeUiSession> {
     frame_count: Arc<AtomicUsize>,
     video_format: CodecFormat,
     elevation_requested: bool,
+    fps_control: FpsControl,
+    decode_fps: Arc<AtomicUsize>,
 }
 
 impl<T: InvokeUiSession> Remote<T> {
@@ -76,6 +78,7 @@ impl<T: InvokeUiSession> Remote<T> {
         receiver: mpsc::UnboundedReceiver<Data>,
         sender: mpsc::UnboundedSender<Data>,
         frame_count: Arc<AtomicUsize>,
+        decode_fps: Arc<AtomicUsize>,
     ) -> Self {
         Self {
             handler,
@@ -100,6 +103,8 @@ impl<T: InvokeUiSession> Remote<T> {
             stop_voice_call_sender: None,
             voice_call_request_timestamp: None,
             elevation_requested: false,
+            fps_control: Default::default(),
+            decode_fps,
         }
     }
 
@@ -147,6 +152,7 @@ impl<T: InvokeUiSession> Remote<T> {
                 let mut rx_clip_client = rx_clip_client_lock.lock().await;
 
                 let mut status_timer = time::interval(Duration::new(1, 0));
+                let mut fps_instant = Instant::now();
 
                 loop {
                     tokio::select! {
@@ -224,9 +230,18 @@ impl<T: InvokeUiSession> Remote<T> {
                             }
                         }
                         _ = status_timer.tick() => {
-                            let speed = self.data_count.swap(0, Ordering::Relaxed);
+                            self.fps_control();
+                            let elapsed = fps_instant.elapsed().as_millis();
+                            if elapsed < 1000 {
+                                continue;
+                            }
+                            fps_instant = Instant::now();
+                            let mut speed = self.data_count.swap(0, Ordering::Relaxed);
+                            speed = speed * 1000 / elapsed as usize;
                             let speed = format!("{:.2}kB/s", speed as f32 / 1024 as f32);
-                            let fps = self.frame_count.swap(0, Ordering::Relaxed) as _;
+                            let mut fps = self.frame_count.swap(0, Ordering::Relaxed) as _;
+                            // Correcting the inaccuracy of status_timer
+                            fps = fps * 1000 / elapsed as i32;
                             self.handler.update_quality_status(QualityStatus {
                                 speed:Some(speed),
                                 fps:Some(fps),
@@ -826,6 +841,53 @@ impl<T: InvokeUiSession> Remote<T> {
             None => false,
         }
     }
+    #[inline]
+    fn fps_control(&mut self) {
+        let len = self.video_queue.len();
+        let ctl = &mut self.fps_control;
+        // Current full speed decoding fps
+        let decode_fps = self.decode_fps.load(std::sync::atomic::Ordering::Relaxed);
+        // 500ms
+        let debounce = if decode_fps > 10 { decode_fps / 2 } else { 5 };
+        if len < debounce || decode_fps == 0 {
+            return;
+        }
+        let mut refresh = false;
+        // First setting , or the length of the queue still increases after setting, or exceed the size of the last setting again
+        if ctl.set_times < 10 // enough
+            && (ctl.set_times == 0
+                || (len > ctl.last_queue_size && ctl.last_set_instant.elapsed().as_secs() > 30))
+        {
+            // 80% fps to ensure decoding is faster than encoding
+            let mut custom_fps = decode_fps as i32 * 4 / 5;
+            if custom_fps < 1 {
+                custom_fps = 1;
+            }
+            // send custom fps
+            let mut misc = Misc::new();
+            misc.set_option(OptionMessage {
+                custom_fps,
+                ..Default::default()
+            });
+            let mut msg = Message::new();
+            msg.set_misc(misc);
+            self.sender.send(Data::Message(msg)).ok();
+            ctl.last_queue_size = len;
+            ctl.set_times += 1;
+            ctl.last_set_instant = Instant::now();
+            refresh = true;
+        }
+        // send refresh
+        if ctl.refresh_times < 10 // enough
+            && (refresh
+                || (len > self.video_queue.len() / 2
+                    && ctl.last_refresh_instant.elapsed().as_secs() > 30))
+        {
+            self.handler.refresh_video();
+            ctl.refresh_times += 1;
+            ctl.last_refresh_instant = Instant::now();
+        }
+    }
 
     async fn handle_msg_from_peer(&mut self, data: &[u8], peer: &mut Stream) -> bool {
         if let Ok(msg_in) = Message::parse_from_bytes(&data) {
@@ -1489,3 +1551,23 @@ impl RemoveJob {
         }
     }
 }
+
+struct FpsControl {
+    last_queue_size: usize,
+    set_times: usize,
+    refresh_times: usize,
+    last_set_instant: Instant,
+    last_refresh_instant: Instant,
+}
+
+impl Default for FpsControl {
+    fn default() -> Self {
+        Self {
+            last_queue_size: Default::default(),
+            set_times: Default::default(),
+            refresh_times: Default::default(),
+            last_set_instant: Instant::now(),
+            last_refresh_instant: Instant::now(),
+        }
+    }
+}

src/server/video_qos.rs

@@ -1,7 +1,7 @@
 use super::*;
 use std::time::Duration;
 pub const FPS: u8 = 30;
-pub const MIN_FPS: u8 = 10;
+pub const MIN_FPS: u8 = 1;
 pub const MAX_FPS: u8 = 120;
 trait Percent {
     fn as_percent(&self) -> u32;
@@ -221,7 +221,9 @@ impl VideoQoS {
     }
 
     pub fn reset(&mut self) {
-        *self = Default::default();
+        self.fps = FPS;
+        self.user_fps = FPS;
+        self.updated = true;
     }
 
     pub fn check_abr_config(&mut self) -> bool {

src/server/video_service.rs

@@ -547,7 +547,7 @@ fn run(sp: GenericService) -> ResultType<()> {
         check_uac_switch(c.privacy_mode_id, c._capturer_privacy_mode_id)?;
 
         let mut video_qos = VIDEO_QOS.lock().unwrap();
-        if video_qos.check_if_updated() {
+        if video_qos.check_if_updated() && video_qos.target_bitrate > 0 {
             log::debug!(
                 "qos is updated, target_bitrate:{}, fps:{}",
                 video_qos.target_bitrate,

src/ui_session_interface.rs

@@ -1162,7 +1162,7 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>) {
     let frame_count = Arc::new(AtomicUsize::new(0));
     let frame_count_cl = frame_count.clone();
     let ui_handler = handler.ui_handler.clone();
-    let (video_sender, audio_sender, video_queue) =
+    let (video_sender, audio_sender, video_queue, decode_fps) =
         start_video_audio_threads(move |data: &mut Vec<u8>| {
             frame_count_cl.fetch_add(1, Ordering::Relaxed);
             ui_handler.on_rgba(data);
@@ -1176,6 +1176,7 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>) {
         receiver,
         sender,
         frame_count,
+        decode_fps,
     );
     remote.io_loop(&key, &token).await;
     remote.sync_jobs_status_to_local().await;