Optimize RFDETR nano segmentation GPU inference pipeline

inference/core/interfaces/stream/inference_pipeline.py

@@ -479,6 +479,7 @@ def init_with_workflow(
         video_source_properties: Optional[Dict[str, float]] = None,
         workflow_init_parameters: Optional[Dict[str, Any]] = None,
         workflows_thread_pool_workers: int = 4,
+        max_inflight_workflow_batches: int = 2,
         cancel_thread_pool_tasks_on_exit: bool = True,
         video_metadata_input_name: str = "video_metadata",
         batch_collection_timeout: Optional[float] = None,
@@ -547,6 +548,9 @@ def init_with_workflow(
                 with custom plugins.
             workflows_thread_pool_workers (int): Number of workers for workflows thread pool which is used
                 by workflows blocks to run background tasks.
+            max_inflight_workflow_batches (int): Number of workflow frame batches that can be processed concurrently
+                by the inference pipeline. Values greater than one allow CPU-heavy workflow result processing for
+                one frame batch to overlap with GPU work for another frame batch, while preserving output order.
             cancel_thread_pool_tasks_on_exit (bool): Flag to decide if unstated background tasks should be
                 canceled at the end of InferencePipeline processing. By default, when video file ends or
                 pipeline is stopped, tasks that has not started will be cancelled.
@@ -690,6 +694,7 @@ def init_with_workflow(
             batch_collection_timeout=batch_collection_timeout,
             predictions_queue_size=predictions_queue_size,
             decoding_buffer_size=decoding_buffer_size,
+            inference_thread_pool_workers=max_inflight_workflow_batches,
         )
 
     @classmethod
@@ -710,6 +715,7 @@ def init_with_custom_logic(
         sink_mode: SinkMode = SinkMode.ADAPTIVE,
         predictions_queue_size: int = PREDICTIONS_QUEUE_SIZE,
         decoding_buffer_size: int = DEFAULT_BUFFER_SIZE,
+        inference_thread_pool_workers: int = 1,
     ) -> "InferencePipeline":
         """
         This class creates the abstraction for making inferences from given workflow against video stream.
@@ -780,6 +786,8 @@ def init_with_custom_logic(
                 default value is taken from INFERENCE_PIPELINE_PREDICTIONS_QUEUE_SIZE env variable
             decoding_buffer_size (int): size of video source decoding buffer
                 default value is taken from VIDEO_SOURCE_BUFFER_SIZE env variable
+            inference_thread_pool_workers (int): Number of frame batches that can be processed concurrently.
+                Output dispatching order is preserved.
 
         Other ENV variables involved in low-level configuration:
         * INFERENCE_PIPELINE_PREDICTIONS_QUEUE_SIZE - size of buffer for predictions that are ready for dispatching
@@ -826,6 +834,7 @@ def init_with_custom_logic(
             on_pipeline_end=on_pipeline_end,
             batch_collection_timeout=batch_collection_timeout,
             sink_mode=sink_mode,
+            inference_thread_pool_workers=inference_thread_pool_workers,
         )
 
     def __init__(
@@ -841,6 +850,7 @@ def __init__(
         max_fps: Optional[float] = None,
         batch_collection_timeout: Optional[float] = None,
         sink_mode: SinkMode = SinkMode.ADAPTIVE,
+        inference_thread_pool_workers: int = 1,
     ):
         self._on_video_frame = on_video_frame
         self._video_sources = video_sources
@@ -858,6 +868,12 @@ def __init__(
         self._on_pipeline_end = on_pipeline_end
         self._batch_collection_timeout = batch_collection_timeout
         self._sink_mode = sink_mode
+        try:
+            self._inference_thread_pool_workers = max(
+                1, int(inference_thread_pool_workers)
+            )
+        except (TypeError, ValueError):
+            self._inference_thread_pool_workers = 1
 
     def start(self, use_main_thread: bool = True) -> None:
         self._stop = False
@@ -910,26 +926,52 @@ def _execute_inference(self) -> None:
             status_update_handlers=self._status_update_handlers,
         )
         logger.info(f"Inference thread started")
+        inference_executor = None
         try:
-            for video_frames in self._generate_frames():
-                self._watchdog.on_model_inference_started(
-                    frames=video_frames,
-                )
-                predictions = self._on_video_frame(video_frames)
-                self._watchdog.on_model_prediction_ready(
-                    frames=video_frames,
-                )
-                self._predictions_queue.put((predictions, video_frames))
-                send_inference_pipeline_status_update(
-                    severity=UpdateSeverity.DEBUG,
-                    event_type=INFERENCE_COMPLETED_EVENT,
-                    payload={
-                        "frames_ids": [f.frame_id for f in video_frames],
-                        "frames_timestamps": [f.frame_timestamp for f in video_frames],
-                        "sources_id": [f.source_id for f in video_frames],
-                    },
-                    status_update_handlers=self._status_update_handlers,
+            if self._inference_thread_pool_workers == 1:
+                for video_frames in self._generate_frames():
+                    predictions, video_frames = self._run_inference_on_video_frames(
+                        video_frames=video_frames
+                    )
+                    self._emit_inference_result(
+                        predictions=predictions,
+                        video_frames=video_frames,
+                    )
+            else:
+                inference_executor = ThreadPoolExecutor(
+                    max_workers=self._inference_thread_pool_workers
                 )
+                pending_predictions = {}
+                next_submitted_batch = 0
+                next_dispatched_batch = 0
+                for video_frames in self._generate_frames():
+                    while (
+                        len(pending_predictions) >= self._inference_thread_pool_workers
+                    ):
+                        predictions, predicted_video_frames = pending_predictions.pop(
+                            next_dispatched_batch
+                        ).result()
+                        self._emit_inference_result(
+                            predictions=predictions,
+                            video_frames=predicted_video_frames,
+                        )
+                        next_dispatched_batch += 1
+                    pending_predictions[next_submitted_batch] = (
+                        inference_executor.submit(
+                            self._run_inference_on_video_frames,
+                            video_frames=video_frames,
+                        )
+                    )
+                    next_submitted_batch += 1
+                while next_dispatched_batch < next_submitted_batch:
+                    predictions, predicted_video_frames = pending_predictions.pop(
+                        next_dispatched_batch
+                    ).result()
+                    self._emit_inference_result(
+                        predictions=predictions,
+                        video_frames=predicted_video_frames,
+                    )
+                    next_dispatched_batch += 1
 
         except Exception as error:
             payload = {
@@ -945,6 +987,8 @@ def _execute_inference(self) -> None:
             )
             logger.exception(f"Encountered inference error: {error}")
         finally:
+            if inference_executor is not None:
+                inference_executor.shutdown(wait=True, cancel_futures=True)
             self._predictions_queue.put(None)
             send_inference_pipeline_status_update(
                 severity=UpdateSeverity.INFO,
@@ -953,6 +997,36 @@ def _execute_inference(self) -> None:
             )
             logger.info(f"Inference thread finished")
 
+    def _run_inference_on_video_frames(
+        self,
+        video_frames: List[VideoFrame],
+    ) -> Tuple[List[AnyPrediction], List[VideoFrame]]:
+        self._watchdog.on_model_inference_started(
+            frames=video_frames,
+        )
+        predictions = self._on_video_frame(video_frames)
+        self._watchdog.on_model_prediction_ready(
+            frames=video_frames,
+        )
+        return predictions, video_frames
+
+    def _emit_inference_result(
+        self,
+        predictions: List[AnyPrediction],
+        video_frames: List[VideoFrame],
+    ) -> None:
+        self._predictions_queue.put((predictions, video_frames))
+        send_inference_pipeline_status_update(
+            severity=UpdateSeverity.DEBUG,
+            event_type=INFERENCE_COMPLETED_EVENT,
+            payload={
+                "frames_ids": [f.frame_id for f in video_frames],
+                "frames_timestamps": [f.frame_timestamp for f in video_frames],
+                "sources_id": [f.source_id for f in video_frames],
+            },
+            status_update_handlers=self._status_update_handlers,
+        )
+
     def _dispatch_inference_results(self) -> None:
         while True:
             inference_results: Optional[

inference/core/interfaces/stream/model_handlers/workflows.py

@@ -1,11 +1,19 @@
-from typing import List, Optional
+from typing import Callable, Dict, List, Optional
 
 from inference.core.interfaces.camera.entities import VideoFrame
 from inference.core.workflows.execution_engine.core import ExecutionEngine
-from inference.core.workflows.execution_engine.entities.base import VideoMetadata
+from inference.core.workflows.execution_engine.entities.base import (
+    ImageParentMetadata,
+    VideoMetadata,
+    WorkflowImageData,
+)
 
 
 class WorkflowRunner:
+    def __init__(self) -> None:
+        self._fast_path_cache: Dict[
+            int, Optional[Callable[[List[VideoFrame]], List[dict]]]
+        ] = {}
 
     def run_workflow(
         self,
@@ -17,6 +25,13 @@ def run_workflow(
         serialize_results: bool = False,
         _is_preview: bool = False,
     ) -> List[dict]:
+        if not workflows_parameters and not serialize_results and not _is_preview:
+            fast_path = self._get_single_step_fast_path(
+                execution_engine=execution_engine,
+                image_input_name=image_input_name,
+            )
+            if fast_path is not None:
+                return fast_path(video_frames)
         if workflows_parameters is None:
             workflows_parameters = {}
         # TODO: pass fps reflecting each stream to workflows_parameters
@@ -58,3 +73,94 @@ def run_workflow(
             serialize_results=serialize_results,
             _is_preview=_is_preview,
         )
+
+    def _get_single_step_fast_path(
+        self,
+        execution_engine: ExecutionEngine,
+        image_input_name: str,
+    ) -> Optional[Callable[[List[VideoFrame]], List[dict]]]:
+        cache_key = id(execution_engine)
+        if cache_key not in self._fast_path_cache:
+            self._fast_path_cache[cache_key] = self._build_single_step_fast_path(
+                execution_engine=execution_engine,
+                image_input_name=image_input_name,
+            )
+        return self._fast_path_cache[cache_key]
+
+    @staticmethod
+    def _build_single_step_fast_path(
+        execution_engine: ExecutionEngine,
+        image_input_name: str,
+    ) -> Optional[Callable[[List[VideoFrame]], List[dict]]]:
+        inner_engine = getattr(execution_engine, "_engine", None)
+        compiled_workflow = getattr(inner_engine, "_compiled_workflow", None)
+        if compiled_workflow is None:
+            return None
+        if compiled_workflow.input_substitutions:
+            return None
+        if len(compiled_workflow.steps) != 1:
+            return None
+        if len(compiled_workflow.workflow_definition.inputs) != 1:
+            return None
+        if len(compiled_workflow.workflow_definition.outputs) != 1:
+            return None
+        step_name, initialised_step = next(iter(compiled_workflow.steps.items()))
+        manifest = initialised_step.manifest
+        if (
+            getattr(manifest, "type", None)
+            != "roboflow_core/roboflow_instance_segmentation_model@v3"
+        ):
+            return None
+        if getattr(manifest, "images", None) != f"$inputs.{image_input_name}":
+            return None
+        output = compiled_workflow.workflow_definition.outputs[0]
+        if (
+            output.name != "predictions"
+            or output.selector != f"$steps.{step_name}.predictions"
+        ):
+            return None
+        step = initialised_step.step
+
+        def run_single_step_workflow(video_frames: List[VideoFrame]) -> List[dict]:
+            workflow_images = []
+            for idx, video_frame in enumerate(video_frames):
+                video_metadata = VideoMetadata(
+                    video_identifier=(
+                        str(video_frame.source_id)
+                        if video_frame.source_id
+                        else "default_source"
+                    ),
+                    frame_number=video_frame.frame_id,
+                    frame_timestamp=video_frame.frame_timestamp,
+                    fps=video_frame.fps,
+                    measured_fps=video_frame.measured_fps,
+                    comes_from_video_file=video_frame.comes_from_video_file,
+                )
+                parent_id = f"{image_input_name}.[{idx}]"
+                parent_metadata = ImageParentMetadata(parent_id=parent_id)
+                workflow_images.append(
+                    WorkflowImageData(
+                        parent_metadata=parent_metadata,
+                        workflow_root_ancestor_metadata=parent_metadata,
+                        numpy_image=video_frame.image,
+                        video_metadata=video_metadata,
+                    )
+                )
+            step_results = step.run(
+                images=workflow_images,
+                model_id=manifest.model_id,
+                confidence_mode=manifest.confidence_mode,
+                custom_confidence=manifest.custom_confidence,
+                class_agnostic_nms=manifest.class_agnostic_nms,
+                class_filter=manifest.class_filter,
+                iou_threshold=manifest.iou_threshold,
+                max_detections=manifest.max_detections,
+                max_candidates=manifest.max_candidates,
+                mask_decode_mode=manifest.mask_decode_mode,
+                tradeoff_factor=manifest.tradeoff_factor,
+                disable_active_learning=manifest.disable_active_learning,
+                active_learning_target_dataset=manifest.active_learning_target_dataset,
+            )
+            return [{"predictions": result["predictions"]} for result in step_results]
+
+        return run_single_step_workflow

inference/core/models/inference_models_adapters.py

@@ -280,7 +280,10 @@ def map_inference_kwargs(self, kwargs: dict) -> dict:
             disable_static_crop=kwargs.get("disable_preproc_static_crop", False),
         )
         kwargs["pre_processing_overrides"] = pre_processing_overrides
-        if "rle" in self._model.supported_mask_formats:
+        if (
+            kwargs.get("response_mask_format") == "rle"
+            and "rle" in self._model.supported_mask_formats
+        ):
             kwargs["mask_format"] = "rle"
         return kwargs
 

inference/core/utils/postprocess.py

@@ -52,11 +52,6 @@ def masks2poly(masks: np.ndarray) -> List[np.ndarray]:
                 m_bool = np.ascontiguousarray(m_bool)
             m_uint8 = m_bool.view(np.uint8)
 
-        # Quickly skip empty masks
-        if not np.any(m_uint8):
-            segments.append(np.zeros((0, 2), dtype=np.float32))
-            continue
-
         segments.append(mask2poly(m_uint8))
     return segments
 
@@ -88,11 +83,6 @@ def masks2multipoly(masks: np.ndarray) -> List[np.ndarray]:
                 m_bool = np.ascontiguousarray(m_bool)
             m_uint8 = m_bool.view(np.uint8)
 
-        # Quickly skip empty masks
-        if not np.any(m_uint8):
-            segments.append([np.zeros((0, 2), dtype=np.float32)])
-            continue
-
         segments.append(mask2multipoly(m_uint8))
     return segments
 

inference/core/workflows/core_steps/common/utils.py

@@ -118,7 +118,10 @@ def convert_inference_detections_batch_to_sv_detections(
             raw_predictions = filter_out_invalid_polygons(predictions=raw_predictions)
         parent_ids = [d.get(PARENT_ID_KEY, "") for d in raw_predictions]
         detection_ids = [
-            d.get(DETECTION_ID_KEY, str(uuid.uuid4())) for d in raw_predictions
+            d[DETECTION_ID_KEY]
+            if DETECTION_ID_KEY in d
+            else str(uuid.uuid4())
+            for d in raw_predictions
         ]
         detections[DETECTION_ID_KEY] = np.array(detection_ids)
         detections[PARENT_ID_KEY] = np.array(parent_ids)