非同期と同期のタイミングモデル

GraphとRunは、処理を実行するための2つの方法を提供します。

1回の実行: Graph.run(input, ...)は、入力を送信し、1回の呼び出しで出力が完了するのを待ちます。入力なしのGraph.run()は、EOS（End of Stream）に達するまで、ソースが所有するグラフを実行します。
再利用可能な実行: Graph.build(...)は、実行可能なRunを返します。アプリケーションは、入力を送信し、出力を取得し、入力を閉じ、残りの処理を実行し、測定を行い、実行を停止します。

どちらのモードも、同じGraphプランと、同じハードウェアを使用します。違いは、ループを誰が所有するかです。

1回の実行モード

1つの入力があり、最短の正しいパスを求める場合は、1回の実行モードを使用します。

simaai::neat::Graph graph("classifier");
graph.add(simaai::neat::nodes::Input("image"));
graph.add(model);
graph.add(simaai::neat::nodes::Output("classes"));

simaai::neat::TensorList out = graph.run(
    simaai::neat::TensorList{image_tensor});

ソースが所有するグラフの場合、引数なしでGraph.run()を呼び出します。

simaai::neat::Graph graph("file_job");
graph.add(source_fragment);
graph.add(model);
graph.add(sink_fragment);

graph.run();  // Blocks until EOS or error.

以下の用途には、ワンショットモードを使用してください。

スモークテスト
短時間の検証実行
完了まで実行されるべき、ソースコードが所有するジョブ
長期間にわたってランタイムハンドルを管理する必要がないコード

再利用可能な実行モード

アプリケーションがループを所有している場合は、Graph.build(...) を使用してください。

auto run = graph.build();

while (have_more_inputs()) {
  run.push("image", simaai::neat::TensorList{next_tensor()});

  if (auto out = run.pull("classes", /*timeout_ms=*/0)) {
    consume(*out);
  }
}

run.close_input();
while (auto out = run.pull("classes", /*timeout_ms=*/1000)) {
  consume(*out);
}
run.close();

以下の用途には、再利用可能なモードを使用します。

ライブビデオ、RTSP、またはカメラ入力
アプリケーションが処理速度を制御するストリーム処理
複数の入力または出力を持つグラフ
スループットテスト
測定、エクスポート、ドレイン、および停止制御

プッシュタイミング

Run::push(...) は、入力がグラフの境界で受け入れられた後に処理を終了します。入力がすべてのノードを通過するまで待機することはありません。

入力キューがいっぱいの場合：

OverflowPolicy::Block は、プロデューサーにバックプレッシャーを適用します。
OverflowPolicy::DropIncoming は、新しい入力を拒否します。
OverflowPolicy::KeepLatest は、古いキューに入れられた入力を破棄し、ライブパスを最新の状態に保ちます。
try_push(...) は、ブロックする代わりに false を返します。

ソースに合ったポリシーを選択します。ファイルおよびバッチジョブでは通常、Block が必要です。ライブストリームでは通常、最新性を保つポリシーが必要です。

インラインキュー

RunOptions::overflow_policy は Run の入力にあるキューを制御します。グラフノード間のキューを制御するには、設定済みの Queue を追加します。

simaai::neat::QueueOptions queue;
queue.max_buffers = 1;
queue.overflow_policy = simaai::neat::OverflowPolicy::KeepLatest;
graph.add(simaai::neat::nodes::Queue(queue));

queue = pyneat.QueueOptions()
queue.max_buffers = 1
queue.overflow_policy = pyneat.OverflowPolicy.KeepLatest
graph.add(pyneat.nodes.queue(queue))

max_buffers を設定しないままにすると、既定のキュー制限が使用されます。正の値を設定すると、バッファー数が唯一の容量制限になります。Block は空きができるまで待機し、KeepLatest は待機中の最も古いバッファーを破棄し、DropIncoming は新しいバッファーを破棄します。

プルタイミング

Run::pull(...) は、出力境界から次の利用可能な Sample を返します。

アプリケーションは入力の送信を一時停止し、同じ Run で再開できます。出力がないだけで失敗することはありません。有限の pull() タイムアウトはその待機を終了しますが、Run は停止しません。RunOptions::input_timeout_ms は、入力モードの Graph::run() と、シードを渡すビルドの対応する事前検証での出力待機時間を制御します。Run::run() と Model::run() は呼び出しごとのタイムアウトを使用し、pull(-1) と同様に -1 は無制限です。処理の停滞を検出する必要がある場合は、未完了の処理に期限を設定してください。

タイムアウトと EOS が同じ「サンプルなし」パスを共有できる場合は、便利なオーバーロードを使用します。

if (auto sample = run.pull("classes", /*timeout_ms=*/1000)) {
  consume(*sample);
}

タイムアウト、クローズ、エラーが発生した場合に、それぞれ異なる処理を行う必要がある場合は、構造化されたステータスオーバーロードを使用してください。

simaai::neat::Sample sample;
simaai::neat::PullError error;

switch (run.pull("classes", /*timeout_ms=*/1000, sample, &error)) {
case simaai::neat::PullStatus::Ok:
  consume(sample);
  break;
case simaai::neat::PullStatus::Timeout:
  break;
case simaai::neat::PullStatus::Closed:
  break;
case simaai::neat::PullStatus::Error:
  throw std::runtime_error(error.message);
}

隠れたフォールバック処理は行わない：複数の出力を持つグラフには、明確な単一の出力がない限り、名前付きのpull("output", ...)が必要。

テレメトリー：担当するループを計測する

アプリケーションが担当するワークロードの周囲にRun::start_measurement(...)を使用する。返されるMeasureReportは、公開されているタイミング情報であり、以下が含まれる：

エンドツーエンドのプッシュから出力までの遅延とスループット。
プッシュ、プル、およびドロップされたサンプルのランタイムカウンタ。
MeasureOptionsで要求された場合のプラグイン/カーネルおよびエッジのタイミング。
非推奨のRunOptions::power_monitorが有効な場合の電力。

TelemetryMonitorを使うと、同じ期間のボードの電力、温度、CPU使用率、メモリを記録できる。

セットアップ、ファイルダウンロード、フレームごとのロギング、およびレポートのエクスポートは、計測対象のホットループの外で行う。ただし、質問がエンドツーエンドのアプリケーションコストである場合は除く。

1回の実行モード​

再利用可能な実行モード​

プッシュタイミング​

インラインキュー​

プルタイミング​

テレメトリー：担当するループを計測する​

関連資料​