curvefs/client: fix the problem that chunkCacheManager has been released in advance during the Release process of dataCache

previously, s3 compact task stores a raw pointer of copyset,
so after copyset is been purged, s3 compact task will cause
curvefs-metaserver process exited.

now, let s3 compact task store a shared copyset node.

.bazelversion

@@ -0,0 +1 @@
+0.17.2

CHANGELOG-2.0.md

@@ -0,0 +1,26 @@
+# CHANGELOG of v2.0
+
+The content of Curve v2.0 includes CurveBS v1.3, CurveFS v0.2.0 and some other content listed below.
+Previous change logs can be found at [CHANGELOG-1.3](https://github.com/opencurve/curve/blob/master/CHANGELOG-1.3.md)
+
+## new features
+
+- [CurveFS: support replace a server which doesn't work anymore.](https://github.com/opencurve/curve/pull/954)
+- [CurveFS: support internal and external services.](https://github.com/opencurve/curve/issues/973)
+
+
+## optimization
+
+- [CurveFS client: update inode asynchronously.](https://github.com/opencurve/curve/pull/1020)
+- [Turn off batch delete object in the config file.](https://github.com/opencurve/curve/pull/997)
+- [Remove create filesystem at mount process and use curvefs-tool instead.](https://github.com/opencurve/curve/pull/899)
+- [Optimizing delete filesystem confirmation of curvefs-tool.](https://github.com/opencurve/curve/issues/843)
+- [Optimizing leaky bucket algorithm of quality of service.](https://github.com/opencurve/curve/pull/1045)
+
+
+## bug fix
+
+- [Fix insert inode fail when load deleting partition.](https://github.com/opencurve/curve/pull/997)
+- [Fix read fail when write cache of diskcache is enabled, because metadata maybe newer than data in S3.](https://github.com/opencurve/curve/pull/1006)
+- [Fix CurveFS mds heartbeat doesn't delete copyset creating.](https://github.com/opencurve/curve/pull/1011)
+- [Fix CurveFS create copyset when exist already.](https://github.com/opencurve/curve/pull/1002)

curvefs/conf/client.conf

@@ -99,6 +99,7 @@ s3.intervalSec=3
 s3.flushIntervalSec=5
 s3.writeCacheMaxByte=838860800
 s3.readCacheMaxByte=209715200
+s3.dataCrc=true
 s3.endpoint=
 s3.bucket_name=
 s3.ak=

curvefs/conf/tools.conf

@@ -24,5 +24,5 @@ s3.ak=ak
 s3.sk=sk
 s3.endpoint=endpoint
 s3.bucket_name=bucket
-s3.blocksize=1048576
-s3.chunksize=4194304
+s3.blocksize=4194304
+s3.chunksize=67108864

curvefs/src/client/common/config.cpp

@@ -158,6 +158,8 @@ void InitS3Option(Configuration *conf, S3Option *s3Opt) {
                               &s3Opt->s3ClientAdaptorOpt.nearfullRatio);
     conf->GetValueFatalIfFail("s3.baseSleepUs",
                               &s3Opt->s3ClientAdaptorOpt.baseSleepUs);
+    conf->GetValueFatalIfFail("s3.dataCrc",
+                              &s3Opt->s3DataCrc);
     ::curve::common::InitS3AdaptorOption(conf, &s3Opt->s3AdaptrOpt);
     InitDiskCacheOption(conf, &s3Opt->s3ClientAdaptorOpt.diskCacheOpt);
 }

curvefs/src/client/common/config.h

@@ -120,6 +120,7 @@ struct S3ClientAdaptorOption {
 struct S3Option {
     S3ClientAdaptorOption s3ClientAdaptorOpt;
     S3AdapterOption s3AdaptrOpt;
+    bool s3DataCrc;
 };
 
 struct VolumeOption {

curvefs/src/client/fuse_client.cpp

@@ -151,15 +151,8 @@ CURVEFS_ERROR FuseClient::FuseOpInit(void *userdata,
     FSStatusCode ret = mdsClient_->GetFsInfo(fsName, &fsInfo);
     if (ret != FSStatusCode::OK) {
         if (FSStatusCode::NOT_FOUND == ret) {
-            LOG(INFO) << "The fsName not exist, try to CreateFs"
-                      << ", fsName = " << fsName;
-
-            CURVEFS_ERROR ret2 = CreateFs(userdata, &fsInfo);
-            if (ret2 != CURVEFS_ERROR::OK) {
-                LOG(ERROR) << "CreateFs failed, ret = " << ret2
-                           << ", fsName = " << fsName;
-                return ret2;
-            }
+            LOG(ERROR) << "The fsName not exist, fsName = " << fsName;
+            return CURVEFS_ERROR::NOTEXIST;
         } else {
             LOG(ERROR) << "GetFsInfo failed, FSStatusCode = " << ret
                        << ", FSStatusCode_Name = "

curvefs/src/client/fuse_s3_client.cpp

@@ -33,7 +33,7 @@ CURVEFS_ERROR FuseS3Client::Init(const FuseClientOption &option) {
     if (ret != CURVEFS_ERROR::OK) {
         return ret;
     }
-    s3Client_ = std::make_shared<S3ClientImpl>();
+    s3Client_ = std::make_shared<S3ClientImpl>(option.s3Opt.s3DataCrc);
     s3Client_->Init(option.s3Opt.s3AdaptrOpt);
     ret = s3Adaptor_->Init(option.s3Opt.s3ClientAdaptorOpt, s3Client_.get(),
                            inodeManager_, mdsClient_);

curvefs/src/client/rpcclient/metaserver_client.cpp

@@ -586,8 +586,15 @@ void MetaServerClientImpl::UpdateInodeAsync(const Inode &inode,
         MetaServerOpType::UpdateInode, task, inode.fsid(), inode.inodeid());
     auto excutor = std::make_shared<UpdateInodeExcutor>(opt_,
         metaCache_, channelManager_, taskCtx);
-    TaskExecutorDone *taskDone = new TaskExecutorDone(excutor, done);
-    excutor->DoAsyncRPCTask(taskDone);
+    TaskExecutorDone *taskDone = new TaskExecutorDone(
+        excutor, done);
+    brpc::ClosureGuard taskDone_guard(taskDone);
+    int ret = excutor->DoAsyncRPCTask(taskDone);
+    if (ret < 0) {
+        taskDone->SetRetCode(ret);
+        return;
+    }
+    taskDone_guard.release();
 }
 
 MetaStatusCode MetaServerClientImpl::GetOrModifyS3ChunkInfo(
@@ -741,8 +748,15 @@ void MetaServerClientImpl::GetOrModifyS3ChunkInfoAsync(
         MetaServerOpType::GetOrModifyS3ChunkInfo, task, fsId, inodeId);
     auto excutor = std::make_shared<GetOrModifyS3ChunkInfoExcutor>(opt_,
         metaCache_, channelManager_, taskCtx);
-    TaskExecutorDone *taskDone = new TaskExecutorDone(excutor, done);
-    excutor->DoAsyncRPCTask(taskDone);
+    TaskExecutorDone *taskDone = new TaskExecutorDone(
+        excutor, done);
+    brpc::ClosureGuard taskDone_guard(taskDone);
+    int ret = excutor->DoAsyncRPCTask(taskDone);
+    if (ret < 0) {
+        taskDone->SetRetCode(ret);
+        return;
+    }
+    taskDone_guard.release();
 }
 
 MetaStatusCode MetaServerClientImpl::CreateInode(const InodeParam &param,

curvefs/src/client/rpcclient/task_excutor.cpp

@@ -79,37 +79,38 @@ int TaskExecutor::DoRPCTask() {
     return retCode;
 }
 
-void TaskExecutor::DoAsyncRPCTask(TaskExecutorDone *done) {
-    brpc::ClosureGuard done_guard(done);
+int TaskExecutor::DoAsyncRPCTask(TaskExecutorDone *done) {
     task_->rpcTimeoutMs = opt_.rpcTimeoutMS;
+
     int retCode = -1;
 
-    if (task_->retryTimes++ > opt_.maxRetry) {
-        LOG(ERROR) << task_->TaskContextStr()
-                   << " retry times exceeds the limit";
-        done->SetRetCode(retCode);
-        return;
-    }
+    do {
+        if (task_->retryTimes++ > opt_.maxRetry) {
+            LOG(ERROR) << task_->TaskContextStr()
+                       << " retry times exceeds the limit";
+            break;
+        }
 
-    if (!HasValidTarget() && !GetTarget()) {
-        LOG(WARNING) << "get target fail for " << task_->TaskContextStr()
-                     << ", sleep and retry";
-        done->SetRetCode(retCode);
-        return;
-    }
+        if (!HasValidTarget() && !GetTarget()) {
+            LOG(WARNING) << "get target fail for " << task_->TaskContextStr()
+                         << ", sleep and retry";
+            bthread_usleep(opt_.retryIntervalUS);
+            continue;
+        }
 
-    auto channel = channelManager_->GetOrCreateChannel(
-        task_->target.metaServerID, task_->target.endPoint);
-    if (!channel) {
-        LOG(WARNING) << "GetOrCreateChannel fail for "
-                     << task_->TaskContextStr() << ", sleep and retry";
-        done->SetRetCode(retCode);
-        return;
-    }
+        auto channel = channelManager_->GetOrCreateChannel(
+            task_->target.metaServerID, task_->target.endPoint);
+        if (!channel) {
+            LOG(WARNING) << "GetOrCreateChannel fail for "
+                         << task_->TaskContextStr() << ", sleep and retry";
+            bthread_usleep(opt_.retryIntervalUS);
+            continue;
+        }
+        retCode = ExcuteTask(channel.get(), done);
+        break;
+    } while (true);
 
-    ExcuteTask(channel.get(), done);
-    done_guard.release();
-    return;
+    return retCode;
 }
 
 bool TaskExecutor::OnReturn(int retCode) {
@@ -303,7 +304,11 @@ void TaskExecutorDone::Run() {
     needRetry = excutor_->OnReturn(code_);
     if (needRetry) {
         excutor_->PreProcessBeforeRetry(code_);
-        excutor_->DoAsyncRPCTask(this);
+        code_ = excutor_->DoAsyncRPCTask(this);
+        if (code_ < 0) {
+            done_->SetMetaStatusCode(ConvertToMetaStatusCode(code_));
+            return;
+        }
         self_guard.release();
         done_guard.release();
     } else {

curvefs/src/client/rpcclient/task_excutor.h

@@ -106,7 +106,7 @@ class TaskExecutor {
     }
 
     int DoRPCTask();
-    void DoAsyncRPCTask(TaskExecutorDone *done);
+    int DoAsyncRPCTask(TaskExecutorDone *done);
 
     bool OnReturn(int retCode);
     void PreProcessBeforeRetry(int retCode);

curvefs/src/client/s3/client_s3.cpp

@@ -19,11 +19,15 @@
  * Created Date: 21-5-31
  * Author: huyao
  */
+#include "src/common/crc32.h"
 #include "curvefs/src/client/s3/client_s3.h"
 
 namespace curvefs {
 
 namespace client {
+
+using ::curve::common::CRC32;
+
 void S3ClientImpl::Init(const curve::common::S3AdapterOption &option) {
     s3Adapter_->Init(option);
 }
@@ -71,6 +75,11 @@ int S3ClientImpl::Download(const std::string &name, char *buf, uint64_t offset,
         LOG(ERROR) << "download error:" << ret;
     }
 
+    if (dataCrc_) {
+        uint32_t checkSum = CRC32(buf, length);
+        VLOG(3) << "download name: " << name << " ,crc: " << checkSum;
+    }
+
     VLOG(9) << "download end, ret:" << ret << ",length:" << length;
     return ret;
 }

curvefs/src/client/s3/client_s3.h

@@ -45,12 +45,14 @@ class S3Client {
                          uint64_t length) = 0;
     virtual void DownloadAsync(
         std::shared_ptr<GetObjectAsyncContext> context) = 0;
+    virtual bool IsDataCrc() = 0;
 };
 
 class S3ClientImpl : public S3Client {
  public:
-    S3ClientImpl() : S3Client() {
+    explicit S3ClientImpl(bool dataCrc) : S3Client() {
         s3Adapter_ = std::make_shared<curve::common::S3Adapter>();
+        dataCrc_ = dataCrc;
     }
     virtual ~S3ClientImpl() {}
     void Init(const curve::common::S3AdapterOption& option);
@@ -63,9 +65,12 @@ class S3ClientImpl : public S3Client {
     void SetAdapter(std::shared_ptr<curve::common::S3Adapter> adapter) {
         s3Adapter_ = adapter;
     }
-
+    bool IsDataCrc() {
+        return dataCrc_;
+    }
  private:
     std::shared_ptr<curve::common::S3Adapter> s3Adapter_;
+    bool dataCrc_;
 };
 
 }  // namespace client

curvefs/src/client/s3/client_s3_cache_manager.cpp

@@ -19,17 +19,17 @@
  * Created Date: 21-8-18
  * Author: huyao
  */
-
-#include "curvefs/src/client/s3/client_s3_cache_manager.h"
-
 #include <utility>
-
+#include "src/common/crc32.h"
+#include "curvefs/src/client/s3/client_s3_cache_manager.h"
 #include "curvefs/src/client/s3/client_s3_adaptor.h"
 #include "curvefs/src/common/s3util.h"
 
 namespace curvefs {
 namespace client {
 
+using ::curve::common::CRC32;
+
 FileCacheManagerPtr FsCacheManager::FindFileCacheManager(uint64_t inodeId) {
     ReadLockGuard readLockGuard(rwLock_);
 
@@ -120,16 +120,17 @@ void FsCacheManager::Get(std::list<DataCachePtr>::iterator iter) {
                                  lruReadDataCacheList_, iter);
 }
 
-void FsCacheManager::Delete(std::list<DataCachePtr>::iterator iter) {
+bool FsCacheManager::Delete(std::list<DataCachePtr>::iterator iter) {
     std::lock_guard<std::mutex> lk(lruMtx_);
 
     if (!(*iter)->InReadCache()) {
-        return;
+        return false;
     }
 
     (*iter)->SetReadCacheState(false);
     lruByte_ -= (*iter)->GetActualLen();
     lruReadDataCacheList_.erase(iter);
+    return true;
 }
 
 CURVEFS_ERROR FsCacheManager::FsSync(bool force) {
@@ -518,8 +519,15 @@ int FileCacheManager::ReadFromS3(const std::vector<S3ReadRequest> &requests,
         std::vector<uint64_t> &DataCacheVec = dataCacheMapIter.second;
         WriteLockGuard writeLockGuard(chunkCacheManager->rwLockChunk_);
         for (auto &chunkPos : DataCacheVec) {
+            if (s3ClientAdaptor_->GetS3Client()->IsDataCrc()) {
+                int checkSum = CRC32((*responses)[i].GetDataBuf(),
+                                     (*responses)[i].GetBufLen());
+                VLOG(3) << "response data chunkPos: " << chunkPos
+                        << ", crc: " << checkSum
+                        << ", len: " << (*responses)[i].GetBufLen();
+            }
             DataCachePtr dataCache = std::make_shared<DataCache>(
-            s3ClientAdaptor_, chunkCacheManager.get(), chunkPos,
+            s3ClientAdaptor_, chunkCacheManager, chunkPos,
             (*responses)[i].GetBufLen(), (*responses)[i].GetDataBuf());
             chunkCacheManager->AddReadDataCache(dataCache);
             i++;
@@ -1204,11 +1212,10 @@ DataCachePtr ChunkCacheManager::FindWriteableDataCache(
 void ChunkCacheManager::WriteNewDataCache(S3ClientAdaptorImpl *s3ClientAdaptor,
                                           uint32_t chunkPos, uint32_t len,
                                           const char *data) {
-    DataCachePtr dataCache =
-        std::make_shared<DataCache>(s3ClientAdaptor, this, chunkPos, len, data);
+    DataCachePtr dataCache = std::make_shared<DataCache>(
+        s3ClientAdaptor, this->shared_from_this(), chunkPos, len, data);
     VLOG(9) << "WriteNewDataCache chunkPos:" << chunkPos << ", len:" << len
-            << ", new len:" << dataCache->GetLen()
-            << ",chunkIndex:" << index_;
+            << ", new len:" << dataCache->GetLen() << ",chunkIndex:" << index_;
     WriteLockGuard writeLockGuard(rwLockWrite_);
 
     auto ret = dataWCacheMap_.emplace(chunkPos, dataCache);
@@ -1249,8 +1256,9 @@ void ChunkCacheManager::AddReadDataCache(DataCachePtr dataCache) {
     for (auto key : deleteKeyVec) {
         auto iter = dataRCacheMap_.find(key);
         std::list<DataCachePtr>::iterator dcpIter = iter->second;
-        s3ClientAdaptor_->GetFsCacheManager()->Delete(dcpIter);
-        dataRCacheMap_.erase(iter);
+        if (s3ClientAdaptor_->GetFsCacheManager()->Delete(dcpIter)) {
+            dataRCacheMap_.erase(iter);
+        }
     }
     std::list<DataCachePtr>::iterator outIter;
     bool ret =
@@ -1287,9 +1295,10 @@ void ChunkCacheManager::ReleaseCache() {
     WriteLockGuard writeLockGuard(rwLockRead_);
     auto iter = dataRCacheMap_.begin();
     for (; iter != dataRCacheMap_.end(); iter++) {
-        s3ClientAdaptor_->GetFsCacheManager()->Delete(iter->second);
+        if (s3ClientAdaptor_->GetFsCacheManager()->Delete(iter->second)) {
+            dataRCacheMap_.erase(iter);
+        }
     }
-    dataRCacheMap_.clear();
 }
 
 void ChunkCacheManager::TruncateCache(uint64_t chunkPos) {
@@ -1331,8 +1340,9 @@ void ChunkCacheManager::TruncateReadCache(uint64_t chunkPos) {
         uint64_t dcLen = (*rIter->second)->GetLen();
         uint64_t dcActualLen = (*rIter->second)->GetActualLen();
         if ((dcChunkPos + dcLen) > chunkPos) {
-            s3ClientAdaptor_->GetFsCacheManager()->Delete(rIter->second);
-            dataRCacheMap_.erase(next(rIter).base());
+            if (s3ClientAdaptor_->GetFsCacheManager()->Delete(rIter->second)) {
+                dataRCacheMap_.erase(next(rIter).base());
+            }
         } else {
             break;
         }
@@ -1415,7 +1425,7 @@ void ChunkCacheManager::UpdateWriteCacheMap(uint64_t oldChunkPos,
 }
 
 DataCache::DataCache(S3ClientAdaptorImpl *s3ClientAdaptor,
-                     ChunkCacheManager *chunkCacheManager, uint64_t chunkPos,
+                     ChunkCacheManagerPtr chunkCacheManager, uint64_t chunkPos,
                      uint64_t len, const char *data)
     : s3ClientAdaptor_(s3ClientAdaptor), chunkCacheManager_(chunkCacheManager),
       dirty_(true), delete_(false), inReadCache_(false) {

curvefs/src/client/s3/client_s3_cache_manager.h

@@ -98,7 +98,7 @@ using PageDataMap = std::map<uint64_t, PageData *>;
 class DataCache : public std::enable_shared_from_this<DataCache> {
  public:
     DataCache(S3ClientAdaptorImpl *s3ClientAdaptor,
-              ChunkCacheManager *chunkCacheManager, uint64_t chunkPos,
+              ChunkCacheManagerPtr chunkCacheManager, uint64_t chunkPos,
               uint64_t len, const char *data);
     virtual ~DataCache() {
         auto iter = dataMap_.begin();
@@ -171,7 +171,7 @@ class DataCache : public std::enable_shared_from_this<DataCache> {
 
  private:
     S3ClientAdaptorImpl *s3ClientAdaptor_;
-    ChunkCacheManager* chunkCacheManager_;
+    ChunkCacheManagerPtr chunkCacheManager_;
     uint64_t chunkPos_;  // useful chunkPos
     uint64_t len_;  // useful len
     uint64_t actualChunkPos_;  // after alignment the actual chunkPos
@@ -203,7 +203,8 @@ class S3ReadResponse {
     uint64_t len_;
 };
 
-class ChunkCacheManager {
+class ChunkCacheManager
+    : public std::enable_shared_from_this<ChunkCacheManager> {
  public:
     ChunkCacheManager(uint64_t index, S3ClientAdaptorImpl *s3ClientAdaptor)
         : index_(index), s3ClientAdaptor_(s3ClientAdaptor) {}
@@ -324,7 +325,7 @@ class FsCacheManager {
 
     bool Set(DataCachePtr dataCache,
              std::list<DataCachePtr>::iterator *outIter);
-    void Delete(std::list<DataCachePtr>::iterator iter);
+    bool Delete(std::list<DataCachePtr>::iterator iter);
     void Get(std::list<DataCachePtr>::iterator iter);
 
     CURVEFS_ERROR FsSync(bool force);

curvefs/src/client/s3/disk_cache_manager.cpp

@@ -302,12 +302,18 @@ bool DiskCacheManager::IsDiskCacheFull() {
 }
 
 bool DiskCacheManager::IsDiskCacheSafe() {
+    int64_t ratio = diskFsUsedRatio_.load(std::memory_order_seq_cst);
     uint64_t usedBytes = GetDiskUsedbytes();
-    if (usedBytes <= (safeRatio_ * maxUsableSpaceBytes_ / 100)) {
+    if ((usedBytes < (safeRatio_ * maxUsableSpaceBytes_ / 100))
+      && (ratio < safeRatio_)) {
         VLOG(3) << "disk cache is safe"
-                << ", usedBytes is: " << usedBytes;
+                << ", usedBytes is: " << usedBytes
+                << ", use ratio is: " << ratio;
         return true;
     }
+    VLOG(3) << "disk cache is not safe"
+                << ", usedBytes is: " << usedBytes
+                << ", use ratio is: " << ratio;
     return false;
 }
 

curvefs/src/metaserver/copyset/copyset_node.cpp

@@ -82,7 +82,12 @@ CopysetNode::CopysetNode(PoolId poolId, CopysetId copysetId,
       ongoingConfChange_(),
       metric_(absl::make_unique<OperatorApplyMetric>(poolId_, copysetId_)) {}
 
-CopysetNode::~CopysetNode() { Stop(); }
+CopysetNode::~CopysetNode() {
+    Stop();
+    raftNode_.reset();
+    applyQueue_.reset();
+    metaStore_.reset();
+}
 
 bool CopysetNode::Init(const CopysetNodeOptions& options) {
     options_ = options;
@@ -108,7 +113,7 @@ bool CopysetNode::Init(const CopysetNodeOptions& options) {
     }
 
     // create metastore
-    metaStore_ = absl::make_unique<MetaStoreImpl>();
+    metaStore_ = absl::make_unique<MetaStoreImpl>(this);
 
     InitRaftNodeOptions();
 
@@ -141,13 +146,15 @@ void CopysetNode::Stop() {
     if (raftNode_) {
         raftNode_->shutdown(nullptr);
         raftNode_->join();
-        raftNode_.reset();
     }
 
     if (applyQueue_) {
         applyQueue_->Flush();
         applyQueue_->Stop();
-        applyQueue_.reset();
+    }
+
+    if (metaStore_) {
+        metaStore_->Clear();
     }
 }
 

curvefs/src/metaserver/copyset/copyset_node.h

@@ -58,7 +58,7 @@ class CopysetNode : public braft::StateMachine {
                 const braft::Configuration& conf,
                 CopysetNodeManager* nodeManager);
 
-    ~CopysetNode();
+    ~CopysetNode() override;
 
     bool Init(const CopysetNodeOptions& options);
 

curvefs/src/metaserver/copyset/copyset_node_manager.cpp

@@ -27,7 +27,6 @@
 #include <string>
 #include <utility>
 
-#include "absl/memory/memory.h"
 #include "curvefs/src/metaserver/copyset/copyset_reloader.h"
 #include "curvefs/src/metaserver/copyset/raft_cli_service2.h"
 #include "curvefs/src/metaserver/copyset/utils.h"
@@ -39,13 +38,6 @@ namespace copyset {
 
 using ::curve::common::TimeUtility;
 
-CopysetNodeManager::CopysetNodeManager()
-    : options_(),
-      running_(false),
-      loadFinished_(false),
-      lock_(),
-      copysets_() {}
-
 bool CopysetNodeManager::IsLoadFinished() const {
     return loadFinished_.load(std::memory_order_acquire);
 }
@@ -164,6 +156,18 @@ CopysetNode* CopysetNodeManager::GetCopysetNode(PoolId poolId,
     return nullptr;
 }
 
+std::shared_ptr<CopysetNode> CopysetNodeManager::GetSharedCopysetNode(
+    PoolId poolId, CopysetId copysetId) {
+    ReadLockGuard lock(lock_);
+
+    auto it = copysets_.find(ToGroupId(poolId, copysetId));
+    if (it != copysets_.end()) {
+        return it->second;
+    }
+
+    return nullptr;
+}
+
 int CopysetNodeManager::IsCopysetNodeExist(
     const CreateCopysetRequest::Copyset& copyset) {
     ReadLockGuard lock(lock_);
@@ -175,16 +179,16 @@ int CopysetNodeManager::IsCopysetNodeExist(
         auto copysetNode = iter->second.get();
         std::vector<Peer> peers;
         copysetNode->ListPeers(&peers);
-        if (peers.size() != copyset.peers_size()) {
+        if (peers.size() != static_cast<size_t>(copyset.peers_size())) {
             return -1;
         }
 
         for (int i = 0; i < copyset.peers_size(); i++) {
-            auto cspeer = copyset.peers(i);
-            auto iter = std::find_if(
-            peers.begin(), peers.end(),
-            [&cspeer](const Peer& p) { return
-                cspeer.address() == p.address();});
+            const auto& cspeer = copyset.peers(i);
+            auto iter = std::find_if(peers.begin(), peers.end(),
+                                     [&cspeer](const Peer& p) {
+                                         return cspeer.address() == p.address();
+                                     });
             if (iter == peers.end()) {
                 return -1;
             }
@@ -193,12 +197,6 @@ int CopysetNodeManager::IsCopysetNodeExist(
     return 1;
 }
 
-bool CopysetNodeManager::IsCopysetNodeExist(PoolId poolId,
-                                            CopysetId copysetId) const {
-    ReadLockGuard lock(lock_);
-    return copysets_.count(ToGroupId(poolId, copysetId)) != 0;
-}
-
 bool CopysetNodeManager::CreateCopysetNode(PoolId poolId, CopysetId copysetId,
                                            const braft::Configuration& conf,
                                            bool checkLoadFinish) {
@@ -208,48 +206,30 @@ bool CopysetNodeManager::CreateCopysetNode(PoolId poolId, CopysetId copysetId,
         return false;
     }
 
-    if (IsCopysetNodeExist(poolId, copysetId)) {
+    braft::GroupId groupId = ToGroupId(poolId, copysetId);
+    std::shared_ptr<CopysetNode> copysetNode;
+
+    WriteLockGuard lock(lock_);
+    if (copysets_.count(groupId) != 0) {
         LOG(WARNING) << "Copyset node already exists: "
                      << ToGroupIdString(poolId, copysetId);
         return false;
     }
 
-    braft::GroupId groupId = ToGroupId(poolId, copysetId);
-    CopysetNode* node = nullptr;
-
-    {
-        WriteLockGuard lock(lock_);
-        if (copysets_.count(groupId) != 0) {
-            LOG(WARNING) << "Copyset node already exists: "
-                         << ToGroupNid(poolId, copysetId);
-            return false;
-        }
-
-        auto copysetNode =
-            absl::make_unique<CopysetNode>(poolId, copysetId, conf, this);
-        node = copysetNode.get();
-        copysets_.emplace(groupId, std::move(copysetNode));
-    }
-
-    auto removeNode = [&]() {
-        WriteLockGuard lock(lock_);
-        copysets_.erase(groupId);
-    };
-
-    if (!node->Init(options_)) {
-        removeNode();
+    copysetNode = std::make_shared<CopysetNode>(poolId, copysetId, conf, this);
+    if (!copysetNode->Init(options_)) {
         LOG(ERROR) << "Copyset " << ToGroupIdString(poolId, copysetId)
                    << "init failed";
         return false;
     }
 
-    if (!node->Start()) {
-        removeNode();
+    if (!copysetNode->Start()) {
         LOG(ERROR) << "Copyset " << ToGroupIdString(poolId, copysetId)
                    << " start failed";
         return false;
     }
 
+    copysets_.emplace(groupId, std::move(copysetNode));
     LOG(INFO) << "Create copyset success "
               << ToGroupIdString(poolId, copysetId);
     return true;

curvefs/src/metaserver/copyset/copyset_node_manager.h

@@ -57,14 +57,15 @@ class CopysetNodeManager {
 
     virtual CopysetNode* GetCopysetNode(PoolId poolId, CopysetId copysetId);
 
+    virtual std::shared_ptr<CopysetNode> GetSharedCopysetNode(
+        PoolId poolId, CopysetId copysetId);
+
     /**
      * @return 0: not exist; 1: key exist and peers are exactly same;
      * -1: key exist but peers are not exactly same
      */
     int IsCopysetNodeExist(const CreateCopysetRequest::Copyset& copyset);
 
-    bool IsCopysetNodeExist(PoolId poolId, CopysetId copysetId) const;
-
     bool CreateCopysetNode(PoolId poolId, CopysetId copysetId,
                            const braft::Configuration& conf,
                            bool checkLoadFinish = true);
@@ -78,8 +79,14 @@ class CopysetNodeManager {
     virtual bool IsLoadFinished() const;
 
  public:
-    CopysetNodeManager();
+    CopysetNodeManager()
+        : options_(),
+          running_(false),
+          loadFinished_(false),
+          lock_(),
+          copysets_() {}
 
+ public:
     /**
      * @brief Add raft related services to server
      */
@@ -91,7 +98,7 @@ class CopysetNodeManager {
 
  private:
     using CopysetNodeMap =
-        std::unordered_map<braft::GroupId, std::unique_ptr<CopysetNode>>;
+        std::unordered_map<braft::GroupId, std::shared_ptr<CopysetNode>>;
 
     CopysetNodeOptions options_;
 

curvefs/src/metaserver/metastore.cpp

@@ -28,10 +28,10 @@
 #include <vector>
 #include "curvefs/src/metaserver/partition_clean_manager.h"
 #include "curvefs/src/metaserver/storage.h"
+#include "curvefs/src/metaserver/copyset/copyset_node.h"
 
 namespace curvefs {
 namespace metaserver {
-MetaStoreImpl::MetaStoreImpl() {}
 
 // NOTE: if we use set we need define hash function, it's complicate
 using PartitionContainerType = std::unordered_map<uint32_t, PartitionInfo>;
@@ -44,6 +44,8 @@ using InodeIteratorType = MapContainerIterator<InodeContainerType>;
 using DentryIteratorType = SetContainerIterator<DentryContainerType>;
 using PendingTxIteratorType = MapContainerIterator<PendingTxContainerType>;
 
+MetaStoreImpl::MetaStoreImpl(copyset::CopysetNode* node) : copysetNode_(node) {}
+
 bool MetaStoreImpl::LoadPartition(uint32_t partitionId, void* entry) {
     auto partitionInfo = reinterpret_cast<PartitionInfo*>(entry);
     partitionId = partitionInfo->partitionid();
@@ -136,11 +138,8 @@ bool MetaStoreImpl::Load(const std::string& pathname) {
             uint32_t partitionId = it->second->GetPartitionId();
             std::shared_ptr<PartitionCleaner> partitionCleaner =
                 std::make_shared<PartitionCleaner>(GetPartition(partitionId));
-            copyset::CopysetNode* copysetNode =
-                copyset::CopysetNodeManager::GetInstance().GetCopysetNode(
-                    it->second->GetPoolId(), it->second->GetCopySetId());
             PartitionCleanManager::GetInstance().Add(
-                partitionId, partitionCleaner, copysetNode);
+                partitionId, partitionCleaner, copysetNode_);
         }
     }
 
@@ -297,11 +296,8 @@ MetaStatusCode MetaStoreImpl::DeletePartition(
         it->second->ClearDentry();
         std::shared_ptr<PartitionCleaner> partitionCleaner =
             std::make_shared<PartitionCleaner>(GetPartition(partitionId));
-        copyset::CopysetNode* copysetNode =
-            copyset::CopysetNodeManager::GetInstance().GetCopysetNode(
-                it->second->GetPoolId(), it->second->GetCopySetId());
         PartitionCleanManager::GetInstance().Add(partitionId, partitionCleaner,
-                                                 copysetNode);
+                                                 copysetNode_);
         it->second->SetStatus(PartitionStatus::DELETING);
         TrashManager::GetInstance().Remove(partitionId);
         it->second->ClearS3Compact();

curvefs/src/metaserver/metastore.h

@@ -33,6 +33,11 @@
 
 namespace curvefs {
 namespace metaserver {
+
+namespace copyset {
+class CopysetNode;
+}  // namespace copyset
+
 // dentry
 using curvefs::metaserver::GetDentryRequest;
 using curvefs::metaserver::GetDentryResponse;
@@ -122,7 +127,7 @@ class MetaStore {
 
 class MetaStoreImpl : public MetaStore {
  public:
-    MetaStoreImpl();
+    explicit MetaStoreImpl(copyset::CopysetNode* node);
 
     bool Load(const std::string& pathname) override;
     bool Save(const std::string& path,
@@ -202,6 +207,8 @@ class MetaStoreImpl : public MetaStore {
     RWLock rwLock_;  // protect partitionMap_
     std::map<uint32_t, std::shared_ptr<Partition>> partitionMap_;
     std::list<uint32_t> partitionIds_;
+
+    copyset::CopysetNode* copysetNode_;
 };
 }  // namespace metaserver
 }  // namespace curvefs

curvefs/src/metaserver/s3compact_manager.cpp

@@ -152,7 +152,8 @@ void S3CompactManager::Init(std::shared_ptr<Configuration> conf) {
             std::make_shared<S3AdapterManager>(opts_.queueSize, opts_.s3opts);
         s3adapterManager_->Init();
         s3compactworkqueueImpl_ = std::make_shared<S3CompactWorkQueueImpl>(
-            s3adapterManager_, s3infoCache_, opts_);
+            s3adapterManager_, s3infoCache_, opts_,
+            &copyset::CopysetNodeManager::GetInstance());
         inited_ = true;
     } else {
         LOG(INFO) << "s3compact: not enabled";
@@ -210,8 +211,6 @@ void S3CompactManager::Enqueue() {
         }
 
         auto pinfo = s3compact->GetPartition();
-        auto copysetNode = CopysetNodeManager::GetInstance().GetCopysetNode(
-            pinfo.poolid(), pinfo.copysetid());
         // traverse inode container
         auto inodeManager = s3compact->GetInodeManager();
 
@@ -226,7 +225,7 @@ void S3CompactManager::Enqueue() {
         for (const auto& inodeid : inodes) {
             sleeper_.wait_for(std::chrono::milliseconds(opts_.enqueueSleepMS));
             s3compactworkqueueImpl_->Enqueue(
-                inodeManager, InodeKey(fsid, inodeid), pinfo, copysetNode);
+                inodeManager, InodeKey(fsid, inodeid), pinfo);
         }
     }
     {

curvefs/src/metaserver/s3compact_wq_impl.cpp

@@ -33,6 +33,7 @@
 
 #include "curvefs/src/common/s3util.h"
 #include "curvefs/src/metaserver/copyset/meta_operator.h"
+#include "curvefs/src/metaserver/copyset/copyset_node_manager.h"
 
 using curve::common::Configuration;
 using curve::common::InitS3AdaptorOption;
@@ -45,8 +46,7 @@ namespace curvefs {
 namespace metaserver {
 
 void S3CompactWorkQueueImpl::Enqueue(std::shared_ptr<InodeManager> inodeManager,
-                                     InodeKey inodeKey, PartitionInfo pinfo,
-                                     CopysetNode* copysetNode) {
+                                     InodeKey inodeKey, PartitionInfo pinfo) {
     std::unique_lock<std::mutex> guard(mutex_);
 
     // inodeKey already in working queue, just return
@@ -58,12 +58,26 @@ void S3CompactWorkQueueImpl::Enqueue(std::shared_ptr<InodeManager> inodeManager,
     while (IsFullUnlock()) {
         notFull_.wait(guard);
     }
+
+    auto copysetNode = copysetNodeMgr_->GetSharedCopysetNode(pinfo.poolid(),
+                                                             pinfo.copysetid());
+    if (!copysetNode) {
+        VLOG(6) << "Copyset node not found, poolid: " << pinfo.poolid()
+                << ", copysetid: " << pinfo.copysetid()
+                << ", fsid: " << inodeKey.fsId
+                << ", inodeid: " << inodeKey.inodeId;
+        return;
+    }
+
     compactingInodes_.push_back(inodeKey);
+
     struct S3CompactTask t {
         inodeManager, inodeKey, pinfo,
             std::make_shared<CopysetNodeWrapper>(copysetNode)
     };
-    auto task = std::bind(&S3CompactWorkQueueImpl::CompactChunks, this, t);
+
+    auto task =
+        std::bind(&S3CompactWorkQueueImpl::CompactChunks, this, std::move(t));
     // am i copysetnode leader?_
     queue_.push_back(std::move(task));
     notEmpty_.notify_one();
@@ -101,7 +115,8 @@ std::vector<uint64_t> S3CompactWorkQueueImpl::GetNeedCompact(
             VLOG(9) << "s3compact: reach max chunks to compact per time";
             break;
         }
-        if (item.second.s3chunks_size() > opts_.fragmentThreshold) {
+        if (static_cast<uint64_t>(item.second.s3chunks_size()) >
+            opts_.fragmentThreshold) {
             needCompact.push_back(item.first);
         }
     }

curvefs/src/metaserver/s3compact_wq_impl.h

@@ -47,25 +47,25 @@ using curve::common::S3Adapter;
 using curve::common::S3AdapterOption;
 using curve::common::TaskThreadPool;
 using curvefs::metaserver::copyset::CopysetNode;
+using curvefs::metaserver::copyset::CopysetNodeManager;
 
 namespace curvefs {
 namespace metaserver {
 
 class CopysetNodeWrapper {
  public:
-    explicit CopysetNodeWrapper(CopysetNode* copysetNode)
+    explicit CopysetNodeWrapper(const std::shared_ptr<CopysetNode>& copysetNode)
         : copysetNode_(copysetNode) {}
     virtual ~CopysetNodeWrapper() {}
-    CopysetNode* copysetNode_;
+    std::shared_ptr<CopysetNode> copysetNode_;
     virtual bool IsLeaderTerm() {
-        if (copysetNode_ == nullptr) return false;
-        return copysetNode_->IsLeaderTerm();
+        return copysetNode_ && copysetNode_->IsLeaderTerm();
     }
     virtual bool IsValid() {
         return copysetNode_ != nullptr;
     }
     CopysetNode* Get() {
-        return copysetNode_;
+        return copysetNode_.get();
     }
 };
 
@@ -73,20 +73,24 @@ class S3CompactWorkQueueImpl : public TaskThreadPool<> {
  public:
     S3CompactWorkQueueImpl(std::shared_ptr<S3AdapterManager> s3adapterManager,
                            std::shared_ptr<S3InfoCache> s3infoCache,
-                           const S3CompactWorkQueueOption& opts)
+                           const S3CompactWorkQueueOption& opts,
+                           copyset::CopysetNodeManager* nodeMgr)
         : s3adapterManager_(s3adapterManager),
           s3infoCache_(s3infoCache),
-          opts_(opts) {}
+          opts_(opts),
+          copysetNodeMgr_(nodeMgr) {}
 
     std::shared_ptr<S3AdapterManager> s3adapterManager_;
     std::shared_ptr<S3InfoCache> s3infoCache_;
     S3CompactWorkQueueOption opts_;
     std::deque<InodeKey> compactingInodes_;
+    copyset::CopysetNodeManager* copysetNodeMgr_;
     void Enqueue(std::shared_ptr<InodeManager> inodeManager, InodeKey inodeKey,
-                 PartitionInfo pinfo, CopysetNode* copyset);
+                 PartitionInfo pinfo);
     std::function<void()> Dequeue();
     void ThreadFunc();
 
+    // compact task for one partition
     struct S3CompactTask {
         std::shared_ptr<InodeManager> inodeManager;
         InodeKey inodeKey;

curvefs/src/tools/create/curvefs_create_fs.cpp

@@ -127,9 +127,16 @@ int CreateFsTool::Init() {
     }
 
     AddRequest(request);
+
+    SetController();
+
     return ret;
 }
 
+void CreateFsTool::SetController() {
+    controller_->set_timeout_ms(FLAGS_rpcTimeoutMs);
+}
+
 bool CreateFsTool::AfterSendRequestToHost(const std::string& host) {
     bool ret = true;
     if (controller_->Failed()) {

curvefs/src/tools/create/curvefs_create_fs.h

@@ -50,6 +50,7 @@ class CreateFsTool : public CurvefsToolRpc<curvefs::mds::CreateFsRequest,
  protected:
     void AddUpdateFlags() override;
     bool AfterSendRequestToHost(const std::string& host) override;
+    void SetController() override;
 };
 
 }  // namespace create

curvefs/src/tools/curvefs_tool.h

@@ -211,6 +211,7 @@ class CurvefsToolRpc : public CurvefsTool {
                 return true;
             }
             controller_->Reset();
+            SetController();
         }
         // send request to all host failed
         return false;

curvefs/test/client/client_s3_test.cpp

@@ -40,7 +40,7 @@ class ClientS3Test : public testing::Test {
     ClientS3Test() {}
     ~ClientS3Test() {}
     virtual void SetUp() {
-        client_ = new S3ClientImpl();
+        client_ = new S3ClientImpl(false);
         s3Client_ = std::make_shared<MockS3Adapter>();
         client_->SetAdapter(s3Client_);
         Aws::InitAPI(awsOptions_);

curvefs/test/client/fs_cache_manager_test.cpp

@@ -59,7 +59,7 @@ TEST(FsCacheManagerTest, test_read_lru_cache_size) {
 
         for (size_t i = 0; i < maxReadCacheByte / smallDataCacheByte; ++i) {
             manager.Set(std::make_shared<DataCache>(s3ClientAdaptor_,
-                                                    mockCacheMgr.get(), 0,
+                                                    mockCacheMgr, 0,
                                                     smallDataCacheByte, buf),
                         &outIter);
         }
@@ -73,7 +73,7 @@ TEST(FsCacheManagerTest, test_read_lru_cache_size) {
             .Times(expectCallTimes)
             .WillRepeatedly(Invoke([&counter](uint64_t) { counter.Signal(); }));
         manager.Set(std::make_shared<DataCache>(s3ClientAdaptor_,
-                                                mockCacheMgr.get(), 0,
+                                                mockCacheMgr, 0,
                                                 dataCacheByte, buf),
                     &outIter);
 
@@ -89,7 +89,7 @@ TEST(FsCacheManagerTest, test_read_lru_cache_size) {
             .WillRepeatedly(Invoke([&counter](uint64_t) { counter.Signal(); }));
 
         manager.Set(std::make_shared<DataCache>(s3ClientAdaptor_,
-                                                mockCacheMgr.get(), 0,
+                                                mockCacheMgr, 0,
                                                 dataCacheByte, buf),
                     &outIter);
         counter.Wait();

curvefs/test/client/mock_client_s3.h

@@ -50,6 +50,7 @@ class MockS3Client : public S3Client {
                                char* buf, uint64_t offset, uint64_t length));
     MOCK_METHOD1(DownloadAsync, void(
                  std::shared_ptr<GetObjectAsyncContext> context));
+    MOCK_METHOD0(IsDataCrc, bool());
 };
 
 

curvefs/test/client/rpcclient/metaserver_client_test.cpp

@@ -749,6 +749,69 @@ TEST_F(MetaServerClientImplTest, test_UpdateInode) {
     ASSERT_EQ(MetaStatusCode::RPC_ERROR, status);
 }
 
+TEST_F(MetaServerClientImplTest, test_GetOrModifyS3ChunkInfo) {
+    uint32_t fsId = 1;
+    uint64_t inodeId = 100;
+    google::protobuf::Map<
+        uint64_t, S3ChunkInfoList> s3ChunkInfos;
+    bool returnS3ChunkInfoMap = true;
+    google::protobuf::Map<
+        uint64_t, S3ChunkInfoList> out;
+    uint64_t applyIndex = 10;
+
+    // test1: success
+    curvefs::metaserver::GetOrModifyS3ChunkInfoResponse response;
+    response.set_statuscode(curvefs::metaserver::OK);
+    response.set_appliedindex(applyIndex);
+    EXPECT_CALL(mockMetaServerService_, GetOrModifyS3ChunkInfo(_, _, _, _))
+        .WillOnce(DoAll(
+            SetArgPointee<2>(response),
+            Invoke(SetRpcService<
+                curvefs::metaserver::GetOrModifyS3ChunkInfoRequest,
+                curvefs::metaserver::GetOrModifyS3ChunkInfoResponse>)));
+    EXPECT_CALL(*mockMetacache_.get(), GetTarget(_, _, _, _, _))
+        .WillRepeatedly(DoAll(SetArgPointee<2>(target_),
+                              SetArgPointee<3>(applyIndex), Return(true)));
+    EXPECT_CALL(*mockMetacache_.get(), UpdateApplyIndex(_, _));
+
+    MetaStatusCode status = metaserverCli_.GetOrModifyS3ChunkInfo(
+        fsId, inodeId, s3ChunkInfos, returnS3ChunkInfoMap, &out);
+
+    ASSERT_EQ(MetaStatusCode::OK, status);
+
+    // test2: overload
+    response.set_statuscode(curvefs::metaserver::OVERLOAD);
+    EXPECT_CALL(mockMetaServerService_, GetOrModifyS3ChunkInfo(_, _, _, _))
+        .WillRepeatedly(DoAll(
+            SetArgPointee<2>(response),
+            Invoke(SetRpcService<
+                curvefs::metaserver::GetOrModifyS3ChunkInfoRequest,
+                curvefs::metaserver::GetOrModifyS3ChunkInfoResponse>)));
+    status = metaserverCli_.GetOrModifyS3ChunkInfo(
+        fsId, inodeId, s3ChunkInfos, returnS3ChunkInfoMap, &out);
+    ASSERT_EQ(MetaStatusCode::OVERLOAD, status);
+
+    // test3: has no applyIndex
+    response.set_statuscode(curvefs::metaserver::OK);
+    response.clear_appliedindex();
+    EXPECT_CALL(mockMetaServerService_, GetOrModifyS3ChunkInfo(_, _, _, _))
+        .WillRepeatedly(DoAll(
+            SetArgPointee<2>(response),
+            Invoke(SetRpcService<
+                curvefs::metaserver::GetOrModifyS3ChunkInfoRequest,
+                curvefs::metaserver::GetOrModifyS3ChunkInfoResponse>)));
+    status = metaserverCli_.GetOrModifyS3ChunkInfo(
+        fsId, inodeId, s3ChunkInfos, returnS3ChunkInfoMap, &out);
+    ASSERT_EQ(MetaStatusCode::RPC_ERROR, status);
+
+    // test4: get target always fail
+    EXPECT_CALL(*mockMetacache_.get(), GetTarget(_, _, _, _, _))
+        .WillRepeatedly(Return(false));
+    status = metaserverCli_.GetOrModifyS3ChunkInfo(
+        fsId, inodeId, s3ChunkInfos, returnS3ChunkInfoMap, &out);
+    ASSERT_EQ(MetaStatusCode::RPC_ERROR, status);
+}
+
 TEST_F(MetaServerClientImplTest, test_CreateInode) {
     // in
     InodeParam inode;

curvefs/test/client/rpcclient/mock_metaserver_service.h

@@ -83,6 +83,12 @@ class MockMetaServerService : public curvefs::metaserver::MetaServerService {
                       const ::curvefs::metaserver::DeleteInodeRequest *request,
                       ::curvefs::metaserver::DeleteInodeResponse *response,
                       ::google::protobuf::Closure *done));
+
+    MOCK_METHOD4(GetOrModifyS3ChunkInfo,
+        void(::google::protobuf::RpcController *controller,
+            const ::curvefs::metaserver::GetOrModifyS3ChunkInfoRequest *request,
+             ::curvefs::metaserver::GetOrModifyS3ChunkInfoResponse *response,
+             ::google::protobuf::Closure *done));
 };
 }  // namespace rpcclient
 }  // namespace client

curvefs/test/client/test_disk_cache_manager.cpp

@@ -292,7 +292,25 @@ TEST_F(TestDiskCacheManager, IsDiskCacheFull) {
 }
 
 TEST_F(TestDiskCacheManager, IsDiskCacheSafe) {
-    int ret = diskCacheManager_->IsDiskCacheSafe();
+    S3ClientAdaptorOption option;
+    option.diskCacheOpt.diskCacheType = (DiskCacheType)2;
+    option.diskCacheOpt.cacheDir = "/mnt/test_unit";
+    option.diskCacheOpt.trimCheckIntervalSec = 1;
+    option.diskCacheOpt.fullRatio = 0;
+    option.diskCacheOpt.safeRatio = 0;
+    option.diskCacheOpt.maxUsableSpaceBytes = 0;
+    option.diskCacheOpt.cmdTimeoutSec = 5;
+    option.diskCacheOpt.asyncLoadPeriodMs = 10;
+    S3Client *client = nullptr;
+    diskCacheManager_->Init(client, option);
+    bool ret = diskCacheManager_->IsDiskCacheSafe();
+    ASSERT_EQ(false, ret);
+
+    option.diskCacheOpt.fullRatio = 100;
+    option.diskCacheOpt.safeRatio = 99;
+    option.diskCacheOpt.maxUsableSpaceBytes = 100000000;
+    diskCacheManager_->Init(client, option);
+    ret = diskCacheManager_->IsDiskCacheSafe();
     ASSERT_EQ(true, ret);
 }
 

curvefs/test/client/test_fuse_client.cpp

@@ -188,29 +188,12 @@ TEST_F(TestFuseVolumeClient, FuseOpInit_when_fs_not_exist) {
     EXPECT_CALL(*mdsClient_, GetFsInfo(fsName, _))
         .WillOnce(Return(FSStatusCode::NOT_FOUND));
 
-    EXPECT_CALL(*blockDeviceClient_, Stat(volName, user, _))
-        .WillOnce(Return(CURVEFS_ERROR::OK));
-
-    EXPECT_CALL(*mdsClient_, CreateFs(_, _, _))
-        .WillOnce(Return(FSStatusCode::OK));
-
     FsInfo fsInfoExp;
     fsInfoExp.set_fsid(100);
     fsInfoExp.set_fsname(fsName);
-    EXPECT_CALL(*mdsClient_, MountFs(fsName, _, _))
-        .WillOnce(DoAll(SetArgPointee<2>(fsInfoExp), Return(FSStatusCode::OK)));
-
-    EXPECT_CALL(*blockDeviceClient_, Open(volName, user))
-        .WillOnce(Return(CURVEFS_ERROR::OK));
 
     CURVEFS_ERROR ret = client_->FuseOpInit(&mOpts, nullptr);
-    ASSERT_EQ(CURVEFS_ERROR::OK, ret);
-
-    auto fsInfo = client_->GetFsInfo();
-    ASSERT_NE(fsInfo, nullptr);
-
-    ASSERT_EQ(fsInfo->fsid(), fsInfoExp.fsid());
-    ASSERT_EQ(fsInfo->fsname(), fsInfoExp.fsname());
+    ASSERT_EQ(CURVEFS_ERROR::NOTEXIST, ret);
 }
 
 TEST_F(TestFuseVolumeClient, FuseOpDestroy) {
@@ -1768,23 +1751,12 @@ TEST_F(TestFuseS3Client, FuseOpInit_when_fs_not_exist) {
     EXPECT_CALL(*mdsClient_, GetFsInfo(fsName, _))
         .WillOnce(Return(FSStatusCode::NOT_FOUND));
 
-    EXPECT_CALL(*mdsClient_, CreateFsS3(_, _, _))
-        .WillOnce(Return(FSStatusCode::OK));
-
     FsInfo fsInfoExp;
     fsInfoExp.set_fsid(100);
     fsInfoExp.set_fsname(fsName);
-    EXPECT_CALL(*mdsClient_, MountFs(fsName, _, _))
-        .WillOnce(DoAll(SetArgPointee<2>(fsInfoExp), Return(FSStatusCode::OK)));
 
     CURVEFS_ERROR ret = client_->FuseOpInit(&mOpts, nullptr);
-    ASSERT_EQ(CURVEFS_ERROR::OK, ret);
-
-    auto fsInfo = client_->GetFsInfo();
-    ASSERT_NE(fsInfo, nullptr);
-
-    ASSERT_EQ(fsInfo->fsid(), fsInfoExp.fsid());
-    ASSERT_EQ(fsInfo->fsname(), fsInfoExp.fsname());
+    ASSERT_EQ(CURVEFS_ERROR::NOTEXIST, ret);
 }
 
 TEST_F(TestFuseS3Client, FuseOpDestroy) {

curvefs/test/metaserver/BUILD

@@ -156,5 +156,6 @@ cc_test(
         "//curvefs/src/metaserver:curvefs_metaserver",
         "@com_google_googletest//:gtest_main",
         "@com_google_googletest//:gtest",
+        "//curvefs/test/metaserver/copyset/mock:metaserver_copyset_test_mock",
     ],
 )

curvefs/test/metaserver/copyset/copyset_node_snapshot_test.cpp

@@ -345,6 +345,7 @@ TEST_F(CopysetNodeRaftSnapshotTest,
     node->ListPeers(&peers);
     EXPECT_EQ(3, peers.size());
     EXPECT_EQ(epochBefore, node->GetConfEpoch());
+    node->SetMetaStore(nullptr);
 }
 
 TEST_F(CopysetNodeRaftSnapshotTest, SnapshotLoadTest_MetaStoreLoadFailed) {
@@ -372,6 +373,7 @@ TEST_F(CopysetNodeRaftSnapshotTest, SnapshotLoadTest_MetaStoreLoadFailed) {
         .Times(0);
 
     EXPECT_NE(0, node->on_snapshot_load(&reader));
+    node->SetMetaStore(nullptr);
 }
 
 }  // namespace copyset

curvefs/test/metaserver/copyset/copyset_node_test.cpp

@@ -45,6 +45,7 @@ using ::testing::Invoke;
 using ::testing::Return;
 using ::testing::SetArgPointee;
 using ::testing::SetArrayArgument;
+using ::testing::AtLeast;
 
 using ::curve::fs::MockLocalFileSystem;
 
@@ -542,6 +543,41 @@ TEST_F(CopysetNodeTest, StartWithoutInitReturnFailed) {
     EXPECT_FALSE(node.Start());
 }
 
+namespace {
+
+struct FakeClosure : public braft::Closure {
+    void Run() override {}
+};
+
+};  // namespace
+
+TEST_F(CopysetNodeTest, ProposeAfterStopWontFatal) {
+    CopysetNode node(poolId_, copysetId_, conf_, &mockNodeManager_);
+
+    EXPECT_TRUE(node.Init(options_));
+
+    auto* mockRaftNode = new MockRaftNode();
+    node.SetRaftNode(mockRaftNode);
+
+    EXPECT_CALL(*mockRaftNode, init(_))
+        .WillOnce(Return(0));
+
+    ASSERT_EQ(true, node.Start());
+
+    EXPECT_CALL(*mockRaftNode, shutdown(_))
+        .Times(AtLeast(1));
+    EXPECT_CALL(*mockRaftNode, join())
+        .Times(AtLeast(1));
+
+    node.Stop();
+
+    FakeClosure fakeDone;
+    braft::Task task;
+    task.done = &fakeDone;
+
+    ASSERT_NO_FATAL_FAILURE({ node.Propose(task); });
+}
+
 }  // namespace copyset
 }  // namespace metaserver
 }  // namespace curvefs

curvefs/test/metaserver/copyset/meta_operator_test.cpp

@@ -91,6 +91,7 @@ using ::testing::_;
 using ::testing::DoAll;
 using ::testing::Invoke;
 using ::testing::Return;
+using ::testing::AtLeast;
 
 class MetaOperatorTest : public testing::Test {
  protected:
@@ -154,6 +155,9 @@ TEST_F(MetaOperatorTest, OnApplyErrorTest) {
     mock::MockMetaStore* mockMetaStore = new mock::MockMetaStore();
     node.SetMetaStore(mockMetaStore);
 
+    ON_CALL(*mockMetaStore, Clear())
+        .WillByDefault(Return(true));
+
     brpc::Controller cntl;
 
 #define OPERATOR_ON_APPLY_TEST(TYPE)                                           \
@@ -257,6 +261,9 @@ TEST_F(MetaOperatorTest, OnApplyFromLogErrorTest) {
     mock::MockMetaStore* mockMetaStore = new mock::MockMetaStore();
     node.SetMetaStore(mockMetaStore);
 
+    ON_CALL(*mockMetaStore, Clear())
+        .WillByDefault(Return(true));
+
     brpc::Controller cntl;
 
 #define OPERATOR_ON_APPLY_FROM_LOG_TEST(TYPE)                                \
@@ -378,12 +385,14 @@ TEST_F(MetaOperatorTest, PropostTest_RequestCanBypassProcess) {
     auto* mockRaftNode = new MockRaftNode();
     node.SetRaftNode(mockRaftNode);
 
+    ON_CALL(*mockMetaStore, Clear())
+        .WillByDefault(Return(true));
     EXPECT_CALL(*mockRaftNode, apply(_))
         .Times(0);
     EXPECT_CALL(*mockRaftNode, shutdown(_))
-        .Times(1);
+        .Times(AtLeast(1));
     EXPECT_CALL(*mockRaftNode, join())
-        .Times(1);
+        .Times(AtLeast(1));
     EXPECT_CALL(*mockMetaStore, GetDentry(_, _))
         .WillOnce(Return(MetaStatusCode::OK));
 

curvefs/test/metaserver/metaserver_test.cpp

@@ -54,8 +54,8 @@ namespace metaserver {
 class MetaserverTest : public ::testing::Test {
  protected:
     void SetUp() override {
-        metaserverListenAddr_ = "127.0.0.1:6702";
-        topologyServiceAddr_ = "127.0.0.1:6700";
+        metaserverListenAddr_ = "127.0.0.1:16702";
+        topologyServiceAddr_ = "127.0.0.1:16700";
         ASSERT_EQ(0, server_.AddService(&mockTopologyService_,
                                         brpc::SERVER_DOESNT_OWN_SERVICE));
         ASSERT_EQ(0, server_.AddService(&mockHeartbeatService_,

curvefs/test/metaserver/metastore_test.cpp

@@ -135,7 +135,7 @@ class MetastoreTest : public ::testing::Test {
 };
 
 TEST_F(MetastoreTest, partition) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
     CreatePartitionRequest createPartitionRequest;
     CreatePartitionResponse createPartitionResponse;
     PartitionInfo partitionInfo;
@@ -230,7 +230,7 @@ TEST_F(MetastoreTest, partition) {
 }
 
 TEST_F(MetastoreTest, test_inode) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
 
     // create partition1 partition2
     CreatePartitionRequest createPartitionRequest;
@@ -464,7 +464,7 @@ TEST_F(MetastoreTest, test_inode) {
 }
 
 TEST_F(MetastoreTest, test_dentry) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
 
     // create partition1 partition2
     CreatePartitionRequest createPartitionRequest;
@@ -689,7 +689,7 @@ TEST_F(MetastoreTest, test_dentry) {
 }
 
 TEST_F(MetastoreTest, persist_success) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
     uint32_t partitionId = 4;
     uint32_t partitionId2 = 2;
     // create partition1
@@ -803,7 +803,7 @@ TEST_F(MetastoreTest, persist_success) {
     ASSERT_TRUE(done.IsSuccess());
 
     // load MetaStoreImpl to new meta
-    MetaStoreImpl metastoreNew;
+    MetaStoreImpl metastoreNew(nullptr);
     LOG(INFO) << "MetastoreTest test Load";
     ASSERT_TRUE(metastoreNew.Load("./metastore_test"));
 
@@ -827,7 +827,7 @@ TEST_F(MetastoreTest, persist_success) {
 }
 
 TEST_F(MetastoreTest, persist_deleting_partition_success) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
     uint32_t partitionId = 4;
     uint32_t partitionId2 = 2;
     // create partition1
@@ -955,7 +955,7 @@ TEST_F(MetastoreTest, persist_deleting_partition_success) {
     ASSERT_TRUE(done.IsSuccess());
 
     // load MetaStoreImpl to new meta
-    MetaStoreImpl metastoreNew;
+    MetaStoreImpl metastoreNew(nullptr);
     LOG(INFO) << "MetastoreTest test Load";
     ASSERT_TRUE(metastoreNew.Load("./metastore_test"));
 
@@ -979,7 +979,7 @@ TEST_F(MetastoreTest, persist_deleting_partition_success) {
 }
 
 TEST_F(MetastoreTest, persist_partition_fail) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
     uint32_t partitionId = 4;
     // create partition1
     CreatePartitionRequest createPartitionRequest;
@@ -1003,7 +1003,7 @@ TEST_F(MetastoreTest, persist_partition_fail) {
 }
 
 TEST_F(MetastoreTest, persist_dentry_fail) {
-    MetaStoreImpl metastore;
+    MetaStoreImpl metastore(nullptr);
     uint32_t partitionId = 4;
 
     // create partition1

curvefs/test/metaserver/mock_metaserver_s3.h

@@ -43,6 +43,7 @@ class MockS3Client : public S3Client {
     MOCK_METHOD1(Init, void(const curve::common::S3AdapterOption &options));
     MOCK_METHOD1(Delete, int(const std::string &name));
     MOCK_METHOD1(DeleteBatch, int(const std::list<std::string>& nameList));
+    MOCK_METHOD0(IsDataCrc, bool());
 };
 }  // namespace metaserver
 }  // namespace curvefs

curvefs/test/metaserver/mock_s3compactwq_impl.h

@@ -44,8 +44,8 @@ class MockS3CompactWorkQueueImpl : public S3CompactWorkQueueImpl {
     MockS3CompactWorkQueueImpl(
         std::shared_ptr<S3AdapterManager> s3AdapterManager,
         std::shared_ptr<S3InfoCache> s3infoCache,
-        const S3CompactWorkQueueOption& opts)
-        : S3CompactWorkQueueImpl(s3AdapterManager, s3infoCache, opts) {}
+        const S3CompactWorkQueueOption& opts, copyset::CopysetNodeManager* mgr)
+        : S3CompactWorkQueueImpl(s3AdapterManager, s3infoCache, opts, mgr) {}
     MetaStatusCode UpdateInode(
         CopysetNode* copysetNode, const PartitionInfo& pinfo, uint64_t inodeId,
         ::google::protobuf::Map<uint64_t, S3ChunkInfoList>&& s3ChunkInfoAdd,

curvefs/test/metaserver/s3compactwq_test.cpp

@@ -25,10 +25,14 @@
 
 #include "curvefs/src/metaserver/s3compact_manager.h"
 #include "curvefs/src/metaserver/s3compact_wq_impl.h"
+#include "curvefs/test/metaserver/copyset/mock/mock_copyset_node_manager.h"
 #include "curvefs/test/metaserver/mock_s3_adapter.h"
 #include "curvefs/test/metaserver/mock_s3compactwq_impl.h"
 #include "curvefs/test/metaserver/mock_s3infocache.h"
 
+using ::curvefs::metaserver::copyset::CopysetNode;
+using ::curvefs::metaserver::copyset::CopysetNodeManager;
+using ::curvefs::metaserver::copyset::MockCopysetNodeManager;
 using ::testing::_;
 using ::testing::AtLeast;
 using ::testing::DoAll;
@@ -64,10 +68,13 @@ class S3CompactWorkQueueImplTest : public ::testing::Test {
         inodeStorage_ = std::make_shared<MemoryInodeStorage>();
         trash_ = std::make_shared<TrashImpl>(inodeStorage_);
         inodeManager_ = std::make_shared<InodeManager>(inodeStorage_, trash_);
-        impl_ = std::make_shared<S3CompactWorkQueueImpl>(s3adapterManager_,
-                                                         s3infoCache_, opts_);
+        impl_ = std::make_shared<S3CompactWorkQueueImpl>(
+            s3adapterManager_, s3infoCache_, opts_,
+            &copyset::CopysetNodeManager::GetInstance());
+        mockCopystNodeManager_ = std::make_shared<MockCopysetNodeManager>();
         mockImpl_ = std::make_shared<MockS3CompactWorkQueueImpl>(
-            s3adapterManager_, s3infoCache_, opts_);
+            s3adapterManager_, s3infoCache_, opts_,
+            mockCopystNodeManager_.get());
         mockCopysetNodeWrapper_ = std::make_shared<MockCopysetNodeWrapper>();
     }
 
@@ -87,12 +94,13 @@ class S3CompactWorkQueueImplTest : public ::testing::Test {
     std::shared_ptr<S3CompactWorkQueueImpl> impl_;
     std::shared_ptr<MockS3CompactWorkQueueImpl> mockImpl_;
     std::shared_ptr<MockCopysetNodeWrapper> mockCopysetNodeWrapper_;
+    std::shared_ptr<MockCopysetNodeManager> mockCopystNodeManager_;
 };
 
 TEST_F(S3CompactWorkQueueImplTest, test_CopysetNodeWrapper) {
     braft::Configuration c;
-    CopysetNode n(0, 0, c, nullptr);
-    CopysetNodeWrapper cw1(&n);
+    auto n = std::make_shared<CopysetNode>(0, 0, c, nullptr);
+    CopysetNodeWrapper cw1(n);
     ASSERT_EQ(cw1.IsLeaderTerm(), false);
     ASSERT_EQ(cw1.IsValid(), true);
     CopysetNodeWrapper cw2(nullptr);

curvefs/test/tools/curvefs_umount_fs_tool_test.cpp

@@ -62,7 +62,7 @@ class UmountfsToolTest : public testing::Test {
     }
 
  protected:
-    std::string addr_ = "127.0.0.1:6791";
+    std::string addr_ = "127.0.0.1:16701";
     brpc::Server server_;
     MockMdsService mockMdsService_;
     UmountFsTool ut_;