Remove locks
This commit is contained in:
parent
6b5b69acba
commit
e3c17e9923
|
|
@ -1,123 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
|
||||
/**
|
||||
* A binary mutex with the following properties:
|
||||
*
|
||||
* Exposes two different {@link Lock}s: LEFT, RIGHT.
|
||||
*
|
||||
* When LEFT is held other threads can acquire LEFT but thread trying to acquire RIGHT will be
|
||||
* blocked. When RIGHT is held other threads can acquire RIGHT but thread trying to acquire LEFT
|
||||
* will be blocked.
|
||||
*/
|
||||
public class LeftRightLock {
|
||||
|
||||
public static final int ACQUISITION_FAILED = -1;
|
||||
public static final int ACQUISITION_SUCCEEDED = 1;
|
||||
|
||||
private final LeftRightSync sync = new LeftRightSync();
|
||||
|
||||
public void lockLeft() {
|
||||
sync.acquireShared(LockSide.LEFT.getV());
|
||||
}
|
||||
|
||||
public void lockRight() {
|
||||
sync.acquireShared(LockSide.RIGHT.getV());
|
||||
}
|
||||
|
||||
public void releaseLeft() {
|
||||
sync.releaseShared(LockSide.LEFT.getV());
|
||||
}
|
||||
|
||||
public void releaseRight() {
|
||||
sync.releaseShared(LockSide.RIGHT.getV());
|
||||
}
|
||||
|
||||
public boolean tryLockLeft() {
|
||||
return sync.tryAcquireShared(LockSide.LEFT) == ACQUISITION_SUCCEEDED;
|
||||
}
|
||||
|
||||
public boolean tryLockRight() {
|
||||
return sync.tryAcquireShared(LockSide.RIGHT) == ACQUISITION_SUCCEEDED;
|
||||
}
|
||||
|
||||
private enum LockSide {
|
||||
LEFT(-1), NONE(0), RIGHT(1);
|
||||
|
||||
private final int v;
|
||||
|
||||
LockSide(int v) {
|
||||
this.v = v;
|
||||
}
|
||||
|
||||
public int getV() {
|
||||
return v;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* Keep count the count of threads holding either the LEFT or the RIGHT lock.
|
||||
* </p>
|
||||
*
|
||||
* <li>A state ({@link AbstractQueuedSynchronizer#getState()}) greater than 0 means one or more threads are holding RIGHT lock. </li>
|
||||
* <li>A state ({@link AbstractQueuedSynchronizer#getState()}) lower than 0 means one or more threads are holding LEFT lock.</li>
|
||||
* <li>A state ({@link AbstractQueuedSynchronizer#getState()}) equal to zero means no thread is holding any lock.</li>
|
||||
*/
|
||||
private static final class LeftRightSync extends AbstractQueuedSynchronizer {
|
||||
|
||||
|
||||
@Override
|
||||
protected int tryAcquireShared(int requiredSide) {
|
||||
return (tryChangeThreadCountHoldingCurrentLock(requiredSide, ChangeType.ADD) ? ACQUISITION_SUCCEEDED : ACQUISITION_FAILED);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean tryReleaseShared(int requiredSide) {
|
||||
return tryChangeThreadCountHoldingCurrentLock(requiredSide, ChangeType.REMOVE);
|
||||
}
|
||||
|
||||
public boolean tryChangeThreadCountHoldingCurrentLock(int requiredSide, ChangeType changeType) {
|
||||
if (requiredSide != 1 && requiredSide != -1)
|
||||
throw new AssertionError("You can either lock LEFT or RIGHT (-1 or +1)");
|
||||
|
||||
int curState;
|
||||
int newState;
|
||||
do {
|
||||
curState = this.getState();
|
||||
if (!sameSide(curState, requiredSide)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (changeType == ChangeType.ADD) {
|
||||
newState = curState + requiredSide;
|
||||
} else {
|
||||
newState = curState - requiredSide;
|
||||
}
|
||||
//TODO: protect against int overflow (hopefully you won't have so many threads)
|
||||
} while (!this.compareAndSetState(curState, newState));
|
||||
return true;
|
||||
}
|
||||
|
||||
final int tryAcquireShared(LockSide lockSide) {
|
||||
return this.tryAcquireShared(lockSide.getV());
|
||||
}
|
||||
|
||||
final boolean tryReleaseShared(LockSide lockSide) {
|
||||
return this.tryReleaseShared(lockSide.getV());
|
||||
}
|
||||
|
||||
private boolean sameSide(int curState, int requiredSide) {
|
||||
return curState == 0 || sameSign(curState, requiredSide);
|
||||
}
|
||||
|
||||
private boolean sameSign(int a, int b) {
|
||||
return (a >= 0) ^ (b < 0);
|
||||
}
|
||||
|
||||
public enum ChangeType {
|
||||
ADD, REMOVE
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,372 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
|
||||
import it.cavallium.concurrentlocks.ReadWriteUpdateLock;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
/**
|
||||
* Fine-grained locking on data or on the entire collection for a snapshottable collection
|
||||
* @param <T>
|
||||
*/
|
||||
public class SnapshottableCollectionLock<T> {
|
||||
|
||||
private final TransitReadWriteLock fullLock;
|
||||
private final TransitReadWriteLock dataLock;
|
||||
private final Striped<ReadWriteUpdateLock> transactionLock;
|
||||
|
||||
public SnapshottableCollectionLock(int stripedSize) {
|
||||
this.transactionLock = Striped.readWriteUpdateLock(stripedSize);
|
||||
this.fullLock = new TransitReadWriteLock();
|
||||
this.dataLock = new TransitReadWriteLock();
|
||||
}
|
||||
|
||||
public enum DataLockType {
|
||||
READ,
|
||||
/**
|
||||
* "Update" allow reads but blocks writes and other updates.
|
||||
*/
|
||||
UPDATE,
|
||||
WRITE
|
||||
}
|
||||
|
||||
public enum FullLockType {
|
||||
/** <p>This lock is a read lock for "full", nothing for "data"</p>
|
||||
* <p>Useful for coarse iterators</p>
|
||||
*
|
||||
* <ul>
|
||||
* <li>full<ul>
|
||||
* <li>read: allow</li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* <li>data<ul>
|
||||
* <li>read/update: allow</li>
|
||||
* <li>write: allow</li>
|
||||
* </ul></li>
|
||||
* </ul>
|
||||
*/
|
||||
DENY_ONLY_FULL_WRITE,
|
||||
/** <p>This lock is a read lock for "full" and a read lock for "data"</p>
|
||||
*
|
||||
* <ul>
|
||||
* <li>full<ul>
|
||||
* <li>read: allow</li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* <li>data<ul>
|
||||
* <li>read/update: allow</li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* </ul>
|
||||
*/
|
||||
ALLOW_ONLY_READ,
|
||||
/** <p>This lock is a write lock for "full" and a read lock for "data"</p>
|
||||
*
|
||||
* <ul>
|
||||
* <li>full<ul>
|
||||
* <li>read: <u>deny</u></li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* <li>data<ul>
|
||||
* <li>read/update: allow</li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* </ul>
|
||||
*/
|
||||
ALLOW_ONLY_ITEMS_READ,
|
||||
/** <p>This lock is a read lock for "full" and a write lock for "data"</p>
|
||||
*
|
||||
* <ul>
|
||||
* <li>full<ul>
|
||||
* <li>read: allow</li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* <li>data<ul>
|
||||
* <li>read/update: <u>deny</u></li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* </ul>
|
||||
*/
|
||||
ALLOW_ONLY_FULL_READ,
|
||||
/** <p>This lock is a write lock for "full" and a write lock for "data"</p>
|
||||
*
|
||||
* <ul>
|
||||
* <li>full<ul>
|
||||
* <li>read: <u>deny</u></li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* <li>data<ul>
|
||||
* <li>read/update: <u>deny</u></li>
|
||||
* <li>write: <u>deny</u></li>
|
||||
* </ul></li>
|
||||
* </ul>
|
||||
*/
|
||||
EXCLUSIVE;
|
||||
|
||||
public static final Set<FullLockType> READ_FULL_TYPES = Set.of(
|
||||
DENY_ONLY_FULL_WRITE,
|
||||
ALLOW_ONLY_READ,
|
||||
ALLOW_ONLY_FULL_READ
|
||||
);
|
||||
public static final Set<FullLockType> WRITE_FULL_TYPES = Set.of(ALLOW_ONLY_ITEMS_READ, EXCLUSIVE);
|
||||
public static final Set<FullLockType> READ_DATA_TYPES = Set.of(ALLOW_ONLY_READ, ALLOW_ONLY_ITEMS_READ);
|
||||
public static final Set<FullLockType> WRITE_DATA_TYPES = Set.of(ALLOW_ONLY_FULL_READ, EXCLUSIVE);
|
||||
public static final Set<FullLockType> NOOP_DATA_TYPES = Set.of(DENY_ONLY_FULL_WRITE);
|
||||
}
|
||||
|
||||
public void fullLock(Object snapshot, FullLockType fullLockMode) {
|
||||
if (snapshot == null) {
|
||||
// Transit on full locks
|
||||
switch (fullLockMode) {
|
||||
case DENY_ONLY_FULL_WRITE:
|
||||
case ALLOW_ONLY_READ:
|
||||
case ALLOW_ONLY_FULL_READ:
|
||||
fullLock.startTransitRead();
|
||||
break;
|
||||
case ALLOW_ONLY_ITEMS_READ:
|
||||
case EXCLUSIVE:
|
||||
break;
|
||||
default:
|
||||
throw new IllegalStateException("Unexpected value: " + fullLockMode);
|
||||
}
|
||||
// Manage full locks
|
||||
switch (fullLockMode) {
|
||||
case DENY_ONLY_FULL_WRITE:
|
||||
case ALLOW_ONLY_READ:
|
||||
case ALLOW_ONLY_FULL_READ:
|
||||
fullLock.disallowTransitWrite();
|
||||
break;
|
||||
case ALLOW_ONLY_ITEMS_READ:
|
||||
case EXCLUSIVE:
|
||||
fullLock.disallowTransit();
|
||||
break;
|
||||
default:
|
||||
throw new IllegalStateException("Unexpected value: " + fullLockMode);
|
||||
}
|
||||
// Manage data locks
|
||||
switch (fullLockMode) {
|
||||
case DENY_ONLY_FULL_WRITE:
|
||||
break;
|
||||
case ALLOW_ONLY_READ:
|
||||
case ALLOW_ONLY_ITEMS_READ:
|
||||
dataLock.disallowTransitWrite();
|
||||
break;
|
||||
case ALLOW_ONLY_FULL_READ:
|
||||
case EXCLUSIVE:
|
||||
dataLock.disallowTransit();
|
||||
break;
|
||||
default:
|
||||
throw new IllegalStateException("Unexpected value: " + fullLockMode);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void fullUnlock(Object snapshot, FullLockType fullLockMode) {
|
||||
if (snapshot == null) {
|
||||
// Manage full locks
|
||||
switch (fullLockMode) {
|
||||
case DENY_ONLY_FULL_WRITE:
|
||||
case ALLOW_ONLY_READ:
|
||||
case ALLOW_ONLY_FULL_READ:
|
||||
fullLock.reAllowTransitWrite();
|
||||
break;
|
||||
case ALLOW_ONLY_ITEMS_READ:
|
||||
case EXCLUSIVE:
|
||||
fullLock.reAllowTransit();
|
||||
break;
|
||||
default:
|
||||
throw new IllegalStateException("Unexpected value: " + fullLockMode);
|
||||
}
|
||||
// Transit on full locks
|
||||
switch (fullLockMode) {
|
||||
case DENY_ONLY_FULL_WRITE:
|
||||
case ALLOW_ONLY_READ:
|
||||
case ALLOW_ONLY_FULL_READ:
|
||||
fullLock.endTransitRead();
|
||||
break;
|
||||
case ALLOW_ONLY_ITEMS_READ:
|
||||
case EXCLUSIVE:
|
||||
break;
|
||||
default:
|
||||
throw new IllegalStateException("Unexpected value: " + fullLockMode);
|
||||
}
|
||||
// Manage data locks
|
||||
switch (fullLockMode) {
|
||||
case DENY_ONLY_FULL_WRITE:
|
||||
break;
|
||||
case ALLOW_ONLY_READ:
|
||||
case ALLOW_ONLY_ITEMS_READ:
|
||||
dataLock.reAllowTransitWrite();
|
||||
break;
|
||||
case ALLOW_ONLY_FULL_READ:
|
||||
case EXCLUSIVE:
|
||||
dataLock.reAllowTransit();
|
||||
break;
|
||||
default:
|
||||
throw new IllegalStateException("Unexpected value: " + fullLockMode);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void partialLockKey(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
|
||||
switch (dataLockMode) {
|
||||
case READ:
|
||||
dataLock.startTransitRead();
|
||||
lock.readLock().lock();
|
||||
break;
|
||||
//todo: check carefully if an update lock can lock dataLock only for data read.
|
||||
// If it can, replace startTransitWrite with startTransitRead here.
|
||||
case UPDATE:
|
||||
dataLock.startTransitWrite();
|
||||
lock.updateLock().lock();
|
||||
break;
|
||||
case WRITE:
|
||||
dataLock.startTransitWrite();
|
||||
lock.writeLock().lock();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private void partialUnlockKey(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
|
||||
switch (dataLockMode) {
|
||||
case READ:
|
||||
lock.readLock().unlock();
|
||||
dataLock.endTransitRead();
|
||||
break;
|
||||
//todo: check carefully if an update lock can lock dataLock only for data read.
|
||||
// If it can, replace startTransitWrite with startTransitRead here.
|
||||
case UPDATE:
|
||||
lock.updateLock().unlock();
|
||||
dataLock.endTransitWrite();
|
||||
break;
|
||||
case WRITE:
|
||||
lock.writeLock().unlock();
|
||||
dataLock.endTransitWrite();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private void partialLockKeyUpgrade(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
|
||||
switch (dataLockMode) {
|
||||
case READ:
|
||||
lock.readLock().lock();
|
||||
break;
|
||||
case UPDATE:
|
||||
lock.updateLock().lock();
|
||||
break;
|
||||
case WRITE:
|
||||
lock.writeLock().lock();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private void partialUnlockKeyUpgrade(DataLockType dataLockMode, ReadWriteUpdateLock lock) {
|
||||
switch (dataLockMode) {
|
||||
case READ:
|
||||
lock.readLock().unlock();
|
||||
break;
|
||||
case UPDATE:
|
||||
lock.writeLock().unlock();
|
||||
break;
|
||||
case WRITE:
|
||||
lock.writeLock().unlock();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
public void dataLock(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
var lock = transactionLock.get(key);
|
||||
partialLockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataLockAt(Object snapshot, DataLockType dataLockMode, int index) {
|
||||
if (snapshot == null) {
|
||||
var lock = transactionLock.getAt(index);
|
||||
partialLockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataLockUpgrade(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
var lock = transactionLock.get(key);
|
||||
partialLockKeyUpgrade(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataLockUpgradeAt(Object snapshot, DataLockType dataLockMode, int index) {
|
||||
if (snapshot == null) {
|
||||
var lock = transactionLock.getAt(index);
|
||||
partialLockKeyUpgrade(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataUnlock(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
var lock = transactionLock.get(key);
|
||||
partialUnlockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataUnlockAt(Object snapshot, DataLockType dataLockMode, int index) {
|
||||
if (snapshot == null) {
|
||||
var lock = transactionLock.getAt(index);
|
||||
partialUnlockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataUnlockUpgrade(Object snapshot, DataLockType dataLockMode, @NotNull T key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
var lock = transactionLock.get(key);
|
||||
partialUnlockKeyUpgrade(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataUnlockUpgradeAt(Object snapshot, DataLockType dataLockMode, @NotNull int index) {
|
||||
if (snapshot == null) {
|
||||
var lock = transactionLock.getAt(index);
|
||||
partialUnlockKeyUpgrade(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
|
||||
public void dataLockMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<? extends T> key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
for (ReadWriteUpdateLock lock : transactionLock.bulkGet(key)) {
|
||||
partialLockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void dataLockAtMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<Integer> key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
for (ReadWriteUpdateLock lock : transactionLock.bulkGetAt(key)) {
|
||||
partialLockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void dataUnlockMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<? extends T> key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
for (ReadWriteUpdateLock lock : transactionLock.bulkGet(key)) {
|
||||
partialUnlockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void dataUnlockAtMulti(Object snapshot, DataLockType dataLockMode, @NotNull Iterable<Integer> key) {
|
||||
if (snapshot == null) {
|
||||
Objects.requireNonNull(key, () -> "Lock key must not be null");
|
||||
for (ReadWriteUpdateLock lock : transactionLock.bulkGetAt(key)) {
|
||||
partialUnlockKey(dataLockMode, lock);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,55 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class TransitLock {
|
||||
|
||||
private final FlexibleCountDownLatch unfinishedTransits = new FlexibleCountDownLatch(0);
|
||||
private final Semaphore permits = new Semaphore(Integer.MAX_VALUE);
|
||||
private final AtomicInteger allowedTransits = new AtomicInteger(1);
|
||||
private final Object synchronization = new Object();
|
||||
|
||||
public TransitLock() {
|
||||
}
|
||||
|
||||
public void allowTransit() {
|
||||
var allowedTransits = this.allowedTransits.incrementAndGet();
|
||||
if (allowedTransits == 1) {
|
||||
var permitsToRelease = Math.max(0, Integer.MAX_VALUE - (permits.availablePermits() + 10000));
|
||||
permits.release(permitsToRelease);
|
||||
} else if (allowedTransits > 1) {
|
||||
throw new IllegalStateException("Already allowed transit!");
|
||||
}
|
||||
}
|
||||
|
||||
public void disallowTransit() {
|
||||
synchronized (synchronization) {
|
||||
var allowedTransits = this.allowedTransits.decrementAndGet();
|
||||
if (allowedTransits == 0) {
|
||||
unfinishedTransits.await();
|
||||
permits.drainPermits();
|
||||
} else if (allowedTransits > 0) {
|
||||
throw new IllegalStateException("Transits are still allowed for an unknown reason!");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void transit() {
|
||||
synchronized (synchronization) {
|
||||
startTransit();
|
||||
endTransit();
|
||||
}
|
||||
}
|
||||
|
||||
public void startTransit() {
|
||||
synchronized (synchronization) {
|
||||
unfinishedTransits.grow();
|
||||
permits.acquireUninterruptibly();
|
||||
}
|
||||
}
|
||||
|
||||
public void endTransit() {
|
||||
unfinishedTransits.countDown();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,69 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
|
||||
public class TransitReadWriteLock {
|
||||
|
||||
private final TransitLock writeLock;
|
||||
private final TransitLock readLock;
|
||||
private final Object synchronization = new Object();
|
||||
|
||||
public TransitReadWriteLock() {
|
||||
this.writeLock = new TransitLock();
|
||||
this.readLock = new TransitLock();
|
||||
}
|
||||
|
||||
public void reAllowTransit() {
|
||||
this.readLock.allowTransit();
|
||||
this.writeLock.allowTransit();
|
||||
}
|
||||
|
||||
public void disallowTransit() {
|
||||
synchronized (synchronization) {
|
||||
this.writeLock.disallowTransit();
|
||||
this.readLock.disallowTransit();
|
||||
}
|
||||
}
|
||||
|
||||
public void reAllowTransitRead() {
|
||||
this.readLock.allowTransit();
|
||||
}
|
||||
|
||||
public void disallowTransitRead() {
|
||||
this.readLock.disallowTransit();
|
||||
}
|
||||
|
||||
public void reAllowTransitWrite() {
|
||||
this.writeLock.allowTransit();
|
||||
}
|
||||
|
||||
public void disallowTransitWrite() {
|
||||
this.writeLock.disallowTransit();
|
||||
}
|
||||
|
||||
public void transitWrite() {
|
||||
this.writeLock.transit();
|
||||
}
|
||||
|
||||
public void transitRead() {
|
||||
this.readLock.transit();
|
||||
}
|
||||
|
||||
public void startTransitWrite() {
|
||||
synchronized (synchronization) {
|
||||
this.writeLock.startTransit();
|
||||
}
|
||||
}
|
||||
|
||||
public void startTransitRead() {
|
||||
synchronized (synchronization) {
|
||||
this.readLock.startTransit();
|
||||
}
|
||||
}
|
||||
|
||||
public void endTransitWrite() {
|
||||
writeLock.endTransit();
|
||||
}
|
||||
|
||||
public void endTransitRead() {
|
||||
readLock.endTransit();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,184 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.Timeout;
|
||||
|
||||
public class LeftRightLockTest {
|
||||
|
||||
int logLineSequenceNumber = 0;
|
||||
private LeftRightLock sut = new LeftRightLock();
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void acquiringLeftLockExcludeAcquiringRightLock() throws Exception {
|
||||
sut.lockLeft();
|
||||
|
||||
|
||||
Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockRight());
|
||||
assertFalse(task.get(), "I shouldn't be able to acquire the RIGHT lock!");
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void acquiringRightLockExcludeAcquiringLeftLock() throws Exception {
|
||||
sut.lockRight();
|
||||
Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockLeft());
|
||||
assertFalse(task.get(), "I shouldn't be able to acquire the LEFT lock!");
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void theLockShouldBeReentrant() throws Exception {
|
||||
sut.lockLeft();
|
||||
assertTrue(sut.tryLockLeft());
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void multipleThreadShouldBeAbleToAcquireTheSameLock_Right() throws Exception {
|
||||
sut.lockRight();
|
||||
Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockRight());
|
||||
assertTrue(task.get());
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void multipleThreadShouldBeAbleToAcquireTheSameLock_left() throws Exception {
|
||||
sut.lockLeft();
|
||||
Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> sut.tryLockLeft());
|
||||
assertTrue(task.get());
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void shouldKeepCountOfAllTheThreadsHoldingTheSide() throws Exception {
|
||||
|
||||
CountDownLatch latchA = new CountDownLatch(1);
|
||||
CountDownLatch latchB = new CountDownLatch(1);
|
||||
|
||||
|
||||
Thread threadA = spawnThreadToAcquireLeftLock(latchA, sut);
|
||||
Thread threadB = spawnThreadToAcquireLeftLock(latchB, sut);
|
||||
|
||||
System.out.println("Both threads have acquired the left lock.");
|
||||
|
||||
try {
|
||||
latchA.countDown();
|
||||
threadA.join();
|
||||
boolean acqStatus = sut.tryLockRight();
|
||||
System.out.println("The right lock was " + (acqStatus ? "" : "not") + " acquired");
|
||||
assertFalse(acqStatus, "There is still a thread holding the left lock. This shouldn't succeed.");
|
||||
} finally {
|
||||
latchB.countDown();
|
||||
threadB.join();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void shouldBlockThreadsTryingToAcquireLeftIfRightIsHeld() throws Exception {
|
||||
sut.lockLeft();
|
||||
|
||||
CountDownLatch taskStartedLatch = new CountDownLatch(1);
|
||||
|
||||
final Future<Boolean> task = Executors.newSingleThreadExecutor().submit(() -> {
|
||||
taskStartedLatch.countDown();
|
||||
sut.lockRight();
|
||||
return false;
|
||||
});
|
||||
|
||||
taskStartedLatch.await();
|
||||
Thread.sleep(100);
|
||||
|
||||
assertFalse(task.isDone());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldBeFreeAfterRelease() throws Exception {
|
||||
sut.lockLeft();
|
||||
sut.releaseLeft();
|
||||
assertTrue(sut.tryLockRight());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldBeFreeAfterMultipleThreadsReleaseIt() throws Exception {
|
||||
CountDownLatch latch = new CountDownLatch(1);
|
||||
|
||||
final Thread thread1 = spawnThreadToAcquireLeftLock(latch, sut);
|
||||
final Thread thread2 = spawnThreadToAcquireLeftLock(latch, sut);
|
||||
|
||||
latch.countDown();
|
||||
|
||||
thread1.join();
|
||||
thread2.join();
|
||||
|
||||
assertTrue(sut.tryLockRight());
|
||||
|
||||
}
|
||||
|
||||
@Timeout(2000)
|
||||
@Test()
|
||||
public void lockShouldBeReleasedIfNoThreadIsHoldingIt() throws Exception {
|
||||
CountDownLatch releaseLeftLatch = new CountDownLatch(1);
|
||||
CountDownLatch rightLockTaskIsRunning = new CountDownLatch(1);
|
||||
|
||||
Thread leftLockThread1 = spawnThreadToAcquireLeftLock(releaseLeftLatch, sut);
|
||||
Thread leftLockThread2 = spawnThreadToAcquireLeftLock(releaseLeftLatch, sut);
|
||||
|
||||
Future<Boolean> acquireRightLockTask = Executors.newSingleThreadExecutor().submit(() -> {
|
||||
if (sut.tryLockRight())
|
||||
throw new AssertionError("The left lock should be still held, I shouldn't be able to acquire right a this point.");
|
||||
printSynchronously("Going to be blocked on right lock");
|
||||
rightLockTaskIsRunning.countDown();
|
||||
sut.lockRight();
|
||||
printSynchronously("Lock acquired!");
|
||||
return true;
|
||||
});
|
||||
|
||||
rightLockTaskIsRunning.await();
|
||||
|
||||
releaseLeftLatch.countDown();
|
||||
leftLockThread1.join();
|
||||
leftLockThread2.join();
|
||||
|
||||
assertTrue(acquireRightLockTask.get());
|
||||
}
|
||||
|
||||
private synchronized void printSynchronously(String str) {
|
||||
|
||||
System.out.println(logLineSequenceNumber++ + ")" + str);
|
||||
System.out.flush();
|
||||
}
|
||||
|
||||
private Thread spawnThreadToAcquireLeftLock(CountDownLatch releaseLockLatch, LeftRightLock lock) throws InterruptedException {
|
||||
CountDownLatch lockAcquiredLatch = new CountDownLatch(1);
|
||||
final Thread thread = spawnThreadToAcquireLeftLock(releaseLockLatch, lockAcquiredLatch, lock);
|
||||
lockAcquiredLatch.await();
|
||||
return thread;
|
||||
}
|
||||
|
||||
private Thread spawnThreadToAcquireLeftLock(CountDownLatch releaseLockLatch, CountDownLatch lockAcquiredLatch, LeftRightLock lock) {
|
||||
final Thread thread = new Thread(() -> {
|
||||
lock.lockLeft();
|
||||
printSynchronously("Thread " + Thread.currentThread() + " Acquired left lock");
|
||||
try {
|
||||
lockAcquiredLatch.countDown();
|
||||
releaseLockLatch.await();
|
||||
} catch (InterruptedException ignore) {
|
||||
} finally {
|
||||
lock.releaseLeft();
|
||||
}
|
||||
|
||||
printSynchronously("Thread " + Thread.currentThread() + " RELEASED left lock");
|
||||
});
|
||||
thread.start();
|
||||
return thread;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,132 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.warp.commonutils.locks.SnapshottableCollectionLock.DataLockType;
|
||||
import org.warp.commonutils.locks.SnapshottableCollectionLock.FullLockType;
|
||||
|
||||
public class SnapshottableCollectionLockTest {
|
||||
|
||||
@Test
|
||||
public void testNoop() {
|
||||
var lock = new SnapshottableCollectionLock<Long>(50);
|
||||
checkSituation(lock, true, true, true, true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDenyOnlyFullWrite() {
|
||||
var lock = new SnapshottableCollectionLock<Long>(50);
|
||||
lock.fullLock(null, FullLockType.DENY_ONLY_FULL_WRITE);
|
||||
checkSituation(lock, true, false, true, true);
|
||||
lock.fullUnlock(null, FullLockType.DENY_ONLY_FULL_WRITE);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAllowOnlyRead() {
|
||||
var lock = new SnapshottableCollectionLock<Long>(50);
|
||||
lock.fullLock(null, FullLockType.ALLOW_ONLY_READ);
|
||||
checkSituation(lock, true, false, true, false);
|
||||
lock.fullUnlock(null, FullLockType.ALLOW_ONLY_READ);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAllowOnlyItemsRead() {
|
||||
var lock = new SnapshottableCollectionLock<Long>(50);
|
||||
lock.fullLock(null, FullLockType.ALLOW_ONLY_ITEMS_READ);
|
||||
checkSituation(lock, false, false, true, false);
|
||||
lock.fullUnlock(null, FullLockType.ALLOW_ONLY_ITEMS_READ);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAllowOnlyFullRead() {
|
||||
var lock = new SnapshottableCollectionLock<Long>(50);
|
||||
lock.fullLock(null, FullLockType.ALLOW_ONLY_FULL_READ);
|
||||
checkSituation(lock, true, false, false, false);
|
||||
lock.fullUnlock(null, FullLockType.ALLOW_ONLY_FULL_READ);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExclusive() {
|
||||
var lock = new SnapshottableCollectionLock<Long>(50);
|
||||
lock.fullLock(null, FullLockType.EXCLUSIVE);
|
||||
checkSituation(lock, false, false, false, false);
|
||||
lock.fullUnlock(null, FullLockType.EXCLUSIVE);
|
||||
}
|
||||
|
||||
private void checkSituation(SnapshottableCollectionLock<Long> lock, boolean fullReadAllow, boolean fullWriteAllow, boolean dataReadAllow, boolean dataWriteAllow) {
|
||||
|
||||
var pool = Executors.newWorkStealingPool();
|
||||
|
||||
for (FullLockType readFullType : FullLockType.READ_FULL_TYPES) {
|
||||
if (fullReadAllow) {
|
||||
Assertions.assertDoesNotThrow(() -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.fullLock(null, readFullType);
|
||||
lock.fullUnlock(null, readFullType);
|
||||
}, pool).get(2, TimeUnit.SECONDS);
|
||||
});
|
||||
} else {
|
||||
Assertions.assertThrows(TimeoutException.class, () -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.fullLock(null, readFullType);
|
||||
lock.fullUnlock(null, readFullType);
|
||||
}, pool).get(50, TimeUnit.MILLISECONDS);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
for (FullLockType writeFullType : FullLockType.WRITE_FULL_TYPES) {
|
||||
if (fullWriteAllow) {
|
||||
Assertions.assertDoesNotThrow(() -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.fullLock(null, writeFullType);
|
||||
lock.fullUnlock(null, writeFullType);
|
||||
}, pool).get(2, TimeUnit.SECONDS);
|
||||
});
|
||||
} else {
|
||||
Assertions.assertThrows(TimeoutException.class, () -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.fullLock(null, writeFullType);
|
||||
lock.fullUnlock(null, writeFullType);
|
||||
}, pool).get(50, TimeUnit.MILLISECONDS);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (dataReadAllow) {
|
||||
Assertions.assertDoesNotThrow(() -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.dataLock(null, DataLockType.READ, 1L);
|
||||
lock.dataUnlock(null, DataLockType.READ, 1L);
|
||||
}, pool).get(2, TimeUnit.SECONDS);
|
||||
});
|
||||
} else {
|
||||
Assertions.assertThrows(TimeoutException.class, () -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.dataLock(null, DataLockType.READ, 1L);
|
||||
lock.dataUnlock(null, DataLockType.READ, 1L);
|
||||
}, pool).get(50, TimeUnit.MILLISECONDS);
|
||||
});
|
||||
}
|
||||
|
||||
if (dataWriteAllow) {
|
||||
Assertions.assertDoesNotThrow(() -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.dataLock(null, DataLockType.WRITE, 1L);
|
||||
lock.dataUnlock(null, DataLockType.WRITE, 1L);
|
||||
}, pool).get(2, TimeUnit.SECONDS);
|
||||
});
|
||||
} else {
|
||||
Assertions.assertThrows(TimeoutException.class, () -> {
|
||||
CompletableFuture.runAsync(() -> {
|
||||
lock.dataLock(null, DataLockType.WRITE, 1L);
|
||||
lock.dataUnlock(null, DataLockType.WRITE, 1L);
|
||||
}, pool).get(50, TimeUnit.MILLISECONDS);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,156 +0,0 @@
|
|||
package org.warp.commonutils.locks;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
public class TransitLockTest {
|
||||
|
||||
@Test
|
||||
public void testTransit() {
|
||||
var lock = new TransitLock();
|
||||
lock.transit();
|
||||
lock.transit();
|
||||
lock.transit();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPartialTransit() {
|
||||
var lock = new TransitLock();
|
||||
lock.startTransit();
|
||||
lock.endTransit();
|
||||
lock.startTransit();
|
||||
lock.endTransit();
|
||||
lock.startTransit();
|
||||
lock.endTransit();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAllowTransit() {
|
||||
var lock = new TransitLock();
|
||||
lock.disallowTransit();
|
||||
lock.allowTransit();
|
||||
lock.transit();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMultiAllowTransit() {
|
||||
Assertions.assertThrows(IllegalStateException.class, () -> {
|
||||
var lock = new TransitLock();
|
||||
lock.allowTransit();
|
||||
lock.allowTransit();
|
||||
lock.allowTransit();
|
||||
lock.transit();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMultiDisallowAllowTransit() {
|
||||
var lock = new TransitLock();
|
||||
lock.disallowTransit();
|
||||
lock.allowTransit();
|
||||
lock.disallowTransit();
|
||||
lock.disallowTransit();
|
||||
lock.allowTransit();
|
||||
lock.allowTransit();
|
||||
lock.transit();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDisallowTransit() {
|
||||
var lock = new TransitLock();
|
||||
lock.disallowTransit();
|
||||
for (int i = 0; i < 10; i++) {
|
||||
Assertions.assertThrows(TimeoutException.class, () -> {
|
||||
CompletableFuture.runAsync(lock::transit).get(5, TimeUnit.MILLISECONDS);
|
||||
}, "Disallowed transit didn't block a transit");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMultiDisallowTransit() {
|
||||
var lock = new TransitLock();
|
||||
lock.disallowTransit();
|
||||
lock.disallowTransit();
|
||||
lock.disallowTransit();
|
||||
for (int i = 0; i < 10; i++) {
|
||||
Assertions.assertThrows(TimeoutException.class, () -> {
|
||||
CompletableFuture.runAsync(lock::transit).get(5, TimeUnit.MILLISECONDS);
|
||||
}, "Disallowed transit didn't block a transit");
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDeadlocks() {
|
||||
var lock = new TransitLock();
|
||||
AtomicInteger alreadyRunningTransits = new AtomicInteger();
|
||||
|
||||
var pool = Executors.newFixedThreadPool(500);
|
||||
AtomicReference<Exception> failure = new AtomicReference<>();
|
||||
for (int i = 0; i < 100; i++) {
|
||||
int iF = i;
|
||||
pool.submit(() -> {
|
||||
try {
|
||||
for (int j = 0; j < 500; j++) {
|
||||
if (iF % 2 == 0) {
|
||||
lock.startTransit();
|
||||
alreadyRunningTransits.getAndIncrement();
|
||||
alreadyRunningTransits.decrementAndGet();
|
||||
lock.endTransit();
|
||||
} else {
|
||||
lock.disallowTransit();
|
||||
Assertions.assertEquals(0, alreadyRunningTransits.get());
|
||||
lock.allowTransit();
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
failure.set(e);
|
||||
}
|
||||
});
|
||||
}
|
||||
pool.shutdown();
|
||||
if (failure.get() != null) throw new AssertionError(failure.get());
|
||||
Assertions.assertDoesNotThrow(() -> pool.awaitTermination(10, TimeUnit.SECONDS));
|
||||
Assertions.assertTrue(pool.isTerminated());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testParallelTransit() {
|
||||
var lock = new TransitLock();
|
||||
AtomicInteger alreadyRunningTransits = new AtomicInteger();
|
||||
|
||||
var pool = Executors.newFixedThreadPool(100);
|
||||
AtomicReference<AssertionError> failure = new AtomicReference<>();
|
||||
for (int i = 0; i < 100; i++) {
|
||||
pool.submit(() -> {
|
||||
try {
|
||||
lock.startTransit();
|
||||
alreadyRunningTransits.getAndIncrement();
|
||||
try {
|
||||
Thread.sleep(20);
|
||||
} catch (InterruptedException e) {
|
||||
Assertions.fail(e);
|
||||
}
|
||||
alreadyRunningTransits.decrementAndGet();
|
||||
lock.endTransit();
|
||||
} catch (AssertionError e) {
|
||||
e.printStackTrace();
|
||||
failure.set(e);
|
||||
}
|
||||
});
|
||||
}
|
||||
lock.disallowTransit();
|
||||
Assertions.assertEquals(0, alreadyRunningTransits.get());
|
||||
lock.allowTransit();
|
||||
pool.shutdown();
|
||||
if (failure.get() != null) throw failure.get();
|
||||
Assertions.assertDoesNotThrow(() -> pool.awaitTermination(10, TimeUnit.SECONDS));
|
||||
Assertions.assertTrue(pool.isTerminated());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue