fix: never self-move the parked output tuple
push_outputs ended with
else pending_ = std::move(result);
and the retry path calls it as push_outputs(std::move(*pending_), …), so on
that path `result` is the parked tuple itself. The assignment was a
self-move-assignment. std::tuple's is elementwise, and libstdc++'s
std::vector does not guard against self-move: _M_move_assign swaps its data
into a temporary, which is then destroyed. The vector ends up empty.
So the first park was clean — the argument there is a local temporary — and
the second erased the payload. The value was still delivered, still in
order, still counted, just empty. Downstream cannot distinguish that from a
frame on which the node genuinely found nothing, which is why it would never
surface as an error: in scene-actor-extraction it reads as "no faces in this
frame" and the run completes with a quietly wrong answer.
Scope, stated precisely because I first got it wrong: this needs a node with
*two or more* outputs. With one output the only thing that resubmits a
parked node is that output's own space callback, which by definition fires
when there is room, so the retry always succeeds and never reassigns. With
two, output A draining resubmits the node while output B is still full — the
retry skips A (already delivered, tracked in pending_done_) and fails on B,
and that is the reassignment that eats B's payload.
Every node in the scene-actor-extraction pipeline currently has exactly one
output, and the fanout is a separate class that does not use pending_, so
this is latent there rather than active. It is reachable by any multi-output
node under backpressure, which the library supports and documents.
Verified in both directions: on 6a4f45f the parked payload arrives with size
0; here it arrives intact. The test drives raw channels rather than consumer
nodes so each step is forced rather than raced, and both channels are
capacity 1 — Channel fires the space callback only on the full->not-full
edge, so a roomy channel A would never resubmit the node and the retry would
never happen at all.
This commit is contained in:
@@ -530,7 +530,15 @@ private:
|
||||
push_one_out<Is>(std::get<Is>(std::move(result))),
|
||||
all = all && pending_done_[Is]), ...);
|
||||
if (all) { pending_.reset(); pending_done_.fill(false); }
|
||||
else pending_ = std::move(result);
|
||||
// The retry path calls this as push_outputs(std::move(*pending_), …), so
|
||||
// on that path `result` *is* the parked tuple. Assigning it to itself is
|
||||
// a self-move-assignment, which for std::tuple is elementwise — and
|
||||
// libstdc++'s std::vector does not guard against it: it swaps its data
|
||||
// into a temporary and leaves the vector empty. A value that failed to
|
||||
// push twice would therefore be delivered with its payload silently
|
||||
// erased, which downstream reads as a legitimately empty result rather
|
||||
// than as a loss. Only store when it is not already stored.
|
||||
else if (!pending_ || &result != &*pending_) pending_ = std::move(result);
|
||||
}
|
||||
|
||||
/// Returns false when the ring was full and the value was NOT taken; the
|
||||
@@ -1012,7 +1020,15 @@ private:
|
||||
push_one_out<Is>(std::get<Is>(std::move(result))),
|
||||
all = all && pending_done_[Is]), ...);
|
||||
if (all) { pending_.reset(); pending_done_.fill(false); }
|
||||
else pending_ = std::move(result);
|
||||
// The retry path calls this as push_outputs(std::move(*pending_), …), so
|
||||
// on that path `result` *is* the parked tuple. Assigning it to itself is
|
||||
// a self-move-assignment, which for std::tuple is elementwise — and
|
||||
// libstdc++'s std::vector does not guard against it: it swaps its data
|
||||
// into a temporary and leaves the vector empty. A value that failed to
|
||||
// push twice would therefore be delivered with its payload silently
|
||||
// erased, which downstream reads as a legitimately empty result rather
|
||||
// than as a loss. Only store when it is not already stored.
|
||||
else if (!pending_ || &result != &*pending_) pending_ = std::move(result);
|
||||
}
|
||||
/// Returns false when the ring was full and the value was NOT taken; the
|
||||
/// caller must keep it and retry after the channel signals space.
|
||||
|
||||
@@ -524,3 +524,81 @@ TEST_CASE("node snapshot fields line up with the values nodes supply",
|
||||
CHECK_FALSE(snap.queued);
|
||||
CHECK_FALSE(snap.wake_pending);
|
||||
}
|
||||
|
||||
// Regression: a value parked twice must keep its payload.
|
||||
//
|
||||
// push_outputs ends with
|
||||
//
|
||||
// else pending_ = std::move(result);
|
||||
//
|
||||
// and the retry path calls it as push_outputs(std::move(*pending_), …), so on
|
||||
// that path `result` is the parked tuple itself. The assignment was therefore a
|
||||
// self-move-assignment. std::tuple's is elementwise, and libstdc++'s
|
||||
// std::vector does not guard against self-move: it swaps its data into a
|
||||
// temporary and leaves the vector empty. So the first park was clean (the
|
||||
// argument is a local temporary) and the second erased the payload.
|
||||
//
|
||||
// The value was still delivered, still in order, still counted — just empty.
|
||||
// Downstream cannot distinguish that from a frame on which the node genuinely
|
||||
// found nothing, which is why it never surfaced as an error: in
|
||||
// scene-actor-extraction it reads as "no faces in this frame" and the run
|
||||
// completes with a quietly wrong answer.
|
||||
//
|
||||
// Reaching it needs *two* outputs. With one, the only thing that resubmits a
|
||||
// parked node is that output's own space callback, which by definition fires
|
||||
// when there is room — so the retry always succeeds and never reassigns. With
|
||||
// two, output A draining resubmits the node while output B is still full: the
|
||||
// retry skips A (already delivered, tracked in pending_done_) and fails on B,
|
||||
// and that is the reassignment that eats B's payload.
|
||||
//
|
||||
// Driven through raw channels rather than consumer nodes so each step is
|
||||
// forced rather than raced: B is pre-filled and stays full for exactly as long
|
||||
// as the test wants it to.
|
||||
namespace {
|
||||
|
||||
struct TwoPayloads {
|
||||
static constexpr std::string_view label() { return "two_payloads"; }
|
||||
std::tuple<std::vector<int>, std::vector<int>> operator()() {
|
||||
return {std::vector<int>(4, 1), std::vector<int>(4, 2)};
|
||||
}
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST_CASE("a twice-parked value keeps its payload", "[pool_node][backpressure]") {
|
||||
auto pool = std::make_shared<ThreadPool>(2);
|
||||
pool->start();
|
||||
|
||||
TwoPayloads fn;
|
||||
auto node = make_pool_node(fn, pool);
|
||||
|
||||
// Both capacity 1. A must be *full* for its pop to signal space at all —
|
||||
// Channel fires the space callback only on the full->not-full edge, so a
|
||||
// roomy A would never resubmit the node and the retry would never happen.
|
||||
Channel<std::vector<int>> out_a(1), out_b(1);
|
||||
node.set_output_channel<0>(&out_a);
|
||||
node.set_output_channel<1>(&out_b);
|
||||
|
||||
// B is full before the node ever runs, so the very first firing parks.
|
||||
out_b.push(std::vector<int>(4, 99));
|
||||
|
||||
node.start();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
|
||||
// Draining A resubmits the node while B is still full: this is the retry
|
||||
// that reassigned the tuple to itself.
|
||||
(void)out_a.pop();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
|
||||
// Now let B through and collect what the node had been holding for it.
|
||||
(void)out_b.pop(); // the pre-fill
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
std::vector<int> parked = out_b.pop(); // the value parked across two tries
|
||||
|
||||
node.stop();
|
||||
pool->stop();
|
||||
|
||||
INFO("parked payload size " << parked.size());
|
||||
CHECK(parked.size() == 4);
|
||||
if (parked.size() == 4) CHECK(parked[0] == 2);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user