From 5665501e8e91c8a1d49550724494308e816dac5f Mon Sep 17 00:00:00 2001 From: gavinchou Date: Sat, 11 Jul 2026 09:40:52 +0800 Subject: [PATCH] [improvement](be) Support dynamic file cache capacity ### What problem does this PR solve? Issue Number: N/A Related PR: N/A Problem Summary: File cache paths configured with zero capacity only resolved the disk size at startup. Runtime disk expansion or shrink did not update the effective capacity or queue limits, manual reset returned free-form text, and synchronous shrink could hold the cache lock while deleting many blocks. This change adds explicit AUTO/MANUAL capacity policy, a shared disk snapshot and resize planner, atomic BlockFileCache quota publication, generation checks for automatic refresh, batched asynchronous shrink, on-demand Factory aggregation, and structured reset/info HTTP responses. ### Release note Support automatic runtime file cache capacity updates for disk-backed paths configured with zero capacity. The file cache reset API now accepts zero to restore AUTO mode and returns structured capacity details, and op=info exposes per-path capacity, disk, queue, and resize state. ### Check List (For Author) - Test: - Build: source custom_env.sh && sh build.sh --be -j 16 - Unit Test: BlockFileCacheTest.* and FileCacheActionTest.* (141 tests, 138 passed, 3 disabled) - New Unit Test smoke: capacity policy, SyncPoint disk resize, Factory preflight, and FileCacheActionTest.* (16 passed) - Full BE Unit Test: 9495 tests, 9438 passed, 57 skipped - Format: build-support/check-format.sh - Behavior changed: Yes. Disk cache capacity 0 is a persistent AUTO mode, reset capacity 0 restores AUTO mode, queue quotas update with capacity, and shrink converges asynchronously. - Does this need documentation: Yes. File Cache operations documentation should describe AUTO mode, structured reset output, and op=info. --- be/src/io/cache/block_file_cache.cpp | 516 +++++++++++++----- be/src/io/cache/block_file_cache.h | 33 +- be/src/io/cache/block_file_cache_factory.cpp | 297 ++++++---- be/src/io/cache/block_file_cache_factory.h | 11 +- be/src/io/cache/file_cache_common.cpp | 269 ++++++++- be/src/io/cache/file_cache_common.h | 134 +++++ .../service/http/action/file_cache_action.cpp | 101 +++- be/src/storage/options.cpp | 8 +- be/test/io/cache/block_file_cache_test.cpp | 343 +++++++++++- .../io/cache/block_file_cache_test_common.h | 1 - .../cache/block_file_cache_test_lru_dump.cpp | 1 - ...ched_remote_file_reader_lock_wait_test.cpp | 3 - .../cached_remote_file_reader_peer_test.cpp | 1 - .../cache/cached_remote_file_reader_test.cpp | 1 - .../io/fs/packed_file_concurrency_test.cpp | 1 - .../service/http/file_cache_action_test.cpp | 134 ++++- 16 files changed, 1560 insertions(+), 294 deletions(-) diff --git a/be/src/io/cache/block_file_cache.cpp b/be/src/io/cache/block_file_cache.cpp index df8bfb12b70731..3aeb61131a9be5 100644 --- a/be/src/io/cache/block_file_cache.cpp +++ b/be/src/io/cache/block_file_cache.cpp @@ -22,36 +22,28 @@ #include +#include // IWYU pragma: keep #include #include #include -#include - -#include "common/status.h" -#include "cpp/sync_point.h" -#include "runtime/exec_env.h" - -#if defined(__APPLE__) -#include -#else -#include -#endif - -#include // IWYU pragma: keep #include #include +#include #include "common/cast_set.h" #include "common/check.h" #include "common/config.h" #include "common/logging.h" +#include "common/status.h" #include "core/uint128.h" +#include "cpp/sync_point.h" #include "exec/common/sip_hash.h" #include "io/cache/block_file_cache_ttl_mgr.h" #include "io/cache/file_block.h" #include "io/cache/file_cache_common.h" #include "io/cache/fs_file_cache_storage.h" #include "io/cache/mem_file_cache_storage.h" +#include "runtime/exec_env.h" #include "runtime/runtime_profile.h" #include "util/concurrency_stats.h" #include "util/stack_util.h" @@ -193,10 +185,28 @@ BlockFileCache::BlockFileCache(const std::string& cache_base_path, : _cache_base_path(cache_base_path), _capacity(cache_settings.capacity), _max_file_block_size(cache_settings.max_file_block_size) { + _capacity_state.policy = make_file_cache_capacity_policy(cache_settings); + auto& policy = _capacity_state.policy; + _max_file_block_size = policy.max_file_block_size; + _capacity_state.last_resize.time_ms = UnixMillis(); + _cur_cache_size_metrics = std::make_shared>(_cache_base_path.c_str(), "file_cache_cache_size", 0); _cache_capacity_metrics = std::make_shared>( _cache_base_path.c_str(), "file_cache_capacity", _capacity); + _resize_pending_bytes_metrics = std::make_shared>( + _cache_base_path.c_str(), "file_cache_resize_pending_bytes", 0); + _disk_total_capacity_metrics = std::make_shared>( + _cache_base_path.c_str(), "file_cache_disk_total_capacity", 0); + _disk_available_capacity_metrics = std::make_shared>( + _cache_base_path.c_str(), "file_cache_disk_available_capacity", 0); + _capacity_mode_metrics = std::make_shared>( + _cache_base_path.c_str(), "file_cache_capacity_mode", + policy.mode == FileCacheCapacityMode::AUTO ? 1 : 0); + _auto_resize_success_metrics = std::make_shared>( + _cache_base_path.c_str(), "file_cache_auto_resize_success_total"); + _auto_resize_failure_metrics = std::make_shared>( + _cache_base_path.c_str(), "file_cache_auto_resize_failure_total"); _cur_ttl_cache_size_metrics = std::make_shared>( _cache_base_path.c_str(), "file_cache_ttl_cache_size", 0); _cur_normal_queue_element_count_metrics = std::make_shared>( @@ -1708,6 +1718,7 @@ void BlockFileCache::remove(FileBlockSPtr file_block, T& cache_lock, U& block_lo if (FileCacheType::TTL == type) { _cur_ttl_size -= file_block->range().size(); } + update_resize_pending_unlocked(cache_lock); auto it = _files.find(hash); if (it != _files.end()) { it->second.erase(file_block->offset()); @@ -1939,111 +1950,291 @@ void BlockFileCache::change_cache_type(const UInt128Wrapper& hash, size_t offset } } -// @brief: get a path's disk capacity used percent, inode used percent -// @param: path -// @param: percent.first disk used percent, percent.second inode used percent -int disk_used_percentage(const std::string& path, std::pair* percent) { - struct statfs stat; - int ret = statfs(path.c_str(), &stat); - if (ret != 0) { - return ret; +std::string BlockFileCache::reset_capacity(size_t new_capacity) { + return reset_capacity(new_capacity, false); +} + +std::string BlockFileCache::reset_capacity(size_t new_capacity, bool auto_capacity) { + BlockFileCacheResetRequest request; + request.requested_capacity = auto_capacity ? 0 : new_capacity; + request.source = FileCacheResizeSource::HTTP; + FileCachePathResizeResult result; + auto st = reset_capacity(request, &result); + if (!st.ok()) { + return st.to_string(); } - // https://github.com/coreutils/coreutils/blob/master/src/df.c#L1195 - // v->used = stat.f_blocks - stat.f_bfree - // nonroot_total = stat.f_blocks - stat.f_bfree + stat.f_bavail - uintmax_t u100 = (stat.f_blocks - stat.f_bfree) * 100; - uintmax_t nonroot_total = stat.f_blocks - stat.f_bfree + stat.f_bavail; - int capacity_percentage = int(u100 / nonroot_total + (u100 % nonroot_total != 0)); + return fmt::format( + "finish reset_capacity, path={} old_capacity={} new_capacity={} mode={} " + "pending_eviction_bytes={}", + result.path, result.old_capacity, result.new_capacity, + file_cache_capacity_mode_to_string(result.mode), result.pending_eviction_bytes); +} - unsigned long long inode_free = stat.f_ffree; - unsigned long long inode_total = stat.f_files; - int inode_percentage = cast_set(inode_free * 100 / inode_total); - percent->first = capacity_percentage; - percent->second = 100 - inode_percentage; +Status BlockFileCache::prepare_reset_capacity( + const BlockFileCacheResetRequest& request, + const std::optional& observed_disk, FileCacheResizePlan* plan) const { + DCHECK(plan != nullptr); + FileCacheCapacityPolicy policy; + { + std::lock_guard cache_lock(_mutex); + policy = _capacity_state.policy; + } - // Add sync point for testing - TEST_SYNC_POINT_CALLBACK("BlockFileCache::disk_used_percentage:1", percent); + std::optional disk_state = observed_disk; + if (policy.storage != "memory" && !disk_state.has_value()) { + FileCacheDiskState state; + RETURN_IF_ERROR(get_file_cache_disk_state(_cache_base_path, &state)); + disk_state = state; + } + return build_file_cache_resize_plan(policy, request, disk_state, plan); +} - return 0; +Status BlockFileCache::apply_reset_capacity(const FileCacheResizePlan& plan, + FileCachePathResizeResult* result) { + DCHECK(result != nullptr); + SCOPED_CACHE_LOCK(_mutex, this); + return apply_reset_capacity_unlocked(plan, result, cache_lock); } -std::string BlockFileCache::reset_capacity(size_t new_capacity) { - using namespace std::chrono; - int64_t space_released = 0; - size_t old_capacity = 0; - std::stringstream ss; - ss << "finish reset_capacity, path=" << _cache_base_path; - auto adjust_start_time = steady_clock::time_point(); +Status BlockFileCache::reset_capacity(const BlockFileCacheResetRequest& request, + FileCachePathResizeResult* result) { + FileCacheResizePlan plan; + auto st = prepare_reset_capacity(request, std::nullopt, &plan); + if (!st.ok()) { + record_resize_failure(request.source, st); + return st; + } + st = apply_reset_capacity(plan, result); + if (!st.ok()) { + record_resize_failure(request.source, st); + } + return st; +} + +// Publish the complete resize state under one lock so readers never observe a partial update. +// NOLINTNEXTLINE(readability-function-size) +Status BlockFileCache::apply_reset_capacity_unlocked(const FileCacheResizePlan& plan, + FileCachePathResizeResult* result, + std::lock_guard& cache_lock) { + *result = FileCachePathResizeResult {}; + result->path = _cache_base_path; + result->old_capacity = _capacity; + result->mode = _capacity_state.policy.mode; + result->requested_capacity = _capacity_state.policy.requested_capacity; + result->clamped_by_disk = plan.clamped_by_disk; + if (plan.disk_state.has_value()) { + result->disk_total_capacity = plan.disk_state->total_capacity; + result->disk_available_capacity = plan.disk_state->available_capacity; + update_disk_state_unlocked(*plan.disk_state, cache_lock); + } + + if (plan.expected_generation.has_value() && + (*plan.expected_generation != _capacity_state.generation || + (plan.source == FileCacheResizeSource::AUTO_REFRESH && + _capacity_state.policy.mode != FileCacheCapacityMode::AUTO))) { + result->new_capacity = _capacity; + result->used_bytes = _cur_cache_size; + result->pending_eviction_bytes = + _cur_cache_size > _capacity ? _cur_cache_size - _capacity : 0; + result->skipped = true; + return Status::OK(); + } + + const auto& settings = plan.next_settings; + const auto& next_policy = plan.next_policy; + const bool changed = + _capacity != settings.capacity || _capacity_state.policy.mode != next_policy.mode || + _capacity_state.policy.requested_capacity != next_policy.requested_capacity || + _disposable_queue.max_size != settings.disposable_queue_size || + _disposable_queue.max_element_size != settings.disposable_queue_elements || + _normal_queue.max_size != settings.query_queue_size || + _normal_queue.max_element_size != settings.query_queue_elements || + _index_queue.max_size != settings.index_queue_size || + _index_queue.max_element_size != settings.index_queue_elements || + _ttl_queue.max_size != settings.ttl_queue_size || + _ttl_queue.max_element_size != settings.ttl_queue_elements; + + if (changed) { + _capacity_state.policy = next_policy; + ++_capacity_state.generation; + _capacity = settings.capacity; + _disposable_queue.max_size = settings.disposable_queue_size; + _disposable_queue.max_element_size = settings.disposable_queue_elements; + _normal_queue.max_size = settings.query_queue_size; + _normal_queue.max_element_size = settings.query_queue_elements; + _index_queue.max_size = settings.index_queue_size; + _index_queue.max_element_size = settings.index_queue_elements; + _ttl_queue.max_size = settings.ttl_queue_size; + _ttl_queue.max_element_size = settings.ttl_queue_elements; + _capacity_state.last_resize = { + .source = plan.source, .time_ms = UnixMillis(), .status = "OK", .message = ""}; + _cache_capacity_metrics->set_value(_capacity); + _capacity_mode_metrics->set_value(next_policy.mode == FileCacheCapacityMode::AUTO ? 1 : 0); + update_resize_pending_unlocked(cache_lock); + if (plan.source == FileCacheResizeSource::AUTO_REFRESH) { + *_auto_resize_success_metrics << 1; + } + + const auto log_message = fmt::format( + "File cache capacity changed. source={} mode={} path={} old_capacity={} " + "new_capacity={} disk_total_capacity={} pending_eviction_bytes={}", + file_cache_resize_source_to_string(plan.source), + file_cache_capacity_mode_to_string(next_policy.mode), _cache_base_path, + result->old_capacity, _capacity, result->disk_total_capacity, + _capacity_state.pending_eviction_bytes); + if (plan.clamped_by_disk) { + LOG(WARNING) << log_message << " requested_capacity=" << next_policy.requested_capacity; + } else { + LOG(INFO) << log_message; + } + } else if (plan.source == FileCacheResizeSource::STARTUP || + plan.source == FileCacheResizeSource::RELOAD) { + _capacity_state.last_resize = { + .source = plan.source, .time_ms = UnixMillis(), .status = "OK", .message = ""}; + } + + result->mode = _capacity_state.policy.mode; + result->requested_capacity = _capacity_state.policy.requested_capacity; + result->new_capacity = _capacity; + result->used_bytes = _cur_cache_size; + result->pending_eviction_bytes = _cur_cache_size > _capacity ? _cur_cache_size - _capacity : 0; + result->changed = changed; + return Status::OK(); +} + +void BlockFileCache::update_resize_pending_unlocked(std::lock_guard& /*cache_lock*/) { + _capacity_state.pending_eviction_bytes = + _cur_cache_size > _capacity ? _cur_cache_size - _capacity : 0; + _resize_pending_bytes_metrics->set_value(_capacity_state.pending_eviction_bytes); +} + +void BlockFileCache::update_disk_state_unlocked(const FileCacheDiskState& disk_state, + std::lock_guard& /*cache_lock*/) { + _capacity_state.last_disk_state = disk_state; + _disk_total_capacity_metrics->set_value(disk_state.total_capacity); + _disk_available_capacity_metrics->set_value(disk_state.available_capacity); +} + +void BlockFileCache::record_resize_failure(FileCacheResizeSource source, const Status& status) { + SCOPED_CACHE_LOCK(_mutex, this); + _capacity_state.last_resize = {.source = source, + .time_ms = UnixMillis(), + .status = "ERROR", + .message = status.to_string()}; + if (source == FileCacheResizeSource::AUTO_REFRESH) { + *_auto_resize_failure_metrics << 1; + } +} + +size_t BlockFileCache::capacity() const { + std::lock_guard lock(_mutex); + return _capacity; +} + +Status BlockFileCache::get_runtime_info(FileCacheRuntimeInfo* info) const { + DCHECK(info != nullptr); + std::lock_guard cache_lock(_mutex); + info->path = _cache_base_path; + info->storage = _storage->get_type() == FileCacheStorageType::MEMORY ? "memory" : "disk"; + info->capacity_mode = _capacity_state.policy.mode; + info->requested_capacity = _capacity_state.policy.requested_capacity; + info->capacity_generation = _capacity_state.generation; + info->capacity = _capacity; + info->current_size = _cur_cache_size; + info->pending_eviction_size = _cur_cache_size > _capacity ? _cur_cache_size - _capacity : 0; + info->max_file_block_size = _max_file_block_size; + info->disk_resource_limit_mode = _disk_resource_limit_mode; + info->need_evict_in_advance = _need_evict_cache_in_advance; + info->disk_state = _capacity_state.last_disk_state; + info->last_resize = _capacity_state.last_resize; + auto fill_queue_info = [](const LRUQueue& queue, size_t percent, + FileCacheQueueRuntimeInfo* queue_info) { + queue_info->percent = percent; + queue_info->max_size = queue.max_size; + queue_info->current_size = queue.cache_size; + queue_info->max_elements = queue.max_element_size; + queue_info->current_elements = queue.queue.size(); + }; + fill_queue_info(_disposable_queue, _capacity_state.policy.disposable_percent, + &info->queues[FileCacheType::DISPOSABLE]); + fill_queue_info(_normal_queue, _capacity_state.policy.normal_percent, + &info->queues[FileCacheType::NORMAL]); + fill_queue_info(_index_queue, _capacity_state.policy.index_percent, + &info->queues[FileCacheType::INDEX]); + fill_queue_info(_ttl_queue, _capacity_state.policy.ttl_percent, + &info->queues[FileCacheType::TTL]); + return Status::OK(); +} + +void BlockFileCache::maybe_reset_capacity_from_disk(const FileCacheDiskState& disk_state, + uint64_t expected_generation) { + if (disk_state.total_capacity == 0) { + bool record_failure = false; + { + std::lock_guard cache_lock(_mutex); + record_failure = _capacity_state.policy.mode == FileCacheCapacityMode::AUTO && + _capacity_state.generation == expected_generation; + } + if (record_failure) { + record_resize_failure(FileCacheResizeSource::AUTO_REFRESH, + Status::IOError("file cache disk capacity is zero")); + } + LOG_EVERY_N(WARNING, 10) << "Skip file cache auto resize because disk capacity is zero. " + << "path=" << _cache_base_path; + return; + } + + FileCacheCapacityPolicy policy; { SCOPED_CACHE_LOCK(_mutex, this); - if (new_capacity < _capacity && new_capacity < _cur_cache_size) { - int64_t need_remove_size = _cur_cache_size - new_capacity; - auto remove_blocks = [&](LRUQueue& queue) -> int64_t { - int64_t queue_released = 0; - std::vector to_evict; - for (const auto& [entry_key, entry_offset, entry_size] : queue) { - if (need_remove_size <= 0) { - break; - } - need_remove_size -= entry_size; - space_released += entry_size; - queue_released += entry_size; - auto* cell = get_cell(entry_key, entry_offset, cache_lock); - if (!cell->releasable()) { - cell->file_block->set_deleting(); - continue; - } - to_evict.push_back(cell); - } - for (auto& cell : to_evict) { - FileBlockSPtr file_block = cell->file_block; - std::lock_guard block_lock(file_block->_mutex); - remove(file_block, cache_lock, block_lock); - } - return queue_released; - }; - int64_t queue_released = remove_blocks(_disposable_queue); - ss << " disposable_queue released " << queue_released; - queue_released = remove_blocks(_normal_queue); - ss << " normal_queue released " << queue_released; - queue_released = remove_blocks(_index_queue); - ss << " index_queue released " << queue_released; - queue_released = remove_blocks(_ttl_queue); - ss << " ttl_queue released " << queue_released; - - _disk_resource_limit_mode = true; - _disk_limit_mode_metrics->set_value(1); - ss << " total_space_released=" << space_released; - } - old_capacity = _capacity; - _capacity = new_capacity; - _cache_capacity_metrics->set_value(_capacity); + update_disk_state_unlocked(disk_state, cache_lock); + if (_capacity_state.policy.mode != FileCacheCapacityMode::AUTO || + _capacity_state.generation != expected_generation || + _capacity == disk_state.total_capacity) { + return; + } + policy = _capacity_state.policy; + } + + BlockFileCacheResetRequest request; + request.source = FileCacheResizeSource::AUTO_REFRESH; + request.expected_generation = expected_generation; + FileCacheResizePlan plan; + auto st = build_file_cache_resize_plan(policy, request, disk_state, &plan); + if (!st.ok()) { + record_resize_failure(request.source, st); + return; + } + FileCachePathResizeResult result; + st = apply_reset_capacity(plan, &result); + if (!st.ok()) { + record_resize_failure(request.source, st); } - auto use_time = duration_cast(steady_clock::time_point() - adjust_start_time); - LOG(INFO) << "Finish tag deleted block. path=" << _cache_base_path - << " use_time=" << cast_set(use_time.count()); - ss << " old_capacity=" << old_capacity << " new_capacity=" << new_capacity; - LOG(INFO) << ss.str(); - return ss.str(); } void BlockFileCache::check_disk_resource_limit() { if (_storage->get_type() != FileCacheStorageType::DISK) { return; } + FileCacheDiskState disk_state; + auto st = get_file_cache_disk_state(_cache_base_path, &disk_state); + if (!st.ok()) { + LOG_WARNING("").error(st); + return; + } + check_disk_resource_limit(disk_state); +} + +void BlockFileCache::check_disk_resource_limit(const FileCacheDiskState& disk_state) { + SCOPED_CACHE_LOCK(_mutex, this); bool previous_mode = _disk_resource_limit_mode; if (_capacity > _cur_cache_size) { _disk_resource_limit_mode = false; _disk_limit_mode_metrics->set_value(0); } - std::pair percent; - int ret = disk_used_percentage(_cache_base_path, &percent); - if (ret != 0) { - LOG_ERROR("").tag("file cache path", _cache_base_path).tag("error", strerror(errno)); - return; - } - auto [space_percentage, inode_percentage] = percent; + auto [space_percentage, inode_percentage] = + std::pair {disk_state.disk_used_percent, disk_state.inode_used_percent}; auto is_insufficient = [](const int& percentage) { return percentage >= config::file_cache_enter_disk_resource_limit_mode_percent; }; @@ -2099,19 +2290,12 @@ void BlockFileCache::check_disk_resource_limit() { } } -void BlockFileCache::check_need_evict_cache_in_advance() { - if (_storage->get_type() != FileCacheStorageType::DISK) { - return; - } - - std::pair percent; - int ret = disk_used_percentage(_cache_base_path, &percent); - if (ret != 0) { - LOG_ERROR("").tag("file cache path", _cache_base_path).tag("error", strerror(errno)); - return; - } - auto [space_percentage, inode_percentage] = percent; - int size_percentage = static_cast(_cur_cache_size * 100 / _capacity); +void BlockFileCache::check_need_evict_cache_in_advance(const FileCacheDiskState& disk_state) { + SCOPED_CACHE_LOCK(_mutex, this); + auto [space_percentage, inode_percentage] = + std::pair {disk_state.disk_used_percent, disk_state.inode_used_percent}; + int size_percentage = + _capacity == 0 ? 100 : static_cast(_cur_cache_size * 100 / _capacity); auto is_insufficient = [](const int& percentage) { return percentage >= config::file_cache_enter_need_evict_cache_in_advance_percent; }; @@ -2175,18 +2359,70 @@ void BlockFileCache::check_need_evict_cache_in_advance() { } } +void BlockFileCache::check_need_evict_cache_in_advance() { + if (_storage->get_type() != FileCacheStorageType::DISK) { + return; + } + FileCacheDiskState disk_state; + auto st = get_file_cache_disk_state(_cache_base_path, &disk_state); + if (!st.ok()) { + LOG_WARNING("").error(st); + return; + } + check_need_evict_cache_in_advance(disk_state); +} + +void BlockFileCache::run_background_monitor_once() { + if (_storage->get_type() != FileCacheStorageType::DISK) { + std::lock_guard cache_lock(_mutex); + _need_evict_cache_in_advance = false; + _need_evict_cache_in_advance_metrics->set_value(0); + return; + } + + uint64_t expected_generation = 0; + bool auto_capacity = false; + { + std::lock_guard cache_lock(_mutex); + expected_generation = _capacity_state.generation; + auto_capacity = _capacity_state.policy.mode == FileCacheCapacityMode::AUTO; + } + + FileCacheDiskState disk_state; + auto st = get_file_cache_disk_state(_cache_base_path, &disk_state); + if (!st.ok()) { + LOG_EVERY_N(WARNING, 10) << "Failed to get file cache disk state. path=" << _cache_base_path + << " error=" << st; + if (auto_capacity) { + bool generation_matches = false; + { + std::lock_guard cache_lock(_mutex); + generation_matches = _capacity_state.policy.mode == FileCacheCapacityMode::AUTO && + _capacity_state.generation == expected_generation; + } + if (generation_matches) { + record_resize_failure(FileCacheResizeSource::AUTO_REFRESH, st); + } + } + return; + } + maybe_reset_capacity_from_disk(disk_state, expected_generation); + check_disk_resource_limit(disk_state); + if (config::enable_evict_file_cache_in_advance) { + check_need_evict_cache_in_advance(disk_state); + } else { + std::lock_guard cache_lock(_mutex); + _need_evict_cache_in_advance = false; + _need_evict_cache_in_advance_metrics->set_value(0); + } +} + void BlockFileCache::run_background_monitor() { Thread::set_self_name("run_background_monitor"); while (!_close) { int64_t interval_ms = config::file_cache_background_monitor_interval_ms; TEST_SYNC_POINT_CALLBACK("BlockFileCache::set_sleep_time", &interval_ms); - check_disk_resource_limit(); - if (config::enable_evict_file_cache_in_advance) { - check_need_evict_cache_in_advance(); - } else { - _need_evict_cache_in_advance = false; - _need_evict_cache_in_advance_metrics->set_value(0); - } + run_background_monitor_once(); { std::unique_lock close_lock(_close_mtx); @@ -2296,6 +2532,38 @@ void BlockFileCache::run_background_gc() { } } +void BlockFileCache::evict_over_capacity(size_t batch_size, + std::lock_guard& cache_lock) { + if (_cur_cache_size <= _capacity) { + update_resize_pending_unlocked(cache_lock); + return; + } + + const size_t target_size = std::min(_cur_cache_size - _capacity, batch_size); + size_t selected_size = 0; + std::vector to_evict; + auto collect = [&](LRUQueue& queue) { + for (const auto& [entry_key, entry_offset, entry_size] : queue) { + if (selected_size >= target_size) { + break; + } + auto* cell = get_cell(entry_key, entry_offset, cache_lock); + if (cell != nullptr && cell->releasable()) { + to_evict.push_back(cell); + selected_size += entry_size; + } + } + }; + + collect(_disposable_queue); + collect(_normal_queue); + collect(_index_queue); + collect(_ttl_queue); + std::string reason = "capacity resize"; + remove_file_blocks(to_evict, cache_lock, false, reason); + update_resize_pending_unlocked(cache_lock); +} + void BlockFileCache::run_background_evict_in_advance() { Thread::set_self_name("run_background_evict_in_advance"); LOG(INFO) << "Starting background evict in advance thread"; @@ -2313,10 +2581,8 @@ void BlockFileCache::run_background_evict_in_advance() { } batch = config::file_cache_evict_in_advance_batch_bytes; - // Skip if eviction not needed or too many pending recycles - if (!_need_evict_cache_in_advance || - _recycle_keys.size_approx() >= - config::file_cache_evict_in_advance_recycle_keys_num_threshold) { + if (_recycle_keys.size_approx() >= + config::file_cache_evict_in_advance_recycle_keys_num_threshold) { continue; } @@ -2324,7 +2590,13 @@ void BlockFileCache::run_background_evict_in_advance() { { SCOPED_CACHE_LOCK(_mutex, this); SCOPED_RAW_TIMER(&duration_ns); - try_evict_in_advance(batch, cache_lock); + if (_cur_cache_size > _capacity) { + const size_t resize_batch = batch > 0 ? cast_set(batch) + : std::max(_max_file_block_size, 1); + evict_over_capacity(resize_batch, cache_lock); + } else if (_need_evict_cache_in_advance) { + try_evict_in_advance(batch, cache_lock); + } } *_evict_in_advance_latency_us << (duration_ns / 1000); } diff --git a/be/src/io/cache/block_file_cache.h b/be/src/io/cache/block_file_cache.h index 9a01d07d136a62..a4f4265c240966 100644 --- a/be/src/io/cache/block_file_cache.h +++ b/be/src/io/cache/block_file_cache.h @@ -212,7 +212,7 @@ class BlockFileCache { Status initialize(); /// Cache capacity in bytes. - [[nodiscard]] size_t capacity() const { return _capacity; } + [[nodiscard]] size_t capacity() const; // try to release all releasable block // it maybe hang the io/system @@ -264,6 +264,16 @@ class BlockFileCache { * @returns summary message */ std::string reset_capacity(size_t new_capacity); + std::string reset_capacity(size_t new_capacity, bool auto_capacity); + + Status prepare_reset_capacity(const BlockFileCacheResetRequest& request, + const std::optional& observed_disk, + FileCacheResizePlan* plan) const; + Status apply_reset_capacity(const FileCacheResizePlan& plan, FileCachePathResizeResult* result); + Status reset_capacity(const BlockFileCacheResetRequest& request, + FileCachePathResizeResult* result); + + Status get_runtime_info(FileCacheRuntimeInfo* info) const; std::map get_blocks_by_key(const UInt128Wrapper& hash); @@ -467,8 +477,12 @@ class BlockFileCache { size_t get_used_cache_size_unlocked(FileCacheType type, std::lock_guard& cache_lock) const; + void check_disk_resource_limit(const FileCacheDiskState& disk_state); + void check_need_evict_cache_in_advance(const FileCacheDiskState& disk_state); void check_disk_resource_limit(); void check_need_evict_cache_in_advance(); + void maybe_reset_capacity_from_disk(const FileCacheDiskState& disk_state, + uint64_t expected_generation); size_t get_available_cache_size_unlocked(FileCacheType type, std::lock_guard& cache_lock) const; @@ -479,6 +493,7 @@ class BlockFileCache { bool need_to_move(FileCacheType cell_type, FileCacheType query_type) const; void run_background_monitor(); + void run_background_monitor_once(); void run_background_gc(); void run_background_lru_log_replay(); size_t replay_lru_logs_once(); @@ -522,10 +537,20 @@ class BlockFileCache { void clear_need_update_lru_blocks(); + Status apply_reset_capacity_unlocked(const FileCacheResizePlan& plan, + FileCachePathResizeResult* result, + std::lock_guard& cache_lock); + void update_resize_pending_unlocked(std::lock_guard& cache_lock); + void update_disk_state_unlocked(const FileCacheDiskState& disk_state, + std::lock_guard& cache_lock); + void record_resize_failure(FileCacheResizeSource source, const Status& status); + void evict_over_capacity(size_t batch_size, std::lock_guard& cache_lock); + // info std::string _cache_base_path; size_t _capacity = 0; size_t _max_file_block_size = 0; + FileCacheCapacityState _capacity_state; mutable std::mutex _mutex; bool _close {false}; @@ -572,6 +597,12 @@ class BlockFileCache { // metrics std::shared_ptr> _cache_capacity_metrics; + std::shared_ptr> _resize_pending_bytes_metrics; + std::shared_ptr> _disk_total_capacity_metrics; + std::shared_ptr> _disk_available_capacity_metrics; + std::shared_ptr> _capacity_mode_metrics; + std::shared_ptr> _auto_resize_success_metrics; + std::shared_ptr> _auto_resize_failure_metrics; std::shared_ptr> _cur_cache_size_metrics; std::shared_ptr> _cur_ttl_cache_size_metrics; std::shared_ptr> _cur_ttl_cache_lru_queue_cache_size_metrics; diff --git a/be/src/io/cache/block_file_cache_factory.cpp b/be/src/io/cache/block_file_cache_factory.cpp index 0ffc9cea365d6f..f2339f8d68c833 100644 --- a/be/src/io/cache/block_file_cache_factory.cpp +++ b/be/src/io/cache/block_file_cache_factory.cpp @@ -22,20 +22,15 @@ #include -#include -#include -#if defined(__APPLE__) -#include -#else -#include -#endif - #include #include #include #include +#include #include +#include +#include "common/cast_set.h" #include "common/config.h" #include "core/block/block.h" #include "information_schema/schema_scanner_helper.h" @@ -72,7 +67,14 @@ size_t FileCacheFactory::try_release(const std::string& base_path) { Status FileCacheFactory::create_file_cache(const std::string& cache_base_path, FileCacheSettings file_cache_settings) { - if (file_cache_settings.storage == "memory") { + std::lock_guard reset_lock(_reset_mtx); + if (file_cache_settings.storage != "memory" && file_cache_settings.capacity == 0) { + file_cache_settings.auto_capacity = true; + file_cache_settings.requested_capacity = 0; + } + auto policy = make_file_cache_capacity_policy(file_cache_settings); + file_cache_settings.storage = policy.storage; + if (policy.storage == "memory") { if (cache_base_path != "memory") { LOG(WARNING) << "memory storage must use memory path"; return Status::InvalidArgument("memory storage must use memory path"); @@ -89,73 +91,123 @@ Status FileCacheFactory::create_file_cache(const std::string& cache_base_path, RETURN_IF_ERROR(fs->delete_directory(cache_base_path)); RETURN_IF_ERROR(fs->create_directory(cache_base_path)); } + } - struct statfs stat; - if (statfs(cache_base_path.c_str(), &stat) < 0) { - LOG_ERROR("").tag("file cache path", cache_base_path).tag("error", strerror(errno)); - return Status::IOError("{} statfs error {}", cache_base_path, strerror(errno)); - } -#if defined(__APPLE__) - const auto block_size = stat.f_bsize; -#else - const auto block_size = stat.f_frsize ? stat.f_frsize : stat.f_bsize; -#endif - size_t disk_capacity = static_cast(stat.f_blocks) * static_cast(block_size); - if (file_cache_settings.capacity == 0 || disk_capacity < file_cache_settings.capacity) { - LOG_INFO( - "The cache {} config size {} is larger than disk size {} or zero, recalc " - "it.", - cache_base_path, file_cache_settings.capacity, disk_capacity); - file_cache_settings = get_file_cache_settings(disk_capacity, - file_cache_settings.max_query_cache_size); - } - LOG(INFO) << "[FileCache] path: " << cache_base_path - << " total_size: " << file_cache_settings.capacity - << " disk_total_size: " << disk_capacity; + std::optional disk_state; + if (policy.storage != "memory") { + FileCacheDiskState state; + RETURN_IF_ERROR(get_file_cache_disk_state(cache_base_path, &state)); + disk_state = state; + } + uint64_t effective_capacity = 0; + bool clamped_by_disk = false; + RETURN_IF_ERROR( + resolve_file_cache_capacity(policy, disk_state, &effective_capacity, &clamped_by_disk)); + if (file_cache_settings.capacity != effective_capacity || + policy.mode == FileCacheCapacityMode::AUTO) { + RETURN_IF_ERROR( + build_file_cache_settings(effective_capacity, policy, &file_cache_settings)); + } else { + file_cache_settings.requested_capacity = policy.requested_capacity; + file_cache_settings.auto_capacity = policy.mode == FileCacheCapacityMode::AUTO; + file_cache_settings.max_file_block_size = policy.max_file_block_size; } + if (clamped_by_disk) { + LOG(WARNING) << "File cache capacity is clamped by disk. path=" << cache_base_path + << " requested_capacity=" << policy.requested_capacity + << " disk_total_capacity=" << disk_state->total_capacity; + } + auto cache = std::make_unique(cache_base_path, file_cache_settings); + FileCacheResizePlan initial_plan; + initial_plan.next_policy = policy; + initial_plan.next_settings = file_cache_settings; + initial_plan.disk_state = disk_state; + initial_plan.source = FileCacheResizeSource::STARTUP; + initial_plan.clamped_by_disk = clamped_by_disk; + FileCachePathResizeResult initial_result; + RETURN_IF_ERROR(cache->apply_reset_capacity(initial_plan, &initial_result)); RETURN_IF_ERROR(cache->initialize()); { std::lock_guard lock(_mtx); _path_to_cache[cache_base_path] = cache.get(); _caches.push_back(std::move(cache)); - _capacity += file_cache_settings.capacity; } return Status::OK(); } Status FileCacheFactory::reload_file_cache(const std::vector& cache_base_paths) { + std::lock_guard reset_lock(_reset_mtx); { - std::unique_lock lock(_mtx); + std::lock_guard topology_lock(_mtx); for (const auto& cache_path : cache_base_paths) { if (_path_to_cache.find(cache_path.path) == _path_to_cache.end()) { return Status::InternalError( "Current file cache not support file cache num changes"); } } + } - for (const auto& cache_path : cache_base_paths) { - auto cache_map_iter = _path_to_cache.find(cache_path.path); - auto cache_iter = std::find_if(_caches.begin(), _caches.end(), - [cache_map_iter](const auto& cache_uptr) { - return cache_uptr.get() == cache_map_iter->second; - }); - - if (cache_iter == _caches.end()) { - return Status::InternalError("Target relaod cache in path {} may has been released", - cache_path.path); - } + struct PreparedReplacement { + std::string path; + FileCacheSettings settings; + FileCacheCapacityPolicy policy; + std::optional disk_state; + bool clamped_by_disk = false; + }; + std::vector replacements; + replacements.reserve(cache_base_paths.size()); + for (const auto& cache_path : cache_base_paths) { + auto settings = cache_path.init_settings(); + auto policy = make_file_cache_capacity_policy(settings); + std::optional disk_state; + if (policy.storage != "memory") { + FileCacheDiskState state; + RETURN_IF_ERROR(get_file_cache_disk_state(cache_path.path, &state)); + disk_state = state; + } + uint64_t effective_capacity = 0; + bool clamped_by_disk = false; + RETURN_IF_ERROR(resolve_file_cache_capacity(policy, disk_state, &effective_capacity, + &clamped_by_disk)); + if (settings.capacity != effective_capacity || policy.mode == FileCacheCapacityMode::AUTO) { + RETURN_IF_ERROR(build_file_cache_settings(effective_capacity, policy, &settings)); + } - // deconstruct target reload first - *cache_iter = std::unique_ptr(); - // after deconstruct the BlockFileCache, construct the BlockFileCache again - *cache_iter = - std::make_unique(cache_path.path, cache_path.init_settings()); - cache_map_iter->second = cache_iter->get(); + replacements.push_back({.path = cache_path.path, + .settings = std::move(settings), + .policy = std::move(policy), + .disk_state = disk_state, + .clamped_by_disk = clamped_by_disk}); + } - RETURN_IF_ERROR(cache_iter->get()->initialize()); + std::lock_guard topology_lock(_mtx); + for (auto& prepared : replacements) { + auto cache_map_iter = _path_to_cache.find(prepared.path); + auto cache_iter = std::find_if(_caches.begin(), _caches.end(), + [cache_map_iter](const auto& cache_uptr) { + return cache_uptr.get() == cache_map_iter->second; + }); + + if (cache_iter == _caches.end()) { + return Status::InternalError("Target reload cache in path {} may have been released", + prepared.path); } + cache_map_iter->second = nullptr; + cache_iter->reset(); + auto replacement = std::make_unique(prepared.path, prepared.settings); + FileCacheResizePlan initial_plan; + initial_plan.next_policy = prepared.policy; + initial_plan.next_settings = prepared.settings; + initial_plan.disk_state = prepared.disk_state; + initial_plan.source = FileCacheResizeSource::RELOAD; + initial_plan.clamped_by_disk = prepared.clamped_by_disk; + FileCachePathResizeResult initial_result; + RETURN_IF_ERROR(replacement->apply_reset_capacity(initial_plan, &initial_result)); + RETURN_IF_ERROR(replacement->initialize()); + *cache_iter = std::move(replacement); + cache_map_iter->second = cache_iter->get(); } return Status::OK(); @@ -312,66 +364,107 @@ std::vector FileCacheFactory::get_base_paths() { return paths; } -std::string validate_capacity(const std::string& path, int64_t new_capacity, - int64_t& valid_capacity) { - struct statfs stat; - if (statfs(path.c_str(), &stat) < 0) { - auto ret = fmt::format("reset capacity {} statfs error {}. ", path, strerror(errno)); - LOG_ERROR(ret); - valid_capacity = 0; // caller will handle the error - return ret; +std::string FileCacheFactory::reset_capacity(const std::string& path, int64_t new_capacity) { + std::string result; + auto st = reset_capacity(path, new_capacity, &result); + return st.ok() ? result : st.to_string(); +} + +Status FileCacheFactory::reset_capacity(const std::string& path, int64_t new_capacity, + std::string* result) { + DCHECK(result != nullptr); + if (new_capacity < 0) { + return Status::InvalidArgument("file cache capacity must not be negative"); } -#if defined(__APPLE__) - const auto block_size = stat.f_bsize; -#else - const auto block_size = stat.f_frsize ? stat.f_frsize : stat.f_bsize; -#endif - size_t disk_capacity = static_cast(stat.f_blocks) * static_cast(block_size); - if (new_capacity == 0 || disk_capacity < new_capacity) { - auto ret = fmt::format( - "The cache {} config size {} is larger than disk size {} or zero, recalc " - "it to disk size. ", - path, new_capacity, disk_capacity); - valid_capacity = disk_capacity; - LOG_WARNING(ret); - return ret; + + FileCacheResetResult reset_result; + RETURN_IF_ERROR(reset_capacity(path, cast_set(new_capacity), &reset_result)); + std::stringstream ss; + for (const auto& cache : reset_result.caches) { + ss << fmt::format( + "finish reset_capacity, path={} old_capacity={} new_capacity={} mode={} " + "pending_eviction_bytes={} clamped_by_disk={} changed={}\n", + cache.path, cache.old_capacity, cache.new_capacity, + file_cache_capacity_mode_to_string(cache.mode), cache.pending_eviction_bytes, + cache.clamped_by_disk, cache.changed); } - valid_capacity = new_capacity; - return ""; + *result = ss.str(); + return Status::OK(); } -std::string FileCacheFactory::reset_capacity(const std::string& path, int64_t new_capacity) { - std::stringstream ss; - size_t total_capacity = 0; - if (path.empty()) { - for (auto& [p, cache] : _path_to_cache) { - int64_t valid_capacity = 0; - ss << validate_capacity(p, new_capacity, valid_capacity); - if (valid_capacity <= 0) { - return ss.str(); +Status FileCacheFactory::reset_capacity(const std::string& path, uint64_t new_capacity, + FileCacheResetResult* result) { + DCHECK(result != nullptr); + std::lock_guard reset_lock(_reset_mtx); + + std::vector caches; + { + std::lock_guard topology_lock(_mtx); + if (!path.empty()) { + auto iter = _path_to_cache.find(path); + if (iter == _path_to_cache.end()) { + return Status::InvalidArgument("unknown file cache path {}", path); } - ss << cache->reset_capacity(valid_capacity); - total_capacity += cache->capacity(); - } - _capacity = total_capacity; - return ss.str(); - } else { - if (auto iter = _path_to_cache.find(path); iter != _path_to_cache.end()) { - int64_t valid_capacity = 0; - ss << validate_capacity(path, new_capacity, valid_capacity); - if (valid_capacity <= 0) { - return ss.str(); + caches.push_back(iter->second); + } else { + caches.reserve(_caches.size()); + for (const auto& cache : _caches) { + caches.push_back(cache.get()); } - ss << iter->second->reset_capacity(valid_capacity); + } + } + std::ranges::sort(caches, {}, &BlockFileCache::get_base_path); + + BlockFileCacheResetRequest request; + request.requested_capacity = new_capacity; + request.source = FileCacheResizeSource::HTTP; + std::vector plans; + plans.reserve(caches.size()); + for (auto* cache : caches) { + plans.emplace_back(); + RETURN_IF_ERROR(cache->prepare_reset_capacity(request, std::nullopt, &plans.back())); + } - for (auto& [p, cache] : _path_to_cache) { - total_capacity += cache->capacity(); - } - _capacity = total_capacity; - return ss.str(); + result->caches.clear(); + result->caches.reserve(caches.size()); + for (size_t i = 0; i < caches.size(); ++i) { + result->caches.emplace_back(); + RETURN_IF_ERROR(caches[i]->apply_reset_capacity(plans[i], &result->caches.back())); + } + result->total_capacity = get_capacity(); + return Status::OK(); +} + +size_t FileCacheFactory::get_capacity() const { + std::lock_guard lock(_mtx); + size_t total_capacity = 0; + for (const auto& cache : _caches) { + total_capacity += cache->capacity(); + } + return total_capacity; +} + +Status FileCacheFactory::get_cache_infos(const std::string& path, + std::vector* infos) const { + DCHECK(infos != nullptr); + infos->clear(); + std::lock_guard lock(_mtx); + if (!path.empty()) { + auto iter = _path_to_cache.find(path); + if (iter == _path_to_cache.end()) { + return Status::InvalidArgument("unknown file cache path {}", path); } + infos->emplace_back(); + return iter->second->get_runtime_info(&infos->back()); } - return "Unknown the cache path " + path; + + infos->reserve(_caches.size()); + for (const auto& cache : _caches) { + infos->emplace_back(); + RETURN_IF_ERROR(cache->get_runtime_info(&infos->back())); + } + std::ranges::sort(*infos, {}, &FileCacheRuntimeInfo::path); + return Status::OK(); } void FileCacheFactory::get_cache_stats_block(Block* block) { diff --git a/be/src/io/cache/block_file_cache_factory.h b/be/src/io/cache/block_file_cache_factory.h index f5be334de8b75f..906dab03c0cf7d 100644 --- a/be/src/io/cache/block_file_cache_factory.h +++ b/be/src/io/cache/block_file_cache_factory.h @@ -62,7 +62,7 @@ class FileCacheFactory { return _caches[cur_index % _caches.size()]->get_base_path(); } - [[nodiscard]] size_t get_capacity() const { return _capacity; } + [[nodiscard]] size_t get_capacity() const; [[nodiscard]] size_t get_cache_instance_size() const { return _caches.size(); } @@ -103,6 +103,11 @@ class FileCacheFactory { * @return summary message */ std::string reset_capacity(const std::string& path, int64_t new_capacity); + Status reset_capacity(const std::string& path, int64_t new_capacity, std::string* result); + Status reset_capacity(const std::string& path, uint64_t new_capacity, + FileCacheResetResult* result); + + Status get_cache_infos(const std::string& path, std::vector* infos) const; void get_cache_stats_block(Block* block); @@ -114,10 +119,10 @@ class FileCacheFactory { FileCacheFactory(const FileCacheFactory&) = delete; private: - std::mutex _mtx; + std::mutex _reset_mtx; + mutable std::mutex _mtx; std::vector> _caches; std::unordered_map _path_to_cache; - size_t _capacity = 0; std::atomic_size_t _next_index {0}; // use for round-robin }; diff --git a/be/src/io/cache/file_cache_common.cpp b/be/src/io/cache/file_cache_common.cpp index 948bd4e144ecde..aadd5a50137ecb 100644 --- a/be/src/io/cache/file_cache_common.cpp +++ b/be/src/io/cache/file_cache_common.cpp @@ -20,12 +20,47 @@ #include "io/cache/file_cache_common.h" +#include +#include +#include + +#if defined(__APPLE__) +#include +#else +#include +#endif + #include "common/config.h" +#include "cpp/sync_point.h" #include "exec/common/hex.h" #include "io/cache/block_file_cache.h" namespace doris::io { +std::string file_cache_capacity_mode_to_string(FileCacheCapacityMode mode) { + switch (mode) { + case FileCacheCapacityMode::AUTO: + return "AUTO"; + case FileCacheCapacityMode::MANUAL: + return "MANUAL"; + } + return "MANUAL"; +} + +std::string file_cache_resize_source_to_string(FileCacheResizeSource source) { + switch (source) { + case FileCacheResizeSource::STARTUP: + return "STARTUP"; + case FileCacheResizeSource::AUTO_REFRESH: + return "AUTO_REFRESH"; + case FileCacheResizeSource::HTTP: + return "HTTP"; + case FileCacheResizeSource::RELOAD: + return "RELOAD"; + } + return "STARTUP"; +} + std::string cache_type_to_surfix(FileCacheType type) { switch (type) { case FileCacheType::INDEX: @@ -82,7 +117,8 @@ std::string cache_type_to_string(FileCacheType type) { std::string FileCacheSettings::to_string() const { std::stringstream ss; - ss << "capacity: " << capacity << ", max_file_block_size: " << max_file_block_size + ss << "capacity: " << capacity << ", requested_capacity: " << requested_capacity + << ", max_file_block_size: " << max_file_block_size << ", max_query_cache_size: " << max_query_cache_size << ", disposable_queue_size: " << disposable_queue_size << ", disposable_queue_elements: " << disposable_queue_elements @@ -90,7 +126,10 @@ std::string FileCacheSettings::to_string() const { << ", index_queue_elements: " << index_queue_elements << ", ttl_queue_size: " << ttl_queue_size << ", ttl_queue_elements: " << ttl_queue_elements << ", query_queue_size: " << query_queue_size - << ", query_queue_elements: " << query_queue_elements << ", storage: " << storage; + << ", query_queue_elements: " << query_queue_elements + << ", normal_percent: " << normal_percent << ", disposable_percent: " << disposable_percent + << ", index_percent: " << index_percent << ", ttl_percent: " << ttl_percent + << ", auto_capacity: " << auto_capacity << ", storage: " << storage; return ss.str(); } @@ -99,36 +138,214 @@ FileCacheSettings get_file_cache_settings(size_t capacity, size_t max_query_cach size_t index_percent, size_t ttl_percent, const std::string& storage) { io::FileCacheSettings settings; - if (capacity == 0) { - return settings; - } - settings.capacity = capacity; + settings.requested_capacity = capacity; settings.max_file_block_size = config::file_cache_each_block_size; settings.max_query_cache_size = max_query_cache_size; - size_t per_size = settings.capacity / 100; - settings.disposable_queue_size = per_size * disposable_percent; - settings.disposable_queue_elements = - std::max(settings.disposable_queue_size / settings.max_file_block_size, - REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS); - - settings.index_queue_size = per_size * index_percent; - settings.index_queue_elements = - std::max(settings.index_queue_size / settings.max_file_block_size, - REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS); - - settings.ttl_queue_size = per_size * ttl_percent; - settings.ttl_queue_elements = std::max(settings.ttl_queue_size / settings.max_file_block_size, - REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS); - - settings.query_queue_size = settings.capacity - settings.disposable_queue_size - - settings.index_queue_size - settings.ttl_queue_size; - settings.query_queue_elements = - std::max(settings.query_queue_size / settings.max_file_block_size, - REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS); + settings.normal_percent = normal_percent; + settings.disposable_percent = disposable_percent; + settings.index_percent = index_percent; + settings.ttl_percent = ttl_percent; settings.storage = storage; + if (capacity > 0) { + FileCacheCapacityPolicy policy; + policy.mode = FileCacheCapacityMode::MANUAL; + policy.requested_capacity = capacity; + policy.normal_percent = normal_percent; + policy.disposable_percent = disposable_percent; + policy.index_percent = index_percent; + policy.ttl_percent = ttl_percent; + policy.max_query_cache_size = max_query_cache_size; + policy.max_file_block_size = settings.max_file_block_size; + policy.storage = storage; + auto st = build_file_cache_settings(capacity, policy, &settings); + DCHECK(st.ok()) << st; + } return settings; } +Status get_file_cache_disk_state(const std::string& path, FileCacheDiskState* state) { + DCHECK(state != nullptr); + Status injected_status; + TEST_SYNC_POINT_CALLBACK("BlockFileCache::get_file_cache_disk_state:status", &injected_status); + RETURN_IF_ERROR(injected_status); + + struct statfs stat; + if (statfs(path.c_str(), &stat) != 0) { + return Status::IOError("{} statfs error {}", path, strerror(errno)); + } +#if defined(__APPLE__) + const auto block_size = stat.f_bsize; +#else + const auto block_size = stat.f_frsize ? stat.f_frsize : stat.f_bsize; +#endif + const auto checked_bytes = [&](uint64_t blocks, uint64_t* bytes) -> Status { + const auto size = static_cast(block_size); + if (size != 0 && blocks > std::numeric_limits::max() / size) { + return Status::IOError("{} statfs capacity overflows uint64", path); + } + *bytes = blocks * size; + return Status::OK(); + }; + RETURN_IF_ERROR(checked_bytes(static_cast(stat.f_blocks), &state->total_capacity)); + RETURN_IF_ERROR( + checked_bytes(static_cast(stat.f_bavail), &state->available_capacity)); + + const uintmax_t used_blocks = stat.f_blocks - stat.f_bfree; + const uintmax_t nonroot_total = used_blocks + stat.f_bavail; + const unsigned __int128 used_times_100 = static_cast(used_blocks) * 100; + state->disk_used_percent = nonroot_total == 0 + ? 0 + : static_cast(used_times_100 / nonroot_total + + (used_times_100 % nonroot_total != 0)); + state->inode_used_percent = + stat.f_files == 0 ? 0 + : 100 - static_cast(static_cast(stat.f_ffree) * 100 / + static_cast(stat.f_files)); + + std::pair legacy_percent {state->disk_used_percent, state->inode_used_percent}; + TEST_SYNC_POINT_CALLBACK("BlockFileCache::disk_used_percentage:1", &legacy_percent); + state->disk_used_percent = legacy_percent.first; + state->inode_used_percent = legacy_percent.second; + TEST_SYNC_POINT_CALLBACK("BlockFileCache::get_file_cache_disk_state", state, &path); + return Status::OK(); +} + +FileCacheCapacityPolicy make_file_cache_capacity_policy(const FileCacheSettings& settings) { + FileCacheCapacityPolicy policy; + policy.mode = + settings.auto_capacity ? FileCacheCapacityMode::AUTO : FileCacheCapacityMode::MANUAL; + policy.requested_capacity = + policy.mode == FileCacheCapacityMode::AUTO + ? 0 + : (settings.requested_capacity == 0 ? settings.capacity + : settings.requested_capacity); + policy.normal_percent = settings.normal_percent; + policy.disposable_percent = settings.disposable_percent; + policy.index_percent = settings.index_percent; + policy.ttl_percent = settings.ttl_percent; + policy.max_query_cache_size = settings.max_query_cache_size; + policy.max_file_block_size = settings.max_file_block_size == 0 + ? config::file_cache_each_block_size + : settings.max_file_block_size; + policy.storage = settings.storage == "memory" ? "memory" : "disk"; + return policy; +} + +Status resolve_file_cache_capacity(const FileCacheCapacityPolicy& policy, + const std::optional& disk_state, + uint64_t* effective_capacity, bool* clamped_by_disk) { + DCHECK(effective_capacity != nullptr); + DCHECK(clamped_by_disk != nullptr); + *clamped_by_disk = false; + + if (policy.storage == "memory") { + if (policy.mode == FileCacheCapacityMode::AUTO) { + return Status::InvalidArgument("memory file cache does not support automatic capacity"); + } + *effective_capacity = policy.requested_capacity; + } else { + if (!disk_state.has_value()) { + return Status::IOError("file cache disk state is unavailable"); + } + if (disk_state->total_capacity == 0) { + return Status::IOError("file cache disk capacity is zero"); + } + if (policy.mode == FileCacheCapacityMode::AUTO) { + *effective_capacity = disk_state->total_capacity; + } else { + *effective_capacity = std::min(policy.requested_capacity, disk_state->total_capacity); + *clamped_by_disk = policy.requested_capacity > disk_state->total_capacity; + } + } + + if (*effective_capacity == 0) { + return Status::InvalidArgument("file cache capacity must be greater than zero"); + } + if (*effective_capacity > std::numeric_limits::max()) { + return Status::InvalidArgument("file cache capacity exceeds size_t"); + } + return Status::OK(); +} + +Status build_file_cache_settings(uint64_t effective_capacity, const FileCacheCapacityPolicy& policy, + FileCacheSettings* settings) { + DCHECK(settings != nullptr); + if (effective_capacity == 0 || effective_capacity > std::numeric_limits::max()) { + return Status::InvalidArgument("invalid effective file cache capacity {}", + effective_capacity); + } + if (policy.max_file_block_size == 0) { + return Status::InvalidArgument("max file cache block size must be greater than zero"); + } + if (policy.normal_percent + policy.disposable_percent + policy.index_percent + + policy.ttl_percent != + 100) { + return Status::InvalidArgument("file cache queue percentages must sum to 100"); + } + + FileCacheSettings next; + next.capacity = static_cast(effective_capacity); + next.requested_capacity = policy.requested_capacity; + next.max_file_block_size = policy.max_file_block_size; + next.max_query_cache_size = policy.max_query_cache_size; + next.normal_percent = policy.normal_percent; + next.disposable_percent = policy.disposable_percent; + next.index_percent = policy.index_percent; + next.ttl_percent = policy.ttl_percent; + next.auto_capacity = policy.mode == FileCacheCapacityMode::AUTO; + next.storage = policy.storage; + + const size_t per_size = next.capacity / 100; + next.disposable_queue_size = per_size * policy.disposable_percent; + next.index_queue_size = per_size * policy.index_percent; + next.ttl_queue_size = per_size * policy.ttl_percent; + next.query_queue_size = next.capacity - next.disposable_queue_size - next.index_queue_size - + next.ttl_queue_size; + const auto queue_elements = [&](size_t queue_size) { + return std::max(queue_size / next.max_file_block_size, + REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS); + }; + next.disposable_queue_elements = queue_elements(next.disposable_queue_size); + next.index_queue_elements = queue_elements(next.index_queue_size); + next.ttl_queue_elements = queue_elements(next.ttl_queue_size); + next.query_queue_elements = queue_elements(next.query_queue_size); + *settings = std::move(next); + return Status::OK(); +} + +Status build_file_cache_resize_plan(const FileCacheCapacityPolicy& current_policy, + const BlockFileCacheResetRequest& request, + const std::optional& disk_state, + FileCacheResizePlan* plan) { + DCHECK(plan != nullptr); + FileCacheCapacityPolicy next_policy = current_policy; + if (request.source == FileCacheResizeSource::AUTO_REFRESH) { + if (current_policy.mode != FileCacheCapacityMode::AUTO) { + return Status::InvalidArgument("manual file cache cannot be auto refreshed"); + } + } else { + next_policy.mode = request.requested_capacity == 0 ? FileCacheCapacityMode::AUTO + : FileCacheCapacityMode::MANUAL; + next_policy.requested_capacity = request.requested_capacity; + } + + uint64_t effective_capacity = 0; + bool clamped_by_disk = false; + RETURN_IF_ERROR(resolve_file_cache_capacity(next_policy, disk_state, &effective_capacity, + &clamped_by_disk)); + + FileCacheResizePlan next_plan; + next_plan.next_policy = next_policy; + RETURN_IF_ERROR( + build_file_cache_settings(effective_capacity, next_policy, &next_plan.next_settings)); + next_plan.disk_state = disk_state; + next_plan.expected_generation = request.expected_generation; + next_plan.source = request.source; + next_plan.clamped_by_disk = clamped_by_disk; + *plan = std::move(next_plan); + return Status::OK(); +} + std::string UInt128Wrapper::to_string() const { return get_hex_uint_lowercase(value_); } diff --git a/be/src/io/cache/file_cache_common.h b/be/src/io/cache/file_cache_common.h index 4c60d2e86f3e60..49894ae334d876 100644 --- a/be/src/io/cache/file_cache_common.h +++ b/be/src/io/cache/file_cache_common.h @@ -19,9 +19,13 @@ // and modified by Doris #pragma once +#include #include +#include +#include #include +#include "common/status.h" #include "core/uint128.h" #include "io/io_common.h" @@ -42,6 +46,22 @@ enum FileCacheType { DISPOSABLE = 0, TTL = 3, }; + +enum class FileCacheCapacityMode { + AUTO, + MANUAL, +}; + +enum class FileCacheResizeSource { + STARTUP, + AUTO_REFRESH, + HTTP, + RELOAD, +}; + +std::string file_cache_capacity_mode_to_string(FileCacheCapacityMode mode); +std::string file_cache_resize_source_to_string(FileCacheResizeSource source); + std::string cache_type_to_surfix(FileCacheType type); FileCacheType surfix_to_cache_type(const std::string& str); @@ -126,6 +146,7 @@ struct FileCacheKey { struct FileCacheSettings { size_t capacity {0}; + uint64_t requested_capacity {0}; size_t disposable_queue_size {0}; size_t disposable_queue_elements {0}; size_t index_queue_size {0}; @@ -136,6 +157,11 @@ struct FileCacheSettings { size_t ttl_queue_elements {0}; size_t max_file_block_size {0}; size_t max_query_cache_size {0}; + size_t normal_percent {DEFAULT_NORMAL_PERCENT}; + size_t disposable_percent {DEFAULT_DISPOSABLE_PERCENT}; + size_t index_percent {DEFAULT_INDEX_PERCENT}; + size_t ttl_percent {DEFAULT_TTL_PERCENT}; + bool auto_capacity {false}; std::string storage; // to string @@ -149,6 +175,114 @@ FileCacheSettings get_file_cache_settings(size_t capacity, size_t max_query_cach size_t ttl_percent = DEFAULT_TTL_PERCENT, const std::string& storage = "disk"); +struct FileCacheDiskState { + uint64_t total_capacity {0}; + uint64_t available_capacity {0}; + int disk_used_percent {0}; + int inode_used_percent {0}; +}; + +Status get_file_cache_disk_state(const std::string& path, FileCacheDiskState* state); + +struct FileCacheCapacityPolicy { + FileCacheCapacityMode mode {FileCacheCapacityMode::MANUAL}; + uint64_t requested_capacity {0}; + size_t normal_percent {DEFAULT_NORMAL_PERCENT}; + size_t disposable_percent {DEFAULT_DISPOSABLE_PERCENT}; + size_t index_percent {DEFAULT_INDEX_PERCENT}; + size_t ttl_percent {DEFAULT_TTL_PERCENT}; + size_t max_query_cache_size {0}; + size_t max_file_block_size {FILE_CACHE_MAX_FILE_BLOCK_SIZE}; + std::string storage {"disk"}; +}; + +FileCacheCapacityPolicy make_file_cache_capacity_policy(const FileCacheSettings& settings); + +struct FileCacheResizeAudit { + FileCacheResizeSource source {FileCacheResizeSource::STARTUP}; + int64_t time_ms {0}; + std::string status {"OK"}; + std::string message; +}; + +struct FileCacheCapacityState { + FileCacheCapacityPolicy policy; + uint64_t generation {0}; + std::optional last_disk_state; + FileCacheResizeAudit last_resize; + uint64_t pending_eviction_bytes {0}; +}; + +struct BlockFileCacheResetRequest { + uint64_t requested_capacity {0}; + FileCacheResizeSource source {FileCacheResizeSource::HTTP}; + std::optional expected_generation; +}; + +struct FileCacheResizePlan { + FileCacheCapacityPolicy next_policy; + FileCacheSettings next_settings; + std::optional disk_state; + std::optional expected_generation; + FileCacheResizeSource source {FileCacheResizeSource::HTTP}; + bool clamped_by_disk {false}; +}; + +struct FileCachePathResizeResult { + std::string path; + FileCacheCapacityMode mode {FileCacheCapacityMode::MANUAL}; + uint64_t requested_capacity {0}; + uint64_t old_capacity {0}; + uint64_t new_capacity {0}; + uint64_t disk_total_capacity {0}; + uint64_t disk_available_capacity {0}; + uint64_t used_bytes {0}; + uint64_t pending_eviction_bytes {0}; + bool clamped_by_disk {false}; + bool changed {false}; + bool skipped {false}; +}; + +struct FileCacheResetResult { + uint64_t total_capacity {0}; + std::vector caches; +}; + +Status resolve_file_cache_capacity(const FileCacheCapacityPolicy& policy, + const std::optional& disk_state, + uint64_t* effective_capacity, bool* clamped_by_disk); +Status build_file_cache_settings(uint64_t effective_capacity, const FileCacheCapacityPolicy& policy, + FileCacheSettings* settings); +Status build_file_cache_resize_plan(const FileCacheCapacityPolicy& current_policy, + const BlockFileCacheResetRequest& request, + const std::optional& disk_state, + FileCacheResizePlan* plan); + +struct FileCacheQueueRuntimeInfo { + size_t percent {0}; + size_t max_size {0}; + size_t current_size {0}; + size_t max_elements {0}; + size_t current_elements {0}; +}; + +struct FileCacheRuntimeInfo { + std::string path; + std::string storage; + FileCacheCapacityMode capacity_mode {FileCacheCapacityMode::MANUAL}; + uint64_t requested_capacity {0}; + uint64_t capacity_generation {0}; + uint64_t capacity {0}; + uint64_t current_size {0}; + uint64_t pending_eviction_size {0}; + uint64_t max_file_block_size {0}; + bool disk_resource_limit_mode {false}; + bool need_evict_in_advance {false}; + std::optional disk_state; + FileCacheResizeAudit last_resize; + std::array queues; +}; + struct CacheContext { CacheContext(const IOContext* io_context) { if (io_context->expiration_time != 0) { diff --git a/be/src/service/http/action/file_cache_action.cpp b/be/src/service/http/action/file_cache_action.cpp index d291bc8df1d08d..372760282894ef 100644 --- a/be/src/service/http/action/file_cache_action.cpp +++ b/be/src/service/http/action/file_cache_action.cpp @@ -48,6 +48,7 @@ constexpr static std::string_view SYNC = "sync"; constexpr static std::string_view PATH = "path"; constexpr static std::string_view CLEAR = "clear"; constexpr static std::string_view RESET = "reset"; +constexpr static std::string_view INFO = "info"; constexpr static std::string_view HASH = "hash"; constexpr static std::string_view LIST_CACHE = "list_cache"; constexpr static std::string_view LIST_BASE_PATHS = "list_base_paths"; @@ -60,6 +61,87 @@ constexpr static std::string_view DUMP = "dump"; constexpr static std::string_view VALUE = "value"; constexpr static std::string_view RELOAD = "reload"; +std::string file_cache_infos_to_json(const std::vector& infos, + const std::string& message = "") { + EasyJson json; + json["status"] = "OK"; + if (!message.empty()) { + json["message"] = message; + } + uint64_t total_capacity = 0; + auto caches = json["caches"]; + caches.SetArray(); + auto add_queue = [](EasyJson queues, const std::string& name, + const io::FileCacheQueueRuntimeInfo& info) { + auto queue = queues[name]; + queue.SetObject(); + queue["percent"] = static_cast(info.percent); + queue["max_bytes"] = static_cast(info.max_size); + queue["used_bytes"] = static_cast(info.current_size); + queue["max_elements"] = static_cast(info.max_elements); + queue["elements"] = static_cast(info.current_elements); + }; + for (const auto& info : infos) { + total_capacity += info.capacity; + auto cache = caches.PushBack(EasyJson::kObject); + cache["path"] = info.path; + cache["storage"] = info.storage; + cache["capacity_mode"] = io::file_cache_capacity_mode_to_string(info.capacity_mode); + cache["requested_capacity"] = info.requested_capacity; + cache["capacity_generation"] = info.capacity_generation; + cache["effective_capacity"] = info.capacity; + cache["used_bytes"] = info.current_size; + cache["pending_eviction_bytes"] = static_cast(info.pending_eviction_size); + cache["max_file_block_size"] = info.max_file_block_size; + cache["disk_resource_limit_mode"] = info.disk_resource_limit_mode; + cache["need_evict_in_advance"] = info.need_evict_in_advance; + if (info.disk_state.has_value()) { + cache["disk_total_bytes"] = info.disk_state->total_capacity; + cache["disk_available_bytes"] = info.disk_state->available_capacity; + } else { + cache["disk_total_bytes"]; + cache["disk_available_bytes"]; + } + auto last_resize = cache["last_resize"]; + last_resize.SetObject(); + last_resize["source"] = io::file_cache_resize_source_to_string(info.last_resize.source); + last_resize["time_ms"] = info.last_resize.time_ms; + last_resize["status"] = info.last_resize.status; + last_resize["message"] = info.last_resize.message; + auto queues = cache["queues"]; + queues.SetObject(); + add_queue(queues, "normal", info.queues[io::FileCacheType::NORMAL]); + add_queue(queues, "disposable", info.queues[io::FileCacheType::DISPOSABLE]); + add_queue(queues, "index", info.queues[io::FileCacheType::INDEX]); + add_queue(queues, "ttl", info.queues[io::FileCacheType::TTL]); + } + json["total_capacity"] = total_capacity; + return json.ToString(); +} + +std::string file_cache_reset_result_to_json(const io::FileCacheResetResult& result) { + EasyJson json; + json["status"] = "OK"; + json["total_capacity"] = result.total_capacity; + auto caches = json["caches"]; + caches.SetArray(); + for (const auto& info : result.caches) { + auto cache = caches.PushBack(EasyJson::kObject); + cache["path"] = info.path; + cache["capacity_mode"] = io::file_cache_capacity_mode_to_string(info.mode); + cache["requested_capacity"] = info.requested_capacity; + cache["old_capacity"] = info.old_capacity; + cache["effective_capacity"] = info.new_capacity; + cache["disk_total_bytes"] = info.disk_total_capacity; + cache["disk_available_bytes"] = info.disk_available_capacity; + cache["used_bytes"] = info.used_bytes; + cache["pending_eviction_bytes"] = info.pending_eviction_bytes; + cache["clamped_by_disk"] = info.clamped_by_disk; + cache["changed"] = info.changed; + } + return json.ToString(); +} + Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metrics) { const std::string header_json(HEADER_JSON); req->add_output_header(HttpHeaders::CONTENT_TYPE, header_json.c_str()); @@ -105,20 +187,29 @@ Status FileCacheAction::_handle_header(HttpRequest* req, std::string* json_metri int64_t new_capacity = 0; bool parse = true; try { - new_capacity = std::stoll(capacity); + size_t parsed_chars = 0; + new_capacity = std::stoll(capacity, &parsed_chars); + parse = parsed_chars == capacity.size(); } catch (...) { parse = false; } - if (!parse || new_capacity <= 0) { + if (!parse || new_capacity < 0) { st = Status::InvalidArgument( "The capacity {} failed to be parsed, the capacity needs to be in " - "the interval (0, INT64_MAX]", + "the interval [0, INT64_MAX]", capacity); } else { const std::string& path = req->param(std::string(PATH)); - auto ret = io::FileCacheFactory::instance()->reset_capacity(path, new_capacity); - LOG(INFO) << ret; + io::FileCacheResetResult reset_result; + RETURN_IF_ERROR(io::FileCacheFactory::instance()->reset_capacity( + path, static_cast(new_capacity), &reset_result)); + *json_metrics = file_cache_reset_result_to_json(reset_result); } + } else if (operation == INFO) { + const std::string& path = req->param(std::string(PATH)); + std::vector infos; + RETURN_IF_ERROR(io::FileCacheFactory::instance()->get_cache_infos(path, &infos)); + *json_metrics = file_cache_infos_to_json(infos); } else if (operation == HASH) { const std::string& segment_path = req->param(std::string(VALUE)); if (segment_path.empty()) { diff --git a/be/src/storage/options.cpp b/be/src/storage/options.cpp index f8593f18247593..440369f6385313 100644 --- a/be/src/storage/options.cpp +++ b/be/src/storage/options.cpp @@ -306,8 +306,12 @@ Status parse_conf_cache_paths(const std::string& config_path, std::vector_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } io::BlockFileCache* create_cached_remote_reader_test_cache(const fs::path& cache_path, @@ -243,7 +242,6 @@ io::FileCacheSettings cached_remote_reader_cache_settings() { void reset_file_cache_factory_for_test() { FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } Status create_cached_remote_reader_cache(const std::string& cache_path, BlockFileCache** cache) { @@ -3192,7 +3190,6 @@ TEST_F(BlockFileCacheTest, clear_file_cache_sync_factory_rejects_concurrent_sync auto* factory = FileCacheFactory::instance(); factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; io::FileCacheSettings settings; settings.query_queue_size = 1024; @@ -3247,7 +3244,6 @@ TEST_F(BlockFileCacheTest, clear_file_cache_sync_factory_rejects_concurrent_sync sp->clear_all_call_backs(); factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; if (fs::exists(my_cache_path)) { fs::remove_all(my_cache_path); } @@ -3589,7 +3585,6 @@ TEST_F(BlockFileCacheTest, test_factory_1) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, test_factory_2) { @@ -3627,7 +3622,6 @@ TEST_F(BlockFileCacheTest, test_factory_2) { config::clear_file_cache = false; FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, test_factory_3) { @@ -3661,7 +3655,6 @@ TEST_F(BlockFileCacheTest, test_factory_3) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, test_hash_key) { @@ -3803,7 +3796,6 @@ TEST_F(BlockFileCacheTest, test_query_limit) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, state_to_string) { @@ -4192,7 +4184,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_accepts_null_io_context) { @@ -4246,7 +4237,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_accepts_null_io_context) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_tail) { @@ -4317,7 +4307,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_tail) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_error_handle) { @@ -4402,7 +4391,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_error_handle) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } extern bvar::Adder g_read_cache_self_heal_on_not_found; @@ -4728,7 +4716,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_no_self_heal_on_non_not_fou } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_failed_read_does_not_update_metrics) { @@ -4845,7 +4832,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_init) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_concurrent) { @@ -4930,7 +4916,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_concurrent) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_concurrent_2) { @@ -5014,7 +4999,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_concurrent_2) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, test_hot_data) { @@ -5587,7 +5571,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_opt_lock) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; config::enable_read_cache_file_directly = false; } @@ -5772,6 +5755,313 @@ TEST_F(BlockFileCacheTest, remove_from_other_queue_2) { } } +TEST_F(BlockFileCacheTest, capacity_policy_and_settings) { + FileCacheCapacityPolicy policy; + policy.mode = FileCacheCapacityMode::AUTO; + policy.requested_capacity = 0; + policy.normal_percent = 41; + policy.disposable_percent = 7; + policy.index_percent = 11; + policy.ttl_percent = 41; + policy.max_file_block_size = 10; + policy.storage = "disk"; + FileCacheDiskState disk_state {.total_capacity = 1000, + .available_capacity = 600, + .disk_used_percent = 40, + .inode_used_percent = 10}; + + uint64_t effective_capacity = 0; + bool clamped_by_disk = false; + ASSERT_TRUE( + resolve_file_cache_capacity(policy, disk_state, &effective_capacity, &clamped_by_disk)); + EXPECT_EQ(effective_capacity, 1000); + EXPECT_FALSE(clamped_by_disk); + + FileCacheSettings settings; + ASSERT_TRUE(build_file_cache_settings(effective_capacity, policy, &settings)); + EXPECT_EQ(settings.disposable_queue_size, 70); + EXPECT_EQ(settings.index_queue_size, 110); + EXPECT_EQ(settings.ttl_queue_size, 410); + EXPECT_EQ(settings.query_queue_size, 410); + EXPECT_EQ(settings.disposable_queue_size + settings.index_queue_size + settings.ttl_queue_size + + settings.query_queue_size, + settings.capacity); + EXPECT_EQ(settings.query_queue_elements, REMOTE_FS_OBJECTS_CACHE_DEFAULT_ELEMENTS); + + ASSERT_TRUE(build_file_cache_settings(1003, policy, &settings)); + EXPECT_EQ(settings.query_queue_size, 413); + EXPECT_EQ(settings.disposable_queue_size + settings.index_queue_size + settings.ttl_queue_size + + settings.query_queue_size, + settings.capacity); + + policy.max_file_block_size = 1; + ASSERT_TRUE(build_file_cache_settings(20'000'000, policy, &settings)); + EXPECT_EQ(settings.query_queue_elements, 8'200'000); + policy.max_file_block_size = 10; + + policy.mode = FileCacheCapacityMode::MANUAL; + policy.requested_capacity = 1200; + ASSERT_TRUE( + resolve_file_cache_capacity(policy, disk_state, &effective_capacity, &clamped_by_disk)); + EXPECT_EQ(effective_capacity, 1000); + EXPECT_TRUE(clamped_by_disk); + + policy.requested_capacity = 800; + ASSERT_TRUE( + resolve_file_cache_capacity(policy, disk_state, &effective_capacity, &clamped_by_disk)); + EXPECT_EQ(effective_capacity, 800); + EXPECT_FALSE(clamped_by_disk); + + policy.storage = "memory"; + ASSERT_TRUE(resolve_file_cache_capacity(policy, std::nullopt, &effective_capacity, + &clamped_by_disk)); + EXPECT_EQ(effective_capacity, 800); + policy.mode = FileCacheCapacityMode::AUTO; + EXPECT_FALSE(resolve_file_cache_capacity(policy, std::nullopt, &effective_capacity, + &clamped_by_disk)); + + policy.storage = "disk"; + FileCacheDiskState zero_disk; + EXPECT_FALSE( + resolve_file_cache_capacity(policy, zero_disk, &effective_capacity, &clamped_by_disk)); +} + +TEST_F(BlockFileCacheTest, auto_resize_capacity_with_sync_point_disk_state) { + const std::string resize_cache_path = caches_dir / "auto_resize_cache" / ""; + if (fs::exists(resize_cache_path)) { + fs::remove_all(resize_cache_path); + } + fs::create_directories(resize_cache_path); + + const int32_t original_evict_interval = config::file_cache_evict_in_advance_interval_ms; + config::file_cache_evict_in_advance_interval_ms = 60 * 60 * 1000; + Defer restore_evict_interval { + [&] { config::file_cache_evict_in_advance_interval_ms = original_evict_interval; }}; + + std::atomic disk_capacity {1000}; + std::atomic disk_state_calls {0}; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard disk_state_guard; + SyncPoint::CallbackGuard sleep_guard; + sync_point->set_call_back( + "BlockFileCache::get_file_cache_disk_state", + [&](auto&& args) { + auto* state = try_any_cast(args[0]); + state->total_capacity = disk_capacity.load(); + state->available_capacity = state->total_capacity / 2; + state->disk_used_percent = 10; + state->inode_used_percent = 10; + ++disk_state_calls; + }, + &disk_state_guard); + sync_point->set_call_back( + "BlockFileCache::set_sleep_time", + [](auto&& args) { *try_any_cast(args[0]) = 60 * 60 * 1000; }, &sleep_guard); + sync_point->enable_processing(); + + FileCacheCapacityPolicy policy; + policy.mode = FileCacheCapacityMode::AUTO; + policy.max_file_block_size = 10; + policy.storage = "disk"; + FileCacheSettings settings; + ASSERT_TRUE(build_file_cache_settings(1000, policy, &settings)); + + { + BlockFileCache cache(resize_cache_path, settings); + ASSERT_TRUE(cache.initialize()); + wait_until_cache_ready(cache); + for (int i = 0; i < 100 && disk_state_calls.load() == 0; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + ASSERT_GT(disk_state_calls.load(), 0); + + int calls_before_tick = disk_state_calls.load(); + disk_capacity = 2000; + cache.run_background_monitor_once(); + EXPECT_EQ(disk_state_calls.load(), calls_before_tick + 1); + EXPECT_EQ(cache.capacity(), 2000); + EXPECT_EQ(cache._disposable_queue.max_size, 100); + EXPECT_EQ(cache._index_queue.max_size, 100); + EXPECT_EQ(cache._ttl_queue.max_size, 1000); + EXPECT_EQ(cache._normal_queue.max_size, 800); + + CacheContext context; + ReadStatistics stats; + context.stats = &stats; + { + auto holder = + cache.get_or_set(BlockFileCache::hash("auto-resize-data"), 0, 900, context); + for (const auto& block : holder.file_blocks) { + ASSERT_EQ(block->get_or_set_downloader(), FileBlock::get_caller_id()); + std::string data(block->range().size(), 'x'); + ASSERT_TRUE(block->append(Slice(data)).ok()); + ASSERT_TRUE(block->finalize().ok()); + } + } + EXPECT_EQ(cache._cur_cache_size, 900); + + calls_before_tick = disk_state_calls.load(); + disk_capacity = 600; + cache.run_background_monitor_once(); + EXPECT_EQ(disk_state_calls.load(), calls_before_tick + 1); + EXPECT_EQ(cache.capacity(), 600); + EXPECT_EQ(cache._disposable_queue.max_size, 30); + EXPECT_EQ(cache._index_queue.max_size, 30); + EXPECT_EQ(cache._ttl_queue.max_size, 300); + EXPECT_EQ(cache._normal_queue.max_size, 240); + EXPECT_EQ(cache._capacity_state.pending_eviction_bytes, 300); + { + std::lock_guard cache_lock(cache._mutex); + cache.evict_over_capacity(300, cache_lock); + } + EXPECT_EQ(cache._cur_cache_size, 600); + EXPECT_EQ(cache._capacity_state.pending_eviction_bytes, 0); + + disk_capacity = 3000; + BlockFileCacheResetRequest manual_request {.requested_capacity = 700, + .source = FileCacheResizeSource::HTTP, + .expected_generation = std::nullopt}; + FileCachePathResizeResult manual_result; + ASSERT_TRUE(cache.reset_capacity(manual_request, &manual_result)); + EXPECT_TRUE(manual_result.changed); + EXPECT_EQ(manual_result.mode, FileCacheCapacityMode::MANUAL); + EXPECT_EQ(cache.capacity(), 700); + + calls_before_tick = disk_state_calls.load(); + cache.run_background_monitor_once(); + EXPECT_EQ(disk_state_calls.load(), calls_before_tick + 1); + EXPECT_EQ(cache.capacity(), 700); + + BlockFileCacheResetRequest auto_request {.requested_capacity = 0, + .source = FileCacheResizeSource::HTTP, + .expected_generation = std::nullopt}; + FileCachePathResizeResult auto_result; + ASSERT_TRUE(cache.reset_capacity(auto_request, &auto_result)); + EXPECT_EQ(auto_result.mode, FileCacheCapacityMode::AUTO); + EXPECT_EQ(cache.capacity(), 3000); + + FileCacheCapacityPolicy stale_policy; + uint64_t stale_generation = 0; + { + std::lock_guard cache_lock(cache._mutex); + stale_policy = cache._capacity_state.policy; + stale_generation = cache._capacity_state.generation; + } + FileCacheDiskState stale_disk {.total_capacity = 4000, + .available_capacity = 2000, + .disk_used_percent = 10, + .inode_used_percent = 10}; + BlockFileCacheResetRequest stale_request {.source = FileCacheResizeSource::AUTO_REFRESH, + .expected_generation = stale_generation}; + FileCacheResizePlan stale_plan; + ASSERT_TRUE( + build_file_cache_resize_plan(stale_policy, stale_request, stale_disk, &stale_plan)); + manual_request.requested_capacity = 500; + ASSERT_TRUE(cache.reset_capacity(manual_request, &manual_result)); + FileCachePathResizeResult stale_result; + ASSERT_TRUE(cache.apply_reset_capacity(stale_plan, &stale_result)); + EXPECT_TRUE(stale_result.skipped); + EXPECT_EQ(cache.capacity(), 500); + + FileCacheRuntimeInfo info; + ASSERT_TRUE(cache.get_runtime_info(&info)); + EXPECT_EQ(info.capacity_mode, FileCacheCapacityMode::MANUAL); + EXPECT_EQ(info.requested_capacity, 500); + ASSERT_TRUE(info.disk_state.has_value()); + EXPECT_EQ(info.disk_state->total_capacity, 4000); + } + + fs::remove_all(resize_cache_path); +} + +TEST_F(BlockFileCacheTest, factory_aggregates_dynamic_capacity_and_preflights_all_paths) { + reset_file_cache_factory_for_test(); + const std::string path1 = caches_dir / "factory_resize_cache1" / ""; + const std::string path2 = caches_dir / "factory_resize_cache2" / ""; + fs::remove_all(path1); + fs::remove_all(path2); + fs::create_directories(path1); + fs::create_directories(path2); + + std::atomic disk1_capacity {1000}; + std::atomic disk2_capacity {2000}; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard disk_state_guard; + SyncPoint::CallbackGuard sleep_guard; + sync_point->set_call_back( + "BlockFileCache::get_file_cache_disk_state", + [&](auto&& args) { + auto* state = try_any_cast(args[0]); + const auto* path = try_any_cast(args[1]); + state->total_capacity = + *path == path1 ? disk1_capacity.load() : disk2_capacity.load(); + state->available_capacity = state->total_capacity / 2; + state->disk_used_percent = 10; + state->inode_used_percent = 10; + }, + &disk_state_guard); + sync_point->set_call_back( + "BlockFileCache::set_sleep_time", + [](auto&& args) { *try_any_cast(args[0]) = 60 * 60 * 1000; }, &sleep_guard); + sync_point->enable_processing(); + + FileCacheSettings auto_settings = get_file_cache_settings(0, 0); + auto_settings.auto_capacity = true; + ASSERT_TRUE(FileCacheFactory::instance()->create_file_cache(path1, auto_settings)); + ASSERT_TRUE(FileCacheFactory::instance()->create_file_cache(path2, auto_settings)); + EXPECT_EQ(FileCacheFactory::instance()->get_capacity(), 3000); + + auto* cache1 = FileCacheFactory::instance()->get_by_path(path1); + ASSERT_NE(cache1, nullptr); + disk1_capacity = 1500; + cache1->run_background_monitor_once(); + EXPECT_EQ(cache1->capacity(), 1500); + EXPECT_EQ(FileCacheFactory::instance()->get_capacity(), 3500); + + FileCacheResetResult manual_result; + ASSERT_TRUE( + FileCacheFactory::instance()->reset_capacity(path1, uint64_t {500}, &manual_result)); + ASSERT_EQ(manual_result.caches.size(), 1); + EXPECT_EQ(manual_result.caches[0].mode, FileCacheCapacityMode::MANUAL); + EXPECT_EQ(manual_result.total_capacity, 2500); + + { + SyncPoint::CallbackGuard status_guard; + sync_point->set_call_back( + "BlockFileCache::get_file_cache_disk_state:status", + [](auto&& args) { + *try_any_cast(args[0]) = Status::IOError("injected disk error"); + }, + &status_guard); + FileCacheResetResult failed_result; + Status status = + FileCacheFactory::instance()->reset_capacity(path1, uint64_t {400}, &failed_result); + EXPECT_TRUE(status.is()); + EXPECT_EQ(cache1->capacity(), 500); + } + + FileCacheCapacityPolicy memory_policy; + memory_policy.mode = FileCacheCapacityMode::MANUAL; + memory_policy.requested_capacity = 100; + memory_policy.max_file_block_size = 10; + memory_policy.storage = "memory"; + FileCacheSettings memory_settings; + ASSERT_TRUE(build_file_cache_settings(100, memory_policy, &memory_settings)); + ASSERT_TRUE(FileCacheFactory::instance()->create_file_cache("memory", memory_settings)); + const size_t capacity_before_failed_reset = FileCacheFactory::instance()->get_capacity(); + + FileCacheResetResult auto_result; + Status status = FileCacheFactory::instance()->reset_capacity("", uint64_t {0}, &auto_result); + EXPECT_TRUE(status.is()); + EXPECT_EQ(FileCacheFactory::instance()->get_capacity(), capacity_before_failed_reset); + EXPECT_EQ(cache1->capacity(), 500); + + FileCacheFactory::instance()->clear_file_caches(true); + reset_file_cache_factory_for_test(); + fs::remove_all(path1); + fs::remove_all(path2); +} + TEST_F(BlockFileCacheTest, reset_capacity) { if (fs::exists(cache_base_path)) { fs::remove_all(cache_base_path); @@ -5797,8 +6087,11 @@ TEST_F(BlockFileCacheTest, reset_capacity) { auto key = io::BlockFileCache::hash("key1"); auto key2 = io::BlockFileCache::hash("key2"); io::BlockFileCache cache(cache_base_path, settings); + const int32_t original_evict_interval = config::file_cache_evict_in_advance_interval_ms; + config::file_cache_evict_in_advance_interval_ms = 60 * 60 * 1000; auto sp = SyncPoint::get_instance(); - Defer defer {[sp] { + Defer defer {[sp, original_evict_interval] { + config::file_cache_evict_in_advance_interval_ms = original_evict_interval; sp->clear_call_back("BlockFileCache::set_remove_batch"); sp->clear_call_back("BlockFileCache::set_sleep_time"); }}; @@ -5841,7 +6134,15 @@ TEST_F(BlockFileCacheTest, reset_capacity) { } std::cout << cache.reset_capacity(30) << std::endl; + EXPECT_EQ(cache.capacity(), 30); + EXPECT_EQ(cache._cur_cache_size, 90); + EXPECT_EQ(cache._capacity_state.pending_eviction_bytes, 60); + { + std::lock_guard cache_lock(cache._mutex); + cache.evict_over_capacity(60, cache_lock); + } EXPECT_EQ(cache._cur_cache_size, 30); + EXPECT_EQ(cache._capacity_state.pending_eviction_bytes, 0); if (fs::exists(cache_base_path)) { fs::remove_all(cache_base_path); } @@ -8073,7 +8374,6 @@ TEST_F(BlockFileCacheTest, reader_dryrun_when_download_file_cache) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; config::enable_reader_dryrun_when_download_file_cache = org; } @@ -8561,7 +8861,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_ttl_index) { // Then clean up internal state (following the pattern from other tests) FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, cached_remote_file_reader_normal_index) { @@ -8639,7 +8938,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_normal_index) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, test_reset_capacity) { @@ -8717,7 +9015,6 @@ TEST_F(BlockFileCacheTest, test_reset_capacity) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, DISABLE_cached_remote_file_reader_direct_read_and_evict_cache) { @@ -8794,7 +9091,6 @@ TEST_F(BlockFileCacheTest, DISABLE_cached_remote_file_reader_direct_read_and_evi } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } extern bvar::Adder g_read_cache_direct_whole_num; @@ -8916,7 +9212,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_direct_read_bytes_check) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, finalize_empty_block) { diff --git a/be/test/io/cache/block_file_cache_test_common.h b/be/test/io/cache/block_file_cache_test_common.h index 17de4bb814b51e..5d0b7c76dd8b1c 100644 --- a/be/test/io/cache/block_file_cache_test_common.h +++ b/be/test/io/cache/block_file_cache_test_common.h @@ -70,7 +70,6 @@ namespace doris::io { namespace fs = std::filesystem; -extern int disk_used_percentage(const std::string& path, std::pair* percent); extern fs::path caches_dir; extern std::string cache_base_path; extern std::string tmp_file; diff --git a/be/test/io/cache/block_file_cache_test_lru_dump.cpp b/be/test/io/cache/block_file_cache_test_lru_dump.cpp index 3a581d77c76590..16aa94fba34059 100644 --- a/be/test/io/cache/block_file_cache_test_lru_dump.cpp +++ b/be/test/io/cache/block_file_cache_test_lru_dump.cpp @@ -659,7 +659,6 @@ TEST_F(BlockFileCacheTest, cached_remote_file_reader_direct_read_order_check) { } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } TEST_F(BlockFileCacheTest, get_or_set_hit_order_check) { diff --git a/be/test/io/cache/cached_remote_file_reader_lock_wait_test.cpp b/be/test/io/cache/cached_remote_file_reader_lock_wait_test.cpp index 97b846a1db2e73..0cb01fea833db0 100644 --- a/be/test/io/cache/cached_remote_file_reader_lock_wait_test.cpp +++ b/be/test/io/cache/cached_remote_file_reader_lock_wait_test.cpp @@ -172,7 +172,6 @@ class CachedRemoteFileReaderLockWaitTest : public testing::Test { factory->clear_file_caches(true); factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; } if (_owns_factory) { ExecEnv::GetInstance()->_file_cache_factory = nullptr; @@ -193,7 +192,6 @@ class CachedRemoteFileReaderLockWaitTest : public testing::Test { factory->clear_file_caches(true); factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; } } @@ -204,7 +202,6 @@ class CachedRemoteFileReaderLockWaitTest : public testing::Test { factory->clear_file_caches(true); factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; constexpr size_t kCapacityBytes = 512ULL * 1024ULL * 1024ULL; auto settings = get_file_cache_settings(kCapacityBytes, 0, DEFAULT_NORMAL_PERCENT, diff --git a/be/test/io/cache/cached_remote_file_reader_peer_test.cpp b/be/test/io/cache/cached_remote_file_reader_peer_test.cpp index ebf96a59e2794e..b39066b3fa416d 100644 --- a/be/test/io/cache/cached_remote_file_reader_peer_test.cpp +++ b/be/test/io/cache/cached_remote_file_reader_peer_test.cpp @@ -109,7 +109,6 @@ FileCacheSettings create_peer_test_settings(size_t block_size, std::string stora void clear_cached_remote_reader_factory() { FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; } BlockFileCache* create_peer_test_cache(const fs::path& cache_path, size_t block_size, diff --git a/be/test/io/cache/cached_remote_file_reader_test.cpp b/be/test/io/cache/cached_remote_file_reader_test.cpp index 84cce46c9980aa..1e9350c3a87985 100644 --- a/be/test/io/cache/cached_remote_file_reader_test.cpp +++ b/be/test/io/cache/cached_remote_file_reader_test.cpp @@ -118,7 +118,6 @@ TEST_F(BlockFileCacheTest, } FileCacheFactory::instance()->_caches.clear(); FileCacheFactory::instance()->_path_to_cache.clear(); - FileCacheFactory::instance()->_capacity = 0; config::enable_read_cache_file_directly = false; } diff --git a/be/test/io/fs/packed_file_concurrency_test.cpp b/be/test/io/fs/packed_file_concurrency_test.cpp index 14c41d53a7b442..b7f1575bd560b9 100644 --- a/be/test/io/fs/packed_file_concurrency_test.cpp +++ b/be/test/io/fs/packed_file_concurrency_test.cpp @@ -566,7 +566,6 @@ void cleanup_file_cache_resources() { if (factory != nullptr) { factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; } g_cache_initialized = false; g_cache_base_path.clear(); diff --git a/be/test/service/http/file_cache_action_test.cpp b/be/test/service/http/file_cache_action_test.cpp index ce4f8b02372728..de8f55f1255944 100644 --- a/be/test/service/http/file_cache_action_test.cpp +++ b/be/test/service/http/file_cache_action_test.cpp @@ -28,6 +28,7 @@ #include #include "common/config.h" +#include "cpp/sync_point.h" #include "io/cache/block_file_cache.h" #include "io/cache/block_file_cache_factory.h" #include "io/cache/file_cache_common.h" @@ -102,6 +103,7 @@ class FileCacheActionTest : public testing::Test { if (doris::ExecEnv::GetInstance()->file_cache_factory() != nullptr) { reset_file_cache_factory(); } + (void)std::filesystem::remove_all(_base_path); } protected: @@ -118,7 +120,6 @@ class FileCacheActionTest : public testing::Test { factory->clear_file_caches(true); factory->_caches.clear(); factory->_path_to_cache.clear(); - factory->_capacity = 0; _cache = nullptr; } @@ -242,4 +243,135 @@ TEST_F(FileCacheActionTest, clear_value_uses_async_remove) { EXPECT_TRUE(_cache->get_blocks_by_key(hash).empty()); } +TEST_F(FileCacheActionTest, info_returns_capacity_policy_and_queues) { + HttpRequest req(_evhttp_req); + std::string json_metrics; + req._params["op"] = "info"; + + Status status = _action->_handle_header(&req, &json_metrics); + + ASSERT_TRUE(status.ok()) << status; + EXPECT_NE(json_metrics.find("\"status\":\"OK\""), std::string::npos); + EXPECT_NE(json_metrics.find("\"path\":\"memory\""), std::string::npos); + EXPECT_NE(json_metrics.find("\"storage\":\"memory\""), std::string::npos); + EXPECT_NE(json_metrics.find("\"capacity_mode\":\"MANUAL\""), std::string::npos); + EXPECT_NE(json_metrics.find("\"requested_capacity\":1048576"), std::string::npos); + EXPECT_NE(json_metrics.find("\"effective_capacity\":1048576"), std::string::npos); + EXPECT_NE(json_metrics.find("\"disk_total_bytes\":null"), std::string::npos); + EXPECT_NE(json_metrics.find("\"normal\":{"), std::string::npos); + EXPECT_NE(json_metrics.find("\"last_resize\":{"), std::string::npos); +} + +TEST_F(FileCacheActionTest, info_by_path_and_unknown_path) { + HttpRequest req(_evhttp_req); + std::string json_metrics; + req._params["op"] = "info"; + req._params["path"] = _base_path; + ASSERT_TRUE(_action->_handle_header(&req, &json_metrics)); + EXPECT_NE(json_metrics.find("\"path\":\"memory\""), std::string::npos); + + req._params["path"] = "/not/a/cache"; + Status status = _action->_handle_header(&req, &json_metrics); + EXPECT_TRUE(status.is()); + EXPECT_NE(status.to_string().find("unknown file cache path"), std::string::npos); +} + +TEST_F(FileCacheActionTest, reset_manual_returns_structured_result) { + HttpRequest req(_evhttp_req); + std::string json_metrics; + req._params["op"] = "reset"; + req._params["path"] = _base_path; + req._params["capacity"] = "9007199254740993"; + + Status status = _action->_handle_header(&req, &json_metrics); + + ASSERT_TRUE(status.ok()) << status; + EXPECT_EQ(_cache->capacity(), 9007199254740993ULL); + EXPECT_NE(json_metrics.find("\"capacity_mode\":\"MANUAL\""), std::string::npos); + EXPECT_NE(json_metrics.find("\"requested_capacity\":9007199254740993"), std::string::npos); + EXPECT_NE(json_metrics.find("\"old_capacity\":1048576"), std::string::npos); + EXPECT_NE(json_metrics.find("\"effective_capacity\":9007199254740993"), std::string::npos); + EXPECT_NE(json_metrics.find("\"changed\":true"), std::string::npos); +} + +TEST_F(FileCacheActionTest, reset_rejects_invalid_values_and_unknown_path) { + HttpRequest req(_evhttp_req); + req._params["op"] = "reset"; + req._params["path"] = _base_path; + for (const std::string& value : {"", "-1", "abc", "1junk", "9223372036854775808"}) { + std::string json_metrics; + req._params["capacity"] = value; + Status status = _action->_handle_header(&req, &json_metrics); + EXPECT_TRUE(status.is()) << value << ": " << status; + EXPECT_EQ(_cache->capacity(), 1024 * 1024); + } + + std::string json_metrics; + req._params["capacity"] = "1024"; + req._params["path"] = "/not/a/cache"; + Status status = _action->_handle_header(&req, &json_metrics); + EXPECT_TRUE(status.is()); + EXPECT_EQ(_cache->capacity(), 1024 * 1024); +} + +TEST_F(FileCacheActionTest, reset_zero_enables_auto_mode_for_disk_cache) { + reset_file_cache_factory(); + std::filesystem::create_directories(_base_path); + + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard disk_state_guard; + SyncPoint::CallbackGuard sleep_guard; + sync_point->set_call_back( + "BlockFileCache::get_file_cache_disk_state", + [](auto&& args) { + auto* state = try_any_cast(args[0]); + state->total_capacity = 4096; + state->available_capacity = 2048; + state->disk_used_percent = 10; + state->inode_used_percent = 10; + }, + &disk_state_guard); + sync_point->set_call_back( + "BlockFileCache::set_sleep_time", + [](auto&& args) { *try_any_cast(args[0]) = 60 * 60 * 1000; }, &sleep_guard); + sync_point->enable_processing(); + + io::FileCacheCapacityPolicy policy; + policy.mode = io::FileCacheCapacityMode::MANUAL; + policy.requested_capacity = 1024; + policy.max_file_block_size = 64; + policy.storage = "disk"; + io::FileCacheSettings settings; + ASSERT_TRUE(io::build_file_cache_settings(1024, policy, &settings)); + ASSERT_TRUE(io::FileCacheFactory::instance()->create_file_cache(_base_path, settings)); + _cache = io::FileCacheFactory::instance()->get_by_path(_base_path); + ASSERT_NE(_cache, nullptr); + + HttpRequest req(_evhttp_req); + std::string json_metrics; + req._params["op"] = "reset"; + req._params["path"] = _base_path; + req._params["capacity"] = "0"; + Status status = _action->_handle_header(&req, &json_metrics); + + ASSERT_TRUE(status.ok()) << status; + EXPECT_EQ(_cache->capacity(), 4096); + EXPECT_NE(json_metrics.find("\"capacity_mode\":\"AUTO\""), std::string::npos); + EXPECT_NE(json_metrics.find("\"requested_capacity\":0"), std::string::npos); + EXPECT_NE(json_metrics.find("\"effective_capacity\":4096"), std::string::npos); +} + +TEST_F(FileCacheActionTest, reset_zero_rejects_memory_cache) { + HttpRequest req(_evhttp_req); + std::string json_metrics; + req._params["op"] = "reset"; + req._params["path"] = _base_path; + req._params["capacity"] = "0"; + + Status status = _action->_handle_header(&req, &json_metrics); + + EXPECT_TRUE(status.is()); + EXPECT_EQ(_cache->capacity(), 1024 * 1024); +} + } // namespace doris