#pragma once #include "diagnostics.hpp" #include #include #include #include #include #include #include #include #include namespace kpn { // ── Storage policy ──────────────────────────────────────────────────────────── template struct channel_storage_policy { static constexpr bool by_value = std::is_trivially_copyable_v && sizeof(T) <= 8; }; template using channel_storage_t = std::conditional_t< channel_storage_policy::by_value, T, std::shared_ptr >; // ── Exceptions ──────────────────────────────────────────────────────────────── class ChannelOverflowError : public std::runtime_error { public: explicit ChannelOverflowError(std::size_t capacity) : std::runtime_error("channel overflow: capacity " + std::to_string(capacity) + " exceeded") {} ChannelOverflowError(std::size_t capacity, std::string context) : std::runtime_error(std::move(context) + ": capacity " + std::to_string(capacity) + " exceeded") {} }; class ChannelClosedError : public std::runtime_error { public: ChannelClosedError() : std::runtime_error("channel closed") {} }; // ── Channel ─────────────────────────────────────────────────────────────────── template class Channel { public: using storage_type = channel_storage_t; explicit Channel(std::size_t capacity = 5) : capacity_(capacity) {} Channel(const Channel&) = delete; Channel& operator=(const Channel&) = delete; // Push a value. // - If channel is disabled (accepting_ == false): silently drop, return immediately. // - If channel is full: throw ChannelOverflowError. void push(T value) { if (!accepting_.load(std::memory_order_relaxed)) { stats_.record_drop(); return; } std::unique_lock lock(mutex_); if (!accepting_.load(std::memory_order_relaxed)) { stats_.record_drop(); return; } if (queue_.size() >= capacity_) { stats_.record_overflow(); throw ChannelOverflowError(capacity_); } queue_.push(make_storage(std::move(value))); stats_.record_push(queue_.size()); lock.unlock(); cv_.notify_one(); } // Blocking pop. Unblocks when an item is available or the channel is disabled. // Throws ChannelClosedError if disabled and queue is empty. T pop() { std::unique_lock lock(mutex_); cv_.wait(lock, [this] { return !queue_.empty() || !accepting_.load(std::memory_order_relaxed); }); if (queue_.empty()) throw ChannelClosedError{}; T value = extract(std::move(queue_.front())); queue_.pop(); stats_.record_pop(); return value; } // Non-blocking pop with timeout. For watchdog/display use only — not used in run_loop. bool try_pop(T& out, std::chrono::milliseconds timeout) { std::unique_lock lock(mutex_); if (!cv_.wait_for(lock, timeout, [this] { return !queue_.empty() || !accepting_.load(std::memory_order_relaxed); })) return false; if (queue_.empty()) return false; out = extract(std::move(queue_.front())); queue_.pop(); stats_.record_pop(); return true; } // Enable the channel (called by consumer node on start()). void enable() { accepting_.store(true, std::memory_order_relaxed); } // Disable the channel: drop all queued items, unblock any waiting pop(). // Called by consumer node on stop(). Producer push() will silently drop after this. void disable() { accepting_.store(false, std::memory_order_relaxed); { std::lock_guard lock(mutex_); while (!queue_.empty()) queue_.pop(); } cv_.notify_all(); } std::size_t size() const { std::lock_guard lock(mutex_); return queue_.size(); } std::size_t capacity() const { return capacity_; } bool is_accepting() const { return accepting_.load(std::memory_order_relaxed); } const ChannelStats& stats() const { return stats_; } ChannelSnapshot snapshot(const std::string& name) const { std::lock_guard lock(mutex_); return { name, capacity_, queue_.size(), stats_.peak_fill.load(std::memory_order_relaxed), stats_.pushes.load(std::memory_order_relaxed), stats_.drops.load(std::memory_order_relaxed), stats_.overflows.load(std::memory_order_relaxed), stats_.pops.load(std::memory_order_relaxed), sizeof(T), // payload bytes — sizeof(T) regardless of storage policy }; } private: static storage_type make_storage(T&& v) { if constexpr (channel_storage_policy::by_value) return std::move(v); else return std::make_shared(std::move(v)); } static T extract(storage_type&& s) { if constexpr (channel_storage_policy::by_value) return std::move(s); else return *s; } const std::size_t capacity_; std::queue queue_; std::atomic accepting_{true}; mutable std::mutex mutex_; std::condition_variable cv_; ChannelStats stats_; }; // ── Channel probe — type-erased snapshot accessor ───────────────────────────── // Used by both Network and StaticNetwork for diagnostics. struct IChannelProbe { virtual ~IChannelProbe() = default; virtual ChannelSnapshot snapshot() const = 0; }; template struct ChannelProbe : IChannelProbe { const Channel& ch; std::string name; ChannelProbe(const Channel& c, std::string n) : ch(c), name(std::move(n)) {} ChannelSnapshot snapshot() const override { return ch.snapshot(name); } }; } // namespace kpn