This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 4aeec0a391a branch-4.0:[fix](file-cache): speed up LRU restore startup
path(pick#65174) (#66129)
4aeec0a391a is described below
commit 4aeec0a391a9aec60db6091be15c2fecfcdac0b7
Author: zhengyu <[email protected]>
AuthorDate: Tue Jul 28 23:08:51 2026 +0800
branch-4.0:[fix](file-cache): speed up LRU restore startup path(pick#65174)
(#66129)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #65174
---
be/src/io/cache/block_file_cache_factory.cpp | 137 ++++++++++++++++++++++-----
be/src/io/cache/block_file_cache_factory.h | 7 ++
be/src/io/cache/cache_lru_dumper.cpp | 18 ++--
be/src/io/cache/cache_lru_dumper.h | 4 +-
be/src/runtime/exec_env_init.cpp | 28 ++----
be/test/io/cache/block_file_cache_test.cpp | 67 +++++++++++++
be/test/io/cache/cache_lru_dumper_test.cpp | 42 ++++++++
7 files changed, 251 insertions(+), 52 deletions(-)
diff --git a/be/src/io/cache/block_file_cache_factory.cpp
b/be/src/io/cache/block_file_cache_factory.cpp
index 93ec201a784..3f60612a790 100644
--- a/be/src/io/cache/block_file_cache_factory.cpp
+++ b/be/src/io/cache/block_file_cache_factory.cpp
@@ -33,7 +33,10 @@
#include <algorithm>
#include <atomic>
#include <execution>
+#include <functional>
#include <ostream>
+#include <thread>
+#include <unordered_set>
#include <utility>
#include "common/config.h"
@@ -41,6 +44,7 @@
#include "io/cache/file_cache_common.h"
#include "io/fs/local_file_system.h"
#include "runtime/exec_env.h"
+#include "runtime/thread_context.h"
#include "service/backend_options.h"
#include "util/slice.h"
#include "vec/core/block.h"
@@ -50,28 +54,16 @@ class TUniqueId;
namespace io {
-FileCacheFactory* FileCacheFactory::instance() {
- return ExecEnv::GetInstance()->file_cache_factory();
-}
+namespace {
-size_t FileCacheFactory::try_release() {
- int elements = 0;
- for (auto& cache : _caches) {
- elements += cache->try_release();
- }
- return elements;
-}
-
-size_t FileCacheFactory::try_release(const std::string& base_path) {
- auto iter = _path_to_cache.find(base_path);
- if (iter != _path_to_cache.end()) {
- return iter->second->try_release();
- }
- return 0;
-}
+struct BuiltFileCache {
+ std::string cache_base_path;
+ FileCacheSettings settings;
+ std::unique_ptr<BlockFileCache> cache;
+};
-Status FileCacheFactory::create_file_cache(const std::string& cache_base_path,
- FileCacheSettings
file_cache_settings) {
+Status build_file_cache(const std::string& cache_base_path, FileCacheSettings
file_cache_settings,
+ BuiltFileCache* built_cache) {
if (file_cache_settings.storage == "memory") {
if (cache_base_path != "memory") {
LOG(WARNING) << "memory storage must use memory path";
@@ -110,13 +102,112 @@ Status FileCacheFactory::create_file_cache(const
std::string& cache_base_path,
<< " total_size: " << file_cache_settings.capacity
<< " disk_total_size: " << disk_capacity;
}
+
auto cache = std::make_unique<BlockFileCache>(cache_base_path,
file_cache_settings);
RETURN_IF_ERROR(cache->initialize());
+ built_cache->cache_base_path = cache_base_path;
+ built_cache->settings = file_cache_settings;
+ built_cache->cache = std::move(cache);
+ return Status::OK();
+}
+
+} // namespace
+
+FileCacheFactory* FileCacheFactory::instance() {
+ return ExecEnv::GetInstance()->file_cache_factory();
+}
+
+size_t FileCacheFactory::try_release() {
+ int elements = 0;
+ for (auto& cache : _caches) {
+ elements += cache->try_release();
+ }
+ return elements;
+}
+
+size_t FileCacheFactory::try_release(const std::string& base_path) {
+ auto iter = _path_to_cache.find(base_path);
+ if (iter != _path_to_cache.end()) {
+ return iter->second->try_release();
+ }
+ return 0;
+}
+
+Status FileCacheFactory::create_file_cache(const std::string& cache_base_path,
+ FileCacheSettings
file_cache_settings) {
+ BuiltFileCache built_cache;
+ RETURN_IF_ERROR(build_file_cache(cache_base_path, file_cache_settings,
&built_cache));
+ {
+ std::lock_guard lock(_mtx);
+ _path_to_cache[built_cache.cache_base_path] = built_cache.cache.get();
+ _capacity += built_cache.settings.capacity;
+ _caches.push_back(std::move(built_cache.cache));
+ }
+
+ return Status::OK();
+}
+
+Status FileCacheFactory::create_file_caches(
+ const std::vector<CachePath>& cache_paths,
+ const std::function<bool(const std::string&, const Status&)>&
should_ignore_error) {
+ struct BuildResult {
+ std::string cache_base_path;
+ FileCacheSettings settings;
+ BuiltFileCache built_cache;
+ Status status;
+ bool skip = false;
+ };
+
+ std::vector<BuildResult> results;
+ results.reserve(cache_paths.size());
+ std::unordered_set<std::string> cache_path_set;
+ for (const auto& cache_path : cache_paths) {
+ if (cache_path_set.find(cache_path.path) != cache_path_set.end()) {
+ LOG(WARNING) << fmt::format("cache path {} is duplicate",
cache_path.path);
+ continue;
+ }
+
+ cache_path_set.emplace(cache_path.path);
+ auto& result = results.emplace_back();
+ result.cache_base_path = cache_path.path;
+ result.settings = cache_path.init_settings();
+ }
+
+ std::vector<std::thread> workers;
+ workers.reserve(results.size());
+ for (auto& result : results) {
+ auto* result_ptr = &result;
+ workers.emplace_back([result_ptr]() {
+ SCOPED_INIT_THREAD_CONTEXT();
+ result_ptr->status = build_file_cache(result_ptr->cache_base_path,
result_ptr->settings,
+ &result_ptr->built_cache);
+ });
+ }
+
+ for (auto& worker : workers) {
+ worker.join();
+ }
+
+ for (auto& result : results) {
+ if (!result.status.ok()) {
+ if (should_ignore_error &&
should_ignore_error(result.cache_base_path, result.status)) {
+ result.skip = true;
+ continue;
+ }
+ return result.status;
+ }
+ }
+
{
std::lock_guard lock(_mtx);
- _path_to_cache[cache_base_path] = cache.get();
- _caches.push_back(std::move(cache));
- _capacity += file_cache_settings.capacity;
+ for (auto& result : results) {
+ if (result.skip) {
+ continue;
+ }
+ _path_to_cache[result.built_cache.cache_base_path] =
result.built_cache.cache.get();
+ _capacity += result.built_cache.settings.capacity;
+ _caches.push_back(std::move(result.built_cache.cache));
+ }
}
return Status::OK();
diff --git a/be/src/io/cache/block_file_cache_factory.h
b/be/src/io/cache/block_file_cache_factory.h
index f2882f0a29a..6daa314b80b 100644
--- a/be/src/io/cache/block_file_cache_factory.h
+++ b/be/src/io/cache/block_file_cache_factory.h
@@ -20,6 +20,7 @@
#pragma once
+#include <functional>
#include <memory>
#include <mutex>
#include <optional>
@@ -51,6 +52,10 @@ public:
Status create_file_cache(const std::string& cache_base_path,
FileCacheSettings file_cache_settings);
+ Status create_file_caches(
+ const std::vector<CachePath>& cache_paths,
+ const std::function<bool(const std::string&, const Status&)>&
should_ignore_error);
+
Status reload_file_cache(const std::vector<CachePath>& cache_base_paths);
size_t try_release();
@@ -107,6 +112,8 @@ public:
void get_cache_stats_block(vectorized::Block* block);
+ const std::vector<std::unique_ptr<BlockFileCache>>& get_caches() const {
return _caches; }
+
FileCacheFactory() = default;
FileCacheFactory& operator=(const FileCacheFactory&) = delete;
FileCacheFactory(const FileCacheFactory&) = delete;
diff --git a/be/src/io/cache/cache_lru_dumper.cpp
b/be/src/io/cache/cache_lru_dumper.cpp
index 1e4a3c2e0ce..69ceb388027 100644
--- a/be/src/io/cache/cache_lru_dumper.cpp
+++ b/be/src/io/cache/cache_lru_dumper.cpp
@@ -393,6 +393,8 @@ Status CacheLRUDumper::parse_dump_footer(std::ifstream& in,
std::string& filenam
RETURN_IF_ERROR(check_ifstream_status(in, filename));
_parse_meta.Clear();
_current_parse_group.Clear();
+ _parse_group_index = 0;
+ _parse_entry_index = 0;
if (!_parse_meta.ParseFromString(meta_serialized)) {
std::string warn_msg = std::string(
fmt::format("LRU dump file meta parse failed, file={}, skip
restore", filename));
@@ -407,13 +409,13 @@ Status CacheLRUDumper::parse_dump_footer(std::ifstream&
in, std::string& filenam
Status CacheLRUDumper::parse_one_lru_entry(std::ifstream& in, std::string&
filename,
UInt128Wrapper& hash, size_t&
offset, size_t& size) {
- // Read next group if current is empty
- if (_current_parse_group.entries_size() == 0) {
- if (_parse_meta.group_offset_size_size() == 0) {
+ // Read next group if current group has been fully consumed.
+ if (_parse_entry_index >= _current_parse_group.entries_size()) {
+ if (_parse_group_index >= _parse_meta.group_offset_size_size()) {
return Status::EndOfFile("No more entries");
}
- auto group_info = _parse_meta.group_offset_size(0);
+ const auto& group_info =
_parse_meta.group_offset_size(_parse_group_index++);
in.seekg(group_info.offset(), std::ios::beg);
std::string group_serialized(group_info.size(), '\0');
in.read(&group_serialized[0], group_serialized.size());
@@ -431,20 +433,16 @@ Status CacheLRUDumper::parse_one_lru_entry(std::ifstream&
in, std::string& filen
LOG(WARNING) << warn_msg;
return Status::InternalError(warn_msg);
}
-
- // Remove processed group info
-
_parse_meta.mutable_group_offset_size()->erase(_parse_meta.group_offset_size().begin());
+ _parse_entry_index = 0;
}
// Get next entry from current group
VLOG_DEBUG << "After deserialization: " <<
_current_parse_group.DebugString();
- auto entry = _current_parse_group.entries(0);
+ const auto& entry = _current_parse_group.entries(_parse_entry_index++);
hash = UInt128Wrapper((static_cast<uint128_t>(entry.hash().high()) << 64)
| entry.hash().low());
offset = entry.offset();
size = entry.size();
- // Remove processed entry
-
_current_parse_group.mutable_entries()->erase(_current_parse_group.entries().begin());
return Status::OK();
}
diff --git a/be/src/io/cache/cache_lru_dumper.h
b/be/src/io/cache/cache_lru_dumper.h
index d9addff614c..d4589c09955 100644
--- a/be/src/io/cache/cache_lru_dumper.h
+++ b/be/src/io/cache/cache_lru_dumper.h
@@ -86,6 +86,8 @@ private:
// For parsing
doris::io::cache::LRUDumpEntryGroupPb _current_parse_group;
doris::io::cache::LRUDumpMetaPb _parse_meta;
+ int _parse_group_index = 0;
+ int _parse_entry_index = 0;
BlockFileCache* _mgr;
LRUQueueRecorder* _recorder;
@@ -93,4 +95,4 @@ private:
std::string _start_time;
bool _is_first_dump = true;
};
-} // namespace doris::io
\ No newline at end of file
+} // namespace doris::io
diff --git a/be/src/runtime/exec_env_init.cpp b/be/src/runtime/exec_env_init.cpp
index 1630cc50d8a..994dcbab071 100644
--- a/be/src/runtime/exec_env_init.cpp
+++ b/be/src/runtime/exec_env_init.cpp
@@ -498,7 +498,6 @@ void
ExecEnv::init_file_cache_factory(std::vector<doris::CachePath>& cache_paths
config::file_cache_each_block_size,
config::s3_write_buffer_size);
exit(-1);
}
- std::unordered_set<std::string> cache_path_set;
Status rest =
doris::parse_conf_cache_paths(doris::config::file_cache_path, cache_paths);
if (!rest) {
throw Exception(
@@ -506,23 +505,16 @@ void
ExecEnv::init_file_cache_factory(std::vector<doris::CachePath>& cache_paths
doris::config::file_cache_path,
rest.msg()));
}
- doris::Status cache_status;
- for (auto& cache_path : cache_paths) {
- if (cache_path_set.find(cache_path.path) != cache_path_set.end()) {
- LOG(WARNING) << fmt::format("cache path {} is duplicate",
cache_path.path);
- continue;
- }
-
- cache_status =
doris::io::FileCacheFactory::instance()->create_file_cache(
- cache_path.path, cache_path.init_settings());
- if (!cache_status.ok()) {
- if (!doris::config::ignore_broken_disk) {
- throw Exception(
- Status::FatalError("failed to init file cache, err:
{}", cache_status));
- }
- LOG(WARNING) << "failed to init file cache, err: " << cache_status;
- }
- cache_path_set.emplace(cache_path.path);
+ auto cache_status =
doris::io::FileCacheFactory::instance()->create_file_caches(
+ cache_paths, [](const std::string&, const Status& status) {
+ if (!doris::config::ignore_broken_disk) {
+ return false;
+ }
+ LOG(WARNING) << "failed to init file cache, err: " << status;
+ return true;
+ });
+ if (!cache_status.ok()) {
+ throw Exception(Status::FatalError("failed to init file cache, err:
{}", cache_status));
}
}
diff --git a/be/test/io/cache/block_file_cache_test.cpp
b/be/test/io/cache/block_file_cache_test.cpp
index 3f51263736e..81c58de85ec 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -3665,6 +3665,73 @@ TEST_F(BlockFileCacheTest, test_factory_1) {
FileCacheFactory::instance()->_capacity = 0;
}
+TEST_F(BlockFileCacheTest, create_file_caches_preserves_config_order) {
+ reset_file_cache_factory_for_test();
+ std::string cache_path2 = caches_dir / "cache2" / "";
+ std::string cache_path3 = caches_dir / "cache3" / "";
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ if (fs::exists(cache_path2)) {
+ fs::remove_all(cache_path2);
+ }
+ if (fs::exists(cache_path3)) {
+ fs::remove_all(cache_path3);
+ }
+ Defer cleanup {[&] {
+ reset_file_cache_factory_for_test();
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ if (fs::exists(cache_path2)) {
+ fs::remove_all(cache_path2);
+ }
+ if (fs::exists(cache_path3)) {
+ fs::remove_all(cache_path3);
+ }
+ }};
+
+ std::vector<CachePath> cache_paths;
+ cache_paths.emplace_back(cache_base_path, 90, 30, DEFAULT_NORMAL_PERCENT,
+ DEFAULT_DISPOSABLE_PERCENT,
DEFAULT_INDEX_PERCENT, DEFAULT_TTL_PERCENT,
+ "disk");
+ cache_paths.emplace_back(cache_path2, 120, 30, DEFAULT_NORMAL_PERCENT,
+ DEFAULT_DISPOSABLE_PERCENT,
DEFAULT_INDEX_PERCENT, DEFAULT_TTL_PERCENT,
+ "disk");
+ cache_paths.emplace_back(cache_base_path, 90, 30, DEFAULT_NORMAL_PERCENT,
+ DEFAULT_DISPOSABLE_PERCENT,
DEFAULT_INDEX_PERCENT, DEFAULT_TTL_PERCENT,
+ "disk");
+ cache_paths.emplace_back(cache_path3, 150, 30, DEFAULT_NORMAL_PERCENT,
+ DEFAULT_DISPOSABLE_PERCENT,
DEFAULT_INDEX_PERCENT, DEFAULT_TTL_PERCENT,
+ "disk");
+
+ ASSERT_TRUE(FileCacheFactory::instance()
+ ->create_file_caches(cache_paths, [](const
std::string&,
+ const Status&) {
return false; })
+ .ok());
+
+ const auto& caches = FileCacheFactory::instance()->get_caches();
+ ASSERT_EQ(caches.size(), 3);
+ EXPECT_EQ(caches[0]->get_base_path(), cache_base_path);
+ EXPECT_EQ(caches[1]->get_base_path(), cache_path2);
+ EXPECT_EQ(caches[2]->get_base_path(), cache_path3);
+
EXPECT_EQ(FileCacheFactory::instance()->get_by_path(cache_base_path)->get_base_path(),
+ cache_base_path);
+
EXPECT_EQ(FileCacheFactory::instance()->get_by_path(cache_path2)->get_base_path(),
cache_path2);
+
EXPECT_EQ(FileCacheFactory::instance()->get_by_path(cache_path3)->get_base_path(),
cache_path3);
+ EXPECT_EQ(FileCacheFactory::instance()->get_capacity(), 360);
+
+ for (const auto& cache : caches) {
+ wait_until_cache_ready(*cache);
+ }
+
+ for (int i = 0; i < 64; ++i) {
+ auto key = io::BlockFileCache::hash("factory_order_key_" +
std::to_string(i));
+ const auto expected_path = caches[KeyHash()(key) %
caches.size()]->get_base_path();
+
EXPECT_EQ(FileCacheFactory::instance()->get_by_path(key)->get_base_path(),
expected_path);
+ }
+}
+
TEST_F(BlockFileCacheTest, test_factory_2) {
if (fs::exists(cache_base_path)) {
fs::remove_all(cache_base_path);
diff --git a/be/test/io/cache/cache_lru_dumper_test.cpp
b/be/test/io/cache/cache_lru_dumper_test.cpp
index e2fdfb6a7e7..e8a6bb4e674 100644
--- a/be/test/io/cache/cache_lru_dumper_test.cpp
+++ b/be/test/io/cache/cache_lru_dumper_test.cpp
@@ -160,6 +160,48 @@ TEST_F(CacheLRUDumperTest, test_dump_and_restore_queue) {
}
}
+TEST_F(CacheLRUDumperTest,
test_parse_multiple_groups_without_mutating_repeated_fields) {
+ const auto old_tail_record_num =
config::file_cache_background_lru_dump_tail_record_num;
+ Defer defer {[old_tail_record_num] {
+ config::file_cache_background_lru_dump_tail_record_num =
old_tail_record_num;
+ }};
+ config::file_cache_background_lru_dump_tail_record_num = 10001;
+
+ LRUQueue src_queue;
+ std::string queue_name = "normal";
+ std::lock_guard<std::mutex> lock(_mutex);
+ UInt128Wrapper hash(987654321ULL);
+
+ for (size_t i = 0; i < 10001; ++i) {
+ src_queue.add(hash, i * 4096, 4096 + i, lock);
+ }
+
+ dumper->do_dump_queue(src_queue, queue_name);
+
+ std::string filename = fmt::format("{}lru_dump_{}.tail", test_dir,
queue_name);
+ std::ifstream in(filename, std::ios::binary);
+ ASSERT_TRUE(in);
+
+ size_t entry_num = 0;
+ ASSERT_TRUE(dumper->parse_dump_footer(in, filename, entry_num).ok());
+ ASSERT_EQ(entry_num, 10001);
+ ASSERT_EQ(dumper->_parse_meta.group_offset_size_size(), 2);
+
+ UInt128Wrapper parsed_hash;
+ size_t offset = 0;
+ size_t size = 0;
+ for (size_t i = 0; i < entry_num; ++i) {
+ ASSERT_TRUE(dumper->parse_one_lru_entry(in, filename, parsed_hash,
offset, size).ok());
+ EXPECT_EQ(parsed_hash, hash);
+ EXPECT_EQ(offset, i * 4096);
+ EXPECT_EQ(size, 4096 + i);
+ }
+
+ EXPECT_EQ(dumper->_parse_meta.group_offset_size_size(), 2);
+ EXPECT_EQ(dumper->_parse_group_index, 2);
+ EXPECT_EQ(dumper->_parse_entry_index, 1);
+}
+
TEST_F(CacheLRUDumperTest,
test_lru_log_record_disabled_keeps_existing_backlog) {
const auto old_tail_record_num =
config::file_cache_background_lru_dump_tail_record_num;
const auto old_queue_limit =
config::file_cache_background_lru_log_queue_max_size;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
