From 61d33c0ab7a62511569ada0df2e761a1bdcbbc7c Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Tue, 20 May 2025 14:21:37 +0200 Subject: [PATCH 01/10] Improvements to webrtc stability --- inference/core/interfaces/http/http_api.py | 2 + .../manager_app/inference_pipeline_manager.py | 8 +- .../stream_manager/manager_app/webrtc.py | 162 +++++++++--------- inference/core/utils/async_utils.py | 32 +++- 4 files changed, 109 insertions(+), 95 deletions(-) diff --git a/inference/core/interfaces/http/http_api.py b/inference/core/interfaces/http/http_api.py index 92910847b7..ed7ebb94b4 100644 --- a/inference/core/interfaces/http/http_api.py +++ b/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( diff --git a/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py b/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py index 4b4e73beb6..60305b220b 100644 --- a/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py +++ b/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,10 +265,11 @@ 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, + processing_timeout=0.005, + fps_probe_frames=10, ), loop, ) @@ -279,7 +280,6 @@ def start_loop(loop: asyncio.AbstractEventLoop): 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( diff --git a/inference/core/interfaces/stream_manager/manager_app/webrtc.py b/inference/core/interfaces/stream_manager/manager_app/webrtc.py index 1da416bba0..6032057ad6 100644 --- a/inference/core/interfaces/stream_manager/manager_app/webrtc.py +++ b/inference/core/interfaces/stream_manager/manager_app/webrtc.py @@ -2,8 +2,9 @@ import concurrent.futures import time from threading import Event -from typing import Dict, Optional, Tuple +from typing import Dict, List, Optional, Tuple +import cv2 as cv import numpy as np from aiortc import ( RTCConfiguration, @@ -30,34 +31,60 @@ from inference.core.utils.function import experimental +FALLBACK_FPS: float = 10 + + +def overlay_text_on_frame(frame: VideoFrame, text: List[str]): + result_frame = frame.to_ndarray(format="bgr24") + for i, l in enumerate(text): + result_frame = cv.putText( + result_frame, + l, + (10, 20 + 30 * i), + cv.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 255, 0), + 2, + ) + return result_frame + + class VideoTransformTrack(VideoStreamTrack): def __init__( self, to_inference_queue: "SyncAsyncQueue[VideoFrame]", from_inference_queue: "SyncAsyncQueue[np.ndarray]", asyncio_loop: asyncio.AbstractEventLoop, - webrtc_peer_timeout: float = 1, + processing_timeout: float = 0.2, fps_probe_frames: int = 10, webcam_fps: Optional[float] = None, *args, **kwargs, ): super().__init__(*args, **kwargs) - if not webrtc_peer_timeout: - webrtc_peer_timeout = 1 - self.webrtc_peer_timeout: float = webrtc_peer_timeout + if not processing_timeout: + processing_timeout = 0.005 + self.processing_timeout: float = processing_timeout self.track: Optional[RemoteStreamTrack] = None + self._track_active: bool = True + self._id = time.time_ns() self._processed = 0 + self.to_inference_queue: "SyncAsyncQueue[VideoFrame]" = to_inference_queue self.from_inference_queue: "SyncAsyncQueue[np.ndarray]" = from_inference_queue + self._asyncio_loop = asyncio_loop self._pool = concurrent.futures.ThreadPoolExecutor() - self._track_active: bool = True + self._fps_probe_frames = fps_probe_frames + self._fps_probe_t1: Optional[float] = None + self._fps_probe_t2: Optional[float] = None self.incoming_stream_fps: Optional[float] = webcam_fps + self._last_frame: Optional[VideoFrame] = None + def set_track(self, track: RemoteStreamTrack): if not self.track: self.track = track @@ -66,69 +93,50 @@ def close(self): self._track_active = False async def recv(self): + frame = await self.track.recv() + self._processed += 1 if not self.incoming_stream_fps: - logger.debug("Probing incoming stream FPS") - t1 = 0 - t2 = 0 - for i in range(self._fps_probe_frames): - try: - frame: VideoFrame = await asyncio.wait_for( - self.track.recv(), self.webrtc_peer_timeout - ) - except (asyncio.TimeoutError, MediaStreamError): - logger.info( - "Timeout while waiting to receive frames sent through webrtc peer connection; assuming peer disconnected." - ) - self.close() - raise MediaStreamError - # drop first frame - if i == 1: - t1 = time.time() - t2 = time.time() - if t1 == t2: - logger.info( - "All frames probed in the same time - could not calculate fps." - ) - raise MediaStreamError - self.incoming_stream_fps = (self._fps_probe_frames - 1) / (t2 - t1) - logger.debug("Incoming stream fps: %s", self.incoming_stream_fps) - - while self._track_active: - try: - frame: VideoFrame = await asyncio.wait_for( - self.track.recv(), self.webrtc_peer_timeout + if not self._fps_probe_t1: + logger.debug("Probing incoming stream FPS") + if self._processed == 1: + self._fps_probe_t1 = time.time() + if self._processed == self._fps_probe_frames: + self._fps_probe_t2 = time.time() + if self._fps_probe_t1 == self._fps_probe_t2: + logger.warning( + "All frames probed in the same time - could not calculate fps, assuming fallback %s FPS.", + FALLBACK_FPS, ) - except (asyncio.TimeoutError, MediaStreamError): - logger.info( - "Timeout while waiting to receive frames sent through webrtc peer connection; assuming peer disconnected." + self.incoming_stream_fps = FALLBACK_FPS + elif self._fps_probe_t1 is not None and self._fps_probe_t2 is not None: + self.incoming_stream_fps = (self._fps_probe_frames - 1) / ( + self._fps_probe_t2 - self._fps_probe_t1 ) - self.close() - raise MediaStreamError + logger.info("Incoming stream fps: %s", self.incoming_stream_fps) + if not await self.to_inference_queue.async_full(): await self.to_inference_queue.async_put(frame) - from_inference_queue_empty = await self.from_inference_queue.async_empty() - if not from_inference_queue_empty: - break - - while self._track_active: - try: - res: np.ndarray = await asyncio.wait_for( - self.from_inference_queue.async_get(), self.webrtc_peer_timeout - ) - break - except asyncio.TimeoutError: - continue - if not self._track_active: - logger.info( - "Received close request while waiting to receive frames from inference pipeline; assuming termination." + try: + np_frame = await self.from_inference_queue.async_get( + timeout=self.processing_timeout ) - raise MediaStreamError - - new_frame = VideoFrame.from_ndarray(res, format="bgr24") - new_frame.pts = frame.pts + new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24") + self._last_frame = new_frame + except asyncio.TimeoutError: + logger.warning( + "Timeout while waiting for inference result, serving last frame" + ) + if self._last_frame: + new_frame = self._last_frame + else: + np_frame = overlay_text_on_frame( + frame, ["Inference pipeline is starting..."] + ) + new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24") + new_frame.pts = self._processed new_frame.time_base = frame.time_base - self._processed += 1 + return new_frame @@ -142,7 +150,6 @@ def __init__( to_inference_queue: "SyncAsyncQueue[VideoFrame]", stop_event: Event, webrtc_video_transform_track: VideoTransformTrack, - webrtc_peer_timeout: float, ): self.to_inference_queue: "SyncAsyncQueue[VideoFrame]" = to_inference_queue self._stop_event = stop_event @@ -150,7 +157,6 @@ def __init__( self._h: Optional[int] = None self._video_transform_track = webrtc_video_transform_track self._is_opened = True - self.webrtc_peer_timeout = webrtc_peer_timeout def grab(self) -> bool: if self._stop_event.is_set(): @@ -158,13 +164,9 @@ def grab(self) -> bool: self._is_opened = False return False - try: - res = self.to_inference_queue.sync_get(timeout=self.webrtc_peer_timeout) - if res is None: - self._stop_event.set() - return False - except asyncio.TimeoutError: - logger.error("Timeout while grabbing frame, considering source depleted.") + res = self.to_inference_queue.sync_get() + if res is None: + self._stop_event.set() return False return True @@ -174,15 +176,9 @@ def retrieve(self) -> Tuple[bool, Optional[np.ndarray]]: self._is_opened = False return False, None - try: - frame: VideoFrame = self.to_inference_queue.sync_get( - timeout=self.webrtc_peer_timeout - ) - if frame is None: - self._stop_event.set() - return False, None - except asyncio.TimeoutError: - logger.error("Timeout while retrieving frame, considering source depleted.") + frame: VideoFrame = self.to_inference_queue.sync_get() + if frame is None: + self._stop_event.set() return False, None img = frame.to_ndarray(format="bgr24") @@ -218,20 +214,22 @@ def __init__(self, video_transform_track: VideoTransformTrack, *args, **kwargs): async def init_rtc_peer_connection( webrtc_offer: WebRTCOffer, - webrtc_turn_config: WebRTCTURNConfig, to_inference_queue: "SyncAsyncQueue[VideoFrame]", from_inference_queue: "SyncAsyncQueue[np.ndarray]", - webrtc_peer_timeout: float, feedback_stop_event: Event, asyncio_loop: asyncio.AbstractEventLoop, + webrtc_turn_config: Optional[WebRTCTURNConfig] = None, webcam_fps: Optional[float] = None, + processing_timeout: float = 0.005, + fps_probe_frames: int = 10, ) -> RTCPeerConnectionWithFPS: video_transform_track = VideoTransformTrack( to_inference_queue=to_inference_queue, from_inference_queue=from_inference_queue, asyncio_loop=asyncio_loop, - webrtc_peer_timeout=webrtc_peer_timeout, webcam_fps=webcam_fps, + processing_timeout=processing_timeout, + fps_probe_frames=fps_probe_frames, ) if webrtc_turn_config: diff --git a/inference/core/utils/async_utils.py b/inference/core/utils/async_utils.py index 6c82bf167b..49298c4dca 100644 --- a/inference/core/utils/async_utils.py +++ b/inference/core/utils/async_utils.py @@ -20,26 +20,31 @@ async def async_lock( lock.release() -async def create_async_queue() -> asyncio.Queue: - return asyncio.Queue() +async def create_async_queue(maxsize: int = 0) -> asyncio.Queue: + return asyncio.Queue(maxsize=maxsize) class Queue: - def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None): + def __init__( + self, loop: Optional[asyncio.AbstractEventLoop] = None, maxsize: int = 0 + ): self._loop = loop if not self._loop: self._loop = asyncio.get_running_loop() self._queue = asyncio.run_coroutine_threadsafe( - create_async_queue(), self._loop + create_async_queue(maxsize=maxsize), self._loop ).result() def sync_put_nowait(self, item): - self._loop.call_soon(self._queue.put_nowait, item) + self._queue.put_nowait(item) def sync_put(self, item): asyncio.run_coroutine_threadsafe(self._queue.put(item), self._loop).result() - def sync_get(self, timeout: float): + def sync_get_nowait(self): + return self._queue.get_nowait() + + def sync_get(self, timeout: Optional[float] = None): return asyncio.run_coroutine_threadsafe( asyncio.wait_for(self._queue.get(), timeout), self._loop ).result() @@ -47,14 +52,23 @@ def sync_get(self, timeout: float): def sync_empty(self): return self._queue.empty() - def async_put_nowait(self, item): + def sync_full(self): + return self._queue.full() + + async def async_put_nowait(self, item): self._queue.put_nowait(item) async def async_put(self, item): await self._queue.put(item) - async def async_get(self): - return await self._queue.get() + async def async_get_nowait(self): + return self._queue.get_nowait() + + async def async_get(self, timeout: Optional[float] = None): + return await asyncio.wait_for(self._queue.get(), timeout) async def async_empty(self): return self._queue.empty() + + async def async_full(self): + return self._queue.full() From 83c9693ee94f90ea0b0cc671c29d91f026c44e50 Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Tue, 20 May 2025 14:40:12 +0200 Subject: [PATCH 02/10] formatting --- inference/core/interfaces/stream_manager/manager_app/webrtc.py | 1 - 1 file changed, 1 deletion(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/webrtc.py b/inference/core/interfaces/stream_manager/manager_app/webrtc.py index 6032057ad6..c18c8fc18d 100644 --- a/inference/core/interfaces/stream_manager/manager_app/webrtc.py +++ b/inference/core/interfaces/stream_manager/manager_app/webrtc.py @@ -30,7 +30,6 @@ from inference.core.utils.async_utils import Queue as SyncAsyncQueue from inference.core.utils.function import experimental - FALLBACK_FPS: float = 10 From 1855bd725281449235de6c296e1c635f3daee841 Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Wed, 21 May 2025 19:18:04 +0200 Subject: [PATCH 03/10] Safe-guard against pipeline that was already removed --- inference/core/interfaces/stream_manager/manager_app/app.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/inference/core/interfaces/stream_manager/manager_app/app.py b/inference/core/interfaces/stream_manager/manager_app/app.py index 7465c8005a..f45b135095 100644 --- a/inference/core/interfaces/stream_manager/manager_app/app.py +++ b/inference/core/interfaces/stream_manager/manager_app/app.py @@ -253,6 +253,11 @@ def _terminate_pipeline( ) with PROCESSES_TABLE_LOCK: # termination ended + if pipeline_id not in self._processes_table: + logger.warning( + f"Pipeline {pipeline_id} already removed from processes table." + ) + return pipeline = self._processes_table[pipeline_id] pipeline.is_terminating = False serialised_response = prepare_response( From 637fad8f0c0d5af605a3bb6f5cc0d481f86351e6 Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 13:36:10 +0200 Subject: [PATCH 04/10] Do not terminate execution of pipeline termination if pipeline id was already removed from process table --- inference/core/interfaces/stream_manager/manager_app/app.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/app.py b/inference/core/interfaces/stream_manager/manager_app/app.py index f45b135095..7c30b874ab 100644 --- a/inference/core/interfaces/stream_manager/manager_app/app.py +++ b/inference/core/interfaces/stream_manager/manager_app/app.py @@ -257,9 +257,9 @@ def _terminate_pipeline( logger.warning( f"Pipeline {pipeline_id} already removed from processes table." ) - return - pipeline = self._processes_table[pipeline_id] - pipeline.is_terminating = False + 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 ) From 8d83b0d1f6b8d74818d550464563d62291a8088a Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 13:37:31 +0200 Subject: [PATCH 05/10] Expose processing timeout and fps probe frames to request model with sane defaults --- .../stream_manager/manager_app/entities.py | 2 ++ .../manager_app/inference_pipeline_manager.py | 25 ++++++++++--------- 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/entities.py b/inference/core/interfaces/stream_manager/manager_app/entities.py index 64f345d68e..7c44a96891 100644 --- a/inference/core/interfaces/stream_manager/manager_app/entities.py +++ b/inference/core/interfaces/stream_manager/manager_app/entities.py @@ -108,6 +108,8 @@ 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 class ConsumeResultsPayload(BaseModel): diff --git a/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py b/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py index 60305b220b..ac01bd9351 100644 --- a/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py +++ b/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py @@ -268,13 +268,24 @@ def start_loop(loop: asyncio.AbstractEventLoop): feedback_stop_event=stop_event, asyncio_loop=loop, webcam_fps=webcam_fps, - processing_timeout=0.005, - fps_probe_frames=10, + 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, @@ -349,16 +360,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, From 7343a27af37f2a570bcb2abca7f4db989392ea0a Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 13:44:08 +0200 Subject: [PATCH 06/10] cycle frames in to-inference queue if they are not consumed on time; only log timeout warning if inference pipeline is already processing frames --- .../stream_manager/manager_app/webrtc.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/webrtc.py b/inference/core/interfaces/stream_manager/manager_app/webrtc.py index c18c8fc18d..9c0035935c 100644 --- a/inference/core/interfaces/stream_manager/manager_app/webrtc.py +++ b/inference/core/interfaces/stream_manager/manager_app/webrtc.py @@ -54,15 +54,13 @@ def __init__( to_inference_queue: "SyncAsyncQueue[VideoFrame]", from_inference_queue: "SyncAsyncQueue[np.ndarray]", asyncio_loop: asyncio.AbstractEventLoop, - processing_timeout: float = 0.2, - fps_probe_frames: int = 10, + processing_timeout: float, + fps_probe_frames: int, webcam_fps: Optional[float] = None, *args, **kwargs, ): super().__init__(*args, **kwargs) - if not processing_timeout: - processing_timeout = 0.005 self.processing_timeout: float = processing_timeout self.track: Optional[RemoteStreamTrack] = None @@ -115,6 +113,9 @@ 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: + await self.to_inference_queue.async_get_nowait() + await self.to_inference_queue.async_put_nowait(frame) try: np_frame = await self.from_inference_queue.async_get( @@ -123,10 +124,10 @@ async def recv(self): new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24") self._last_frame = new_frame except asyncio.TimeoutError: - logger.warning( - "Timeout while waiting for inference result, serving last frame" - ) if self._last_frame: + logger.warning( + "Timeout while waiting for inference result, serving last frame" + ) new_frame = self._last_frame else: np_frame = overlay_text_on_frame( @@ -217,10 +218,10 @@ async def init_rtc_peer_connection( from_inference_queue: "SyncAsyncQueue[np.ndarray]", feedback_stop_event: Event, asyncio_loop: asyncio.AbstractEventLoop, + processing_timeout: float, + fps_probe_frames: int, webrtc_turn_config: Optional[WebRTCTURNConfig] = None, webcam_fps: Optional[float] = None, - processing_timeout: float = 0.005, - fps_probe_frames: int = 10, ) -> RTCPeerConnectionWithFPS: video_transform_track = VideoTransformTrack( to_inference_queue=to_inference_queue, From 906bc2f30392c5c4093882e7b13eaae803148ba9 Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 13:44:33 +0200 Subject: [PATCH 07/10] remove duplicated log entries --- inference/core/logger.py | 1 + 1 file changed, 1 insertion(+) diff --git a/inference/core/logger.py b/inference/core/logger.py index bf3251908d..76bfd8aa99 100644 --- a/inference/core/logger.py +++ b/inference/core/logger.py @@ -144,6 +144,7 @@ def structlog_exception_formatter( handler = logging.StreamHandler() handler.setFormatter(NoTracebackFormatter("[%(levelname)s] %(message)s")) bounded_logger._logger.addHandler(handler) + bounded_logger._logger.propagate = False else: logger = logging.getLogger("inference") logger.setLevel(LOG_LEVEL) From 14434123f19abba4e1ae4edf97e708bce57debfe Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 15:05:22 +0200 Subject: [PATCH 08/10] When frames are not processed quick enough display overlay message on resulting frames --- .../stream_manager/manager_app/entities.py | 2 + .../manager_app/inference_pipeline_manager.py | 2 + .../stream_manager/manager_app/webrtc.py | 56 +++++++++++++++---- 3 files changed, 49 insertions(+), 11 deletions(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/entities.py b/inference/core/interfaces/stream_manager/manager_app/entities.py index 7c44a96891..906762b9ab 100644 --- a/inference/core/interfaces/stream_manager/manager_app/entities.py +++ b/inference/core/interfaces/stream_manager/manager_app/entities.py @@ -110,6 +110,8 @@ class InitialiseWebRTCPipelinePayload(InitialisePipelinePayload): 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): diff --git a/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py b/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py index ac01bd9351..077c681985 100644 --- a/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py +++ b/inference/core/interfaces/stream_manager/manager_app/inference_pipeline_manager.py @@ -268,6 +268,8 @@ def start_loop(loop: asyncio.AbstractEventLoop): 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, ), diff --git a/inference/core/interfaces/stream_manager/manager_app/webrtc.py b/inference/core/interfaces/stream_manager/manager_app/webrtc.py index 9c0035935c..c300ac5a9e 100644 --- a/inference/core/interfaces/stream_manager/manager_app/webrtc.py +++ b/inference/core/interfaces/stream_manager/manager_app/webrtc.py @@ -33,11 +33,10 @@ FALLBACK_FPS: float = 10 -def overlay_text_on_frame(frame: VideoFrame, text: List[str]): - result_frame = frame.to_ndarray(format="bgr24") +def overlay_text_on_np_frame(frame: np.ndarray, text: List[str]): for i, l in enumerate(text): - result_frame = cv.putText( - result_frame, + frame = cv.putText( + frame, l, (10, 20 + 30 * i), cv.FONT_HERSHEY_SIMPLEX, @@ -45,7 +44,7 @@ def overlay_text_on_frame(frame: VideoFrame, text: List[str]): (0, 255, 0), 2, ) - return result_frame + return frame class VideoTransformTrack(VideoStreamTrack): @@ -56,6 +55,8 @@ 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, @@ -81,6 +82,10 @@ 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 def set_track(self, track: RemoteStreamTrack): if not self.track: @@ -90,7 +95,7 @@ def close(self): self._track_active = False async def recv(self): - frame = await self.track.recv() + frame: VideoFrame = await self.track.recv() self._processed += 1 if not self.incoming_stream_fps: if not self._fps_probe_t1: @@ -117,23 +122,48 @@ async def recv(self): await self.to_inference_queue.async_get_nowait() await self.to_inference_queue.async_put_nowait(frame) + np_frame: Optional[np.ndarray] = None try: np_frame = await self.from_inference_queue.async_get( timeout=self.processing_timeout ) 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: if self._last_frame: - logger.warning( - "Timeout while waiting for inference result, serving 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..."] + 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 = self._last_frame + 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: - np_frame = overlay_text_on_frame( - frame, ["Inference pipeline is starting..."] + 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.pts = self._processed new_frame.time_base = frame.time_base @@ -220,6 +250,8 @@ 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, ) -> RTCPeerConnectionWithFPS: @@ -230,6 +262,8 @@ 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: From 1687aaf62c61c3ebcce9bbce1d64d46f71a5e1d9 Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 15:05:37 +0200 Subject: [PATCH 09/10] Return 0 as RAM usage if process PID cannot be found --- inference/core/interfaces/stream_manager/manager_app/app.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/app.py b/inference/core/interfaces/stream_manager/manager_app/app.py index 7c30b874ab..0651155fe7 100644 --- a/inference/core/interfaces/stream_manager/manager_app/app.py +++ b/inference/core/interfaces/stream_manager/manager_app/app.py @@ -542,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: From 478b69e2860c235cfed08e1a1cdc5007556f4bf3 Mon Sep 17 00:00:00 2001 From: Grzegorz Klimaszewski <166530809+grzegorz-roboflow@users.noreply.github.com> Date: Thu, 22 May 2025 15:08:11 +0200 Subject: [PATCH 10/10] formatting --- .../stream_manager/manager_app/webrtc.py | 23 ++++++++++++++----- 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/inference/core/interfaces/stream_manager/manager_app/webrtc.py b/inference/core/interfaces/stream_manager/manager_app/webrtc.py index c300ac5a9e..30cb7cdc1d 100644 --- a/inference/core/interfaces/stream_manager/manager_app/webrtc.py +++ b/inference/core/interfaces/stream_manager/manager_app/webrtc.py @@ -141,24 +141,35 @@ async def recv(self): 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..."] + workflow_too_slow_message = [ + "Workflow is too heavy to process all frames on time..." + ] 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..."] + 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: + 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 + 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: + 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 + self._last_frame.to_ndarray(format="bgr24"), + workflow_too_slow_message, ) new_frame = VideoFrame.from_ndarray(np_frame, format="bgr24") else: