[feat] capacity check for nfsstore (#315)

* capacity check

* recycle

---------

Co-authored-by: lijiachen <lijiachen19@huawei.com>

Author: Lijiachen1018

ucm/store/infra/file/file.cc

@@ -53,6 +53,16 @@ Status File::Access(const std::string& path, const int32_t mode)
     return FileImpl{path}.Access(mode);
 }
 
+Status File::Stat(const std::string& path, IFile::FileStat& st)
+{
+    FileImpl file{path};
+    auto status = file.Open(IFile::OpenFlag::READ_ONLY);
+    if (status.Failure()) { return status; }
+    status = file.Stat(st);
+    file.Close();
+    return status;
+}
+
 Status File::Read(const std::string& path, const size_t offset, const size_t length,
                   uintptr_t address, const bool directIo)
 {

ucm/store/infra/file/file.h

@@ -36,6 +36,7 @@ class File {
     static Status RmDir(const std::string& path);
     static Status Rename(const std::string& path, const std::string& newName);
     static Status Access(const std::string& path, const int32_t mode);
+    static Status Stat(const std::string& path, IFile::FileStat& st);
     static Status Read(const std::string& path, const size_t offset, const size_t length,
                        uintptr_t address, const bool directIo = false);
     static Status Write(const std::string& path, const size_t offset, const size_t length,

ucm/store/infra/status/status.h

@@ -42,6 +42,7 @@ class Status {
         ESERIALIZE = UC_MAKE_STATUS_CODE(6),
         EDESERIALIZE = UC_MAKE_STATUS_CODE(7),
         EUNSUPPORTED = UC_MAKE_STATUS_CODE(8),
+        ENOSPACE = UC_MAKE_STATUS_CODE(9),
 #undef UC_MAKE_STATUS_CODE
     };
 
@@ -101,7 +102,11 @@ class Status {
         static Status s{Code::EUNSUPPORTED};
         return s;
     }
-
+    static Status& NoSpace()
+    {
+        static Status s{Code::ENOSPACE};
+        return s;
+    }
 public:
     Status(const Status& status) { this->code_ = status.code_; }
     Status& operator=(const Status& status)

ucm/store/infra/template/topn_heap.h

@@ -0,0 +1,120 @@
+/**
+ * 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_TOP_N_HEAP_H
+#define UNIFIEDCACHE_TOP_N_HEAP_H
+
+#include <algorithm>
+#include <cstdint>
+
+namespace UC {
+
+template <typename T, typename Compare = std::less<T>>
+class TopNHeap {
+public:
+    using ValueType = T;
+    using SizeType = uint32_t;
+    using ConstRef = const T&;
+
+private:
+    using IndexType = uint32_t;
+    std::vector<ValueType> val_{};
+    std::vector<IndexType> idx_{};
+    SizeType capacity_{0};
+    SizeType size_{0};
+    Compare cmp_{};
+
+public:
+    explicit TopNHeap(const SizeType capacity) noexcept(
+        std::is_nothrow_default_constructible_v<Compare>)
+    : capacity_{capacity} {
+        val_.resize(capacity);
+        idx_.resize(capacity);
+    }
+    TopNHeap(const TopNHeap&) = delete;
+    TopNHeap(const TopNHeap&&) = delete;
+    TopNHeap& operator=(const TopNHeap&) = delete;
+    TopNHeap& operator=(const TopNHeap&&) = delete;
+    ~TopNHeap() { Clear(); }
+    
+    SizeType Size() const noexcept { return size_; }
+    SizeType Capacity() const noexcept { return capacity_; }
+    bool Empty() const noexcept { return size_ == 0; }
+
+    void Push(ConstRef value) noexcept {
+        if (size_ < capacity_) {
+            val_[size_] = value;
+            idx_[size_] = size_;
+            SiftUp(size_);
+            size_++;
+            return;
+        }
+        if (cmp_(val_[idx_.front()], value)) {
+            val_[idx_.front()] = value;
+            SiftDown(0);
+        }
+    }
+    ConstRef Top() const noexcept { return val_[idx_.front()]; }
+    void Pop() noexcept {
+        idx_[0] = idx_[--size_];
+        if (size_) { SiftDown(0); }
+    }
+    void Clear() noexcept { size_ = 0; }
+private:
+    static IndexType Parent(IndexType i) noexcept { return (i - 1) / 2; }
+    static IndexType Left(IndexType i) noexcept { return 2 * i + 1; }
+    static IndexType Right(IndexType i) noexcept { return 2 * i + 2; }
+    void SiftUp(IndexType i) noexcept {
+        auto pos = i;
+        while (pos > 0) {
+            auto p = Parent(pos);
+            if (!cmp_(val_[idx_[pos]], val_[idx_[p]])) { break; }
+            std::swap(idx_[pos], idx_[p]);
+            pos = p;
+        }
+    }
+    void SiftDown(IndexType i) noexcept {
+        auto pos = i;
+        for (;;) {
+            auto l = Left(pos);
+            auto r = Right(pos);
+            auto best = pos;
+            if (l < size_ && cmp_(val_[idx_[l]], val_[idx_[best]])) { best = l; }
+            if (r < size_ && cmp_(val_[idx_[r]], val_[idx_[best]])) { best = r; }
+            if (best == pos) { break; }
+            std::swap(idx_[pos], idx_[best]);
+            pos = best;
+        }
+    }
+};
+
+template <typename T, size_t N, typename Compare = std::less<T>>
+class TopNFixedHeap : public TopNHeap<T, Compare> {
+public:
+    TopNFixedHeap() : TopNHeap<T, Compare>{N} {}
+};
+
+} // namespace UC
+
+#endif

ucm/store/nfsstore/cc/api/nfsstore.cc

@@ -35,7 +35,8 @@ class NFSStoreImpl : public NFSStore {
     int32_t Setup(const Config& config)
     {
         auto status = this->spaceMgr_.Setup(config.storageBackends, config.kvcacheBlockSize,
-                                            config.tempDumpDirEnable);
+                                            config.tempDumpDirEnable, config.storageCapacity,
+                                            config.recycleEnable, config.recycleThresholdRatio);
         if (status.Failure()) {
             UC_ERROR("Failed({}) to setup SpaceManager.", status);
             return status.Underlying();
@@ -120,6 +121,9 @@ class NFSStoreImpl : public NFSStore {
         UC_INFO("Set UC::TempDumpDirEnable to {}.", config.tempDumpDirEnable);
         UC_INFO("Set UC::HotnessInterval to {}.", config.hotnessInterval);
         UC_INFO("Set UC::HotnessEnable to {}.", config.hotnessEnable);
+        UC_INFO("Set UC::storageCapacity to {}.", config.storageCapacity);
+        UC_INFO("Set UC::RecycleEnable to {}.", config.recycleEnable);
+        UC_INFO("Set UC::RecycleThreshold to {}.", config.recycleThresholdRatio);
     }
 
 private:

ucm/store/nfsstore/cc/api/nfsstore.h

@@ -43,13 +43,17 @@ class NFSStore : public CCStore {
         bool tempDumpDirEnable;
         bool hotnessEnable;
         size_t hotnessInterval;
+        size_t storageCapacity;
+        bool recycleEnable;
+        float recycleThresholdRatio;
 
         Config(const std::vector<std::string>& storageBackends, const size_t kvcacheBlockSize,
                const bool transferEnable)
             : storageBackends{storageBackends}, kvcacheBlockSize{kvcacheBlockSize},
               transferEnable{transferEnable}, transferDeviceId{-1}, transferStreamNumber{32},
               transferIoSize{262144}, transferBufferNumber{512}, transferTimeoutMs{30000},
-              tempDumpDirEnable{false}, hotnessEnable{true}, hotnessInterval{60}
+              tempDumpDirEnable{false}, hotnessEnable{true}, hotnessInterval{60},
+              storageCapacity{0}, recycleEnable{true}, recycleThresholdRatio{0.7f}
         {
         }
     };

ucm/store/nfsstore/cc/domain/space/space_layout.h

@@ -24,18 +24,25 @@
 #ifndef UNIFIEDCACHE_SPACE_LAYOUT_H
 #define UNIFIEDCACHE_SPACE_LAYOUT_H
 
+#include <memory>
 #include <string>
 #include <vector>
 #include "status/status.h"
 
 namespace UC {
 
 class SpaceLayout {
+public:
+    struct DataIterator;
 public:
     virtual ~SpaceLayout() = default;
     virtual Status Setup(const std::vector<std::string>& storageBackends) = 0;
     virtual std::string DataFileParent(const std::string& blockId, bool activated) const = 0;
     virtual std::string DataFilePath(const std::string& blockId, bool activated) const = 0;
+    virtual std::string ClusterPropertyFilePath() const = 0;
+    virtual std::shared_ptr<DataIterator> CreateFilePathIterator() const = 0;
+    virtual std::string NextDataFilePath(std::shared_ptr<DataIterator> iter) const = 0;
+    virtual bool IsActivatedFile(const std::string& filePath) const = 0;
 };
 
 } // namespace UC

ucm/store/nfsstore/cc/domain/space/space_manager.cc

@@ -40,7 +40,8 @@ std::unique_ptr<SpaceLayout> MakeSpaceLayout(const bool tempDumpDirEnable)
 }
 
 Status SpaceManager::Setup(const std::vector<std::string>& storageBackends, const size_t blockSize,
-                           const bool tempDumpDirEnable)
+                           const bool tempDumpDirEnable, const size_t storageCapacity,
+                           const bool recycleEnable, const float recycleThresholdRatio)
 {
     if (blockSize == 0) {
         UC_ERROR("Invalid block size({}).", blockSize);
@@ -50,20 +51,34 @@ Status SpaceManager::Setup(const std::vector<std::string>& storageBackends, cons
     if (!this->layout_) { return Status::OutOfMemory(); }
     auto status = this->layout_->Setup(storageBackends);
     if (status.Failure()) { return status; }
+    status = this->property_.Setup(this->layout_->ClusterPropertyFilePath());
+    if (recycleEnable && storageCapacity > 0) {
+        auto totalBlocks = storageCapacity / blockSize;
+        status = this->recycle_.Setup(this->GetSpaceLayout(), totalBlocks, [this] {
+            this->property_.DecreaseCapacity(this->blockSize_);
+        });
+        if (status.Failure()) { return status; }
+    }
+    if (status.Failure()) { return status; }
     this->blockSize_ = blockSize;
+    this->capacity_ = storageCapacity;
+    this->recycleEnable_ = recycleEnable;
+    this->capacityRecycleThreshold_ = static_cast<size_t>(storageCapacity * recycleThresholdRatio);
     return Status::OK();
 }
 
-Status SpaceManager::NewBlock(const std::string& blockId) const
+Status SpaceManager::NewBlock(const std::string& blockId)
 {
+    Status status = this->CapacityCheck();
+    if (status.Failure()) { return status; }
     constexpr auto activated = true;
     auto parent = File::Make(this->layout_->DataFileParent(blockId, activated));
     auto file = File::Make(this->layout_->DataFilePath(blockId, activated));
     if (!parent || !file) {
         UC_ERROR("Failed to new block({}).", blockId);
         return Status::OutOfMemory();
     }
-    auto status = parent->MkDir();
+    status = parent->MkDir();
     if (status == Status::DuplicateKey()) { status = Status::OK(); }
     if (status.Failure()) {
         UC_ERROR("Failed({}) to new block({}).", status, blockId);
@@ -88,10 +103,11 @@ Status SpaceManager::NewBlock(const std::string& blockId) const
         UC_ERROR("Failed({}) to new block({}).", status, blockId);
         return status;
     }
+    this->property_.IncreaseCapacity(this->blockSize_);
     return Status::OK();
 }
 
-Status SpaceManager::CommitBlock(const std::string& blockId, bool success) const
+Status SpaceManager::CommitBlock(const std::string& blockId, bool success)
 {
     const auto activatedParent = this->layout_->DataFileParent(blockId, true);
     const auto activatedFile = this->layout_->DataFilePath(blockId, true);
@@ -112,6 +128,7 @@ Status SpaceManager::CommitBlock(const std::string& blockId, bool success) const
     if (status.Failure()) {
         UC_ERROR("Failed({}) to {} block({}).", status, success ? "commit" : "cancel", blockId);
     }
+    this->property_.DecreaseCapacity(this->blockSize_);
     return status;
 }
 
@@ -136,4 +153,20 @@ bool SpaceManager::LookupBlock(const std::string& blockId) const
 
 const SpaceLayout* SpaceManager::GetSpaceLayout() const { return this->layout_.get(); }
 
+Status SpaceManager::CapacityCheck()
+{
+    if (this->capacity_ == 0) { return Status::OK(); }
+    
+    const size_t used = this->property_.GetCapacity();
+    if (this->recycleEnable_ && used >= this->capacityRecycleThreshold_) {
+        this->recycle_.Trigger();
+    }
+    if (used > this->capacity_ - this->blockSize_) {
+        UC_ERROR("Capacity is not enough, capacity: {}, current: {}, block size: {}.", 
+                 this->capacity_, used, this->blockSize_);
+        return Status::NoSpace();
+    }
+    return Status::OK();
+}
+
 } // namespace UC

ucm/store/nfsstore/cc/domain/space/space_manager.h

@@ -26,22 +26,32 @@
 
 #include <memory>
 #include "space_layout.h"
+#include "space_property.h"
 #include "status/status.h"
+#include "space_recycle.h"
 
 namespace UC {
 
 class SpaceManager {
 public:
     Status Setup(const std::vector<std::string>& storageBackends, const size_t blockSize,
-                 const bool tempDumpDirEnable);
-    Status NewBlock(const std::string& blockId) const;
-    Status CommitBlock(const std::string& blockId, bool success = true) const;
+                 const bool tempDumpDirEnable, const size_t storageCapacity = 0,
+                 const bool recycleEnable = false, const float recycleThresholdRatio = 0.7f);
+    Status NewBlock(const std::string& blockId);
+    Status CommitBlock(const std::string& blockId, bool success = true);
     bool LookupBlock(const std::string& blockId) const;
     const SpaceLayout* GetSpaceLayout() const;
 
+private:
+    Status CapacityCheck();
 private:
     std::unique_ptr<SpaceLayout> layout_;
+    SpaceProperty property_;
+    SpaceRecycle recycle_;
     size_t blockSize_;
+    size_t capacity_;
+    bool recycleEnable_;
+    size_t capacityRecycleThreshold_;
 };
 
 } // namespace UC