You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2022/08/30 06:57:44 UTC
[doris] branch revert-12168-20220829_fix_podarray_resetpeak created (now e7231d49e5)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a change to branch revert-12168-20220829_fix_podarray_resetpeak
in repository https://gitbox.apache.org/repos/asf/doris.git
at e7231d49e5 Revert "[enhancement](memtracker) Improve performance of tracking real physical memory of PODArray #12168"
This branch includes the following new commits:
new e7231d49e5 Revert "[enhancement](memtracker) Improve performance of tracking real physical memory of PODArray #12168"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 01/01: Revert "[enhancement](memtracker) Improve performance of tracking real physical memory of PODArray #12168"
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch revert-12168-20220829_fix_podarray_resetpeak
in repository https://gitbox.apache.org/repos/asf/doris.git
commit e7231d49e5a335a5ba5b838e6ea3349cf83db103
Author: yiguolei <67...@qq.com>
AuthorDate: Tue Aug 30 14:57:36 2022 +0800
Revert "[enhancement](memtracker) Improve performance of tracking real physical memory of PODArray #12168"
This reverts commit 8370115cf61ad78e2ee76af53484085a12f676af.
---
be/src/exec/hash_table.cpp | 2 +-
be/src/exec/partitioned_aggregation_node.cc | 6 +++---
be/src/exec/partitioned_hash_table.cc | 2 +-
be/src/runtime/buffered_block_mgr2.cc | 14 +++++++-------
be/src/runtime/exec_env.h | 3 ---
be/src/runtime/exec_env_init.cpp | 1 -
be/src/runtime/mem_pool.cpp | 8 ++++----
be/src/runtime/mem_pool.h | 4 ++--
be/src/runtime/memory/mem_tracker.cpp | 15 +++++++++------
be/src/runtime/memory/mem_tracker_limiter.cpp | 2 +-
be/src/runtime/memory/mem_tracker_limiter.h | 2 +-
be/src/runtime/memory/thread_mem_tracker_mgr.cpp | 6 ++----
be/src/runtime/memory/thread_mem_tracker_mgr.h | 11 ++++-------
be/src/runtime/runtime_state.cpp | 13 +++++--------
be/src/runtime/thread_context.h | 12 ++++++------
be/src/service/doris_main.cpp | 2 +-
be/src/vec/common/pod_array.h | 10 +++++-----
17 files changed, 52 insertions(+), 61 deletions(-)
diff --git a/be/src/exec/hash_table.cpp b/be/src/exec/hash_table.cpp
index 2aa195bebc..b50b03460e 100644
--- a/be/src/exec/hash_table.cpp
+++ b/be/src/exec/hash_table.cpp
@@ -175,7 +175,7 @@ Status HashTable::resize_buckets(int64_t num_buckets) {
int64_t old_num_buckets = _num_buckets;
int64_t delta_bytes = (num_buckets - old_num_buckets) * sizeof(Bucket);
- Status st = thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->check_limit(
+ Status st = thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->check_limit(
delta_bytes);
if (!st) {
LOG_EVERY_N(WARNING, 100) << "resize bucket failed: " << st.to_string();
diff --git a/be/src/exec/partitioned_aggregation_node.cc b/be/src/exec/partitioned_aggregation_node.cc
index 624f7c71ce..16332151b0 100644
--- a/be/src/exec/partitioned_aggregation_node.cc
+++ b/be/src/exec/partitioned_aggregation_node.cc
@@ -911,13 +911,13 @@ Tuple* PartitionedAggregationNode::ConstructIntermediateTuple(
<< "Backend: " << BackendOptions::get_localhost() << ", "
<< "fragment: " << print_id(state_->fragment_instance_id()) << " "
<< "Used: "
- << thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->consumption()
+ << thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->consumption()
<< ", Limit: "
- << thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->limit() << ". "
+ << thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->limit() << ". "
<< "You can change the limit by session variable exec_mem_limit.";
string details = Substitute(str.str(), _id, tuple_data_size);
*status = thread_context()
- ->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()
+ ->_thread_mem_tracker_mgr->limiter_mem_tracker()
->mem_limit_exceeded(state_, details, tuple_data_size);
return nullptr;
}
diff --git a/be/src/exec/partitioned_hash_table.cc b/be/src/exec/partitioned_hash_table.cc
index a662764907..b78622e137 100644
--- a/be/src/exec/partitioned_hash_table.cc
+++ b/be/src/exec/partitioned_hash_table.cc
@@ -307,7 +307,7 @@ Status PartitionedHashTableCtx::ExprValuesCache::Init(RuntimeState* state,
MAX_EXPR_VALUES_ARRAY_SIZE / expr_values_bytes_per_row_));
int mem_usage = MemUsage(capacity_, expr_values_bytes_per_row_, num_exprs_);
- if (UNLIKELY(!thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->check_limit(
+ if (UNLIKELY(!thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->check_limit(
mem_usage))) {
capacity_ = 0;
string details = Substitute(
diff --git a/be/src/runtime/buffered_block_mgr2.cc b/be/src/runtime/buffered_block_mgr2.cc
index fa2d1b70b0..e610cf3803 100644
--- a/be/src/runtime/buffered_block_mgr2.cc
+++ b/be/src/runtime/buffered_block_mgr2.cc
@@ -251,7 +251,7 @@ int64_t BufferedBlockMgr2::remaining_unreserved_buffers() const {
int64_t num_buffers =
_free_io_buffers.size() + _unpinned_blocks.size() + _non_local_outstanding_writes;
num_buffers +=
- thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->spare_capacity() /
+ thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->spare_capacity() /
max_block_size();
num_buffers -= _unfullfilled_reserved_buffers;
return num_buffers;
@@ -358,9 +358,9 @@ Status BufferedBlockMgr2::get_new_block(Client* client, Block* unpin_block, Bloc
if (len > 0 && len < _max_block_size) {
DCHECK(unpin_block == nullptr);
- Status st = thread_context()
- ->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()
- ->check_limit(len);
+ Status st =
+ thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->check_limit(
+ len);
WARN_IF_ERROR(st, "get_new_block failed");
if (st) {
client->_tracker->consume(len);
@@ -986,7 +986,7 @@ Status BufferedBlockMgr2::find_buffer(unique_lock<mutex>& lock, BufferDescriptor
// First, try to allocate a new buffer.
if (_free_io_buffers.size() < _block_write_threshold &&
- thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->check_limit(
+ thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->check_limit(
_max_block_size)) {
_mem_tracker->consume(_max_block_size);
uint8_t* new_buffer = new uint8_t[_max_block_size];
@@ -1155,9 +1155,9 @@ string BufferedBlockMgr2::debug_internal() const {
<< " Unfullfilled reserved buffers: " << _unfullfilled_reserved_buffers << endl
<< " BUffer Block Mgr Used memory: " << _mem_tracker->consumption()
<< " Instance remaining memory: "
- << thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->spare_capacity()
+ << thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->spare_capacity()
<< " (#blocks="
- << (thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->spare_capacity() /
+ << (thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->spare_capacity() /
_max_block_size)
<< ")" << endl
<< " Block write threshold: " << _block_write_threshold;
diff --git a/be/src/runtime/exec_env.h b/be/src/runtime/exec_env.h
index a86889f35d..cd32b9beff 100644
--- a/be/src/runtime/exec_env.h
+++ b/be/src/runtime/exec_env.h
@@ -114,10 +114,8 @@ public:
}
std::shared_ptr<MemTrackerLimiter> process_mem_tracker() { return _process_mem_tracker; }
- MemTrackerLimiter* process_mem_tracker_raw() { return _process_mem_tracker_raw; }
void set_process_mem_tracker(const std::shared_ptr<MemTrackerLimiter>& tracker) {
_process_mem_tracker = tracker;
- _process_mem_tracker_raw = tracker.get();
}
std::shared_ptr<MemTrackerLimiter> query_pool_mem_tracker() { return _query_pool_mem_tracker; }
std::shared_ptr<MemTrackerLimiter> load_pool_mem_tracker() { return _load_pool_mem_tracker; }
@@ -196,7 +194,6 @@ private:
// The ancestor for all trackers. Every tracker is visible from the process down.
// Not limit total memory by process tracker, and it's just used to track virtual memory of process.
std::shared_ptr<MemTrackerLimiter> _process_mem_tracker;
- MemTrackerLimiter* _process_mem_tracker_raw;
// The ancestor for all querys tracker.
std::shared_ptr<MemTrackerLimiter> _query_pool_mem_tracker;
// The ancestor for all load tracker.
diff --git a/be/src/runtime/exec_env_init.cpp b/be/src/runtime/exec_env_init.cpp
index 0aedb39b49..0fab11f3ee 100644
--- a/be/src/runtime/exec_env_init.cpp
+++ b/be/src/runtime/exec_env_init.cpp
@@ -205,7 +205,6 @@ Status ExecEnv::_init_mem_tracker() {
}
_process_mem_tracker =
std::make_shared<MemTrackerLimiter>(global_memory_limit_bytes, "Process");
- _process_mem_tracker_raw = _process_mem_tracker.get();
thread_context()->_thread_mem_tracker_mgr->init();
thread_context()->_thread_mem_tracker_mgr->set_check_attach(false);
#if defined(USE_MEM_TRACKER) && !defined(__SANITIZE_ADDRESS__) && !defined(ADDRESS_SANITIZER) && \
diff --git a/be/src/runtime/mem_pool.cpp b/be/src/runtime/mem_pool.cpp
index 768cdccb18..6fae53bba4 100644
--- a/be/src/runtime/mem_pool.cpp
+++ b/be/src/runtime/mem_pool.cpp
@@ -67,7 +67,7 @@ MemPool::~MemPool() {
ChunkAllocator::instance()->free(chunk.chunk);
}
THREAD_MEM_TRACKER_TRANSFER_FROM(total_bytes_released - peak_allocated_bytes_,
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
if (_mem_tracker) _mem_tracker->release(total_bytes_released);
DorisMetrics::instance()->memory_pool_bytes_total->increment(-total_bytes_released);
}
@@ -89,7 +89,7 @@ void MemPool::free_all() {
ChunkAllocator::instance()->free(chunk.chunk);
}
THREAD_MEM_TRACKER_TRANSFER_FROM(total_bytes_released - peak_allocated_bytes_,
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
if (_mem_tracker) _mem_tracker->release(total_bytes_released);
chunks_.clear();
next_chunk_size_ = INITIAL_CHUNK_SIZE;
@@ -141,7 +141,7 @@ Status MemPool::find_chunk(size_t min_size, bool check_limits) {
chunk_size = BitUtil::RoundUpToPowerOfTwo(chunk_size);
if (check_limits &&
- !thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->check_limit(
+ !thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->check_limit(
chunk_size)) {
return Status::MemoryAllocFailed("MemPool find new chunk {} bytes faild, exceed limit",
chunk_size);
@@ -150,7 +150,7 @@ Status MemPool::find_chunk(size_t min_size, bool check_limits) {
// Allocate a new chunk. Return early if allocate fails.
Chunk chunk;
RETURN_IF_ERROR(ChunkAllocator::instance()->allocate(chunk_size, &chunk));
- THREAD_MEM_TRACKER_TRANSFER_TO(chunk_size, ExecEnv::GetInstance()->process_mem_tracker_raw());
+ THREAD_MEM_TRACKER_TRANSFER_TO(chunk_size, ExecEnv::GetInstance()->process_mem_tracker().get());
if (_mem_tracker) _mem_tracker->consume(chunk_size);
ASAN_POISON_MEMORY_REGION(chunk.data, chunk_size);
// Put it before the first free chunk. If no free chunks, it goes at the end.
diff --git a/be/src/runtime/mem_pool.h b/be/src/runtime/mem_pool.h
index 98e2d5a271..5ad56ef934 100644
--- a/be/src/runtime/mem_pool.h
+++ b/be/src/runtime/mem_pool.h
@@ -212,9 +212,9 @@ private:
bool check_integrity(bool check_current_chunk_empty);
void reset_peak() {
- if (total_allocated_bytes_ - peak_allocated_bytes_ > 65536) {
+ if (total_allocated_bytes_ - peak_allocated_bytes_ > 4096) {
THREAD_MEM_TRACKER_TRANSFER_FROM(total_allocated_bytes_ - peak_allocated_bytes_,
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
peak_allocated_bytes_ = total_allocated_bytes_;
}
}
diff --git a/be/src/runtime/memory/mem_tracker.cpp b/be/src/runtime/memory/mem_tracker.cpp
index b282b87758..5ef28ee7da 100644
--- a/be/src/runtime/memory/mem_tracker.cpp
+++ b/be/src/runtime/memory/mem_tracker.cpp
@@ -58,12 +58,15 @@ MemTracker::MemTracker(const std::string& label, RuntimeProfile* profile) {
_consumption = profile->AddSharedHighWaterMarkCounter(COUNTER_NAME, TUnit::BYTES);
}
- DCHECK(thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker() != nullptr);
- _label = fmt::format(
- "{} | {}", label,
- thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->label());
- _bind_group_num =
- thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->group_num();
+ if (thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()) {
+ _label = fmt::format(
+ "{} | {}", label,
+ thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->label());
+ } else {
+ _label = label + " | ";
+ }
+
+ _bind_group_num = thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->group_num();
{
std::lock_guard<std::mutex> l(mem_tracker_pool[_bind_group_num].group_lock);
_tracker_group_it = mem_tracker_pool[_bind_group_num].trackers.insert(
diff --git a/be/src/runtime/memory/mem_tracker_limiter.cpp b/be/src/runtime/memory/mem_tracker_limiter.cpp
index 7597ecfb39..13e888959a 100644
--- a/be/src/runtime/memory/mem_tracker_limiter.cpp
+++ b/be/src/runtime/memory/mem_tracker_limiter.cpp
@@ -303,7 +303,7 @@ Status MemTrackerLimiter::mem_limit_exceeded(const std::string& msg,
// The limit of the current tracker and parents is less than 0, the consume will not fail,
// and the current process memory has no excess limit.
detail += fmt::format("unknown exceed reason, executing_msg:<{}>", msg);
- print_log_usage_tracker = ExecEnv::GetInstance()->process_mem_tracker_raw();
+ print_log_usage_tracker = ExecEnv::GetInstance()->process_mem_tracker().get();
}
auto st = MemTrackerLimiter::mem_limit_exceeded_construct(detail);
if (print_log_usage_tracker != nullptr)
diff --git a/be/src/runtime/memory/mem_tracker_limiter.h b/be/src/runtime/memory/mem_tracker_limiter.h
index 9cbc688b71..65506964a8 100644
--- a/be/src/runtime/memory/mem_tracker_limiter.h
+++ b/be/src/runtime/memory/mem_tracker_limiter.h
@@ -73,7 +73,7 @@ public:
auto st = Status::MemoryLimitExceeded(
"process memory used {} exceed limit {}, failed_alloc_size={}",
PerfCounters::get_vm_rss(), MemInfo::mem_limit(), bytes);
- ExecEnv::GetInstance()->process_mem_tracker_raw()->print_log_usage(st.get_error_msg());
+ ExecEnv::GetInstance()->process_mem_tracker()->print_log_usage(st.get_error_msg());
return st;
}
return Status::OK();
diff --git a/be/src/runtime/memory/thread_mem_tracker_mgr.cpp b/be/src/runtime/memory/thread_mem_tracker_mgr.cpp
index 5cbc76eadc..025a8248d4 100644
--- a/be/src/runtime/memory/thread_mem_tracker_mgr.cpp
+++ b/be/src/runtime/memory/thread_mem_tracker_mgr.cpp
@@ -32,7 +32,6 @@ void ThreadMemTrackerMgr::attach_limiter_tracker(
_task_id = task_id;
_fragment_instance_id = fragment_instance_id;
_limiter_tracker = mem_tracker;
- _limiter_tracker_raw = mem_tracker.get();
}
void ThreadMemTrackerMgr::detach_limiter_tracker() {
@@ -40,7 +39,6 @@ void ThreadMemTrackerMgr::detach_limiter_tracker() {
_task_id = "";
_fragment_instance_id = TUniqueId();
_limiter_tracker = ExecEnv::GetInstance()->process_mem_tracker();
- _limiter_tracker_raw = ExecEnv::GetInstance()->process_mem_tracker_raw();
}
void ThreadMemTrackerMgr::exceeded_cancel_task(const std::string& cancel_details) {
@@ -56,9 +54,9 @@ void ThreadMemTrackerMgr::exceeded(Status failed_try_consume_st) {
_cb_func();
}
if (is_attach_query()) {
- auto st = _limiter_tracker_raw->mem_limit_exceeded(
+ auto st = _limiter_tracker->mem_limit_exceeded(
fmt::format("exec node:<{}>", last_consumer_tracker()),
- _limiter_tracker_raw->parent().get(), failed_try_consume_st);
+ _limiter_tracker->parent().get(), failed_try_consume_st);
exceeded_cancel_task(st.get_error_msg());
_check_limit = false; // Make sure it will only be canceled once
}
diff --git a/be/src/runtime/memory/thread_mem_tracker_mgr.h b/be/src/runtime/memory/thread_mem_tracker_mgr.h
index 13ecea46b4..5e02e2cd77 100644
--- a/be/src/runtime/memory/thread_mem_tracker_mgr.h
+++ b/be/src/runtime/memory/thread_mem_tracker_mgr.h
@@ -50,7 +50,7 @@ public:
// only for tcmalloc hook
static void consume_no_attach(int64_t size) {
- ExecEnv::GetInstance()->process_mem_tracker_raw()->consume(size);
+ ExecEnv::GetInstance()->process_mem_tracker()->consume(size);
}
// After thread initialization, calling `init` again must call `clear_untracked_mems` first
@@ -84,7 +84,6 @@ public:
bool is_attach_query() { return _fragment_instance_id != TUniqueId(); }
std::shared_ptr<MemTrackerLimiter> limiter_mem_tracker() { return _limiter_tracker; }
- MemTrackerLimiter* limiter_mem_tracker_raw() { return _limiter_tracker_raw; }
void set_check_limit(bool check_limit) { _check_limit = check_limit; }
void set_check_attach(bool check_attach) { _check_attach = check_attach; }
@@ -114,7 +113,6 @@ private:
int64_t _untracked_mem = 0;
std::shared_ptr<MemTrackerLimiter> _limiter_tracker;
- MemTrackerLimiter* _limiter_tracker_raw;
std::vector<MemTracker*> _consumer_tracker_stack;
// If true, call memtracker try_consume, otherwise call consume.
@@ -131,7 +129,6 @@ inline void ThreadMemTrackerMgr::init() {
DCHECK(_consumer_tracker_stack.empty());
_task_id = "";
_limiter_tracker = ExecEnv::GetInstance()->process_mem_tracker();
- _limiter_tracker_raw = ExecEnv::GetInstance()->process_mem_tracker_raw();
_check_limit = true;
}
@@ -182,15 +179,15 @@ inline void ThreadMemTrackerMgr::flush_untracked_mem() {
// DCHECK(!_check_attach || btls_key != EMPTY_BTLS_KEY ||
// _limiter_tracker->label() != "Process");
#endif
- Status st = _limiter_tracker_raw->try_consume(_untracked_mem);
+ Status st = _limiter_tracker->try_consume(_untracked_mem);
if (!st) {
// The memory has been allocated, so when TryConsume fails, need to continue to complete
// the consume to ensure the accuracy of the statistics.
- _limiter_tracker_raw->consume(_untracked_mem);
+ _limiter_tracker->consume(_untracked_mem);
exceeded(st);
}
} else {
- _limiter_tracker_raw->consume(_untracked_mem);
+ _limiter_tracker->consume(_untracked_mem);
}
for (auto tracker : _consumer_tracker_stack) {
tracker->consume(_untracked_mem);
diff --git a/be/src/runtime/runtime_state.cpp b/be/src/runtime/runtime_state.cpp
index b07a8cda87..e8a5e2846c 100644
--- a/be/src/runtime/runtime_state.cpp
+++ b/be/src/runtime/runtime_state.cpp
@@ -203,14 +203,13 @@ Status RuntimeState::init(const TUniqueId& fragment_instance_id, const TQueryOpt
Status RuntimeState::init_mem_trackers(const TUniqueId& query_id) {
bool has_query_mem_tracker = _query_options.__isset.mem_limit && (_query_options.mem_limit > 0);
int64_t bytes_limit = has_query_mem_tracker ? _query_options.mem_limit : -1;
- if (bytes_limit > ExecEnv::GetInstance()->process_mem_tracker_raw()->limit()) {
+ if (bytes_limit > ExecEnv::GetInstance()->process_mem_tracker()->limit()) {
VLOG_NOTICE << "Query memory limit " << PrettyPrinter::print(bytes_limit, TUnit::BYTES)
<< " exceeds process memory limit of "
- << PrettyPrinter::print(
- ExecEnv::GetInstance()->process_mem_tracker_raw()->limit(),
- TUnit::BYTES)
+ << PrettyPrinter::print(ExecEnv::GetInstance()->process_mem_tracker()->limit(),
+ TUnit::BYTES)
<< ". Using process memory limit instead";
- bytes_limit = ExecEnv::GetInstance()->process_mem_tracker_raw()->limit();
+ bytes_limit = ExecEnv::GetInstance()->process_mem_tracker()->limit();
}
auto mem_tracker_counter = ADD_COUNTER(&_profile, "MemoryLimit", TUnit::BYTES);
mem_tracker_counter->set(bytes_limit);
@@ -300,9 +299,7 @@ Status RuntimeState::set_mem_limit_exceeded(const std::string& msg) {
Status RuntimeState::check_query_state(const std::string& msg) {
// TODO: it would be nice if this also checked for cancellation, but doing so breaks
// cases where we use Status::Cancelled("Cancelled") to indicate that the limit was reached.
- if (thread_context()
- ->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()
- ->any_limit_exceeded()) {
+ if (thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->any_limit_exceeded()) {
RETURN_LIMIT_EXCEEDED(this, msg);
}
return query_status();
diff --git a/be/src/runtime/thread_context.h b/be/src/runtime/thread_context.h
index b097910f24..1fe2f82530 100644
--- a/be/src/runtime/thread_context.h
+++ b/be/src/runtime/thread_context.h
@@ -136,7 +136,7 @@ public:
const std::shared_ptr<MemTrackerLimiter>& mem_tracker) {
DCHECK((_type == TaskType::UNKNOWN || _type == TaskType::BRPC) && _task_id == "")
<< ",new tracker label: " << mem_tracker->label() << ",old tracker label: "
- << _thread_mem_tracker_mgr->limiter_mem_tracker_raw()->label();
+ << _thread_mem_tracker_mgr->limiter_mem_tracker()->label();
DCHECK(type != TaskType::UNKNOWN);
_type = type;
_task_id = task_id;
@@ -256,15 +256,15 @@ public:
doris::thread_context()->_thread_mem_tracker_mgr->consume(size)
#define RELEASE_THREAD_MEM_TRACKER(size) \
doris::thread_context()->_thread_mem_tracker_mgr->consume(-size)
-#define THREAD_MEM_TRACKER_TRANSFER_TO(size, tracker) \
- doris::thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw()->transfer_to( \
- size, tracker)
+#define THREAD_MEM_TRACKER_TRANSFER_TO(size, tracker) \
+ doris::thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker()->transfer_to(size, \
+ tracker)
#define THREAD_MEM_TRACKER_TRANSFER_FROM(size, tracker) \
tracker->transfer_to( \
- size, doris::thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker_raw())
+ size, doris::thread_context()->_thread_mem_tracker_mgr->limiter_mem_tracker().get())
#define RETURN_LIMIT_EXCEEDED(state, msg, ...) \
return doris::thread_context() \
- ->_thread_mem_tracker_mgr->limiter_mem_tracker_raw() \
+ ->_thread_mem_tracker_mgr->limiter_mem_tracker() \
->mem_limit_exceeded( \
state, \
fmt::format("exec node:<{}>, {}", \
diff --git a/be/src/service/doris_main.cpp b/be/src/service/doris_main.cpp
index 680778989c..f6fa80e52c 100644
--- a/be/src/service/doris_main.cpp
+++ b/be/src/service/doris_main.cpp
@@ -502,7 +502,7 @@ int main(int argc, char** argv) {
// this will cause coredump for ASAN build when running regression test,
// disable temporarily.
doris::ExecEnv::GetInstance()->task_pool_mem_tracker_registry()->logout_task_mem_tracker();
- doris::ExecEnv::GetInstance()->process_mem_tracker_raw()->enable_print_log_usage();
+ doris::ExecEnv::GetInstance()->process_mem_tracker()->enable_print_log_usage();
sleep(1);
}
diff --git a/be/src/vec/common/pod_array.h b/be/src/vec/common/pod_array.h
index 2f842040a6..84ced225e1 100644
--- a/be/src/vec/common/pod_array.h
+++ b/be/src/vec/common/pod_array.h
@@ -113,9 +113,9 @@ protected:
}
inline void reset_peak() {
- if (UNLIKELY(c_end - c_end_peak > 65536)) {
+ if (UNLIKELY(c_end - c_end_peak > 4096)) {
THREAD_MEM_TRACKER_TRANSFER_FROM(c_end - c_end_peak,
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
c_end_peak = c_end;
}
}
@@ -127,7 +127,7 @@ protected:
template <typename... TAllocatorParams>
void alloc(size_t bytes, TAllocatorParams&&... allocator_params) {
THREAD_MEM_TRACKER_TRANSFER_TO(bytes - pad_right - pad_left,
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
c_start = c_end = c_end_peak =
reinterpret_cast<char*>(TAllocator::alloc(
bytes, std::forward<TAllocatorParams>(allocator_params)...)) +
@@ -144,7 +144,7 @@ protected:
TAllocator::free(c_start - pad_left, allocated_bytes());
THREAD_MEM_TRACKER_TRANSFER_FROM(c_end_of_storage - c_end_peak,
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
}
template <typename... TAllocatorParams>
@@ -157,7 +157,7 @@ protected:
unprotect();
THREAD_MEM_TRACKER_TRANSFER_TO(bytes - allocated_bytes(),
- ExecEnv::GetInstance()->process_mem_tracker_raw());
+ ExecEnv::GetInstance()->process_mem_tracker().get());
ptrdiff_t end_diff = c_end - c_start;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org