< Summary

Information
Class: include
Assembly: holoflow
File(s): File 1: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\include\holoflow\core\tasks.hh
File 2: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\include\holoflow\core\tensor.hh
File 3: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\include\holoflow\runtime\graph_exec.hh
Line coverage
70%
Covered lines: 17
Uncovered lines: 7
Coverable lines: 24
Total lines: 735
Line coverage: 70.8%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

File(s)

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\include\holoflow\core\tasks.hh

#LineLine coverage
 1// Copyright 2025 Digital Holography Foundation
 2//
 3// Licensed under the Apache License, Version 2.0 (the "License");
 4// you may not use this file except in compliance with the License.
 5// You may obtain a copy of the License at
 6//
 7//     http://www.apache.org/licenses/LICENSE-2.0
 8//
 9// Unless required by applicable law or agreed to in writing, software
 10// distributed under the License is distributed on an "AS IS" BASIS,
 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12// See the License for the specific language governing permissions and
 13// limitations under the License.
 14
 15/// @file tasks.hh
 16/// @brief Interfaces and runtime contexts for Holoflow tasks.
 17///
 18/// A task consumes zero or more input tensors and produces zero or more output
 19/// tensors. Tasks are either synchronous (blocking) or asynchronous
 20/// (decoupled push/pop).
 21///
 22/// @section overview Overview
 23/// - Contexts: @ref holoflow::core::SyncCtx, @ref holoflow::core::AsyncPushCtx,
 24///   @ref holoflow::core::AsyncPopCtx
 25/// - Control flow: @ref holoflow::core::OpResult
 26/// - Task interfaces: @ref holoflow::core::ITask, @ref
 27///   holoflow::core::ISyncTask, @ref holoflow::core::IAsyncTask
 28/// - Factories: @ref holoflow::core::ITaskFactory,
 29///   @ref holoflow::core::ISyncTaskFactory, @ref holoflow::core::IAsyncTaskFactory
 30/// - Inference result: @ref holoflow::core::InferResult
 31///
 32/// @section lifecycle Lifecycle (single source of truth)
 33/// - **Inputs**: Scheduler provides views. For owned inputs, it must call
 34///   @ref holoflow::core::ITask::acquire_input() and place the returned view in
 35///   the context.
 36/// - **Execution**:
 37///   - Sync: scheduler calls @ref holoflow::core::ISyncTask::execute().
 38///   - Async push: scheduler calls @ref holoflow::core::IAsyncTask::try_push().
 39///   - Async pop: scheduler calls @ref holoflow::core::IAsyncTask::try_pop().
 40/// - **Outputs**: Task writes views to the context. For owned outputs the task
 41///   overwrites slots with owned views. After downstream use, the scheduler
 42///   must call @ref holoflow::core::ITask::release_output() per owned slot.
 43/// - **Cancellation**: cooperative via `cancelled`.
 44///
 45/// @section ownership Ownership
 46/// - Acquire/release apply only to indices marked owned by factory inference.
 47/// - Calling acquire/release on non-owned indices is undefined.
 48/// - At most one acquired input group and one unreleased output group exist at
 49///   a time under SPSC assumptions.
 50/// - @todo Define policy for discarding an acquired input on cancellation
 51///   before use (abort API vs implicit discard vs push-then-drop vs aggregation
 52///   commit).
 53///
 54/// @section errors Errors vs control flow
 55/// - Exceptions report validation/runtime errors (indices, shapes, allocation,
 56///   internal failures).
 57/// - @ref holoflow::core::OpResult is only for control flow: NotReady,
 58///   Cancelled, Eof.
 59///
 60/// @section concurrency Concurrency
 61/// - Assumed SPSC for async tasks. MPMC is undefined behavior.
 62/// - @todo Document/extend semantics if MPMC is added later.
 63
 64#pragma once
 65
 66#include <atomic>
 67#include <cuda_runtime.h>
 68#include <memory>
 69#include <nlohmann/json.hpp>
 70#include <optional>
 71#include <span>
 72#include <vector>
 73
 74#include "driver_types.h"
 75#include "holoflow/core/tensor.hh"
 76#include "holoflow_event/router.hh"
 77
 78namespace holoflow::core {
 79
 80/// Runtime execution context for a synchronous task.
 81struct SyncCtx {
 82  std::span<TView>             inputs;    ///< Scheduler-provided input views; some may be owned.
 83  std::span<TView>             outputs;   ///< Output slots; owned outputs are written by the task.
 84  std::atomic<bool>           *cancelled; ///< Non-null cancellation flag.
 85  holoflow_event::EventWriter *event_writer; ///< Event writer for emitting events.
 86  holoflow_event::EventReader *event_reader; ///< Event reader for receiving events.
 87};
 88
 89/// Runtime execution context for an asynchronous task push operation.
 90struct AsyncPushCtx {
 91  std::span<TView>   inputs;    ///< Scheduler-provided input views; some may be owned.
 92  std::atomic<bool> *cancelled; ///< Non-null cancellation flag.
 93};
 94
 95/// Runtime execution context for an asynchronous task pop operation.
 96struct AsyncPopCtx {
 97  std::span<TView>   outputs;   ///< Output slots; owned outputs are written by the task.
 98  std::atomic<bool> *cancelled; ///< Non-null cancellation flag.
 99};
 100
 101/// Possible expected outcomes of a task operation.
 102enum class OpResult : uint8_t {
 103  Ok,        ///< Completed successfully.
 104  NotReady,  ///< Nothing to do now; caller may retry.
 105  Cancelled, ///< Aborted due to cancellation.
 106  Eof,       ///< End of stream.
 107};
 108
 109class IOStorageAccess {
 110public:
 1111  virtual ~IOStorageAccess() = default;
 112
 113  [[nodiscard]] virtual Storage &owned_input_storage(size_t index)  = 0;
 114  [[nodiscard]] virtual Storage &owned_output_storage(size_t index) = 0;
 115};
 116
 117/// @brief Abstract base interface for tasks with optional tensor ownership.
 118///
 119/// Provides ownership hooks for inputs and outputs. Only indices declared as
 120/// owned by the task (via factory inference) may be acquired or released.
 121///
 122/// @warning Calling acquire/release on non-owned indices is undefined behavior.
 123/// @warning Not calling acquire/release on owned indices is undefined behavior.
 124///
 125/// @par Ownership lifecycle
 126/// - Inputs (owned):
 127///   - Acquire with @ref acquire_input(int) before execution or push.
 128///   - Publish the acquired pointer through @ref IOStorageAccess::owned_input_storage.
 129///   - The acquired view must not be used after the operation returns.
 130/// - Outputs (owned):
 131///   - Publish the output pointer through @ref IOStorageAccess::owned_output_storage.
 132///   - Context TViews share that stable Storage and observe pointer updates directly.
 133///   - After downstream consumption, scheduler calls @ref release_output(int).
 134///   - The task controls pointer cleanup and the lifetime of its owned memory.
 135/// @todo Define rollback semantics on cancellation before use.
 136class ITask {
 137public:
 1138  virtual ~ITask() = default;
 139
 140  /// Acquire a writable view for an **owned** input index.
 141  /// Valid only for indices marked owned by inference.
 142  /// The view must reference the compiler-provided Storage for that index.
 143  /// It signals readiness; scheduler contexts already share the same Storage.
 144  /// @returns view or std::nullopt if not currently acquirable.
 145  /// @throws std::out_of_range on bad index.
 146  [[nodiscard]] virtual std::optional<TView> acquire_input(int index);
 147
 148  /// Release a produced **owned** output at @p index after downstream use.
 149  /// @throws std::out_of_range on bad index.
 150  virtual void release_output(int index);
 151
 152  void bind_logger(std::shared_ptr<spdlog::logger> logger);
 153
 154  void bind_storage_access(IOStorageAccess *storage_access);
 155
 156protected:
 157  spdlog::logger *logger();
 158
 159  [[nodiscard]] IOStorageAccess &storage_access();
 160
 161private:
 162  std::shared_ptr<spdlog::logger> logger_         = nullptr;
 163  IOStorageAccess                *storage_access_ = nullptr;
 164};
 165
 166/// @brief Interface for synchronous (blocking) tasks.
 167///
 168/// A synchronous task consumes inputs and produces outputs within a single
 169/// blocking call to @ref execute().
 170///
 171/// @par Synchronous lifecycle
 172/// 1. Scheduler optionally acquires owned inputs via
 173///    @ref ITask::acquire_input(int).
 174/// 2. Scheduler calls @ref execute().
 175/// 3. Task may write owned output views into @ref SyncCtx::outputs,
 176///    overwriting the corresponding slots.
 177/// 4. Scheduler consumes outputs and calls @ref ITask::release_output(int)
 178///    for each owned output slot.
 179///
 180/// @note Inputs acquired via @ref ITask::acquire_input(int) must not be used
 181///       after @ref execute() returns.
 182class ISyncTask : public ITask {
 183public:
 1184  virtual ~ISyncTask() = default;
 185
 186  /// Execute in a blocking manner.
 187  /// Overwrites owned output slots in ctx.outputs.
 188  /// @returns control-flow result; errors via exceptions.
 189  [[nodiscard]] virtual OpResult execute(SyncCtx &ctx) = 0;
 190};
 191
 192/// @brief Interface for asynchronous (decoupled) tasks.
 193///
 194/// Asynchronous tasks split their interaction into a producer-side push and a
 195/// consumer-side pop. Push submits inputs. Pop retrieves available outputs.
 196///
 197/// @par Push lifecycle (producer)
 198/// 1. Scheduler optionally acquires owned inputs via
 199///    @ref ITask::acquire_input(int).
 200/// 2. Scheduler calls @ref try_push().
 201/// 3. Task consumes submitted inputs.
 202///
 203/// @par Pop lifecycle (consumer)
 204/// 1. Scheduler calls @ref try_pop().
 205/// 2. On success, task writes outputs, overwriting slots for owned outputs.
 206/// 3. Scheduler consumes outputs and calls @ref ITask::release_output(int) for
 207///    each owned output slot.
 208class IAsyncTask : public ITask {
 209public:
 1210  virtual ~IAsyncTask() = default;
 211
 212  /// Producer-side submission.
 213  /// @returns control-flow result; errors via exceptions.
 214  [[nodiscard]] virtual OpResult try_push(AsyncPushCtx &ctx) = 0;
 215
 216  /// Consumer-side retrieval.
 217  /// Overwrites owned output slots in ctx.outputs.
 218  /// @returns control-flow result; errors via exceptions.
 219  [[nodiscard]] virtual OpResult try_pop(AsyncPopCtx &ctx) = 0;
 220};
 221
 222/// Describes an in-place link between an input and output tensor.
 223struct InPlace {
 224  int in_idx;  ///< Input index reused/aliased.
 225  int out_idx; ///< Output index reusing input storage.
 226};
 227
 228/// Kind of task: synchronous or asynchronous.
 229enum class TaskKind {
 230  Sync,  /// Synchronous
 231  Async, /// Asynchronous
 232};
 233
 234/// Result of task inference from a factory infer function. This provides
 235/// information about the task's input and output tensor shapes and other
 236/// properties to the scheduling system / compiler.
 237struct InferResult {
 238  std::vector<TDesc>   input_descs;   ///< Inferred input tensor descriptions
 239  std::vector<TDesc>   output_descs;  ///< Inferred output tensor descriptions
 240  std::vector<InPlace> in_place;      ///< In-place input-output tensor pairs
 241  std::vector<bool>    owned_inputs;  ///< Ownership status of input tensors
 242  std::vector<bool>    owned_outputs; ///< Ownership status of output tensors
 243  TaskKind             kind;          ///< Kind of task (sync or async)
 244  /// Async producer capability. When true, try_push synchronizes its producer stream before any
 245  /// result that lets the scheduler advance. NotReady retries need not synchronize.
 246  bool synchronizes_producer_stream = false;
 247};
 248
 249/// Context for sync task creation.
 250struct SyncCreateCtx {
 251  cudaStream_t stream = static_cast<cudaStream_t>(0); ///< CUDA stream for task execution
 252};
 253
 254/// Context for async task creation.
 255struct AsyncCreateCtx {
 256  /// CUDA streams for producer side.
 257  cudaStream_t producer_stream = static_cast<cudaStream_t>(0);
 258  /// CUDA streams for consumer side.
 259  cudaStream_t consumer_stream = static_cast<cudaStream_t>(0);
 260};
 261
 262/// Base factory interface. Provides common inference API.
 263class ITaskFactory {
 264public:
 1265  virtual ~ITaskFactory() = default;
 266
 267  /// Infer metadata for a task without constructing it.
 268  ///
 269  /// @param input_descs  Upstream input tensor descriptors.
 270  /// @param jsettings    Task configuration (read-only view).
 271  /// @return             Inference result (I/O descs, ownership, in-place).
 272  /// @throws std::invalid_argument on inference failure.
 273  [[nodiscard]]
 274  virtual InferResult infer(std::span<const TDesc> input_descs,
 275                            const nlohmann::json  &jsettings) const = 0;
 276};
 277
 278/// Factory for synchronous tasks.
 279class ISyncTaskFactory : public ITaskFactory {
 280public:
 1281  virtual ~ISyncTaskFactory() = default;
 282
 283  /// Create a new synchronous task instance.
 284  ///
 285  /// @param input_descs  Upstream input tensor descriptors.
 286  /// @param jsettings    Task configuration.
 287  /// @param ctx          Runtime handles (sync stream).
 288  /// @return             New task ready for `ISyncTask::execute()`.
 289  virtual std::unique_ptr<ISyncTask> create(std::span<const TDesc> input_descs,
 290                                            const nlohmann::json  &jsettings,
 291                                            const SyncCreateCtx   &ctx) const = 0;
 292
 293  /// Update or replace an existing synchronous task.
 294  ///
 295  /// Use when inputs/settings change. Reuse internal allocations if possible.
 296  /// Otherwise, construct a replacement and return it.
 297  ///
 298  /// @param old_task     Existing task to reuse or replace.
 299  /// @param input_descs  New input descriptors.
 300  /// @param jsettings    New configuration.
 301  /// @param ctx          Runtime handles (sync stream).
 302  /// @return             Updated task. `old_task` must not be used afterwards.
 303  virtual std::unique_ptr<ISyncTask> update(std::unique_ptr<ISyncTask> old_task,
 304                                            std::span<const TDesc>     input_descs,
 305                                            const nlohmann::json      &jsettings,
 306                                            const SyncCreateCtx       &ctx) const;
 307};
 308
 309/// Factory for asynchronous tasks.
 310class IAsyncTaskFactory : public ITaskFactory {
 311public:
 1312  virtual ~IAsyncTaskFactory() = default;
 313
 314  /// Create a new asynchronous task instance.
 315  ///
 316  /// @param input_descs  Upstream input tensor descriptors.
 317  /// @param jsettings    Task configuration.
 318  /// @param ctx          Runtime handles (producer and consumer streams).
 319  /// @return             New task ready for `try_push()/try_pop()`
 320  virtual std::unique_ptr<IAsyncTask> create(std::span<const TDesc> input_descs,
 321                                             const nlohmann::json  &jsettings,
 322                                             const AsyncCreateCtx  &ctx) const = 0;
 323
 324  /// Update or replace an existing asynchronous task.
 325  ///
 326  /// @param old_task     Existing task to reuse or replace.
 327  /// @param input_descs  New input descriptors.
 328  /// @param jsettings    New configuration.
 329  /// @param ctx          Runtime handles (producer and consumer streams).
 330  /// @return             Updated task. `old_task` must not be used afterwards.
 331  virtual std::unique_ptr<IAsyncTask> update(std::unique_ptr<IAsyncTask> old_task,
 332                                             std::span<const TDesc>      input_descs,
 333                                             const nlohmann::json       &jsettings,
 334                                             const AsyncCreateCtx       &ctx) const;
 335};
 336
 337} // namespace holoflow::core

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\include\holoflow\core\tensor.hh

#LineLine coverage
 1// Copyright 2025 Digital Holography Foundation
 2//
 3// Licensed under the Apache License, Version 2.0 (the "License");
 4// you may not use this file except in compliance with the License.
 5// You may obtain a copy of the License at
 6//
 7//     http://www.apache.org/licenses/LICENSE-2.0
 8//
 9// Unless required by applicable law or agreed to in writing, software
 10// distributed under the License is distributed on an "AS IS" BASIS,
 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12// See the License for the specific language governing permissions and
 13// limitations under the License.
 14
 15#pragma once
 16
 17#include <cstdint>
 18#include <functional>
 19#include <memory>
 20#include <nlohmann/json.hpp>
 21#include <string_view>
 22#include <vector>
 23
 24#include "curaii/cuda.hh"
 25
 26namespace holoflow::core {
 27
 28/// Supported tensor element types.
 29enum class DType : uint8_t {
 30  U8,   ///< Unsigned 8-bit integer
 31  U16,  ///< Unsigned 16-bit integer
 32  F32,  ///< 32-bit float
 33  CF32, ///< 32-bit complex float
 34};
 35
 36/// Returns the size in bytes of a given DType.
 37[[nodiscard]] size_t size_of(DType dtype) noexcept;
 38
 39/// Returns a human-readable name for the DType.
 40[[nodiscard]] std::string_view to_string(DType dtype) noexcept;
 41
 42void to_json(nlohmann::json &j, DType dtype);
 43void from_json(const nlohmann::json &j, DType &dtype);
 44
 45/// Where memory lives: host (CPU) or device (GPU).
 46enum class MemLoc : uint8_t {
 47  Host,   ///< CPU-accessible memory
 48  Device, ///< GPU device memory
 49};
 50
 51/// Returns a human-readable name for the MemLoc.
 52[[nodiscard]] std::string_view to_string(MemLoc loc) noexcept;
 53
 54void to_json(nlohmann::json &j, MemLoc loc);
 55void from_json(const nlohmann::json &j, MemLoc &loc);
 56
 57struct Storage {
 58  MemLoc     mem_loc; ///< Memory location
 59  size_t     bytes;   ///< Size in bytes
 60  std::byte *ptr;     ///< Pointer to memory
 61};
 62
 63/// Describes a multi-dimensional array (tensor).
 64struct TDesc {
 65  std::vector<size_t> shape;      ///< The shape of the tensor (dimensions)
 66  DType               dtype;      ///< The data type of the tensor elements
 67  MemLoc              mem_loc;    ///< The memory location of the tensor
 68  std::vector<size_t> strides;    ///< The strides of the tensor (in bytes)
 169  size_t              offset = 0; ///< Byte offset from the start of the storage
 70
 171  TDesc() = default;
 72  TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc);
 73  TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc, std::vector<size_t> strides);
 74  TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc, size_t offset);
 75  TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc, std::vector<size_t> strides,
 76        size_t offset);
 77
 78  /// Returns the rank (number of dimensions) of the tensor.
 79  size_t rank() const noexcept;
 80
 81  /// Returns the total number of elements in the tensor.
 82  size_t num_elements() const;
 83
 84  /// Returns the total size in bytes of the tensor data.
 85  size_t num_bytes() const;
 86};
 87
 88void to_json(nlohmann::json &j, const TDesc &desc);
 89void from_json(const nlohmann::json &j, TDesc &desc);
 90
 91/// A non-owning view into tensor data.
 92struct TView {
 93  TDesc    desc    = {};      ///< Description of the tensor
 94  Storage *storage = nullptr; ///< Reference to the underlying storage
 95
 96  std::byte *data();
 97  bool       is_nullptr();
 98};
 99
 100/// A multi-dimensional array (tensor) holding data in either host or device
 101/// memory.
 102class Tensor {
 103public:
 104  /// Constructs a Tensor with the given descriptor. Allocates memory
 105  /// according to the descriptor's memory location.
 106  explicit Tensor(const TDesc &desc);
 107
 108  /// Returns a mutable pointer to the tensor data.
 109  [[nodiscard]] void *data() noexcept;
 110
 111  /// Returns a constant pointer to the tensor data.
 112  [[nodiscard]] const void *data() const noexcept;
 113
 114  /// Returns the tensor descriptor.
 115  [[nodiscard]] const TDesc &desc() const noexcept;
 116
 117  /// Returns a non-owning mutable view into the tensor data.
 118  [[nodiscard]] TView view() noexcept;
 119
 120private:
 121  using HData = curaii::unique_host_ptr<std::byte>;
 122  using DData = curaii::unique_device_ptr<std::byte>;
 123
 124  TDesc                    desc_;    /// Descriptor of the tensor
 125  HData                    h_data_;  ///< Host memory (if applicable)
 126  DData                    d_data_;  ///< Device memory (if applicable)
 127  std::byte               *data_;    ///< Raw pointer to the tensor data
 128  std::unique_ptr<Storage> storage_; ///< Underlying storage information
 129};
 130
 131} // namespace holoflow::core

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\include\holoflow\runtime\graph_exec.hh

#LineLine coverage
 1// Copyright 2025 Digital Holography Foundation
 2//
 3// Licensed under the Apache License, Version 2.0 (the "License");
 4// you may not use this file except in compliance with the License.
 5// You may obtain a copy of the License at
 6//
 7//     http://www.apache.org/licenses/LICENSE-2.0
 8//
 9// Unless required by applicable law or agreed to in writing, software
 10// distributed under the License is distributed on an "AS IS" BASIS,
 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12// See the License for the specific language governing permissions and
 13// limitations under the License.
 14
 15#pragma once
 16
 17#include <atomic>
 18#include <boost/graph/adjacency_list.hpp>
 19#include <chrono>
 20#include <condition_variable>
 21#include <cstddef>
 22#include <cstdint>
 23#include <cuda_runtime.h>
 24#include <map>
 25#include <memory>
 26#include <mutex>
 27#include <optional>
 28#include <span>
 29#include <string>
 30#include <thread>
 31#include <utility>
 32#include <variant>
 33#include <vector>
 34
 35#include "curaii/cuda.hh"
 36#include "driver_types.h"
 37#include "holoflow/core/graph_spec.hh"
 38#include "holoflow/core/tasks.hh"
 39#include "holoflow/core/tensor.hh"
 40#include "holoflow_event/router.hh"
 41
 42namespace holoflow::runtime {
 43
 44struct NodePlan {
 45  core::NodeSpec    spec;     ///< Node specification.
 46  core::InferResult infer;    ///< Inference metadata.
 47  std::vector<int>  in_tids;  ///< Input tensor IDs.
 48  std::vector<int>  out_tids; ///< Output tensor IDs.
 49};
 50
 51struct EdgePlan {
 52  core::EdgeSpec spec; ///< Edge specification.
 53  core::TDesc    desc; ///< Tensor descriptor.
 54  int            tid;  ///< Tensor ID.
 55};
 56
 57using GraphPlan = boost::adjacency_list<boost::vecS,           // OutEdgeList
 58                                        boost::vecS,           // VertexList
 59                                        boost::bidirectionalS, // Directed graph
 60                                        NodePlan,              // Vertex properties
 61                                        EdgePlan               // Edge properties
 62                                        >;
 63
 64/// Represents a block of memory used during graph execution.
 65struct MemoryBlock {
 66  core::MemLoc mem_loc;    ///< Memory location (host or device).
 67  size_t       size_bytes; ///< Size of the memory block in bytes.
 68
 69  curaii::unique_host_ptr<std::byte>   h_data; ///< Host memory (if applicable).
 70  curaii::unique_device_ptr<std::byte> d_data; ///< Device memory (if applicable).
 71
 72  void *get(); ///< Returns a mutable pointer to the memory block.
 73};
 74
 75struct ExecResouces {
 76  std::map<size_t, MemoryBlock>                    memory_blocks; ///< StorageID -> MemoryBlock.
 77  std::map<size_t, std::unique_ptr<core::Storage>> storages;      ///< StorageID -> Storage.
 78  std::map<size_t, core::TDesc>                    tensor_descs;  ///< TensorID -> TDesc.
 79  std::map<size_t, size_t>                         tid_to_sid;    ///< TensorID -> StorageID.
 80
 81  std::map<std::string, std::unique_ptr<core::IOStorageAccess>> node_storage_adapters;
 82  std::map<size_t, curaii::CudaStream>                          streams; ///< CUDA streams by ID.
 83  std::map<std::string, std::unique_ptr<core::ITask>>           tasks;   ///< Task instances by ID.
 84  // std::map<int, core::Tensor>                         tensors; ///< Allocated tensors by ID.
 85};
 86
 87struct Section {
 88  int                                       id;         ///< Section ID.
 89  std::string                               name;       ///< Section name (for logging).
 90  cudaStream_t                              stream;     ///< CUDA stream for this section.
 91  std::vector<GraphPlan::vertex_descriptor> sync_topo;  ///< Synchronous nodes in topological order.
 92  std::vector<GraphPlan::vertex_descriptor> async_cons; ///< Asynchronous consumer nodes.
 93  std::vector<GraphPlan::vertex_descriptor> async_prod; ///< Asynchronous producer nodes
 94  bool has_synchronizing_async_producer = false; ///< Producer supplies the section stream barrier.
 95};
 96
 97struct SyncRt {
 98  core::ISyncTask         *task = nullptr;
 99  std::vector<core::TView> in_views;
 100  std::vector<core::TView> out_views;
 101  core::SyncCtx            ctx{};
 102};
 103
 104struct AsyncRt {
 105  core::IAsyncTask        *task = nullptr;
 106  std::vector<core::TView> in_views;
 107  std::vector<core::TView> out_views;
 108  core::AsyncPushCtx       pctx{};
 109  core::AsyncPopCtx        xctx{};
 110};
 111
 112using NodeRt = std::variant<SyncRt, AsyncRt>;
 113
 114struct NodeMetrics {
 115  double   average_duration_ms                = 0.0;
 116  double   runs_per_second                    = 0.0;
 117  double   host_throughput_bytes_per_second   = 0.0;
 118  double   device_throughput_bytes_per_second = 0.0;
 119  uint64_t sample_count                       = 0;
 120};
 121
 122class Scheduler {
 123public:
 124  Scheduler(const GraphPlan &graph, const std::vector<Section> &sections, ExecResouces &res,
 125            std::chrono::milliseconds metrics_interval = std::chrono::milliseconds{1000});
 126
 127  ~Scheduler();
 128
 129  void set_metrics_interval(std::chrono::milliseconds interval);
 130  [[nodiscard]] std::map<std::string, NodeMetrics> metrics() const;
 131
 132  void start();
 133  void request_stop();
 134  void wait();
 135
 136  bool is_running() const;
 137  bool stop_requested() const;
 138
 139  [[nodiscard]] bool ui_try_send(const std::string &node_id, nlohmann::json &&data) noexcept;
 140
 141  [[nodiscard]] std::optional<holoflow_event::Event> ui_try_receive() noexcept;
 142
 143private:
 144  void init_tviews();
 145  void build_event_handles();
 146  void build_nodes_rts();
 147  void reset_metrics_state();
 148  void start_metrics_thread();
 149  void stop_metrics_thread();
 150  void metrics_loop();
 151  void aggregate_metrics(double interval_seconds);
 152  void record_node_sample(std::size_t idx, uint64_t duration_ns, uint64_t host_bytes,
 153                          uint64_t device_bytes);
 154  static std::pair<uint64_t, uint64_t> sum_bytes(std::span<const core::TView> views);
 155
 156  void run_router();
 157
 158  void run_section(int section_id);
 159
 160  /// This function acquires all owned inputs for the given node.
 161  /// Owning tasks publish memory by updating their compiler-provided Storage;
 162  /// all scheduler TViews retain pointers to that stable Storage object.
 163  /// This function blocks until all owned inputs are acquired.
 164  /// @warning If stop_ is set while waiting, the function returns early,
 165  /// and some owned inputs may not be acquired.
 166  /// @warning This function must be called on a synchronous or asynchronous
 167  /// producer node only.
 168  void acquire_owned_inputs(GraphPlan::vertex_descriptor v);
 169
 170  /// This function releases all owned outputs for the given node.
 171  /// Pointer cleanup remains the owning task's responsibility.
 172  /// This function does not block.
 173  void release_owned_outputs(GraphPlan::vertex_descriptor v);
 174
 175  /// Executes a synchronous node.
 176  /// @warning This function must be called on a synchronous node only.
 177  [[nodiscard]] core::OpResult run_sync(GraphPlan::vertex_descriptor v);
 178
 179  /// Executes an asynchronous consumer node.
 180  /// @warning This function must be called on an asynchronous consumer node only.
 181  [[nodiscard]] core::OpResult run_async_cons(GraphPlan::vertex_descriptor v);
 182
 183  /// Executes an asynchronous producer node.
 184  /// @warning This function must be called on an asynchronous producer node only.
 185  [[nodiscard]] core::OpResult run_async_prod(GraphPlan::vertex_descriptor v);
 186
 187private:
 1188  std::atomic<bool>           running_{false}; ///< True if the scheduler is running.
 1189  std::atomic<bool>           stop_{false};    ///< True if a stop has been requested.
 190  const GraphPlan            &graph_;          ///< The computational graph to execute.
 191  const std::vector<Section> &sections_;       ///< Execution sections.
 192  ExecResouces               &res_;            ///< Execution resources (streams, tasks, tensors).
 193
 194  /// Stable TViews for all tensors by their IDs. Copies across node contexts
 195  /// observe ownership changes through their shared Storage pointers.
 196  std::vector<core::TView> tviews_;
 197
 198  std::vector<NodeRt>      node_rts_; ///< Runtime data for each node.
 199  std::vector<std::string> node_names_;
 200
 201  struct NodeMetricAccumulator {
 1202    std::atomic<uint64_t> duration_ns{0};
 1203    std::atomic<uint64_t> run_count{0};
 1204    std::atomic<uint64_t> host_bytes{0};
 1205    std::atomic<uint64_t> device_bytes{0};
 206
 1207    NodeMetricAccumulator() = default;
 208
 0209    NodeMetricAccumulator(const NodeMetricAccumulator &other) {
 0210      duration_ns.store(other.duration_ns.load(std::memory_order_relaxed),
 211                        std::memory_order_relaxed);
 0212      run_count.store(other.run_count.load(std::memory_order_relaxed), std::memory_order_relaxed);
 0213      host_bytes.store(other.host_bytes.load(std::memory_order_relaxed), std::memory_order_relaxed);
 0214      device_bytes.store(other.device_bytes.load(std::memory_order_relaxed),
 215                         std::memory_order_relaxed);
 0216    }
 217
 218    NodeMetricAccumulator &operator=(const NodeMetricAccumulator &other) {
 219      if (this == &other) {
 220        return *this;
 221      }
 222      duration_ns.store(other.duration_ns.load(std::memory_order_relaxed),
 223                        std::memory_order_relaxed);
 224      run_count.store(other.run_count.load(std::memory_order_relaxed), std::memory_order_relaxed);
 225      host_bytes.store(other.host_bytes.load(std::memory_order_relaxed), std::memory_order_relaxed);
 226      device_bytes.store(other.device_bytes.load(std::memory_order_relaxed),
 227                         std::memory_order_relaxed);
 228      return *this;
 229    }
 230
 0231    NodeMetricAccumulator(NodeMetricAccumulator &&other) noexcept : NodeMetricAccumulator(other) {}
 232
 233    NodeMetricAccumulator &operator=(NodeMetricAccumulator &&other) noexcept {
 234      return (*this = other);
 235    }
 236  };
 237
 238  std::vector<NodeMetricAccumulator> metric_accumulators_;
 239
 240  mutable std::mutex                 metrics_mutex_;
 241  std::map<std::string, NodeMetrics> latest_metrics_;
 242  std::chrono::milliseconds          metrics_interval_;
 1243  std::atomic<bool>                  metrics_running_{false};
 244  std::thread                        metrics_thread_;
 245  std::condition_variable            metrics_cv_;
 246  mutable std::mutex                 metrics_thread_mutex_;
 247
 248  std::vector<std::thread> threads_; ///< Threads for each section.
 249
 250  holoflow_event::Router                                     router_;
 251  std::map<std::string, holoflow_event::Router::NodeHandles> event_handles_;
 252};
 253
 254} // namespace holoflow::runtime
 255
 256template <> struct fmt::formatter<holoflow::runtime::NodeMetrics> {
 257  constexpr auto parse(format_parse_context &ctx) { return ctx.begin(); }
 258
 259  template <typename FormatContext>
 260  auto format(const holoflow::runtime::NodeMetrics &m, FormatContext &ctx) const {
 261    return fmt::format_to(
 262        ctx.out(),
 263        "{{avg: {:.3f} ms, rps: {:.3f}, host: {:.3f} B/s, device: {:.3f} B/s, samples: {}}}",
 264        m.average_duration_ms, m.runs_per_second, m.host_throughput_bytes_per_second,
 265        m.device_throughput_bytes_per_second, m.sample_count);
 266  }
 267};
Generated by: ReportGenerator 5.5.10.0
9/16/2026 - 1:44:14 PM
GitHub | reportgenerator.io

Methods/Properties