Merge pull request #1305 from roboflow/fix/improve-webrtc-stability

Improvements to webrtc stability

Author: grzegorz-roboflow

inference/core/interfaces/http/http_api.py

@@ -1493,9 +1493,11 @@ async def initialise(request: InitialisePipelinePayload) -> CommandResponse:
             async def initialise_webrtc_inference_pipeline(
                 request: InitialiseWebRTCPipelinePayload,
             ) -> CommandResponse:
+                logger.debug("Received initialise webrtc inference pipeline request")
                 resp = await self.stream_manager_client.initialise_webrtc_pipeline(
                     initialisation_request=request
                 )
+                logger.debug("Returning initialise webrtc inference pipeline response")
                 return resp
 
             @app.post(

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

@@ -253,8 +253,13 @@ def _terminate_pipeline(
             )
             with PROCESSES_TABLE_LOCK:
                 # termination ended
-                pipeline = self._processes_table[pipeline_id]
-                pipeline.is_terminating = False
+                if pipeline_id not in self._processes_table:
+                    logger.warning(
+                        f"Pipeline {pipeline_id} already removed from processes table."
+                    )
+                else:
+                    pipeline = self._processes_table[pipeline_id]
+                    pipeline.is_terminating = False
         serialised_response = prepare_response(
             request_id=request_id, response=response, pipeline_id=pipeline_id
         )
@@ -537,7 +542,10 @@ def spawn_managed_pipeline_process(
 
 
 def _get_process_memory_usage_mb(process: Process) -> int:
-    return psutil.Process(process.pid).memory_info().rss / (1024 * 1024)
+    try:
+        return psutil.Process(process.pid).memory_info().rss / (1024 * 1024)
+    except psutil.NoSuchProcess:
+        return 0
 
 
 def start(expected_warmed_up_pipelines: int = 0) -> None:

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

@@ -108,6 +108,10 @@ class InitialiseWebRTCPipelinePayload(InitialisePipelinePayload):
     data_output: Optional[List[Optional[str]]] = Field(default_factory=list)
     webrtc_peer_timeout: float = 1
     webcam_fps: Optional[float] = None
+    processing_timeout: float = 0.005
+    fps_probe_frames: int = 10
+    max_consecutive_timeouts: int = 30
+    min_consecutive_on_time: int = 5
 
 
 class ConsumeResultsPayload(BaseModel):

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

@@ -254,8 +254,8 @@ def start_loop(loop: asyncio.AbstractEventLoop):
             webrtc_offer = parsed_payload.webrtc_offer
             webrtc_turn_config = parsed_payload.webrtc_turn_config
             webcam_fps = parsed_payload.webcam_fps
-            to_inference_queue = SyncAsyncQueue(loop=loop)
-            from_inference_queue = SyncAsyncQueue(loop=loop)
+            to_inference_queue = SyncAsyncQueue(loop=loop, maxsize=10)
+            from_inference_queue = SyncAsyncQueue(loop=loop, maxsize=10)
 
             stop_event = Event()
 
@@ -265,21 +265,34 @@ def start_loop(loop: asyncio.AbstractEventLoop):
                     webrtc_turn_config=webrtc_turn_config,
                     to_inference_queue=to_inference_queue,
                     from_inference_queue=from_inference_queue,
-                    webrtc_peer_timeout=parsed_payload.webrtc_peer_timeout,
                     feedback_stop_event=stop_event,
                     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,
                 ),
                 loop,
             )
             peer_connection: RTCPeerConnectionWithFPS = future.result()
 
+            self._responses_queue.put(
+                (
+                    request_id,
+                    {
+                        STATUS_KEY: OperationStatus.SUCCESS,
+                        "sdp": peer_connection.localDescription.sdp,
+                        "type": peer_connection.localDescription.type,
+                    },
+                )
+            )
+
             webrtc_producer = partial(
                 WebRTCVideoFrameProducer,
                 to_inference_queue=to_inference_queue,
                 stop_event=stop_event,
                 webrtc_video_transform_track=peer_connection.video_transform_track,
-                webrtc_peer_timeout=parsed_payload.webrtc_peer_timeout,
             )
 
             def webrtc_sink(
@@ -349,16 +362,6 @@ def webrtc_sink(
                 decoding_buffer_size=parsed_payload.decoding_buffer_size,
             )
             self._inference_pipeline.start(use_main_thread=False)
-            self._responses_queue.put(
-                (
-                    request_id,
-                    {
-                        STATUS_KEY: OperationStatus.SUCCESS,
-                        "sdp": peer_connection.localDescription.sdp,
-                        "type": peer_connection.localDescription.type,
-                    },
-                )
-            )
             logger.info(f"WebRTC pipeline initialised. request_id={request_id}...")
         except (
             ValidationError,