Threading

[andrey-churkin]

import time
from pathlib import Path
from functools import partial
import numpy as np
import openvino.runtime as ov
from openvino.tools.pot.api.samples.object_detection.data_loader import COCOLoader
from openvino.tools.pot.api.samples.object_detection.metric import MAP
import nncf
import threading


FILE = Path(__file__).resolve()
ROOT = FILE.parent.relative_to(Path.cwd())
MODEL_DIR = ROOT / 'ssd_mobilenet_v1_fpn_quantization'
DATASET_DIR = ROOT / 'coco2017' / 'val2017'
ANNOTATION_FILE = ROOT / 'coco2017' / 'instances_val2017.json'


ie = ov.Core()


def main():
    ir_model_xml = ROOT / 'public' / 'ssd_mobilenet_v1_fpn_coco' / 'FP32' / 'ssd_mobilenet_v1_fpn_coco.xml'
    ov_model = ie.read_model(model=ir_model_xml)


    dataset_config = {
        'images_path': f'{DATASET_DIR}/',
        'annotation_path': ANNOTATION_FILE,
    }
    data_source = COCOLoader(dataset_config)
    def transform_fn(data):
        images, _ = data
        return np.expand_dims(images, 0)


    # val_func = partial(validation_fn, labels=data_source.labels)
    val_func = partial(async_validation_fn, labels=data_source.labels)

    dataset = nncf.Dataset(data_source, transform_fn)

    compiled_model = ie.compile_model(ov_model, device_name='CPU')

    start_time = time.time()

    ans_1 = {}
    ans_2 = {}

    if True:
        t1 = threading.Thread(target=task_1, args=(compiled_model, dataset, val_func, ans_1))
        t2 = threading.Thread(target=task_2, args=(compiled_model, dataset, val_func, ans_2))

        t1.start()
        t2.start()

        t1.join()
        t2.join()
    else:
        # Time: 00:12:51
        task_1(compiled_model, dataset, val_func, ans_1)
        t1_time = time.time() - start_time
        print('Time:', time.strftime("%H:%M:%S", time.gmtime(t1_time)))
        task_2(compiled_model, dataset, val_func, ans_2)

    finish_time = time.time()
    total = finish_time - start_time
    print('Time:', time.strftime("%H:%M:%S", time.gmtime(total)))
    print(ans_1)

# Time: 00:05:55
def task_1(compiled_model, dataset, val_func, ans):
    print('Task 1: Started')
    ans['value'] = val_func(compiled_model, dataset.get_data())
    print('Task 1: Ended')

# Time: 00:06:56
def task_2(compiled_model, dataset, val_func, ans):
    print('Task 2: Started')
    for i in range(5000):
        ans[i] = val_func(compiled_model, dataset.get_data([i]))
    print('Task 2: Ended')


# Time: 00:11:49
# 0.3555770739989489
def validation_fn(compiled_model: ov.CompiledModel, dataset, labels) -> float:
    output_layer = compiled_model.output(0)
    metric = MAP(91, labels)

    for images, labels in dataset:
        input_data = np.expand_dims(images, 0)
        output = compiled_model([input_data])[output_layer]
        metric.update(output, [labels])

    return metric.avg_value['MAP'].item()


# Time: 00:11:28
# 0.3555770739989489
def async_validation_fn(compiled_model: ov.CompiledModel, dataset, labels) -> float:
    output_layer = compiled_model.output(0)
    metric = MAP(91, labels)

    def callback(request, labels):
        output = request.results
        output = output[output_layer]
        metric.update(output, [labels])

    requests_num = compiled_model.get_property('OPTIMAL_NUMBER_OF_INFER_REQUESTS')
    infer_queue = ov.AsyncInferQueue(compiled_model, requests_num)
    infer_queue.set_callback(callback)

    for images, labels in dataset:
        input_data = np.expand_dims(images, 0)
        infer_queue.start_async([input_data], labels)
    infer_queue.wait_all()

    return metric.avg_value['MAP'].item()


if __name__ == '__main__':
    main()

[andrey-churkin]

baseline: 00:05:59

Num Threads	Time
1	00:11:33
2	00:11:20
4	00:14:34
6	00:15:42

import time
from pathlib import Path
import multiprocessing as mp

import numpy as np
import openvino.runtime as ov

from openvino.tools.pot.api.samples.object_detection.data_loader import COCOLoader
from openvino.tools.pot.api.samples.object_detection.metric import MAP


ROOT = Path(__file__).parent.resolve()
DATASET_DIR = ROOT / 'coco2017' / 'val2017'
ANNOTATION_FILE = ROOT / 'coco2017' / 'instances_val2017.json'


def load_model() -> ov.Model:
    ir_model_xml = ROOT / 'public' / 'ssd_mobilenet_v1_fpn_coco' / 'FP32' / 'ssd_mobilenet_v1_fpn_coco.xml'
    model = ov.Core().read_model(model=ir_model_xml)
    return model


def main():
    model = load_model()

    dataset_config = {
        'images_path': f'{DATASET_DIR}/',
        'annotation_path': ANNOTATION_FILE,
    }
    data_source = COCOLoader(dataset_config)

    if False:
        start_time = time.time()
        metric = task(model, data_source, 1)
        finish_time = time.time()
        total = finish_time - start_time
        print('Time:', time.strftime("%H:%M:%S", time.gmtime(total)))
        print(f'Metric: {metric}')
        exit(0)

    p1 = mp.Process(target=task, args=(model, data_source, 4))
    p2 = mp.Process(target=task, args=(model, data_source, 4))

    start_time = time.time()
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    finish_time = time.time()
    total = finish_time - start_time
    print('Time:', time.strftime("%H:%M:%S", time.gmtime(total)))


def task(model: ov.Model, data_source, num_threads: int):
    config = {}
    if num_threads > 1:
        config['AFFINITY'] = 'NUMA'
        config['INFERENCE_NUM_THREADS'] = mp.cpu_count() / num_threads

    compiled_model = ov.Core().compile_model(model, device_name='CPU', config=config)

    output_layer = compiled_model.output(0)
    metric = MAP(91, data_source.labels)

    for images, labels in data_source:
        inputs = np.expand_dims(images, 0)
        output = compiled_model([inputs])[output_layer]
        metric.update(output, [labels])

    return metric.avg_value['MAP'].item()


if __name__ == '__main__':
    main()