< Summary

Information
Class: src
Assembly: holoflow
File(s): File 1: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\core\graph_spec.cc
File 2: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\core\registry.cc
File 3: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\core\tasks.cc
File 4: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\core\tensor.cc
File 5: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\logger.cc
File 6: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\runtime\compiler.cc
File 7: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\runtime\graph_display.cc
File 8: C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\runtime\graph_exec.cc
Line coverage
86%
Covered lines: 1468
Uncovered lines: 230
Coverable lines: 1698
Total lines: 3227
Line coverage: 86.4%
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\src\core\graph_spec.cc

#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#include "holoflow/core/graph_spec.hh"
 16
 17#include <algorithm>
 18#include <boost/graph/adjacency_list.hpp>
 19#include <boost/graph/graph_traits.hpp>
 20#include <cctype>
 21#include <stdexcept>
 22#include <string>
 23#include <unordered_map>
 24#include <utility>
 25#include <vector>
 26
 27#include "bug.hh"
 28
 29namespace holoflow::core {
 30
 31namespace {
 32
 133[[noreturn]] void json_error(const std::string &msg) {
 134  throw std::runtime_error("GraphSpec JSON: " + msg);
 035}
 36
 137inline void require(bool cond, const std::string &msg) {
 138  if (!cond)
 139    json_error(msg);
 140}
 41
 142inline bool has_object_key(const nlohmann::json &j, const char *key) {
 143  return j.is_object() && j.contains(key) && !j.at(key).is_null();
 144}
 45
 146inline std::vector<std::string> sorted_object_keys(const nlohmann::json &obj) {
 147  std::vector<std::string> keys;
 148  keys.reserve(obj.size());
 149  for (auto it = obj.begin(); it != obj.end(); ++it)
 150    keys.push_back(it.key());
 151  std::ranges::sort(keys);
 152  return keys;
 153}
 54
 55} // namespace
 56
 157nlohmann::json to_json(const GraphSpec &g, const GraphSpecWriteOptions &opts) {
 58  // clang-format off
 159  HOLOFLOW_CHECK(opts.include_node_names, "include_node_names=false not supported in this format");
 160  HOLOFLOW_CHECK(opts.include_node_kinds, "include_node_kinds=false not supported in this format");
 161  HOLOFLOW_CHECK(opts.include_node_settings, "include_node_settings=false not supported in this format");
 162  HOLOFLOW_CHECK(opts.include_edge_indices, "include_edge_indices=false not supported in this format");
 63  // clang-format on
 64
 165  nlohmann::json j;
 166  j["nodes"] = nlohmann::json::object();
 167  j["edges"] = nlohmann::json::array();
 68
 69  // Nodes
 170  for (auto vd : boost::make_iterator_range(vertices(g))) {
 171    const auto &n = g[vd];
 172    require(!n.name.empty(), "node has empty name");
 173    nlohmann::json node_obj = nlohmann::json::object();
 74
 175    if (opts.include_node_kinds) {
 176      node_obj["type"] = n.kind;
 77    }
 78
 179    if (opts.include_node_settings) {
 180      node_obj["params"] = n.settings.is_null() ? nlohmann::json::object() : n.settings;
 81    }
 82
 183    if (!n.debug) {
 184      node_obj["debug"] = false;
 85    }
 86
 187    const std::string key = n.name;
 188    j["nodes"][key]       = std::move(node_obj);
 189  }
 90
 91  // Edges
 192  for (auto ed : boost::make_iterator_range(edges(g))) {
 193    const auto  src = source(ed, g);
 194    const auto  dst = target(ed, g);
 195    const auto &e   = g[ed];
 96
 197    const auto &src_name = g[src].name;
 198    const auto &dst_name = g[dst].name;
 99
 1100    require(!src_name.empty() && !dst_name.empty(), "edge connects unnamed node(s)");
 101
 1102    nlohmann::json edge_obj;
 1103    edge_obj["from"] = src_name;
 1104    edge_obj["to"]   = dst_name;
 1105    edge_obj["out"]  = e.out_idx;
 1106    edge_obj["in"]   = e.in_idx;
 1107    j["edges"].push_back(std::move(edge_obj));
 1108  }
 109
 1110  return j;
 1111}
 112
 1113GraphSpec from_json(const nlohmann::json &j) {
 1114  require(j.is_object(), "root must be an object");
 1115  require(has_object_key(j, "nodes"), "missing required key 'nodes'");
 1116  require(j.at("nodes").is_object(), "'nodes' must be an object");
 117
 1118  const auto &jnodes = j.at("nodes");
 1119  const auto &jedges = j.contains("edges") ? j.at("edges") : nlohmann::json::array();
 1120  require(jedges.is_array(), "'edges' must be an array if present");
 121
 1122  GraphSpec g;
 123  using Vertex = boost::graph_traits<GraphSpec>::vertex_descriptor;
 1124  std::unordered_map<std::string, Vertex> name_to_v;
 1125  name_to_v.reserve(jnodes.size());
 126
 127  // Deterministic vertex numbering: sort node keys.
 1128  for (const auto &name : sorted_object_keys(jnodes)) {
 1129    require(!name.empty(), "node key (name) must not be empty");
 130
 1131    const auto &node_json = jnodes.at(name);
 1132    require(node_json.is_object(), "node '" + name + "' must be an object");
 1133    require(has_object_key(node_json, "type"), "node '" + name + "': missing 'type'");
 1134    require(node_json.at("type").is_string(), "node '" + name + "': 'type' must be a string");
 1135    require(node_json.contains("params"), "node '" + name + "': missing 'params'");
 136
 1137    NodeSpec spec;
 1138    spec.name = name;
 1139    spec.kind = node_json.at("type").get<std::string>();
 140
 1141    spec.settings = node_json.at("params");
 1142    require(spec.settings.is_object() || spec.settings.is_array() || spec.settings.is_primitive() ||
 143                spec.settings.is_null(),
 144            "node '" + name + "': 'params' must be valid JSON");
 145
 1146    if (spec.settings.is_null())
 1147      spec.settings = nlohmann::json::object();
 148
 1149    if (node_json.contains("debug")) {
 1150      require(node_json.at("debug").is_boolean(), "node '" + name + "': 'debug' must be boolean");
 1151      spec.debug = node_json.at("debug").get<bool>();
 1152    } else {
 1153      spec.debug = true;
 154    }
 155
 1156    const auto v              = add_vertex(std::move(spec), g);
 1157    const auto [it, inserted] = name_to_v.emplace(name, v);
 1158    require(inserted, "duplicate node name '" + name + "'");
 1159  }
 160
 161  // Add edges
 1162  for (std::size_t i = 0; i < jedges.size(); ++i) {
 1163    const auto &e = jedges.at(i);
 1164    require(e.is_object(), "edge[" + std::to_string(i) + "] must be an object");
 1165    require(has_object_key(e, "from"), "edge[" + std::to_string(i) + "]: missing 'from'");
 1166    require(has_object_key(e, "to"), "edge[" + std::to_string(i) + "]: missing 'to'");
 1167    require(e.at("from").is_string(), "edge[" + std::to_string(i) + "]: 'from' must be string");
 1168    require(e.at("to").is_string(), "edge[" + std::to_string(i) + "]: 'to' must be string");
 169
 1170    const auto from    = e.at("from").get<std::string>();
 1171    const auto to      = e.at("to").get<std::string>();
 1172    const auto it_from = name_to_v.find(from);
 1173    const auto it_to   = name_to_v.find(to);
 174
 1175    require(it_from != name_to_v.end(),
 176            "edge[" + std::to_string(i) + "]: unknown 'from' node '" + from + "'");
 1177    require(it_to != name_to_v.end(),
 178            "edge[" + std::to_string(i) + "]: unknown 'to' node '" + to + "'");
 179
 1180    EdgeSpec es{};
 1181    require(e.at("out").is_number_integer(), "edge[" + std::to_string(i) + "]: 'out' must be int");
 1182    require(e.at("in").is_number_integer(), "edge[" + std::to_string(i) + "]: 'in' must be int");
 1183    es.out_idx = e.at("out").get<int>();
 1184    es.in_idx  = e.at("in").get<int>();
 1185    add_edge(it_from->second, it_to->second, es, g);
 1186  }
 187
 1188  return g;
 1189}
 190} // namespace holoflow::core

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\core\registry.cc

#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#include "holoflow/core/registry.hh"
 16
 17#include <stdexcept>
 18#include <utility>
 19
 20namespace holoflow::core {
 21
 122void Registry::register_sync(const Key &kind, SyncPtr factory) {
 123  if (!factory) {
 124    throw std::invalid_argument("null sync factory");
 25  }
 126  if (sync_factories_.find(kind) != sync_factories_.end() ||
 27      async_factories_.find(kind) != async_factories_.end()) {
 128    throw std::invalid_argument("factory kind already registered: " + kind);
 29  }
 130  sync_factories_.emplace(kind, std::move(factory));
 131}
 32
 133void Registry::register_async(const Key &kind, AsyncPtr factory) {
 134  if (!factory) {
 135    throw std::invalid_argument("null async factory");
 36  }
 137  if (sync_factories_.find(kind) != sync_factories_.end() ||
 38      async_factories_.find(kind) != async_factories_.end()) {
 139    throw std::invalid_argument("factory kind already registered: " + kind);
 40  }
 141  async_factories_.emplace(kind, std::move(factory));
 142}
 43
 144const ISyncTaskFactory &Registry::get_sync(const Key &kind) const {
 145  auto it = sync_factories_.find(kind);
 146  if (it == sync_factories_.end()) {
 147    throw std::out_of_range("unknown sync kind: " + kind);
 48  }
 149  return *(it->second);
 150}
 51
 152const IAsyncTaskFactory &Registry::get_async(const Key &kind) const {
 153  auto it = async_factories_.find(kind);
 154  if (it == async_factories_.end()) {
 155    throw std::out_of_range("unknown async kind: " + kind);
 56  }
 157  return *(it->second);
 158}
 59
 160const ITaskFactory &Registry::get(const Key &kind) const {
 161  if (is_sync_registered(kind)) {
 162    return get_sync(kind);
 63  }
 164  if (is_async_registered(kind)) {
 165    return get_async(kind);
 66  }
 167  throw std::out_of_range("unknown kind: " + kind);
 168}
 69
 170bool Registry::is_sync_registered(const Key &kind) const noexcept {
 171  return sync_factories_.find(kind) != sync_factories_.end();
 172}
 73
 174bool Registry::is_async_registered(const Key &kind) const noexcept {
 175  return async_factories_.find(kind) != async_factories_.end();
 176}
 77
 178bool Registry::is_registered(const Key &kind) const noexcept {
 179  return is_sync_registered(kind) || is_async_registered(kind);
 180}
 81
 82} // namespace holoflow::core

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

#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#include "holoflow/core/tasks.hh"
 16
 17#include <optional>
 18
 19#include "bug.hh"
 20#include "holoflow/core/tensor.hh"
 21
 22namespace holoflow::core {
 23
 124std::optional<TView> ITask::acquire_input(int) {
 125  throw std::out_of_range("Input index out of range");
 026}
 27
 128void ITask::release_output(int) { throw std::out_of_range("Output index out of range"); }
 29
 130void ITask::bind_logger(std::shared_ptr<spdlog::logger> logger) {
 131  HOLOFLOW_CHECK(logger != nullptr, "Cannot bind null logger to task");
 132  logger_ = std::move(logger);
 133}
 34
 135void ITask::bind_storage_access(IOStorageAccess *storage_access) {
 136  HOLOFLOW_CHECK(storage_access != nullptr, "Cannot bind null storage access to task");
 137  storage_access_ = storage_access;
 138}
 39
 140spdlog::logger *ITask::logger() {
 141  HOLOFLOW_CHECK(logger_, "Logger not bound to task");
 142  return logger_.get();
 143}
 44
 145[[nodiscard]] IOStorageAccess &ITask::storage_access() {
 146  HOLOFLOW_CHECK(storage_access_ != nullptr, "Storage access not bound to task");
 147  return *storage_access_;
 148}
 49
 50std::unique_ptr<ISyncTask> ISyncTaskFactory::update(std::unique_ptr<ISyncTask>,
 51                                                    std::span<const TDesc> input_descs,
 52                                                    const nlohmann::json  &jsettings,
 153                                                    const SyncCreateCtx   &ctx) const {
 154  return create(input_descs, jsettings, ctx);
 155}
 56
 57std::unique_ptr<IAsyncTask> IAsyncTaskFactory::update(std::unique_ptr<IAsyncTask>,
 58                                                      std::span<const TDesc> input_descs,
 59                                                      const nlohmann::json  &jsettings,
 160                                                      const AsyncCreateCtx  &ctx) const {
 161  return create(input_descs, jsettings, ctx);
 162}
 63
 64} // namespace holoflow::core

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

#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#include "holoflow/core/tensor.hh"
 16
 17#include <limits>
 18#include <stdexcept>
 19
 20#include "bug.hh"
 21
 22namespace holoflow::core {
 23
 124size_t size_of(DType dtype) noexcept {
 125  switch (dtype) {
 26  case DType::U8:
 127    return 1;
 28  case DType::U16:
 129    return 2;
 30  case DType::F32:
 131    return 4;
 32  case DType::CF32:
 133    return 8;
 34  }
 35
 036  HOLOFLOW_UNREACHABLE();
 137}
 38
 139std::string_view to_string(DType dtype) noexcept {
 140  switch (dtype) {
 41  case DType::U8:
 142    return "U8";
 43  case DType::U16:
 144    return "U16";
 45  case DType::F32:
 146    return "F32";
 47  case DType::CF32:
 148    return "CF32";
 49  }
 50
 051  HOLOFLOW_UNREACHABLE();
 152}
 53
 154void to_json(nlohmann::json &j, DType dtype) { j = to_string(dtype); }
 55
 156void from_json(const nlohmann::json &j, DType &dtype) {
 157  std::string_view s = j.get<std::string_view>();
 158  if (s == "U8") {
 159    dtype = DType::U8;
 160  } else if (s == "U16") {
 161    dtype = DType::U16;
 162  } else if (s == "F32") {
 163    dtype = DType::F32;
 164  } else if (s == "CF32") {
 165    dtype = DType::CF32;
 166  } else {
 167    throw std::invalid_argument("invalid DType string");
 68  }
 169}
 70
 171std::string_view to_string(MemLoc loc) noexcept {
 172  switch (loc) {
 73  case MemLoc::Host:
 174    return "Host";
 75  case MemLoc::Device:
 176    return "Device";
 77  }
 78
 079  HOLOFLOW_UNREACHABLE();
 180}
 81
 182void to_json(nlohmann::json &j, MemLoc loc) { j = to_string(loc); }
 83
 184void from_json(const nlohmann::json &j, MemLoc &loc) {
 185  std::string_view s = j.get<std::string_view>();
 186  if (s == "Host") {
 187    loc = MemLoc::Host;
 188  } else if (s == "Device") {
 189    loc = MemLoc::Device;
 190  } else {
 191    throw std::invalid_argument("invalid MemLoc string");
 92  }
 193}
 94
 195void to_json(nlohmann::json &j, const TDesc &desc) {
 196  j = nlohmann::json{
 97      {"shape", desc.shape},
 98      {"dtype", desc.dtype},
 99      {"mem_loc", desc.mem_loc},
 100      {"strides", desc.strides},
 101  };
 1102}
 103
 1104void from_json(const nlohmann::json &j, TDesc &desc) {
 1105  j.at("shape").get_to(desc.shape);
 1106  j.at("dtype").get_to(desc.dtype);
 1107  j.at("mem_loc").get_to(desc.mem_loc);
 1108  j.at("strides").get_to(desc.strides);
 1109}
 110
 111namespace {
 1112std::vector<std::size_t> make_default_strides(const std::vector<std::size_t> &shape, DType dtype) {
 1113  std::vector<std::size_t> strides(shape.size());
 114
 1115  std::size_t stride = size_of(dtype);
 1116  for (std::size_t i = shape.size(); i-- > 0;) {
 1117    strides[i] = stride;
 1118    stride *= shape[i];
 1119  }
 120
 1121  return strides;
 1122}
 123} // namespace
 124
 125TDesc::TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc)
 1126    : shape(std::move(shape)), dtype(dtype), mem_loc(mem_loc),
 1127      strides(make_default_strides(this->shape, this->dtype)), offset(0) {}
 128
 129TDesc::TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc, std::vector<size_t> strides)
 1130    : shape(std::move(shape)), dtype(dtype), mem_loc(mem_loc), strides(std::move(strides)),
 1131      offset(0) {}
 132
 133TDesc::TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc, size_t offset)
 1134    : shape(std::move(shape)), dtype(dtype), mem_loc(mem_loc),
 1135      strides(make_default_strides(this->shape, this->dtype)), offset(offset) {}
 136
 137TDesc::TDesc(std::vector<size_t> shape, DType dtype, MemLoc mem_loc, std::vector<size_t> strides,
 138             size_t offset)
 1139    : shape(std::move(shape)), dtype(dtype), mem_loc(mem_loc), strides(std::move(strides)),
 1140      offset(offset) {}
 141
 1142size_t TDesc::rank() const noexcept { return shape.size(); }
 143
 1144size_t TDesc::num_elements() const {
 1145  constexpr size_t max = std::numeric_limits<size_t>::max();
 1146  size_t           n   = 1;
 1147  for (auto d : shape) {
 1148    if (d == 0)
 1149      return 0;
 1150    if (n > max / d) {
 1151      throw std::overflow_error("num_elements overflow");
 152    }
 1153    n *= d;
 1154  }
 1155  return n;
 1156}
 157
 1158size_t TDesc::num_bytes() const {
 1159  if (shape.empty()) {
 1160    return 0;
 161  }
 162
 1163  return strides[0] * shape[0];
 1164}
 165
 1166std::byte *TView::data() {
 1167  HOLOFLOW_CHECK(storage != nullptr, "TView has null storage");
 1168  HOLOFLOW_CHECK(storage->ptr != nullptr, "TView has null data pointer");
 1169  return storage->ptr + desc.offset;
 1170}
 171
 1172bool TView::is_nullptr() { return storage == nullptr || storage->ptr == nullptr; }
 173
 1174Tensor::Tensor(const TDesc &desc) : desc_(desc), data_(nullptr) {
 1175  switch (desc_.mem_loc) {
 176  case MemLoc::Host:
 1177    h_data_ = curaii::make_unique_host_ptr<std::byte>(desc_.num_bytes());
 1178    data_   = h_data_.get();
 1179    break;
 180  case MemLoc::Device:
 1181    d_data_ = curaii::make_unique_device_ptr<std::byte>(desc_.num_bytes());
 1182    data_   = d_data_.get();
 183    break;
 184  }
 185
 1186  storage_ = std::make_unique<Storage>(Storage{desc_.mem_loc, desc_.num_bytes(), data_});
 1187}
 188
 1189void *Tensor::data() noexcept { return data_; }
 190
 1191const void *Tensor::data() const noexcept { return data_; }
 192
 1193const TDesc &Tensor::desc() const noexcept { return desc_; }
 194
 1195TView Tensor::view() noexcept { return {desc_, storage_.get()}; }
 196
 197} // namespace holoflow::core

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\logger.cc

#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#include "logger.hh"
 16
 17#include <spdlog/sinks/stdout_color_sinks.h>
 18#include <spdlog/spdlog.h>
 19
 20namespace holoflow {
 21
 122std::shared_ptr<spdlog::logger> logger() {
 123  static std::shared_ptr<spdlog::logger> logger = [] {
 24    auto sink = std::make_shared<spdlog::sinks::stdout_color_sink_mt>();
 25    sink->set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%n] [thread %t] [%^%l%$] %v");
 26
 27    auto log = std::make_shared<spdlog::logger>("holoflow", sink);
 28    log->set_level(spdlog::default_logger()->level());
 29    log->flush_on(spdlog::level::warn);
 30
 31    spdlog::register_logger(log);
 32
 33    return log;
 134  }();
 135  return logger;
 136}
 37
 38} // namespace holoflow

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\runtime\compiler.cc

#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#include "holoflow/runtime/compiler.hh"
 16
 17#include <boost/graph/adjacency_list.hpp>
 18#include <boost/graph/breadth_first_search.hpp>
 19#include <boost/graph/topological_sort.hpp>
 20#include <chrono>
 21#include <format>
 22#include <fstream>
 23#include <iostream>
 24#include <mutex>
 25#include <numeric>
 26#include <nvtx3/nvtx3.hpp>
 27#include <queue>
 28#include <ranges>
 29#include <set>
 30#include <stack>
 31#include <thread>
 32#include <type_traits>
 33#include <unordered_map>
 34#include <unordered_set>
 35#include <variant>
 36
 37#include "curaii/cuda.hh"
 38#include "holoflow/core/graph_spec.hh"
 39#include "holoflow/core/registry.hh"
 40#include "holoflow/core/tasks.hh"
 41#include "holoflow/core/tensor.hh"
 42#include "holoflow/runtime/graph_display.hh"
 43#include "holoflow/runtime/graph_exec.hh"
 44#include "spdlog/sinks/basic_file_sink.h"
 45#include "spdlog/sinks/stdout_color_sinks.h"
 46#include "spdlog/spdlog.h"
 47
 48namespace holoflow::runtime {
 49
 50// -------------------------------------------------------------------------------------------------
 51// Internal Types & Error Handling
 52// -------------------------------------------------------------------------------------------------
 53
 54class CompilerException : public std::runtime_error {
 55public:
 56  using std::runtime_error::runtime_error;
 57};
 58
 59// -------------------------------------------------------------------------------------------------
 60// Profiling Data Structures
 61// -------------------------------------------------------------------------------------------------
 62
 63struct TraceEvent {
 64  std::string name;
 65  std::string category;
 66  long long   start_us;
 67  long long   dur_us;
 68  uint32_t    tid;
 69};
 70
 71class CompilationProfiler {
 72public:
 173  void add_event(std::string name, std::string category, long long start_us, long long dur_us) {
 174    std::lock_guard<std::mutex> lock(mutex_);
 175    uint32_t tid = static_cast<uint32_t>(std::hash<std::thread::id>{}(std::this_thread::get_id()));
 176    events_.push_back({std::move(name), std::move(category), start_us, dur_us, tid});
 177  }
 78
 179  void log_summary(std::shared_ptr<spdlog::logger> &logger) const {
 180    if (!logger || events_.empty())
 081      return;
 82
 183    double total_time_ms = 0.0;
 184    for (const auto &ev : events_) {
 185      if (ev.name == "Total Compilation") {
 186        total_time_ms = ev.dur_us / 1000.0;
 187        break;
 88      }
 089    }
 90
 191    logger->info("{:=^60}", " Compilation Passes Summary ");
 192    logger->info("{:<30} | {:>12} | {:>10}", "Pass Name", "Time (ms)", "% Total");
 193    logger->info("{:-^60}", "");
 94
 195    for (const auto &ev : events_) {
 196      if (ev.category != "pass" && ev.name != "Total Compilation")
 197        continue;
 98
 199      double dur_ms  = ev.dur_us / 1000.0;
 1100      double percent = (total_time_ms > 0) ? (dur_ms / total_time_ms) * 100.0 : 0.0;
 101
 1102      if (ev.name == "Total Compilation") {
 1103        logger->info("{:-^60}", "");
 104      }
 1105      logger->info("{:<30} | {:>12.3f} | {:>9.2f}%", ev.name, dur_ms, percent);
 1106    }
 1107    logger->info("{:=^60}", "");
 1108  }
 109
 1110  void dump_chrome_tracing(const std::filesystem::path &filepath) const {
 1111    std::ofstream out(filepath);
 1112    if (!out.is_open())
 0113      return;
 114
 1115    out << "[\n";
 1116    for (size_t i = 0; i < events_.size(); ++i) {
 1117      const auto &ev = events_[i];
 1118      out << "  {"
 119          << "\"name\": \"" << ev.name << "\", "
 120          << "\"cat\": \"" << ev.category << "\", "
 121          << "\"ph\": \"X\", "
 122          << "\"ts\": " << ev.start_us << ", "
 123          << "\"dur\": " << ev.dur_us << ", "
 124          << "\"pid\": 1, "
 125          << "\"tid\": " << ev.tid << "}";
 1126      if (i < events_.size() - 1)
 1127        out << ",";
 1128      out << "\n";
 1129    }
 1130    out << "]\n";
 1131  }
 132
 133private:
 134  std::vector<TraceEvent> events_;
 135  std::mutex              mutex_;
 136};
 137
 138// -------------------------------------------------------------------------------------------------
 139// Observability & Scoped Tracer
 140// -------------------------------------------------------------------------------------------------
 141
 142class ScopedTrace {
 143public:
 144  using Clock       = std::chrono::steady_clock;
 145  using SystemClock = std::chrono::system_clock;
 146
 147  ScopedTrace(std::string name, std::string category, std::shared_ptr<spdlog::logger> logger,
 148              CompilationProfiler *profiler)
 1149      : name_(std::move(name)), category_(std::move(category)), logger_(std::move(logger)),
 1150        profiler_(profiler) {
 151
 1152    start_time_ = Clock::now();
 1153    start_us_   = std::chrono::time_point_cast<std::chrono::microseconds>(SystemClock::now())
 154                      .time_since_epoch()
 155                      .count();
 156
 1157    if (logger_ && category_ == "pass") {
 1158      logger_->trace(">> Begin Pass: {}", name_);
 159    }
 160
 1161    nvtxRangePush(name_.c_str());
 1162  }
 163
 1164  ~ScopedTrace() {
 1165    auto end_time = Clock::now();
 1166    auto dur_us =
 167        std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time_).count();
 168
 1169    if (logger_ && category_ == "pass") {
 1170      logger_->info("<< End Pass:   {} ({:.3f} ms)", name_, dur_us / 1000.0);
 171    }
 172
 1173    if (profiler_) {
 1174      profiler_->add_event(name_, category_, start_us_, dur_us);
 175    }
 176
 1177    nvtxRangePop();
 1178  }
 179
 180private:
 181  std::string                     name_;
 182  std::string                     category_;
 183  std::shared_ptr<spdlog::logger> logger_;
 184  CompilationProfiler            *profiler_;
 185  Clock::time_point               start_time_;
 186  long long                       start_us_;
 187};
 188
 189// -------------------------------------------------------------------------------------------------
 190// Storage Adapter for owning tasks
 191// -------------------------------------------------------------------------------------------------
 192class TaskStorageAdapter : public core::IOStorageAccess {
 193public:
 194  TaskStorageAdapter(std::vector<int> in_tids, std::vector<int> out_tids, ExecResouces &resources);
 195  [[nodiscard]] core::Storage &owned_input_storage(size_t index) override;
 196  [[nodiscard]] core::Storage &owned_output_storage(size_t index) override;
 197
 198private:
 199  std::vector<int> in_tids_;
 200  std::vector<int> out_tids_;
 201  ExecResouces    &res_;
 202};
 203
 204TaskStorageAdapter::TaskStorageAdapter(std::vector<int> in_tids, std::vector<int> out_tids,
 205                                       ExecResouces &resources)
 1206    : in_tids_(std::move(in_tids)), out_tids_(std::move(out_tids)), res_(resources) {}
 207
 1208core::Storage &TaskStorageAdapter::owned_input_storage(size_t index) {
 1209  if (index >= in_tids_.size()) {
 1210    throw std::out_of_range("Input index out of range in TaskStorageAdapter");
 211  }
 1212  size_t tid = in_tids_[index];
 1213  size_t sid = res_.tid_to_sid.at(tid);
 1214  return *res_.storages.at(sid);
 1215}
 216
 1217core::Storage &TaskStorageAdapter::owned_output_storage(size_t index) {
 1218  if (index >= out_tids_.size()) {
 1219    throw std::out_of_range("Output index out of range in TaskStorageAdapter");
 220  }
 1221  size_t tid = out_tids_[index];
 1222  size_t sid = res_.tid_to_sid.at(tid);
 1223  return *res_.storages.at(sid);
 1224}
 225
 226// -------------------------------------------------------------------------------------------------
 227// Compiler Declaration (PIMPL)
 228// -------------------------------------------------------------------------------------------------
 229
 230class Compiler::Impl {
 231public:
 232  Impl(core::Registry &registry, Compiler::Config config);
 233
 234  std::unique_ptr<CompilerOutput> run(const core::GraphSpec          &gspec,
 235                                      std::unique_ptr<CompilerOutput> prev);
 236
 237private:
 238  // --- State ---
 239  core::Registry                 &registry_;
 240  Compiler::Config                config_;
 241  std::shared_ptr<spdlog::logger> logger_;
 242  CompilationProfiler             profiler_;
 243
 1244  const core::GraphSpec          *gspec_ = nullptr;
 245  std::unique_ptr<CompilerOutput> prev_;
 246  std::unique_ptr<CompilerOutput> out_;
 247
 248  // Auxiliary Map: Node Name -> Section ID
 249  std::unordered_map<std::string, size_t> node_to_section_map_;
 250
 251  // --- Helpers ---
 252  void        setup_logging();
 253  ScopedTrace trace_scope(std::string name, std::string category = "pass");
 254  void        dump_graphviz(const std::string &filename);
 255  template <class TaskInterface, class Factory, class Ctx>
 256  std::unique_ptr<core::ITask> create_or_update_task(Factory &factory, const NodePlan &np,
 257                                                     const Ctx &ctx);
 258
 259  // --- Pass Declarations ---
 260  void validate_spec();
 261  void build_graph_structure();
 262  void run_type_inference();
 263  void assign_tensor_ids();
 264  void assign_storage_ids();
 265  void verify_buffer_consistency();
 266  void allocate_buffers();
 267  void create_storage_adapters();
 268  void partition_sections();
 269  void assign_streams();
 270  void instantiate_tasks();
 271  void bind_tasks();
 272
 273  // Generic Pass Runner
 274  template <typename Func> void run_pass(const char *name, Func &&fn) {
 275    auto scope = trace_scope(name, "pass");
 276    fn();
 277  }
 278};
 279
 280// -------------------------------------------------------------------------------------------------
 281// Compiler Implementation (PIMPL)
 282// -------------------------------------------------------------------------------------------------
 283
 284Compiler::Impl::Impl(core::Registry &registry, Compiler::Config config)
 1285    : registry_(registry), config_(std::move(config)) {
 1286  setup_logging();
 1287}
 288
 289std::unique_ptr<CompilerOutput> Compiler::Impl::run(const core::GraphSpec          &gspec,
 290                                                    std::unique_ptr<CompilerOutput> prev) {
 291  gspec_ = &gspec;
 292  prev_  = std::move(prev);
 293  out_   = std::make_unique<CompilerOutput>();
 294
 295  // Use optional to control exactly when the trace ends without double-destruction
 296  std::optional<ScopedTrace> total_trace;
 297  total_trace.emplace(trace_scope("Total Compilation", "lifecycle"));
 298
 299  try {
 300    run_pass("Validate Spec", [&] { validate_spec(); });
 301    run_pass("Build Graph Plan", [&] { build_graph_structure(); });
 302    run_pass("Type Inference", [&] { run_type_inference(); });
 303
 304    run_pass("Tensor IDs", [&] { assign_tensor_ids(); });
 305    run_pass("Storage Mapping", [&] { assign_storage_ids(); });
 306    run_pass("Buffer Consistency", [&] { verify_buffer_consistency(); });
 307    run_pass("Buffer Allocation", [&] { allocate_buffers(); });
 308
 309    run_pass("Storage Adapters", [&] { create_storage_adapters(); });
 310
 311    run_pass("Section Partitioning", [&] { partition_sections(); });
 312    run_pass("Stream Assignment", [&] { assign_streams(); });
 313    run_pass("Task Instantiation", [&] { instantiate_tasks(); });
 314    run_pass("Task Binding", [&] { bind_tasks(); });
 315
 316    if (config_.dump_dot_on_failure) {
 317      run_pass("Dump Graphviz", [&] { dump_graphviz("compilation_success.dot"); });
 318    }
 319  } catch (const std::exception &e) {
 320    logger_->error("Compilation Failed: {}", e.what());
 321    if (config_.dump_dot_on_failure) {
 322      run_pass("Dump Graphviz", [&] { dump_graphviz("compilation_failure.dot"); });
 323    }
 324
 325    total_trace.reset(); // Stop timer before throwing
 326
 327    try {
 328      CUDA_CHECK(cudaDeviceSynchronize());
 329    } catch (const std::exception &cuda_e) {
 330      logger_->error("CUDA error during cleanup: {}", cuda_e.what());
 331    }
 332
 333    try {
 334      CUDA_CHECK(cudaGetLastError());
 335    } catch (const std::exception &cuda_e) {
 336      logger_->error("CUDA error during cleanup: {}", cuda_e.what());
 337    }
 338
 339    logger_->flush();
 340    throw;
 341  }
 342
 343  // Stop the total compilation timer safely
 344  total_trace.reset();
 345
 346  if (config_.enable_profiling) {
 347    // profiler_.log_summary(logger_);
 348    run_pass("Dump log summary", [&] { profiler_.log_summary(logger_); });
 349    if (!config_.log_dir.empty()) {
 350      // profiler_.dump_chrome_tracing(config_.log_dir / config_.trace_filename);
 351      run_pass("Dump Chrome Tracing",
 352               [&] { profiler_.dump_chrome_tracing(config_.log_dir / config_.trace_filename); });
 353    }
 354  }
 355
 356  return std::move(out_);
 357}
 358
 1359void Compiler::Impl::setup_logging() {
 1360  if (spdlog::get("compiler")) {
 0361    spdlog::drop("compiler");
 362  }
 363
 1364  if (!config_.log_dir.empty()) {
 1365    std::filesystem::create_directories(config_.log_dir);
 1366    auto path = config_.log_dir / "compiler.log";
 1367    logger_   = spdlog::basic_logger_mt("compiler", path.string(), true);
 1368  } else {
 1369    logger_ = spdlog::stdout_color_mt("compiler");
 370  }
 1371  logger_->set_level(config_.verbose_tracing ? spdlog::level::trace : spdlog::level::info);
 1372}
 373
 1374ScopedTrace Compiler::Impl::trace_scope(std::string name, std::string category) {
 1375  return ScopedTrace(std::move(name), std::move(category), logger_,
 376                     config_.enable_profiling ? &profiler_ : nullptr);
 1377}
 378
 1379void Compiler::Impl::dump_graphviz(const std::string &filename) {
 1380  if (config_.log_dir.empty()) {
 0381    return;
 382  }
 383
 1384  std::ofstream file(config_.log_dir / filename);
 1385  if (!file.is_open()) {
 0386    return;
 387  }
 388
 1389  const auto graph_name = std::filesystem::path(filename).stem().string();
 1390  file << to_dot(*out_, GraphCompiledDumpPreferences{}, graph_name);
 1391}
 392
 393// -------------------------------------------------------------------------------------------------
 394// Pass: Validate Spec
 395// -------------------------------------------------------------------------------------------------
 1396void Compiler::Impl::validate_spec() {
 1397  std::unordered_set<std::string> names;
 1398  std::unordered_set<std::string> edge_dsts;
 399
 1400  auto vertices = boost::make_iterator_range(boost::vertices(*gspec_));
 1401  for (const auto &v : vertices) {
 1402    const auto &ns = (*gspec_)[v];
 1403    if (!names.insert(ns.name).second) {
 1404      throw CompilerException(std::format("Duplicate node name: '{}'", ns.name));
 405    }
 1406    if (!registry_.is_registered(ns.kind)) {
 1407      throw CompilerException(std::format("Unknown node kind '{}'", ns.kind));
 408    }
 1409  }
 410
 1411  auto edges = boost::make_iterator_range(boost::edges(*gspec_));
 1412  for (const auto &e : edges) {
 1413    const auto &es    = (*gspec_)[e];
 1414    const auto  dst   = (*gspec_)[boost::target(e, *gspec_)];
 1415    std::string label = std::format("{}:{}", dst.name, es.in_idx);
 416
 1417    if (!edge_dsts.insert(label).second) {
 1418      throw CompilerException(std::format("Multiple edges targeting: {}", label));
 419    }
 1420  }
 1421}
 422
 423// -------------------------------------------------------------------------------------------------
 424// Pass: Build Graph Structure
 425// -------------------------------------------------------------------------------------------------
 1426void Compiler::Impl::build_graph_structure() {
 427  using VSpec = core::GraphSpec::vertex_descriptor;
 428  using VPlan = GraphPlan::vertex_descriptor;
 1429  std::map<VSpec, VPlan> v_map;
 1430  auto                  &g = out_->graph;
 431
 1432  for (auto v : boost::make_iterator_range(boost::vertices(*gspec_))) {
 1433    NodePlan np;
 1434    np.spec  = (*gspec_)[v];
 1435    v_map[v] = boost::add_vertex(np, g);
 1436  }
 437
 1438  for (auto e : boost::make_iterator_range(boost::edges(*gspec_))) {
 1439    const auto &es  = (*gspec_)[e];
 1440    const auto  src = v_map.at(boost::source(e, *gspec_));
 1441    const auto  dst = v_map.at(boost::target(e, *gspec_));
 1442    EdgePlan    ep;
 1443    ep.spec = es;
 1444    boost::add_edge(src, dst, ep, g);
 1445  }
 1446}
 447
 448// -------------------------------------------------------------------------------------------------
 449// Pass: Type Inference
 450// -------------------------------------------------------------------------------------------------
 1451void Compiler::Impl::run_type_inference() {
 1452  auto                                     &g = out_->graph;
 1453  std::vector<GraphPlan::vertex_descriptor> topo_order;
 454
 455  try {
 1456    boost::topological_sort(g, std::back_inserter(topo_order));
 1457  } catch (const boost::not_a_dag &) {
 1458    throw CompilerException("Graph contains a cycle (loop), which is not allowed.");
 0459  }
 460
 1461  for (auto v : std::views::reverse(topo_order)) {
 1462    auto &node       = g[v];
 1463    auto  node_trace = trace_scope(std::format("Infer: {}", node.spec.name), "detail");
 464
 1465    auto                     in_degree = boost::in_degree(v, g);
 1466    std::vector<core::TDesc> input_descs(in_degree);
 467
 1468    for (auto e : boost::make_iterator_range(boost::in_edges(v, g))) {
 1469      const auto &edge_plan = g[e];
 1470      if (edge_plan.spec.in_idx >= input_descs.size()) {
 0471        throw CompilerException("Input index out of bounds");
 472      }
 1473      input_descs[edge_plan.spec.in_idx] = edge_plan.desc;
 1474    }
 475
 1476    const auto &factory = registry_.get(node.spec.kind);
 1477    node.infer          = factory.infer(input_descs, node.spec.settings);
 478
 1479    for (auto e : boost::make_iterator_range(boost::out_edges(v, g))) {
 1480      auto &edge_plan = g[e];
 1481      if (edge_plan.spec.out_idx >= node.infer.output_descs.size()) {
 1482        throw CompilerException("Output index out of bounds");
 483      }
 1484      edge_plan.desc = node.infer.output_descs[edge_plan.spec.out_idx];
 1485    }
 1486  }
 1487}
 488
 489// -------------------------------------------------------------------------------------------------
 490// Pass: Assign Tensor IDs
 491// -------------------------------------------------------------------------------------------------
 1492void Compiler::Impl::assign_tensor_ids() {
 1493  auto &g        = out_->graph;
 1494  auto &res      = out_->resources;
 1495  int   next_tid = 0;
 496
 1497  std::vector<GraphPlan::vertex_descriptor> topo;
 1498  boost::topological_sort(g, std::back_inserter(topo));
 499
 1500  for (auto v : std::views::reverse(topo)) {
 1501    auto &node = g[v];
 502
 1503    node.in_tids.resize(node.infer.input_descs.size());
 1504    for (auto e : boost::make_iterator_range(boost::in_edges(v, g))) {
 1505      const auto &ep               = g[e];
 1506      node.in_tids[ep.spec.in_idx] = ep.tid;
 1507      res.tensor_descs[ep.tid]     = ep.desc;
 1508    }
 509
 1510    node.out_tids.resize(node.infer.output_descs.size());
 1511    auto out_edges = boost::out_edges(v, g);
 512
 1513    for (size_t i = 0; i < node.out_tids.size(); ++i) {
 1514      int tid               = next_tid++;
 1515      node.out_tids[i]      = tid;
 1516      res.tensor_descs[tid] = node.infer.output_descs[i];
 517
 1518      for (auto e : boost::make_iterator_range(out_edges)) {
 1519        if (g[e].spec.out_idx == static_cast<int>(i)) {
 1520          g[e].tid = tid;
 521        }
 1522      }
 1523    }
 1524  }
 1525}
 526
 527// -------------------------------------------------------------------------------------------------
 528// Pass: Assign Storage IDs
 529// -------------------------------------------------------------------------------------------------
 1530void Compiler::Impl::assign_storage_ids() {
 1531  auto &g   = out_->graph;
 1532  auto &res = out_->resources;
 533
 1534  res.tid_to_sid.clear();
 1535  int next_sid = 0;
 536
 1537  std::vector<GraphPlan::vertex_descriptor> topo;
 1538  boost::topological_sort(g, std::back_inserter(topo));
 539
 1540  for (auto v : std::views::reverse(topo)) {
 1541    auto &node = g[v];
 542
 1543    for (size_t out_idx = 0; out_idx < node.out_tids.size(); ++out_idx) {
 1544      int out_tid = node.out_tids[out_idx];
 1545      int sid     = -1;
 546
 1547      for (const auto &ip : node.infer.in_place) {
 0548        if (ip.out_idx == static_cast<int>(out_idx)) {
 0549          int in_tid = node.in_tids[ip.in_idx];
 550
 0551          if (res.tid_to_sid.contains(in_tid)) {
 0552            sid = (int)res.tid_to_sid.at(in_tid);
 0553          } else {
 0554            throw CompilerException(
 555                std::format("Node '{}': In-place input TID {} has no Storage ID assigned.",
 556                            node.spec.name, in_tid));
 557          }
 0558          break;
 559        }
 0560      }
 561
 1562      if (sid == -1) {
 1563        sid = next_sid++;
 564      }
 1565      res.tid_to_sid[out_tid] = sid;
 1566    }
 1567  }
 1568}
 569
 1570void Compiler::Impl::verify_buffer_consistency() {
 1571  std::map<size_t, std::vector<std::string>> owners;
 1572  auto                                      &g   = out_->graph;
 1573  auto                                      &res = out_->resources;
 574
 1575  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 1576    const auto &node = g[v];
 1577    for (size_t i = 0; i < node.infer.owned_inputs.size(); ++i) {
 1578      if (node.infer.owned_inputs[i]) {
 1579        owners[res.tid_to_sid.at(node.in_tids[i])].push_back(node.spec.name + ":in");
 580      }
 1581    }
 1582    for (size_t i = 0; i < node.infer.owned_outputs.size(); ++i) {
 1583      if (node.infer.owned_outputs[i]) {
 1584        owners[res.tid_to_sid.at(node.out_tids[i])].push_back(node.spec.name + ":out");
 585      }
 1586    }
 1587  }
 588
 1589  for (const auto &[sid, nodeList] : owners) {
 1590    if (nodeList.size() > 1) {
 1591      throw CompilerException(std::format("Storage ID {} has multiple owners", sid));
 592    }
 1593  }
 1594}
 595
 596// void Compiler::Impl::allocate_buffers() {
 597//   auto &g   = out_->graph;
 598//   auto &res = out_->resources;
 599
 600//   res.memory_blocks.clear();
 601//   res.storages.clear();
 602
 603//   std::unordered_set<size_t> user_managed_sids;
 604//   for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 605//     const auto &node = g[v];
 606//     for (size_t i = 0; i < node.out_tids.size(); ++i) {
 607//       if (node.infer.owned_outputs[i]) {
 608//         int    tid = node.out_tids[i];
 609//         size_t sid = res.tid_to_sid.at(tid);
 610//         user_managed_sids.insert(sid);
 611//       }
 612//     }
 613//     for (size_t i = 0; i < node.in_tids.size(); ++i) {
 614//       if (node.infer.owned_inputs[i]) {
 615//         int    tid = node.in_tids[i];
 616//         size_t sid = res.tid_to_sid.at(tid);
 617//         user_managed_sids.insert(sid);
 618//       }
 619//     }
 620//   }
 621
 622//   std::map<size_t, size_t> sid_to_rep_tid;
 623//   for (const auto &[tid, sid] : res.tid_to_sid) {
 624//     if (!sid_to_rep_tid.count(sid))
 625//       sid_to_rep_tid[sid] = tid;
 626//   }
 627
 628//   for (const auto &[sid, tid] : sid_to_rep_tid) {
 629//     auto        alloc_scope = trace_scope(std::format("Alloc SID {}", sid), "detail");
 630//     const auto &desc        = res.tensor_descs.at(tid);
 631
 632//     auto storage     = std::make_unique<core::Storage>();
 633//     storage->mem_loc = desc.mem_loc;
 634//     storage->bytes   = desc.num_bytes();
 635//     storage->ptr     = nullptr;
 636
 637//     if (!user_managed_sids.contains(sid)) {
 638//       MemoryBlock block;
 639//       block.mem_loc    = desc.mem_loc;
 640//       block.size_bytes = desc.num_bytes();
 641
 642//       logger_->info("Allocating {} bytes for SID {} at {:?} memory", block.size_bytes, sid,
 643//                     to_string(desc.mem_loc));
 644
 645//       // Use a standard scope block to control the RAII timer
 646//       {
 647//         auto sys_scope = trace_scope(
 648//             desc.mem_loc == core::MemLoc::Host ? "Host Malloc" : "Device Malloc", "syscall");
 649//         if (desc.mem_loc == core::MemLoc::Host) {
 650//           block.h_data = curaii::make_unique_host_ptr<std::byte>(block.size_bytes);
 651//         } else {
 652//           block.d_data = curaii::make_unique_device_ptr<std::byte>(block.size_bytes);
 653//         }
 654//       } // sys_scope naturally destructs here!
 655
 656//       storage->ptr = static_cast<std::byte *>(block.get());
 657//       res.memory_blocks.emplace(sid, std::move(block));
 658//     } else {
 659//       logger_->info("SID {} is user-managed; skipping allocation.", sid);
 660//     }
 661
 662//     res.storages.emplace(sid, std::move(storage));
 663//   }
 664
 665//   // Temp test, trigger a 1b cuda memcopy to see how it shows up in the profiler
 666//   CUDA_CHECK(cudaDeviceSynchronize());
 667//   if (res.memory_blocks.size() >= 2) {
 668//     auto &block1 = res.memory_blocks.begin()->second;
 669//     auto &block2 = std::next(res.memory_blocks.begin())->second;
 670//     if (block1.mem_loc == core::MemLoc::Device && block2.mem_loc == core::MemLoc::Device) {
 671//       auto sys_scope = trace_scope("Test Memcpy", "syscall");
 672//       CUDA_CHECK(cudaMemcpy(block2.get(), block1.get(), 1, cudaMemcpyDeviceToDevice));
 673//       CUDA_CHECK(cudaDeviceSynchronize());
 674//     }
 675//   }
 676// }
 677
 1678void Compiler::Impl::allocate_buffers() {
 1679  auto &g   = out_->graph;
 1680  auto &res = out_->resources;
 681
 1682  res.memory_blocks.clear();
 1683  res.storages.clear();
 684
 685  // 1. Identify user-managed SIDs
 1686  std::unordered_set<size_t> user_managed_sids;
 1687  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 1688    const auto &node = g[v];
 1689    for (size_t i = 0; i < node.out_tids.size(); ++i) {
 1690      if (node.infer.owned_outputs[i]) {
 1691        user_managed_sids.insert(res.tid_to_sid.at(node.out_tids[i]));
 692      }
 1693    }
 1694    for (size_t i = 0; i < node.in_tids.size(); ++i) {
 1695      if (node.infer.owned_inputs[i]) {
 1696        user_managed_sids.insert(res.tid_to_sid.at(node.in_tids[i]));
 697      }
 1698    }
 1699  }
 700
 701  // 2. Map SID to representative TID
 1702  std::map<size_t, size_t> sid_to_rep_tid;
 1703  for (const auto &[tid, sid] : res.tid_to_sid) {
 1704    if (!sid_to_rep_tid.count(sid)) {
 1705      sid_to_rep_tid[sid] = tid;
 706    }
 1707  }
 708
 709  // 3. Build a pool of scavengable blocks from prev_
 710  // Key: {MemLoc, size_in_bytes}
 1711  std::multimap<std::pair<core::MemLoc, size_t>, MemoryBlock> free_blocks;
 1712  if (prev_) {
 1713    for (auto &[prev_sid, block] : prev_->resources.memory_blocks) {
 1714      free_blocks.emplace(std::make_pair(block.mem_loc, block.size_bytes), std::move(block));
 1715    }
 716  }
 717
 718  // 4. Allocate or Scavenge
 1719  for (const auto &[sid, tid] : sid_to_rep_tid) {
 1720    auto        alloc_scope = trace_scope(std::format("Alloc SID {}", sid), "detail");
 1721    const auto &desc        = res.tensor_descs.at(tid);
 722
 1723    auto storage     = std::make_unique<core::Storage>();
 1724    storage->mem_loc = desc.mem_loc;
 1725    storage->bytes   = desc.num_bytes();
 1726    storage->ptr     = nullptr;
 727
 1728    if (!user_managed_sids.contains(sid)) {
 1729      MemoryBlock block;
 1730      auto        pool_key = std::make_pair(desc.mem_loc, desc.num_bytes());
 1731      auto        it       = free_blocks.find(pool_key);
 732
 1733      if (it != free_blocks.end()) {
 734        // We found an exact match! Scavenge it.
 1735        logger_->info("Reusing {} bytes for SID {} at {:?} memory", desc.num_bytes(), sid,
 736                      to_string(desc.mem_loc));
 1737        block = std::move(it->second);
 1738        free_blocks.erase(it);
 1739      } else {
 740        // No match found, allocate fresh memory.
 1741        block.mem_loc    = desc.mem_loc;
 1742        block.size_bytes = desc.num_bytes();
 743
 1744        logger_->info("Allocating {} bytes for SID {} at {:?} memory", block.size_bytes, sid,
 745                      to_string(desc.mem_loc));
 746
 747        {
 1748          auto sys_scope = trace_scope(
 749              desc.mem_loc == core::MemLoc::Host ? "Host Malloc" : "Device Malloc", "syscall");
 1750          if (desc.mem_loc == core::MemLoc::Host) {
 1751            block.h_data = curaii::make_unique_host_ptr<std::byte>(block.size_bytes);
 1752          } else {
 1753            block.d_data = curaii::make_unique_device_ptr<std::byte>(block.size_bytes);
 754          }
 1755        }
 756      }
 757
 1758      storage->ptr = static_cast<std::byte *>(block.get());
 1759      res.memory_blocks.emplace(sid, std::move(block));
 1760    } else {
 1761      logger_->info("SID {} is user-managed; skipping allocation.", sid);
 762    }
 763
 1764    res.storages.emplace(sid, std::move(storage));
 1765  }
 766
 767  // Temp test, trigger a 1b cuda memcopy to see how it shows up in the profiler
 1768  CUDA_CHECK(cudaDeviceSynchronize());
 1769  if (res.memory_blocks.size() >= 2) {
 1770    auto &block1 = res.memory_blocks.begin()->second;
 1771    auto &block2 = std::next(res.memory_blocks.begin())->second;
 1772    if (block1.mem_loc == core::MemLoc::Device && block2.mem_loc == core::MemLoc::Device) {
 1773      auto sys_scope = trace_scope("Test Memcpy", "syscall");
 1774      CUDA_CHECK(cudaMemcpy(block2.get(), block1.get(), 1, cudaMemcpyDeviceToDevice));
 1775      CUDA_CHECK(cudaDeviceSynchronize());
 1776    }
 777  }
 778
 779  // Any blocks left inside `free_blocks` will naturally go out of scope here and
 780  // safely deallocate, meaning memory for removed nodes is properly cleaned up.
 1781}
 782
 783// -------------------------------------------------------------------------------------------------
 784// Pass: Section Partitioning
 785// -------------------------------------------------------------------------------------------------
 1786void Compiler::Impl::partition_sections() {
 1787  auto &g         = out_->graph;
 1788  auto  num_verts = boost::num_vertices(g);
 789
 1790  for (auto e : boost::make_iterator_range(boost::edges(g))) {
 1791    const auto source = boost::source(e, g);
 1792    const auto target = boost::target(e, g);
 1793    if (g[source].infer.kind == core::TaskKind::Async &&
 794        g[target].infer.kind == core::TaskKind::Async) {
 1795      throw CompilerException(
 796          std::format("Consecutive asynchronous nodes '{}' and '{}' are not supported",
 797                      g[source].spec.name, g[target].spec.name));
 798    }
 1799  }
 800
 1801  std::vector<size_t> parent(num_verts);
 1802  std::iota(parent.begin(), parent.end(), 0);
 803
 804  auto find = [&](size_t i) {
 805    while (i != parent[i]) {
 806      parent[i] = parent[parent[i]];
 807      i         = parent[i];
 808    }
 809    return i;
 1810  };
 811
 812  auto unite = [&](size_t i, size_t j) {
 813    size_t root_i = find(i);
 814    size_t root_j = find(j);
 815    if (root_i != root_j)
 816      parent[root_i] = root_j;
 1817  };
 818
 1819  for (auto e : boost::make_iterator_range(boost::edges(g))) {
 1820    auto u = boost::source(e, g);
 1821    auto v = boost::target(e, g);
 1822    if (g[u].infer.kind == core::TaskKind::Sync && g[v].infer.kind == core::TaskKind::Sync) {
 1823      unite(u, v);
 824    }
 1825  }
 826
 1827  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 1828    if (g[v].infer.kind == core::TaskKind::Async) {
 1829      std::vector<size_t> sync_preds;
 1830      for (auto e : boost::make_iterator_range(boost::in_edges(v, g))) {
 1831        auto p = boost::source(e, g);
 1832        if (g[p].infer.kind == core::TaskKind::Sync)
 1833          sync_preds.push_back(p);
 1834      }
 1835      if (!sync_preds.empty()) {
 1836        for (size_t i = 1; i < sync_preds.size(); ++i)
 0837          unite(sync_preds[0], sync_preds[i]);
 838      }
 839
 1840      std::vector<size_t> sync_succs;
 1841      for (auto e : boost::make_iterator_range(boost::out_edges(v, g))) {
 1842        auto s = boost::target(e, g);
 1843        if (g[s].infer.kind == core::TaskKind::Sync)
 1844          sync_succs.push_back(s);
 1845      }
 1846      if (!sync_succs.empty()) {
 1847        for (size_t i = 1; i < sync_succs.size(); ++i)
 0848          unite(sync_succs[0], sync_succs[i]);
 849      }
 1850    }
 1851  }
 852
 1853  std::map<size_t, size_t> root_to_section_id;
 1854  out_->sections.clear();
 1855  int next_sec_id = 0;
 1856  node_to_section_map_.clear();
 857
 858  auto get_section_id = [&](size_t v_idx) {
 859    size_t root = find(v_idx);
 860    if (root_to_section_id.find(root) == root_to_section_id.end()) {
 861      Section s;
 862      s.id   = next_sec_id;
 863      s.name = std::format("section-{}", next_sec_id);
 864      out_->sections.push_back(s);
 865      root_to_section_id[root] = next_sec_id++;
 866    }
 867    return root_to_section_id[root];
 1868  };
 869
 1870  std::vector<GraphPlan::vertex_descriptor> topo;
 1871  boost::topological_sort(g, std::back_inserter(topo));
 872
 1873  for (auto v : std::views::reverse(topo)) {
 1874    auto &np = g[v];
 1875    if (np.infer.kind == core::TaskKind::Sync) {
 1876      size_t sec_id = get_section_id(v);
 1877      out_->sections[sec_id].sync_topo.push_back(v);
 1878      node_to_section_map_[np.spec.name] = sec_id;
 879    }
 1880  }
 881
 1882  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 1883    if (g[v].infer.kind != core::TaskKind::Async)
 1884      continue;
 885
 1886    std::set<size_t> unique_cons_sections;
 1887    for (auto e : boost::make_iterator_range(boost::out_edges(v, g))) {
 1888      auto s = boost::target(e, g);
 1889      if (g[s].infer.kind == core::TaskKind::Sync) {
 1890        unique_cons_sections.insert(get_section_id(s));
 891      }
 1892    }
 1893    for (size_t sec_id : unique_cons_sections) {
 1894      out_->sections[sec_id].async_cons.push_back(v);
 1895    }
 896
 1897    std::set<size_t> unique_prod_sections;
 1898    for (auto e : boost::make_iterator_range(boost::in_edges(v, g))) {
 1899      auto p = boost::source(e, g);
 1900      if (g[p].infer.kind == core::TaskKind::Sync) {
 1901        unique_prod_sections.insert(get_section_id(p));
 902      }
 1903    }
 1904    for (size_t sec_id : unique_prod_sections) {
 1905      out_->sections[sec_id].async_prod.push_back(v);
 1906    }
 1907  }
 908
 1909  for (auto &section : out_->sections) {
 910    // A synchronizing producer must run before ordinary producers: its barrier covers all preceding
 911    // sync-node work before an ordinary queue is allowed to publish its GPU-backed input.
 912    const auto ordinary_begin =
 1913        std::stable_partition(section.async_prod.begin(), section.async_prod.end(),
 914                              [&](auto v) { return g[v].infer.synchronizes_producer_stream; });
 1915    section.has_synchronizing_async_producer = ordinary_begin != section.async_prod.begin();
 1916  }
 1917}
 918
 919// void Compiler::Impl::assign_streams() {
 920//   out_->resources.streams.clear();
 921//   for (auto &sec : out_->sections) {
 922//     curaii::CudaStream stream;
 923//     sec.stream = stream.get();
 924//     out_->resources.streams.emplace(sec.id, std::move(stream));
 925//   }
 926// }
 927
 1928void Compiler::Impl::assign_streams() {
 1929  out_->resources.streams.clear();
 930
 931  // 1. Only set up the iterators if prev_ actually exists
 1932  if (prev_) {
 1933    auto prev_it  = prev_->resources.streams.begin();
 1934    auto prev_end = prev_->resources.streams.end();
 935
 1936    for (auto &sec : out_->sections) {
 1937      if (prev_it != prev_end) {
 938        // Scavenge an existing stream
 1939        auto &old_stream = prev_it->second;
 1940        sec.stream       = old_stream.get();
 941
 1942        out_->resources.streams.emplace(sec.id, std::move(old_stream));
 1943        ++prev_it;
 1944      } else {
 945        // Fallback: Create a new stream (ran out of old ones)
 0946        curaii::CudaStream stream;
 0947        sec.stream = stream.get();
 0948        out_->resources.streams.emplace(sec.id, std::move(stream));
 0949      }
 1950    }
 1951  } else {
 952    // 2. No previous graph at all, just create fresh streams for everything
 1953    for (auto &sec : out_->sections) {
 1954      curaii::CudaStream stream;
 1955      sec.stream = stream.get();
 1956      out_->resources.streams.emplace(sec.id, std::move(stream));
 1957    }
 958  }
 1959}
 960
 1961void Compiler::Impl::create_storage_adapters() {
 1962  auto &g   = out_->graph;
 1963  auto &res = out_->resources;
 964
 1965  res.node_storage_adapters.clear();
 966
 1967  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 1968    const auto &np      = g[v];
 1969    auto        adapter = std::make_unique<TaskStorageAdapter>(np.in_tids, np.out_tids, res);
 1970    res.node_storage_adapters.emplace(np.spec.name, std::move(adapter));
 1971  }
 1972}
 973
 974template <class To, class From>
 1975std::unique_ptr<To> dynamic_unique_ptr_cast(std::unique_ptr<From> &&ptr) noexcept {
 1976  if (auto casted = dynamic_cast<To *>(ptr.get())) {
 1977    ptr.release();
 1978    return std::unique_ptr<To>(casted);
 979  }
 0980  return nullptr;
 1981}
 982
 983template <class TaskInterface, class Factory, class Ctx>
 984std::unique_ptr<core::ITask>
 1985Compiler::Impl::create_or_update_task(Factory &factory, const NodePlan &np, const Ctx &ctx) {
 986
 987  // 1. Helper to synchronize the correct streams based on Ctx type
 988  auto sync_streams = [&]() {
 989    if constexpr (std::is_same_v<Ctx, core::SyncCreateCtx>) {
 990      if (ctx.stream) {
 991        CUDA_CHECK(cudaStreamSynchronize(ctx.stream));
 992      } else {
 993        logger_->warn("SyncCreateCtx has null stream; skipping synchronization.");
 994      }
 995    } else if constexpr (std::is_same_v<Ctx, core::AsyncCreateCtx>) {
 996      if (ctx.producer_stream) {
 997        CUDA_CHECK(cudaStreamSynchronize(ctx.producer_stream));
 998      } else {
 999        logger_->warn("AsyncCreateCtx has null producer_stream; skipping synchronization.");
 1000      }
 1001      if (ctx.consumer_stream) {
 1002        CUDA_CHECK(cudaStreamSynchronize(ctx.consumer_stream));
 1003      } else {
 1004        logger_->warn("AsyncCreateCtx has null consumer_stream; skipping synchronization.");
 1005      }
 1006    }
 11007  };
 1008
 1009  // 2. Helper to accurately profile Task Creation
 1010  auto do_create = [&]() {
 1011    auto scope = trace_scope(std::format("Create Task: {}", np.spec.name), "detail");
 1012    auto task  = factory.create(np.infer.input_descs, np.spec.settings, ctx);
 1013    sync_streams();
 1014    return task;
 11015  }; // scope naturally destructs here, capturing the fully synced time
 1016
 1017  // 3. Helper to accurately profile Task Updating
 1018  auto do_update = [&](std::unique_ptr<TaskInterface> prev_task) {
 1019    auto scope = trace_scope(std::format("Update Task: {}", np.spec.name), "detail");
 1020    auto task  = factory.update(std::move(prev_task), np.infer.input_descs, np.spec.settings, ctx);
 1021    sync_streams();
 1022    return task;
 11023  };
 1024
 1025  // --- Main Logic ---
 1026
 11027  std::unique_ptr<core::ITask> *prev_ptr_ref = nullptr;
 11028  if (prev_ && !prev_->resources.tasks.empty()) {
 11029    auto &prev_tasks = prev_->resources.tasks;
 11030    if (auto it = prev_tasks.find(np.spec.name); it != prev_tasks.end()) {
 11031      prev_ptr_ref = &it->second;
 1032    }
 1033  }
 1034
 11035  if (!prev_ptr_ref) {
 11036    return do_create();
 1037  }
 1038
 11039  bool kind_mismatch       = false;
 11040  bool found_in_prev_graph = false;
 1041
 11042  auto [vi, vi_end] = boost::vertices(prev_->graph);
 11043  for (; vi != vi_end; ++vi) {
 11044    const NodePlan &prev_node = prev_->graph[*vi];
 11045    if (prev_node.spec.name == np.spec.name) {
 11046      found_in_prev_graph = true;
 11047      if (prev_node.spec.kind != np.spec.kind) {
 01048        kind_mismatch = true;
 1049      }
 11050      break;
 1051    }
 11052  }
 1053
 11054  if (!found_in_prev_graph || kind_mismatch) {
 01055    return do_create();
 1056  }
 1057
 11058  auto prev_task_typed = dynamic_unique_ptr_cast<TaskInterface>(std::move(*prev_ptr_ref));
 1059
 11060  if (!prev_task_typed) {
 01061    return do_create();
 1062  }
 1063
 11064  return do_update(std::move(prev_task_typed));
 11065}
 1066
 11067void Compiler::Impl::instantiate_tasks() {
 11068  auto &g     = out_->graph;
 11069  auto &tasks = out_->resources.tasks;
 1070
 11071  tasks.clear();
 1072
 11073  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 11074    const auto &np = g[v];
 1075
 11076    if (np.infer.kind == core::TaskKind::Sync) {
 11077      size_t sid    = node_to_section_map_.at(np.spec.name);
 11078      auto   stream = out_->resources.streams.at(sid).get();
 1079
 11080      core::SyncCreateCtx ctx{.stream = stream};
 11081      auto               &factory = registry_.get_sync(np.spec.kind);
 1082
 11083      auto task = create_or_update_task<core::ISyncTask>(factory, np, ctx);
 11084      tasks.emplace(np.spec.name, std::move(task));
 1085
 11086    } else if (np.infer.kind == core::TaskKind::Async) {
 11087      void *prod_stream = nullptr;
 11088      for (auto e : boost::make_iterator_range(boost::in_edges(v, g))) {
 11089        auto p = boost::source(e, g);
 11090        if (g[p].infer.kind == core::TaskKind::Sync) {
 11091          size_t sid  = node_to_section_map_.at(g[p].spec.name);
 11092          prod_stream = out_->resources.streams.at(sid).get();
 11093          break;
 1094        }
 01095      }
 1096
 11097      void *cons_stream = nullptr;
 11098      for (auto e : boost::make_iterator_range(boost::out_edges(v, g))) {
 11099        auto s = boost::target(e, g);
 11100        if (g[s].infer.kind == core::TaskKind::Sync) {
 11101          size_t sid  = node_to_section_map_.at(g[s].spec.name);
 11102          cons_stream = out_->resources.streams.at(sid).get();
 11103          break;
 1104        }
 01105      }
 1106
 11107      core::AsyncCreateCtx ctx{.producer_stream = static_cast<cudaStream_t>(prod_stream),
 11108                               .consumer_stream = static_cast<cudaStream_t>(cons_stream)};
 1109
 11110      auto &factory = registry_.get_async(np.spec.kind);
 1111
 11112      auto task = create_or_update_task<core::IAsyncTask>(factory, np, ctx);
 11113      tasks.emplace(np.spec.name, std::move(task));
 11114    }
 11115  }
 11116}
 1117
 1118std::shared_ptr<spdlog::logger> create_task_logger(const std::string &node_name,
 11119                                                   const std::string &node_kind) {
 11120  auto sink = std::make_shared<spdlog::sinks::stdout_color_sink_mt>();
 11121  sink->set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%n] [thread %t] [%^%l%$] %v");
 11122  auto logger_name = fmt::format("TaskLogger-{}-{}", node_kind, node_name);
 11123  auto logger      = std::make_shared<spdlog::logger>(logger_name, sink);
 11124  logger->set_level(spdlog::default_logger()->level());
 11125  return logger;
 11126}
 1127
 11128void Compiler::Impl::bind_tasks() {
 11129  auto &g   = out_->graph;
 11130  auto &res = out_->resources;
 1131
 11132  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 11133    const auto &np = g[v];
 1134
 11135    auto it_task = res.tasks.find(np.spec.name);
 11136    if (it_task == res.tasks.end())
 01137      continue;
 11138    core::ITask *task = it_task->second.get();
 1139
 11140    if (auto it_adapter = res.node_storage_adapters.find(np.spec.name);
 11141        it_adapter != res.node_storage_adapters.end()) {
 11142      task->bind_storage_access(it_adapter->second.get());
 1143    }
 1144
 11145    auto logger = create_task_logger(np.spec.name, np.spec.kind);
 11146    task->bind_logger(std::move(logger));
 11147  }
 11148}
 1149
 1150// -------------------------------------------------------------------------------------------------
 1151// Public API PIMPL forwarding
 1152// -------------------------------------------------------------------------------------------------
 1153
 1154Compiler::Compiler(core::Registry &registry, Config config)
 11155    : impl_(std::make_unique<Impl>(registry, std::move(config))) {}
 1156
 11157Compiler::~Compiler() = default;
 1158
 1159std::unique_ptr<CompilerOutput> Compiler::compile(const core::GraphSpec          &gspec,
 11160                                                  std::unique_ptr<CompilerOutput> prev) {
 11161  return impl_->run(gspec, std::move(prev));
 11162}
 1163
 1164} // namespace holoflow::runtime

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holoflow\src\runtime\graph_display.cc

#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#include "holoflow/runtime/graph_display.hh"
 16
 17#include "holoflow/core/tensor.hh"
 18
 19#include <algorithm>
 20#include <boost/graph/graph_traits.hpp>
 21#include <charconv>
 22#include <cmath>
 23#include <cstdint>
 24#include <format>
 25#include <functional>
 26#include <iomanip>
 27#include <memory>
 28#include <queue>
 29#include <sstream>
 30#include <string>
 31#include <string_view>
 32#include <type_traits>
 33#include <unordered_set>
 34#include <vector>
 35
 36#include "holoflow/runtime/compiler.hh"
 37
 38namespace {
 39
 40using namespace holoflow::core;
 41using GraphCompiledDumpPreferences = holoflow::runtime::GraphCompiledDumpPreferences;
 42
 143static std::string replace_newlines_escaped_with_l(const std::string &s) {
 144  std::string out;
 145  out.reserve(s.size());
 146  for (auto it = s.begin(); it != s.end(); ++it) {
 147    const char c = *it;
 148    if (c == '\\') {
 149      if (it + 1 != s.end() && *(it + 1) == 'n') {
 150        out += "\\l";
 151        ++it;
 152      } else {
 153        out += "\\";
 54      }
 155    } else {
 156      out += c;
 57    }
 158  }
 159  return out;
 160}
 61
 162static std::string replace_newlines_with_l(const std::string &s) {
 163  std::string out;
 164  out.reserve(s.size());
 165  for (char c : s) {
 166    if (c == '\n') {
 167      out += "\\l";
 168    } else {
 169      out += c;
 70    }
 171  }
 172  return out;
 173}
 74
 75// reuse escape helper
 176static std::string escape_for_label(const std::string &s) {
 177  std::string out;
 178  out.reserve(s.size());
 179  for (auto it = s.begin(); it != s.end(); ++it) {
 180    const char c = *it;
 181    switch (c) {
 82    case '\\':
 183      if (it + 1 != s.end() && *(it + 1) == 'l') {
 184        out += "\\l";
 185        ++it;
 186      } else {
 187        out += "\\\\";
 88      }
 189      break;
 90    case '"':
 191      out += "\\\"";
 192      break;
 93    case '\n':
 194      out += "\\n";
 195      break;
 96    case '\r':
 097      break;
 98    default:
 199      out += c;
 100      break;
 101    }
 1102  }
 1103  return out;
 1104}
 105
 1106static void round_json_floating_point_values(nlohmann::json &value, int precision) {
 1107  if (value.is_array() || value.is_object()) {
 1108    for (auto &child : value) {
 1109      round_json_floating_point_values(child, precision);
 1110    }
 1111    return;
 112  }
 113
 1114  if (!value.is_number_float()) {
 1115    return;
 116  }
 117
 0118  const double number = value.get<double>();
 0119  if (!std::isfinite(number)) {
 0120    return;
 121  }
 122
 123  char       buffer[64];
 0124  const auto result = std::to_chars(buffer, buffer + sizeof(buffer), number,
 125                                    std::chars_format::scientific, precision);
 0126  if (result.ec != std::errc{}) {
 0127    return;
 128  }
 129
 0130  double     rounded = number;
 0131  const auto parsed  = std::from_chars(buffer, result.ptr, rounded, std::chars_format::scientific);
 0132  if (parsed.ec == std::errc{}) {
 0133    value = rounded;
 134  }
 1135}
 136
 137static std::string dump_json_with_floating_point_precision(const nlohmann::json &value,
 1138                                                           int                   precision) {
 1139  auto rounded = value;
 1140  round_json_floating_point_values(rounded, precision);
 1141  return rounded.dump(2);
 1142}
 143
 144std::string tdesc_to_string(const TDesc &d) {
 145  std::ostringstream ss;
 146  ss << "{" << "\\n";
 147  ss << "  shape: " << escape_for_label(nlohmann::json(d.shape).dump()) << ",\\n";
 148  ss << "  dtype: " << escape_for_label(nlohmann::json(d.dtype).dump()) << ",\\n";
 149  ss << "  mem_loc: " << escape_for_label(nlohmann::json(d.mem_loc).dump()) << ",\\n";
 150  ss << "  strides: " << escape_for_label(nlohmann::json(d.strides).dump()) << "\\n";
 151  ss << "}";
 152
 153  return ss.str();
 154}
 155
 1156std::string format_tdesc(const TDesc &d) {
 1157  std::ostringstream ss;
 1158  ss << "{\\n";
 1159  ss << "  shape: " << escape_for_label(nlohmann::json(d.shape).dump()) << ",\\n";
 1160  ss << "  dtype: " << escape_for_label(nlohmann::json(d.dtype).dump()) << "\\n";
 1161  ss << "  mem_loc: " << escape_for_label(nlohmann::json(d.mem_loc).dump()) << "\\n";
 1162  ss << "  strides: " << escape_for_label(nlohmann::json(d.strides).dump()) << "\\n";
 1163  ss << "  offset: " << d.offset << "\\n";
 1164  ss << "}";
 1165  return ss.str();
 1166}
 167
 168} // namespace
 169namespace holoflow::runtime {
 170
 1171static bool uses_section_layout(const GraphCompiledDumpPreferences &prefs) {
 1172  return prefs.dump_section_info && prefs.layout != GraphCompiledDumpPreferences::Layout::Normal;
 1173}
 174
 1175static bool uses_block_layout(const GraphCompiledDumpPreferences &prefs) {
 1176  return prefs.dump_section_info && prefs.layout == GraphCompiledDumpPreferences::Layout::Block;
 1177}
 178
 1179static bool uses_snake_layout(const GraphCompiledDumpPreferences &prefs) {
 1180  return prefs.dump_section_info && prefs.layout == GraphCompiledDumpPreferences::Layout::Snake;
 1181}
 182
 183static void write_compiled_graph_header(std::ostringstream                 &ss,
 184                                        const GraphCompiledDumpPreferences &prefs,
 1185                                        const std::string &title = "holoflow_compiled_graph") {
 1186  ss << "digraph " << title << " {\n";
 1187  if (!uses_section_layout(prefs) &&
 188      prefs.rankdir == GraphCompiledDumpPreferences::Rankdir::LeftToRight)
 1189    ss << "  rankdir=LR;\n";
 190  else
 0191    ss << "  rankdir=TB;\n";
 192
 1193  ss << "  compound=true;\n";
 1194  if (uses_block_layout(prefs) || uses_snake_layout(prefs)) {
 0195    ss << "  newrank=true;\n";
 196  }
 1197  if (uses_block_layout(prefs)) {
 0198    ss << "  splines=polyline;\n";
 1199  } else if (uses_snake_layout(prefs) && prefs.dump_edge_descriptions) {
 0200    ss << "  splines=line;\n";
 201  }
 1202  if (uses_section_layout(prefs)) {
 0203    ss << "  nodesep=0.8;\n";
 0204    ss << "  ranksep=1.2;\n";
 205  }
 1206  ss << "  node [fontname=\"Helvetica\", shape=box, style=filled];\n";
 1207  ss << "  edge [fontname=\"Helvetica\"];\n\n";
 1208}
 209
 210static void write_compiled_nodes(std::ostringstream &ss, const runtime::GraphPlan &g,
 211                                 const holoflow::runtime::ExecResouces &res,
 1212                                 const GraphCompiledDumpPreferences    &prefs) {
 213
 214  auto fmt_id = [&](int tid) -> std::string {
 215    if (res.tid_to_sid.contains(tid)) {
 216      return std::format("{}(s:{})", tid, res.tid_to_sid.at(tid));
 217    }
 218    return std::to_string(tid);
 1219  };
 220
 221  auto get_visual_id = [&](size_t v, bool is_source) -> std::string {
 222    if (g[v].infer.kind == core::TaskKind::Async) {
 223      return is_source ? std::format("v{}_out", v) : std::format("v{}_in", v);
 224    }
 225    return std::format("v{}", v);
 1226  };
 227
 1228  std::string ids_line = "";
 1229  for (auto v : boost::make_iterator_range(boost::vertices(g))) {
 1230    const auto &np = g[v];
 231
 1232    std::ostringstream label_base;
 1233    if (prefs.dump_node_name)
 1234      label_base << (np.spec.name.empty() ? "(unnamed)" : np.spec.name);
 235
 1236    if (prefs.dump_node_settings && np.spec.debug && !np.spec.settings.is_null() &&
 237        !(np.spec.settings.is_object() && np.spec.settings.empty())) {
 1238      if (!label_base.str().empty()) {
 1239        label_base << "\n";
 240      }
 1241      label_base << replace_newlines_with_l(dump_json_with_floating_point_precision(
 242                        np.spec.settings, prefs.floating_point_precision))
 243                 << "\\l";
 244    }
 245
 1246    if (prefs.dump_node_in_out_tids) {
 1247      std::string in_str = "[";
 1248      for (size_t i = 0; i < np.in_tids.size(); ++i) {
 1249        in_str += (i ? "," : "") + fmt_id(np.in_tids[i]);
 1250      }
 1251      in_str += "]";
 252
 1253      std::string out_str = "[";
 1254      for (size_t i = 0; i < np.out_tids.size(); ++i) {
 1255        const int   out_tid = np.out_tids[i];
 1256        std::string id_text = fmt_id(out_tid);
 257
 1258        bool is_alias = false;
 1259        if (res.tid_to_sid.count(out_tid)) {
 1260          const size_t out_sid = res.tid_to_sid.at(out_tid);
 1261          for (int in_tid : np.in_tids) {
 0262            if (res.tid_to_sid.count(in_tid) && res.tid_to_sid.at(in_tid) == out_sid) {
 0263              is_alias = true;
 0264              break;
 265            }
 0266          }
 267        }
 268
 1269        out_str += (i ? "," : "") + id_text + (is_alias ? "*" : "");
 1270      }
 1271      out_str += "]";
 272
 1273      ids_line = "\nIn: " + in_str + "\nOut: " + out_str;
 1274    }
 275
 1276    if (prefs.dump_node_kind) {
 1277      if (np.infer.kind == core::TaskKind::Async) {
 1278        const std::string label_in = label_base.str() + "\n(Producer/Write)" + ids_line + "\n";
 1279        ss << std::format("  v{}_in [label=\"{}\", shape=invhouse, fillcolor=\"#e6f2ff\", "
 280                          "color=\"#0066cc\", style=\"filled,dashed\"];\n",
 281                          v, escape_for_label(label_in));
 282
 1283        const std::string label_out = label_base.str() + "\n(Consumer/Read)" + ids_line + "\n";
 1284        ss << std::format("  v{}_out [label=\"{}\", shape=house, fillcolor=\"#ffe6e6\", "
 285                          "color=\"#cc0000\", style=\"filled,dashed\"];\n",
 286                          v, escape_for_label(label_out));
 287
 1288        if (uses_block_layout(prefs)) {
 0289          ss << std::format("  v{}_in:e -> v{}_out:w [style=dotted, color=\"#888888\", "
 290                            "penwidth=2, arrowhead=none, label=\"Async Signal\", "
 291                            "constraint=false];\n",
 292                            v, v);
 0293        } else {
 1294          ss << std::format("  v{}_in -> v{}_out [style=dotted, color=\"#888888\", penwidth=2, "
 295                            "arrowhead=none, label=\"Async Signal\"];\n",
 296                            v, v);
 297        }
 1298      } else {
 1299        const std::string label = label_base.str() + "\n(" + np.spec.kind + ")" + ids_line + "\n";
 1300        ss << std::format("  v{} [label=\"{}\", fillcolor=\"#ccffcc\"];\n", v,
 301                          escape_for_label(label));
 1302      }
 303    }
 1304  }
 1305}
 306
 307static std::vector<size_t>
 308get_section_layout_order(const std::vector<runtime::Section> &sections);
 309
 310static void write_compiled_edges(std::ostringstream                    &ss,
 311                                 const runtime::GraphPlan              &g,
 312                                 const holoflow::runtime::ExecResouces &res,
 313                                 const GraphCompiledDumpPreferences    &prefs,
 1314                                 const std::vector<runtime::Section>   &sections) {
 315
 316  auto get_visual_id = [&](size_t v, bool is_source) -> std::string {
 317    if (g[v].infer.kind == core::TaskKind::Async) {
 318      return is_source ? std::format("v{}_out", v) : std::format("v{}_in", v);
 319    }
 320    return std::format("v{}", v);
 1321  };
 322
 1323  std::vector<size_t> section_positions(sections.size());
 1324  if (uses_snake_layout(prefs)) {
 0325    const auto order = get_section_layout_order(sections);
 0326    for (size_t position = 0; position < order.size(); ++position) {
 0327      section_positions[order[position]] = position;
 0328    }
 0329  }
 330
 331  auto contains_vertex = [](const auto &vertices, auto vertex) {
 332    return std::find(vertices.begin(), vertices.end(), vertex) != vertices.end();
 333  };
 334
 1335  for (auto e : boost::make_iterator_range(boost::edges(g))) {
 1336    const auto  u  = boost::source(e, g);
 1337    const auto  v  = boost::target(e, g);
 1338    const auto &ep = g[e];
 339
 1340    const std::string u_vis = get_visual_id(u, true);
 1341    const std::string v_vis = get_visual_id(v, false);
 342
 1343    bool reverse_edge = false;
 1344    if (uses_snake_layout(prefs)) {
 0345      for (size_t section_idx = 0; section_idx < sections.size(); ++section_idx) {
 0346        const auto &section = sections[section_idx];
 0347        const bool  contains_source =
 348            g[u].infer.kind == core::TaskKind::Async
 349                ? contains_vertex(section.async_cons, u)
 350                : contains_vertex(section.sync_topo, u);
 0351        const bool contains_target =
 352            g[v].infer.kind == core::TaskKind::Async
 353                ? contains_vertex(section.async_prod, v)
 354                : contains_vertex(section.sync_topo, v);
 0355        if (contains_source && contains_target) {
 0356          reverse_edge = section_positions[section_idx] % 2 != 0;
 0357          break;
 358        }
 0359      }
 360    }
 361
 1362    std::ostringstream edge_lbl;
 1363    edge_lbl << "tid:" << ep.tid;
 1364    if (res.tid_to_sid.count(ep.tid)) {
 1365      edge_lbl << " (s:" << res.tid_to_sid.at(ep.tid) << ")";
 366    }
 1367    edge_lbl << "\\n" << format_tdesc(ep.desc);
 368
 1369    ss << std::format("  {} -> {} ", reverse_edge ? v_vis : u_vis,
 370                      reverse_edge ? u_vis : v_vis);
 1371    if (reverse_edge) {
 0372      ss << "[dir=back]";
 373    }
 1374    if (prefs.dump_edge_indices) {
 1375      ss << std::format("[taillabel=\"{}\", headlabel=\"{}\"]",
 376                        reverse_edge ? ep.spec.in_idx : ep.spec.out_idx,
 377                        reverse_edge ? ep.spec.out_idx : ep.spec.in_idx);
 378    }
 1379    if (prefs.dump_edge_descriptions) {
 1380      auto formated = replace_newlines_escaped_with_l(edge_lbl.str());
 1381      ss << std::format("[label=\"{}\\l\"]", formated);
 1382    }
 1383    ss << ";\n";
 1384  }
 1385}
 386
 387static void write_compiled_resources(std::ostringstream                    &ss,
 1388                                     const holoflow::runtime::ExecResouces &res) {
 1389  ss << "  // --- resources summary ---\n";
 390
 1391  ss << "  // streams: ";
 1392  bool first = true;
 1393  for (const auto &[id, stream] : res.streams) {
 394    (void)stream;
 1395    ss << (first ? "" : ", ") << id;
 1396    first = false;
 1397  }
 1398  ss << "\n";
 399
 1400  ss << "  // tasks: ";
 1401  first = true;
 1402  for (const auto &[name, task] : res.tasks) {
 403    (void)task;
 1404    ss << (first ? "" : ", ") << name;
 1405    first = false;
 1406  }
 1407  ss << "\n\n";
 1408}
 409
 0410static std::vector<size_t> get_section_layout_order(const std::vector<runtime::Section> &sections) {
 0411  std::vector<std::vector<size_t>> successors(sections.size());
 0412  std::vector<size_t>              indegrees(sections.size(), 0);
 413
 0414  for (size_t source_idx = 0; source_idx < sections.size(); ++source_idx) {
 0415    for (const auto vertex : sections[source_idx].async_prod) {
 0416      for (size_t target_idx = 0; target_idx < sections.size(); ++target_idx) {
 417        if (source_idx == target_idx ||
 418            std::find(sections[target_idx].async_cons.begin(),
 419                      sections[target_idx].async_cons.end(),
 0420                      vertex) == sections[target_idx].async_cons.end() ||
 421            std::find(successors[source_idx].begin(), successors[source_idx].end(), target_idx) !=
 422                successors[source_idx].end()) {
 0423          continue;
 424        }
 425
 0426        successors[source_idx].push_back(target_idx);
 0427        ++indegrees[target_idx];
 0428      }
 0429    }
 0430  }
 431
 0432  std::priority_queue<size_t, std::vector<size_t>, std::greater<>> ready;
 0433  for (size_t section_idx = 0; section_idx < sections.size(); ++section_idx) {
 0434    if (indegrees[section_idx] == 0) {
 0435      ready.push(section_idx);
 436    }
 0437  }
 438
 0439  std::vector<size_t> order;
 0440  order.reserve(sections.size());
 0441  while (!ready.empty()) {
 0442    const size_t section_idx = ready.top();
 0443    ready.pop();
 0444    order.push_back(section_idx);
 445
 0446    for (const size_t successor : successors[section_idx]) {
 0447      if (--indegrees[successor] == 0) {
 0448        ready.push(successor);
 449      }
 0450    }
 0451  }
 452
 0453  if (order.size() != sections.size()) {
 0454    for (size_t section_idx = 0; section_idx < sections.size(); ++section_idx) {
 0455      if (std::find(order.begin(), order.end(), section_idx) == order.end()) {
 0456        order.push_back(section_idx);
 457      }
 0458    }
 459  }
 0460  return order;
 0461}
 462
 463static void write_compiled_sections(std::ostringstream                  &ss,
 464                                    const std::vector<runtime::Section> &sections,
 1465                                    const GraphCompiledDumpPreferences  &prefs) {
 1466  const bool row_layout   = uses_section_layout(prefs);
 1467  const bool block_layout = uses_block_layout(prefs);
 1468  const bool snake_layout = uses_snake_layout(prefs);
 469
 1470  std::vector<size_t> section_positions(sections.size());
 1471  if (snake_layout) {
 0472    const auto order = get_section_layout_order(sections);
 0473    for (size_t position = 0; position < order.size(); ++position) {
 0474      section_positions[order[position]] = position;
 0475    }
 0476  }
 477
 1478  for (size_t section_idx = 0; section_idx < sections.size(); ++section_idx) {
 1479    const auto &sec = sections[section_idx];
 1480    ss << std::format("  subgraph cluster_section_{} {{\n", sec.id);
 481
 1482    ss << std::format("    label=\"Section {}", sec.id);
 1483    if (!sec.name.empty()) {
 1484      ss << ": " << escape_for_label(sec.name);
 485    }
 1486    if (prefs.dump_section_stream_addr) {
 1487      ss << std::format(" (Stream {})", (void *)sec.stream);
 488    }
 1489    ss << std::format("\\l\";\n");
 490
 1491    ss << "    style=rounded; color=gray; bgcolor=\"#f8f8f8\";\n";
 492
 1493    if (row_layout) {
 0494      ss << "    { rank=same;\n";
 495    }
 496
 1497    std::vector<std::string> visual_ids;
 498    auto append_visual_ids = [&](const auto &vertices, std::string_view suffix) {
 499      for (const auto vertex : vertices) {
 500        visual_ids.push_back(std::format("v{}{}", vertex, suffix));
 501      }
 1502    };
 503    auto append_visual_ids_reversed = [&](const auto &vertices, std::string_view suffix) {
 504      for (auto vertex = vertices.rbegin(); vertex != vertices.rend(); ++vertex) {
 505        visual_ids.push_back(std::format("v{}{}", *vertex, suffix));
 506      }
 1507    };
 508
 1509    if (block_layout) {
 0510      ss << std::format("      section_layout_{}_left [shape=point, width=0, height=0, "
 511                        "label=\"\", style=invis, group=section_layout_left];\n",
 512                        section_idx);
 0513      append_visual_ids(sec.async_cons, "_out");
 0514      append_visual_ids(sec.sync_topo, "");
 0515      append_visual_ids(sec.async_prod, "_in");
 1516    } else if (snake_layout && section_positions[section_idx] % 2 != 0) {
 0517      append_visual_ids_reversed(sec.async_prod, "_in");
 0518      append_visual_ids_reversed(sec.sync_topo, "");
 0519      append_visual_ids_reversed(sec.async_cons, "_out");
 1520    } else if (snake_layout) {
 0521      append_visual_ids(sec.async_cons, "_out");
 0522      append_visual_ids(sec.sync_topo, "");
 0523      append_visual_ids(sec.async_prod, "_in");
 0524    } else {
 1525      append_visual_ids(sec.sync_topo, "");
 1526      append_visual_ids(sec.async_prod, "_in");
 1527      append_visual_ids(sec.async_cons, "_out");
 528    }
 529
 1530    for (const auto &visual_id : visual_ids) {
 1531      ss << std::format("      {};\n", visual_id);
 1532    }
 533
 1534    if (block_layout && !visual_ids.empty()) {
 0535      ss << std::format("      section_layout_{}_left", section_idx);
 0536      for (const auto &visual_id : visual_ids) {
 0537        ss << " -> " << visual_id;
 0538      }
 0539      ss << " [style=invis, weight=1000];\n";
 1540    } else if (snake_layout && visual_ids.size() > 1) {
 0541      ss << "      " << visual_ids.front();
 0542      for (size_t visual_idx = 1; visual_idx < visual_ids.size(); ++visual_idx) {
 0543        ss << " -> " << visual_ids[visual_idx];
 0544      }
 0545      ss << " [style=invis, weight=1000];\n";
 546    }
 1547    if (row_layout) {
 0548      ss << "    }\n";
 549    }
 1550    ss << "  }\n";
 1551  }
 552
 1553  if (block_layout && sections.size() > 1) {
 0554    const auto order = get_section_layout_order(sections);
 0555    ss << std::format("  section_layout_{}_left", order.front());
 0556    for (size_t order_idx = 1; order_idx < order.size(); ++order_idx) {
 0557      ss << std::format(" -> section_layout_{}_left", order[order_idx]);
 0558    }
 0559    ss << " [style=invis, weight=100000];\n";
 0560  }
 1561}
 562
 563std::string to_dot(const CompilerOutput &out, const GraphCompiledDumpPreferences &prefs,
 1564                   std::string filename) {
 1565  std::ostringstream ss;
 1566  write_compiled_graph_header(ss, prefs, filename);
 567
 1568  if (prefs.dump_resource_info) {
 1569    write_compiled_resources(ss, out.resources);
 570  }
 1571  write_compiled_nodes(ss, out.graph, out.resources, prefs);
 1572  ss << "\n";
 1573  write_compiled_edges(ss, out.graph, out.resources, prefs, out.sections);
 1574  ss << "\n";
 1575  if (prefs.dump_section_info) {
 1576    write_compiled_sections(ss, out.sections, prefs);
 577  }
 1578  ss << "}\n";
 1579  return ss.str();
 1580}
 581
 582} // namespace holoflow::runtime
 583
 584namespace holoflow::core {
 585
 1586static void write_graph_header(std::ostringstream &ss, const GraphSpecDumpPreferences &dump_prefs) {
 1587  ss << "digraph holoflow_graph {\n";
 1588  if (dump_prefs.rankdir == GraphSpecDumpPreferences::Rankdir::LeftToRight) {
 1589    ss << "  rankdir=LR;\n";
 1590  } else {
 0591    ss << "  rankdir=TB;\n";
 592  }
 1593  ss << "  node [shape=box, fontname=\"Helvetica\"];\n";
 1594  ss << "  edge [fontname=\"Helvetica\"];\n\n";
 1595}
 596
 597static void write_nodes(std::ostringstream &ss, const GraphSpec &g,
 1598                        const GraphSpecDumpPreferences &dump_prefs) {
 599  using vertex_iter_t = boost::graph_traits<GraphSpec>::vertex_iterator;
 1600  vertex_iter_t vi, vi_end;
 1601  for (boost::tie(vi, vi_end) = boost::vertices(g); vi != vi_end; ++vi) {
 1602    auto            v  = *vi;
 1603    const NodeSpec &ns = g[v];
 604
 1605    std::ostringstream label;
 1606    if (dump_prefs.dump_node_name) {
 1607      if (!ns.name.empty())
 1608        label << ns.name;
 609      else
 1610        label << "(unnamed)";
 611    }
 612
 1613    if (dump_prefs.dump_node_kind && !ns.kind.empty()) {
 1614      if (!label.str().empty())
 1615        label << "\n";
 1616      label << "(" << ns.kind << ")\n";
 617    }
 618
 1619    if (dump_prefs.dump_node_settings && ns.debug && !ns.settings.is_null() &&
 620        !(ns.settings.is_object() && ns.settings.empty())) {
 1621      std::string settings_dump =
 622          dump_json_with_floating_point_precision(ns.settings, dump_prefs.floating_point_precision);
 1623      label << replace_newlines_with_l(settings_dump) << "\\l";
 1624    }
 625
 1626    std::string esc_label = escape_for_label(label.str());
 1627    ss << "  v" << v << " [label=\"" << esc_label << "\"];\n";
 1628  }
 1629  ss << "\n";
 1630}
 631
 632static void write_edges(std::ostringstream &ss, const GraphSpec &g,
 1633                        const GraphSpecDumpPreferences &dump_prefs) {
 634  using edge_iter_t = boost::graph_traits<GraphSpec>::edge_iterator;
 1635  edge_iter_t ei, ei_end;
 1636  for (boost::tie(ei, ei_end) = boost::edges(g); ei != ei_end; ++ei) {
 1637    auto            e  = *ei;
 1638    auto            s  = boost::source(e, g);
 1639    auto            t  = boost::target(e, g);
 1640    const EdgeSpec &es = g[e];
 641
 1642    ss << "  v" << s << " -> v" << t;
 1643    if (dump_prefs.dump_edge_indices) {
 1644      ss << " [taillabel=\"" << escape_for_label(std::to_string(es.out_idx)) << "\""
 645         << " headlabel=\"" << escape_for_label(std::to_string(es.in_idx)) << "\"]";
 646    }
 1647    ss << ";\n";
 1648  }
 1649}
 650
 1651std::string to_dot(const GraphSpec &g, const GraphSpecDumpPreferences &dump_prefs) {
 1652  std::ostringstream ss;
 1653  write_graph_header(ss, dump_prefs);
 1654  write_nodes(ss, g, dump_prefs);
 1655  write_edges(ss, g, dump_prefs);
 1656  ss << "}\n";
 1657  return ss.str();
 1658}
 659
 660} // namespace holoflow::core

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

#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#define NOMINMAX
 16
 17#include "holoflow/runtime/graph_exec.hh"
 18
 19#include <algorithm>
 20#include <boost/graph/adjacency_list.hpp>
 21#include <boost/graph/graph_traits.hpp>
 22#include <chrono>
 23#include <map>
 24#include <mutex>
 25#include <nvtx3/nvtx3.hpp>
 26#include <vector>
 27#include <windows.h>
 28
 29#include "boost/graph/properties.hpp"
 30#include "boost/range/iterator_range_core.hpp"
 31#include "bug.hh"
 32#include "cuda_runtime_api.h"
 33#include "curaii/cuda.hh"
 34#include "holoflow/core/tasks.hh"
 35#include "holoflow/core/tensor.hh"
 36#include "logger.hh"
 37
 38namespace holoflow::runtime {
 39namespace {
 40
 41inline std::string node_context_msg(const holoflow::runtime::GraphPlan             &g,
 42                                    holoflow::runtime::GraphPlan::vertex_descriptor v,
 43                                    std::string_view phase, int section_id,
 044                                    std::string_view section_name) {
 045  const auto &np       = g[v];
 046  const auto  vertex_i = static_cast<std::size_t>(boost::get(boost::vertex_index, g, v));
 047  const auto  tid      = ::GetCurrentThreadId();
 48
 049  return std::format("Exception in node '{}'\n"
 50                     "  phase: {}\n"
 51                     "  section: {} (id={})\n"
 52                     "  vertex_idx: {}\n"
 53                     "  thread_id: {}\n",
 54                     np.spec.name, phase, section_name, section_id, vertex_i, tid);
 055}
 56
 57[[noreturn]] inline void
 58rethrow_with_node_context(const holoflow::runtime::GraphPlan             &g,
 59                          holoflow::runtime::GraphPlan::vertex_descriptor v, std::string_view phase,
 060                          int section_id, std::string_view section_name) {
 61  try {
 062    throw; // rethrow current exception
 063  } catch (const std::exception &e) {
 064    throw std::runtime_error(node_context_msg(g, v, phase, section_id, section_name) +
 65                             std::string{"  what: "} + e.what());
 066  } catch (...) {
 067    throw std::runtime_error(node_context_msg(g, v, phase, section_id, section_name) +
 68                             "  what: <non-std exception>");
 069  }
 070}
 71
 172size_t count_distinct_tids(const GraphPlan &graph) {
 173  std::set<int> tids;
 174  for (const auto &v : boost::make_iterator_range(boost::vertices(graph))) {
 175    const auto &np   = graph[v];
 176    const auto tids_ = std::array{std::span{np.in_tids}, std::span{np.out_tids}} | std::views::join;
 77
 178    for (const auto tid : tids_) {
 179      tids.insert(tid);
 180    }
 181  }
 182  return tids.size();
 183}
 84
 185std::string task_range_name(std::string_view operation, const core::NodeSpec &spec) {
 186  return std::format("{}: {} ({})", operation, spec.name, spec.kind);
 187}
 88
 089[[noreturn]] inline void log_and_abort_current_exception(std::string_view thread_name) {
 90  try {
 091    throw;
 092  } catch (const std::exception &e) {
 093    logger()->critical("[{}] Fatal runtime exception:\n{}", thread_name, e.what());
 094  } catch (...) {
 095    logger()->critical("[{}] Fatal runtime exception: <non-std exception>", thread_name);
 096  }
 97
 098  logger()->flush();
 099  std::abort();
 0100}
 101
 102} // namespace
 103
 1104void *MemoryBlock::get() {
 1105  void *ptr = nullptr;
 106
 1107  switch (mem_loc) {
 108  case core::MemLoc::Host:
 1109    ptr = h_data.get();
 1110    break;
 111  case core::MemLoc::Device:
 1112    ptr = d_data.get();
 113    break;
 114  }
 115
 1116  HOLOFLOW_CHECK(ptr != nullptr, "MemoryBlock pointer is null");
 1117  return ptr;
 1118}
 119
 120Scheduler::Scheduler(const GraphPlan &graph, const std::vector<Section> &sections,
 121                     ExecResouces &resources, std::chrono::milliseconds metrics_interval)
 1122    : graph_(graph), sections_(sections), res_(resources), metrics_interval_(metrics_interval) {
 123
 1124  if (metrics_interval_.count() <= 0) {
 1125    metrics_interval_ = std::chrono::milliseconds{1};
 126  }
 127
 128  // init_tensor_tables();
 129  // bind_resource_tensors();
 1130  init_tviews();
 1131  build_event_handles();
 1132  build_nodes_rts();
 1133}
 134
 1135Scheduler::~Scheduler() {
 1136  if (is_running()) {
 0137    request_stop();
 0138    wait();
 0139  } else {
 1140    stop_metrics_thread();
 141  }
 1142}
 143
 0144void Scheduler::set_metrics_interval(std::chrono::milliseconds interval) {
 0145  if (interval.count() <= 0) {
 0146    interval = std::chrono::milliseconds{1};
 147  }
 148  {
 0149    std::lock_guard<std::mutex> lock(metrics_thread_mutex_);
 0150    metrics_interval_ = interval;
 0151  }
 0152  metrics_cv_.notify_all();
 0153}
 154
 1155std::map<std::string, NodeMetrics> Scheduler::metrics() const {
 1156  std::lock_guard<std::mutex> lock(metrics_mutex_);
 1157  return latest_metrics_;
 1158}
 159
 1160void Scheduler::start() {
 1161  logger()->info("[Scheduler::start] Starting scheduler");
 1162  if (running_.exchange(true)) {
 1163    logger()->warn("[Scheduler::start] Scheduler is already running");
 1164    return;
 165  }
 166
 1167  stop_.store(false);
 1168  reset_metrics_state();
 1169  start_metrics_thread();
 1170  threads_.reserve(sections_.size());
 1171  for (size_t i = 0; i < sections_.size(); i++) {
 1172    threads_.emplace_back(&Scheduler::run_section, this, static_cast<int>(i));
 1173  }
 174
 1175  threads_.emplace_back(&Scheduler::run_router, this);
 1176}
 177
 1178void Scheduler::request_stop() {
 1179  logger()->info("[Scheduler::request_stop] Requesting scheduler to stop");
 1180  if (!running_.load()) {
 1181    logger()->warn("[Scheduler::request_stop] Scheduler is not running");
 1182    return;
 183  }
 1184  if (stop_.exchange(true)) {
 1185    logger()->warn("[Scheduler::request_stop] Stop already requested");
 186    return;
 187  }
 1188}
 189
 1190void Scheduler::wait() {
 1191  logger()->info("[Scheduler::wait] Waiting for scheduler to stop");
 1192  for (auto &t : threads_) {
 1193    auto tid = GetThreadId(static_cast<HANDLE>(t.native_handle()));
 1194    logger()->debug("[Scheduler::wait] Joining thread {}...", tid);
 1195    t.join();
 1196    logger()->debug("[Scheduler::wait] Thread {} joined", tid);
 1197  }
 198
 1199  threads_.clear();
 1200  logger()->info("[Scheduler::wait] Scheduler stopped");
 1201  running_.store(false);
 1202  stop_metrics_thread();
 203  // TODO: Is this really the best place to reset running_?
 1204}
 205
 1206bool Scheduler::is_running() const { return running_.load(); }
 207
 1208bool Scheduler::stop_requested() const { return stop_.load(); }
 209
 0210bool Scheduler::ui_try_send(const std::string &node_id, nlohmann::json &&data) noexcept {
 0211  return router_.ui_try_send(node_id, std::move(data));
 0212}
 213
 0214std::optional<holoflow_event::Event> Scheduler::ui_try_receive() noexcept {
 0215  return router_.ui_try_receive();
 0216}
 217
 1218void Scheduler::init_tviews() {
 1219  auto nb_tids = count_distinct_tids(graph_);
 1220  tviews_.assign(nb_tids, core::TView{});
 221
 1222  for (auto &[tid, tdesc] : res_.tensor_descs) {
 1223    auto sid        = res_.tid_to_sid.at(tid);
 1224    auto storage    = res_.storages.at(sid).get();
 1225    tviews_.at(tid) = core::TView{tdesc, storage};
 1226  }
 1227}
 228
 229// void Scheduler::init_tensor_tables() {
 230//   auto nb_tids = count_distinct_tids(graph_);
 231//   storages_.assign(nb_tids, core::Storage{});
 232//   tviews_.assign(nb_tids, core::TView{});
 233// }
 234
 235// void Scheduler::bind_resource_tensors() {
 236// for (auto &[tid, tensor] : res_.tensors) {
 237//   auto v = tensor.view();
 238
 239//   storages_.at(tid) = *v.storage;
 240//   tviews_.at(tid)   = core::TView{v.desc, &storages_.at(tid)};
 241// }
 242// }
 243
 1244void Scheduler::build_event_handles() {
 1245  event_handles_.clear();
 1246  for (auto v : boost::make_iterator_range(boost::vertices(graph_))) {
 1247    const auto &np = graph_[v];
 1248    event_handles_.emplace(np.spec.name, router_.bind_node(np.spec.name));
 1249  }
 1250}
 251
 1252void Scheduler::build_nodes_rts() {
 1253  const auto num_vertices = boost::num_vertices(graph_);
 1254  node_rts_.resize(num_vertices);
 1255  node_names_.resize(num_vertices);
 1256  metric_accumulators_.resize(num_vertices);
 257
 1258  for (auto v : boost::make_iterator_range(boost::vertices(graph_))) {
 1259    const auto idx      = boost::get(boost::vertex_index, graph_, v);
 1260    const auto np       = graph_[v];
 1261    node_names_.at(idx) = np.spec.name;
 1262    auto *task          = res_.tasks.at(np.spec.name).get();
 1263    HOLOFLOW_CHECK(task != nullptr, "Task for node {} is null", np.spec.name);
 264
 265    // Build in_views and out_views
 1266    std::vector<core::TView> in_views(np.in_tids.size());
 1267    std::vector<core::TView> out_views(np.out_tids.size());
 268
 1269    for (size_t i = 0; i < np.in_tids.size(); ++i) {
 1270      int tid     = np.in_tids[i];
 1271      in_views[i] = tviews_.at(tid);
 1272    }
 273
 1274    for (size_t i = 0; i < np.out_tids.size(); ++i) {
 1275      int tid      = np.out_tids[i];
 1276      out_views[i] = tviews_.at(tid);
 1277    }
 278
 1279    if (auto *st = dynamic_cast<core::ISyncTask *>(task)) {
 1280      SyncRt srt;
 1281      srt.task             = st;
 1282      srt.in_views         = in_views;
 1283      srt.out_views        = out_views;
 1284      srt.ctx.inputs       = srt.in_views;
 1285      srt.ctx.outputs      = srt.out_views;
 1286      srt.ctx.cancelled    = &stop_;
 1287      srt.ctx.event_writer = &event_handles_.at(np.spec.name).out;
 1288      srt.ctx.event_reader = &event_handles_.at(np.spec.name).in;
 1289      node_rts_.at(idx)    = std::move(srt);
 1290    } else if (auto *at = dynamic_cast<core::IAsyncTask *>(task)) {
 1291      AsyncRt art;
 1292      art.task           = at;
 1293      art.in_views       = in_views;
 1294      art.out_views      = out_views;
 1295      art.pctx.inputs    = art.in_views;
 1296      art.pctx.cancelled = &stop_;
 1297      art.xctx.outputs   = art.out_views;
 1298      art.xctx.cancelled = &stop_;
 1299      node_rts_.at(idx)  = std::move(art);
 1300    } else {
 0301      HOLOFLOW_BUG("Task for node {} is neither sync nor async", np.spec.name);
 302    }
 1303  }
 1304}
 305
 1306void Scheduler::run_router() {
 307  try {
 1308    logger()->info("[Scheduler::run_router] Starting event router");
 1309    while (!stop_.load()) {
 1310      router_.tick();
 1311      std::this_thread::sleep_for(std::chrono::milliseconds(1));
 1312    }
 1313    logger()->info("[Scheduler::run_router] Event router stopped");
 0314  } catch (...) {
 0315    stop_.store(true);
 0316    log_and_abort_current_exception("Scheduler::run_router");
 0317  }
 1318}
 319
 1320void Scheduler::run_section(int section_id) {
 1321  const auto &sec    = sections_.at(section_id);
 1322  auto        stream = sec.stream;
 323
 324  // Define a consistent, professional color palette for your timeline
 1325  constexpr nvtx3::color color_section{0x555555}; // Dark Gray
 1326  constexpr nvtx3::color color_acquire{0xFF8C00}; // Dark Orange
 1327  constexpr nvtx3::color color_async_c{0x1E90FF}; // Dodger Blue
 1328  constexpr nvtx3::color color_sync{0x32CD32};    // Lime Green
 1329  constexpr nvtx3::color color_async_p{0x8A2BE2}; // Blue Violet
 1330  constexpr nvtx3::color color_release{0xFF4500}; // Orange Red
 331
 332  try {
 1333    while (!stop_.load()) {
 1334      std::vector<GraphPlan::vertex_descriptor> produced_owned_outputs;
 1335      logger()->trace("[Scheduler::run_section] Running section {}", sec.name);
 336
 337      // 1. Outer Section Range
 338      // Automatically popped at the end of this while-loop iteration,
 339      // safely handling the 'break' statements below.
 1340      nvtx3::scoped_range section_range{nvtx3::event_attributes{sec.name.c_str(), color_section}};
 341
 342      // 2. Acquire owned inputs
 343      {
 1344        nvtx3::scoped_range r{nvtx3::event_attributes{"Acquire owned inputs", color_acquire}};
 1345        for (auto v : sec.sync_topo) {
 346          try {
 1347            acquire_owned_inputs(v);
 0348          } catch (...) {
 0349            rethrow_with_node_context(graph_, v, "acquire_owned_inputs", section_id, sec.name);
 0350          }
 1351        }
 1352        for (auto v : sec.async_prod) {
 353          try {
 1354            acquire_owned_inputs(v);
 0355          } catch (...) {
 0356            rethrow_with_node_context(graph_, v, "acquire_owned_inputs", section_id, sec.name);
 0357          }
 1358        }
 1359      } // <-- Range automatically pops here
 360
 1361      if (stop_.load()) {
 1362        break; // Safe! `section_range` will cleanly pop on its way out.
 363      }
 364
 365      // 3. Execute async consumers
 366      {
 1367        nvtx3::scoped_range r{nvtx3::event_attributes{"Execute async consumers", color_async_c}};
 1368        for (auto v : sec.async_cons) {
 369          try {
 1370            if (run_async_cons(v) == core::OpResult::Ok) {
 1371              produced_owned_outputs.push_back(v);
 372            }
 0373          } catch (...) {
 0374            rethrow_with_node_context(graph_, v, "execute_async_consumer", section_id, sec.name);
 0375          }
 1376          if (stop_.load())
 1377            break;
 1378        }
 1379      }
 380
 381      // 4. Execute sync nodes
 1382      if (!stop_.load()) {
 1383        nvtx3::scoped_range r{nvtx3::event_attributes{"Execute sync nodes", color_sync}};
 1384        for (auto v : sec.sync_topo) {
 385          try {
 1386            if (run_sync(v) == core::OpResult::Ok) {
 1387              produced_owned_outputs.push_back(v);
 388            }
 0389          } catch (...) {
 0390            rethrow_with_node_context(graph_, v, "execute_sync", section_id, sec.name);
 0391          }
 1392          if (stop_.load())
 1393            break;
 1394        }
 395
 396        // A compiler-ordered synchronizing producer runs first below, so its later barrier covers
 397        // both these kernels and its own CUDA launch while preserving safe publication by ordinary
 398        // producers. Sections without that capability retain the explicit scheduler barrier.
 1399        if (!sec.has_synchronizing_async_producer) {
 1400          CUDA_CHECK(cudaStreamSynchronize(stream));
 401        }
 1402      }
 403
 404      // 5. Execute async producers
 1405      if (!stop_.load()) {
 1406        nvtx3::scoped_range r{nvtx3::event_attributes{"Execute async producers", color_async_p}};
 1407        for (auto v : sec.async_prod) {
 408          try {
 1409            (void)run_async_prod(v);
 0410          } catch (...) {
 0411            rethrow_with_node_context(graph_, v, "execute_async_producer", section_id, sec.name);
 0412          }
 1413          if (stop_.load())
 1414            break;
 1415        }
 1416      }
 417
 418      // 6. Release only outputs produced successfully in this iteration.
 419      {
 1420        nvtx3::scoped_range r{nvtx3::event_attributes{"Release owned outputs", color_release}};
 1421        for (auto v : produced_owned_outputs) {
 422          try {
 1423            release_owned_outputs(v);
 0424          } catch (...) {
 0425            rethrow_with_node_context(graph_, v, "release_owned_outputs", section_id, sec.name);
 0426          }
 1427        }
 1428      }
 429
 1430      if (stop_.load())
 1431        break;
 1432    }
 0433  } catch (...) {
 0434    stop_.store(true);
 0435    log_and_abort_current_exception(
 436        std::format("Scheduler::run_section section={} id={}", sec.name, section_id));
 0437  }
 1438}
 439
 440// void Scheduler::run_section(int section_id) {
 441//   const auto &sec    = sections_.at(section_id);
 442//   auto        stream = sec.stream;
 443
 444//   // Nodes in sections are topologically sorted, so we can execute them in order.
 445//   // However, owned inputs are used as outputs for former nodes, so we need to
 446//   // acquire them first.
 447//   // Owned outputs are used as inputs for later nodes, so we need to release
 448//   // them last.
 449
 450//   // TODO: How to handle end of stream (Eof)? Do we need to propagate it?
 451//   // Do we need to stop the scheduler when we reach Eof for every node?
 452//   // Do we need to notify nodes when we reach Eof for their inputs?
 453
 454//   // TODO: How to handle stream synchronization? Should asynchronous tasks
 455//   // be responsible for synchronizing push-stream before enabling to pop data?
 456//   // Or should the scheduler be the sole responsible for synchronizing streams?
 457
 458//   // TODO: How to properly collect metrics on given tasks run on cuda streams?
 459//   // Should we use cuda events?
 460
 461//   while (!stop_.load()) {
 462//     logger()->trace("[Scheduler::run_section] Running section {}", sec.name);
 463//     nvtxRangePush(sec.name.c_str());
 464
 465//     // Acquire owned inputs.
 466//     //
 467//     // - We do not acquire owned-inputs of async consumers here, as they
 468//     //   used in the former section only.
 469//     //
 470//     // - It is mandatory to check stop_ after acquiring inputs, as
 471//     //   the scheduler may have been requested to stop while waiting
 472//     //   for owned inputs to become available. This leads to undefined
 473//     //   behavior if we proceed to execute nodes after stop_ was set.
 474//     nvtxRangePush("Acquire owned inputs");
 475//     for (auto v : sec.sync_topo) {
 476//       acquire_owned_inputs(v);
 477//     }
 478//     for (auto v : sec.async_prod) {
 479//       acquire_owned_inputs(v);
 480//     }
 481//     nvtxRangePop();
 482//     if (stop_.load()) {
 483//       break;
 484//     }
 485
 486//     // Execute nodes.
 487//     //
 488//     // - We know the nodes are topologically sorted within the section, so we
 489//     //   can execute them in order. The topological order also takes into account
 490//     //   in-place operations, so inputs or siblings is not changed before they are
 491//     //   executed.
 492//     //
 493//     // - It is mandatory to syncronize the stream before running async producers,
 494//     //   as async consumers from the next section may depend on work done on this stream,
 495//     //   and we have no guarantee that the async producer at the end of this section
 496//     //   will synchronize the stream before pushing data (it may not be cuda-related).
 497//     nvtxRangePush("Execute async consumers");
 498//     for (auto v : sec.async_cons) {
 499//       run_async_cons(v);
 500//     }
 501//     if (stop_.load()) {
 502//       break;
 503//     }
 504//     nvtxRangePop();
 505
 506//     nvtxRangePush("Execute sync nodes");
 507//     for (auto v : sec.sync_topo) {
 508//       run_sync(v);
 509//     }
 510
 511//     CUDA_CHECK(cudaStreamSynchronize(stream));
 512//     nvtxRangePop();
 513
 514//     nvtxRangePush("Execute async producers");
 515//     for (auto v : sec.async_prod) {
 516//       run_async_prod(v);
 517//     }
 518//     nvtxRangePop();
 519
 520//     // Release owned outputs.
 521//     //
 522//     // - We do not release owned-outputs of async producers here, as they
 523//     // used only in the next section.
 524//     nvtxRangePush("Release owned outputs");
 525//     for (auto v : sec.sync_topo) {
 526//       release_owned_outputs(v);
 527//     }
 528//     for (auto v : sec.async_cons) {
 529//       release_owned_outputs(v);
 530//     }
 531//     nvtxRangePop();
 532//     nvtxRangePop();
 533//   }
 534// }
 535
 1536void Scheduler::acquire_owned_inputs(GraphPlan::vertex_descriptor v) {
 1537  const auto  idx        = boost::get(boost::vertex_index, graph_, v);
 1538  const auto &np         = graph_[v];
 1539  auto       &nrt        = node_rts_.at(idx);
 1540  const auto &owned_mask = np.infer.owned_inputs;
 1541  auto       *task       = std::holds_alternative<SyncRt>(nrt)
 542                               ? static_cast<core::ITask *>(std::get<SyncRt>(nrt).task)
 543                               : static_cast<core::ITask *>(std::get<AsyncRt>(nrt).task);
 544
 1545  for (size_t i = 0; i < owned_mask.size(); i++) {
 1546    if (!owned_mask.at(i))
 1547      continue;
 548
 1549    std::optional<core::TView> tview = task->acquire_input(static_cast<int>(i));
 1550    while (!tview.has_value()) {
 1551      if (stop_.load())
 0552        return;
 1553      tview = task->acquire_input(static_cast<int>(i));
 1554    }
 1555  }
 1556}
 557
 1558void Scheduler::release_owned_outputs(GraphPlan::vertex_descriptor v) {
 1559  const auto  idx        = boost::get(boost::vertex_index, graph_, v);
 1560  const auto &np         = graph_[v];
 1561  auto       &nrt        = node_rts_.at(idx);
 1562  const auto &owned_mask = np.infer.owned_outputs;
 1563  auto       *task       = std::holds_alternative<SyncRt>(nrt)
 564                               ? static_cast<core::ITask *>(std::get<SyncRt>(nrt).task)
 565                               : static_cast<core::ITask *>(std::get<AsyncRt>(nrt).task);
 566
 1567  for (size_t i = 0; i < owned_mask.size(); i++) {
 1568    if (!owned_mask.at(i))
 1569      continue;
 570
 1571    task->release_output(static_cast<int>(i));
 1572  }
 1573}
 574
 1575core::OpResult Scheduler::run_sync(GraphPlan::vertex_descriptor v) {
 576  using clock     = std::chrono::high_resolution_clock;
 1577  const auto  idx = boost::get(boost::vertex_index, graph_, v);
 1578  const auto &np  = graph_[v];
 1579  auto       &nrt = node_rts_.at(idx);
 1580  HOLOFLOW_CHECK(std::holds_alternative<SyncRt>(nrt));
 1581  auto &srt = std::get<SyncRt>(nrt);
 582
 1583  logger()->trace("[Scheduler::run_sync] Executing node '{}'", np.spec.name);
 1584  const auto range_name = task_range_name("sync execute", np.spec);
 1585  auto       t0         = clock::now();
 1586  auto       r          = core::OpResult::Cancelled;
 587  {
 1588    nvtx3::scoped_range task_range{range_name.c_str()};
 1589    r = srt.task->execute(srt.ctx);
 1590  }
 1591  auto t1 = clock::now();
 592
 1593  switch (r) {
 594  case core::OpResult::Cancelled:
 1595    logger()->debug("[Scheduler::run_sync] Node '{}' execution cancelled", np.spec.name);
 1596    stop_.store(true);
 1597    break;
 598  case core::OpResult::Eof:
 1599    logger()->debug("[Scheduler::run_sync] Node '{}' reached end of stream", np.spec.name);
 1600    stop_.store(true);
 1601    break;
 602  case core::OpResult::NotReady:
 1603    logger()->error(
 604        "[Scheduler::run_sync] The synchronous task '{}' returned NotReady, which is not allowed",
 605        np.spec.name);
 1606    stop_.store(true);
 607    break;
 608  case core::OpResult::Ok:
 609    // All good.
 610    break;
 611  }
 612
 1613  if (r == core::OpResult::Ok) {
 1614    const auto duration_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
 1615    auto [host_in, device_in] =
 1616        sum_bytes(std::span<const core::TView>(srt.in_views.data(), srt.in_views.size()));
 1617    auto [host_out, device_out] =
 1618        sum_bytes(std::span<const core::TView>(srt.out_views.data(), srt.out_views.size()));
 1619    record_node_sample(idx, static_cast<uint64_t>(duration_ns), host_in + host_out,
 620                       device_in + device_out);
 621  }
 1622  return r;
 1623}
 624
 1625core::OpResult Scheduler::run_async_cons(GraphPlan::vertex_descriptor v) {
 626  using clock     = std::chrono::high_resolution_clock;
 1627  const auto  idx = boost::get(boost::vertex_index, graph_, v);
 1628  const auto &np  = graph_[v];
 1629  auto       &nrt = node_rts_.at(idx);
 1630  HOLOFLOW_CHECK(std::holds_alternative<AsyncRt>(nrt));
 1631  auto &art = std::get<AsyncRt>(nrt);
 632
 1633  logger()->trace("[Scheduler::run_async_cons] Executing node '{}'", np.spec.name);
 1634  const auto range_name = task_range_name("async pop", np.spec);
 1635  auto       t0         = clock::now();
 1636  auto       r          = core::OpResult::Cancelled;
 637  {
 1638    nvtx3::scoped_range task_range{range_name.c_str()};
 1639    r = art.task->try_pop(art.xctx);
 1640    while (r == core::OpResult::NotReady) {
 1641      if (stop_.load())
 0642        return core::OpResult::Cancelled;
 1643      r = art.task->try_pop(art.xctx);
 1644    }
 1645  }
 1646  auto t1 = clock::now();
 647
 1648  switch (r) {
 649  case core::OpResult::Cancelled:
 0650    logger()->debug("[Scheduler::run_async_cons] Node '{}' execution cancelled", np.spec.name);
 0651    stop_.store(true);
 0652    break;
 653  case core::OpResult::Eof:
 0654    logger()->debug("[Scheduler::run_async_cons] Node '{}' reached end of stream", np.spec.name);
 0655    stop_.store(true);
 0656    break;
 657  case core::OpResult::NotReady:
 0658    HOLOFLOW_UNREACHABLE();
 659    break;
 660  case core::OpResult::Ok:
 661    // All good.
 662    break;
 663  }
 664
 1665  if (r == core::OpResult::Ok) {
 1666    const auto duration_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
 1667    auto [host_out, device_out] =
 1668        sum_bytes(std::span<const core::TView>(art.out_views.data(), art.out_views.size()));
 1669    record_node_sample(idx, static_cast<uint64_t>(duration_ns), host_out, device_out);
 670  }
 1671  return r;
 1672}
 673
 1674core::OpResult Scheduler::run_async_prod(GraphPlan::vertex_descriptor v) {
 675  using clock     = std::chrono::high_resolution_clock;
 1676  const auto  idx = boost::get(boost::vertex_index, graph_, v);
 1677  const auto &np  = graph_[v];
 1678  auto       &nrt = node_rts_.at(idx);
 1679  HOLOFLOW_CHECK(std::holds_alternative<AsyncRt>(nrt));
 1680  auto &art = std::get<AsyncRt>(nrt);
 681
 1682  logger()->trace("[Scheduler::run_async_prod] Executing node '{}'", np.spec.name);
 1683  const auto range_name = task_range_name("async push", np.spec);
 1684  auto       t0         = clock::now();
 1685  auto       r          = core::OpResult::Cancelled;
 686  {
 1687    nvtx3::scoped_range task_range{range_name.c_str()};
 1688    r = art.task->try_push(art.pctx);
 1689    while (r == core::OpResult::NotReady) {
 1690      if (stop_.load())
 1691        return core::OpResult::Cancelled;
 1692      r = art.task->try_push(art.pctx);
 1693    }
 1694  }
 1695  auto t1 = clock::now();
 696
 1697  switch (r) {
 698  case core::OpResult::Cancelled:
 0699    logger()->debug("[Scheduler::run_async_prod] Node '{}' execution cancelled", np.spec.name);
 0700    stop_.store(true);
 0701    break;
 702  case core::OpResult::Eof:
 0703    logger()->debug("[Scheduler::run_async_prod] Node '{}' reached end of stream", np.spec.name);
 0704    stop_.store(true);
 0705    break;
 706  case core::OpResult::NotReady:
 0707    HOLOFLOW_UNREACHABLE();
 708    break;
 709  case core::OpResult::Ok:
 710    // All good.
 711    break;
 712  }
 713
 1714  if (r == core::OpResult::Ok) {
 1715    const auto duration_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
 1716    auto [host_in, device_in] =
 1717        sum_bytes(std::span<const core::TView>(art.in_views.data(), art.in_views.size()));
 1718    record_node_sample(idx, static_cast<uint64_t>(duration_ns), host_in, device_in);
 719  }
 1720  return r;
 1721}
 722
 1723void Scheduler::reset_metrics_state() {
 1724  auto num_vertices = boost::num_vertices(graph_);
 1725  metric_accumulators_.clear();
 1726  metric_accumulators_.resize(num_vertices);
 727
 1728  std::lock_guard<std::mutex> lock(metrics_mutex_);
 1729  latest_metrics_.clear();
 1730  for (const auto &name : node_names_) {
 1731    latest_metrics_.emplace(name, NodeMetrics{});
 1732  }
 1733}
 734
 1735void Scheduler::start_metrics_thread() {
 1736  if (metrics_interval_.count() <= 0) {
 0737    return;
 738  }
 1739  bool expected = false;
 1740  if (!metrics_running_.compare_exchange_strong(expected, true)) {
 0741    return;
 742  }
 1743  metrics_thread_ = std::thread(&Scheduler::metrics_loop, this);
 1744}
 745
 1746void Scheduler::stop_metrics_thread() {
 1747  bool expected = true;
 1748  if (!metrics_running_.compare_exchange_strong(expected, false)) {
 1749    return;
 750  }
 1751  metrics_cv_.notify_all();
 1752  if (metrics_thread_.joinable()) {
 1753    metrics_thread_.join();
 754  }
 1755  metrics_thread_ = std::thread();
 1756}
 757
 1758void Scheduler::metrics_loop() {
 1759  auto                         last = std::chrono::steady_clock::now();
 1760  std::unique_lock<std::mutex> lock(metrics_thread_mutex_);
 1761  while (metrics_running_.load()) {
 1762    auto interval = metrics_interval_;
 1763    if (metrics_cv_.wait_for(lock, interval, [this] { return !metrics_running_.load(); })) {
 1764      break;
 765    }
 1766    auto now     = std::chrono::steady_clock::now();
 1767    auto elapsed = std::chrono::duration<double>(now - last).count();
 1768    last         = now;
 1769    aggregate_metrics(elapsed);
 1770  }
 1771  auto now     = std::chrono::steady_clock::now();
 1772  auto elapsed = std::chrono::duration<double>(now - last).count();
 1773  aggregate_metrics(elapsed);
 1774}
 775
 1776void Scheduler::aggregate_metrics(double interval_seconds) {
 1777  if (interval_seconds <= 0.0) {
 0778    interval_seconds = static_cast<double>(metrics_interval_.count()) / 1000.0;
 779  }
 780
 1781  std::map<std::string, NodeMetrics> snapshot;
 782
 1783  for (std::size_t idx = 0; idx < metric_accumulators_.size(); ++idx) {
 1784    auto      &acc          = metric_accumulators_.at(idx);
 1785    const auto duration_ns  = acc.duration_ns.exchange(0, std::memory_order_relaxed);
 1786    const auto runs         = acc.run_count.exchange(0, std::memory_order_relaxed);
 1787    const auto host_bytes   = acc.host_bytes.exchange(0, std::memory_order_relaxed);
 1788    const auto device_bytes = acc.device_bytes.exchange(0, std::memory_order_relaxed);
 789
 1790    NodeMetrics metrics;
 1791    metrics.sample_count = runs;
 1792    if (runs > 0) {
 1793      metrics.average_duration_ms =
 794          static_cast<double>(duration_ns) / static_cast<double>(runs) / 1'000'000.0;
 795    }
 1796    if (interval_seconds > 0.0) {
 1797      metrics.runs_per_second                  = static_cast<double>(runs) / interval_seconds;
 1798      metrics.host_throughput_bytes_per_second = static_cast<double>(host_bytes) / interval_seconds;
 1799      metrics.device_throughput_bytes_per_second =
 800          static_cast<double>(device_bytes) / interval_seconds;
 801    }
 1802    snapshot.emplace(node_names_.at(idx), metrics);
 1803  }
 804
 1805  std::lock_guard<std::mutex> lock(metrics_mutex_);
 1806  latest_metrics_ = std::move(snapshot);
 1807}
 808
 809void Scheduler::record_node_sample(std::size_t idx, uint64_t duration_ns, uint64_t host_bytes,
 1810                                   uint64_t device_bytes) {
 1811  auto &acc = metric_accumulators_.at(idx);
 1812  acc.duration_ns.fetch_add(duration_ns, std::memory_order_relaxed);
 1813  acc.run_count.fetch_add(1, std::memory_order_relaxed);
 1814  acc.host_bytes.fetch_add(host_bytes, std::memory_order_relaxed);
 1815  acc.device_bytes.fetch_add(device_bytes, std::memory_order_relaxed);
 1816}
 817
 1818std::pair<uint64_t, uint64_t> Scheduler::sum_bytes(std::span<const core::TView> views) {
 1819  uint64_t host_total   = 0;
 1820  uint64_t device_total = 0;
 1821  for (const auto &view : views) {
 1822    const auto bytes = static_cast<uint64_t>(view.desc.num_bytes());
 1823    if (view.desc.mem_loc == core::MemLoc::Device) {
 1824      device_total += bytes;
 1825    } else {
 1826      host_total += bytes;
 827    }
 1828  }
 1829  return {host_total, device_total};
 1830}
 831
 832} // namespace holoflow::runtime
Generated by: ReportGenerator 5.5.10.0
9/16/2026 - 1:44:14 PM
GitHub | reportgenerator.io

Methods/Properties