[RIP-70-3]Extract adaptive lock mechanism #8663
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #8663 +/- ##
=============================================
- Coverage 47.44% 47.40% -0.05%
- Complexity 11613 11625 +12
=============================================
Files 1284 1288 +4
Lines 90025 90156 +131
Branches 11585 11603 +18
=============================================
+ Hits 42713 42735 +22
- Misses 42075 42160 +85
- Partials 5237 5261 +24 ☔ View full report in Codecov by Sentry. |
# Conflicts: # store/src/main/java/org/apache/rocketmq/store/lock/AdaptiveLockImpl.java # store/src/main/java/org/apache/rocketmq/store/lock/CollisionRetreatLock.java
| if (this.adaptiveLock instanceof CollisionRetreatLock) { | ||
| CollisionRetreatLock lock = (CollisionRetreatLock) this.adaptiveLock; | ||
| int base = Math.min(200 + tps / 200, 500); | ||
| if (lock.getNumberOfRetreat(slot) * base >= tps) { | ||
| if (lock.isAdapt()) { | ||
| lock.adapt(true); | ||
| } else { | ||
| this.tpsSwapCriticalPoint = tps; | ||
| needSwap = true; | ||
| } | ||
| } else if (lock.getNumberOfRetreat(slot) * base * 3 / 2 <= tps) { | ||
| lock.adapt(false); | ||
| } | ||
| lock.setNumberOfRetreat(slot, 0); | ||
| } else { | ||
| if (tps <= this.tpsSwapCriticalPoint * 4 / 5) { | ||
| needSwap = true; | ||
| } |
There was a problem hiding this comment.
Too many magic numbers, it should be optimized.
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| public class CollisionRetreatLock implements AdaptiveBackOffLock { |
There was a problem hiding this comment.
What is RetreatLock? Is it a misspelling of reentrant lock?
| private final List<AtomicInteger> numberOfRetreat; | ||
|
|
||
| public CollisionRetreatLock() { | ||
| this.initOptimalDegree = 1000; |
There was a problem hiding this comment.
You'd better make these numbers as constants, like OPTIMAL_DEGREE = 1000, MAX_OPTIMAL_DEGREE = 10000
common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
Outdated
Show resolved
Hide resolved
|
|
||
| public interface AdaptiveBackOffSpinLock extends PutMessageLock { | ||
|
|
||
| void lock(); |
There was a problem hiding this comment.
The PutMessageLock class includes methods for locking and unlocking.
| @@ -130,7 +132,11 @@ protected PutMessageThreadLocal initialValue() { | |||
| return new PutMessageThreadLocal(defaultMessageStore.getMessageStoreConfig()); | |||
| } | |||
| }; | |||
| this.putMessageLock = messageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock(); | |||
|
|
|||
| AdaptiveBackOffSpinLock adaptiveBackOffSpinLock = new AdaptiveBackOffSpinLockImpl(); | |||
There was a problem hiding this comment.
PutMessageLock adaptiveBackOffSpinLock = new AdaptiveBackOffSpinLockImpl() may be better
| } | ||
| } | ||
|
|
||
| if (needSwap) { |
There was a problem hiding this comment.
Add some comments to explain the principles and the rationale behind these numbers.
|
|
||
| public AdaptiveBackOffSpinLockImpl() { | ||
| this.locks = new HashMap<>(); | ||
| this.locks.put("Reentrant", new BackOffReentrantLock()); |
There was a problem hiding this comment.
These strings use class constants.
… adaptive_lock # Conflicts: # store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
Which Issue(s) This PR Fixes
Fixes #8442
Brief Description
How Did You Test This Change?
Test document ：https://shimo.im/docs/ZzkLMQ4RwwUa87AQ/