mirror of
https://github.com/apache/rocketmq.git
synced 2026-08-29 04:13:00 +08:00
* extract the adaptive lock
* extract the adaptive lock
* feat(): perfect the adaptive lock
* feat(): perfect the adaptive lock
* Optimized code type
* Optimized code type
* Optimized code type
* fix fail test
* Optimize the adaptive locking mechanism logic
* Optimize the adaptive locking mechanism logic
* feat:Adaptive locking mechanism adjustment
* feat:Adaptive locking mechanism adjustment
* feat:Adaptive locking mechanism adjustment
* Optimize the adaptive locking mechanism logic
* Optimize the adaptive locking mechanism logic
* Optimize the adaptive locking mechanism logic
* feat:Supports the hot activation of ABS locks
* feat:Supports the hot activation of ABS locks
* feat:Supports the hot activation of ABS locks
* feat:Supports the hot activation of ABS locks
* Optimize code style
* Optimize code style
* Optimize code style
* Optimize code style
* Optimize code style
* Optimize code style
* Updated the locking mechanism name
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* delete unused import
* Optimize the logic of switching to spin locks
* Revert "Optimize the logic of switching to spin locks"
This reverts commit 1d7bac5c2f.
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimize the logic of switching to spin locks
* Optimized locking logic
* Optimized locking logic
* Optimized locking logic
* fix test
* fix test
* fix test
* fix test
* Optimize code style
* Optimize code style
* fix test
* fix test
* optimize client rebalancing logic
---------
Co-authored-by: wanghuaiyuan <wanghuaiyuan@xiaomi.com>
This commit is contained in:
hqbfz
@@ -62,6 +62,7 @@ import org.apache.rocketmq.store.exception.ConsumeQueueException;
|
||||
import org.apache.rocketmq.store.exception.StoreException;
|
||||
import org.apache.rocketmq.store.ha.HAService;
|
||||
import org.apache.rocketmq.store.ha.autoswitch.AutoSwitchHAService;
|
||||
import org.apache.rocketmq.store.lock.AdaptiveBackOffSpinLockImpl;
|
||||
import org.apache.rocketmq.store.logfile.MappedFile;
|
||||
import org.apache.rocketmq.store.util.LibC;
|
||||
import org.rocksdb.RocksDBException;
|
||||
@@ -130,7 +131,11 @@ public class CommitLog implements Swappable {
|
||||
return new PutMessageThreadLocal(defaultMessageStore.getMessageStoreConfig());
|
||||
}
|
||||
};
|
||||
this.putMessageLock = messageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();
|
||||
|
||||
PutMessageLock adaptiveBackOffSpinLock = new AdaptiveBackOffSpinLockImpl();
|
||||
|
||||
this.putMessageLock = messageStore.getMessageStoreConfig().getUseABSLock() ? adaptiveBackOffSpinLock :
|
||||
messageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();
|
||||
|
||||
this.flushDiskWatcher = new FlushDiskWatcher();
|
||||
|
||||
|
||||
@@ -445,6 +445,17 @@ public class MessageStoreConfig {
|
||||
*/
|
||||
private String bottomMostCompressionTypeForConsumeQueueStore = "zstd";
|
||||
|
||||
/**
|
||||
* Spin number in the retreat strategy of spin lock
|
||||
* Default is 1000
|
||||
*/
|
||||
private int spinLockCollisionRetreatOptimalDegree = 1000;
|
||||
|
||||
/**
|
||||
* Use AdaptiveBackOffLock
|
||||
**/
|
||||
private boolean useABSLock = false;
|
||||
|
||||
public boolean isRocksdbCQDoubleWriteEnable() {
|
||||
return rocksdbCQDoubleWriteEnable;
|
||||
}
|
||||
@@ -1898,4 +1909,20 @@ public class MessageStoreConfig {
|
||||
public void setBottomMostCompressionTypeForConsumeQueueStore(String bottomMostCompressionTypeForConsumeQueueStore) {
|
||||
this.bottomMostCompressionTypeForConsumeQueueStore = bottomMostCompressionTypeForConsumeQueueStore;
|
||||
}
|
||||
|
||||
public int getSpinLockCollisionRetreatOptimalDegree() {
|
||||
return spinLockCollisionRetreatOptimalDegree;
|
||||
}
|
||||
|
||||
public void setSpinLockCollisionRetreatOptimalDegree(int spinLockCollisionRetreatOptimalDegree) {
|
||||
this.spinLockCollisionRetreatOptimalDegree = spinLockCollisionRetreatOptimalDegree;
|
||||
}
|
||||
|
||||
public void setUseABSLock(boolean useABSLock) {
|
||||
this.useABSLock = useABSLock;
|
||||
}
|
||||
|
||||
public boolean getUseABSLock() {
|
||||
return useABSLock;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You 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 org.apache.rocketmq.store.lock;
|
||||
|
||||
import org.apache.rocketmq.store.PutMessageLock;
|
||||
import org.apache.rocketmq.store.config.MessageStoreConfig;
|
||||
|
||||
public interface AdaptiveBackOffSpinLock extends PutMessageLock {
|
||||
/**
|
||||
* Configuration update
|
||||
* @param messageStoreConfig
|
||||
*/
|
||||
default void update(MessageStoreConfig messageStoreConfig) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Locking mechanism switching
|
||||
*/
|
||||
default void swap() {
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You 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 org.apache.rocketmq.store.lock;
|
||||
|
||||
import org.apache.rocketmq.store.config.MessageStoreConfig;
|
||||
|
||||
import java.time.LocalTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class AdaptiveBackOffSpinLockImpl implements AdaptiveBackOffSpinLock {
|
||||
private AdaptiveBackOffSpinLock adaptiveLock;
|
||||
//state
|
||||
private AtomicBoolean state = new AtomicBoolean(true);
|
||||
|
||||
// Used to determine the switchover between a mutex lock and a spin lock
|
||||
private final static float SWAP_SPIN_LOCK_RATIO = 0.8f;
|
||||
|
||||
// It is used to adjust the spin number K of the escape spin lock
|
||||
// When (retreat number / TPS) <= (1 / BASE_SWAP_ADAPTIVE_RATIO * SPIN_LOCK_ADAPTIVE_RATIO), K is decreased
|
||||
private final static int SPIN_LOCK_ADAPTIVE_RATIO = 4;
|
||||
|
||||
// It is used to adjust the spin number K of the escape spin lock
|
||||
// When (retreat number / TPS) >= (1 / BASE_SWAP_ADAPTIVE_RATIO), K is increased
|
||||
private final static int BASE_SWAP_LOCK_RATIO = 320;
|
||||
|
||||
private final static String BACK_OFF_SPIN_LOCK = "SpinLock";
|
||||
|
||||
private final static String REENTRANT_LOCK = "ReentrantLock";
|
||||
|
||||
private Map<String, AdaptiveBackOffSpinLock> locks;
|
||||
|
||||
private final List<AtomicInteger> tpsTable;
|
||||
|
||||
private final List<Map<Thread, Byte>> threadTable;
|
||||
|
||||
private int swapCriticalPoint;
|
||||
|
||||
private AtomicInteger currentThreadNum = new AtomicInteger(0);
|
||||
|
||||
private AtomicBoolean isOpen = new AtomicBoolean(true);
|
||||
|
||||
public AdaptiveBackOffSpinLockImpl() {
|
||||
this.locks = new HashMap<>();
|
||||
this.locks.put(REENTRANT_LOCK, new BackOffReentrantLock());
|
||||
this.locks.put(BACK_OFF_SPIN_LOCK, new BackOffSpinLock());
|
||||
|
||||
this.threadTable = new ArrayList<>(2);
|
||||
this.threadTable.add(new ConcurrentHashMap<>());
|
||||
this.threadTable.add(new ConcurrentHashMap<>());
|
||||
|
||||
this.tpsTable = new ArrayList<>(2);
|
||||
this.tpsTable.add(new AtomicInteger(0));
|
||||
this.tpsTable.add(new AtomicInteger(0));
|
||||
|
||||
adaptiveLock = this.locks.get(BACK_OFF_SPIN_LOCK);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void lock() {
|
||||
int slot = LocalTime.now().getSecond() % 2;
|
||||
this.threadTable.get(slot).putIfAbsent(Thread.currentThread(), Byte.MAX_VALUE);
|
||||
this.tpsTable.get(slot).getAndIncrement();
|
||||
boolean state;
|
||||
do {
|
||||
state = this.state.get();
|
||||
} while (!state);
|
||||
|
||||
currentThreadNum.incrementAndGet();
|
||||
this.adaptiveLock.lock();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unlock() {
|
||||
this.adaptiveLock.unlock();
|
||||
currentThreadNum.decrementAndGet();
|
||||
if (isOpen.get()) {
|
||||
swap();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void update(MessageStoreConfig messageStoreConfig) {
|
||||
this.adaptiveLock.update(messageStoreConfig);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void swap() {
|
||||
if (!this.state.get()) {
|
||||
return;
|
||||
}
|
||||
boolean needSwap = false;
|
||||
int slot = 1 - LocalTime.now().getSecond() % 2;
|
||||
int tps = this.tpsTable.get(slot).get() + 1;
|
||||
int threadNum = this.threadTable.get(slot).size();
|
||||
this.tpsTable.get(slot).set(-1);
|
||||
this.threadTable.get(slot).clear();
|
||||
if (tps == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.adaptiveLock instanceof BackOffSpinLock) {
|
||||
BackOffSpinLock lock = (BackOffSpinLock) this.adaptiveLock;
|
||||
// Avoid frequent adjustment of K, and make a reasonable range through experiments
|
||||
// reasonable range : (retreat number / TPS) > (1 / BASE_SWAP_ADAPTIVE_RATIO * SPIN_LOCK_ADAPTIVE_RATIO) &&
|
||||
// (retreat number / TPS) < (1 / BASE_SWAP_ADAPTIVE_RATIO)
|
||||
if (lock.getNumberOfRetreat(slot) * BASE_SWAP_LOCK_RATIO >= tps) {
|
||||
if (lock.isAdapt()) {
|
||||
lock.adapt(true);
|
||||
} else {
|
||||
// It is used to switch between mutex lock and spin lock
|
||||
this.swapCriticalPoint = tps * threadNum;
|
||||
needSwap = true;
|
||||
}
|
||||
} else if (lock.getNumberOfRetreat(slot) * BASE_SWAP_LOCK_RATIO * SPIN_LOCK_ADAPTIVE_RATIO <= tps) {
|
||||
lock.adapt(false);
|
||||
}
|
||||
lock.setNumberOfRetreat(slot, 0);
|
||||
} else {
|
||||
if (tps * threadNum <= this.swapCriticalPoint * SWAP_SPIN_LOCK_RATIO) {
|
||||
needSwap = true;
|
||||
}
|
||||
}
|
||||
|
||||
if (needSwap) {
|
||||
if (this.state.compareAndSet(true, false)) {
|
||||
// Ensures that no threads are in contention locks as well as in critical zones
|
||||
int currentThreadNum;
|
||||
do {
|
||||
currentThreadNum = this.currentThreadNum.get();
|
||||
} while (currentThreadNum != 0);
|
||||
|
||||
try {
|
||||
if (this.adaptiveLock instanceof BackOffSpinLock) {
|
||||
this.adaptiveLock = this.locks.get(REENTRANT_LOCK);
|
||||
} else {
|
||||
this.adaptiveLock = this.locks.get(BACK_OFF_SPIN_LOCK);
|
||||
((BackOffSpinLock) this.adaptiveLock).adapt(false);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
//ignore
|
||||
} finally {
|
||||
this.state.compareAndSet(false, true);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public List<AdaptiveBackOffSpinLock> getLocks() {
|
||||
return (List<AdaptiveBackOffSpinLock>) this.locks.values();
|
||||
}
|
||||
|
||||
public void setLocks(Map<String, AdaptiveBackOffSpinLock> locks) {
|
||||
this.locks = locks;
|
||||
}
|
||||
|
||||
public boolean getState() {
|
||||
return this.state.get();
|
||||
}
|
||||
|
||||
public void setState(boolean state) {
|
||||
this.state.set(state);
|
||||
}
|
||||
|
||||
public AdaptiveBackOffSpinLock getAdaptiveLock() {
|
||||
return adaptiveLock;
|
||||
}
|
||||
|
||||
public List<AtomicInteger> getTpsTable() {
|
||||
return tpsTable;
|
||||
}
|
||||
|
||||
public void setSwapCriticalPoint(int swapCriticalPoint) {
|
||||
this.swapCriticalPoint = swapCriticalPoint;
|
||||
}
|
||||
|
||||
public int getSwapCriticalPoint() {
|
||||
return swapCriticalPoint;
|
||||
}
|
||||
|
||||
public boolean isOpen() {
|
||||
return this.isOpen.get();
|
||||
}
|
||||
|
||||
public void setOpen(boolean open) {
|
||||
this.isOpen.set(open);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You 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 org.apache.rocketmq.store.lock;
|
||||
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
public class BackOffReentrantLock implements AdaptiveBackOffSpinLock {
|
||||
private ReentrantLock putMessageNormalLock = new ReentrantLock(); // NonfairSync
|
||||
|
||||
@Override
|
||||
public void lock() {
|
||||
putMessageNormalLock.lock();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unlock() {
|
||||
putMessageNormalLock.unlock();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You 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 org.apache.rocketmq.store.lock;
|
||||
|
||||
import org.apache.rocketmq.store.config.MessageStoreConfig;
|
||||
|
||||
import java.time.LocalTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class BackOffSpinLock implements AdaptiveBackOffSpinLock {
|
||||
|
||||
private AtomicBoolean putMessageSpinLock = new AtomicBoolean(true);
|
||||
|
||||
private int optimalDegree;
|
||||
|
||||
private final static int INITIAL_DEGREE = 1000;
|
||||
|
||||
private final static int MAX_OPTIMAL_DEGREE = 10000;
|
||||
|
||||
private final List<AtomicInteger> numberOfRetreat;
|
||||
|
||||
public BackOffSpinLock() {
|
||||
this.optimalDegree = INITIAL_DEGREE;
|
||||
|
||||
numberOfRetreat = new ArrayList<>(2);
|
||||
numberOfRetreat.add(new AtomicInteger(0));
|
||||
numberOfRetreat.add(new AtomicInteger(0));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void lock() {
|
||||
int spinDegree = this.optimalDegree;
|
||||
while (true) {
|
||||
for (int i = 0; i < spinDegree; i++) {
|
||||
if (this.putMessageSpinLock.compareAndSet(true, false)) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
numberOfRetreat.get(LocalTime.now().getSecond() % 2).getAndIncrement();
|
||||
try {
|
||||
Thread.sleep(0);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void unlock() {
|
||||
this.putMessageSpinLock.compareAndSet(false, true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void update(MessageStoreConfig messageStoreConfig) {
|
||||
this.optimalDegree = messageStoreConfig.getSpinLockCollisionRetreatOptimalDegree();
|
||||
}
|
||||
|
||||
public int getOptimalDegree() {
|
||||
return this.optimalDegree;
|
||||
}
|
||||
|
||||
public void setOptimalDegree(int optimalDegree) {
|
||||
this.optimalDegree = optimalDegree;
|
||||
}
|
||||
|
||||
public boolean isAdapt() {
|
||||
return optimalDegree < MAX_OPTIMAL_DEGREE;
|
||||
}
|
||||
|
||||
public synchronized void adapt(boolean isRise) {
|
||||
if (isRise) {
|
||||
if (optimalDegree * 2 <= MAX_OPTIMAL_DEGREE) {
|
||||
optimalDegree *= 2;
|
||||
} else {
|
||||
if (optimalDegree + INITIAL_DEGREE <= MAX_OPTIMAL_DEGREE) {
|
||||
optimalDegree += INITIAL_DEGREE;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if (optimalDegree >= 2 * INITIAL_DEGREE) {
|
||||
optimalDegree -= INITIAL_DEGREE;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public int getNumberOfRetreat(int pos) {
|
||||
return numberOfRetreat.get(pos).get();
|
||||
}
|
||||
|
||||
public void setNumberOfRetreat(int pos, int size) {
|
||||
this.numberOfRetreat.get(pos).set(size);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You 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 org.apache.rocketmq.store.lock;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
public class AdaptiveLockTest {
|
||||
|
||||
AdaptiveBackOffSpinLockImpl adaptiveLock;
|
||||
|
||||
@Before
|
||||
public void init() {
|
||||
adaptiveLock = new AdaptiveBackOffSpinLockImpl();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAdaptiveLock() throws InterruptedException {
|
||||
assertTrue(adaptiveLock.getAdaptiveLock() instanceof BackOffSpinLock);
|
||||
CountDownLatch countDownLatch = new CountDownLatch(1);
|
||||
adaptiveLock.lock();
|
||||
new Thread(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
adaptiveLock.lock();
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException e) {
|
||||
//ignore
|
||||
}
|
||||
adaptiveLock.unlock();
|
||||
countDownLatch.countDown();
|
||||
}
|
||||
}).start();
|
||||
Thread.sleep(1000);
|
||||
adaptiveLock.unlock();
|
||||
assertEquals(2000, ((BackOffSpinLock) adaptiveLock.getAdaptiveLock()).getOptimalDegree());
|
||||
countDownLatch.await();
|
||||
|
||||
for (int i = 0; i <= 5; i++) {
|
||||
CountDownLatch countDownLatch1 = new CountDownLatch(1);
|
||||
adaptiveLock.lock();
|
||||
new Thread(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
adaptiveLock.lock();
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException e) {
|
||||
//ignore
|
||||
}
|
||||
adaptiveLock.unlock();
|
||||
countDownLatch1.countDown();
|
||||
}
|
||||
}).start();
|
||||
Thread.sleep(1000);
|
||||
adaptiveLock.unlock();
|
||||
countDownLatch1.await();
|
||||
}
|
||||
assertTrue(adaptiveLock.getAdaptiveLock() instanceof BackOffReentrantLock);
|
||||
|
||||
adaptiveLock.lock();
|
||||
Thread.sleep(1000);
|
||||
adaptiveLock.unlock();
|
||||
assertTrue(adaptiveLock.getAdaptiveLock() instanceof BackOffSpinLock);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user