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 2024/09/20 12:25:04 UTC
(doris) branch spill_and_reserve updated: flush memtable before spill disk
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch spill_and_reserve
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/spill_and_reserve by this push:
new 66d7d786dc3 flush memtable before spill disk
66d7d786dc3 is described below
commit 66d7d786dc309801c4209e6fe4d11756a62247a6
Author: yiguolei <yi...@gmail.com>
AuthorDate: Thu Sep 19 16:46:30 2024 +0800
flush memtable before spill disk
---
be/src/olap/memtable_memory_limiter.cpp | 32 ++++++++++--
be/src/olap/memtable_memory_limiter.h | 6 ++-
be/src/olap/memtable_writer.h | 8 +++
be/src/runtime/workload_group/workload_group.h | 20 ++++++++
.../workload_group/workload_group_manager.cpp | 60 ++++++++++++++++++++--
.../workload_group/workload_group_manager.h | 9 ++++
6 files changed, 127 insertions(+), 8 deletions(-)
diff --git a/be/src/olap/memtable_memory_limiter.cpp b/be/src/olap/memtable_memory_limiter.cpp
index 213d5aaab9a..d5b2a587702 100644
--- a/be/src/olap/memtable_memory_limiter.cpp
+++ b/be/src/olap/memtable_memory_limiter.cpp
@@ -22,6 +22,7 @@
#include "common/config.h"
#include "olap/memtable.h"
#include "olap/memtable_writer.h"
+#include "runtime/workload_group/workload_group_manager.h"
#include "util/doris_metrics.h"
#include "util/mem_info.h"
#include "util/metrics.h"
@@ -139,7 +140,7 @@ void MemTableMemoryLimiter::handle_memtable_flush() {
<< ", active: " << PrettyPrinter::print_bytes(_active_mem_usage)
<< ", queue: " << PrettyPrinter::print_bytes(_queue_mem_usage)
<< ", flush: " << PrettyPrinter::print_bytes(_flush_mem_usage);
- _flush_active_memtables(need_flush);
+ _flush_active_memtables(0, need_flush);
}
} while (_hard_limit_reached());
g_memtable_memory_limit_waiting_threads << -1;
@@ -149,7 +150,13 @@ void MemTableMemoryLimiter::handle_memtable_flush() {
LOG(INFO) << "waited " << time_ms << " ms for memtable memory limit";
}
-void MemTableMemoryLimiter::_flush_active_memtables(int64_t need_flush) {
+void MemTableMemoryLimiter::flush_workload_group_memtables(uint64_t wg_id,
+ int64_t need_flush_bytes) {
+ std::unique_lock<std::mutex> l(_lock);
+ _flush_active_memtables(wg_id, need_flush_bytes);
+}
+
+void MemTableMemoryLimiter::_flush_active_memtables(uint64_t wg_id, int64_t need_flush) {
if (need_flush <= 0) {
return;
}
@@ -181,6 +188,10 @@ void MemTableMemoryLimiter::_flush_active_memtables(int64_t need_flush) {
if (w == nullptr) {
continue;
}
+ // If wg id is specified, but wg id not match, then not need flush
+ if (wg_id != 0 && w->workload_group_id() != wg_id) {
+ continue;
+ }
int64_t mem = w->active_memtable_mem_consumption();
if (mem < sort_mem * 0.9) {
// if the memtable writer just got flushed, don't flush it again
@@ -234,17 +245,30 @@ void MemTableMemoryLimiter::_refresh_mem_tracker() {
_flush_mem_usage = 0;
_queue_mem_usage = 0;
_active_mem_usage = 0;
+ std::map<uint64_t, doris::MemtableUsage> wg_mem_usages;
_active_writers.clear();
for (auto it = _writers.begin(); it != _writers.end();) {
if (auto writer = it->lock()) {
+ if (wg_mem_usages.find(writer->workload_group_id()) == wg_mem_usages.end()) {
+ wg_mem_usages.insert({writer->workload_group_id(), {0, 0, 0}});
+ }
+ auto& wg_mem_usage = wg_mem_usages.find(writer->workload_group_id())->second;
+
// The memtable is currently used by writer to insert blocks.
auto active_usage = writer->active_memtable_mem_consumption();
+ wg_mem_usage.active_mem_usage += active_usage;
_active_mem_usage += active_usage;
if (active_usage > 0) {
_active_writers.push_back(writer);
}
- _flush_mem_usage += writer->mem_consumption(MemType::FLUSH);
- _queue_mem_usage += writer->mem_consumption(MemType::WRITE_FINISHED);
+
+ auto flush_usage = writer->mem_consumption(MemType::FLUSH);
+ wg_mem_usage.flush_mem_usage += flush_usage;
+ _flush_mem_usage += flush_usage;
+
+ auto write_usage = writer->mem_consumption(MemType::WRITE_FINISHED);
+ wg_mem_usage.queue_mem_usage += write_usage;
+ _queue_mem_usage += write_usage;
++it;
} else {
*it = std::move(_writers.back());
diff --git a/be/src/olap/memtable_memory_limiter.h b/be/src/olap/memtable_memory_limiter.h
index 1e32cb165e4..143edaa8fe5 100644
--- a/be/src/olap/memtable_memory_limiter.h
+++ b/be/src/olap/memtable_memory_limiter.h
@@ -39,8 +39,12 @@ public:
// check if the total mem consumption exceeds limit.
// If yes, it will flush memtable to try to reduce memory consumption.
+ // Every write operation will call this API to check if need flush memtable OR hang
+ // when memory is not available.
void handle_memtable_flush();
+ void flush_workload_group_memtables(uint64_t wg_id, int64_t need_flush_bytes);
+
void register_writer(std::weak_ptr<MemTableWriter> writer);
void refresh_mem_tracker();
@@ -57,7 +61,7 @@ private:
bool _hard_limit_reached();
bool _load_usage_low();
int64_t _need_flush();
- void _flush_active_memtables(int64_t need_flush);
+ void _flush_active_memtables(uint64_t wg_id, int64_t need_flush);
void _refresh_mem_tracker();
std::mutex _lock;
diff --git a/be/src/olap/memtable_writer.h b/be/src/olap/memtable_writer.h
index ec44348b4a9..ae06a04e86d 100644
--- a/be/src/olap/memtable_writer.h
+++ b/be/src/olap/memtable_writer.h
@@ -106,6 +106,14 @@ public:
uint64_t flush_running_count() const;
+ uint64_t workload_group_id() const {
+ auto wg = _query_thread_context.wg_wptr.lock();
+ if (wg != nullptr) {
+ return wg->id();
+ }
+ return 0;
+ }
+
private:
// push a full memtable to flush executor
Status _flush_memtable_async();
diff --git a/be/src/runtime/workload_group/workload_group.h b/be/src/runtime/workload_group/workload_group.h
index fa7fc787870..abbb585692b 100644
--- a/be/src/runtime/workload_group/workload_group.h
+++ b/be/src/runtime/workload_group/workload_group.h
@@ -135,6 +135,20 @@ public:
_spill_high_watermark.load(std::memory_order_relaxed) / 100));
}
+ void update_load_mem_usage(int64_t active_bytes, int64_t queue_bytes, int64_t flush_bytes) {
+ std::unique_lock<std::shared_mutex> wlock(_mutex);
+ _active_mem_usage = active_bytes;
+ _queue_mem_usage = queue_bytes;
+ _flush_mem_usage = flush_bytes;
+ }
+
+ void get_load_mem_usage(int64_t* active_bytes, int64_t* queue_bytes, int64_t* flush_bytes) {
+ std::shared_lock<std::shared_mutex> r_lock(_mutex);
+ *active_bytes += _active_mem_usage;
+ *queue_bytes += _queue_mem_usage;
+ *flush_bytes += _flush_mem_usage;
+ }
+
std::string debug_string() const;
std::string memory_debug_string() const;
@@ -225,6 +239,12 @@ private:
std::string _name;
int64_t _version;
int64_t _memory_limit; // bytes
+
+ // memory used by load memtable
+ int64_t _active_mem_usage = 0;
+ int64_t _queue_mem_usage = 0;
+ int64_t _flush_mem_usage = 0;
+
// `weighted_memory_limit` less than or equal to _memory_limit, calculate after exclude public memory.
// more detailed description in `refresh_wg_weighted_memory_limit`.
std::atomic<int64_t> _weighted_memory_limit {0}; //
diff --git a/be/src/runtime/workload_group/workload_group_manager.cpp b/be/src/runtime/workload_group/workload_group_manager.cpp
index 6fe8c7a51eb..9baf193d721 100644
--- a/be/src/runtime/workload_group/workload_group_manager.cpp
+++ b/be/src/runtime/workload_group/workload_group_manager.cpp
@@ -404,11 +404,48 @@ void WorkloadGroupMgr::handle_paused_queries() {
}
// 2. If memtable size larger than 10% of wg's limit, then flush memtable and wait.
+ MemTableMemoryLimiter* memtable_limiter =
+ doris::ExecEnv::GetInstance()->memtable_memory_limiter();
+ // Not use memlimit, should use high water mark.
+ int64_t wg_high_water_mark_limit =
+ wg->memory_limit() * wg->spill_threshold_high_water_mark() / 100;
+ int64_t memtable_active_bytes = 0;
+ int64_t memtable_write_bytes = 0;
+ int64_t memtable_flush_bytes = 0;
+ wg->get_load_mem_usage(&memtable_active_bytes, &memtable_write_bytes,
+ &memtable_flush_bytes);
+ // TODO: should add a signal in memtable limiter to prevent new batch
+ // For example, streamload, it will not reserve many memory, but it will occupy many memtable memory.
+ // TODO: 0.2 should be a workload group properties. For example, the group is optimized for load,then the value
+ // should be larged, if the group is optimized for query, then the value should be smaller.
+ int64_t max_wg_memtable_bytes =
+ std::max<int64_t>(100 * 1024 * 1024, (int64_t)(wg_high_water_mark_limit * 0.2));
+ if (memtable_active_bytes + memtable_write_bytes + memtable_flush_bytes >
+ max_wg_memtable_bytes) {
+ // There are many table in flush queue, just waiting them flush finished.
+ if (memtable_write_bytes + memtable_flush_bytes > max_wg_memtable_bytes / 2) {
+ LOG_EVERY_T(INFO, 60)
+ << wg->name() << " load memtable size is: " << memtable_active_bytes << ", "
+ << memtable_write_bytes << ", " << memtable_flush_bytes
+ << ", wait for flush finished to release more memory";
+ continue;
+ } else {
+ // Flush 50% active bytes, it means flush some memtables(currently written) to flush queue.
+ memtable_limiter->flush_workload_group_memtables(
+ wg->id(), (int64_t)(memtable_active_bytes * 0.5));
+ LOG_EVERY_T(INFO, 60)
+ << wg->name() << " load memtable size is: " << memtable_active_bytes << ", "
+ << memtable_write_bytes << ", " << memtable_flush_bytes
+ << ", flush half of active memtable to revoke memory";
+ continue;
+ }
+ }
// If the wg enable memory overcommit, then not spill, just cancel query.
- if (wg->enable_memory_overcommit()) {
- continue;
- }
+ // if (wg->enable_memory_overcommit()) {
+ // TODO should cancel top query here.
+ // continue;
+ //}
std::shared_ptr<QueryContext> max_revocable_query;
std::shared_ptr<QueryContext> max_memory_usage_query;
@@ -513,4 +550,21 @@ void WorkloadGroupMgr::stop() {
}
}
+void WorkloadGroupMgr::update_load_memtable_usage(
+ const std::map<uint64_t, MemtableUsage>& wg_memtable_usages) {
+ // Use readlock here, because it will not modify workload_groups
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto it = _workload_groups.begin(); it != _workload_groups.end(); ++it) {
+ auto wg_usage = wg_memtable_usages.find(it->first);
+ if (wg_usage != wg_memtable_usages.end()) {
+ it->second->update_load_mem_usage(wg_usage->second.active_mem_usage,
+ wg_usage->second.queue_mem_usage,
+ wg_usage->second.flush_mem_usage);
+ } else {
+ // Not anything in memtable limiter, then set to 0
+ it->second->update_load_mem_usage(0, 0, 0);
+ }
+ }
+}
+
} // namespace doris
diff --git a/be/src/runtime/workload_group/workload_group_manager.h b/be/src/runtime/workload_group/workload_group_manager.h
index 68d4f932de0..6e51a1f3767 100644
--- a/be/src/runtime/workload_group/workload_group_manager.h
+++ b/be/src/runtime/workload_group/workload_group_manager.h
@@ -62,6 +62,13 @@ private:
std::string query_id_;
};
+class MemtableUsage {
+public:
+ int64_t active_mem_usage = 0;
+ int64_t queue_mem_usage = 0;
+ int64_t flush_mem_usage = 0;
+};
+
class WorkloadGroupMgr {
public:
WorkloadGroupMgr() = default;
@@ -92,6 +99,8 @@ public:
void handle_paused_queries();
+ void update_load_memtable_usage(const std::map<uint64_t, MemtableUsage>& wg_memtable_usages);
+
private:
std::shared_mutex _group_mutex;
std::unordered_map<uint64_t, WorkloadGroupPtr> _workload_groups;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org