github-actions[bot] commented on code in PR #64496:
URL: https://github.com/apache/doris/pull/64496#discussion_r3614829924
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -841,6 +818,17 @@ FileBlocks BlockFileCache::split_range_into_cells(const
UInt128Wrapper& hash,
std::lock_guard<std::mutex>&
cache_lock) {
DCHECK(size > 0);
+ CacheContext block_context = context;
+ if (auto file_iter = _files.find(hash);
+ file_iter != _files.end() && !file_iter->second.empty()) {
+ auto& first_block = file_iter->second.begin()->second.file_block;
Review Comment:
[P1] Avoid choosing the TTL expiration by lowest offset
Startup can load disjoint offsets from two expiration directories for the
same hash: it deduplicates only identical hash/offset pairs, while
`_key_to_time` records whichever directory was inserted first. This `begin()`
is ordered by offset instead, so new holes can be written under a different
expiration than TTL GC tracks. The duplicate checker then refuses the hash
because its in-memory expirations already differ, leaving the split state
indefinitely. Please detect this mixed state and either reconcile it against
one explicit canonical expiration or avoid caching new holes until it is
repaired.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2283,341 @@ 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)));
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100));
+ 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)));
+ 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) {
+ return result;
+ }
+
+ _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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>
+ ttl_dirs_by_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_by_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 {
+ ++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());
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to scan file cache disk. path=" <<
_cache_base_path
+ << ", error=" << st;
+ }
+ finalize_disk_scan_ttl_checker(&ttl_dirs_by_hash, &actions, &result);
+ drain_disk_scan_actions(&actions, &result);
+ }
+
+ *_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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>*
+ ttl_dirs_by_hash,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ if (entry.expiration_time > 0) {
+ (*ttl_dirs_by_hash)[entry.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);
Review Comment:
[P1] Do not treat a partial startup load as authoritative
`_async_open_done` is set even when `load_cache_info_into_memory()` has
skipped a prefix, key directory, or file after an enumeration/stat error. If
that transient error clears before this scan, the valid persisted files are
visible here but absent from `_files`, so files older than the grace period are
classified as disk-only and deleted. This is after loading "completed", so the
existing gate does not help. Please track a successful/complete startup image
(or retry/reconcile skipped paths) before enabling destructive disk-vs-memory
repair.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2283,341 @@ 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)));
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100));
+ 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)));
+ 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) {
+ return result;
+ }
+
+ _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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>
+ ttl_dirs_by_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_by_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 {
+ ++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());
Review Comment:
[P1] Make a running scan cancellable during shutdown
Once this call starts, neither callback nor `scan_disk_cache()` observes
`_close`, and token-bucket sleeps are also uninterruptible. The destructor sets
`_close` and then joins this thread, so shutdown waits for the remainder of the
whole cache traversal. At the default 1 MiB block size and 2,000-QPS limit, a
populated 1 TiB cache is about 1,048,576 callbacks, or roughly 8.7 minutes.
Please thread cancellation through enumeration and limiter waits so destruction
can stop a round promptly, and cover it with a shutdown-during-scan test.
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -287,6 +340,142 @@ Status
FSFileCacheStorage::change_key_meta_expiration(const FileCacheKey& key,
return Status::OK();
}
+Status FSFileCacheStorage::scan_disk_cache(DiskScanKeyDirCallback on_key_dir,
+ DiskScanBlockFileCallback
on_block_file,
+ TokenBucketRateLimiterHolder*
scan_limiter) {
+ auto consume_scan_token = [&]() {
+ if (scan_limiter != nullptr) {
+ scan_limiter->add(1);
+ }
+ };
+
+ auto scan_key_dir = [&](const std::string& prefix,
+ const std::filesystem::path& key_dir_path) ->
Status {
+ std::error_code ec;
+ if (!std::filesystem::is_directory(key_dir_path, ec) || ec) {
+ return Status::OK();
+ }
+
+ UInt128Wrapper hash;
+ uint64_t expiration_time = 0;
+ if (!parse_key_dir_name(key_dir_path.filename().native(), &hash,
&expiration_time)) {
+ return Status::OK();
+ }
+
+ DiskScanKeyDirEntry key_dir_entry;
+ key_dir_entry.prefix = prefix;
+ key_dir_entry.hash = hash;
+ key_dir_entry.expiration_time = expiration_time;
+ key_dir_entry.key_dir = key_dir_path;
+ RETURN_IF_ERROR(on_key_dir(key_dir_entry));
+
+ std::vector<std::string> block_files;
+ consume_scan_token();
+ auto st = collect_directory_entries(key_dir_path, block_files);
Review Comment:
[P2] Rate-limit while enumerating each key directory
`collect_directory_entries()` exhausts the directory and stores every full
path before the following loop consumes any per-file tokens. A highly
fragmented object can therefore cause an unthrottled `readdir` burst and a very
large temporary vector despite a low scan QPS (at the default 1 MiB block size,
a 1 TiB object is roughly one million entries). Please stream or batch the
iterator and consume a token before each advancement/inspection, as the startup
loader already does with bounded batches.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2283,341 @@ 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)));
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100));
+ 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)));
+ 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) {
+ return result;
+ }
+
+ _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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>
Review Comment:
[P2] Bound TTL checker state during the scan
`file_cache_disk_scan_max_pending_repairs` bounds only `actions`; this map
retains a copied path entry for every positive-expiration key directory until
the entire cache scan returns. A high-cardinality TTL cache can therefore add
hundreds of MB of temporary nodes/vectors/paths each round. In the v2 layout
every directory for a hash is confined to its sorted three-digit prefix, so
completed hash or prefix groups can be finalized and erased incrementally.
Please bound this accumulator independently of the repair-action queue and add
a high-cardinality test.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2283,341 @@ 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)));
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100));
+ 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)));
+ 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) {
+ return result;
+ }
+
+ _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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>
+ ttl_dirs_by_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_by_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 {
+ ++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());
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to scan file cache disk. path=" <<
_cache_base_path
+ << ", error=" << st;
+ }
+ finalize_disk_scan_ttl_checker(&ttl_dirs_by_hash, &actions, &result);
+ drain_disk_scan_actions(&actions, &result);
+ }
+
+ *_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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>*
+ ttl_dirs_by_hash,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ if (entry.expiration_time > 0) {
Review Comment:
[P2] Include `_0` when detecting duplicate hash directories
A positive-expiration directory paired with `<hash>_0` produces only one
entry here, so the directory checker sees no duplicate. The disk-memory checker
also compares only hash/offset, ignoring expiration/path; when both directories
contain the same offset, both physical files look represented by the single
in-memory block. Thus either a canonical TTL plus stale NORMAL path or the
reverse survives every round. Please group expiration zero too and compare each
disk entry with the in-memory block's expiration/type before removing the
noncanonical path; add restart tests for both canonical directions.
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -287,6 +340,142 @@ Status
FSFileCacheStorage::change_key_meta_expiration(const FileCacheKey& key,
return Status::OK();
}
+Status FSFileCacheStorage::scan_disk_cache(DiskScanKeyDirCallback on_key_dir,
+ DiskScanBlockFileCallback
on_block_file,
+ TokenBucketRateLimiterHolder*
scan_limiter) {
+ auto consume_scan_token = [&]() {
+ if (scan_limiter != nullptr) {
+ scan_limiter->add(1);
+ }
+ };
+
+ auto scan_key_dir = [&](const std::string& prefix,
+ const std::filesystem::path& key_dir_path) ->
Status {
+ std::error_code ec;
+ if (!std::filesystem::is_directory(key_dir_path, ec) || ec) {
+ return Status::OK();
+ }
+
+ UInt128Wrapper hash;
+ uint64_t expiration_time = 0;
+ if (!parse_key_dir_name(key_dir_path.filename().native(), &hash,
&expiration_time)) {
+ return Status::OK();
+ }
+
+ DiskScanKeyDirEntry key_dir_entry;
+ key_dir_entry.prefix = prefix;
+ key_dir_entry.hash = hash;
+ key_dir_entry.expiration_time = expiration_time;
+ key_dir_entry.key_dir = key_dir_path;
+ RETURN_IF_ERROR(on_key_dir(key_dir_entry));
+
+ std::vector<std::string> block_files;
+ consume_scan_token();
+ auto st = collect_directory_entries(key_dir_path, block_files);
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to list file cache key dir " <<
key_dir_path
+ << ", error=" << st;
+ return Status::OK();
+ }
+ for (const auto& block_file : block_files) {
+ consume_scan_token();
+ std::filesystem::path block_file_path(block_file);
+ if (!std::filesystem::is_regular_file(block_file_path, ec) || ec) {
+ continue;
+ }
+ int64_t file_size = 0;
+ auto identity = get_file_identity(block_file_path, 0);
+ if (!identity) {
+ continue;
+ }
+ file_size = static_cast<int64_t>(identity->size);
+ size_t offset = 0;
+ bool is_tmp = false;
+ FileCacheType cache_type = FileCacheType::NORMAL;
+ if (!parse_filename_suffix_to_cache_type(fs,
block_file_path.filename().native(),
+ expiration_time,
file_size, &offset, &is_tmp,
+ &cache_type, false)) {
+ continue;
+ }
+
+ DiskScanBlockFileEntry block_entry;
+ block_entry.hash = hash;
+ block_entry.offset = offset;
+ block_entry.expiration_time = expiration_time;
+ block_entry.cache_type = cache_type;
+ block_entry.is_tmp = is_tmp;
+ block_entry.identity = *identity;
+ block_entry.file_path = block_file_path;
+ block_entry.key_dir = key_dir_path;
+ RETURN_IF_ERROR(on_block_file(block_entry));
+ }
+ return Status::OK();
+ };
+
+ if constexpr (USE_CACHE_VERSION2) {
+ std::vector<std::string> prefix_entries;
+ consume_scan_token();
+ RETURN_IF_ERROR(collect_directory_entries(_cache_base_path,
prefix_entries));
+ std::sort(prefix_entries.begin(), prefix_entries.end());
+ for (const auto& prefix_path_str : prefix_entries) {
+ std::filesystem::path prefix_path(prefix_path_str);
+ std::error_code ec;
+ if (!std::filesystem::is_directory(prefix_path, ec) || ec) {
+ continue;
+ }
+ auto prefix = prefix_path.filename().native();
+ if (prefix.size() != KEY_PREFIX_LENGTH) {
+ continue;
+ }
+
+ std::vector<std::string> key_dir_entries;
+ consume_scan_token();
+ auto st = collect_directory_entries(prefix_path, key_dir_entries);
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to list file cache prefix dir " <<
prefix_path
+ << ", error=" << st;
+ continue;
+ }
+ std::sort(key_dir_entries.begin(), key_dir_entries.end());
+ for (const auto& key_dir : key_dir_entries) {
+ RETURN_IF_ERROR(scan_key_dir(prefix, key_dir));
+ }
+ }
+ } else {
+ std::vector<std::string> key_dir_entries;
+ consume_scan_token();
+ RETURN_IF_ERROR(collect_directory_entries(_cache_base_path,
key_dir_entries));
+ std::sort(key_dir_entries.begin(), key_dir_entries.end());
+ for (const auto& key_dir : key_dir_entries) {
+ RETURN_IF_ERROR(scan_key_dir("", key_dir));
+ }
+ }
+ return Status::OK();
+}
+
+bool FSFileCacheStorage::has_active_writer(const UInt128Wrapper& hash, size_t
offset) {
+ std::lock_guard lock(_mtx);
+ return _key_to_writer.contains(std::make_pair(hash, offset));
+}
+
+bool FSFileCacheStorage::has_active_writer_for_hash(const UInt128Wrapper&
hash) {
+ std::lock_guard lock(_mtx);
+ for (const auto& [key, _] : _key_to_writer) {
+ if (key.first == hash) {
+ return true;
+ }
+ }
+ return false;
+}
+
+Status FSFileCacheStorage::delete_file_for_disk_scan(const
std::filesystem::path& path) {
+ return fs->delete_file(path);
Review Comment:
[P2] Remove empty key directories after repairing their last file
The normal `remove()` path deletes a key directory after its last file is
removed, but this repair path only unlinks the leaf. If a disk-only file is the
directory's sole entry, `<hash>_<expiration>` then remains indefinitely: there
is no memory cell to trigger normal cleanup, a lone TTL directory does not
qualify as a duplicate, and `_0` is excluded from that checker. At high orphan
cardinality this retains one directory inode per key and makes every later
round traverse all of them again. Please attempt a non-recursive
empty-directory removal after the unlink, and serialize it at `(hash,
expiration)` directory scope with all offsets' writer creation/finalization (or
retry append if the directory vanishes); a per-`(hash, offset)` claim is not
enough. Extend the orphan-file test to assert that the parent disappears.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2283,341 @@ 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)));
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100));
+ 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)));
+ 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) {
+ return result;
+ }
+
+ _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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>
+ ttl_dirs_by_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_by_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 {
+ ++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());
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to scan file cache disk. path=" <<
_cache_base_path
+ << ", error=" << st;
+ }
+ finalize_disk_scan_ttl_checker(&ttl_dirs_by_hash, &actions, &result);
+ drain_disk_scan_actions(&actions, &result);
+ }
+
+ *_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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>*
+ ttl_dirs_by_hash,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ if (entry.expiration_time > 0) {
+ (*ttl_dirs_by_hash)[entry.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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>*
+ ttl_dirs_by_hash,
+ std::vector<DiskScanRepairAction>* actions, DiskScanRoundResult*
result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ for (auto& [hash, dirs] : *ttl_dirs_by_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());
+ if (dirs.size() <= 1) {
+ continue;
+ }
+ uint64_t canonical_expiration_time = 0;
+ if (!disk_scan_hash_has_stable_canonical_expiration(hash,
&canonical_expiration_time)) {
+ continue;
+ }
+ bool canonical_on_disk = false;
+ for (const auto& dir : dirs) {
+ canonical_on_disk |= dir.expiration_time ==
canonical_expiration_time;
+ }
+ if (!canonical_on_disk) {
+ continue;
+ }
+ 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_by_hash->clear();
+}
+
+bool BlockFileCache::disk_scan_hash_has_stable_canonical_expiration(
+ const UInt128Wrapper& hash, uint64_t* canonical_expiration_time) {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ 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) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ _disk_scan_repair_limiter->add(1);
+ if (action.type == DiskScanRepairActionType::DELETE_FILE) {
+ if (!disk_scan_file_is_deletable(action)) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ auto st = _storage->delete_file_for_disk_scan(action.path);
+ if (st.ok()) {
+ ++result->repaired_files;
+ if (action.observed_identity) {
+ result->deleted_bytes += action.observed_identity->size;
+ }
+ } else {
+ ++result->skipped_candidates;
+ }
+ } else {
+ if (!disk_scan_dir_is_deletable(action)) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ auto st = _storage->delete_dir_for_disk_scan(action.path);
+ if (st.ok()) {
+ ++result->repaired_dirs;
+ } else {
+ ++result->skipped_candidates;
+ }
+ }
+ }
+ actions->clear();
+}
+
+bool BlockFileCache::disk_scan_file_is_deletable(const DiskScanRepairAction&
action) {
+ if (!action.offset) {
+ return false;
+ }
+ {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ 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) {
+ uint64_t canonical_expiration_time = 0;
+ if (!disk_scan_hash_has_stable_canonical_expiration(action.hash,
&canonical_expiration_time)) {
+ 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));
+ return std::filesystem::exists(canonical_path);
Review Comment:
[P1] Keep filesystem errors from escaping the repair thread
This is the throwing overload of `std::filesystem::exists`. A permission or
I/O error while checking an inconsistent cache path can throw
`filesystem_error`; there is no exception boundary in
`run_disk_scan_repair_once()` or the background thread, so the exception
invokes `std::terminate` and takes down the BE. Please use the `error_code`
overload (or the storage filesystem abstraction), log/measure the error, and
treat the directory as not deletable.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2285,6 +2283,341 @@ 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)));
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100));
+ 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)));
+ 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) {
+ return result;
+ }
+
+ _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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>
+ ttl_dirs_by_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_by_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 {
+ ++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());
+ if (!st.ok()) {
+ LOG(WARNING) << "Failed to scan file cache disk. path=" <<
_cache_base_path
+ << ", error=" << st;
+ }
+ finalize_disk_scan_ttl_checker(&ttl_dirs_by_hash, &actions, &result);
+ drain_disk_scan_actions(&actions, &result);
+ }
+
+ *_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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>*
+ ttl_dirs_by_hash,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ if (entry.expiration_time > 0) {
+ (*ttl_dirs_by_hash)[entry.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::unordered_map<UInt128Wrapper, std::vector<DiskScanKeyDirEntry>,
KeyHash>*
+ ttl_dirs_by_hash,
+ std::vector<DiskScanRepairAction>* actions, DiskScanRoundResult*
result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ for (auto& [hash, dirs] : *ttl_dirs_by_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());
+ if (dirs.size() <= 1) {
+ continue;
+ }
+ uint64_t canonical_expiration_time = 0;
+ if (!disk_scan_hash_has_stable_canonical_expiration(hash,
&canonical_expiration_time)) {
+ continue;
+ }
+ bool canonical_on_disk = false;
+ for (const auto& dir : dirs) {
+ canonical_on_disk |= dir.expiration_time ==
canonical_expiration_time;
+ }
+ if (!canonical_on_disk) {
+ continue;
+ }
+ 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_by_hash->clear();
+}
+
+bool BlockFileCache::disk_scan_hash_has_stable_canonical_expiration(
+ const UInt128Wrapper& hash, uint64_t* canonical_expiration_time) {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ 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) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ _disk_scan_repair_limiter->add(1);
+ if (action.type == DiskScanRepairActionType::DELETE_FILE) {
+ if (!disk_scan_file_is_deletable(action)) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ auto st = _storage->delete_file_for_disk_scan(action.path);
Review Comment:
[P1] Serialize validation with the final file unlink
All cache-map, active-writer, and inode checks have completed before this
pathname deletion. In that gap another request can insert the same hash/offset,
write its tmp file, and rename it over the observed final path; this call then
unlinks the new inode and leaves a `DOWNLOADED` in-memory block with no backing
file. This is distinct from the existing recursive TTL-directory race because
it is the new single-file repair path. Please hold a per-hash/offset repair
claim through an identity-aware unlink (and block publish/finalize on that
claim), with a sync-point race test.
--
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]
