Mirrors/blakeblackshear.frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-04-28 23:06:13 +02:00
Code Issues Projects Releases Wiki Activity

[MemryX] Clean shutdown of detector process (#21035)

Browse Source 
* update code for clean exit

* ruff format

* remove unused time import

* update stop_event handling

* remove hasattr check
This commit is contained in:
Abinila Siva
2025-11-25 12:25:07 -05:00
committed by GitHub
parent 8520ade5c4
commit fe47620153
2 changed files with 78 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
frigate/object_detection/base.py
View File

@@ -43,6 +43,7 @@ class BaseLocalDetector(ObjectDetector):
self,
detector_config: BaseDetectorConfig = None,
labels: str = None,
stop_event: MpEvent = None,
):
self.fps = EventsPerSecond()
if labels is None:
@@ -60,6 +61,10 @@ class BaseLocalDetector(ObjectDetector):
self.detect_api = create_detector(detector_config)
# If the detector supports stop_event, pass it
if hasattr(self.detect_api, "set_stop_event") and stop_event:
self.detect_api.set_stop_event(stop_event)
def _transform_input(self, tensor_input: np.ndarray) -> np.ndarray:
if self.input_transform:
tensor_input = np.transpose(tensor_input, self.input_transform)
@@ -240,6 +245,10 @@ class AsyncDetectorRunner(FrigateProcess):
while not self.stop_event.is_set():
connection_id, detections = self._detector.async_receive_output()
# Handle timeout case (queue.Empty) - just continue
if connection_id is None:
continue
if not self.send_times:
# guard; shouldn't happen if send/recv are balanced
continue
@@ -266,21 +275,38 @@ class AsyncDetectorRunner(FrigateProcess):
self._frame_manager = SharedMemoryFrameManager()
self._publisher = ObjectDetectorPublisher()
self._detector = AsyncLocalObjectDetector(detector_config=self.detector_config)
self._detector = AsyncLocalObjectDetector(
detector_config=self.detector_config, stop_event=self.stop_event
)
for name in self.cameras:
self.create_output_shm(name)
t_detect = threading.Thread(target=self._detect_worker, daemon=True)
t_result = threading.Thread(target=self._result_worker, daemon=True)
t_detect = threading.Thread(target=self._detect_worker, daemon=False)
t_result = threading.Thread(target=self._result_worker, daemon=False)
t_detect.start()
t_result.start()
while not self.stop_event.is_set():
time.sleep(0.5)
try:
while not self.stop_event.is_set():
time.sleep(0.5)
self._publisher.stop()
logger.info("Exited async detection process...")
logger.info(
"Stop event detected, waiting for detector threads to finish..."
)
# Wait for threads to finish processing
t_detect.join(timeout=5)
t_result.join(timeout=5)
# Shutdown the AsyncDetector
self._detector.detect_api.shutdown()
self._publisher.stop()
except Exception as e:
logger.error(f"Error during async detector shutdown: {e}")
finally:
logger.info("Exited Async detection process...")
class ObjectDetectProcess:
@@ -308,7 +334,7 @@ class ObjectDetectProcess:
# if the process has already exited on its own, just return
if self.detect_process and self.detect_process.exitcode:
return
self.detect_process.terminate()
logging.info("Waiting for detection process to exit gracefully...")
self.detect_process.join(timeout=30)
if self.detect_process.exitcode is None: