This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 2a5e992a3a6 [Enhancement](cloud) Add concurrency for tablet stats
processing (#43903)
2a5e992a3a6 is described below
commit 2a5e992a3a67662d40c81933eb0568a7429eb026
Author: abmdocrt <[email protected]>
AuthorDate: Fri Nov 29 10:22:33 2024 +0800
[Enhancement](cloud) Add concurrency for tablet stats processing (#43903)
---
.../main/java/org/apache/doris/common/Config.java | 4 ++
.../apache/doris/catalog/CloudTabletStatMgr.java | 54 ++++++++++++++--------
2 files changed, 39 insertions(+), 19 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 3caa137c858..2ef391bb7ab 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -3222,6 +3222,10 @@ public class Config extends ConfigBase {
+ "model is set to 300 times, which is approximately 5 minutes by
default."})
public static int auto_start_wait_to_resume_times = 300;
+ @ConfField(description = {"Get tablet stat task的最大并发数。",
+ "Maximal concurrent num of get tablet stat job."})
+ public static int max_get_tablet_stat_task_threads_num = 4;
+
// ATTN: DONOT add any config not related to cloud mode here
// ATTN: DONOT add any config not related to cloud mode here
// ATTN: DONOT add any config not related to cloud mode here
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/CloudTabletStatMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/CloudTabletStatMgr.java
index 0ba2686060e..309b433572c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/CloudTabletStatMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/CloudTabletStatMgr.java
@@ -36,6 +36,10 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
/*
* CloudTabletStatMgr is for collecting tablet(replica) statistics from
backends.
@@ -47,6 +51,9 @@ public class CloudTabletStatMgr extends MasterDaemon {
// <(dbId, tableId) -> OlapTable.Statistics>
private volatile Map<Pair<Long, Long>, OlapTable.Statistics>
cloudTableStatsMap = new HashMap<>();
+ private static final ExecutorService GET_TABLET_STATS_THREAD_POOL =
Executors.newFixedThreadPool(
+ Config.max_get_tablet_stat_task_threads_num);
+
public CloudTabletStatMgr() {
super("cloud tablet stat mgr",
Config.tablet_stat_update_interval_second * 1000);
}
@@ -103,28 +110,37 @@ public class CloudTabletStatMgr extends MasterDaemon {
reqList.add(builder.build());
}
+ List<Future<Void>> futures = new ArrayList<>();
for (GetTabletStatsRequest req : reqList) {
- GetTabletStatsResponse resp;
- try {
- resp = getTabletStats(req);
- } catch (RpcException e) {
- LOG.info("get tablet stats exception:", e);
- continue;
- }
-
- if (resp.getStatus().getCode() != MetaServiceCode.OK) {
- continue;
- }
-
- if (LOG.isDebugEnabled()) {
- int i = 0;
- for (TabletIndexPB idx : req.getTabletIdxList()) {
- LOG.debug("db_id: {} table_id: {} index_id: {} tablet_id:
{} size: {}",
- idx.getDbId(), idx.getTableId(), idx.getIndexId(),
idx.getTabletId(),
- resp.getTabletStats(i++).getDataSize());
+ futures.add(GET_TABLET_STATS_THREAD_POOL.submit(() -> {
+ GetTabletStatsResponse resp =
GetTabletStatsResponse.newBuilder().build();
+ try {
+ resp = getTabletStats(req);
+ } catch (RpcException e) {
+ LOG.warn("get tablet stats exception:", e);
+ }
+ if (resp.getStatus().getCode() != MetaServiceCode.OK) {
+ LOG.warn("get tablet stats return failed.");
+ }
+ if (LOG.isDebugEnabled()) {
+ int i = 0;
+ for (TabletIndexPB idx : req.getTabletIdxList()) {
+ LOG.debug("db_id: {} table_id: {} index_id: {}
tablet_id: {} size: {}",
+ idx.getDbId(), idx.getTableId(),
idx.getIndexId(),
+ idx.getTabletId(),
resp.getTabletStats(i++).getDataSize());
+ }
}
+ updateTabletStat(resp);
+ return null;
+ }));
+ }
+
+ try {
+ for (Future<Void> future : futures) {
+ future.get();
}
- updateTabletStat(resp);
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.error("Error waiting for get tablet stats tasks to complete",
e);
}
LOG.info("finished to get tablet stat of all backends. cost: {} ms",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
