Skip to content

[manager] fair eviction across instances within an instance group #183

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

Draft
oldsharp wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

[manager] fair eviction across instances within an instance group #183

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
2 changes: 2 additions & 0 deletions kv_cache_manager/config/cache_reclaim_strategy.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ bool CacheReclaimStrategy::FromRapidValue(const rapidjson::Value &rapid_value) {
KVCM_JSON_GET_MACRO(rapid_value, "reclaim_step_size", reclaim_step_size_);
KVCM_JSON_GET_MACRO(rapid_value, "reclaim_step_percentage", reclaim_step_percentage_);
KVCM_JSON_GET_MACRO(rapid_value, "delay_before_delete_ms", delay_before_delete_ms_);
KVCM_JSON_GET_MACRO(rapid_value, "enable_instance_fairness", enable_instance_fairness_);
return true;
}

Expand All @@ -23,6 +24,7 @@ void CacheReclaimStrategy::ToRapidWriter(rapidjson::Writer<rapidjson::StringBuff
Put(writer, "reclaim_step_size", reclaim_step_size_);
Put(writer, "reclaim_step_percentage", reclaim_step_percentage_);
Put(writer, "delay_before_delete_ms", delay_before_delete_ms_);
Put(writer, "enable_instance_fairness", enable_instance_fairness_);
}

bool CacheReclaimStrategy::ValidateRequiredFields(std::string &invalid_fields) const {
Expand Down
11 changes: 9 additions & 2 deletions kv_cache_manager/config/cache_reclaim_strategy.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,16 @@ class CacheReclaimStrategy : public Jsonizable {
int32_t trigger_period_seconds,
int32_t reclaim_step_size,
int32_t reclaim_step_percentage,
int32_t delay_before_delete_ms = 0)
int32_t delay_before_delete_ms = 0,
bool enable_instance_fairness = true)
: storage_unique_name_(storage_unique_name)
, reclaim_policy_(reclaim_policy)
, trigger_strategy_(trigger_strategy)
, trigger_period_seconds_(trigger_period_seconds)
, reclaim_step_size_(reclaim_step_size)
, reclaim_step_percentage_(reclaim_step_percentage)
, delay_before_delete_ms_(delay_before_delete_ms) {}
, delay_before_delete_ms_(delay_before_delete_ms)
, enable_instance_fairness_(enable_instance_fairness) {}

~CacheReclaimStrategy() override;

Expand All @@ -52,6 +54,7 @@ class CacheReclaimStrategy : public Jsonizable {
int32_t reclaim_step_size() const { return reclaim_step_size_; }
int32_t reclaim_step_percentage() const { return reclaim_step_percentage_; }
int32_t delay_before_delete_ms() const { return delay_before_delete_ms_; }
bool enable_instance_fairness() const { return enable_instance_fairness_; }

// Setters
void set_storage_unique_name(const std::string &storage_unique_name) { storage_unique_name_ = storage_unique_name; }
Expand All @@ -67,6 +70,9 @@ class CacheReclaimStrategy : public Jsonizable {
void set_delay_before_delete_ms(int32_t delay_before_delete_ms) {
delay_before_delete_ms_ = delay_before_delete_ms;
}
void set_enable_instance_fairness(bool enable_instance_fairness) {
enable_instance_fairness_ = enable_instance_fairness;
}

public:
bool FromRapidValue(const rapidjson::Value &rapid_value) override;
Expand All @@ -82,6 +88,7 @@ class CacheReclaimStrategy : public Jsonizable {
int32_t reclaim_step_size_;
int32_t reclaim_step_percentage_;
int32_t delay_before_delete_ms_;
bool enable_instance_fairness_ = true;
};

} // namespace kv_cache_manager
247 changes: 191 additions & 56 deletions kv_cache_manager/manager/cache_reclaimer.cc
View file
Open in desktop

Large diffs are not rendered by default.

80 changes: 72 additions & 8 deletions kv_cache_manager/manager/cache_reclaimer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -377,22 +377,56 @@ class CacheReclaimer {
array_t_ water_level_exceed_by_type_;
};

// instance group key & storage usage data, with a per-instance
// breakdown so ComputeInstanceReclaimPlan can weight eviction by
// absolute usage without doing a second MetaIndexer pass
struct GroupUsageData {
struct InstanceUsage {
std::shared_ptr<const InstanceInfo> instance_info;
std::size_t used_key_cnt = 0;
std::size_t max_key_cnt = 0;
std::size_t used_byte_sz = 0;
std::array<std::size_t, static_cast<std::size_t>(DataStorageType::COUNT)> used_byte_sz_by_type{};
};

std::size_t grp_used_key_cnt_;
std::size_t grp_max_key_cnt_;
std::size_t grp_used_byte_sz_;
std::vector<InstanceUsage> per_instance_;

GroupUsageData();
~GroupUsageData() = default;

[[nodiscard]] std::size_t GetGroupUsageByType(const DataStorageType &type) const noexcept;
void AddGroupUsageByType(const DataStorageType &type, std::size_t value) noexcept;

// group storage usage data array aggregated by storage type
// slot 0: DATA_STORAGE_TYPE_UNKNOWN **UNUSED**
// slot 1: DATA_STORAGE_TYPE_HF3FS usage data
// slot 2: DATA_STORAGE_TYPE_MOONCAKE usage data
// slot 3: DATA_STORAGE_TYPE_TAIR_MEMPOOL usage data
// slot 4: DATA_STORAGE_TYPE_NFS usage data
// slot 5: DATA_STORAGE_TYPE_VCNS_HF3FS **UNUSED** (merged into HF3FS)
// slot 6: DATA_STORAGE_TYPE_DUMMY usage data (testing only)
std::array<std::size_t, static_cast<std::size_t>(DataStorageType::COUNT)> grp_storage_usage_by_type_;
};

/**
* @brief Calculate the group's WaterLevelExceed data
*
* @param request_context The context of the request
* @param instance_group_name Name of the instance group to check
* @param instance_group_quota Quota info for the instance group
* @param reclaim_strategy Strategy to use for reclamation decisions
* @param instance_infos Vector of instance information
* @param group_usage_data Pre-computed group usage data
* @return shared pointer to the result WaterLevelExceed data
*/
[[nodiscard]] std::shared_ptr<WaterLevelExceed>
GetWaterLevelExceed(const RequestContext *request_context,
const std::string &instance_group_name,
const InstanceGroupQuota &instance_group_quota,
const std::shared_ptr<CacheReclaimStrategy> &reclaim_strategy,
const std::vector<std::shared_ptr<const InstanceInfo>> &instance_infos) noexcept;
const GroupUsageData &group_usage_data) noexcept;

/**
* @brief Determine if a reclamation should be triggered for an
Expand Down Expand Up @@ -425,7 +459,8 @@ class CacheReclaimer {
void ReclaimByLRU(const std::shared_ptr<RequestContext> &request_context,
const std::shared_ptr<const InstanceInfo> &instance_info,
const WaterLevelExceed &water_level_exceed,
std::int32_t delay_before_delete_ms) noexcept;
std::int32_t delay_before_delete_ms,
std::size_t batch_size_override = 0) noexcept;

/**
* @brief Reclaim cache entries using LFU (Least Frequently Used)
Expand All @@ -438,11 +473,14 @@ class CacheReclaimer {
* @param instance_info Instance information to process
* @param water_level_exceed the detailed trigger result
* @param delay_before_delete_ms delay milliseconds for executor
* @param batch_size_override per-instance batch size (0 means use
* the atomic batching_size_); set by the fair-eviction planner
*/
void ReclaimByLFU(const std::shared_ptr<RequestContext> &request_context,
const std::shared_ptr<const InstanceInfo> &instance_info,
const WaterLevelExceed &water_level_exceed,
std::int32_t delay_before_delete_ms) noexcept;
std::int32_t delay_before_delete_ms,
std::size_t batch_size_override = 0) noexcept;

/**
* @brief Reclaim cache entries using TTL (Time To Live) strategy
Expand All @@ -454,11 +492,14 @@ class CacheReclaimer {
* @param instance_info Instance information to process
* @param water_level_exceed the detailed trigger result
* @param delay_before_delete_ms delay milliseconds for executor
* @param batch_size_override per-instance batch size (0 means use
* the atomic batching_size_); set by the fair-eviction planner
*/
void ReclaimByTTL(const std::shared_ptr<RequestContext> &request_context,
const std::shared_ptr<const InstanceInfo> &instance_info,
const WaterLevelExceed &water_level_exceed,
std::int32_t delay_before_delete_ms) noexcept;
std::int32_t delay_before_delete_ms,
std::size_t batch_size_override = 0) noexcept;

bool TryReclaimOnGroup(const std::shared_ptr<RequestContext> &request_context,
const std::shared_ptr<const InstanceGroup> &instance_group) noexcept;
Expand Down Expand Up @@ -487,7 +528,8 @@ class CacheReclaimer {
const std::vector<std::int64_t> &sampled_keys,
const std::vector<std::map<std::string, std::string>> &property_maps,
std::vector<std::int64_t> &out_batch,
AgeStats &out_lru_age_stats) const noexcept;
AgeStats &out_lru_age_stats,
std::size_t batch_size_override = 0) const noexcept;

bool FilterLocID(RequestContext *request_context,
const std::shared_ptr<const InstanceInfo> &instance_info,
Expand All @@ -500,12 +542,34 @@ class CacheReclaimer {
const std::shared_ptr<const InstanceInfo> &instance_info,
const CacheLocationDelRequest &req) noexcept;

struct GroupUsageData;

[[nodiscard]] std::shared_ptr<GroupUsageData>
GetGroupUsageData(const RequestContext *request_context,
const std::vector<std::shared_ptr<const InstanceInfo>> &instance_infos) const noexcept;

// per-instance batch plan returned by ComputeInstanceReclaimPlan
struct InstanceReclaimPlan {
std::shared_ptr<const InstanceInfo> instance_info;
std::size_t batch_size = 0; // 0 means skip this instance
};

/**
* @brief Build a per-instance reclaim plan that biases eviction
* toward instances contributing more to the over-watermark state
*
* When fairness is disabled in the strategy, returns one plan per
* instance with batch_size = batching_size_ (legacy behavior).
* When enabled, allocates a global budget of batching_size_ * N
* keys across instances proportionally to their absolute usage.
* Instances with zero usage are skipped; non-zero contributors are
* floored at one batch.
*/
[[nodiscard]] std::vector<InstanceReclaimPlan>
ComputeInstanceReclaimPlan(const std::shared_ptr<CacheReclaimStrategy> &reclaim_strategy,
const GroupUsageData &group_usage_data,
const WaterLevelExceed &water_level_exceed,
const std::vector<std::shared_ptr<const InstanceInfo>> &instance_infos)
const noexcept;

KVCM_COUNTER_METRICS_FOR_CACHE_RECLAIMER(reclaim_cron_count)
KVCM_COUNTER_METRICS_FOR_CACHE_RECLAIMER(reclaim_job_count)
KVCM_COUNTER_METRICS_FOR_CACHE_RECLAIMER(block_submit_count)
Expand Down
Loading
Loading