執行 PCIe 非同步推論

欄位	值
類別	PCIe 協同處理
難度	初級
預估閱讀時間	15 minutes
標籤	PCIe, asynchronous, throughput, detection

本教學會重新使用 YOLOv8s 影像加框解碼設定和 640x480 街景，這些來自教學 024。它提交一個重複的影像，因此儲存和影像解碼不會扭曲 PCIe 測量。

操作指南

設定一個檢測模型

一次載入影像，設定卡端 COCO 預處理和 YOLOv8 框解碼，然後在佇列 0 上建立一個 Model。缺少的檔案和卡啟動錯誤會在測量開始之前停止程式。

pcie_host/tutorials/025_run_pcie_inference_async/run_pcie_inference_async.cpp

const Args args = parse_args(argc, argv);
if (!std::filesystem::is_regular_file(kModelPath)) {
  throw std::runtime_error(std::string("model does not exist: ") + kModelPath);
}
const cv::Mat image = cv::imread(kImagePath, cv::IMREAD_COLOR);
if (image.empty()) {
  throw std::runtime_error(std::string("OpenCV could not decode: ") + kImagePath);
}
pcie::ConnectionOptions connection;
connection.card_id = args.card_id;
pcie::Model model(kModelPath, detection_options(), connection);
model.build(kBuildTimeoutMs);

預熱管線

執行幾個完整的檢測，但不計時。預熱可以消除模型啟動和第一個緩衝區效應，從而更準確地報告工作負載。

pcie_host/tutorials/025_run_pcie_inference_async/run_pcie_inference_async.cpp

for (int index = 0; index < kWarmupFrames; ++index) {
  (void)detection_count(model.run(image, kPullTimeoutMs));
}

同時提交和檢索

一個執行緒使用 push() 提交影像，而另一個執行緒使用具有有限超時的 pull() 檢索 BBOX 輸出。一個小型應用程式擁有的 FIFO 儲存每個有序提交的開始時間。任何拒絕、超時或格式不正確的結果都會關閉模型並喚醒另一個執行緒。

該範例僅依賴於正常的 Model 流程控制行為；應用程式中沒有佇列深度調整。

pcie_host/tutorials/025_run_pcie_inference_async/run_pcie_inference_async.cpp

const BenchmarkResult result = measure(model, image, kMeasuredFrames);

報告完成的工作

僅在兩個執行緒都完成並且所有已接受的影像都已檢索後，才停止計時。每秒幀數使用已完成輸出的數量。平均延遲從每次提交嘗試開始，直到其匹配的有序結果到達為止。

pcie_host/tutorials/025_run_pcie_inference_async/run_pcie_inference_async.cpp

std::cout << "completed=" << result.completed << '\n';
std::cout << std::fixed << std::setprecision(2) << "elapsed_seconds=" << result.elapsed_seconds
          << '\n'
          << "throughput_fps=" << result.completed / result.elapsed_seconds << '\n'
          << "average_latency_ms=" << result.average_latency_ms << '\n'
          << "total_detections=" << result.total_detections << '\n';

執行

安裝 PCIe 主機套件，並按照 教學設定 中所述下載教學套件。從解壓縮後的 PCIe extras 根目錄中，下載 YOLOv8s（如果尚未存在）：

sima-cli modelzoo get yolo_v8s

該程式需要此目錄中的確切路徑 yolo_v8s_mpk.tar.gz。如果 Model Zoo 使用了不同的名稱或位置，請將下載的檔案複製到指定位置：

cp /absolute/path/to/downloaded-yolov8s-archive.tar.gz yolo_v8s_mpk.tar.gz
test -f yolo_v8s_mpk.tar.gz

執行 Python：

source ~/pyneatpcie/bin/activate
python3 share/sima-pcie-host/tutorials/025_run_pcie_inference_async/run_pcie_inference_async.py

執行預建的 C++ 教學：

./lib/sima-pcie-host/tutorials/tutorial_025_run_pcie_inference_async

或者重新編譯它：

./build.sh --target tutorial_025_run_pcie_inference_async
./build/tutorials-standalone/tutorial_025_run_pcie_inference_async

確切的計時取決於主機和卡，但兩個程式都使用相同的測量邊界並輸出：

completed=1000
elapsed_seconds=...
throughput_fps=...
average_latency_ms=...
total_detections=...
[OK] 025_run_pcie_inference_async

教學程式總是使用五個幀進行預熱，然後測量 1,000 個完成的幀。僅在要使用另一個卡時，才傳遞 --card N。

實務應用

保持提交和檢索的平衡。如果應用程式無限期地推送而不進行拉取，則正常的反壓最終會減慢提交速度。一個專用的消費者也可以使故障變得簡單：有限的超時可以識別出停滯的結果，並且關閉模型即使生產者正在等待，也會釋放佇列 0。

為了獲得具有代表性的基準測試，請將重複的幀替換為固定的影像集，並將磁碟讀取操作放在計時區域之外。繼續執行 同時執行多個模型，以同時執行兩個不同的模型。

完整原始碼

顯示完整原始碼程式

pcie_host/tutorials/025_run_pcie_inference_async/run_pcie_inference_async.cpp

// Measure completed YOLOv8s detections with asynchronous PCIe push/pull.
//
// Usage:
//   tutorial_025_run_pcie_inference_async

#include <simaai/neat/pcie/Model.h>

#include <opencv2/imgcodecs.hpp>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <cstring>
#include <cstdlib>
#include <deque>
#include <exception>
#include <filesystem>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <numeric>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>

namespace pcie = simaai::neat::pcie;

namespace {

using Clock = std::chrono::steady_clock;
constexpr int kBuildTimeoutMs = 180000;
constexpr int kPullTimeoutMs = 30000;
constexpr int kWarmupFrames = 5;
constexpr int kMeasuredFrames = 1000;
constexpr char kModelPath[] = "yolo_v8s_mpk.tar.gz";
constexpr char kImagePath[] = "share/sima-pcie-host/tutorials/assets/street-scene.png";

struct Args {
  int card_id = 0;
};

std::string require_value(int argc, char** argv, int& index, const char* option) {
  if (index + 1 >= argc) {
    throw std::runtime_error(std::string("missing value for ") + option);
  }
  return argv[++index];
}

Args parse_args(int argc, char** argv) {
  Args args;
  for (int index = 1; index < argc; ++index) {
    const std::string arg = argv[index];
    if (arg == "--card") {
      args.card_id = std::stoi(require_value(argc, argv, index, "--card"));
    } else if (arg == "-h" || arg == "--help") {
      std::cout << "Usage: " << argv[0] << " [--card 0]\n";
      std::exit(0);
    } else {
      throw std::runtime_error("unknown argument: " + arg);
    }
  }
  return args;
}

pcie::ModelOptions detection_options() {
  pcie::ModelOptions options;
  options.preprocess.kind = pcie::InputKind::Image;
  options.preprocess.color_convert.input_format = pcie::ColorFormat::BGR;
  options.preprocess.color_convert.output_format = pcie::ColorFormat::RGB;
  options.preprocess.resize.enable = pcie::AutoFlag::On;
  options.preprocess.resize.mode = pcie::ResizeMode::Letterbox;
  options.preprocess.normalize.preset = pcie::NormalizePreset::COCO_YOLO;
  options.decode_type = pcie::BoxDecodeType::YoloV8;
  options.score_threshold = 0.25F;
  options.nms_iou_threshold = 0.45F;
  options.top_k = 100;
  return options;
}

std::uint32_t detection_count(const pcie::TensorList& outputs) {
  if (outputs.size() != 1 || outputs[0].data == nullptr || outputs[0].byte_offset < 0) {
    throw std::runtime_error("boxdecode must return one populated BBOX tensor");
  }
  const auto& tensor = outputs[0];
  const auto offset = static_cast<std::size_t>(tensor.byte_offset);
  if (offset > tensor.size_bytes || tensor.size_bytes - offset < sizeof(std::uint32_t)) {
    throw std::runtime_error("BBOX tensor is too small");
  }
  std::uint32_t count = 0;
  std::memcpy(&count, static_cast<const std::uint8_t*>(tensor.data) + offset, sizeof(count));
  constexpr std::size_t record_size = 24;
  if (count > (tensor.size_bytes - offset - 4) / record_size) {
    throw std::runtime_error("BBOX detection count exceeds its payload");
  }
  return count;
}

struct BenchmarkResult {
  std::size_t completed = 0;
  double elapsed_seconds = 0.0;
  double average_latency_ms = 0.0;
  std::uint64_t total_detections = 0;
};

BenchmarkResult measure(pcie::Model& model, const cv::Mat& image, const int frame_count) {
  std::deque<Clock::time_point> submitted;
  std::mutex submitted_mutex;
  std::mutex failure_mutex;
  std::exception_ptr first_failure;
  std::atomic<bool> cancelled = false;
  std::vector<double> latency_ms;
  latency_ms.reserve(static_cast<std::size_t>(frame_count));
  std::uint64_t total_detections = 0;

  auto fail = [&](std::exception_ptr failure) {
    {
      std::lock_guard<std::mutex> lock(failure_mutex);
      if (!first_failure) {
        first_failure = std::move(failure);
      }
    }
    cancelled = true;
    model.close();
  };

  const auto benchmark_start = Clock::now();
  std::thread producer([&] {
    try {
      for (int index = 0; index < frame_count && !cancelled; ++index) {
        const auto started = Clock::now();
        {
          std::lock_guard<std::mutex> lock(submitted_mutex);
          submitted.push_back(started);
        }
        if (!model.push(image)) {
          throw std::runtime_error("push rejected frame " + std::to_string(index));
        }
      }
    } catch (...) {
      fail(std::current_exception());
    }
  });

  std::thread consumer([&] {
    try {
      for (int index = 0; index < frame_count && !cancelled; ++index) {
        auto outputs = model.pull(kPullTimeoutMs);
        if (!outputs) {
          throw std::runtime_error("pull timed out for frame " + std::to_string(index));
        }
        Clock::time_point started;
        {
          std::lock_guard<std::mutex> lock(submitted_mutex);
          if (submitted.empty()) {
            throw std::runtime_error("completion arrived without a submission record");
          }
          started = submitted.front();
          submitted.pop_front();
        }
        total_detections += detection_count(*outputs);
        latency_ms.push_back(
            std::chrono::duration<double, std::milli>(Clock::now() - started).count());
      }
    } catch (...) {
      fail(std::current_exception());
    }
  });

  producer.join();
  consumer.join();
  const auto benchmark_end = Clock::now();
  if (first_failure) {
    std::rethrow_exception(first_failure);
  }
  if (latency_ms.size() != static_cast<std::size_t>(frame_count)) {
    throw std::runtime_error("not every submitted frame completed");
  }

  BenchmarkResult result;
  result.completed = latency_ms.size();
  result.elapsed_seconds = std::chrono::duration<double>(benchmark_end - benchmark_start).count();
  result.average_latency_ms =
      std::accumulate(latency_ms.begin(), latency_ms.end(), 0.0) / latency_ms.size();
  result.total_detections = total_detections;
  return result;
}

} // namespace

int main(int argc, char** argv) {
  try {
    const Args args = parse_args(argc, argv);
    if (!std::filesystem::is_regular_file(kModelPath)) {
      throw std::runtime_error(std::string("model does not exist: ") + kModelPath);
    }
    const cv::Mat image = cv::imread(kImagePath, cv::IMREAD_COLOR);
    if (image.empty()) {
      throw std::runtime_error(std::string("OpenCV could not decode: ") + kImagePath);
    }
    pcie::ConnectionOptions connection;
    connection.card_id = args.card_id;
    pcie::Model model(kModelPath, detection_options(), connection);
    model.build(kBuildTimeoutMs);

    for (int index = 0; index < kWarmupFrames; ++index) {
      (void)detection_count(model.run(image, kPullTimeoutMs));
    }

    // CORE LOGIC
    const BenchmarkResult result = measure(model, image, kMeasuredFrames);

    std::cout << "completed=" << result.completed << '\n';
    std::cout << std::fixed << std::setprecision(2) << "elapsed_seconds=" << result.elapsed_seconds
              << '\n'
              << "throughput_fps=" << result.completed / result.elapsed_seconds << '\n'
              << "average_latency_ms=" << result.average_latency_ms << '\n'
              << "total_detections=" << result.total_detections << '\n';

    model.close();
    std::cout << "[OK] 025_run_pcie_inference_async\n";
    return 0;
  } catch (const std::exception& error) {
    std::cerr << "[FAIL] " << error.what() << '\n';
    return 1;
  }
}

來源

• C++
• Python
• README