[feat][client] Add aws thread pool option and bthread work num option

Author: chuandew

conf/client.conf

@@ -96,6 +96,8 @@ fuseClient.warmupThreadsNum=10
 # if no disk cache this option will be ignored
 fuseClient.in_time_warmup=false
 
+fuseClient.bthread_worker_num=0
+
 # the write throttle bps of fuseClient, default no limit
 fuseClient.throttle.avgWriteBytes=0
 # the write burst bps of fuseClient, default no limit
@@ -194,7 +196,6 @@ data_stream.slice.stay_in_memory_max_second=5
 data_stream.page.size=65536
 data_stream.page.total_size_mb=1024
 data_stream.page.use_pool=true
-data_stream.s3.async_upload_workers=256
 # }
 
 #### block cache
@@ -276,6 +277,9 @@ s3.maxReadRetryIntervalMs = 1000
 # retry interval
 s3.readRetryIntervalMs = 100
 
+s3.use_thread_pool=true
+s3.async_thread_num_in_thread_pool=256
+
 s3.enableTelemetry=false
 
 #### common

conf/mds.conf

@@ -123,7 +123,7 @@ s3.requestTimeout=10000
 # Off = 0,Fatal = 1,Error = 2,Warn = 3,Info = 4,Debug = 5,Trace = 6
 s3.logLevel=4
 s3.logPrefix=/data/logs/dingofs/aws_  # __DINGOADM_TEMPLATE__ /dingofs/client/logs/aws_ __DINGOADM_TEMPLATE__
-data_stream.s3.async_upload_workers=30
+s3.async_thread_num_in_thread_pool=10
 # limit all inflight async requests' bytes, |0| means not limited
 s3.maxAsyncRequestInflightBytes=104857600
 # throttle

conf/metaserver.conf

@@ -20,6 +20,7 @@ s3.requestTimeout=10000
 s3.logLevel=4
 s3.logPrefix=/tmp/dingofs/metaserver/aws_
 data_stream.s3.async_upload_workers=10
+s3.async_thread_num_in_thread_pool=10
 # throttle
 s3.throttle.iopsTotalLimit=0
 s3.throttle.iopsReadLimit=0

src/aws/s3_adapter.cpp

@@ -100,7 +100,20 @@ void InitS3AdaptorOptionExceptS3InfoOption(Configuration* conf,
          !conf->GetIntValue("s3.connectTimeout", &s3_opt->connectTimeout));
   LOG_IF(FATAL,
          !conf->GetIntValue("s3.requestTimeout", &s3_opt->requestTimeout));
-  LOG_IF(FATAL, !conf->GetIntValue("data_stream.s3.async_upload_workers",
+
+  if (!conf->GetBoolValue("s3.use_thread_pool", &s3_opt->use_thread_pool)) {
+    LOG(INFO) << "Not found s3.use_thread_pool in conf, use default "
+              << (s3_opt->use_thread_pool ? "true" : "false");
+  }
+
+  if (!conf->GetIntValue("s3.async_thread_num_in_thread_pool",
+                         &s3_opt->asyncThreadNum)) {
+    LOG(INFO)
+        << "Not found s3.async_thread_num_in_thread_pool in conf, use default"
+        << s3_opt->asyncThreadNum;
+  }
+
+  LOG_IF(FATAL, !conf->GetIntValue("s3.async_thread_num_in_thread_pool",
                                    &s3_opt->asyncThreadNum));
   LOG_IF(FATAL, !conf->GetUInt64Value("s3.throttle.iopsTotalLimit",
                                       &s3_opt->iopsTotalLimit));
@@ -173,11 +186,14 @@ void S3Adapter::Init(const S3AdapterOption& option) {
   clientCfg_->connectTimeoutMs = option.connectTimeout;
   clientCfg_->requestTimeoutMs = option.requestTimeout;
   clientCfg_->endpointOverride = s3Address_;
-  int async_thread_num = option.asyncThreadNum;
-  LOG(INFO) << "S3Adapter init thread num = " << async_thread_num << std::endl;
-  clientCfg_->executor =
-      Aws::MakeShared<Aws::Utils::Threading::PooledThreadExecutor>(
-          "S3Adapter.S3Client", async_thread_num);
+
+  if (option.use_thread_pool) {
+    LOG(INFO) << "S3Adapter init async thread pool thread num = "
+              << option.asyncThreadNum;
+    clientCfg_->executor =
+        Aws::MakeShared<Aws::Utils::Threading::PooledThreadExecutor>(
+            "S3Adapter.S3Client", option.asyncThreadNum);
+  }
 
   if (option.enableTelemetry) {
     LOG(INFO) << "Enable telemetry for aws s3 adapter";

src/aws/s3_adapter.h

@@ -82,7 +82,8 @@ struct S3AdapterOption {
   int maxConnections;
   int connectTimeout;
   int requestTimeout;
-  int asyncThreadNum;
+  bool use_thread_pool{true};
+  int asyncThreadNum{256};
   uint64_t maxAsyncRequestInflightBytes;
   uint64_t iopsTotalLimit;
   uint64_t iopsReadLimit;

src/client/common/config.cpp

@@ -429,13 +429,19 @@ void InitFuseClientOption(Configuration* conf, FuseClientOption* clientOption) {
   conf->GetValueFatalIfFail("fuseClient.warmupThreadsNum",
                             &clientOption->warmupThreadsNum);
 
-  if (!conf->GetBoolValue("fuseClient.in_time_warmup",
-                          &FLAGS_in_time_warmup)) {
+  if (!conf->GetBoolValue("fuseClient.in_time_warmup", &FLAGS_in_time_warmup)) {
     LOG(INFO) << "Not found `fuseClient.in_time_warmup` in conf, default to "
                  "false";
     FLAGS_in_time_warmup = false;
   }
 
+  if (!conf->GetIntValue("fuseClient.bthread_worker_num",
+                         &FLAGS_bthread_worker_num)) {
+    FLAGS_bthread_worker_num = 0;
+    LOG(INFO)
+        << "Not found `fuseClient.bthread_worker_num` in conf, default to 0";
+  }
+
   LOG_IF(WARNING, conf->GetBoolValue("fuseClient.enableSplice",
                                      &clientOption->enableFuseSplice))
       << "Not found `fuseClient.enableSplice` in conf, use default value `"

src/client/common/dynamic_config.cpp

@@ -125,6 +125,9 @@ DEFINE_validator(s3_prefetch, &PassBool);
 DEFINE_bool(in_time_warmup, false, "in time warmup inode when enable");
 DEFINE_validator(in_time_warmup, &PassBool);
 
+
+DEFINE_int32(bthread_worker_num, 0, "bthread worker num");
+
 }  // namespace common
 }  // namespace client
 }  // namespace dingofs

src/client/common/dynamic_config.h

@@ -24,6 +24,7 @@
 #define DINGOFS_SRC_CLIENT_COMMON_DYNAMIC_CONFIG_H_
 
 #include <gflags/gflags.h>
+#include <gflags/gflags_declare.h>
 
 namespace dingofs {
 namespace client {
@@ -80,6 +81,8 @@ DECLARE_bool(s3_prefetch);
 
 DECLARE_bool(in_time_warmup);
 
+DECLARE_int32(bthread_worker_num);
+
 }  // namespace common
 }  // namespace client
 }  // namespace dingofs

src/client/dingo_fuse_op.cpp

@@ -24,15 +24,18 @@
 
 #include <glog/logging.h>
 
+#include <cstdint>
 #include <cstring>
 #include <memory>
 #include <string>
 #include <vector>
 
 #include "aws/s3_access_log.h"
+#include "bthread/bthread.h"
 #include "client/blockcache/log.h"
 #include "client/common/common.h"
 #include "client/common/config.h"
+#include "client/common/dynamic_config.h"
 #include "client/filesystem/access_log.h"
 #include "client/filesystem/error.h"
 #include "client/filesystem/meta.h"
@@ -82,6 +85,10 @@ static FuseClient* g_client_instance = nullptr;
 static FuseClientOption* g_fuse_client_option = nullptr;
 static ClientOpMetric* g_clientOpMetric = nullptr;
 
+namespace bthread {
+DECLARE_int32(bthread_concurrency);
+}  // namespace bthread
+
 namespace {
 
 void InitFuseConnInfo(struct fuse_conn_info* conn, const FuseConnInfo& option) {
@@ -174,8 +181,8 @@ int InitLog(const char* conf_path, const char* argv0) {
 
   bool succ = InitAccessLog(FLAGS_log_dir) &&
               InitBlockCacheLog(FLAGS_log_dir) &&
-              dingofs::aws::InitS3AccessLog(FLAGS_log_dir)
-              && dingofs::stub::InitMetaAccessLog(FLAGS_log_dir);
+              dingofs::aws::InitS3AccessLog(FLAGS_log_dir) &&
+              dingofs::stub::InitMetaAccessLog(FLAGS_log_dir);
   if (!succ) {
     return -1;
   }
@@ -199,6 +206,13 @@ int InitFuseClient(const struct MountOption* mount_option) {
   g_fuse_client_option = new FuseClientOption();
   dingofs::client::common::InitFuseClientOption(&conf, g_fuse_client_option);
 
+  int32_t bthread_worker_num =
+      dingofs::client::common::FLAGS_bthread_worker_num;
+  if (bthread_worker_num > 0) {
+    bthread::FLAGS_bthread_concurrency = bthread_worker_num;
+    LOG(INFO) << "set bthread concurrency to " << bthread_worker_num;
+  }
+
   auto fs_info = std::make_shared<FsInfo>();
   if (GetFsInfo(mount_option->fsName, fs_info.get()) != 0) {
     return -1;

src/client/fuse_s3_client.cpp

@@ -26,6 +26,7 @@
 
 #include "client/blockcache/block_cache.h"
 #include "client/blockcache/s3_client.h"
+#include "client/common/dynamic_config.h"
 #include "client/datastream/data_stream.h"
 #include "client/filesystem/meta.h"
 #include "client/kvclient/memcache_client.h"
@@ -566,6 +567,9 @@ DINGOFS_ERROR FuseS3Client::InitBrpcServer() {
   }
 
   brpc::ServerOptions brpc_server_options;
+  if (common::FLAGS_bthread_worker_num > 0) {
+    brpc_server_options.num_threads = common::FLAGS_bthread_worker_num;
+  }
 
   uint32_t listen_port = 0;
   if (!StartBrpcServer(server_, &brpc_server_options,