github-actions[bot] commented on code in PR #64496:
URL: https://github.com/apache/doris/pull/64496#discussion_r3665098971
##########
be/src/common/config.cpp:
##########
@@ -1189,6 +1189,15 @@
DEFINE_mBool(enable_reader_dryrun_when_download_file_cache, "true");
DEFINE_mInt64(file_cache_background_monitor_interval_ms, "5000");
DEFINE_mInt64(file_cache_background_ttl_gc_interval_ms, "3000");
DEFINE_mInt64(file_cache_background_ttl_gc_batch, "1000");
+DEFINE_mBool(enable_file_cache_disk_scan_repair, "false");
+DEFINE_mInt64(file_cache_disk_scan_interval_ms, "1800000");
Review Comment:
[P2] Reject unsafe negative disk-scan settings
These new mutable signed settings have no validators. Downstream clamping
makes a negative interval run every 1 ms, negative scan/repair QPS become zero
(which disables limiting), and negative grace protect no young files. A
malformed live update can therefore turn repair into a near-continuous,
unlimited destructive pass. Please reject out-of-range values at both startup
and the live-update boundary, and define/test the intended zero semantics.
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -37,6 +47,185 @@
namespace doris::io {
+namespace {
+
+constexpr std::string_view DISK_SCAN_QUARANTINE_MARKER =
".disk_scan_quarantine.";
+
+class ScopedFd {
+public:
+ ScopedFd() = default;
+ explicit ScopedFd(int fd) : _fd(fd) {}
+ ~ScopedFd() { reset(); }
+ ScopedFd(const ScopedFd&) = delete;
+ ScopedFd& operator=(const ScopedFd&) = delete;
+ ScopedFd& operator=(ScopedFd&& other) noexcept {
+ reset(other.release());
+ return *this;
+ }
+
+ int get() const { return _fd; }
+ int release() { return std::exchange(_fd, -1); }
+ void reset(int fd = -1) {
+ if (_fd >= 0) {
+ ::close(_fd);
+ }
+ _fd = fd;
+ }
+
+private:
+ int _fd = -1;
+};
+
+struct DirCloser {
+ void operator()(DIR* dir) const {
+ if (dir != nullptr) {
+ ::closedir(dir);
+ }
+ }
+};
+
+using ScopedDir = std::unique_ptr<DIR, DirCloser>;
+
+Status open_disk_scan_parent_no_follow(int disk_scan_root_fd, const
std::string& cache_base_path,
+ const std::filesystem::path& target,
ScopedFd* parent_fd,
+ std::string* name) {
+ const auto root =
std::filesystem::path(cache_base_path).lexically_normal();
+ const auto relative = target.lexically_normal().lexically_relative(root);
+ if (relative.empty() || relative == "." || relative.is_absolute()) {
+ return Status::InvalidArgument("disk scan path is outside cache root:
{}", target.native());
+ }
+ for (const auto& component : relative) {
+ if (component == "." || component == ".." || component.empty()) {
+ return Status::InvalidArgument("disk scan path is outside cache
root: {}",
+ target.native());
+ }
+ }
+
+ auto target_name = relative.filename().native();
+ if (target_name.empty()) {
+ return Status::InvalidArgument("invalid disk scan path: {}",
target.native());
+ }
+
+ ScopedFd current(::fcntl(disk_scan_root_fd, F_DUPFD_CLOEXEC, 0));
+ if (current.get() < 0) {
+ const int error = errno;
+ return Status::IOError("failed to duplicate file cache root {},
error={}", root.native(),
+ std::strerror(error));
+ }
+ for (const auto& component : relative.parent_path()) {
+ const auto component_name = component.native();
+ int next = ::openat(current.get(), component_name.c_str(),
+ O_RDONLY | O_DIRECTORY | O_CLOEXEC | O_NOFOLLOW);
Review Comment:
[P1] Refuse mount crossings during rooted traversal
`O_NOFOLLOW` rejects symlinks, but it still opens bind mounts and nested
mount points. If a cache prefix/key directory is mounted from elsewhere, both
`lstat()` and rooted `fstatat()` see the same mounted inode, so the later
quarantine rename/unlink operates in that external tree; recursive quarantine
GC likewise descends into mounted children. Please reject mount transitions at
every component (for example with a mount-ID check or `RESOLVE_NO_XDEV`) and
add a bind-mount sentinel test.
##########
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),
Review Comment:
[P2] Preserve token debt across scan rounds
Each call to `reset(rate, rate, 0)` replaces the limiter with a bucket whose
`_remain_tokens` starts at a full one-second burst. Because the scan interval
is mutable and may be below one second, repair QPS 100 with a 100 ms interval
can admit 100 repairs every round (about 1000/s), rather than refilling only
the 10 tokens earned. Please preserve bucket state across rounds and
reconfigure without granting a fresh burst; a consecutive-round subsecond test
would catch this.
--
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]
