diff --git a/include/cachinglayer/CacheSlot.h b/include/cachinglayer/CacheSlot.h index bd38163..4c68ef2 100644 --- a/include/cachinglayer/CacheSlot.h +++ b/include/cachinglayer/CacheSlot.h @@ -31,7 +31,7 @@ #include #include -#include "cachinglayer/LoadingOverheadTracker.h" +#include "cachinglayer/LoadingOverhead.h" #include "cachinglayer/Metrics.h" #include "cachinglayer/Translator.h" #include "cachinglayer/Utils.h" @@ -68,7 +68,8 @@ class CacheSlot final : public std::enable_shared_from_this> { self_reserve_(self_reserve), storage_usage_tracking_enabled_(storage_usage_tracking_enabled), loading_timeout_(loading_timeout), - warmup_loading_timeout_(warmup_loading_timeout) { + warmup_loading_timeout_(warmup_loading_timeout), + loading_overhead_config_(translator_->meta()->loading_overhead_config) { if (const auto& metric_attribution = translator_->meta()->metric_attribution; metric_attribution && !metric_attribution->shard.empty()) { shard_disk_usage_metric_ = @@ -78,13 +79,11 @@ class CacheSlot final : public std::enable_shared_from_this> { for (cid_t i = 0; i < static_cast(translator_->num_cells()); ++i) { cells_.push_back(std::make_unique(this, i)); } + if (loading_overhead_config_) { + dlist_->BindLoadingOverheadGroups(*loading_overhead_config_); + } monitor::cache_slot_count(cell_data_type_, storage_type_).Increment(); monitor::cache_cell_count(cell_data_type_, storage_type_).Increment(translator_->num_cells()); - // Register after all potentially-throwing operations, so that if the constructor - // fails, we don't leak a ref_count (destructor won't run for incomplete objects). - if (auto& lo = translator_->meta()->loading_overhead) { - overhead_handle_ = dlist_->RegisterLoadingOverhead(*lo); - } } CacheSlot(const CacheSlot&) = delete; @@ -345,7 +344,9 @@ class CacheSlot final : public std::enable_shared_from_this> { } ~CacheSlot() { - dlist_->UnregisterLoadingOverhead(overhead_handle_); + if (loading_overhead_config_) { + dlist_->UnbindLoadingOverheadGroups(*loading_overhead_config_); + } monitor::cache_slot_count(cell_data_type_, storage_type_).Decrement(); monitor::cache_cell_count(cell_data_type_, storage_type_).Decrement(translator_->num_cells()); } @@ -353,6 +354,11 @@ class CacheSlot final : public std::enable_shared_from_this> { private: friend class CellAccessor; + [[nodiscard]] const LoadingOverheadConfig* + loadingOverheadConfig() const noexcept { + return loading_overhead_config_ ? &loading_overhead_config_.value() : nullptr; + } + [[nodiscard]] std::vector AllCellIds() const { std::vector cids(translator_->num_cells()); @@ -437,8 +443,8 @@ class CacheSlot final : public std::enable_shared_from_this> { RunLoad(OpContext* ctx, std::unordered_set&& cids, std::chrono::milliseconds timeout) { // loaded_resource: the estimated final resource usage (from .first), reserved unconditionally. // loading_overhead: the estimated temporary overhead during loading (from .second). - // Configured dimensions are group-capped by LoadingOverheadTracker; omitted dimensions - // pass through unchanged. The tracker returns the combined incremental DList delta. + // Configured dimensions are capped by their loading-overhead Group; omitted dimensions + // pass through unchanged. DList derives the incremental reservation from the Group target transition. std::vector loading_cids; try { auto start = std::chrono::steady_clock::now(); @@ -484,8 +490,8 @@ class CacheSlot final : public std::enable_shared_from_this> { } // loaded_resource is reserved unconditionally from DList (no capping). - // loading_overhead goes through the tracker: configured dimensions return the change - // in min(sum, UB), while omitted dimensions return their full overhead. + // loading_overhead goes through its Groups: configured dimensions return the change + // in min(sum, bound), while omitted dimensions return their full overhead. auto loaded_resource = essential_loaded_resource + bonus_loaded_resource; auto loading_overhead = essential_loading_overhead + bonus_loading_overhead; @@ -500,27 +506,28 @@ class CacheSlot final : public std::enable_shared_from_this> { // If that fails, fall back to essential-only with the real timeout. // This avoids blocking in the waiting queue for a bonus attempt that could retry immediately. auto reserve_timeout = bonus_cids.empty() ? timeout : std::chrono::milliseconds(0); - auto actual_dlist_reserve = SemiInlineGet(dlist_->ReserveLoadingResourceWithTimeout( - loaded_resource, loading_overhead, overhead_handle_, reserve_timeout, ctx)); - bool reservation_success = actual_dlist_reserve.AnyGTZero(); + auto reservation = SemiInlineGet(dlist_->ReserveLoadingResourceWithTimeout( + loaded_resource, loading_overhead, loadingOverheadConfig(), reserve_timeout, ctx)); + auto actual_dlist_reserve = reservation.reserved; + bool reservation_success = reservation.success; - // Guard: releases DList + tracker atomically. All by-ref so bonus retry + // Guard: releases DList + Groups atomically. All by-ref so bonus retry // updates are reflected automatically. bool metrics_tracked = false; size_t loading_cids_count = 0; + ResourceUsage metric_loading_resource{}; auto defer_release = folly::makeGuard([&]() { if (!reservation_success) { return; } try { - auto released_resource = - dlist_->ReleaseLoadingResource(loaded_resource, loading_overhead, overhead_handle_); + dlist_->ReleaseLoadingResource(loaded_resource, loading_overhead, loadingOverheadConfig()); if (metrics_tracked) { monitor::cache_cell_loading_count(cell_data_type_, storage_type_).Decrement(loading_cids_count); monitor::cache_loading_bytes(cell_data_type_, StorageType::MEMORY) - .Decrement(released_resource.memory_bytes); + .Decrement(metric_loading_resource.memory_bytes); monitor::cache_loading_bytes(cell_data_type_, StorageType::DISK) - .Decrement(released_resource.file_bytes); + .Decrement(metric_loading_resource.file_bytes); } } catch (...) { auto ew = folly::exception_wrapper(std::current_exception()); @@ -539,9 +546,10 @@ class CacheSlot final : public std::enable_shared_from_this> { "essential loading resource"); loaded_resource = essential_loaded_resource; loading_overhead = essential_loading_overhead; - actual_dlist_reserve = SemiInlineGet(dlist_->ReserveLoadingResourceWithTimeout( - loaded_resource, loading_overhead, overhead_handle_, timeout, ctx)); - reservation_success = actual_dlist_reserve.AnyGTZero(); + reservation = SemiInlineGet(dlist_->ReserveLoadingResourceWithTimeout( + loaded_resource, loading_overhead, loadingOverheadConfig(), timeout, ctx)); + actual_dlist_reserve = reservation.reserved; + reservation_success = reservation.success; } else { loading_cids.insert(loading_cids.end(), bonus_cids.begin(), bonus_cids.end()); } @@ -562,9 +570,14 @@ class CacheSlot final : public std::enable_shared_from_this> { loading_overhead.ToString(), actual_dlist_reserve.ToString()); } + // Track the active request estimate rather than the mutable Group reservation. + // Group reconfiguration may change DList bookkeeping while this request is in flight, + // but the metric must decrement the same value that it incremented. + metric_loading_resource = loaded_resource + loading_overhead; monitor::cache_loading_bytes(cell_data_type_, StorageType::MEMORY) - .Increment(actual_dlist_reserve.memory_bytes); - monitor::cache_loading_bytes(cell_data_type_, StorageType::DISK).Increment(actual_dlist_reserve.file_bytes); + .Increment(metric_loading_resource.memory_bytes); + monitor::cache_loading_bytes(cell_data_type_, StorageType::DISK) + .Increment(metric_loading_resource.file_bytes); loading_cids_count = loading_cids.size(); monitor::cache_cell_loading_count(cell_data_type_, storage_type_).Increment(loading_cids_count); metrics_tracked = true; @@ -719,7 +732,8 @@ class CacheSlot final : public std::enable_shared_from_this> { const bool storage_usage_tracking_enabled_; std::chrono::milliseconds loading_timeout_{100000}; std::chrono::milliseconds warmup_loading_timeout_{0}; - uint64_t overhead_handle_{0}; + // Bind and Unbind must use the same runtime-unit metadata even if Meta is later modified. + std::optional loading_overhead_config_; std::atomic warmup_called_{false}; std::atomic skip_pin_{false}; }; diff --git a/include/cachinglayer/LoadingOverhead.h b/include/cachinglayer/LoadingOverhead.h index 7d114d8..d8350c4 100644 --- a/include/cachinglayer/LoadingOverhead.h +++ b/include/cachinglayer/LoadingOverhead.h @@ -11,8 +11,13 @@ #pragma once +#include #include +#include +#include #include +#include +#include #include #include @@ -20,29 +25,207 @@ namespace milvus::cachinglayer { -struct LoadingOverheadDimensionConfig { - int64_t upper_bound; - std::string group; -}; +/** @brief Immutable Group policy used to derive a reservation bound. */ +class LoadingOverheadPolicy { + public: + enum class Kind { + kFixed, + kPassthrough, + kBudget, + kExecutor, + }; -// A missing dimension is not group-capped. Its loading overhead passes through -// unchanged and is still checked against the DList resource limit. -struct LoadingOverheadConfig { - LoadingOverheadConfig() = default; + static LoadingOverheadPolicy + Fixed(int64_t upper_bound) { + return {Kind::kFixed, ValidateNonNegative(upper_bound, "fixed upper bound")}; + } + + static LoadingOverheadPolicy + Passthrough() { + return {Kind::kPassthrough, 0}; + } + + static LoadingOverheadPolicy + Budget(int64_t capacity_bytes) { + return {Kind::kBudget, ValidateNonNegative(capacity_bytes, "Budget capacity")}; + } + + static LoadingOverheadPolicy + Executor(int64_t configured_workers) { + return {Kind::kExecutor, ValidateNonNegative(configured_workers, "executor worker count")}; + } - LoadingOverheadConfig(std::optional memory, - std::optional file) - : memory(std::move(memory)), file(std::move(file)) { + [[nodiscard]] int64_t + ResolveBound(int64_t max_runtime_unit_bytes) const noexcept { + switch (kind_) { + case Kind::kFixed: + return value_; + case Kind::kPassthrough: + return std::numeric_limits::max(); + case Kind::kBudget: + if (max_runtime_unit_bytes < 0 || value_ == 0) { + return std::numeric_limits::max(); + } + return std::max(value_, max_runtime_unit_bytes); + case Kind::kExecutor: + return SaturatingMultiply(value_, max_runtime_unit_bytes); + } + return std::numeric_limits::max(); } - // Compatibility entry point for callers using the original shared-group config. - LoadingOverheadConfig(const ResourceUsage& upper_bound, std::string group) - : memory(LoadingOverheadDimensionConfig{upper_bound.memory_bytes, group}), - file(LoadingOverheadDimensionConfig{upper_bound.file_bytes, std::move(group)}) { + [[nodiscard]] bool + RequiresRuntimeUnitBound() const noexcept { + return kind_ == Kind::kBudget || kind_ == Kind::kExecutor; } - std::optional memory; - std::optional file; + private: + LoadingOverheadPolicy(Kind kind, int64_t value) : kind_(kind), value_(value) { + } + + static int64_t + ValidateNonNegative(int64_t value, const char* name) { + if (value < 0) { + throw std::invalid_argument(std::string("LoadingOverheadPolicy ") + name + " must be non-negative"); + } + return value; + } + + static int64_t + SaturatingMultiply(int64_t lhs, int64_t rhs) noexcept { + if (lhs < 0 || rhs < 0) { + return std::numeric_limits::max(); + } + if (lhs == 0 || rhs == 0) { + return 0; + } + if (lhs > std::numeric_limits::max() / rhs) { + return std::numeric_limits::max(); + } + return lhs * rhs; + } + + Kind kind_; + int64_t value_; +}; + +/** @brief Resource dimension managed by a loading-overhead Group. */ +enum class LoadingOverheadDimension { + kMemory, + kFile, +}; + +enum class LoadingOverheadUpdateResult; + +namespace internal { +class DList; +} + +// Opaque shared state for one independently created loading-overhead Group. +// DList serializes mutations with admission accounting under its list mutex. +class LoadingOverheadGroup { + public: + LoadingOverheadGroup(const LoadingOverheadGroup&) = delete; + LoadingOverheadGroup& + operator=(const LoadingOverheadGroup&) = delete; + LoadingOverheadGroup(LoadingOverheadGroup&&) = delete; + LoadingOverheadGroup& + operator=(LoadingOverheadGroup&&) = delete; + ~LoadingOverheadGroup() = default; + + private: + friend class internal::DList; + + LoadingOverheadGroup(const internal::DList* owner, LoadingOverheadDimension dimension, LoadingOverheadPolicy policy) + : owner_(owner), dimension_(dimension), policy_(std::move(policy)) { + } + + void + validateBinding(const internal::DList* owner, LoadingOverheadDimension dimension, + const std::optional& max_runtime_unit) const; + + void + bind(const std::optional& max_runtime_unit); + + void + unbind(const internal::DList* owner, LoadingOverheadDimension dimension, + const std::optional& max_runtime_unit) noexcept; + + LoadingOverheadUpdateResult + updatePolicy(const internal::DList* owner, LoadingOverheadPolicy policy); + + int64_t + reserve(int64_t overhead); + + void + rollbackReserve(int64_t overhead, int64_t reserved) noexcept; + + int64_t + release(int64_t overhead) noexcept; + + [[nodiscard]] int64_t + computeTarget() const noexcept; + + [[nodiscard]] bool + hasMissingRuntimeUnit() const noexcept; + + const internal::DList* owner_; + LoadingOverheadDimension dimension_; + LoadingOverheadPolicy policy_; + int64_t sum_of_overhead_{0}; + std::multiset runtime_unit_bounds_; + int64_t max_runtime_unit_{0}; + int64_t overhead_reserved_{0}; + uint64_t binding_count_{0}; +}; + +/** + * @brief Translator binding to one loading-overhead Group. + */ +struct LoadingOverheadGroupBinding { + /** + * @brief Group created independently before this binding is attached. + */ + std::shared_ptr group; + + /** + * @brief Optional conservative bound for one runtime unit from this binding. + * + * The Group caches the maximum value across attached bindings. A + * policy that bounds Budget acquisitions or executor tasks requires + * this value on every binding; binding or policy replacement is + * rejected otherwise. Fixed and Passthrough Groups do not require it. It + * must describe one runtime unit, not the sum of a multi-cell load request. + * + * @pre When present, the value is non-negative. + */ + std::optional max_runtime_unit; +}; + +/** + * @brief Loading-overhead configuration for a Translator. + * + * Each resource dimension binds independently to a Group. An absent dimension is + * not assigned to a Group; its request-local loading overhead passes through + * unchanged and remains subject to DList admission limits. + */ +struct LoadingOverheadConfig { + /** @brief Memory-dimension binding, or std::nullopt for request-local passthrough. */ + std::optional memory; + + /** @brief File-dimension binding, or std::nullopt for request-local passthrough. */ + std::optional file; +}; + +/** @brief Result of updating a loading-overhead Group. */ +enum class LoadingOverheadUpdateResult { + /** The desired policy was committed. */ + kApplied, + + /** The requested policy is incompatible with the Group's bindings. */ + kIncompatiblePolicy, + + /** The Group handle was invalid. */ + kInvalidArgument, }; } // namespace milvus::cachinglayer diff --git a/include/cachinglayer/LoadingOverheadTracker.h b/include/cachinglayer/LoadingOverheadTracker.h deleted file mode 100644 index c42f05f..0000000 --- a/include/cachinglayer/LoadingOverheadTracker.h +++ /dev/null @@ -1,262 +0,0 @@ -// Copyright (C) 2019-2026 Zilliz. All rights reserved. -// -// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software distributed under the License -// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express -// or implied. See the License for the specific language governing permissions and limitations under the License - -#pragma once - -#include -#include -#include -#include -#include - -#include "cachinglayer/LoadingOverhead.h" -#include "cachinglayer/Utils.h" -#include "log/Log.h" - -namespace milvus::cachinglayer { - -// Manages per-dimension, per-group loading overhead reservation with an upper bound (UB). -// -// Memory and file groups are independent even when they use the same string key. -// Registration returns a composite uint64_t handle for O(1) lookup on the hot path. -// -// The total loading overhead reserved from DList for each configured dimension is capped: -// DList dimension reservation = min(sum_of_overhead, dimension_UB) -// An unconfigured dimension passes through unchanged. INT64_MAX is an explicit -// unlimited upper bound; re-registering a group can only increase its upper bound. -// -// Each Reserve/Release call returns the incremental delta to apply to DList. -// The tracker directly tracks `overhead_reserved` (actual amount of overhead currently -// reserved in DList) to ensure correctness. -// -// The compatibility Register overload configures both dimensions. A missing -// dimension in LoadingOverheadConfig preserves pass-through behavior. -class LoadingOverheadTracker { - public: - static inline const ResourceUsage kUnlimited{std::numeric_limits::max(), - std::numeric_limits::max()}; - - static constexpr uint64_t kInvalidHandle = 0; - - uint64_t - Register(const LoadingOverheadConfig& config) { - std::lock_guard lock(mtx_); - RegistrationState registration; - if (config.memory.has_value()) { - registration.memory_group_handle = - registerDimensionGroup(memory_name_to_group_handle_, config.memory.value(), "memory"); - } - if (config.file.has_value()) { - registration.file_group_handle = - registerDimensionGroup(file_name_to_group_handle_, config.file.value(), "file"); - } - - auto handle = next_registration_handle_++; - registration_state_[handle] = registration; - return handle; - } - - // Compatibility entry point. The two dimensions use independent group state. - uint64_t - Register(const std::string& group, const ResourceUsage& upper_bound) { - return Register(LoadingOverheadConfig{LoadingOverheadDimensionConfig{upper_bound.memory_bytes, group}, - LoadingOverheadDimensionConfig{upper_bound.file_bytes, group}}); - } - - // Called before loading. Returns the delta to reserve from DList for loading overhead. - ResourceUsage - Reserve(uint64_t handle, const ResourceUsage& loading_overhead) { - std::lock_guard lock(mtx_); - auto it = registration_state_.find(handle); - if (it == registration_state_.end()) { - return loading_overhead; - } - auto delta = loading_overhead; - if (it->second.memory_group_handle != kInvalidHandle) { - delta.memory_bytes = reserveDimension(it->second.memory_group_handle, loading_overhead.memory_bytes); - } - if (it->second.file_group_handle != kInvalidHandle) { - delta.file_bytes = reserveDimension(it->second.file_group_handle, loading_overhead.file_bytes); - } - return delta; - } - - // Called to release a previous Reserve. - // Returns the delta to release from DList for loading overhead. - ResourceUsage - Release(uint64_t handle, const ResourceUsage& loading_overhead) { - std::lock_guard lock(mtx_); - auto it = registration_state_.find(handle); - if (it == registration_state_.end()) { - return loading_overhead; - } - auto delta = loading_overhead; - if (it->second.memory_group_handle != kInvalidHandle) { - delta.memory_bytes = releaseDimension(it->second.memory_group_handle, loading_overhead.memory_bytes); - } - if (it->second.file_group_handle != kInvalidHandle) { - delta.file_bytes = releaseDimension(it->second.file_group_handle, loading_overhead.file_bytes); - } - return delta; - } - - bool - HasFiniteUpperBound(uint64_t handle) const { - std::lock_guard lock(mtx_); - return !(getUpperBoundLocked(handle) == kUnlimited); - } - - ResourceUsage - GetUpperBound(uint64_t handle) const { - std::lock_guard lock(mtx_); - return getUpperBoundLocked(handle); - } - - // Decrement ref count for a group. When ref count reaches 0, the group is - // unconditionally removed. Safe to call from CacheSlot destructor. - void - Unregister(uint64_t handle) { - if (handle == kInvalidHandle) { - return; - } - std::lock_guard lock(mtx_); - auto it = registration_state_.find(handle); - if (it == registration_state_.end()) { - return; - } - unregisterDimensionGroup(memory_name_to_group_handle_, it->second.memory_group_handle, "memory"); - unregisterDimensionGroup(file_name_to_group_handle_, it->second.file_group_handle, "file"); - registration_state_.erase(it); - } - - private: - struct DimensionGroupState { - int64_t upper_bound{0}; - int64_t sum_of_overhead{0}; - int64_t overhead_reserved{0}; - uint64_t ref_count{0}; - std::string group_name; - }; - - struct RegistrationState { - uint64_t memory_group_handle{kInvalidHandle}; - uint64_t file_group_handle{kInvalidHandle}; - }; - - uint64_t - registerDimensionGroup(std::unordered_map& name_to_handle, - const LoadingOverheadDimensionConfig& config, const char* dimension) { - auto it = name_to_handle.find(config.group); - if (it != name_to_handle.end()) { - auto& state = dimension_group_state_[it->second]; - state.ref_count++; - if (state.upper_bound < config.upper_bound) { - LOG_WARN( - "[MCL] LoadingOverheadTracker {} UB mismatch for group '{}' (handle {}): existing={}, new={}. " - "Taking max.", - dimension, config.group, it->second, state.upper_bound, config.upper_bound); - state.upper_bound = config.upper_bound; - } else { - LOG_DEBUG("[MCL] LoadingOverheadTracker re-registered {} group '{}' (handle {}, refs={}), UB unchanged", - dimension, config.group, it->second, state.ref_count); - } - return it->second; - } - - auto handle = next_group_handle_++; - name_to_handle[config.group] = handle; - dimension_group_state_[handle] = DimensionGroupState{config.upper_bound, 0, 0, 1, config.group}; - LOG_INFO("[MCL] LoadingOverheadTracker registered {} group '{}' (handle {}, refs=1): UB={}", dimension, - config.group, handle, config.upper_bound); - return handle; - } - - int64_t - reserveDimension(uint64_t group_handle, int64_t overhead) { - auto& state = dimension_group_state_.at(group_handle); - state.sum_of_overhead += overhead; - auto target = std::min(std::max(state.sum_of_overhead, int64_t{0}), state.upper_bound); - auto delta = std::max(target - state.overhead_reserved, int64_t{0}); - state.overhead_reserved += delta; - return delta; - } - - int64_t - releaseDimension(uint64_t group_handle, int64_t overhead) { - auto& state = dimension_group_state_.at(group_handle); - state.sum_of_overhead -= overhead; - if (state.sum_of_overhead < 0) { - LOG_ERROR("[MCL] LoadingOverheadTracker Release group handle {}: sum_of_overhead < 0", group_handle); - state.sum_of_overhead = 0; - } - auto target = std::min(state.sum_of_overhead, state.upper_bound); - auto delta = std::max(state.overhead_reserved - target, int64_t{0}); - state.overhead_reserved -= delta; - return delta; - } - - ResourceUsage - getUpperBoundLocked(uint64_t registration_handle) const { - auto it = registration_state_.find(registration_handle); - if (it == registration_state_.end()) { - return kUnlimited; - } - ResourceUsage result = kUnlimited; - if (it->second.memory_group_handle != kInvalidHandle) { - result.memory_bytes = dimension_group_state_.at(it->second.memory_group_handle).upper_bound; - } - if (it->second.file_group_handle != kInvalidHandle) { - result.file_bytes = dimension_group_state_.at(it->second.file_group_handle).upper_bound; - } - return result; - } - - void - unregisterDimensionGroup(std::unordered_map& name_to_handle, uint64_t group_handle, - const char* dimension) { - if (group_handle == kInvalidHandle) { - return; - } - auto it = dimension_group_state_.find(group_handle); - if (it == dimension_group_state_.end()) { - return; - } - auto& state = it->second; - if (state.ref_count > 0) { - state.ref_count--; - } - if (state.ref_count > 0) { - LOG_DEBUG("[MCL] LoadingOverheadTracker {} group handle {} ref_count decremented to {}", dimension, - group_handle, state.ref_count); - return; - } - if (state.sum_of_overhead > 0 || state.overhead_reserved > 0) { - LOG_ERROR( - "[MCL] LoadingOverheadTracker {} group handle {} ref_count=0 with residual reservations: " - "sum_of_overhead={}, overhead_reserved={}. Cleaning up anyway to avoid leak.", - dimension, group_handle, state.sum_of_overhead, state.overhead_reserved); - } - LOG_INFO("[MCL] LoadingOverheadTracker unregistered {} group '{}' (handle {})", dimension, state.group_name, - group_handle); - name_to_handle.erase(state.group_name); - dimension_group_state_.erase(it); - } - - mutable std::mutex mtx_; - std::unordered_map memory_name_to_group_handle_; - std::unordered_map file_name_to_group_handle_; - std::unordered_map dimension_group_state_; - std::unordered_map registration_state_; - uint64_t next_group_handle_{1}; - uint64_t next_registration_handle_{1}; -}; - -} // namespace milvus::cachinglayer diff --git a/include/cachinglayer/Manager.h b/include/cachinglayer/Manager.h index 8560bfa..e67cafd 100644 --- a/include/cachinglayer/Manager.h +++ b/include/cachinglayer/Manager.h @@ -16,7 +16,7 @@ #include #include "cachinglayer/CacheSlot.h" -#include "cachinglayer/LoadingOverheadTracker.h" +#include "cachinglayer/LoadingOverhead.h" #include "cachinglayer/TieredStorageConfig.h" #include "cachinglayer/Translator.h" #include "cachinglayer/Utils.h" @@ -43,6 +43,21 @@ class Manager { UpdateConfig(std::chrono::milliseconds loading_timeout, std::chrono::milliseconds warmup_loading_timeout, bool storage_usage_tracking_enabled, CacheWarmupPolicies warmup_policies); + // Creates one Group before it is referenced by any Translator binding. + static std::shared_ptr + CreateLoadingOverheadGroup(LoadingOverheadDimension dimension, LoadingOverheadPolicy policy); + + // Replaces the policy of an existing Group while preserving its runtime + // state. Calling contract for the authoritative owner of that Group: + // - Serialize all reconfigurations and build the policy from the latest + // configured Budget/TP limit in the same serialized section. + // - Expansion: update the Group before increasing the actual Budget/TP. + // Existing reservation reconciles on the next request Reserve. + // - Tightening: restrict the actual Budget/TP first, then update the Group + // with the desired policy. Existing reservation drains lazily with inflight work. + static LoadingOverheadUpdateResult + UpdateLoadingOverheadGroup(const std::shared_ptr& group, LoadingOverheadPolicy policy); + ~Manager(); Manager(const Manager&) = delete; @@ -146,7 +161,6 @@ class Manager { std::shared_ptr dlist_{nullptr}; std::shared_ptr prefetch_pool_{nullptr}; - std::shared_ptr loading_overhead_tracker_ = std::make_shared(); bool eviction_enabled_{false}; }; // class Manager diff --git a/include/cachinglayer/Translator.h b/include/cachinglayer/Translator.h index 8f74b15..7b1ed7c 100644 --- a/include/cachinglayer/Translator.h +++ b/include/cachinglayer/Translator.h @@ -42,22 +42,22 @@ struct Meta { // Whether the translator supports strategy based eviction. // Does not affect manual eviction. bool support_eviction; - // Loading overhead configuration for this translator. + // Loading-overhead configuration for this translator. // Each configured resource dimension is capped across CacheSlots sharing its group. // An omitted dimension passes through unchanged and remains subject to DList admission. - // If the whole config is not set, no capping is applied (existing behavior). - std::optional loading_overhead; + // If the config is not set, no capping is applied (existing behavior). + std::optional loading_overhead_config; std::optional metric_attribution; explicit Meta(StorageType storage_type, CellIdMappingMode cell_id_mapping_mode, CellDataType cell_data_type, CacheWarmupPolicy cache_warmup_policy, bool support_eviction, - std::optional loading_overhead = std::nullopt, + std::optional loading_overhead_config = std::nullopt, std::optional metric_attribution = std::nullopt) : storage_type(storage_type), cell_id_mapping_mode(cell_id_mapping_mode), cell_data_type(cell_data_type), cache_warmup_policy(cache_warmup_policy), support_eviction(support_eviction), - loading_overhead(std::move(loading_overhead)), + loading_overhead_config(std::move(loading_overhead_config)), metric_attribution(std::move(metric_attribution)) { } }; @@ -74,14 +74,14 @@ class Translator { // For resource reservation when a cell is about to be loaded. // Returns {loaded_usage, loading_overhead}: // - loaded_usage (first): the final resource usage after the cell is fully loaded and in cache. - // - loading_overhead (second): the *temporary* resource usage during loading (e.g., preprocessing buffers), - // excluding the final loaded usage. This is the extra overhead that only exists during the loading phase. - // When a loading_overhead dimension is configured in Meta, the total reservation across all CacheSlots - // sharing that dimension's group is capped at its upper bound. Omitted dimensions pass through unchanged. + // - loading_overhead (second): a conservative upper bound for the *temporary* resource usage during loading + // (e.g., preprocessing buffers), excluding the final loaded usage. For grouped dimensions it must cover the + // request's actual transient usage from successful DList reservation until the paired Release. + // When a loading_overhead dimension is configured in Meta, the total reservation across all CacheSlots sharing + // that dimension's group is governed by the Group policy. Omitted dimensions pass through unchanged. // If a cell is about to be pinned and loaded, and there are not enough resource for it, EvictionManager - // will try to evict some other cells to make space. Thus this estimation should generally be greater - // than or equal to the actual size. If the estimation is smaller than the actual size, with insufficient - // resource reserved, the load may fail. + // will try to evict some other cells to make space. Both estimates must be greater than or equal to the actual + // usage. Underestimation can break admission safety and may make the load fail. virtual std::pair estimated_byte_size_of_cell(cid_t cid) const = 0; // must be unique to identify a CacheSlot. diff --git a/include/cachinglayer/lrucache/DList.h b/include/cachinglayer/lrucache/DList.h index 2fc5cbb..b638ea5 100644 --- a/include/cachinglayer/lrucache/DList.h +++ b/include/cachinglayer/lrucache/DList.h @@ -17,14 +17,16 @@ #include #include +#include #include #include #include #include +#include #include #include -#include "cachinglayer/LoadingOverheadTracker.h" +#include "cachinglayer/LoadingOverhead.h" #include "cachinglayer/Metrics.h" #include "cachinglayer/Utils.h" #include "cachinglayer/lrucache/ListNode.h" @@ -33,6 +35,13 @@ namespace milvus::cachinglayer::internal { +struct LoadingResourceReservationResult { + // Explicit because a successful policy-derived reservation may be zero. + bool success{false}; + // Unscaled request reservation corresponding to the applied transition. + ResourceUsage reserved; +}; + class DList : public std::enable_shared_from_this { public: DList(bool eviction_enabled, ResourceUsage max_memory, ResourceUsage low_watermark, ResourceUsage high_watermark, @@ -68,35 +77,21 @@ class DList : public std::enable_shared_from_this { } } - // Must be called during initialization, before any Reserve/Release calls. - // Not thread-safe with concurrent Reserve/Release. void - SetLoadingOverheadTracker(std::shared_ptr tracker) { - loading_overhead_tracker_ = std::move(tracker); - } + BindLoadingOverheadGroups(const LoadingOverheadConfig& config); - uint64_t - RegisterLoadingOverhead(const LoadingOverheadConfig& config) { - if (loading_overhead_tracker_) { - return loading_overhead_tracker_->Register(config); - } - return LoadingOverheadTracker::kInvalidHandle; - } + void + UnbindLoadingOverheadGroups(const LoadingOverheadConfig& config); - uint64_t - RegisterLoadingOverhead(const std::string& group, const ResourceUsage& upper_bound) { - if (loading_overhead_tracker_) { - return loading_overhead_tracker_->Register(group, upper_bound); - } - return LoadingOverheadTracker::kInvalidHandle; - } + // Creates one Group independently before Translator bindings reference it. + std::shared_ptr + CreateLoadingOverheadGroup(LoadingOverheadDimension dimension, LoadingOverheadPolicy policy); - void - UnregisterLoadingOverhead(uint64_t overhead_handle) { - if (loading_overhead_tracker_) { - loading_overhead_tracker_->Unregister(overhead_handle); - } - } + // Replaces an existing Group policy. The next Reserve or Release reconciles + // its reservation with the new target. The caller must follow Manager's + // serialized owner and Budget/TP ordering contract. + LoadingOverheadUpdateResult + UpdateLoadingOverheadGroup(const std::shared_ptr& group, LoadingOverheadPolicy policy); ~DList() { // waiting requests should be cleared before event base thread is stopped @@ -136,7 +131,8 @@ class DList : public std::enable_shared_from_this { bool UpdateMaxLimit(const ResourceUsage& new_limit); - // Update low/high watermark does not trigger eviction, thus will not fail. + // Updating a watermark retries queued reservations and may evict cache + // entries while satisfying them. Validation failure is reported by exception. void UpdateLowWatermark(const ResourceUsage& new_low_watermark); @@ -158,19 +154,20 @@ class DList : public std::enable_shared_from_this { ReserveLoadingResourceWithTimeout(const ResourceUsage& size, std::chrono::milliseconds timeout, OpContext* ctx = nullptr); - // Reserve with loading overhead tracker integration. - // Space check uses (loaded + overhead) * factor as upper bound. - // Actual reservation uses (loaded + delta) * factor, where delta = tracker->Reserve() or overhead if no tracker. - // Returns the actual reserved size (zero = failure). - folly::SemiFuture + // Reserve with loading-overhead Group integration. + // Group-managed overhead is capped across the bound Group before the + // request total is scaled by loading_resource_factor. + // A successful reservation may reserve zero bytes, so success is explicit. + folly::SemiFuture ReserveLoadingResourceWithTimeout(const ResourceUsage& loaded, const ResourceUsage& overhead, - uint64_t overhead_handle, std::chrono::milliseconds timeout, + const LoadingOverheadConfig* config, std::chrono::milliseconds timeout, OpContext* ctx = nullptr); - // Release with loading overhead tracker integration. - // Returns the actual unscaled size released (loaded + tracker_delta). + // Release with loading-overhead Group integration. + // Returns the actual unscaled size released (loaded + Group delta). ResourceUsage - ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& overhead, uint64_t overhead_handle); + ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& overhead, + const LoadingOverheadConfig* config); // Release resource used for loading, called after loading a cell. void @@ -215,37 +212,36 @@ class DList : public std::enable_shared_from_this { // Waiting request for timeout-based memory reservation struct WaitingRequest { - ResourceUsage required_size; // loaded + overhead (for space check) - ResourceUsage loaded; // loaded portion (for tracker-aware path) - ResourceUsage overhead; // overhead portion (for tracker-aware path) - uint64_t overhead_handle{0}; + ResourceUsage required_size; // initial policy-derived scaled requirement used for queue ordering + ResourceUsage loaded; // loaded portion (for Group-aware path) + ResourceUsage overhead; // overhead portion (for Group-aware path) + // Queued requests own a copy so they never retain a pointer into Translator Meta. + std::optional loading_overhead_config; std::chrono::steady_clock::time_point deadline; folly::Promise bool_promise; - folly::Promise resource_promise; + folly::Promise resource_promise; bool use_resource_promise{false}; uint64_t request_id; std::optional cancel_cb{std::nullopt}; - // Legacy constructor (no tracker) + // Request-local constructor. WaitingRequest(ResourceUsage size, std::chrono::steady_clock::time_point dl, folly::Promise p, uint64_t id) : required_size(size), deadline(dl), bool_promise(std::move(p)), - resource_promise(folly::Promise::makeEmpty()), + resource_promise(folly::Promise::makeEmpty()), request_id(id) { } - // Tracker-aware constructor. - // required_size is scaled by loading_resource_factor to match the legacy path, - // ensuring consistent queue ordering between legacy and tracker-aware requests. - WaitingRequest(ResourceUsage loaded, ResourceUsage overhead, uint64_t overhead_handle, - std::chrono::steady_clock::time_point dl, folly::Promise p, uint64_t id, - float loading_resource_factor) - : required_size((loaded + overhead) * loading_resource_factor), + // Group-aware constructor. + WaitingRequest(ResourceUsage required_size, ResourceUsage loaded, ResourceUsage overhead, + const LoadingOverheadConfig* config, std::chrono::steady_clock::time_point dl, + folly::Promise p, uint64_t id) + : required_size(required_size), loaded(loaded), overhead(overhead), - overhead_handle(overhead_handle), + loading_overhead_config(config ? std::make_optional(*config) : std::nullopt), deadline(dl), bool_promise(folly::Promise::makeEmpty()), resource_promise(std::move(p)), @@ -253,16 +249,27 @@ class DList : public std::enable_shared_from_this { request_id(id) { } + const LoadingOverheadConfig* + loadingOverheadConfig() const { + return loading_overhead_config ? &loading_overhead_config.value() : nullptr; + } + void setValue(bool success, ResourceUsage actual = {}) { if (use_resource_promise) { - resource_promise.setValue(success ? actual : ResourceUsage{}); + resource_promise.setValue( + LoadingResourceReservationResult{success, success ? actual : ResourceUsage{}}); } else { bool_promise.setValue(success); } } }; + struct LoadingResourceReservationAttempt { + LoadingResourceReservationResult result; + ResourceUsage required_size; + }; + // Comparator for priority queue (smaller size and earlier deadline have higher priority) struct WaitingRequestComparator { bool @@ -272,8 +279,10 @@ class DList : public std::enable_shared_from_this { return a->deadline > b->deadline; } // Second priority: resource size (smaller size has higher priority) - int64_t total_a = a->required_size.memory_bytes + a->required_size.file_bytes; - int64_t total_b = b->required_size.memory_bytes + b->required_size.file_bytes; + const auto total_a = static_cast(std::max(a->required_size.memory_bytes, int64_t{0})) + + static_cast(std::max(a->required_size.file_bytes, int64_t{0})); + const auto total_b = static_cast(std::max(b->required_size.memory_bytes, int64_t{0})) + + static_cast(std::max(b->required_size.file_bytes, int64_t{0})); return total_a > total_b; } }; @@ -287,12 +296,25 @@ class DList : public std::enable_shared_from_this { std::pair reserveResourceInternalImpl(const ResourceUsage& size, std::function rollback); - // Reserve with tracker under lock. Space check uses loaded + overhead, - // actual reservation uses loaded + tracker delta. Returns actual reserved (zero = failed). - // Returns {success, unscaled_reserved}. Scaled amount is added to total_loading_size_ internally. - std::pair - reserveResourceInternalWithTracker(const ResourceUsage& loaded, const ResourceUsage& overhead, - uint64_t overhead_handle, LoadingOverheadTracker* tracker); + void + validateLoadingOverheadBinding(const std::optional& binding, + LoadingOverheadDimension dimension) const; + + ResourceUsage + reserveLoadingOverhead(const LoadingOverheadConfig& config, const ResourceUsage& overhead); + + void + rollbackLoadingOverhead(const LoadingOverheadConfig& config, const ResourceUsage& overhead, + const ResourceUsage& reserved) noexcept; + + ResourceUsage + releaseLoadingOverhead(const LoadingOverheadConfig& config, const ResourceUsage& overhead); + + // Reserve with Groups under lock using the factor-adjusted target transition. + // Returns the explicit result and the checked scaled requirement attempted. + LoadingResourceReservationAttempt + reserveResourceInternalWithOverhead(const ResourceUsage& loaded, const ResourceUsage& overhead, + const LoadingOverheadConfig* config); void evictionLoop(); @@ -312,6 +334,12 @@ class DList : public std::enable_shared_from_this { std::vector> handleWaitingRequests(); + // Fail one queued request and immediately retry requests behind it. + // Must be called with list_mtx_ held; returned requests are destroyed by + // the caller after releasing the lock. + std::vector> + failWaitingRequest(uint64_t request_id, const char* reason); + // Clear all waiting requests (used in destructor) void clearWaitingQueue(); @@ -356,8 +384,6 @@ class DList : public std::enable_shared_from_this { ListNode* head_ = nullptr; ListNode* tail_ = nullptr; - std::shared_ptr loading_overhead_tracker_; - // TODO(tiered storage 3): benchmark folly::DistributedMutex for this usecase. mutable std::mutex list_mtx_; std::atomic max_resource_limit_; diff --git a/src/cachinglayer/LoadingOverhead.cpp b/src/cachinglayer/LoadingOverhead.cpp new file mode 100644 index 0000000..d465318 --- /dev/null +++ b/src/cachinglayer/LoadingOverhead.cpp @@ -0,0 +1,139 @@ +// Copyright (C) 2019-2026 Zilliz. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software distributed under the License +// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express +// or implied. See the License for the specific language governing permissions and limitations under the License + +#include "cachinglayer/LoadingOverhead.h" + +#include "log/Log.h" + +namespace milvus::cachinglayer { + +namespace { + +const char* +DimensionName(LoadingOverheadDimension dimension) noexcept { + return dimension == LoadingOverheadDimension::kMemory ? "memory" : "file"; +} + +} // namespace + +void +LoadingOverheadGroup::validateBinding(const internal::DList* owner, LoadingOverheadDimension dimension, + const std::optional& max_runtime_unit) const { + if (owner_ != owner || dimension_ != dimension) { + throw std::invalid_argument("loading-overhead binding requires a Group from the matching dimension"); + } + if (max_runtime_unit.has_value() && max_runtime_unit.value() < 0) { + throw std::invalid_argument("loading-overhead binding runtime-unit bound must be non-negative"); + } + if (!max_runtime_unit.has_value() && policy_.RequiresRuntimeUnitBound()) { + throw std::invalid_argument("bounded loading-overhead Group binding requires max_runtime_unit"); + } +} + +void +LoadingOverheadGroup::bind(const std::optional& max_runtime_unit) { + if (max_runtime_unit.has_value()) { + runtime_unit_bounds_.insert(max_runtime_unit.value()); + max_runtime_unit_ = std::max(max_runtime_unit_, max_runtime_unit.value()); + } + ++binding_count_; + LOG_DEBUG("[MCL] LoadingOverheadGroup bound {} Group (bindings={})", DimensionName(dimension_), binding_count_); +} + +void +LoadingOverheadGroup::unbind(const internal::DList* owner, LoadingOverheadDimension dimension, + const std::optional& max_runtime_unit) noexcept { + if (owner_ != owner || dimension_ != dimension) { + LOG_ERROR("[MCL] LoadingOverheadGroup cannot unbind invalid {} Group", DimensionName(dimension)); + return; + } + + if (max_runtime_unit.has_value()) { + const auto unit_it = runtime_unit_bounds_.find(max_runtime_unit.value()); + if (unit_it == runtime_unit_bounds_.end()) { + LOG_ERROR("[MCL] LoadingOverheadGroup runtime-unit binding {} not found", max_runtime_unit.value()); + } else { + runtime_unit_bounds_.erase(unit_it); + max_runtime_unit_ = runtime_unit_bounds_.empty() ? int64_t{0} : *runtime_unit_bounds_.rbegin(); + } + } + + if (binding_count_ > 0) { + --binding_count_; + } + LOG_DEBUG("[MCL] LoadingOverheadGroup unbound {} Group (bindings={})", DimensionName(dimension_), binding_count_); + if (binding_count_ == 0 && (sum_of_overhead_ > 0 || overhead_reserved_ > 0)) { + LOG_ERROR( + "[MCL] LoadingOverheadGroup {} Group binding_count=0 with residual reservations: " + "sum_of_overhead={}, overhead_reserved={}", + DimensionName(dimension_), sum_of_overhead_, overhead_reserved_); + } +} + +LoadingOverheadUpdateResult +LoadingOverheadGroup::updatePolicy(const internal::DList* owner, LoadingOverheadPolicy policy) { + if (owner_ != owner) { + return LoadingOverheadUpdateResult::kInvalidArgument; + } + if (policy.RequiresRuntimeUnitBound() && hasMissingRuntimeUnit()) { + return LoadingOverheadUpdateResult::kIncompatiblePolicy; + } + policy_ = std::move(policy); + return LoadingOverheadUpdateResult::kApplied; +} + +int64_t +LoadingOverheadGroup::reserve(int64_t overhead) { + sum_of_overhead_ += overhead; + const auto delta = std::max(computeTarget() - overhead_reserved_, int64_t{0}); + overhead_reserved_ += delta; + return delta; +} + +void +LoadingOverheadGroup::rollbackReserve(int64_t overhead, int64_t reserved) noexcept { + sum_of_overhead_ -= overhead; + if (sum_of_overhead_ < 0) { + LOG_ERROR("[MCL] LoadingOverheadGroup Reserve rollback: sum_of_overhead < 0"); + sum_of_overhead_ = 0; + } + overhead_reserved_ -= reserved; + if (overhead_reserved_ < 0) { + LOG_ERROR("[MCL] LoadingOverheadGroup Reserve rollback: overhead_reserved < 0"); + overhead_reserved_ = 0; + } +} + +int64_t +LoadingOverheadGroup::release(int64_t overhead) noexcept { + sum_of_overhead_ -= overhead; + if (sum_of_overhead_ < 0) { + LOG_ERROR("[MCL] LoadingOverheadGroup Release: sum_of_overhead < 0"); + sum_of_overhead_ = 0; + } + + const auto delta = std::max(overhead_reserved_ - computeTarget(), int64_t{0}); + overhead_reserved_ -= delta; + return delta; +} + +int64_t +LoadingOverheadGroup::computeTarget() const noexcept { + const auto bound = policy_.ResolveBound(max_runtime_unit_); + return std::min(std::max(sum_of_overhead_, int64_t{0}), bound); +} + +bool +LoadingOverheadGroup::hasMissingRuntimeUnit() const noexcept { + return runtime_unit_bounds_.size() != binding_count_; +} + +} // namespace milvus::cachinglayer diff --git a/src/cachinglayer/Manager.cpp b/src/cachinglayer/Manager.cpp index 961e360..0ee5092 100644 --- a/src/cachinglayer/Manager.cpp +++ b/src/cachinglayer/Manager.cpp @@ -58,7 +58,6 @@ Manager::ConfigureTieredStorage(CacheWarmupPolicies warmup_policies, CacheLimit manager.dlist_ = std::make_shared(eviction_enabled, max, low_watermark, high_watermark, eviction_config); - manager.dlist_->SetLoadingOverheadTracker(manager.loading_overhead_tracker_); LOG_INFO( "[MCL] Configured Tiered Storage manager with " @@ -91,6 +90,24 @@ Manager::UpdateConfig(std::chrono::milliseconds loading_timeout, std::chrono::mi warmup_policies.ToString()); } +std::shared_ptr +Manager::CreateLoadingOverheadGroup(LoadingOverheadDimension dimension, LoadingOverheadPolicy policy) { + auto& manager = GetInstance(); + if (!manager.dlist_) { + return nullptr; + } + return manager.dlist_->CreateLoadingOverheadGroup(dimension, std::move(policy)); +} + +LoadingOverheadUpdateResult +Manager::UpdateLoadingOverheadGroup(const std::shared_ptr& group, LoadingOverheadPolicy policy) { + auto& manager = GetInstance(); + if (!manager.dlist_) { + return LoadingOverheadUpdateResult::kInvalidArgument; + } + return manager.dlist_->UpdateLoadingOverheadGroup(group, std::move(policy)); +} + size_t Manager::memory_overhead() const { // TODO(tiered storage 2): calculate memory overhead diff --git a/src/cachinglayer/Metrics.cpp b/src/cachinglayer/Metrics.cpp index 9293be0..57d0e3f 100644 --- a/src/cachinglayer/Metrics.cpp +++ b/src/cachinglayer/Metrics.cpp @@ -37,7 +37,8 @@ DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_LOCATION(internal_cache_low_watermark_bytes, DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_slot_count, "[cpp]cache slot count"); DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_count, "[cpp]cache cell count"); DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_loaded_bytes, "[cpp]cache loaded bytes"); -DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_loading_bytes, "[cpp]cache loading bytes"); +DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION( + internal_cache_loading_bytes, "[cpp]estimated resource bytes held by active cache load requests"); DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_loading_count, "[cpp]cache cell loading count"); DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_loaded_count, diff --git a/src/cachinglayer/lrucache/DList.cpp b/src/cachinglayer/lrucache/DList.cpp index ddbc814..bafcbf6 100644 --- a/src/cachinglayer/lrucache/DList.cpp +++ b/src/cachinglayer/lrucache/DList.cpp @@ -14,6 +14,7 @@ #include #include +#include #include #include @@ -35,35 +36,89 @@ ClampNonNegative(std::atomic& counter, LogFn&& log_fn) { } } -folly::SemiFuture +std::shared_ptr +DList::CreateLoadingOverheadGroup(LoadingOverheadDimension dimension, LoadingOverheadPolicy policy) { + if (dimension != LoadingOverheadDimension::kMemory && dimension != LoadingOverheadDimension::kFile) { + return nullptr; + } + return std::shared_ptr(new LoadingOverheadGroup(this, dimension, std::move(policy))); +} + +void +DList::BindLoadingOverheadGroups(const LoadingOverheadConfig& config) { + std::lock_guard lock(list_mtx_); + validateLoadingOverheadBinding(config.memory, LoadingOverheadDimension::kMemory); + validateLoadingOverheadBinding(config.file, LoadingOverheadDimension::kFile); + if (config.memory.has_value()) { + config.memory->group->bind(config.memory->max_runtime_unit); + } + if (config.file.has_value()) { + config.file->group->bind(config.file->max_runtime_unit); + } +} + +void +DList::UnbindLoadingOverheadGroups(const LoadingOverheadConfig& config) { + std::lock_guard lock(list_mtx_); + if (config.memory.has_value()) { + if (config.memory->group) { + config.memory->group->unbind(this, LoadingOverheadDimension::kMemory, config.memory->max_runtime_unit); + } else { + LOG_ERROR("[MCL] LoadingOverheadGroup cannot unbind invalid memory Group"); + } + } + if (config.file.has_value()) { + if (config.file->group) { + config.file->group->unbind(this, LoadingOverheadDimension::kFile, config.file->max_runtime_unit); + } else { + LOG_ERROR("[MCL] LoadingOverheadGroup cannot unbind invalid file Group"); + } + } +} + +LoadingOverheadUpdateResult +DList::UpdateLoadingOverheadGroup(const std::shared_ptr& group, LoadingOverheadPolicy policy) { + std::vector> to_destroy; + LoadingOverheadUpdateResult result; + { + std::unique_lock lock(list_mtx_); + result = group ? group->updatePolicy(this, std::move(policy)) : LoadingOverheadUpdateResult::kInvalidArgument; + if (result == LoadingOverheadUpdateResult::kApplied) { + to_destroy = handleWaitingRequests(); + } + } + return result; +} + +folly::SemiFuture DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& loaded, const ResourceUsage& overhead, - uint64_t overhead_handle, std::chrono::milliseconds timeout, OpContext* ctx) { - // Quick reject: if even loaded alone (minimum possible) exceeds capacity, fail fast. - auto min_possible = loaded * eviction_config_.loading_resource_factor; + const LoadingOverheadConfig* config, std::chrono::milliseconds timeout, + OpContext* ctx) { std::unique_lock lock(list_mtx_); - if (!max_resource_limit_.load().CanHold(min_possible)) { - LOG_ERROR("[MCL] Failed to reserve loaded={} as it exceeds max_memory_={}.", loaded.ToString(), - max_resource_limit_.load().ToString()); - return folly::makeSemiFuture(ResourceUsage{}); + auto attempt = reserveResourceInternalWithOverhead(loaded, overhead, config); + if (attempt.result.success) { + return folly::makeSemiFuture(attempt.result); } - auto [success, actual] = - reserveResourceInternalWithTracker(loaded, overhead, overhead_handle, loading_overhead_tracker_.get()); - if (success) { - return folly::makeSemiFuture(actual); + + if (!max_resource_limit_.load().CanHold(attempt.required_size)) { + LOG_ERROR( + "[MCL] Failed to reserve Group-aware loading resource because the policy-derived requirement={} " + "exceeds capacity={}", + attempt.required_size.ToString(), max_resource_limit_.load().ToString()); + return folly::makeSemiFuture(LoadingResourceReservationResult{}); } if (timeout.count() == 0) { - return folly::makeSemiFuture(ResourceUsage{}); + return folly::makeSemiFuture(LoadingResourceReservationResult{}); } auto deadline = timeout.count() > 0 ? std::chrono::steady_clock::now() + timeout : std::chrono::steady_clock::time_point::max(); - auto [promise, future] = folly::makePromiseContract(); + auto [promise, future] = folly::makePromiseContract(); uint64_t request_id = next_request_id_.fetch_add(1); - auto waiting_request = - std::make_unique(loaded, overhead, overhead_handle, deadline, std::move(promise), request_id, - eviction_config_.loading_resource_factor); + auto waiting_request = std::make_unique(attempt.required_size, loaded, overhead, config, deadline, + std::move(promise), request_id); WaitingRequest* request_ptr = waiting_request.get(); waiting_requests_map_[request_id] = request_ptr; waiting_queue_.push(std::move(waiting_request)); @@ -76,19 +131,19 @@ DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& loaded, const Reso event_base_thread_->getEventBase()->runInEventBaseThread([weak_self, request_id, timeout]() { auto self = weak_self.lock(); - if (!self) + if (!self) { return; + } self->event_base_thread_->getEventBase()->runAfterDelay( [weak_self, request_id]() { auto self = weak_self.lock(); - if (!self) + if (!self) { return; - std::unique_lock lock(self->list_mtx_); - auto it = self->waiting_requests_map_.find(request_id); - if (it != self->waiting_requests_map_.end()) { - LOG_WARN("[MCL] Reserve Request {} timed out.", request_id); - it->second->setValue(false); - self->waiting_requests_map_.erase(it); + } + std::vector> to_destroy; + { + std::unique_lock lock(self->list_mtx_); + to_destroy = self->failWaitingRequest(request_id, "timed out"); } }, static_cast(timeout.count())); @@ -98,19 +153,19 @@ DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& loaded, const Reso if (ctx && ctx->cancellation_token.canBeCancelled()) { request_ptr->cancel_cb.emplace(ctx->cancellation_token, [weak_self, request_id]() { auto self = weak_self.lock(); - if (!self) + if (!self) { return; + } self->event_base_thread_->getEventBase()->runInEventBaseThread([weak_self, request_id]() { auto self = weak_self.lock(); - if (!self) - return; - std::unique_lock lock(self->list_mtx_); - auto it = self->waiting_requests_map_.find(request_id); - if (it == self->waiting_requests_map_.end()) + if (!self) { return; - LOG_WARN("[MCL] Request {} cancelled.", request_id); - it->second->setValue(false); - self->waiting_requests_map_.erase(it); + } + std::vector> to_destroy; + { + std::unique_lock lock(self->list_mtx_); + to_destroy = self->failWaitingRequest(request_id, "cancelled"); + } }); }); } @@ -123,7 +178,7 @@ DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& original_size, std OpContext* ctx) { // NOTE: we can reserve more loading resources than the original request size by adjusting the // loading_resource_factor to avoid potential problems from bad resource estimation. - auto size = original_size * eviction_config_.loading_resource_factor; + const auto size = original_size * eviction_config_.loading_resource_factor; // Try immediate reservation; if it fails, enqueue atomically under the same lock // to avoid a race window where resources could be released and notified between @@ -173,15 +228,10 @@ DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& original_size, std if (!self) { return; // DList already destroyed } - std::unique_lock lock(self->list_mtx_); - auto it = self->waiting_requests_map_.find(request_id); - if (it != self->waiting_requests_map_.end()) { - LOG_WARN( - "[MCL] Reserve Request {} of size {} timed out, " - "notifying failure.", - request_id, it->second->required_size.ToString()); - it->second->setValue(false); - self->waiting_requests_map_.erase(it); + std::vector> to_destroy; + { + std::unique_lock lock(self->list_mtx_); + to_destroy = self->failWaitingRequest(request_id, "timed out"); } }, static_cast(timeout.count())); @@ -202,14 +252,11 @@ DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& original_size, std if (!self) { return; // DList already destroyed } - std::unique_lock lock(self->list_mtx_); - auto it = self->waiting_requests_map_.find(request_id); - if (it == self->waiting_requests_map_.end()) { - return; + std::vector> to_destroy; + { + std::unique_lock lock(self->list_mtx_); + to_destroy = self->failWaitingRequest(request_id, "cancelled"); } - LOG_WARN("[MCL] Request {} cancelled, notifying failure.", request_id); - it->second->setValue(false); - self->waiting_requests_map_.erase(it); }); }); } @@ -224,41 +271,102 @@ DList::reserveResourceInternal(const ResourceUsage& size) { return success; } -std::pair -DList::reserveResourceInternalWithTracker(const ResourceUsage& loaded, const ResourceUsage& overhead, - uint64_t overhead_handle, LoadingOverheadTracker* tracker) { - // Compute tracker delta first so space check uses the actual amount, not the uncapped overhead. - // This avoids rejecting loads that would fit after tracker capping (e.g., delta=0 when group is saturated). +void +DList::validateLoadingOverheadBinding(const std::optional& binding, + LoadingOverheadDimension dimension) const { + if (!binding.has_value()) { + return; + } + if (!binding->group) { + throw std::invalid_argument("loading-overhead binding requires a Group from the matching dimension"); + } + binding->group->validateBinding(this, dimension, binding->max_runtime_unit); +} + +ResourceUsage +DList::reserveLoadingOverhead(const LoadingOverheadConfig& config, const ResourceUsage& overhead) { + if (!overhead.AllGEZero()) { + throw std::invalid_argument("loading overhead must be non-negative"); + } + + validateLoadingOverheadBinding(config.memory, LoadingOverheadDimension::kMemory); + validateLoadingOverheadBinding(config.file, LoadingOverheadDimension::kFile); + + auto delta = overhead; + if (config.memory.has_value()) { + delta.memory_bytes = config.memory->group->reserve(overhead.memory_bytes); + } + if (config.file.has_value()) { + delta.file_bytes = config.file->group->reserve(overhead.file_bytes); + } + return delta; +} + +void +DList::rollbackLoadingOverhead(const LoadingOverheadConfig& config, const ResourceUsage& overhead, + const ResourceUsage& reserved) noexcept { + if (config.memory.has_value() && config.memory->group) { + config.memory->group->rollbackReserve(overhead.memory_bytes, reserved.memory_bytes); + } + if (config.file.has_value() && config.file->group) { + config.file->group->rollbackReserve(overhead.file_bytes, reserved.file_bytes); + } +} + +ResourceUsage +DList::releaseLoadingOverhead(const LoadingOverheadConfig& config, const ResourceUsage& overhead) { + validateLoadingOverheadBinding(config.memory, LoadingOverheadDimension::kMemory); + validateLoadingOverheadBinding(config.file, LoadingOverheadDimension::kFile); + auto delta = overhead; - if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && tracker != nullptr) { - delta = tracker->Reserve(overhead_handle, overhead); + if (config.memory.has_value()) { + delta.memory_bytes = config.memory->group->release(overhead.memory_bytes); + } + if (config.file.has_value()) { + delta.file_bytes = config.file->group->release(overhead.file_bytes); } - auto actual_size = (loaded + delta) * eviction_config_.loading_resource_factor; + return delta; +} - auto rollback = [overhead_handle, overhead, tracker]() { - if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && tracker != nullptr) { - tracker->Release(overhead_handle, overhead); +DList::LoadingResourceReservationAttempt +DList::reserveResourceInternalWithOverhead(const ResourceUsage& loaded, const ResourceUsage& overhead, + const LoadingOverheadConfig* config) { + const auto delta = config != nullptr ? reserveLoadingOverhead(*config, overhead) : overhead; + + auto rollback = [this, config, overhead, delta]() { + if (config != nullptr) { + rollbackLoadingOverhead(*config, overhead, delta); } }; - auto [success, _] = reserveResourceInternalImpl(actual_size, rollback); + const auto unscaled = loaded + delta; + const auto scaled = unscaled * eviction_config_.loading_resource_factor; + + if (!max_resource_limit_.load().CanHold(scaled)) { + rollback(); + return {.result = LoadingResourceReservationResult{}, .required_size = scaled}; + } + + auto [success, _] = reserveResourceInternalImpl(scaled, rollback); if (!success) { - return {false, {}}; + return {.result = LoadingResourceReservationResult{}, .required_size = scaled}; } - auto unscaled = loaded + delta; - LOG_TRACE("[MCL] reserve with tracker: loaded={}, overhead={}, delta={}, unscaled={}, scaled={}, total_loading={}", - loaded.ToString(), overhead.ToString(), delta.ToString(), unscaled.ToString(), actual_size.ToString(), - total_loading_size_.load().ToString()); - return {true, unscaled}; + LOG_TRACE( + "[MCL] reserve with loading-overhead Groups: loaded={}, overhead={}, delta={}, unscaled={}, scaled={}, " + "total_loading={}", + loaded.ToString(), overhead.ToString(), delta.ToString(), unscaled.ToString(), scaled.ToString(), + total_loading_size_.load().ToString()); + return {.result = LoadingResourceReservationResult{.success = true, .reserved = unscaled}, .required_size = scaled}; } std::pair DList::reserveResourceInternalImpl(const ResourceUsage& size, std::function rollback) { auto using_resources = total_loaded_size_.load() + total_loading_size_.load(); + auto required_resources = using_resources + size; // Combined logical and physical memory limit check - bool logical_limit_exceeded = !max_resource_limit_.load().CanHold(using_resources + size); + bool logical_limit_exceeded = !max_resource_limit_.load().CanHold(required_resources); auto physical_eviction_needed = checkPhysicalMemoryLimit(size); // If either limit is exceeded, attempt unified eviction @@ -270,8 +378,8 @@ DList::reserveResourceInternalImpl(const ResourceUsage& size, std::function list_lock(list_mtx_); - auto using_resources = total_loaded_size_.load() + total_loading_size_.load(); - if (!new_limit.CanHold(using_resources)) { - // positive means amount owed - auto deficit = using_resources - new_limit; - // deficit is the hard limit of eviction, if we cannot evict deficit, we give - // up the limit change. - if (!tryEvict(deficit, deficit).AnyGTZero()) { - return false; + std::vector> to_destroy; + { + std::unique_lock list_lock(list_mtx_); + auto using_resources = total_loaded_size_.load() + total_loading_size_.load(); + if (!new_limit.CanHold(using_resources)) { + // positive means amount owed + auto deficit = using_resources - new_limit; + // deficit is the hard limit of eviction, if we cannot evict deficit, we give + // up the limit change. + if (!tryEvict(deficit, deficit).AnyGTZero()) { + return false; + } } + LOG_INFO("[MCL] UpdateMaxLimit: from {} to {}", max_resource_limit_.load().ToString(), new_limit.ToString()); + max_resource_limit_ = new_limit; + cachinglayer::monitor::cache_capacity_bytes(StorageType::MEMORY).Set(max_resource_limit_.load().memory_bytes); + cachinglayer::monitor::cache_capacity_bytes(StorageType::DISK).Set(max_resource_limit_.load().file_bytes); + to_destroy = handleWaitingRequests(); } - LOG_INFO("[MCL] UpdateMaxLimit: from {} to {}", max_resource_limit_.load().ToString(), new_limit.ToString()); - max_resource_limit_ = new_limit; - cachinglayer::monitor::cache_capacity_bytes(StorageType::MEMORY).Set(max_resource_limit_.load().memory_bytes); - cachinglayer::monitor::cache_capacity_bytes(StorageType::DISK).Set(max_resource_limit_.load().file_bytes); return true; } void DList::UpdateLowWatermark(const ResourceUsage& new_low_watermark) { - std::unique_lock list_lock(list_mtx_); - AssertInfo(new_low_watermark.AllGEZero(), - "[MCL] low watermark must be greater than or " - "equal to 0. new_low_watermark: {}", - new_low_watermark.ToString()); - AssertInfo((high_watermark_.load() - new_low_watermark).AllGEZero(), - "[MCL] low watermark must be less than or equal to high " - "watermark. new_low_watermark: {}, high_watermark: {}", - new_low_watermark.ToString(), high_watermark_.load().ToString()); - LOG_INFO("[MCL] UpdateLowWatermark: from {} to {}", low_watermark_.load().ToString(), new_low_watermark.ToString()); - low_watermark_ = new_low_watermark; - cachinglayer::monitor::cache_low_watermark_bytes(StorageType::MEMORY).Set(low_watermark_.load().memory_bytes); - cachinglayer::monitor::cache_low_watermark_bytes(StorageType::DISK).Set(low_watermark_.load().file_bytes); + std::vector> to_destroy; + { + std::unique_lock list_lock(list_mtx_); + AssertInfo(new_low_watermark.AllGEZero(), + "[MCL] low watermark must be greater than or " + "equal to 0. new_low_watermark: {}", + new_low_watermark.ToString()); + AssertInfo((high_watermark_.load() - new_low_watermark).AllGEZero(), + "[MCL] low watermark must be less than or equal to high " + "watermark. new_low_watermark: {}, high_watermark: {}", + new_low_watermark.ToString(), high_watermark_.load().ToString()); + LOG_INFO("[MCL] UpdateLowWatermark: from {} to {}", low_watermark_.load().ToString(), + new_low_watermark.ToString()); + low_watermark_ = new_low_watermark; + cachinglayer::monitor::cache_low_watermark_bytes(StorageType::MEMORY).Set(low_watermark_.load().memory_bytes); + cachinglayer::monitor::cache_low_watermark_bytes(StorageType::DISK).Set(low_watermark_.load().file_bytes); + to_destroy = handleWaitingRequests(); + } } void DList::UpdateHighWatermark(const ResourceUsage& new_high_watermark) { - std::unique_lock list_lock(list_mtx_); - AssertInfo((new_high_watermark - low_watermark_.load()).AllGEZero(), - "[MCL] high watermark must be greater than or " - "equal to low watermark. new_high_watermark: {}, low_watermark: {}", - new_high_watermark.ToString(), low_watermark_.load().ToString()); - AssertInfo((max_resource_limit_.load() - new_high_watermark).AllGEZero(), - "[MCL] high watermark must be less than or equal to max " - "resource limit. new_high_watermark: {}, max_resource_limit: {}", - new_high_watermark.ToString(), max_resource_limit_.load().ToString()); - LOG_INFO("[MCL] UpdateHighWatermark: from {} to {}", high_watermark_.load().ToString(), - new_high_watermark.ToString()); - high_watermark_ = new_high_watermark; - cachinglayer::monitor::cache_high_watermark_bytes(StorageType::MEMORY).Set(high_watermark_.load().memory_bytes); - cachinglayer::monitor::cache_high_watermark_bytes(StorageType::DISK).Set(high_watermark_.load().file_bytes); + std::vector> to_destroy; + { + std::unique_lock list_lock(list_mtx_); + AssertInfo((new_high_watermark - low_watermark_.load()).AllGEZero(), + "[MCL] high watermark must be greater than or " + "equal to low watermark. new_high_watermark: {}, low_watermark: {}", + new_high_watermark.ToString(), low_watermark_.load().ToString()); + AssertInfo((max_resource_limit_.load() - new_high_watermark).AllGEZero(), + "[MCL] high watermark must be less than or equal to max " + "resource limit. new_high_watermark: {}, max_resource_limit: {}", + new_high_watermark.ToString(), max_resource_limit_.load().ToString()); + LOG_INFO("[MCL] UpdateHighWatermark: from {} to {}", high_watermark_.load().ToString(), + new_high_watermark.ToString()); + high_watermark_ = new_high_watermark; + cachinglayer::monitor::cache_high_watermark_bytes(StorageType::MEMORY).Set(high_watermark_.load().memory_bytes); + cachinglayer::monitor::cache_high_watermark_bytes(StorageType::DISK).Set(high_watermark_.load().file_bytes); + to_destroy = handleWaitingRequests(); + } } ResourceUsage -DList::ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& overhead, uint64_t overhead_handle) { +DList::ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& overhead, + const LoadingOverheadConfig* config) { std::vector> to_destroy; - ResourceUsage unscaled{}; + auto delta = overhead; + ResourceUsage unscaled; { std::unique_lock lock(list_mtx_); - auto delta = overhead; - if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && loading_overhead_tracker_) { - delta = loading_overhead_tracker_->Release(overhead_handle, overhead); + if (config != nullptr) { + delta = releaseLoadingOverhead(*config, overhead); } unscaled = loaded + delta; - auto actual = unscaled * eviction_config_.loading_resource_factor; - total_loading_size_ -= actual; + total_loading_size_ -= unscaled * eviction_config_.loading_resource_factor; ClampNonNegative(total_loading_size_, [&](const ResourceUsage& curr) { - LOG_ERROR( - "[MCL] total_loading_size_ negative after tracker release: loaded={}, overhead={}, delta={}, " - "current={}", - loaded.ToString(), overhead.ToString(), delta.ToString(), curr.ToString()); + LOG_ERROR("[MCL] total_loading_size_ negative after Group release: loaded={}, overhead={}, current={}", + loaded.ToString(), overhead.ToString(), curr.ToString()); }); to_destroy = handleWaitingRequests(); } @@ -640,13 +760,13 @@ DList::ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& void DList::ReleaseLoadingResource(const ResourceUsage& loading_size) { - auto size = loading_size * eviction_config_.loading_resource_factor; - total_loading_size_ -= size; + const auto scaled_size = loading_size * eviction_config_.loading_resource_factor; + total_loading_size_ -= scaled_size; ClampNonNegative(total_loading_size_, [&](const ResourceUsage& curr) { LOG_ERROR( "[MCL] total_loading_size_ became negative after release: release_scaled={}, original_release={}, " "loading_resource_factor={}, current_total_loading={}", - size.ToString(), loading_size.ToString(), eviction_config_.loading_resource_factor, curr.ToString()); + scaled_size.ToString(), loading_size.ToString(), eviction_config_.loading_resource_factor, curr.ToString()); }); // Notify waiting requests that resources are available std::vector> to_destroy; @@ -820,16 +940,17 @@ DList::handleWaitingRequests() { continue; } - // Try to fulfill: tracker-aware path computes delta under lock, - // legacy path uses required_size directly. + // Try to fulfill: Group-aware path computes delta under lock; + // request-local path uses required_size directly. bool fulfilled = false; ResourceUsage actual{}; + auto attempted_requirement = request_ptr_ref->required_size; if (request_ptr_ref->use_resource_promise) { - auto [ok, unscaled] = - reserveResourceInternalWithTracker(request_ptr_ref->loaded, request_ptr_ref->overhead, - request_ptr_ref->overhead_handle, loading_overhead_tracker_.get()); - fulfilled = ok; - actual = unscaled; + auto attempt = reserveResourceInternalWithOverhead(request_ptr_ref->loaded, request_ptr_ref->overhead, + request_ptr_ref->loadingOverheadConfig()); + fulfilled = attempt.result.success; + actual = attempt.result.reserved; + attempted_requirement = attempt.required_size; } else { if (reserveResourceInternal(request_ptr_ref->required_size)) { fulfilled = true; @@ -844,28 +965,29 @@ DList::handleWaitingRequests() { request->setValue(true, actual); } else { // Request was already handled by timeout/cancel, rollback. - // Legacy: actual is already scaled. Tracker-aware: actual is unscaled, needs scaling. LOG_WARN("[MCL] Request {} was already handled by timeout/cancel, rolling back.", request->request_id); - auto rollback = - request->use_resource_promise ? actual * eviction_config_.loading_resource_factor : actual; - total_loading_size_ -= rollback; - if (request->overhead_handle != LoadingOverheadTracker::kInvalidHandle && loading_overhead_tracker_) { - loading_overhead_tracker_->Release(request->overhead_handle, request->overhead); + if (request->use_resource_promise) { + auto delta = request->overhead; + if (const auto* config = request->loadingOverheadConfig(); config != nullptr) { + delta = releaseLoadingOverhead(*config, request->overhead); + } + total_loading_size_ -= (request->loaded + delta) * eviction_config_.loading_resource_factor; + } else { + total_loading_size_ -= actual; } } requests_to_destroy.push_back(std::move(request)); waiting_queue_.pop(); } else { - // Check if this request is permanently impossible (required size exceeds capacity). - // Use required_size for both paths: (loaded + overhead) * factor for tracker-aware, - // which is the upper bound of what the request could need. - if (!max_resource_limit_.load().CanHold(request_ptr_ref->required_size)) { + const auto required_size = + request_ptr_ref->use_resource_promise ? attempted_requirement : request_ptr_ref->required_size; + if (!max_resource_limit_.load().CanHold(required_size)) { auto request = std::move(request_ptr_ref); if (waiting_requests_map_.erase(request->request_id) > 0) { LOG_WARN( - "[MCL] Request {} is permanently impossible (required_size={} > capacity={}), " + "[MCL] Request {} is permanently impossible (required_size={} capacity={}), " "failing immediately.", - request->request_id, request->required_size.ToString(), max_resource_limit_.load().ToString()); + request->request_id, required_size.ToString(), max_resource_limit_.load().ToString()); request->setValue(false); } requests_to_destroy.push_back(std::move(request)); @@ -873,7 +995,7 @@ DList::handleWaitingRequests() { continue; } LOG_DEBUG("[MCL] Request {} of size {} cannot be satisfied, breaking.", request_ptr_ref->request_id, - request_ptr_ref->required_size.ToString()); + required_size.ToString()); // Cannot satisfy right now but may succeed later. // The queue is ordered by deadline, so stop here. break; @@ -883,6 +1005,18 @@ DList::handleWaitingRequests() { return requests_to_destroy; } +std::vector> +DList::failWaitingRequest(uint64_t request_id, const char* reason) { + auto it = waiting_requests_map_.find(request_id); + if (it == waiting_requests_map_.end()) { + return {}; + } + LOG_WARN("[MCL] Waiting request {} {}.", request_id, reason); + it->second->setValue(false); + waiting_requests_map_.erase(it); + return handleWaitingRequests(); +} + void DList::clearWaitingQueue() { // Move requests out while holding the lock, then destroy them outside the lock diff --git a/test/test_cachinglayer/CMakeLists.txt b/test/test_cachinglayer/CMakeLists.txt index 7ce6c7c..bf0a9fd 100644 --- a/test/test_cachinglayer/CMakeLists.txt +++ b/test/test_cachinglayer/CMakeLists.txt @@ -17,7 +17,8 @@ set(CACHINGLAYER_TEST_FILES test_dlist.cpp test_cache_slot.cpp test_tiered_storage_config.cpp - test_loading_overhead_tracker.cpp + test_loading_overhead_policy.cpp + test_loading_overhead_group.cpp test_utils.cpp ) diff --git a/test/test_cachinglayer/cachinglayer_test_utils.h b/test/test_cachinglayer/cachinglayer_test_utils.h index aa717f5..feeb961 100644 --- a/test/test_cachinglayer/cachinglayer_test_utils.h +++ b/test/test_cachinglayer/cachinglayer_test_utils.h @@ -98,6 +98,11 @@ class DListTestFriend { } } + static std::unique_lock + test_lock_list(DList* dlist) { + return std::unique_lock(dlist->list_mtx_); + } + // nodes are from tail to head static void verify_list(DList* dlist, std::vector nodes) { diff --git a/test/test_cachinglayer/test_cache_slot.cpp b/test/test_cachinglayer/test_cache_slot.cpp index 268f8b0..216decb 100644 --- a/test/test_cachinglayer/test_cache_slot.cpp +++ b/test/test_cachinglayer/test_cache_slot.cpp @@ -143,6 +143,10 @@ class MockTranslator : public Translator { requested_cids_.push_back(cids); } + if (load_start_callback_) { + load_start_callback_(); + } + if (load_should_throw_) { throw std::runtime_error("Simulated load error"); } @@ -189,13 +193,12 @@ class MockTranslator : public Translator { loading_overhead_ = loading_overhead; } void - SetLoadingOverheadConfig(const std::string& group, const ResourceUsage& upper_bound) { - meta_.loading_overhead = LoadingOverheadConfig{LoadingOverheadDimensionConfig{upper_bound.memory_bytes, group}, - LoadingOverheadDimensionConfig{upper_bound.file_bytes, group}}; + SetLoadingOverheadConfig(LoadingOverheadConfig config) { + meta_.loading_overhead_config = std::move(config); } void - SetLoadingOverheadConfig(LoadingOverheadConfig config) { - meta_.loading_overhead = std::move(config); + SetLoadStartCallback(std::function callback) { + load_start_callback_ = std::move(callback); } int GetCellsCallCount() const { @@ -223,6 +226,7 @@ class MockTranslator : public Translator { Meta meta_; std::unordered_map cid_load_delay_ms_; + std::function load_start_callback_; bool load_should_throw_ = false; int cells_storage_bytes_throw_on_cid_ = -1; // -1 means no throw std::unordered_map> extra_cids_; @@ -2148,23 +2152,24 @@ TEST(WarmupTimeoutTest, SyncWarmupBestEffortResourceAvailable) { } } -// Test that CacheSlot correctly integrates with LoadingOverheadTracker when +// Test that CacheSlot correctly integrates with a loading-overhead Group when // the translator reports non-zero loading overhead. -TEST(CacheSlotTrackerTest, LoadingOverheadTrackerIntegration) { +TEST(CacheSlotLoadingOverheadTest, GroupIntegration) { ResourceUsage limit{10000, 0}; auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); - auto tracker = std::make_shared(); - dlist->SetLoadingOverheadTracker(tracker); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Fixed(500)); + ASSERT_NE(group, nullptr); const int64_t cell_loaded_size = 100; const int64_t cell_loading_overhead = 200; auto translator = std::make_unique( std::vector>{{0, cell_loaded_size}, {1, cell_loaded_size}}, - std::unordered_map{{0, 0}, {1, 1}}, "test_tracker_integration", StorageType::MEMORY); + std::unordered_map{{0, 0}, {1, 1}}, "test_group_integration", StorageType::MEMORY); translator->SetLoadingOverhead({cell_loading_overhead, 0}); - translator->SetLoadingOverheadConfig("test_group", {500, 0}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{LoadingOverheadGroupBinding{group}, std::nullopt}); auto* translator_ptr = translator.get(); auto cache_slot = @@ -2173,12 +2178,12 @@ TEST(CacheSlotTrackerTest, LoadingOverheadTrackerIntegration) { auto op_ctx = std::make_unique(); - // Pin cell 0: should reserve loaded(100) + overhead delta from tracker + // Pin cell 0: should reserve loaded(100) + overhead delta from the Group. auto accessor = cache_slot->PinCellsDirect(op_ctx.get(), {0}); ASSERT_NE(accessor, nullptr); EXPECT_EQ(accessor->get_cell_of(0)->data, 0); - // Pin cell 1: should also go through tracker + // Pin cell 1: should also go through the Group. auto accessor2 = cache_slot->PinCellsDirect(op_ctx.get(), {1}); ASSERT_NE(accessor2, nullptr); EXPECT_EQ(accessor2->get_cell_of(1)->data, 10); @@ -2186,17 +2191,156 @@ TEST(CacheSlotTrackerTest, LoadingOverheadTrackerIntegration) { EXPECT_EQ(translator_ptr->GetCellsCallCount(), 2); } -TEST(CacheSlotTrackerTest, PassthroughFileOverheadParticipatesInAdmission) { +TEST(CacheSlotLoadingOverheadTest, LoadingMetricReturnsToBaselineAfterGroupReconfiguration) { + ResourceUsage limit{10000, 0}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Passthrough()); + ASSERT_NE(group, nullptr); + + auto translator = std::make_unique(std::vector>{{0, 100}}, + std::unordered_map{{0, 0}}, + "test_loading_metric_policy_update", StorageType::MEMORY); + translator->SetLoadingOverhead({200, 0}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{group, 200}, + std::nullopt, + }); + + std::promise load_started; + std::promise finish_load; + auto finish_load_future = finish_load.get_future().share(); + translator->SetLoadStartCallback([&]() { + load_started.set_value(); + finish_load_future.wait(); + }); + + auto cache_slot = + std::make_shared>(std::move(translator), dlist.get(), true, true, false, + std::chrono::milliseconds(5000), std::chrono::milliseconds(0)); + auto& loading_bytes = monitor::cache_loading_bytes(CellDataType::OTHER, StorageType::MEMORY); + const auto baseline = loading_bytes.Value(); + + auto load = std::async(std::launch::async, [&]() { + auto op_ctx = std::make_unique(); + return cache_slot->PinCellsDirect(op_ctx.get(), {0}); + }); + load_started.get_future().wait(); + EXPECT_EQ(loading_bytes.Value(), baseline + 300); + + EXPECT_EQ(dlist->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(0)), + LoadingOverheadUpdateResult::kApplied); + + finish_load.set_value(); + EXPECT_NE(load.get(), nullptr); + EXPECT_EQ(loading_bytes.Value(), baseline); +} + +TEST(CacheSlotLoadingOverheadTest, InvalidGroupBindingDoesNotLeakSlotMetrics) { + ResourceUsage limit{100, 0}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto& slot_count = monitor::cache_slot_count(CellDataType::OTHER, StorageType::MEMORY); + auto& cell_count = monitor::cache_cell_count(CellDataType::OTHER, StorageType::MEMORY); + const auto slot_baseline = slot_count.Value(); + const auto cell_baseline = cell_count.Value(); + + auto translator = std::make_unique(std::vector>{{0, 1}}, + std::unordered_map{{0, 0}}, + "test_binding_metric_rollback", StorageType::MEMORY); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{}, + std::nullopt, + }); + + EXPECT_THROW(std::make_shared>(std::move(translator), dlist.get(), true, true, false, + std::chrono::milliseconds(5000), std::chrono::milliseconds(0)), + std::invalid_argument); + EXPECT_EQ(slot_count.Value(), slot_baseline); + EXPECT_EQ(cell_count.Value(), cell_baseline); + + // Keep the global gauges isolated even when running against the buggy implementation. + slot_count.Decrement(slot_count.Value() - slot_baseline); + cell_count.Decrement(cell_count.Value() - cell_baseline); +} + +TEST(CacheSlotLoadingOverheadTest, CacheSlotUnbindUsesOriginalConfigMetadata) { + ResourceUsage limit{1000, 0}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(group, nullptr); + + auto translator = std::make_unique(std::vector>{{0, 1}}, + std::unordered_map{{0, 0}}, + "test_binding_snapshot", StorageType::MEMORY); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{group, 100}, + std::nullopt, + }); + auto cache_slot = + std::make_shared>(std::move(translator), dlist.get(), true, true, false, + std::chrono::milliseconds(5000), std::chrono::milliseconds(0)); + + LoadingOverheadConfig other_binding{ + LoadingOverheadGroupBinding{group, 50}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(other_binding); + + cache_slot->meta()->loading_overhead_config->memory->max_runtime_unit = 50; + cache_slot.reset(); + + auto reservation = + std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{500, 0}, &other_binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(reservation.success); + EXPECT_EQ(reservation.reserved, (ResourceUsage{50, 0})); + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{500, 0}, &other_binding), + (ResourceUsage{50, 0})); + dlist->UnbindLoadingOverheadGroups(other_binding); +} + +TEST(CacheSlotLoadingOverheadTest, ZeroByteReservationIsSuccessful) { + ResourceUsage limit{100, 0}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Executor(0)); + ASSERT_NE(group, nullptr); + + auto translator = std::make_unique(std::vector>{{0, 0}}, + std::unordered_map{{0, 0}}, + "test_zero_byte_reservation", StorageType::MEMORY); + translator->SetLoadingOverhead({100, 0}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{group, 100}, + std::nullopt, + }); + auto* translator_ptr = translator.get(); + + auto cache_slot = + std::make_shared>(std::move(translator), dlist.get(), true, true, false, + std::chrono::milliseconds(5000), std::chrono::milliseconds(0)); + auto op_ctx = std::make_unique(); + + auto accessor = cache_slot->PinCellsDirect(op_ctx.get(), {0}); + ASSERT_NE(accessor, nullptr); + EXPECT_EQ(translator_ptr->GetCellsCallCount(), 1); + EXPECT_EQ(DListTestFriend::get_loading_memory(*dlist), ResourceUsage{}); +} + +TEST(CacheSlotLoadingOverheadTest, PassthroughFileOverheadParticipatesInAdmission) { ResourceUsage limit{110, 200}; auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); - dlist->SetLoadingOverheadTracker(std::make_shared()); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Fixed(100)); + ASSERT_NE(group, nullptr); auto translator = std::make_unique(std::vector>{{0, 10}}, std::unordered_map{{0, 0}}, "test_dimension_admission", StorageType::MEMORY); translator->SetLoadingOverhead({200, 200}); - translator->SetLoadingOverheadConfig( - LoadingOverheadConfig{LoadingOverheadDimensionConfig{100, "load_transient"}, std::nullopt}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{LoadingOverheadGroupBinding{group}, std::nullopt}); auto cache_slot = std::make_shared>(std::move(translator), dlist.get(), true, true, false, std::chrono::milliseconds(0), std::chrono::milliseconds(0)); @@ -2205,17 +2349,18 @@ TEST(CacheSlotTrackerTest, PassthroughFileOverheadParticipatesInAdmission) { EXPECT_NO_THROW(cache_slot->PinCellsDirect(op_ctx.get(), {0})); } -TEST(CacheSlotTrackerTest, InsufficientDiskRejectsPassthroughFileOverhead) { +TEST(CacheSlotLoadingOverheadTest, InsufficientDiskRejectsPassthroughFileOverhead) { ResourceUsage limit{1000, 199}; auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); - dlist->SetLoadingOverheadTracker(std::make_shared()); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Fixed(100)); + ASSERT_NE(group, nullptr); auto translator = std::make_unique(std::vector>{{0, 10}}, std::unordered_map{{0, 0}}, "test_dimension_disk_reject", StorageType::MEMORY); translator->SetLoadingOverhead({200, 200}); - translator->SetLoadingOverheadConfig( - LoadingOverheadConfig{LoadingOverheadDimensionConfig{100, "load_transient"}, std::nullopt}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{LoadingOverheadGroupBinding{group}, std::nullopt}); auto* translator_ptr = translator.get(); auto cache_slot = std::make_shared>(std::move(translator), dlist.get(), true, true, false, @@ -2226,22 +2371,23 @@ TEST(CacheSlotTrackerTest, InsufficientDiskRejectsPassthroughFileOverhead) { EXPECT_EQ(translator_ptr->GetCellsCallCount(), 0); } -// Test that tracker state is properly cleaned up when load throws an exception. -TEST(CacheSlotTrackerTest, LoadingOverheadTrackerCleanupOnException) { +// Test that Group state is properly cleaned up when load throws an exception. +TEST(CacheSlotLoadingOverheadTest, GroupCleanupOnException) { ResourceUsage limit{10000, 0}; auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); - auto tracker = std::make_shared(); - dlist->SetLoadingOverheadTracker(tracker); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Fixed(500)); + ASSERT_NE(group, nullptr); const int64_t cell_loaded_size = 100; const int64_t cell_loading_overhead = 200; auto translator = std::make_unique(std::vector>{{0, cell_loaded_size}}, std::unordered_map{{0, 0}}, - "test_tracker_exception", StorageType::MEMORY); + "test_group_exception", StorageType::MEMORY); translator->SetLoadingOverhead({cell_loading_overhead, 0}); - translator->SetLoadingOverheadConfig("test_group", {500, 0}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{LoadingOverheadGroupBinding{group}, std::nullopt}); translator->SetShouldThrow(true); auto cache_slot = @@ -2250,29 +2396,36 @@ TEST(CacheSlotTrackerTest, LoadingOverheadTrackerCleanupOnException) { auto op_ctx = std::make_unique(); - // Pin should fail because translator throws, but tracker state must be cleaned up. + // Pin should fail because translator throws, but Group state must be cleaned up. EXPECT_ANY_THROW(cache_slot->PinCellsDirect(op_ctx.get(), {0})); - // Register again to get handle for verification (CacheSlot registered internally). - auto handle = tracker->Register("test_group", {500, 0}); + // Bind again for verification (CacheSlot is already bound internally). + LoadingOverheadConfig verification_binding{LoadingOverheadGroupBinding{group}, std::nullopt}; + dlist->BindLoadingOverheadGroups(verification_binding); - // Verify tracker state is clean by reserving the full UB. - auto delta = tracker->Reserve(handle, {500, 0}); - EXPECT_EQ(delta.memory_bytes, 500); + // Verify Group state is clean by reserving the full Group bound. + auto reserve = + std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{500, 0}, &verification_binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(reserve.success); + EXPECT_EQ(reserve.reserved.memory_bytes, 500); - auto release = tracker->Release(handle, {500, 0}); + auto release = dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{500, 0}, &verification_binding); EXPECT_EQ(release.memory_bytes, 500); + dlist->UnbindLoadingOverheadGroups(verification_binding); } -// Test bonus cells retry path with tracker: essential+bonus exceeds DList capacity, +// Test bonus cells retry path with a Group: essential+bonus exceeds DList capacity, // falls back to essential-only which succeeds. -TEST(CacheSlotTrackerTest, BonusCellsRetryWithTracker) { - // Tight capacity: can hold 2 cells (200 bytes) + overhead (up to UB=100), but not 3 cells (300). +TEST(CacheSlotLoadingOverheadTest, BonusCellsRetryWithGroup) { + // Tight capacity: can hold 2 cells (200 bytes) + overhead (up to bound=100), but not 3 cells (300). ResourceUsage limit{350, 0}; auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); - auto tracker = std::make_shared(); - dlist->SetLoadingOverheadTracker(tracker); + auto group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Fixed(100)); + ASSERT_NE(group, nullptr); const int64_t cell_loaded_size = 100; const int64_t cell_loading_overhead = 50; @@ -2285,7 +2438,7 @@ TEST(CacheSlotTrackerTest, BonusCellsRetryWithTracker) { std::vector>{{0, cell_loaded_size}, {1, cell_loaded_size}, {2, cell_loaded_size}}, std::unordered_map{{0, 0}, {1, 1}, {2, 2}}, "test_bonus_retry", StorageType::MEMORY); translator->SetLoadingOverhead({cell_loading_overhead, 0}); - translator->SetLoadingOverheadConfig("test_bonus_retry", {100, 0}); + translator->SetLoadingOverheadConfig(LoadingOverheadConfig{LoadingOverheadGroupBinding{group}, std::nullopt}); translator->SetExtraReturnCids({{0, {1, 2}}}); auto cache_slot = diff --git a/test/test_cachinglayer/test_dlist.cpp b/test/test_cachinglayer/test_dlist.cpp index d6710da..717c883 100644 --- a/test/test_cachinglayer/test_dlist.cpp +++ b/test/test_cachinglayer/test_dlist.cpp @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -15,6 +16,12 @@ using milvus::cachinglayer::cid_t; using milvus::cachinglayer::EvictionConfig; +using milvus::cachinglayer::LoadingOverheadConfig; +using milvus::cachinglayer::LoadingOverheadDimension; +using milvus::cachinglayer::LoadingOverheadGroup; +using milvus::cachinglayer::LoadingOverheadGroupBinding; +using milvus::cachinglayer::LoadingOverheadPolicy; +using milvus::cachinglayer::LoadingOverheadUpdateResult; using milvus::cachinglayer::ResourceUsage; using milvus::cachinglayer::internal::DList; using milvus::cachinglayer::internal::DListTestFriend; @@ -71,6 +78,11 @@ class DListTest : public ::testing::Test { dlist->ReleaseLoadingResource(size); } + std::shared_ptr + CreateMemoryGroup(LoadingOverheadPolicy policy) { + return dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, std::move(policy)); + } + // Helper to create a mock node, simulate loading it, and add it to the list. // Returns a raw pointer, but ownership is managed by the shared_ptr in managed_nodes. MockListNode* @@ -134,6 +146,230 @@ TEST_F(DListTest, Initialization) { EXPECT_EQ(DLF::get_tail(*dlist), nullptr); } +TEST_F(DListTest, LoadingOverheadGroupNeedsNoSeparateTracker) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(group, nullptr); + + LoadingOverheadConfig config{ + LoadingOverheadGroupBinding{group, 10}, + std::nullopt, + }; + EXPECT_NO_THROW(dlist->BindLoadingOverheadGroups(config)); + dlist->UnbindLoadingOverheadGroups(config); +} + +TEST_F(DListTest, PolicyTighteningDrainsExistingReservationWithInflightWork) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Passthrough()); + ASSERT_NE(group, nullptr); + + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{group, 50}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(binding); + auto reservation = std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{100, 0}, &binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(reservation.success); + EXPECT_EQ(reservation.reserved, (ResourceUsage{100, 0})); + + auto waiter = dlist->ReserveLoadingResourceWithTimeout({50, 0}, std::chrono::seconds(1)); + EXPECT_FALSE(waiter.isReady()); + + EXPECT_EQ(dlist->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(get_loading_memory(), (ResourceUsage{100, 0})); + EXPECT_FALSE(waiter.isReady()); + + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{100, 0}, &binding), (ResourceUsage{100, 0})); + EXPECT_TRUE(std::move(waiter).get()); + EXPECT_EQ(get_loading_memory(), (ResourceUsage{50, 0})); + dlist->ReleaseLoadingResource({50, 0}); + EXPECT_EQ(get_loading_memory(), ResourceUsage{}); +} + +TEST_F(DListTest, PolicyExpansionReconcilesOnNextReserveAndRollsBackFailure) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(group, nullptr); + + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{group, 50}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(binding); + auto reservation = std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{500, 0}, &binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(reservation.success); + EXPECT_EQ(reservation.reserved, (ResourceUsage{50, 0})); + + EXPECT_EQ(dlist->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(3)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(get_loading_memory(), (ResourceUsage{50, 0})); + + auto failed = std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{1, 0}, &binding, std::chrono::milliseconds(0))) + .get(); + EXPECT_FALSE(failed.success); + EXPECT_EQ(get_loading_memory(), (ResourceUsage{50, 0})); + + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{500, 0}, &binding), (ResourceUsage{50, 0})); +} + +TEST_F(DListTest, FailedReserveRollsBackActiveDemand) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(group, nullptr); + + ASSERT_TRUE(std::move(dlist->ReserveLoadingResourceWithTimeout({50, 0}, std::chrono::milliseconds(0))).get()); + LoadingOverheadConfig failed_binding{ + LoadingOverheadGroupBinding{group, 80}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(failed_binding); + auto failed = std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{500, 0}, &failed_binding, std::chrono::milliseconds(0))) + .get(); + EXPECT_FALSE(failed.success); + dlist->ReleaseLoadingResource({50, 0}); + + LoadingOverheadConfig admitted_binding{ + LoadingOverheadGroupBinding{group, 10}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(admitted_binding); + auto admitted = + std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{500, 0}, &admitted_binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(admitted.success); + // Runtime-unit bounds follow attached bindings, so the failed binding's + // 80-byte bound remains while its CacheSlot is bound. + EXPECT_EQ(admitted.reserved, (ResourceUsage{80, 0})); + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{500, 0}, &admitted_binding), + (ResourceUsage{80, 0})); +} + +TEST_F(DListTest, GroupsAreDimensionIndependent) { + auto memory_group = CreateMemoryGroup(LoadingOverheadPolicy::Passthrough()); + auto file_group = + dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kFile, LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(memory_group, nullptr); + ASSERT_NE(file_group, nullptr); + + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{memory_group, 50}, + LoadingOverheadGroupBinding{file_group, 10}, + }; + dlist->BindLoadingOverheadGroups(binding); + auto reservation = std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{100, 50}, &binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(reservation.success); + EXPECT_EQ(reservation.reserved, (ResourceUsage{100, 10})); + + EXPECT_EQ(dlist->UpdateLoadingOverheadGroup(memory_group, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(dlist->UpdateLoadingOverheadGroup(file_group, LoadingOverheadPolicy::Passthrough()), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(get_loading_memory(), (ResourceUsage{100, 10})); + + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{100, 50}, &binding), (ResourceUsage{100, 10})); + EXPECT_EQ(get_loading_memory(), ResourceUsage{}); +} + +TEST_F(DListTest, FractionalLoadingFactorScalesGroupDelta) { + auto scaled_config = eviction_config_; + scaled_config.loading_resource_factor = 1.5F; + auto scaled_dlist = std::make_shared(true, initial_limit, low_watermark, high_watermark, scaled_config); + + auto group = + scaled_dlist->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(group, nullptr); + + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{group, 1}, + std::nullopt, + }; + scaled_dlist->BindLoadingOverheadGroups(binding); + auto reservation = std::move(scaled_dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{2, 0}, &binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(reservation.success); + EXPECT_EQ(reservation.reserved, (ResourceUsage{1, 0})); + EXPECT_EQ(DLF::get_loading_memory(*scaled_dlist), (ResourceUsage{2, 0})); + + ASSERT_EQ(scaled_dlist->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Passthrough()), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(DLF::get_loading_memory(*scaled_dlist), (ResourceUsage{2, 0})); + + ASSERT_EQ(scaled_dlist->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(0)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(DLF::get_loading_memory(*scaled_dlist), (ResourceUsage{2, 0})); + + EXPECT_EQ(scaled_dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{2, 0}, &binding), + (ResourceUsage{1, 0})); + EXPECT_EQ(DLF::get_loading_memory(*scaled_dlist), ResourceUsage{}); +} + +TEST_F(DListTest, BindingChangesSerializeWithAdmissionTransactions) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + ASSERT_NE(group, nullptr); + + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{group, 10}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(binding); + + auto expect_serialized = [&](auto&& operation) { + auto list_lock = DLF::test_lock_list(dlist.get()); + std::promise operation_started; + auto operation_started_future = operation_started.get_future(); + auto operation_future = std::async(std::launch::async, [&]() { + operation_started.set_value(); + operation(); + }); + operation_started_future.wait(); + EXPECT_EQ(operation_future.wait_for(std::chrono::milliseconds(50)), std::future_status::timeout); + + list_lock.unlock(); + operation_future.get(); + }; + + expect_serialized([&]() { dlist->UnbindLoadingOverheadGroups(binding); }); + + expect_serialized([&]() { dlist->BindLoadingOverheadGroups(binding); }); + dlist->UnbindLoadingOverheadGroups(binding); +} + +TEST_F(DListTest, GroupReconfigurationReprocessesWaitersWithNewBound) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Passthrough()); + ASSERT_NE(group, nullptr); + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{group}, + std::nullopt, + }; + dlist->BindLoadingOverheadGroups(binding); + + auto active = std::move(dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{80, 0}, &binding, std::chrono::milliseconds(0))) + .get(); + ASSERT_TRUE(active.success); + auto waiter = dlist->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{30, 0}, &binding, std::chrono::milliseconds(-1)); + ASSERT_FALSE(waiter.isReady()); + + EXPECT_EQ(dlist->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Fixed(80)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_TRUE(waiter.isReady()); + const auto waiter_result = std::move(waiter).get(); + EXPECT_TRUE(waiter_result.success); + EXPECT_EQ(waiter_result.reserved, ResourceUsage{}); + + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{30, 0}, &binding), ResourceUsage{}); + EXPECT_EQ(dlist->ReleaseLoadingResource(/*loaded=*/{}, /*overhead=*/{80, 0}, &binding), (ResourceUsage{80, 0})); +} + TEST_F(DListTest, UpdateMaxLimitIncrease) { MockListNode* node1 = add_and_load_node({10, 5}); EXPECT_EQ(get_used_memory(), node1->loaded_size()); @@ -146,6 +382,20 @@ TEST_F(DListTest, UpdateMaxLimitIncrease) { EXPECT_EQ(get_loading_memory(), ResourceUsage{}); } +TEST_F(DListTest, UpdateMaxLimitIncreaseReprocessesWaiters) { + ASSERT_TRUE(reserveLoadingMemorySync({80, 0})); + + auto waiter = dlist->ReserveLoadingResourceWithTimeout({50, 0}, std::chrono::milliseconds(-1)); + ASSERT_FALSE(waiter.isReady()); + + ASSERT_TRUE(dlist->UpdateMaxLimit({150, 50})); + ASSERT_TRUE(waiter.isReady()); + EXPECT_TRUE(std::move(waiter).get()); + EXPECT_EQ(get_loading_memory(), (ResourceUsage{130, 0})); + dlist->ReleaseLoadingResource({50, 0}); + dlist->ReleaseLoadingResource({80, 0}); +} + TEST_F(DListTest, UpdateMaxLimitDecreaseNoEviction) { MockListNode* node1 = add_and_load_node({10, 5}); ResourceUsage current_usage = node1->loaded_size(); @@ -814,6 +1064,55 @@ TEST_F(DListTest, ReserveWithCancellationTokenCancelledWhileWaiting) { EXPECT_EQ(get_loading_memory(), ResourceUsage{}); } +TEST_F(DListTest, TimeoutOfHeadWaiterReprocessesFollowingRequest) { + auto config = eviction_config_; + config.eviction_interval = std::chrono::milliseconds(60000); + dlist = std::make_shared(true, initial_limit, low_watermark, high_watermark, config); + + ASSERT_TRUE(reserveLoadingMemorySync({70, 0})); + ASSERT_TRUE(reserveLoadingMemorySync({30, 0})); + + auto head = dlist->ReserveLoadingResourceWithTimeout({80, 0}, std::chrono::milliseconds(50)); + auto following = dlist->ReserveLoadingResourceWithTimeout({20, 0}, std::chrono::milliseconds(-1)); + dlist->ReleaseLoadingResource({30, 0}); + ASSERT_FALSE(following.isReady()); + + EXPECT_FALSE(std::move(head).get()); + for (int i = 0; i < 50 && !following.isReady(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + ASSERT_TRUE(following.isReady()); + EXPECT_TRUE(std::move(following).get()); + dlist->ReleaseLoadingResource({20, 0}); + dlist->ReleaseLoadingResource({70, 0}); +} + +TEST_F(DListTest, CancellationOfHeadWaiterReprocessesFollowingRequest) { + auto config = eviction_config_; + config.eviction_interval = std::chrono::milliseconds(60000); + dlist = std::make_shared(true, initial_limit, low_watermark, high_watermark, config); + + ASSERT_TRUE(reserveLoadingMemorySync({70, 0})); + ASSERT_TRUE(reserveLoadingMemorySync({30, 0})); + + folly::CancellationSource cancel_source; + auto op_ctx = std::make_unique(cancel_source.getToken()); + auto head = dlist->ReserveLoadingResourceWithTimeout({80, 0}, std::chrono::seconds(5), op_ctx.get()); + auto following = dlist->ReserveLoadingResourceWithTimeout({20, 0}, std::chrono::milliseconds(-1)); + dlist->ReleaseLoadingResource({30, 0}); + ASSERT_FALSE(following.isReady()); + + cancel_source.requestCancellation(); + EXPECT_FALSE(std::move(head).get()); + for (int i = 0; i < 50 && !following.isReady(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + ASSERT_TRUE(following.isReady()); + EXPECT_TRUE(std::move(following).get()); + dlist->ReleaseLoadingResource({20, 0}); + dlist->ReleaseLoadingResource({70, 0}); +} + TEST_F(DListTest, ReserveWithCancellationTokenSucceedsBeforeCancel) { // Test that if reservation succeeds before cancellation, it returns true folly::CancellationSource cancel_source; diff --git a/test/test_cachinglayer/test_loading_overhead_group.cpp b/test/test_cachinglayer/test_loading_overhead_group.cpp new file mode 100644 index 0000000..f672f5a --- /dev/null +++ b/test/test_cachinglayer/test_loading_overhead_group.cpp @@ -0,0 +1,303 @@ +#include + +#include +#include +#include +#include +#include +#include + +#include "cachinglayer/LoadingOverhead.h" +#include "cachinglayer/Utils.h" +#include "cachinglayer/lrucache/DList.h" + +using namespace milvus::cachinglayer; +using milvus::cachinglayer::internal::DList; + +class LoadingOverheadGroupTest : public ::testing::Test { + protected: + std::shared_ptr + CreateMemoryGroup(const LoadingOverheadPolicy& policy) { + return dlist_->CreateLoadingOverheadGroup(LoadingOverheadDimension::kMemory, policy); + } + + LoadingOverheadConfig + BindMemory(const std::shared_ptr& group, + std::optional max_runtime_unit = std::nullopt) { + LoadingOverheadConfig binding{ + LoadingOverheadGroupBinding{group, max_runtime_unit}, + std::nullopt, + }; + dlist_->BindLoadingOverheadGroups(binding); + return binding; + } + + ResourceUsage + Reserve(const LoadingOverheadConfig& binding, ResourceUsage overhead) { + auto result = std::move(dlist_->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, overhead, &binding, std::chrono::milliseconds(0))) + .get(); + EXPECT_TRUE(result.success); + return result.reserved; + } + + ResourceUsage + Release(const LoadingOverheadConfig& binding, ResourceUsage overhead) { + return dlist_->ReleaseLoadingResource(/*loaded=*/{}, overhead, &binding); + } + + void + ReserveMemory(const LoadingOverheadConfig& binding, int count, int64_t overhead) { + for (int i = 0; i < count; ++i) { + Reserve(binding, {overhead, 0}); + } + } + + std::shared_ptr dlist_ = std::make_shared(false, ResourceUsage{1'000'000, 1'000'000}, ResourceUsage{}, + ResourceUsage{}, EvictionConfig{}); +}; + +TEST_F(LoadingOverheadGroupTest, BindingRequiresGroup) { + EXPECT_THROW(dlist_->BindLoadingOverheadGroups(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{}, + std::nullopt, + }), + std::invalid_argument); +} + +TEST_F(LoadingOverheadGroupTest, FailedBindingDoesNotAttachConfiguredDimensions) { + auto memory = CreateMemoryGroup(LoadingOverheadPolicy::Passthrough()); + + EXPECT_THROW(dlist_->BindLoadingOverheadGroups(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{memory}, + LoadingOverheadGroupBinding{memory}, + }), + std::invalid_argument); + + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(memory, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kApplied); +} + +TEST_F(LoadingOverheadGroupTest, FixedGroupCapsAndReleasesReservation) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Fixed(200)); + auto binding = BindMemory(group); + + EXPECT_EQ(Reserve(binding, {150, 0}), (ResourceUsage{150, 0})); + EXPECT_EQ(Reserve(binding, {100, 0}), (ResourceUsage{50, 0})); + EXPECT_EQ(Reserve(binding, {100, 0}), ResourceUsage{}); + + EXPECT_EQ(Release(binding, {100, 0}), ResourceUsage{}); + EXPECT_EQ(Release(binding, {100, 0}), (ResourceUsage{50, 0})); + EXPECT_EQ(Release(binding, {150, 0}), (ResourceUsage{150, 0})); +} + +TEST_F(LoadingOverheadGroupTest, GroupsReserveIndependently) { + auto vector_group = CreateMemoryGroup(LoadingOverheadPolicy::Fixed(200)); + auto scalar_group = CreateMemoryGroup(LoadingOverheadPolicy::Fixed(100)); + + auto vector = BindMemory(vector_group); + auto scalar = BindMemory(scalar_group); + + EXPECT_EQ(Reserve(vector, {300, 0}), (ResourceUsage{200, 0})); + EXPECT_EQ(Reserve(scalar, {300, 0}), (ResourceUsage{100, 0})); +} + +TEST_F(LoadingOverheadGroupTest, ConcurrentReserveReleasePreservesAccounting) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Fixed(200)); + auto binding = BindMemory(group); + + constexpr int kThreadCount = 10; + constexpr int kOperationsPerThread = 100; + std::vector threads; + std::atomic total_reserved{0}; + std::atomic total_released{0}; + + threads.reserve(kThreadCount); + for (int i = 0; i < kThreadCount; ++i) { + threads.emplace_back([&]() { + for (int j = 0; j < kOperationsPerThread; ++j) { + total_reserved += Reserve(binding, {10, 0}).memory_bytes; + } + for (int j = 0; j < kOperationsPerThread; ++j) { + total_released += Release(binding, {10, 0}).memory_bytes; + } + }); + } + + for (auto& thread : threads) { + thread.join(); + } + EXPECT_EQ(total_reserved.load(), total_released.load()); +} + +TEST_F(LoadingOverheadGroupTest, BindingRejectsInvalidMetadata) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Passthrough()); + + EXPECT_THROW(dlist_->BindLoadingOverheadGroups(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{group, -1}, + std::nullopt, + }), + std::invalid_argument); +} + +TEST_F(LoadingOverheadGroupTest, GroupUsesBoundRuntimeUnit) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + auto binding = BindMemory(group, 100); + + EXPECT_EQ(Reserve(binding, {300, 0}), (ResourceUsage{100, 0})); + EXPECT_EQ(Release(binding, {300, 0}), (ResourceUsage{100, 0})); +} + +TEST_F(LoadingOverheadGroupTest, BudgetAndExecutorPoliciesCanReplaceEachOther) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Budget(200)); + auto config = BindMemory(group, 100); + + EXPECT_EQ(Reserve(config, {500, 0}), (ResourceUsage{200, 0})); + + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(Release(config, {100, 0}), (ResourceUsage{100, 0})); + + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Budget(300)), + LoadingOverheadUpdateResult::kApplied); + EXPECT_EQ(Reserve(config, {100, 0}), (ResourceUsage{200, 0})); + EXPECT_EQ(Release(config, {500, 0}), (ResourceUsage{300, 0})); +} + +TEST_F(LoadingOverheadGroupTest, BoundedGroupRequiresBoundRuntimeUnit) { + auto executor = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + + EXPECT_THROW(dlist_->BindLoadingOverheadGroups(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{executor}, + std::nullopt, + }), + std::invalid_argument); + + auto budget = CreateMemoryGroup(LoadingOverheadPolicy::Budget(100)); + EXPECT_THROW(dlist_->BindLoadingOverheadGroups(LoadingOverheadConfig{ + LoadingOverheadGroupBinding{budget}, + std::nullopt, + }), + std::invalid_argument); +} + +TEST_F(LoadingOverheadGroupTest, PassthroughAllowsMissingRuntimeUnitButBoundedReconfigurationDoesNot) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Passthrough()); + auto binding = BindMemory(group); + + EXPECT_EQ(Reserve(binding, {300, 0}), (ResourceUsage{300, 0})); + EXPECT_EQ(Release(binding, {300, 0}), (ResourceUsage{300, 0})); + + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kIncompatiblePolicy); + + dlist_->UnbindLoadingOverheadGroups(binding); + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kApplied); +} + +TEST_F(LoadingOverheadGroupTest, GroupCachesMaximumAcrossRuntimeUnitBounds) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + + auto large = BindMemory(group, 300); + auto small = BindMemory(group, 100); + + EXPECT_EQ(Reserve(small, {500, 0}), (ResourceUsage{300, 0})); + EXPECT_EQ(Release(small, {500, 0}), (ResourceUsage{300, 0})); + + dlist_->UnbindLoadingOverheadGroups(large); + EXPECT_EQ(Reserve(small, {500, 0}), (ResourceUsage{100, 0})); + EXPECT_EQ(Release(small, {500, 0}), (ResourceUsage{100, 0})); +} + +TEST_F(LoadingOverheadGroupTest, GroupSurvivesWithoutBindings) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + auto first = BindMemory(group, 100); + dlist_->UnbindLoadingOverheadGroups(first); + + auto second = BindMemory(group, 50); + EXPECT_EQ(Reserve(second, {300, 0}), (ResourceUsage{50, 0})); + EXPECT_EQ(Release(second, {300, 0}), (ResourceUsage{50, 0})); +} + +TEST_F(LoadingOverheadGroupTest, PolicyTighteningAppliesImmediately) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(4)); + auto binding = BindMemory(group, 100); + ReserveMemory(binding, 5, 100); + + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(1)), + LoadingOverheadUpdateResult::kApplied); + + EXPECT_EQ(Release(binding, {100, 0}), (ResourceUsage{300, 0})); + EXPECT_EQ(Release(binding, {100, 0}), ResourceUsage{}); + EXPECT_EQ(Release(binding, {100, 0}), ResourceUsage{}); + EXPECT_EQ(Release(binding, {100, 0}), ResourceUsage{}); + EXPECT_EQ(Release(binding, {100, 0}), (ResourceUsage{100, 0})); +} + +TEST_F(LoadingOverheadGroupTest, PolicyExpansionReconcilesOnNextReserve) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + auto binding = BindMemory(group, 100); + ReserveMemory(binding, 5, 100); + + EXPECT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(2)), + LoadingOverheadUpdateResult::kApplied); + + EXPECT_EQ(Reserve(binding, {100, 0}), (ResourceUsage{100, 0})); + EXPECT_EQ(Release(binding, {100, 0}), ResourceUsage{}); + EXPECT_EQ(Release(binding, {500, 0}), (ResourceUsage{200, 0})); +} + +TEST_F(LoadingOverheadGroupTest, FailedReserveRevertsDeltaAfterPolicyExpansion) { + auto group = CreateMemoryGroup(LoadingOverheadPolicy::Executor(1)); + auto binding = BindMemory(group, 100); + ASSERT_EQ(Reserve(binding, {500, 0}), (ResourceUsage{100, 0})); + ASSERT_EQ(dlist_->UpdateLoadingOverheadGroup(group, LoadingOverheadPolicy::Executor(4)), + LoadingOverheadUpdateResult::kApplied); + + ASSERT_TRUE(std::move(dlist_->ReserveLoadingResourceWithTimeout({999'700, 0}, std::chrono::milliseconds(0))).get()); + const auto failed = std::move(dlist_->ReserveLoadingResourceWithTimeout( + /*loaded=*/{}, /*overhead=*/{100, 0}, &binding, std::chrono::milliseconds(0))) + .get(); + EXPECT_FALSE(failed.success); + dlist_->ReleaseLoadingResource({999'700, 0}); + + EXPECT_EQ(Release(binding, {500, 0}), (ResourceUsage{100, 0})); +} + +TEST_F(LoadingOverheadGroupTest, DimensionsUseIndependentGroupsWhileAbsentFilePassesThrough) { + auto memory_group = CreateMemoryGroup(LoadingOverheadPolicy::Fixed(200)); + auto file_group = + dlist_->CreateLoadingOverheadGroup(LoadingOverheadDimension::kFile, LoadingOverheadPolicy::Fixed(50)); + + auto scalar_binding = BindMemory(memory_group); + LoadingOverheadConfig field_binding{ + LoadingOverheadGroupBinding{memory_group}, + LoadingOverheadGroupBinding{file_group}, + }; + dlist_->BindLoadingOverheadGroups(field_binding); + + auto scalar_first = Reserve(scalar_binding, {150, 100}); + EXPECT_EQ(scalar_first, (ResourceUsage{150, 100})); + + auto field_first = Reserve(field_binding, {100, 40}); + EXPECT_EQ(field_first, (ResourceUsage{50, 40})); + + auto scalar_second = Reserve(scalar_binding, {100, 200}); + EXPECT_EQ(scalar_second, (ResourceUsage{0, 200})); + + auto field_second = Reserve(field_binding, {0, 20}); + EXPECT_EQ(field_second, (ResourceUsage{0, 10})); + + auto scalar_first_release = Release(scalar_binding, {150, 100}); + EXPECT_EQ(scalar_first_release, (ResourceUsage{0, 100})); + + auto field_first_release = Release(field_binding, {100, 40}); + EXPECT_EQ(field_first_release, (ResourceUsage{100, 30})); + + auto scalar_second_release = Release(scalar_binding, {100, 200}); + EXPECT_EQ(scalar_second_release, (ResourceUsage{100, 200})); + + auto field_second_release = Release(field_binding, {0, 20}); + EXPECT_EQ(field_second_release, (ResourceUsage{0, 20})); +} diff --git a/test/test_cachinglayer/test_loading_overhead_policy.cpp b/test/test_cachinglayer/test_loading_overhead_policy.cpp new file mode 100644 index 0000000..1011abc --- /dev/null +++ b/test/test_cachinglayer/test_loading_overhead_policy.cpp @@ -0,0 +1,66 @@ +#include + +#include +#include +#include + +#include "cachinglayer/LoadingOverhead.h" + +namespace milvus::cachinglayer { +namespace { + +TEST(LoadingOverheadPolicyTest, FixedUsesConfiguredBound) { + const auto policy = LoadingOverheadPolicy::Fixed(400); + + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/100), 400); + EXPECT_FALSE(policy.RequiresRuntimeUnitBound()); +} + +TEST(LoadingOverheadPolicyTest, FixedRejectsNegativeBound) { + EXPECT_THROW(LoadingOverheadPolicy::Fixed(-1), std::invalid_argument); +} + +TEST(LoadingOverheadPolicyTest, PassthroughUsesUnlimitedBound) { + const auto policy = LoadingOverheadPolicy::Passthrough(); + + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/100), std::numeric_limits::max()); + EXPECT_FALSE(policy.RequiresRuntimeUnitBound()); +} + +TEST(LoadingOverheadPolicyTest, BudgetUsesLargerOfCapacityAndRuntimeUnit) { + const auto policy = LoadingOverheadPolicy::Budget(400); + + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/100), 400); + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/500), 500); + EXPECT_TRUE(policy.RequiresRuntimeUnitBound()); +} + +TEST(LoadingOverheadPolicyTest, ZeroBudgetUsesUnlimitedBound) { + const auto policy = LoadingOverheadPolicy::Budget(0); + + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/50), std::numeric_limits::max()); +} + +TEST(LoadingOverheadPolicyTest, BudgetRejectsNegativeCapacity) { + EXPECT_THROW(LoadingOverheadPolicy::Budget(-1), std::invalid_argument); +} + +TEST(LoadingOverheadPolicyTest, ExecutorScalesRuntimeUnitByWorkerCount) { + const auto policy = LoadingOverheadPolicy::Executor(6); + + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/100), 600); + EXPECT_TRUE(policy.RequiresRuntimeUnitBound()); +} + +TEST(LoadingOverheadPolicyTest, ExecutorSaturatesMultiplication) { + const auto policy = LoadingOverheadPolicy::Executor(std::numeric_limits::max()); + + EXPECT_EQ(policy.ResolveBound(/*max_runtime_unit_bytes=*/2), std::numeric_limits::max()); +} + +TEST(LoadingOverheadPolicyTest, ExecutorRejectsNegativeWorkerCount) { + EXPECT_THROW(LoadingOverheadPolicy::Executor(-1), std::invalid_argument); +} + +} // namespace +} // namespace milvus::cachinglayer diff --git a/test/test_cachinglayer/test_loading_overhead_tracker.cpp b/test/test_cachinglayer/test_loading_overhead_tracker.cpp deleted file mode 100644 index bf0a9f1..0000000 --- a/test/test_cachinglayer/test_loading_overhead_tracker.cpp +++ /dev/null @@ -1,380 +0,0 @@ -#include - -#include -#include - -#include "cachinglayer/LoadingOverhead.h" -#include "cachinglayer/LoadingOverheadTracker.h" -#include "cachinglayer/Utils.h" - -using namespace milvus::cachinglayer; - -class LoadingOverheadTrackerTest : public ::testing::Test { - protected: - LoadingOverheadTracker tracker_; -}; - -TEST_F(LoadingOverheadTrackerTest, NoUpperBoundPassThrough) { - // Without registering a UB, all amounts pass through unchanged. - auto handle = tracker_.Register("vector_index", LoadingOverheadTracker::kUnlimited); - auto delta = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(delta.memory_bytes, 100); - EXPECT_EQ(delta.file_bytes, 0); - - auto release = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(release.memory_bytes, 100); - EXPECT_EQ(release.file_bytes, 0); -} - -TEST_F(LoadingOverheadTrackerTest, BasicCapping) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - // First reserve: 100, sum=100 <= UB=200, full amount passes through - auto d1 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d1.memory_bytes, 100); - - // Second reserve: 100, sum=200 <= UB=200, full amount passes through - auto d2 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d2.memory_bytes, 100); - - // Third reserve: 100, sum=300 > UB=200, capped: delta = 0 - auto d3 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d3.memory_bytes, 0); - - // Fourth reserve: 100, sum=400 > UB=200, capped: delta = 0 - auto d4 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d4.memory_bytes, 0); -} - -TEST_F(LoadingOverheadTrackerTest, BasicRelease) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - // Reserve 4x100, total sum=400, actual DList reserve = 200 - tracker_.Reserve(handle, {100, 0}); - tracker_.Reserve(handle, {100, 0}); - tracker_.Reserve(handle, {100, 0}); - tracker_.Reserve(handle, {100, 0}); - - // Release first 100: sum 400->300, both >= UB, release 0 - auto r1 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r1.memory_bytes, 0); - - // Release second 100: sum 300->200, release 0 - auto r2 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r2.memory_bytes, 0); - - // Release third 100: sum 200->100, release 100 - auto r3 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r3.memory_bytes, 100); - - // Release fourth 100: sum 100->0, release 100 - auto r4 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r4.memory_bytes, 100); -} - -TEST_F(LoadingOverheadTrackerTest, TotalReservedEqualsReleased) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - int64_t total_reserved = 0; - int64_t total_released = 0; - - // Reserve 10 x 100 - for (int i = 0; i < 10; i++) { - total_reserved += tracker_.Reserve(handle, {100, 0}).memory_bytes; - } - // Should have reserved exactly UB = 200 - EXPECT_EQ(total_reserved, 200); - - // Release all 10 - for (int i = 0; i < 10; i++) { - total_released += tracker_.Release(handle, {100, 0}).memory_bytes; - } - // Total released should equal total reserved - EXPECT_EQ(total_released, 200); -} - -TEST_F(LoadingOverheadTrackerTest, PartialCapping) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - // Reserve 150: sum=150 <= UB=200, full amount - auto d1 = tracker_.Reserve(handle, {150, 0}); - EXPECT_EQ(d1.memory_bytes, 150); - - // Reserve 100: sum=250 > UB=200, delta = 200-150 = 50 - auto d2 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d2.memory_bytes, 50); -} - -TEST_F(LoadingOverheadTrackerTest, ReleaseUndoesReserve) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - // Reserve 150: actual = 150 - auto d1 = tracker_.Reserve(handle, {150, 0}); - EXPECT_EQ(d1.memory_bytes, 150); - - // Release undoes the reserve: sum 150->0, release = 150 - auto undo = tracker_.Release(handle, {150, 0}); - EXPECT_EQ(undo.memory_bytes, 150); -} - -TEST_F(LoadingOverheadTrackerTest, MultipleTypes) { - auto vec_handle = tracker_.Register("vector_index", {200, 0}); - auto scalar_handle = tracker_.Register("scalar_field", {100, 0}); - - // Types are tracked independently - auto d1 = tracker_.Reserve(vec_handle, {200, 0}); - EXPECT_EQ(d1.memory_bytes, 200); - - auto d2 = tracker_.Reserve(scalar_handle, {100, 0}); - EXPECT_EQ(d2.memory_bytes, 100); - - // Both at UB, further reserves return 0 - auto d3 = tracker_.Reserve(vec_handle, {100, 0}); - EXPECT_EQ(d3.memory_bytes, 0); - - auto d4 = tracker_.Reserve(scalar_handle, {50, 0}); - EXPECT_EQ(d4.memory_bytes, 0); -} - -TEST_F(LoadingOverheadTrackerTest, RegisterUpperBoundTakesMax) { - auto handle = tracker_.Register("vector_index", {100, 50}); - handle = tracker_.Register("vector_index", {200, 30}); - - // UB should be {200, 50} (max per dimension) - auto ub = tracker_.GetUpperBound(handle); - EXPECT_EQ(ub.memory_bytes, 200); - EXPECT_EQ(ub.file_bytes, 50); - - auto d1 = tracker_.Reserve(handle, {200, 50}); - EXPECT_EQ(d1.memory_bytes, 200); - EXPECT_EQ(d1.file_bytes, 50); - - // Next reserve should be fully capped - auto d2 = tracker_.Reserve(handle, {100, 100}); - EXPECT_EQ(d2.memory_bytes, 0); - EXPECT_EQ(d2.file_bytes, 0); -} - -TEST_F(LoadingOverheadTrackerTest, HasFiniteUpperBound) { - auto unlimited_handle = tracker_.Register("vector_index", LoadingOverheadTracker::kUnlimited); - EXPECT_FALSE(tracker_.HasFiniteUpperBound(unlimited_handle)); - - auto finite_handle = tracker_.Register("finite_vector_index", {200, 0}); - EXPECT_TRUE(tracker_.HasFiniteUpperBound(finite_handle)); - - auto scalar_handle = tracker_.Register("scalar_field", LoadingOverheadTracker::kUnlimited); - EXPECT_FALSE(tracker_.HasFiniteUpperBound(scalar_handle)); -} - -TEST_F(LoadingOverheadTrackerTest, ConcurrentReserveRelease) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - const int num_threads = 10; - const int ops_per_thread = 100; - std::vector threads; - std::atomic total_reserved{0}; - std::atomic total_released{0}; - - threads.reserve(num_threads); - for (int i = 0; i < num_threads; i++) { - threads.emplace_back([&]() { - for (int j = 0; j < ops_per_thread; j++) { - auto reserved = tracker_.Reserve(handle, {10, 0}); - total_reserved += reserved.memory_bytes; - } - for (int j = 0; j < ops_per_thread; j++) { - auto released = tracker_.Release(handle, {10, 0}); - total_released += released.memory_bytes; - } - }); - } - - for (auto& t : threads) { - t.join(); - } - - // Total reserved must equal total released - EXPECT_EQ(total_reserved.load(), total_released.load()); -} - -TEST_F(LoadingOverheadTrackerTest, DefaultUnlimitedUBFallback) { - // Register with kUnlimited -> unlimited, behaves like no capping. - auto handle = tracker_.Register("vector_index", LoadingOverheadTracker::kUnlimited); - - EXPECT_FALSE(tracker_.HasFiniteUpperBound(handle)); - - auto d1 = tracker_.Reserve(handle, {1000000000, 0}); - EXPECT_EQ(d1.memory_bytes, 1000000000); - - auto d2 = tracker_.Reserve(handle, {2000000000, 0}); - EXPECT_EQ(d2.memory_bytes, 2000000000); - - auto r1 = tracker_.Release(handle, {1000000000, 0}); - EXPECT_EQ(r1.memory_bytes, 1000000000); - - auto r2 = tracker_.Release(handle, {2000000000, 0}); - EXPECT_EQ(r2.memory_bytes, 2000000000); -} - -TEST_F(LoadingOverheadTrackerTest, RegisterUnlimitedThenFiniteKeepsUnlimited) { - auto handle = tracker_.Register("vector_index", LoadingOverheadTracker::kUnlimited); - EXPECT_FALSE(tracker_.HasFiniteUpperBound(handle)); - - handle = tracker_.Register("vector_index", {200, 0}); - EXPECT_FALSE(tracker_.HasFiniteUpperBound(handle)); - - // INT64_MAX is an explicit unlimited upper bound. Use a missing dimension - // when loading overhead should pass through without joining a capped group. - auto d1 = tracker_.Reserve(handle, {300, 0}); - EXPECT_EQ(d1.memory_bytes, 300); -} - -TEST_F(LoadingOverheadTrackerTest, UnregisteredTypeAutoCreatesUnlimited) { - auto handle = tracker_.Register("scalar_field", LoadingOverheadTracker::kUnlimited); - auto d1 = tracker_.Reserve(handle, {500, 0}); - EXPECT_EQ(d1.memory_bytes, 500); - - auto r1 = tracker_.Release(handle, {500, 0}); - EXPECT_EQ(r1.memory_bytes, 500); - - EXPECT_FALSE(tracker_.HasFiniteUpperBound(handle)); -} - -TEST_F(LoadingOverheadTrackerTest, UBChangesMidFlight) { - auto handle = tracker_.Register("vector_index", {200, 0}); - - // Reserve 3x100 under UB=200: dlist gets 100+100+0 = 200 - auto d1 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d1.memory_bytes, 100); - auto d2 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d2.memory_bytes, 100); - auto d3 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d3.memory_bytes, 0); - - int64_t total_reserved = d1.memory_bytes + d2.memory_bytes + d3.memory_bytes; - EXPECT_EQ(total_reserved, 200); - - // UB changes to 400 - handle = tracker_.Register("vector_index", {400, 0}); - - // Release all 3: should release exactly 200 total - auto r1 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r1.memory_bytes, 0); - auto r2 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r2.memory_bytes, 100); - auto r3 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r3.memory_bytes, 100); - - int64_t total_released = r1.memory_bytes + r2.memory_bytes + r3.memory_bytes; - EXPECT_EQ(total_released, 200); - EXPECT_EQ(total_reserved, total_released); -} - -TEST_F(LoadingOverheadTrackerTest, UnlimitedUBDoesNotDecreaseMidFlight) { - auto handle = tracker_.Register("vector_index", LoadingOverheadTracker::kUnlimited); - auto d1 = tracker_.Reserve(handle, {1000, 0}); - EXPECT_EQ(d1.memory_bytes, 1000); - auto d2 = tracker_.Reserve(handle, {1000, 0}); - EXPECT_EQ(d2.memory_bytes, 1000); - - // A later finite registration must not lower an explicit unlimited upper bound. - handle = tracker_.Register("vector_index", {200, 0}); - - auto d3 = tracker_.Reserve(handle, {100, 0}); - EXPECT_EQ(d3.memory_bytes, 100); - - auto r1 = tracker_.Release(handle, {1000, 0}); - EXPECT_EQ(r1.memory_bytes, 1000); - auto r2 = tracker_.Release(handle, {1000, 0}); - EXPECT_EQ(r2.memory_bytes, 1000); - auto r3 = tracker_.Release(handle, {100, 0}); - EXPECT_EQ(r3.memory_bytes, 100); - - int64_t total_reserved = d1.memory_bytes + d2.memory_bytes + d3.memory_bytes; - int64_t total_released = r1.memory_bytes + r2.memory_bytes + r3.memory_bytes; - EXPECT_EQ(total_reserved, 2100); - EXPECT_EQ(total_released, 2100); -} - -TEST_F(LoadingOverheadTrackerTest, GetUpperBound) { - auto unlimited_handle = tracker_.Register("vector_index", LoadingOverheadTracker::kUnlimited); - EXPECT_EQ(tracker_.GetUpperBound(unlimited_handle), LoadingOverheadTracker::kUnlimited); - - auto finite_handle = tracker_.Register("finite_vector_index", {200, 100}); - auto ub = tracker_.GetUpperBound(finite_handle); - EXPECT_EQ(ub.memory_bytes, 200); - EXPECT_EQ(ub.file_bytes, 100); -} - -TEST_F(LoadingOverheadTrackerTest, PartialUnlimitedRemainsUnlimitedRegardlessOfRegistrationOrder) { - constexpr auto kMax = std::numeric_limits::max(); - - auto max_first = tracker_.Register("max_first", {kMax, 0}); - tracker_.Register("max_first", {200, 0}); - EXPECT_EQ(tracker_.GetUpperBound(max_first), (ResourceUsage{kMax, 0})); - - auto finite_first = tracker_.Register("finite_first", {200, 0}); - tracker_.Register("finite_first", {kMax, 0}); - EXPECT_EQ(tracker_.GetUpperBound(finite_first), (ResourceUsage{kMax, 0})); -} - -TEST_F(LoadingOverheadTrackerTest, LegacyConfigConstructorConfiguresBothDimensions) { - ResourceUsage upper_bound{200, 50}; - LoadingOverheadConfig config(upper_bound, "legacy_group"); - - ASSERT_TRUE(config.memory.has_value()); - EXPECT_EQ(config.memory->upper_bound, 200); - EXPECT_EQ(config.memory->group, "legacy_group"); - ASSERT_TRUE(config.file.has_value()); - EXPECT_EQ(config.file->upper_bound, 50); - EXPECT_EQ(config.file->group, "legacy_group"); - - auto handle = tracker_.Register(config); - EXPECT_EQ(tracker_.GetUpperBound(handle), upper_bound); -} - -TEST_F(LoadingOverheadTrackerTest, DimensionsShareMemoryWhileScalarFilePassesThrough) { - auto scalar_handle = - tracker_.Register(LoadingOverheadConfig{LoadingOverheadDimensionConfig{200, "load_transient"}, std::nullopt}); - auto field_handle = tracker_.Register(LoadingOverheadConfig{LoadingOverheadDimensionConfig{200, "load_transient"}, - LoadingOverheadDimensionConfig{50, "load_transient"}}); - - auto scalar_first = tracker_.Reserve(scalar_handle, {150, 100}); - EXPECT_EQ(scalar_first, (ResourceUsage{150, 100})); - - auto field_first = tracker_.Reserve(field_handle, {100, 40}); - EXPECT_EQ(field_first, (ResourceUsage{50, 40})); - - auto scalar_second = tracker_.Reserve(scalar_handle, {100, 200}); - EXPECT_EQ(scalar_second, (ResourceUsage{0, 200})); - - auto field_second = tracker_.Reserve(field_handle, {0, 20}); - EXPECT_EQ(field_second, (ResourceUsage{0, 10})); - - auto scalar_first_release = tracker_.Release(scalar_handle, {150, 100}); - EXPECT_EQ(scalar_first_release, (ResourceUsage{0, 100})); - - auto field_first_release = tracker_.Release(field_handle, {100, 40}); - EXPECT_EQ(field_first_release, (ResourceUsage{100, 30})); - - auto scalar_second_release = tracker_.Release(scalar_handle, {100, 200}); - EXPECT_EQ(scalar_second_release, (ResourceUsage{100, 200})); - - auto field_second_release = tracker_.Release(field_handle, {0, 20}); - EXPECT_EQ(field_second_release, (ResourceUsage{0, 20})); -} - -TEST_F(LoadingOverheadTrackerTest, PassthroughRegistrationDoesNotPolluteFiniteFileGroup) { - auto field_handle = tracker_.Register(LoadingOverheadConfig{LoadingOverheadDimensionConfig{200, "load_transient"}, - LoadingOverheadDimensionConfig{50, "load_transient"}}); - auto scalar_handle = - tracker_.Register(LoadingOverheadConfig{LoadingOverheadDimensionConfig{200, "load_transient"}, std::nullopt}); - - EXPECT_EQ(tracker_.GetUpperBound(field_handle), (ResourceUsage{200, 50})); - EXPECT_EQ(tracker_.GetUpperBound(scalar_handle), (ResourceUsage{200, std::numeric_limits::max()})); - - auto scalar = tracker_.Reserve(scalar_handle, {0, 100}); - auto field = tracker_.Reserve(field_handle, {0, 100}); - EXPECT_EQ(scalar.file_bytes, 100); - EXPECT_EQ(field.file_bytes, 50); -}