Skip to content

Commit

Permalink
1
Browse files Browse the repository at this point in the history
  • Loading branch information
xinyiZzz committed Dec 19, 2023
1 parent 71b7dcf commit 738ab8c
Show file tree
Hide file tree
Showing 3 changed files with 31 additions and 4 deletions.
19 changes: 19 additions & 0 deletions be/src/common/daemon.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@

namespace doris {

CountDownLatch Daemon::_je_purge_dirty_pages_thread_latch {1};

void Daemon::tcmalloc_gc_thread() {
// TODO All cache GC wish to be supported
#if !defined(ADDRESS_SANITIZER) && !defined(LEAK_SANITIZER) && !defined(THREAD_SANITIZER) && \
Expand Down Expand Up @@ -352,6 +354,18 @@ void Daemon::block_spill_gc_thread() {
}
}

void Daemon::je_purge_dirty_pages_thread() const {
_je_purge_dirty_pages_thread_latch.reset(1);
do {
_je_purge_dirty_pages_thread_latch.wait();
if (_is_stopped) {
break;
}
doris::MemInfo::je_purge_all_arena_dirty_pages();
_je_purge_dirty_pages_thread_latch.reset(1);
} while (true);
}

void Daemon::start() {
Status st;
st = Thread::create(
Expand Down Expand Up @@ -381,6 +395,9 @@ void Daemon::start() {
st = Thread::create(
"Daemon", "block_spill_gc_thread", [this]() { this->block_spill_gc_thread(); },
&_threads.emplace_back());
st = Thread::create(
"Daemon", "je_purge_dirty_pages_thread",
[this]() { this->je_purge_dirty_pages_thread(); }, &_threads.emplace_back());
CHECK(st.ok()) << st;
}

Expand All @@ -390,7 +407,9 @@ void Daemon::stop() {
LOG(INFO) << "Doris daemon stop returned since no bg threads latch.";
return;
}
_is_stopped = true;
_stop_background_threads_latch.count_down();
_je_purge_dirty_pages_thread_latch.count_down();
for (auto&& t : _threads) {
if (t) {
t->join();
Expand Down
7 changes: 7 additions & 0 deletions be/src/common/daemon.h
Original file line number Diff line number Diff line change
Expand Up @@ -36,15 +36,22 @@ class Daemon {
// Stop background threads
void stop();

static void count_down_je_purge_dirty_pages_thread_latch() {
_je_purge_dirty_pages_thread_latch.count_down();
}

private:
void tcmalloc_gc_thread();
void memory_maintenance_thread();
void memory_gc_thread();
void memtable_memory_limiter_tracker_refresh_thread();
void calculate_metrics_thread();
void block_spill_gc_thread();
void je_purge_dirty_pages_thread() const;

CountDownLatch _stop_background_threads_latch;
static CountDownLatch _je_purge_dirty_pages_thread_latch;
bool _is_stopped {};
std::vector<scoped_refptr<Thread>> _threads;
};
} // namespace doris
9 changes: 5 additions & 4 deletions be/src/util/mem_info.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
#include <vector>

#include "common/config.h"
#include "common/daemon.h"
#include "common/status.h"
#include "gutil/strings/split.h"
#include "runtime/exec_env.h"
Expand Down Expand Up @@ -129,7 +130,7 @@ bool MemInfo::process_minor_gc() {
std::string pre_sys_mem_available = MemInfo::sys_mem_available_str();

Defer defer {[&]() {
je_purge_all_arena_dirty_pages();
Daemon::count_down_je_purge_dirty_pages_thread_latch();
std::stringstream ss;
profile->pretty_print(&ss);
LOG(INFO) << fmt::format(
Expand All @@ -139,7 +140,7 @@ bool MemInfo::process_minor_gc() {
}};

freed_mem += CacheManager::instance()->for_each_cache_prune_stale(profile.get());
je_purge_all_arena_dirty_pages();
Daemon::count_down_je_purge_dirty_pages_thread_latch();
if (freed_mem > _s_process_minor_gc_size) {
return true;
}
Expand Down Expand Up @@ -180,7 +181,7 @@ bool MemInfo::process_full_gc() {
std::string pre_sys_mem_available = MemInfo::sys_mem_available_str();

Defer defer {[&]() {
je_purge_all_arena_dirty_pages();
Daemon::count_down_je_purge_dirty_pages_thread_latch();
std::stringstream ss;
profile->pretty_print(&ss);
LOG(INFO) << fmt::format(
Expand All @@ -190,7 +191,7 @@ bool MemInfo::process_full_gc() {
}};

freed_mem += CacheManager::instance()->for_each_cache_prune_all(profile.get());
je_purge_all_arena_dirty_pages();
Daemon::count_down_je_purge_dirty_pages_thread_latch();
if (freed_mem > _s_process_full_gc_size) {
return true;
}
Expand Down

0 comments on commit 738ab8c

Please sign in to comment.