fix: deduplicate eviction loop, fix logical-limit recheck, and improve release accuracy

sparknack · sparknack · commit 1a179baba878 · 2026-04-02T07:05:04.000Z
Extract reserveResourceInternalImpl to eliminate duplicated eviction logic between
reserveResourceInternal and reserveResourceInternalWithTracker. Fix a bug where
logical_limit_exceeded was unconditionally cleared after eviction instead of being
rechecked. Return actual unscaled release size from ReleaseLoadingResource for
accurate metrics tracking. Also improve exception safety in CacheSlot constructor
and optimize LoadingOverheadTracker::Unregister from O(n) to O(1).

Signed-off-by: Shawn Wang &lt;shawn.wang@zilliz.com&gt;
diff --git a/include/cachinglayer/CacheSlot.h b/include/cachinglayer/CacheSlot.h
@@ -68,15 +68,17 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
           storage_usage_tracking_enabled_(storage_usage_tracking_enabled),
           loading_timeout_(loading_timeout),
           warmup_loading_timeout_(warmup_loading_timeout) {
-        if (auto& lo = translator_->meta()->loading_overhead) {
-            overhead_handle_ = dlist_->RegisterLoadingOverhead(lo->group, lo->upper_bound);
-        }
         cells_.reserve(translator_->num_cells());
         for (cid_t i = 0; i < static_cast<cid_t>(translator_->num_cells()); ++i) {
             cells_.push_back(std::make_unique<CacheCell>(this, i));
         }
         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->group, lo->upper_bound);
+        }
     }
 
     CacheSlot(const CacheSlot&) = delete;
@@ -485,13 +487,14 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
                     return;
                 }
                 try {
-                    dlist_->ReleaseLoadingResource(loaded_resource, loading_overhead, overhead_handle_);
+                    auto released_resource =
+                        dlist_->ReleaseLoadingResource(loaded_resource, loading_overhead, overhead_handle_);
                     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(actual_dlist_reserve.memory_bytes);
+                            .Decrement(released_resource.memory_bytes);
                         monitor::cache_loading_bytes(cell_data_type_, StorageType::DISK)
-                            .Decrement(actual_dlist_reserve.file_bytes);
+                            .Decrement(released_resource.file_bytes);
                     }
                 } catch (...) {
                     auto ew = folly::exception_wrapper(std::current_exception());
diff --git a/include/cachinglayer/LoadingOverheadTracker.h b/include/cachinglayer/LoadingOverheadTracker.h
@@ -58,10 +58,12 @@ class LoadingOverheadTracker {
                          it->second, state.ref_count, upper_bound.ToString());
             } else if (state.upper_bound.memory_bytes < upper_bound.memory_bytes ||
                        state.upper_bound.file_bytes < upper_bound.file_bytes) {
+                LOG_WARN(
+                    "[MCL] LoadingOverheadTracker UB mismatch for group '{}' (handle {}): existing={}, new={}. "
+                    "Taking max per dimension.",
+                    group, it->second, state.upper_bound.ToString(), upper_bound.ToString());
                 state.upper_bound.memory_bytes = std::max(state.upper_bound.memory_bytes, upper_bound.memory_bytes);
                 state.upper_bound.file_bytes = std::max(state.upper_bound.file_bytes, upper_bound.file_bytes);
-                LOG_INFO("[MCL] LoadingOverheadTracker widened UB for group '{}' (handle {}, refs={}): {}", group,
-                         it->second, state.ref_count, state.upper_bound.ToString());
             } else {
                 LOG_DEBUG("[MCL] LoadingOverheadTracker re-registered group '{}' (handle {}, refs={}), UB unchanged",
                           group, it->second, state.ref_count);
@@ -70,7 +72,7 @@ class LoadingOverheadTracker {
         }
         auto handle = next_handle_++;
         name_to_handle_[group] = handle;
-        handle_state_[handle] = GroupState{upper_bound, {}, {}, 1};
+        handle_state_[handle] = GroupState{upper_bound, {}, {}, 1, group};
         LOG_INFO("[MCL] LoadingOverheadTracker registered group '{}' (handle {}, refs=1): UB={}", group, handle,
                  upper_bound.ToString());
         return handle;
@@ -166,13 +168,8 @@ class LoadingOverheadTracker {
                 "sum_of_overhead={}, overhead_reserved={}. Cleaning up anyway to avoid leak.",
                 handle, state.sum_of_overhead.ToString(), state.overhead_reserved.ToString());
         }
-        for (auto nit = name_to_handle_.begin(); nit != name_to_handle_.end(); ++nit) {
-            if (nit->second == handle) {
-                LOG_INFO("[MCL] LoadingOverheadTracker unregistered group '{}' (handle {})", nit->first, handle);
-                name_to_handle_.erase(nit);
-                break;
-            }
-        }
+        LOG_INFO("[MCL] LoadingOverheadTracker unregistered group '{}' (handle {})", state.group_name, handle);
+        name_to_handle_.erase(state.group_name);
         handle_state_.erase(it);
     }
 
@@ -182,6 +179,7 @@ class LoadingOverheadTracker {
         ResourceUsage sum_of_overhead;
         ResourceUsage overhead_reserved;
         uint64_t ref_count{0};
+        std::string group_name;
     };
 
     static ResourceUsage
diff --git a/include/cachinglayer/lrucache/DList.h b/include/cachinglayer/lrucache/DList.h
@@ -160,7 +160,8 @@ class DList : public std::enable_shared_from_this<DList> {
                                       OpContext* ctx = nullptr);
 
     // Release with loading overhead tracker integration.
-    void
+    // Returns the actual unscaled size released (loaded + tracker_delta).
+    ResourceUsage
     ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& overhead, uint64_t overhead_handle);
 
     // Release resource used for loading, called after loading a cell.
@@ -273,6 +274,11 @@ class DList : public std::enable_shared_from_this<DList> {
     bool
     reserveResourceInternal(const ResourceUsage& size);
 
+    // Common implementation for resource reservation with eviction.
+    // Returns {success, scaled_size_reserved}.
+    std::pair<bool, ResourceUsage>
+    reserveResourceInternalImpl(const ResourceUsage& size, std::function<void()> 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.
diff --git a/src/cachinglayer/lrucache/DList.cpp b/src/cachinglayer/lrucache/DList.cpp
@@ -219,6 +219,42 @@ DList::ReserveLoadingResourceWithTimeout(const ResourceUsage& original_size, std
 
 bool
 DList::reserveResourceInternal(const ResourceUsage& size) {
+    auto rollback = []() {};
+    auto [success, _] = reserveResourceInternalImpl(size, rollback);
+    return success;
+}
+
+std::pair<bool, ResourceUsage>
+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).
+    auto delta = overhead;
+    if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && tracker != nullptr) {
+        delta = tracker->Reserve(overhead_handle, overhead);
+    }
+    auto actual_size = (loaded + delta) * eviction_config_.loading_resource_factor;
+
+    auto rollback = [overhead_handle, overhead, tracker]() {
+        if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && tracker != nullptr) {
+            tracker->Release(overhead_handle, overhead);
+        }
+    };
+
+    auto [success, _] = reserveResourceInternalImpl(actual_size, rollback);
+    if (!success) {
+        return {false, {}};
+    }
+
+    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};
+}
+
+std::pair<bool, ResourceUsage>
+DList::reserveResourceInternalImpl(const ResourceUsage& size, std::function<void()> rollback) {
     auto using_resources = total_loaded_size_.load() + total_loading_size_.load();
 
     // Combined logical and physical memory limit check
@@ -268,10 +304,13 @@ DList::reserveResourceInternal(const ResourceUsage& size) {
                 "[MCL] reserve resource with size={} failed due to all zero evicted_size, "
                 "eviction_target={}, min_eviction={}",
                 size.ToString(), eviction_target.ToString(), min_eviction.ToString());
-            return false;
+            rollback();
+            return {false, {}};
         }
+
+        using_resources -= evicted_size;
         // logical limit is accurate, thus we can guarantee after one successful eviction, logical limit is satisfied.
-        logical_limit_exceeded = false;
+        logical_limit_exceeded = !max_resource_limit_.load().CanHold(using_resources + size);
 
         if (!physical_eviction_needed.AnyGTZero()) {
             // we only need to evict for logical limit and we have succeeded.
@@ -290,79 +329,10 @@ DList::reserveResourceInternal(const ResourceUsage& size) {
     }
 
     total_loading_size_ += size;
-    LOG_TRACE("[MCL] reserve resource with size={} success, total_loading_size={}, total_loaded_size={}",
-              size.ToString(), total_loading_size_.load().ToString(), total_loaded_size_.load().ToString());
-
-    return true;
-}
-
-std::pair<bool, ResourceUsage>
-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).
-    auto delta = overhead;
-    if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && tracker != nullptr) {
-        delta = tracker->Reserve(overhead_handle, overhead);
-    }
-    auto actual_size = (loaded + delta) * eviction_config_.loading_resource_factor;
-    auto using_resources = total_loaded_size_.load() + total_loading_size_.load();
+    LOG_TRACE("[MCL] reserve resource success, size={}, total_loading_size={}, total_loaded_size={}", size.ToString(),
+              total_loading_size_.load().ToString(), total_loaded_size_.load().ToString());
 
-    bool logical_limit_exceeded = !max_resource_limit_.load().CanHold(using_resources + actual_size);
-    auto physical_eviction_needed = checkPhysicalMemoryLimit(actual_size);
-
-    while (logical_limit_exceeded || physical_eviction_needed.AnyGTZero()) {
-        ResourceUsage eviction_target;
-        ResourceUsage min_eviction;
-
-        if (logical_limit_exceeded) {
-            eviction_target = using_resources + actual_size - low_watermark_;
-            min_eviction = using_resources + actual_size - max_resource_limit_.load();
-            if (eviction_target.memory_bytes < 0)
-                eviction_target.memory_bytes = 0;
-            if (eviction_target.file_bytes < 0)
-                eviction_target.file_bytes = 0;
-            if (min_eviction.memory_bytes < 0)
-                min_eviction.memory_bytes = 0;
-            if (min_eviction.file_bytes < 0)
-                min_eviction.file_bytes = 0;
-        }
-
-        if (physical_eviction_needed.AnyGTZero()) {
-            eviction_target.memory_bytes =
-                std::max(eviction_target.memory_bytes, physical_eviction_needed.memory_bytes);
-            eviction_target.file_bytes = std::max(eviction_target.file_bytes, physical_eviction_needed.file_bytes);
-            min_eviction.memory_bytes = std::max(min_eviction.memory_bytes, physical_eviction_needed.memory_bytes);
-            min_eviction.file_bytes = std::max(min_eviction.file_bytes, physical_eviction_needed.file_bytes);
-        }
-
-        ResourceUsage evicted_size = tryEvict(eviction_target, min_eviction);
-        if (!evicted_size.AnyGTZero()) {
-            LOG_WARN(
-                "[MCL] reserve with tracker failed: loaded={}, overhead={}, delta={}, eviction_target={}, "
-                "min_eviction={}",
-                loaded.ToString(), overhead.ToString(), delta.ToString(), eviction_target.ToString(),
-                min_eviction.ToString());
-            // Rollback tracker state on failure
-            if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && tracker != nullptr) {
-                tracker->Release(overhead_handle, overhead);
-            }
-            return {false, {}};
-        }
-        logical_limit_exceeded = false;
-
-        if (!physical_eviction_needed.AnyGTZero())
-            break;
-        if (physical_eviction_needed = checkPhysicalMemoryLimit(actual_size); !physical_eviction_needed.AnyGTZero())
-            break;
-    }
-
-    total_loading_size_ += actual_size;
-    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};
+    return {true, size};
 }
 
 void
@@ -643,16 +613,18 @@ DList::UpdateHighWatermark(const ResourceUsage& new_high_watermark) {
     cachinglayer::monitor::cache_high_watermark_bytes(StorageType::DISK).Set(high_watermark_.load().file_bytes);
 }
 
-void
+ResourceUsage
 DList::ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage& overhead, uint64_t overhead_handle) {
     std::vector<std::unique_ptr<WaitingRequest>> to_destroy;
+    ResourceUsage unscaled{};
     {
         std::unique_lock<std::mutex> lock(list_mtx_);
         auto delta = overhead;
         if (overhead_handle != LoadingOverheadTracker::kInvalidHandle && loading_overhead_tracker_) {
             delta = loading_overhead_tracker_->Release(overhead_handle, overhead);
         }
-        auto actual = (loaded + delta) * eviction_config_.loading_resource_factor;
+        unscaled = loaded + delta;
+        auto actual = unscaled * eviction_config_.loading_resource_factor;
         total_loading_size_ -= actual;
         ClampNonNegative(total_loading_size_, [&](const ResourceUsage& curr) {
             LOG_ERROR(
@@ -663,6 +635,7 @@ DList::ReleaseLoadingResource(const ResourceUsage& loaded, const ResourceUsage&
         to_destroy = handleWaitingRequests();
     }
     // Destroy requests outside lock to avoid deadlock with cancel callbacks
+    return unscaled;
 }
 
 void
@@ -884,14 +857,16 @@ DList::handleWaitingRequests() {
             waiting_queue_.pop();
         } else {
             // Check if this request is permanently impossible (required size exceeds capacity).
-            // If so, fail it immediately instead of blocking the entire queue until timeout.
+            // 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)) {
                 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={}), "
                         "failing immediately.",
-                        request->request_id, request->required_size.ToString(), max_resource_limit_.load().ToString());
+                        request->request_id, request->required_size.ToString(),
+                        max_resource_limit_.load().ToString());
                     request->setValue(false);
                 }
                 requests_to_destroy.push_back(std::move(request));
