chunkserver: fix crc issues

opencurve · Jul 9, 2021 · dc58284 · dc58284
1 parent ead087b
commit dc58284
Show file tree

Hide file tree

Showing 20 changed files with 350 additions and 183 deletions.
diff --git a/curvefs_python/BUILD_bak b/curvefs_python/BUILD_bak
@@ -54,6 +54,7 @@ cc_library(
         "//proto:nameserver2_cc_proto",
         "//proto:common_cc_proto",
         "//proto:topology_cc_proto",
+        "//proto:scan_cc_proto",
         "//proto:chunkserver-cc-protos",
         "//src/client:curve_client"
     ],
@@ -79,6 +80,7 @@ cc_library(
         "-lnameserver2_proto",
         "-ltopology_proto",
         "-lcommon_proto",
+        "-lscan_proto",
         "-lchunkserver-protos",
         "-lprotobuf",
         "-lprotobuf_lite",

diff --git a/proto/BUILD b/proto/BUILD
@@ -68,7 +68,8 @@ proto_library(
         "chunkserver.proto",
         "curve_storage.proto",
     ]),
-    deps = [":common_proto"],
+    deps = [":common_proto",
+            ":scan_proto"],
     visibility = ["//visibility:public"],
 )
 
@@ -104,4 +105,3 @@ proto_library(
     name = "scan_proto",
     srcs = ["scan.proto"],        
 )
-
diff --git a/proto/chunk.proto b/proto/chunk.proto
@@ -65,6 +65,7 @@ message ChunkRequest {
     optional uint64 sendScanMapTimeoutMs = 14;         // for scan chunk
     optional uint32 sendScanMapRetryTimes= 15;         // for scan chunk
     optional uint64 sendScanMapRetryIntervalUs = 16;   // for scan chunk
+    optional bool readMetaPage = 17;                   // for scan chunk
 };
 
 enum CHUNK_OP_STATUS {

diff --git a/proto/heartbeat.proto b/proto/heartbeat.proto
@@ -22,6 +22,7 @@
 
 syntax = "proto2";
 import "proto/common.proto";
+import "proto/scan.proto";
 package curve.mds.heartbeat;
 
 option cc_generic_services = true;
@@ -43,6 +44,8 @@ message CopySetInfo {
     optional bool scaning = 8;
     // timestamp for last success scan (seconds)
     optional uint64 lastScanSec = 9;
+    // failed crc check scanmap
+    repeated chunkserver.ScanMap scanMap = 10;
 };
 
 message ConfigChangeInfo {

diff --git a/src/chunkserver/chunkserver.cpp b/src/chunkserver/chunkserver.cpp
@@ -613,6 +613,8 @@ void ChunkServer::InitScanOptions(
         &scanOptions->intervalSec));
     LOG_IF(FATAL, !conf->GetUInt64Value("copyset.scan_size_byte",
         &scanOptions->scanSize));
+    LOG_IF(FATAL, !conf->GetUInt32Value("global.meta_page_size",
+        &scanOptions->chunkMetaPageSize));
     LOG_IF(FATAL, !conf->GetUInt64Value("copyset.scan_rpc_timeout_ms",
         &scanOptions->timeoutMs));
     LOG_IF(FATAL, !conf->GetUInt32Value("copyset.scan_rpc_retry_times",

diff --git a/src/chunkserver/copyset_node.cpp b/src/chunkserver/copyset_node.cpp
@@ -518,6 +518,10 @@ uint64_t CopysetNode::GetLastScan() const {
     return lastScanSec_;
 }
 
+std::vector<ScanMap>& CopysetNode::GetFailedScanMap() {
+    return failedScanMaps_;
+}
+
 std::string CopysetNode::GetCopysetDir() const {
     return copysetDirPath_;
 }

diff --git a/src/chunkserver/copyset_node.h b/src/chunkserver/copyset_node.h
@@ -43,6 +43,7 @@
 #include "proto/heartbeat.pb.h"
 #include "proto/chunk.pb.h"
 #include "proto/common.pb.h"
+#include "proto/scan.pb.h"
 
 namespace curve {
 namespace chunkserver {
@@ -152,6 +153,8 @@ class CopysetNode : public braft::StateMachine,
 
     virtual uint64_t GetLastScan() const;
 
+    virtual std::vector<ScanMap>& GetFailedScanMap();
+
     /**
      * 返回复制组数据目录
      * @return
@@ -281,7 +284,7 @@ class CopysetNode : public braft::StateMachine,
      * @param peers:返回的成员列表(输出参数)
      * @return
      */
-    void ListPeers(std::vector<Peer>* peers);
+    virtual void ListPeers(std::vector<Peer>* peers);
 
     /**
      * @brief initialize raft node options corresponding to the copyset node
@@ -458,6 +461,8 @@ class CopysetNode : public braft::StateMachine,
     bool scaning_;
     // last scan time
     uint64_t lastScanSec_;
+    // failed check scanmap
+    std::vector<ScanMap> failedScanMaps_;
 };
 
 }  // namespace chunkserver

diff --git a/src/chunkserver/datastore/chunkserver_chunkfile.cpp b/src/chunkserver/datastore/chunkserver_chunkfile.cpp
@@ -484,6 +484,18 @@ CSErrorCode CSChunkFile::Read(char * buf, off_t offset, size_t length) {
     return CSErrorCode::Success;
 }
 
+CSErrorCode CSChunkFile::ReadMetaPage(char * buf) {
+    ReadLockGuard readGuard(rwLock_);
+    int rc = readMetaPage(buf);
+    if (rc < 0) {
+        LOG(ERROR) << "Read chunk meta page failed."
+                   << "ChunkID: " << chunkId_
+                   << ",chunk sn: " << metaPage_.sn;
+        return CSErrorCode::InternalError;
+    }
+    return CSErrorCode::Success;
+}
+
 CSErrorCode CSChunkFile::ReadSpecifiedChunk(SequenceNum sn,
                                             char * buf,
                                             off_t offset,

diff --git a/src/chunkserver/datastore/chunkserver_chunkfile.h b/src/chunkserver/datastore/chunkserver_chunkfile.h
@@ -184,6 +184,15 @@ class CSChunkFile {
      * @return: return error code
      */
     CSErrorCode Read(char * buf, off_t offset, size_t length);
+
+    /**
+     * Read chunk meta data
+     * There may be concurrency, add read lock
+     * @param buf: the data read
+     * @return: return error code
+     */
+    CSErrorCode ReadMetaPage(char * buf);
+
     /**
      * Read the chunk of the specified Sequence
      * There may be concurrency, add read lock

diff --git a/src/chunkserver/datastore/chunkserver_datastore.cpp b/src/chunkserver/datastore/chunkserver_datastore.cpp
@@ -160,6 +160,22 @@ CSErrorCode CSDataStore::ReadChunk(ChunkID id,
     return CSErrorCode::Success;
 }
 
+CSErrorCode CSDataStore::ReadChunkMetaPage(ChunkID id, SequenceNum sn,
+                                           char * buf) {
+    auto chunkFile = metaCache_.Get(id);
+    if (chunkFile == nullptr) {
+        return CSErrorCode::ChunkNotExistError;
+    }
+
+    CSErrorCode errorCode = chunkFile->ReadMetaPage(buf);
+    if (errorCode != CSErrorCode::Success) {
+        LOG(WARNING) << "Read chunk meta page failed."
+                     << "ChunkID = " << id;
+        return errorCode;
+    }
+    return CSErrorCode::Success;
+}
+
 CSErrorCode CSDataStore::ReadSnapshotChunk(ChunkID id,
                                            SequenceNum sn,
                                            char * buf,

diff --git a/src/chunkserver/datastore/chunkserver_datastore.h b/src/chunkserver/datastore/chunkserver_datastore.h
@@ -187,6 +187,19 @@ class CSDataStore {
                                   char * buf,
                                   off_t offset,
                                   size_t length);
+
+    /**
+     * Read the metadata of the current chunk
+     * @param id: the chunk id to be read
+     * @param sn: used to record trace, not used in actual logic processing,
+     *             indicating the sequence number of the current user file
+     * @param buf: the content of the data read
+     * @return: return error code
+     */
+    virtual CSErrorCode ReadChunkMetaPage(ChunkID id,
+                                          SequenceNum sn,
+                                          char * buf);
+
     /**
      * Read the data of the specified sequence, it may read the current
      * chunk file, or it may read the snapshot file

diff --git a/src/chunkserver/heartbeat.cpp b/src/chunkserver/heartbeat.cpp
@@ -154,6 +154,12 @@ int Heartbeat::BuildCopysetInfo(curve::mds::heartbeat::CopySetInfo* info,
     if (copyset->GetLastScan() > 0) {
         info->set_lastscansec(copyset->GetLastScan());
     }
+    auto failedScanMaps = copyset->GetFailedScanMap();
+    if (!failedScanMaps.empty()) {
+        for (auto &map : failedScanMaps) {
+            info->add_scanmap()->CopyFrom(map);
+        }
+    }
 
     std::vector<Peer> peers;
     copyset->ListPeers(&peers);

diff --git a/src/chunkserver/heartbeat.h b/src/chunkserver/heartbeat.h
@@ -40,6 +40,7 @@
 #include "src/common/concurrent/concurrent.h"
 #include "src/chunkserver/scan_manager.h"
 #include "proto/heartbeat.pb.h"
+#include "proto/scan.pb.h"
 
 using ::curve::common::Thread;
 

diff --git a/src/chunkserver/op_request.cpp b/src/chunkserver/op_request.cpp
@@ -876,12 +876,20 @@ void ScanChunkRequest::OnApply(uint64_t index,
     std::unique_ptr<char[]> readBuffer(new(std::nothrow)char[size]);
     CHECK(nullptr != readBuffer)
         << "new readBuffer failed " << strerror(errno);
+    // scan chunk metapage or user data
+    auto ret = 0;
+    if (request_->has_readmetapage() && request_->readmetapage()) {
+        ret = datastore_->ReadChunkMetaPage(request_->chunkid(),
+                                            request_->sn(),
+                                            readBuffer.get());
+    } else {
+        ret = datastore_->ReadChunk(request_->chunkid(),
+                                    request_->sn(),
+                                    readBuffer.get(),
+                                    request_->offset(),
+                                    size);
+    }
 
-    auto ret = datastore_->ReadChunk(request_->chunkid(),
-                                     request_->sn(),
-                                     readBuffer.get(),
-                                     request_->offset(),
-                                     size);
     if (CSErrorCode::Success == ret) {
         crc = ::curve::common::CRC32(readBuffer.get(), size);
         // build scanmap
@@ -924,11 +932,20 @@ void ScanChunkRequest::OnApplyFromLog(std::shared_ptr<CSDataStore> datastore,  /
     CHECK(nullptr != readBuffer)
         << "new readBuffer failed " << strerror(errno);
 
-    auto ret = datastore->ReadChunk(request.chunkid(),
+    // scan chunk metapage or user data
+    auto ret = 0;
+    if (request.has_readmetapage() && request.readmetapage()) {
+        ret = datastore->ReadChunkMetaPage(request.chunkid(),
+                                            request.sn(),
+                                            readBuffer.get());
+    } else {
+        ret = datastore->ReadChunk(request.chunkid(),
                                     request.sn(),
                                     readBuffer.get(),
                                     request.offset(),
                                     size);
+    }
+
     if (CSErrorCode::Success == ret) {
         crc = ::curve::common::CRC32(readBuffer.get(), size);
         BuildAndSendScanMap(request, index_, crc);