Chunk pool format asyn

Signed-off-by: yyyyufeng <[email protected]>
opencurve · Oct 8, 2023 · 7d85dc8 · 7d85dc8
1 parent e822715
commit 7d85dc8
Show file tree

Hide file tree

Showing 24 changed files with 610 additions and 34 deletions.
diff --git a/conf/chunkserver.conf b/conf/chunkserver.conf
@@ -207,6 +207,10 @@ chunkfilepool.clean.enable=true
 chunkfilepool.clean.bytes_per_write=4096
 # The throttle iops for cleaning chunk (4KB/IO)
 chunkfilepool.clean.throttle_iops=500
+# The size of chunkfilepool
+chunkfilepool.chunk_file_pool_size=10GB
+# The thread num for format chunks
+chunkfilepool.thread_num=2
 
 #
 # WAL file pool

diff --git a/conf/chunkserver.conf.example b/conf/chunkserver.conf.example
@@ -199,6 +199,10 @@ chunkfilepool.clean.enable=true
 chunkfilepool.clean.bytes_per_write=4096
 # The throttle iops for cleaning chunk (4KB/IO)
 chunkfilepool.clean.throttle_iops=500
+# The size of chunkfilepool
+chunkfilepool.chunk_file_pool_size=0
+# The thread num for format chunks
+chunkfilepool.thread_num=2
 
 #
 # WAL file pool

diff --git a/proto/heartbeat.proto b/proto/heartbeat.proto
@@ -89,6 +89,8 @@ message ChunkServerStatisticInfo {
     required uint64 chunkSizeTrashedBytes = 7;
     // chunkfilepool的大小
     optional uint64 chunkFilepoolSize = 8;
+    // chunkfilepool格式化进度
+    optional uint32 chunkFilepoolFormatRate = 9;
 };
 
 message ChunkServerHeartbeatRequest {

diff --git a/proto/topology.proto b/proto/topology.proto
@@ -501,6 +501,21 @@ message GetCopySetsInChunkServerRequest {
     optional uint32 port = 3;
 }
 
+message ChunkFormatStatus {
+    required string ip = 1;
+    required uint32 port = 2;
+    required uint32 chunkServerID = 3;
+    required uint32 formatRate = 4;
+}
+
+message ListChunkFormatStatusRequest {
+
+}
+
+message ListChunkFormatStatusResponse {
+    repeated ChunkFormatStatus chunkFormatStatus = 1;
+}
+
 message GetCopySetsInChunkServerResponse {
     required sint32 statusCode = 1;
     repeated common.CopysetInfo copysetInfos = 2;
@@ -595,4 +610,5 @@ service TopologyService {
     rpc GetClusterInfo(GetClusterInfoRequest) returns (GetClusterInfoResponse);
     rpc SetCopysetsAvailFlag(SetCopysetsAvailFlagRequest) returns (SetCopysetsAvailFlagResponse);
     rpc ListUnAvailCopySets(ListUnAvailCopySetsRequest) returns (ListUnAvailCopySetsResponse);
+    rpc ListChunkFormatStatus(ListChunkFormatStatusRequest) returns (ListChunkFormatStatusResponse);
 }
diff --git a/src/chunkserver/chunkserver.cpp b/src/chunkserver/chunkserver.cpp
@@ -44,6 +44,7 @@
 #include "src/chunkserver/raftsnapshot/curve_snapshot_storage.h"
 #include "src/chunkserver/raftlog/curve_segment_log_storage.h"
 #include "src/common/curve_version.h"
+#include "src/common/bytes_convert.h"
 
 using ::curve::fs::LocalFileSystem;
 using ::curve::fs::LocalFileSystemOption;
@@ -65,6 +66,9 @@ DEFINE_string(raftSnapshotUri, "curve://./0/copysets", "raft snapshot uri");
 DEFINE_string(raftLogUri, "curve://./0/copysets", "raft log uri");
 DEFINE_string(recycleUri, "local://./0/recycler" , "recycle uri");
 DEFINE_string(chunkFilePoolDir, "./0/", "chunk file pool location");
+DEFINE_string(chunkFilePoolSize, "1GB", "format percent for chunkfillpool.");
+DEFINE_uint32(chunkFormatThreadNum,
+    1, "number of threads while file pool formatting");
 DEFINE_string(chunkFilePoolMetaPath,
     "./chunkfilepool.meta", "chunk file pool meta path");
 DEFINE_string(logPath, "./0/chunkserver.log-", "log file path");
@@ -479,8 +483,6 @@ void ChunkServer::Stop() {
     brpc::AskToQuit();
 }
 
-
-
 void ChunkServer::InitChunkFilePoolOptions(
     common::Configuration *conf, FilePoolOptions *chunkFilePoolOptions) {
     LOG_IF(FATAL, !conf->GetUInt32Value("global.chunk_size",
@@ -513,6 +515,20 @@ void ChunkServer::InitChunkFilePoolOptions(
             "chunkfilepool.meta_path", &metaUri));
         ::memcpy(
             chunkFilePoolOptions->metaPath, metaUri.c_str(), metaUri.size());
+
+        std::string chunkFilePoolUri;
+        LOG_IF(FATAL, !conf->GetStringValue(
+            "chunkfilepool.chunk_file_pool_dir", &chunkFilePoolUri));
+
+        ::memcpy(chunkFilePoolOptions->filePoolDir,
+                 chunkFilePoolUri.c_str(),
+                 chunkFilePoolUri.size());
+        std::string pool_size;
+        LOG_IF(FATAL, !conf->GetStringValue("chunkfilepool.chunk_file_pool_size", &pool_size)); // NOLINT
+        LOG_IF(FATAL, !curve::common::ToNumbericByte(pool_size,
+            &chunkFilePoolOptions->filePoolSize));
+        LOG_IF(FATAL, !conf->GetUInt32Value("chunkfilepool.chunk_file_pool_format_thread_num", // NOLINT
+            &chunkFilePoolOptions->formatThreadNum));
         LOG_IF(FATAL, !conf->GetBoolValue("chunkfilepool.clean.enable",
             &chunkFilePoolOptions->needClean));
         LOG_IF(FATAL, !conf->GetUInt32Value("chunkfilepool.clean.bytes_per_write",  // NOLINT
@@ -833,6 +849,20 @@ void ChunkServer::LoadConfigFromCmdline(common::Configuration *conf) {
         << "chunkFilePoolDir must be set when run chunkserver in command.";
     }
 
+    if (GetCommandLineFlagInfo("chunkFilePoolSize", &info)) {
+        conf->SetStringValue(
+            "chunkfilepool.chunk_file_pool_size", FLAGS_chunkFilePoolSize);
+    } else {
+        LOG(FATAL)
+        << "chunkFilePoolSize must be set when run chunkserver in command.";
+    }
+
+    if (GetCommandLineFlagInfo("chunkFormatThreadNum", &info)) {
+        conf->SetUInt64Value(
+            "chunkfilepool.chunk_file_pool_format_thread_num",
+            FLAGS_chunkFormatThreadNum);
+    }
+
     if (GetCommandLineFlagInfo("chunkFilePoolMetaPath", &info) &&
         !info.is_default) {
         conf->SetStringValue(