github-actions[bot] commented on code in PR #64496:
URL: https://github.com/apache/doris/pull/64496#discussion_r3662595951
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -291,6 +390,346 @@ 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,
+ DiskScanQuarantineCallback
on_quarantine,
+ TokenBucketRateLimiterHolder*
scan_limiter,
+ DiskScanCancellationCallback
should_cancel) {
+ auto is_cancelled = [&]() { return should_cancel && should_cancel(); };
+ auto consume_scan_token = [&]() -> bool {
+ if (is_cancelled()) {
+ return false;
+ }
+ TEST_SYNC_POINT_CALLBACK("FSFileCacheStorage::before_disk_scan_token");
+ if (scan_limiter != nullptr) {
+ scan_limiter->add(1, should_cancel);
+ }
+ return !is_cancelled();
+ };
+
+ auto scan_key_dir = [&](const std::string& prefix,
+ const std::filesystem::path& key_dir_path) ->
Status {
+ if (is_cancelled()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ 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;
+ if (!consume_scan_token()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ auto st = collect_directory_entries(key_dir_path, block_files,
should_cancel);
+ if (!st.ok()) {
+ if (st.is<ErrorCode::CANCELLED>()) {
+ return st;
+ }
+ LOG(WARNING) << "Failed to list file cache key dir " <<
key_dir_path
+ << ", error=" << st;
+ return Status::OK();
+ }
+ for (const auto& block_file : block_files) {
+ if (!consume_scan_token()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ std::filesystem::path block_file_path(block_file);
+ std::string quarantined_file_name;
+ if (parse_disk_scan_quarantine(block_file_path,
&quarantined_file_name) &&
+ is_cache_block_file_name(quarantined_file_name)) {
+ auto file_status =
std::filesystem::symlink_status(block_file_path, ec);
+ if (ec) {
+ continue;
+ }
+ RETURN_IF_ERROR(on_quarantine(
+ {block_file_path,
+ file_status.type() ==
std::filesystem::file_type::directory}));
+ continue;
+ }
+ 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;
+ if (!consume_scan_token()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ RETURN_IF_ERROR(collect_directory_entries(_cache_base_path,
prefix_entries, should_cancel));
+ std::sort(prefix_entries.begin(), prefix_entries.end());
+ for (const auto& prefix_path_str : prefix_entries) {
+ if (is_cancelled()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ std::filesystem::path prefix_path(prefix_path_str);
+ std::error_code ec;
+ if (!std::filesystem::is_directory(prefix_path, ec) || ec) {
Review Comment:
[P1] Keep destructive scans inside the cache root
`std::filesystem::is_directory()` follows symlinks here (and in
`scan_key_dir()`), so a post-startup prefix such as `<cache>/abc -> /outside`
is traversed as a normal cache prefix. If `/outside/<hash>_0/0` is absent from
`_files`, the queued action retains the lexical `<cache>/abc/.../0` path, but
`stat`, quarantine `rename`, and deletion resolve the intermediate symlink and
remove the external file. The later `symlink_status()` checks only protect
quarantine entries/recursive children, not these ancestors. Please require real
no-follow prefix/key directories and enforce resolved cache-root containment
again at mutation time (ideally with descriptor-relative no-follow operations),
with v1/v2 symlink tests asserting an external sentinel is untouched.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2298,6 +2305,450 @@ void BlockFileCache::run_background_ttl_gc() {
}
}
+void BlockFileCache::run_background_disk_scan_repair() {
+ Thread::set_self_name("run_disk_scan");
+ if (config::file_cache_disk_scan_initial_jitter_ms > 0) {
+ std::random_device rd;
+ std::mt19937 gen(rd());
+ std::uniform_int_distribution<int64_t> dist(
+ 0, std::max<int64_t>(0,
config::file_cache_disk_scan_initial_jitter_ms));
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(dist(gen)),
+ [this] { return
_close.load(std::memory_order_relaxed); });
+ }
+
+ while (!_close) {
+ if (!_async_open_done) {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(100),
+ [this] { return
_close.load(std::memory_order_relaxed); });
+ continue;
+ }
+
+ if (config::enable_file_cache_disk_scan_repair) {
+ auto result = run_disk_scan_repair_once();
+ if (result.repaired_files > 0 || result.repaired_dirs > 0) {
+ LOG(INFO) << "File cache disk scan repair finished. path=" <<
_cache_base_path
+ << " repaired_files=" << result.repaired_files
+ << " repaired_dirs=" << result.repaired_dirs
+ << " skipped_candidates=" <<
result.skipped_candidates;
+ }
+ }
+
+ int64_t interval_ms = config::file_cache_disk_scan_interval_ms;
+ TEST_SYNC_POINT_CALLBACK("BlockFileCache::set_disk_scan_sleep_time",
&interval_ms);
+ {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock,
+
std::chrono::milliseconds(std::max<int64_t>(interval_ms, 1)),
+ [this] { return
_close.load(std::memory_order_relaxed); });
+ if (_close) {
+ break;
+ }
+ }
+ }
+}
+
+DiskScanRoundResult BlockFileCache::run_disk_scan_repair_once() {
+ DiskScanRoundResult result;
+ if (!config::enable_file_cache_disk_scan_repair ||
+ _storage->get_type() != FileCacheStorageType::DISK ||
!_async_open_done ||
+ !_async_open_success) {
+ return result;
+ }
+ auto should_cancel = [this]() {
+ return _close.load(std::memory_order_relaxed) ||
+ !config::enable_file_cache_disk_scan_repair;
+ };
+
+ _disk_scan_scan_limiter->reset(std::max<int64_t>(0,
config::file_cache_disk_scan_scan_rate_qps),
+ std::max<int64_t>(0,
config::file_cache_disk_scan_scan_rate_qps),
+ 0);
+ _disk_scan_repair_limiter->reset(
+ std::max<int64_t>(0, config::file_cache_disk_scan_repair_rate_qps),
+ std::max<int64_t>(0,
config::file_cache_disk_scan_repair_rate_qps), 0);
+
+ int64_t duration_ns = 0;
+ {
+ SCOPED_RAW_TIMER(&duration_ns);
+ std::vector<DiskScanRepairAction> actions;
+ actions.reserve(std::min<int64_t>(
+ std::max<int64_t>(1,
config::file_cache_disk_scan_max_pending_repairs), 1024));
+ std::vector<DiskScanKeyDirEntry> ttl_dirs_for_hash;
+
+ std::string last_prefix;
+ auto on_key_dir = [&](const DiskScanKeyDirEntry& entry) -> Status {
+ if (should_cancel()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ ++result.scanned_key_dirs;
+ if (last_prefix != entry.prefix) {
+ last_prefix = entry.prefix;
+ ++result.scanned_prefix_dirs;
+ }
+ if (!ttl_dirs_for_hash.empty() && ttl_dirs_for_hash.front().hash
!= entry.hash) {
+ finalize_disk_scan_ttl_checker(&ttl_dirs_for_hash, &actions,
&result);
+ }
+ run_disk_scan_checkers(entry, &actions, &ttl_dirs_for_hash,
&result);
+ if (actions.size() >= static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(&actions, &result);
+ }
+ return Status::OK();
+ };
+ auto on_block_file = [&](const DiskScanBlockFileEntry& entry) ->
Status {
+ if (should_cancel()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ ++result.scanned_files;
+ run_disk_scan_checkers(entry, &actions, &result);
+ if (actions.size() >= static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(&actions, &result);
+ }
+ return Status::OK();
+ };
+ auto on_quarantine = [&](const DiskScanQuarantineEntry& entry) ->
Status {
+ if (should_cancel()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ DiskScanRepairAction action;
+ action.type = entry.is_directory ?
DiskScanRepairActionType::DELETE_DIR
+ :
DiskScanRepairActionType::DELETE_FILE;
+ action.path = entry.path;
+ action.key_dir = entry.path;
+ action.reason = DiskScanRepairReason::QUARANTINE_GC;
+ actions.push_back(std::move(action));
+ ++result.candidates;
+ if (actions.size() >= static_cast<size_t>(std::max<int64_t>(
+ 1,
config::file_cache_disk_scan_max_pending_repairs))) {
+ drain_disk_scan_actions(&actions, &result);
+ }
+ return Status::OK();
+ };
+ auto st = _storage->scan_disk_cache(on_key_dir, on_block_file,
on_quarantine,
+ _disk_scan_scan_limiter.get(),
should_cancel);
+ if (!st.ok() && !st.is<ErrorCode::CANCELLED>()) {
+ LOG(WARNING) << "Failed to scan file cache disk. path=" <<
_cache_base_path
+ << ", error=" << st;
+ }
+ if (!should_cancel()) {
+ finalize_disk_scan_ttl_checker(&ttl_dirs_for_hash, &actions,
&result);
+ drain_disk_scan_actions(&actions, &result);
+ } else {
+ result.skipped_candidates += actions.size();
+ actions.clear();
+ }
+ }
+
+ *_disk_scan_latency_us << (duration_ns / 1000);
+ *_disk_scan_rounds << 1;
+ *_disk_scan_scanned_prefix_dirs << result.scanned_prefix_dirs;
+ *_disk_scan_scanned_key_dirs << result.scanned_key_dirs;
+ *_disk_scan_scanned_files << result.scanned_files;
+ *_disk_scan_candidates << result.candidates;
+ *_disk_scan_repaired_files << result.repaired_files;
+ *_disk_scan_repaired_dirs << result.repaired_dirs;
+ *_disk_scan_skipped_candidates << result.skipped_candidates;
+ *_disk_scan_deleted_bytes << result.deleted_bytes;
+ return result;
+}
+
+void BlockFileCache::run_disk_scan_checkers(const DiskScanKeyDirEntry& entry,
+ std::vector<DiskScanRepairAction>*
/* actions */,
+ std::vector<DiskScanKeyDirEntry>*
ttl_dirs_for_hash,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker) {
+ return;
+ }
+ if (entry.expiration_time > 0) {
+ DCHECK(ttl_dirs_for_hash->empty() || ttl_dirs_for_hash->front().hash
== entry.hash);
+ ttl_dirs_for_hash->push_back(entry);
+ }
+}
+
+void BlockFileCache::run_disk_scan_checkers(const DiskScanBlockFileEntry&
entry,
+ std::vector<DiskScanRepairAction>*
actions,
+ DiskScanRoundResult* result) {
+ if (!config::file_cache_disk_scan_enable_disk_memory_checker) {
+ return;
+ }
+
+ if (is_younger_than_grace(entry.identity,
config::file_cache_disk_scan_grace_seconds)) {
+ return;
+ }
+
+ if (entry.is_tmp) {
+ bool skip = false;
+ {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ auto file_iter = _files.find(entry.hash);
+ if (file_iter != _files.end()) {
+ auto block_iter = file_iter->second.find(entry.offset);
+ if (block_iter != file_iter->second.end()) {
+ std::lock_guard
block_lock(block_iter->second.file_block->_mutex);
+ auto state =
block_iter->second.file_block->state_unlock(block_lock);
+ skip = state == FileBlock::State::EMPTY ||
+ state == FileBlock::State::DOWNLOADING;
+ }
+ }
+ }
+ if (skip || _storage->has_active_writer(entry.hash, entry.offset)) {
+ return;
+ }
+ actions->push_back({DiskScanRepairActionType::DELETE_FILE, entry.hash,
entry.offset,
+ entry.expiration_time, entry.file_path,
entry.key_dir, entry.identity,
+ DiskScanRepairReason::OLD_TMP_FILE});
+ ++result->candidates;
+ return;
+ }
+
+ bool disk_only = false;
+ {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ auto file_iter = _files.find(entry.hash);
+ disk_only = file_iter == _files.end() ||
!file_iter->second.contains(entry.offset);
+ }
+ if (!disk_only) {
+ return;
+ }
+ actions->push_back({DiskScanRepairActionType::DELETE_FILE, entry.hash,
entry.offset,
+ entry.expiration_time, entry.file_path, entry.key_dir,
entry.identity,
+ DiskScanRepairReason::DISK_ONLY_FILE});
+ ++result->candidates;
+}
+
+void BlockFileCache::finalize_disk_scan_ttl_checker(
+ std::vector<DiskScanKeyDirEntry>* ttl_dirs_for_hash,
+ std::vector<DiskScanRepairAction>* actions, DiskScanRoundResult*
result) {
+ if (!config::file_cache_disk_scan_enable_ttl_duplicate_checker ||
ttl_dirs_for_hash->empty()) {
+ ttl_dirs_for_hash->clear();
+ return;
+ }
+
+ auto& dirs = *ttl_dirs_for_hash;
+ 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;
+ };
+ const bool is_leftover_quarantine = action.reason ==
DiskScanRepairReason::QUARANTINE_GC;
+ std::filesystem::path quarantine_path = action.path;
+ Status st = Status::OK();
+ if (!is_leftover_quarantine) {
+ _disk_scan_repair_limiter->add(1, should_cancel);
+ if (should_cancel()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ st = quarantine_disk_scan_action(action, &quarantine_path);
+ if (!st.ok()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ }
+ if (action.type == DiskScanRepairActionType::DELETE_FILE) {
+ if (is_leftover_quarantine) {
+ _disk_scan_repair_limiter->add(1, should_cancel);
+ if (should_cancel()) {
+ ++result->skipped_candidates;
+ continue;
+ }
+ }
+ st = _storage->delete_file_for_disk_scan(quarantine_path);
+ if (st.ok()) {
+ ++result->repaired_files;
+ if (action.observed_identity) {
Review Comment:
[P2] Make deleted-byte accounting match actual deletions
This is the only `deleted_bytes` increment. Recovered `QUARANTINE_GC`
actions have no identity, and `DELETE_DIR` returns no child-byte total, so a
round can delete a multi-TiB tree while the metric remains zero. In the other
direction, normal discovery/revalidation follows a leaf symlink via
`is_regular_file`/`stat`, but quarantine rename/remove unlinks only the link
while this branch charges the untouched target's full size. Please use
no-follow classification/identity for leaf entries and return the regular-file
bytes actually removed by recursive/resumed deletion, with undercount and
leaf-symlink overcount tests.
--
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]
