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

seawinde · Nov 12, 2023 · 7502d54 · 7502d54
1 parent 33d1051
commit 7502d54
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 6 deletions.
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
@@ -34,7 +34,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 {
 
@@ -162,7 +164,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
@@ -123,13 +123,13 @@ 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);
     static_cast<void>(
             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));
 
@@ -504,4 +504,4 @@ void ScannerScheduler::_deregister_metrics() {
     DEREGISTER_HOOK_METRIC(group_local_scan_thread_pool_thread_num);
 }
 
-} // namespace doris::vectorized
+} // namespace doris::vectorized
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -76,6 +76,8 @@ class ScannerScheduler {
         return _task_group_local_scan_queue.get();
     }
 
+    int remote_thread_pool_max_size() const { return _remote_thread_pool_max_size; }
+
 private:
     // scheduling thread function
     void _schedule_thread(int queue_id);
@@ -118,6 +120,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;
 };
 
 struct SimplifiedScanTask {