From 04e791b153fbac95896a25d872a78d174e6b8e64 Mon Sep 17 00:00:00 2001 From: hutiefang76 <137664623+hutiefang76@users.noreply.github.com> Date: Wed, 7 Oct 2026 16:23:48 +0800 Subject: [PATCH 1/2] Preserve shared parameter thread counters when removing a rule --- .../block/flow/param/ParameterMetric.java | 24 +++++++- .../flow/param/ParamFlowRuleManagerTest.java | 41 +++++++++++++ .../block/flow/param/ParameterMetricTest.java | 59 +++++++++++++++++++ 3 files changed, 123 insertions(+), 1 deletion(-) diff --git a/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java b/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java index a5ab902c27..7c0f3b5b27 100644 --- a/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java +++ b/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java @@ -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; @@ -91,8 +92,29 @@ public void clearForRule(ParamFlowRule rule) { 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. + Integer paramIdx = rule.getParamIdx(); + if (!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) { diff --git a/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerTest.java b/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerTest.java index fe21297d5a..2cb4f89fe3 100644 --- a/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerTest.java +++ b/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerTest.java @@ -96,6 +96,47 @@ public void testLoadParamRulesClearingUnusedMetricsForRule() { } + @Test + public void testLoadParamRulesKeepingSharedThreadCountForSurvivingRule() { + final String resA = "resA"; + final int sharedParamIdx = 0; + + // Two distinct rules share the same parameter index. + ParamFlowRule ruleA = new ParamFlowRule(resA) + .setCount(10) + .setParamIdx(sharedParamIdx) + .setGrade(RuleConstant.FLOW_GRADE_THREAD); + ParamFlowRule ruleB = new ParamFlowRule(resA) + .setCount(5) + .setParamIdx(sharedParamIdx); + + ParamFlowRuleManager.loadRules(Arrays.asList(ruleA, ruleB)); + + // Initialize the metrics for both rules, as ParameterMetricStorage#initParamMetricsFor does + // for the loaded rules. + ParameterMetric metric = new ParameterMetric(); + metric.initialize(ruleA); + metric.initialize(ruleB); + ParameterMetricStorage.getMetricsMap().put(resA, metric); + + // Seed a known thread count on the shared parameter index. + Object value = "v"; + metric.addThreadCount(value); + metric.addThreadCount(value); + long expectedThreadCount = metric.getThreadCount(sharedParamIdx, value); + assertTrue("Test fixture should have a positive thread count", expectedThreadCount > 0); + + // Reload with only rule A: rule B is removed while rule A survives on the same index. + ParamFlowRuleManager.loadRules(Collections.singletonList(ruleA)); + + ParameterMetric metricAfter = ParameterMetricStorage.getParamMetricForResource(resA); + assertNotNull(metricAfter); + assertEquals("The thread count of the surviving rule should be preserved", + expectedThreadCount, metricAfter.getThreadCount(sharedParamIdx, value)); + assertNotNull("The surviving rule's time counter should be preserved", metricAfter.getRuleTimeCounter(ruleA)); + assertNull("The removed rule's time counter should be dropped", metricAfter.getRuleTimeCounter(ruleB)); + } + @Test public void testLoadParamRulesAndGet() { final String resA = "abc"; diff --git a/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetricTest.java b/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetricTest.java index 8c7e7d8c64..824b13ae4d 100644 --- a/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetricTest.java +++ b/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetricTest.java @@ -18,6 +18,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -76,6 +77,64 @@ public void testInitAndClearParameterMetric() { assertEquals(0, metric.getRuleTokenCounterMap().size()); } + @Test + public void testClearForRuleKeepsSharedThreadCount() { + ParameterMetric metric = new ParameterMetric(); + + // Two distinct rules share the same parameter index. + ParamFlowRule ruleA = new ParamFlowRule("abc").setParamIdx(1); + ParamFlowRule ruleB = new ParamFlowRule("abc").setParamIdx(1).setCount(5); + metric.initialize(ruleA); + metric.initialize(ruleB); + + CacheMap threadCountMap = metric.getThreadCountMap().get(1); + assertNotNull(threadCountMap); + threadCountMap.put("v", new AtomicInteger(2)); + + // Clearing rule B should keep the shared thread counter, as rule A still uses index 1. + metric.clearForRule(ruleB); + assertNull("The removed rule's time counter should be cleared", metric.getRuleTimeCounter(ruleB)); + assertNull("The removed rule's token counter should be cleared", metric.getRuleTokenCounterMap().get(ruleB)); + assertNotNull("The surviving rule's time counter should be kept", metric.getRuleTimeCounter(ruleA)); + assertSame("The shared thread counter should be kept for the surviving rule", + threadCountMap, metric.getThreadCountMap().get(1)); + assertEquals(2L, metric.getThreadCount(1, "v")); + + // Once the last rule using index 1 is cleared, the shared counter should be removed. + metric.clearForRule(ruleA); + assertNull("The thread counter should be cleared when no rule uses the index", + metric.getThreadCountMap().get(1)); + assertEquals(0L, metric.getThreadCount(1, "v")); + } + + @Test + public void testClearForRuleRemovesThreadCountOfUnusedIndex() { + ParameterMetric metric = new ParameterMetric(); + + ParamFlowRule ruleA = new ParamFlowRule("abc").setParamIdx(1); + // Rule C uses a different, unique index. + ParamFlowRule ruleC = new ParamFlowRule("abc").setParamIdx(2); + metric.initialize(ruleA); + metric.initialize(ruleC); + + CacheMap sharedThreadCount = metric.getThreadCountMap().get(1); + CacheMap uniqueThreadCount = metric.getThreadCountMap().get(2); + assertNotNull(sharedThreadCount); + assertNotNull(uniqueThreadCount); + sharedThreadCount.put("v", new AtomicInteger(2)); + uniqueThreadCount.put("w", new AtomicInteger(3)); + + // Index 2 is only used by rule C, so its counter should be removed. + metric.clearForRule(ruleC); + assertNull("The thread counter of a unique index should be cleared", + metric.getThreadCountMap().get(2)); + assertEquals(0L, metric.getThreadCount(2, "w")); + // Index 1 is still used by rule A, so it should be untouched. + assertSame("The thread counter of an index still in use should be kept", + sharedThreadCount, metric.getThreadCountMap().get(1)); + assertEquals(2L, metric.getThreadCount(1, "v")); + } + @Test public void testAddAndDecreaseThreadCountCommon() { testAddAndDecreaseThreadCount(PARAM_TYPE_NORMAL); From d0a6d0a7fdc3cff12df7b9de3ee0ed1853f45158 Mon Sep 17 00:00:00 2001 From: hutiefang76 <137664623+hutiefang76@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:35:29 +0800 Subject: [PATCH 2/2] fix: retain in-flight parameter counts across rule reloads --- .../flow/param/ParamFlowRuleManager.java | 14 +- .../block/flow/param/ParameterMetric.java | 18 +- .../param/ParamFlowRuleManagerReloadTest.java | 525 ++++++++++++++++++ 3 files changed, 554 insertions(+), 3 deletions(-) create mode 100644 sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerReloadTest.java diff --git a/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManager.java b/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManager.java index a8e72e1379..dd2807d65f 100644 --- a/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManager.java +++ b/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManager.java @@ -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; @@ -129,13 +130,24 @@ private Map> aggregateAndPrepareParamRules(List rules, Integer paramIdx) { + for (ParamFlowRule rule : rules) { + if (Objects.equals(rule.getParamIdx(), paramIdx)) { + return true; + } + } + return false; + } } private ParamFlowRuleManager() {} diff --git a/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java b/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java index 7c0f3b5b27..3438db437b 100644 --- a/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java +++ b/sentinel-extension/sentinel-parameter-flow-control/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParameterMetric.java @@ -89,15 +89,29 @@ public CacheMap 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); // 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. + // 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 (!isParamIdxInUse(paramIdx)) { + if (!paramIdxUsedByNewRules && !isParamIdxInUse(paramIdx)) { threadCountMap.remove(paramIdx); } } diff --git a/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerReloadTest.java b/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerReloadTest.java new file mode 100644 index 0000000000..abc7a34c73 --- /dev/null +++ b/sentinel-extension/sentinel-parameter-flow-control/src/test/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowRuleManagerReloadTest.java @@ -0,0 +1,525 @@ +/* + * Copyright 1999-2018 Alibaba Group Holding Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.alibaba.csp.sentinel.slots.block.flow.param; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import com.alibaba.csp.sentinel.Entry; +import com.alibaba.csp.sentinel.EntryType; +import com.alibaba.csp.sentinel.SphU; +import com.alibaba.csp.sentinel.context.ContextUtil; +import com.alibaba.csp.sentinel.node.DefaultNode; +import com.alibaba.csp.sentinel.slotchain.ProcessorSlotEntryCallback; +import com.alibaba.csp.sentinel.slotchain.ProcessorSlotExitCallback; +import com.alibaba.csp.sentinel.slots.block.BlockException; +import com.alibaba.csp.sentinel.slots.block.RuleConstant; +import com.alibaba.csp.sentinel.slots.statistic.ParamFlowStatisticEntryCallback; +import com.alibaba.csp.sentinel.slots.statistic.ParamFlowStatisticExitCallback; +import com.alibaba.csp.sentinel.slots.statistic.StatisticSlotCallbackRegistry; +import com.alibaba.csp.sentinel.slots.statistic.cache.CacheMap; + +/** + * Test cases for reloading rules of {@link ParamFlowRuleManager} when several rules of the same + * resource share the thread counter of one parameter index. + * + *

The thread counter of a parameter index is shared by all the rules that use that index, so a + * reload has to keep the counter as long as one of those rules is still in use. Otherwise the + * count of the entries that are still in flight is lost and the resource is under-counted, which + * allows more entries to pass than the configured threshold.

+ */ +public class ParamFlowRuleManagerReloadTest { + + private static final String CONTEXT_NAME = "param-flow-rule-manager-reload-test"; + private static final int PARAM_IDX = 0; + private static final long VALUE = 100L; + + private final List openEntries = new ArrayList(); + private List previousRules; + private Map previousMetrics; + private ProcessorSlotEntryCallback addedEntryCallback; + private ProcessorSlotExitCallback addedExitCallback; + + @Before + public void setUp() { + ContextUtil.exit(); + + // Keep the process state of the rules and metrics so that it can be restored afterwards. + previousRules = ParamFlowRuleManager.getRules(); + previousMetrics = new HashMap(ParameterMetricStorage.getMetricsMap()); + // Detach the saved metrics before the rules are reset: the metrics of the previous rules + // must not be modified by this test. + ParameterMetricStorage.getMetricsMap().clear(); + ParamFlowRuleManager.loadRules(null); + ParameterMetricStorage.getMetricsMap().clear(); + + addedEntryCallback = installEntryCallbackIfAbsent(); + addedExitCallback = installExitCallbackIfAbsent(); + + ContextUtil.enter(CONTEXT_NAME); + } + + @After + public void tearDown() { + try { + closeOpenEntries(); + ContextUtil.exit(); + } finally { + try { + ParamFlowRuleManager.loadRules(null); + ParameterMetricStorage.getMetricsMap().clear(); + if (previousRules != null) { + ParamFlowRuleManager.loadRules(previousRules); + } + if (previousMetrics != null) { + ParameterMetricStorage.getMetricsMap().putAll(previousMetrics); + } + previousRules = null; + previousMetrics = null; + } finally { + removeCallbacksIfAdded(); + } + } + } + + @Test + public void testReloadDroppingSiblingKeepsSharedThreadCounter() throws Exception { + final String resource = "paramFlowReloadKeepSharedCounter"; + ParamFlowRule survivor = threadRule(resource, 3); + ParamFlowRule sibling = threadRule(resource, 10); + checkSiblingRemovalKeepsSharedThreadCounter(resource, Arrays.asList(survivor, sibling), survivor, + sibling); + } + + @Test + public void testReloadDroppingSiblingKeepsSharedThreadCounterInReversedOrder() throws Exception { + final String resource = "paramFlowReloadKeepSharedCounterReversed"; + ParamFlowRule survivor = threadRule(resource, 3); + ParamFlowRule sibling = threadRule(resource, 10); + checkSiblingRemovalKeepsSharedThreadCounter(resource, Arrays.asList(sibling, survivor), survivor, + sibling); + } + + @Test + public void testReloadWithTighterLimitKeepsInFlightCount() throws Exception { + final String resource = "paramFlowReloadTighterLimit"; + ParamFlowRule oldCompanion = threadRule(resource, 10); + ParamFlowRule replaced = threadRule(resource, 5); + ParamFlowRule removed = threadRule(resource, 7); + checkTighterReloadKeepsInFlightCount(resource, Arrays.asList(oldCompanion, replaced, removed), + oldCompanion, replaced, removed, threadRule(resource, 2)); + } + + @Test + public void testReloadWithTighterLimitKeepsInFlightCountInReversedOrder() throws Exception { + final String resource = "paramFlowReloadTighterLimitReversed"; + ParamFlowRule oldCompanion = threadRule(resource, 10); + ParamFlowRule replaced = threadRule(resource, 5); + ParamFlowRule removed = threadRule(resource, 7); + checkTighterReloadKeepsInFlightCount(resource, Arrays.asList(removed, replaced, oldCompanion), + oldCompanion, replaced, removed, threadRule(resource, 2)); + } + + @Test + public void testReloadWithValueEqualReplacementRuleKeepsSharedCounter() throws Exception { + final String resource = "paramFlowReloadEqualReplacement"; + ParamFlowRule keeper = threadRule(resource, 10); + ParamFlowRule sibling = threadRule(resource, 5); + ParamFlowRuleManager.loadRules(Arrays.asList(keeper, sibling)); + + Entry first = enter(resource, VALUE); + Entry second = enter(resource, VALUE); + + ParameterMetric metric = metricOf(resource); + assertNotNull(metric); + CacheMap counter = threadCounterOf(metric, PARAM_IDX); + assertNotNull(counter); + CacheMap keeperTimeCounter = metric.getRuleTimeCounter(keeper); + assertNotNull(keeperTimeCounter); + assertEquals(2L, metric.getThreadCount(PARAM_IDX, VALUE)); + + // A rule object that is equal to an already loaded rule reuses the counters of that rule. + ParamFlowRule replacement = threadRule(resource, 10); + assertNotSame(keeper, replacement); + assertEquals(keeper, replacement); + + ParamFlowRuleManager.loadRules(Collections.singletonList(replacement)); + + ParameterMetric metricAfter = metricOf(resource); + assertSame(metric, metricAfter); + assertSame("The equal rule must reuse the counters of the loaded instance", keeperTimeCounter, + metricAfter.getRuleTimeCounter(replacement)); + assertSame("The shared thread counter must be kept", counter, threadCounterOf(metricAfter, PARAM_IDX)); + assertEquals(2L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + assertNull(metricAfter.getRuleTimeCounter(sibling)); + + Entry third = enter(resource, VALUE); + assertEquals(3L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + + exit(third, VALUE); + assertEquals(2L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + exit(second, VALUE); + assertEquals(1L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + exit(first, VALUE); + assertEquals(0L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + } + + @Test + public void testReloadRemovingRuleOfLastParamIdxKeepsOtherIndexCounter() throws Exception { + final String resource = "paramFlowReloadParamIdxCleanup"; + final long firstValue = 100L; + final long lastValue = 200L; + ParamFlowRule firstIdxRule = threadRule(resource, 5); + ParamFlowRule lastIdxRule = threadRule(resource, 5); + lastIdxRule.setParamIdx(1); + ParamFlowRuleManager.loadRules(Arrays.asList(firstIdxRule, lastIdxRule)); + + Entry first = enter(resource, firstValue, lastValue); + Entry second = enter(resource, firstValue, lastValue); + + ParameterMetric metric = metricOf(resource); + assertNotNull(metric); + CacheMap firstIdxCounter = threadCounterOf(metric, 0); + assertNotNull(firstIdxCounter); + assertNotNull(threadCounterOf(metric, 1)); + assertEquals(2L, metric.getThreadCount(0, firstValue)); + assertEquals(2L, metric.getThreadCount(1, lastValue)); + + // Only the rule of the last parameter index is dropped. + ParamFlowRuleManager.loadRules(Collections.singletonList(firstIdxRule)); + + ParameterMetric metricAfter = metricOf(resource); + assertSame(metric, metricAfter); + assertSame("The counter of the index that stays in use must be kept", firstIdxCounter, + threadCounterOf(metricAfter, 0)); + assertNull("The counter of the index that is no longer used must be removed", + threadCounterOf(metricAfter, 1)); + assertEquals(2L, metricAfter.getThreadCount(0, firstValue)); + assertEquals(0L, metricAfter.getThreadCount(1, lastValue)); + + exit(second, firstValue, lastValue); + assertEquals(1L, metricAfter.getThreadCount(0, firstValue)); + assertEquals(0L, metricAfter.getThreadCount(1, lastValue)); + exit(first, firstValue, lastValue); + assertEquals(0L, metricAfter.getThreadCount(0, firstValue)); + assertEquals(0L, metricAfter.getThreadCount(1, lastValue)); + assertSame(firstIdxCounter, threadCounterOf(metricAfter, 0)); + } + + @Test + public void testReloadRemovingResourceClearsMetricAndKeepsExitsSafe() throws Exception { + final String resource = "paramFlowReloadResourceRemoved"; + final String otherResource = "paramFlowReloadResourceOther"; + ParamFlowRule rule = threadRule(resource, 2); + ParamFlowRuleManager.loadRules(Collections.singletonList(rule)); + + Entry first = enter(resource, VALUE); + Entry second = enter(resource, VALUE); + + ParameterMetric metric = metricOf(resource); + assertNotNull(metric); + assertEquals(2L, metric.getThreadCount(PARAM_IDX, VALUE)); + + ParamFlowRule otherRule = threadRule(otherResource, 3); + ParamFlowRuleManager.loadRules(Collections.singletonList(otherRule)); + + assertNull("The metric of the removed resource must be cleared", metricOf(resource)); + assertFalse(ParamFlowRuleManager.hasRules(resource)); + assertTrue(ParamFlowRuleManager.hasRules(otherResource)); + assertEquals("The detached metric must not be updated any more", 2L, + metric.getThreadCount(PARAM_IDX, VALUE)); + + // The in-flight entries of the removed resource exit without touching any metric. + exit(second, VALUE); + exit(first, VALUE); + assertNull(metricOf(resource)); + assertEquals("The detached metric must not be updated any more", 2L, + metric.getThreadCount(PARAM_IDX, VALUE)); + + // Without rules the resource is not controlled and no metric is created again. + Entry entry = enter(resource, VALUE); + assertNull(metricOf(resource)); + exit(entry, VALUE); + assertNull(metricOf(resource)); + } + + @Test + public void testReloadWithoutInFlightEntriesKeepsNoPhantomCount() throws Exception { + final String resource = "paramFlowReloadNoInFlight"; + ParamFlowRule keeper = threadRule(resource, 2); + ParamFlowRule sibling = threadRule(resource, 9); + ParamFlowRuleManager.loadRules(Arrays.asList(keeper, sibling)); + assertNull("Loading rules must not create the parameter metric", metricOf(resource)); + + ParamFlowRuleManager.loadRules(Collections.singletonList(keeper)); + assertNull(metricOf(resource)); + assertEquals(1, ParamFlowRuleManager.getRulesOfResource(resource).size()); + + Entry entry = enter(resource, VALUE); + ParameterMetric metric = metricOf(resource); + assertNotNull(metric); + assertEquals(1L, metric.getThreadCount(PARAM_IDX, VALUE)); + + exit(entry, VALUE); + assertEquals(0L, metric.getThreadCount(PARAM_IDX, VALUE)); + CacheMap counter = threadCounterOf(metric, PARAM_IDX); + assertNotNull(counter); + assertEquals("The count of the value must be released on exit", 0, counter.size()); + } + + /** + * Both rules of the resource share the thread counter of the parameter index. Dropping the + * sibling must neither reset the counter nor replace the holder of the count of the entries + * that are still in flight. + */ + private void checkSiblingRemovalKeepsSharedThreadCounter(String resource, List initialRules, + ParamFlowRule survivor, ParamFlowRule sibling) + throws Exception { + ParamFlowRuleManager.loadRules(initialRules); + + Entry first = enter(resource, VALUE); + Entry second = enter(resource, VALUE); + + ParameterMetric metric = metricOf(resource); + assertNotNull(metric); + CacheMap counter = threadCounterOf(metric, PARAM_IDX); + assertNotNull(counter); + AtomicInteger valueCounter = counter.get(VALUE); + assertNotNull(valueCounter); + assertEquals(2, valueCounter.get()); + assertEquals(2L, metric.getThreadCount(PARAM_IDX, VALUE)); + + ParamFlowRuleManager.loadRules(Collections.singletonList(survivor)); + + ParameterMetric metricAfter = metricOf(resource); + assertSame("The reload must keep the metric of the surviving rule", metric, metricAfter); + CacheMap counterAfter = threadCounterOf(metricAfter, PARAM_IDX); + assertSame("The thread counter of the parameter index must be kept", counter, counterAfter); + assertSame("The holder of the in-flight count must not be replaced", valueCounter, + counterAfter.get(VALUE)); + assertEquals("The in-flight count must be retained", 2L, + metricAfter.getThreadCount(PARAM_IDX, VALUE)); + assertNotNull(metricAfter.getRuleTimeCounter(survivor)); + assertNull("The removed sibling must release its own counters", metricAfter.getRuleTimeCounter(sibling)); + + Entry third = enter(resource, VALUE); + assertEquals("The limit of the surviving rule applies to the retained count", 3L, + metricAfter.getThreadCount(PARAM_IDX, VALUE)); + + exit(third, VALUE); + assertEquals(2L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + exit(second, VALUE); + assertEquals(1L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + exit(first, VALUE); + assertEquals(0L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + assertSame("Exiting must not detach the shared counter", counter, + threadCounterOf(metricAfter, PARAM_IDX)); + } + + /** + * Reloading the resource with the single tighter rule drops every old rule instance of the + * parameter index by value equality: the companion rule, the replaced one and the removed one + * are all gone, while the new rule still loads the same parameter index and stays lazily + * initialized until its first entry. The in-flight count of the value must survive this + * reload, so the next entry for the same value is blocked by the new limit although the new + * rule does not hold the retained entries itself. + */ + private void checkTighterReloadKeepsInFlightCount(String resource, List initialRules, + ParamFlowRule oldCompanion, ParamFlowRule replaced, + ParamFlowRule removed, ParamFlowRule tighter) + throws Exception { + ParamFlowRuleManager.loadRules(initialRules); + + Entry first = enter(resource, VALUE); + Entry second = enter(resource, VALUE); + + ParameterMetric metric = metricOf(resource); + assertNotNull(metric); + CacheMap counter = threadCounterOf(metric, PARAM_IDX); + assertNotNull(counter); + AtomicInteger valueCounter = counter.get(VALUE); + assertNotNull(valueCounter); + assertEquals(2, valueCounter.get()); + // All the rules of the resource rely on the same counter of the parameter index. + assertNotNull(metric.getRuleTimeCounter(oldCompanion)); + assertNotNull(metric.getRuleTimeCounter(replaced)); + assertNotNull(metric.getRuleTimeCounter(removed)); + + // No old rule value survives this reload: the only rule that keeps the parameter index of + // the resource loaded is the new one, which is not initialized before its first entry. + ParamFlowRuleManager.loadRules(Collections.singletonList(tighter)); + + assertExtraEntryBlocked(resource, VALUE); + ParameterMetric metricAfter = metricOf(resource); + assertSame(metric, metricAfter); + CacheMap counterAfter = threadCounterOf(metricAfter, PARAM_IDX); + assertSame("The reload must keep the shared thread counter of the parameter index", counter, + counterAfter); + assertSame(valueCounter, counterAfter.get(VALUE)); + assertEquals("The in-flight count must survive the reload", 2L, + metricAfter.getThreadCount(PARAM_IDX, VALUE)); + assertNull("The replaced rule must release its own counters", metricAfter.getRuleTimeCounter(replaced)); + assertNull("The removed rule must release its own counters", metricAfter.getRuleTimeCounter(removed)); + assertNotNull("The checked rule is initialized by the blocked entry", metricAfter.getRuleTimeCounter(tighter)); + + assertEquals("A blocked entry must not change the in-flight count", 2L, + metricAfter.getThreadCount(PARAM_IDX, VALUE)); + + exit(second, VALUE); + assertEquals(1L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + exit(first, VALUE); + assertEquals(0L, metricAfter.getThreadCount(PARAM_IDX, VALUE)); + } + + private static ParamFlowRule threadRule(String resource, int count) { + ParamFlowRule rule = new ParamFlowRule(resource); + rule.setCount(count); + rule.setParamIdx(PARAM_IDX); + rule.setGrade(RuleConstant.FLOW_GRADE_THREAD); + return rule; + } + + private static ParameterMetric metricOf(String resource) { + return ParameterMetricStorage.getParamMetricForResource(resource); + } + + private static CacheMap threadCounterOf(ParameterMetric metric, int paramIdx) { + return metric.getThreadCountMap().get(paramIdx); + } + + private Entry enter(String resource, Object... args) throws BlockException { + Entry entry = SphU.entry(resource, EntryType.IN, 1, args); + openEntries.add(new OpenEntry(entry, args)); + return entry; + } + + private void exit(Entry entry, Object... args) throws Exception { + forgetOpenEntry(entry); + entry.exit(1, args); + } + + private void forgetOpenEntry(Entry entry) { + for (int i = 0; i < openEntries.size(); i++) { + if (openEntries.get(i).entry == entry) { + openEntries.remove(i); + return; + } + } + } + + private void closeOpenEntries() { + for (int i = openEntries.size() - 1; i >= 0; i--) { + OpenEntry openEntry = openEntries.get(i); + try { + openEntry.entry.exit(1, openEntry.args); + } catch (Throwable ignored) { + // Ignore: the cleanup of a failed test must not hide the assertion error. + } + } + openEntries.clear(); + } + + private static void assertExtraEntryBlocked(String resource, Object value) throws Exception { + try { + Entry entry = SphU.entry(resource, EntryType.IN, 1, value); + entry.exit(1, value); + fail("The extra entry should be blocked by the parameter flow rule"); + } catch (BlockException ex) { + assertTrue("Expect a ParamFlowException but was " + ex.getClass().getName(), + ex instanceof ParamFlowException); + assertEquals(resource, ((ParamFlowException) ex).getResourceName()); + } + } + + private static ProcessorSlotEntryCallback installEntryCallbackIfAbsent() { + for (ProcessorSlotEntryCallback registered : StatisticSlotCallbackRegistry + .getEntryCallbacks()) { + if (registered instanceof ParamFlowStatisticEntryCallback) { + return null; + } + } + ProcessorSlotEntryCallback callback = new ParamFlowStatisticEntryCallback(); + StatisticSlotCallbackRegistry.addEntryCallback(ParamFlowStatisticEntryCallback.class.getName(), callback); + return callback; + } + + private static ProcessorSlotExitCallback installExitCallbackIfAbsent() { + for (ProcessorSlotExitCallback registered : StatisticSlotCallbackRegistry.getExitCallbacks()) { + if (registered instanceof ParamFlowStatisticExitCallback) { + return null; + } + } + ProcessorSlotExitCallback callback = new ParamFlowStatisticExitCallback(); + StatisticSlotCallbackRegistry.addExitCallback(ParamFlowStatisticExitCallback.class.getName(), callback); + return callback; + } + + private void removeCallbacksIfAdded() { + if (addedEntryCallback != null) { + // Only unregister the instance installed by this test. + if (containsIdentity(StatisticSlotCallbackRegistry.getEntryCallbacks(), addedEntryCallback)) { + StatisticSlotCallbackRegistry.removeEntryCallback(ParamFlowStatisticEntryCallback.class.getName()); + } + addedEntryCallback = null; + } + if (addedExitCallback != null) { + if (containsIdentity(StatisticSlotCallbackRegistry.getExitCallbacks(), addedExitCallback)) { + StatisticSlotCallbackRegistry.removeExitCallback(ParamFlowStatisticExitCallback.class.getName()); + } + addedExitCallback = null; + } + } + + private static boolean containsIdentity(Collection registered, Object callback) { + for (Object item : registered) { + if (item == callback) { + return true; + } + } + return false; + } + + private static final class OpenEntry { + + private final Entry entry; + private final Object[] args; + + private OpenEntry(Entry entry, Object[] args) { + this.entry = entry; + this.args = args; + } + } +}