Compare commits
1
Commits
27f884496d
...
c9aa246322
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c9aa246322 |
+15
-3
@@ -267,10 +267,22 @@ public:
|
|||||||
// has_eof_, the consumer is the only one that clears it, so observing
|
// 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
|
// false here means the consumer has finished with the storage and will
|
||||||
// not touch it again until this store publishes the next token.
|
// 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;
|
return SentinelResult::SlotBusy;
|
||||||
}
|
|
||||||
eof_value_ = make_storage(std::move(value));
|
eof_value_ = make_storage(std::move(value));
|
||||||
has_eof_.store(true, std::memory_order_release);
|
has_eof_.store(true, std::memory_order_release);
|
||||||
// Wake a consumer blocked in pop(): the sentinel is now deliverable even
|
// Wake a consumer blocked in pop(): the sentinel is now deliverable even
|
||||||
|
|||||||
+14
-16
@@ -622,14 +622,13 @@ private:
|
|||||||
// downstream pop() forever. Deliver them out-of-band (push_sentinel),
|
// downstream pop() forever. Deliver them out-of-band (push_sentinel),
|
||||||
// which never overflows and never blocks this node's worker thread.
|
// which never overflows and never blocks this node's worker thread.
|
||||||
if (is_sentinel_value(val)) {
|
if (is_sentinel_value(val)) {
|
||||||
// A refused sentinel is a protocol error, not backpressure, so it
|
// Not parked and not reported. Parking would spin against a slot
|
||||||
// is reported rather than parked and retried — retrying would spin
|
// only the consumer can free; reporting would cry data loss on the
|
||||||
// forever against a slot only the consumer can free, and there is
|
// normal steady state, since a source at end of input keeps being
|
||||||
// no correct value to deliver second anyway. Closed is normal
|
// polled and keeps returning EOF, so the token is re-offered on
|
||||||
// during teardown and stays quiet.
|
// every firing. Refusing a re-offer loses nothing — the pending
|
||||||
if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
|
// token says the same thing. See Channel::try_push_sentinel.
|
||||||
::SentinelResult::SlotBusy)
|
ch->try_push_sentinel(val);
|
||||||
fire_callbacks(event_callbacks_);
|
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
// Backpressure without parking the worker. A full channel means the
|
// 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),
|
// downstream pop() forever. Deliver them out-of-band (push_sentinel),
|
||||||
// which never overflows and never blocks this node's worker thread.
|
// which never overflows and never blocks this node's worker thread.
|
||||||
if (is_sentinel_value(val)) {
|
if (is_sentinel_value(val)) {
|
||||||
// A refused sentinel is a protocol error, not backpressure, so it
|
// Not parked and not reported. Parking would spin against a slot
|
||||||
// is reported rather than parked and retried — retrying would spin
|
// only the consumer can free; reporting would cry data loss on the
|
||||||
// forever against a slot only the consumer can free, and there is
|
// normal steady state, since a source at end of input keeps being
|
||||||
// no correct value to deliver second anyway. Closed is normal
|
// polled and keeps returning EOF, so the token is re-offered on
|
||||||
// during teardown and stays quiet.
|
// every firing. Refusing a re-offer loses nothing — the pending
|
||||||
if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
|
// token says the same thing. See Channel::try_push_sentinel.
|
||||||
::SentinelResult::SlotBusy)
|
ch->try_push_sentinel(val);
|
||||||
fire_callbacks(event_callbacks_);
|
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
// See the note on the typed overload above: park rather than block.
|
// See the note on the typed overload above: park rather than block.
|
||||||
|
|||||||
+16
-5
@@ -271,15 +271,26 @@ TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]")
|
|||||||
CHECK(out == 3);
|
CHECK(out == 3);
|
||||||
}
|
}
|
||||||
|
|
||||||
TEST_CASE("a refused sentinel is counted as a drop", "[channel][sentinel]") {
|
TEST_CASE("a refused sentinel is not counted as a drop", "[channel][sentinel]") {
|
||||||
// Visibility matters more here than for a dropped value: the refusal means
|
// A refusal means a token arrived while an equivalent one was already
|
||||||
// a control token went nowhere, and the only alternative to a counter is
|
// pending — not that anything was lost. Counting it as a drop was wrong in
|
||||||
// for it to vanish.
|
// 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<int> ch(4);
|
Channel<int> ch(4);
|
||||||
REQUIRE(ch.push_sentinel(1));
|
REQUIRE(ch.push_sentinel(1));
|
||||||
const auto before = ch.stats().drops.load();
|
const auto before = ch.stats().drops.load();
|
||||||
REQUIRE_FALSE(ch.push_sentinel(2));
|
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]") {
|
TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") {
|
||||||
|
|||||||
Reference in New Issue
Block a user