优化：解耦帧调度与推理队列，提升单卡实时推理稳定性

core/tensorrt_engine.py

@@ -116,7 +116,7 @@ class TensorRTEngine:
                 if self._context is not None:
                     self._release_resources()
                 
-                self._cuda_context = cuda.Device(0).make_context()
+                self._cuda_context = cuda.Device(self.config.device_id).make_context()
                 self._stream = cuda.Stream()
                 
                 TRT_LOGGER = trt.Logger(trt.Logger.WARNING)

core/video_stream.py

@@ -102,6 +102,7 @@ class RTSPStreamReader:
         
         self._logger = get_logger("video_stream")
         self._lock = threading.Lock()
+        self._latest_frame: Optional[VideoFrame] = None
     
     @property
     def is_connected(self) -> bool:
@@ -221,6 +222,8 @@ class RTSPStreamReader:
                             pass
 
                     self._frame_buffer.put_nowait(frame_obj)
+                    with self._lock:
+                        self._latest_frame = frame_obj
 
                     if self._on_frame_callback:
                         self._on_frame_callback(frame_obj)
@@ -293,6 +296,9 @@ class RTSPStreamReader:
                 self._frame_buffer.get_nowait()
             except queue.Empty:
                 break
+
+        with self._lock:
+            self._latest_frame = None
         
         self._logger.info(f"视频流已停止: {self.camera_id}")
     
@@ -305,16 +311,11 @@ class RTSPStreamReader:
     
     def get_latest_frame(self, timeout: float = 1.0) -> Optional[VideoFrame]:
         """获取最新帧(丢弃中间帧)"""
-        try:
-            while True:
-                try:
-                    frame = self._frame_buffer.get_nowait()
-                    if self._frame_buffer.empty():
-                        return frame
-                except queue.Empty:
-                    return None
-        except Exception:
-            return None
+        del timeout
+        with self._lock:
+            frame = self._latest_frame
+            self._latest_frame = None
+        return frame
     
     def get_frame_batch(self, max_count: int = 8, 
                         timeout: float = 2.0) -> List[VideoFrame]:

docs/edge-inference-optimization.md

@@ -0,0 +1,172 @@
+# Edge 推理结构优化说明
+
+## 目标
+
+本次优化只调整 `security-ai-edge` 内部实时推理链路，目标是：
+
+- 保持现有功能不变
+- 保持与其他项目的现有对接方式不变
+- 提升单卡场景下的实时稳定性
+- 降低解码线程被 CPU 预处理拖慢的风险
+- 在高并发下优先处理最新帧，避免旧帧堆积
+
+本次未修改以下外部行为：
+
+- WVP 下发摄像头与 ROI 配置的方式
+- Redis / MQTT / HTTP 对接协议
+- 告警生成、告警去重、截图回调、心跳上报的外部接口
+- TensorRT engine 文件加载方式
+
+## 改动概览
+
+### 1. 解耦读帧线程与推理预处理
+
+优化前：
+
+- `RTSPStreamReader` 读到帧后，直接通过回调进入 `_process_frame`
+- `_process_frame` 内部会做 ROI 遍历、裁剪、预处理、入推理队列
+- 当 ROI 多、CPU 紧张时，读帧线程容易被预处理阻塞
+
+优化后：
+
+- `RTSPStreamReader` 只负责读帧和保存“最新帧”
+- 新增中心调度线程 `FrameScheduler`
+- 调度线程从每路流提取最新帧，再统一调用 `_process_frame`
+
+收益：
+
+- 解码线程不再承担重 CPU 预处理
+- 拉流稳定性更高
+- 更符合实时视频系统“最新帧优先”的处理模型
+
+### 2. 每路流改为“最新帧消费”
+
+优化前：
+
+- `get_latest_frame()` 通过清空队列获取最后一帧
+- 在高并发下，队列遍历和频繁出队会增加额外开销
+
+优化后：
+
+- 流对象内部维护 `_latest_frame`
+- 调度线程直接消费这份最新帧引用
+- 中间帧天然被覆盖，不再积压到后面阶段
+
+收益：
+
+- 获取最新帧成本更低
+- 更适合实时系统而非离线批处理
+
+### 3. ROI 预处理结果复用
+
+优化前：
+
+- 同一 ROI 下如果绑定多个算法，会重复执行 `preprocess_single`
+
+优化后：
+
+- 先筛出启用的 `bind`
+- 对每个 ROI 只执行一次预处理
+- 多个 `bind` 复用同一份裁剪图和缩放信息
+
+收益：
+
+- 降低 CPU crop / resize / normalize 开销
+- ROI 绑定越多，收益越明显
+
+### 4. 推理队列增加拥塞保护
+
+优化前：
+
+- ROI 任务持续追加到 `_batch_roi_queue`
+- 在输入高于推理吞吐时，可能出现旧任务堆积
+
+优化后：
+
+- 为 `_batch_roi_queue` 增加最大挂起量限制
+- 超限时丢弃最旧 ROI 任务
+
+收益：
+
+- 控制排队长度
+- 避免系统在高压时持续处理过期画面
+
+说明：
+
+- 这是实时系统常见策略，目标是“宁可丢旧帧，不处理过期帧”
+
+### 5. TensorRT 设备选择修正
+
+优化前：
+
+- `TensorRT` 初始化时固定使用 `cuda.Device(0)`
+- 即使配置里指定了 `device_id`，实际也不会生效
+
+优化后：
+
+- 改为使用 `self.config.device_id`
+
+收益：
+
+- 保留现有单卡行为
+- 为后续指定 GPU 设备提供正确基础
+
+## 当前结构
+
+优化后的主链路为：
+
+1. `RTSPStreamReader` 拉流并保存最新帧
+2. `FrameScheduler` 周期性轮询所有流
+3. `_process_frame` 生成 ROI 推理任务
+4. `_batch_roi_queue` 聚合任务并做拥塞保护
+5. `InferenceWorker` 做小批量 TensorRT 推理
+6. `PostProcessor` + `AlgorithmManager` 输出业务告警
+7. `ResultReporter` / `AlarmUploadWorker` 继续负责上报
+
+## 兼容性说明
+
+本次优化没有改变这些关键约束：
+
+- 摄像头仍然按 ROI 配置启动和停用
+- 算法处理入口 `_handle_detections` 不变
+- 现有告警冷却、ROI 去重、resolve 机制不变
+- 现有外部平台无需修改配置或接口
+
+因此它属于“内部调度优化”，而不是“系统集成改造”。
+
+## 新增运行指标
+
+`get_status()` 新增了以下统计项：
+
+- `dropped_frames`
+- `dropped_roi_tasks`
+- `inference_batches`
+- `scheduler_cycles`
+
+这些指标用于判断：
+
+- 是否存在旧帧丢弃
+- 是否出现 ROI 队列拥塞
+- 推理批次是否稳定
+- 调度线程是否持续运行
+
+## 仍可继续优化的方向
+
+本次改造重点是“在当前架构下提升实时性和稳定性”，后续还可以继续做：
+
+- 将 OpenCV/FFmpeg CPU 解码切换到 GPU 硬解
+- 将调度参数做成配置项，例如最大帧年龄、调度周期、最大挂起 ROI 数
+- 为不同摄像头增加优先级调度
+- 为高频 ROI 增加更细粒度缓存和复用
+- 为推理链路补充阶段耗时统计：调度、预处理、推理、后处理
+
+## 适用场景
+
+这版结构更适合：
+
+- 单卡部署
+- 多路 RTSP 并发
+- 小 batch 实时推理
+- 需要低延迟而非离线全量处理的场景
+
+如果后续要切到 GPU 解码，当前这版调度结构也更容易继续演进，因为“读帧”和“处理帧”已经拆开了。

main.py

@@ -59,6 +59,7 @@ class EdgeInferenceService:
         self._debug_http_server = None
         self._debug_http_thread: Optional[threading.Thread] = None
         self._heartbeat_thread: Optional[threading.Thread] = None
+        self._scheduler_thread: Optional[threading.Thread] = None
         
         self._processing_threads: Dict[str, threading.Thread] = {}
         self._stop_event = threading.Event()
@@ -68,6 +69,10 @@ class EdgeInferenceService:
             "total_frames_processed": 0,
             "total_alerts_generated": 0,
             "uptime_seconds": 0,
+            "dropped_frames": 0,
+            "dropped_roi_tasks": 0,
+            "scheduler_cycles": 0,
+            "inference_batches": 0,
         }
         
         self._batch_roi_queue: List[tuple] = []
@@ -76,6 +81,9 @@ class EdgeInferenceService:
         self._inference_thread: Optional[threading.Thread] = None
         self._max_batch_size = 8
         self._batch_timeout_sec = 0.05  # 50ms 攒批窗口
+        self._scheduler_interval_sec = 0.01
+        self._max_frame_age_sec = 0.5
+        self._max_pending_roi_items = self._max_batch_size * 32
 
         # 摄像头级别告警去重：同一摄像头+告警类型在冷却期内只上报一次
         self._camera_alert_cooldown: Dict[str, datetime] = {}
@@ -459,7 +467,6 @@ class EdgeInferenceService:
                     camera_id=camera.camera_id,
                     rtsp_url=camera.rtsp_url,
                     target_fps=self._settings.video_stream.default_fps,
-                    on_frame_callback=self._create_frame_callback(camera.camera_id)
                 )
                 self._logger.info(f"已添加摄像头: {camera.camera_id}")
                 success_count += 1
@@ -525,7 +532,6 @@ class EdgeInferenceService:
                         camera_id=camera.camera_id,
                         rtsp_url=camera.rtsp_url,
                         target_fps=self._settings.video_stream.default_fps,
-                        on_frame_callback=self._create_frame_callback(camera.camera_id)
                     )
                     # 立即启动新添加的流
                     self._stream_manager._streams[camera.camera_id].start()
@@ -586,12 +592,24 @@ class EdgeInferenceService:
         except Exception as e:
             self._logger.error(f"清理摄像头流失败: {e}")
 
-    def _create_frame_callback(self, camera_id: str):
-        """创建帧处理回调"""
-        def callback(frame):
-            self._process_frame(camera_id, frame)
-        return callback
-    
+    def _enqueue_roi_items(self, roi_items: List[tuple]):
+        """向推理队列追加 ROI 任务，并在拥塞时丢弃最旧任务。"""
+        if not roi_items:
+            return
+
+        with self._batch_lock:
+            overflow = max(
+                0,
+                len(self._batch_roi_queue) + len(roi_items) - self._max_pending_roi_items
+            )
+            if overflow > 0:
+                del self._batch_roi_queue[:overflow]
+                self._performance_stats["dropped_roi_tasks"] += overflow
+
+            self._batch_roi_queue.extend(roi_items)
+
+        self._batch_event.set()
+
     def _process_frame(self, camera_id: str, frame: VideoFrame):
         """处理视频帧 - 批量处理多 ROI"""
         try:
@@ -607,32 +625,53 @@ class EdgeInferenceService:
                     continue
                 if not roi.bindings:
                     continue
-                
-                for bind in roi.bindings:
-                    if not bind.enabled:
-                        continue
-                    
-                    try:
-                        cropped, scale_info = self._preprocessor.preprocess_single(
-                            frame.image, roi
-                        )
-                        roi_items.append((camera_id, roi, bind, frame, cropped, scale_info))
-                    except Exception as e:
-                        self._logger.error(f"预处理 ROI 失败 {roi.roi_id}: {e}")
+
+                enabled_binds = [bind for bind in roi.bindings if bind.enabled]
+                if not enabled_binds:
+                    continue
+
+                try:
+                    cropped, scale_info = self._preprocessor.preprocess_single(
+                        frame.image, roi
+                    )
+                except Exception as e:
+                    self._logger.error(f"预处理 ROI 失败 {roi.roi_id}: {e}")
+                    continue
+
+                for bind in enabled_binds:
+                    roi_items.append((camera_id, roi, bind, frame, cropped, scale_info))
             
             if not roi_items:
                 return
 
-            with self._batch_lock:
-                self._batch_roi_queue.extend(roi_items)
-
-            # 通知推理线程有新数据
-            self._batch_event.set()
-
+            self._enqueue_roi_items(roi_items)
             self._performance_stats["total_frames_processed"] += 1
             
         except Exception as e:
             self._logger.error(f"处理帧失败 {camera_id}: {e}")
+
+    def _scheduler_worker(self):
+        """中心调度线程：只消费各路最新帧，避免解码线程被预处理阻塞。"""
+        while not self._stop_event.is_set():
+            had_frame = False
+
+            for stream in self._stream_manager.get_all_streams():
+                frame = stream.get_latest_frame(timeout=0.0)
+                if frame is None:
+                    continue
+
+                frame_age = (datetime.now() - frame.timestamp).total_seconds()
+                if frame_age > self._max_frame_age_sec:
+                    self._performance_stats["dropped_frames"] += 1
+                    continue
+
+                had_frame = True
+                self._process_frame(stream.camera_id, frame)
+
+            self._performance_stats["scheduler_cycles"] += 1
+
+            if not had_frame:
+                self._stop_event.wait(self._scheduler_interval_sec)
     
     def _batch_process_rois(self):
         """批量处理 ROI - 真正的 batch 推理（按 max_batch_size 分块）"""
@@ -661,6 +700,11 @@ class EdgeInferenceService:
 
                 # 一次性推理整个 batch
                 outputs, inference_time_ms = engine.infer(batch_data)
+                self._performance_stats["inference_batches"] += 1
+                self._logger.log_inference_latency(
+                    inference_time_ms,
+                    batch_size=len(chunk),
+                )
 
                 # 诊断：输出原始推理结果形状（非告警诊断日志，使用 DEBUG 级别）
                 import numpy as np
@@ -938,6 +982,15 @@ class EdgeInferenceService:
         self._stop_event.clear()
         
         self._load_cameras()
+        self._stream_manager.start_all()
+
+        self._scheduler_thread = threading.Thread(
+            target=self._scheduler_worker,
+            name="FrameScheduler",
+            daemon=True
+        )
+        self._scheduler_thread.start()
+        self._logger.info("帧调度线程已启动")
 
         # 启动独立推理线程（生产者-消费者模式）
         self._inference_thread = threading.Thread(
@@ -948,8 +1001,6 @@ class EdgeInferenceService:
         self._inference_thread.start()
         self._logger.info("推理线程已启动")
 
-        self._stream_manager.start_all()
-
         self._logger.info("Edge_Inference_Service 已启动")
 
         self._register_signal_handlers()
@@ -979,6 +1030,9 @@ class EdgeInferenceService:
         self._stop_event.set()
         self._batch_event.set()  # 唤醒推理线程以退出
 
+        if self._scheduler_thread and self._scheduler_thread.is_alive():
+            self._scheduler_thread.join(timeout=5)
+
         if self._inference_thread and self._inference_thread.is_alive():
             self._inference_thread.join(timeout=5)
 
@@ -1029,6 +1083,10 @@ class EdgeInferenceService:
             "uptime_seconds": self._performance_stats["uptime_seconds"],
             "total_frames_processed": self._performance_stats["total_frames_processed"],
             "total_alerts_generated": self._performance_stats["total_alerts_generated"],
+            "dropped_frames": self._performance_stats["dropped_frames"],
+            "dropped_roi_tasks": self._performance_stats["dropped_roi_tasks"],
+            "inference_batches": self._performance_stats["inference_batches"],
+            "scheduler_cycles": self._performance_stats["scheduler_cycles"],
             "stream_manager": (
                 self._stream_manager.get_statistics() 
                 if self._stream_manager else {}