Chunk pool format asyn

Signed-off-by: yyyyufeng <[email protected]>

conf/chunkserver.conf

@@ -194,7 +194,7 @@ rconcurrentapply.queuedepth=1
 # 是否开启从chunkfilepool获取chunk，一般是true
 chunkfilepool.enable_get_chunk_from_pool=true
 # chunkfilepool目录
-chunkfilepool.chunk_file_pool_dir=./0/  # __CURVEADM_TEMPLATE__ ${prefix}/data __CURVEADM_TEMPLATE__
+chunkfilepool.chunk_file_pool_dir=./0/chunks  # __CURVEADM_TEMPLATE__ ${prefix}/data __CURVEADM_TEMPLATE__
 # chunkfilepool meta文件路径
 chunkfilepool.meta_path=./chunkfilepool.meta  # __CURVEADM_TEMPLATE__ ${prefix}/data/chunkfilepool.meta __CURVEADM_TEMPLATE__
 # chunkfilepool meta文件大小
@@ -207,6 +207,14 @@ chunkfilepool.clean.enable=true
 chunkfilepool.clean.bytes_per_write=4096
 # The throttle iops for cleaning chunk (4KB/IO)
 chunkfilepool.clean.throttle_iops=500
+# Whether allocate filePool by percent of disk size.
+chunkfilepool.allocated_by_percent=true
+# Preallocate storage percent of total disk
+chunkfilepool.allocate_percent=80
+# Preallocate storage size of chunkfilepool (None/KB/MB/GB/TB)
+chunkfilepool.chunk_file_pool_size=1GB
+# The thread num for format chunks
+chunkfilepool.thread_num=1
 
 #
 # WAL file pool
@@ -229,6 +237,14 @@ walfilepool.metapage_size=4096
 walfilepool.meta_file_size=4096
 # WAL filepool get chunk最大重试次数
 walfilepool.retry_times=5
+# Whether allocate filePool by percent of disk size.
+walfilepool.allocated_by_percent=true
+# Preallocate storage percent of total disk
+walfilepool.allocate_percent=90
+# Preallocate storage size size of walfilepool (None/KB/MB/GB/TB)
+walfilepool.wal_file_pool_size=0
+# The thread num for format chunks
+walfilepool.thread_num=1
 
 #
 # trash settings

conf/chunkserver.conf.example

@@ -186,7 +186,7 @@ rconcurrentapply.queuedepth=1
 # 是否开启从chunkfilepool获取chunk，一般是true
 chunkfilepool.enable_get_chunk_from_pool=true
 # chunkfilepool目录
-chunkfilepool.chunk_file_pool_dir=./0/
+chunkfilepool.chunk_file_pool_dir=./0/chunks
 # chunkfilepool meta文件路径
 #chunkfilepool.meta_path=./chunkfilepool.meta
 # chunkfilepool meta文件大小
@@ -199,6 +199,14 @@ chunkfilepool.clean.enable=true
 chunkfilepool.clean.bytes_per_write=4096
 # The throttle iops for cleaning chunk (4KB/IO)
 chunkfilepool.clean.throttle_iops=500
+# Whether allocate filePool by percent of disk size.
+chunkfilepool.allocated_by_percent=true
+# Preallocate storage percent of total disk
+chunkfilepool.allocate_percent=80
+# Preallocate storage size of chunkfilepool (None/KB/MB/GB/TB)
+chunkfilepool.chunk_file_pool_size=1GB
+# The thread num for format chunks
+chunkfilepool.thread_num=1
 
 #
 # WAL file pool
@@ -221,6 +229,14 @@ walfilepool.metapage_size=4096
 walfilepool.meta_file_size=4096
 # WAL filepool get chunk最大重试次数
 walfilepool.retry_times=5
+# Whether allocate filePool by percent of disk size.
+walfilepool.allocated_by_percent=true
+# Preallocate storage percent of total disk
+walfilepool.allocate_percent=10
+# Preallocate storage size size of walfilepool (None/KB/MB/GB/TB)
+walfilepool.wal_file_pool_size=0
+# The thread num for format chunks
+walfilepool.thread_num=1
 
 #
 # trash settings

proto/heartbeat.proto

@@ -89,6 +89,8 @@ message ChunkServerStatisticInfo {
     required uint64 chunkSizeTrashedBytes = 7;
     // chunkfilepool的大小
     optional uint64 chunkFilepoolSize = 8;
+    // percentage of chunkfilepool formatting
+    optional uint32 chunkFilepoolFormatPercent = 9;
 };
 
 message ChunkServerHeartbeatRequest {

proto/topology.proto

@@ -501,6 +501,19 @@ message GetCopySetsInChunkServerRequest {
     optional uint32 port = 3;
 }
 
+message ChunkFormatStatus {
+    required string ip = 1;
+    required uint32 port = 2;
+    required uint32 chunkServerID = 3;
+    required uint32 formatPercent = 4;
+}
+
+message ListChunkFormatStatusRequest {}
+
+message ListChunkFormatStatusResponse {
+    repeated ChunkFormatStatus chunkFormatStatus = 1;
+}
+
 message GetCopySetsInChunkServerResponse {
     required sint32 statusCode = 1;
     repeated common.CopysetInfo copysetInfos = 2;
@@ -595,4 +608,5 @@ service TopologyService {
     rpc GetClusterInfo(GetClusterInfoRequest) returns (GetClusterInfoResponse);
     rpc SetCopysetsAvailFlag(SetCopysetsAvailFlagRequest) returns (SetCopysetsAvailFlagResponse);
     rpc ListUnAvailCopySets(ListUnAvailCopySetsRequest) returns (ListUnAvailCopySetsResponse);
+    rpc ListChunkFormatStatus(ListChunkFormatStatusRequest) returns (ListChunkFormatStatusResponse);
 }

src/chunkserver/chunkserver.cpp

@@ -20,30 +20,31 @@
  * Author: lixiaocui
  */
 
-#include <glog/logging.h>
+#include "src/chunkserver/chunkserver.h"
 
-#include <butil/endpoint.h>
 #include <braft/builtin_service_impl.h>
 #include <braft/raft_service.h>
 #include <braft/storage.h>
+#include <butil/endpoint.h>
+#include <glog/logging.h>
 
 #include <memory>
 
-#include "src/chunkserver/chunkserver.h"
-#include "src/chunkserver/chunkserver_metrics.h"
-#include "src/chunkserver/chunkserver_service.h"
-#include "src/chunkserver/copyset_service.h"
-#include "src/chunkserver/chunk_service.h"
 #include "src/chunkserver/braft_cli_service.h"
 #include "src/chunkserver/braft_cli_service2.h"
+#include "src/chunkserver/chunk_service.h"
 #include "src/chunkserver/chunkserver_helper.h"
-#include "src/common/concurrent/task_thread_pool.h"
-#include "src/common/uri_parser.h"
-#include "src/chunkserver/raftsnapshot/curve_snapshot_attachment.h"
+#include "src/chunkserver/chunkserver_metrics.h"
+#include "src/chunkserver/chunkserver_service.h"
+#include "src/chunkserver/copyset_service.h"
+#include "src/chunkserver/raftlog/curve_segment_log_storage.h"
 #include "src/chunkserver/raftsnapshot/curve_file_service.h"
+#include "src/chunkserver/raftsnapshot/curve_snapshot_attachment.h"
 #include "src/chunkserver/raftsnapshot/curve_snapshot_storage.h"
-#include "src/chunkserver/raftlog/curve_segment_log_storage.h"
+#include "src/common/bytes_convert.h"
+#include "src/common/concurrent/task_thread_pool.h"
 #include "src/common/curve_version.h"
+#include "src/common/uri_parser.h"
 
 using ::curve::fs::LocalFileSystem;
 using ::curve::fs::LocalFileSystemOption;
@@ -65,6 +66,10 @@ 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_int32(chunkFilePoolAllocatedPercent, 80,
+             "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 +484,6 @@ void ChunkServer::Stop() {
     brpc::AskToQuit();
 }
 
-
-
 void ChunkServer::InitChunkFilePoolOptions(
     common::Configuration *conf, FilePoolOptions *chunkFilePoolOptions) {
     LOG_IF(FATAL, !conf->GetUInt32Value("global.chunk_size",
@@ -513,13 +516,56 @@ 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));
+        LOG_IF(FATAL, !curve::common::ToNumbericByte(
+                          pool_size, &chunkFilePoolOptions->filePoolSize));
+        LOG_IF(FATAL,
+               !conf->GetBoolValue("chunkfilepool.allocated_by_percent",
+                                   &chunkFilePoolOptions->allocatedByPercent));
+        LOG_IF(FATAL,
+               !conf->GetUInt32Value("chunkfilepool.allocate_percent",
+                                     &chunkFilePoolOptions->allocatedPercent));
+        LOG_IF(FATAL, !conf->GetUInt32Value(
+                          "chunkfilepool.chunk_file_pool_format_thread_num",
+                          &chunkFilePoolOptions->formatThreadNum));
         LOG_IF(FATAL, !conf->GetBoolValue("chunkfilepool.clean.enable",
             &chunkFilePoolOptions->needClean));
-        LOG_IF(FATAL, !conf->GetUInt32Value("chunkfilepool.clean.bytes_per_write",  // NOLINT
-            &chunkFilePoolOptions->bytesPerWrite));
+        LOG_IF(FATAL,
+               !conf->GetUInt32Value("chunkfilepool.clean.bytes_per_write",
+                                     &chunkFilePoolOptions->bytesPerWrite));
         LOG_IF(FATAL, !conf->GetUInt32Value("chunkfilepool.clean.throttle_iops",
             &chunkFilePoolOptions->iops4clean));
 
+        std::string copysetUri;
+        LOG_IF(FATAL,
+               !conf->GetStringValue("copyset.raft_snapshot_uri", &copysetUri));
+        curve::common::UriParser::ParseUri(copysetUri,
+                                           &chunkFilePoolOptions->copysetDir);
+
+        std::string recycleUri;
+        LOG_IF(FATAL,
+               !conf->GetStringValue("copyset.recycler_uri", &recycleUri));
+        curve::common::UriParser::ParseUri(recycleUri,
+                                           &chunkFilePoolOptions->recycleDir);
+
+        bool useChunkFilePoolAsWalPool;
+        LOG_IF(FATAL, !conf->GetBoolValue("walfilepool.use_chunk_file_pool",
+                                          &useChunkFilePoolAsWalPool));
+
+        chunkFilePoolOptions->isAllocated = [=](const std::string& filename) {
+            return Trash::IsChunkOrSnapShotFile(filename) ||
+                   (useChunkFilePoolAsWalPool && Trash::IsWALFile(filename));
+        };
+
         if (0 == chunkFilePoolOptions->bytesPerWrite
             || chunkFilePoolOptions->bytesPerWrite > 1 * 1024 * 1024
             || 0 != chunkFilePoolOptions->bytesPerWrite % 4096) {
@@ -565,6 +611,36 @@ void ChunkServer::InitWalFilePoolOptions(
         std::string metaUri;
         LOG_IF(FATAL, !conf->GetStringValue(
             "walfilepool.meta_path", &metaUri));
+
+        std::string pool_size;
+        LOG_IF(FATAL, !conf->GetStringValue("walfilepool.chunk_file_pool_size",
+                                            &pool_size));
+        LOG_IF(FATAL, !curve::common::ToNumbericByte(
+                          pool_size, &walPoolOptions->filePoolSize));
+        LOG_IF(FATAL, !conf->GetUInt64Value("walfilepool.wal_file_pool_size",
+                                            &walPoolOptions->filePoolSize));
+        LOG_IF(FATAL, !conf->GetBoolValue("walfilepool.allocated_by_percent",
+                                          &walPoolOptions->allocatedByPercent));
+        LOG_IF(FATAL, !conf->GetUInt32Value("walfilepool.allocated_percent",
+                                            &walPoolOptions->allocatedPercent));
+        LOG_IF(FATAL, !conf->GetUInt32Value("walfilepool.thread_num",
+                                            &walPoolOptions->formatThreadNum));
+
+        std::string copysetUri;
+        LOG_IF(FATAL,
+               !conf->GetStringValue("copyset.raft_log_uri", &copysetUri));
+        curve::common::UriParser::ParseUri(copysetUri,
+                                           &walPoolOptions->copysetDir);
+
+        std::string recycleUri;
+        LOG_IF(FATAL,
+               !conf->GetStringValue("copyset.recycler_uri", &recycleUri));
+        curve::common::UriParser::ParseUri(recycleUri,
+                                           &walPoolOptions->recycleDir);
+
+        walPoolOptions->isAllocated = [](const string& filename) {
+            return Trash::IsWALFile(filename);
+        };
         ::memcpy(
             walPoolOptions->metaPath, metaUri.c_str(), metaUri.size());
     }
@@ -833,6 +909,16 @@ void ChunkServer::LoadConfigFromCmdline(common::Configuration *conf) {
         << "chunkFilePoolDir must be set when run chunkserver in command.";
     }
 
+    if (GetCommandLineFlagInfo("chunkFilePoolAllocatedPercent", &info)) {
+        conf->SetUInt32Value("chunkfilepool.allocate_percent",
+                             FLAGS_chunkFilePoolAllocatedPercent);
+    }
+
+    if (GetCommandLineFlagInfo("chunkFormatThreadNum", &info)) {
+        conf->SetUInt64Value("chunkfilepool.chunk_file_pool_format_thread_num",
+                             FLAGS_chunkFormatThreadNum);
+    }
+
     if (GetCommandLineFlagInfo("chunkFilePoolMetaPath", &info) &&
         !info.is_default) {
         conf->SetStringValue(