GDS-fixed

FengHao · FengHao · commit cb681d5a3a80 · 2025-10-29T05:35:07.000-07:00
diff --git a/ucm/store/device/cuda/cuda_device.cu b/ucm/store/device/cuda/cuda_device.cu
@@ -32,7 +32,7 @@
 #include <cstring>
 #include <unordered_map>
 #include <cstdlib>
-#include "infra/template/sharded_handle_recorder.h"
+#include "infra/template/handle_recorder.h"
 
 #define CUDA_TRANS_UNIT_SIZE (sizeof(uint64_t) * 2)
 #define CUDA_TRANS_BLOCK_NUMBER (32)
@@ -311,13 +311,17 @@ private:
     cudaStream_t stream_;
 };
 
+void DeviceFactory::Setup(bool useDirect)
+{
+    if (useDirect) {
+            CudaDevice::InitGdsOnce();
+    }
+}
+
 std::unique_ptr<IDevice> DeviceFactory::Make(const int32_t deviceId, const size_t bufferSize,
-                                             const size_t bufferNumber, bool transferUseDirect)
+                                             const size_t bufferNumber)
 {
     try {
-        if (transferUseDirect) {
-            CudaDevice::InitGdsOnce();
-        }
         return std::make_unique<CudaDevice>(deviceId, bufferSize, bufferNumber);
     } catch (const std::exception& e) {
         UC_ERROR("Failed({}) to make cuda device({},{},{}).", e.what(), deviceId, bufferSize,
diff --git a/ucm/store/device/idevice.h b/ucm/store/device/idevice.h
@@ -61,8 +61,9 @@ class IDevice {
 
 class DeviceFactory {
 public:
+    static void Setup(bool useDirect = false);
     static std::unique_ptr<IDevice> Make(const int32_t deviceId, const size_t bufferSize,
-                                         const size_t bufferNumber, bool transferUseDirect = false);
+                                         const size_t bufferNumber);
 };
 
 } // namespace UC
diff --git a/ucm/store/infra/template/handle_recorder.h b/ucm/store/infra/template/handle_recorder.h
diff --git a/ucm/store/nfsstore/cc/api/nfsstore.cc b/ucm/store/nfsstore/cc/api/nfsstore.cc
@@ -44,7 +44,7 @@ class NFSStoreImpl : public NFSStore {
             status =
                 this->transMgr_.Setup(config.transferDeviceId, config.transferStreamNumber,
                                       config.transferIoSize, config.transferBufferNumber,
-                                      this->spaceMgr_.GetSpaceLayout(), config.transferTimeoutMs, config.transferUseDirect);
+                                      this->spaceMgr_.GetSpaceLayout(), config.transferTimeoutMs, config.useDirect);
             if (status.Failure()) {
                 UC_ERROR("Failed({}) to setup TsfTaskManager.", status);
                 return status.Underlying();
diff --git a/ucm/store/nfsstore/cc/api/nfsstore.h b/ucm/store/nfsstore/cc/api/nfsstore.h
@@ -43,14 +43,14 @@ class NFSStore : public CCStore {
         bool tempDumpDirEnable;
         bool hotnessEnable;
         size_t hotnessInterval;
-        bool transferUseDirect;
+        bool useDirect;
 
         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}, transferUseDirect{false}
+              tempDumpDirEnable{false}, hotnessEnable{true}, hotnessInterval{60}, useDirect{false}
         {
         }
     };
diff --git a/ucm/store/nfsstore/cc/domain/trans/directstorage_queue.cc b/ucm/store/nfsstore/cc/domain/trans/directstorage_queue.cc
@@ -31,14 +31,14 @@
 namespace UC {
 
 Status DirectStorageQueue::Setup(const int32_t deviceId, const size_t bufferSize, const size_t bufferNumber,
-                         TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool transferUseDirect)
+                         TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool useDirect)
 {
     this->deviceId_ = deviceId;
     this->bufferSize_ = bufferSize;
     this->bufferNumber_ = bufferNumber;
     this->failureSet_ = failureSet;
     this->layout_ = layout;
-    this->transferUseDirect_ = transferUseDirect;
+    this->useDirect = useDirect;
     auto success =
         this->backend_.SetWorkerInitFn([this](auto& device) { return this->Init(device); })
             .SetWorkerFn([this](auto& shard, const auto& device) { this->Work(shard, device); })
@@ -52,7 +52,8 @@ void DirectStorageQueue::Push(std::list<Task::Shard>& shards) noexcept { this->b
 bool DirectStorageQueue::Init(Device& device)
 {
     if (this->deviceId_ < 0) { return true; }
-    device = DeviceFactory::Make(this->deviceId_, this->bufferSize_, this->bufferNumber_, this->transferUseDirect_);
+    DeviceFactory::Setup(useDirect);
+    device = DeviceFactory::Make(this->deviceId_, this->bufferSize_, this->bufferNumber_);
     if (!device) { return false; }
     return device->Setup().Success();
 }
diff --git a/ucm/store/nfsstore/cc/domain/trans/directstorage_queue.h b/ucm/store/nfsstore/cc/domain/trans/directstorage_queue.h
@@ -34,7 +34,7 @@
 #include <unordered_map>
 #include <mutex>
 #include <string>
-#include "infra/template/sharded_handle_recorder.h"
+#include "infra/template/handle_recorder.h"
 
 namespace UC {
 
@@ -46,11 +46,11 @@ class DirectStorageQueue : public TaskQueue {
     TaskSet* failureSet_{nullptr};
     const SpaceLayout* layout_{nullptr};
     ThreadPool<Task::Shard, Device> backend_{};
-    bool transferUseDirect_{false};
+    bool useDirect{false};
 
 public:
     Status Setup(const int32_t deviceId, const size_t bufferSize, const size_t bufferNumber,
-                 TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool transferUseDirect);
+                 TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool useDirect);
     void Push(std::list<Task::Shard>& shards) noexcept override;
 
 private:
diff --git a/ucm/store/nfsstore/cc/domain/trans/posix_queue.cc b/ucm/store/nfsstore/cc/domain/trans/posix_queue.cc
@@ -35,14 +35,14 @@ bool IsAligned(const T value)
 }
 
 Status PosixQueue::Setup(const int32_t deviceId, const size_t bufferSize, const size_t bufferNumber,
-                         TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool transferUseDirect)
+                         TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool useDirect)
 {
     this->deviceId_ = deviceId;
     this->bufferSize_ = bufferSize;
     this->bufferNumber_ = bufferNumber;
     this->failureSet_ = failureSet;
     this->layout_ = layout;
-    this->transferUseDirect_ = transferUseDirect;
+    this->useDirect = useDirect;
     auto success =
         this->backend_.SetWorkerInitFn([this](auto& device) { return this->Init(device); })
             .SetWorkerFn([this](auto& shard, const auto& device) { this->Work(shard, device); })
@@ -56,7 +56,7 @@ void PosixQueue::Push(std::list<Task::Shard>& shards) noexcept { this->backend_.
 bool PosixQueue::Init(Device& device)
 {
     if (this->deviceId_ < 0) { return true; }
-    device = DeviceFactory::Make(this->deviceId_, this->bufferSize_, this->bufferNumber_, this->transferUseDirect_);
+    device = DeviceFactory::Make(this->deviceId_, this->bufferSize_, this->bufferNumber_);
     if (!device) { return false; }
     return device->Setup().Success();
 }
diff --git a/ucm/store/nfsstore/cc/domain/trans/posix_queue.h b/ucm/store/nfsstore/cc/domain/trans/posix_queue.h
@@ -41,11 +41,11 @@ class PosixQueue : public TaskQueue {
     TaskSet* failureSet_{nullptr};
     const SpaceLayout* layout_{nullptr};
     ThreadPool<Task::Shard, Device> backend_{};
-    bool transferUseDirect_{false};
+    bool useDirect{false};
 
 public:
     Status Setup(const int32_t deviceId, const size_t bufferSize, const size_t bufferNumber,
-                 TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool transferUseDirect);
+                 TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool useDirect);
     void Push(std::list<Task::Shard>& shards) noexcept override;
 
 private:
diff --git a/ucm/store/nfsstore/cc/domain/trans/trans_manager.h b/ucm/store/nfsstore/cc/domain/trans/trans_manager.h
@@ -33,16 +33,16 @@ namespace UC {
 class TransManager : public TaskManager {
 public:
     Status Setup(const int32_t deviceId, const size_t streamNumber, const size_t ioSize,
-                 const size_t bufferNumber, const SpaceLayout* layout, const size_t timeoutMs, bool transferUseDirect)
+                 const size_t bufferNumber, const SpaceLayout* layout, const size_t timeoutMs, bool useDirect)
     {
         this->timeoutMs_ = timeoutMs;
         auto status = Status::OK();
-        if(transferUseDirect)
+        if(useDirect)
         {
             for (size_t i = 0; i < streamNumber; i++) {
                 auto q = std::make_shared<DirectStorageQueue>();
                 status =
-                    q->Setup(deviceId, ioSize, bufferNumber, &this->failureSet_, layout, timeoutMs, transferUseDirect);
+                    q->Setup(deviceId, ioSize, bufferNumber, &this->failureSet_, layout, timeoutMs, useDirect);
                 if (status.Failure()) { break; }
                 this->queues_.emplace_back(std::move(q));
             }
@@ -52,7 +52,7 @@ class TransManager : public TaskManager {
             for (size_t i = 0; i < streamNumber; i++) {
                 auto q = std::make_shared<PosixQueue>();
                 status =
-                    q->Setup(deviceId, ioSize, bufferNumber, &this->failureSet_, layout, timeoutMs, transferUseDirect);
+                    q->Setup(deviceId, ioSize, bufferNumber, &this->failureSet_, layout, timeoutMs, useDirect);
                 if (status.Failure()) { break; }
                 this->queues_.emplace_back(std::move(q));
             }
diff --git a/ucm/store/nfsstore/cpy/nfsstore.py.cc b/ucm/store/nfsstore/cpy/nfsstore.py.cc
@@ -125,7 +125,7 @@ PYBIND11_MODULE(ucmnfsstore, module)
     config.def_readwrite("tempDumpDirEnable", &UC::NFSStorePy::Config::tempDumpDirEnable);
     config.def_readwrite("hotnessEnable", &UC::NFSStorePy::Config::hotnessEnable);
     config.def_readwrite("hotnessInterval", &UC::NFSStorePy::Config::hotnessInterval);
-    config.def_readwrite("transferUseDirect", &UC::NFSStorePy::Config::transferUseDirect);
+    config.def_readwrite("useDirect", &UC::NFSStorePy::Config::useDirect);
     store.def(py::init<>());
     store.def("CCStoreImpl", &UC::NFSStorePy::CCStoreImpl);
     store.def("Setup", &UC::NFSStorePy::Setup);
diff --git a/ucm/store/nfsstore/nfsstore_connector.py b/ucm/store/nfsstore/nfsstore_connector.py
@@ -45,15 +45,15 @@ def __init__(self, config: Dict):
         ]
         block_size = int(config["kv_block_size"])
         transfer_enable = True if config["role"] == "worker" else False
-        transferUseDirect = config.get("transferUseDirect", False)
+        useDirect = config.get("useDirect", False)
         transferStreamNumber = config.get("transferStreamNumber", 32)
         param = ucmnfsstore.NFSStore.Config(
             storage_backends, block_size, transfer_enable
         )
         if transfer_enable:
             param.transferDeviceId = config["device"]
             param.transferIoSize = config["io_size"]
-            param.transferUseDirect = transferUseDirect
+            param.useDirect = useDirect
             param.transferStreamNumber = transferStreamNumber
         ret = self.store.Setup(param)
         if ret != 0:

Original file line number	Diff line number	Diff line change
`@@ -35,14 +35,14 @@ bool IsAligned(const T value)`
`35`	`35`	`}`
`36`	`36`
`37`	`37`	`Status PosixQueue::Setup(const int32_t deviceId, const size_t bufferSize, const size_t bufferNumber,`
`38`		`- TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool transferUseDirect)`
	`38`	`+ TaskSet* failureSet, const SpaceLayout* layout, const size_t timeoutMs, bool useDirect)`
`39`	`39`	`{`
`40`	`40`	`this->deviceId_ = deviceId;`
`41`	`41`	`this->bufferSize_ = bufferSize;`
`42`	`42`	`this->bufferNumber_ = bufferNumber;`
`43`	`43`	`this->failureSet_ = failureSet;`
`44`	`44`	`this->layout_ = layout;`
`45`		`- this->transferUseDirect_ = transferUseDirect;`
	`45`	`+ this->useDirect = useDirect;`
`46`	`46`	`auto success =`
`47`	`47`	`this->backend_.SetWorkerInitFn([this](auto& device) { return this->Init(device); })`
`48`	`48`	`.SetWorkerFn([this](auto& shard, const auto& device) { this->Work(shard, device); })`
`@@ -56,7 +56,7 @@ void PosixQueue::Push(std::list<Task::Shard>& shards) noexcept { this->backend_.`
`56`	`56`	`bool PosixQueue::Init(Device& device)`
`57`	`57`	`{`
`58`	`58`	`if (this->deviceId_ < 0) { return true; }`
`59`		`- device = DeviceFactory::Make(this->deviceId_, this->bufferSize_, this->bufferNumber_, this->transferUseDirect_);`
	`59`	`+ device = DeviceFactory::Make(this->deviceId_, this->bufferSize_, this->bufferNumber_);`
`60`	`60`	`if (!device) { return false; }`
`61`	`61`	`return device->Setup().Success();`
`62`	`62`	`}`