diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index d586c1e..4dff40c 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -267,10 +267,22 @@ public: // has_eof_, the consumer is the only one that clears it, so observing // false here means the consumer has finished with the storage and will // not touch it again until this store publishes the next token. - if (has_eof_.load(std::memory_order_acquire)) { - stats_.record_drop(); + // + // Not counted as a drop, and this is the important part. A source that + // has reached the end of its input keeps being polled and keeps + // returning EOF — that is the normal steady state, not an error — so a + // token arriving while one is already pending is a *re-offer*, and + // refusing it loses nothing: the pending token carries the same + // meaning and is already on its way. Counting it as a drop made a + // clean run report data loss and exit non-zero. + // + // The cost of that choice, stated plainly: a genuinely distinct second + // token would also be refused silently, and the channel cannot tell the + // two apart. Re-offering is the case that actually occurs here, and the + // delivery guarantee that matters — the first token arrives — holds + // either way. + if (has_eof_.load(std::memory_order_acquire)) return SentinelResult::SlotBusy; - } 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 diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index e6113b5..45115aa 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -622,14 +622,13 @@ private: // 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)) { - // A refused sentinel is a protocol error, not backpressure, so it - // is reported rather than parked and retried — retrying would spin - // forever against a slot only the consumer can free, and there is - // no correct value to deliver second anyway. Closed is normal - // during teardown and stays quiet. - if (ch->try_push_sentinel(val) == Channel> - ::SentinelResult::SlotBusy) - fire_callbacks(event_callbacks_); + // Not parked and not reported. Parking would spin against a slot + // only the consumer can free; reporting would cry data loss on the + // normal steady state, since a source at end of input keeps being + // polled and keeps returning EOF, so the token is re-offered on + // every firing. Refusing a re-offer loses nothing — the pending + // token says the same thing. See Channel::try_push_sentinel. + ch->try_push_sentinel(val); return true; } // Backpressure without parking the worker. A full channel means the @@ -1207,14 +1206,13 @@ private: // 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)) { - // A refused sentinel is a protocol error, not backpressure, so it - // is reported rather than parked and retried — retrying would spin - // forever against a slot only the consumer can free, and there is - // no correct value to deliver second anyway. Closed is normal - // during teardown and stays quiet. - if (ch->try_push_sentinel(val) == Channel> - ::SentinelResult::SlotBusy) - fire_callbacks(event_callbacks_); + // Not parked and not reported. Parking would spin against a slot + // only the consumer can free; reporting would cry data loss on the + // normal steady state, since a source at end of input keeps being + // polled and keeps returning EOF, so the token is re-offered on + // every firing. Refusing a re-offer loses nothing — the pending + // token says the same thing. See Channel::try_push_sentinel. + ch->try_push_sentinel(val); return true; } // See the note on the typed overload above: park rather than block. diff --git a/tests/test_channel.cpp b/tests/test_channel.cpp index 7be801e..7b660c7 100644 --- a/tests/test_channel.cpp +++ b/tests/test_channel.cpp @@ -271,15 +271,26 @@ TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]") CHECK(out == 3); } -TEST_CASE("a refused sentinel is counted as a drop", "[channel][sentinel]") { - // Visibility matters more here than for a dropped value: the refusal means - // a control token went nowhere, and the only alternative to a counter is - // for it to vanish. +TEST_CASE("a refused sentinel is not counted as a drop", "[channel][sentinel]") { + // A refusal means a token arrived while an equivalent one was already + // pending — not that anything was lost. Counting it as a drop was wrong in + // a way that showed up immediately on real content: a source at the end of + // its input keeps being polled and keeps returning EOF, so the token is + // re-offered on every firing, and the pipeline reported hundreds of dropped + // frames on a clean run and exited non-zero. + // + // The delivery guarantee is unaffected: the first token is pending and will + // arrive. Only the accounting changed. Channel ch(4); REQUIRE(ch.push_sentinel(1)); const auto before = ch.stats().drops.load(); REQUIRE_FALSE(ch.push_sentinel(2)); - CHECK(ch.stats().drops.load() == before + 1); + CHECK(ch.stats().drops.load() == before); + + // And the one that was accepted is still the one delivered. + int out = 0; + REQUIRE(ch.try_pop_now(out)); + CHECK(out == 1); } TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") {