diff --git a/be/src/pipeline/pipeline_fragment_context.cpp b/be/src/pipeline/pipeline_fragment_context.cpp index 51cfd73e549d79..3bf54ed7ece00b 100644 --- a/be/src/pipeline/pipeline_fragment_context.cpp +++ b/be/src/pipeline/pipeline_fragment_context.cpp @@ -1031,6 +1031,10 @@ Status PipelineFragmentContext::_create_data_sink(ObjectPool* pool, const TDataS } case TDataSinkType::GROUP_COMMIT_BLOCK_SINK: { DCHECK(thrift_sink.__isset.olap_table_sink); +#ifndef NDEBUG + DCHECK(state->get_query_ctx() != nullptr); + state->get_query_ctx()->query_mem_tracker->is_group_commit_load = true; +#endif _sink.reset( new GroupCommitBlockSinkOperatorX(next_sink_operator_id(), row_desc, output_exprs)); break; @@ -1177,6 +1181,10 @@ Status PipelineFragmentContext::_create_operator(ObjectPool* pool, const TPlanNo break; } case TPlanNodeType::GROUP_COMMIT_SCAN_NODE: { +#ifndef NDEBUG + DCHECK(_query_ctx != nullptr); + _query_ctx->query_mem_tracker->is_group_commit_load = true; +#endif op.reset(new GroupCommitOperatorX(pool, tnode, next_operator_id(), descs, _num_instances)); RETURN_IF_ERROR(cur_pipe->add_operator(op)); if (request.__isset.parallel_instances) { diff --git a/be/src/runtime/memory/mem_tracker_limiter.cpp b/be/src/runtime/memory/mem_tracker_limiter.cpp index cc695a6fdd51e1..6df577f8a50347 100644 --- a/be/src/runtime/memory/mem_tracker_limiter.cpp +++ b/be/src/runtime/memory/mem_tracker_limiter.cpp @@ -114,6 +114,8 @@ MemTrackerLimiter::~MemTrackerLimiter() { "mem tracker not equal to 0 when mem tracker destruct, this usually means that " "memory tracking is inaccurate and SCOPED_ATTACH_TASK and " "SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER are not used correctly. " + "If the log is truncated, search for `Address Sanitizer` in the be.INFO log to see " + "more information." "1. For query and load, memory leaks may have occurred, it is expected that the query " "mem tracker will be bound to the thread context using SCOPED_ATTACH_TASK and " "SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER before all memory alloc and free. " @@ -127,7 +129,7 @@ MemTrackerLimiter::~MemTrackerLimiter() { if (_consumption->current_value() != 0) { // TODO, expect mem tracker equal to 0 at the load/compaction/etc. task end. #ifndef NDEBUG - if (_type == Type::QUERY) { + if (_type == Type::QUERY || (_type == Type::LOAD && !is_group_commit_load)) { std::string err_msg = fmt::format("mem tracker label: {}, consumption: {}, peak consumption: {}, {}.", label(), _consumption->current_value(), _consumption->peak_value(), @@ -140,11 +142,11 @@ MemTrackerLimiter::~MemTrackerLimiter() { } _consumption->set(0); #ifndef NDEBUG - } else if (!_address_sanitizers.empty()) { - LOG(INFO) << "[Address Sanitizer] consumption is 0, but address sanitizers not empty. " - << ", mem tracker label: " << _label - << ", peak consumption: " << _consumption->peak_value() - << print_address_sanitizers(); + } else if (!_address_sanitizers.empty() && !is_group_commit_load) { + LOG(FATAL) << "[Address Sanitizer] consumption is 0, but address sanitizers not empty. " + << ", mem tracker label: " << _label + << ", peak consumption: " << _consumption->peak_value() + << print_address_sanitizers(); #endif } memory_memtrackerlimiter_cnt << -1; @@ -152,17 +154,17 @@ MemTrackerLimiter::~MemTrackerLimiter() { #ifndef NDEBUG void MemTrackerLimiter::add_address_sanitizers(void* buf, size_t size) { - if (_type == Type::QUERY) { + if (_type == Type::QUERY || (_type == Type::LOAD && !is_group_commit_load)) { std::lock_guard l(_address_sanitizers_mtx); auto it = _address_sanitizers.find(buf); if (it != _address_sanitizers.end()) { - LOG(INFO) << "[Address Sanitizer] memory buf repeat add, mem tracker label: " << _label - << ", consumption: " << _consumption->current_value() - << ", peak consumption: " << _consumption->peak_value() << ", buf: " << buf - << ", size: " << size << ", old buf: " << it->first - << ", old size: " << it->second.size - << ", new stack_trace: " << get_stack_trace(1, "DISABLED") - << ", old stack_trace: " << it->second.stack_trace; + _error_address_sanitizers.emplace_back( + fmt::format("[Address Sanitizer] memory buf repeat add, mem tracker label: {}, " + "consumption: {}, peak consumption: {}, buf: {}, size: {}, old " + "buf: {}, old size: {}, new stack_trace: {}, old stack_trace: {}.", + _label, _consumption->current_value(), _consumption->peak_value(), + buf, size, it->first, it->second.size, + get_stack_trace(1, "FULL_WITH_INLINE"), it->second.stack_trace)); } // if alignment not equal to 0, maybe usable_size > size. @@ -174,26 +176,26 @@ void MemTrackerLimiter::add_address_sanitizers(void* buf, size_t size) { } void MemTrackerLimiter::remove_address_sanitizers(void* buf, size_t size) { - if (_type == Type::QUERY) { + if (_type == Type::QUERY || (_type == Type::LOAD && !is_group_commit_load)) { std::lock_guard l(_address_sanitizers_mtx); auto it = _address_sanitizers.find(buf); if (it != _address_sanitizers.end()) { if (it->second.size != size) { - LOG(INFO) << "[Address Sanitizer] free memory buf size inaccurate, mem tracker " - "label: " - << _label << ", consumption: " << _consumption->current_value() - << ", peak consumption: " << _consumption->peak_value() - << ", buf: " << buf << ", size: " << size << ", old buf: " << it->first - << ", old size: " << it->second.size - << ", new stack_trace: " << get_stack_trace(1, "DISABLED") - << ", old stack_trace: " << it->second.stack_trace; + _error_address_sanitizers.emplace_back(fmt::format( + "[Address Sanitizer] free memory buf size inaccurate, mem tracker label: " + "{}, consumption: {}, peak consumption: {}, buf: {}, size: {}, old buf: " + "{}, old size: {}, new stack_trace: {}, old stack_trace: {}.", + _label, _consumption->current_value(), _consumption->peak_value(), buf, + size, it->first, it->second.size, get_stack_trace(1, "FULL_WITH_INLINE"), + it->second.stack_trace)); } _address_sanitizers.erase(buf); } else { - LOG(INFO) << "[Address Sanitizer] memory buf not exist, mem tracker label: " << _label - << ", consumption: " << _consumption->current_value() - << ", peak consumption: " << _consumption->peak_value() << ", buf: " << buf - << ", size: " << size << ", stack_trace: " << get_stack_trace(1, "DISABLED"); + _error_address_sanitizers.emplace_back(fmt::format( + "[Address Sanitizer] memory buf not exist, mem tracker label: {}, consumption: " + "{}, peak consumption: {}, buf: {}, size: {}, stack_trace: {}.", + _label, _consumption->current_value(), _consumption->peak_value(), buf, size, + get_stack_trace(1, "FULL_WITH_INLINE"))); } } } @@ -201,9 +203,20 @@ void MemTrackerLimiter::remove_address_sanitizers(void* buf, size_t size) { std::string MemTrackerLimiter::print_address_sanitizers() { std::lock_guard l(_address_sanitizers_mtx); std::string detail = "[Address Sanitizer]:"; + detail += "\n memory not be freed:"; for (const auto& it : _address_sanitizers) { - detail += fmt::format("\n {}, size {}, strack trace: {}", it.first, it.second.size, - it.second.stack_trace); + auto msg = fmt::format( + "\n [Address Sanitizer] buf not be freed, mem tracker label: {}, consumption: " + "{}, peak consumption: {}, buf: {}, size {}, strack trace: {}", + _label, _consumption->current_value(), _consumption->peak_value(), it.first, + it.second.size, it.second.stack_trace); + LOG(INFO) << msg; + detail += msg; + } + detail += "\n incorrect memory alloc and free:"; + for (const auto& err_msg : _error_address_sanitizers) { + LOG(INFO) << err_msg; + detail += fmt::format("\n {}", err_msg); } return detail; } diff --git a/be/src/runtime/memory/mem_tracker_limiter.h b/be/src/runtime/memory/mem_tracker_limiter.h index e5c5cb1bc0369c..344f3dc92b6670 100644 --- a/be/src/runtime/memory/mem_tracker_limiter.h +++ b/be/src/runtime/memory/mem_tracker_limiter.h @@ -209,6 +209,7 @@ class MemTrackerLimiter final : public MemTracker { void add_address_sanitizers(void* buf, size_t size); void remove_address_sanitizers(void* buf, size_t size); std::string print_address_sanitizers(); + bool is_group_commit_load {false}; #endif std::string debug_string() override { @@ -260,6 +261,7 @@ class MemTrackerLimiter final : public MemTracker { std::mutex _address_sanitizers_mtx; std::unordered_map _address_sanitizers; + std::vector _error_address_sanitizers; #endif };