GC

Author: UESTC-AHao

examples/ucm_config_example.yaml

@@ -17,6 +17,7 @@ ucm_connectors:
       storage_backends: "/mnt/test"
       io_direct: false
       # cache_buffer_capacity_gb: 256
+      # posix_capacity_gb: 1024
 
 # When you use UcmNfsStore, you should set enable_event_sync to false.
 enable_event_sync: true

ucm/integration/vllm/ucm_connector.py

@@ -193,6 +193,11 @@ def __init__(self, vllm_config: "VllmConfig", role: KVConnectorRole):
         )
         self.tp_size = self._vllm_config.parallel_config.tensor_parallel_size
         self.kv_cache_dtype: torch.dtype = None
+        self.num_head = vllm_config.model_config.get_num_kv_heads(
+            vllm_config.parallel_config
+        )
+        self.head_size = vllm_config.model_config.get_head_size()
+        self.element_size = vllm_config.model_config.dtype.itemsize
 
         if current_platform.is_cuda_alike():
             logger.info("CUDA device is available.")
@@ -300,6 +305,20 @@ def _create_store(
             config["shard_size"] = kv_cache_layout.shard_size * self.blocks_per_chunk
             config["block_size"] = kv_cache_layout.block_size * self.blocks_per_chunk
             config["local_rank_size"] = self.tp_size if self.is_mla else 1
+        else:
+            config_base = self.block_size * self.element_size * self.head_size
+            config["block_size"] = (
+                config_base
+                * self.num_layers
+                * (1 if self.is_mla else self.num_head * 2)
+                * self.blocks_per_chunk
+            )
+        # GC only enabled for Scheduler with data_parallel_rank == 0
+        if config.get("posix_gc_enable", False):
+            dp_rank = self._vllm_config.parallel_config.data_parallel_rank
+            if self._role == KVConnectorRole.WORKER or dp_rank != 0:
+                config["posix_gc_enable"] = False
+
         logger.info(f"create {name} with config: {config}")
         return UcmConnectorFactoryV1.create_connector(name, config, module_path)
 

ucm/store/posix/cc/global_config.h

@@ -43,6 +43,14 @@ struct Config {
     size_t commitConcurrency{4};
     size_t timeoutMs{30000};
     size_t dataDirShardBytes{3};
+    bool posixGcEnable{true};
+    double posixGcRecyclePercent{0.1};
+    size_t posixGcConcurrency{16};
+    size_t posixGcCheckIntervalSec{30};
+    size_t posixCapacityGb{0};
+    double posixGcTriggerThresholdRatio{0.7};
+    size_t posixGcMaxRecycleCountPerShard{1000000};
+    double posixGcShardSampleRatio{0.1};
 };
 
 }  // namespace UC::PosixStore

ucm/store/posix/cc/hotness_tracker.cc

@@ -0,0 +1,78 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ * */
+#include "hotness_tracker.h"
+#include <utime.h>
+#include "logger/logger.h"
+
+namespace UC::PosixStore {
+
+HotnessTracker::~HotnessTracker()
+{
+    stop_.store(true);
+    if (utimeWorker_.joinable()) { utimeWorker_.join(); }
+}
+
+Status HotnessTracker::Setup(const SpaceLayout* layout)
+{
+    layout_ = layout;
+
+    utimeWorker_ = std::thread(&HotnessTracker::UtimeWorkerLoop, this);
+    return Status::OK();
+}
+
+void HotnessTracker::Touch(const Detail::BlockId& blockId)
+{
+    std::lock_guard<std::mutex> lock(queueMtx_);
+    produceQueue_.push_back(blockId);
+}
+
+void HotnessTracker::UtimeWorkerLoop()
+{
+    std::deque<Detail::BlockId> consumeQueue;
+    constexpr size_t kSpinLimit = 16;
+    size_t spinCount = 0;
+    while (!stop_.load()) {
+        {
+            std::lock_guard<std::mutex> lock(queueMtx_);
+            consumeQueue.swap(produceQueue_);
+        }
+        if (consumeQueue.empty()) {
+            if (++spinCount < kSpinLimit) {
+                std::this_thread::yield();
+            } else {
+                std::this_thread::sleep_for(std::chrono::microseconds(100));
+                spinCount = 0;
+            }
+            continue;
+        }
+        spinCount = 0;
+        while (!consumeQueue.empty()) {
+            auto filePath = layout_->DataFilePath(consumeQueue.front(), false);
+            utime(filePath.c_str(), nullptr);
+            consumeQueue.pop_front();
+        }
+    }
+}
+
+}  // namespace UC::PosixStore

ucm/store/posix/cc/hotness_tracker.h

@@ -0,0 +1,60 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ * */
+#ifndef UNIFIEDCACHE_POSIX_STORE_CC_HOTNESS_TRACKER_H
+#define UNIFIEDCACHE_POSIX_STORE_CC_HOTNESS_TRACKER_H
+
+#include <atomic>
+#include <deque>
+#include <mutex>
+#include <thread>
+#include "space_layout.h"
+#include "type/types.h"
+
+namespace UC::PosixStore {
+
+class HotnessTracker {
+public:
+    HotnessTracker() = default;
+    HotnessTracker(const HotnessTracker&) = delete;
+    HotnessTracker& operator=(const HotnessTracker&) = delete;
+    ~HotnessTracker();
+
+    Status Setup(const SpaceLayout* layout);
+    void Touch(const Detail::BlockId& blockId);
+
+private:
+    void UtimeWorkerLoop();
+
+    const SpaceLayout* layout_{nullptr};
+
+    std::deque<Detail::BlockId> produceQueue_;
+    std::mutex queueMtx_;
+
+    std::atomic<bool> stop_{false};
+    std::thread utimeWorker_;
+};
+
+}  // namespace UC::PosixStore
+
+#endif

ucm/store/posix/cc/posix_store.cc

@@ -43,14 +43,17 @@ class PosixStore : public StoreV1 {
             UC_ERROR("Failed to check config params: {}.", s);
             return s;
         }
+        transEnable_ = config.deviceId >= 0;
+
+        ShowConfig(config);
+
         s = spaceMgr_.Setup(config);
         if (s.Failure()) [[unlikely]] { return s; }
-        transEnable_ = config.deviceId >= 0;
+
         if (transEnable_) {
             s = transMgr_.Setup(config, spaceMgr_.GetLayout());
             if (s.Failure()) [[unlikely]] { return s; }
         }
-        ShowConfig(config);
         return Status::OK();
     }
     std::string Readme() const override { return "PosixStore"; }
@@ -115,6 +118,15 @@ class PosixStore : public StoreV1 {
         inConfig.GetNumber("posix_commit_concurrency", config.commitConcurrency);
         inConfig.GetNumber("timeout_ms", config.timeoutMs);
         inConfig.GetNumber("data_dir_shard_bytes", config.dataDirShardBytes);
+        inConfig.Get("posix_gc_enable", config.posixGcEnable);
+        inConfig.Get("posix_gc_recycle_percent", config.posixGcRecyclePercent);
+        inConfig.GetNumber("posix_gc_concurrency", config.posixGcConcurrency);
+        inConfig.GetNumber("posix_gc_check_interval_sec", config.posixGcCheckIntervalSec);
+        inConfig.GetNumber("posix_capacity_gb", config.posixCapacityGb);
+        inConfig.Get("posix_gc_trigger_threshold_ratio", config.posixGcTriggerThresholdRatio);
+        inConfig.GetNumber("posix_gc_max_recycle_count_per_shard",
+                           config.posixGcMaxRecycleCountPerShard);
+        inConfig.Get("posix_gc_shard_sample_ratio", config.posixGcShardSampleRatio);
         return config;
     }
     Status CheckConfig(const Config& config)

ucm/store/posix/cc/shard_gc.cc

@@ -0,0 +1,155 @@
+/**
+ * MIT License
+ *
+ * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ * */
+#include "shard_gc.h"
+#include "logger/logger.h"
+#include "posix_file.h"
+
+namespace UC::PosixStore {
+
+ShardGarbageCollector::~ShardGarbageCollector() { StopBackgroundCheck(); }
+
+Status ShardGarbageCollector::Setup(const SpaceLayout* layout, const ShardGCConfig& config)
+{
+    layout_ = layout;
+    config_ = config;
+    shards_ = layout_->RelativeRoots();
+
+    if (shards_.empty()) { return Status::InvalidParam("no shards available"); }
+    if (config_.recyclePercent <= 0 || config_.recyclePercent > 1.0f) {
+        return Status::InvalidParam("invalid recycle percent");
+    }
+    if (config_.gcConcurrency == 0) { return Status::InvalidParam("invalid gc concurrency"); }
+
+    auto success = gcPool_.SetWorkerFn([this](ShardTaskContext& ctx, auto&) { ProcessTask(ctx); })
+                       .SetNWorker(config_.gcConcurrency)
+                       .Run();
+
+    if (!success) { return Status::Error("failed to start gc thread pool"); }
+
+    UC_INFO(
+        "ShardGC setup: recyclePercent={}, concurrency={}, maxFileCount={}, "
+        "thresholdRatio={}, checkIntervalSec={}",
+        config_.recyclePercent, config_.gcConcurrency, config_.maxFileCount, config_.thresholdRatio,
+        config_.gcCheckIntervalSec);
+
+    if (config_.maxFileCount > 0) {
+        gcCheckWorker_ = std::thread(&ShardGarbageCollector::GCCheckLoop, this);
+        while (!stop_.load() && ShouldTrigger()) { Execute(); }
+    }
+
+    return Status::OK();
+}
+
+void ShardGarbageCollector::StopBackgroundCheck()
+{
+    {
+        std::lock_guard<std::mutex> lock(gcCheckMtx_);
+        stop_ = true;
+    }
+    gcCheckCv_.notify_all();
+    if (gcCheckWorker_.joinable()) { gcCheckWorker_.join(); }
+}
+
+void ShardGarbageCollector::GCCheckLoop()
+{
+    while (!stop_.load()) {
+        {
+            std::unique_lock<std::mutex> lock(gcCheckMtx_);
+            gcCheckCv_.wait_for(lock, std::chrono::seconds(config_.gcCheckIntervalSec),
+                                [this] { return stop_.load(); });
+        }
+        if (stop_.load()) { break; }
+        while (!stop_.load() && ShouldTrigger()) { Execute(); }
+    }
+}
+
+void ShardGarbageCollector::Execute()
+{
+    std::shared_ptr<Latch> waiter;
+
+    try {
+        waiter = std::make_shared<Latch>();
+    } catch (const std::exception& e) {
+        UC_ERROR("Failed to allocate latch: ", e.what());
+        return;
+    }
+
+    waiter->Set(shards_.size());
+
+    for (const auto& shard : shards_) {
+        gcPool_.Push({ShardTaskContext::Type::GC, shard, waiter, nullptr});
+    }
+
+    waiter->Wait();
+}
+
+bool ShardGarbageCollector::ShouldTrigger()
+{
+    auto sampleShards = layout_->SampleShards(config_.shardSampleRatio);
+    if (sampleShards.empty()) { return false; }
+
+    std::shared_ptr<Latch> waiter;
+
+    try {
+        waiter = std::make_shared<Latch>();
+    } catch (const std::exception& e) {
+        UC_ERROR("Failed to allocate latch: {}", e.what());
+        return false;
+    }
+
+    std::atomic<size_t> sampledFiles{0};
+    waiter->Set(sampleShards.size());
+
+    for (const auto& shard : sampleShards) {
+        gcPool_.Push({ShardTaskContext::Type::SAMPLE, shard, waiter, &sampledFiles});
+    }
+
+    waiter->Wait();
+
+    size_t avgFilesPerShard = sampledFiles.load() / sampleShards.size();
+    size_t thresholdFilesPerShard = config_.maxFileCount / shards_.size();
+
+    UC_INFO("GC sampling: avgFiles/shard={}, threshold={}", avgFilesPerShard,
+            static_cast<size_t>(thresholdFilesPerShard * config_.thresholdRatio));
+
+    return avgFilesPerShard >= static_cast<size_t>(thresholdFilesPerShard * config_.thresholdRatio);
+}
+
+void ShardGarbageCollector::ProcessTask(ShardTaskContext& ctx)
+{
+    if (ctx.type == ShardTaskContext::Type::SAMPLE) {
+        size_t count = layout_->CountFilesInShard(ctx.shard);
+        ctx.sampledFiles->fetch_add(count, std::memory_order_relaxed);
+    } else {
+        auto filesToDelete = layout_->GetOldestFiles(ctx.shard, config_.recyclePercent,
+                                                     config_.maxRecycleCountPerShard);
+        for (const auto& blockId : filesToDelete) {
+            PosixFile{layout_->DataFilePath(blockId, false)}.Remove();
+        }
+    }
+
+    ctx.waiter->Done();
+}
+
+}  // namespace UC::PosixStore