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