inference will be slow if use threading.thread or threadpool

[YeWenxuan64]

i want to use another thread to aviod the block the gui of the main thread. but found the qnncontext's inference in another thread is slower than it in main thread.

[quic-zhanweiw]

Which model are you using? Which platform are you running? Can you provide code snippets?

[YeWenxuan64]

here is a sample of inference in thread, use the same yolo8 model in ai-engine-direct-helper/samples/linux/python/yolov8_det.
fill the path of model in mian func, switch the thread mode(threading_test.init(use_thread=True) in main func) or none thread mode to compare the differences of fps

this sample's fps differences could be 20fps (thread mode:65fps, none thread mode:85fps) .in my other models the differences could be 2 times.(my other models are yolo11-det, yolo11-pose, retinaface-mobile, nanotrack-v3, salience-msi_net)

my platform is Radxa dragon Q6A(QCS6490). OS is radxa os "test5" of the dev-board(https://github.com/radxa-build/radxa-dragon-q6a/releases)

qairt version 2.37.1.250807 or v2.38.0.250901

import os
import time
from pathlib import Path
from queue import Queue
from threading import Thread
import numpy as np
import cv2

from qai_appbuilder import QNNContext, Runtime, LogLevel, ProfilingLevel, PerfProfile, QNNConfig




class qnn_threading_test:
    def __init__(self, model_path):
        self.model_path = model_path

        self.inputs = None
        self.outputs = None

        self.thread_qnn = None
        self.use_thread = True


    def init(self, use_thread=True):
        self.use_thread = use_thread

        model_file = Path(self.model_path)
        model_name = model_file.stem

        QNNConfig.Config('None', Runtime.HTP, LogLevel.ERROR, ProfilingLevel.BASIC)

        self.qnn_context = QNNContext(model_name=model_name, model_path=self.model_path)

        self.queue_in = Queue()
        self.queue_out = Queue()

        if self.use_thread:
            self.thread_qnn = Thread(target=self.qnn_thread, daemon=True)
            self.thread_qnn.start()

        PerfProfile.SetPerfProfileGlobal(PerfProfile.BURST)

    def qnn_thread(self):

        while True:
            input_list = self.queue_in.get() # use queue as signal instead of data to save performance

            if input_list is None:
                break

            outputs:list = self.qnn_context.Inference(self.inputs)

            self.outputs = outputs
            self.queue_out.put(True) # use queue as signal instead of data to save performance

    def put(self, input_data:list[np.ndarray]):
        #input_data = [np.transpose(tensor, (0, 3, 1, 2)).copy(order='C') for tensor in input_data]

        if self.use_thread:
            self.inputs = input_data
            self.queue_in.put(True)
        else:
            self.outputs = self.qnn_context.Inference(input_data)
            self.queue_out.put(True)

    def get(self, block:bool=False) -> list[np.ndarray]|None:
        result_list = None

        if block is True:
            self.queue_out.get()

        result_list = self.outputs
        self.outputs = None

        return result_list
    
    def release(self):
        self.outputs = None

        if self.thread_qnn is not None:
            self.queue_in.put(None)
            self.thread_qnn.join()

            del self.qnn_context

            print(f'qnn_context:  released')

            PerfProfile.RelPerfProfileGlobal()


class_map = {
    0: "person",
    1: "bicycle",
    2: "car",
    3: "motorcycle",
    4: "airplane",
    5: "bus",
    6: "train",
    7: "truck",
    8: "boat",
    9: "traffic light",
    10: "fire hydrant",
    11: "stop sign",
    12: "parking meter",
    13: "bench",
    14: "bird",
    15: "cat",
    16: "dog",
    17: "horse",
    18: "sheep",
    19: "cow",
    20: "elephant",
    21: "bear",
    22: "zebra",
    23: "giraffe",
    24: "backpack",
    25: "umbrella",
    26: "handbag",
    27: "tie",
    28: "suitcase",
    29: "frisbee",
    30: "skis",
    31: "snowboard",
    32: "sports ball",
    33: "kite",
    34: "baseball bat",
    35: "baseball glove",
    36: "skateboard",
    37: "surfboard",
    38: "tennis racket",
    39: "bottle",
    40: "wine glass",
    41: "cup",
    42: "fork",
    43: "knife",
    44: "spoon",
    45: "bowl",
    46: "banana",
    47: "apple",
    48: "sandwich",
    49: "orange",
    50: "broccoli",
    51: "carrot",
    52: "hot dog",
    53: "pizza",
    54: "donut",
    55: "cake",
    56: "chair",
    57: "couch",
    58: "potted plant",
    59: "bed",
    60: "dining table",
    61: "toilet",
    62: "tv",
    63: "laptop",
    64: "mouse",
    65: "remote",
    66: "keyboard",
    67: "cell phone",
    68: "microwave",
    69: "oven",
    70: "toaster",
    71: "sink",
    72: "refrigerator",
    73: "book",
    74: "clock",
    75: "vase",
    76: "scissors",
    77: "teddy bear",
    78: "hair drier",
    79: "toothbrush"
}

def preprocess_frame(frame: np.ndarray) -> np.ndarray:
    """
    Preprocess a frame for inference.
    Args:
        frame: Input frame in BGR format (H, W, C)
    Returns:
        Preprocessed frame as numpy array (1, H, W, C) in RGB format
    """
    image_size = 640

    # Convert BGR to RGB
    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

    # Get original dimensions
    original_height, original_width = frame_rgb.shape[:2]

    # Calculate scaling factor to maintain aspect ratio
    scale = min(image_size / original_width, image_size / original_height)

    # Calculate new dimensions
    new_width = int(original_width * scale)
    new_height = int(original_height * scale)

    # Resize the image
    resized_frame = cv2.resize(frame_rgb, (new_width, new_height), interpolation=cv2.INTER_LINEAR)

    # Create a new image with the target size and fill with black (or gray)
    padded_frame = np.full((image_size, image_size, 3), 114, dtype=np.uint8)  # Gray padding (114, 114, 114)

    # Calculate the position to place the resized image
    x_offset = (image_size - new_width) // 2
    y_offset = (image_size - new_height) // 2

    # Place the resized image in the center
    padded_frame[y_offset:y_offset+new_height, x_offset:x_offset+new_width] = resized_frame

    # Normalize to [0, 1] and convert to float32
    normalized_frame = padded_frame.astype(np.float32) / 255.0

    # Add batch dimension
    batched_frame = np.expand_dims(normalized_frame, axis=0)

    return batched_frame, scale, x_offset, y_offset

def draw_box_from_xyxy(
    frame: np.ndarray,
    top_left: np.ndarray | tuple[int, int],
    bottom_right: np.ndarray | tuple[int, int],
    color: tuple[int, int, int] = (0, 0, 0),
    size: int = 3,
    text:str =None
):
    """
    Draw a box using the provided top left / bottom right points to compute the box.

    Parameters:
        frame: np.ndarray
            np array (H W C x uint8, BGR)

        box: np.ndarray | torch.Tensor
            array (4), where layout is
                [xc, yc, h, w]

        color: Tuple[int, int, int]
            Color of drawn points and connection lines (RGB)

        size: int
            Size of drawn points and connection lines BGR channel layout

        text: None | str
            Overlay text at the top of the box.

    Returns:
        None; modifies frame in place.
    """
    if not isinstance(top_left, tuple):
        top_left = (int(top_left[0].item()), int(top_left[1].item()))
    if not isinstance(bottom_right, tuple):
        bottom_right = (int(bottom_right[0].item()), int(bottom_right[1].item()))

    cv2.rectangle(frame, top_left, bottom_right, color, size)
    if text is not None:
        cv2.putText(frame, text, (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, size)


def after_inference(frame, model_output:list[np.ndarray], width, height, scale, x_offset, y_offset):
    # Convert outputs to tensors
    pred_boxes = model_output[2].reshape(-1, 4)  # Remove batch dimension
    pred_scores = model_output[0].reshape(-1)    # Remove batch dimension
    pred_class_idx = model_output[1].reshape(-1) # Remove batch dimension

    # Create batch indices (all zeros since we have single batch)
    batch_indices = np.zeros(len(pred_boxes), dtype=np.int32)

    # Convert boxes to [x1, y1, x2, y2] format if needed
    # pred_boxes = convert_to_xyxy(pred_boxes)  # Uncomment if needed

    # Non Maximum Suppression using NMSBoxesBatched
    indices = cv2.dnn.NMSBoxesBatched(
        pred_boxes.tolist(),
        pred_scores.tolist(),
        batch_indices.tolist(),
        score_threshold=0.35,
        nms_threshold=0.7
    )
    #print(pred_scores.max())

    # Convert indices to flat array if needed
    if len(indices) > 0:
        indices = indices.flatten()
    else:
        return frame  # No detections

    # Filter boxes, scores and class indices using NMS results
    pred_boxes = pred_boxes[indices]
    pred_scores = pred_scores[indices]
    pred_class_idx = pred_class_idx[indices]

    # Draw boxes on the original frame
    for box, score, class_idx in zip(pred_boxes, pred_scores, pred_class_idx):
        # Convert box coordinates from model output space to original frame space
        x1, y1, x2, y2 = box

        # Adjust for padding and scaling
        x1 = (x1 - x_offset) / scale
        y1 = (y1 - y_offset) / scale
        x2 = (x2 - x_offset) / scale
        y2 = (y2 - y_offset) / scale

        # Ensure coordinates are within frame bounds
        x1 = max(0, min(x1, width))
        y1 = max(0, min(y1, height))
        x2 = max(0, min(x2, width))
        y2 = max(0, min(y2, height))

        class_idx_item = class_idx.item()
        class_name = class_map.get(class_idx_item, "Unknown")

        draw_box_from_xyxy(
            frame,
            (int(x1), int(y1)),
            (int(x2), int(y2)),
            color=(0, 255, 0),
            size=2,
            text=f'{score.item():.2f} {class_name}'
        )

    return frame

def main():
    current_path = os.path.dirname(os.path.abspath(__file__))
    model_path = os.path.join(current_path, 'models/yolov8_det.bin')

    threading_test = qnn_threading_test(model_path)

    cap = cv2.VideoCapture('balcony.mp4')
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))\
    
    threading_test.init(use_thread=True) # args: use_thread means whether to use threading or not

    time_list = []
    last_time = time.time()

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        processed_frame, scale, x_offset, y_offset = preprocess_frame(frame)

        time1 = time.time()
        threading_test.put([processed_frame])

        model_output = threading_test.get(block=True)
        time2 = time.time()

        time_list.append(1 / (time2 - time1 + 1e-6))

        if model_output is not None:
            frame = after_inference(frame, model_output, width, height, scale, x_offset, y_offset)


        cv2.imshow('YOLOv8 Detection', frame)

        if time2 - last_time >= 1.0:
            last_time = time2
            time_list_array = np.array(time_list, dtype=np.float32)
            fps = np.mean(time_list_array)
            time_list = time_list[:len(time_list)//2]
            print(fps)

        # Press 'q' to exit
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break

    threading_test.release()
    cap.release()
    cv2.destroyAllWindows()


print('test')
if __name__ == '__main__':
    main()