비동기적으로 추론 실행
| 필드 | 값 |
|---|---|
| 범주 | 모델 및 추론 |
| 난이도 | 초급 |
| 예상 소요 시간 | 10-15 minutes |
| 레이블 | async, push-pull, throughput, runtime |
제1장에서는 단일 동기 호출을 사용하여 모델을 실행했습니다. 즉, 하나의 프레임을 전달하고 결과가 반환될 때까지 기다립니다. 이는 간단하지만 컴퓨팅 자원을 낭비합니다. 입력 데이터를 생성하는 스레드와 출력 데이터를 소비하는 스레드가 동일하므로 두 작업이 동시에 수행될 수 없습니다. 이 장에서는 동일한 ResNet-50 모델을 사용하되, 두 작업을 분리하여 처리량을 높이는 파이프라인으로 만듭니다.
메커니즘은 비동기 Run입니다. 모델을 Async 모드에서 Graph로 build()한 다음, 두 개의 독립적인 호출(생성자에서 push(...), 소비자에서 pull(...))을 사용하여 실행합니다. 결과적으로 생성자 스레드는 런타임이 허용하는 한 최대한 빠르게 프레임을 제공하고, 메인 스레드는 예측 결과를 가져오며, 마지막으로 pushed=N pulled=N 라인을 통해 데이터 손실이 없었음을 확인합니다.
둘러보기
모델 로드
제1장과 동일하게 아카이브에서 Model을 생성하는 것으로 시작하지만, 여기서는 include_input 및 include_output이 설정된 RouteOptions도 선언합니다. 이러한 플래그는 모델이 그래프에 포함될 때 자체 입력 및 출력 경계를 노출하도록 지시하므로, 주변 파이프라인이 프레임을 입력하고 텐서를 출력할 수 있습니다.
simaai::neat::Model model(model_path, build_options(size));
simaai::neat::Model::RouteOptions route_opt;
route_opt.include_input = true;
route_opt.include_output = true;
비동기 파이프라인 구축
Model은 직접 push/pull 방식으로 실행할 수 없습니다. Run을 사용해야 합니다. 모델을 graph.add(model.graph(route_opt))를 통해 새 Graph로 래핑한 다음, 대표 프레임을 사용하여 build(...)합니다. 샘플 프레임을 전달하면 build()가 구체적인 텐서 모양을 미리 협상할 수 있습니다. 반환된 Run은 두 스레드가 공유할 핸들입니다.
simaai::neat::Graph graph;
graph.add(model.graph(route_opt));
auto run = graph.build(std::vector<cv::Mat>{frames.front()});
생성자에서 프레임 전송
생성자의 유일한 작업은 입력을 제공하는 것입니다. 준비된 프레임을 반복하고 각 프레임에 대해 push(...)를 호출한 다음, 더 이상 프레임이 없음을 알리기 위해 close_input()를 호출하는 스레드를 생성합니다. 이 신호는 소비자가 언제 중지해야 하는지 알 수 있도록 합니다. 생성자가 독립적으로 실행되므로 다음 프레임을 보내기 전에 결과를 기다리지 않습니다.
std::thread가 루프를 실행합니다. 원자 pushed 카운터와 producer_done 플래그가 업데이트되므로 메인 스레드는 잠금 없이 진행 상황을 관찰할 수 있습니다.
std::atomic<int> pushed{0};
std::atomic<bool> producer_done{false};
std::thread producer([&]() {
for (const cv::Mat& f : frames) {
run.push(std::vector<cv::Mat>{f});
pushed.fetch_add(1, std::memory_order_relaxed);
}
run.close_input();
producer_done.store(true);
});
소비자에서 결과 가져오기
메인 스레드는 결과를 소비합니 다. pull(timeout_ms=2000)를 호출하는 루프를 실행합니다. 이 함수는 사용 가능한 다음 출력을 반환하거나, 지정된 시간 내에 결과가 없으면 아무것도 반환하지 않습니다. 빈 결과가 반환되면 생성자가 완료되었는지 확인합니다. 완료되었으면 루프를 중지하고, 그렇지 않으면 계속 기다립니다. 각 실제 결과는 상위 1개 클래스 인덱스로 축소되고 출력됩니다. 루프가 끝나면 생성자 스레드를 종료하고 pushed == pulled가 올바르게 수행되었는지 확인합니다.
pull()은 optional<Sample>을 반환합니다. 바이트를 읽기 전에 tensors_from_sample(...)을 사용하여 텐서를 추출합니다.
int pulled = 0;
while (pulled < n) {
auto out = run.pull(/*timeout_ms=*/2000);
if (!out.has_value()) {
if (producer_done.load())
break;
continue;
}
std::cout << "top1=" << top1_from_output(*out) << "\n";
++pulled;
}
producer.join();
실행
실행하면 각 프레임마다 하나의 top1= 줄이 표시되고, 그 뒤에 푸시/풀 집계 결과가 나타납니다. Neat 설치 디렉터리(share/ 및 lib/가 포함된 디렉터리)에서 Python 및 C++(미리 빌드된 버전) 명령을 실행합니다. 그런 다음 저장소 루트에서 소스 코드를 빌드하는 명령을 실행합니다.
C++ (prebuilt):
./lib/sima-neat/tutorials/tutorial_002_run_inference_async \
--model /tmp/resnet_50.tar.gz --n 4
C++ (build from source):
./build.sh --target tutorial_002_run_inference_async
./build/tutorials-standalone/tutorial_002_run_inference_async \
--model /tmp/resnet_50.tar.gz --n 4
예상 출력(정확한 인덱스는 이미지에 따라 다름. C++ 빌드는 pushed=... 필드를 추가하고, Python 빌드는 pulled=...만 출력함):
top1=285
top1=285
top1=285
top1=285
pushed=4 pulled=4
[OK] 002_run_inference_async
이 장의 C++ 소스 코드를 사용자 지정 CMakeLists.txt 파일을 사용하여 자신의 프로젝트에 통합하려면 (별도의 추가 폴더는 필요하지 않음), 랜딩 페이지의 튜토리얼 실행 방법을 참조하십시오.
실전 활용
이 장에서는 비동기 푸시/풀 표면을 사용합니다. 동일한 모델을 결 정론적 합성 입력으로 측정하려면 모델 성능 평가를 계속 진행하십시오. 전체 빌드 대 실행 및 동기 대 비동기 모델과 함께 완전한 RunOptions 표면에 대한 내용은 첫 번째 그래프 빌드를 참조하십시오. 큐 깊이, 오버플로 정책 및 부하 상태에서의 측정에 대한 내용은 처리량 및 큐 깊이 조정을 참조하십시오.
전체 소스
전체 소스 프로그램 표시
// Async push/pull: producer thread pushes frames, main thread pulls outputs.
//
// Usage:
// tutorial_002_run_inference_async --model /path/to/resnet_50.tar.gz [--image /path/to.jpg] [--n
// 4]
#include "neat.h"
#include <opencv2/core.hpp>
#include <opencv2/imgcodecs.hpp>
#include <opencv2/imgproc.hpp>
#include <atomic>
#include <cstring>
#include <exception>
#include <filesystem>
#include <iostream>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>
namespace fs = std::filesystem;
namespace {
bool get_arg(int argc, char** argv, const std::string& key, std::string& out) {
for (int i = 1; i + 1 < argc; ++i) {
if (key == argv[i]) {
out = argv[i + 1];
return true;
}
}
return false;
}
int parse_int_arg(int argc, char** argv, const std::string& key, int def) {
std::string value;
if (!get_arg(argc, argv, key, value))
return def;
return std::stoi(value);
}
cv::Mat load_rgb(const fs::path& image_path, int size) {
cv::Mat bgr = cv::imread(image_path.string(), cv::IMREAD_COLOR);
if (bgr.empty())
throw std::runtime_error("failed to read image: " + image_path.string());
if (bgr.cols != size || bgr.rows != size) {
cv::resize(bgr, bgr, cv::Size(size, size), 0, 0, cv::INTER_AREA);
}
cv::Mat rgb;
cv::cvtColor(bgr, rgb, cv::COLOR_BGR2RGB);
if (!rgb.isContinuous())
rgb = rgb.clone();
return rgb;
}
simaai::neat::Model::Options build_options(int size) {
simaai::neat::Model::Options opt;
opt.preprocess.color_convert.input_format = simaai::neat::PreprocessColorFormat::RGB;
opt.preprocess.input_max_width = size;
opt.preprocess.input_max_height = size;
opt.preprocess.input_max_depth = 3;
opt.preprocess.normalize.mean = {0.485f, 0.456f, 0.406f};
opt.preprocess.normalize.stddev = {0.229f, 0.224f, 0.225f};
return opt;
}
int top1_from_output(const simaai::neat::Sample& out) {
if (simaai::neat::tensors_from_sample(out, true).empty())
throw std::runtime_error("no tensor output");
const simaai::neat::Mapping m = simaai::neat::tensors_from_sample(out, true).front().map_read();
const size_t n = m.size_bytes / sizeof(float);
const float* p = reinterpret_cast<const float*>(m.data);
int best = 0;
for (size_t i = 1; i < n && i < 1000; ++i) {
if (p[i] > p[best])
best = static_cast<int>(i);
}
return best;
}
} // namespace
int main(int argc, char** argv) {
try {
std::string model_path, image;
if (!get_arg(argc, argv, "--model", model_path)) {
std::cerr
<< "Usage: tutorial_002_run_inference_async --model <path> [--image <path>] [--n <n>]\n";
return 1;
}
get_arg(argc, argv, "--image", image);
const int n = parse_int_arg(argc, argv, "--n", 4);
const int size = 224;
cv::Mat frame = image.empty() ? cv::Mat(size, size, CV_8UC3, cv::Scalar(99, 99, 99))
: load_rgb(image, size);
std::vector<cv::Mat> frames(n, frame);
// CORE LOGIC
// Build a Graph around the model and run it async: one producer thread pushes,
// the main thread pulls outputs.
simaai::neat::Model model(model_path, build_options(size));
simaai::neat::Model::RouteOptions route_opt;
route_opt.include_input = true;
route_opt.include_output = true;
simaai::neat::Graph graph;
graph.add(model.graph(route_opt));
auto run = graph.build(std::vector<cv::Mat>{frames.front()});
std::atomic<int> pushed{0};
std::atomic<bool> producer_done{false};
std::thread producer([&]() {
for (const cv::Mat& f : frames) {
run.push(std::vector<cv::Mat>{f});
pushed.fetch_add(1, std::memory_order_relaxed);
}
run.close_input();
producer_done.store(true);
});
int pulled = 0;
while (pulled < n) {
auto out = run.pull(/*timeout_ms=*/2000);
if (!out.has_value()) {
if (producer_done.load())
break;
continue;
}
std::cout << "top1=" << top1_from_output(*out) << "\n";
++pulled;
}
producer.join();
std::cout << "pushed=" << pushed.load() << " pulled=" << pulled << "\n";
if (pulled != n)
throw std::runtime_error("pulled=" + std::to_string(pulled) +
" != pushed=" + std::to_string(pushed.load()));
std::cout << "[OK] 002_run_inference_async\n";
return 0;
} catch (const std::exception& e) {
std::cerr << "[FAIL] " << e.what() << "\n";
return 1;
}
}