Skip to content

Channels

A Channel<T> is a lock-free SPSC (single-producer, single-consumer) ring buffer with atomic wait/notify.

Semantics

  • Bounded: fixed capacity set at construction. Default is 5 items.
  • Backpressure: when full, push() throws ChannelOverflowError immediately — no blocking, no spin.
  • Blocking consumer: pop() blocks until an item is available or the channel is disabled.
  • Disable: channel.disable() stops accepting pushes and unblocks any waiting pop() with ChannelClosedError.

Storage policy

Small trivially-copyable types (≤ 8 bytes) are stored by value. Larger types are heap-allocated and passed via shared_ptr<const T> — one allocation per push, zero-copy fan-out:

// Override: store Tag by value despite being a struct
// (it's trivially copyable and small — this just makes the policy explicit)
template<>
struct kpn::channel_storage_policy<Tag> {
    static constexpr bool by_value = true;
};

Specialize kpn::ChannelDataSize<T> for accurate bandwidth reporting on heap-owning types:

template<>
struct kpn::ChannelDataSize<cv::Mat> {
    static std::size_t bytes(const cv::Mat& m) { return m.total() * m.elemSize(); }
};

Named ports

in<"name"> and out<"name"> tag nodes for readable wiring:

    // tokenise: no inputs, one named output "words"
    auto tok = make_node<tokenise>(out<"words">{}, 4);

    // count_words: named input "words", named outputs "count" and "words"
    auto cnt = make_node<count_words>(in<"words">{}, out<"count", "words">{}, 4);

    // report: two named inputs
    auto snk = make_node<report>(in<"count", "words">{}, 4);

Named ports are checked at compile time — a typo in a port name is a compile error.

Capacity tuning

Set capacity per node at construction:

auto node = make_node<my_func>(/*capacity=*/20);

Capacity is rounded up internally to the next power of two. Monitor fill levels via diagnostics to tune for your workload — a too-small capacity causes overflows; a too-large one wastes memory and hides producer/consumer speed mismatches.

Spin count

Channel spins for up to ~4 µs (200 pause hints at ~20 ns each on x86) before sleeping on a futex. Set to 0 for power-constrained or predominantly-idle pipelines:

Channel<int> ch(/*capacity=*/5, /*spin_count=*/0);