以非同步方式執行推論

欄位	值
類別	模型與推論
難度	初級
預估閱讀時間	10-15 minutes
標籤	async, push-pull, throughput, runtime

第一章執行了一個模型，使用單一同步呼叫：傳入一幀，並阻塞直到結果傳回。這很簡單，但會浪費運算資源——產生輸入的執行緒和處理輸出的執行緒是同一個執行緒，因此它們永遠無法同時執行。本章保留完全相同的 ResNet-50 模型，並將其轉換為以吞吐量為導向的管線，方法是將這兩個工作分開。

其機制是異步 Run：你 build() Graph 在 Async 模式，然後透過兩個獨立的呼叫來驅動它—— push(...) 來自一位製作人 pull(...) 來自消費者。最後，您將有一個生產者執行緒，以最快的速度將影格傳輸到執行階段，而執行階段會接受這些影格，同時主要執行緒會提取預測結果，以及一個最終的 pushed=N pulled=N 這條線證明了沒有任何東西遺失。

操作指南

載入模型

我們從與第 001 章完全相同的方式開始——建置一個 Model 從檔案中擷取——但我們在此也聲明一個 RouteOptions 搭配 include_input 以及 include_output 設定。這些旗標會指示模型在組成圖時，公開其自身的輸入和輸出邊界，以便周圍的管線可以將影格推送進來，並將張量提取出來。

tutorials/002_run_inference_async/run_inference_async.cpp

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 無法直接透過推/拉方式來驅動； Run 是。我們將模型包裝在一個全新的 Graph 透過 graph.add(model.graph(route_opt))，然後 build(...) 將其與一個代表性的框架搭配。傳遞樣本框架可讓 build() 事先協商好具體的張量形狀。傳回的 Run 是處理器將共用的控制項。

tutorials/002_run_inference_async/run_inference_async.cpp

simaai::neat::Graph graph;
graph.add(model.graph(route_opt));

auto run = graph.build(std::vector<cv::Mat>{frames.front()});

從生產者推送框架

製作者的唯一任務是提供輸入。我們啟動一個執行緒，該執行緒會循環處理準備好的影格，並呼叫。 push(...) 對於每一個，然後再呼叫 close_input() 以表示不再有更多影格即將傳輸——這個訊號讓接收者知道何時應該停止。由於產生器是獨立運作，因此在傳送下一個影格之前，它不會等待任何結果。

C++： std::thread 執行迴圈；一個原子 pushed 計數器和一個 producer_done 旗標會隨著進程更新，因此主要執行緒可以在沒有鎖的情況下觀察進度。

Python： threading.Thread 命名為 frame_producer 執行迴圈；之後，消費者會進行檢查。 thread.is_alive() 以檢測完成情況。

tutorials/002_run_inference_async/run_inference_async.cpp

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)，該函數會回傳下一個可用的輸出，或者如果在逾時期間沒有任何輸出，則回傳空值。在沒有輸出的情況下，我們會檢查生產者是否已完成——如果是，我們就會停止；否則，我們會繼續等待。每個實際結果都會被縮減為一個 top-1 類別索引，然後進行輸出。在循環結束後，我們會加入生產者，並確認 pushed == pulled。

C++： pull() 會回傳一個 optional<Sample>；在讀取位元組之前，使用 tensors_from_sample(...) 提取張量。

Python： pull() 會回傳一個 Sample 或 None；sample.tensor.to_numpy() 會將陣列傳遞給 argmax。

tutorials/002_run_inference_async/run_inference_async.cpp

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 介面，請參閱 建置您的第一個圖。對於佇列深度、溢位原則以及負載下的測量，請參閱 調整輸送量和佇列深度。

完整原始碼

顯示完整原始碼程式

tutorials/002_run_inference_async/run_inference_async.cpp

// 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;
  }
}

來源

• C++
• Python
• README