First attempt

include/kpn/channel.hpp

@@ -0,0 +1,175 @@
+#pragma once
+#include "diagnostics.hpp"
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <stdexcept>
+#include <string>
+#include <type_traits>
+
+namespace kpn {
+
+// ── Storage policy ────────────────────────────────────────────────────────────
+
+template<typename T>
+struct channel_storage_policy {
+    static constexpr bool by_value =
+        std::is_trivially_copyable_v<T> && sizeof(T) <= 8;
+};
+
+template<typename T>
+using channel_storage_t = std::conditional_t<
+    channel_storage_policy<T>::by_value,
+    T,
+    std::shared_ptr<const T>
+>;
+
+// ── 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<typename T>
+class Channel {
+public:
+    using storage_type = channel_storage_t<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<T>::by_value)
+            return std::move(v);
+        else
+            return std::make_shared<const T>(std::move(v));
+    }
+
+    static T extract(storage_type&& s) {
+        if constexpr (channel_storage_policy<T>::by_value)
+            return std::move(s);
+        else
+            return *s;
+    }
+
+    const std::size_t            capacity_;
+    std::queue<storage_type>     queue_;
+    std::atomic<bool>            accepting_{true};
+    mutable std::mutex           mutex_;
+    std::condition_variable      cv_;
+    ChannelStats                 stats_;
+};
+
+} // namespace kpn