[bugfix] clean up shm remnants to fix "BufferOut" error in PCStore (#512)

PCStore uses shared memory to share data in DRAM. If the service terminates abnormally and residual files of the shared data are not cleaned up, it will cause newly started services to report BufferOut errors.

Author: mag1c-h

ucm/store/pcstore/cc/api/pcstore.cc

@@ -37,11 +37,15 @@ class PcStoreImpl : public PcStore {
         auto status = this->spaceMgr_.Setup(config.storageBackends, config.kvcacheBlockSize);
         if (status.Failure()) { return status.Underlying(); }
         if (config.transferEnable) {
+            if (config.uniqueId.empty()) {
+                UC_ERROR("UniqueId is required.");
+                return Status::InvalidParam().Underlying();
+            }
             status = this->transMgr_.Setup(
                 config.transferLocalRankSize, config.transferDeviceId, config.transferStreamNumber,
                 config.kvcacheBlockSize, config.transferIoSize, config.transferIoDirect,
                 config.transferBufferNumber, this->spaceMgr_.GetSpaceLayout(),
-                config.transferTimeoutMs, config.transferScatterGatherEnable);
+                config.transferTimeoutMs, config.transferScatterGatherEnable, config.uniqueId);
             if (status.Failure()) { return status.Underlying(); }
         }
         this->ShowConfig(config);
@@ -86,6 +90,7 @@ class PcStoreImpl : public PcStore {
         UC_INFO("Set UC::BlockSize to {}.", config.kvcacheBlockSize);
         UC_INFO("Set UC::TransferEnable to {}.", config.transferEnable);
         if (!config.transferEnable) { return; }
+        UC_INFO("Set UC::UniqueId to {}.", config.uniqueId);
         UC_INFO("Set UC::IoSize to {}.", config.transferIoSize);
         UC_INFO("Set UC::IoDirect to {}.", config.transferIoDirect);
         UC_INFO("Set UC::LocalRankSize to {}.", config.transferLocalRankSize);
@@ -112,4 +117,4 @@ int32_t PcStore::Setup(const Config& config)
     return impl->Setup(config);
 }
 
-} // namespace UC
+}  // namespace UC

ucm/store/pcstore/cc/api/pcstore.h

@@ -35,6 +35,7 @@ class PcStore : CCStore<TransTask> {
         std::vector<std::string> storageBackends;
         size_t kvcacheBlockSize;
         bool transferEnable;
+        std::string uniqueId{};
         size_t transferIoSize{262144};
         bool transferIoDirect{false};
         size_t transferLocalRankSize{1};
@@ -46,7 +47,8 @@ class PcStore : CCStore<TransTask> {
 
         Config(const std::vector<std::string>& storageBackends, const size_t kvcacheBlockSize,
                const bool transferEnable)
-            : storageBackends{storageBackends}, kvcacheBlockSize{kvcacheBlockSize},
+            : storageBackends{storageBackends},
+              kvcacheBlockSize{kvcacheBlockSize},
               transferEnable{transferEnable}
         {
         }
@@ -87,6 +89,6 @@ class PcStore : CCStore<TransTask> {
     PcStore* impl_{nullptr};
 };
 
-} // namespace UC
+}  // namespace UC
 
 #endif

ucm/store/pcstore/cc/domain/file/file.cc

@@ -94,4 +94,4 @@ void File::ShmUnlink(const std::string& path) { FileImpl{path}.ShmUnlink(); }
 
 void File::Remove(const std::string& path) { FileImpl{path}.Remove(); }
 
-} // namespace UC
+}  // namespace UC

ucm/store/pcstore/cc/domain/file/file.h

@@ -47,6 +47,6 @@ class File {
     static void Remove(const std::string& path);
 };
 
-} // namespace UC
+}  // namespace UC
 
 #endif

ucm/store/pcstore/cc/domain/file/ifile.h

@@ -76,6 +76,6 @@ class IFile {
     std::string path_;
 };
 
-} // namespace UC
+}  // namespace UC
 
 #endif

ucm/store/pcstore/cc/domain/file/posix_file.cc

@@ -25,8 +25,8 @@
 #include <sys/mman.h>
 #include <sys/stat.h>
 #include <sys/xattr.h>
-#include <utime.h>
 #include <unistd.h>
+#include <utime.h>
 #include "logger/logger.h"
 
 namespace UC {
@@ -242,4 +242,4 @@ Status PosixFile::UpdateTime()
     return Status::OK();
 }
 
-} // namespace UC
+}  // namespace UC

ucm/store/pcstore/cc/domain/file/posix_file.h

@@ -53,6 +53,6 @@ class PosixFile : public IFile {
     int32_t handle_;
 };
 
-} // namespace UC
+}  // namespace UC
 
-#endif // UNIFIEDCACHE_POSIX_FILE_H
+#endif  // UNIFIEDCACHE_POSIX_FILE_H

ucm/store/pcstore/cc/domain/trans/share_buffer.cc

@@ -24,6 +24,7 @@
 #include "share_buffer.h"
 #include <atomic>
 #include <chrono>
+#include <filesystem>
 #include <thread>
 #include <unistd.h>
 #include "file/file.h"
@@ -97,21 +98,36 @@ struct ShareBufferHeader {
     ShareBlockHeader headers[0];
 };
 
-inline std::string GenShareBufferName(const size_t blockSize, const size_t blockNumber,
-                                      const bool ioDirect, const size_t nSharer)
+const inline std::string& ShmPrefix() noexcept
 {
-    return fmt::format("uc.buf-{}-{}-{}-{:04x}", blockSize, blockNumber, ioDirect, nSharer);
+    static std::string prefix{"uc_shm_pcstore_"};
+    return prefix;
+}
+void CleanUpShmFileExceptMe(const std::string& me)
+{
+    namespace fs = std::filesystem;
+    std::string_view prefix = ShmPrefix();
+    fs::path shmDir = "/dev/shm";
+    if (!fs::exists(shmDir)) { return; }
+    for (const auto& entry : fs::directory_iterator(shmDir)) {
+        const auto& name = entry.path().filename().string();
+        if (entry.is_regular_file() && (name.compare(0, prefix.length(), prefix) == 0) &&
+            name != me) {
+            fs::remove(entry.path());
+        }
+    }
 }
 
 Status ShareBuffer::Setup(const size_t blockSize, const size_t blockNumber, const bool ioDirect,
-                          const size_t nSharer)
+                          const size_t nSharer, const std::string& uniqueId)
 {
     this->blockSize_ = blockSize;
     this->blockNumber_ = blockNumber;
     this->ioDirect_ = ioDirect;
     this->nSharer_ = nSharer;
     this->addr_ = nullptr;
-    this->shmName_ = GenShareBufferName(blockSize, blockNumber, ioDirect, nSharer);
+    this->shmName_ = ShmPrefix() + uniqueId;
+    CleanUpShmFileExceptMe(this->shmName_);
     auto file = File::Make(this->shmName_);
     if (!file) { return Status::OutOfMemory(); }
     auto flags = IFile::OpenFlag::CREATE | IFile::OpenFlag::EXCL | IFile::OpenFlag::READ_WRITE;
@@ -305,4 +321,4 @@ uintptr_t ShareBuffer::Reader::GetData()
     return (uintptr_t)header->Data();
 }
 
-} // namespace UC
+}  // namespace UC

ucm/store/pcstore/cc/domain/trans/share_buffer.h

@@ -49,7 +49,11 @@ class ShareBuffer {
     private:
         Reader(const std::string& block, const std::string& path, const size_t length,
                const bool ioDirect, const size_t nSharer, void* addr)
-            : block_{block}, path_{path}, length_{length}, ioDirect_{ioDirect}, nSharer_{nSharer},
+            : block_{block},
+              path_{path},
+              length_{length},
+              ioDirect_{ioDirect},
+              nSharer_{nSharer},
               addr_{addr}
         {
         }
@@ -58,7 +62,7 @@ class ShareBuffer {
 
 public:
     Status Setup(const size_t blockSize, const size_t blockNumber, const bool ioDirect,
-                 const size_t nSharer);
+                 const size_t nSharer, const std::string& uniqueId);
     ~ShareBuffer();
     std::shared_ptr<Reader> MakeReader(const std::string& block, const std::string& path);
 
@@ -80,6 +84,6 @@ class ShareBuffer {
     void* addr_;
 };
 
-} // namespace UC
+}  // namespace UC
 
 #endif

ucm/store/pcstore/cc/domain/trans/trans_manager.cc

@@ -29,12 +29,13 @@ namespace UC {
 Status TransManager::Setup(const size_t rankSize, const int32_t deviceId, const size_t streamNumber,
                            const size_t blockSize, const size_t ioSize, const bool ioDirect,
                            const size_t bufferNumber, const SpaceLayout* layout,
-                           const size_t timeoutMs, const bool scatterGatherEnable)
+                           const size_t timeoutMs, const bool scatterGatherEnable,
+                           const std::string& uniqueId)
 {
     auto s = Status::OK();
     if (rankSize > 1) {
         s = this->shareQueue_.Setup(rankSize, deviceId, streamNumber, blockSize, ioSize, ioDirect,
-                                    bufferNumber, layout, &this->failureSet_);
+                                    bufferNumber, layout, &this->failureSet_, uniqueId);
         if (s.Failure()) { return s; }
     }
     s = this->queue_.Setup(deviceId, streamNumber, blockSize, ioSize, ioDirect, bufferNumber,
@@ -115,4 +116,4 @@ Status TransManager::Check(const size_t taskId, bool& finish) noexcept
     return Status::OK();
 }
 
-} // namespace UC
+}  // namespace UC