【chunkserver】split read and write thread in concurrentApplyModel

opencurve · Sep 23, 2020 · 8e5ff23 · 8e5ff23
1 parent 012200c
commit 8e5ff23
Show file tree

Hide file tree

Showing 30 changed files with 642 additions and 666 deletions.
diff --git a/conf/chunkserver.conf.example b/conf/chunkserver.conf.example
@@ -141,11 +141,14 @@ storeng.sync_write=false
 
 #
 # Concurrent apply module
-#
-# 并发模块的并发度，一般是10
-concurrentapply.size=10
-# 并发模块线程的队列深度
-concurrentapply.queuedepth=1
+# 并发模块写线程的并发度，一般是10
+wconcurrentapply.size=10
+# 并发模块写线程的队列深度
+wconcurrentapply.queuedepth=1
+# 并发模块读线程的并发度，一般是5
+rconcurrentapply.size=5
+# 并发模块读线程的队列深度
+rconcurrentapply.queuedepth=1
 
 #
 # Chunkfile pool

diff --git a/coverage/test.py b/coverage/test.py
diff --git a/coverage/tmp.py b/coverage/tmp.py
diff --git a/curve-ansible/roles/generate_config/defaults/main.yml b/curve-ansible/roles/generate_config/defaults/main.yml
@@ -114,8 +114,10 @@ chunkserver_s3_config_path: /etc/curve/cs_s3.conf
 chunkserver_fs_enable_renameat2: true
 chunkserver_metric_onoff: true
 chunkserver_storeng_sync_write: false
-chunkserver_concurrentapply_size: 10
-chunkserver_concurrentapply_queuedepth: 1
+chunkserver_wconcurrentapply_size: 10
+chunkserver_wconcurrentapply_queuedepth: 1
+chunkserver_rconcurrentapply_size: 5
+chunkserver_rconcurrentapply_queuedepth: 1
 chunkserver_chunkfilepool_enable_get_chunk_from_pool: true
 chunkserver_chunkfilepool_chunk_file_pool_dir: ./0/
 chunkserver_chunkfilepool_cpmeta_file_size: 4096

diff --git a/curve-ansible/roles/generate_config/templates/chunkserver.conf.j2 b/curve-ansible/roles/generate_config/templates/chunkserver.conf.j2
@@ -149,9 +149,13 @@ storeng.sync_write={{ chunkserver_storeng_sync_write }}
 # Concurrent apply module
 #
 # 并发模块的并发度，一般是10
-concurrentapply.size={{ chunkserver_concurrentapply_size }}
+wconcurrentapply.size={{ chunkserver_wconcurrentapply_size }}
 # 并发模块线程的队列深度
-concurrentapply.queuedepth={{ chunkserver_concurrentapply_queuedepth }}
+wconcurrentapply.queuedepth={{ chunkserver_wconcurrentapply_queuedepth }}
+# 并发模块读线程的并发度，一般是5
+rconcurrentapply.size={{ chunkserver_rconcurrentapply_size }}
+# 并发模块读线程的队列深度
+rconcurrentapply.queuedepth={{ chunkserver_rconcurrentapply_queuedepth }}
 
 #
 # Chunkfile pool

diff --git a/deploy/local/chunkserver/conf/chunkserver.conf.0 b/deploy/local/chunkserver/conf/chunkserver.conf.0
@@ -113,6 +113,11 @@ storeng.sync_write=false
 #
 concurrentapply.size=10
 concurrentapply.queuedepth=1
+wconcurrentapply.size=10
+wconcurrentapply.queuedepth=1
+rconcurrentapply.size=5
+rconcurrentapply.queuedepth=1
+
 
 #
 # Chunkfile pool

diff --git a/deploy/local/chunkserver/conf/chunkserver.conf.1 b/deploy/local/chunkserver/conf/chunkserver.conf.1
@@ -111,8 +111,10 @@ storeng.sync_write=false
 #
 # Concurrent apply module
 #
-concurrentapply.size=10
-concurrentapply.queuedepth=1
+wconcurrentapply.size=10
+wconcurrentapply.queuedepth=1
+rconcurrentapply.size=5
+rconcurrentapply.queuedepth=1
 
 #
 # Chunkfile pool

diff --git a/deploy/local/chunkserver/conf/chunkserver.conf.2 b/deploy/local/chunkserver/conf/chunkserver.conf.2
@@ -111,8 +111,10 @@ storeng.sync_write=false
 #
 # Concurrent apply module
 #
-concurrentapply.size=10
-concurrentapply.queuedepth=1
+wconcurrentapply.size=10
+wconcurrentapply.queuedepth=1
+rconcurrentapply.size=5
+rconcurrentapply.queuedepth=1
 
 #
 # Chunkfile pool

diff --git a/src/chunkserver/BUILD b/src/chunkserver/BUILD
@@ -65,6 +65,7 @@ cc_library(
         "//proto:chunkserver-cc-protos",
         "//proto:topology_cc_proto",
         "//src/chunkserver/datastore:chunkserver_datastore",
+        "//src/chunkserver/concurrent_apply:chunkserver_concurrent_apply",
         "//src/chunkserver/raftsnapshot:chunkserver-raft-snapshot",
         "//src/common:curve_common",
         "//src/common:curve_s3_adapter",
@@ -108,6 +109,7 @@ cc_library(
         "//proto:chunkserver-cc-protos",
         "//proto:topology_cc_proto",
         "//src/chunkserver/datastore:chunkserver_datastore",
+        "//src/chunkserver/concurrent_apply:chunkserver_concurrent_apply",
         "//src/chunkserver/raftsnapshot:chunkserver-raft-snapshot",
         "//src/common:curve_common",
         "//src/common:curve_s3_adapter",
@@ -144,6 +146,7 @@ cc_binary(
         "//proto:chunkserver-cc-protos",
         "//src/chunkserver:chunkserver-lib",
         "//src/chunkserver/datastore:chunkserver_datastore",
+        "//src/chunkserver/concurrent_apply:chunkserver_concurrent_apply",
         "//src/chunkserver/raftsnapshot:chunkserver-raft-snapshot",
         "//src/common:curve_common",
         "//src/common:curve_s3_adapter",

diff --git a/src/chunkserver/chunkserver.cpp b/src/chunkserver/chunkserver.cpp
@@ -47,6 +47,7 @@ using ::curve::fs::LocalFileSystem;
 using ::curve::fs::LocalFileSystemOption;
 using ::curve::fs::LocalFsFactory;
 using ::curve::fs::FileSystemType;
+using ::curve::chunkserver::concurrent::ConcurrentApplyModule;
 
 DEFINE_string(conf, "ChunkServer.conf", "Path of configuration file");
 DEFINE_string(chunkServerIp, "127.0.0.1", "chunkserver ip");
@@ -104,11 +105,9 @@ int ChunkServer::Run(int argc, char** argv) {
 
     // 初始化并发持久模块
     ConcurrentApplyModule concurrentapply;
-    int size;
-    LOG_IF(FATAL, !conf.GetIntValue("concurrentapply.size", &size));
-    int qdepth;
-    LOG_IF(FATAL, !conf.GetIntValue("concurrentapply.queuedepth", &qdepth));
-    LOG_IF(FATAL, false == concurrentapply.Init(size, qdepth))
+    ConcurrentApplyOption concurrentApplyOptions;
+    InitConcurrentApplyOptions(&conf, &concurrentApplyOptions);
+    LOG_IF(FATAL, false == concurrentapply.Init(concurrentApplyOptions))
         << "Failed to initialize concurrentapply module!";
 
     // 初始化本地文件系统
@@ -380,6 +379,8 @@ void ChunkServer::Stop() {
     brpc::AskToQuit();
 }
 
+
+
 void ChunkServer::InitChunkFilePoolOptions(
     common::Configuration *conf, ChunkfilePoolOptions *chunkFilePoolOptions) {
     LOG_IF(FATAL, !conf->GetUInt32Value("global.chunk_size",
@@ -408,6 +409,18 @@ void ChunkServer::InitChunkFilePoolOptions(
     }
 }
 
+void ChunkServer::InitConcurrentApplyOptions(common::Configuration *conf,
+        ConcurrentApplyOption *concurrentApplyOptions) {
+    LOG_IF(FATAL, !conf->GetIntValue(
+        "rconcurrentapply.size", &concurrentApplyOptions->rconcurrentsize));
+    LOG_IF(FATAL, !conf->GetIntValue(
+        "wconcurrentapply.size", &concurrentApplyOptions->wconcurrentsize));
+    LOG_IF(FATAL, !conf->GetIntValue(
+        "rconcurrentapply.queuedepth", &concurrentApplyOptions->rqueuedepth));
+    LOG_IF(FATAL, !conf->GetIntValue(
+        "wconcurrentapply.queuedepth", &concurrentApplyOptions->wqueuedepth));
+}
+
 void ChunkServer::InitCopysetNodeOptions(
     common::Configuration *conf, CopysetNodeOptions *copysetNodeOptions) {
     LOG_IF(FATAL, !conf->GetStringValue("global.ip", &copysetNodeOptions->ip));

diff --git a/src/chunkserver/chunkserver.h b/src/chunkserver/chunkserver.h
@@ -32,6 +32,9 @@
 #include "src/chunkserver/register.h"
 #include "src/chunkserver/trash.h"
 #include "src/chunkserver/chunkserver_metrics.h"
+#include "src/chunkserver/concurrent_apply/concurrent_apply.h"
+
+using ::curve::chunkserver::concurrent::ConcurrentApplyOption;
 
 namespace curve {
 namespace chunkserver {
@@ -56,6 +59,9 @@ class ChunkServer {
     void InitChunkFilePoolOptions(common::Configuration *conf,
         ChunkfilePoolOptions *chunkFilePoolOptions);
 
+    void InitConcurrentApplyOptions(common::Configuration *conf,
+        ConcurrentApplyOption *concurrentApplyOption);
+
     void InitCopysetNodeOptions(common::Configuration *conf,
         CopysetNodeOptions *copysetNodeOptions);
 

diff --git a/src/chunkserver/concurrent_apply.cpp b/src/chunkserver/concurrent_apply.cpp