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); +}