Renderer Engineering

#include "prism_engineering/asset_pipeline.hpp"
#include <algorithm>
#include <new>
#include <stdexcept>
#include <system_error>
#include <utility>

namespace prism_engineering {
std::expected<std::unique_ptr<AssetPipeline>, Error>
AssetPipeline::create(PipelineOptions options) {
  if (options.workers == 0 || options.workers > worker_limit)
    return std::unexpected(Error::invalid_worker_count);
  try {
    auto pipeline = std::unique_ptr<AssetPipeline>{new AssetPipeline};
    try {
      for (std::size_t i = 0; i < options.workers; ++i) {
        if (options.fail_launch_after == i)
          throw std::system_error{std::make_error_code(std::errc::resource_unavailable_try_again)};
        // The nonmovable pipeline remains at this address until all workers join.
        pipeline->workers_[i] = std::jthread{[owner = pipeline.get()] {
          try { owner->run_worker(); }
          catch (...) {
            owner->failed_.store(true);
            owner->stop_.request_stop();
            owner->changed_.notify_all();
          }
        }};
      }
    } catch (...) {
      pipeline->shutdown();
      throw;
    }
    return pipeline;
  } catch (const std::bad_alloc&) { return std::unexpected(Error::allocation); }
    catch (const std::system_error&) { return std::unexpected(Error::thread_start); }
}

AssetPipeline::~AssetPipeline() { shutdown(); }

std::expected<void, Admission> AssetPipeline::try_submit(ImageRequest request) {
  if (request.width < 2 || request.width > extent_limit ||
      request.height < 2 || request.height > extent_limit)
    return std::unexpected(Admission::invalid_dimensions);
  std::lock_guard lock{mutex_};
  if (!accepting_ || stop_.stop_requested()) return std::unexpected(Admission::closed);
  if (request.generation <= last_generation_) return std::unexpected(Admission::invalid_generation);
  auto slot = std::find_if(slots_.begin(), slots_.end(), [](const auto& item) { return item.phase == Phase::empty; });
  if (slot == slots_.end()) return std::unexpected(Admission::full);
  last_generation_ = request.generation; // Never increment an exhausted identifier.
  slot->request = std::move(request);
  slot->phase = Phase::queued;
  changed_.notify_one();
  return {};
}

bool AssetPipeline::has_completed_locked() const {
  return std::any_of(slots_.begin(), slots_.end(), [](const auto& item) { return item.phase == Phase::complete; });
}
std::optional<Completion> AssetPipeline::take_locked() {
  auto slot = std::find_if(slots_.begin(), slots_.end(), [](const auto& item) { return item.phase == Phase::complete; });
  if (slot == slots_.end()) return std::nullopt;
  Completion completion = std::move(slot->result);
  *slot = Slot{}; // Only taking a terminal result restores admission capacity.
  return completion;
}
std::optional<Completion> AssetPipeline::try_take() {
  std::lock_guard lock{mutex_};
  return take_locked();
}
std::optional<Completion> AssetPipeline::wait_take() {
  std::unique_lock lock{mutex_};
  changed_.wait(lock, [this] { return has_completed_locked() || joined_ || failed_.load(); });
  return take_locked();
}
void AssetPipeline::request_stop() {
  { std::lock_guard lock{mutex_}; accepting_ = false; }
  stop_.request_stop();
  changed_.notify_all();
}
void AssetPipeline::shutdown() {
  request_stop();
  for (auto& worker : workers_) if (worker.joinable()) worker.join();
  {
    std::lock_guard lock{mutex_};
    for (auto& slot : slots_) {
      if (slot.phase == Phase::queued || slot.phase == Phase::running) {
        slot.result = Completion{slot.request.generation, failed_.load() ? Status::failed : Status::cancelled, {}};
        slot.request.gate.reset();
        slot.phase = Phase::complete;
      }
    }
    joined_ = true;
  }
  changed_.notify_all();
}
void AssetPipeline::run_worker() {
  const auto token = stop_.get_token();
  for (;;) {
    Slot* slot{};
    ImageRequest request;
    {
      std::unique_lock lock{mutex_};
      if (!changed_.wait(lock, token, [this] {
        return std::any_of(slots_.begin(), slots_.end(), [](const auto& item) { return item.phase == Phase::queued; });
      }) || token.stop_requested()) return;
      slot = &*std::find_if(slots_.begin(), slots_.end(), [](const auto& item) { return item.phase == Phase::queued; });
      slot->phase = Phase::running;
      request = slot->request;
    }
    Completion result{request.generation, Status::succeeded, {}};
    try {
      if (request.gate && !request.gate->arrive_and_wait(token)) result.status = Status::cancelled;
      if (token.stop_requested()) result.status = Status::cancelled;
      if (result.status == Status::succeeded) {
        if (request.inject_failure) throw std::runtime_error{"injected image preparation failure"};
        result.image = PreparedImage{request.width, request.height, std::vector<Rgb>(static_cast<std::size_t>(request.width) * request.height)};
        for (std::uint32_t y = 0; y < request.height; ++y) {
          if (token.stop_requested()) { result.status = Status::cancelled; result.image = {}; break; }
          for (std::uint32_t x = 0; x < request.width; ++x)
            result.image.pixels[static_cast<std::size_t>(y) * request.width + x] = Rgb{
              static_cast<std::uint8_t>(255U * x / (request.width - 1U)),
              static_cast<std::uint8_t>(255U * y / (request.height - 1U)), request.blue};
        }
      }
    } catch (...) { result.status = Status::failed; result.image = {}; }
    {
      std::lock_guard lock{mutex_};
      slot->result = std::move(result);
      slot->request.gate.reset();
      slot->phase = Phase::complete;
    }
    changed_.notify_all();
  }
}

std::expected<void, Admission> Gallery::request(AssetPipeline& pipeline, ImageRequest request) {
  const auto generation = request.generation;
  auto accepted = pipeline.try_submit(std::move(request));
  if (!accepted) return accepted;
  requested_ = generation;
  settled_ = false;
  latest_ = Status::pending;
  return {};
}
bool Gallery::apply(Completion completion) {
  if (completion.generation != requested_ || settled_ || completion.status == Status::pending) return false;
  settled_ = true;
  latest_ = completion.status;
  if (completion.status != Status::succeeded) return false;
  image_ = std::move(completion.image);
  published_ = completion.generation;
  return true;
}
} // namespace prism_engineering