github-actions[bot] commented on code in PR #64496:
URL: https://github.com/apache/doris/pull/64496#discussion_r3670971105
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -750,8 +1383,8 @@ void
FSFileCacheStorage::load_cache_info_into_memory(BlockFileCache* _mgr) const
BatchLoadArgs args;
args.ctx = context;
args.hash = hash;
- args.key_path = key_it->path();
- args.offset_path = offset_it->path();
+ args.key_path = key_path;
+ args.offset_path = offset_path;
Review Comment:
[P1] Keep duplicate sizes bound to the surviving TTL path
During startup the first `(hash, offset)` fixes this cell's expiration/path,
but later duplicate handling lets a same-offset file from another positive-TTL
directory resize that cell. The new duplicate checker then treats the first
expiration as stable and deletes the later directory. If the later file was
larger, the surviving cell advertises a larger range than its file;
`LocalFileReader` clamps at EOF and `FSFileCacheStorage::read()` enforces the
byte count only with `DCHECK`, so release reads can return success with the
tail unfilled. If it was smaller, range and cache accounting still come from
the deleted sibling. Please resize only from the same canonical path, or mark
the startup image inconsistent and re-stat the selected canonical file before
repair. Add a restart test with the same offset and unequal sizes in two
positive-TTL directories.
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -291,6 +507,391 @@ 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;
+ const auto key_dir_status =
std::filesystem::symlink_status(key_dir_path, ec);
+ if (ec || key_dir_status.type() !=
std::filesystem::file_type::directory) {
+ 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;
+ }
+ const auto block_file_status =
std::filesystem::symlink_status(block_file_path, ec);
+ if (ec || block_file_status.type() !=
std::filesystem::file_type::regular) {
+ 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;
+ const auto prefix_status =
std::filesystem::symlink_status(prefix_path, ec);
+ if (ec || prefix_status.type() !=
std::filesystem::file_type::directory) {
+ continue;
+ }
+ auto prefix = prefix_path.filename().native();
+ if (prefix.size() != KEY_PREFIX_LENGTH) {
+ continue;
+ }
+
+ std::vector<std::string> key_dir_entries;
+ if (!consume_scan_token()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ auto st = collect_directory_entries(prefix_path, key_dir_entries,
should_cancel);
+ if (!st.ok()) {
+ if (st.is<ErrorCode::CANCELLED>()) {
+ return st;
+ }
+ 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) {
+ std::filesystem::path key_dir_path(key_dir);
+ std::string quarantined_key_dir_name;
+ UInt128Wrapper quarantined_hash;
+ uint64_t quarantined_expiration = 0;
+ if (parse_disk_scan_quarantine(key_dir_path,
&quarantined_key_dir_name) &&
+ parse_key_dir_name(quarantined_key_dir_name,
&quarantined_hash,
+ &quarantined_expiration)) {
+ auto key_dir_status =
std::filesystem::symlink_status(key_dir_path, ec);
+ if (ec) {
+ continue;
+ }
+ RETURN_IF_ERROR(on_quarantine(
+ {key_dir_path,
+ key_dir_status.type() ==
std::filesystem::file_type::directory}));
+ continue;
+ }
+ RETURN_IF_ERROR(scan_key_dir(prefix, key_dir));
+ }
+ }
+ } else {
+ std::vector<std::string> key_dir_entries;
+ if (!consume_scan_token()) {
+ return Status::Cancelled("file cache disk scan cancelled");
+ }
+ RETURN_IF_ERROR(
+ collect_directory_entries(_cache_base_path, key_dir_entries,
should_cancel));
+ std::sort(key_dir_entries.begin(), key_dir_entries.end());
+ for (const auto& key_dir : key_dir_entries) {
+ std::filesystem::path key_dir_path(key_dir);
+ std::string quarantined_key_dir_name;
+ UInt128Wrapper quarantined_hash;
+ uint64_t quarantined_expiration = 0;
+ if (parse_disk_scan_quarantine(key_dir_path,
&quarantined_key_dir_name) &&
+ parse_key_dir_name(quarantined_key_dir_name, &quarantined_hash,
+ &quarantined_expiration)) {
+ std::error_code ec;
+ auto key_dir_status =
std::filesystem::symlink_status(key_dir_path, ec);
+ if (ec) {
+ continue;
+ }
+ RETURN_IF_ERROR(on_quarantine(
+ {key_dir_path,
+ key_dir_status.type() ==
std::filesystem::file_type::directory}));
+ continue;
+ }
+ 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);
+ auto key = std::make_pair(hash, offset);
+ return _key_to_writer.contains(key) || _publishing_writers.contains(key);
+}
+
+bool FSFileCacheStorage::has_active_writer_for_hash(const UInt128Wrapper&
hash) {
+ std::lock_guard lock(_mtx);
+ for (const auto& [key, _] : _key_to_writer) {
Review Comment:
[P2] Keep cache-wide repair validation independent of unrelated writers
Each stale TTL-directory action reaches this scan while
`quarantine_disk_scan_action()` holds `BlockFileCache::_mutex`. Walking both
cache-wide writer containers makes `W` active/publishing writers and `D` stale
directories cost `O(W*D)` unrelated comparisons with all `get_or_set()` and
metadata work excluded. This can also wait on the same storage `_mtx` that
`append()` holds across `create_directory()` and `create_file()`, so slow
filesystem creation for another hash stalls the cache under `_mutex`. The
per-action repair token does not bound either wait. Please maintain an O(1)
per-hash active/publishing count or generation that does not wait behind
unrelated writer creation, and add a many-writer/many-stale-directory latency
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]
