From 80c2b1fb2f1834631282fbbf6b5ec35ca2a4c8b2 Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Wed, 5 Aug 2026 16:14:50 +0200 Subject: [PATCH] fix: try_push must distinguish delivered from discarded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit try_push returned bool, and returned *true* for a closed channel — so "the value arrived" and "the value was thrown away because nobody is listening" were the same answer. Every caller was nonetheless correct, because both cases mean "stop trying, do not park and retry". But nothing above the channel could tell the two apart: a node counting successful pushes counted discards among them, and the only record of the loss was the channel's own drop counter, visible solely to whoever read the diagnostics table. Now a three-way PushResult { Taken, Full, Closed }, matching the shape SentinelResult already uses. Behaviour is unchanged at every call site — each treats Closed the same as Taken, and only Full parks — but the distinction is now available to anyone who needs it, and a scoped enum means a future caller cannot silently reintroduce the conflation with `if (push)`. deliver_one benefits immediately: it no longer reaches its teardown path for a closed channel, only for one that is still full, so the last-ditch throwing push it does there to record the loss now records an overflow rather than a drop the channel had already counted. --- include/kpn/branch.hpp | 24 ++++++++++++++++-------- include/kpn/channel.hpp | 23 ++++++++++++++++++----- include/kpn/fanout.hpp | 5 ++++- include/kpn/pool_node.hpp | 10 ++++++++-- tests/test_channel.cpp | 25 +++++++++++++++++++++++++ 5 files changed, 71 insertions(+), 16 deletions(-) diff --git a/include/kpn/branch.hpp b/include/kpn/branch.hpp index 676a1f8..81903ae 100644 --- a/include/kpn/branch.hpp +++ b/include/kpn/branch.hpp @@ -52,16 +52,24 @@ bool deliver_one(Channel* ch, T& val, const std::atomic& stop_flag, } const auto park_from = clock_t::now(); for (;;) { - if (ch->try_push(val)) { - parked = duration_t(clock_t::now() - park_from); - return true; + switch (ch->try_push(val)) { + case Channel::PushResult::Taken: + parked = duration_t(clock_t::now() - park_from); + return true; + case Channel::PushResult::Closed: + // Nobody is listening any more; the channel has recorded the + // drop. Retrying would spin until teardown noticed. + parked = duration_t(clock_t::now() - park_from); + return false; + case Channel::PushResult::Full: + break; // fall through to the retry logic } if (stop_flag.load(std::memory_order_relaxed)) { - // Teardown with work in hand. One last throwing push, purely so the - // channel's own stats record the loss (drop if it is disabled, - // overflow if it is merely full). The point of the lossless path is - // that a loss is never invisible, and a silent return here would - // reintroduce exactly the hole this function exists to close. + // Teardown with work in hand and the output still full. One last + // throwing push, purely so the channel's own stats record the + // overflow — the point of the lossless path is that a loss is never + // invisible, and a silent return here would reintroduce exactly the + // hole this function exists to close. try { ch->push(std::move(val)); } catch (const ChannelOverflowError&) {} parked = duration_t(clock_t::now() - park_from); diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index afaf9fe..e7420ff 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -165,13 +165,26 @@ public: head_.load(std::memory_order_acquire) < capacity_; } - /// Non-blocking, lossless push. Returns false when the ring is full, having + /// Outcome of a non-blocking push. + /// + /// try_push used to return bool, and returned *true* for a closed channel — + /// so "delivered" and "discarded because nobody is listening" were the same + /// answer. Both mean "stop trying", which is why the callers were correct, + /// but neither they nor the producer's own accounting could tell a value + /// that arrived from one that was thrown away. Only the channel's drop + /// counter knew. + enum class PushResult { Taken, Full, Closed }; + + /// Non-blocking, lossless push. Returns Full when the ring is full, having /// changed nothing — the caller keeps the value and retries when woken. - bool try_push(T& value) { - if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); return true; } + PushResult try_push(T& value) { + if (!accepting_.load(std::memory_order_acquire)) { + stats_.record_drop(); + return PushResult::Closed; + } const std::size_t t = tail_.load(std::memory_order_relaxed); const std::size_t h = head_.load(std::memory_order_acquire); - if (t - h >= capacity_) return false; + if (t - h >= capacity_) return PushResult::Full; const std::size_t data_bytes = ChannelDataSize::bytes(value); const bool was_empty = (t == h); @@ -181,7 +194,7 @@ public: wake_.fetch_add(1, std::memory_order_release); wake_.notify_one(); if (was_empty && push_callback_) push_callback_(); - return true; + return PushResult::Taken; } // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to diff --git a/include/kpn/fanout.hpp b/include/kpn/fanout.hpp index 1d6bf9e..f9f2c46 100644 --- a/include/kpn/fanout.hpp +++ b/include/kpn/fanout.hpp @@ -160,7 +160,10 @@ private: for (;;) { for (std::size_t i = 0; i < N; ++i) { if (!pending[i]) continue; - if (out_channels_[i]->try_push(*pending[i])) { + // Taken or Closed both mean "stop trying" — delivered, or gone + // with the drop recorded. Only Full is worth another pass. + if (out_channels_[i]->try_push(*pending[i]) + != Channel::PushResult::Full) { pending[i].reset(); --outstanding; } diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index c875b9b..e6113b5 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -641,7 +641,10 @@ private: // to run the consumer that would drain the channel. That is the // hold-and-wait deadlock channel.hpp warns about for sentinels; it // applies to data pushes just as much. - return ch->try_push(val); + // Closed counts as "stop trying", not as delivered: the value is gone + // and the channel has recorded the drop. Only Full means park and retry. + return ch->try_push(val) != Channel> + ::PushResult::Full; } template @@ -1215,7 +1218,10 @@ private: return true; } // See the note on the typed overload above: park rather than block. - return ch->try_push(val); + // Closed counts as "stop trying", not as delivered: the value is gone + // and the channel has recorded the drop. Only Full means park and retry. + return ch->try_push(val) != Channel> + ::PushResult::Full; } Obj& obj_; diff --git a/tests/test_channel.cpp b/tests/test_channel.cpp index 8150b6d..7be801e 100644 --- a/tests/test_channel.cpp +++ b/tests/test_channel.cpp @@ -298,3 +298,28 @@ TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][senti CHECK(ch.try_push_sentinel(third) == Channel::SentinelResult::Closed); CHECK(third == "eof-3"); } + +// Regression: try_push must distinguish delivered from discarded. +// +// It returned bool, and returned *true* for a closed channel — so "the value +// arrived" and "the value was thrown away because nobody is listening" were the +// same answer. Every caller was nonetheless correct, because both cases mean +// "stop trying"; but nothing above the channel could tell the two apart, and a +// producer counting successful pushes counted discards among them. Only the +// channel's own drop counter knew, and only if someone read the diagnostics. +TEST_CASE("try_push distinguishes taken, full and closed", "[channel]") { + Channel ch(2); + int v = 1; + + CHECK(ch.try_push(v) == Channel::PushResult::Taken); + CHECK(ch.try_push(v) == Channel::PushResult::Taken); + // Ring is full: the value is untouched and the caller keeps it. + CHECK(ch.try_push(v) == Channel::PushResult::Full); + CHECK(v == 1); + + ch.disable(); + const auto drops_before = ch.stats().drops.load(); + CHECK(ch.try_push(v) == Channel::PushResult::Closed); + // Discarded, and recorded as such rather than reported as a delivery. + CHECK(ch.stats().drops.load() == drops_before + 1); +}