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/app.py b/inference/core/interfaces/stream_manager/manager_app/app.py index 7465c8005a..0651155fe7 100644 --- a/inference/core/interfaces/stream_manager/manager_app/app.py +++ b/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: diff --git a/inference/core/interfaces/stream_manager/manager_app/entities.py b/inference/core/interfaces/stream_manager/manager_app/entities.py index 64f345d68e..906762b9ab 100644 --- a/inference/core/interfaces/stream_manager/manager_app/entities.py +++ b/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): 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..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 @@ -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, diff --git a/inference/core/interfaces/stream_manager/manager_app/webrtc.py b/inference/core/interfaces/stream_manager/manager_app/webrtc.py index 1da416bba0..30cb7cdc1d 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, @@ -29,6 +30,22 @@ from inference.core.utils.async_utils import Queue as SyncAsyncQueue from inference.core.utils.function import experimental +FALLBACK_FPS: float = 10 + + +def overlay_text_on_np_frame(frame: np.ndarray, text: List[str]): + for i, l in enumerate(text): + frame = cv.putText( + frame, + l, + (10, 20 + 30 * i), + cv.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 255, 0), + 2, + ) + return frame + class VideoTransformTrack(VideoStreamTrack): def __init__( @@ -36,28 +53,40 @@ def __init__( to_inference_queue: "SyncAsyncQueue[VideoFrame]", from_inference_queue: "SyncAsyncQueue[np.ndarray]", asyncio_loop: asyncio.AbstractEventLoop, - webrtc_peer_timeout: float = 1, - fps_probe_frames: int = 10, + 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, ): super().__init__(*args, **kwargs) - if not webrtc_peer_timeout: - webrtc_peer_timeout = 1 - self.webrtc_peer_timeout: float = webrtc_peer_timeout + 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 + 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: self.track = track @@ -66,69 +95,89 @@ def close(self): self._track_active = False async def recv(self): + frame: VideoFrame = 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) + elif not self._last_frame: + await self.to_inference_queue.async_get_nowait() + await self.to_inference_queue.async_put_nowait(frame) - from_inference_queue_empty = await self.from_inference_queue.async_empty() - if not from_inference_queue_empty: - break + 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 - while self._track_active: - try: - res: np.ndarray = await asyncio.wait_for( - self.from_inference_queue.async_get(), self.webrtc_peer_timeout + 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: + 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..."], ) - 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." - ) - raise MediaStreamError + 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(res, format="bgr24") - new_frame.pts = frame.pts + new_frame.pts = self._processed new_frame.time_base = frame.time_base - self._processed += 1 + return new_frame @@ -142,7 +191,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 +198,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 +205,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 +217,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 +255,26 @@ 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, + 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: 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, + max_consecutive_timeouts=max_consecutive_timeouts, + min_consecutive_on_time=min_consecutive_on_time, ) if webrtc_turn_config: 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) 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()