github-actions[bot] commented on code in PR #64496:
URL: https://github.com/apache/doris/pull/64496#discussion_r3653019067
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -851,23 +848,24 @@ FileBlocks BlockFileCache::split_range_into_cells(const
UInt128Wrapper& hash,
while (current_pos < end_pos_non_included) {
current_size = std::min(remaining_size, _max_file_block_size);
remaining_size -= current_size;
- state = try_reserve(hash, context, current_pos, current_size,
cache_lock)
+ state = try_reserve(hash, block_context, current_pos, current_size,
cache_lock)
Review Comment:
[P2] Skip reservation when metadata already forced no-cache
The mixed-image guard above can set `state` to `SKIP_CACHE`, but this still
calls `try_reserve()` for every hole. A successful reservation may
synchronously evict real LRU cells and add `(hash, offset, size)` to the query
context; the following branch then creates only a detached `SKIP_CACHE` block,
so no cache cell owns or releases that reservation. Reading a hole in an
inconsistent startup image can therefore evict useful data and consume query
quota even though the hole is intentionally not cached. Please bypass
`try_reserve()` when `state` is already `SKIP_CACHE`, and cover a
full-cache/query-limit case.
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -707,37 +972,54 @@ void
FSFileCacheStorage::load_cache_info_into_memory(BlockFileCache* _mgr) const
batch_load_buffer.clear();
};
- auto scan_file_cache = [&](std::filesystem::directory_iterator& key_it) {
+ auto scan_file_cache = [&](const std::vector<std::string>& key_paths) {
TEST_SYNC_POINT_CALLBACK("BlockFileCache::TmpFile1");
- for (; key_it != std::filesystem::directory_iterator(); ++key_it) {
- auto key_with_suffix = key_it->path().filename().native();
- auto delim_pos = key_with_suffix.find('_');
- DCHECK(delim_pos != std::string::npos);
- std::string key_str = key_with_suffix.substr(0, delim_pos);
- std::string expiration_time_str = key_with_suffix.substr(delim_pos
+ 1);
- auto hash =
UInt128Wrapper(vectorized::unhex_uint<uint128_t>(key_str.c_str()));
+ for (const auto& key_path_str : key_paths) {
+ std::filesystem::path key_path(key_path_str);
std::error_code ec;
- std::filesystem::directory_iterator offset_it(key_it->path(), ec);
- if (ec) [[unlikely]] {
- LOG(WARNING) << "filesystem error, failed to remove file,
file=" << key_it->path()
- << " error=" << ec.message();
+ if (!std::filesystem::is_directory(key_path, ec)) {
+ if (ec) {
+ auto st = Status::IOError("failed to stat cache key
directory {}, error={}",
+ key_path.native(), ec.message());
+ LOG(WARNING) << st;
+ remember_load_error(st);
+ }
+ continue;
+ }
+
+ UInt128Wrapper hash;
+ uint64_t expiration_time = 0;
+ if (!parse_key_dir_name(key_path.filename().native(), &hash,
&expiration_time)) {
+ continue;
+ }
+
+ std::vector<std::string> offset_paths;
+ auto list_st = collect_directory_entries(key_path, offset_paths);
Review Comment:
[P2] Keep startup directory discovery memory-bounded
This rewrite collects every block path in a key directory before the
10,000-entry `batch_load_buffer` can process anything; the prefix loop
similarly retains every key path in that prefix. The base loader streamed both
directory iterators, so a highly fragmented multi-TiB object can now allocate
millions of copied full paths (hundreds of MiB) during startup. This is a
separate instance from the existing periodic-scan enumeration thread, which
explicitly cited the startup loader as bounded. Please preserve the new error
accounting while consuming directory entries in bounded batches at both levels.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2292,417 @@ 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 st = _storage->scan_disk_cache(on_key_dir, on_block_file,
+ _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;
+ 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) {
+ std::lock_guard block_lock(cell.file_block->_mutex);
+ if (cell.file_block->state_unlock(block_lock) !=
FileBlock::State::DOWNLOADED) {
+ return false;
+ }
+ auto expiration_time = cell.file_block->expiration_time();
+ if (!has_canonical) {
+ *canonical_expiration_time = expiration_time;
+ has_canonical = true;
+ } else if (*canonical_expiration_time != expiration_time) {
+ return false;
+ }
+ }
+ return has_canonical;
+}
+
+void
BlockFileCache::drain_disk_scan_actions(std::vector<DiskScanRepairAction>*
actions,
+ DiskScanRoundResult* result) {
+ for (const auto& action : *actions) {
+ if (_close || !config::enable_file_cache_disk_scan_repair) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ auto should_cancel = [this]() {
+ return _close.load(std::memory_order_relaxed) ||
+ !config::enable_file_cache_disk_scan_repair;
+ };
+ _disk_scan_repair_limiter->add(1, should_cancel);
+ if (should_cancel()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ std::filesystem::path quarantine_path;
+ auto st = quarantine_disk_scan_action(action, &quarantine_path);
+ if (!st.ok()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ if (action.type == DiskScanRepairActionType::DELETE_FILE) {
+ st = _storage->delete_file_for_disk_scan(quarantine_path);
+ if (st.ok()) {
+ ++result->repaired_files;
+ if (action.observed_identity) {
+ result->deleted_bytes += action.observed_identity->size;
+ }
+ } else {
+ LOG(WARNING) << "Failed to delete quarantined file cache file.
path="
+ << quarantine_path << ", error=" << st;
+ ++result->skipped_candidates;
+ }
+ } else {
+ st = _storage->delete_dir_for_disk_scan(quarantine_path);
Review Comment:
[P1] Rate-limit and cancel recursive quarantine deletion
A whole stale TTL key directory is represented by one action and consumes
one repair token, but this call forwards to `std::filesystem::remove_all()`. A
multi-TiB object can put millions of block files behind that single token; once
deletion starts, neither shutdown nor the global/per-checker switches are
observed until the entire tree is gone, so repair QPS is bypassed and the cache
destructor can still block on the worker join. This is after quarantine,
separate from the existing traversal/limiter-wait thread. Please unlink
children in cancellable, rate-limited batches and make a partially drained
quarantine recoverable by a later round.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2292,417 @@ 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 st = _storage->scan_disk_cache(on_key_dir, on_block_file,
+ _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;
+ 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) {
+ std::lock_guard block_lock(cell.file_block->_mutex);
+ if (cell.file_block->state_unlock(block_lock) !=
FileBlock::State::DOWNLOADED) {
+ return false;
+ }
+ auto expiration_time = cell.file_block->expiration_time();
+ if (!has_canonical) {
+ *canonical_expiration_time = expiration_time;
+ has_canonical = true;
+ } else if (*canonical_expiration_time != expiration_time) {
+ return false;
+ }
+ }
+ return has_canonical;
+}
+
+void
BlockFileCache::drain_disk_scan_actions(std::vector<DiskScanRepairAction>*
actions,
+ DiskScanRoundResult* result) {
+ for (const auto& action : *actions) {
+ if (_close || !config::enable_file_cache_disk_scan_repair) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ auto should_cancel = [this]() {
+ return _close.load(std::memory_order_relaxed) ||
+ !config::enable_file_cache_disk_scan_repair;
+ };
+ _disk_scan_repair_limiter->add(1, should_cancel);
+ if (should_cancel()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ std::filesystem::path quarantine_path;
+ auto st = quarantine_disk_scan_action(action, &quarantine_path);
+ if (!st.ok()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ if (action.type == DiskScanRepairActionType::DELETE_FILE) {
+ st = _storage->delete_file_for_disk_scan(quarantine_path);
+ if (st.ok()) {
+ ++result->repaired_files;
+ if (action.observed_identity) {
+ result->deleted_bytes += action.observed_identity->size;
+ }
+ } else {
+ LOG(WARNING) << "Failed to delete quarantined file cache file.
path="
+ << quarantine_path << ", error=" << st;
+ ++result->skipped_candidates;
+ }
+ } else {
+ st = _storage->delete_dir_for_disk_scan(quarantine_path);
+ if (st.ok()) {
+ ++result->repaired_dirs;
+ } else {
+ LOG(WARNING) << "Failed to delete quarantined file cache
directory. path="
+ << quarantine_path << ", error=" << st;
+ ++result->skipped_candidates;
+ }
+ }
+ }
+ actions->clear();
+}
+
+bool BlockFileCache::disk_scan_file_is_deletable(const DiskScanRepairAction&
action) {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ return disk_scan_file_is_deletable_unlocked(action, cache_lock);
+}
+
+bool BlockFileCache::disk_scan_file_is_deletable_unlocked(
+ const DiskScanRepairAction& action, std::lock_guard<std::mutex>& /*
cache_lock */) {
+ if (!action.offset) {
+ return false;
+ }
+ auto file_iter = _files.find(action.hash);
+ if (file_iter != _files.end()) {
+ auto block_iter = file_iter->second.find(*action.offset);
+ if (block_iter != file_iter->second.end()) {
+ if (action.reason != DiskScanRepairReason::OLD_TMP_FILE) {
+ return false;
+ }
+ std::lock_guard block_lock(block_iter->second.file_block->_mutex);
+ auto state =
block_iter->second.file_block->state_unlock(block_lock);
+ if (state == FileBlock::State::EMPTY || state ==
FileBlock::State::DOWNLOADING) {
+ return false;
+ }
+ }
+ }
+ if (_storage->has_active_writer(action.hash, *action.offset)) {
+ return false;
+ }
+ auto identity = get_current_identity(action.path);
+ if (!identity) {
+ return false;
+ }
+ if (action.observed_identity && !same_identity(*action.observed_identity,
*identity)) {
+ return false;
+ }
+ return !is_younger_than_grace(*identity,
config::file_cache_disk_scan_grace_seconds);
+}
+
+bool BlockFileCache::disk_scan_dir_is_deletable(const DiskScanRepairAction&
action) {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ return disk_scan_dir_is_deletable_unlocked(action, cache_lock);
+}
+
+bool BlockFileCache::disk_scan_dir_is_deletable_unlocked(const
DiskScanRepairAction& action,
+
std::lock_guard<std::mutex>& cache_lock) {
+ uint64_t canonical_expiration_time = 0;
+ if (!disk_scan_hash_has_stable_canonical_expiration_unlocked(
+ action.hash, &canonical_expiration_time, cache_lock)) {
+ return false;
+ }
+ if (_storage->has_active_writer_for_hash(action.hash)) {
+ return false;
+ }
+ if (action.expiration_time == canonical_expiration_time) {
+ return false;
+ }
+ std::filesystem::path canonical_path =
+ action.path.parent_path() /
+ (action.hash.to_string() + "_" +
std::to_string(canonical_expiration_time));
+ bool exists = false;
+ auto st = _storage->exists_for_disk_scan(canonical_path, &exists);
+
TEST_SYNC_POINT_CALLBACK("BlockFileCache::disk_scan_canonical_path_exists",
&st, &exists);
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to check canonical file cache directory.
path=" << canonical_path
+ << ", error=" << st;
+ return false;
+ }
+ return exists;
+}
+
+Status BlockFileCache::quarantine_disk_scan_action(const DiskScanRepairAction&
action,
+ std::filesystem::path*
quarantine_path) {
+ TEST_SYNC_POINT_CALLBACK("BlockFileCache::before_disk_scan_quarantine",
&action);
+ SCOPED_CACHE_LOCK(_mutex, this);
+ bool deletable = action.type == DiskScanRepairActionType::DELETE_FILE
+ ? disk_scan_file_is_deletable_unlocked(action,
cache_lock)
+ : disk_scan_dir_is_deletable_unlocked(action,
cache_lock);
+ if (!deletable) {
+ return Status::InternalError("disk scan repair action is no longer
safe");
+ }
+
+ static std::atomic<uint64_t> quarantine_id {0};
+ *quarantine_path = action.path.parent_path() /
+ (action.path.filename().native() +
".disk_scan_quarantine." +
Review Comment:
[P2] Recover quarantines left by a crash or delete failure
The rename happens before deletion, but these generated siblings have no
recovery path if deletion fails or the BE exits in between. A quarantined key
directory is rejected by `parse_key_dir_name()` and becomes invisible forever,
while a file such as `0.disk_scan_quarantine.7` is permissively parsed as
offset 0 by `parse_filename_suffix_to_cache_type()` and can be loaded as a
`DOWNLOADED` cell whose canonical filename does not exist; later repair then
skips it because that offset is in `_files`. The process-static ID also resets
after restart. Please use a dedicated sweepable quarantine namespace (or
explicitly retry these names), make normal filename parsing require full
numeric/exact-suffix consumption, and add delete-error plus crash/restart
coverage.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2292,417 @@ 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 st = _storage->scan_disk_cache(on_key_dir, on_block_file,
+ _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;
+ 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) {
+ std::lock_guard block_lock(cell.file_block->_mutex);
+ if (cell.file_block->state_unlock(block_lock) !=
FileBlock::State::DOWNLOADED) {
+ return false;
+ }
+ auto expiration_time = cell.file_block->expiration_time();
+ if (!has_canonical) {
+ *canonical_expiration_time = expiration_time;
+ has_canonical = true;
+ } else if (*canonical_expiration_time != expiration_time) {
+ return false;
+ }
+ }
+ return has_canonical;
+}
+
+void
BlockFileCache::drain_disk_scan_actions(std::vector<DiskScanRepairAction>*
actions,
+ DiskScanRoundResult* result) {
+ for (const auto& action : *actions) {
+ if (_close || !config::enable_file_cache_disk_scan_repair) {
Review Comment:
[P2] Stop queued repairs when their checker is disabled
The per-checker switches are mutable, but after discovery this drain path
checks only `_close` and the global repair flag. If an operator disables
`file_cache_disk_scan_enable_disk_memory_checker` or
`file_cache_disk_scan_enable_ttl_duplicate_checker`, up to the full pending
batch can keep waiting on the limiter and deleting files/directories for that
disabled checker. Please make cancellation reason-aware before and during the
limiter wait, and add a test that flips each checker flag after its actions
have been queued.
##########
be/test/io/cache/block_file_cache_test.cpp:
##########
@@ -206,6 +209,110 @@ void complete_into_memory(const io::FileBlocksHolder&
holder) {
}
}
+fs::path key_dir_path(const UInt128Wrapper& key, uint64_t expiration_time) {
+ auto key_str = key.to_string();
+ if constexpr (FSFileCacheStorage::USE_CACHE_VERSION2) {
+ return fs::path(cache_base_path) /
+ key_str.substr(0, FSFileCacheStorage::KEY_PREFIX_LENGTH) /
+ (key_str + "_" + std::to_string(expiration_time));
+ }
+ return fs::path(cache_base_path) / (key_str + "_" +
std::to_string(expiration_time));
+}
+
+void create_cache_file(const fs::path& file_path, std::string_view content =
"00000") {
+
ASSERT_TRUE(global_local_filesystem()->create_directory(file_path.parent_path(),
false).ok());
+ FileWriterPtr writer;
+ ASSERT_TRUE(global_local_filesystem()->create_file(file_path,
&writer).ok());
+ ASSERT_TRUE(writer->append(Slice(content.data(), content.size())).ok());
+ ASSERT_TRUE(writer->close().ok());
+}
+
+void rewrite_cache_file(const fs::path& file_path, std::string_view content) {
+ std::ofstream file(file_path, std::ios::binary | std::ios::trunc);
+ ASSERT_TRUE(file.good());
+ file.write(content.data(), content.size());
+ ASSERT_TRUE(file.good());
+}
+
+UInt128Wrapper find_key_with_different_prefix(std::string_view base,
std::string_view excluded) {
+ for (size_t i = 0; i < 10000; ++i) {
+ auto key = io::BlockFileCache::hash(std::string(base) + "_" +
std::to_string(i));
+ if (key.to_string().substr(0, FSFileCacheStorage::KEY_PREFIX_LENGTH)
!= excluded) {
+ return key;
+ }
+ }
+ CHECK(false) << "failed to find key with a different prefix";
+}
+
+class DiskScanConfigGuard {
+public:
+ DiskScanConfigGuard()
+ : enable(config::enable_file_cache_disk_scan_repair),
+
initial_jitter_ms(config::file_cache_disk_scan_initial_jitter_ms),
+ scan_rate_qps(config::file_cache_disk_scan_scan_rate_qps),
+ repair_rate_qps(config::file_cache_disk_scan_repair_rate_qps),
+ grace_seconds(config::file_cache_disk_scan_grace_seconds),
+
max_pending_repairs(config::file_cache_disk_scan_max_pending_repairs),
+
enable_ttl_duplicate(config::file_cache_disk_scan_enable_ttl_duplicate_checker),
+
enable_disk_memory(config::file_cache_disk_scan_enable_disk_memory_checker) {
+ config::enable_file_cache_disk_scan_repair = false;
+ config::file_cache_disk_scan_initial_jitter_ms = 0;
+ config::file_cache_disk_scan_scan_rate_qps = 0;
+ config::file_cache_disk_scan_repair_rate_qps = 0;
+ config::file_cache_disk_scan_grace_seconds = 0;
+ config::file_cache_disk_scan_max_pending_repairs = 10000;
+ config::file_cache_disk_scan_enable_ttl_duplicate_checker = true;
+ config::file_cache_disk_scan_enable_disk_memory_checker = true;
+ }
+
+ ~DiskScanConfigGuard() {
+ config::enable_file_cache_disk_scan_repair = enable;
+ config::file_cache_disk_scan_initial_jitter_ms = initial_jitter_ms;
+ config::file_cache_disk_scan_scan_rate_qps = scan_rate_qps;
+ config::file_cache_disk_scan_repair_rate_qps = repair_rate_qps;
+ config::file_cache_disk_scan_grace_seconds = grace_seconds;
+ config::file_cache_disk_scan_max_pending_repairs = max_pending_repairs;
+ config::file_cache_disk_scan_enable_ttl_duplicate_checker =
enable_ttl_duplicate;
+ config::file_cache_disk_scan_enable_disk_memory_checker =
enable_disk_memory;
+ }
+
+private:
+ bool enable;
+ int64_t initial_jitter_ms;
+ int64_t scan_rate_qps;
+ int64_t repair_rate_qps;
+ int64_t grace_seconds;
+ int64_t max_pending_repairs;
+ bool enable_ttl_duplicate;
+ bool enable_disk_memory;
+};
+
+io::DiskScanRoundResult run_disk_scan_once_for_test(io::BlockFileCache& cache)
{
+ config::enable_file_cache_disk_scan_repair = true;
Review Comment:
[P2] Isolate synchronous scan tests from the background worker
Every initialized cache starts `run_background_disk_scan_repair()`, and this
guard sets initial jitter to zero. This helper then publishes the global enable
flag before invoking the same scan synchronously. A background worker delayed
in the 100 ms startup poll can observe that flag at the same time, quarantine
the fixture or consume a sync-point callback first, and make the explicit call
return counts such as zero instead of one. Please let the test-only entry
bypass the production enable gate without toggling the worker-visible flag, or
explicitly park/disable the instance's background scan thread.
--
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]
