Skip to content

enhance: support transactional loading overhead policies #109

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sparknack wants to merge 1 commit into
base: master
Choose a base branch
from
Open

enhance: support transactional loading overhead policies #109

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 40 additions & 26 deletions include/cachinglayer/CacheSlot.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@
#include <utility>
#include <vector>

#include "cachinglayer/LoadingOverheadTracker.h"
#include "cachinglayer/LoadingOverhead.h"
#include "cachinglayer/Metrics.h"
#include "cachinglayer/Translator.h"
#include "cachinglayer/Utils.h"
Expand Down Expand Up @@ -68,7 +68,8 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
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_ =
Expand All @@ -78,13 +79,11 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
for (cid_t i = 0; i < static_cast<cid_t>(translator_->num_cells()); ++i) {
cells_.push_back(std::make_unique<CacheCell>(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;
Expand Down Expand Up @@ -345,14 +344,21 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
}

~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());
}

private:
friend class CellAccessor<CellT>;

[[nodiscard]] const LoadingOverheadConfig*
loadingOverheadConfig() const noexcept {
return loading_overhead_config_ ? &loading_overhead_config_.value() : nullptr;
}

[[nodiscard]] std::vector<cid_t>
AllCellIds() const {
std::vector<cid_t> cids(translator_->num_cells());
Expand Down Expand Up @@ -437,8 +443,8 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
RunLoad(OpContext* ctx, std::unordered_set<cid_t>&& 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<cid_t> loading_cids;
try {
auto start = std::chrono::steady_clock::now();
Expand Down Expand Up @@ -484,8 +490,8 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
}

// 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;

Expand All @@ -500,27 +506,28 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
// 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());
Expand All @@ -539,9 +546,10 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
"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());
}
Expand All @@ -562,9 +570,14 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
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;
Expand Down Expand Up @@ -719,7 +732,8 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
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<LoadingOverheadConfig> loading_overhead_config_;
std::atomic<bool> warmup_called_{false};
std::atomic<bool> skip_pin_{false};
};
Expand Down
217 changes: 200 additions & 17 deletions include/cachinglayer/LoadingOverhead.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,38 +11,221 @@

#pragma once

#include <algorithm>
#include <cstdint>
#include <limits>
#include <memory>
#include <optional>
#include <set>
#include <stdexcept>
#include <string>
#include <utility>

#include "cachinglayer/Utils.h"

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<LoadingOverheadDimensionConfig> memory,
std::optional<LoadingOverheadDimensionConfig> 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<int64_t>::max();
case Kind::kBudget:
if (max_runtime_unit_bytes < 0 || value_ == 0) {
return std::numeric_limits<int64_t>::max();
}
return std::max(value_, max_runtime_unit_bytes);
case Kind::kExecutor:
return SaturatingMultiply(value_, max_runtime_unit_bytes);
}
return std::numeric_limits<int64_t>::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<LoadingOverheadDimensionConfig> memory;
std::optional<LoadingOverheadDimensionConfig> 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<int64_t>::max();
}
if (lhs == 0 || rhs == 0) {
return 0;
}
if (lhs > std::numeric_limits<int64_t>::max() / rhs) {
return std::numeric_limits<int64_t>::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<int64_t>& max_runtime_unit) const;

void
bind(const std::optional<int64_t>& max_runtime_unit);

void
unbind(const internal::DList* owner, LoadingOverheadDimension dimension,
const std::optional<int64_t>& 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<int64_t> 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<LoadingOverheadGroup> 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<int64_t> 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<LoadingOverheadGroupBinding> memory;

/** @brief File-dimension binding, or std::nullopt for request-local passthrough. */
std::optional<LoadingOverheadGroupBinding> 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
Loading
Loading