Ignore build_test/ (out-of-tree test build dir)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitignore

@@ -1,5 +1,6 @@
 # Build output
 build/
+build_test/
 build_debug/
 site/
 # Python

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Duncan Tourolle
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -467,7 +467,7 @@ not data.
 
 Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | 1.7 | **1.4** |
 | chain depth-4 | 2.5 | **2.2** |
@@ -479,7 +479,7 @@ Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
 Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | **2.1** | 3.5 |
 | chain depth-4 | **4.3** | 5.2 |
@@ -489,7 +489,7 @@ Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 | wide fanout-4 | 3.4 | **1.9** |
 | diamond (2×2) | **4.1** | 6.1 |
 
-KPN private pools beat TBB for every chain and diamond topology at 100 µs/node, and
+KPN++ pools beat TBB for every chain and diamond topology at 100 µs/node, and
 match TBB within ~20% at 10 µs/node for shallow chains. TBB retains an edge on wide
 fanout (serial dispatch loop vs. work-stealing pool) and at extreme oversubscription
 depths (chain-32 at 10 µs). The remaining gap at light work is the cost of
@@ -578,3 +578,33 @@ examples/
 scripts/
   render_readme.py   — regenerates README.md from README.md.in
 ```
+
+---
+
+## Contributing
+
+Contributions are welcome. This project is hosted on a self-hosted Gitea
+instance that accepts sign-in and registration with a GitHub account, so you
+can log in with your existing GitHub identity to open issues and pull requests.
+
+If you change any code that appears in a README snippet, edit `README.md.in`
+(the template) rather than `README.md` directly, then regenerate:
+
+```bash
+cmake --build build --target readme   # or: python scripts/render_readme.py
+```
+
+---
+
+## Acknowledgments
+
+AI tooling was used heavily throughout the development of this project,
+including the design, implementation, tests, and documentation. All output
+has been reviewed, but please keep this in mind when reading or building on the
+code.
+
+---
+
+## License
+
+Released under the [MIT License](LICENSE). Copyright (c) 2026 Duncan Tourolle.

README.md.in

@@ -289,7 +289,7 @@ not data.
 
 Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | 1.7 | **1.4** |
 | chain depth-4 | 2.5 | **2.2** |
@@ -301,7 +301,7 @@ Overhead µs/item at **work_us = 10** (framework overhead dominates):
 
 Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 
-| Topology | KPN private | TBB |
+| Topology | KPN++ | TBB |
 |---|---|---|
 | chain depth-1 | **2.1** | 3.5 |
 | chain depth-4 | **4.3** | 5.2 |
@@ -311,7 +311,7 @@ Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):
 | wide fanout-4 | 3.4 | **1.9** |
 | diamond (2×2) | **4.1** | 6.1 |
 
-KPN private pools beat TBB for every chain and diamond topology at 100 µs/node, and
+KPN++ pools beat TBB for every chain and diamond topology at 100 µs/node, and
 match TBB within ~20% at 10 µs/node for shallow chains. TBB retains an edge on wide
 fanout (serial dispatch loop vs. work-stealing pool) and at extreme oversubscription
 depths (chain-32 at 10 µs). The remaining gap at light work is the cost of
@@ -400,3 +400,33 @@ examples/
 scripts/
   render_readme.py   — regenerates README.md from README.md.in
 ```
+
+---
+
+## Contributing
+
+Contributions are welcome. This project is hosted on a self-hosted Gitea
+instance that accepts sign-in and registration with a GitHub account, so you
+can log in with your existing GitHub identity to open issues and pull requests.
+
+If you change any code that appears in a README snippet, edit `README.md.in`
+(the template) rather than `README.md` directly, then regenerate:
+
+```bash
+cmake --build build --target readme   # or: python scripts/render_readme.py
+```
+
+---
+
+## Acknowledgments
+
+AI tooling was used heavily throughout the development of this project,
+including the design, implementation, tests, and documentation. All output
+has been reviewed, but please keep this in mind when reading or building on the
+code.
+
+---
+
+## License
+
+Released under the [MIT License](LICENSE). Copyright (c) 2026 Duncan Tourolle.

include/kpn/channel.hpp

@@ -136,6 +136,36 @@ public:
             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() {
@@ -148,21 +178,28 @@ public:
             // 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.
+                { T s; if (take_sentinel(s)) return s; }
+
                 if (!accepting_.load(std::memory_order_acquire))
                     throw ChannelClosedError{};
 
-                for (std::size_t s = 0; s < spin_count_; ++s) {
+                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() or disable() fires.
+                    // Still empty after spin — sleep until push()/push_sentinel()
-                    // Re-check tail after loading w to guard against a lost wakeup.
+                    // 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;
                 }
@@ -190,9 +227,12 @@ public:
     }
 
     // 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 false;
+        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();
@@ -217,12 +257,21 @@ public:
         push_callback_ = std::move(cb);
     }
 
-    // Size derived lazily from ring indices — no separate counter on the hot path.
+    // 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);
     }
-    std::size_t approx_size() const { return size(); }
+
+    // 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); }
@@ -260,6 +309,17 @@ private:
             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_;
@@ -267,8 +327,16 @@ private:
     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};

include/kpn/pool_node.hpp

@@ -22,6 +22,38 @@
 
 namespace kpn {
 
+// ── Sentinel detection ────────────────────────────────────────────────────────
+// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
+// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
+// source Frame) or nested one level under a `.source` member (`v.source.eof`,
+// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap
+// the originating Frame). Sentinels are delivered losslessly and non-blockingly
+// via Channel::push_sentinel() instead of the throwing push(), so backpressure
+// can never drop the token that unblocks downstream teardown.
+//
+// Types with neither shape are never treated as sentinels — both traits are
+// SFINAE-safe and the runtime check compiles away to `false` for them, so this
+// stays a no-op for pipelines that don't use an eof convention.
+template<typename T, typename = void>
+struct has_eof_field : std::false_type {};
+template<typename T>
+struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
+    : std::true_type {};
+
+template<typename T, typename = void>
+struct has_source_eof_field : std::false_type {};
+template<typename T>
+struct has_source_eof_field<T,
+    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
+    : std::true_type {};
+
+template<typename T>
+constexpr bool is_sentinel_value(const T& v) {
+    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
+    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
+    else                                               return false;
+}
+
 // ── PoolNode ──────────────────────────────────────────────────────────────────
 //
 // Reactive alternative to Node<>. Instead of owning a blocked thread, the node
@@ -361,6 +393,13 @@ private:
     void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
         if (!ch) return;
+        // Sentinels (EOF) must never be dropped: a lost token wedges every
+        // downstream pop() forever. Deliver them out-of-band (push_sentinel),
+        // which never overflows and never blocks this node's worker thread.
+        if (is_sentinel_value(val)) {
+            ch->push_sentinel(std::move(val));
+            return;
+        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {
@@ -660,6 +699,13 @@ private:
     void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
         if (!ch) return;
+        // Sentinels (EOF) must never be dropped: a lost token wedges every
+        // downstream pop() forever. Deliver them out-of-band (push_sentinel),
+        // which never overflows and never blocks this node's worker thread.
+        if (is_sentinel_value(val)) {
+            ch->push_sentinel(std::move(val));
+            return;
+        }
         try {
             ch->push(std::move(val));
         } catch (const ChannelOverflowError&) {

mkdocs.yml

@@ -1,7 +1,7 @@
 site_name: KPN++
 site_description: A C++20 Kahn Process Network library
-repo_url: https://github.com/yourusername/kpn
+repo_url: https://gitea.tourolle.paris/dtourolle/KPN
-repo_name: kpn
+repo_name: dtourolle/KPN
 
 theme:
   name: material

scripts/render_readme.py

@@ -6,9 +6,9 @@ Directive format (in README.md.in):
   <!-- @snippet path/to/file.cpp snippet_name -->
 
 Snippet tags (in C++ source files):
-  // [snippet: snippet_name]
+  // --8<-- [start:snippet_name]
   ...content...
-  // [/snippet: snippet_name]
+  // --8<-- [end:snippet_name]
 
 In Python files use # instead of //.
 """
@@ -33,8 +33,8 @@ def comment_prefix(path: Path) -> str:
 
 def extract_snippet(file_path: Path, name: str) -> str:
     prefix = comment_prefix(file_path)
-    open_tag  = f'{prefix} [snippet: {name}]'
+    open_tag  = f'{prefix} --8<-- [start:{name}]'
-    close_tag = f'{prefix} [/snippet: {name}]'
+    close_tag = f'{prefix} --8<-- [end:{name}]'
 
     text = file_path.read_text()
     lines = text.splitlines()

tests/test_channel.cpp

@@ -175,3 +175,66 @@ TEST_CASE("bandwidth_mbs returns 0 when elapsed_s is zero or negative", "[channe
     REQUIRE(snap.bandwidth_mbs(0.0) == 0.0);
     REQUIRE(snap.bandwidth_mbs(-1.0) == 0.0);
 }
+
+TEST_CASE("push_sentinel never overflows even on a full channel", "[channel][sentinel]") {
+    Channel<int> ch(2);
+    ch.push(1);
+    ch.push(2);   // channel full — a plain push(3) would throw ChannelOverflowError
+
+    // The sentinel is stored out-of-band, so it neither throws nor blocks the
+    // caller — the exact property an EOF token needs under backpressure. This
+    // returns immediately with the ring still full.
+    REQUIRE(ch.push_sentinel(99));
+    REQUIRE(ch.size() == 2);   // sentinel did not consume ring capacity
+}
+
+TEST_CASE("push_sentinel is delivered after all ring data, in order", "[channel][sentinel]") {
+    Channel<int> ch(4);
+    ch.push(1);
+    ch.push(2);
+    ch.push_sentinel(99);   // enqueue EOF while data is still buffered
+
+    // Data drains first; the sentinel arrives only once the ring is empty.
+    REQUIRE(ch.pop() == 1);
+    REQUIRE(ch.pop() == 2);
+    REQUIRE(ch.pop() == 99);
+}
+
+TEST_CASE("push_sentinel wakes a blocked pop", "[channel][sentinel]") {
+    Channel<int> ch(2);   // empty
+    std::thread producer([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        ch.push_sentinel(99);   // must wake a consumer parked on an empty ring
+    });
+    REQUIRE(ch.pop() == 99);
+    producer.join();
+}
+
+TEST_CASE("approx_size counts a pending sentinel so consumers stay schedulable",
+          "[channel][sentinel]") {
+    Channel<int> ch(4);
+    REQUIRE(ch.approx_size() == 0);
+    ch.push_sentinel(99);
+    // Node readiness checks call approx_size(); it must report the out-of-band
+    // sentinel as consumable work even though it holds no ring slot.
+    REQUIRE(ch.approx_size() == 1);
+    REQUIRE(ch.size() == 0);       // ...but the ring itself is still empty
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));
+    REQUIRE(out == 99);
+    REQUIRE(ch.approx_size() == 0);
+}
+
+TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty",
+          "[channel][sentinel]") {
+    Channel<int> ch(2);
+    ch.push(1);
+    ch.push_sentinel(99);
+
+    int out = 0;
+    REQUIRE(ch.try_pop_now(out));   // ring data first
+    REQUIRE(out == 1);
+    REQUIRE(ch.try_pop_now(out));   // then the sentinel
+    REQUIRE(out == 99);
+    REQUIRE_FALSE(ch.try_pop_now(out));   // nothing left
+}