github-actions[bot] commented on code in PR #64496:
URL: https://github.com/apache/doris/pull/64496#discussion_r3666659466
##########
be/test/io/cache/block_file_cache_test.cpp:
##########
@@ -2521,207 +2635,1270 @@ TEST_F(BlockFileCacheTest, ttl_change_to_normal) {
auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
if (auto storage =
dynamic_cast<FSFileCacheStorage*>(cache._storage.get());
storage != nullptr) {
- std::string dir = storage->get_path_in_local_cache(key2, 0);
+ std::string dir = storage->get_path_in_local_cache(key2,
ttl_expiration_time);
+ EXPECT_TRUE(
+ fs::exists(storage->get_path_in_local_cache(dir, 50,
io::FileCacheType::TTL)));
+ std::string normal_dir = storage->get_path_in_local_cache(key2, 0);
+ EXPECT_FALSE(fs::exists(
+ storage->get_path_in_local_cache(normal_dir, 50,
io::FileCacheType::NORMAL)));
+ }
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ EXPECT_EQ(blocks[0]->expiration_time(), ttl_expiration_time);
+ std::string buffer(10, '1');
+ EXPECT_TRUE(blocks[0]->read(Slice(buffer.data(), 10), 0).ok());
+ EXPECT_EQ(buffer, std::string(10, '0'));
+ }
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest, ttl_change_to_normal_memory_storage) {
+ test_file_cache_memory_storage(io::FileCacheType::NORMAL);
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 30;
+ settings.query_queue_elements = 5;
+ settings.capacity = 30;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ settings.storage = "memory";
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.query_id = query_id;
+ int64_t cur_time = UnixSeconds();
+ context.expiration_time = cur_time + 180;
+ const int64_t ttl_expiration_time = context.expiration_time;
+ auto key2 = io::BlockFileCache::hash("key2");
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ };
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ {
+ auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::EMPTY);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download_into_memory(blocks[0]);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ }
+ {
+ context.cache_type = io::FileCacheType::NORMAL;
+ context.expiration_time = 0;
+ auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ EXPECT_EQ(blocks[0]->expiration_time(), ttl_expiration_time);
+ std::string buffer(10, '1');
+ EXPECT_TRUE(blocks[0]->read(Slice(buffer.data(), 10), 0).ok());
+ EXPECT_EQ(buffer, std::string(10, '0'));
+ }
+}
+
+TEST_F(BlockFileCacheTest, ttl_change_expiration_time) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ test_file_cache(io::FileCacheType::NORMAL);
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 30;
+ settings.query_queue_elements = 5;
+ settings.ttl_queue_size = 30;
+ settings.ttl_queue_elements = 5;
+ settings.capacity = 60;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.query_id = query_id;
+ int64_t cur_time = UnixSeconds();
+ context.expiration_time = cur_time + 180;
+ const int64_t ttl_expiration_time = context.expiration_time;
+ int64_t change_time = cur_time + 120;
+ auto key2 = io::BlockFileCache::hash("key2");
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ };
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ {
+ auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ // std::cout << "current cache size:" << cache.get_used_cache_size()
<< std::endl;
+ std::cout << "cache capacity:" << cache.capacity() << std::endl;
+ auto map = cache.get_stats_unsafe();
+ for (auto& [key, value] : map) {
+ std::cout << key << " : " << value << std::endl;
+ }
+ auto key1 = io::BlockFileCache::hash("key1");
+ std::cout << cache.dump_structure(key1) << std::endl;
+ std::cout << cache.dump_structure(key2) << std::endl;
+
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::EMPTY);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ }
+ {
+ context.cache_type = io::FileCacheType::TTL;
+ context.expiration_time = change_time;
+ auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
+ if (auto storage =
dynamic_cast<FSFileCacheStorage*>(cache._storage.get());
+ storage != nullptr) {
+ std::string old_dir = storage->get_path_in_local_cache(key2,
ttl_expiration_time);
EXPECT_TRUE(fs::exists(
- storage->get_path_in_local_cache(dir, 50,
io::FileCacheType::NORMAL)));
+ storage->get_path_in_local_cache(old_dir, 50,
io::FileCacheType::TTL)));
+ std::string new_dir = storage->get_path_in_local_cache(key2,
change_time);
+ EXPECT_FALSE(fs::exists(
+ storage->get_path_in_local_cache(new_dir, 50,
io::FileCacheType::TTL)));
}
auto blocks = fromHolder(holder);
ASSERT_EQ(blocks.size(), 1);
assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
- EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::NORMAL);
- EXPECT_EQ(blocks[0]->expiration_time(), 0);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ EXPECT_EQ(blocks[0]->expiration_time(), ttl_expiration_time);
+ std::string buffer(10, '1');
+ EXPECT_TRUE(blocks[0]->read(Slice(buffer.data(), 10), 0).ok());
+ EXPECT_EQ(buffer, std::string(10, '0'));
+ }
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest, ttl_change_expiration_time_memory_storage) {
+ test_file_cache_memory_storage(io::FileCacheType::NORMAL);
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 30;
+ settings.query_queue_elements = 5;
+ settings.capacity = 30;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ settings.storage = "memory";
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.query_id = query_id;
+ int64_t cur_time = UnixSeconds();
+ context.expiration_time = cur_time + 180;
+ const int64_t ttl_expiration_time = context.expiration_time;
+ int64_t change_time = cur_time + 120;
+ auto key2 = io::BlockFileCache::hash("key2");
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ };
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ {
+ auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::EMPTY);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download_into_memory(blocks[0]);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ }
+ {
+ context.cache_type = io::FileCacheType::TTL;
+ context.expiration_time = change_time;
+ auto holder = cache.get_or_set(key2, 50, 10, context); /// Add range
[50, 59]
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ assert_range(1, blocks[0], io::FileBlock::Range(50, 59),
io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ EXPECT_EQ(blocks[0]->expiration_time(), ttl_expiration_time);
std::string buffer(10, '1');
EXPECT_TRUE(blocks[0]->read(Slice(buffer.data(), 10), 0).ok());
EXPECT_EQ(buffer, std::string(10, '0'));
}
+}
+
+TEST_F(BlockFileCacheTest, ttl_read_path_does_not_modify_existing_ttl) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.query_id = query_id;
+ context.expiration_time = UnixSeconds() + 300;
+ const int64_t old_expiration_time = context.expiration_time;
+ auto key =
io::BlockFileCache::hash("ttl_read_path_does_not_modify_existing_ttl");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ {
+ auto holder = cache.get_or_set(key, 0, 10, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ ASSERT_EQ(blocks[0]->expiration_time(), old_expiration_time);
+ }
+
+ context.expiration_time = old_expiration_time + 300;
+ {
+ auto holder = cache.get_or_set(key, 0, 10, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ EXPECT_EQ(blocks[0]->cache_type(), io::FileCacheType::TTL);
+ EXPECT_EQ(blocks[0]->expiration_time(), old_expiration_time);
+ EXPECT_TRUE(fs::exists(key_dir_path(key, old_expiration_time) / "0"));
+ EXPECT_FALSE(fs::exists(key_dir_path(key, context.expiration_time) /
"0"));
+ }
+
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest, ttl_hole_inherits_existing_hash_expiration) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.query_id = query_id;
+ context.expiration_time = UnixSeconds() + 300;
+ const int64_t old_expiration_time = context.expiration_time;
+ auto key =
io::BlockFileCache::hash("ttl_hole_inherits_existing_hash_expiration");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ {
+ auto holder = cache.get_or_set(key, 0, 10, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ }
+
+ context.expiration_time = old_expiration_time + 300;
+ {
+ auto holder = cache.get_or_set(key, 0, 20, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 2);
+ EXPECT_EQ(blocks[0]->expiration_time(), old_expiration_time);
+ EXPECT_EQ(blocks[1]->expiration_time(), old_expiration_time);
+ ASSERT_TRUE(blocks[1]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[1]);
+ EXPECT_TRUE(fs::exists(key_dir_path(key, old_expiration_time) / "10"));
+ EXPECT_FALSE(fs::exists(key_dir_path(key, context.expiration_time) /
"10"));
+ }
+
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_removes_disk_only_orphan_file) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_removes_disk_only_orphan_file");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ auto file_path = key_dir_path(key, 0) / "0";
+ create_cache_file(file_path);
+ ASSERT_TRUE(fs::exists(file_path));
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 1);
+ EXPECT_FALSE(fs::exists(file_path));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_keeps_young_tmp_file) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+ config::file_cache_disk_scan_grace_seconds = 3600;
+
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_keeps_young_tmp_file");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ auto tmp_path = key_dir_path(key, 0) / "0_tmp";
+ create_cache_file(tmp_path);
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(tmp_path));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_keeps_active_writer_tmp_file) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::NORMAL;
+ context.query_id = query_id;
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_keeps_active_writer_tmp_file");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ auto holder = cache.get_or_set(key, 0, 5, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ ASSERT_TRUE(blocks[0]->append(Slice("abcde", 5)).ok());
+ auto tmp_path = key_dir_path(key, 0) / "0_tmp";
+ ASSERT_TRUE(fs::exists(tmp_path));
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(tmp_path));
+}
+
+TEST_F(BlockFileCacheTest,
disk_scan_repair_removes_old_tmp_with_downloaded_block) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::NORMAL;
+ context.query_id = query_id;
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_removes_old_tmp_with_downloaded_block");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ {
+ auto holder = cache.get_or_set(key, 0, 5, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ }
+ auto tmp_path = key_dir_path(key, 0) / "0_tmp";
+ create_cache_file(tmp_path);
+ ASSERT_TRUE(fs::exists(tmp_path));
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 1);
+ EXPECT_TRUE(fs::exists(key_dir_path(key, 0) / "0"));
+ EXPECT_FALSE(fs::exists(tmp_path));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_removes_non_canonical_ttl_dir) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.query_id = query_id;
+ context.expiration_time = UnixSeconds() + 300;
+ auto canonical_expiration_time = context.expiration_time;
+ auto stale_expiration_time = canonical_expiration_time + 300;
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_removes_non_canonical_ttl_dir");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ {
+ auto holder = cache.get_or_set(key, 0, 5, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ }
+ create_cache_file(key_dir_path(key, stale_expiration_time) / "100");
+ ASSERT_TRUE(fs::exists(key_dir_path(key, canonical_expiration_time)));
+ ASSERT_TRUE(fs::exists(key_dir_path(key, stale_expiration_time)));
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_dirs, 1);
+ EXPECT_TRUE(fs::exists(key_dir_path(key, canonical_expiration_time)));
+ EXPECT_FALSE(fs::exists(key_dir_path(key, stale_expiration_time)));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_keeps_mismatched_memory_block) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ TUniqueId query_id;
+ query_id.hi = 1;
+ query_id.lo = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 60;
+ settings.query_queue_elements = 10;
+ settings.ttl_queue_size = 60;
+ settings.ttl_queue_elements = 10;
+ settings.capacity = 120;
+ settings.max_file_block_size = 30;
+ settings.max_query_cache_size = 30;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::NORMAL;
+ context.query_id = query_id;
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_keeps_mismatched_memory_block");
+
+ io::BlockFileCache cache(cache_base_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ for (int i = 0; i < 100; i++) {
+ if (cache.get_async_open_success()) {
+ break;
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ }
+ ASSERT_TRUE(cache.get_async_open_success());
+
+ {
+ auto holder = cache.get_or_set(key, 0, 5, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_TRUE(blocks[0]->get_or_set_downloader() ==
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ }
+ auto file_path = key_dir_path(key, 0) / "0";
+ ASSERT_TRUE(fs::exists(file_path));
+ rewrite_cache_file(file_path, "0000000000");
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(file_path));
+}
+
+TEST_F(BlockFileCacheTest, ttl_mixed_expiration_hole_skips_cache) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ auto key =
io::BlockFileCache::hash("ttl_mixed_expiration_hole_skips_cache");
+ auto expiration1 = UnixSeconds() + 300;
+ auto expiration2 = expiration1 + 300;
+ create_cache_file(key_dir_path(key, expiration1) / "0");
+ create_cache_file(key_dir_path(key, expiration2) / "10");
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.expiration_time = expiration2 + 300;
+ auto holder = cache.get_or_set(key, 0, 15, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 3);
+ EXPECT_EQ(blocks[0]->state_unsafe(), io::FileBlock::State::DOWNLOADED);
+ EXPECT_EQ(blocks[1]->state_unsafe(), io::FileBlock::State::SKIP_CACHE);
+ EXPECT_EQ(blocks[2]->state_unsafe(), io::FileBlock::State::DOWNLOADED);
+ EXPECT_FALSE(fs::exists(key_dir_path(key, expiration1) / "5"));
+ EXPECT_FALSE(fs::exists(key_dir_path(key, expiration2) / "5"));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_skips_incomplete_startup_image) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+ auto sp = SyncPoint::get_instance();
+ Defer defer {[sp] { sp->clear_all_call_backs(); }};
+ sp->set_call_back("FSFileCacheStorage::load_cache_info_into_memory",
[](auto&& args) {
+ *try_any_cast<Status*>(args[0]) = Status::IOError("inject incomplete
startup image");
+ });
+ sp->enable_processing();
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_skips_incomplete_startup_image");
+ auto file_path = key_dir_path(key, 0) / "0";
+ create_cache_file(file_path);
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(file_path));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_repair_keeps_ttl_dir_on_exists_error) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+ auto sp = SyncPoint::get_instance();
+ Defer defer {[sp] { sp->clear_all_call_backs(); }};
+
+ auto key =
io::BlockFileCache::hash("disk_scan_repair_keeps_ttl_dir_on_exists_error");
+ auto canonical_expiration = UnixSeconds() + 300;
+ auto stale_expiration = canonical_expiration + 300;
+ io::CacheContext context;
+ ReadStatistics rstats;
+ context.stats = &rstats;
+ context.cache_type = io::FileCacheType::TTL;
+ context.expiration_time = canonical_expiration;
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+ {
+ auto holder = cache.get_or_set(key, 0, 5, context);
+ auto blocks = fromHolder(holder);
+ ASSERT_EQ(blocks.size(), 1);
+ ASSERT_EQ(blocks[0]->get_or_set_downloader(),
io::FileBlock::get_caller_id());
+ download(blocks[0]);
+ }
+ create_cache_file(key_dir_path(key, stale_expiration) / "10");
+
+ sp->set_call_back("BlockFileCache::disk_scan_canonical_path_exists",
[](auto&& args) {
+ *try_any_cast<Status*>(args[0]) = Status::IOError("inject exists
error");
+ });
+ sp->enable_processing();
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_dirs, 0);
+ EXPECT_TRUE(fs::exists(key_dir_path(key, stale_expiration)));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_dynamic_disable_cancels_round) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+ auto sp = SyncPoint::get_instance();
+ Defer defer {[sp] { sp->clear_all_call_backs(); }};
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+ auto key =
io::BlockFileCache::hash("disk_scan_dynamic_disable_cancels_round");
+ auto file_path = key_dir_path(key, 0) / "0";
+ create_cache_file(file_path);
+
+ std::atomic<int> token_count {0};
+ sp->set_call_back("FSFileCacheStorage::before_disk_scan_token",
[&](auto&&) {
+ if (++token_count == 2) {
+ config::enable_file_cache_disk_scan_repair = false;
+ }
+ });
+ sp->enable_processing();
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(file_path));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_does_not_follow_v2_prefix_symlink) {
+ static_assert(FSFileCacheStorage::USE_CACHE_VERSION2);
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ auto outside_prefix = caches_dir / "disk_scan_outside_prefix";
+ if (fs::exists(outside_prefix)) {
+ fs::remove_all(outside_prefix);
+ }
+ Defer cleanup {[&] {
+ std::error_code ec;
+ fs::remove_all(outside_prefix, ec);
+ }};
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+
+ auto key =
io::BlockFileCache::hash("disk_scan_does_not_follow_v2_prefix_symlink");
+ auto key_string = key.to_string();
+ auto outside_sentinel = outside_prefix / (key_string + "_0") / "0";
+ create_cache_file(outside_sentinel, "outside");
+ auto prefix_link =
+ fs::path(cache_base_path) / key_string.substr(0,
FSFileCacheStorage::KEY_PREFIX_LENGTH);
+ std::error_code ec;
+ fs::create_directory_symlink(outside_prefix, prefix_link, ec);
+ ASSERT_FALSE(ec) << ec.message();
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.candidates, 0);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(outside_sentinel));
+ EXPECT_EQ(fs::file_size(outside_sentinel), 7);
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_does_not_follow_key_dir_symlink) {
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ auto outside_key_dir = caches_dir / "disk_scan_outside_key_dir";
+ if (fs::exists(outside_key_dir)) {
+ fs::remove_all(outside_key_dir);
+ }
+ Defer cleanup {[&] {
+ std::error_code ec;
+ fs::remove_all(outside_key_dir, ec);
+ }};
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+
+ auto key =
io::BlockFileCache::hash("disk_scan_does_not_follow_key_dir_symlink");
+ auto outside_sentinel = outside_key_dir / "0";
+ create_cache_file(outside_sentinel, "outside");
+ auto key_dir_link = key_dir_path(key, 0);
+ fs::create_directories(key_dir_link.parent_path());
+ std::error_code ec;
+ fs::create_directory_symlink(outside_key_dir, key_dir_link, ec);
+ ASSERT_FALSE(ec) << ec.message();
+
+ auto result = run_disk_scan_once_for_test(cache);
+ EXPECT_EQ(result.candidates, 0);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_TRUE(fs::exists(outside_sentinel));
+ EXPECT_EQ(fs::file_size(outside_sentinel), 7);
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_mutation_rejects_replaced_prefix_symlink)
{
+ static_assert(FSFileCacheStorage::USE_CACHE_VERSION2);
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+ auto sp = SyncPoint::get_instance();
+ Defer clear_sync_points {[sp] { sp->clear_all_call_backs(); }};
+
+ auto outside_prefix = caches_dir / "disk_scan_mutation_outside_prefix";
+ if (fs::exists(outside_prefix)) {
+ fs::remove_all(outside_prefix);
+ }
+ Defer cleanup {[&] {
+ std::error_code ec;
+ fs::remove_all(outside_prefix, ec);
+ }};
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+
+ auto key =
io::BlockFileCache::hash("disk_scan_mutation_rejects_replaced_prefix_symlink");
+ auto file_path = key_dir_path(key, 0) / "0";
+ create_cache_file(file_path, "inside");
+ auto key_string = key.to_string();
+ auto outside_sentinel = outside_prefix / (key_string + "_0") / "0";
+ create_cache_file(outside_sentinel, "outside");
+ auto prefix_path = file_path.parent_path().parent_path();
+ auto saved_prefix = fs::path(prefix_path.native() + ".saved");
+
+ bool replaced = false;
+ sp->set_call_back("BlockFileCache::before_disk_scan_quarantine_rename",
[&](auto&&) {
+ ASSERT_FALSE(replaced);
+ std::error_code ec;
+ fs::rename(prefix_path, saved_prefix, ec);
+ ASSERT_FALSE(ec) << ec.message();
+ fs::create_directory_symlink(outside_prefix, prefix_path, ec);
+ ASSERT_FALSE(ec) << ec.message();
+ replaced = true;
+ });
+ sp->enable_processing();
+
+ auto result = run_disk_scan_once_for_test(cache);
+ ASSERT_TRUE(replaced);
+ EXPECT_EQ(result.repaired_files, 0);
+ EXPECT_EQ(result.skipped_candidates, 1);
+ EXPECT_TRUE(fs::exists(outside_sentinel));
+ EXPECT_EQ(fs::file_size(outside_sentinel), 7);
+
+ std::error_code ec;
+ fs::remove(prefix_path, ec);
+ ASSERT_FALSE(ec) << ec.message();
+ fs::rename(saved_prefix, prefix_path, ec);
+ ASSERT_FALSE(ec) << ec.message();
+ EXPECT_TRUE(fs::exists(file_path));
+}
+
+TEST_F(BlockFileCacheTest, disk_scan_does_not_cross_bind_mount_prefix) {
+ static_assert(FSFileCacheStorage::USE_CACHE_VERSION2);
+ if (fs::exists(cache_base_path)) {
+ fs::remove_all(cache_base_path);
+ }
+ fs::create_directories(cache_base_path);
+ DiskScanConfigGuard config_guard;
+
+ auto outside_prefix = caches_dir / "disk_scan_bind_mount_outside_prefix";
+ if (fs::exists(outside_prefix)) {
+ fs::remove_all(outside_prefix);
+ }
+ Defer cleanup {[&] {
+ std::error_code ec;
+ fs::remove_all(outside_prefix, ec);
+ }};
+
+ io::BlockFileCache cache(cache_base_path, disk_scan_test_settings());
+ ASSERT_TRUE(cache.initialize());
+ wait_for_async_open(cache);
+
+ auto key =
io::BlockFileCache::hash("disk_scan_does_not_cross_bind_mount_prefix");
+ auto key_string = key.to_string();
+ auto outside_sentinel = outside_prefix / (key_string + "_0") / "0";
+ create_cache_file(outside_sentinel, "outside");
+ auto prefix_mount =
+ fs::path(cache_base_path) / key_string.substr(0,
FSFileCacheStorage::KEY_PREFIX_LENGTH);
+ fs::create_directories(prefix_mount);
+ if (::mount(outside_prefix.c_str(), prefix_mount.c_str(), nullptr,
MS_BIND, nullptr) != 0) {
Review Comment:
[P1] Guard the bind-mount tests on non-Linux builds
This file is compiled into the macOS `doris_be_test`:
`be/test/CMakeLists.txt` globs every test `.cpp` and its macOS branch excludes
only `system_metrics_test.cpp`, while `be-ut-mac.yml` runs that target for this
PR's `be/**` changes. Darwin's `mount` has a different signature and does not
provide Linux `MS_BIND`, `umount2`, or `MNT_DETACH`, so this translation unit
fails to compile before the runtime `GTEST_SKIP()` can help (the second new
bind-mount test has the same issue). Please guard the `<sys/mount.h>` include
and both test bodies with the repository's Linux platform macro. Coordinate
that with the production `openat2` capability gate so the remaining repair
tests do not expect this Linux-only mutation path to succeed on macOS.
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -37,6 +47,202 @@
namespace doris::io {
+namespace {
+
+constexpr std::string_view DISK_SCAN_QUARANTINE_MARKER =
".disk_scan_quarantine.";
+
+class ScopedFd {
+public:
+ ScopedFd() = default;
+ explicit ScopedFd(int fd) : _fd(fd) {}
+ ~ScopedFd() { reset(); }
+ ScopedFd(const ScopedFd&) = delete;
+ ScopedFd& operator=(const ScopedFd&) = delete;
+ ScopedFd& operator=(ScopedFd&& other) noexcept {
+ reset(other.release());
+ return *this;
+ }
+
+ int get() const { return _fd; }
+ int release() { return std::exchange(_fd, -1); }
+ void reset(int fd = -1) {
+ if (_fd >= 0) {
+ ::close(_fd);
+ }
+ _fd = fd;
+ }
+
+private:
+ int _fd = -1;
+};
+
+struct DirCloser {
+ void operator()(DIR* dir) const {
+ if (dir != nullptr) {
+ ::closedir(dir);
+ }
+ }
+};
+
+using ScopedDir = std::unique_ptr<DIR, DirCloser>;
+
+struct DiskScanOpenHow {
+ uint64_t flags;
+ uint64_t mode;
+ uint64_t resolve;
+};
+
+constexpr uint64_t DISK_SCAN_RESOLVE_NO_XDEV = 0x01;
+// The Doris toolchain sysroot predates openat2 definitions. Linux assigned
437 to openat2
+// across the architectures supported by Doris.
+constexpr long DISK_SCAN_SYS_OPENAT2 = 437;
+
+int open_disk_scan_dir_no_follow_no_xdev(int parent_fd, const char* name) {
+ DiskScanOpenHow how {};
+ how.flags = O_RDONLY | O_DIRECTORY | O_CLOEXEC | O_NOFOLLOW;
+ how.resolve = DISK_SCAN_RESOLVE_NO_XDEV;
+ return static_cast<int>(::syscall(DISK_SCAN_SYS_OPENAT2, parent_fd, name,
&how, sizeof(how)));
Review Comment:
[P2] Gate repair on `openat2` capability
[`openat2` and `RESOLVE_NO_XDEV` are available only since Linux
5.6](https://man7.org/linux/man-pages/man2/openat2.2.html), while [current
Doris deployment prerequisites allow CentOS 7+ and Ubuntu
20.04+](https://doris.apache.org/docs/dev/install/deploy-manually/separating-storage-compute-deploy-manually/).
On a pre-5.6 kernel this call returns `ENOSYS`. Every V2 mutation needs a
parent walk here, and `drain_disk_scan_actions()` silently reduces the
pre-quarantine failure to `skipped_candidates`, so V2 repair scans and skips
every candidate. In V1 a top-level TTL directory can be renamed before
recursive deletion fails to reopen it; that failure is logged, but the
quarantine is stranded and retried in later rounds. Please probe
platform/capability support once and expose an unavailable feature as
disabled/error, or provide a fail-closed fallback that preserves both no-follow
and no-mount-crossing guarantees. Gate non-Linux repair tests on the same
capability and add an `ENOSYS` ca
se.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2298,6 +2305,450 @@ void BlockFileCache::run_background_ttl_gc() {
}
}
+void BlockFileCache::run_background_disk_scan_repair() {
+ Thread::set_self_name("run_disk_scan");
+ if (config::file_cache_disk_scan_initial_jitter_ms > 0) {
+ std::random_device rd;
+ std::mt19937 gen(rd());
+ std::uniform_int_distribution<int64_t> dist(
+ 0, std::max<int64_t>(0,
config::file_cache_disk_scan_initial_jitter_ms));
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(dist(gen)),
+ [this] { return
_close.load(std::memory_order_relaxed); });
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100),
+ [this] { return
_close.load(std::memory_order_relaxed); });
+ continue;
+ }
+
+ if (config::enable_file_cache_disk_scan_repair) {
+ auto result = run_disk_scan_repair_once();
+ if (result.repaired_files > 0 || result.repaired_dirs > 0) {
+ LOG(INFO) << "File cache disk scan repair finished. path=" <<
_cache_base_path
+ << " repaired_files=" << result.repaired_files
+ << " repaired_dirs=" << result.repaired_dirs
+ << " skipped_candidates=" <<
result.skipped_candidates;
+ }
+ }
+
+ int64_t interval_ms = config::file_cache_disk_scan_interval_ms;
+ TEST_SYNC_POINT_CALLBACK("BlockFileCache::set_disk_scan_sleep_time",
&interval_ms);
+ {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock,
+
std::chrono::milliseconds(std::max<int64_t>(interval_ms, 1)),
+ [this] { return
_close.load(std::memory_order_relaxed); });
+ if (_close) {
+ break;
+ }
+ }
+ }
+}
+
+DiskScanRoundResult BlockFileCache::run_disk_scan_repair_once() {
+ DiskScanRoundResult result;
+ if (!config::enable_file_cache_disk_scan_repair ||
+ _storage->get_type() != FileCacheStorageType::DISK ||
!_async_open_done ||
+ !_async_open_success) {
+ return result;
+ }
+ auto should_cancel = [this]() {
+ return _close.load(std::memory_order_relaxed) ||
+ !config::enable_file_cache_disk_scan_repair;
+ };
+
+ _disk_scan_scan_limiter->reset(std::max<int64_t>(0,
config::file_cache_disk_scan_scan_rate_qps),
+ std::max<int64_t>(0,
config::file_cache_disk_scan_scan_rate_qps),
+ 0);
+ _disk_scan_repair_limiter->reset(
+ std::max<int64_t>(0, config::file_cache_disk_scan_repair_rate_qps),
+ std::max<int64_t>(0,
config::file_cache_disk_scan_repair_rate_qps), 0);
+
+ int64_t duration_ns = 0;
+ {
+ SCOPED_RAW_TIMER(&duration_ns);
+ std::vector<DiskScanRepairAction> actions;
+ actions.reserve(std::min<int64_t>(
+ std::max<int64_t>(1,
config::file_cache_disk_scan_max_pending_repairs), 1024));
+ std::vector<DiskScanKeyDirEntry> ttl_dirs_for_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ if (should_cancel()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ if (!ttl_dirs_for_hash.empty() && ttl_dirs_for_hash.front().hash
!= entry.hash) {
+ finalize_disk_scan_ttl_checker(&ttl_dirs_for_hash, &actions,
&result);
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_for_hash,
&result);
+ if (actions.size() >= static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(&actions, &result);
+ }
+ return Status::OK();
+ };
+ auto on_block_file = [&](const DiskScanBlockFileEntry& entry) ->
Status {
+ if (should_cancel()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ ++result.scanned_files;
+ run_disk_scan_checkers(entry, &actions, &result);
+ if (actions.size() >= static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(&actions, &result);
+ }
+ return Status::OK();
+ };
+ auto on_quarantine = [&](const DiskScanQuarantineEntry& entry) ->
Status {
+ if (should_cancel()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ DiskScanRepairAction action;
+ action.type = entry.is_directory ?
DiskScanRepairActionType::DELETE_DIR
+ :
DiskScanRepairActionType::DELETE_FILE;
+ action.path = entry.path;
+ action.key_dir = entry.path;
+ action.reason = DiskScanRepairReason::QUARANTINE_GC;
+ actions.push_back(std::move(action));
+ ++result.candidates;
+ if (actions.size() >= static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(&actions, &result);
+ }
+ return Status::OK();
+ };
+ auto st = _storage->scan_disk_cache(on_key_dir, on_block_file,
on_quarantine,
+ _disk_scan_scan_limiter.get(),
should_cancel);
+ if (!st.ok() && !st.is<ErrorCode::CANCELLED>()) {
+ LOG(WARNING) << "Failed to scan file cache disk. path=" <<
_cache_base_path
+ << ", error=" << st;
+ }
+ if (!should_cancel()) {
+ finalize_disk_scan_ttl_checker(&ttl_dirs_for_hash, &actions,
&result);
+ drain_disk_scan_actions(&actions, &result);
+ } else {
+ result.skipped_candidates += actions.size();
+ actions.clear();
+ }
+ }
+
+ *_disk_scan_latency_us << (duration_ns / 1000);
+ *_disk_scan_rounds << 1;
+ *_disk_scan_scanned_prefix_dirs << result.scanned_prefix_dirs;
+ *_disk_scan_scanned_key_dirs << result.scanned_key_dirs;
+ *_disk_scan_scanned_files << result.scanned_files;
+ *_disk_scan_candidates << result.candidates;
+ *_disk_scan_repaired_files << result.repaired_files;
+ *_disk_scan_repaired_dirs << result.repaired_dirs;
+ *_disk_scan_skipped_candidates << result.skipped_candidates;
+ *_disk_scan_deleted_bytes << result.deleted_bytes;
+ return result;
+}
+
+void BlockFileCache::run_disk_scan_checkers(const DiskScanKeyDirEntry& entry,
+ std::vector<DiskScanRepairAction>*
/* actions */,
+ std::vector<DiskScanKeyDirEntry>*
ttl_dirs_for_hash,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ if (entry.expiration_time > 0) {
+ DCHECK(ttl_dirs_for_hash->empty() || ttl_dirs_for_hash->front().hash
== entry.hash);
+ ttl_dirs_for_hash->push_back(entry);
+ }
+}
+
+void BlockFileCache::run_disk_scan_checkers(const DiskScanBlockFileEntry&
entry,
+ std::vector<DiskScanRepairAction>*
actions,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_disk_memory_checker) {
+ return;
+ }
+
+ if (is_younger_than_grace(entry.identity,
config::file_cache_disk_scan_grace_seconds)) {
+ return;
+ }
+
+ if (entry.is_tmp) {
+ bool skip = false;
+ {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ auto file_iter = _files.find(entry.hash);
+ if (file_iter != _files.end()) {
+ auto block_iter = file_iter->second.find(entry.offset);
+ if (block_iter != file_iter->second.end()) {
+ std::lock_guard
block_lock(block_iter->second.file_block->_mutex);
+ auto state =
block_iter->second.file_block->state_unlock(block_lock);
+ skip = state == FileBlock::State::EMPTY ||
+ state == FileBlock::State::DOWNLOADING;
+ }
+ }
+ }
+ if (skip || _storage->has_active_writer(entry.hash, entry.offset)) {
+ return;
+ }
+ actions->push_back({DiskScanRepairActionType::DELETE_FILE, entry.hash,
entry.offset,
+ entry.expiration_time, entry.file_path,
entry.key_dir, entry.identity,
+ DiskScanRepairReason::OLD_TMP_FILE});
+ ++result->candidates;
+ return;
+ }
+
+ bool disk_only = false;
+ {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ auto file_iter = _files.find(entry.hash);
+ disk_only = file_iter == _files.end() ||
!file_iter->second.contains(entry.offset);
+ }
+ if (!disk_only) {
+ return;
+ }
+ actions->push_back({DiskScanRepairActionType::DELETE_FILE, entry.hash,
entry.offset,
+ entry.expiration_time, entry.file_path, entry.key_dir,
entry.identity,
+ DiskScanRepairReason::DISK_ONLY_FILE});
+ ++result->candidates;
+}
+
+void BlockFileCache::finalize_disk_scan_ttl_checker(
+ std::vector<DiskScanKeyDirEntry>* ttl_dirs_for_hash,
+ std::vector<DiskScanRepairAction>* actions, DiskScanRoundResult*
result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker ||
ttl_dirs_for_hash->empty()) {
+ ttl_dirs_for_hash->clear();
+ return;
+ }
+
+ auto& dirs = *ttl_dirs_for_hash;
+ [[maybe_unused]] auto ttl_group_size = dirs.size();
+ TEST_SYNC_POINT_CALLBACK("BlockFileCache::finalize_disk_scan_ttl_group",
&ttl_group_size);
+ const auto hash = dirs.front().hash;
+ std::sort(dirs.begin(), dirs.end(), [](const auto& lhs, const auto& rhs) {
+ return lhs.expiration_time < rhs.expiration_time;
+ });
+ dirs.erase(std::unique(dirs.begin(), dirs.end(),
+ [](const auto& lhs, const auto& rhs) {
+ return lhs.expiration_time ==
rhs.expiration_time;
+ }),
+ dirs.end());
+ uint64_t canonical_expiration_time = 0;
+ if (dirs.size() > 1 &&
+ disk_scan_hash_has_stable_canonical_expiration(hash,
&canonical_expiration_time)) {
+ bool canonical_on_disk = false;
+ for (const auto& dir : dirs) {
+ canonical_on_disk |= dir.expiration_time ==
canonical_expiration_time;
+ }
+ if (canonical_on_disk) {
+ for (const auto& dir : dirs) {
+ if (dir.expiration_time == canonical_expiration_time) {
+ continue;
+ }
+ actions->push_back({DiskScanRepairActionType::DELETE_DIR,
hash, std::nullopt,
+ dir.expiration_time, dir.key_dir,
dir.key_dir, std::nullopt,
+ DiskScanRepairReason::TTL_DUPLICATE_DIR});
+ ++result->candidates;
+ if (actions->size() >=
+ static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(actions, result);
+ }
+ }
+ }
+ }
+ ttl_dirs_for_hash->clear();
+}
+
+bool BlockFileCache::disk_scan_hash_has_stable_canonical_expiration(
+ const UInt128Wrapper& hash, uint64_t* canonical_expiration_time) {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ return disk_scan_hash_has_stable_canonical_expiration_unlocked(hash,
canonical_expiration_time,
+ cache_lock);
+}
+
+bool BlockFileCache::disk_scan_hash_has_stable_canonical_expiration_unlocked(
+ const UInt128Wrapper& hash, uint64_t* canonical_expiration_time,
+ std::lock_guard<std::mutex>& /* cache_lock */) {
+ auto file_iter = _files.find(hash);
+ if (file_iter == _files.end() || file_iter->second.empty()) {
+ return false;
+ }
+
+ bool has_canonical = false;
+ for (auto& [_, cell] : file_iter->second) {
Review Comment:
[P2] Bound canonical validation under the cache-wide mutex
This walks every cached offset for the hash and takes each block lock while
`_mutex` excludes all `get_or_set` and cache-metadata work. The duplicate
checker runs one walk uncharged by the repair limiter when forming the group,
then every stale-directory action repeats it during final quarantine
revalidation after consuming only one repair token. With B cached blocks and D
stale directories, that is B*(D+1) block-lock operations; each O(B) critical
section is internally unbounded and cannot observe shutdown or global repair
disable until it finishes. A highly fragmented large object can therefore stall
the cache data plane despite the action-QPS limit, while the new tests use only
one in-memory block per hash. Please maintain constant-time
canonical/mixed-state metadata or a generation, or use bounded snapshots plus
one versioned revalidation for the whole group; add a fragmented-hash
concurrency test that bounds request and shutdown latency.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
