非同期で推論を実行

項目	値
カテゴリ	モデルと推論
難易度	初級
推定所要時間	10-15 minutes
ラベル	async, push-pull, throughput, runtime

第001章では、単一の同期呼び出しでモデルを実行しました。つまり、1つのフレームを渡し、結果が返ってくるまで処理をブロックします。これは単純ですが、計算リソースが無駄になります。入力データを生成するスレッドと、出力データを処理するスレッドは同じスレッドであるため、並行して処理することはできません。この章では、同じResNet-50モデルを使用し、それらの2つのジョブを分割することで、スループットを重視したパイプラインに変換します。

そのメカニズムは、非同期のRunです。モデルをAsyncモードでGraphにbuild()し、次に、プロデューサーからの2つの独立した呼び出し（push(...)）とコンシューマーからの呼び出し（pull(...)）で駆動します。最終的には、プロデューサースレッドがランタイムが受け入れる速度でフレームを供給し、メインスレッドが予測結果を抽出するようになり、最後にpushed=N pulled=N行が表示され、データが失われていないことが確認されます。

ウォークスルー

モデルのロード

第001章と同様に、アーカイブからModelを構築することから始めますが、ここではinclude_inputとinclude_outputを設定したRouteOptionsも宣言します。これらのフラグは、モデルがグラフに組み込まれたときに、独自の入力と出力の境界を公開するように指示します。これにより、周囲のパイプラインはフレームをプッシュインし、テンソルをプルアウトできます。

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.add(model.graph(route_opt))を介して新しいGraphでラップし、次に代表的なフレームを使用して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()を呼び出して、これ以上フレームが来ないことを通知します。このシグナルは、コンシューマーがいつ停止するかを判断するために使用されます。プロデューサーは独立して実行されるため、次のフレームを送信する前に結果を待つ必要はありません。

std::threadがループを実行します。アトミックなpushedカウンターとproducer_doneフラグが更新され、メインスレッドはロックなしで進捗状況を監視できます。

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) を呼び出すループがあり、これは次の利用可能な出力を返します。タイムアウト内にデータが到着しない場合は何も返しません。データが空の場合、プロデューサーが処理を終了したかどうかを確認します。終了している場合は停止し、そうでない場合は待機を続けます。各実際の結果は、上位1つのクラスのインデックスに集約され、出力されます。ループの後に、プロデューサーと結合し、pushed == pulled が確認されます。

pull() は optional<Sample> を返します。バイトを読み取る前に、tensors_from_sample(...) を使用してテンソルを抽出します。

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();

実行

実行すると、各フレームに対して1つの 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