zanmato1984 commented on code in PR #51498:
URL: https://github.com/apache/arrow/pull/51498#discussion_r4118187633
##########
cpp/src/arrow/util/async_generator.h:
##########
@@ -1338,8 +1363,11 @@ class MergedGenerator {
// Now we have given up the lock and we can take all the actions we
decided we
// need to take.
+ if (signaled_error) {
+ RunErrorSignaledHookForTesting();
+ }
if (should_mark_final_error) {
- state->MarkFinalError(maybe_next->status(), std::move(sink));
+ state->DeliverFinalError(maybe_next->status(), std::move(sink));
Review Comment:
Could we also make the waiter-present path atomic with respect to later
terminal pulls?
When `sink` is valid, `SignalErrorUnlocked` sets `broken` while holding the
state mutex, but the error callback is not registered on `all_finished` until
`DeliverFinalError` runs after the mutex is released. A concurrent `operator()`
in that interval observes `broken` with an OK `final_error` and registers its
`IterationEnd` continuation on `all_finished` first.
Since `FutureImpl` runs callbacks in registration order, the newer terminal
future can complete before the older waiting future receives its error. This
violates the `AsyncGenerator` contract that a terminal result must not complete
while an earlier returned future is still outstanding.
The new hook makes this deterministic to reproduce:
1. Use one inner generator whose first future is pending.
2. Call `merged()` once, leaving an existing waiter.
3. Call `merged()` again from `error_signaled_hook_for_testing`.
4. Fail the pending future and record callback completion order.
Expected: `error, terminal`
Actual on this commit: `terminal, error`
Could we register or persist the waiting error as part of the locked state
transition so that later terminal pulls cannot overtake it, and add a
regression test for this waiter-present path? The same ordering should also be
checked for the outer-error path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
