1つのグラフで複数のストリームを実行する
| 項目 | 値 |
|---|---|
| カテゴリ | グラフとパイプライン |
| 難易度 | 上級 |
| 推定所要時間 | 20-25 minutes |
| ラベル | graph, multistream, scheduler, join |
前の章では、1つの入力をプッシュし、1つの出力をプルしました。実際のマルチカメラおよび並列ブランチシステムはより複雑です。複数のストリームが独立して進行し、それらの結果は、下流のシステムがそれを使用する前に、正しく結合される必要があります。この章では、それを決定的に行うための結合プリミティブ、つまり、2つの名前付き入力と1つの名前付き出力を持つ結合グラフを示します。このグラフは、両方の側が一致するフレームを生成した場合にのみ、バンドルを出力します。
プッシュするすべてのサンプルには、stream_idとframe_idが含まれます。結合ポリシーByFrameは、名前付きの入力（leftとright）の両方が、同じframe_idを持つサンプルを配信するまで待ち、その後、正確に1つの結合された バンドルを出力します。最終的には、結合グラフを構築し、2つの入力を通じて決定的なストリーム/フレームごとのワークロードを分散させ、結合されたバンドルをプルして、出力数と各バンドルが2つのフィールドを持つことを検証します。
ウォークスルー
結合グラフの構築
graphs::Combine（C++）/ graphs.combine（Python）は、通常のパブリックGraphフラグメントを返します。その形状（2つの名前付き入力、1つの名前付き出力、および結合ポリシー）以外には、特別な点は何もありません。入力名として["left", "right"]、出力名として"combined"を渡し、フレームIDの一致を選択するためにCombinePolicy.ByFrameを渡します。describe()を出力すると、結果のトポロジーが表示され、build()は、その説明を実行可能なハンドルに変換します。グラフはデフォルトで非同期に実行されるため、各ストリームは独立して進捗できます。
出力キューは制限されています。ワークロード全体に必要なキュー領域を割り当てる代わりに、この例では、次のペアをプッシュする前に、各結合されたバンドルをプルします。プロデューサーとコンシュ ーマーは一緒に進むため、フレーム数が増加しても、メモリ使用量は制限されたままになります。
CombinePolicy.ByFrameは、Sample.frame_idに基づいて一致します。CombinePolicy.ByPtsは、フレームが明確なフレームインデックスを共有しない場合に、プレゼンテーションタイムスタンプ（Sample.pts_ns）に基づいて一致させる代替手段です。
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();
ストリームのプッシュ
次に、ワークロードを実行します。各フレームと各ストリームについて、そのstream_idと一意のframe_idがタグ付けされた小さな決定的なRGBサンプルを合成し、それを両方の名前付き入力にプッシュします。IDは決定的に計算されるため（frame * streams + sid）、結合は、一意のペアを見つけることができます。leftフレームNには、常に一致するrightフレームNがあります。一致するrightをプッシュした後、そのペアの結合された出力をドレインしてから、次のペアに進みます。
各サンプルは、frame_idとstream_idが設定されたTensor（HWC、UInt8、RGB）をラップしたSampleとして明示的に構築されます。run.push("left", sample)は、run.last_error()に対して確認すべきboolを返します。
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());
}
各結合されたバンドルをプルする
各一致するペアがプッシュされる直後に、名前付きの出力"combined"から一度プルします。各成功したプルは、ランタイムが両方の入力がそのフレームを送信した後に送信したバンドルを返します。生成と同時に処理することで、バッファリングされた出力キューがいっぱいになり、入力側にバックプレッシャーが伝播するのを防ぎます。両方の例では、すべてのバンドルに2つの結合されたフィールドが含まれていることを確認し、その後close()を呼び出して、実行をクリーンに終了します。予想されるバンドルの数はstreams * framesと等しく、ペアリングが削除されなかったことを証明します。
run.pull("combined", timeout_ms)は、オプションのバンドルを返します。bundle.stream_idとbundle.fields.size()を読み取り、各バンドルに2つのフィールドがあることを確認します。
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++ビルドではグラフの説明も出力されます。両方のビルドでは、最初のいくつかのバンドルが出力されます）。
received=32 fields=2
[OK] 015_run_multiple_streams
カスタムのCMakeLists.txt（追加のフォルダーは不要）を使用して、この章のC++ソースを独自のプロジェクトに統合する方法については、ランディングページにあるチュートリアルの実行方法を参照してください。
完全なソース
完全なソースプログラムを表示
// 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;
}
}