Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import com.alibaba.csp.sentinel.log.RecordLog;
Expand Down Expand Up @@ -129,13 +130,24 @@ private Map<String, List<ParamFlowRule>> aggregateAndPrepareParamRules(List<Para
for (ParamFlowRule rule : oldRuleList) {
ParameterMetric parameterMetric = ParameterMetricStorage.getParamMetricForResource(resource);
if (parameterMetric != null) {
parameterMetric.clearForRule(rule);
// The new rules of the resource may not be initialized yet, so the shared
// thread counter of the index must be kept while they still use it.
parameterMetric.clearForRule(rule, isParamIdxLoaded(newRuleList, rule.getParamIdx()));
}
}
}

return newRuleMap;
}

private static boolean isParamIdxLoaded(List<ParamFlowRule> rules, Integer paramIdx) {
for (ParamFlowRule rule : rules) {
if (Objects.equals(rule.getParamIdx(), paramIdx)) {
return true;
}
}
return false;
}
}

private ParamFlowRuleManager() {}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
Expand Down Expand Up @@ -88,11 +89,46 @@ public CacheMap<Object, AtomicLong> getRuleTimeCounter(ParamFlowRule rule) {
}

public void clearForRule(ParamFlowRule rule) {
clearForRule(rule, false);
}

/**
* Clear the counters of a rule that is not loaded any more. The shared thread counter of the
* parameter index is kept as long as another rule of the metric or one of the rules that are
* loaded by the same update still uses that index, even if the latter is not initialized yet.
*
* @param rule the rule that is removed
* @param paramIdxUsedByNewRules whether the rules loaded by the same update use the parameter index of the rule
*/
void clearForRule(ParamFlowRule rule, boolean paramIdxUsedByNewRules) {
synchronized (lock) {
ruleTimeCounters.remove(rule);
ruleTokenCounter.remove(rule);
threadCountMap.remove(rule.getParamIdx());
// The thread counters are shared by all the rules with the same parameter index,
// so the counter of the index should only be removed when no other rule still
// relies on it. Otherwise the in-flight thread count of the surviving rules
// would be dropped, making them under-counted. The rules loaded by the update are
// not initialized before their first entry, so they are taken into account
// explicitly.
Integer paramIdx = rule.getParamIdx();
if (!paramIdxUsedByNewRules && !isParamIdxInUse(paramIdx)) {
threadCountMap.remove(paramIdx);
}
}
}

private boolean isParamIdxInUse(Integer paramIdx) {
for (ParamFlowRule rule : ruleTimeCounters.keySet()) {
if (Objects.equals(rule.getParamIdx(), paramIdx)) {
return true;
}
}
for (ParamFlowRule rule : ruleTokenCounter.keySet()) {
if (Objects.equals(rule.getParamIdx(), paramIdx)) {
return true;
}
}
return false;
}

public void initialize(ParamFlowRule rule) {
Expand Down
Loading