This is an automated email from the ASF dual-hosted git repository.
liaoxin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9e7345888c8 [opt](load) use notify to replace polling for FlushToken
(#35796)
9e7345888c8 is described below
commit 9e7345888c814f60d36e64146447bec35d6edc4f
Author: Xin Liao <[email protected]>
AuthorDate: Mon Jun 3 19:34:51 2024 +0800
[opt](load) use notify to replace polling for FlushToken (#35796)
---
be/src/olap/memtable_flush_executor.cpp | 21 ++++++++++-----------
be/src/olap/memtable_flush_executor.h | 4 ++++
2 files changed, 14 insertions(+), 11 deletions(-)
diff --git a/be/src/olap/memtable_flush_executor.cpp
b/be/src/olap/memtable_flush_executor.cpp
index 40128fc204a..dc9545d0b34 100644
--- a/be/src/olap/memtable_flush_executor.cpp
+++ b/be/src/olap/memtable_flush_executor.cpp
@@ -95,6 +95,7 @@ Status FlushToken::submit(std::unique_ptr<MemTable>
mem_table) {
this, std::move(mem_table), _rowset_writer->allocate_segment_id(),
submit_task_time);
Status ret = _thread_pool->submit(std::move(task));
if (ret.ok()) {
+ // _wait_running_task_finish was executed after this function, so no
need to notify _cond here
_stats.flush_running_count++;
}
return ret;
@@ -103,16 +104,8 @@ Status FlushToken::submit(std::unique_ptr<MemTable>
mem_table) {
// NOTE: FlushToken's submit/cancel/wait run in one thread,
// so we don't need to make them mutually exclusive, std::atomic is enough.
void FlushToken::_wait_running_task_finish() {
- while (true) {
- int64_t flush_running_count = _stats.flush_running_count.load();
- if (flush_running_count < 0) {
- LOG(ERROR) << "flush_running_count < 0, this is not expected!";
- }
- if (flush_running_count == 0) {
- break;
- }
- std::this_thread::sleep_for(std::chrono::milliseconds(50));
- }
+ std::unique_lock<std::mutex> lock(_mutex);
+ _cond.wait(lock, [&]() { return _stats.flush_running_count.load() == 0; });
}
void FlushToken::cancel() {
@@ -155,7 +148,13 @@ Status FlushToken::_do_flush_memtable(MemTable* memtable,
int32_t segment_id, in
void FlushToken::_flush_memtable(std::unique_ptr<MemTable> memtable_ptr,
int32_t segment_id,
int64_t submit_task_time) {
- Defer defer {[&]() { _stats.flush_running_count--; }};
+ Defer defer {[&]() {
+ std::lock_guard<std::mutex> lock(_mutex);
+ _stats.flush_running_count--;
+ if (_stats.flush_running_count == 0) {
+ _cond.notify_one();
+ }
+ }};
if (_is_shutdown()) {
return;
}
diff --git a/be/src/olap/memtable_flush_executor.h
b/be/src/olap/memtable_flush_executor.h
index 1a529127ea5..b647c23deb9 100644
--- a/be/src/olap/memtable_flush_executor.h
+++ b/be/src/olap/memtable_flush_executor.h
@@ -18,6 +18,7 @@
#pragma once
#include <atomic>
+#include <condition_variable>
#include <cstdint>
#include <iosfwd>
#include <memory>
@@ -101,6 +102,9 @@ private:
std::atomic<bool> _shutdown = false;
ThreadPool* _thread_pool = nullptr;
+
+ std::mutex _mutex;
+ std::condition_variable _cond;
};
// MemTableFlushExecutor is responsible for flushing memtables to disk.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
