[feat][mdsv2] Add truncate for open/setattr.

Author: rock-git

src/client/vfs/meta/v2/filesystem.cpp

@@ -14,6 +14,8 @@
 
 #include "client/vfs/meta/v2/filesystem.h"
 
+#include <fcntl.h>
+
 #include <cstdint>
 #include <string>
 #include <utility>
@@ -336,6 +338,10 @@ Status MDSV2FileSystem::MkNod(Ino parent, const std::string& name, uint32_t uid,
 Status MDSV2FileSystem::Open(Ino ino, int flags, uint64_t fh) {
   LOG(INFO) << fmt::format("[meta.{}] open ino({}).", name_, ino);
 
+  if ((flags & O_TRUNC) && !(flags & O_WRONLY || flags & O_RDWR)) {
+    return Status::NoPermission("O_TRUNC without O_WRONLY or O_RDWR");
+  }
+
   std::string session_id;
   auto status = mds_client_->Open(ino, flags, session_id);
   if (!status.ok()) {

src/client/vfs/meta/v2/mds_client.cpp

@@ -384,6 +384,8 @@ Status MDSClient::Open(Ino ino, int flags, std::string& session_id) {
 
   session_id = response.session_id();
 
+  parent_memo_->UpsertVersion(ino, response.version());
+
   return Status::OK();
 }
 

src/mdsv2/common/type.h

@@ -29,13 +29,15 @@ namespace mdsv2 {
 using Ino = uint64_t;
 using AttrType = pb::mdsv2::Inode;
 using DentryType = pb::mdsv2::Dentry;
+using SliceType = pb::mdsv2::Slice;
 using ChunkType = pb::mdsv2::Chunk;
 using FsInfoType = pb::mdsv2::FsInfo;
 using TrashSliceList = pb::mdsv2::TrashSliceList;
 using QuotaEntry = pb::mdsv2::Quota;
 using UsageEntry = pb::mdsv2::Usage;
 using MdsEntry = pb::mdsv2::MDS;
 using ClientEntry = pb::mdsv2::Client;
+using FileSessionEntry = pb::mdsv2::FileSession;
 
 inline bool IsDir(Ino ino) { return (ino & 1) == 1; }
 inline bool IsFile(Ino ino) { return (ino & 1) == 0; }

src/mdsv2/filesystem/file_session.cc

@@ -14,19 +14,18 @@
 
 #include "mdsv2/filesystem/file_session.h"
 
-#include <fmt/format.h>
-#include <glog/logging.h>
-
 #include <cstdint>
 #include <string>
 #include <utility>
 
 #include "dingofs/error.pb.h"
-#include "mdsv2/common/codec.h"
+#include "fmt/format.h"
 #include "mdsv2/common/logging.h"
 #include "mdsv2/common/status.h"
-#include "mdsv2/storage/storage.h"
+#include "mdsv2/common/type.h"
+#include "mdsv2/filesystem/store_operation.h"
 #include "utils/concurrent/concurrent.h"
+#include "utils/uuid.h"
 
 namespace dingofs {
 namespace mdsv2 {
@@ -39,26 +38,20 @@ static const std::string kFileSessionCacheCountMetricsName = "dingofs_file_sessi
 static const std::string kFileSessionTatalCountMetricsName = "dingofs_file_session_total_count";
 static const std::string kFileSessionCountMetricsName = "dingofs_file_session_count";
 
-std::string FileSession::EncodeKey(uint32_t fs_id, uint64_t ino, const std::string& session_id) {
-  return MetaCodec::EncodeFileSessionKey(fs_id, ino, session_id);
-}
-
-std::string FileSession::EncodeKey() const { return MetaCodec::EncodeFileSessionKey(fs_id_, ino_, session_id_); }
+static FileSessionPtr NewFileSession(uint64_t ino, const std::string& client_id) {
+  auto file_session = std::make_shared<FileSessionEntry>();
+  file_session->set_session_id(utils::UUIDGenerator::GenerateUUID());
+  file_session->set_ino(ino);
+  file_session->set_client_id(client_id);
 
-std::string FileSession::EncodeValue() const {
-  pb::mdsv2::FileSession file_session;
-  file_session.set_session_id(session_id_);
-  file_session.set_ino(ino_);
-  file_session.set_client_id(client_id_);
-
-  return MetaCodec::EncodeFileSessionValue(file_session);
+  return file_session;
 }
 
 FileSessionCache::FileSessionCache() : count_metrics_(kFileSessionCacheCountMetricsName) {}
 
 bool FileSessionCache::Put(FileSessionPtr file_session) {
   utils::WriteLockGuard guard(lock_);
-  auto key = Key{.ino = file_session->Ino(), .session_id = file_session->SessionId()};
+  auto key = Key{.ino = file_session->ino(), .session_id = file_session->session_id()};
   auto it = file_session_map_.find(key);
   if (it != file_session_map_.end()) {
     return false;
@@ -74,7 +67,7 @@ bool FileSessionCache::Put(FileSessionPtr file_session) {
 void FileSessionCache::Upsert(FileSessionPtr file_session) {
   utils::WriteLockGuard guard(lock_);
 
-  auto key = Key{.ino = file_session->Ino(), .session_id = file_session->SessionId()};
+  auto key = Key{.ino = file_session->ino(), .session_id = file_session->session_id()};
 
   auto it = file_session_map_.find(key);
   if (it == file_session_map_.end()) {
@@ -150,21 +143,14 @@ bool FileSessionCache::IsExist(uint64_t ino, const std::string& session_id) {
   return file_session_map_.find(key) != file_session_map_.end();
 }
 
-FileSessionManager::FileSessionManager(uint32_t fs_id, KVStorageSPtr kv_storage)
+FileSessionManager::FileSessionManager(uint32_t fs_id, OperationProcessorSPtr operation_processor)
     : fs_id_(fs_id),
-      kv_storage_(kv_storage),
+      operation_processor_(operation_processor),
       total_count_metrics_(kFileSessionTatalCountMetricsName),
       count_metrics_(kFileSessionCountMetricsName) {}
 
 Status FileSessionManager::Create(uint64_t ino, const std::string& client_id, FileSessionPtr& file_session) {
-  file_session = FileSession::New(fs_id_, ino, client_id);
-
-  KVStorage::WriteOption write_option;
-  auto status = kv_storage_->Put(write_option, file_session->EncodeKey(), file_session->EncodeValue());
-  if (!status.ok()) {
-    DINGO_LOG(ERROR) << fmt::format("[filesession] create fail, {}/{} {}", ino, client_id, status.error_str());
-    return status;
-  }
+  file_session = NewFileSession(ino, client_id);
 
   // add to cache
   CHECK(file_session_cache_.Put(file_session))
@@ -230,12 +216,6 @@ Status FileSessionManager::IsExist(uint64_t ino, bool just_cache, bool& is_exist
 }
 
 Status FileSessionManager::Delete(uint64_t ino, const std::string& session_id) {
-  auto status = kv_storage_->Delete(FileSession::EncodeKey(fs_id_, ino, session_id));
-  if (!status.ok()) {
-    DINGO_LOG(ERROR) << fmt::format("[filesession] delete fail, {}/{} {}", ino, session_id, status.error_str());
-    return status;
-  }
-
   // delete cache
   file_session_cache_.Delete(ino, session_id);
 
@@ -245,95 +225,64 @@ Status FileSessionManager::Delete(uint64_t ino, const std::string& session_id) {
 }
 
 Status FileSessionManager::Delete(uint64_t ino) {
-  std::vector<FileSessionPtr> file_sessions;
-  auto status = GetFileSessionsFromStore(ino, file_sessions);
-  if (!status.ok()) {
-    return status;
-  }
-
-  if (file_sessions.empty()) {
-    return Status::OK();
-  }
-
-  int retry = 0;
-  do {
-    auto txn = kv_storage_->NewTxn();
-
-    for (auto& file_session : file_sessions) {
-      txn->Delete(file_session->EncodeKey());
-    }
-
-    status = txn->Commit();
-    if (status.error_code() != pb::error::ESTORE_MAYBE_RETRY) {
-      break;
-    }
-
-    ++retry;
-  } while (retry < FLAGS_txn_max_retry_times);
+  // delete cache
+  file_session_cache_.Delete(ino);
 
-  if (status.ok()) {
-    // delete cache
-    file_session_cache_.Delete(ino);
+  // count_metrics_ << (0 - static_cast<int64_t>(file_sessions.size()));
 
-    count_metrics_ << (0 - static_cast<int64_t>(file_sessions.size()));
-  }
-
-  return status;
+  return Status::OK();
 }
 
 Status FileSessionManager::GetFileSessionsFromStore(uint64_t ino, std::vector<FileSessionPtr>& file_sessions) {
-  Range range;
-  MetaCodec::GetFileSessionRange(fs_id_, ino, range.start_key, range.end_key);
+  Trace trace;
+  ScanFileSessionOperation operation(trace, fs_id_, ino);
 
-  auto txn = kv_storage_->NewTxn();
-  std::vector<KeyValue> kvs;
-  do {
-    kvs.clear();
-
-    auto status = txn->Scan(range, FLAGS_fs_scan_batch_size, kvs);
-    if (!status.ok()) {
-      DINGO_LOG(ERROR) << fmt::format("[filesession] scan fail, {} {}", ino, status.error_str());
-      break;
-    }
+  auto status = operation_processor_->RunAlone(&operation);
+  if (!status.ok()) {
+    DINGO_LOG(ERROR) << fmt::format("[filesession] scan file session fail, status({}).", status.error_str());
+    return status;
+  }
 
-    for (auto& kv : kvs) {
-      file_sessions.push_back(FileSession::New(fs_id_, MetaCodec::DecodeFileSessionValue(kv.value)));
-    }
+  auto& result = operation.GetResult();
 
-  } while (kvs.size() >= FLAGS_fs_scan_batch_size);
+  for (const auto& file_session : result.file_sessions) {
+    file_sessions.push_back(std::make_shared<FileSessionEntry>(file_session));
+  }
 
   return Status::OK();
 }
 
 Status FileSessionManager::GetFileSessionFromStore(uint64_t ino, const std::string& session_id,
                                                    FileSessionPtr& file_session) {
-  std::string key = FileSession::EncodeKey(fs_id_, ino, session_id);
+  Trace trace;
+  GetFileSessionOperation operation(trace, fs_id_, ino, session_id);
 
-  std::string value;
-  auto status = kv_storage_->Get(key, value);
+  auto status = operation_processor_->RunAlone(&operation);
   if (!status.ok()) {
-    DINGO_LOG(ERROR) << fmt::format("[filesession] get fail, {}/{} {}", ino, session_id, status.error_str());
+    DINGO_LOG(ERROR) << fmt::format("[filesession] get file session fail, status({}).", status.error_str());
     return status;
   }
 
-  file_session = FileSession::New(fs_id_, MetaCodec::DecodeFileSessionValue(value));
+  auto& result = operation.GetResult();
+
+  file_session = std::make_shared<FileSessionEntry>(result.file_session);
 
   return Status::OK();
 }
 
 Status FileSessionManager::IsExistFromStore(uint64_t ino, bool& is_exist) {
-  Range range;
-  MetaCodec::GetFileSessionRange(fs_id_, ino, range.start_key, range.end_key);
+  Trace trace;
+  ScanFileSessionOperation operation(trace, fs_id_, ino);
 
-  std::vector<KeyValue> kvs;
-  auto txn = kv_storage_->NewTxn();
-  auto status = txn->Scan(range, 1, kvs);
+  auto status = operation_processor_->RunAlone(&operation);
   if (!status.ok()) {
-    DINGO_LOG(ERROR) << fmt::format("[filesession] scan fail, {} {}", ino, status.error_str());
+    DINGO_LOG(ERROR) << fmt::format("[filesession] scan file session fail, status({}).", status.error_str());
     return status;
   }
 
-  is_exist = !kvs.empty();
+  auto& result = operation.GetResult();
+
+  is_exist = !result.file_sessions.empty();
 
   return Status::OK();
 }

src/mdsv2/filesystem/file_session.h

@@ -20,48 +20,15 @@
 #include <string>
 #include <vector>
 
-#include "dingofs/mdsv2.pb.h"
 #include "mdsv2/common/status.h"
-#include "mdsv2/storage/storage.h"
+#include "mdsv2/common/type.h"
+#include "mdsv2/filesystem/store_operation.h"
 #include "utils/concurrent/concurrent.h"
-#include "utils/uuid.h"
 
 namespace dingofs {
 namespace mdsv2 {
 
-class FileSession;
-using FileSessionPtr = std::shared_ptr<FileSession>;
-
-// represent file session for client open file
-class FileSession {
- public:
-  FileSession() = default;
-  FileSession(uint32_t fs_id, const std::string& session_id, uint64_t ino, const std::string& client_id)
-      : fs_id_(fs_id), session_id_(session_id), ino_(ino), client_id_(client_id) {}
-
-  static FileSessionPtr New(uint32_t fs_id, uint64_t ino, const std::string& client_id) {
-    return std::make_shared<FileSession>(fs_id, utils::UUIDGenerator::GenerateUUID(), ino, client_id);
-  }
-
-  static FileSessionPtr New(uint32_t fs_id, const pb::mdsv2::FileSession& file_session) {
-    return std::make_shared<FileSession>(fs_id, file_session.session_id(), file_session.ino(),
-                                         file_session.client_id());
-  }
-
-  const std::string& SessionId() const { return session_id_; }
-  uint64_t Ino() const { return ino_; }
-  const std::string& ClientId() const { return client_id_; }
-
-  static std::string EncodeKey(uint32_t fs_id, uint64_t ino, const std::string& session_id);
-  std::string EncodeKey() const;
-  std::string EncodeValue() const;
-
- private:
-  uint32_t fs_id_;
-  uint64_t ino_;
-  std::string session_id_;
-  std::string client_id_;
-};
+using FileSessionPtr = std::shared_ptr<FileSessionEntry>;
 
 // cache file session
 class FileSessionCache {
@@ -85,6 +52,7 @@ class FileSessionCache {
   void Upsert(FileSessionPtr file_session);
   void Delete(uint64_t ino, const std::string& session_id);
   void Delete(uint64_t ino);
+
   FileSessionPtr Get(uint64_t ino, const std::string& session_id);
   std::vector<FileSessionPtr> Get(uint64_t ino);
   bool IsExist(uint64_t ino);
@@ -106,22 +74,23 @@ using FileSessionManagerUPtr = std::unique_ptr<FileSessionManager>;
 // persist store file session and cache file session
 class FileSessionManager {
  public:
-  FileSessionManager(uint32_t fs_id, KVStorageSPtr kv_storage);
+  FileSessionManager(uint32_t fs_id, OperationProcessorSPtr operation_processor);
   ~FileSessionManager() = default;
 
   FileSessionManager(const FileSessionManager&) = delete;
   FileSessionManager& operator=(const FileSessionManager&) = delete;
   FileSessionManager(FileSessionManager&&) = delete;
   FileSessionManager& operator=(FileSessionManager&&) = delete;
 
-  static FileSessionManagerUPtr New(uint32_t fs_id, KVStorageSPtr kv_storage) {
-    return std::make_unique<FileSessionManager>(fs_id, kv_storage);
+  static FileSessionManagerUPtr New(uint32_t fs_id, OperationProcessorSPtr operation_processor) {
+    return std::make_unique<FileSessionManager>(fs_id, operation_processor);
   }
 
   Status Create(uint64_t ino, const std::string& client_id, FileSessionPtr& file_session);
   Status IsExist(uint64_t ino, bool just_cache, bool& is_exist);
   Status Delete(uint64_t ino, const std::string& session_id);
   Status Delete(uint64_t ino);
+
   FileSessionPtr Get(uint64_t ino, const std::string& session_id, bool just_cache = false);
   std::vector<FileSessionPtr> Get(uint64_t ino, bool just_cache = false);
 
@@ -134,12 +103,11 @@ class FileSessionManager {
 
   uint32_t fs_id_;
 
-  // store file session
-  KVStorageSPtr kv_storage_;
-
   // cache file session
   FileSessionCache file_session_cache_;
 
+  OperationProcessorSPtr operation_processor_;
+
   // statistics
   bvar::Adder<uint64_t> total_count_metrics_;
   bvar::Adder<int64_t> count_metrics_;