在單一圖中執行多個串流

欄位	值
類別	圖形與管線
難度	進階
預估閱讀時間	20-25 minutes
標籤	graph, multistream, scheduler, join

前幾章介紹了單一輸入和單一輸出。真正的多攝影機和並行分支系統則更複雜：多個串流獨立進行，它們的結果必須在任何後續處理之前正確地重新合併。本章將展示用於實現此目的的合併基本元件——一個具有兩個命名輸入和一個命名輸出的合併圖，該圖僅在雙方都產生匹配的影格時才會輸出一個組合。

您推送的每個樣本都帶有一個 stream_id 和一個 frame_id。合併策略 ByFrame 會等待，直到兩個命名輸入（left 和 right）都傳遞了具有相同 frame_id 的樣本，然後輸出精確一個組合。最後，您將建立一個合併圖，通過其兩個輸入將確定性的每個串流/每個影格的工作負載分發出去，並將合併的組合拉回——驗證輸出計數，並驗證每個組合都包含兩個欄位。

操作指南

建立合併圖

graphs::Combine (C++) / graphs.combine (Python) 傳回一個普通的公共 Graph 片段——除了其形狀之外，它沒有什麼特別之處：兩個命名輸入、一個命名輸出和一個合併策略。我們將 ["left", "right"] 作為輸入名稱，將 "combined" 作為輸出名稱，並將 CombinePolicy.ByFrame 傳遞給它，以選擇影格 ID 匹配。列印 describe() 會顯示生成的拓撲結構，而 build() 會將描述轉換為可執行的句柄。該圖預設以非同步方式運行，因此每個串流都可以獨立地進行。

輸出佇列是有限制的。與為整個工作負載分配足夠的佇列空間不同，此範例在推送下一對之前，會先拉取每個合併的組合。生產者和消費者同步進行，因此隨著影格計數的增加，記憶體使用量保持在有限的範圍內。

CombinePolicy.ByFrame 根據 Sample.frame_id 進行匹配；CombinePolicy.ByPts 是另一種替代方案，它根據呈現時間戳 (Sample.pts_ns) 進行匹配，當影格不共享乾淨的影格索引時使用。

tutorials/015_run_multiple_streams/run_multiple_streams.cpp

simaai::neat::Graph graph = simaai::neat::graphs::Combine({"left", "right"}, "combined",
                                                          simaai::neat::CombinePolicy::ByFrame);

std::cout << graph.describe() << "\n";

const int expected = streams * frames;
simaai::neat::Run run = graph.build();

推送串流

現在我們驅動工作負載。對於每個影格和每個串流，我們都會合成一個小的確定性 RGB 樣本，並標記其 stream_id 和一個唯一的 frame_id，然後將其推送到兩個命名輸入中。由於 ID 是確定性地計算的 (frame * streams + sid)，因此合併具有明確的配對關係——left 影格 N 始終具有匹配的 right 影格 N。在匹配的 right 推送之後，我們會在移動到下一對之前，先清空該對的合併輸出。

**C++：**每個樣本都是明確地建構而成，作為一個 Sample，它包含一個 Tensor（HWC、UInt8、RGB），並設定了 frame_id 和 stream_id；run.push("left", sample) 會傳回一個布林值，您應該將其與 run.last_error() 進行比較。

Python：make_rgb_sample(...) 透過 Tensor.from_numpy(...) 從 NumPy 陣列建構 Sample；run.push("left", [sample]) 接受一個樣本列表。

tutorials/015_run_multiple_streams/run_multiple_streams.cpp

if (!run.push("left", make_rgb_sample(std::to_string(sid), logical_frame))) {
  throw std::runtime_error("left push failed: " + run.last_error());
}
if (!run.push("right", make_rgb_sample(std::to_string(sid), logical_frame))) {
  throw std::runtime_error("right push failed: " + run.last_error());
}

提取每個已合併的 bundle

在每次配對成功推送後，我們會從指定的輸出 "combined" 提取一次。每次成功的提取都會傳回執行階段在兩個輸入都傳遞了該幀之後發出的 bundle。在產生資料的同時進行提取，可以防止受限的輸出佇列填滿，並將反壓傳播到輸入端。兩個範例都會驗證每個 bundle 是否包含兩個已合併的欄位，然後呼叫 close() 以乾淨地結束執行。預期的 bundle 數量等於 streams * frames，證明沒有遺漏任何配對。

C++：run.pull("combined", timeout_ms) 傳回一個可選的 bundle；我們讀取 bundle.stream_id 和 bundle.fields.size()，並驗證每個 bundle 是否具有兩個欄位。

Python：run.pull("combined", 2000) 傳回 bundle 或 None；如果發生逾時，該範例會立即失敗，並驗證每個 bundle 的欄位數量。

tutorials/015_run_multiple_streams/run_multiple_streams.cpp

auto maybe_bundle = run.pull("combined", /*timeout_ms=*/2000);
if (!maybe_bundle.has_value()) {
  throw std::runtime_error("timed out waiting for combined output: " + run.last_error());
}
const auto& bundle = *maybe_bundle;
const int fields = static_cast<int>(bundle.fields.size());
if (fields != 2)
  throw std::runtime_error("joined bundle should contain two fields");
if (first_fields < 0)
  first_fields = fields;
++received;
if (received <= 4) {
  std::cout << "bundle stream=" << bundle.stream_id << " fields=" << fields << "\n";
}

執行

本章不需要模型封存檔。從 Neat 安裝根目錄（包含 share/ 和 lib/ 的目錄）執行 Python 和 C++（預先建置） 命令；從 程式碼庫根目錄 執行 從原始碼建置 命令。

C++ (prebuilt):

./lib/sima-neat/tutorials/tutorial_015_run_multiple_streams \
  --streams 8 --frames 4

C++ (build from source):

./build.sh --target tutorial_015_run_multiple_streams
./build/tutorials-standalone/tutorial_015_run_multiple_streams \
  --streams 8 --frames 4

預期的輸出（C++ 建置也會列印圖的描述；兩個建置都會列印前幾個 bundle）：

received=32 fields=2
[OK] 015_run_multiple_streams

若要將本章的 C++ 原始碼整合到您自己的專案中，並使用自訂的 CMakeLists.txt（不需要額外的資料夾），請參閱登陸頁面上的 如何執行教學。

完整原始碼

顯示完整原始碼程式

tutorials/015_run_multiple_streams/run_multiple_streams.cpp

// Multistream public Graph: named inputs -> Combine(ByFrame) -> named output bundle.
//
// Usage:
//   tutorial_015_run_multiple_streams [--streams 8] [--frames 4]

#include "neat.h"

#include <cstdint>
#include <iostream>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>

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

std::vector<int64_t> contiguous_strides_bytes(const std::vector<int64_t>& shape,
                                              int64_t elem_bytes) {
  std::vector<int64_t> strides(shape.size(), 0);
  int64_t stride = elem_bytes;
  for (int i = static_cast<int>(shape.size()) - 1; i >= 0; --i) {
    strides[static_cast<size_t>(i)] = stride;
    stride *= shape[static_cast<size_t>(i)];
  }
  return strides;
}

simaai::neat::Sample make_rgb_sample(const std::string& stream_id, int frame_id) {
  const int w = 8;
  const int h = 6;
  const int c = 3;
  const std::size_t bytes = static_cast<std::size_t>(w) * h * c;

  simaai::neat::Tensor t;
  t.device = {simaai::neat::DeviceType::CPU, 0};
  t.dtype = simaai::neat::TensorDType::UInt8;
  t.layout = simaai::neat::TensorLayout::HWC;
  t.shape = {h, w, c};
  t.semantic.image = simaai::neat::ImageSpec{simaai::neat::ImageSpec::PixelFormat::RGB, ""};
  t.storage = simaai::neat::make_cpu_owned_storage(bytes);
  t.strides_bytes = contiguous_strides_bytes(t.shape, 1);
  t.read_only = false;
  {
    auto map = t.map(simaai::neat::MapMode::Write);
    auto* p = static_cast<std::uint8_t*>(map.data);
    for (std::size_t i = 0; i < bytes; ++i)
      p[i] = static_cast<std::uint8_t>(i % 255);
  }
  t.read_only = true;

  simaai::neat::Sample sample;
  sample.kind = simaai::neat::SampleKind::Tensor;
  sample.tensor = std::move(t);
  sample.frame_id = frame_id;
  sample.stream_id = stream_id;
  return sample;
}

} // namespace

int main(int argc, char** argv) {
  try {
    const int streams = parse_int_arg(argc, argv, "--streams", 8);
    const int frames = parse_int_arg(argc, argv, "--frames", 4);

    // CORE LOGIC
    // `graphs::Combine` is a normal public Graph fragment. It declares two
    // named inputs ("left", "right") and one named output ("combined"). ByFrame
    // means the runtime emits one bundle only after both inputs have delivered
    // samples with the same Sample::frame_id.
    simaai::neat::Graph graph = simaai::neat::graphs::Combine({"left", "right"}, "combined",
                                                              simaai::neat::CombinePolicy::ByFrame);

    std::cout << graph.describe() << "\n";

    const int expected = streams * frames;
    simaai::neat::Run run = graph.build();

    int received = 0;
    int first_fields = -1;
    for (int frame = 0; frame < frames; ++frame) {
      for (int sid = 0; sid < streams; ++sid) {
        const int logical_frame = frame * streams + sid;

        if (!run.push("left", make_rgb_sample(std::to_string(sid), logical_frame))) {
          throw std::runtime_error("left push failed: " + run.last_error());
        }
        if (!run.push("right", make_rgb_sample(std::to_string(sid), logical_frame))) {
          throw std::runtime_error("right push failed: " + run.last_error());
        }

        auto maybe_bundle = run.pull("combined", /*timeout_ms=*/2000);
        if (!maybe_bundle.has_value()) {
          throw std::runtime_error("timed out waiting for combined output: " + run.last_error());
        }
        const auto& bundle = *maybe_bundle;
        const int fields = static_cast<int>(bundle.fields.size());
        if (fields != 2)
          throw std::runtime_error("joined bundle should contain two fields");
        if (first_fields < 0)
          first_fields = fields;
        ++received;
        if (received <= 4) {
          std::cout << "bundle stream=" << bundle.stream_id << " fields=" << fields << "\n";
        }
      }
    }

    run.close();

    if (received != expected)
      throw std::runtime_error("expected=" + std::to_string(expected) +
                               " received=" + std::to_string(received));
    if (first_fields != 2)
      throw std::runtime_error("join should emit a two-field bundle");

    std::cout << "received=" << received << " fields=" << first_fields << "\n";
    std::cout << "[OK] 015_run_multiple_streams\n";
    return 0;
  } catch (const std::exception& e) {
    std::cerr << "[FAIL] " << e.what() << "\n";
    return 1;
  }
}

來源

• C++
• Python
• README