【chunkserver】split read thread and write thread in concurrentApplyModel

opencurve · Aug 26, 2020 · f874e36 · f874e36
1 parent d0bf4c1
commit f874e36
Show file tree

Hide file tree

Showing 25 changed files with 586 additions and 631 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/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
@@ -41,6 +41,7 @@
 #include "src/chunkserver/raftsnapshot/curve_snapshot_attachment.h"
 #include "src/chunkserver/raftsnapshot/curve_file_service.h"
 #include "src/chunkserver/raftsnapshot/curve_snapshot_storage.h"
+#include "src/chunkserver/concurrent_apply/concurrent_apply.h"
 #include "src/common/curve_version.h"
 
 using ::curve::fs::LocalFileSystem;
@@ -103,12 +104,14 @@ int ChunkServer::Run(int argc, char** argv) {
         << "Failed to init chunkserver metric.";
 
     // 初始化并发持久模块
-    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))
+    ::curve::chunkserver::concurrent::ConcurrentApplyModule concurrentapply;
+    int rsize, wsize;
+    LOG_IF(FATAL, !conf.GetIntValue("rconcurrentapply.size", &rsize));
+    LOG_IF(FATAL, !conf.GetIntValue("wconcurrentapply.size", &wsize));
+    int rqdepth, wqdepth;
+    LOG_IF(FATAL, !conf.GetIntValue("rconcurrentapply.queuedepth", &rqdepth));
+    LOG_IF(FATAL, !conf.GetIntValue("wconcurrentapply.queuedepth", &wqdepth));
+    LOG_IF(FATAL, false == concurrentapply.Init(wsize, wqdepth, rsize, rqdepth))
         << "Failed to initialize concurrentapply module!";
 
     // 初始化本地文件系统

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