[opt](scanner) increase the connection num of s3 client

apache · Nov 12, 2023 · c359c5d · c359c5d
1 parent 92581d3
commit c359c5d
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 4 deletions.
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
@@ -33,7 +33,9 @@
 
 #include "common/config.h"
 #include "common/logging.h"
+#include "runtime/exec_env.h"
 #include "s3_uri.h"
+#include "vec/exec/scan/scanner_scheduler.h"
 
 namespace doris {
 
@@ -149,7 +151,14 @@ std::shared_ptr<Aws::S3::S3Client> S3ClientFactory::create(const S3Conf& s3_conf
  if (s3_conf.max_connections > 0) {
  aws_config.maxConnections = s3_conf.max_connections;
  } else {
- aws_config.maxConnections = config::doris_remote_scanner_thread_pool_thread_num;
+#ifdef BE_TEST
+ // the S3Client may shared by many threads.
+ // So need to set the number of connections large enough.
+ aws_config.maxConnections = config::doris_scanner_thread_pool_thread_num;
+#else
+ aws_config.maxConnections =
+ ExecEnv::GetInstance()->scanner_scheduler()->remote_thread_pool_max_size();
+#endif
  }
 
  if (s3_conf.request_timeout_ms > 0) {

diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -99,11 +99,12 @@ Status ScannerScheduler::init(ExecEnv* env) {
  config::doris_scanner_thread_pool_queue_size, "local_scan"));
 
  // 3. remote scan thread pool
+ _remote_thread_pool_max_size = config::doris_max_remote_scanner_thread_pool_thread_num != -1
+ ? config::doris_max_remote_scanner_thread_pool_thread_num
+ : std::max(512, CpuInfo::num_cores() * 10);
  ThreadPoolBuilder("RemoteScanThreadPool")
  .set_min_threads(config::doris_scanner_thread_pool_thread_num) // 48 default
- .set_max_threads(config::doris_max_remote_scanner_thread_pool_thread_num != -1
- ? config::doris_max_remote_scanner_thread_pool_thread_num
- : std::max(512, CpuInfo::num_cores() * 10))
+ .set_max_threads(_remote_thread_pool_max_size)
  .set_max_queue_size(config::doris_scanner_thread_pool_queue_size)
  .build(&_remote_scan_thread_pool);
 

diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -67,6 +67,8 @@ class ScannerScheduler {
  std::unique_ptr<ThreadPoolToken> new_limited_scan_pool_token(ThreadPool::ExecutionMode mode,
  int max_concurrency);
 
+ int remote_thread_pool_max_size() const { return _remote_thread_pool_max_size; }
+
 private:
  // scheduling thread function
  void _schedule_thread(int queue_id);
@@ -101,6 +103,7 @@ class ScannerScheduler {
  // true is the scheduler is closed.
  std::atomic_bool _is_closed = {false};
  bool _is_init = false;
+ int _remote_thread_pool_max_size;
 };
 
 } // namespace doris::vectorized