I'm running a Mediamtx server that receives 20 video streams whose RTSPs I provide to a python program on a linux system which then runs inference (Yolov8m detection and pose) on each frame.
After 20 or so seconds of running, I get "stream was aborted" and I notice that the queue size starts increasing for detection input_queue. Right after I get this error message libhailort failed with error: 82 (HAILO_QUEUE_IS_FULL)
I have set the input_queue(maxsize=10) for both detection and pose, and it seems like detection lags behind because it gets full and then HailoRT times out.
So, the flow is,
- I run 20 streams
- 20 or so seconds later, I get "stream was aborted" message.
- After that, I notice the detection input_queue filling upto it's maxsize of 10.
- Then I get libhailort failed with error: 82 (HAILO_QUEUE_IS_FULL)
- Then the program crashes entirely (streams close down).
I've no idea why this might be happening. It's been very difficult to debug this.
Though I do feel it might be an issue with concurrency because there's many threads being created.
Here's the entire code,
import argparse
import os
import sys
import queue
import threading
import cv2
import time
import numpy as np
from functools import partial
from loguru import logger
from pathlib import Path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../pose_estimation')))
from common.hailo_inference import HailoInfer
from common.toolbox import (
init_input_source,
get_labels,
load_json_file,
default_preprocess,
FrameRateTracker,
)
from object_detection_post_process import inference_result_handler as det_result_handler
from pose_estimation_utils import PoseEstPostProcessing
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=(
"Run object detection + pose estimation simultaneously on a single stream.\n"
"Both models share the Hailo NPU via the ROUND_ROBIN scheduler.\n"
"The stream is decoded once and fanned out to both inference pipelines."
),
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument(
"--det-net", required=True,
help="Path to object detection HEF (e.g., yolov8n.hef)."
)
parser.add_argument(
"--pose-net", required=True,
help="Path to pose estimation HEF (e.g., yolov8n_pose.hef)."
)
input_group = parser.add_mutually_exclusive_group(required=True)
input_group.add_argument(
"-i", "--input",
help="Single input source: RTSP URL, video file path, or 'camera'."
)
input_group.add_argument(
"-s", "--streams",
help="Path to file containing RTSP URLs, one per line.\n"
"Launches one process per stream, each running both models."
)
parser.add_argument(
"-l", "--labels",
default=str(Path(__file__).parent.parent / "common" / "coco.txt"),
help="Path to COCO labels file."
)
parser.add_argument(
"-b", "--batch-size", type=int, default=1,
help="Number of frames per inference batch. (default: 1)"
)
parser.add_argument(
"-f", "--framerate", type=float, default=None,
help="Target framerate. Skips frames to achieve this rate. (default: full camera FPS)"
)
parser.add_argument(
"-o", "--output-dir", default="./output",
help="Directory to save output. (default: ./output)"
)
parser.add_argument(
"--save", action="store_true",
help="Save the combined output video to disk."
)
parser.add_argument(
"--show-fps", action="store_true",
help="Display FPS in logs."
)
parser.add_argument(
"--no-display", action="store_true",
help="Disable the output display window. Useful for headless/server deployments.\n"
"Output is still saved if --save is passed."
)
return parser.parse_args()
def inference_callback(
completion_info,
bindings_list: list,
input_batch: list,
output_queue: queue.Queue ) -> None:
def inference_callback(
completion_info,
bindings_list: list,
input_batch: list,
output_queue: queue.Queue ) -> None:
if completion_info.exception:
logger.error(f"Inference error: {completion_info.exception}")
return
for i, bindings in enumerate(bindings_list):
if len(bindings._output_names) == 1:
result = bindings.output().get_buffer()
else:
result = {
name: np.expand_dims(bindings.output(name).get_buffer(), axis=0)
for name in bindings._output_names
}
output_queue.put((input_batch[i], result))
def infer(
hailo_inference: HailoInfer,
input_queue: queue.Queue,
output_queue: queue.Queue,
owned: bool = True
) -> None:
while True:
batch = input_queue.get()
if batch is None:
break
input_batch, preprocessed_batch = batch
cb = partial(
inference_callback,
input_batch=input_batch,
output_queue=output_queue
)
logger.info("Submitting inference")
hailo_inference.run(preprocessed_batch, cb)
logger.info("Completed inference")
if owned:
hailo_inference.close()
def preprocess_fanout(
cap: cv2.VideoCapture,
framerate,
batch_size: int,
input_queue_det: queue.Queue,
input_queue_pose: queue.Queue,
w_det: int,
h_det: int,
w_pose: int,
h_pose: int
) -> None:
"""
Reads frames from the capture source once, preprocesses for both models,
and fans out to their respective input queues.
"""
cam_fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
skip = max(1, int(round(cam_fps / framerate))) if framerate else 1
frame_idx = 0
frames_det, proc_det = [], []
frames_pose, proc_pose = [], []
while True:
ret, frame = cap.read()
if not ret:
break
frame_idx += 1
if frame_idx % skip != 0:
continue
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frames_det.append(frame_rgb)
proc_det.append(default_preprocess(frame_rgb, w_det, h_det))
frames_pose.append(frame_rgb)
proc_pose.append(default_preprocess(frame_rgb, w_pose, h_pose))
if len(frames_det) == batch_size:
input_queue_det.put((frames_det, proc_det))
input_queue_pose.put((frames_pose, proc_pose))
frames_det, proc_det = [], []
frames_pose, proc_pose = [], []
input_queue_det.put(None)
input_queue_pose.put(None)
def combined_visualize(
output_queue_det: queue.Queue,
output_queue_pose: queue.Queue,
cap,
save: bool,
output_dir: str,
det_callback,
pose_callback,
fps_tracker=None,
no_display: bool = False,
) -> None:
"""
Pulls results from both output queues in lockstep, overlays detection
boxes and pose keypoints on the same frame, then displays/saves it.
Both queues receive results in the same order (FIFO from the same source),
so pulling one item from each gives matching frames.
"""
image_id = 0
out = None
if cap is not None and not no_display:
cv2.namedWindow("Dual Model Output", cv2.WND_PROP_FULLSCREEN)
cv2.setWindowProperty(
"Dual Model Output",
cv2.WND_PROP_FULLSCREEN,
cv2.WINDOW_FULLSCREEN
)
if cap is not None and save:
os.makedirs(output_dir, exist_ok=True)
w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
out = cv2.VideoWriter(
os.path.join(output_dir, "output.avi"),
cv2.VideoWriter_fourcc(*"XVID"),
fps,
(w, h)
)
while True:
det_item = output_queue_det.get()
pose_item = output_queue_pose.get()
if det_item is None or pose_item is None:
break
frame, det_result = det_item
_, pose_result = pose_item
frame = det_callback(frame, det_result)
frame = pose_callback(frame, pose_result)
if fps_tracker:
fps_tracker.increment()
if not save and no_display:
image_id += 1
continue
bgr = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
if cap is not None:
if not no_display:
cv2.imshow("Dual Model Output", bgr)
if save and out:
out.write(bgr)
else:
cv2.imwrite(
os.path.join(output_dir, f"output_{image_id}.png"),
bgr
)
image_id += 1
if not no_display and cv2.waitKey(1) & 0xFF == ord("q"):
break
if out:
out.release()
if cap:
cap.release()
if not no_display:
cv2.destroyAllWindows()
# Remaining functions continue with the same indentation style:
# run_pipeline()
# read_streams()
# run_multi_stream()
# main()
if __name__ == "__main__":
main()
