hubgeter commented on code in PR #69682:
URL: https://github.com/apache/doris/pull/69682#discussion_r4412059032
##########
be/src/exec/exchange/vdata_stream_recvr.cpp:
##########
@@ -96,34 +96,40 @@ Status VDataStreamRecvr::SenderQueue::get_batch(Block*
block, bool* eos) {
block_item = std::move(_block_queue.front());
_block_queue.pop_front();
}
- BlockUPtr next_block;
- RETURN_IF_ERROR(block_item.get_block(next_block));
+
size_t block_byte_size = block_item.block_byte_size();
- COUNTER_UPDATE(_recvr->_deserialize_row_batch_timer,
block_item.deserialize_time());
- COUNTER_UPDATE(_recvr->_decompress_timer, block_item.decompress_time());
- COUNTER_UPDATE(_recvr->_decompress_bytes, block_item.decompress_bytes());
- _recvr->_memory_used_counter->update(-(int64_t)block_byte_size);
- INJECT_MOCK_SLEEP(std::lock_guard<std::mutex> l(_lock));
- sub_blocks_memory_usage(block_byte_size);
- _record_debug_info();
- if (_block_queue.empty() && _source_dependency) {
- if (!_is_cancelled && _num_remaining_senders > 0) {
- _source_dependency->block();
+ Defer release_block {[&]() {
+ COUNTER_UPDATE(_recvr->_deserialize_row_batch_timer,
block_item.deserialize_time());
+ COUNTER_UPDATE(_recvr->_decompress_timer,
block_item.decompress_time());
+ COUNTER_UPDATE(_recvr->_decompress_bytes,
block_item.decompress_bytes());
+ _recvr->_memory_used_counter->update(-(int64_t)block_byte_size);
+ {
+ INJECT_MOCK_SLEEP(std::lock_guard<std::mutex> l(_lock));
+ sub_blocks_memory_usage(block_byte_size);
+ if (_block_queue.empty() && _source_dependency) {
+ if (!_is_cancelled && _num_remaining_senders > 0) {
+ _source_dependency->block();
+ }
+ }
}
- }
+ _record_debug_info();
- if (!_pending_closures.empty()) {
- auto closure_pair = _pending_closures.front();
- closure_pair.first->Run();
- int64_t elapse_time = closure_pair.second.elapsed_time();
- if (_recvr->_max_wait_to_process_time->value() < elapse_time) {
- _recvr->_max_wait_to_process_time->set(elapse_time);
+ if (!_pending_closures.empty()) {
Review Comment:
Fixed in f17eb83b138bfae9ece0fbc02fc196a8bc5e9a30. Callback selection and
removal now hold the queue mutex, and only the detached callback runs after
unlocking. Debug recording and the entry debug-state check also hold that
mutex. Added deterministic success/error cases that enqueue another remote
block and follow remove_sender -> cancel_stream -> cancel while the first
callback runs, then check that each callback executes once.
The old P0 (1083217), cloud P0 (1083219) and external (1083218) logs all
show an ASAN double-free in _record_debug_info concurrent with local
add_block/get_batch, directly matching the missing lock. DEBUG production
compilation and targeted analyzer checks passed; focused DEBUG BE UT is still
building.
--
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]
