[Feat] Call scatter gather interface in dramstore (#324)

* refactor: reusable transport abstraction & optimized NSFStore pipeline

* add memory pool

* rewrite dramstore (1st version)

* fix

* fix

* fix

* fix

* fix

* add comment for later development

* fix

* fix

* fix

* fix

* fix

* complete dramstore_connector.py

* fix

* fix CMakeLists.txt

* fix

* fix

* fix

* fix

* rewrite dram test:dump (1/2)

* format

* fix

* rewrite dram test:fetch (2/2)

* modify dram test script

* fix

* fix

* naive try: add device intf into memory pool

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* update memPool intf

* fix

* fix

* fix

* try reconstruct threadPool in dramstore (naive version)

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* remove redundant comments

---------

Co-authored-by: Mag1c.H <hemajun815@163.com>

Author: ChenyuZhu1

test/test_ucm_dram.py

@@ -1,7 +1,10 @@
 file(GLOB_RECURSE UCMSTORE_DRAM_CC_SOURCE_FILES "./cc/*.cc")
 add_library(dramstore STATIC ${UCMSTORE_DRAM_CC_SOURCE_FILES})
-target_include_directories(dramstore PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/cc)
-target_link_libraries(dramstore PUBLIC storeinfra storetask)
+target_include_directories(dramstore PUBLIC 
+    ${CMAKE_CURRENT_SOURCE_DIR}/cc/api
+    ${CMAKE_CURRENT_SOURCE_DIR}/cc/domain
+)
+target_link_libraries(dramstore PUBLIC storeinfra storedevice storetask)
 
 file(GLOB_RECURSE UCMSTORE_DRAM_CPY_SOURCE_FILES "./cpy/*.cc")
 pybind11_add_module(ucmdramstore ${UCMSTORE_DRAM_CPY_SOURCE_FILES})

ucm/store/dramstore/CMakeLists.txt

@@ -24,27 +24,70 @@
 #include "dramstore.h"
 #include "logger/logger.h"
 #include "status/status.h"
+#include "trans/dram_trans_manager.h"
+#include "memory/memory_pool.h"
 
 namespace UC {
 
 class DRAMStoreImpl : public DRAMStore {
 public:
-    int32_t Setup(const size_t ioSize, const size_t capacity, const int32_t deviceId) { return -1; }
-    int32_t Alloc(const std::string& block) override { return -1; }
-    bool Lookup(const std::string& block) override { return false; }
-    void Commit(const std::string& block, const bool success) override {}
+    int32_t Setup(const Config& config) {
+        auto status = this->memPool_.Setup(config.deviceId, config.capacity, config.blockSize);
+        if (status.Failure()) {
+            UC_ERROR("Failed({}) to setup MemoryPool.", status);
+            return status.Underlying();
+        }
+        status = this->transMgr_.Setup(config.deviceId, config.streamNumber, &this->memPool_, config.timeoutMs);
+        if (status.Failure()) {
+            UC_ERROR("Failed({}) to setup TsfTaskManager.", status);
+            return status.Underlying();
+        }
+        return Status::OK().Underlying();
+    }
+    int32_t Alloc(const std::string& block) override { return this->memPool_.NewBlock(block).Underlying(); }
+    bool Lookup(const std::string& block) override { return this->memPool_.LookupBlock(block); }
+    void Commit(const std::string& block, const bool success) override { this->memPool_.CommitBlock(block, success).Underlying(); }
     std::list<int32_t> Alloc(const std::list<std::string>& blocks) override
     {
-        return std::list<int32_t>();
+        std::list<int32_t> results;
+        for (const auto &block : blocks) {
+            results.emplace_back(this->Alloc(block));
+        }
+        return results;
     }
     std::list<bool> Lookup(const std::list<std::string>& blocks) override
     {
-        return std::list<bool>();
+        std::list<bool> founds;
+        for (const auto &block : blocks) {
+            founds.emplace_back(this->Lookup(block));
+        }
+        return founds;
+    }
+    void Commit(const std::list<std::string>& blocks, const bool success) override {
+        for (const auto &block : blocks) {
+            this->Commit(block, success);
+        }
+    }
+    size_t Submit(Task&& task) override {
+        auto taskId = Task::invalid;
+        auto status = this->transMgr_.Submit(std::move(task), taskId);
+        if (status.Failure()) { taskId = Task::invalid; }
+        return taskId;    }
+
+    int32_t Wait(const size_t task) override {
+        return this->transMgr_.Wait(task).Underlying();
+    }
+
+    int32_t Check(const size_t task, bool& finish) override {
+        return this->transMgr_.Check(task, finish).Underlying();
     }
-    void Commit(const std::list<std::string>& blocks, const bool success) override {}
-    size_t Submit(Task&& task) override { return 0; }
-    int32_t Wait(const size_t task) override { return -1; }
-    int32_t Check(const size_t task, bool& finish) override { return -1; }
+
+
+private:
+
+    DramTransManager transMgr_;
+    MemoryPool memPool_;
+
 };
 
 int32_t DRAMStore::Setup(const Config& config)
@@ -55,7 +98,7 @@ int32_t DRAMStore::Setup(const Config& config)
         return Status::OutOfMemory().Underlying();
     }
     this->impl_ = impl;
-    return impl->Setup(config.ioSize, config.capacity, config.deviceId);
+    return impl->Setup(config);
 }
 
 } // namespace UC

ucm/store/dramstore/cc/api/dramstore.cc

@@ -31,11 +31,13 @@ namespace UC {
 class DRAMStore : public CCStore {
 public:
     struct Config {
-        size_t ioSize;
         size_t capacity;
+        size_t blockSize;
         int32_t deviceId;
-        Config(const size_t ioSize, const size_t capacity)
-            : ioSize{ioSize}, capacity{capacity}, deviceId{-1}
+        size_t streamNumber;
+        size_t timeoutMs;
+        Config(const size_t capacity, const size_t blockSize, const int32_t deviceId, const size_t streamNumber, const size_t timeoutMs)
+            : capacity{capacity}, blockSize{blockSize}, deviceId{deviceId}, streamNumber{streamNumber}, timeoutMs{timeoutMs}
         {
         }
     };