About inaccurate control over concurrent thread counts

我写了下面的测试类做本地测试，其中grade 配置值是RuleConstant.FLOW_GRADE_THREAD，Count 配置值是10，用
System.out.println("concurrent request count max record:" + requestCountMaxRecord.intValue()); 输出当前正在执行业务处理的线程数量，测试结果输出的最大值是13

```
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.csp.sentinel.slots.block.RuleConstant;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;

public class RateLimiterTest1 {
final static String RESQUEST_RESOURCE = "HelloWorld";

static class Resquest {
String resource;
Object param;
public Resquest(String resource) {
this.resource = resource;
}
public Resquest(String resource, Object param) {
this.resource = resource;
this.param = param;
}
public String getResource() {
return this.resource;
}
public Object getParam() {
return param;
}
}
static class Response {
int code;
Object param;
Throwable throwable;
public Response(int code, Object param) {
this.code = code;
this.param = param;
}
public Response(int code, Throwable throwable) {
this.code = code;
this.throwable = throwable;
}
public int getCode() {
return code;
}
}
interface BussProxy {
public Response handle(Resquest resquest);
}

static {
List rules = new ArrayList();
FlowRule rule = new FlowRule();
rule.setResource(RESQUEST_RESOURCE);
rule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
rule.setCount(10);
rules.add(rule);
FlowRuleManager.loadRules(rules);

//先调一次，试试是不是首次初始化有并发问题
Entry entry = null;
try {
entry = SphU.entry(RESQUEST_RESOURCE);
} catch (Exception e) {
} finally {
if (entry != null) {
entry.exit();
}
}
}

static class RequestInterceptor {
private AtomicInteger requestCountMaxRecord = new AtomicInteger(0);
private AtomicInteger requestCountCurrentRecord = new AtomicInteger(0);

public Response acceptResuqest(Resquest resquest, BussProxy proxy) {
Entry entry = null;
boolean isAddedRequestCountCurrentRecord = false;
try {
entry = SphU.entry(resquest.getResource());
int requestCountCurrentRecordIntVal = requestCountCurrentRecord.incrementAndGet();
isAddedRequestCountCurrentRecord = true;
if (requestCountMaxRecord.intValue() < requestCountCurrentRecordIntVal) {//测试用不同步，能输出最大值即可
requestCountMaxRecord.set(requestCountCurrentRecordIntVal);
System.out.println("concurrent request count max record:" + requestCountMaxRecord.intValue());
}
return proxy.handle(resquest);
} catch (BlockException e) {
return new Response(9, e);
} finally {
if (isAddedRequestCountCurrentRecord) {
requestCountCurrentRecord.decrementAndGet();
}
if (entry != null) {
entry.exit();
}
}
}
}

public static void main(String[] args) throws Exception {
int threadCount = 150;
final AtomicInteger requestCountRecord = new AtomicInteger();
final AtomicInteger requestSucRecord = new AtomicInteger();
final AtomicInteger requestErrRecord = new AtomicInteger();
RequestInterceptor interceptor = new RequestInterceptor();
ExecutorService threadPool = Executors.newFixedThreadPool(threadCount, new ThreadFactory() {
final AtomicInteger threadOrder = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "thread-" + threadOrder.incrementAndGet());
//System.out.println(thread.getName() + " created.");
return thread;
}
});
for (int i = 0; i < threadCount; i++) {
threadPool.submit(new Runnable() {
long beginTimestamp = System.currentTimeMillis();
@Override
public void run() {
try {
while (System.currentTimeMillis() - beginTimestamp < TimeUnit.SECONDS.toMillis(10)) {
requestCountRecord.incrementAndGet();
Response response = interceptor.acceptResuqest(new Resquest(RESQUEST_RESOURCE), new BussProxy() {
@Override
public Response handle(Resquest resquest) {
try {
Thread.sleep(5);
} catch (InterruptedException e) {
}
return new Response(0, "success");
}
});
if (response.getCode() == 0) {
requestSucRecord.incrementAndGet();
} else {
requestErrRecord.incrementAndGet();
}
}
} catch (Throwable e) {
e.printStackTrace();
}
}
});
}
threadPool.shutdown();
while (true) {
if (threadPool.isTerminated()) {
System.out.println("request count record:" + requestCountRecord.intValue() + "\n"
+ "request success record:" + requestSucRecord.intValue() + "\n"
+ "request error record:" + requestErrRecord.intValue());
break;
}
Thread.sleep(100);
}
}

}
```

检查代码后怀疑可能是
com.alibaba.csp.sentinel.slots.statistic.StatisticSlot 的entry和exit方法处理的问题
如下1、2、3 标记处，处理逻辑是先判断当前节点正在执行的线程数量是否大于配置值，是则抛出异常，否通过且当前正在执行的线程数量加1
对于并发场景，可能有多个线程并行运行在1 处，其线程数量加curThreadNum当前值已大于配置限定值，但是由于这些线程都未执行到2处，都未累计入curThreadNum，因此都通过检查了

```
public class StatisticSlot extends AbstractLinkedProcessorSlot {

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
try {
// Do some checking.
fireEntry(context, resourceWrapper, node, count, prioritized, args); ---- 1、这里会先判断当前curThreadNum是否大于配置值

// Request passed, add thread count and pass count.
node.increaseThreadNum(); ---- 2、这里对curThreadNum加1
node.addPassRequest(count);

if (context.getCurEntry().getOriginNode() != null) {
// Add count for origin node.
context.getCurEntry().getOriginNode().increaseThreadNum();
context.getCurEntry().getOriginNode().addPassRequest(count);
}

if (resourceWrapper.getType() == EntryType.IN) {
// Add count for global inbound entry node for global statistics.
Constants.ENTRY_NODE.increaseThreadNum();
Constants.ENTRY_NODE.addPassRequest(count);
}

// Handle pass event with registered entry callback handlers.
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onPass(context, resourceWrapper, node, count, args);
}
} catch (PriorityWaitException ex) {
node.increaseThreadNum();
if (context.getCurEntry().getOriginNode() != null) {
// Add count for origin node.
context.getCurEntry().getOriginNode().increaseThreadNum();
}

if (resourceWrapper.getType() == EntryType.IN) {
// Add count for global inbound entry node for global statistics.
Constants.ENTRY_NODE.increaseThreadNum();
}
// Handle pass event with registered entry callback handlers.
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onPass(context, resourceWrapper, node, count, args);
}
} catch (BlockException e) {
// Blocked, set block exception to current entry.
context.getCurEntry().setError(e);

// Add block count.
node.increaseBlockQps(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseBlockQps(count);
}

if (resourceWrapper.getType() == EntryType.IN) {
// Add count for global inbound entry node for global statistics.
Constants.ENTRY_NODE.increaseBlockQps(count);
}

// Handle block event with registered entry callback handlers.
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onBlocked(e, context, resourceWrapper, node, count, args);
}

throw e;
} catch (Throwable e) {
// Unexpected error, set error to current entry.
context.getCurEntry().setError(e);

// This should not happen.
node.increaseExceptionQps(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseExceptionQps(count);
}

if (resourceWrapper.getType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseExceptionQps(count);
}
throw e;
}
}

@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
DefaultNode node = (DefaultNode)context.getCurNode();

if (context.getCurEntry().getError() == null) {
// Calculate response time (max RT is TIME_DROP_VALVE).
long rt = TimeUtil.currentTimeMillis() - context.getCurEntry().getCreateTime();
if (rt > Constants.TIME_DROP_VALVE) {
rt = Constants.TIME_DROP_VALVE;
}

// Record response time and success count.
node.addRtAndSuccess(rt, count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().addRtAndSuccess(rt, count);
}

node.decreaseThreadNum(); ----- 3、这里对curThreadNum 减1

if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().decreaseThreadNum();
}

if (resourceWrapper.getType() == EntryType.IN) {
Constants.ENTRY_NODE.addRtAndSuccess(rt, count);
Constants.ENTRY_NODE.decreaseThreadNum();
}
} else {
// Error may happen.
}

// Handle exit event with registered exit callback handlers.
Collection exitCallbacks = StatisticSlotCallbackRegistry.getExitCallbacks();
for (ProcessorSlotExitCallback handler : exitCallbacks) {
handler.onExit(context, resourceWrapper, count, args);
}

fireExit(context, resourceWrapper, count);
}
}
```
在1处 和 2处中间增加2ms延迟后，重跑测试类，输出当前正在执行业务处理的线程数量，结果最大值是135
```
... 其它省略
// Do some checking.
fireEntry(context, resourceWrapper, node, count, prioritized, args); ---- 1、这里会先判断当前curThreadNum是否大于配置值
Thread.sleep(2); ---- 加2ms 延迟
// Request passed, add thread count and pass count.
node.increaseThreadNum(); ---- 2、这里对curThreadNum加1
node.addPassRequest(count);
... 其它省略
```

下面是基于上述问题修改后的代码和测试类，思路是把原来的先判断curThreadNum值是否超限，再做累计，改成线程计数先加1，再判断curThreadNum值是否超限，同时exit方法对当前线程计数减1操作也做相应调整
最后为了确保无论检查通过与否 exit方法都会在最后被调用，因此除了修改com.alibaba.csp.sentinel.slots.statistic.StatisticSlot 类，还需要修改 com.alibaba.csp.sentinel.CtSph 、com.alibaba.csp.sentinel.node.StatisticNode 和测试类调用方式
修改后测试类的System.out.println("concurrent request count max record:" + requestCountMaxRecord.intValue()); 输出当前正在执行业务处理的线程数量是正确值10

具体如下：
修改 com.alibaba.csp.sentinel.CtSph 类，删除如下1、2 处代码
```
... 其它省略
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
throws BlockException {
Context context = ContextUtil.getContext();
if (context instanceof NullContext) {
// The {@link NullContext} indicates that the amount of context has exceeded the threshold,
// so here init the entry only. No rule checking will be done.
return new CtEntry(resourceWrapper, null, context);
}

if (context == null) {
// Using default context.
context = MyContextUtil.myEnter(Constants.CONTEXT_DEFAULT_NAME, "", resourceWrapper.getType());
}

// Global switch is close, no rule checking will do.
if (!Constants.ON) {
return new CtEntry(resourceWrapper, null, context);
}

ProcessorSlot chain = lookProcessChain(resourceWrapper);

/*
* Means amount of resources (slot chain) exceeds {@link Constants.MAX_SLOT_CHAIN_SIZE},
* so no rule checking will be done.
*/
if (chain == null) {
return new CtEntry(resourceWrapper, null, context);
}

Entry e = new CtEntry(resourceWrapper, chain, context);
try {
chain.entry(context, resourceWrapper, null, count, prioritized, args);
} catch (BlockException e1) {
// e.exit(count, args); ---- 1、删除此行
// throw e1; ---- 2、删除此行
} catch (Throwable e1) {
// This should not happen, unless there are errors existing in Sentinel internal.
RecordLog.info("Sentinel unexpected exception", e1);
}
return e;
}
... 其它省略
```

修改 com.alibaba.csp.sentinel.node.StatisticNode 类

```
public class StatisticNode implements Node {

/**
* Holds statistics of the recent {@code INTERVAL} seconds. The {@code INTERVAL} is divided into time spans
* by given {@code sampleCount}.
*/
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
IntervalProperty.INTERVAL);

/**
* Holds statistics of the recent 60 seconds. The windowLengthInMs is deliberately set to 1000 milliseconds,
* meaning each bucket per second, in this way we can get accurate statistics of each second.
*/
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);

/**
* The counter for thread count.
*/
private AtomicInteger curThreadNum = new AtomicInteger(-1); ---- 1、修改 原是private AtomicInteger curThreadNum = new AtomicInteger(0);

private static ThreadLocal, Integer>> curThreadNumRecord = new ThreadLocal, Integer>>() { ---- 2、新增
protected Map, Integer> initialValue() { ---- 3、新增
return new HashMap, Integer>(2); ---- 4、新增
}; ---- 5、新增
}; ---- 6、新增

/**
* The last timestamp when metrics were fetched.
*/
private long lastFetchTime = -1;

@Override
public Map metrics() {
// The fetch operation is thread-safe under a single-thread scheduler pool.
long currentTime = TimeUtil.currentTimeMillis();
currentTime = currentTime - currentTime % 1000;
Map metrics = new ConcurrentHashMap<>();
List nodesOfEverySecond = rollingCounterInMinute.details();
long newLastFetchTime = lastFetchTime;
// Iterate metrics of all resources, filter valid metrics (not-empty and up-to-date).
for (MetricNode node : nodesOfEverySecond) {
if (isNodeInTime(node, currentTime) && isValidMetricNode(node)) {
metrics.put(node.getTimestamp(), node);
newLastFetchTime = Math.max(newLastFetchTime, node.getTimestamp());
}
}
lastFetchTime = newLastFetchTime;

return metrics;
}

private boolean isNodeInTime(MetricNode node, long currentTime) {
return node.getTimestamp() > lastFetchTime && node.getTimestamp() < currentTime;
}

private boolean isValidMetricNode(MetricNode node) {
return node.getPassQps() > 0 || node.getBlockQps() > 0 || node.getSuccessQps() > 0
|| node.getExceptionQps() > 0 || node.getRt() > 0 || node.getOccupiedPassQps() > 0;
}

@Override
public void reset() {
rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);
}

@Override
public long totalRequest() {
long totalRequest = rollingCounterInMinute.pass() + rollingCounterInMinute.block();
return totalRequest;
}

@Override
public long blockRequest() {
return rollingCounterInMinute.block();
}

@Override
public double blockQps() {
return rollingCounterInSecond.block() / rollingCounterInSecond.getWindowIntervalInSec();
}

@Override
public double previousBlockQps() {
return this.rollingCounterInMinute.previousWindowBlock();
}

@Override
public double previousPassQps() {
return this.rollingCounterInMinute.previousWindowPass();
}

@Override
public double totalQps() {
return passQps() + blockQps();
}

@Override
public long totalSuccess() {
return rollingCounterInMinute.success();
}

@Override
public double exceptionQps() {
return rollingCounterInSecond.exception() / rollingCounterInSecond.getWindowIntervalInSec();
}

@Override
public long totalException() {
return rollingCounterInMinute.exception();
}

@Override
public double passQps() {
return rollingCounterInSecond.pass() / rollingCounterInSecond.getWindowIntervalInSec();
}

@Override
public long totalPass() {
return rollingCounterInMinute.pass();
}

@Override
public double successQps() {
return rollingCounterInSecond.success() / rollingCounterInSecond.getWindowIntervalInSec();
}

@Override
public double maxSuccessQps() {
return rollingCounterInSecond.maxSuccess() * rollingCounterInSecond.getSampleCount();
}

@Override
public double occupiedPassQps() {
return rollingCounterInSecond.occupiedPass() / rollingCounterInSecond.getWindowIntervalInSec();
}

@Override
public double avgRt() {
long successCount = rollingCounterInSecond.success();
if (successCount == 0) {
return 0;
}

return rollingCounterInSecond.rt() * 1.0 / successCount;
}

@Override
public double minRt() {
return rollingCounterInSecond.minRt();
}

@Override
public int curThreadNum() {
return curThreadNumRecord.get().get(this.getClass()); ---- 7、修改 原是 return curThreadNum.get();
}

@Override
public void addPassRequest(int count) {
rollingCounterInSecond.addPass(count);
rollingCounterInMinute.addPass(count);
}

@Override
public void addRtAndSuccess(long rt, int successCount) {
rollingCounterInSecond.addSuccess(successCount);
rollingCounterInSecond.addRT(rt);

rollingCounterInMinute.addSuccess(successCount);
rollingCounterInMinute.addRT(rt);
}

@Override
public void increaseBlockQps(int count) {
rollingCounterInSecond.addBlock(count);
rollingCounterInMinute.addBlock(count);
}

@Override
public void increaseExceptionQps(int count) {
rollingCounterInSecond.addException(count);
rollingCounterInMinute.addException(count);
}

@Override
public void increaseThreadNum() {
Thread.yield(); ---- 8、新增
curThreadNumRecord.get().put(this.getClass(), curThreadNum.incrementAndGet()); ---- 9、修改 原是 curThreadNum.incrementAndGet();
}

@Override
public void decreaseThreadNum() {
curThreadNum.decrementAndGet();
}

@Override
public void debug() {
rollingCounterInSecond.debug();
}

@Override
public long tryOccupyNext(long currentTime, int acquireCount, double threshold) {
double maxCount = threshold * IntervalProperty.INTERVAL / 1000;
long currentBorrow = rollingCounterInSecond.waiting();
if (currentBorrow >= maxCount) {
return OccupyTimeoutProperty.getOccupyTimeout();
}

int windowLength = IntervalProperty.INTERVAL / SampleCountProperty.SAMPLE_COUNT;
long earliestTime = currentTime - currentTime % windowLength + windowLength - IntervalProperty.INTERVAL;

int idx = 0;
/*
* Note: here {@code currentPass} may be less than it really is NOW, because time difference
* since call rollingCounterInSecond.pass(). So in high concurrency, the following code may
* lead more tokens be borrowed.
*/
long currentPass = rollingCounterInSecond.pass();
while (earliestTime < currentTime) {
long waitInMs = idx * windowLength + windowLength - currentTime % windowLength;
if (waitInMs >= OccupyTimeoutProperty.getOccupyTimeout()) {
break;
}
long windowPass = rollingCounterInSecond.getWindowPass(earliestTime);
if (currentPass + currentBorrow + acquireCount - windowPass <= maxCount) {
return waitInMs;
}
earliestTime += windowLength;
currentPass -= windowPass;
idx++;
}

return OccupyTimeoutProperty.getOccupyTimeout();
}

@Override
public long waiting() {
return rollingCounterInSecond.waiting();
}

@Override
public void addWaitingRequest(long futureTime, int acquireCount) {
rollingCounterInSecond.addWaiting(futureTime, acquireCount);
}

@Override
public void addOccupiedPass(int acquireCount) {
rollingCounterInMinute.addOccupiedPass(acquireCount);
rollingCounterInMinute.addPass(acquireCount);
}
}
```

修改 com.alibaba.csp.sentinel.slots.statistic.StatisticSlot 类

```
package com.alibaba.csp.sentinel.slots.statistic;

import java.util.Collection;

import com.alibaba.csp.sentinel.slotchain.ProcessorSlotEntryCallback;
import com.alibaba.csp.sentinel.slotchain.ProcessorSlotExitCallback;
import com.alibaba.csp.sentinel.slots.block.flow.PriorityWaitException;
import com.alibaba.csp.sentinel.util.TimeUtil;
import com.alibaba.csp.sentinel.Constants;
import com.alibaba.csp.sentinel.EntryType;
import com.alibaba.csp.sentinel.context.Context;
import com.alibaba.csp.sentinel.node.ClusterNode;
import com.alibaba.csp.sentinel.node.DefaultNode;
import com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot;
import com.alibaba.csp.sentinel.slotchain.ResourceWrapper;
import com.alibaba.csp.sentinel.slots.block.BlockException;

public class StatisticSlot extends AbstractLinkedProcessorSlot {

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
boolean prioritized, Object... args) throws Throwable {
try {

node.increaseThreadNum(); ---- 修改，原此行内容置于fireEntry 后，现修改成调用 fireEntry 方法之前

// Do some checking.
fireEntry(context, resourceWrapper, node, count, prioritized, args);

node.addPassRequest(count);

if (context.getCurEntry().getOriginNode() != null) {
// Add count for origin node.
context.getCurEntry().getOriginNode().increaseThreadNum();
context.getCurEntry().getOriginNode().addPassRequest(count);
}

if (resourceWrapper.getType() == EntryType.IN) {
// Add count for global inbound entry node for global statistics.
Constants.ENTRY_NODE.increaseThreadNum();
Constants.ENTRY_NODE.addPassRequest(count);
}

// Handle pass event with registered entry callback handlers.
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onPass(context, resourceWrapper, node, count, args);
}
} catch (PriorityWaitException ex) {
ex.printStackTrace();
node.increaseThreadNum();
if (context.getCurEntry().getOriginNode() != null) {
// Add count for origin node.
context.getCurEntry().getOriginNode().increaseThreadNum();
}

if (resourceWrapper.getType() == EntryType.IN) {
// Add count for global inbound entry node for global statistics.
Constants.ENTRY_NODE.increaseThreadNum();
}
// Handle pass event with registered entry callback handlers.
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onPass(context, resourceWrapper, node, count, args);
}
} catch (BlockException e) {

// Blocked, set block exception to current entry.
context.getCurEntry().setError(e);

// Add block count.
node.increaseBlockQps(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseBlockQps(count);
}

if (resourceWrapper.getType() == EntryType.IN) {
// Add count for global inbound entry node for global statistics.
Constants.ENTRY_NODE.increaseBlockQps(count);
}

// Handle block event with registered entry callback handlers.
for (ProcessorSlotEntryCallback handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
handler.onBlocked(e, context, resourceWrapper, node, count, args);
}

throw e;
} catch (Throwable e) {
e.printStackTrace();
// Unexpected error, set error to current entry.
context.getCurEntry().setError(e);

// This should not happen.
node.increaseExceptionQps(count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().increaseExceptionQps(count);
}

if (resourceWrapper.getType() == EntryType.IN) {
Constants.ENTRY_NODE.increaseExceptionQps(count);
}
throw e;
}
}

@Override
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
DefaultNode node = (DefaultNode)context.getCurNode();

if (context.getCurEntry().getError() == null) {
// Calculate response time (max RT is TIME_DROP_VALVE).
long rt = TimeUtil.currentTimeMillis() - context.getCurEntry().getCreateTime();
if (rt > Constants.TIME_DROP_VALVE) {
rt = Constants.TIME_DROP_VALVE;
}

// Record response time and success count.
node.addRtAndSuccess(rt, count);
if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().addRtAndSuccess(rt, count);
}

if (context.getCurEntry().getOriginNode() != null) {
context.getCurEntry().getOriginNode().decreaseThreadNum();
}

if (resourceWrapper.getType() == EntryType.IN) {
Constants.ENTRY_NODE.addRtAndSuccess(rt, count);
Constants.ENTRY_NODE.decreaseThreadNum();
}
} else {
// Error may happen.
}

node.decreaseThreadNum(); ---- 修改，原此行内容置于if (context.getCurEntry().getError() == null) 块中，现修改成置于if块外部

// Handle exit event with registered exit callback handlers.
Collection exitCallbacks = StatisticSlotCallbackRegistry.getExitCallbacks();
for (ProcessorSlotExitCallback handler : exitCallbacks) {
handler.onExit(context, resourceWrapper, count, args);
}

fireExit(context, resourceWrapper, count);
}
}
```

测试类
```
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.slots.block.RuleConstant;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;

public class RateLimiterTest2 {
final static String RESQUEST_RESOURCE = "HelloWorld";

static class Resquest {
String resource;
Object param;
public Resquest(String resource) {
this.resource = resource;
}
public Resquest(String resource, Object param) {
this.resource = resource;
this.param = param;
}
public String getResource() {
return this.resource;
}
public Object getParam() {
return param;
}
}
static class Response {
int code;
Object param;
Throwable throwable;
public Response(int code, Object param) {
this.code = code;
this.param = param;
}
public Response(int code, Throwable throwable) {
this.code = code;
this.throwable = throwable;
}
public int getCode() {
return code;
}
}
interface BussProxy {
public Response handle(Resquest resquest);
}

static {
List rules = new ArrayList();
FlowRule rule = new FlowRule();
rule.setResource(RESQUEST_RESOURCE);
rule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
rule.setCount(10);
rules.add(rule);
FlowRuleManager.loadRules(rules);
}

static class RequestInterceptor {
private AtomicInteger requestCountMaxRecord = new AtomicInteger(0);
private AtomicInteger requestCountCurrentRecord = new AtomicInteger(0);

public Response acceptResuqest(Resquest resquest, BussProxy proxy) {
Entry entry = null;
boolean isAddedRequestCountCurrentRecord = false;
try {
entry = SphU.entry(resquest.getResource());
if (entry.getError() != null) {
throw entry.getError();
}
int requestCountCurrentRecordIntVal = requestCountCurrentRecord.incrementAndGet();
isAddedRequestCountCurrentRecord = true;
if (requestCountMaxRecord.intValue() < requestCountCurrentRecordIntVal) {//测试用不同步，能输出最大值即可
requestCountMaxRecord.set(requestCountCurrentRecordIntVal);
System.out.println("concurrent request count max record:" + requestCountMaxRecord.intValue());
}
return proxy.handle(resquest);
} catch (Throwable e) {
return new Response(9, e);
} finally {
if (isAddedRequestCountCurrentRecord) {
requestCountCurrentRecord.decrementAndGet();
}
entry.exit();
}
}
}

public static void main(String[] args) throws Exception {
int threadCount = 150;
final AtomicInteger requestCountRecord = new AtomicInteger();
final AtomicInteger requestSucRecord = new AtomicInteger();
final AtomicInteger requestErrRecord = new AtomicInteger();
RequestInterceptor interceptor = new RequestInterceptor();
ExecutorService threadPool = Executors.newFixedThreadPool(threadCount, new ThreadFactory() {
final AtomicInteger threadOrder = new AtomicInteger();
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "thread-" + threadOrder.incrementAndGet());
System.out.println(thread.getName() + " created.");
return thread;
}
});
for (int i = 0; i < threadCount; i++) {
threadPool.submit(new Runnable() {
long beginTimestamp = System.currentTimeMillis();
@Override
public void run() {
try {
while (System.currentTimeMillis() - beginTimestamp < TimeUnit.SECONDS.toMillis(10)) {
requestCountRecord.incrementAndGet();
Response response = interceptor.acceptResuqest(new Resquest(RESQUEST_RESOURCE), new BussProxy() {
@Override
public Response handle(Resquest resquest) {
try {
Thread.sleep(5);
} catch (InterruptedException e) {
}
return new Response(0, "success");
}
});
if (response.getCode() == 0) {
requestSucRecord.incrementAndGet();
} else {
requestErrRecord.incrementAndGet();
}
}
} catch (Throwable e) {
e.printStackTrace();
}
}
});
}
threadPool.shutdown();
while (true) {
if (threadPool.isTerminated()) {
System.out.println("request count record:" + requestCountRecord.intValue() + "\n"
+ "request success record:" + requestSucRecord.intValue() + "\n"
+ "request error record:" + requestErrRecord.intValue());
break;
}
Thread.sleep(100);
}
}

}
```

Start with the entry and exit paths in com.alibaba.csp.sentinel.slots.statistic.StatisticSlot, then trace the related handling in com.alibaba.csp.sentinel.CtSph and com.alibaba.csp.sentinel.node.StatisticNode. Use the provided RateLimiterTest1 concurrency test and verify that a thread limit of 10 cannot be exceeded while entry and exit accounting remains consistent.

Written by the indexing model from the issue text.