< Summary

Line coverage
43%
Covered lines: 958
Uncovered lines: 1247
Coverable lines: 2205
Total lines: 5020
Line coverage: 43.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\holotask\src\asyncs\batch_queue.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 "holotask/asyncs/batch_queue.hh"
 16
 17#include <atomic>
 18#include <cstddef>
 19#include <memory>
 20#include <numeric>
 21#include <optional>
 22
 23#include "bug.hh"
 24#include "curaii/cuda.hh"
 25#include "holoflow/core/tasks.hh"
 26#include "holoflow/core/tensor.hh"
 27#include "logger.hh"
 28
 29#ifndef CACHE_LINE_SIZE
 30#define CACHE_LINE_SIZE 64
 31#endif
 32
 33namespace holotask::asyncs {
 34
 35template <typename T> using DevPtr  = curaii::unique_device_ptr<T>;
 36template <typename T> using HostPtr = curaii::unique_host_ptr<T>;
 37
 038void to_json(nlohmann::json &j, const BatchQueueSettings &bqs) {
 039  j = nlohmann::json{
 40      {"target_capacity", bqs.target_capacity},
 41      {"output_size", bqs.output_size},
 42      {"output_stride", bqs.output_stride},
 43  };
 044}
 45
 046void from_json(const nlohmann::json &j, BatchQueueSettings &bqs) {
 047  j.at("target_capacity").get_to(bqs.target_capacity);
 048  j.at("output_size").get_to(bqs.output_size);
 049  j.at("output_stride").get_to(bqs.output_stride);
 050}
 51
 52namespace {
 53
 54class BatchQueue : public holoflow::core::IAsyncTask {
 55public:
 56  BatchQueue(BatchQueueSettings settings, holoflow::core::TDesc idesc, holoflow::core::TDesc odesc,
 57             HostPtr<std::byte> &&h_buf, DevPtr<std::byte> &&d_buf, std::byte *buf, size_t nb_slots,
 58             size_t input_size, size_t element_size)
 059      : settings_(std::move(settings)), idesc_(std::move(idesc)), odesc_(std::move(odesc)),
 060        h_buf_(std::move(h_buf)), d_buf_(std::move(d_buf)), buf_(buf), nb_slots_(nb_slots),
 061        input_size_(input_size), element_size_(element_size) {}
 62
 63  std::optional<holoflow::core::TView> acquire_input(int index) override;
 64  void                                 release_output(int index) override;
 65  holoflow::core::OpResult             try_push(holoflow::core::AsyncPushCtx &ctx) override;
 66  holoflow::core::OpResult             try_pop(holoflow::core::AsyncPopCtx &ctx) override;
 67
 068  const holoflow::core::TDesc &idesc() const { return idesc_; }
 069  size_t                       nb_slots() const { return nb_slots_; }
 070  size_t                       element_size() const { return element_size_; }
 071  HostPtr<std::byte>           take_host_buffer() { return std::move(h_buf_); }
 072  DevPtr<std::byte>            take_device_buffer() { return std::move(d_buf_); }
 073  std::byte                   *buffer() const { return buf_; }
 74
 75private:
 76  size_t writer_size() const;
 77  size_t reader_size() const;
 78
 79  BatchQueueSettings    settings_;
 80  holoflow::core::TDesc idesc_;
 81  holoflow::core::TDesc odesc_;
 82  HostPtr<std::byte>    h_buf_;
 83  DevPtr<std::byte>     d_buf_;
 84  std::byte            *buf_;
 85  size_t                nb_slots_;
 86  size_t                input_size_;
 87  size_t                element_size_;
 88  alignas(CACHE_LINE_SIZE) std::atomic<size_t> write_idx_;
 89  alignas(CACHE_LINE_SIZE) std::atomic<size_t> read_idx_;
 90};
 91
 092std::optional<holoflow::core::TView> BatchQueue::acquire_input(int index) {
 093  if (index != 0) {
 094    throw std::out_of_range("BatchQueue::acquire_input: invalid index");
 95  }
 96
 097  if (nb_slots_ - writer_size() <= input_size_) {
 098    return std::nullopt;
 99  }
 100
 0101  size_t     write_idx = write_idx_.load(std::memory_order_relaxed);
 0102  std::byte *data      = buf_ + write_idx * element_size_;
 0103  auto      &storage   = storage_access().owned_input_storage(0);
 0104  storage.ptr          = data;
 105
 0106  return holoflow::core::TView{
 107      .desc    = idesc_,
 108      .storage = &storage,
 109  };
 0110}
 111
 0112void BatchQueue::release_output(int index) {
 0113  if (index != 0) {
 0114    throw std::out_of_range("BatchQueue::release_output: invalid index");
 115  }
 116
 0117  size_t read_idx      = read_idx_.load(std::memory_order_relaxed);
 0118  size_t next_read_idx = read_idx + settings_.output_stride;
 0119  if (next_read_idx == nb_slots_) {
 0120    next_read_idx = 0;
 121  }
 0122  auto &storage = storage_access().owned_output_storage(0);
 0123  storage.ptr   = nullptr;
 0124  read_idx_.store(next_read_idx, std::memory_order_release);
 0125}
 126
 0127holoflow::core::OpResult BatchQueue::try_push(holoflow::core::AsyncPushCtx &) {
 0128  size_t write_idx      = write_idx_.load(std::memory_order_relaxed);
 0129  size_t next_write_idx = write_idx + input_size_;
 0130  if (next_write_idx >= nb_slots_) {
 0131    next_write_idx = 0;
 132  }
 0133  auto &storage = storage_access().owned_input_storage(0);
 0134  storage.ptr   = nullptr;
 0135  write_idx_.store(next_write_idx, std::memory_order_release);
 0136  return holoflow::core::OpResult::Ok;
 0137}
 138
 0139holoflow::core::OpResult BatchQueue::try_pop(holoflow::core::AsyncPopCtx &ctx) {
 0140  if (reader_size() < settings_.output_stride) {
 0141    return holoflow::core::OpResult::NotReady;
 142  }
 143
 0144  size_t     read_idx = read_idx_.load(std::memory_order_relaxed);
 0145  std::byte *data     = buf_ + read_idx * element_size_;
 0146  auto      &storage  = storage_access().owned_output_storage(0);
 0147  storage.ptr         = data;
 148
 0149  ctx.outputs[0] = holoflow::core::TView{
 150      .desc    = odesc_,
 151      .storage = &storage,
 152  };
 0153  return holoflow::core::OpResult::Ok;
 0154}
 155
 0156size_t BatchQueue::writer_size() const {
 0157  size_t write_idx = write_idx_.load(std::memory_order_relaxed);
 0158  size_t read_idx  = read_idx_.load(std::memory_order_acquire);
 0159  size_t diff      = write_idx - read_idx;
 0160  if (write_idx < read_idx) {
 0161    diff += nb_slots_;
 162  }
 0163  return diff;
 0164}
 165
 0166size_t BatchQueue::reader_size() const {
 0167  size_t write_idx = write_idx_.load(std::memory_order_acquire);
 0168  size_t read_idx  = read_idx_.load(std::memory_order_relaxed);
 0169  size_t diff      = write_idx - read_idx;
 0170  if (write_idx < read_idx) {
 0171    diff += nb_slots_;
 172  }
 0173  return diff;
 0174}
 175
 0176int lcm_above(int x, int y, int k) {
 0177  auto base = std::lcm(x, y);
 0178  HOLOVIBES_CHECK(base > 0, "lcm_above: lcm overflow");
 0179  auto mult = (k + base - 1) / base;
 0180  return base * mult;
 0181}
 182
 0183bool is_contiguous(const holoflow::core::TDesc &desc) {
 0184  holoflow::core::TDesc contiguous(desc.shape, desc.dtype, desc.mem_loc, desc.offset);
 0185  return desc.strides == contiguous.strides;
 0186}
 187
 188} // namespace
 189
 190holoflow::core::InferResult
 191BatchQueueFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 0192                         const nlohmann::json                  &jsettings) const {
 193  const auto check = [&](bool condition, const std::string &msg) {
 194    if (!condition) {
 195      logger()->error("[BatchQueueFactory::infer] error: {}", msg);
 196      throw std::invalid_argument("BatchQueueFactory inference error: " + msg);
 197    }
 198  };
 199
 0200  auto settings = jsettings.get<BatchQueueSettings>();
 201
 202  // Validate
 0203  check(input_descs.size() == 1, "BatchQueue task must have exactly one input");
 0204  check(input_descs[0].rank() > 0, "BatchQueue task input must have rank > 0");
 0205  check(is_contiguous(input_descs[0]), "BatchQueue task input must be contiguous");
 0206  check(settings.target_capacity > 0, "BatchQueue task target capacity must be > 0");
 0207  check(settings.output_size > 0, "BatchQueue task output size must be > 0");
 0208  check(settings.output_stride > 0, "BatchQueue task output stride must be > 0");
 0209  auto is_factor = settings.output_stride % settings.output_size == 0;
 0210  check(is_factor, "BatchQueue task output stride must be a multiple of output size");
 211
 212  // Success
 0213  auto odesc     = input_descs[0];
 0214  odesc.shape[0] = settings.output_size;
 0215  return holoflow::core::InferResult{
 216      .input_descs   = {input_descs[0]},
 217      .output_descs  = {odesc},
 218      .in_place      = {},
 219      .owned_inputs  = {true},
 220      .owned_outputs = {true},
 221      .kind          = holoflow::core::TaskKind::Async,
 222  };
 0223}
 224
 225std::unique_ptr<holoflow::core::IAsyncTask>
 226BatchQueueFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 227                          const nlohmann::json                  &jsettings,
 0228                          const holoflow::core::AsyncCreateCtx &) const {
 0229  auto infer    = this->infer(input_descs, jsettings);
 0230  auto settings = jsettings.get<BatchQueueSettings>();
 231
 232  // Compute n_slots such that:
 233  // - nb_slots >= target_capacity
 234  // - nb_slots % stride == 0
 235  // - nb_slots % input_descs[0].shape[0] == 0
 0236  int x        = static_cast<int>(input_descs[0].shape[0]);
 0237  int y        = settings.output_stride;
 0238  int k        = settings.target_capacity + x;
 0239  int nb_slots = lcm_above(x, y, k);
 240
 241  // Setup buffers
 0242  size_t input_size   = static_cast<int>(input_descs[0].shape[0]);
 0243  size_t element_size = static_cast<int>(input_descs[0].num_bytes() / input_size);
 0244  size_t bytes        = nb_slots * element_size;
 245
 0246  logger()->debug(
 247      "[BatchQueueFactory::create] Creating BatchQueue with {} slots, capacity={}, input_size={}, "
 248      "output_size={}, output_stride={}, "
 249      "element_size={}, "
 250      "total_bytes={}",
 251      nb_slots, settings.target_capacity, input_size, settings.output_size, settings.output_stride,
 252      element_size, bytes);
 253
 0254  HostPtr<std::byte> h_buf = nullptr;
 0255  DevPtr<std::byte>  d_buf = nullptr;
 0256  std::byte         *buf   = nullptr;
 0257  switch (input_descs[0].mem_loc) {
 258  case holoflow::core::MemLoc::Host:
 0259    h_buf = curaii::make_unique_host_ptr<std::byte>(bytes);
 0260    buf   = h_buf.get();
 0261    break;
 262  case holoflow::core::MemLoc::Device:
 0263    d_buf = curaii::make_unique_device_ptr<std::byte>(bytes);
 0264    buf   = d_buf.get();
 265    break;
 266  }
 267
 0268  return std::make_unique<BatchQueue>(settings, input_descs[0], infer.output_descs[0],
 269                                      std::move(h_buf), std::move(d_buf), buf, nb_slots, input_size,
 270                                      element_size);
 0271}
 272
 273std::unique_ptr<holoflow::core::IAsyncTask>
 274BatchQueueFactory::update(std::unique_ptr<holoflow::core::IAsyncTask> old_task,
 275                          std::span<const holoflow::core::TDesc>      input_descs,
 276                          const nlohmann::json                       &jsettings,
 0277                          const holoflow::core::AsyncCreateCtx       &ctx) const {
 0278  auto infer    = this->infer(input_descs, jsettings);
 0279  auto settings = jsettings.get<BatchQueueSettings>();
 0280  auto old_bq   = dynamic_cast<BatchQueue *>(old_task.get());
 0281  if (old_bq == nullptr) {
 0282    return this->create(input_descs, jsettings, ctx);
 283  }
 284
 285  // Update
 0286  int    x            = static_cast<int>(input_descs[0].shape[0]);
 0287  int    y            = settings.output_stride;
 0288  int    k            = settings.target_capacity + x;
 0289  int    nb_slots     = lcm_above(x, y, k);
 0290  size_t input_size   = static_cast<int>(input_descs[0].shape[0]);
 0291  size_t element_size = static_cast<int>(input_descs[0].num_bytes() / input_size);
 0292  size_t bytes        = nb_slots * element_size;
 0293  bool   same_buffer  = (bytes == old_bq->nb_slots() * old_bq->element_size()) &&
 294                     (input_descs[0].mem_loc == old_bq->idesc().mem_loc);
 295
 0296  if (same_buffer) {
 0297    logger()->debug("[BatchQueueFactory::update] Reusing existing BatchQueue task");
 0298    return std::make_unique<BatchQueue>(settings, input_descs[0], infer.output_descs[0],
 299                                        old_bq->take_host_buffer(), old_bq->take_device_buffer(),
 300                                        old_bq->buffer(), nb_slots, input_size, element_size);
 301  }
 302
 303  // Fallback to recreate
 0304  logger()->debug("[BatchQueueFactory::update] Recreating BatchQueue task");
 0305  return this->create(input_descs, jsettings, ctx);
 0306}
 307
 308} // namespace holotask::asyncs

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\asyncs\dual_reader_batch_queue.cc

#LineLine coverage
 1// Copyright 2026 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 "holotask/asyncs/dual_reader_batch_queue.hh"
 16
 17#include <algorithm>
 18#include <atomic>
 19#include <cstddef>
 20#include <cstdint>
 21#include <memory>
 22#include <numeric>
 23#include <optional>
 24#include <stdexcept>
 25#include <string>
 26#include <utility>
 27
 28#include "curaii/cuda.hh"
 29#include "logger.hh"
 30
 31#ifndef CACHE_LINE_SIZE
 32#define CACHE_LINE_SIZE 64
 33#endif
 34
 35namespace holotask::asyncs {
 36
 37template <typename T> using DevPtr  = curaii::unique_device_ptr<T>;
 38template <typename T> using HostPtr = curaii::unique_host_ptr<T>;
 39
 140void to_json(nlohmann::json &j, const DualReaderBatchQueueSettings &s) {
 141  j = nlohmann::json{{"target_capacity", s.target_capacity}, {"window_size", s.window_size}};
 142}
 43
 144void from_json(const nlohmann::json &j, DualReaderBatchQueueSettings &s) {
 145  j.at("target_capacity").get_to(s.target_capacity);
 146  j.at("window_size").get_to(s.window_size);
 147}
 48
 49namespace {
 50
 151bool is_contiguous(const holoflow::core::TDesc &desc) {
 152  holoflow::core::TDesc contiguous(desc.shape, desc.dtype, desc.mem_loc, desc.offset);
 153  return desc.strides == contiguous.strides;
 154}
 55
 156size_t lcm_above(size_t x, size_t y, size_t minimum) {
 157  const size_t base = std::lcm(x, y);
 158  if (base == 0) {
 059    throw std::overflow_error("DualReaderBatchQueue: capacity alignment overflow");
 60  }
 161  return ((minimum + base - 1) / base) * base;
 162}
 63
 64class DualReaderBatchQueue final : public holoflow::core::IAsyncTask {
 65public:
 66  DualReaderBatchQueue(DualReaderBatchQueueSettings settings, holoflow::core::TDesc input_desc,
 67                       holoflow::core::TDesc output_desc, HostPtr<std::byte> host_buffer,
 68                       DevPtr<std::byte> device_buffer, HostPtr<std::byte> host_scratch,
 69                       DevPtr<std::byte> device_scratch, std::byte *buffer, std::byte *scratch,
 70                       size_t slot_count, size_t input_size, size_t element_size)
 171      : settings_(settings), input_desc_(std::move(input_desc)),
 172        output_desc_(std::move(output_desc)), host_buffer_(std::move(host_buffer)),
 173        device_buffer_(std::move(device_buffer)), host_scratch_(std::move(host_scratch)),
 174        device_scratch_(std::move(device_scratch)), buffer_(buffer), scratch_(scratch),
 175        slot_count_(slot_count), input_size_(input_size), element_size_(element_size),
 176        delay_((settings_.window_size - 1) / 2) {}
 77
 178  std::optional<holoflow::core::TView> acquire_input(int index) override {
 179    if (index != 0) {
 080      throw std::out_of_range("DualReaderBatchQueue::acquire_input: invalid index");
 81    }
 182    if (slot_count_ - writer_size() <= input_size_) {
 083      return std::nullopt;
 84    }
 85
 186    const size_t write_idx = write_idx_.load(std::memory_order_relaxed);
 187    auto        &storage   = storage_access().owned_input_storage(0);
 188    storage.ptr            = buffer_ + write_idx * element_size_;
 189    return holoflow::core::TView{.desc = input_desc_, .storage = &storage};
 190  }
 91
 192  void release_output(int index) override {
 193    if (index != 0 && index != 1) {
 094      throw std::out_of_range("DualReaderBatchQueue::release_output: invalid index");
 95    }
 196    if (!pop_active_) {
 097      throw std::logic_error("DualReaderBatchQueue::release_output: no active output");
 98    }
 99
 1100    auto &storage = storage_access().owned_output_storage(static_cast<size_t>(index));
 1101    storage.ptr   = nullptr;
 1102    if (index == 0) {
 1103      current_released_ = true;
 1104    } else {
 1105      delayed_released_ = true;
 106    }
 107
 1108    if (!current_released_ || !delayed_released_) {
 1109      return;
 110    }
 111
 1112    current_read_idx_ = increment(current_read_idx_);
 1113    if (delayed_active_) {
 1114      const size_t delayed_read_idx = delayed_read_idx_.load(std::memory_order_relaxed);
 1115      delayed_read_idx_.store(increment(delayed_read_idx), std::memory_order_release);
 116    }
 1117    ++sequence_;
 1118    pop_active_ = false;
 1119  }
 120
 1121  holoflow::core::OpResult try_push(holoflow::core::AsyncPushCtx &) override {
 1122    size_t next_write_idx = write_idx_.load(std::memory_order_relaxed) + input_size_;
 1123    if (next_write_idx >= slot_count_) {
 1124      next_write_idx = 0;
 125    }
 1126    storage_access().owned_input_storage(0).ptr = nullptr;
 1127    write_idx_.store(next_write_idx, std::memory_order_release);
 1128    return holoflow::core::OpResult::Ok;
 1129  }
 130
 1131  holoflow::core::OpResult try_pop(holoflow::core::AsyncPopCtx &ctx) override {
 1132    if (reader_size(current_read_idx_) < 1) {
 0133      return holoflow::core::OpResult::NotReady;
 134    }
 1135    if (pop_active_) {
 0136      throw std::logic_error("DualReaderBatchQueue::try_pop: prior output was not released");
 137    }
 138
 1139    auto &current_storage         = storage_access().owned_output_storage(0);
 1140    auto &delayed_storage         = storage_access().owned_output_storage(1);
 1141    current_storage.ptr           = buffer_ + current_read_idx_ * element_size_;
 1142    delayed_active_               = sequence_ >= delay_;
 1143    const size_t delayed_read_idx = delayed_read_idx_.load(std::memory_order_relaxed);
 1144    delayed_storage.ptr = delayed_active_ ? buffer_ + delayed_read_idx * element_size_ : scratch_;
 145
 1146    ctx.outputs[0] = {.desc = output_desc_, .storage = &current_storage};
 1147    ctx.outputs[1] = {.desc = output_desc_, .storage = &delayed_storage};
 1148    *reinterpret_cast<std::uint8_t *>(ctx.outputs[2].data()) =
 149        sequence_ >= settings_.window_size - 1 ? std::uint8_t{1} : std::uint8_t{0};
 150
 1151    current_released_ = false;
 1152    delayed_released_ = false;
 1153    pop_active_       = true;
 1154    return holoflow::core::OpResult::Ok;
 1155  }
 156
 157private:
 1158  size_t increment(size_t index) const { return index + 1 == slot_count_ ? 0 : index + 1; }
 159
 1160  size_t distance(size_t begin, size_t end) const {
 1161    return end >= begin ? end - begin : end + slot_count_ - begin;
 1162  }
 163
 1164  size_t writer_size() const {
 1165    const size_t write_idx        = write_idx_.load(std::memory_order_relaxed);
 1166    const size_t delayed_read_idx = delayed_read_idx_.load(std::memory_order_acquire);
 1167    return distance(delayed_read_idx, write_idx);
 1168  }
 169
 1170  size_t reader_size(size_t read_idx) const {
 1171    const size_t write_idx = write_idx_.load(std::memory_order_acquire);
 1172    return distance(read_idx, write_idx);
 1173  }
 174
 175  DualReaderBatchQueueSettings settings_;
 176  holoflow::core::TDesc        input_desc_;
 177  holoflow::core::TDesc        output_desc_;
 178  HostPtr<std::byte>           host_buffer_;
 179  DevPtr<std::byte>            device_buffer_;
 180  HostPtr<std::byte>           host_scratch_;
 181  DevPtr<std::byte>            device_scratch_;
 182  std::byte                   *buffer_;
 183  std::byte                   *scratch_;
 184  size_t                       slot_count_;
 185  size_t                       input_size_;
 186  size_t                       element_size_;
 187  size_t                       delay_;
 1188  alignas(CACHE_LINE_SIZE) std::atomic<size_t> write_idx_{0};
 1189  size_t current_read_idx_ = 0;
 1190  alignas(CACHE_LINE_SIZE) std::atomic<size_t> delayed_read_idx_{0};
 1191  size_t sequence_         = 0;
 1192  bool   pop_active_       = false;
 1193  bool   delayed_active_   = false;
 1194  bool   current_released_ = false;
 1195  bool   delayed_released_ = false;
 196};
 197
 198} // namespace
 199
 200holoflow::core::InferResult
 201DualReaderBatchQueueFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 1202                                   const nlohmann::json                  &jsettings) const {
 203  const auto check = [&](bool condition, const std::string &message) {
 204    if (!condition) {
 205      logger()->error("[DualReaderBatchQueueFactory::infer] error: {}", message);
 206      throw std::invalid_argument("DualReaderBatchQueueFactory inference error: " + message);
 207    }
 208  };
 209
 1210  const auto settings = jsettings.get<DualReaderBatchQueueSettings>();
 1211  check(input_descs.size() == 1, "task must have exactly one input");
 1212  const auto &input = input_descs[0];
 1213  check(input.rank() > 0, "input rank must be positive");
 1214  check(input.shape[0] > 0, "input leading dimension must be positive");
 1215  check(is_contiguous(input), "input must be contiguous");
 1216  check(settings.target_capacity > 0, "target_capacity must be positive");
 1217  check(settings.window_size > 0, "window_size must be positive");
 218
 1219  auto output     = input;
 1220  output.shape[0] = 1;
 1221  const holoflow::core::TDesc valid({1}, holoflow::core::DType::U8, holoflow::core::MemLoc::Host);
 1222  return {
 223      .input_descs   = {input},
 224      .output_descs  = {output, output, valid},
 225      .in_place      = {},
 226      .owned_inputs  = {true},
 227      .owned_outputs = {true, true, false},
 228      .kind          = holoflow::core::TaskKind::Async,
 229  };
 1230}
 231
 232std::unique_ptr<holoflow::core::IAsyncTask>
 233DualReaderBatchQueueFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 234                                    const nlohmann::json                  &jsettings,
 1235                                    const holoflow::core::AsyncCreateCtx  &ctx) const {
 1236  const auto   infer        = this->infer(input_descs, jsettings);
 1237  const auto   settings     = jsettings.get<DualReaderBatchQueueSettings>();
 1238  const auto  &input        = input_descs[0];
 1239  const size_t input_size   = input.shape[0];
 1240  const size_t element_size = input.num_bytes() / input_size;
 1241  const size_t delay        = (settings.window_size - 1) / 2;
 1242  const size_t slot_count =
 243      lcm_above(input_size, size_t{1}, settings.target_capacity + input_size + delay + size_t{1});
 1244  const size_t bytes = slot_count * element_size;
 245
 1246  HostPtr<std::byte> host_buffer;
 1247  DevPtr<std::byte>  device_buffer;
 1248  HostPtr<std::byte> host_scratch;
 1249  DevPtr<std::byte>  device_scratch;
 1250  std::byte         *buffer  = nullptr;
 1251  std::byte         *scratch = nullptr;
 252
 1253  if (input.mem_loc == holoflow::core::MemLoc::Host) {
 1254    host_buffer  = curaii::make_unique_host_ptr<std::byte>(bytes);
 1255    host_scratch = curaii::make_unique_host_ptr<std::byte>(element_size);
 1256    buffer       = host_buffer.get();
 1257    scratch      = host_scratch.get();
 1258    std::fill_n(scratch, element_size, std::byte{0});
 1259  } else {
 0260    device_buffer  = curaii::make_unique_device_ptr<std::byte>(bytes);
 0261    device_scratch = curaii::make_unique_device_ptr<std::byte>(element_size);
 0262    buffer         = device_buffer.get();
 0263    scratch        = device_scratch.get();
 0264    CUDA_CHECK(cudaMemsetAsync(scratch, 0, element_size, ctx.consumer_stream));
 0265    CUDA_CHECK(cudaStreamSynchronize(ctx.consumer_stream));
 266  }
 267
 1268  logger()->debug("[DualReaderBatchQueueFactory::create] slots={}, input_size={}, delay={}, "
 269                  "window_size={}, total_bytes={}",
 270                  slot_count, input_size, delay, settings.window_size, bytes);
 271
 1272  return std::make_unique<DualReaderBatchQueue>(
 273      settings, input, infer.output_descs[0], std::move(host_buffer), std::move(device_buffer),
 274      std::move(host_scratch), std::move(device_scratch), buffer, scratch, slot_count, input_size,
 275      element_size);
 1276}
 277
 278std::unique_ptr<holoflow::core::IAsyncTask> DualReaderBatchQueueFactory::update(
 279    std::unique_ptr<holoflow::core::IAsyncTask>, std::span<const holoflow::core::TDesc> input_descs,
 0280    const nlohmann::json &jsettings, const holoflow::core::AsyncCreateCtx &ctx) const {
 0281  return create(input_descs, jsettings, ctx);
 0282}
 283
 284} // namespace holotask::asyncs

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\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 holotask {
 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>("holotask", 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 holotask

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\sinks\holofile.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 "holotask/sinks/holofile.hh"
 16
 17// #include <Windows.h>
 18#include <cerrno>
 19#include <chrono>
 20#include <cstddef>
 21#include <cstdint>
 22#include <cstdio>
 23#include <cstring>
 24#include <memory>
 25#include <omp.h>
 26#include <span>
 27#include <system_error>
 28#include <unordered_set>
 29
 30#include "bug.hh"
 31#include "curaii/cuda.hh"
 32#include "holofile/holofile.hh"
 33#include "holoflow/core/tasks.hh"
 34#include "logger.hh"
 35
 36template <typename T> using HostPtr = curaii::unique_host_ptr<T>;
 37
 38namespace holotask::sinks {
 39
 40// -------------------------------------------------------------------------------------------------
 41// JSON serialization
 42// -------------------------------------------------------------------------------------------------
 43
 144void to_json(nlohmann::json &j, const HolofileSettings &hs) {
 145  j = nlohmann::json{
 46      {"path", hs.path},
 47      {"count", hs.count},
 48      {"pipeline_settings", hs.pipeline_settings},
 49      {"use_buffer", hs.use_buffer},
 50  };
 151}
 52
 153void from_json(const nlohmann::json &j, HolofileSettings &hs) {
 154  j.at("path").get_to(hs.path);
 155  j.at("count").get_to(hs.count);
 156  j.at("pipeline_settings").get_to(hs.pipeline_settings);
 157  if (j.contains("use_buffer"))
 158    j.at("use_buffer").get_to(hs.use_buffer);
 159}
 60
 61// -------------------------------------------------------------------------------------------------
 62// Private implementation
 63// -------------------------------------------------------------------------------------------------
 64
 65namespace {
 66
 067void mt_memcpy(void *dst, const void *src, const std::size_t n) {
 068  constexpr int NUM_THREADS = 2;
 069  auto         *dst_bytes   = static_cast<std::uint8_t *>(dst);
 070  const auto   *src_bytes   = static_cast<const std::uint8_t *>(src);
 71
 072  const std::size_t chunk_size = n / NUM_THREADS;
 073  const std::size_t remainder  = n % NUM_THREADS;
 74
 075#pragma omp parallel num_threads(NUM_THREADS)
 76  {
 077    const int         tid       = omp_get_thread_num();
 078    const std::size_t offset    = tid * chunk_size;
 079    std::size_t       this_size = chunk_size;
 80
 081    if (tid == NUM_THREADS - 1) {
 082      this_size += remainder;
 83    }
 84
 085    if (this_size > 0) {
 086      std::memcpy(dst_bytes + offset, src_bytes + offset, this_size);
 87    }
 088  }
 089}
 90
 91struct RecordingGeometry {
 92  uint8_t  bits_per_pixel;
 93  uint32_t frame_width;
 94  uint32_t frame_height;
 95};
 96
 097holofile::Header make_header(int count, const RecordingGeometry &g) {
 098  const auto frame_count = static_cast<uint32_t>(count);
 099  const auto data_size =
 100      static_cast<uint64_t>(frame_count) * g.frame_height * g.frame_width * g.bits_per_pixel / 8;
 0101  return holofile::Header{
 102      .magic_number       = holofile::Header::MAGIC_NUMBER_LE,
 103      .version            = holofile::Header::CURRENT_VERSION,
 104      .bits_per_pixel     = g.bits_per_pixel,
 105      .frame_width        = g.frame_width,
 106      .frame_height       = g.frame_height,
 107      .frame_count        = frame_count,
 108      .data_size_in_bytes = data_size,
 109      .endianness         = holofile::Header::LITTLE_ENDIAN,
 110  };
 0111}
 112
 1113uint8_t bits_per_pixel_for(holoflow::core::DType dtype) {
 1114  switch (dtype) {
 115  case holoflow::core::DType::U8:
 1116    return 8;
 117  case holoflow::core::DType::U16:
 0118    return 16;
 119  default:
 0120    HOLOVIBES_BUG("Unsupported Holofile dtype: {}", static_cast<int>(dtype));
 121  }
 1122}
 123
 1124RecordingGeometry recording_geometry_from_desc(const holoflow::core::TDesc &desc) {
 1125  return RecordingGeometry{
 126      .bits_per_pixel = bits_per_pixel_for(desc.dtype),
 127      .frame_width    = static_cast<uint32_t>(desc.shape[2]),
 128      .frame_height   = static_cast<uint32_t>(desc.shape[1]),
 129  };
 1130}
 131
 1132size_t recording_frame_byte_size(const RecordingGeometry &geometry) {
 1133  return static_cast<size_t>(geometry.frame_width) * geometry.frame_height *
 134         geometry.bits_per_pixel / 8;
 1135}
 136
 1137bool same_geometry(const RecordingGeometry &a, const RecordingGeometry &b) {
 1138  return a.bits_per_pixel == b.bits_per_pixel && a.frame_width == b.frame_width &&
 139         a.frame_height == b.frame_height;
 1140}
 141
 1142void check(bool condition, const std::string &msg) {
 1143  if (!condition) {
 0144    logger()->error("[HolofileFactory] {}", msg);
 0145    throw std::invalid_argument("HolofileFactory error: " + msg);
 146  }
 1147}
 148
 149// -------------------------------------------------------------------------------------------------
 150// Buffer-then-flush writer task
 151// -------------------------------------------------------------------------------------------------
 152//
 153// Two-phase recording:
 154//   1. Accumulate: each execute() call memcpys incoming frames into the preallocated ring buffer.
 155//   2. Flush:      once all frames are collected, write the entire buffer to disk in one shot.
 156//
 157// No background thread. The flush happens synchronously inside execute(), so the pipeline will
 158// stall for one cycle while the file is written. For typical recording sizes this is acceptable
 159// and avoids all async complexity.
 160
 161class HolofileWriter : public holoflow::core::ISyncTask {
 162public:
 163  HolofileWriter(const HolofileSettings &settings, const RecordingGeometry &geometry,
 164                 size_t frame_byte_size)
 1165      : settings_(settings), geometry_(geometry), frame_byte_size_(frame_byte_size),
 1166        ring_(curaii::make_unique_host_ptr<uint8_t>(static_cast<size_t>(settings.count) *
 1167                                                    frame_byte_size)) {
 168    // Touch every page so the OS actually backs the allocation with physical memory.
 169    // Without this, the first recording pays the page-fault cost.
 1170    const size_t total_bytes = static_cast<size_t>(settings.count) * frame_byte_size_;
 171
 172    // 2. The Un-Optimizable Demand-Zero Pre-faulter
 173    // We cast to volatile so the compiler is strictly forbidden from removing this loop.
 1174    volatile uint8_t *v_ptr = static_cast<volatile uint8_t *>(ring_.get());
 175
 176    // Step through memory one 4KB page at a time (Windows page size)
 1177    for (size_t i = 0; i < total_bytes; i += 4096) {
 178      // We MUST write a non-zero value to force the OS to physically allocate the page.
 179      // If we write 0, Windows memory deduplicators might map it to a shared zero-page.
 1180      v_ptr[i] = 1;
 1181    }
 1182  }
 183
 1184  ~HolofileWriter() override = default;
 185
 0186  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 0187    handle_events(ctx);
 188
 0189    if (!recording_)
 0190      return holoflow::core::OpResult::Ok;
 191
 192    // --- Phase 1: Accumulate frames into the buffer ---
 0193    if (frames_buffered_ < settings_.count) {
 0194      const auto  remaining  = settings_.count - frames_buffered_;
 0195      const auto  batch_size = static_cast<int>(ctx.inputs[0].desc.shape[0]);
 0196      const auto  to_copy    = std::min(remaining, batch_size);
 0197      const auto *idata      = reinterpret_cast<const uint8_t *>(ctx.inputs[0].data());
 198
 0199      auto start = std::chrono::steady_clock::now();
 0200      mt_memcpy(ring_.get() + static_cast<size_t>(frames_buffered_) * frame_byte_size_, idata,
 201                static_cast<size_t>(to_copy) * frame_byte_size_);
 0202      auto end         = std::chrono::steady_clock::now();
 0203      auto duration_us = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
 0204      logger()->trace("[HolofileWriter] Buffered {} frames ({} us)", to_copy, duration_us);
 0205      frames_buffered_ += to_copy;
 206    }
 207
 208    // --- Phase 2: Buffer full â€” flush everything to disk ---
 0209    if (frames_buffered_ >= settings_.count) {
 0210      logger()->info("[HolofileWriter] Frame buffer full ({} frames); flushing to disk...",
 211                     settings_.count);
 212      try {
 0213        flush_to_disk();
 0214        emit_finished_event(ctx);
 0215        logger()->info("[HolofileWriter] Recording complete: {} frames written to {}",
 216                       frames_buffered_, settings_.path);
 0217      } catch (const std::exception &e) {
 0218        logger()->error("[HolofileWriter] Failed to finalize recording at {}: {}", settings_.path,
 219                        e.what());
 0220        emit_failed_event(ctx, e.what());
 0221      }
 0222      reset();
 223    }
 224
 0225    return holoflow::core::OpResult::Ok;
 0226  }
 227
 228  bool can_reuse(const HolofileSettings &settings, const RecordingGeometry &geometry,
 1229                 size_t frame_byte_size) const {
 1230    return settings.use_buffer == settings_.use_buffer && settings.count == settings_.count &&
 231           same_geometry(geometry, geometry_) && frame_byte_size == frame_byte_size_;
 1232  }
 233
 1234  void update_settings(const HolofileSettings &settings) {
 1235    const auto active_path = settings_.path;
 236
 1237    settings_ = settings;
 1238    if (recording_) {
 0239      settings_.path = active_path;
 240    }
 1241  }
 242
 243private:
 0244  void flush_to_disk() {
 0245    const auto header = make_header(settings_.count, geometry_);
 0246    const auto footer = holofile::Footer{.pipeline_settings = settings_.pipeline_settings};
 247
 0248    holofile::Writer writer(settings_.path, header, footer);
 0249    writer.write_frames(ring_.get(), static_cast<size_t>(settings_.count));
 0250    writer.write_footer();
 0251  }
 252
 0253  void handle_events(holoflow::core::SyncCtx &ctx) {
 0254    if (!ctx.event_reader)
 0255      return;
 256
 0257    while (true) {
 0258      auto event = ctx.event_reader->try_pop();
 0259      if (!event.has_value())
 0260        break;
 261
 0262      HOLOVIBES_CHECK(event->direction == holoflow_event::EventDirection::ToNode,
 263                      "Unexpected event direction");
 264
 0265      const auto type = event->data.at("type").get<std::string>();
 266
 0267      if (type == "start_recording") {
 0268        if (recording_) {
 0269          logger()->error("[HolofileWriter] Ignoring duplicate start_recording event");
 0270          emit_failed_event(ctx, "Recording already in progress");
 0271          continue;
 272        }
 273
 0274        const auto record_path = event->data.value("record_path", std::string{});
 0275        if (record_path.empty()) {
 0276          logger()->error("[HolofileWriter] Rejecting start_recording event with empty path");
 0277          emit_failed_event(ctx, "Cannot start recording: empty path");
 0278          continue;
 279        }
 280
 0281        if (settings_.count <= 0) {
 0282          const auto message = "Cannot start recording: invalid frame count (" +
 283                               std::to_string(settings_.count) + ")";
 0284          logger()->error("[HolofileWriter] {}", message);
 0285          emit_failed_event(ctx, message);
 0286          continue;
 287        }
 288
 0289        settings_.path   = record_path;
 0290        frames_buffered_ = 0;
 0291        recording_       = true;
 292
 0293      } else if (type == "stop_recording") {
 0294        if (!recording_) {
 0295          logger()->warn("[HolofileWriter] Ignoring stop_recording event while idle");
 0296          continue;
 297        }
 298
 0299        recording_       = false;
 0300        frames_buffered_ = 0;
 301
 0302        if (!settings_.path.empty() && std::remove(settings_.path.c_str()) != 0 &&
 303            errno != ENOENT) {
 0304          std::error_code ec(errno, std::generic_category());
 0305          logger()->error("[HolofileWriter] Failed to remove incomplete recording at {}: {}",
 306                          settings_.path, ec.message());
 0307          emit_failed_event(ctx, "Failed to remove incomplete recording at " + settings_.path +
 308                                     ": " + ec.message());
 309        }
 310
 0311      } else {
 0312        HOLOVIBES_BUG("Unknown event type: {}", type);
 313      }
 0314    }
 0315  }
 316
 0317  void emit_finished_event(holoflow::core::SyncCtx &ctx) {
 0318    auto event = holoflow_event::Event{
 319        .direction = holoflow_event::EventDirection::ToUi,
 320        .node_id   = "",
 321        .data =
 322            nlohmann::json{
 323                {"type", "recording_finished"},
 324                {"path", settings_.path},
 325                {"frames_written", frames_buffered_},
 326            },
 327        .ts = std::chrono::steady_clock::now(),
 328    };
 0329    HOLOVIBES_CHECK(ctx.event_writer->try_push(std::move(event)),
 330                    "Failed to emit recording_finished event");
 0331  }
 332
 0333  void emit_failed_event(holoflow::core::SyncCtx &ctx, const std::string &message) {
 0334    auto event = holoflow_event::Event{
 335        .direction = holoflow_event::EventDirection::ToUi,
 336        .node_id   = "",
 337        .data =
 338            nlohmann::json{
 339                {"type", "recording_failed"},
 340                {"path", settings_.path},
 341                {"message", message},
 342            },
 343        .ts = std::chrono::steady_clock::now(),
 344    };
 0345    HOLOVIBES_CHECK(ctx.event_writer->try_push(std::move(event)),
 346                    "Failed to emit recording_failed event");
 0347  }
 348
 0349  void reset() {
 0350    recording_       = false;
 0351    frames_buffered_ = 0;
 0352  }
 353
 354  HolofileSettings  settings_;
 355  RecordingGeometry geometry_;
 356  size_t            frame_byte_size_;
 357  HostPtr<uint8_t>  ring_; ///< Preallocated at construction; reused across recordings.
 358
 1359  bool recording_       = false;
 1360  int  frames_buffered_ = 0;
 361};
 362
 363} // namespace
 364
 365// -------------------------------------------------------------------------------------------------
 366// Factory
 367// -------------------------------------------------------------------------------------------------
 368
 369holoflow::core::InferResult
 370HolofileFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 1371                       const nlohmann::json                  &jsettings) const {
 1372  auto settings = jsettings.get<HolofileSettings>();
 373
 1374  check(settings.use_buffer, "use_buffer must be true (synchronous writer is not supported)");
 1375  check(input_descs.size() == 1, "expected exactly one input tensor");
 376
 1377  auto &idesc = input_descs[0];
 1378  check(idesc.shape.size() == 3, "input tensor must have rank 3 (batch, height, width)");
 1379  check(idesc.mem_loc == holoflow::core::MemLoc::Host, "input tensor must be in Host memory");
 380
 1381  static const std::unordered_set<holoflow::core::DType> supported_dtypes = {
 382      holoflow::core::DType::U8,
 383      holoflow::core::DType::U16,
 384  };
 1385  check(supported_dtypes.contains(idesc.dtype),
 386        "unsupported input dtype: " + std::to_string(static_cast<int>(idesc.dtype)));
 387
 1388  const auto batch_size = static_cast<int>(idesc.shape[0]);
 1389  check(settings.count % batch_size == 0, "frame count (" + std::to_string(settings.count) +
 390                                              ") must be divisible by batch size (" +
 391                                              std::to_string(batch_size) + ")");
 392
 1393  return holoflow::core::InferResult{
 394      .input_descs   = {idesc},
 395      .output_descs  = {},
 396      .in_place      = {},
 397      .owned_inputs  = {false},
 398      .owned_outputs = {},
 399      .kind          = holoflow::core::TaskKind::Sync,
 400  };
 1401}
 402
 403std::unique_ptr<holoflow::core::ISyncTask>
 404HolofileFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 405                        const nlohmann::json                  &jsettings,
 1406                        const holoflow::core::SyncCreateCtx &) const {
 1407  auto  settings = jsettings.get<HolofileSettings>();
 1408  auto &idesc    = input_descs[0];
 409
 1410  const auto geometry        = recording_geometry_from_desc(idesc);
 1411  const auto frame_byte_size = recording_frame_byte_size(geometry);
 412
 1413  logger()->info("[HolofileFactory::create] Buffer-then-flush writer ({} frames)", settings.count);
 1414  return std::make_unique<HolofileWriter>(settings, geometry, frame_byte_size);
 1415}
 416
 417std::unique_ptr<holoflow::core::ISyncTask>
 418HolofileFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 419                        std::span<const holoflow::core::TDesc>     input_descs,
 420                        const nlohmann::json                      &jsettings,
 1421                        const holoflow::core::SyncCreateCtx       &ctx) const {
 1422  auto *old = dynamic_cast<HolofileWriter *>(old_task.get());
 1423  if (old == nullptr)
 0424    return create(input_descs, jsettings, ctx);
 425
 1426  infer(input_descs, jsettings);
 427
 1428  const auto settings        = jsettings.get<HolofileSettings>();
 1429  const auto geometry        = recording_geometry_from_desc(input_descs[0]);
 1430  const auto frame_byte_size = recording_frame_byte_size(geometry);
 431
 1432  if (!old->can_reuse(settings, geometry, frame_byte_size)) {
 1433    logger()->debug("[HolofileFactory::update] Recreating writer because buffer shape changed");
 1434    return create(input_descs, jsettings, ctx);
 435  }
 436
 1437  logger()->debug("[HolofileFactory::update] Reusing existing Holofile writer buffer");
 1438  old->update_settings(settings);
 1439  return old_task;
 1440}
 441
 442} // namespace holotask::sinks
#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 "holotask/sources/ametek_s710_euresys_coaxlink_octo.hh"
 16
 17#ifdef HOLOTASK_HAS_EGRABBER
 18
 19#include <EGrabber.h>
 20#include <EuresysGenapiErrorFormats.h>
 21#include <format>
 22#include <fstream>
 23#include <map>
 24#include <optional>
 25#include <sstream>
 26
 27#include "bug.hh"
 28#include "curaii/cuda.hh"
 29#include "logger.hh"
 30
 31template <typename T> using HostPtr = curaii::unique_host_ptr<T>;
 32
 33namespace holotask::sources {
 34
 35// -------------------------------------------------------------------------------------------------
 36// JSON serialization
 37// -------------------------------------------------------------------------------------------------
 38
 039void to_json(nlohmann::json &j, const AmetekS710EuresysCoaxlinkOctoSettings &s) {
 040  j = nlohmann::json{{"cfg_path", s.cfg_path}};
 041}
 42
 043void from_json(const nlohmann::json &j, AmetekS710EuresysCoaxlinkOctoSettings &s) {
 044  j.at("cfg_path").get_to(s.cfg_path);
 045}
 46
 47// -------------------------------------------------------------------------------------------------
 48// Private implementation types
 49// -------------------------------------------------------------------------------------------------
 50
 51namespace {
 52
 053void check(bool condition, const std::string &msg) {
 054  if (!condition) {
 055    logger()->error("[AmetekS710EuresysCoaxlinkOctoFactory] error: {}", msg);
 056    throw std::invalid_argument("AmetekS710EuresysCoaxlinkOctoFactory error: " + msg);
 57  }
 058}
 59
 060std::string format_genapi_error(const Euresys::genapi_error &err) {
 061  std::ostringstream oss;
 062  oss << "GenApi error: code=" << err.genapi_error_code << ", what=\"" << err.what() << "\"";
 63
 064  size_t count = err.parameter_count();
 065  if (count > 0) {
 066    oss << ", parameters=[";
 067    for (size_t i = 0; i < count; ++i) {
 068      oss << "{";
 069      auto type = err.parameter_type(i);
 070      switch (type) {
 71      case GenTL::EuresysCustomGenTL::GENAPI_ERROR_PARAMETER_TYPE_STRING:
 072        oss << "string:" << err.string_parameter(i);
 073        break;
 74      case GenTL::EuresysCustomGenTL::GENAPI_ERROR_PARAMETER_TYPE_INTEGER:
 075        oss << "int:" << err.integer_parameter(i);
 076        break;
 77      case GenTL::EuresysCustomGenTL::GENAPI_ERROR_PARAMETER_TYPE_FLOAT:
 078        oss << "float:" << err.float_parameter(i);
 079        break;
 80      default:
 081        oss << "unknown";
 82        break;
 83      }
 084      oss << "}";
 085      if (i + 1 < count)
 086        oss << ", ";
 087    }
 088    oss << "]";
 89  }
 90
 091  return oss.str();
 092}
 93
 94std::optional<Euresys::EGrabberCameraInfo> find_camera(Euresys::EGenTL   &gentl,
 095                                                       const std::string &camera_name) {
 96  using namespace Euresys;
 97
 098  EGrabberDiscovery discovery(gentl);
 099  discovery.discover();
 0100  for (int i = 0; i < discovery.cameraCount(); ++i) {
 0101    auto info = discovery.cameras(i);
 0102    auto g    = EGrabber(info);
 0103    auto name = g.getString<DeviceModule>("DeviceModelName");
 0104    if (name == camera_name) {
 0105      return info;
 106    }
 0107  }
 108
 0109  return std::nullopt;
 0110}
 111
 0112void configure_grabber(Euresys::EGrabberCameraInfo &info, const nlohmann::json &cfg) {
 113  using namespace Euresys;
 114
 115  // Bank mapping: 2-grabber vs 4-grabber configurations
 0116  const std::map<std::size_t, std::string> banks_map = {
 117      {2, "Banks_AB"},
 118      {4, "Banks_ABCD"},
 119  };
 120
 121  // Extract configuration parameters
 0122  const auto width                 = cfg.at("Width").get<std::size_t>();
 0123  const auto height                = cfg.at("Height").get<std::size_t>();
 0124  const auto nb_grabbers           = info.grabbers.size();
 0125  const auto pixel_format          = cfg.at("PixelFormat").get<std::string>();
 0126  const auto trigger_source        = cfg.at("TriggerSource").get<std::string>();
 0127  const auto trigger_mode          = cfg.at("TriggerMode").get<std::string>();
 0128  const auto exposure_time         = cfg.at("ExposureTime").get<std::size_t>();
 0129  const auto cycle_min_period      = cfg.at("CycleMinimumPeriod").get<std::size_t>();
 0130  const auto gain_selector         = cfg.at("GainSelector").get<std::string>();
 0131  const auto gain                  = cfg.at("Gain").get<float>();
 0132  const auto balance_white_marker  = cfg.at("BalanceWhiteMarker").get<std::string>();
 0133  const auto flat_field_correction = cfg.at("FlatFieldCorrection").get<std::string>();
 0134  const auto buffer_part_count     = cfg.at("BufferPartCount").get<std::size_t>();
 135
 136  // Computed parameters
 0137  const auto stripe_height            = height / nb_grabbers;
 0138  const auto stripe_pitch             = height;
 0139  const auto block_height             = stripe_height;
 0140  const auto camera_control_method    = "RC";
 0141  const auto exposure_readout_overlap = "True";
 0142  const auto error_selector           = "All";
 0143  const auto stripe_arrangement       = "Geometry_1X_2YM";
 0144  const auto statistics_sampling_sel  = "LastSecond";
 0145  const auto lut_configuration        = "M_10x8";
 0146  const auto banks                    = banks_map.at(nb_grabbers);
 147
 148  try {
 149    // Device-level master configuration
 0150    auto g = Euresys::EGrabber(info);
 0151    g.execute<Euresys::DeviceModule>("DeviceReset");
 0152    g.setString<Euresys::RemoteModule>("Banks", banks);
 0153    g.setString<Euresys::DeviceModule>("CameraControlMethod", camera_control_method);
 154
 0155    if (trigger_source == "SWTRIGGER") {
 0156      g.setString<Euresys::DeviceModule>("ErrorSelector", error_selector);
 0157      g.setInteger<Euresys::DeviceModule>("CycleMinimumPeriod", cycle_min_period);
 0158      g.setString<Euresys::DeviceModule>("ExposureReadoutOverlap", exposure_readout_overlap);
 159    }
 160
 161    // Remote configuration
 0162    g.setInteger<Euresys::RemoteModule>("Width", width);
 0163    g.setInteger<Euresys::RemoteModule>("Height", height / nb_grabbers);
 0164    g.setString<Euresys::RemoteModule>("PixelFormat", pixel_format);
 0165    g.setString<Euresys::RemoteModule>("TriggerMode", trigger_mode);
 0166    g.setString<Euresys::RemoteModule>("TriggerSource", trigger_source);
 0167    g.setInteger<Euresys::RemoteModule>("ExposureTime", exposure_time);
 0168    g.setString<Euresys::RemoteModule>("BalanceWhiteMarker", balance_white_marker);
 0169    g.setString<Euresys::RemoteModule>("GainSelector", gain_selector);
 0170    g.setFloat<Euresys::RemoteModule>("Gain", gain);
 0171    g.setString<Euresys::RemoteModule>("FlatFieldCorrection", flat_field_correction);
 172
 173    // Stream configuration
 0174    g.setString<Euresys::StreamModule>("StripeArrangement", stripe_arrangement);
 0175    g.setInteger<Euresys::StreamModule>("LinePitch", width);
 0176    g.setInteger<Euresys::StreamModule>("LineWidth", width);
 0177    g.setInteger<Euresys::StreamModule>("StripeHeight", stripe_height);
 0178    g.setInteger<Euresys::StreamModule>("StripePitch", stripe_pitch);
 0179    g.setInteger<Euresys::StreamModule>("BlockHeight", block_height);
 0180    g.setString<Euresys::StreamModule>("LUTConfiguration", lut_configuration);
 0181    g.setInteger<Euresys::StreamModule>("BufferPartCount", buffer_part_count);
 0182    g.setString<Euresys::StreamModule>("StatisticsSamplingSelector", statistics_sampling_sel);
 0183  } catch (const Euresys::genapi_error &e) {
 0184    throw std::runtime_error(format_genapi_error(e));
 0185  }
 0186}
 187
 188HostPtr<uint8_t> allocate_buffers(Euresys::EGrabber<> &g, std::size_t nb_buffers,
 0189                                  std::size_t buffer_size) {
 0190  const auto size    = buffer_size * nb_buffers;
 0191  auto       buffers = curaii::make_unique_host_ptr<uint8_t>(size);
 192
 0193  for (size_t buf_idx = 0; buf_idx < nb_buffers; ++buf_idx) {
 0194    auto *buff_ptr = buffers.get() + buf_idx * buffer_size;
 0195    auto  memory   = Euresys::UserMemory(buff_ptr, buffer_size);
 0196    g.announceAndQueue(memory);
 0197  }
 198
 0199  return buffers;
 0200}
 201
 202} // namespace
 203
 204// -------------------------------------------------------------------------------------------------
 205// Task implementation (private)
 206// -------------------------------------------------------------------------------------------------
 207
 208class AmetekS710EuresysCoaxlinkOcto : public holoflow::core::ISyncTask {
 209public:
 210  AmetekS710EuresysCoaxlinkOcto(const AmetekS710EuresysCoaxlinkOctoSettings &settings,
 211                                HostPtr<uint8_t>                           &&buffers,
 212                                std::unique_ptr<Euresys::EGenTL>           &&gentl,
 213                                std::unique_ptr<Euresys::EGrabber<>>       &&grabber,
 214                                const nlohmann::json                        &cfg)
 0215      : settings_(settings), buffers_(std::move(buffers)), gentl_(std::move(gentl)),
 0216        grabber_(std::move(grabber)), running_(false), cfg_(cfg) {
 0217    HOLOVIBES_CHECK(grabber_ != nullptr);
 0218    HOLOVIBES_CHECK(gentl_ != nullptr);
 0219    HOLOVIBES_CHECK(buffers_ != nullptr);
 0220  }
 221
 0222  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 223    using namespace Euresys;
 0224    constexpr auto DELIVERED = ge::BUFFER_INFO_CUSTOM_NUM_DELIVERED_PARTS;
 0225    constexpr auto TIMESTAMP = GenTL::BUFFER_INFO_TIMESTAMP;
 226
 0227    if (!running_) {
 0228      grabber_->start();
 0229      running_ = true;
 230    }
 231
 0232    while (!ctx.cancelled->load()) {
 233      try {
 0234        auto buffer    = ScopedBuffer(*grabber_, 1000);
 0235        auto delivered = buffer.getInfo<uint64_t>(DELIVERED);
 0236        auto ts        = buffer.getInfo<uint64_t>(TIMESTAMP);
 237
 0238        logger()->trace(
 239            "[AmetekS710EuresysCoaxlinkOcto::execute] buffer with {} parts, timestamp {}",
 240            delivered, ts);
 241
 0242        const auto *idata = buffer.getInfo<void *>(GenTL::BUFFER_INFO_BASE);
 0243        auto       *odata = ctx.outputs[0].data();
 0244        std::memcpy(odata, idata, ctx.outputs[0].desc.num_bytes());
 0245        return holoflow::core::OpResult::Ok;
 0246      } catch (const Euresys::genapi_error &err) {
 0247        logger()->error("[AmetekS710EuresysCoaxlinkOcto::execute] GenApi error: {}", err.what());
 0248      } catch (const Euresys::gentl_error &err) {
 0249        logger()->error("[AmetekS710EuresysCoaxlinkOcto::execute] GenTL error: {}", err.what());
 0250      } catch (const std::exception &err) {
 0251        logger()->error("[AmetekS710EuresysCoaxlinkOcto::execute] error: {}", err.what());
 0252      }
 0253    }
 0254    return holoflow::core::OpResult::Cancelled;
 0255  }
 256
 257  // Expose config for update() comparison
 0258  const nlohmann::json &get_cfg() const { return cfg_; }
 259
 260private:
 261  AmetekS710EuresysCoaxlinkOctoSettings settings_;
 262  HostPtr<uint8_t>                      buffers_;
 263  std::unique_ptr<Euresys::EGenTL>      gentl_;
 264  std::unique_ptr<Euresys::EGrabber<>>  grabber_;
 265  bool                                  running_;
 266  nlohmann::json                        cfg_;
 267};
 268
 269// -------------------------------------------------------------------------------------------------
 270// Factory implementation
 271// -------------------------------------------------------------------------------------------------
 272
 273holoflow::core::InferResult
 274AmetekS710EuresysCoaxlinkOctoFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 0275                                            const nlohmann::json &jsettings) const {
 276  // clang-format off
 0277  check(input_descs.size() == 0, "expected zero input tensors");
 278
 0279  auto settings = jsettings.get<AmetekS710EuresysCoaxlinkOctoSettings>();
 0280  check(!settings.cfg_path.empty(), "cfg_path is empty");
 281
 0282  std::ifstream cfg_file(settings.cfg_path);
 0283  check(cfg_file.is_open(), std::format("could not open config file: {}", settings.cfg_path));
 284
 0285  auto        cfg    = nlohmann::json::parse(cfg_file).at("s710");
 0286  std::string format = cfg.at("PixelFormat");
 287  // clang-format on
 288
 0289  static const std::map<std::string, holoflow::core::DType> dtypes = {
 290      {"Mono8", holoflow::core::DType::U8},
 291      {"Mono16", holoflow::core::DType::U16},
 292  };
 293
 0294  check(dtypes.contains(format), "unsupported PixelFormat: " + format);
 295
 0296  const auto batch_size = cfg.at("BufferPartCount").get<size_t>();
 0297  const auto height     = cfg.at("Height").get<size_t>();
 0298  const auto width      = cfg.at("Width").get<size_t>();
 0299  const auto dtype      = dtypes.at(format);
 0300  const auto loc        = holoflow::core::MemLoc::Host;
 301
 0302  holoflow::core::TDesc odesc({batch_size, height, width}, dtype, loc);
 303
 0304  return holoflow::core::InferResult{
 305      .input_descs   = {},
 306      .output_descs  = {odesc},
 307      .in_place      = {},
 308      .owned_inputs  = {},
 309      .owned_outputs = {false},
 310      .kind          = holoflow::core::TaskKind::Sync,
 311  };
 0312}
 313
 314std::unique_ptr<holoflow::core::ISyncTask>
 315AmetekS710EuresysCoaxlinkOctoFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 316                                             const nlohmann::json                  &jsettings,
 0317                                             const holoflow::core::SyncCreateCtx   &ctx) const {
 318  (void)ctx;
 319
 0320  auto settings = jsettings.get<AmetekS710EuresysCoaxlinkOctoSettings>();
 0321  auto cfg_file = std::ifstream(settings.cfg_path);
 0322  auto cfg      = nlohmann::json::parse(cfg_file).at("s710");
 323
 324  // Setup GenTL
 0325  auto gentl       = std::make_unique<Euresys::EGenTL>();
 0326  auto camera_info = find_camera(*gentl, "Phantom S710");
 327
 0328  check(camera_info.has_value(), "could not find Phantom S710 camera");
 329
 0330  configure_grabber(*camera_info, cfg);
 0331  auto grabber      = std::make_unique<Euresys::EGrabber<>>(*camera_info);
 0332  auto infer_result = this->infer(input_descs, jsettings);
 0333  auto buffer_size  = infer_result.output_descs[0].num_bytes();
 0334  auto buffers      = allocate_buffers(*grabber, cfg.at("BufferPartCount"), buffer_size);
 335
 0336  return std::make_unique<AmetekS710EuresysCoaxlinkOcto>(settings, std::move(buffers),
 337                                                         std::move(gentl), std::move(grabber), cfg);
 0338}
 339
 340std::unique_ptr<holoflow::core::ISyncTask>
 341AmetekS710EuresysCoaxlinkOctoFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 342                                             std::span<const holoflow::core::TDesc>     input_descs,
 343                                             const nlohmann::json                      &jsettings,
 0344                                             const holoflow::core::SyncCreateCtx       &ctx) const {
 345  (void)ctx;
 346
 0347  auto *old = dynamic_cast<AmetekS710EuresysCoaxlinkOcto *>(old_task.get());
 0348  if (old == nullptr || input_descs.size() != 0) {
 0349    return create(input_descs, jsettings, ctx);
 350  }
 351
 0352  auto settings = jsettings.get<AmetekS710EuresysCoaxlinkOctoSettings>();
 0353  auto cfg_file = std::ifstream(settings.cfg_path);
 0354  auto cfg      = nlohmann::json::parse(cfg_file).at("s710");
 355
 0356  if (cfg == old->get_cfg()) {
 0357    return old_task; // Reuse if config unchanged
 358  }
 359
 0360  return create(input_descs, jsettings, ctx);
 0361}
 362
 363} // namespace holotask::sources
 364
 365#else
 366
 367#include <stdexcept>
 368
 369namespace holotask::sources {
 370
 371void to_json(nlohmann::json &j, const AmetekS710EuresysCoaxlinkOctoSettings &s) {
 372  j = nlohmann::json{{"cfg_path", s.cfg_path}};
 373}
 374
 375void from_json(const nlohmann::json &j, AmetekS710EuresysCoaxlinkOctoSettings &s) {
 376  j.at("cfg_path").get_to(s.cfg_path);
 377}
 378
 379holoflow::core::InferResult
 380AmetekS710EuresysCoaxlinkOctoFactory::infer(std::span<const holoflow::core::TDesc>,
 381                                            const nlohmann::json &) const {
 382  throw std::logic_error("holotask library was built without EGrabber support");
 383}
 384
 385std::unique_ptr<holoflow::core::ISyncTask>
 386AmetekS710EuresysCoaxlinkOctoFactory::create(std::span<const holoflow::core::TDesc>,
 387                                             const nlohmann::json &,
 388                                             const holoflow::core::SyncCreateCtx &) const {
 389  throw std::logic_error("holotask library was built without EGrabber support");
 390}
 391
 392std::unique_ptr<holoflow::core::ISyncTask> AmetekS710EuresysCoaxlinkOctoFactory::update(
 393    std::unique_ptr<holoflow::core::ISyncTask>, std::span<const holoflow::core::TDesc>,
 394    const nlohmann::json &, const holoflow::core::SyncCreateCtx &) const {
 395  throw std::logic_error("holotask library was built without EGrabber support");
 396}
 397
 398} // namespace holotask::sources
 399
 400#endif
#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 "holotask/sources/ametek_s711_euresys_coaxlink_qsfp+.hh"
 16
 17#ifdef HOLOTASK_HAS_EGRABBER
 18
 19#include <EGrabber.h>
 20#include <EuresysGenapiErrorFormats.h>
 21
 22#include <cstdint>
 23#include <format>
 24#include <fstream>
 25#include <map>
 26#include <optional>
 27#include <sstream>
 28#include <string>
 29#include <utility>
 30#include <vector>
 31
 32#include "bug.hh"
 33#include "curaii/cuda.hh"
 34#include "logger.hh"
 35
 36template <typename T> using HostPtr = curaii::unique_host_ptr<T>;
 37
 38namespace holotask::sources {
 39
 40// -------------------------------------------------------------------------------------------------
 41// JSON serialization
 42// -------------------------------------------------------------------------------------------------
 43
 044void to_json(nlohmann::json &j, const AmetekS711EuresysCoaxlinkQSFPSettings &s) {
 045  j = nlohmann::json{{"cfg_path", s.cfg_path}};
 046}
 47
 048void from_json(const nlohmann::json &j, AmetekS711EuresysCoaxlinkQSFPSettings &s) {
 049  j.at("cfg_path").get_to(s.cfg_path);
 050}
 51
 52// -------------------------------------------------------------------------------------------------
 53// Private implementation types
 54// -------------------------------------------------------------------------------------------------
 55
 56namespace {
 57
 058void check(bool condition, const std::string &msg) {
 059  if (!condition) {
 060    logger()->error("[AmetekS711EuresysCoaxlinkQSFPFactory] error: {}", msg);
 061    throw std::invalid_argument("AmetekS711EuresysCoaxlinkQSFPFactory error: " + msg);
 62  }
 063}
 64
 065std::string format_genapi_error(const Euresys::genapi_error &err) {
 066  std::ostringstream oss;
 067  oss << "GenApi error: code=" << err.genapi_error_code << ", what=\"" << err.what() << "\"";
 68
 069  const size_t count = err.parameter_count();
 070  if (count > 0) {
 071    oss << ", parameters=[";
 072    for (size_t i = 0; i < count; ++i) {
 073      oss << "{";
 074      switch (err.parameter_type(i)) {
 75      case GenTL::EuresysCustomGenTL::GENAPI_ERROR_PARAMETER_TYPE_STRING:
 076        oss << "string:" << err.string_parameter(i);
 077        break;
 78      case GenTL::EuresysCustomGenTL::GENAPI_ERROR_PARAMETER_TYPE_INTEGER:
 079        oss << "int:" << err.integer_parameter(i);
 080        break;
 81      case GenTL::EuresysCustomGenTL::GENAPI_ERROR_PARAMETER_TYPE_FLOAT:
 082        oss << "float:" << err.float_parameter(i);
 083        break;
 84      default:
 085        oss << "unknown";
 86        break;
 87      }
 088      oss << "}";
 089      if (i + 1 < count) {
 090        oss << ", ";
 91      }
 092    }
 093    oss << "]";
 94  }
 95
 096  return oss.str();
 097}
 98
 99struct RuntimeConfig {
 100  std::string                camera_model_name;
 101  std::size_t                expected_grabber_count;
 102  std::size_t                nb_buffers;
 103  std::size_t                buffer_part_count;
 104  std::size_t                final_height;
 105  std::size_t                width;
 106  std::string                pixel_format;
 107  std::size_t                bytes_per_pixel;
 108  std::string                banks;
 109  std::string                trigger_source;
 110  std::string                trigger_mode;
 111  std::string                trigger_selector;
 112  std::string                gain_selector;
 113  float                      gain;
 114  std::string                flat_field_correction;
 115  std::optional<std::string> balance_white_marker;
 116  double                     exposure_time;
 117  std::int64_t               cycle_minimum_period;
 118  std::vector<std::size_t>   offsets;
 119  std::size_t                line_width;
 120  std::size_t                line_pitch;
 121  std::size_t                stripe_height;
 122  std::size_t                stripe_pitch;
 123  std::size_t                block_height;
 124  std::string                stripe_arrangement;
 125  std::size_t                pop_timeout_ms;
 126
 0127  [[nodiscard]] std::size_t camera_height() const {
 0128    if (banks == "Banks_AB") {
 0129      check(final_height % 2 == 0, "final height must be even in Banks_AB mode");
 0130      return final_height / 2;
 131    }
 0132    return final_height;
 0133  }
 134};
 135
 0136nlohmann::json load_cfg(const std::string &cfg_path) {
 0137  check(!cfg_path.empty(), "cfg_path is empty");
 138
 0139  std::ifstream cfg_file(cfg_path);
 0140  check(cfg_file.is_open(), std::format("could not open config file: {}", cfg_path));
 141
 0142  auto root = nlohmann::json::parse(cfg_file);
 0143  check(root.contains("s711"), "config file does not contain top-level key 's711'");
 0144  return root.at("s711");
 0145}
 146
 0147RuntimeConfig parse_cfg(const nlohmann::json &cfg) {
 0148  static const std::map<std::string, std::size_t> pixel_format_map = {
 149      {"Mono8", 1},
 150      {"Mono16", 2},
 151  };
 152
 0153  const auto pixel_format = cfg.at("PixelFormat").get<std::string>();
 0154  check(pixel_format_map.contains(pixel_format), "unsupported PixelFormat: " + pixel_format);
 155
 156  RuntimeConfig out{
 0157      .camera_model_name      = cfg.value("CameraModelName", std::string("Phantom S711")),
 0158      .expected_grabber_count = cfg.value("ExpectedGrabberCount", std::size_t(2)),
 0159      .nb_buffers             = cfg.at("NbBuffers").get<std::size_t>(),
 0160      .buffer_part_count      = cfg.at("BufferPartCount").get<std::size_t>(),
 0161      .final_height           = cfg.value("FinalHeight", cfg.at("Height").get<std::size_t>()),
 0162      .width                  = cfg.at("Width").get<std::size_t>(),
 0163      .pixel_format           = pixel_format,
 0164      .bytes_per_pixel        = pixel_format_map.at(pixel_format),
 0165      .banks                  = cfg.value("Banks", std::string("Banks_AB")),
 0166      .trigger_source         = cfg.at("TriggerSource").get<std::string>(),
 0167      .trigger_mode           = cfg.at("TriggerMode").get<std::string>(),
 0168      .trigger_selector       = cfg.at("TriggerSelector").get<std::string>(),
 0169      .gain_selector          = cfg.at("GainSelector").get<std::string>(),
 0170      .gain                   = cfg.at("Gain").get<float>(),
 0171      .flat_field_correction  = cfg.at("FlatFieldCorrection").get<std::string>(),
 172      .balance_white_marker =
 173          cfg.contains("BalanceWhiteMarker")
 0174              ? std::optional<std::string>(cfg.at("BalanceWhiteMarker").get<std::string>())
 175              : std::nullopt,
 0176      .exposure_time        = cfg.at("ExposureTime").get<double>(),
 0177      .cycle_minimum_period = cfg.at("CycleMinimumPeriod").get<std::int64_t>(),
 0178      .offsets              = cfg.at("Offsets").get<std::vector<std::size_t>>(),
 0179      .line_width           = cfg.value("LineWidth", cfg.at("Width").get<std::size_t>() *
 180                                                         pixel_format_map.at(pixel_format)),
 0181      .line_pitch           = cfg.value("LinePitch", cfg.at("Width").get<std::size_t>() *
 182                                                         pixel_format_map.at(pixel_format)),
 0183      .stripe_height        = cfg.value("StripeHeight", std::size_t(8)),
 0184      .stripe_pitch         = cfg.value("StripePitch", std::size_t(16)),
 0185      .block_height         = cfg.value("BlockHeight", std::size_t(8)),
 0186      .stripe_arrangement   = cfg.value("StripeArrangement", std::string("Geometry_1X_2YM")),
 0187      .pop_timeout_ms       = cfg.value("PopTimeoutMs", std::size_t(1000)),
 188  };
 189
 0190  check(out.expected_grabber_count == 2,
 191        "only two-grabber S711 Banks_AB acquisition is implemented");
 0192  check(out.offsets.size() == out.expected_grabber_count,
 193        "Offsets array size must match expected grabber count");
 0194  check(out.banks == "Banks_AB", "only Banks_AB mode is implemented");
 195
 0196  return out;
 0197}
 198
 0199nlohmann::json normalized_cfg_json(const RuntimeConfig &cfg) {
 0200  return nlohmann::json{
 201      {"CameraModelName", cfg.camera_model_name},
 202      {"ExpectedGrabberCount", cfg.expected_grabber_count},
 203      {"NbBuffers", cfg.nb_buffers},
 204      {"BufferPartCount", cfg.buffer_part_count},
 205      {"Height", cfg.final_height},
 206      {"Width", cfg.width},
 207      {"PixelFormat", cfg.pixel_format},
 208      {"Banks", cfg.banks},
 209      {"TriggerSource", cfg.trigger_source},
 210      {"TriggerMode", cfg.trigger_mode},
 211      {"TriggerSelector", cfg.trigger_selector},
 212      {"GainSelector", cfg.gain_selector},
 213      {"Gain", cfg.gain},
 214      {"FlatFieldCorrection", cfg.flat_field_correction},
 215      {"ExposureTime", cfg.exposure_time},
 216      {"CycleMinimumPeriod", cfg.cycle_minimum_period},
 217      {"Offsets", cfg.offsets},
 218      {"LineWidth", cfg.line_width},
 219      {"LinePitch", cfg.line_pitch},
 220      {"StripeHeight", cfg.stripe_height},
 221      {"StripePitch", cfg.stripe_pitch},
 222      {"BlockHeight", cfg.block_height},
 223      {"StripeArrangement", cfg.stripe_arrangement},
 224      {"PopTimeoutMs", cfg.pop_timeout_ms},
 225  };
 0226}
 227
 0228void dump_cfg(const nlohmann::json &raw_cfg, const RuntimeConfig &cfg) {
 0229  logger()->info("[AmetekS711EuresysCoaxlinkQSFPFactory] loaded config:\n{}", raw_cfg.dump(2));
 0230  logger()->info(
 231      "[AmetekS711EuresysCoaxlinkQSFPFactory] derived config: banks={}, final_height={}, "
 232      "camera_height={}, width={}, pixel_format={}, line_width={}, line_pitch={}, "
 233      "stripe_height={}, stripe_pitch={}, block_height={}, stripe_arrangement={}",
 234      cfg.banks, cfg.final_height, cfg.camera_height(), cfg.width, cfg.pixel_format, cfg.line_width,
 235      cfg.line_pitch, cfg.stripe_height, cfg.stripe_pitch, cfg.block_height,
 236      cfg.stripe_arrangement);
 0237}
 238
 239std::optional<Euresys::EGrabberCameraInfo> find_camera(Euresys::EGenTL   &gentl,
 0240                                                       const std::string &camera_name) {
 241  using namespace Euresys;
 242
 0243  EGrabberDiscovery discovery(gentl);
 0244  discovery.discover();
 245
 0246  for (int i = 0; i < discovery.cameraCount(); ++i) {
 0247    auto info = discovery.cameras(i);
 0248    auto g    = EGrabber<>(info);
 249
 250    try {
 0251      if (g.getString<RemoteModule>("DeviceModelName") == camera_name) {
 0252        return info;
 253      }
 0254    } catch (const Euresys::genapi_error &) {
 255      try {
 0256        if (g.getString<DeviceModule>("DeviceModelName") == camera_name) {
 0257          return info;
 258        }
 0259      } catch (const Euresys::genapi_error &) {
 0260      }
 0261    }
 0262  }
 263
 0264  return std::nullopt;
 0265}
 266
 0267std::size_t find_grabber_index_for_bank(Euresys::EGrabberCameraInfo &info, std::int64_t bank_id) {
 268  using namespace Euresys;
 269
 0270  for (std::size_t i = 0; i < info.grabbers.size(); ++i) {
 0271    EGrabber<> g(info.grabbers[i]);
 0272    if (g.getInteger<RemoteModule>("ConnectedBankID") == bank_id) {
 0273      return i;
 274    }
 0275  }
 276
 0277  throw std::runtime_error(std::format("could not find grabber for ConnectedBankID={}", bank_id));
 0278}
 279
 280template <typename Module>
 0281void dump_string(Euresys::EGrabber<> &g, const std::string &prefix, const char *name) {
 282  try {
 0283    logger()->info("{} {}={}", prefix, name, g.getString<Module>(std::string(name)));
 0284  } catch (const Euresys::genapi_error &e) {
 0285    logger()->info("{} {}=<unavailable: {}>", prefix, name, format_genapi_error(e));
 0286  }
 0287}
 288
 289template <typename Module>
 0290void dump_int(Euresys::EGrabber<> &g, const std::string &prefix, const char *name) {
 291  try {
 0292    logger()->info("{} {}={}", prefix, name, g.getInteger<Module>(std::string(name)));
 0293  } catch (const Euresys::genapi_error &e) {
 0294    logger()->info("{} {}=<unavailable: {}>", prefix, name, format_genapi_error(e));
 0295  }
 0296}
 297
 298template <typename Module>
 0299void dump_float(Euresys::EGrabber<> &g, const std::string &prefix, const char *name) {
 300  try {
 0301    logger()->info("{} {}={}", prefix, name, g.getFloat<Module>(std::string(name)));
 0302  } catch (const Euresys::genapi_error &e) {
 0303    logger()->info("{} {}=<unavailable: {}>", prefix, name, format_genapi_error(e));
 0304  }
 0305}
 306
 0307void dump_state(Euresys::EGrabberCameraInfo &info, const std::string &phase) {
 308  using namespace Euresys;
 309
 0310  logger()->info("[AmetekS711EuresysCoaxlinkQSFPFactory] ===== {} =====", phase);
 311
 0312  for (std::size_t i = 0; i < info.grabbers.size(); ++i) {
 0313    EGrabber<> g(info.grabbers[i]);
 0314    const auto prefix =
 315        std::format("[AmetekS711EuresysCoaxlinkQSFPFactory][{}] grabber[{}]", phase, i);
 316
 0317    dump_int<RemoteModule>(g, prefix, "ConnectedBankID");
 0318    dump_string<RemoteModule>(g, prefix, "Banks");
 0319    dump_int<RemoteModule>(g, prefix, "Width");
 0320    dump_int<RemoteModule>(g, prefix, "Height");
 0321    dump_string<RemoteModule>(g, prefix, "PixelFormat");
 0322    dump_string<RemoteModule>(g, prefix, "TriggerMode");
 0323    dump_string<RemoteModule>(g, prefix, "TriggerSource");
 0324    dump_string<RemoteModule>(g, prefix, "TriggerSelector");
 0325    dump_float<RemoteModule>(g, prefix, "ExposureTime");
 0326    dump_string<RemoteModule>(g, prefix, "GainSelector");
 0327    dump_float<RemoteModule>(g, prefix, "Gain");
 0328    dump_string<RemoteModule>(g, prefix, "FlatFieldCorrection");
 329
 0330    dump_string<DeviceModule>(g, prefix, "CameraControlMethod");
 0331    dump_string<DeviceModule>(g, prefix, "ExposureReadoutOverlap");
 0332    dump_string<DeviceModule>(g, prefix, "ErrorSelector");
 0333    dump_int<DeviceModule>(g, prefix, "CycleMinimumPeriod");
 334
 0335    dump_int<StreamModule>(g, prefix, "BufferPartCount");
 0336    dump_int<StreamModule>(g, prefix, "LineWidth");
 0337    dump_int<StreamModule>(g, prefix, "LinePitch");
 0338    dump_int<StreamModule>(g, prefix, "StripeHeight");
 0339    dump_int<StreamModule>(g, prefix, "StripePitch");
 0340    dump_int<StreamModule>(g, prefix, "BlockHeight");
 0341    dump_int<StreamModule>(g, prefix, "StripeOffset");
 0342    dump_string<StreamModule>(g, prefix, "StripeArrangement");
 0343  }
 0344}
 345
 346template <typename Module>
 347void set_required_string(Euresys::EGrabber<> &g, const std::string &prefix, const char *name,
 0348                         const std::string &value) {
 349  try {
 0350    g.setString<Module>(std::string(name), value);
 0351    logger()->info("{} set {}={}", prefix, name, value);
 0352  } catch (const Euresys::genapi_error &e) {
 0353    throw std::runtime_error(
 354        std::format("{} failed to set {}={}: {}", prefix, name, value, format_genapi_error(e)));
 0355  }
 0356}
 357
 358template <typename Module>
 359void set_required_int(Euresys::EGrabber<> &g, const std::string &prefix, const char *name,
 0360                      std::int64_t value) {
 361  try {
 0362    g.setInteger<Module>(std::string(name), value);
 0363    logger()->info("{} set {}={}", prefix, name, value);
 0364  } catch (const Euresys::genapi_error &e) {
 0365    throw std::runtime_error(
 366        std::format("{} failed to set {}={}: {}", prefix, name, value, format_genapi_error(e)));
 0367  }
 0368}
 369
 370template <typename Module>
 371void set_required_float(Euresys::EGrabber<> &g, const std::string &prefix, const char *name,
 0372                        double value) {
 373  try {
 0374    g.setFloat<Module>(std::string(name), value);
 0375    logger()->info("{} set {}={}", prefix, name, value);
 0376  } catch (const Euresys::genapi_error &e) {
 0377    throw std::runtime_error(
 378        std::format("{} failed to set {}={}: {}", prefix, name, value, format_genapi_error(e)));
 0379  }
 0380}
 381
 382template <typename Module>
 383void set_optional_string(Euresys::EGrabber<> &g, const std::string &prefix, const char *name,
 0384                         const std::string &value) {
 385  try {
 0386    g.setString<Module>(std::string(name), value);
 0387    logger()->info("{} set {}={}", prefix, name, value);
 0388  } catch (const Euresys::genapi_error &e) {
 0389    logger()->warn("{} could not set {}={}: {}", prefix, name, value, format_genapi_error(e));
 0390  }
 0391}
 392
 0393void apply_cfg(Euresys::EGrabberCameraInfo &info, const RuntimeConfig &cfg) {
 394  using namespace Euresys;
 395
 0396  check(info.grabbers.size() == cfg.expected_grabber_count,
 397        std::format("expected {} grabber(s), got {}", cfg.expected_grabber_count,
 398                    info.grabbers.size()));
 399
 0400  const auto bank_a_index = find_grabber_index_for_bank(info, 0);
 0401  const auto bank_b_index = find_grabber_index_for_bank(info, 1);
 402
 0403  EGrabber<> ctrl(info.grabbers[bank_a_index]);
 0404  const auto ctrl_prefix = "[AmetekS711EuresysCoaxlinkQSFPFactory][apply][bankA-control]";
 405
 406  // Camera-side settings are shared across both banks and must be written through bank A.
 0407  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "Banks", cfg.banks);
 0408  set_required_int<RemoteModule>(ctrl, ctrl_prefix, "Width", static_cast<std::int64_t>(cfg.width));
 0409  set_required_int<RemoteModule>(ctrl, ctrl_prefix, "Height",
 410                                 static_cast<std::int64_t>(cfg.camera_height()));
 0411  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "PixelFormat", cfg.pixel_format);
 0412  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "TriggerSelector", cfg.trigger_selector);
 0413  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "TriggerMode", cfg.trigger_mode);
 0414  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "TriggerSource", cfg.trigger_source);
 0415  set_required_float<RemoteModule>(ctrl, ctrl_prefix, "ExposureTime", cfg.exposure_time);
 0416  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "GainSelector", cfg.gain_selector);
 0417  set_required_float<RemoteModule>(ctrl, ctrl_prefix, "Gain", cfg.gain);
 0418  set_required_string<RemoteModule>(ctrl, ctrl_prefix, "FlatFieldCorrection",
 419                                    cfg.flat_field_correction);
 420
 0421  if (cfg.balance_white_marker.has_value()) {
 0422    set_optional_string<RemoteModule>(ctrl, ctrl_prefix, "BalanceWhiteMarker",
 423                                      *cfg.balance_white_marker);
 424  }
 425
 0426  const auto camera_control_method =
 427      cfg.trigger_source == "SWTRIGGER" ? std::string("RC") : std::string("EXTERNAL");
 0428  set_optional_string<DeviceModule>(ctrl, ctrl_prefix, "CameraControlMethod",
 429                                    camera_control_method);
 430
 0431  if (cfg.trigger_source == "SWTRIGGER") {
 0432    set_optional_string<DeviceModule>(ctrl, ctrl_prefix, "ErrorSelector", "All");
 0433    set_optional_string<DeviceModule>(ctrl, ctrl_prefix, "ExposureReadoutOverlap", "True");
 434    try {
 0435      ctrl.setInteger<DeviceModule>("CycleMinimumPeriod", cfg.cycle_minimum_period);
 0436      logger()->info("{} set CycleMinimumPeriod={}", ctrl_prefix, cfg.cycle_minimum_period);
 0437    } catch (const Euresys::genapi_error &e) {
 0438      logger()->warn("{} could not set CycleMinimumPeriod={}: {}", ctrl_prefix,
 439                     cfg.cycle_minimum_period, format_genapi_error(e));
 0440    }
 441  }
 442
 443  auto apply_stream = [&](std::size_t grabber_index, std::size_t stripe_offset) {
 444    EGrabber<> g(info.grabbers[grabber_index]);
 445    const auto prefix =
 446        std::format("[AmetekS711EuresysCoaxlinkQSFPFactory][apply] grabber[{}]", grabber_index);
 447
 448    set_required_int<StreamModule>(g, prefix, "BufferPartCount",
 449                                   static_cast<std::int64_t>(cfg.buffer_part_count));
 450    set_required_int<StreamModule>(g, prefix, "LineWidth",
 451                                   static_cast<std::int64_t>(cfg.line_width));
 452    set_required_int<StreamModule>(g, prefix, "LinePitch",
 453                                   static_cast<std::int64_t>(cfg.line_pitch));
 454    set_required_int<StreamModule>(g, prefix, "StripeHeight",
 455                                   static_cast<std::int64_t>(cfg.stripe_height));
 456    set_required_int<StreamModule>(g, prefix, "StripePitch",
 457                                   static_cast<std::int64_t>(cfg.stripe_pitch));
 458    set_required_int<StreamModule>(g, prefix, "BlockHeight",
 459                                   static_cast<std::int64_t>(cfg.block_height));
 460    set_required_int<StreamModule>(g, prefix, "StripeOffset",
 461                                   static_cast<std::int64_t>(stripe_offset));
 462    set_required_string<StreamModule>(g, prefix, "StripeArrangement", cfg.stripe_arrangement);
 0463  };
 464
 0465  apply_stream(bank_a_index, cfg.offsets[0]);
 0466  apply_stream(bank_b_index, cfg.offsets[1]);
 0467}
 468
 469/**
 470 * Allocate a host-resident buffer pool and announce the exact same buffer slots
 471 * to both banks.
 472 *
 473 * The stream module writes each bank into different stripes of the same logical
 474 * final frame because the stream geometry has already been configured with the
 475 * appropriate StripeOffset / StripePitch / StripeArrangement values.
 476 */
 477HostPtr<uint8_t> allocate_shared_buffers(Euresys::EGrabber<> &grabber_a,
 478                                         Euresys::EGrabber<> &grabber_b, std::size_t nb_buffers,
 0479                                         std::size_t buffer_size) {
 0480  logger()->info(
 481      "[AmetekS711EuresysCoaxlinkQSFPFactory] allocating {} shared host buffers of size {} bytes",
 482      nb_buffers, buffer_size);
 483
 0484  const auto total_size = buffer_size * nb_buffers;
 0485  auto       buffers    = curaii::make_unique_host_ptr<uint8_t>(total_size);
 486
 0487  for (std::size_t buf_idx = 0; buf_idx < nb_buffers; ++buf_idx) {
 0488    auto *base = buffers.get() + buf_idx * buffer_size;
 489
 0490    grabber_a.announceAndQueue(Euresys::UserMemory(base, buffer_size));
 0491    grabber_b.announceAndQueue(Euresys::UserMemory(base, buffer_size));
 492
 0493    logger()->debug("[AmetekS711EuresysCoaxlinkQSFPFactory] announced shared buffer {} at address "
 494                    "{} to both grabbers",
 495                    buf_idx, static_cast<void *>(base));
 0496  }
 497
 0498  return buffers;
 0499}
 500
 501void requeue_buffer_noexcept(Euresys::EGrabber<> &grabber, const Euresys::NewBufferData &data,
 0502                             const char *label) {
 503  try {
 0504    Euresys::Buffer(data).push(grabber);
 0505  } catch (const std::exception &e) {
 0506    logger()->error(
 507        "[AmetekS711EuresysCoaxlinkQSFP] failed to requeue {} buffer while handling an error: {}",
 508        label, e.what());
 0509  }
 0510}
 511
 0512holoflow::core::DType dtype_from_pixel_format(const std::string &pixel_format) {
 0513  static const std::map<std::string, holoflow::core::DType> dtypes = {
 514      {"Mono8", holoflow::core::DType::U8},
 515      {"Mono16", holoflow::core::DType::U16},
 516  };
 517
 0518  check(dtypes.contains(pixel_format), "unsupported PixelFormat: " + pixel_format);
 0519  return dtypes.at(pixel_format);
 0520}
 521
 522} // namespace
 523
 524// -------------------------------------------------------------------------------------------------
 525// Task implementation (private)
 526// -------------------------------------------------------------------------------------------------
 527
 528/**
 529 * Two-bank S711 source task.
 530 *
 531 * This task exposes a host tensor. The configured stream geometry makes both
 532 * grabbers DMA into different stripes of the same final frame buffer. Each
 533 * logical output frame therefore corresponds to one queued buffer slot that is
 534 * announced to both bank A and bank B.
 535 */
 536class AmetekS711EuresysCoaxlinkQSFP : public holoflow::core::ISyncTask {
 537public:
 538  AmetekS711EuresysCoaxlinkQSFP(const AmetekS711EuresysCoaxlinkQSFPSettings &settings,
 539                                RuntimeConfig runtime_cfg, HostPtr<uint8_t> &&buffers,
 540                                std::unique_ptr<Euresys::EGenTL>     &&gentl,
 541                                std::unique_ptr<Euresys::EGrabber<>> &&grabber_a,
 542                                std::unique_ptr<Euresys::EGrabber<>> &&grabber_b,
 543                                std::size_t buffer_size, nlohmann::json normalized_cfg)
 0544      : settings_(settings), runtime_cfg_(std::move(runtime_cfg)), buffers_(std::move(buffers)),
 0545        gentl_(std::move(gentl)), grabber_a_(std::move(grabber_a)),
 0546        grabber_b_(std::move(grabber_b)), buffer_size_(buffer_size), running_(false),
 0547        cfg_(std::move(normalized_cfg)) {
 0548    HOLOVIBES_CHECK(gentl_ != nullptr);
 0549    HOLOVIBES_CHECK(grabber_a_ != nullptr);
 0550    HOLOVIBES_CHECK(grabber_b_ != nullptr);
 0551    HOLOVIBES_CHECK(buffers_ != nullptr);
 0552  }
 553
 0554  ~AmetekS711EuresysCoaxlinkQSFP() override {
 555    try {
 0556      if (running_) {
 0557        grabber_a_->stop();
 0558        grabber_b_->stop();
 559      }
 0560    } catch (const std::exception &e) {
 0561      logger()->warn("[AmetekS711EuresysCoaxlinkQSFP::~AmetekS711EuresysCoaxlinkQSFP] {}",
 562                     e.what());
 0563    }
 0564  }
 565
 0566  std::optional<holoflow::core::TView> acquire_input(int index) override {
 567    (void)index;
 0568    throw std::out_of_range("AmetekS711EuresysCoaxlinkQSFP task has no inputs");
 0569  }
 570
 571  /**
 572   * Re-queue both bank buffers for the frame currently exposed as output 0.
 573   */
 0574  void release_output(int index) override {
 0575    if (index != 0) {
 0576      throw std::out_of_range("AmetekS711EuresysCoaxlinkQSFP task has only one output at index 0");
 577    }
 578
 0579    if (!pending_a_.has_value() || !pending_b_.has_value()) {
 0580      throw std::logic_error("release_output called with no pending two-bank frame");
 581    }
 582
 0583    Euresys::Buffer(*pending_a_).push(*grabber_a_);
 0584    Euresys::Buffer(*pending_b_).push(*grabber_b_);
 0585    pending_a_.reset();
 0586    pending_b_.reset();
 0587  }
 588
 0589  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 590    using namespace Euresys;
 0591    constexpr auto DELIVERED = ge::BUFFER_INFO_CUSTOM_NUM_DELIVERED_PARTS;
 0592    constexpr auto TIMESTAMP = GenTL::BUFFER_INFO_TIMESTAMP;
 593
 0594    HOLOVIBES_CHECK(!pending_a_.has_value() && !pending_b_.has_value(),
 595                    "execute called while previous buffer pair is still held");
 596
 0597    if (!running_) {
 598      // S711 Banks_AB must start bank B first, then bank A.
 0599      grabber_b_->enableEvent<Euresys::NewBufferData>();
 0600      grabber_b_->start();
 0601      grabber_a_->enableEvent<Euresys::NewBufferData>();
 0602      grabber_a_->start();
 0603      running_ = true;
 604    }
 605
 0606    while (!ctx.cancelled->load()) {
 607      try {
 0608        const auto timeout_ms = runtime_cfg_.pop_timeout_ms;
 609
 0610        auto                                  data_a = grabber_a_->pop(timeout_ms);
 0611        std::optional<Euresys::NewBufferData> data_b;
 612        try {
 0613          data_b = grabber_b_->pop(timeout_ms);
 0614        } catch (...) {
 0615          requeue_buffer_noexcept(*grabber_a_, data_a, "bank A");
 0616          throw;
 0617        }
 618
 0619        auto buffer_a = Buffer(data_a);
 0620        auto buffer_b = Buffer(*data_b);
 621
 0622        const auto delivered_a = buffer_a.getInfo<uint64_t>(*grabber_a_, DELIVERED);
 0623        const auto delivered_b = buffer_b.getInfo<uint64_t>(*grabber_b_, DELIVERED);
 0624        const auto ts_a        = buffer_a.getInfo<uint64_t>(*grabber_a_, TIMESTAMP);
 0625        const auto ts_b        = buffer_b.getInfo<uint64_t>(*grabber_b_, TIMESTAMP);
 626
 0627        auto *base_a_v = buffer_a.getInfo<void *>(*grabber_a_, GenTL::BUFFER_INFO_BASE);
 0628        auto *base_b_v = buffer_b.getInfo<void *>(*grabber_b_, GenTL::BUFFER_INFO_BASE);
 0629        auto *base_a   = static_cast<std::byte *>(base_a_v);
 0630        auto *base_b   = static_cast<std::byte *>(base_b_v);
 631
 0632        logger()->trace("[AmetekS711EuresysCoaxlinkQSFP::execute] bankA: delivered={}, ts={}, "
 633                        "base={} | bankB: delivered={}, ts={}, base={}",
 634                        delivered_a, ts_a, static_cast<void *>(base_a), delivered_b, ts_b,
 635                        static_cast<void *>(base_b));
 636
 0637        if (base_a != base_b) {
 0638          requeue_buffer_noexcept(*grabber_a_, data_a, "bank A");
 0639          requeue_buffer_noexcept(*grabber_b_, *data_b, "bank B");
 0640          throw std::runtime_error(
 641              std::format("two-bank frame mismatch: bank A base {} != bank B base {}",
 642                          static_cast<void *>(base_a), static_cast<void *>(base_b)));
 643        }
 644
 0645        auto &storage = storage_access().owned_output_storage(0);
 0646        storage.ptr   = base_a;
 647
 0648        ctx.outputs[0] = holoflow::core::TView{
 649            .desc    = ctx.outputs[0].desc,
 650            .storage = &storage,
 651        };
 652
 0653        pending_a_ = std::move(data_a);
 0654        pending_b_ = std::move(*data_b);
 0655        return holoflow::core::OpResult::Ok;
 656
 0657      } catch (const Euresys::genapi_error &err) {
 0658        logger()->error("[AmetekS711EuresysCoaxlinkQSFP::execute] GenApi error: {}",
 659                        format_genapi_error(err));
 0660      } catch (const Euresys::gentl_error &err) {
 0661        logger()->error("[AmetekS711EuresysCoaxlinkQSFP::execute] GenTL error: {}", err.what());
 0662      } catch (const std::exception &err) {
 0663        logger()->error("[AmetekS711EuresysCoaxlinkQSFP::execute] error: {}", err.what());
 0664      }
 0665    }
 666
 0667    return holoflow::core::OpResult::Cancelled;
 0668  }
 669
 0670  const nlohmann::json &get_cfg() const { return cfg_; }
 671
 672private:
 673  AmetekS711EuresysCoaxlinkQSFPSettings settings_;
 674  RuntimeConfig                         runtime_cfg_;
 675  HostPtr<uint8_t>                      buffers_;
 676  std::unique_ptr<Euresys::EGenTL>      gentl_;
 677  std::unique_ptr<Euresys::EGrabber<>>  grabber_a_;
 678  std::unique_ptr<Euresys::EGrabber<>>  grabber_b_;
 679  std::size_t                           buffer_size_;
 680  bool                                  running_;
 681  nlohmann::json                        cfg_;
 682  std::optional<Euresys::NewBufferData> pending_a_;
 683  std::optional<Euresys::NewBufferData> pending_b_;
 684};
 685
 686// -------------------------------------------------------------------------------------------------
 687// Factory implementation
 688// -------------------------------------------------------------------------------------------------
 689
 690holoflow::core::InferResult
 691AmetekS711EuresysCoaxlinkQSFPFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 0692                                            const nlohmann::json &jsettings) const {
 0693  check(input_descs.empty(), "expected zero input tensors");
 694
 0695  const auto settings    = jsettings.get<AmetekS711EuresysCoaxlinkQSFPSettings>();
 0696  const auto raw_cfg     = load_cfg(settings.cfg_path);
 0697  const auto runtime_cfg = parse_cfg(raw_cfg);
 698
 0699  holoflow::core::TDesc odesc(
 700      {runtime_cfg.buffer_part_count, runtime_cfg.final_height, runtime_cfg.width},
 701      dtype_from_pixel_format(runtime_cfg.pixel_format), holoflow::core::MemLoc::Host);
 702
 0703  return holoflow::core::InferResult{
 704      .input_descs   = {},
 705      .output_descs  = {odesc},
 706      .in_place      = {},
 707      .owned_inputs  = {},
 708      .owned_outputs = {true},
 709      .kind          = holoflow::core::TaskKind::Sync,
 710  };
 0711}
 712
 713std::unique_ptr<holoflow::core::ISyncTask>
 714AmetekS711EuresysCoaxlinkQSFPFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 715                                             const nlohmann::json                  &jsettings,
 0716                                             const holoflow::core::SyncCreateCtx   &ctx) const {
 717  (void)ctx;
 0718  check(input_descs.empty(), "expected zero input tensors");
 719
 0720  const auto settings    = jsettings.get<AmetekS711EuresysCoaxlinkQSFPSettings>();
 0721  const auto raw_cfg     = load_cfg(settings.cfg_path);
 0722  const auto runtime_cfg = parse_cfg(raw_cfg);
 0723  dump_cfg(raw_cfg, runtime_cfg);
 724
 0725  auto gentl       = std::make_unique<Euresys::EGenTL>();
 0726  auto camera_info = find_camera(*gentl, runtime_cfg.camera_model_name);
 0727  check(camera_info.has_value(),
 728        std::format("could not find {} camera", runtime_cfg.camera_model_name));
 729
 0730  dump_state(*camera_info, "before");
 0731  apply_cfg(*camera_info, runtime_cfg);
 0732  dump_state(*camera_info, "after");
 733
 0734  const auto bank_a_index = find_grabber_index_for_bank(*camera_info, 0);
 0735  const auto bank_b_index = find_grabber_index_for_bank(*camera_info, 1);
 736
 0737  auto grabber_a = std::make_unique<Euresys::EGrabber<>>(camera_info->grabbers[bank_a_index]);
 0738  auto grabber_b = std::make_unique<Euresys::EGrabber<>>(camera_info->grabbers[bank_b_index]);
 739
 0740  const holoflow::core::TDesc odesc(
 741      {runtime_cfg.buffer_part_count, runtime_cfg.final_height, runtime_cfg.width},
 742      dtype_from_pixel_format(runtime_cfg.pixel_format), holoflow::core::MemLoc::Host);
 743
 0744  auto buffer_size = odesc.num_bytes();
 0745  auto buffers =
 746      allocate_shared_buffers(*grabber_a, *grabber_b, runtime_cfg.nb_buffers, buffer_size);
 747
 0748  return std::make_unique<AmetekS711EuresysCoaxlinkQSFP>(
 749      settings, runtime_cfg, std::move(buffers), std::move(gentl), std::move(grabber_a),
 750      std::move(grabber_b), buffer_size, normalized_cfg_json(runtime_cfg));
 0751}
 752
 753std::unique_ptr<holoflow::core::ISyncTask>
 754AmetekS711EuresysCoaxlinkQSFPFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 755                                             std::span<const holoflow::core::TDesc>     input_descs,
 756                                             const nlohmann::json                      &jsettings,
 0757                                             const holoflow::core::SyncCreateCtx       &ctx) const {
 758  (void)ctx;
 759
 0760  auto *old = dynamic_cast<AmetekS711EuresysCoaxlinkQSFP *>(old_task.get());
 0761  if (old == nullptr || !input_descs.empty()) {
 0762    return create(input_descs, jsettings, ctx);
 763  }
 764
 0765  const auto settings    = jsettings.get<AmetekS711EuresysCoaxlinkQSFPSettings>();
 0766  const auto raw_cfg     = load_cfg(settings.cfg_path);
 0767  const auto runtime_cfg = parse_cfg(raw_cfg);
 0768  const auto new_cfg     = normalized_cfg_json(runtime_cfg);
 769
 0770  if (new_cfg == old->get_cfg()) {
 0771    return old_task;
 772  }
 773
 0774  return create(input_descs, jsettings, ctx);
 0775}
 776
 777} // namespace holotask::sources
 778
 779#else
 780
 781#include <stdexcept>
 782
 783namespace holotask::sources {
 784
 785void to_json(nlohmann::json &j, const AmetekS711EuresysCoaxlinkQSFPSettings &s) {
 786  j = nlohmann::json{{"cfg_path", s.cfg_path}};
 787}
 788
 789void from_json(const nlohmann::json &j, AmetekS711EuresysCoaxlinkQSFPSettings &s) {
 790  j.at("cfg_path").get_to(s.cfg_path);
 791}
 792
 793holoflow::core::InferResult
 794AmetekS711EuresysCoaxlinkQSFPFactory::infer(std::span<const holoflow::core::TDesc>,
 795                                            const nlohmann::json &) const {
 796  throw std::logic_error("holotask library was built without EGrabber support");
 797}
 798
 799std::unique_ptr<holoflow::core::ISyncTask>
 800AmetekS711EuresysCoaxlinkQSFPFactory::create(std::span<const holoflow::core::TDesc>,
 801                                             const nlohmann::json &,
 802                                             const holoflow::core::SyncCreateCtx &) const {
 803  throw std::logic_error("holotask library was built without EGrabber support");
 804}
 805
 806std::unique_ptr<holoflow::core::ISyncTask> AmetekS711EuresysCoaxlinkQSFPFactory::update(
 807    std::unique_ptr<holoflow::core::ISyncTask>, std::span<const holoflow::core::TDesc>,
 808    const nlohmann::json &, const holoflow::core::SyncCreateCtx &) const {
 809  throw std::logic_error("holotask library was built without EGrabber support");
 810}
 811
 812} // namespace holotask::sources
 813
 814#endif

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\sources\holofile.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 "holotask/sources/holofile.hh"
 16
 17#include <algorithm>
 18#include <chrono>
 19#include <cstddef>
 20#include <cstdint>
 21#include <map>
 22#include <omp.h>
 23#include <thread>
 24
 25#include "bug.hh"
 26#include "logger.hh"
 27
 28#include "curaii/cuda.hh"
 29#include "holofile/holofile.hh"
 30
 31template <typename T> using DevPtr  = curaii::unique_device_ptr<T>;
 32template <typename T> using HostPtr = curaii::unique_host_ptr<T>;
 33
 34namespace holotask::sources {
 35
 36// -------------------------------------------------------------------------------------------------
 37// JSON serialization
 38// -------------------------------------------------------------------------------------------------
 39
 040void to_json(nlohmann::json &j, const HolofileSettings::LoadKind &lk) {
 041  static const std::map<HolofileSettings::LoadKind, std::string> lk_to_str = {
 42      {HolofileSettings::LoadKind::Live, "Live"},
 43      {HolofileSettings::LoadKind::CPUCached, "CPUCached"},
 44      {HolofileSettings::LoadKind::GPUCached, "GPUCached"},
 45  };
 46
 047  HOLOVIBES_CHECK(lk_to_str.contains(lk), "Invalid LoadKind enum value");
 048  j = lk_to_str.at(lk);
 049}
 50
 051void from_json(const nlohmann::json &j, HolofileSettings::LoadKind &lk) {
 052  static const std::map<std::string, HolofileSettings::LoadKind> str_to_lk = {
 53      {"Live", HolofileSettings::LoadKind::Live},
 54      {"CPUCached", HolofileSettings::LoadKind::CPUCached},
 55      {"GPUCached", HolofileSettings::LoadKind::GPUCached},
 56  };
 57
 058  auto key = j.get<std::string>();
 059  if (!str_to_lk.contains(key)) {
 060    throw std::invalid_argument("Invalid LoadKind string: " + key);
 61  }
 062  lk = str_to_lk.at(key);
 063}
 64
 065void to_json(nlohmann::json &j, const HolofileSettings &hs) {
 066  j = {
 67      {"path", hs.path},
 68      {"load_kind", hs.load_kind},
 69      {"start_frame", hs.start_frame},
 70      {"end_frame", hs.end_frame},
 71      {"batch_size", hs.batch_size},
 72      {"keep_cursor", hs.keep_cursor},
 73  };
 74
 075  if (hs.max_fps.has_value()) {
 076    j["max_fps"] = *hs.max_fps;
 77  }
 078}
 79
 080void from_json(const nlohmann::json &j, HolofileSettings &hs) {
 081  j.at("path").get_to(hs.path);
 082  j.at("load_kind").get_to(hs.load_kind);
 083  j.at("start_frame").get_to(hs.start_frame);
 084  j.at("end_frame").get_to(hs.end_frame);
 085  j.at("batch_size").get_to(hs.batch_size);
 086  hs.max_fps = std::nullopt;
 087  if (j.contains("max_fps") && !j.at("max_fps").is_null()) {
 088    hs.max_fps = j.at("max_fps").get<int>();
 89  }
 090  hs.keep_cursor = j.value("keep_cursor", true);
 091}
 92
 93// -------------------------------------------------------------------------------------------------
 94// Private implementation
 95// -------------------------------------------------------------------------------------------------
 96
 97namespace {
 98
 099void check(bool condition, const std::string &msg) {
 0100  if (!condition) {
 0101    logger()->error("[HolofileFactory::infer] error: {}", msg);
 0102    throw std::invalid_argument("HolofileFactory inference error: " + msg);
 103  }
 0104}
 105
 106void mt_memcpy(void *dst, const void *src, const std::size_t n) {
 107  constexpr int NUM_THREADS = 2;
 108  auto         *dst_bytes   = static_cast<std::uint8_t *>(dst);
 109  const auto   *src_bytes   = static_cast<const std::uint8_t *>(src);
 110
 111  const std::size_t chunk_size = n / NUM_THREADS;
 112  const std::size_t remainder  = n % NUM_THREADS;
 113
 114#pragma omp parallel num_threads(NUM_THREADS)
 115  {
 116    const int         tid       = omp_get_thread_num();
 117    const std::size_t offset    = tid * chunk_size;
 118    std::size_t       this_size = chunk_size;
 119
 120    if (tid == NUM_THREADS - 1) {
 121      this_size += remainder;
 122    }
 123
 124    if (this_size > 0) {
 125      std::memcpy(dst_bytes + offset, src_bytes + offset, this_size);
 126    }
 127  }
 128}
 129
 130// -------------------------------------------------------------------------------------------------
 131// Holofile task implementation (private to this translation unit)
 132// -------------------------------------------------------------------------------------------------
 133
 134class Holofile : public holoflow::core::ISyncTask {
 135public:
 136  using Clock = std::chrono::steady_clock;
 137
 138  // -- Configuration ------------------------------------------------------------------------------
 139  HolofileSettings                  settings;
 140  std::unique_ptr<holofile::Reader> reader;
 141  holofile::Header                  header;
 142  int                               frame_idx;
 143  holoflow::core::TDesc             odesc;
 144  std::optional<Clock::time_point>  next_batch_start;
 145
 146  // -- Buffers ------------------------------------------------------------------------------------
 147  std::byte         *buf;   // Non-owning view of the active buffer
 148  HostPtr<std::byte> h_buf; // Owned CPU buffer (if any)
 149  DevPtr<std::byte>  d_buf; // Owned GPU buffer (if any)
 150
 151  cudaStream_t stream; // Stream for GPU transfers
 152
 0153  [[nodiscard]] bool pace_before_batch(holoflow::core::SyncCtx &ctx) {
 0154    if (!settings.max_fps.has_value()) {
 0155      next_batch_start.reset();
 0156      return true;
 157    }
 158
 0159    auto batch_period = std::chrono::duration_cast<Clock::duration>(
 160        std::chrono::duration<double>(static_cast<double>(settings.batch_size) / *settings.max_fps));
 161
 0162    if (batch_period <= Clock::duration::zero()) {
 0163      batch_period = Clock::duration{1};
 164    }
 165
 0166    if (!next_batch_start.has_value()) {
 0167      next_batch_start = Clock::now();
 168    }
 169
 0170    while (Clock::now() < *next_batch_start) {
 0171      if (ctx.cancelled != nullptr && ctx.cancelled->load(std::memory_order_relaxed)) {
 0172        return false;
 173      }
 174
 0175      std::this_thread::yield();
 0176    }
 177
 178    do {
 0179      *next_batch_start += batch_period;
 0180    } while (*next_batch_start < Clock::now());
 181
 0182    return true;
 0183  }
 184
 185  // -- ISyncTask interface ------------------------------------------------------------------------
 0186  std::optional<holoflow::core::TView> acquire_input(int index) override {
 187    (void)index; // Unused since there are no inputs
 0188    throw std::out_of_range("Holofile task has no inputs");
 0189  }
 190
 0191  void release_output(int index) override {
 0192    if (settings.load_kind == HolofileSettings::LoadKind::Live) {
 0193      throw std::logic_error("Cannot release output for Live load kind");
 194    }
 195
 0196    if (index != 0) {
 0197      throw std::out_of_range("Holofile task has only one output at index 0");
 198    }
 0199  }
 200
 0201  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 0202    if (!pace_before_batch(ctx)) {
 0203      return holoflow::core::OpResult::Cancelled;
 204    }
 205
 0206    size_t pixels_per_frame = header.frame_width * header.frame_height;
 0207    size_t bits_per_frame   = pixels_per_frame * header.bits_per_pixel;
 0208    size_t bytes_per_frame  = bits_per_frame / 8;
 209
 210    // Loop back to start when EOF would be bypassed when reading next batch
 0211    if (frame_idx + settings.batch_size > settings.end_frame) {
 0212      if (settings.load_kind == HolofileSettings::LoadKind::Live) {
 0213        reader->seek(settings.start_frame);
 214      }
 0215      frame_idx = settings.start_frame;
 216    }
 217
 218    // Read frames into buffer
 0219    if (settings.load_kind == HolofileSettings::LoadKind::Live) {
 0220      auto *odata = reinterpret_cast<uint8_t *>(ctx.outputs[0].data());
 0221      reader->read_frames(odata, settings.batch_size);
 0222    } else {
 0223      std::byte *data    = buf + frame_idx * bytes_per_frame;
 0224      auto      &storage = storage_access().owned_output_storage(0);
 0225      storage.ptr        = data;
 226
 0227      ctx.outputs[0] = holoflow::core::TView{
 228          .desc    = odesc,
 229          .storage = &storage,
 230      };
 231    }
 232
 0233    frame_idx += settings.batch_size;
 0234    return holoflow::core::OpResult::Ok;
 0235  }
 236};
 237
 238} // namespace
 239
 240// -------------------------------------------------------------------------------------------------
 241// HolofileFactory
 242// -------------------------------------------------------------------------------------------------
 243
 244holoflow::core::InferResult
 245HolofileFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 0246                       const nlohmann::json                  &jsettings) const {
 0247  static const std::map<size_t, holoflow::core::DType> bpp_to_dtype = {
 248      {8, holoflow::core::DType::U8},
 249      {16, holoflow::core::DType::U16},
 250  };
 251
 0252  auto settings = jsettings.get<HolofileSettings>();
 0253  auto reader   = std::make_unique<holofile::Reader>(settings.path);
 0254  auto header   = reader->header();
 255
 256  // clang-format off
 0257  check(input_descs.size() == 0, "Holofile task must have no inputs");
 0258  check(settings.start_frame < settings.end_frame, "Invalid frame range");
 0259  check(settings.batch_size > 0, "Batch size must be positive");
 0260  check(!settings.max_fps.has_value() || *settings.max_fps > 0, "max_fps must be positive when provided");
 0261  check(settings.end_frame <= static_cast<int>(header.frame_count), "end_frame exceeds total frames in file");
 0262  check(settings.batch_size <= (settings.end_frame - settings.start_frame), "Batch size exceeds available frames");
 0263  check(bpp_to_dtype.contains(header.bits_per_pixel), "Unsupported bits_per_pixel: " + std::to_string(header.bits_per_pi
 264  // clang-format on
 265
 0266  holoflow::core::TDesc odesc(
 267      {static_cast<size_t>(settings.batch_size), header.frame_height, header.frame_width},
 268      bpp_to_dtype.at(header.bits_per_pixel),
 269      settings.load_kind == HolofileSettings::LoadKind::GPUCached ? holoflow::core::MemLoc::Device
 270                                                                  : holoflow::core::MemLoc::Host);
 271
 0272  bool owned_output = settings.load_kind != HolofileSettings::LoadKind::Live;
 273
 0274  return holoflow::core::InferResult{
 275      .input_descs   = {},
 276      .output_descs  = {odesc},
 277      .in_place      = {},
 278      .owned_inputs  = {},
 279      .owned_outputs = {owned_output},
 280      .kind          = holoflow::core::TaskKind::Sync,
 281  };
 0282}
 283
 284std::unique_ptr<holoflow::core::ISyncTask>
 285HolofileFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 286                        const nlohmann::json                  &jsettings,
 0287                        const holoflow::core::SyncCreateCtx   &ctx) const {
 0288  auto infer    = this->infer(input_descs, jsettings);
 0289  auto settings = jsettings.get<HolofileSettings>();
 290
 291  // Setup reader
 0292  auto reader = std::make_unique<holofile::Reader>(settings.path);
 0293  auto header = reader->header();
 0294  reader->seek(settings.start_frame);
 295
 0296  size_t pixels_per_frame = header.frame_width * header.frame_height;
 0297  size_t bits_per_frame   = pixels_per_frame * header.bits_per_pixel;
 0298  size_t bytes_per_frame  = bits_per_frame / 8;
 0299  size_t frames_to_load   = settings.end_frame - settings.start_frame;
 0300  size_t bytes_to_load    = frames_to_load * bytes_per_frame;
 301
 302  // Setup buffers
 303  using curaii::make_unique_device_ptr;
 304  using curaii::make_unique_host_ptr;
 0305  std::byte         *buf   = nullptr;
 0306  HostPtr<std::byte> h_buf = nullptr;
 0307  DevPtr<std::byte>  d_buf = nullptr;
 308
 0309  switch (settings.load_kind) {
 310  case HolofileSettings::LoadKind::Live:
 311    // No preloading
 312    break;
 313
 314  case HolofileSettings::LoadKind::CPUCached:
 0315    h_buf = make_unique_host_ptr<std::byte>(bytes_to_load);
 0316    buf   = h_buf.get();
 0317    break;
 318
 319  case HolofileSettings::LoadKind::GPUCached:
 0320    d_buf = make_unique_device_ptr<std::byte>(bytes_to_load);
 0321    buf   = d_buf.get();
 322    break;
 323  }
 324
 325  // Preload if needed
 0326  switch (settings.load_kind) {
 327  case HolofileSettings::LoadKind::Live:
 328    // No preloading
 329    break;
 330
 331  case HolofileSettings::LoadKind::CPUCached:
 0332    reader->read_frames(reinterpret_cast<uint8_t *>(h_buf.get()), frames_to_load);
 0333    break;
 334
 335  case HolofileSettings::LoadKind::GPUCached: {
 0336    auto temp_buf = make_unique_host_ptr<std::byte>(bytes_to_load);
 0337    reader->read_frames(reinterpret_cast<uint8_t *>(temp_buf.get()), frames_to_load);
 0338    CUDA_CHECK(cudaMemcpyAsync(d_buf.get(), temp_buf.get(), bytes_to_load, cudaMemcpyHostToDevice,
 339                               ctx.stream));
 340    // Stream sync removed here!
 0341  } break;
 342  }
 343
 344  // Construct task directly
 0345  auto task       = std::make_unique<Holofile>();
 0346  task->settings  = settings;
 0347  task->reader    = std::move(reader);
 0348  task->header    = header;
 0349  task->frame_idx = settings.start_frame;
 0350  task->odesc     = infer.output_descs[0];
 0351  task->buf       = buf;
 0352  task->h_buf     = std::move(h_buf);
 0353  task->d_buf     = std::move(d_buf);
 0354  task->stream    = ctx.stream;
 0355  task->next_batch_start.reset();
 356
 0357  return task;
 0358}
 359
 360std::unique_ptr<holoflow::core::ISyncTask>
 361HolofileFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 362                        std::span<const holoflow::core::TDesc>     input_descs,
 363                        const nlohmann::json                      &jsettings,
 0364                        const holoflow::core::SyncCreateCtx       &ctx) const {
 0365  auto *old_holofile = dynamic_cast<Holofile *>(old_task.get());
 0366  if (old_holofile == nullptr) {
 0367    return create(input_descs, jsettings, ctx);
 368  }
 369
 0370  auto infer    = this->infer(input_descs, jsettings);
 0371  auto settings = jsettings.get<HolofileSettings>();
 372
 0373  bool is_live     = settings.load_kind == HolofileSettings::LoadKind::Live;
 0374  bool path_same   = (old_holofile->settings.path == settings.path);
 0375  bool kind_same   = (old_holofile->settings.load_kind == settings.load_kind);
 0376  bool start_same  = (old_holofile->settings.start_frame == settings.start_frame);
 0377  bool end_same    = (old_holofile->settings.end_frame == settings.end_frame);
 0378  bool batch_same  = (old_holofile->settings.batch_size == settings.batch_size);
 0379  bool fps_same    = (old_holofile->settings.max_fps == settings.max_fps);
 0380  bool pace_same   = batch_same && fps_same;
 0381  bool bounds_same = start_same && end_same;
 0382  bool can_reuse   = path_same && kind_same && (is_live || bounds_same);
 383
 0384  if (!can_reuse) {
 0385    logger()->debug(
 386        "[HolofileFactory::update] Cannot reuse existing task (path_same={}, kind_same={}, "
 387        "bounds_same={})",
 388        path_same, kind_same, bounds_same);
 0389    return create(input_descs, jsettings, ctx);
 390  }
 391
 0392  logger()->debug("[HolofileFactory::update] Reusing existing Holofile task");
 393
 394  // Transfer ownership of heavy resources
 0395  auto reader = std::move(old_holofile->reader);
 0396  auto header = reader->header();
 0397  auto buf    = old_holofile->buf;
 0398  auto h_buf  = std::move(old_holofile->h_buf);
 0399  auto d_buf  = std::move(old_holofile->d_buf);
 400
 401  // Resolve cursor index
 0402  int frame_idx = settings.start_frame;
 0403  if (settings.keep_cursor) {
 0404    frame_idx = old_holofile->frame_idx;
 405    // Clamp just in case the bounds were shrunk past the current cursor
 0406    if (frame_idx < settings.start_frame || frame_idx >= settings.end_frame) {
 0407      frame_idx = settings.start_frame;
 408    }
 409  }
 410
 0411  if (is_live && frame_idx != old_holofile->frame_idx) {
 0412    logger()->debug(
 413        "[HolofileFactory::update] Seeking reader from frame {} to {} due to cursor change",
 414        old_holofile->frame_idx, frame_idx);
 0415    reader->seek(frame_idx);
 416  }
 417
 418  // Construct task directly
 0419  auto task       = std::make_unique<Holofile>();
 0420  task->settings  = settings;
 0421  task->reader    = std::move(reader);
 0422  task->header    = header;
 0423  task->frame_idx = frame_idx;
 0424  task->odesc     = infer.output_descs[0];
 0425  task->buf       = buf;
 0426  task->h_buf     = std::move(h_buf);
 0427  task->d_buf     = std::move(d_buf);
 0428  task->stream    = ctx.stream;
 0429  task->next_batch_start = pace_same ? old_holofile->next_batch_start : std::nullopt;
 430
 0431  return task;
 0432}
 433
 434} // namespace holotask::sources

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\syncs\cuda_stream_synchronize.cc

#LineLine coverage
 1// Copyright 2026 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 "holotask/syncs/cuda_stream_synchronize.hh"
 16
 17#include <stdexcept>
 18#include <string>
 19
 20#include "bug.hh"
 21#include "logger.hh"
 22
 23namespace holotask::syncs {
 24
 25// -------------------------------------------------------------------------------------------------
 26// JSON serialization
 27// -------------------------------------------------------------------------------------------------
 28
 029void to_json(nlohmann::json &j, const CudaStreamSynchronizeSettings &) { j = nlohmann::json{}; }
 30
 031void from_json(const nlohmann::json &, CudaStreamSynchronizeSettings &) {}
 32
 33namespace {
 34
 035void check(bool condition, const std::string &msg) {
 036  if (!condition) {
 037    logger()->error("[CudaStreamSynchronizeFactory::infer] error: {}", msg);
 038    throw std::invalid_argument("CudaStreamSynchronizeFactory inference error: " + msg);
 39  }
 040}
 41
 42// -------------------------------------------------------------------------------------------------
 43// CudaStreamSynchronize task implementation
 44// -------------------------------------------------------------------------------------------------
 45
 46class CudaStreamSynchronize : public holoflow::core::ISyncTask {
 47public:
 048  explicit CudaStreamSynchronize(cudaStream_t stream) : stream_(stream) {}
 49
 050  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 51    (void)ctx;
 052    CUDA_CHECK(cudaStreamSynchronize(stream_));
 053    return holoflow::core::OpResult::Ok;
 054  };
 55
 056  void         update_stream(cudaStream_t stream) { stream_ = stream; }
 57  cudaStream_t stream() const { return stream_; }
 58
 59private:
 60  cudaStream_t stream_;
 61};
 62
 63} // namespace
 64
 65// -------------------------------------------------------------------------------------------------
 66// CudaStreamSynchronizeFactory
 67// -------------------------------------------------------------------------------------------------
 68
 69holoflow::core::InferResult
 70CudaStreamSynchronizeFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 071                                    const nlohmann::json                  &jsettings) const {
 072  (void)jsettings.get<CudaStreamSynchronizeSettings>();
 073  check(input_descs.size() == 1, "CudaStreamSynchronize task must have exactly one input");
 74
 075  return holoflow::core::InferResult{
 76      .input_descs   = {input_descs[0]},
 77      .output_descs  = {input_descs[0]},
 78      .in_place      = {{0, 0}},
 79      .owned_inputs  = {false},
 80      .owned_outputs = {false},
 81      .kind          = holoflow::core::TaskKind::Sync,
 82  };
 083}
 84
 85std::unique_ptr<holoflow::core::ISyncTask>
 86CudaStreamSynchronizeFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 87                                     const nlohmann::json                  &jsettings,
 088                                     const holoflow::core::SyncCreateCtx   &ctx) const {
 089  (void)this->infer(input_descs, jsettings);
 090  return std::make_unique<CudaStreamSynchronize>(ctx.stream);
 091}
 92
 93std::unique_ptr<holoflow::core::ISyncTask>
 94CudaStreamSynchronizeFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 95                                     std::span<const holoflow::core::TDesc>     input_descs,
 96                                     const nlohmann::json                      &jsettings,
 097                                     const holoflow::core::SyncCreateCtx       &ctx) const {
 098  (void)this->infer(input_descs, jsettings);
 99
 0100  auto *old_sync = dynamic_cast<CudaStreamSynchronize *>(old_task.get());
 0101  if (old_sync == nullptr) {
 0102    return create(input_descs, jsettings, ctx);
 103  }
 104
 0105  old_sync->update_stream(ctx.stream);
 0106  return old_task;
 0107}
 108
 109} // namespace holotask::syncs

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\syncs\memcpy.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 "holotask/syncs/memcpy.hh"
 16
 17#include <cstddef>
 18#include <cstdint>
 19#include <cstring>
 20#include <map>
 21#include <omp.h>
 22#include <stdexcept>
 23#include <string>
 24#include <utility>
 25
 26#include "bug.hh"
 27#include "logger.hh"
 28
 29namespace holotask::syncs {
 30
 31// -------------------------------------------------------------------------------------------------
 32// JSON serialization
 33// -------------------------------------------------------------------------------------------------
 34
 035void to_json(nlohmann::json &j, const MemcpySettings::Target &t) {
 036  static const std::map<MemcpySettings::Target, std::string> t_to_str = {
 37      {MemcpySettings::Target::Host, "Host"},
 38      {MemcpySettings::Target::Device, "Device"},
 39  };
 40
 041  HOLOVIBES_CHECK(t_to_str.contains(t), "Invalid Target enum value");
 042  j = t_to_str.at(t);
 043}
 44
 045void from_json(const nlohmann::json &j, MemcpySettings::Target &t) {
 046  static const std::map<std::string, MemcpySettings::Target> str_to_t = {
 47      {"Host", MemcpySettings::Target::Host},
 48      {"Device", MemcpySettings::Target::Device},
 49  };
 50
 051  auto key = j.get<std::string>();
 052  if (!str_to_t.contains(key)) {
 053    throw std::invalid_argument("Invalid Target string: " + key);
 54  }
 055  t = str_to_t.at(key);
 056}
 57
 058void to_json(nlohmann::json &j, const MemcpySettings &ms) {
 059  j = nlohmann::json{
 60      {"target", ms.target},
 61  };
 062}
 63
 064void from_json(const nlohmann::json &j, MemcpySettings &ms) { j.at("target").get_to(ms.target); }
 65
 66namespace {
 67
 068void check(bool condition, const std::string &msg) {
 069  if (!condition) {
 070    logger()->error("[MemcpyFactory::infer] error: {}", msg);
 071    throw std::invalid_argument("MemcpyFactory inference error: " + msg);
 72  }
 073}
 74
 075void mt_memcpy(void *dst, const void *src, const std::size_t n) {
 076  constexpr int NUM_THREADS = 2;
 077  auto         *dst_bytes   = static_cast<std::uint8_t *>(dst);
 078  const auto   *src_bytes   = static_cast<const std::uint8_t *>(src);
 79
 080  const std::size_t chunk_size = n / NUM_THREADS;
 081  const std::size_t remainder  = n % NUM_THREADS;
 82
 083#pragma omp parallel num_threads(NUM_THREADS)
 84  {
 085    const int         tid       = omp_get_thread_num();
 086    const std::size_t offset    = tid * chunk_size;
 087    std::size_t       this_size = chunk_size;
 88
 089    if (tid == NUM_THREADS - 1) {
 090      this_size += remainder;
 91    }
 92
 093    if (this_size > 0) {
 094      std::memcpy(dst_bytes + offset, src_bytes + offset, this_size);
 95    }
 096  }
 097}
 98
 99} // namespace
 100
 101// -------------------------------------------------------------------------------------------------
 102// Memcpy task implementation
 103// -------------------------------------------------------------------------------------------------
 104
 105class Memcpy : public holoflow::core::ISyncTask {
 106public:
 0107  explicit Memcpy(MemcpySettings settings, cudaStream_t stream)
 0108      : settings_(std::move(settings)), stream_(stream) {}
 109
 0110  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 0111    auto *src       = ctx.inputs[0].data();
 0112    auto *dst       = ctx.outputs[0].data();
 0113    auto  n         = ctx.outputs[0].desc.num_bytes();
 0114    auto  copy_desc = std::make_pair(ctx.inputs[0].desc.mem_loc, settings_.target);
 115    using CopyDesc  = std::pair<holoflow::core::MemLoc, MemcpySettings::Target>;
 0116    static const std::map<CopyDesc, cudaMemcpyKind> copy_map = {
 117        {{holoflow::core::MemLoc::Host, MemcpySettings::Target::Host}, cudaMemcpyHostToHost},
 118        {{holoflow::core::MemLoc::Host, MemcpySettings::Target::Device}, cudaMemcpyHostToDevice},
 119        {{holoflow::core::MemLoc::Device, MemcpySettings::Target::Host}, cudaMemcpyDeviceToHost},
 120        {{holoflow::core::MemLoc::Device, MemcpySettings::Target::Device},
 121         cudaMemcpyDeviceToDevice},
 122    };
 123
 0124    HOLOVIBES_CHECK(copy_map.contains(copy_desc), "Invalid memory copy descriptor");
 0125    const auto kind = copy_map.at(copy_desc);
 0126    if (kind == cudaMemcpyHostToHost) {
 0127      mt_memcpy(dst, src, n);
 0128    } else {
 0129      CUDA_CHECK(cudaMemcpyAsync(dst, src, n, kind, stream_));
 130    }
 131
 0132    return holoflow::core::OpResult::Ok;
 0133  }
 134
 0135  void                  update_stream(cudaStream_t stream) { stream_ = stream; }
 0136  const MemcpySettings &settings() const { return settings_; }
 137
 138private:
 139  MemcpySettings settings_;
 140  cudaStream_t   stream_;
 141};
 142
 143// -------------------------------------------------------------------------------------------------
 144// MemcpyFactory
 145// -------------------------------------------------------------------------------------------------
 146
 147holoflow::core::InferResult MemcpyFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 0148                                                 const nlohmann::json &jsettings) const {
 0149  const auto settings = jsettings.get<MemcpySettings>();
 0150  check(input_descs.size() == 1, "Memcpy task must have exactly one input");
 151
 0152  holoflow::core::TDesc out_desc = input_descs[0];
 0153  out_desc.mem_loc               = settings.target == MemcpySettings::Target::Device
 154                                       ? holoflow::core::MemLoc::Device
 155                                       : holoflow::core::MemLoc::Host;
 156
 0157  return holoflow::core::InferResult{
 158      .input_descs   = {input_descs[0]},
 159      .output_descs  = {out_desc},
 160      .in_place      = {},
 161      .owned_inputs  = {false},
 162      .owned_outputs = {false},
 163      .kind          = holoflow::core::TaskKind::Sync,
 164  };
 0165}
 166
 167std::unique_ptr<holoflow::core::ISyncTask>
 168MemcpyFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 169                      const nlohmann::json                  &jsettings,
 0170                      const holoflow::core::SyncCreateCtx   &ctx) const {
 0171  (void)this->infer(input_descs, jsettings);
 0172  const auto settings = jsettings.get<MemcpySettings>();
 173
 0174  return std::make_unique<Memcpy>(settings, ctx.stream);
 0175}
 176
 177std::unique_ptr<holoflow::core::ISyncTask>
 178MemcpyFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 179                      std::span<const holoflow::core::TDesc>     input_descs,
 180                      const nlohmann::json                      &jsettings,
 0181                      const holoflow::core::SyncCreateCtx       &ctx) const {
 0182  (void)this->infer(input_descs, jsettings);
 183
 0184  auto *old_memcpy = dynamic_cast<Memcpy *>(old_task.get());
 0185  if (old_memcpy == nullptr) {
 0186    return create(input_descs, jsettings, ctx);
 187  }
 188
 0189  const auto settings = jsettings.get<MemcpySettings>();
 0190  if (settings == old_memcpy->settings()) {
 0191    old_memcpy->update_stream(ctx.stream);
 0192    return old_task;
 193  }
 194
 0195  return create(input_descs, jsettings, ctx);
 0196}
 197
 0198} // namespace holotask::syncs

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\syncs\pca.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 "holotask/syncs/pca.hh"
 16
 17#include <cuda.h>
 18#include <nvJitLink.h>
 19#include <nvtx3/nvtx3.hpp>
 20#include <windows.h>
 21
 22#include <algorithm>
 23#include <array>
 24#include <cstddef>
 25#include <cstdint>
 26#include <cstdlib>
 27#include <filesystem>
 28#include <format>
 29#include <limits>
 30#include <list>
 31#include <memory>
 32#include <stdexcept>
 33#include <string>
 34#include <utility>
 35#include <vector>
 36
 37#include "curaii/cublas.hh"
 38#include "curaii/cuda.hh"
 39#include "curaii/cusolver.hh"
 40#include "curaii/nvrtc.hh"
 41
 42#include "logger.hh"
 43
 44namespace holotask::syncs {
 45
 46namespace {
 47
 48// -------------------------------------------------------------------------------------------------
 49// PCA data model
 50// -------------------------------------------------------------------------------------------------
 51
 52// PCA operates on tensors shaped [..., features, height, width]. All leading dimensions are
 53// flattened into independent batches.
 54struct PcaLayout {
 55  size_t feature_axis;
 56  int    features;
 57  int    height;
 58  int    width;
 59  int    samples;
 60  size_t batches;
 61
 62  // Strides in elements between consecutive flattened batches.
 63  long long input_batch_stride;
 64  long long matrix_batch_stride;
 65
 166  explicit PcaLayout(const holoflow::core::TDesc &desc) {
 167    const auto rank = desc.rank();
 68
 169    feature_axis = rank - 3;
 170    features     = static_cast<int>(desc.shape.at(feature_axis));
 171    height       = static_cast<int>(desc.shape.at(rank - 2));
 172    width        = static_cast<int>(desc.shape.at(rank - 1));
 173    samples      = height * width;
 74
 175    batches = 1;
 176    for (size_t axis = 0; axis < feature_axis; ++axis) {
 177      batches *= desc.shape.at(axis);
 178    }
 79
 180    input_batch_stride  = static_cast<long long>(features) * samples;
 181    matrix_batch_stride = static_cast<long long>(features) * features;
 182  }
 83
 184  [[nodiscard]] long long output_batch_stride(int components) const {
 185    return static_cast<long long>(samples) * components;
 186  }
 87};
 88
 89struct PcaWorkspace {
 90  // The matrix buffer contains covariance matrices before HEEV and eigenvectors afterwards.
 91  curaii::unique_device_ptr<float> matrices;
 92  curaii::unique_device_ptr<float> eigenvalues;
 93  curaii::unique_device_ptr<int>   solver_info;
 94
 95  explicit PcaWorkspace(const PcaLayout &layout)
 196      : matrices(curaii::make_unique_device_ptr<float>(
 97            layout.batches * static_cast<size_t>(layout.features) * layout.features)),
 198        eigenvalues(curaii::make_unique_device_ptr<float>(layout.batches * layout.features)),
 199        solver_info(curaii::make_unique_device_ptr<int>(layout.batches)) {}
 100};
 101
 102// -------------------------------------------------------------------------------------------------
 103// CUDA Graph replay
 104// -------------------------------------------------------------------------------------------------
 105
 106using PcaGraphKey = std::array<const void *, 2>;
 107
 108template <typename Enqueue> cudaGraph_t capture_cuda_graph(cudaStream_t stream, Enqueue &&enqueue) {
 109  cudaGraph_t graph     = nullptr;
 110  bool        capturing = false;
 111
 112  try {
 113    CUDA_CHECK(cudaStreamBeginCapture(stream, cudaStreamCaptureModeThreadLocal));
 114    capturing = true;
 115
 116    std::forward<Enqueue>(enqueue)();
 117
 118    CUDA_CHECK(cudaStreamEndCapture(stream, &graph));
 119    capturing = false;
 120    return graph;
 121  } catch (...) {
 122    if (capturing) {
 123      cudaGraph_t discarded = nullptr;
 124      (void)cudaStreamEndCapture(stream, &discarded);
 125      if (discarded != nullptr) {
 126        CUDA_CHECK_NT(cudaGraphDestroy(discarded));
 127      }
 128    } else if (graph != nullptr) {
 129      CUDA_CHECK_NT(cudaGraphDestroy(graph));
 130    }
 131    throw;
 132  }
 133}
 134
 1135[[nodiscard]] bool stream_is_capturing(cudaStream_t stream) {
 1136  cudaStreamCaptureStatus status = cudaStreamCaptureStatusNone;
 1137  CUDA_CHECK(cudaStreamIsCapturing(stream, &status));
 1138  return status != cudaStreamCaptureStatusNone;
 1139}
 140
 141class PcaCudaGraph {
 142public:
 1143  explicit PcaCudaGraph(PcaGraphKey key) : key_(key) {}
 144
 1145  ~PcaCudaGraph() noexcept {
 1146    if (executable_ != nullptr) {
 1147      CUDA_CHECK_NT(cudaGraphExecDestroy(executable_));
 148    }
 1149  }
 150
 151  PcaCudaGraph(const PcaCudaGraph &)            = delete;
 152  PcaCudaGraph &operator=(const PcaCudaGraph &) = delete;
 153
 1154  [[nodiscard]] bool matches(const PcaGraphKey &key) const noexcept {
 1155    return executable_ != nullptr && key_ == key;
 1156  }
 157
 1158  void launch(cudaStream_t stream) const { CUDA_CHECK(cudaGraphLaunch(executable_, stream)); }
 159
 160  template <typename Enqueue> void capture(cudaStream_t stream, Enqueue &&enqueue) {
 161    cudaGraph_t graph = capture_cuda_graph(stream, std::forward<Enqueue>(enqueue));
 162
 163    try {
 164      CUDA_CHECK(cudaGraphInstantiateWithFlags(&executable_, graph, 0));
 165      CUDA_CHECK_NT(cudaGraphDestroy(graph));
 166    } catch (...) {
 167      CUDA_CHECK_NT(cudaGraphDestroy(graph));
 168      throw;
 169    }
 170  }
 171
 172private:
 173  PcaGraphKey     key_{};
 1174  cudaGraphExec_t executable_{nullptr};
 175};
 176
 177class PcaGraphCache {
 178public:
 1179  void invalidate() {
 1180    graphs_.clear();
 1181    capture_enabled_ = true;
 1182  }
 183
 0184  void enable_capture() noexcept { capture_enabled_ = true; }
 185
 1186  [[nodiscard]] bool try_launch(const PcaGraphKey &key, cudaStream_t stream) {
 1187    const auto graph = std::find_if(graphs_.begin(), graphs_.end(),
 188                                    [&](const PcaCudaGraph &entry) { return entry.matches(key); });
 189
 1190    if (graph == graphs_.end()) {
 1191      return false;
 192    }
 193
 1194    graph->launch(stream);
 1195    graphs_.splice(graphs_.begin(), graphs_, graph);
 1196    return true;
 1197  }
 198
 199  template <typename Enqueue>
 200  void try_capture(const PcaGraphKey &key, cudaStream_t stream, Enqueue &&enqueue) {
 201    if (!capture_enabled_) {
 202      return;
 203    }
 204
 205    bool entry_inserted = false;
 206    try {
 207      graphs_.emplace_front(key);
 208      entry_inserted = true;
 209
 210      graphs_.front().capture(stream, std::forward<Enqueue>(enqueue));
 211
 212      if (graphs_.size() > capacity) {
 213        graphs_.pop_back();
 214      }
 215    } catch (const std::exception &error) {
 216      if (entry_inserted) {
 217        graphs_.pop_front();
 218      }
 219      capture_enabled_ = false;
 220      logger()->warn("[Pca] CUDA Graph capture disabled: {}", error.what());
 221    }
 222  }
 223
 224private:
 225  // cuBLAS and cuSOLVER graph nodes embed their buffer addresses. Cache the finite set of rotating
 226  // pipeline buffers instead of attempting to update opaque library nodes.
 227  static constexpr size_t capacity = 128;
 228
 229  bool                    capture_enabled_{true};
 230  std::list<PcaCudaGraph> graphs_;
 231};
 232
 233// -------------------------------------------------------------------------------------------------
 234// Eigensolver abstraction
 235// -------------------------------------------------------------------------------------------------
 236
 237class Eigensolver {
 238public:
 1239  virtual ~Eigensolver() = default;
 240
 241  Eigensolver(const Eigensolver &)            = delete;
 242  Eigensolver &operator=(const Eigensolver &) = delete;
 243
 244  [[nodiscard]] virtual bool is_compatible_stream(cudaStream_t stream) const = 0;
 245
 246  virtual void solve(float *matrices, float *eigenvalues, int *info, cudaStream_t stream) = 0;
 247
 248protected:
 1249  Eigensolver() = default;
 250};
 251
 252std::unique_ptr<Eigensolver> make_eigensolver(const PcaLayout    &layout,
 253                                              const PcaWorkspace &workspace, cudaStream_t stream);
 254
 255// -------------------------------------------------------------------------------------------------
 256// PCA task
 257// -------------------------------------------------------------------------------------------------
 258
 259class PcaTask final : public holoflow::core::ISyncTask {
 260public:
 261  PcaTask(const PcaSettings &settings, const holoflow::core::TDesc &input_desc,
 262          const holoflow::core::SyncCreateCtx &ctx)
 1263      : settings_(settings), input_desc_(input_desc), layout_(input_desc), stream_(ctx.stream),
 1264        workspace_(layout_) {
 1265    CUBLAS_CHECK(cublasSetStream(cublas_.get(), stream_));
 1266    eigensolver_ = make_eigensolver(layout_, workspace_, stream_);
 1267  }
 268
 1269  [[nodiscard]] bool can_reuse(const holoflow::core::TDesc &input_desc, cudaStream_t stream) const {
 1270    return input_desc.shape == input_desc_.shape && input_desc.strides == input_desc_.strides &&
 271           input_desc.dtype == input_desc_.dtype && input_desc.mem_loc == input_desc_.mem_loc &&
 272           eigensolver_->is_compatible_stream(stream);
 1273  }
 274
 1275  void reconfigure(const PcaSettings &settings, cudaStream_t stream) {
 1276    if (settings_ != settings) {
 1277      settings_ = settings;
 1278      graph_cache_.invalidate();
 279    }
 280
 1281    if (stream_ != stream) {
 0282      stream_ = stream;
 0283      CUBLAS_CHECK(cublasSetStream(cublas_.get(), stream_));
 284
 285      // Graph executables are not tied to the stream on which they were captured. Reusing them is
 286      // valid here because can_reuse() already guarantees that the CUDA context is unchanged.
 0287      graph_cache_.enable_capture();
 288    }
 1289  }
 290
 1291  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 1292    nvtx3::scoped_range range("PCA Sync Task");
 293
 1294    if (stream_ == nullptr || stream_is_capturing(stream_)) {
 1295      return enqueue(ctx);
 296    }
 297
 1298    const PcaGraphKey key{ctx.inputs[0].data(), ctx.outputs[0].data()};
 1299    if (graph_cache_.try_launch(key, stream_)) {
 1300      return holoflow::core::OpResult::Ok;
 301    }
 302
 1303    const auto result = enqueue(ctx);
 1304    if (result == holoflow::core::OpResult::Ok) {
 1305      graph_cache_.try_capture(key, stream_, [&]() { (void)enqueue(ctx); });
 306    }
 1307    return result;
 1308  }
 309
 310private:
 1311  holoflow::core::OpResult enqueue(holoflow::core::SyncCtx &ctx) {
 1312    auto *input  = reinterpret_cast<float *>(ctx.inputs[0].data());
 1313    auto *output = reinterpret_cast<float *>(ctx.outputs[0].data());
 314
 1315    enqueue_covariance(input);
 1316    enqueue_eigendecomposition();
 1317    enqueue_projection(input, output);
 318
 1319    return holoflow::core::OpResult::Ok;
 1320  }
 321
 1322  void enqueue_covariance(const float *input) {
 1323    nvtx3::scoped_range range("PCA covariance");
 324
 1325    constexpr float alpha = 1.0f;
 1326    constexpr float beta  = 0.0f;
 327
 328    // Independent GEMMs preserve cuBLAS Split-K selection for the large sample dimension.
 1329    for (size_t batch = 0; batch < layout_.batches; ++batch) {
 1330      const auto input_offset  = static_cast<long long>(batch) * layout_.input_batch_stride;
 1331      const auto matrix_offset = static_cast<long long>(batch) * layout_.matrix_batch_stride;
 332
 1333      CUBLAS_CHECK(cublasGemmEx(cublas_.get(), CUBLAS_OP_T, CUBLAS_OP_N, layout_.features,
 334                                layout_.features, layout_.samples, &alpha, input + input_offset,
 335                                CUDA_R_32F, layout_.samples, input + input_offset, CUDA_R_32F,
 336                                layout_.samples, &beta, workspace_.matrices.get() + matrix_offset,
 337                                CUDA_R_32F, layout_.features, CUBLAS_COMPUTE_32F_FAST_16F,
 338                                CUBLAS_GEMM_DEFAULT));
 1339    }
 1340  }
 341
 1342  void enqueue_eigendecomposition() {
 1343    nvtx3::scoped_range range("PCA eigendecomposition");
 344
 1345    eigensolver_->solve(workspace_.matrices.get(), workspace_.eigenvalues.get(),
 346                        workspace_.solver_info.get(), stream_);
 1347  }
 348
 1349  void enqueue_projection(const float *input, float *output) {
 1350    nvtx3::scoped_range range("PCA projection");
 351
 1352    constexpr float alpha = 1.0f;
 1353    constexpr float beta  = 0.0f;
 354
 1355    const int components = settings_.components();
 356
 1357    const auto eigenvector_offset = static_cast<long long>(settings_.begin) * layout_.features;
 1358    const auto output_stride      = layout_.output_batch_stride(components);
 359
 1360    for (size_t batch = 0; batch < layout_.batches; ++batch) {
 1361      const auto input_offset  = static_cast<long long>(batch) * layout_.input_batch_stride;
 1362      const auto matrix_offset = static_cast<long long>(batch) * layout_.matrix_batch_stride;
 1363      const auto output_offset = static_cast<long long>(batch) * output_stride;
 364
 1365      CUBLAS_CHECK(
 366          cublasGemmEx(cublas_.get(), CUBLAS_OP_N, CUBLAS_OP_N, layout_.samples, components,
 367                       layout_.features, &alpha, input + input_offset, CUDA_R_32F, layout_.samples,
 368                       workspace_.matrices.get() + matrix_offset + eigenvector_offset, CUDA_R_32F,
 369                       layout_.features, &beta, output + output_offset, CUDA_R_32F, layout_.samples,
 370                       CUBLAS_COMPUTE_32F_FAST_16F, CUBLAS_GEMM_DEFAULT));
 1371    }
 1372  }
 373
 374  PcaSettings           settings_;
 375  holoflow::core::TDesc input_desc_;
 376  PcaLayout             layout_;
 377  cudaStream_t          stream_;
 378
 379  curaii::CublasHandle         cublas_;
 380  PcaWorkspace                 workspace_;
 381  std::unique_ptr<Eigensolver> eigensolver_;
 382  PcaGraphCache                graph_cache_;
 383};
 384
 385// -------------------------------------------------------------------------------------------------
 386// CUDA Driver API support
 387// -------------------------------------------------------------------------------------------------
 388
 389std::string driver_error_message(CUresult result, const char *expression, const char *file,
 0390                                 int line) {
 0391  const char *name    = nullptr;
 0392  const char *message = nullptr;
 0393  (void)cuGetErrorName(result, &name);
 0394  (void)cuGetErrorString(result, &message);
 395
 0396  return std::format("CUDA Driver error: {} ({})\n  expression : {}\n  location   : {}:{}",
 397                     message != nullptr ? message : "unknown",
 398                     name != nullptr ? name : std::to_string(static_cast<int>(result)), expression,
 399                     file, line);
 0400}
 401
 1402void driver_check(CUresult result, const char *expression, const char *file, int line) {
 1403  if (result == CUDA_SUCCESS) {
 1404    return;
 405  }
 406
 0407  const auto message = driver_error_message(result, expression, file, line);
 0408  logger()->error("{}", message);
 0409  throw std::runtime_error(message);
 1410}
 411
 412#define PCA_DRIVER_CHECK(expr) driver_check((expr), #expr, __FILE__, __LINE__)
 413
 414class ScopedCudaContext {
 415public:
 1416  explicit ScopedCudaContext(CUcontext context) {
 1417    CUcontext current = nullptr;
 1418    PCA_DRIVER_CHECK(cuCtxGetCurrent(&current));
 419
 1420    if (current != context) {
 0421      PCA_DRIVER_CHECK(cuCtxPushCurrent(context));
 0422      pushed_ = true;
 423    }
 1424  }
 425
 1426  ~ScopedCudaContext() noexcept {
 1427    if (!pushed_) {
 1428      return;
 429    }
 430
 0431    CUcontext  popped = nullptr;
 0432    const auto result = cuCtxPopCurrent(&popped);
 0433    if (result != CUDA_SUCCESS) {
 0434      logger()->critical("{}", driver_error_message(result, "cuCtxPopCurrent", __FILE__, __LINE__));
 0435      std::abort();
 436    }
 1437  }
 438
 439  ScopedCudaContext(const ScopedCudaContext &)            = delete;
 440  ScopedCudaContext &operator=(const ScopedCudaContext &) = delete;
 441
 442private:
 1443  bool pushed_{false};
 444};
 445
 1446CUcontext cuda_context_for_stream(cudaStream_t stream) {
 1447  CUcontext context = nullptr;
 448
 1449  if (stream != nullptr) {
 1450    PCA_DRIVER_CHECK(cuStreamGetCtx(reinterpret_cast<CUstream>(stream), &context));
 1451  } else {
 1452    PCA_DRIVER_CHECK(cuCtxGetCurrent(&context));
 453  }
 454
 1455  if (context == nullptr) {
 0456    throw std::runtime_error("PCA eigensolver requires an active CUDA context");
 457  }
 458
 1459  return context;
 1460}
 461
 1462template <typename T> T read_module_constant(CUmodule module, const char *name) {
 1463  CUdeviceptr address = 0;
 1464  size_t      size    = 0;
 465
 1466  PCA_DRIVER_CHECK(cuModuleGetGlobal(&address, &size, module, name));
 1467  if (size != sizeof(T)) {
 0468    throw std::runtime_error(
 469        std::format("Unexpected size for cuSolverDx module constant {}", name));
 470  }
 471
 1472  T value{};
 1473  PCA_DRIVER_CHECK(cuMemcpyDtoH(&value, address, sizeof(T)));
 1474  return value;
 1475}
 476
 1477CUfunction module_function(CUmodule module, const char *name) {
 1478  CUfunction function = nullptr;
 1479  PCA_DRIVER_CHECK(cuModuleGetFunction(&function, module, name));
 1480  return function;
 1481}
 482
 483// -------------------------------------------------------------------------------------------------
 484// Conventional cuSOLVER eigensolver
 485// -------------------------------------------------------------------------------------------------
 486
 487class CusolverEigensolver final : public Eigensolver {
 488public:
 489  CusolverEigensolver(int features, size_t batches, float *matrices, float *eigenvalues,
 490                      cudaStream_t stream)
 1491      : features_(features), batches_(batches), context_(cuda_context_for_stream(stream)),
 1492        handle_(std::make_unique<curaii::CusolverDnHandle>()),
 1493        params_(std::make_unique<curaii::CusolverDnParams>()) {
 1494    if (batches_ > static_cast<size_t>((std::numeric_limits<int64_t>::max)())) {
 0495      throw std::invalid_argument("[Pca] Batch count exceeds the cuSOLVER API limit");
 496    }
 497
 1498    CUSOLVER_CHECK(cusolverDnSetStream(handle_->get(), stream));
 1499    CUSOLVER_CHECK(cusolverDnXsyevBatched_bufferSize(
 500        handle_->get(), params_->get(), CUSOLVER_EIG_MODE_VECTOR, CUBLAS_FILL_MODE_LOWER, features_,
 501        CUDA_R_32F, matrices, features_, CUDA_R_32F, eigenvalues, CUDA_R_32F,
 502        &device_workspace_size_, &host_workspace_size_, static_cast<int64_t>(batches_)));
 503
 1504    if (device_workspace_size_ != 0) {
 1505      device_workspace_ = curaii::make_unique_device_ptr<std::byte>(device_workspace_size_);
 506    }
 1507    host_workspace_.resize(host_workspace_size_);
 1508  }
 509
 1510  [[nodiscard]] bool is_compatible_stream(cudaStream_t stream) const override {
 1511    return cuda_context_for_stream(stream) == context_;
 1512  }
 513
 1514  void solve(float *matrices, float *eigenvalues, int *info, cudaStream_t stream) override {
 1515    if (!is_compatible_stream(stream)) {
 0516      throw std::invalid_argument(
 517          "[Pca] cuSOLVER eigensolver and PCA stream use different contexts");
 518    }
 519
 1520    CUSOLVER_CHECK(cusolverDnSetStream(handle_->get(), stream));
 1521    CUSOLVER_CHECK(cusolverDnXsyevBatched(
 522        handle_->get(), params_->get(), CUSOLVER_EIG_MODE_VECTOR, CUBLAS_FILL_MODE_LOWER, features_,
 523        CUDA_R_32F, matrices, features_, CUDA_R_32F, eigenvalues, CUDA_R_32F,
 524        device_workspace_.get(), device_workspace_size_, host_workspace_.data(),
 525        host_workspace_size_, info, static_cast<int64_t>(batches_)));
 1526  }
 527
 528private:
 529  int       features_;
 530  size_t    batches_;
 531  CUcontext context_{nullptr};
 532
 533  std::unique_ptr<curaii::CusolverDnHandle> handle_;
 534  std::unique_ptr<curaii::CusolverDnParams> params_;
 535
 536  curaii::unique_device_ptr<std::byte> device_workspace_;
 1537  size_t                               device_workspace_size_{0};
 538  std::vector<std::byte>               host_workspace_;
 1539  size_t                               host_workspace_size_{0};
 540};
 541
 542// -------------------------------------------------------------------------------------------------
 543// cuSolverDx runtime compilation
 544// -------------------------------------------------------------------------------------------------
 545
 546const char *cusolverdx_source();
 547
 548namespace cusolverdx_runtime {
 549
 550struct Assets {
 551  std::filesystem::path cuda_include;
 552  std::filesystem::path cusolverdx_include;
 553  std::filesystem::path cutlass_include;
 554  std::filesystem::path cusolverdx_fatbin;
 555};
 556
 0557std::string jitlink_error_log(nvJitLinkHandle linker) {
 0558  size_t size = 0;
 0559  if (linker == nullptr || nvJitLinkGetErrorLogSize(linker, &size) != NVJITLINK_SUCCESS ||
 560      size == 0) {
 0561    return {};
 562  }
 563
 0564  std::vector<char> log(size);
 0565  if (nvJitLinkGetErrorLog(linker, log.data()) != NVJITLINK_SUCCESS) {
 0566    return {};
 567  }
 568
 0569  return std::string(log.data());
 0570}
 571
 572void jitlink_check(nvJitLinkHandle linker, nvJitLinkResult result, const char *expression,
 1573                   const char *file, int line) {
 1574  if (result == NVJITLINK_SUCCESS) {
 1575    return;
 576  }
 577
 0578  const auto log     = jitlink_error_log(linker);
 0579  const auto message = std::format("nvJitLink error: {} at {}:{}\n  expression : {}\n{}",
 580                                   static_cast<int>(result), file, line, expression, log);
 0581  logger()->error("{}", message);
 0582  throw std::runtime_error(message);
 1583}
 584
 585#define PCA_NVJITLINK_CHECK(linker, expr) jitlink_check((linker), (expr), #expr, __FILE__, __LINE__)
 586
 1587std::filesystem::path executable_directory() {
 1588  std::vector<wchar_t> buffer(MAX_PATH);
 589
 1590  while (true) {
 1591    const auto length =
 592        GetModuleFileNameW(nullptr, buffer.data(), static_cast<DWORD>(buffer.size()));
 593
 1594    if (length == 0) {
 0595      throw std::runtime_error(
 596          std::format("[Pca] Failed to locate the executable (Windows error {})", GetLastError()));
 597    }
 598
 1599    if (length < buffer.size() - 1) {
 1600      return std::filesystem::path(std::wstring(buffer.data(), length)).parent_path();
 601    }
 602
 0603    buffer.resize(buffer.size() * 2);
 0604  }
 1605}
 606
 1607bool assets_exist(const Assets &assets) {
 1608  return std::filesystem::is_directory(assets.cuda_include) &&
 609         std::filesystem::is_regular_file(assets.cuda_include / "cuda_runtime.h") &&
 610         std::filesystem::is_directory(assets.cuda_include / "cccl") &&
 611         std::filesystem::is_regular_file(assets.cusolverdx_include / "cusolverdx.hpp") &&
 612         std::filesystem::is_regular_file(assets.cusolverdx_include / "cusolverdx_io.hpp") &&
 613         std::filesystem::is_directory(assets.cutlass_include) &&
 614         std::filesystem::is_regular_file(assets.cusolverdx_fatbin);
 1615}
 616
 1617Assets locate_assets() {
 1618  const auto installed_root =
 619      (executable_directory() / HOLOFLOW_NVRTC_ASSET_RELATIVE_DIR).lexically_normal();
 620
 621  const Assets installed{
 1622      installed_root / "cuda/include",
 1623      installed_root / "mathdx/include",
 1624      installed_root / "mathdx/cutlass/include",
 1625      installed_root / "mathdx/lib/libcusolverdx.fatbin",
 626  };
 627
 1628  if (assets_exist(installed)) {
 0629    return installed;
 630  }
 631
 632  const Assets build{
 1633      HOLOFLOW_CUDA_INCLUDE_DIR,
 1634      HOLOFLOW_CUSOLVERDX_INCLUDE_DIR,
 1635      HOLOFLOW_CUSOLVERDX_CUTLASS_INCLUDE_DIR,
 1636      HOLOFLOW_CUSOLVERDX_FATBIN,
 637  };
 638
 1639  if (assets_exist(build)) {
 1640    return build;
 641  }
 642
 0643  throw std::runtime_error(std::format(
 644      "[Pca] cuSolverDx runtime-compilation assets are missing. Expected installed assets under "
 645      "'{}' (CUDA headers, cuSolverDx/CommonDx headers, CUTLASS headers, and "
 646      "libcusolverdx.fatbin); build-tree fallback is also unavailable",
 647      installed_root.string()));
 1648}
 649
 650std::vector<char> compile_lto_ir(int features, int solver_sm, int architecture,
 1651                                 const Assets &assets) {
 1652  curaii::NvrtcProgram program(cusolverdx_source(), "pca_heev_kernel.cu");
 653
 1654  std::vector<std::string> options = {
 655      "--std=c++17",
 656      "--device-as-default-execution-space",
 657      "-dlto",
 658      "--relocatable-device-code=true",
 659      std::format("--gpu-architecture=sm_{}", architecture),
 660      std::format("-DPCA_FEATURES={}", features),
 661      std::format("-DPCA_SOLVER_SM={}", solver_sm),
 662      std::format("--include-path={}", assets.cuda_include.string()),
 663      std::format("--include-path={}", (assets.cuda_include / "cccl").string()),
 664      std::format("--include-path={}", assets.cusolverdx_include.string()),
 665      std::format("--include-path={}", assets.cutlass_include.string()),
 666  };
 667
 1668  std::vector<const char *> option_ptrs;
 1669  option_ptrs.reserve(options.size());
 1670  for (const auto &option : options) {
 1671    option_ptrs.push_back(option.c_str());
 1672  }
 673
 1674  const auto compile_result =
 675      nvrtcCompileProgram(program.get(), static_cast<int>(option_ptrs.size()), option_ptrs.data());
 676
 1677  if (compile_result != NVRTC_SUCCESS) {
 0678    size_t log_size = 0;
 0679    NVRTC_CHECK(nvrtcGetProgramLogSize(program.get(), &log_size));
 680
 0681    std::vector<char> log(log_size);
 0682    NVRTC_CHECK(nvrtcGetProgramLog(program.get(), log.data()));
 0683    logger()->error("[Pca] NVRTC compilation log:\n{}", log.data());
 0684    NVRTC_CHECK(compile_result);
 0685  }
 686
 1687  size_t lto_size = 0;
 1688  NVRTC_CHECK(nvrtcGetLTOIRSize(program.get(), &lto_size));
 689
 1690  std::vector<char> lto_ir(lto_size);
 1691  NVRTC_CHECK(nvrtcGetLTOIR(program.get(), lto_ir.data()));
 1692  return lto_ir;
 1693}
 694
 695std::vector<char> link_cubin(const std::vector<char> &lto_ir, int architecture,
 1696                             const Assets &assets) {
 1697  nvJitLinkHandle linker      = nullptr;
 1698  const auto      arch_option = std::format("-arch=sm_{}", architecture);
 1699  const char     *options[]   = {"-lto", arch_option.c_str()};
 700
 1701  PCA_NVJITLINK_CHECK(linker, nvJitLinkCreate(&linker, 2, options));
 702
 703  try {
 1704    PCA_NVJITLINK_CHECK(linker, nvJitLinkAddFile(linker, NVJITLINK_INPUT_FATBIN,
 705                                                 assets.cusolverdx_fatbin.string().c_str()));
 1706    PCA_NVJITLINK_CHECK(linker, nvJitLinkAddData(linker, NVJITLINK_INPUT_LTOIR,
 707                                                 const_cast<char *>(lto_ir.data()), lto_ir.size(),
 708                                                 "pca_heev_lto_ir"));
 1709    PCA_NVJITLINK_CHECK(linker, nvJitLinkComplete(linker));
 710
 1711    size_t cubin_size = 0;
 1712    PCA_NVJITLINK_CHECK(linker, nvJitLinkGetLinkedCubinSize(linker, &cubin_size));
 713
 1714    std::vector<char> cubin(cubin_size);
 1715    PCA_NVJITLINK_CHECK(linker, nvJitLinkGetLinkedCubin(linker, cubin.data()));
 1716    PCA_NVJITLINK_CHECK(linker, nvJitLinkDestroy(&linker));
 1717    return cubin;
 0718  } catch (...) {
 0719    if (linker != nullptr) {
 0720      (void)nvJitLinkDestroy(&linker);
 721    }
 0722    throw;
 0723  }
 1724}
 725
 1726std::vector<char> compile_cubin(int features, int solver_sm, int architecture) {
 1727  const auto assets = locate_assets();
 1728  const auto lto_ir = compile_lto_ir(features, solver_sm, architecture, assets);
 1729  return link_cubin(lto_ir, architecture, assets);
 1730}
 731
 732} // namespace cusolverdx_runtime
 733
 734// -------------------------------------------------------------------------------------------------
 735// cuSolverDx eigensolver
 736// -------------------------------------------------------------------------------------------------
 737
 738struct DxKernel {
 739  CUfunction   function{nullptr};
 740  unsigned int block_x{0};
 741  unsigned int shared_memory{0};
 742};
 743
 744class CusolverDxEigensolver final : public Eigensolver {
 745public:
 1746  CusolverDxEigensolver(int features, size_t batches, cudaStream_t stream)
 1747      : features_(features), batches_(batches) {
 1748    CUDA_CHECK(cudaGetDevice(&device_));
 749
 1750    int major = 0;
 1751    int minor = 0;
 1752    CUDA_CHECK(cudaDeviceGetAttribute(&major, cudaDevAttrComputeCapabilityMajor, device_));
 1753    CUDA_CHECK(cudaDeviceGetAttribute(&minor, cudaDevAttrComputeCapabilityMinor, device_));
 754
 1755    architecture_       = major * 10 + minor;
 1756    const int solver_sm = architecture_ * 10;
 757
 1758    PCA_DRIVER_CHECK(cuInit(0));
 1759    context_ = cuda_context_for_stream(stream);
 760
 1761    ScopedCudaContext context_guard(context_);
 762
 1763    const auto cubin = cusolverdx_runtime::compile_cubin(features_, solver_sm, architecture_);
 1764    PCA_DRIVER_CHECK(cuModuleLoadDataEx(&module_, cubin.data(), 0, nullptr, nullptr));
 765
 766    try {
 1767      load_kernel_configuration();
 1768      configure_shared_memory();
 0769    } catch (...) {
 0770      (void)cuModuleUnload(module_);
 0771      module_ = nullptr;
 0772      throw;
 0773    }
 1774  }
 775
 1776  ~CusolverDxEigensolver() noexcept override {
 1777    if (module_ == nullptr) {
 0778      return;
 779    }
 780
 781    try {
 1782      ScopedCudaContext context_guard(context_);
 1783      const auto        result = cuModuleUnload(module_);
 784
 1785      if (result != CUDA_SUCCESS) {
 0786        logger()->critical("{}",
 787                           driver_error_message(result, "cuModuleUnload", __FILE__, __LINE__));
 0788        std::abort();
 789      }
 1790    } catch (...) {
 0791      std::abort();
 0792    }
 1793  }
 794
 1795  [[nodiscard]] bool is_compatible_stream(cudaStream_t stream) const override {
 1796    return cuda_context_for_stream(stream) == context_;
 1797  }
 798
 1799  void solve(float *matrices, float *eigenvalues, int *info, cudaStream_t stream) override {
 1800    if (!is_compatible_stream(stream)) {
 0801      throw std::invalid_argument(
 802          "[Pca] cuSolverDx eigensolver and PCA stream use different contexts");
 803    }
 804
 1805    const size_t full_batches = (batches_ / batches_per_block_) * batches_per_block_;
 1806    if (full_batches != 0) {
 1807      launch_batched(matrices, eigenvalues, info, full_batches, stream);
 808    }
 809
 1810    const size_t tail_batches = batches_ - full_batches;
 1811    if (tail_batches != 0) {
 1812      launch_tail(matrices, eigenvalues, info, full_batches, tail_batches, stream);
 813    }
 1814  }
 815
 816private:
 1817  void load_kernel_configuration() {
 1818    batched_kernel_ = {
 819        .function      = module_function(module_, "pca_heev_batched"),
 820        .block_x       = read_module_constant<unsigned int>(module_, "pca_batched_block_x"),
 821        .shared_memory = read_module_constant<unsigned int>(module_, "pca_batched_shared_memory"),
 822    };
 823
 1824    tail_kernel_ = {
 825        .function      = module_function(module_, "pca_heev_tail"),
 826        .block_x       = read_module_constant<unsigned int>(module_, "pca_tail_block_x"),
 827        .shared_memory = read_module_constant<unsigned int>(module_, "pca_tail_shared_memory"),
 828    };
 829
 1830    batches_per_block_ = read_module_constant<unsigned int>(module_, "pca_batches_per_block");
 1831  }
 832
 1833  void configure_shared_memory() {
 1834    int max_shared_memory = 0;
 1835    CUDA_CHECK(cudaDeviceGetAttribute(&max_shared_memory, cudaDevAttrMaxSharedMemoryPerBlockOptin,
 836                                      device_));
 837
 1838    const bool uses_batched = batches_ >= batches_per_block_;
 1839    const bool uses_tail    = batches_ < batches_per_block_ || (batches_ % batches_per_block_) != 0;
 840
 1841    if (uses_batched &&
 842        batched_kernel_.shared_memory > static_cast<unsigned int>(max_shared_memory)) {
 0843      throw_shared_memory_error(batched_kernel_.shared_memory, max_shared_memory);
 844    }
 845
 1846    if (uses_tail && tail_kernel_.shared_memory > static_cast<unsigned int>(max_shared_memory)) {
 0847      throw_shared_memory_error(tail_kernel_.shared_memory, max_shared_memory);
 848    }
 849
 1850    if (uses_batched) {
 1851      PCA_DRIVER_CHECK(cuFuncSetAttribute(batched_kernel_.function,
 852                                          CU_FUNC_ATTRIBUTE_MAX_DYNAMIC_SHARED_SIZE_BYTES,
 853                                          static_cast<int>(batched_kernel_.shared_memory)));
 854    }
 855
 1856    if (uses_tail) {
 1857      PCA_DRIVER_CHECK(cuFuncSetAttribute(tail_kernel_.function,
 858                                          CU_FUNC_ATTRIBUTE_MAX_DYNAMIC_SHARED_SIZE_BYTES,
 859                                          static_cast<int>(tail_kernel_.shared_memory)));
 860    }
 1861  }
 862
 0863  [[noreturn]] void throw_shared_memory_error(unsigned int required, int max_shared_memory) const {
 0864    throw std::runtime_error(
 865        std::format("[Pca] cuSolverDx HEEV for {} features requires {} bytes of dynamic "
 866                    "shared memory, but device sm_{} provides {} bytes",
 867                    features_, required, architecture_, max_shared_memory));
 0868  }
 869
 870  void launch_batched(float *matrices, float *eigenvalues, int *info, size_t full_batches,
 1871                      cudaStream_t stream) const {
 1872    CUdeviceptr matrix_arg      = reinterpret_cast<CUdeviceptr>(matrices);
 1873    CUdeviceptr eigenvalues_arg = reinterpret_cast<CUdeviceptr>(eigenvalues);
 1874    CUdeviceptr info_arg        = reinterpret_cast<CUdeviceptr>(info);
 1875    auto        batch_arg       = static_cast<unsigned int>(full_batches);
 1876    void       *arguments[]     = {&matrix_arg, &eigenvalues_arg, &info_arg, &batch_arg};
 877
 1878    PCA_DRIVER_CHECK(cuLaunchKernel(batched_kernel_.function,
 879                                    static_cast<unsigned int>(full_batches / batches_per_block_), 1,
 880                                    1, batched_kernel_.block_x, 1, 1, batched_kernel_.shared_memory,
 881                                    reinterpret_cast<CUstream>(stream), arguments, nullptr));
 1882  }
 883
 884  void launch_tail(float *matrices, float *eigenvalues, int *info, size_t full_batches,
 1885                   size_t tail_batches, cudaStream_t stream) const {
 1886    const size_t matrix_offset = full_batches * static_cast<size_t>(features_) * features_;
 1887    const size_t value_offset  = full_batches * static_cast<size_t>(features_);
 888
 1889    CUdeviceptr matrix_arg      = reinterpret_cast<CUdeviceptr>(matrices + matrix_offset);
 1890    CUdeviceptr eigenvalues_arg = reinterpret_cast<CUdeviceptr>(eigenvalues + value_offset);
 1891    CUdeviceptr info_arg        = reinterpret_cast<CUdeviceptr>(info + full_batches);
 1892    auto        batch_arg       = static_cast<unsigned int>(tail_batches);
 1893    void       *arguments[]     = {&matrix_arg, &eigenvalues_arg, &info_arg, &batch_arg};
 894
 1895    PCA_DRIVER_CHECK(cuLaunchKernel(tail_kernel_.function, static_cast<unsigned int>(tail_batches),
 896                                    1, 1, tail_kernel_.block_x, 1, 1, tail_kernel_.shared_memory,
 897                                    reinterpret_cast<CUstream>(stream), arguments, nullptr));
 1898  }
 899
 900  int       features_;
 901  size_t    batches_;
 1902  int       device_{0};
 1903  int       architecture_{0};
 1904  CUcontext context_{nullptr};
 1905  CUmodule  module_{nullptr};
 906
 907  DxKernel     batched_kernel_;
 908  DxKernel     tail_kernel_;
 1909  unsigned int batches_per_block_{1};
 910};
 911
 912// -------------------------------------------------------------------------------------------------
 913// Eigensolver selection
 914// -------------------------------------------------------------------------------------------------
 915
 916constexpr int cusolverdx_max_features_exclusive = 256;
 917
 918std::unique_ptr<Eigensolver> make_eigensolver(const PcaLayout    &layout,
 1919                                              const PcaWorkspace &workspace, cudaStream_t stream) {
 1920  if (layout.features < cusolverdx_max_features_exclusive) {
 921    try {
 1922      return std::make_unique<CusolverDxEigensolver>(layout.features, layout.batches, stream);
 0923    } catch (const std::exception &error) {
 0924      logger()->warn("[Pca] cuSolverDx initialization failed for depth {}: {}\n"
 925                     "[Pca] Falling back to the conventional cuSOLVER eigensolver",
 926                     layout.features, error.what());
 0927    }
 928  }
 929
 930  try {
 1931    return std::make_unique<CusolverEigensolver>(layout.features, layout.batches,
 932                                                 workspace.matrices.get(),
 933                                                 workspace.eigenvalues.get(), stream);
 0934  } catch (const std::exception &error) {
 0935    logger()->error("[Pca] Failed to initialize any GPU eigensolver for depth {}: {}",
 936                    layout.features, error.what());
 937
 0938    throw std::runtime_error(std::format("PCA could not initialize a GPU eigensolver for depth {}. "
 939                                         "See the terminal log for details.",
 940                                         layout.features));
 0941  }
 1942}
 943
 944// -------------------------------------------------------------------------------------------------
 945// Embedded cuSolverDx device program
 946// -------------------------------------------------------------------------------------------------
 947
 1948const char *cusolverdx_source() {
 949  static constexpr char source[] = R"cusolverdx(
 950#include <cusolverdx.hpp>
 951#include <cusolverdx_io.hpp>
 952
 953using namespace cusolverdx;
 954
 955using Base = decltype(Size<PCA_FEATURES>() + Precision<float>() + Type<type::real>() +
 956                      Function<heev>() + FillMode<fill_mode::lower>() +
 957                      Arrangement<arrangement::col_major>() +
 958                      Job<job::overwrite_vectors>() + SM<PCA_SOLVER_SM>() + Block());
 959using BatchedSolver =
 960    decltype(Base() + BatchesPerBlock<Base::suggested_batches_per_block>());
 961using TailSolver = decltype(Base() + BatchesPerBlock<1>());
 962
 963extern "C" __constant__ unsigned int pca_batched_block_x = BatchedSolver::block_dim.x;
 964extern "C" __constant__ unsigned int pca_batched_shared_memory =
 965    BatchedSolver::shared_memory_size;
 966extern "C" __constant__ unsigned int pca_batches_per_block =
 967    BatchedSolver::batches_per_block;
 968extern "C" __constant__ unsigned int pca_tail_block_x = TailSolver::block_dim.x;
 969extern "C" __constant__ unsigned int pca_tail_shared_memory = TailSolver::shared_memory_size;
 970
 971template <class Solver>
 972__device__ void solve(float *covariance, float *eigenvalues, int *info,
 973                      const unsigned int batches) {
 974  constexpr unsigned int m = Solver::m_size;
 975  constexpr unsigned int batches_per_block = Solver::batches_per_block;
 976  constexpr unsigned int lda_shared = Solver::lda;
 977  constexpr unsigned int matrix_elements = m * m;
 978
 979  const unsigned int batch = blockIdx.x * batches_per_block;
 980  if (batch >= batches) {
 981    return;
 982  }
 983
 984  extern __shared__ __align__(16) cusolverdx::byte shared_memory[];
 985  auto [matrix_shared, eigenvalues_shared, workspace_shared] =
 986      cusolverdx::shared_memory::slice<float, float, float>(
 987          shared_memory, alignof(float), lda_shared * m * batches_per_block, alignof(float),
 988          m * batches_per_block, alignof(float));
 989
 990  float *matrix_global = covariance + matrix_elements * batch;
 991  float *eigenvalues_global = eigenvalues + m * batch;
 992
 993  cusolverdx::copy_2d<Solver, m, m, arrangement::col_major, batches_per_block>(
 994      matrix_global, m, matrix_shared, lda_shared);
 995  __syncthreads();
 996
 997  Solver().execute(matrix_shared, lda_shared, eigenvalues_shared, workspace_shared, info + batch);
 998
 999  cusolverdx::copy_2d<Solver, m, 1, arrangement::col_major, batches_per_block>(
 1000      eigenvalues_shared, m, eigenvalues_global, m);
 1001  __syncthreads();
 1002  cusolverdx::copy_2d<Solver, m, m, arrangement::col_major, batches_per_block>(
 1003      matrix_shared, lda_shared, matrix_global, m);
 1004}
 1005
 1006extern "C" __global__ void pca_heev_batched(float *covariance, float *eigenvalues, int *info,
 1007                                             const unsigned int batches) {
 1008  solve<BatchedSolver>(covariance, eigenvalues, info, batches);
 1009}
 1010
 1011extern "C" __global__ void pca_heev_tail(float *covariance, float *eigenvalues, int *info,
 1012                                          const unsigned int batches) {
 1013  solve<TailSolver>(covariance, eigenvalues, info, batches);
 1014}
 1015)cusolverdx";
 1016
 11017  return source;
 11018}
 1019
 1020#undef PCA_NVJITLINK_CHECK
 1021#undef PCA_DRIVER_CHECK
 1022
 1023} // namespace
 1024
 1025// -------------------------------------------------------------------------------------------------
 1026// JSON serialization
 1027// -------------------------------------------------------------------------------------------------
 1028
 11029void to_json(nlohmann::json &j, const PcaSettings &settings) {
 11030  j = nlohmann::json{
 1031      {"begin", settings.begin},
 1032      {"end", settings.end},
 1033  };
 11034}
 1035
 11036void from_json(const nlohmann::json &j, PcaSettings &settings) {
 11037  j.at("begin").get_to(settings.begin);
 11038  j.at("end").get_to(settings.end);
 11039}
 1040
 1041// -------------------------------------------------------------------------------------------------
 1042// Factory methods
 1043// -------------------------------------------------------------------------------------------------
 1044
 1045holoflow::core::InferResult PcaFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 11046                                              const nlohmann::json &jsettings) const {
 1047  const auto check = [&](bool condition, const std::string &message) {
 1048    if (!condition) {
 1049      logger()->error("[PcaFactory::infer] error: {}", message);
 1050      throw std::invalid_argument("PcaFactory inference error: " + message);
 1051    }
 1052  };
 1053
 11054  const auto settings = jsettings.get<PcaSettings>();
 1055
 11056  check(input_descs.size() == 1, "expected exactly one input");
 1057
 11058  const auto &input_desc = input_descs.front();
 11059  check(input_desc.rank() >= 3, "expected input rank >= 3");
 11060  check(input_desc.dtype == holoflow::core::DType::F32, "PCA currently supports F32 input only");
 11061  check(input_desc.mem_loc == holoflow::core::MemLoc::Device, "expected input in device memory");
 11062  check(settings.begin < settings.end, "expected begin < end");
 11063  check(settings.begin >= 0, "expected begin >= 0");
 1064
 11065  const PcaLayout layout(input_desc);
 11066  check(settings.end <= layout.features, "expected end <= n_features");
 1067
 11068  auto output_shape                    = input_desc.shape;
 11069  output_shape.at(layout.feature_axis) = static_cast<size_t>(settings.components());
 1070
 11071  holoflow::core::TDesc output_desc(output_shape, input_desc.dtype, input_desc.mem_loc);
 1072
 11073  return holoflow::core::InferResult{
 1074      .input_descs   = {input_desc},
 1075      .output_descs  = {output_desc},
 1076      .in_place      = {},
 1077      .owned_inputs  = {false},
 1078      .owned_outputs = {false},
 1079      .kind          = holoflow::core::TaskKind::Sync,
 1080  };
 11081}
 1082
 1083std::unique_ptr<holoflow::core::ISyncTask>
 1084PcaFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 1085                   const nlohmann::json                  &jsettings,
 11086                   const holoflow::core::SyncCreateCtx   &ctx) const {
 11087  this->infer(input_descs, jsettings);
 1088
 11089  return std::make_unique<PcaTask>(jsettings.get<PcaSettings>(), input_descs.front(), ctx);
 11090}
 1091
 1092std::unique_ptr<holoflow::core::ISyncTask>
 1093PcaFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 1094                   std::span<const holoflow::core::TDesc>     input_descs,
 1095                   const nlohmann::json                      &jsettings,
 11096                   const holoflow::core::SyncCreateCtx       &ctx) const {
 11097  this->infer(input_descs, jsettings);
 1098
 11099  const auto  settings   = jsettings.get<PcaSettings>();
 11100  const auto &input_desc = input_descs.front();
 1101
 11102  auto *pca = dynamic_cast<PcaTask *>(old_task.get());
 11103  if (pca != nullptr && pca->can_reuse(input_desc, ctx.stream)) {
 11104    pca->reconfigure(settings, ctx.stream);
 11105    return old_task;
 1106  }
 1107
 01108  return create(input_descs, jsettings, ctx);
 11109}
 1110
 1111} // namespace holotask::syncs

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\syncs\shack_hartmann_geometry.cc

#LineLine coverage
 1// Copyright 2026 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 "syncs/shack_hartmann_geometry.hh"
 16
 17#include <algorithm>
 18
 19namespace holotask::syncs::detail {
 20
 121ShackHartmannGeometry make_shack_hartmann_geometry(const ShackHartmannGeometrySettings &settings) {
 122  const float pitch_x_m = static_cast<float>(settings.stride_x) * settings.dx;
 123  const float pitch_y_m = static_cast<float>(settings.stride_y) * settings.dy;
 24
 125  const float aperture_width_m =
 26      static_cast<float>((settings.sx - 1) * settings.stride_x + settings.subaperture_width) *
 27      settings.dx;
 128  const float aperture_height_m =
 29      static_cast<float>((settings.sy - 1) * settings.stride_y + settings.subaperture_height) *
 30      settings.dy;
 31
 132  ShackHartmannGeometry geometry;
 133  geometry.pupil_radius_m = 0.5f * std::min(aperture_width_m, aperture_height_m);
 134  geometry.samples.reserve(settings.sy * settings.sx);
 35
 136  const float center_x = (static_cast<float>(settings.sx) - 1.0f) * 0.5f;
 137  const float center_y = (static_cast<float>(settings.sy) - 1.0f) * 0.5f;
 38
 139  for (size_t sy = 0; sy < settings.sy; ++sy) {
 140    for (size_t sx = 0; sx < settings.sx; ++sx) {
 141      const float x_m = (static_cast<float>(sx) - center_x) * pitch_x_m;
 142      const float y_m = (static_cast<float>(sy) - center_y) * pitch_y_m;
 143      const float x_n = x_m / geometry.pupil_radius_m;
 144      const float y_n = y_m / geometry.pupil_radius_m;
 45
 146      geometry.samples.push_back({
 47          .x_m    = x_m,
 48          .y_m    = y_m,
 49          .x_n    = x_n,
 50          .y_n    = y_n,
 51          .active = !settings.skip_subapertures_outside_pupil || x_n * x_n + y_n * y_n <= 1.0f,
 52      });
 153    }
 154  }
 55
 156  return geometry;
 157}
 58
 59} // namespace holotask::syncs::detail

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\syncs\zernike_defocus_z_prop.cc

#LineLine coverage
 1// Copyright 2026 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 "holotask/syncs/zernike_defocus_z_prop.hh"
 16
 17#include <algorithm>
 18#include <chrono>
 19#include <cmath>
 20#include <cstddef>
 21#include <stdexcept>
 22#include <string>
 23#include <utility>
 24
 25#include "bug.hh"
 26#include "cuda_runtime_api.h"
 27#include "logger.hh"
 28
 29#ifndef M_PI
 30#define M_PI 3.14159265358979323846
 31#endif
 32
 33namespace holotask::syncs {
 34
 35// -------------------------------------------------------------------------------------------------
 36// JSON serialization
 37// -------------------------------------------------------------------------------------------------
 38
 139void to_json(nlohmann::json &j, const ZernikeDefocusZPropSettings &s) {
 140  j = nlohmann::json{
 41      {"indexes", s.indexes},
 42      {"lambda", s.lambda},
 43      {"z_curr", s.z_curr},
 44      {"pupil_radius", s.pupil_radius},
 45      {"interval_seconds", s.interval_seconds},
 46  };
 147}
 48
 149void from_json(const nlohmann::json &j, ZernikeDefocusZPropSettings &s) {
 150  s.indexes.clear();
 51
 152  if (j.contains("indexes")) {
 153    j.at("indexes").get_to(s.indexes);
 054  } else if (j.contains("indices")) {
 055    j.at("indices").get_to(s.indexes);
 56  }
 57
 158  j.at("lambda").get_to(s.lambda);
 159  j.at("z_curr").get_to(s.z_curr);
 160  j.at("pupil_radius").get_to(s.pupil_radius);
 161  s.interval_seconds = j.value("interval_seconds", 1.0);
 162}
 63
 64namespace {
 65
 66// -------------------------------------------------------------------------------------------------
 67// Helpers
 68// -------------------------------------------------------------------------------------------------
 69
 70constexpr int   kDefocusNollIndex = 4;
 71constexpr float kInverseZEpsilon  = 1e-18f;
 72
 173void check(bool condition, const std::string &msg) {
 174  if (!condition) {
 175    logger()->error("[ZernikeDefocusZPropFactory::infer] error: {}", msg);
 176    throw std::invalid_argument("ZernikeDefocusZPropFactory inference error: " + msg);
 77  }
 178}
 79
 180bool is_c_contiguous(const holoflow::core::TDesc &desc) {
 181  if (desc.shape.size() != desc.strides.size()) {
 082    return false;
 83  }
 84
 185  size_t expected = holoflow::core::size_of(desc.dtype);
 186  for (size_t i = desc.rank(); i-- > 0;) {
 187    if (desc.strides[i] != expected) {
 088      return false;
 89    }
 190    expected *= desc.shape[i];
 191  }
 192  return true;
 193}
 94
 095bool same_desc(const holoflow::core::TDesc &a, const holoflow::core::TDesc &b) {
 096  return a.shape == b.shape && a.strides == b.strides && a.dtype == b.dtype &&
 97         a.mem_loc == b.mem_loc;
 098}
 99
 1100std::size_t defocus_position(const std::vector<int> &indexes) {
 1101  const auto it = std::find(indexes.begin(), indexes.end(), kDefocusNollIndex);
 1102  return static_cast<std::size_t>(std::distance(indexes.begin(), it));
 1103}
 104
 105struct ZPropEstimate {
 106  double delta_inv_z = 0.0;
 107  double z_new       = 0.0;
 108  double delta_z_mm  = 0.0;
 109};
 110
 1111ZPropEstimate estimate_z_prop_from_a4(float a4_rad, const ZernikeDefocusZPropSettings &settings) {
 1112  const double pupil_radius_sq = static_cast<double>(settings.pupil_radius) * settings.pupil_radius;
 1113  const double delta_inv_z     = (2.0 * std::sqrt(3.0) * settings.lambda * a4_rad) /
 114                                 (static_cast<double>(M_PI) * pupil_radius_sq);
 1115  const double inv_z_new       = (1.0 / settings.z_curr) - delta_inv_z;
 1116  const double z_new           = 1.0 / inv_z_new;
 117
 1118  return {
 119      .delta_inv_z = delta_inv_z,
 120      .z_new       = z_new,
 121      .delta_z_mm  = 1000.0 * (z_new - settings.z_curr),
 122  };
 1123}
 124
 125// -------------------------------------------------------------------------------------------------
 126// ZernikeDefocusZProp task implementation
 127// -------------------------------------------------------------------------------------------------
 128
 129class ZernikeDefocusZProp : public holoflow::core::ISyncTask {
 130public:
 1131  ZernikeDefocusZProp(ZernikeDefocusZPropSettings settings, holoflow::core::TDesc idesc,
 132                      cudaStream_t stream)
 1133      : settings_(std::move(settings)), idesc_(std::move(idesc)), stream_(stream) {}
 134
 1135  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 1136    const auto now = std::chrono::steady_clock::now();
 1137    if (now < next_execution_) {
 1138      return holoflow::core::OpResult::Ok;
 139    }
 1140    next_execution_ = now + std::chrono::duration_cast<std::chrono::steady_clock::duration>(
 141                                std::chrono::duration<double>(settings_.interval_seconds));
 142
 1143    auto       &input       = ctx.inputs[0];
 1144    const auto  a4_position = defocus_position(settings_.indexes);
 1145    const auto *src         = reinterpret_cast<const float *>(input.data()) + a4_position;
 146
 1147    float a4_rad = 0.0f;
 1148    switch (input.desc.mem_loc) {
 149    case holoflow::core::MemLoc::Host:
 1150      a4_rad = *src;
 1151      break;
 152    case holoflow::core::MemLoc::Device:
 0153      CUDA_CHECK(cudaMemcpyAsync(&a4_rad, src, sizeof(float), cudaMemcpyDeviceToHost, stream_));
 0154      CUDA_CHECK(cudaStreamSynchronize(stream_));
 0155      break;
 156    default:
 0157      throw std::logic_error("Unsupported memory location for Zernike defocus z_prop estimate");
 158    }
 159
 1160    if (!std::isfinite(a4_rad)) {
 0161      logger()->info("[ZernikeDefocusZPropTask] a4={:.4e} rad; estimated z_prop is undefined",
 162                     a4_rad);
 0163      return holoflow::core::OpResult::Ok;
 164    }
 165
 1166    const auto estimate = estimate_z_prop_from_a4(a4_rad, settings_);
 1167    if (!std::isfinite(estimate.z_new) || std::abs(1.0 / estimate.z_new) <= kInverseZEpsilon) {
 0168      logger()->info("[ZernikeDefocusZPropTask] a4={:.4e} rad; estimated z_prop is undefined",
 169                     a4_rad);
 0170      return holoflow::core::OpResult::Ok;
 171    }
 172
 1173    logger()->info("[ZernikeDefocusZPropTask] a4={:.4e} rad, delta_inv_z={:.4e} 1/m, "
 174                   "z_curr={:.4e} m, z_new={:.4e} m, delta_z={:.4e} mm "
 175                   "(pupil_radius={:.4e} m)",
 176                   a4_rad, estimate.delta_inv_z, settings_.z_curr, estimate.z_new,
 177                   estimate.delta_z_mm, settings_.pupil_radius);
 1178    return holoflow::core::OpResult::Ok;
 1179  }
 180
 0181  const ZernikeDefocusZPropSettings &settings() const { return settings_; }
 0182  const holoflow::core::TDesc       &idesc() const { return idesc_; }
 0183  void                               update_stream(cudaStream_t stream) { stream_ = stream; }
 184
 185private:
 186  ZernikeDefocusZPropSettings           settings_;
 187  holoflow::core::TDesc                 idesc_;
 1188  std::chrono::steady_clock::time_point next_execution_{};
 189  cudaStream_t                          stream_;
 190};
 191
 192} // namespace
 193
 194// -------------------------------------------------------------------------------------------------
 195// ZernikeDefocusZPropFactory
 196// -------------------------------------------------------------------------------------------------
 197
 198holoflow::core::InferResult
 199ZernikeDefocusZPropFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 1200                                  const nlohmann::json                  &jsettings) const {
 1201  const auto settings = jsettings.get<ZernikeDefocusZPropSettings>();
 202
 1203  check(settings.lambda > 0.0f, "lambda must be positive");
 1204  check(settings.z_curr != 0.0f, "z_curr must be non-zero");
 1205  check(settings.pupil_radius > 0.0f, "pupil_radius must be positive");
 1206  check(std::isfinite(settings.interval_seconds) && settings.interval_seconds > 0.0,
 207        "interval_seconds must be positive and finite");
 1208  check(!settings.indexes.empty(), "indexes must not be empty");
 1209  check(std::find(settings.indexes.begin(), settings.indexes.end(), kDefocusNollIndex) !=
 210            settings.indexes.end(),
 211        "indexes must contain Z4 defocus");
 212
 1213  auto unique_indexes = settings.indexes;
 1214  std::sort(unique_indexes.begin(), unique_indexes.end());
 1215  check(std::adjacent_find(unique_indexes.begin(), unique_indexes.end()) == unique_indexes.end(),
 216        "indexes must be unique");
 217
 1218  check(input_descs.size() == 1, "ZernikeDefocusZProp task must have exactly one input");
 219
 1220  const auto &idesc = input_descs[0];
 1221  check(idesc.dtype == holoflow::core::DType::F32, "Input coefficients dtype must be F32");
 1222  check(idesc.mem_loc == holoflow::core::MemLoc::Host ||
 223            idesc.mem_loc == holoflow::core::MemLoc::Device,
 224        "Input coefficients must be in Host or Device memory");
 1225  check(idesc.rank() == 1, "Input coefficients rank must be 1");
 1226  check(is_c_contiguous(idesc), "Input coefficients must be C-contiguous");
 1227  check(idesc.shape[0] == settings.indexes.size(),
 228        "Input coefficient count must match configured indexes size");
 229
 1230  return holoflow::core::InferResult{
 231      .input_descs   = {idesc},
 232      .output_descs  = {},
 233      .in_place      = {},
 234      .owned_inputs  = {false},
 235      .owned_outputs = {},
 236      .kind          = holoflow::core::TaskKind::Sync,
 237  };
 1238}
 239
 240std::unique_ptr<holoflow::core::ISyncTask>
 241ZernikeDefocusZPropFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 242                                   const nlohmann::json                  &jsettings,
 1243                                   const holoflow::core::SyncCreateCtx   &ctx) const {
 1244  (void)infer(input_descs, jsettings);
 245
 1246  auto settings = jsettings.get<ZernikeDefocusZPropSettings>();
 1247  return std::make_unique<ZernikeDefocusZProp>(std::move(settings), input_descs[0], ctx.stream);
 1248}
 249
 250std::unique_ptr<holoflow::core::ISyncTask>
 251ZernikeDefocusZPropFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 252                                   std::span<const holoflow::core::TDesc>     input_descs,
 253                                   const nlohmann::json                      &jsettings,
 0254                                   const holoflow::core::SyncCreateCtx       &ctx) const {
 0255  (void)infer(input_descs, jsettings);
 256
 0257  auto *old = dynamic_cast<ZernikeDefocusZProp *>(old_task.get());
 0258  if (old == nullptr) {
 0259    return create(input_descs, jsettings, ctx);
 260  }
 261
 0262  auto settings = jsettings.get<ZernikeDefocusZPropSettings>();
 0263  if (settings == old->settings() && same_desc(input_descs[0], old->idesc())) {
 0264    old->update_stream(ctx.stream);
 0265    return old_task;
 266  }
 267
 0268  return create(input_descs, jsettings, ctx);
 0269}
 270
 0271} // namespace holotask::syncs

C:\Users\Kremenchuk\actions-runner\_work\Holoflow\Holoflow\src\holotask\src\syncs\zernike_from_slopes.cc

#LineLine coverage
 1// Copyright 2026 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 "holotask/syncs/zernike_from_slopes.hh"
 16
 17#include <algorithm>
 18#include <array>
 19#include <cmath>
 20#include <cstddef>
 21#include <cstdint>
 22#include <span>
 23#include <stdexcept>
 24#include <string>
 25#include <utility>
 26#include <vector>
 27
 28#include "curaii/cuda.hh"
 29#include "logger.hh"
 30#include "syncs/shack_hartmann_geometry.hh"
 31#include "syncs/zernike_from_slopes_gpu.cuh"
 32
 33namespace holotask::syncs {
 34
 35// -------------------------------------------------------------------------------------------------
 36// JSON serialization
 37// -------------------------------------------------------------------------------------------------
 38
 139void to_json(nlohmann::json &j, const ZernikeFromSlopesSettings &s) {
 140  j = nlohmann::json{
 41      {"indexes", s.indexes},
 42      {"lambda", s.lambda},
 43      {"dx", s.dx},
 44      {"dy", s.dy},
 45      {"subaperture_height", s.subaperture_height},
 46      {"subaperture_width", s.subaperture_width},
 47      {"stride_y", s.stride_y},
 48      {"stride_x", s.stride_x},
 49      {"ny", s.ny},
 50      {"nx", s.nx},
 51      {"skip_subapertures_outside_pupil", s.skip_subapertures_outside_pupil},
 52  };
 153}
 54
 155void from_json(const nlohmann::json &j, ZernikeFromSlopesSettings &s) {
 156  s.indexes.clear();
 157  if (j.contains("indexes")) {
 158    j.at("indexes").get_to(s.indexes);
 059  } else if (j.contains("indices")) {
 060    j.at("indices").get_to(s.indexes);
 61  }
 62
 163  j.at("lambda").get_to(s.lambda);
 164  j.at("dx").get_to(s.dx);
 165  j.at("dy").get_to(s.dy);
 166  j.at("subaperture_height").get_to(s.subaperture_height);
 167  j.at("subaperture_width").get_to(s.subaperture_width);
 168  j.at("stride_y").get_to(s.stride_y);
 169  j.at("stride_x").get_to(s.stride_x);
 170  s.ny                              = j.value("ny", size_t{1});
 171  s.nx                              = j.value("nx", size_t{1});
 172  s.skip_subapertures_outside_pupil = j.value("skip_subapertures_outside_pupil", true);
 173}
 74
 75namespace {
 76
 77constexpr size_t kMaxSupportedModes = 9; // Noll indices 2..10
 78
 79template <typename T> using DevPtr = curaii::unique_device_ptr<T>;
 80
 181void check(bool condition, const std::string &message) {
 182  if (!condition) {
 183    logger()->error("[ZernikeFromSlopesFactory::infer] error: {}", message);
 184    throw std::invalid_argument("ZernikeFromSlopesFactory inference error: " + message);
 85  }
 186}
 87
 88struct ZernikeDerivative {
 89  float dx_n = 0.0f;
 90  float dy_n = 0.0f;
 91};
 92
 193ZernikeDerivative eval_zernike_noll_derivative(int noll_index, float x_n, float y_n) {
 194  const float sqrt3 = std::sqrt(3.0f);
 195  const float sqrt6 = std::sqrt(6.0f);
 196  const float sqrt8 = std::sqrt(8.0f);
 97
 198  switch (noll_index) {
 99  case 2:
 0100    return {2.0f, 0.0f};
 101  case 3:
 0102    return {0.0f, 2.0f};
 103  case 4:
 1104    return {4.0f * sqrt3 * x_n, 4.0f * sqrt3 * y_n};
 105  case 5:
 1106    return {2.0f * sqrt6 * y_n, 2.0f * sqrt6 * x_n};
 107  case 6:
 1108    return {2.0f * sqrt6 * x_n, -2.0f * sqrt6 * y_n};
 109  case 7:
 1110    return {6.0f * sqrt8 * x_n * y_n, 3.0f * sqrt8 * (x_n * x_n - y_n * y_n)};
 111  case 8:
 1112    return {6.0f * sqrt8 * x_n * y_n, sqrt8 * (3.0f * x_n * x_n + 9.0f * y_n * y_n - 2.0f)};
 113  case 9:
 1114    return {sqrt8 * (9.0f * x_n * x_n + 3.0f * y_n * y_n - 2.0f), 6.0f * sqrt8 * x_n * y_n};
 115  case 10:
 1116    return {3.0f * sqrt8 * (x_n * x_n - y_n * y_n), -6.0f * sqrt8 * x_n * y_n};
 117  default:
 0118    throw std::invalid_argument("Unsupported Noll index");
 119  }
 1120}
 121
 122std::array<float, kMaxSupportedModes>
 123solve_linear_system(std::array<std::array<float, kMaxSupportedModes>, kMaxSupportedModes> matrix,
 1124                    std::array<float, kMaxSupportedModes> rhs, size_t size) {
 1125  std::array<float, kMaxSupportedModes> solution{};
 1126  constexpr float                       singular_epsilon = 1e-12f;
 127
 1128  for (size_t column = 0; column < size; ++column) {
 1129    size_t pivot      = column;
 1130    float  best_value = std::abs(matrix[column][column]);
 1131    for (size_t row = column + 1; row < size; ++row) {
 1132      const float candidate = std::abs(matrix[row][column]);
 1133      if (candidate > best_value) {
 0134        best_value = candidate;
 0135        pivot      = row;
 136      }
 1137    }
 138
 1139    if (best_value < singular_epsilon) {
 0140      return solution;
 141    }
 142
 1143    if (pivot != column) {
 0144      std::swap(matrix[pivot], matrix[column]);
 0145      std::swap(rhs[pivot], rhs[column]);
 146    }
 147
 1148    const float pivot_value = matrix[column][column];
 1149    for (size_t j = column; j < size; ++j) {
 1150      matrix[column][j] /= pivot_value;
 1151    }
 1152    rhs[column] /= pivot_value;
 153
 1154    for (size_t row = column + 1; row < size; ++row) {
 1155      const float factor = matrix[row][column];
 1156      for (size_t j = column; j < size; ++j) {
 1157        matrix[row][j] -= factor * matrix[column][j];
 1158      }
 1159      rhs[row] -= factor * rhs[column];
 1160    }
 1161  }
 162
 1163  for (int row = static_cast<int>(size) - 1; row >= 0; --row) {
 1164    float value = rhs[static_cast<size_t>(row)];
 1165    for (size_t column = static_cast<size_t>(row) + 1; column < size; ++column) {
 1166      value -= matrix[static_cast<size_t>(row)][column] * solution[column];
 1167    }
 1168    solution[static_cast<size_t>(row)] = value;
 1169  }
 170
 1171  return solution;
 1172}
 173
 1174float load_slope(const holoflow::core::TView &view, size_t sy, size_t sx, size_t component) {
 1175  const auto *bytes = reinterpret_cast<const std::uint8_t *>(view.storage->ptr + view.desc.offset);
 1176  const auto  offset =
 177      sy * view.desc.strides[1] + sx * view.desc.strides[2] + component * view.desc.strides[3];
 1178  return *reinterpret_cast<const float *>(bytes + offset);
 1179}
 180
 181detail::ShackHartmannGeometrySettings geometry_settings(const holoflow::core::TDesc     &input,
 1182                                                        const ZernikeFromSlopesSettings &settings) {
 1183  return {
 184      .sy                              = input.shape[1],
 185      .sx                              = input.shape[2],
 186      .subaperture_height              = settings.subaperture_height,
 187      .subaperture_width               = settings.subaperture_width,
 188      .stride_y                        = settings.stride_y,
 189      .stride_x                        = settings.stride_x,
 190      .dy                              = settings.dy,
 191      .dx                              = settings.dx,
 192      .skip_subapertures_outside_pupil = settings.skip_subapertures_outside_pupil,
 193  };
 1194}
 195
 1196bool same_desc(const holoflow::core::TDesc &a, const holoflow::core::TDesc &b) {
 1197  return a.shape == b.shape && a.strides == b.strides && a.dtype == b.dtype &&
 198         a.mem_loc == b.mem_loc && a.offset == b.offset;
 1199}
 200
 201struct GpuFitData {
 202  std::vector<size_t>                  active_samples;
 203  std::vector<float>                   derivatives_x;
 204  std::vector<float>                   derivatives_y;
 205  std::vector<float>                   regularized_gram;
 206  detail::ZernikeFromSlopesGpuSettings kernel_settings;
 207};
 208
 209GpuFitData make_gpu_fit_data(const holoflow::core::TDesc     &input,
 1210                             const ZernikeFromSlopesSettings &settings) {
 1211  const auto geometry = detail::make_shack_hartmann_geometry(geometry_settings(input, settings));
 212
 1213  std::vector<size_t> observable_positions;
 1214  observable_positions.reserve(settings.indexes.size());
 1215  for (size_t position = 0; position < settings.indexes.size(); ++position) {
 1216    if (settings.indexes[position] != 2 && settings.indexes[position] != 3) {
 1217      observable_positions.push_back(position);
 218    }
 1219  }
 220
 1221  GpuFitData data;
 1222  data.active_samples.reserve(geometry.samples.size());
 1223  for (size_t sample = 0; sample < geometry.samples.size(); ++sample) {
 1224    if (geometry.samples[sample].active) {
 1225      data.active_samples.push_back(sample);
 226    }
 1227  }
 228
 1229  const size_t observable_count = observable_positions.size();
 1230  const size_t active_count     = data.active_samples.size();
 1231  data.derivatives_x.resize(active_count * observable_count);
 1232  data.derivatives_y.resize(active_count * observable_count);
 1233  std::array<float, kMaxSupportedModes> means_x{};
 1234  std::array<float, kMaxSupportedModes> means_y{};
 235
 1236  for (size_t active = 0; active < active_count; ++active) {
 1237    const auto &sample = geometry.samples[data.active_samples[active]];
 1238    for (size_t mode = 0; mode < observable_count; ++mode) {
 1239      const auto derivative = eval_zernike_noll_derivative(
 240          settings.indexes[observable_positions[mode]], sample.x_n, sample.y_n);
 1241      const size_t offset        = active * observable_count + mode;
 1242      data.derivatives_x[offset] = derivative.dx_n / geometry.pupil_radius_m;
 1243      data.derivatives_y[offset] = derivative.dy_n / geometry.pupil_radius_m;
 1244      means_x[mode] += data.derivatives_x[offset];
 1245      means_y[mode] += data.derivatives_y[offset];
 1246    }
 1247  }
 248
 1249  for (size_t mode = 0; mode < observable_count; ++mode) {
 1250    means_x[mode] /= static_cast<float>(active_count);
 1251    means_y[mode] /= static_cast<float>(active_count);
 1252  }
 1253  for (size_t active = 0; active < active_count; ++active) {
 1254    for (size_t mode = 0; mode < observable_count; ++mode) {
 1255      const size_t offset = active * observable_count + mode;
 1256      data.derivatives_x[offset] -= means_x[mode];
 1257      data.derivatives_y[offset] -= means_y[mode];
 1258    }
 1259  }
 260
 1261  data.regularized_gram.assign(observable_count * observable_count, 0.0f);
 1262  for (size_t active = 0; active < active_count; ++active) {
 1263    const size_t derivative_offset = active * observable_count;
 1264    for (size_t i = 0; i < observable_count; ++i) {
 1265      for (size_t j = 0; j < observable_count; ++j) {
 1266        data.regularized_gram[i * observable_count + j] +=
 267            data.derivatives_x[derivative_offset + i] * data.derivatives_x[derivative_offset + j] +
 268            data.derivatives_y[derivative_offset + i] * data.derivatives_y[derivative_offset + j];
 1269      }
 1270    }
 1271  }
 1272  constexpr float ridge = 1e-9f;
 1273  for (size_t mode = 0; mode < observable_count; ++mode) {
 1274    data.regularized_gram[mode * observable_count + mode] += ridge;
 1275  }
 276
 1277  data.kernel_settings = {
 278      .active_count      = active_count,
 279      .observable_count  = observable_count,
 280      .output_count      = settings.indexes.size(),
 281      .sx_count          = input.shape[2],
 282      .stride_y          = input.strides[1],
 283      .stride_x          = input.strides[2],
 284      .stride_component  = input.strides[3],
 285      .radians_per_meter = 2.0f * std::acos(-1.0f) / settings.lambda,
 286  };
 1287  for (size_t mode = 0; mode < observable_count; ++mode) {
 1288    data.kernel_settings.observable_positions[mode] = static_cast<int>(observable_positions[mode]);
 1289  }
 1290  return data;
 1291}
 292
 293// -------------------------------------------------------------------------------------------------
 294// ZernikeFromSlopes task implementation
 295// -------------------------------------------------------------------------------------------------
 296
 297class CpuZernikeFromSlopes : public holoflow::core::ISyncTask {
 298public:
 1299  CpuZernikeFromSlopes(ZernikeFromSlopesSettings settings, holoflow::core::TDesc input_desc,
 300                       cudaStream_t stream)
 1301      : settings_(std::move(settings)), input_desc_(std::move(input_desc)), stream_(stream) {}
 302
 1303  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 1304    CUDA_CHECK(cudaStreamSynchronize(stream_));
 305
 1306    const auto  &input    = ctx.inputs[0].desc;
 1307    const size_t sy_count = input.shape[1];
 1308    const size_t sx_count = input.shape[2];
 1309    const auto geometry = detail::make_shack_hartmann_geometry(geometry_settings(input, settings_));
 310
 1311    std::vector<size_t> observable_positions;
 1312    observable_positions.reserve(settings_.indexes.size());
 1313    for (size_t position = 0; position < settings_.indexes.size(); ++position) {
 1314      if (settings_.indexes[position] != 2 && settings_.indexes[position] != 3) {
 1315        observable_positions.push_back(position);
 316      }
 1317    }
 318
 1319    const size_t observable_count = observable_positions.size();
 1320    std::vector<std::array<float, kMaxSupportedModes>> derivatives_x(geometry.samples.size());
 1321    std::vector<std::array<float, kMaxSupportedModes>> derivatives_y(geometry.samples.size());
 1322    std::array<float, kMaxSupportedModes>              means_x{};
 1323    std::array<float, kMaxSupportedModes>              means_y{};
 1324    size_t                                             active_count = 0;
 325
 1326    for (size_t sample_index = 0; sample_index < geometry.samples.size(); ++sample_index) {
 1327      const auto &sample = geometry.samples[sample_index];
 1328      if (!sample.active) {
 1329        continue;
 330      }
 331
 1332      ++active_count;
 1333      for (size_t mode = 0; mode < observable_count; ++mode) {
 1334        const auto derivative = eval_zernike_noll_derivative(
 335            settings_.indexes[observable_positions[mode]], sample.x_n, sample.y_n);
 1336        derivatives_x[sample_index][mode] = derivative.dx_n / geometry.pupil_radius_m;
 1337        derivatives_y[sample_index][mode] = derivative.dy_n / geometry.pupil_radius_m;
 1338        means_x[mode] += derivatives_x[sample_index][mode];
 1339        means_y[mode] += derivatives_y[sample_index][mode];
 1340      }
 1341    }
 342
 1343    for (size_t mode = 0; mode < observable_count; ++mode) {
 1344      means_x[mode] /= static_cast<float>(active_count);
 1345      means_y[mode] /= static_cast<float>(active_count);
 1346    }
 347
 1348    std::array<std::array<float, kMaxSupportedModes>, kMaxSupportedModes> gtg{};
 1349    std::array<float, kMaxSupportedModes>                                 gts{};
 350
 1351    for (size_t sy = 0; sy < sy_count; ++sy) {
 1352      for (size_t sx = 0; sx < sx_count; ++sx) {
 1353        const size_t sample_index = sy * sx_count + sx;
 1354        if (!geometry.samples[sample_index].active) {
 1355          continue;
 356        }
 357
 1358        const float slope_x = load_slope(ctx.inputs[0], sy, sx, 0);
 1359        const float slope_y = load_slope(ctx.inputs[0], sy, sx, 1);
 360
 1361        for (size_t i = 0; i < observable_count; ++i) {
 1362          const float gx_i = derivatives_x[sample_index][i] - means_x[i];
 1363          const float gy_i = derivatives_y[sample_index][i] - means_y[i];
 1364          for (size_t j = 0; j < observable_count; ++j) {
 1365            const float gx_j = derivatives_x[sample_index][j] - means_x[j];
 1366            const float gy_j = derivatives_y[sample_index][j] - means_y[j];
 1367            gtg[i][j] += gx_i * gx_j + gy_i * gy_j;
 1368          }
 1369          gts[i] += gx_i * slope_x + gy_i * slope_y;
 1370        }
 1371      }
 1372    }
 373
 1374    constexpr float ridge = 1e-9f;
 1375    for (size_t mode = 0; mode < observable_count; ++mode) {
 1376      gtg[mode][mode] += ridge;
 1377    }
 1378    const auto coefficients_m = solve_linear_system(gtg, gts, observable_count);
 379
 1380    auto *output = reinterpret_cast<float *>(ctx.outputs[0].data());
 1381    std::fill_n(output, settings_.indexes.size(), 0.0f);
 1382    const float radians_per_meter = 2.0f * std::acos(-1.0f) / settings_.lambda;
 1383    for (size_t mode = 0; mode < observable_count; ++mode) {
 1384      output[observable_positions[mode]] = coefficients_m[mode] * radians_per_meter;
 1385    }
 386
 1387    return holoflow::core::OpResult::Ok;
 1388  }
 389
 0390  void                             update_stream(cudaStream_t stream) { stream_ = stream; }
 0391  const ZernikeFromSlopesSettings &settings() const { return settings_; }
 0392  const holoflow::core::TDesc     &input_desc() const { return input_desc_; }
 393
 394private:
 395  ZernikeFromSlopesSettings settings_;
 396  holoflow::core::TDesc     input_desc_;
 397  cudaStream_t              stream_;
 398};
 399
 400class GpuZernikeFromSlopes : public holoflow::core::ISyncTask {
 401public:
 402  GpuZernikeFromSlopes(ZernikeFromSlopesSettings settings, holoflow::core::TDesc input_desc,
 403                       GpuFitData fit_data, cudaStream_t stream)
 1404      : settings_(std::move(settings)), input_desc_(std::move(input_desc)),
 1405        kernel_settings_(fit_data.kernel_settings), stream_(stream),
 1406        active_samples_(curaii::make_unique_device_ptr<size_t>(
 407            std::max<size_t>(1, fit_data.active_samples.size()), stream)),
 1408        derivatives_x_(curaii::make_unique_device_ptr<float>(
 409            std::max<size_t>(1, fit_data.derivatives_x.size()), stream)),
 1410        derivatives_y_(curaii::make_unique_device_ptr<float>(
 411            std::max<size_t>(1, fit_data.derivatives_y.size()), stream)),
 1412        regularized_gram_(curaii::make_unique_device_ptr<float>(
 1413            std::max<size_t>(1, fit_data.regularized_gram.size()), stream)) {
 1414    if (!fit_data.active_samples.empty()) {
 1415      CUDA_CHECK(cudaMemcpyAsync(active_samples_.get(), fit_data.active_samples.data(),
 416                                 fit_data.active_samples.size() * sizeof(size_t),
 417                                 cudaMemcpyHostToDevice, stream_));
 418    }
 1419    if (!fit_data.derivatives_x.empty()) {
 1420      CUDA_CHECK(cudaMemcpyAsync(derivatives_x_.get(), fit_data.derivatives_x.data(),
 421                                 fit_data.derivatives_x.size() * sizeof(float),
 422                                 cudaMemcpyHostToDevice, stream_));
 1423      CUDA_CHECK(cudaMemcpyAsync(derivatives_y_.get(), fit_data.derivatives_y.data(),
 424                                 fit_data.derivatives_y.size() * sizeof(float),
 425                                 cudaMemcpyHostToDevice, stream_));
 1426      CUDA_CHECK(cudaMemcpyAsync(regularized_gram_.get(), fit_data.regularized_gram.data(),
 427                                 fit_data.regularized_gram.size() * sizeof(float),
 428                                 cudaMemcpyHostToDevice, stream_));
 429    }
 1430  }
 431
 1432  holoflow::core::OpResult execute(holoflow::core::SyncCtx &ctx) override {
 1433    CUDA_CHECK(detail::launch_zernike_from_slopes_gpu(
 434        reinterpret_cast<const float *>(ctx.inputs[0].data()),
 435        reinterpret_cast<float *>(ctx.outputs[0].data()), active_samples_.get(),
 436        derivatives_x_.get(), derivatives_y_.get(), regularized_gram_.get(), kernel_settings_,
 437        stream_));
 1438    return holoflow::core::OpResult::Ok;
 1439  }
 440
 1441  void                             update_stream(cudaStream_t stream) { stream_ = stream; }
 1442  const ZernikeFromSlopesSettings &settings() const { return settings_; }
 1443  const holoflow::core::TDesc     &input_desc() const { return input_desc_; }
 444
 445private:
 446  ZernikeFromSlopesSettings            settings_;
 447  holoflow::core::TDesc                input_desc_;
 448  detail::ZernikeFromSlopesGpuSettings kernel_settings_;
 449  cudaStream_t                         stream_;
 450  DevPtr<size_t>                       active_samples_;
 451  DevPtr<float>                        derivatives_x_;
 452  DevPtr<float>                        derivatives_y_;
 453  DevPtr<float>                        regularized_gram_;
 454};
 455
 456} // namespace
 457
 458// -------------------------------------------------------------------------------------------------
 459// ZernikeFromSlopesFactory
 460// -------------------------------------------------------------------------------------------------
 461
 462holoflow::core::InferResult
 463ZernikeFromSlopesFactory::infer(std::span<const holoflow::core::TDesc> input_descs,
 1464                                const nlohmann::json                  &jsettings) const {
 1465  const auto settings = jsettings.get<ZernikeFromSlopesSettings>();
 466
 1467  check(input_descs.size() == 1, "task must have exactly one input");
 1468  const auto &input = input_descs[0];
 1469  check(input.mem_loc == holoflow::core::MemLoc::Host ||
 470            input.mem_loc == holoflow::core::MemLoc::Device,
 471        "input memory location must be Host or Device");
 1472  check(input.dtype == holoflow::core::DType::F32, "input dtype must be F32");
 1473  check(input.rank() == 4, "input rank must be 4");
 1474  check(input.shape[0] == 1, "only batch size 1 is supported");
 1475  check(input.shape[1] > 0 && input.shape[2] > 0, "subaperture grid dimensions must be positive");
 1476  check(input.shape[3] == 2, "last input dimension must contain [dW/dx, dW/dy]");
 477
 1478  check(settings.lambda > 0.0f, "wavelength must be > 0");
 1479  check(settings.dx > 0.0f && settings.dy > 0.0f, "pixel pitches must be > 0");
 1480  check(settings.subaperture_height > 0 && settings.subaperture_width > 0,
 481        "subaperture dimensions must be positive");
 1482  check(settings.stride_y > 0 && settings.stride_x > 0, "strides must be positive");
 1483  check(settings.ny == 1 && settings.nx == 1,
 484        "only global fitting with ny = nx = 1 is supported for zero-mean slopes");
 485
 1486  check(!settings.indexes.empty(), "indexes must not be empty");
 1487  check(settings.indexes.size() <= kMaxSupportedModes, "too many requested Zernike modes");
 1488  for (int index : settings.indexes) {
 1489    check(index >= 2 && index <= 10, "only Noll indexes 2..10 are supported");
 1490  }
 1491  auto unique_indexes = settings.indexes;
 1492  std::sort(unique_indexes.begin(), unique_indexes.end());
 1493  check(std::adjacent_find(unique_indexes.begin(), unique_indexes.end()) == unique_indexes.end(),
 494        "indexes must be unique");
 495
 1496  const auto geometry = detail::make_shack_hartmann_geometry(geometry_settings(input, settings));
 1497  check(geometry.pupil_radius_m > 0.0f, "pupil radius must be positive");
 1498  check(std::ranges::any_of(geometry.samples, &detail::ShackHartmannSample::active),
 499        "at least one subaperture must be active");
 500
 1501  holoflow::core::TDesc output({1, 1, settings.indexes.size()}, holoflow::core::DType::F32,
 502                               input.mem_loc);
 1503  return {
 504      .input_descs   = {input},
 505      .output_descs  = {output},
 506      .in_place      = {},
 507      .owned_inputs  = {false},
 508      .owned_outputs = {false},
 509      .kind          = holoflow::core::TaskKind::Sync,
 510  };
 1511}
 512
 513std::unique_ptr<holoflow::core::ISyncTask>
 514ZernikeFromSlopesFactory::create(std::span<const holoflow::core::TDesc> input_descs,
 515                                 const nlohmann::json                  &jsettings,
 1516                                 const holoflow::core::SyncCreateCtx   &ctx) const {
 1517  (void)infer(input_descs, jsettings);
 1518  auto settings = jsettings.get<ZernikeFromSlopesSettings>();
 1519  auto input    = input_descs[0];
 1520  if (input.mem_loc == holoflow::core::MemLoc::Device) {
 1521    return std::make_unique<GpuZernikeFromSlopes>(settings, input,
 522                                                  make_gpu_fit_data(input, settings), ctx.stream);
 523  }
 1524  return std::make_unique<CpuZernikeFromSlopes>(settings, input, ctx.stream);
 1525}
 526
 527std::unique_ptr<holoflow::core::ISyncTask>
 528ZernikeFromSlopesFactory::update(std::unique_ptr<holoflow::core::ISyncTask> old_task,
 529                                 std::span<const holoflow::core::TDesc>     input_descs,
 530                                 const nlohmann::json                      &jsettings,
 1531                                 const holoflow::core::SyncCreateCtx       &ctx) const {
 1532  (void)infer(input_descs, jsettings);
 533
 1534  const auto  settings = jsettings.get<ZernikeFromSlopesSettings>();
 1535  const auto &input    = input_descs[0];
 1536  if (auto *old_cpu = dynamic_cast<CpuZernikeFromSlopes *>(old_task.get());
 537      old_cpu != nullptr && input.mem_loc == holoflow::core::MemLoc::Host &&
 1538      settings == old_cpu->settings() && same_desc(input, old_cpu->input_desc())) {
 0539    old_cpu->update_stream(ctx.stream);
 0540    return old_task;
 541  }
 1542  if (auto *old_gpu = dynamic_cast<GpuZernikeFromSlopes *>(old_task.get());
 543      old_gpu != nullptr && input.mem_loc == holoflow::core::MemLoc::Device &&
 1544      settings == old_gpu->settings() && same_desc(input, old_gpu->input_desc())) {
 1545    old_gpu->update_stream(ctx.stream);
 1546    return old_task;
 547  }
 548
 0549  return create(input_descs, jsettings, ctx);
 1550}
 551
 0552} // namespace holotask::syncs

Methods/Properties