Skip to content

Concurrent Processing

Achieving full utilization when running inference on an NPU is a complex task. Each inference operation inherently involves some waiting time. To understand where this waiting time comes from, look at how the NPU produces an inference result:

  1. Loading (and preprocessing) inputs - host (CPU)
  2. Feeding into the NPU - host-NPU IO (DMA)
  3. Running workloads - NPU
  4. Getting outputs - NPU-host IO (DMA)

The host CPU is actively involved in step 1, while the NPU takes over in step 3. Additional DMA transfers may occur during step 3 if processing on the host is beneficial.

In a multiple inference scenario, where each inference is data-independent, RBLN SDK can optimize the utilization of the aforementioned steps by implementing the concept of concurrency. This involves processing multiple samples simultaneously, thereby filling idle time gaps with other inferences and ensuring maximal utilization of the device.

RBLN SDK offers a straightforward and user-friendly set of asynchronous APIs for achieving concurrency. This document provides an overview of the asynchronous APIs with a PyTorch example demonstrating how to effectively utilize them.

Prerequisite

Before getting started, install the following pip packages:

Usage

Once a deep learning model has been compiled, you obtain an RBLNCompiledModel object. To load it into an asynchronous runtime, call create_async_runtime():

1
2
3
4
import rebel

compiled_model: rebel.RBLNCompiledModel = rebel.compile_from_torch(model, input_info)
runtime: rebel.AsyncRuntime = compiled_model.create_async_runtime()

If the compiled model is already saved as a *.rbln file, you can create the asynchronous runtime directly from it:

1
2
3
import rebel

runtime = rebel.AsyncRuntime("/path/to/saved_file.rbln")

If your program utilizes native asyncio as the event loop, using AsyncRuntime.async_run is an excellent choice for Pythonic programming, as it adheres to the PEP-492 async and await syntax:

1
output = await runtime.async_run(input_)

If the caller program runs on a different type of event loop, such as PyQt or gevent, the invocations and joins need to be managed manually in accordance with the logic. To facilitate this, RBLN SDK also offers AsyncRuntime.run, which is a straightforward asynchronous version of the run function.

1
2
3
4
5
task = runtime.run(input_)

# Do other jobs...

output = task.wait()

Performance Optimization Considerations

When implementing concurrent processing, RBLN SDK provides flexibility in managing input preparation through the parallel parameter in AsyncRuntime. Single-threaded input preparation (parallel=1) is the default behavior. To improve performance in specific scenarios, enable double buffering with two threads (parallel=2).

Double buffering allows one thread to prepare the next input while the NPU processes the current one, which can be beneficial when:

  • Input preprocessing is computationally intensive
  • The model's inference time is comparable to input preparation time
  • DMA transfer times are significant due to:
    • Large input/output tensors that require substantial data transfer between host and NPU
    • Models with multiple inputs/outputs requiring multiple DMA transfers

However, double buffering may not always provide performance benefits and could potentially impact stability with certain models, so benchmark your specific use case before enabling it.

Examples

PyTorch - YOLOv8

YOLOv8 is an object detection model. This example uses YOLOv8 with the COCO dataset. To access the dataset, refer to the COCO dataset download page.

Assume that the images and the annotation file for the 2017/validation set are stored in the current directory:

1
2
images: ./val2017
annotation file: ./annotations/instances_val2017.json

The following is the complete code to run YOLOv8 using the asynchronous runtime APIs:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
from pathlib import Path

import numpy as np
import torch
import rebel
from PIL import Image
from pycocotools.coco import COCO
from ultralytics import YOLO
from ultralytics.data.augment import LetterBox


class COCO2017Dataset:
    def __init__(self, dataset_root: Path):
        image_root = dataset_root / "val2017"
        self.coco_gt = COCO(dataset_root / "annotations" / "instances_val2017.json")

        # Create sample
        self.samples = []
        for image_id in self.coco_gt.getImgIds():
            self.samples.append(
                (
                    image_id,
                    image_root / self.coco_gt.imgs[image_id]["file_name"],
                )
            )

    def __getitem__(self, index):
        image_id, image_path = self.samples[index]
        image = Image.open(image_path)
        letterbox_transform = LetterBox()
        array = np.asarray(image, dtype=np.float32)
        array = letterbox_transform(image=array)
        array = np.transpose(array, [2, 0, 1])
        array = np.ascontiguousarray(array)
        array /= 255
        return array


dataset = COCO2017Dataset(Path("."))
samples = [dataset[i] for i in range(100)]


class YOLOv8Wrapper(torch.nn.Module):
    def __init__(self, model):
        super().__init__()
        self.model = model

    def forward(self, x):
        # In eval mode the detection head returns (predictions, feature maps); keep only the predictions.
        return self.model(x)[0]


model = YOLOv8Wrapper(YOLO("yolov8n.pt").model.eval())
compiled_model = rebel.compile_from_torch(
    model, input_info=[("input", [1, 3, 640, 640], torch.float32)]
)

import asyncio

runtime = rebel.AsyncRuntime(compiled_model)


async def main() -> None:
    tasks = []

    for sample in samples:
        sample = np.expand_dims(sample, axis=0)
        task = runtime.async_run(sample)
        tasks.append(task)

    logits = await asyncio.gather(*tasks)
    return logits


logits = asyncio.run(main())

References