add config option: bolt_io_thread_num

ljcui · Feb 18, 2024 · 58ea184 · 58ea184
1 parent a4be034
commit 58ea184
Show file tree

Hide file tree

Showing 5 changed files with 12 additions and 8 deletions.
diff --git a/src/core/global_config.cpp b/src/core/global_config.cpp
@@ -58,7 +58,8 @@ std::map<std::string, std::string> lgraph::GlobalConfig::FormatAsOptions() const
         AddOption(options, "HA node join(s)", ha_node_join_group_s);
         AddOption(options, "Bootstrap Role", ha_bootstrap_role);
     }
-    AddOption(options, "bolt_port", bolt_port);
+    AddOption(options, "bolt port", bolt_port);
+    AddOption(options, "number of bolt io threads", bolt_io_thread_num);
     return options;
 }
 
@@ -210,6 +211,7 @@ fma_common::Configuration lgraph::GlobalConfig::InitConfig
 
     // bolt
     bolt_port = 0;
+    bolt_io_thread_num = 1;
 
     // parse options
     fma_common::Configuration argparser;
@@ -317,5 +319,7 @@ fma_common::Configuration lgraph::GlobalConfig::InitConfig
         .Comment("Node is witness (donot have data & can not apply request) or not.");
     argparser.Add(bolt_port, "bolt_port", true)
         .Comment("Bolt protocol port.");
+    argparser.Add(bolt_io_thread_num, "bolt_io_thread_num", true)
+        .Comment("Number of bolt io threads.");
     return argparser;
 }
diff --git a/src/core/global_config.h b/src/core/global_config.h
@@ -128,6 +128,7 @@ struct BasicConfigs {
     bool enable_realtime_count{};
     // bolt
     int bolt_port = 0;
+    int bolt_io_thread_num = 1;
 };
 
 template <typename T>

diff --git a/src/server/bolt_server.cpp b/src/server/bolt_server.cpp
@@ -22,17 +22,16 @@ boost::asio::io_service workers;
 static boost::asio::io_service listener(BOOST_ASIO_CONCURRENCY_HINT_UNSAFE);
 extern std::function<void(bolt::BoltConnection &conn, bolt::BoltMsg msg,
                           std::vector<std::any> fields)> BoltHandler;
-bool BoltServer::Start(lgraph::StateMachine* sm, int port) {
+bool BoltServer::Start(lgraph::StateMachine* sm, int port, int io_thread_num) {
     sm_ = sm;
-    port_ = port;
     bolt::MarkersInit();
     std::promise<bool> promise;
     std::future<bool> future = promise.get_future();
-    threads_.emplace_back([this, &promise](){
+    threads_.emplace_back([port, io_thread_num, &promise](){
         bool promise_done = false;
         try {
             bolt::IOService<bolt::BoltConnection, decltype(bolt::BoltHandler)> bolt_service(
-                listener, port_, 1, bolt::BoltHandler);
+                listener, port, io_thread_num, bolt::BoltHandler);
             boost::asio::io_service::work holder(listener);
             LOG_INFO() << "bolt server run";
             promise.set_value(true);

diff --git a/src/server/bolt_server.h b/src/server/bolt_server.h
@@ -29,7 +29,7 @@ class BoltServer final {
     }
     DISABLE_COPY(BoltServer);
     DISABLE_MOVE(BoltServer);
-    bool Start(lgraph::StateMachine* sm, int port);
+    bool Start(lgraph::StateMachine* sm, int port, int io_thread_num);
     void Stop();
     ~BoltServer() {Stop();}
     lgraph::StateMachine* StateMachine() {
@@ -38,7 +38,6 @@ class BoltServer final {
  private:
     BoltServer() = default;
     lgraph::StateMachine* sm_ = nullptr;
-    int port_ = 0;
     std::vector<std::thread> threads_;
     bool stopped_ = false;
 };

diff --git a/src/server/lgraph_server.cpp b/src/server/lgraph_server.cpp
@@ -276,7 +276,8 @@ int LGraphServer::Start() {
 
         if (config_->bolt_port > 0) {
             if (!bolt::BoltServer::Instance().Start(state_machine_.get(),
-                                               config_->bolt_port)) {
+                                               config_->bolt_port,
+                                                    config_->bolt_io_thread_num)) {
                 return -1;
             }
         }