Merge pull request #1444 from roboflow/fix/webrtc-latency-improvements

Webrtc latency for heavy workflows

Author: grzegorz-roboflow

inference/core/interfaces/camera/video_source.py

@@ -550,21 +550,12 @@ def read_frame(self, timeout: Optional[float] = None) -> Optional[VideoFrame]:
             self._fps = source_metadata.source_properties.fps
             if not self._fps or self._fps <= 0 or self._fps > 1000:
                 self._fps = 30  # sane default
-        if not self._is_file:
-            current_timestamp = time.time_ns()
-            if (current_timestamp - self._last_frame_timestamp) / 1e9 < 1 / self._fps:
-                time.sleep(
-                    (1 / self._fps)
-                    - (current_timestamp - self._last_frame_timestamp) / 1e9
-                )
         video_frame: Optional[Union[VideoFrame, str]] = get_from_queue(
             queue=self._frames_buffer,
             on_successful_read=self._video_consumer.notify_frame_consumed,
             timeout=timeout,
             purge=self._buffer_consumption_strategy is BufferConsumptionStrategy.EAGER,
         )
-        if not self._is_file:
-            self._last_frame_timestamp = time.time_ns()
         if video_frame == POISON_PILL:
             raise EndOfStreamError(
                 "Attempted to retrieve frame from stream that already ended."
@@ -926,7 +917,7 @@ def _set_file_mode_buffering_strategies(self) -> None:
 
     def _set_stream_mode_buffering_strategies(self) -> None:
         if self._buffer_filling_strategy is None:
-            self._buffer_filling_strategy = BufferFillingStrategy.DROP_OLDEST
+            self._buffer_filling_strategy = BufferFillingStrategy.ADAPTIVE_DROP_OLDEST
 
     def _video_fps_should_be_sub_sampled(self) -> bool:
         if self._desired_fps is None:

inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py

@@ -280,8 +280,6 @@ def start_loop(loop: asyncio.AbstractEventLoop):
                     from_inference_queue=from_inference_queue,
                     asyncio_loop=loop,
                     webcam_fps=webcam_fps,
-                    max_consecutive_timeouts=parsed_payload.max_consecutive_timeouts,
-                    min_consecutive_on_time=parsed_payload.min_consecutive_on_time,
                     processing_timeout=parsed_payload.processing_timeout,
                     fps_probe_frames=parsed_payload.fps_probe_frames,
                     data_output=data_output,

inference/core/interfaces/stream_manager/manager_app/webrtc.py

@@ -1,5 +1,6 @@
 import asyncio
 import concurrent.futures
+import datetime
 import json
 import time
 from typing import Any, Dict, List, Optional, Tuple, Union
@@ -62,10 +63,27 @@ def get_frame_from_workflow_output(
 ) -> Optional[np.ndarray]:
     step_output = workflow_output.get(frame_output_key)
     if isinstance(step_output, WorkflowImageData):
+        if (
+            step_output.video_metadata
+            and step_output.video_metadata.frame_timestamp is not None
+        ):
+            latency = (
+                datetime.datetime.now() - step_output.video_metadata.frame_timestamp
+            )
+            logger.info("Processing latency: %ss", latency.total_seconds())
         return step_output.numpy_image
     elif isinstance(step_output, dict):
         for frame_output in step_output.values():
             if isinstance(frame_output, WorkflowImageData):
+                if (
+                    frame_output.video_metadata
+                    and frame_output.video_metadata.frame_timestamp is not None
+                ):
+                    latency = (
+                        datetime.datetime.now()
+                        - frame_output.video_metadata.frame_timestamp
+                    )
+                    logger.info("Processing latency: %ss", latency.total_seconds())
                 return frame_output.numpy_image
 
 
@@ -77,8 +95,6 @@ def __init__(
         asyncio_loop: asyncio.AbstractEventLoop,
         processing_timeout: float,
         fps_probe_frames: int,
-        min_consecutive_on_time: int,
-        max_consecutive_timeouts: Optional[int] = None,
         webcam_fps: Optional[float] = None,
         *args,
         **kwargs,
@@ -104,10 +120,6 @@ def __init__(
         self.incoming_stream_fps: Optional[float] = webcam_fps
 
         self._last_frame: Optional[VideoFrame] = None
-        self._consecutive_timeouts: int = 0
-        self._consecutive_on_time: int = 0
-        self._max_consecutive_timeouts: Optional[int] = max_consecutive_timeouts
-        self._min_consecutive_on_time: int = min_consecutive_on_time
 
         self._av_logging_set: bool = False
 
@@ -146,7 +158,7 @@ async def recv(self):
 
         if not await self.to_inference_queue.async_full():
             await self.to_inference_queue.async_put(frame)
-        elif not self._last_frame:
+        else:
             await self.to_inference_queue.async_get_nowait()
             await self.to_inference_queue.async_put_nowait(frame)
 
@@ -157,55 +169,22 @@ async def recv(self):
             )
             new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24")
             self._last_frame = new_frame
-
-            if self._max_consecutive_timeouts:
-                self._consecutive_on_time += 1
-                if self._consecutive_on_time >= self._min_consecutive_on_time:
-                    self._consecutive_timeouts = 0
         except asyncio.TimeoutError:
-            while not await self.to_inference_queue.async_empty():
-                await self.to_inference_queue.async_get_nowait()
-            if self._last_frame:
-                if self._max_consecutive_timeouts:
-                    self._consecutive_timeouts += 1
-                    if self._consecutive_timeouts >= self._max_consecutive_timeouts:
-                        self._consecutive_on_time = 0
-
-        workflow_too_slow_message = [
-            "Workflow is too heavy to process all frames on time..."
-        ]
+            pass
+
         if np_frame is None:
             if not self._last_frame:
                 np_frame = overlay_text_on_np_frame(
                     frame.to_ndarray(format="bgr24"),
                     ["Inference pipeline is starting..."],
                 )
                 new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24")
-            elif (
-                self._max_consecutive_timeouts
-                and self._consecutive_timeouts >= self._max_consecutive_timeouts
-            ):
-                np_frame = overlay_text_on_np_frame(
-                    self._last_frame.to_ndarray(format="bgr24"),
-                    workflow_too_slow_message,
-                )
-                new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24")
             else:
                 new_frame = self._last_frame
         else:
-            if (
-                self._max_consecutive_timeouts
-                and self._consecutive_timeouts >= self._max_consecutive_timeouts
-            ):
-                np_frame = overlay_text_on_np_frame(
-                    self._last_frame.to_ndarray(format="bgr24"),
-                    workflow_too_slow_message,
-                )
-                new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24")
-            else:
-                new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24")
+            new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24")
 
-        new_frame.pts = self._processed
+        new_frame.pts = frame.pts
         new_frame.time_base = frame.time_base
 
         return new_frame
@@ -312,8 +291,6 @@ async def init_rtc_peer_connection(
     asyncio_loop: asyncio.AbstractEventLoop,
     processing_timeout: float,
     fps_probe_frames: int,
-    max_consecutive_timeouts: int,
-    min_consecutive_on_time: int,
     webrtc_turn_config: Optional[WebRTCTURNConfig] = None,
     webcam_fps: Optional[float] = None,
     stream_output: Optional[str] = None,
@@ -326,8 +303,6 @@ async def init_rtc_peer_connection(
         webcam_fps=webcam_fps,
         processing_timeout=processing_timeout,
         fps_probe_frames=fps_probe_frames,
-        max_consecutive_timeouts=max_consecutive_timeouts,
-        min_consecutive_on_time=min_consecutive_on_time,
     )
 
     if webrtc_turn_config:

inference/core/version.py

@@ -1,4 +1,4 @@
-__version__ = "0.51.5"
+__version__ = "0.51.6"
 
 
 if __name__ == "__main__":