This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new c6616775b83 [opt](scanner) increase the connection num of s3 client
#26795 (#26796)
c6616775b83 is described below
commit c6616775b83973bda7151c2d540e861b6c93acf0
Author: Mingyu Chen <[email protected]>
AuthorDate: Sun Nov 12 12:13:39 2023 +0800
[opt](scanner) increase the connection num of s3 client #26795 (#26796)
---
be/src/util/s3_util.cpp | 11 ++++++++++-
be/src/vec/exec/scan/scanner_scheduler.cpp | 7 ++++---
be/src/vec/exec/scan/scanner_scheduler.h | 3 +++
3 files changed, 17 insertions(+), 4 deletions(-)
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
index 25d8d7f45cc..8148ad199ac 100644
--- 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
index 3481128a1d2..504c46c4016 100644
--- 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
index a9792d9df94..81dcf5f8b76 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -67,6 +67,8 @@ public:
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 @@ private:
// 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
