Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3843861810
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -186,53 +181,75 @@ public void
updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
return;
}
long beId = params.backend_id;
- // NOTE(wb) one be sends update request one by one,
- // so there is no need a global lock for beToQueryStatsMap here,
- // just keep one be's put/remove/get is atomic operation is enough
long currentTime = System.currentTimeMillis();
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (beReportInfo == null) {
- beReportInfo = new BeReportInfo(currentTime);
- beToQueryStatsMap.put(beId, beReportInfo);
- } else {
+ // Serialize updates and expiry only per BE. A final update
acknowledged here must not be
+ // removed by a cleaner that made its timeout decision on an older
value.
+ beToQueryStatsMap.compute(beId, (ignored, previousInfo) -> {
+ BeReportInfo beReportInfo = previousInfo == null
+ ? new BeReportInfo(currentTime) : previousInfo;
beReportInfo.beLastReportTime = currentTime;
+ for (Map.Entry<String, TQueryStatisticsResult> entry
+ : params.query_statistics_result_map.entrySet()) {
+ Pair<Long, TQueryStatisticsResult> incoming =
Pair.of(currentTime, entry.getValue());
+ beReportInfo.queryStatsMap.compute(entry.getKey(), (queryId,
previous) ->
+ previous == null ||
shouldReplaceQueryStatistics(previous.second, incoming.second)
+ ? incoming : previous);
+ }
+ return beReportInfo;
+ });
+ }
+
+ private boolean shouldReplaceQueryStatistics(TQueryStatisticsResult
previous,
+ TQueryStatisticsResult incoming) {
+ long previousGeneration = previous.isSetQueryStatisticsGeneration()
+ ? previous.getQueryStatisticsGeneration() : 0;
+ long incomingGeneration = incoming.isSetQueryStatisticsGeneration()
+ ? incoming.getQueryStatisticsGeneration() : 0;
+ if (previousGeneration != incomingGeneration) {
+ return incomingGeneration > previousGeneration;
}
- for (Map.Entry<String, TQueryStatisticsResult> entry :
params.query_statistics_result_map.entrySet()) {
- beReportInfo.queryStatsMap.put(entry.getKey(),
Pair.of(currentTime, entry.getValue()));
+
+ long previousSequence = previous.isSetQueryStatisticsSequence()
+ ? previous.getQueryStatisticsSequence() : 0;
+ long incomingSequence = incoming.isSetQueryStatisticsSequence()
+ ? incoming.getQueryStatisticsSequence() : 0;
+ if (previousSequence != incomingSequence) {
+ return incomingSequence > previousSequence;
}
+
+ // Legacy BEs have no ordering fields; once their terminal snapshot
arrives, an older
+ // in-flight periodic snapshot must not make the same query look
unfinished again.
+ return !previous.isSetQueryFinished() || !previous.isQueryFinished()
+ || (incoming.isSetQueryFinished() &&
incoming.isQueryFinished());
}
- private void clearReportTimeoutBeStatistics() {
+ void clearReportTimeoutBeStatistics() {
// 1 clear report timeout be
Set<Long> currentBeIdSet = beToQueryStatsMap.keySet();
Long currentTime = System.currentTimeMillis();
for (Long beId : currentBeIdSet) {
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (currentTime - beReportInfo.beLastReportTime >
Config.be_report_query_statistics_timeout_ms) {
- beToQueryStatsMap.remove(beId);
- continue;
- }
- Set<String> queryIdSet = beReportInfo.queryStatsMap.keySet();
- for (String queryId : queryIdSet) {
- Pair<Long, TQueryStatisticsResult> pair =
beReportInfo.queryStatsMap.get(queryId);
- long queryLastReportTime = pair.first;
- boolean timeout = currentTime - queryLastReportTime
- > Config.be_report_query_statistics_timeout_ms;
- // Remove query statistics only when both conditions are
satisfied:
- // 1) this query statistics is timeout, and
- // 2) FE no longer has this query in QeProcessorImpl.
- // Example timeline:
- // - t0: query q1 is still running, but one periodic BE report
is delayed for > timeout.
- // - t1: clear thread runs. timeout condition is true, but q1
still exists in FE.
- // - t2: we keep q1 statistics instead of removing it; later
reports can update it again.
- if (timeout && isQueryNotExistInFe(queryId)) {
- beReportInfo.queryStatsMap.remove(queryId);
+ beToQueryStatsMap.computeIfPresent(beId, (ignored, beReportInfo)
-> {
Review Comment:
Fixed in e81012f040. Top-level BE entry creation and empty-shell removal are
now bounded. Statistics updates and timeout expiry use per-query
ConcurrentMap.compute operations; a short per-BE lifecycle read/write lock only
prevents insertion into a shell being removed. Separate latch-controlled tests
verify that both a 1,001-query bulk update and a 1,001-query cleanup cannot
block an independent completion update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
