하나의 그래프에서 여러 스트림 실행
| 필드 | 값 |
|---|---|
| 범주 | 그래프 및 파이프라인 |
| 난이도 | 고급 |
| 예상 소요 시간 | 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"를 전달하고, 프레임 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"에서 한 번 가져옵니다. 각 성공적인 가져오기는 런타임이 두 입력 모두 해당 프레임을 전달한 후에 출력한 번들을 반환합니다. 생성과 동시에 데이터를 소비함으로써 제한된 출력 큐가 가득 차서 입력 측에 역압력이 전파되는 것을 방지합니다. 두 예제 모두 각 번들에 두 개의 결합된 필드가 포함되어 있는지 확인한 다음 close()를 호출하여 실행을 깔끔하게 종료합니다. 예상되는 번들 수는 streams * frames와 같으며, 이는 쌍이 누락되지 않았음을 증명합니다.
run.pull("combined", timeout_ms)는 선택적 번들을 반환합니다. bundle.stream_id 및 bundle.fields.size()를 읽고 각 번들에 두 개의 필드가 있는지 확인합니다.
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;
}
}