You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "kaijchen (via GitHub)" <gi...@apache.org> on 2023/10/24 04:01:23 UTC
[PR] [improve](load) rewrite memtable limiter rules [doris]
kaijchen opened a new pull request, #25814:
URL: https://github.com/apache/doris/pull/25814
## Proposed changes
Issue Number: close #xxx
<!--Describe your changes.-->
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen closed pull request #25814: [improve](load) rewrite memtable memory limiter rules
URL: https://github.com/apache/doris/pull/25814
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen closed pull request #25814: [improve](load) rewrite memtable memory limiter rules
URL: https://github.com/apache/doris/pull/25814
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25814:
URL: https://github.com/apache/doris/pull/25814#issuecomment-1831695333
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #25814:
URL: https://github.com/apache/doris/pull/25814#issuecomment-1831844748
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 44.04 seconds
stream load tsv: 572 seconds loaded 74807831229 Bytes, about 124 MB/s
stream load json: 18 seconds loaded 2358488459 Bytes, about 124 MB/s
stream load orc: 65 seconds loaded 1101869774 Bytes, about 16 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 28.3 seconds inserted 10000000 Rows, about 353K ops/s
storage size: 17162475778 Bytes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25814:
URL: https://github.com/apache/doris/pull/25814#issuecomment-1831697580
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on PR #25814:
URL: https://github.com/apache/doris/pull/25814#issuecomment-1831685096
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [improve](load) rewrite memtable memory limiter rules [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #25814:
URL: https://github.com/apache/doris/pull/25814#discussion_r1369611329
##########
be/src/olap/memtable_memory_limiter.cpp:
##########
@@ -59,135 +59,81 @@
_writers.push_back(writer);
}
+bool MemTableMemoryLimiter::_soft_limit_reached() {
+ return _mem_tracker->consumption() >= _load_soft_mem_limit ||
+ (MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit() &&
+ _mem_tracker->consumption() >= _load_hard_mem_limit / 10);
+}
+
+bool MemTableMemoryLimiter::_hard_limit_reached() {
+ return _mem_tracker->consumption() >= _load_hard_mem_limit ||
+ MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit();
+}
+
void MemTableMemoryLimiter::handle_memtable_flush() {
// Check the soft limit.
DCHECK(_load_soft_mem_limit > 0);
- // Record current memory status.
- int64_t process_soft_mem_limit = MemInfo::soft_mem_limit();
- int64_t proc_mem_no_allocator_cache = MemInfo::proc_mem_no_allocator_cache();
-#ifndef BE_TEST
- // If process memory is almost full but data load don't consume more than 5% (50% * 10%) of
- // total memory, we don't need to flush memtable.
- bool reduce_on_process_soft_mem_limit =
- proc_mem_no_allocator_cache >= process_soft_mem_limit &&
- _mem_tracker->consumption() >= _load_hard_mem_limit / 10;
- if (_mem_tracker->consumption() < _load_soft_mem_limit && !reduce_on_process_soft_mem_limit) {
+ if (!_soft_limit_reached()) {
return;
}
-#endif
- // Indicate whether current thread is reducing mem on hard limit.
- bool reducing_mem_on_hard_limit = false;
- Status st;
- std::vector<WriterMemItem> writers_to_reduce_mem;
{
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> l(_lock);
- while (_should_wait_flush) {
- _wait_flush_cond.wait(l);
- }
- LOG(INFO) << "Reached the one tenth of load hard limit " << _load_hard_mem_limit / 10
- << "and process remaining allocator cache " << proc_mem_no_allocator_cache
- << "reached process soft memory limit " << process_soft_mem_limit
- << ", waited for flush, time_ns:" << timer.elapsed_time();
-#ifndef BE_TEST
- bool hard_limit_reached = _mem_tracker->consumption() >= _load_hard_mem_limit ||
- proc_mem_no_allocator_cache >= process_soft_mem_limit;
- // Some other thread is flushing data, and not reached hard limit now,
- // we don't need to handle mem limit in current thread.
- if (_soft_reduce_mem_in_progress && !hard_limit_reached) {
- return;
- }
-#endif
-
- auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) {
- return lhs.mem_size < rhs.mem_size;
- };
- std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
-
- for (auto it = _writers.begin(); it != _writers.end();) {
- if (auto writer = it->lock()) {
- int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
- mem_heap.emplace(writer, active_memtable_mem);
- ++it;
- } else {
- *it = std::move(_writers.back());
- _writers.pop_back();
+ while (_soft_limit_reached()) {
+ LOG(INFO) << "reached memtable memory soft limit"
+ << " (active: " << PrettyPrinter::print_bytes(_active_mem_usage)
+ << ", write: " << PrettyPrinter::print_bytes(_write_mem_usage)
+ << ", flush: " << PrettyPrinter::print_bytes(_flush_mem_usage) << ")";
+ if (_active_mem_usage >= _load_soft_mem_limit ||
+ (_active_mem_usage > _load_hard_mem_limit / 10 && _hard_limit_reached())) {
+ _flush_active_memtables();
}
+ _soft_limit_end_cond.wait(l);
}
- int64_t mem_to_flushed = _mem_tracker->consumption() / 10;
- int64_t mem_consumption_in_picked_writer = 0;
- while (!mem_heap.empty()) {
- WriterMemItem mem_item = mem_heap.top();
- mem_heap.pop();
- auto writer = mem_item.writer.lock();
- if (!writer) {
- continue;
- }
- int64_t mem_size = mem_item.mem_size;
- writers_to_reduce_mem.emplace_back(writer, mem_size);
- st = writer->flush_async();
- if (!st.ok()) {
- auto err_msg = fmt::format(
- "tablet writer failed to reduce mem consumption by flushing memtable, "
- "tablet_id={}, err={}",
- writer->tablet_id(), st.to_string());
- LOG(WARNING) << err_msg;
- static_cast<void>(writer->cancel_with_status(st));
- }
- mem_consumption_in_picked_writer += mem_size;
- if (mem_consumption_in_picked_writer > mem_to_flushed) {
- break;
- }
- }
- if (writers_to_reduce_mem.empty()) {
- // should not happen, add log to observe
- LOG(WARNING) << "failed to find suitable writers to reduce memory"
- << " when total load mem limit exceed";
- return;
- }
+ timer.stop();
+ int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/olap/memtable_memory_limiter.cpp:
##########
@@ -59,135 +59,81 @@ void MemTableMemoryLimiter::register_writer(std::weak_ptr<MemTableWriter> writer
_writers.push_back(writer);
}
+bool MemTableMemoryLimiter::_soft_limit_reached() {
+ return _mem_tracker->consumption() >= _load_soft_mem_limit ||
+ (MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit() &&
+ _mem_tracker->consumption() >= _load_hard_mem_limit / 10);
+}
+
+bool MemTableMemoryLimiter::_hard_limit_reached() {
+ return _mem_tracker->consumption() >= _load_hard_mem_limit ||
+ MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit();
+}
+
void MemTableMemoryLimiter::handle_memtable_flush() {
// Check the soft limit.
DCHECK(_load_soft_mem_limit > 0);
- // Record current memory status.
- int64_t process_soft_mem_limit = MemInfo::soft_mem_limit();
- int64_t proc_mem_no_allocator_cache = MemInfo::proc_mem_no_allocator_cache();
-#ifndef BE_TEST
- // If process memory is almost full but data load don't consume more than 5% (50% * 10%) of
- // total memory, we don't need to flush memtable.
- bool reduce_on_process_soft_mem_limit =
- proc_mem_no_allocator_cache >= process_soft_mem_limit &&
- _mem_tracker->consumption() >= _load_hard_mem_limit / 10;
- if (_mem_tracker->consumption() < _load_soft_mem_limit && !reduce_on_process_soft_mem_limit) {
+ if (!_soft_limit_reached()) {
return;
}
-#endif
- // Indicate whether current thread is reducing mem on hard limit.
- bool reducing_mem_on_hard_limit = false;
- Status st;
- std::vector<WriterMemItem> writers_to_reduce_mem;
{
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> l(_lock);
- while (_should_wait_flush) {
- _wait_flush_cond.wait(l);
- }
- LOG(INFO) << "Reached the one tenth of load hard limit " << _load_hard_mem_limit / 10
- << "and process remaining allocator cache " << proc_mem_no_allocator_cache
- << "reached process soft memory limit " << process_soft_mem_limit
- << ", waited for flush, time_ns:" << timer.elapsed_time();
-#ifndef BE_TEST
- bool hard_limit_reached = _mem_tracker->consumption() >= _load_hard_mem_limit ||
- proc_mem_no_allocator_cache >= process_soft_mem_limit;
- // Some other thread is flushing data, and not reached hard limit now,
- // we don't need to handle mem limit in current thread.
- if (_soft_reduce_mem_in_progress && !hard_limit_reached) {
- return;
- }
-#endif
-
- auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) {
- return lhs.mem_size < rhs.mem_size;
- };
- std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
-
- for (auto it = _writers.begin(); it != _writers.end();) {
- if (auto writer = it->lock()) {
- int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
- mem_heap.emplace(writer, active_memtable_mem);
- ++it;
- } else {
- *it = std::move(_writers.back());
- _writers.pop_back();
+ while (_soft_limit_reached()) {
+ LOG(INFO) << "reached memtable memory soft limit"
+ << " (active: " << PrettyPrinter::print_bytes(_active_mem_usage)
+ << ", write: " << PrettyPrinter::print_bytes(_write_mem_usage)
+ << ", flush: " << PrettyPrinter::print_bytes(_flush_mem_usage) << ")";
+ if (_active_mem_usage >= _load_soft_mem_limit ||
+ (_active_mem_usage > _load_hard_mem_limit / 10 && _hard_limit_reached())) {
Review Comment:
warning: 10 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
(_active_mem_usage > _load_hard_mem_limit / 10 && _hard_limit_reached())) {
^
```
##########
be/src/olap/memtable_memory_limiter.cpp:
##########
@@ -59,135 +59,81 @@
_writers.push_back(writer);
}
+bool MemTableMemoryLimiter::_soft_limit_reached() {
+ return _mem_tracker->consumption() >= _load_soft_mem_limit ||
+ (MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit() &&
+ _mem_tracker->consumption() >= _load_hard_mem_limit / 10);
+}
+
+bool MemTableMemoryLimiter::_hard_limit_reached() {
+ return _mem_tracker->consumption() >= _load_hard_mem_limit ||
+ MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit();
+}
+
void MemTableMemoryLimiter::handle_memtable_flush() {
// Check the soft limit.
DCHECK(_load_soft_mem_limit > 0);
- // Record current memory status.
- int64_t process_soft_mem_limit = MemInfo::soft_mem_limit();
- int64_t proc_mem_no_allocator_cache = MemInfo::proc_mem_no_allocator_cache();
-#ifndef BE_TEST
- // If process memory is almost full but data load don't consume more than 5% (50% * 10%) of
- // total memory, we don't need to flush memtable.
- bool reduce_on_process_soft_mem_limit =
- proc_mem_no_allocator_cache >= process_soft_mem_limit &&
- _mem_tracker->consumption() >= _load_hard_mem_limit / 10;
- if (_mem_tracker->consumption() < _load_soft_mem_limit && !reduce_on_process_soft_mem_limit) {
+ if (!_soft_limit_reached()) {
return;
}
-#endif
- // Indicate whether current thread is reducing mem on hard limit.
- bool reducing_mem_on_hard_limit = false;
- Status st;
- std::vector<WriterMemItem> writers_to_reduce_mem;
{
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> l(_lock);
- while (_should_wait_flush) {
- _wait_flush_cond.wait(l);
- }
- LOG(INFO) << "Reached the one tenth of load hard limit " << _load_hard_mem_limit / 10
- << "and process remaining allocator cache " << proc_mem_no_allocator_cache
- << "reached process soft memory limit " << process_soft_mem_limit
- << ", waited for flush, time_ns:" << timer.elapsed_time();
-#ifndef BE_TEST
- bool hard_limit_reached = _mem_tracker->consumption() >= _load_hard_mem_limit ||
- proc_mem_no_allocator_cache >= process_soft_mem_limit;
- // Some other thread is flushing data, and not reached hard limit now,
- // we don't need to handle mem limit in current thread.
- if (_soft_reduce_mem_in_progress && !hard_limit_reached) {
- return;
- }
-#endif
-
- auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) {
- return lhs.mem_size < rhs.mem_size;
- };
- std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
-
- for (auto it = _writers.begin(); it != _writers.end();) {
- if (auto writer = it->lock()) {
- int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
- mem_heap.emplace(writer, active_memtable_mem);
- ++it;
- } else {
- *it = std::move(_writers.back());
- _writers.pop_back();
+ while (_soft_limit_reached()) {
+ LOG(INFO) << "reached memtable memory soft limit"
+ << " (active: " << PrettyPrinter::print_bytes(_active_mem_usage)
+ << ", write: " << PrettyPrinter::print_bytes(_write_mem_usage)
+ << ", flush: " << PrettyPrinter::print_bytes(_flush_mem_usage) << ")";
+ if (_active_mem_usage >= _load_soft_mem_limit ||
+ (_active_mem_usage > _load_hard_mem_limit / 10 && _hard_limit_reached())) {
+ _flush_active_memtables();
}
+ _soft_limit_end_cond.wait(l);
}
- int64_t mem_to_flushed = _mem_tracker->consumption() / 10;
- int64_t mem_consumption_in_picked_writer = 0;
- while (!mem_heap.empty()) {
- WriterMemItem mem_item = mem_heap.top();
- mem_heap.pop();
- auto writer = mem_item.writer.lock();
- if (!writer) {
- continue;
- }
- int64_t mem_size = mem_item.mem_size;
- writers_to_reduce_mem.emplace_back(writer, mem_size);
- st = writer->flush_async();
- if (!st.ok()) {
- auto err_msg = fmt::format(
- "tablet writer failed to reduce mem consumption by flushing memtable, "
- "tablet_id={}, err={}",
- writer->tablet_id(), st.to_string());
- LOG(WARNING) << err_msg;
- static_cast<void>(writer->cancel_with_status(st));
- }
- mem_consumption_in_picked_writer += mem_size;
- if (mem_consumption_in_picked_writer > mem_to_flushed) {
- break;
- }
- }
- if (writers_to_reduce_mem.empty()) {
- // should not happen, add log to observe
- LOG(WARNING) << "failed to find suitable writers to reduce memory"
- << " when total load mem limit exceed";
- return;
- }
+ timer.stop();
+ int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
Review Comment:
warning: 1000 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
^
```
##########
be/src/olap/memtable_memory_limiter.cpp:
##########
@@ -59,135 +59,81 @@
_writers.push_back(writer);
}
+bool MemTableMemoryLimiter::_soft_limit_reached() {
+ return _mem_tracker->consumption() >= _load_soft_mem_limit ||
+ (MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit() &&
+ _mem_tracker->consumption() >= _load_hard_mem_limit / 10);
+}
+
+bool MemTableMemoryLimiter::_hard_limit_reached() {
+ return _mem_tracker->consumption() >= _load_hard_mem_limit ||
+ MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit();
+}
+
void MemTableMemoryLimiter::handle_memtable_flush() {
// Check the soft limit.
DCHECK(_load_soft_mem_limit > 0);
- // Record current memory status.
- int64_t process_soft_mem_limit = MemInfo::soft_mem_limit();
- int64_t proc_mem_no_allocator_cache = MemInfo::proc_mem_no_allocator_cache();
-#ifndef BE_TEST
- // If process memory is almost full but data load don't consume more than 5% (50% * 10%) of
- // total memory, we don't need to flush memtable.
- bool reduce_on_process_soft_mem_limit =
- proc_mem_no_allocator_cache >= process_soft_mem_limit &&
- _mem_tracker->consumption() >= _load_hard_mem_limit / 10;
- if (_mem_tracker->consumption() < _load_soft_mem_limit && !reduce_on_process_soft_mem_limit) {
+ if (!_soft_limit_reached()) {
return;
}
-#endif
- // Indicate whether current thread is reducing mem on hard limit.
- bool reducing_mem_on_hard_limit = false;
- Status st;
- std::vector<WriterMemItem> writers_to_reduce_mem;
{
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> l(_lock);
- while (_should_wait_flush) {
- _wait_flush_cond.wait(l);
- }
- LOG(INFO) << "Reached the one tenth of load hard limit " << _load_hard_mem_limit / 10
- << "and process remaining allocator cache " << proc_mem_no_allocator_cache
- << "reached process soft memory limit " << process_soft_mem_limit
- << ", waited for flush, time_ns:" << timer.elapsed_time();
-#ifndef BE_TEST
- bool hard_limit_reached = _mem_tracker->consumption() >= _load_hard_mem_limit ||
- proc_mem_no_allocator_cache >= process_soft_mem_limit;
- // Some other thread is flushing data, and not reached hard limit now,
- // we don't need to handle mem limit in current thread.
- if (_soft_reduce_mem_in_progress && !hard_limit_reached) {
- return;
- }
-#endif
-
- auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) {
- return lhs.mem_size < rhs.mem_size;
- };
- std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
-
- for (auto it = _writers.begin(); it != _writers.end();) {
- if (auto writer = it->lock()) {
- int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
- mem_heap.emplace(writer, active_memtable_mem);
- ++it;
- } else {
- *it = std::move(_writers.back());
- _writers.pop_back();
+ while (_soft_limit_reached()) {
+ LOG(INFO) << "reached memtable memory soft limit"
+ << " (active: " << PrettyPrinter::print_bytes(_active_mem_usage)
+ << ", write: " << PrettyPrinter::print_bytes(_write_mem_usage)
+ << ", flush: " << PrettyPrinter::print_bytes(_flush_mem_usage) << ")";
+ if (_active_mem_usage >= _load_soft_mem_limit ||
+ (_active_mem_usage > _load_hard_mem_limit / 10 && _hard_limit_reached())) {
+ _flush_active_memtables();
}
+ _soft_limit_end_cond.wait(l);
}
- int64_t mem_to_flushed = _mem_tracker->consumption() / 10;
- int64_t mem_consumption_in_picked_writer = 0;
- while (!mem_heap.empty()) {
- WriterMemItem mem_item = mem_heap.top();
- mem_heap.pop();
- auto writer = mem_item.writer.lock();
- if (!writer) {
- continue;
- }
- int64_t mem_size = mem_item.mem_size;
- writers_to_reduce_mem.emplace_back(writer, mem_size);
- st = writer->flush_async();
- if (!st.ok()) {
- auto err_msg = fmt::format(
- "tablet writer failed to reduce mem consumption by flushing memtable, "
- "tablet_id={}, err={}",
- writer->tablet_id(), st.to_string());
- LOG(WARNING) << err_msg;
- static_cast<void>(writer->cancel_with_status(st));
- }
- mem_consumption_in_picked_writer += mem_size;
- if (mem_consumption_in_picked_writer > mem_to_flushed) {
- break;
- }
- }
- if (writers_to_reduce_mem.empty()) {
- // should not happen, add log to observe
- LOG(WARNING) << "failed to find suitable writers to reduce memory"
- << " when total load mem limit exceed";
- return;
- }
+ timer.stop();
+ int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
+ LOG(INFO) << "waited " << time_ms << " ms for memtable memory limit";
+ }
+}
- std::ostringstream oss;
- oss << "reducing memory of " << writers_to_reduce_mem.size()
- << " memtable writers (total mem: "
- << PrettyPrinter::print_bytes(mem_consumption_in_picked_writer)
- << ", max mem: " << PrettyPrinter::print_bytes(writers_to_reduce_mem.front().mem_size)
- << ", min mem:" << PrettyPrinter::print_bytes(writers_to_reduce_mem.back().mem_size)
- << "), ";
- if (proc_mem_no_allocator_cache < process_soft_mem_limit) {
- oss << "because total load mem consumption "
- << PrettyPrinter::print_bytes(_mem_tracker->consumption()) << " has exceeded";
- if (_mem_tracker->consumption() > _load_hard_mem_limit) {
- _should_wait_flush = true;
- reducing_mem_on_hard_limit = true;
- oss << " hard limit: " << PrettyPrinter::print_bytes(_load_hard_mem_limit);
- } else {
- _soft_reduce_mem_in_progress = true;
- oss << " soft limit: " << PrettyPrinter::print_bytes(_load_soft_mem_limit);
- }
+void MemTableMemoryLimiter::_flush_active_memtables() {
+ auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) { return lhs.mem_size < rhs.mem_size; };
+ std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
+
+ _active_mem_usage = 0;
+ for (auto it = _writers.begin(); it != _writers.end();) {
+ if (auto writer = it->lock()) {
+ int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
+ _active_mem_usage += active_memtable_mem;
+ mem_heap.emplace(writer, active_memtable_mem);
+ ++it;
} else {
- _should_wait_flush = true;
- reducing_mem_on_hard_limit = true;
- oss << "because proc_mem_no_allocator_cache consumption "
- << PrettyPrinter::print_bytes(proc_mem_no_allocator_cache)
- << ", has exceeded process soft limit "
- << PrettyPrinter::print_bytes(process_soft_mem_limit)
- << ", total load mem consumption: "
- << PrettyPrinter::print_bytes(_mem_tracker->consumption())
- << ", vm_rss: " << PerfCounters::get_vm_rss_str();
+ *it = std::move(_writers.back());
+ _writers.pop_back();
}
- LOG(INFO) << oss.str();
}
-
- // wait all writers flush without lock
- for (auto item : writers_to_reduce_mem) {
- VLOG_NOTICE << "reducing memory, wait flush mem_size: "
- << PrettyPrinter::print_bytes(item.mem_size);
- auto writer = item.writer.lock();
+ int64_t mem_to_flush = _hard_limit_reached()
+ ? _active_mem_usage - _load_hard_mem_limit / 11
Review Comment:
warning: 11 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
? _active_mem_usage - _load_hard_mem_limit / 11
^
```
##########
be/src/olap/memtable_memory_limiter.cpp:
##########
@@ -59,135 +59,81 @@
_writers.push_back(writer);
}
+bool MemTableMemoryLimiter::_soft_limit_reached() {
+ return _mem_tracker->consumption() >= _load_soft_mem_limit ||
+ (MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit() &&
+ _mem_tracker->consumption() >= _load_hard_mem_limit / 10);
+}
+
+bool MemTableMemoryLimiter::_hard_limit_reached() {
+ return _mem_tracker->consumption() >= _load_hard_mem_limit ||
+ MemInfo::proc_mem_no_allocator_cache() >= MemInfo::soft_mem_limit();
+}
+
void MemTableMemoryLimiter::handle_memtable_flush() {
// Check the soft limit.
DCHECK(_load_soft_mem_limit > 0);
- // Record current memory status.
- int64_t process_soft_mem_limit = MemInfo::soft_mem_limit();
- int64_t proc_mem_no_allocator_cache = MemInfo::proc_mem_no_allocator_cache();
-#ifndef BE_TEST
- // If process memory is almost full but data load don't consume more than 5% (50% * 10%) of
- // total memory, we don't need to flush memtable.
- bool reduce_on_process_soft_mem_limit =
- proc_mem_no_allocator_cache >= process_soft_mem_limit &&
- _mem_tracker->consumption() >= _load_hard_mem_limit / 10;
- if (_mem_tracker->consumption() < _load_soft_mem_limit && !reduce_on_process_soft_mem_limit) {
+ if (!_soft_limit_reached()) {
return;
}
-#endif
- // Indicate whether current thread is reducing mem on hard limit.
- bool reducing_mem_on_hard_limit = false;
- Status st;
- std::vector<WriterMemItem> writers_to_reduce_mem;
{
MonotonicStopWatch timer;
timer.start();
std::unique_lock<std::mutex> l(_lock);
- while (_should_wait_flush) {
- _wait_flush_cond.wait(l);
- }
- LOG(INFO) << "Reached the one tenth of load hard limit " << _load_hard_mem_limit / 10
- << "and process remaining allocator cache " << proc_mem_no_allocator_cache
- << "reached process soft memory limit " << process_soft_mem_limit
- << ", waited for flush, time_ns:" << timer.elapsed_time();
-#ifndef BE_TEST
- bool hard_limit_reached = _mem_tracker->consumption() >= _load_hard_mem_limit ||
- proc_mem_no_allocator_cache >= process_soft_mem_limit;
- // Some other thread is flushing data, and not reached hard limit now,
- // we don't need to handle mem limit in current thread.
- if (_soft_reduce_mem_in_progress && !hard_limit_reached) {
- return;
- }
-#endif
-
- auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) {
- return lhs.mem_size < rhs.mem_size;
- };
- std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
-
- for (auto it = _writers.begin(); it != _writers.end();) {
- if (auto writer = it->lock()) {
- int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
- mem_heap.emplace(writer, active_memtable_mem);
- ++it;
- } else {
- *it = std::move(_writers.back());
- _writers.pop_back();
+ while (_soft_limit_reached()) {
+ LOG(INFO) << "reached memtable memory soft limit"
+ << " (active: " << PrettyPrinter::print_bytes(_active_mem_usage)
+ << ", write: " << PrettyPrinter::print_bytes(_write_mem_usage)
+ << ", flush: " << PrettyPrinter::print_bytes(_flush_mem_usage) << ")";
+ if (_active_mem_usage >= _load_soft_mem_limit ||
+ (_active_mem_usage > _load_hard_mem_limit / 10 && _hard_limit_reached())) {
+ _flush_active_memtables();
}
+ _soft_limit_end_cond.wait(l);
}
- int64_t mem_to_flushed = _mem_tracker->consumption() / 10;
- int64_t mem_consumption_in_picked_writer = 0;
- while (!mem_heap.empty()) {
- WriterMemItem mem_item = mem_heap.top();
- mem_heap.pop();
- auto writer = mem_item.writer.lock();
- if (!writer) {
- continue;
- }
- int64_t mem_size = mem_item.mem_size;
- writers_to_reduce_mem.emplace_back(writer, mem_size);
- st = writer->flush_async();
- if (!st.ok()) {
- auto err_msg = fmt::format(
- "tablet writer failed to reduce mem consumption by flushing memtable, "
- "tablet_id={}, err={}",
- writer->tablet_id(), st.to_string());
- LOG(WARNING) << err_msg;
- static_cast<void>(writer->cancel_with_status(st));
- }
- mem_consumption_in_picked_writer += mem_size;
- if (mem_consumption_in_picked_writer > mem_to_flushed) {
- break;
- }
- }
- if (writers_to_reduce_mem.empty()) {
- // should not happen, add log to observe
- LOG(WARNING) << "failed to find suitable writers to reduce memory"
- << " when total load mem limit exceed";
- return;
- }
+ timer.stop();
+ int64_t time_ms = timer.elapsed_time() / 1000 / 1000;
+ LOG(INFO) << "waited " << time_ms << " ms for memtable memory limit";
+ }
+}
- std::ostringstream oss;
- oss << "reducing memory of " << writers_to_reduce_mem.size()
- << " memtable writers (total mem: "
- << PrettyPrinter::print_bytes(mem_consumption_in_picked_writer)
- << ", max mem: " << PrettyPrinter::print_bytes(writers_to_reduce_mem.front().mem_size)
- << ", min mem:" << PrettyPrinter::print_bytes(writers_to_reduce_mem.back().mem_size)
- << "), ";
- if (proc_mem_no_allocator_cache < process_soft_mem_limit) {
- oss << "because total load mem consumption "
- << PrettyPrinter::print_bytes(_mem_tracker->consumption()) << " has exceeded";
- if (_mem_tracker->consumption() > _load_hard_mem_limit) {
- _should_wait_flush = true;
- reducing_mem_on_hard_limit = true;
- oss << " hard limit: " << PrettyPrinter::print_bytes(_load_hard_mem_limit);
- } else {
- _soft_reduce_mem_in_progress = true;
- oss << " soft limit: " << PrettyPrinter::print_bytes(_load_soft_mem_limit);
- }
+void MemTableMemoryLimiter::_flush_active_memtables() {
+ auto cmp = [](WriterMemItem& lhs, WriterMemItem& rhs) { return lhs.mem_size < rhs.mem_size; };
+ std::priority_queue<WriterMemItem, std::vector<WriterMemItem>, decltype(cmp)> mem_heap(cmp);
+
+ _active_mem_usage = 0;
+ for (auto it = _writers.begin(); it != _writers.end();) {
+ if (auto writer = it->lock()) {
+ int64_t active_memtable_mem = writer->active_memtable_mem_consumption();
+ _active_mem_usage += active_memtable_mem;
+ mem_heap.emplace(writer, active_memtable_mem);
+ ++it;
} else {
- _should_wait_flush = true;
- reducing_mem_on_hard_limit = true;
- oss << "because proc_mem_no_allocator_cache consumption "
- << PrettyPrinter::print_bytes(proc_mem_no_allocator_cache)
- << ", has exceeded process soft limit "
- << PrettyPrinter::print_bytes(process_soft_mem_limit)
- << ", total load mem consumption: "
- << PrettyPrinter::print_bytes(_mem_tracker->consumption())
- << ", vm_rss: " << PerfCounters::get_vm_rss_str();
+ *it = std::move(_writers.back());
+ _writers.pop_back();
}
- LOG(INFO) << oss.str();
}
-
- // wait all writers flush without lock
- for (auto item : writers_to_reduce_mem) {
- VLOG_NOTICE << "reducing memory, wait flush mem_size: "
- << PrettyPrinter::print_bytes(item.mem_size);
- auto writer = item.writer.lock();
+ int64_t mem_to_flush = _hard_limit_reached()
+ ? _active_mem_usage - _load_hard_mem_limit / 11
+ : _active_mem_usage - _load_soft_mem_limit / 5 * 4;
Review Comment:
warning: 5 is a magic number; consider replacing it with a named constant [readability-magic-numbers]
```cpp
: _active_mem_usage - _load_soft_mem_limit / 5 * 4;
^
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org