[ISSUE #1035] [C++] Supports configuring the number of client threads (#1155)

Author: lizhimins

cpp/include/rocketmq/Configuration.h

@@ -51,6 +51,10 @@ class Configuration {
     return tls_;
   }
 
+  std::uint32_t callbackThreads() const {
+    return callback_threads_;
+  }
+
 protected:
   friend class ConfigurationBuilder;
 
@@ -62,6 +66,7 @@ class Configuration {
   CredentialsProviderPtr    credentials_provider_;
   std::chrono::milliseconds request_timeout_{ConfigurationDefaults::RequestTimeout};
   bool tls_ = true;
+  std::uint32_t callback_threads_{2};
 };
 
 class ConfigurationBuilder {
@@ -76,6 +81,8 @@ class ConfigurationBuilder {
 
   ConfigurationBuilder& withSsl(bool with_ssl);
 
+  ConfigurationBuilder& withCallbackThreads(std::uint32_t callback_threads);
+
   Configuration build();
 
 private:

cpp/include/rocketmq/Producer.h

@@ -94,7 +94,7 @@ class ProducerBuilder {
 public:
   ProducerBuilder();
 
-  ProducerBuilder& withConfiguration(Configuration configuration);
+  ProducerBuilder& withConfiguration(const Configuration& configuration);
 
   ProducerBuilder& withTopics(const std::vector<std::string>& topics);
 

cpp/source/base/Configuration.cpp

@@ -48,6 +48,11 @@ ConfigurationBuilder& ConfigurationBuilder::withSsl(bool with_ssl) {
   return *this;
 }
 
+ConfigurationBuilder& ConfigurationBuilder::withCallbackThreads(std::uint32_t callback_threads){
+  configuration_.callback_threads_ = callback_threads;
+  return *this;
+}
+
 Configuration ConfigurationBuilder::build() {
   return std::move(configuration_);
 }

cpp/source/base/ThreadPoolImpl.cpp

@@ -22,8 +22,6 @@
 #include "rocketmq/RocketMQ.h"
 #include "rocketmq/State.h"
 #include "spdlog/spdlog.h"
-#include <atomic>
-#include <cstdint>
 #include <exception>
 #include <system_error>
 
@@ -33,6 +31,7 @@ ThreadPoolImpl::ThreadPoolImpl(std::uint16_t workers)
     : work_guard_(
           absl::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(context_.get_executor())),
       workers_(workers) {
+    SPDLOG_INFO("ThreadPoolImpl created worker threads {}", workers);
 }
 
 void ThreadPoolImpl::start() {

cpp/source/client/ClientManagerImpl.cpp

@@ -32,7 +32,6 @@
 #include "ReceiveMessageContext.h"
 #include "RpcClient.h"
 #include "RpcClientImpl.h"
-#include "Scheduler.h"
 #include "SchedulerImpl.h"
 #include "UtilAll.h"
 #include "google/protobuf/util/time_util.h"
@@ -42,11 +41,11 @@
 
 ROCKETMQ_NAMESPACE_BEGIN
 
-ClientManagerImpl::ClientManagerImpl(std::string resource_namespace, bool with_ssl)
-    : scheduler_(std::make_shared<SchedulerImpl>()),
+ClientManagerImpl::ClientManagerImpl(std::string resource_namespace, bool with_ssl, int thread_count)
+    : scheduler_(std::make_shared<SchedulerImpl>(2)),
       resource_namespace_(std::move(resource_namespace)),
       state_(State::CREATED),
-      callback_thread_pool_(absl::make_unique<ThreadPoolImpl>(std::thread::hardware_concurrency())),
+      callback_thread_pool_(absl::make_unique<ThreadPoolImpl>(thread_count)),
       with_ssl_(with_ssl) {
 
   certificate_verifier_ = grpc::experimental::ExternalCertificateVerifier::Create<InsecureCertificateVerifier>();

cpp/source/client/include/ClientConfig.h

@@ -62,6 +62,7 @@ struct ClientConfig {
   SubscriberConfig subscriber;
   Metric metric;
   bool withSsl;
+  std::uint32_t callback_threads{2};
   std::unique_ptr<opencensus::trace::Sampler> sampler_;
 };
 

cpp/source/client/include/ClientManagerImpl.h

@@ -48,7 +48,7 @@ class ClientManagerImpl : virtual public ClientManager, public std::enable_share
    * effectively.
    * @param resource_namespace Abstract resource namespace, in which this client manager lives.
    */
-  explicit ClientManagerImpl(std::string resource_namespace, bool with_ssl = true);
+  explicit ClientManagerImpl(std::string resource_namespace, bool with_ssl = true, int thread_count = 1);
 
   ~ClientManagerImpl() override;
 

cpp/source/rocketmq/ClientImpl.cpp

@@ -108,7 +108,9 @@ void ClientImpl::start() {
   client_config_.client_id = clientId();
   if (!client_manager_) {
     client_manager_ = std::make_shared<ClientManagerImpl>(
-        client_config_.resource_namespace, client_config_.withSsl);
+        client_config_.resource_namespace,
+        client_config_.withSsl,
+        client_config_.callback_threads);
     client_manager_->start();
   }
 
@@ -118,6 +120,11 @@ void ClientImpl::start() {
     return;
   }
 
+  // A gRPC I/O thread pool is created upon establishing a connection.
+  //   - https://github.com/grpc/grpc/issues/28642
+  //   - https://github.com/grpc/grpc/pull/31662
+  // The source code initializes the number of I/O threads as follows:
+  //   int num_io_threads = grpc_core::Clamp(gpr_cpu_num_cores() / 2, 2u, 16u);
   while (true) {
     createSession(endpoint, false);
     {

cpp/source/rocketmq/Producer.cpp

@@ -76,14 +76,15 @@ ProducerBuilder Producer::newBuilder() {
   return {};
 }
 
-ProducerBuilder::ProducerBuilder() : impl_(std::make_shared<ProducerImpl>()){};
+ProducerBuilder::ProducerBuilder() : impl_(std::make_shared<ProducerImpl>()){}
 
-ProducerBuilder& ProducerBuilder::withConfiguration(Configuration configuration) {
+ProducerBuilder& ProducerBuilder::withConfiguration(const Configuration& configuration) {
   auto name_server_resolver = std::make_shared<StaticNameServerResolver>(configuration.endpoints());
   impl_->withNameServerResolver(std::move(name_server_resolver));
   impl_->withResourceNamespace(configuration.resourceNamespace());
   impl_->withCredentialsProvider(configuration.credentialsProvider());
   impl_->withRequestTimeout(configuration.requestTimeout());
+  impl_->withCallbackThreads(configuration.callbackThreads());
   impl_->withSsl(configuration.withSsl());
   return *this;
 }

cpp/source/rocketmq/PushConsumer.cpp

@@ -44,10 +44,11 @@ PushConsumer PushConsumerBuilder::build() {
   impl->consumeThreadPoolSize(consume_thread_);
   impl->withNameServerResolver(std::make_shared<StaticNameServerResolver>(configuration_.endpoints()));
   impl->withResourceNamespace(configuration_.resourceNamespace());
-  impl->withSsl(configuration_.withSsl());
   impl->withCredentialsProvider(configuration_.credentialsProvider());
   impl->withRequestTimeout(configuration_.requestTimeout());
   impl->withFifoConsumeAccelerator(fifo_consume_accelerator_);
+  impl->withCallbackThreads(configuration_.callbackThreads());
+  impl->withSsl(configuration_.withSsl());
   impl->start();
   return PushConsumer(impl);
 }