This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 66054a5c78f [opt](scanner) increase the connection num of s3 client
(#26795)
66054a5c78f is described below
commit 66054a5c78fff72bbf8ab9a6ccadea4f90937523
Author: Mingyu Chen <[email protected]>
AuthorDate: Sun Nov 12 14:29:11 2023 +0800
[opt](scanner) increase the connection num of s3 client (#26795)
---
be/src/util/s3_util.cpp | 11 ++++++++++-
be/src/vec/exec/scan/scanner_scheduler.cpp | 10 +++++-----
be/src/vec/exec/scan/scanner_scheduler.h | 3 +++
3 files changed, 18 insertions(+), 6 deletions(-)
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
index 830cb4cc4e2..442bb7b205a 100644
--- 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
index e78e8dceffe..6c40ccc242e 100644
--- 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
\ No newline at end of file
+} // namespace doris::vectorized
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h
b/be/src/vec/exec/scan/scanner_scheduler.h
index a6d450bc221..9825085fb1c 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -76,6 +76,8 @@ public:
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 @@ private:
// true is the scheduler is closed.
std::atomic_bool _is_closed = {false};
bool _is_init = false;
+ int _remote_thread_pool_max_size;
};
struct SimplifiedScanTask {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
