dtourolleandClaude Opus 4.8
a0c4bf580e
Channel<T>::pop() surfaced the out-of-band sentinel from its empty branch using the tail_ snapshot taken at the top of the loop. Under contention the producer can push more values *and* the sentinel in the window between that snapshot and take_sentinel(), so pop() could return the sentinel while real values still sat in the ring — the sentinel jumping ahead of values pushed before it. No value was lost (a consumer that keeps draining still receives them, and approx_size() keeps counting them so a PoolNode reschedules), but a consumer treating the sentinel as a hard "last message" barrier would act on EOF early. Re-confirm emptiness against a fresh tail_ load before taking the sentinel. Costs one acquire-load on the empty-ring path only; never runs in steady state. The spin and post-spin takes already reload tail_ on the line above them; try_pop_now() already reads tail_ fresh in the same branch — both were correct and are unchanged. The two sentinel stress cases now assert the strict "sentinel is last, after every value" ordering (previously relaxed to avoid the flake this fixes). Verified TSan-clean (2606 assertions, no data races) over repeated runs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
372 lines
16 KiB
C++
372 lines
16 KiB
C++
#pragma once
|
|
#include "diagnostics.hpp"
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <functional>
|
|
#include <memory>
|
|
#include <stdexcept>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <type_traits>
|
|
|
|
namespace kpn {
|
|
|
|
// ── Data size trait ───────────────────────────────────────────────────────────
|
|
// Returns the number of bytes of logical payload carried by a value.
|
|
// Defaults to sizeof(T), which is correct for PODs and fixed-size types.
|
|
// Specialize for heap-owning types (e.g. cv::Mat) to get accurate bandwidth:
|
|
//
|
|
// template<> struct kpn::ChannelDataSize<cv::Mat> {
|
|
// static std::size_t bytes(const cv::Mat& m) { return m.total() * m.elemSize(); }
|
|
// };
|
|
|
|
template<typename T>
|
|
struct ChannelDataSize {
|
|
static std::size_t bytes(const T&) { return sizeof(T); }
|
|
};
|
|
|
|
// ── 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") {}
|
|
};
|
|
|
|
// ── CPU pause hint ────────────────────────────────────────────────────────────
|
|
// Signals the CPU that this is a spin-wait loop, improving HT sibling throughput
|
|
// and preventing branch-predictor thrash on x86. Falls back to a compiler barrier.
|
|
|
|
[[maybe_unused]] static void spin_hint() noexcept {
|
|
#if defined(__x86_64__) || defined(__i386__)
|
|
__asm__ volatile("pause" ::: "memory");
|
|
#elif defined(__aarch64__) || defined(__arm__)
|
|
__asm__ volatile("yield" ::: "memory");
|
|
#else
|
|
std::atomic_signal_fence(std::memory_order_seq_cst);
|
|
#endif
|
|
}
|
|
|
|
// ── Channel ───────────────────────────────────────────────────────────────────
|
|
// SPSC ring buffer with atomic wait/notify and configurable spin-before-sleep.
|
|
//
|
|
// `spin_count` (constructor arg, default 200): number of pause-hint iterations
|
|
// before falling back to atomic::wait (futex). At ~20 ns/pause on x86 this is
|
|
// ~4 µs. Set to 0 to disable spinning (useful for power-constrained or
|
|
// predominantly-idle pipelines).
|
|
//
|
|
// Memory ordering contract (SPSC):
|
|
// push(): tail_.store(release) pairs with pop()'s tail_.load(acquire)
|
|
// head_.load(acquire) pairs with pop()'s head_.store(release)
|
|
// pop(): head_.store(release) pairs with push()'s head_.load(acquire)
|
|
// tail_.load(acquire) pairs with push()'s tail_.store(release)
|
|
|
|
template<typename T>
|
|
class Channel {
|
|
public:
|
|
using storage_type = channel_storage_t<T>;
|
|
|
|
explicit Channel(std::size_t capacity = 5, std::size_t spin_count = 200)
|
|
: capacity_(capacity), spin_count_(spin_count)
|
|
{
|
|
std::size_t rs = 1;
|
|
while (rs <= capacity) rs <<= 1; // smallest power-of-2 > capacity
|
|
ring_mask_ = rs - 1;
|
|
buf_ = std::make_unique<storage_type[]>(rs);
|
|
}
|
|
|
|
Channel(const Channel&) = delete;
|
|
Channel& operator=(const Channel&) = delete;
|
|
|
|
// Push a value.
|
|
// - If channel is disabled (accepting_ == false): silently drop.
|
|
// - If channel is full (fill >= capacity_): throw ChannelOverflowError.
|
|
void push(T value) {
|
|
if (!accepting_.load(std::memory_order_relaxed)) {
|
|
stats_.record_drop();
|
|
return;
|
|
}
|
|
const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
|
|
const std::size_t t = tail_.load(std::memory_order_relaxed);
|
|
const std::size_t h = head_.load(std::memory_order_acquire);
|
|
|
|
if (!accepting_.load(std::memory_order_acquire)) {
|
|
stats_.record_drop();
|
|
return;
|
|
}
|
|
if (t - h >= capacity_) {
|
|
stats_.record_overflow();
|
|
throw ChannelOverflowError(capacity_);
|
|
}
|
|
|
|
const bool was_empty = (t == h);
|
|
buf_[t & ring_mask_] = make_storage(std::move(value));
|
|
tail_.store(t + 1, std::memory_order_release);
|
|
stats_.record_push(t - h + 1, data_bytes);
|
|
|
|
wake_.fetch_add(1, std::memory_order_release);
|
|
wake_.notify_one();
|
|
|
|
if (was_empty && push_callback_)
|
|
push_callback_();
|
|
}
|
|
|
|
// Lossless, non-blocking delivery for a must-deliver control token (EOF).
|
|
//
|
|
// A sentinel is stored out-of-band — in a dedicated slot that does NOT
|
|
// consume ring capacity — so this can never overflow and never blocks the
|
|
// caller. That distinction is essential: each KPN node has a single worker
|
|
// thread, so a *blocking* push would park that thread and stop it draining
|
|
// its own input, cascading into a hold-and-wait deadlock under backpressure.
|
|
// Setting a flag and returning keeps the worker free to keep popping.
|
|
//
|
|
// The consumer's pop() drains the ring first, then delivers this sentinel,
|
|
// preserving ordering (EOF arrives after all data pushed before it).
|
|
//
|
|
// Only the sole producer may call it (SPSC contract, same as push()).
|
|
// Returns false if the channel is already disabled (token discarded —
|
|
// teardown is in progress, so the sentinel is moot).
|
|
bool push_sentinel(T value) {
|
|
if (!accepting_.load(std::memory_order_acquire)) {
|
|
stats_.record_drop();
|
|
return false;
|
|
}
|
|
eof_value_ = make_storage(std::move(value));
|
|
has_eof_.store(true, std::memory_order_release);
|
|
// Wake a consumer blocked in pop(): the sentinel is now deliverable even
|
|
// though the ring may be empty.
|
|
wake_.fetch_add(1, std::memory_order_release);
|
|
wake_.notify_one();
|
|
if (push_callback_) push_callback_();
|
|
return true;
|
|
}
|
|
|
|
// Blocking pop. Returns when an item is available.
|
|
// Throws ChannelClosedError if the channel is disabled (regardless of fill).
|
|
T pop() {
|
|
for (;;) {
|
|
// Snapshot wake_ BEFORE reading tail_ to prevent lost wakeups.
|
|
const uint32_t w = wake_.load(std::memory_order_relaxed);
|
|
const std::size_t h = head_.load(std::memory_order_relaxed);
|
|
std::size_t t = tail_.load(std::memory_order_acquire);
|
|
|
|
// If empty, spin before sleeping: avoids the futex when the next item
|
|
// arrives within the spin window (~4 µs at default spin_count=200 on x86).
|
|
if (h == t) {
|
|
// Ring drained — deliver any pending out-of-band sentinel (EOF)
|
|
// now, so it always arrives after the data pushed before it.
|
|
//
|
|
// Re-confirm emptiness against a fresh tail_ first: the snapshot
|
|
// at the top of the loop may be stale (the producer can push more
|
|
// values *and* the sentinel in the window since), and the sentinel
|
|
// must never jump ahead of ring values pushed before it. The spin
|
|
// and post-spin takes below already reload tail_ on the line above
|
|
// them; this is the one take that used the loop-top snapshot.
|
|
if (h == tail_.load(std::memory_order_acquire)) {
|
|
T s; if (take_sentinel(s)) return s;
|
|
}
|
|
|
|
if (!accepting_.load(std::memory_order_acquire))
|
|
throw ChannelClosedError{};
|
|
|
|
for (std::size_t si = 0; si < spin_count_; ++si) {
|
|
spin_hint();
|
|
t = tail_.load(std::memory_order_acquire);
|
|
if (t != h) break;
|
|
{ T s; if (take_sentinel(s)) return s; }
|
|
if (!accepting_.load(std::memory_order_relaxed))
|
|
throw ChannelClosedError{};
|
|
}
|
|
|
|
if (h == t) {
|
|
// Still empty after spin — sleep until push()/push_sentinel()
|
|
// or disable() fires. Re-check tail and the sentinel after
|
|
// loading w to guard against a lost wakeup.
|
|
if (tail_.load(std::memory_order_acquire) != h) continue;
|
|
if (has_eof_.load(std::memory_order_acquire)) continue;
|
|
wake_.wait(w, std::memory_order_relaxed);
|
|
continue;
|
|
}
|
|
}
|
|
|
|
// Item available (found immediately or during spin).
|
|
if (!accepting_.load(std::memory_order_acquire))
|
|
throw ChannelClosedError{};
|
|
T value = extract(std::move(buf_[h & ring_mask_]));
|
|
head_.store(h + 1, std::memory_order_release);
|
|
stats_.record_pop();
|
|
return value;
|
|
}
|
|
}
|
|
|
|
// Non-blocking pop with timeout. For watchdog/display use only.
|
|
bool try_pop(T& out, std::chrono::milliseconds timeout) {
|
|
const auto deadline = std::chrono::steady_clock::now() + timeout;
|
|
for (;;) {
|
|
if (try_pop_now(out)) return true;
|
|
if (!accepting_.load(std::memory_order_relaxed)) return false;
|
|
if (std::chrono::steady_clock::now() >= deadline) return false;
|
|
std::this_thread::sleep_for(std::chrono::microseconds(50));
|
|
}
|
|
}
|
|
|
|
// Immediate non-blocking pop. Returns false if the ring is empty.
|
|
// Once the ring is drained, delivers any pending out-of-band sentinel (EOF)
|
|
// so pool nodes — which pop only via this path — still receive the token.
|
|
bool try_pop_now(T& out) {
|
|
const std::size_t h = head_.load(std::memory_order_relaxed);
|
|
if (h == tail_.load(std::memory_order_acquire))
|
|
return take_sentinel(out);
|
|
out = extract(std::move(buf_[h & ring_mask_]));
|
|
head_.store(h + 1, std::memory_order_release);
|
|
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: stop accepting new pushes, unblock any waiting pop().
|
|
// Items already in the ring are abandoned and freed when the Channel is destroyed.
|
|
void disable() {
|
|
accepting_.store(false, std::memory_order_release);
|
|
wake_.fetch_add(1, std::memory_order_release);
|
|
wake_.notify_all();
|
|
}
|
|
|
|
// Register a callback fired when the queue transitions empty→non-empty.
|
|
void set_push_callback(std::function<void()> cb) {
|
|
push_callback_ = std::move(cb);
|
|
}
|
|
|
|
// Ring occupancy, derived lazily from indices — no separate counter on the
|
|
// hot path. Excludes any out-of-band sentinel (that lives outside the ring).
|
|
std::size_t size() const {
|
|
return tail_.load(std::memory_order_relaxed)
|
|
- head_.load(std::memory_order_relaxed);
|
|
}
|
|
|
|
// A pending out-of-band sentinel (EOF) counts as consumable work here even
|
|
// though it holds no ring slot. This is what node readiness checks call, so
|
|
// a channel carrying only a sentinel still schedules its consumer's next
|
|
// fire — without this the sentinel would never be popped and the pipeline
|
|
// would deadlock at teardown.
|
|
std::size_t approx_size() const {
|
|
return size() + (has_eof_.load(std::memory_order_acquire) ? 1u : 0u);
|
|
}
|
|
|
|
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 {
|
|
const std::size_t t = tail_.load(std::memory_order_relaxed);
|
|
const std::size_t h = head_.load(std::memory_order_relaxed);
|
|
return {
|
|
name,
|
|
capacity_,
|
|
t - h,
|
|
stats_.peak_fill.load(std::memory_order_relaxed),
|
|
stats_.pushes.load(std::memory_order_relaxed),
|
|
stats_.bytes_pushed.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),
|
|
};
|
|
}
|
|
|
|
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;
|
|
}
|
|
|
|
// Consume the out-of-band sentinel if one is pending. Consumer-only.
|
|
// Called only when the ring is observed empty, so the sentinel is always
|
|
// delivered after every value pushed before it.
|
|
bool take_sentinel(T& out) {
|
|
if (!has_eof_.load(std::memory_order_acquire)) return false;
|
|
out = extract(std::move(eof_value_));
|
|
has_eof_.store(false, std::memory_order_release);
|
|
stats_.record_pop();
|
|
return true;
|
|
}
|
|
|
|
const std::size_t capacity_;
|
|
const std::size_t spin_count_;
|
|
std::size_t ring_mask_;
|
|
std::unique_ptr<storage_type[]> buf_;
|
|
std::function<void()> push_callback_;
|
|
ChannelStats stats_;
|
|
|
|
// Out-of-band sentinel (EOF): stored outside the ring so its delivery never
|
|
// depends on ring capacity and never blocks the producer. Written by the
|
|
// producer (push_sentinel), read+cleared by the consumer (take_sentinel);
|
|
// has_eof_ is the publish/consume handshake.
|
|
storage_type eof_value_{};
|
|
std::atomic<bool> has_eof_{false};
|
|
|
|
// Separate cache lines: head_ is written only by the consumer;
|
|
// tail_ and wake_ are written only by the producer.
|
|
// wake_ wakes a blocked pop() on enqueue or on a pending sentinel.
|
|
alignas(64) std::atomic<std::size_t> head_{0};
|
|
alignas(64) std::atomic<std::size_t> tail_{0};
|
|
std::atomic<uint32_t> wake_{0};
|
|
std::atomic<bool> accepting_{true};
|
|
};
|
|
|
|
// ── 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<typename T>
|
|
struct ChannelProbe : IChannelProbe {
|
|
const Channel<T>& ch;
|
|
std::string name;
|
|
ChannelProbe(const Channel<T>& c, std::string n) : ch(c), name(std::move(n)) {}
|
|
ChannelSnapshot snapshot() const override { return ch.snapshot(name); }
|
|
};
|
|
|
|
} // namespace kpn
|