Gabriel39 commented on code in PR #67328:
URL: https://github.com/apache/doris/pull/67328#discussion_r3911820678
##########
be/src/runtime/result_block_buffer.cpp:
##########
@@ -103,19 +104,140 @@ Status ResultBlockBuffer<ResultCtxType>::close(const
TUniqueId& id, Status exec_
}
template <typename ResultCtxType>
-void ResultBlockBuffer<ResultCtxType>::cancel(const Status& reason) {
- std::unique_lock<std::mutex> l(_lock);
+void ResultBlockBuffer<ResultCtxType>::cancel(const Status& reason, bool
release_outfile) {
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
- if (_status.ok()) {
- _status = reason;
+ {
+ std::unique_lock<std::mutex> l(_lock);
+ if (_status.ok()) {
+ _status = reason;
+ }
+ _arrow_data_arrival.notify_all();
+ for (auto& ctx : _waiting_rpc) {
+ ctx->on_failure(reason);
+ }
+ _waiting_rpc.clear();
+ _update_dependency();
+ _result_batch_queue.clear();
}
- _arrow_data_arrival.notify_all();
- for (auto& ctx : _waiting_rpc) {
- ctx->on_failure(reason);
+ if (release_outfile) {
+ // Release query result memory before rollback performs potentially
slow remote I/O.
+ release_outfile_cleanup();
}
- _waiting_rpc.clear();
- _update_dependency();
- _result_batch_queue.clear();
+}
+
+template <typename ResultCtxType>
+Status ResultBlockBuffer<ResultCtxType>::add_outfile_cleanup(OutfileCleanup
cleanup) {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
+ bool run_cleanup = false;
+ bool discard_cleanup = false;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ if (_outfile_state == OutfileState::ABORTED) {
+ run_cleanup = true;
+ } else if (_outfile_state == OutfileState::COMMITTED) {
+ discard_cleanup = true;
+ } else {
+ _outfile_cleanups.emplace_back(std::move(cleanup));
+ }
+ }
+ if (discard_cleanup) {
+ cleanup = nullptr;
+ return Status::OK();
+ }
+ if (run_cleanup) {
+ Status status;
+ for (int attempt = 0; attempt < 3; ++attempt) {
+ status = cleanup();
+ if (status.ok()) {
+ return status;
+ }
+ }
+ // Cancellation removed this buffer from the manager, so bounded
inline retries are the
+ // last live-query owner for a cleanup registered after that point.
+ return status;
+ }
+ return Status::OK();
+}
+
+template <typename ResultCtxType>
+Status ResultBlockBuffer<ResultCtxType>::finish_outfile(OutfileOperation
operation) {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);
+ std::vector<OutfileCleanup> cleanups;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ if (operation == OutfileOperation::PREPARE) {
+ if (_outfile_state == OutfileState::ABORTED) {
+ return Status::Cancelled("outfile result buffer was already
aborted");
+ }
+ if (_outfile_state == OutfileState::PENDING) {
+ _outfile_state = OutfileState::PREPARED;
+ }
+ return Status::OK();
+ }
+ if (operation == OutfileOperation::COMMIT) {
+ if (_outfile_state == OutfileState::COMMITTED) {
+ return Status::OK();
+ }
+ if (_outfile_state != OutfileState::PREPARED) {
+ return Status::InternalError("outfile result buffer is not
prepared");
+ }
+ // Keep rollback ownership after this provisional acknowledgement
so an explicit ABORT
+ // can compensate a COMMIT that reached only a subset of receivers.
+ _outfile_state = OutfileState::COMMITTED;
+ return Status::OK();
+ }
+ if (operation == OutfileOperation::ABORT) {
+ if (_outfile_state == OutfileState::ABORTED &&
_outfile_cleanups.empty()) {
Review Comment:
Fixed in d37cfa0cd7. ABORT drains now acquire a per-buffer drain mutex
before inspecting or swapping cleanup callbacks. A concurrent timeout/shutdown
cancellation waits for the active drain, then observes and retries callbacks
reinserted after a failure instead of treating the temporary empty vector as
terminal. A deterministic concurrent regression test was added and passed 100
consecutive ASAN iterations.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
