Run.h File

Run — the live pipeline returned by Graph::build, plus runtime options and diagnostics. More...

Included Headers

#include "nodes/io/Input.h" #include "pipeline/PowerTelemetry.h" #include "pipeline/RuntimeMetrics.h" #include "pipeline/GraphOptions.h" #include <cstdint> #include <cstddef> #include <functional> #include <memory> #include <optional> #include <string> #include <string_view> #include <vector>

Namespaces Index

namespace	cv
namespace	simaai
namespace	neat
namespace	pipeline_internal
namespace	runtime

Classes Index

struct	RunAdvancedOptions
	Advanced runtime tuning knobs (most users never set these). More...
struct	RunAutoExportOptions
	Optional build-time Run JSON export. More...
struct	RunOptions
	Per-Run runtime options. More...
struct	InputDropInfo
	Diagnostic record for a dropped input frame. More...
struct	InputStreamStats
	Per-Run input-side telemetry: counts, drops, and timing averages. More...
struct	RunStats
	Per-Run end-to-end statistics: counts and latency. More...
struct	RunMeasurementSummary
	One-call runtime measurement summary. More...
struct	RunStageStats
	Per-stage timing telemetry — how long each stage takes per sample. More...
struct	RunElementTimingStats
	Per-element timing — finer-grained than per-stage; one row per GStreamer element. More...
struct	RunElementFlowStats
	Per-element data-flow telemetry — buffer and byte counts, plus caps changes. More...
struct	RunElementPadTimingStats
	Per-pad timing — finest-grained telemetry, one row per (element, pad). More...
struct	RunDiagSnapshot
	Aggregate diagnostic snapshot: stages, boundaries, per-element, per-pad. More...
struct	MeasureOptions
	Options for framework-owned runtime measurement. More...
struct	MeasureLatencyStats
	Percentile summary for a measured latency series. More...
struct	MeasurePluginLatency
	Aggregated per-plugin/kernel timing captured during a measurement window. More...
struct	MeasureReport
	Framework-owned report returned by MeasureScope::stop(). More...
class	MeasureScope
	Observation scope for measuring an application-owned push/pull interval. More...
struct	RunReportOptions
	Toggles for what Run::report() includes in its formatted text output. More...
class	Run
	Live pipeline handle: push inputs in, pull outputs out. More...

Description

Run — the live pipeline returned by Graph::build, plus runtime options and diagnostics.

Run is what a Graph becomes when built. It owns the running GStreamer pipeline, its internal threads (typically 5–15: one per element thread boundary, plus dispatcher workers, plus a bus watcher), and its bounded queues. Application code interacts with a Run by push()-ing inputs and pull()-ing outputs — or run() for a synchronous shortcut. A Run can operate in Async mode (continuous pipeline, push/pull at user pace) or Sync mode (one frame in, one result out, repeat).

This header also defines:

• OverflowPolicy — how push() behaves when the input queue is full.
• RunPreset — preset bundles for common workloads (realtime, balanced, reliable).
• RunOptions — runtime knobs (queue depth, overflow policy, output memory, metrics).
• RunStats / InputStreamStats / RunDiagSnapshot — telemetry surfaces.

See Also

Graph::build for how a Run is constructed

See Also

GraphOptions for build-time options (Run takes runtime options here)

See Also

"Runs: the live pipeline (and the timing decision)" (§0.13 of the design deep dive)

File Listing

The file content with the documentation metadata removed is:

1

23#pragma once

24

25#include "nodes/io/Input.h"

26#include "pipeline/PowerTelemetry.h"

27#include "pipeline/RuntimeMetrics.h"

28#include "pipeline/GraphOptions.h"

29

30#include <cstdint>

31#include <cstddef>

32#include <functional>

33#include <memory>

34#include <optional>

35#include <string>

36#include <string_view>

37#include <vector>

38

39namespace cv {

40class Mat;

41} // namespace cv

42

43namespace simaai::neat {

44

45class InputStream;

46class LatencyProfiler;

47class Run;

48struct InputStreamOptions;

49namespace pipeline_internal {

50struct InputRouteProcessor;

52using InputRouteProcessorPtr = std::shared_ptr<const InputRouteProcessor>;

53} // namespace pipeline_internal

54namespace runtime {

55class RunCore;

56} // namespace runtime

57#ifdef SIMA_NEAT_INTERNAL

58namespace run_internal {

59std::shared_ptr<runtime::RunCore> release_core(Run& run);

60std::shared_ptr<const runtime::RunCore> core(const Run& run);

61} // namespace run_internal

62#endif

63

71enum class OverflowPolicy {

72 Block = 0,

73 KeepLatest,

75 DropIncoming,

77};

78

86enum class RunPreset {

87 Realtime,

88 Balanced,

89 Reliable,

90};

91

100enum class OutputMemory {

101 Auto = 0,

102 ZeroCopy,

103 Owned,

104};

105

107struct RunAdvancedOptions {

108 bool copy_input =

109 false;

110 std::size_t max_input_bytes = 0;

111 int sync_num_buffers_override =

112 -1;

125 bool prepare_output_cpu_visible = false;

126};

127

138struct RunAutoExportOptions {

139 std::string path;

140 std::string label;

141 bool include_metrics = true;

142 bool include_power = true;

143 int indent = 2;

144};

145

147using GraphRunAutoExportOptions = RunAutoExportOptions;

148

157struct RunOptions {

158 RunPreset preset = RunPreset::Balanced;

159 int queue_depth = 4;

160 OverflowPolicy overflow_policy =

161 OverflowPolicy::Block;

162 OutputMemory output_memory =

163 OutputMemory::Auto;

164 bool enable_metrics = false;

171 int input_timeout_ms = -1;

185 bool startup_preflight = true;

186 RunAdvancedOptions advanced{};

187 PowerMonitorOptions power_monitor{};

188 RunAutoExportOptions run_export{};

190 RunAutoExportOptions graph_run_export{};

191

198 RunOptions& enable_board_power(int sample_interval_ms = 100) {

199 power_monitor = board_power_monitor_options(sample_interval_ms);

200 return *this;

201 }

202

215 RunOptions& enable_modalix_som_power(int sample_interval_ms = 100) {

216 power_monitor = modalix_som_power_monitor_options(sample_interval_ms);

217 return *this;

218 }

219

223 RunOptions& enable_modalix_dvt_power(int sample_interval_ms = 100) {

224 power_monitor = modalix_dvt_power_monitor_options(sample_interval_ms);

225 return *this;

226 }

227

231 RunOptions& disable_power_monitor() {

232 power_monitor = PowerMonitorOptions{};

233 return *this;

234 }

235

242 std::function<void(const struct InputDropInfo&)> on_input_drop;

243};

244

252struct InputDropInfo {

253 SampleKind kind = SampleKind::Unknown;

254 std::string media_type;

255 std::string format;

256 int width = -1;

257 int height = -1;

258 int depth = -1;

259 int64_t frame_id = -1;

260 std::string stream_id;

261 std::string port_name;

262 std::string reason;

263};

264

272struct InputStreamStats {

273 std::uint64_t push_count = 0;

274 std::uint64_t push_failures = 0;

275 std::uint64_t pull_count = 0;

276 std::uint64_t poll_count = 0;

277 std::uint64_t dropped_frames = 0;

278 std::uint64_t renegotiations = 0;

279 std::uint64_t alloc_grows = 0;

280 std::uint64_t growth_blocked = 0;

281 std::uint64_t renegotiation_blocked =

282 0;

283 double avg_alloc_us = 0.0;

284 double avg_map_us = 0.0;

285 double avg_copy_us = 0.0;

286 double avg_push_us = 0.0;

287 double avg_pull_wait_us = 0.0;

288 double avg_decode_us = 0.0;

289};

290

298struct RunStats {

299 std::uint64_t inputs_enqueued = 0;

300 std::uint64_t inputs_dropped = 0;

301 std::uint64_t inputs_pushed = 0;

302 std::uint64_t outputs_ready =

303 0;

304 std::uint64_t outputs_pulled = 0;

305 std::uint64_t outputs_dropped =

306 0;

307 double avg_latency_ms = 0.0;

308 double min_latency_ms = 0.0;

309 double max_latency_ms = 0.0;

310};

311

321struct RunMeasurementSummary {

322 RunStats stats;

323 InputStreamStats input_stats;

324 double elapsed_seconds = 0.0;

325 double throughput_fps = 0.0;

326 PowerSummary power;

327};

328

333struct RunStageStats {

334 std::string stage_name;

335 std::uint64_t samples = 0;

336 std::uint64_t total_us = 0;

337 std::uint64_t max_us = 0;

338};

339

344struct RunElementTimingStats {

345 std::string element_name;

346 std::uint64_t samples = 0;

347 std::uint64_t total_us =

348 0;

349 std::uint64_t max_us = 0;

350 std::uint64_t min_us = 0;

351 std::uint64_t missed_in = 0;

352 std::uint64_t missed_out = 0;

353};

354

359struct RunElementFlowStats {

360 std::string element_name;

361 std::uint64_t in_buffers = 0;

362 std::uint64_t out_buffers = 0;

363 std::uint64_t in_bytes = 0;

364 std::uint64_t out_bytes = 0;

365 std::uint64_t caps_changes = 0;

366};

367

375struct RunElementPadTimingStats {

376 std::string element_name;

377 std::string pad_name;

378 bool is_sink = false;

379 std::uint64_t samples = 0;

380 std::uint64_t inter_arrival_total_us =

381 0;

382 std::uint64_t inter_arrival_max_us = 0;

383 std::uint64_t queue_wait_samples =

384 0;

385 std::uint64_t queue_wait_total_us =

386 0;

387 std::uint64_t queue_wait_max_us = 0;

388 std::uint64_t bytes = 0;

389};

390

398struct RunDiagSnapshot {

399 std::vector<RunStageStats> stages;

400 std::vector<BoundaryFlowStats> boundaries;

401 std::vector<RunElementTimingStats> element_timings;

402 std::vector<RunElementFlowStats> element_flows;

409 std::vector<RunElementPadTimingStats> element_pad_timings;

410};

411

418struct MeasureOptions {

419 int duration_ms = 10000;

420 int warmup_ms = 1000;

421 int timeout_ms = 5000;

424 bool include_plugin_latency = true;

425 bool include_power = true;

426

429 std::string title = "NEAT measurement";

430 std::string model;

431 std::string input;

432 std::string placement;

433 int logical_batch_size = 1;

434};

435

439struct MeasureLatencyStats {

440 std::size_t count = 0;

441 double avg_ms = 0.0;

442 double p50_ms = 0.0;

443 double p90_ms = 0.0;

444 double p95_ms = 0.0;

445 double p99_ms = 0.0;

446 double max_ms = 0.0;

447};

448

452struct MeasurePluginLatency {

453 std::string name;

454 std::uint64_t calls = 0;

455 double avg_ms = 0.0;

456 double min_ms = 0.0;

457 double max_ms = 0.0;

458};

459

463struct MeasureReport {

464 MeasureOptions options;

465 std::size_t warmup_iterations = 0;

466 std::size_t outputs = 0;

467 double elapsed_s = 0.0;

468 double throughput_batches_per_s = 0.0;

469 double throughput_inferences_per_s = 0.0;

470

471 MeasureLatencyStats end_to_end;

472 MeasureLatencyStats frame_gap;

473 bool latency_samples_collected = false;

474 std::vector<MeasurePluginLatency> plugin_latency;

475

476 std::uint64_t inputs_pushed = 0;

477 std::uint64_t outputs_pulled = 0;

478 std::uint64_t inputs_dropped = 0;

479 std::uint64_t outputs_dropped = 0;

480 RunStats final_run_stats{};

481 PowerSummary power{};

482

484 std::string to_text() const;

485};

486

494class MeasureScope {

495public:

496 MeasureScope(MeasureScope&&) noexcept;

497 MeasureScope& operator=(MeasureScope&&) noexcept;

498 ~MeasureScope();

499

500 MeasureScope(const MeasureScope&) = delete;

501 MeasureScope& operator=(const MeasureScope&) = delete;

502

503 MeasureReport stop();

504 bool stopped() const;

505

506private:

507 friend class Run;

508 struct Impl;

509 explicit MeasureScope(std::unique_ptr<Impl> impl);

510 std::unique_ptr<Impl> impl_;

511};

512

520struct RunReportOptions {

521 bool include_pipeline = true;

522 bool include_stage_timings = true;

523 bool include_element_timings = true;

524 bool include_boundaries = true;

525 bool include_flow_stats = true;

526 bool include_node_reports = false;

527 bool include_next_cpu = false;

528 bool include_queue_depth = true;

529 bool include_num_buffers = true;

530 bool include_run_stats = true;

531 bool include_input_stats = true;

532 bool include_power = true;

533 bool include_system_info = false;

534};

535

562class Run {

563public:

565 Run() = default;

566 Run(const Run&) = delete;

567 Run& operator=(const Run&) = delete;

568

569 Run(Run&&) noexcept;

570 Run& operator=(Run&&) noexcept;

571 ~Run();

572

574 explicit operator bool() const noexcept;

576 bool can_push() const;

578 bool can_pull() const;

580 bool running() const;

582 std::vector<std::string> input_names() const;

584 std::vector<std::string> output_names() const;

585

590 bool push(const std::vector<cv::Mat>& inputs);

592 bool push(std::string_view input_name, const std::vector<cv::Mat>& inputs);

594 bool try_push(const std::vector<cv::Mat>& inputs);

596 bool try_push(std::string_view input_name, const std::vector<cv::Mat>& inputs);

598 bool push(const TensorList& inputs);

600 bool push(std::string_view input_name, const TensorList& inputs);

602 bool try_push(const TensorList& inputs);

604 bool try_push(std::string_view input_name, const TensorList& inputs);

606 bool push(const Sample& msgs);

608 bool push(std::string_view input_name, const Sample& msgs);

610 bool try_push(const Sample& msgs);

612 bool try_push(std::string_view input_name, const Sample& msgs);

614 bool push_holder(const std::shared_ptr<void>& holder);

616 bool try_push_holder(const std::shared_ptr<void>& holder);

619 void close_input();

627 PullStatus pull(int timeout_ms, Sample& out, PullError* err = nullptr);

634 PullStatus pull(std::string_view output_name, int timeout_ms, Sample& out,

635 PullError* err = nullptr);

637 std::optional<Sample> pull(int timeout_ms = -1);

639 std::optional<Sample> pull(std::string_view output_name, int timeout_ms = -1);

641 TensorList pull_tensors(int timeout_ms = -1);

643 TensorList pull_tensors(std::string_view output_name, int timeout_ms = -1);

645 Sample pull_samples(int timeout_ms = -1);

647 Sample pull_samples(std::string_view output_name, int timeout_ms = -1);

649 TensorList run(const std::vector<cv::Mat>& inputs, int timeout_ms = -1);

651 TensorList run(const TensorList& inputs, int timeout_ms = -1);

653 Sample run(const Sample& inputs, int timeout_ms = -1);

654

656 MeasureScope start_measurement(const MeasureOptions& opt = {});

663 int warmup(const std::vector<cv::Mat>& inputs, int warm = -1, int timeout_ms = -1);

664

666 RunStats stats() const;

668 InputStreamStats input_stats() const;

670 RunDiagSnapshot diag_snapshot() const;

672 PowerSummary power_summary() const;

674 RunMeasurementSummary measurement_summary() const;

676 RuntimeMetrics metrics(const RuntimeMetricsOptions& opt = {}) const;

678 std::string metrics_report(const RuntimeMetricsOptions& opt = {},

679 RuntimeMetricsFormat format = RuntimeMetricsFormat::Text) const;

681 std::string metrics_report(RuntimeMetricsFormat format) const;

683 std::string report(const RunReportOptions& opt = {}) const;

685 std::string last_error() const;

687 std::string diagnostics_summary() const;

688

690 void stop();

692 void close();

693

694private:

695 friend class MeasureScope;

696#ifdef SIMA_NEAT_INTERNAL

697 friend std::shared_ptr<runtime::RunCore> run_internal::release_core(Run& run);

698 friend std::shared_ptr<const runtime::RunCore> run_internal::core(const Run& run);

699#endif

700 std::shared_ptr<runtime::RunCore> core_;

701

702 explicit Run(std::shared_ptr<runtime::RunCore> core);

703 void require_async_mode(const char* where) const;

704 void require_async_pull_mode(const char* where) const;

705 void enqueue_run_images(const std::vector<cv::Mat>& inputs);

706 void enqueue_run_tensors(const TensorList& inputs);

707 void enqueue_run_samples(const Sample& inputs);

708 TensorList pull_tensors_strict(int timeout_ms);

709 Sample pull_samples_strict(int timeout_ms);

710 bool push_impl(const cv::Mat& input, bool block);

711 bool push_impl(const simaai::neat::Tensor& input, bool block);

712 bool push_holder_impl(const std::shared_ptr<void>& holder, bool block);

713 bool push_message_impl(const Sample& msg, bool block);

714 bool push_sample_impl(const Sample& msg, bool block);

715 static Run create(InputStream stream, const RunOptions& opt,

716 const struct InputStreamOptions& stream_opt, RunMode mode = RunMode::Async,

717 const std::optional<InputOptions>& tensor_input_opt_for_cv = std::nullopt,

718 pipeline_internal::InputRouteProcessorPtr input_route_processor = nullptr);

719 friend class Graph;

720};

721

722} // namespace simaai::neat

---

Generated via doxygen2docusaurus 2.0.0 by Doxygen 1.9.8.