database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[bugfix] fix tcmalooc hook cancel deadlock (#10514)

Browse Source
This commit is contained in:
Kidd
2022-07-01 10:41:59 +08:00
committed by GitHub
parent a9d23ce337
commit 659e863bd7
3 changed files with 16 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
be/src/runtime/data_stream_recvr.cc
View File

@ -203,6 +203,9 @@ Status DataStreamRecvr::SenderQueue::get_batch(RowBatch** next_batch) {
void DataStreamRecvr::SenderQueue::add_batch(const PRowBatch& pb_batch, int be_number,
int64_t packet_seq,
::google::protobuf::Closure** done) {
// Avoid deadlock when calling SenderQueue::cancel() in tcmalloc hook,
// limit memory via DataStreamRecvr::exceeds_limit.
STOP_CHECK_LIMIT_THREAD_LOCAL_MEM_TRACKER();
lock_guard<mutex> l(_lock);
if (_is_cancelled) {
return;
@ -272,6 +275,9 @@ void DataStreamRecvr::SenderQueue::add_batch(const PRowBatch& pb_batch, int be_n
}
void DataStreamRecvr::SenderQueue::add_batch(RowBatch* batch, bool use_move) {
// Avoid deadlock when calling SenderQueue::cancel() in tcmalloc hook,
// limit memory via DataStreamRecvr::exceeds_limit.
STOP_CHECK_LIMIT_THREAD_LOCAL_MEM_TRACKER();
unique_lock<mutex> l(_lock);
if (_is_cancelled) {
return;

10
be/src/runtime/exec_env_init.cpp
View File

@ -182,14 +182,12 @@ Status ExecEnv::_init_mem_tracker() {
global_memory_limit_bytes = MemInfo::physical_mem();
}
MemTracker::get_process_tracker()->set_limit(global_memory_limit_bytes);
_query_pool_mem_tracker = MemTracker::create_tracker(global_memory_limit_bytes, "QueryPool",
MemTracker::get_process_tracker(),
MemTrackerLevel::OVERVIEW);
_query_pool_mem_tracker = MemTracker::create_tracker(
-1, "QueryPool", MemTracker::get_process_tracker(), MemTrackerLevel::OVERVIEW);
REGISTER_HOOK_METRIC(query_mem_consumption,
[this]() { return _query_pool_mem_tracker->consumption(); });
_load_pool_mem_tracker = MemTracker::create_tracker(global_memory_limit_bytes, "LoadPool",
MemTracker::get_process_tracker(),
MemTrackerLevel::OVERVIEW);
_load_pool_mem_tracker = MemTracker::create_tracker(
-1, "LoadPool", MemTracker::get_process_tracker(), MemTrackerLevel::OVERVIEW);
REGISTER_HOOK_METRIC(load_mem_consumption,
[this]() { return _load_pool_mem_tracker->consumption(); });
LOG(INFO) << "Using global memory limit: "

6
be/src/vec/runtime/vdata_stream_recvr.cpp
View File

@ -90,6 +90,9 @@ Status VDataStreamRecvr::SenderQueue::get_batch(Block** next_block) {
void VDataStreamRecvr::SenderQueue::add_block(const PBlock& pblock, int be_number,
int64_t packet_seq,
::google::protobuf::Closure** done) {
// Avoid deadlock when calling SenderQueue::cancel() in tcmalloc hook,
// limit memory via DataStreamRecvr::exceeds_limit.
STOP_CHECK_LIMIT_THREAD_LOCAL_MEM_TRACKER();
std::lock_guard<std::mutex> l(_lock);
if (_is_cancelled) {
return;
@ -140,6 +143,9 @@ void VDataStreamRecvr::SenderQueue::add_block(const PBlock& pblock, int be_numbe
}
void VDataStreamRecvr::SenderQueue::add_block(Block* block, bool use_move) {
// Avoid deadlock when calling SenderQueue::cancel() in tcmalloc hook,
// limit memory via DataStreamRecvr::exceeds_limit.
STOP_CHECK_LIMIT_THREAD_LOCAL_MEM_TRACKER();
std::unique_lock<std::mutex> l(_lock);
if (_is_cancelled) {
return;