You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2021/10/28 12:27:58 UTC
[ignite-3] branch main updated: IGNITE-15536 Use VarHandle to
replace UNSAFE in IgniteSpinReadWriteLock (#414)
This is an automated email from the ASF dual-hosted git repository.
ibessonov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 0c4ec19 IGNITE-15536 Use VarHandle to replace UNSAFE in IgniteSpinReadWriteLock (#414)
0c4ec19 is described below
commit 0c4ec196041af6faa6034b602e6c377ca6fb2f7c
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Thu Oct 28 16:27:52 2021 +0400
IGNITE-15536 Use VarHandle to replace UNSAFE in IgniteSpinReadWriteLock (#414)
---
.../internal/util/IgniteSpinReadWriteLock.java | 373 ++++++++-------
.../internal/util/IgniteSpinReadWriteLockTest.java | 526 +++++++++++++++++++++
2 files changed, 740 insertions(+), 159 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLock.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLock.java
index f1deb9f..6c9a98c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLock.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLock.java
@@ -17,65 +17,105 @@
package org.apache.ignite.internal.util;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
import java.util.concurrent.TimeUnit;
+
import org.apache.ignite.internal.tostring.S;
/**
- * Spin read-write lock.
+ * <p>Spin read-write lock.
+ * Its blocking methods use the spinwait strategy. When they do so, they are not interruptible (that is, they do not
+ * break their loop on interruption signal).
+ * <p>The locks are reentrant (that is, the same thread can acquire the same lock a few times in a row and then
+ * release them same number of times.
+ * <p>Write lock acquire requests are prioritized over read lock acquire requests. That is, if both read and write lock
+ * acquire requests are received when the write lock is held by someone else, then, on its release,
+ * the write lock attempt will be served first.
*/
public class IgniteSpinReadWriteLock {
- /** */
- private static final long PENDING_WLOCKS_OFFS;
+ /** Signals that nobody currently owns the read lock. */
+ private static final long NO_OWNER = -1;
- /** */
- private static final long STATE_OFFS;
+ /**
+ * State -1 means that the write lock is acquired.
+ *
+ * @see #state
+ */
+ private static final int WRITE_LOCKED = -1;
/**
- * TODO: replace UNSAFE usage using VarHandle https://issues.apache.org/jira/browse/IGNITE-15536
+ * State 0 means that both read and write locks are available for acquiring.
+ *
+ * @see #state
*/
+ private static final int AVAILABLE = 0;
+
+ /** How much time to sleep on each iteration of a spin loop (milliseconds). */
+ private static final int SLEEP_MILLIS = 10;
+
+ /** {@link VarHandle} used to access the {@code pendingWLocks} field. */
+ private static final VarHandle PENDING_WLOCKS_VH;
+
+ /** {@link VarHandle} used to access the {@code state} field. */
+ private static final VarHandle STATE_VH;
+
static {
try {
- STATE_OFFS = GridUnsafe.objectFieldOffset(IgniteSpinReadWriteLock.class.getDeclaredField("state"));
+ STATE_VH = MethodHandles.lookup()
+ .findVarHandle(IgniteSpinReadWriteLock.class, "state", int.class);
- PENDING_WLOCKS_OFFS =
- GridUnsafe.objectFieldOffset(IgniteSpinReadWriteLock.class.getDeclaredField("pendingWLocks"));
+ PENDING_WLOCKS_VH = MethodHandles.lookup()
+ .findVarHandle(IgniteSpinReadWriteLock.class, "pendingWLocks", int.class);
}
- catch (NoSuchFieldException e) {
- throw new Error(e);
+ catch (ReflectiveOperationException e) {
+ throw new ExceptionInInitializerError(e);
}
}
- /** */
- private final ThreadLocal<Integer> readLockEntryCnt = new ThreadLocal<Integer>() {
- @Override protected Integer initialValue() {
- return 0;
- }
- };
+ /** Number of times read lock was acquired, per thread (used to track reentrance). */
+ private final ThreadLocal<Integer> readLockEntryCnt = ThreadLocal.withInitial(() -> 0);
- /** */
+ /**
+ * Main state of the lock.
+ * <ul>
+ * <li>Positive when the read lock has been acquired by at least one thread; in such case, this number equals
+ * to the number of threads holding the read lock. In such state, the read lock may be acquired by any thread,
+ * while an attempt to acquire the write lock will block or fail.</li>
+ * <li>Zero when neither the read lock, nor the write lock has been acquired by any thread. This state allows
+ * a thread to acquire either the read or the write lock at will.</li>
+ * <li>-1 when the write lock has been acquired by exactly one thread. In such state, any attempt to acquire the read
+ * or the write lock by any thread (but the thread holding the write lock) will block or fail.</li>
+ * </ul>
+ */
private volatile int state;
- /** */
+ /**
+ * Number of pending write attempts to acquire the write lock. Currently it is only used to prioritize
+ * write lock attempts over read lock attempts when the write lock has been released (so, if both an attempt to acquire
+ * the write lock and an attempt to acquire the read lock are waiting for write lock to be released, a write lock attempt
+ * will be served first when the release happens).
+ */
private volatile int pendingWLocks;
- /** */
- private long writeLockOwner = -1;
+ /** ID of the thread holding write lock (or {@link #NO_OWNER} if the write lock is not held). */
+ private long writeLockOwner = NO_OWNER;
- /** */
+ /** Number of times the write lock holder locked the write lock (used to track reentrance). */
private int writeLockEntryCnt;
/**
- * Acquires read lock.
+ * Acquires the read lock. If the write lock is held by another thread, this blocks until the write lock is released
+ * (and until all concurrent write locks are acquired and released, as this class pripritizes write lock attempts
+ * over read lock attempts).
*/
@SuppressWarnings("BusyWait")
public void readLock() {
int cnt = readLockEntryCnt.get();
// Read lock reentry or acquiring read lock while holding write lock.
- if (cnt > 0 || Thread.currentThread().getId() == writeLockOwner) {
- assert state > 0 || state == -1;
-
- readLockEntryCnt.set(cnt + 1);
+ if (alreadyHoldingAnyLock(cnt)) {
+ incrementCurrentThreadReadLockCount(cnt);
return;
}
@@ -83,13 +123,13 @@ public class IgniteSpinReadWriteLock {
boolean interrupted = false;
while (true) {
- int cur = state;
+ int curState = state;
- assert cur >= -1;
+ assert curState >= WRITE_LOCKED;
- if (cur == -1 || pendingWLocks > 0) {
+ if (writeLockedOrGoingToBe(curState)) {
try {
- Thread.sleep(10);
+ Thread.sleep(SLEEP_MILLIS);
}
catch (InterruptedException ignored) {
interrupted = true;
@@ -98,7 +138,7 @@ public class IgniteSpinReadWriteLock {
continue;
}
- if (compareAndSet(STATE_OFFS, cur, cur + 1)) {
+ if (tryAdvanceStateToReadLocked(curState)) {
if (interrupted)
Thread.currentThread().interrupt();
@@ -110,29 +150,55 @@ public class IgniteSpinReadWriteLock {
}
/**
- * Tries to acquire read lock.
+ * Whether the current thread already holds any lock.
*
- * @return {@code true} if acquired.
+ * @param currentThreadReadLockAcquiredCount how many times current thread acquired (without releasing yet)
+ * the read lock
+ * @return true if current thread already holds any lock
+ */
+ private boolean alreadyHoldingAnyLock(int currentThreadReadLockAcquiredCount) {
+ return currentThreadReadLockAcquiredCount > 0 || writeLockedByCurrentThread();
+ }
+
+ /***/
+ private void incrementCurrentThreadReadLockCount(int cnt) {
+ assert state > 0 || state == WRITE_LOCKED;
+
+ readLockEntryCnt.set(cnt + 1);
+ }
+
+ /***/
+ private boolean writeLockedOrGoingToBe(int curState) {
+ return curState == WRITE_LOCKED || pendingWLocks > 0;
+ }
+
+ /***/
+ private boolean tryAdvanceStateToReadLocked(int curState) {
+ return compareAndSet(STATE_VH, curState, curState + 1);
+ }
+
+ /**
+ * Tries to acquire the read lock. No spinwait is used if the lock cannot be acquired immediately.
+ *
+ * @return {@code true} if acquired, {@code false} if write lock is already held by someone else
*/
public boolean tryReadLock() {
int cnt = readLockEntryCnt.get();
// Read lock reentry or acquiring read lock while holding write lock.
- if (cnt > 0 || Thread.currentThread().getId() == writeLockOwner) {
- assert state > 0 || state == -1;
-
- readLockEntryCnt.set(cnt + 1);
+ if (alreadyHoldingAnyLock(cnt)) {
+ incrementCurrentThreadReadLockCount(cnt);
return true;
}
while (true) {
- int cur = state;
+ int curState = state;
- if (cur == -1 || pendingWLocks > 0)
+ if (writeLockedOrGoingToBe(curState))
return false;
- if (compareAndSet(STATE_OFFS, cur, cur + 1)) {
+ if (tryAdvanceStateToReadLocked(curState)) {
readLockEntryCnt.set(1);
return true;
@@ -141,7 +207,9 @@ public class IgniteSpinReadWriteLock {
}
/**
- * Read unlock.
+ * Releases the read lock.
+ *
+ * @throws IllegalMonitorStateException thrown if the current thread does not hold the read lock
*/
public void readUnlock() {
int cnt = readLockEntryCnt.get();
@@ -150,8 +218,8 @@ public class IgniteSpinReadWriteLock {
throw new IllegalMonitorStateException();
// Read unlock when holding write lock is performed here.
- if (cnt > 1 || Thread.currentThread().getId() == writeLockOwner) {
- assert state > 0 || state == -1;
+ if (cnt > 1 || writeLockedByCurrentThread()) {
+ assert state > 0 || state == WRITE_LOCKED;
readLockEntryCnt.set(cnt - 1);
@@ -159,11 +227,11 @@ public class IgniteSpinReadWriteLock {
}
while (true) {
- int cur = state;
+ int curState = state;
- assert cur > 0;
+ assert curState > 0;
- if (compareAndSet(STATE_OFFS, cur, cur - 1)) {
+ if (compareAndSet(STATE_VH, curState, curState - 1)) {
readLockEntryCnt.set(0);
return;
@@ -172,102 +240,105 @@ public class IgniteSpinReadWriteLock {
}
/**
- * Acquires write lock.
+ * Acquires the write lock waiting, if needed. The thread will block until all other threads release both read
+ * and write locks.
*/
@SuppressWarnings("BusyWait")
public void writeLock() {
- long threadId = Thread.currentThread().getId();
-
- if (threadId == writeLockOwner) {
- assert state == -1;
-
- writeLockEntryCnt++;
+ if (writeLockedByCurrentThread()) {
+ incrementWriteLockCount();
return;
}
- // Increment pending write locks.
- while (true) {
- int pendingWLocks0 = pendingWLocks;
-
- if (compareAndSet(PENDING_WLOCKS_OFFS, pendingWLocks0, pendingWLocks0 + 1))
- break;
- }
-
boolean interrupted = false;
- while (!compareAndSet(STATE_OFFS, 0, -1)) {
- try {
- Thread.sleep(10);
- }
- catch (InterruptedException ignored) {
- interrupted = true;
+ incrementPendingWriteLocks();
+ try {
+ while (!trySwitchStateToWriteLocked()) {
+ try {
+ Thread.sleep(SLEEP_MILLIS);
+ }
+ catch (InterruptedException ignored) {
+ interrupted = true;
+ }
}
}
-
- // Decrement pending write locks.
- while (true) {
- int pendingWLocks0 = pendingWLocks;
-
- assert pendingWLocks0 > 0;
-
- if (compareAndSet(PENDING_WLOCKS_OFFS, pendingWLocks0, pendingWLocks0 - 1))
- break;
+ finally {
+ decrementPendingWriteLocks();
}
if (interrupted)
Thread.currentThread().interrupt();
- assert writeLockOwner == -1;
-
- writeLockOwner = threadId;
- writeLockEntryCnt = 1;
+ finishWriteLockAcquire();
}
- /**
- * Acquires write lock without sleeping between unsuccessful attempts.
- */
- public void writeLock0() {
- long threadId = Thread.currentThread().getId();
+ /***/
+ private void incrementWriteLockCount() {
+ assert state == WRITE_LOCKED;
- if (threadId == writeLockOwner) {
- assert state == -1;
-
- writeLockEntryCnt++;
-
- return;
- }
+ writeLockEntryCnt++;
+ }
- // Increment pending write locks.
+ /***/
+ private void incrementPendingWriteLocks() {
while (true) {
- int pendingWLocks0 = pendingWLocks;
+ int curPendingWLocks = pendingWLocks;
- if (compareAndSet(PENDING_WLOCKS_OFFS, pendingWLocks0, pendingWLocks0 + 1))
+ if (compareAndSet(PENDING_WLOCKS_VH, curPendingWLocks, curPendingWLocks + 1))
break;
}
+ }
- for (;;) {
- if (compareAndSet(STATE_OFFS, 0, -1))
- break;
- }
+ /***/
+ private boolean trySwitchStateToWriteLocked() {
+ return compareAndSet(STATE_VH, AVAILABLE, WRITE_LOCKED);
+ }
- // Decrement pending write locks.
+ /***/
+ private void decrementPendingWriteLocks() {
while (true) {
- int pendingWLocks0 = pendingWLocks;
+ int curPendingWLocks = pendingWLocks;
- assert pendingWLocks0 > 0;
+ assert curPendingWLocks > 0;
- if (compareAndSet(PENDING_WLOCKS_OFFS, pendingWLocks0, pendingWLocks0 - 1))
+ if (compareAndSet(PENDING_WLOCKS_VH, curPendingWLocks, curPendingWLocks - 1))
break;
}
+ }
- assert writeLockOwner == -1;
+ /***/
+ private void finishWriteLockAcquire() {
+ assert writeLockOwner == NO_OWNER;
- writeLockOwner = threadId;
+ writeLockOwner = Thread.currentThread().getId();
writeLockEntryCnt = 1;
}
/**
+ * Acquires the write lock without sleeping between unsuccessful attempts. Instead, the spinwait eats cycles of
+ * the core it gets at full speed. It is non-interruptible as its {@link #writeLock()} cousin.
+ */
+ public void writeLockBusy() {
+ if (writeLockedByCurrentThread()) {
+ incrementWriteLockCount();
+
+ return;
+ }
+
+ incrementPendingWriteLocks();
+ try {
+ while (!trySwitchStateToWriteLocked());
+ }
+ finally {
+ decrementPendingWriteLocks();
+ }
+
+ finishWriteLockAcquire();
+ }
+
+ /**
* @return {@code True} if blocked by current thread.
*/
public boolean writeLockedByCurrentThread() {
@@ -275,26 +346,20 @@ public class IgniteSpinReadWriteLock {
}
/**
- * Tries to acquire write lock.
+ * Tries to acquire the write lock. Never blocks: if any lock has already been acquired by someone else,
+ * returns {@code false} immediately.
*
- * @return {@code True} if write lock has been acquired.
+ * @return {@code true} if the write lock has been acquired, {@code false} otherwise
*/
public boolean tryWriteLock() {
- long threadId = Thread.currentThread().getId();
-
- if (threadId == writeLockOwner) {
- assert state == -1;
-
- writeLockEntryCnt++;
+ if (writeLockedByCurrentThread()) {
+ incrementWriteLockCount();
return true;
}
- if (compareAndSet(STATE_OFFS, 0, -1)) {
- assert writeLockOwner == -1;
-
- writeLockOwner = threadId;
- writeLockEntryCnt = 1;
+ if (trySwitchStateToWriteLocked()) {
+ finishWriteLockAcquire();
return true;
}
@@ -303,72 +368,53 @@ public class IgniteSpinReadWriteLock {
}
/**
+ * Tries to acquire the write lock with timeout. If it gets the write lock before the timeout expires, then
+ * returns {@code true}. If the timeout expires before the lock becomes available, returns {@code false}.
+ *
* @param timeout Timeout.
* @param unit Unit.
- * @return {@code True} if write lock has been acquired.
+ * @return {@code true} if the write lock has been acquired in time; {@code false} otherwise
* @throws InterruptedException If interrupted.
*/
@SuppressWarnings("BusyWait")
public boolean tryWriteLock(long timeout, TimeUnit unit) throws InterruptedException {
- long threadId = Thread.currentThread().getId();
-
- if (threadId == writeLockOwner) {
- assert state == -1;
-
- writeLockEntryCnt++;
+ if (writeLockedByCurrentThread()) {
+ incrementWriteLockCount();
return true;
}
+ incrementPendingWriteLocks();
try {
- // Increment pending write locks.
- while (true) {
- int pendingWLocks0 = pendingWLocks;
-
- if (compareAndSet(PENDING_WLOCKS_OFFS, pendingWLocks0, pendingWLocks0 + 1))
- break;
- }
-
long startNanos = System.nanoTime();
long timeoutNanos = unit.toNanos(timeout);
while (true) {
- if (compareAndSet(STATE_OFFS, 0, -1)) {
- assert writeLockOwner == -1;
-
- writeLockOwner = threadId;
- writeLockEntryCnt = 1;
+ if (trySwitchStateToWriteLocked()) {
+ finishWriteLockAcquire();
return true;
}
- Thread.sleep(10);
+ Thread.sleep(SLEEP_MILLIS);
if (System.nanoTime() - startNanos >= timeoutNanos)
return false;
}
}
finally {
- // Decrement pending write locks.
- while (true) {
- int pendingWLocks0 = pendingWLocks;
-
- assert pendingWLocks0 > 0;
-
- if (compareAndSet(PENDING_WLOCKS_OFFS, pendingWLocks0, pendingWLocks0 - 1))
- break;
- }
+ decrementPendingWriteLocks();
}
}
/**
- * Releases write lock.
+ * Releases the write lock.
+ *
+ * @throws IllegalMonitorStateException thrown if the current thread does not hold the write lock
*/
public void writeUnlock() {
- long threadId = Thread.currentThread().getId();
-
- if (threadId != writeLockOwner)
+ if (!writeLockedByCurrentThread())
throw new IllegalMonitorStateException();
if (writeLockEntryCnt > 1) {
@@ -378,25 +424,34 @@ public class IgniteSpinReadWriteLock {
}
writeLockEntryCnt = 0;
- writeLockOwner = -1;
+ writeLockOwner = NO_OWNER;
// Current thread holds write and read locks and is releasing
// write lock now.
- int update = readLockEntryCnt.get() > 0 ? 1 : 0;
+ int update = readLockEntryCnt.get() > 0 ? 1 : AVAILABLE;
- boolean b = compareAndSet(STATE_OFFS, -1, update);
+ boolean b = compareAndSet(STATE_VH, WRITE_LOCKED, update);
assert b;
}
/**
- * @param offs Offset.
+ * @param varHandle VarHandle.
* @param expect Expected.
* @param update Update.
* @return {@code True} on success.
*/
- private boolean compareAndSet(long offs, int expect, int update) {
- return GridUnsafe.compareAndSwapInt(this, offs, expect, update);
+ private boolean compareAndSet(VarHandle varHandle, int expect, int update) {
+ return varHandle.compareAndSet(this, expect, update);
+ }
+
+ /**
+ * Returns the count of pending write lock requests count. Only used by tests, should not be used in production code.
+ *
+ * @return count of pending requests to get the write lock
+ */
+ int pendingWriteLocksCount() {
+ return pendingWLocks;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLockTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLockTest.java
new file mode 100644
index 0000000..cfa2f6e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/IgniteSpinReadWriteLockTest.java
@@ -0,0 +1,526 @@
+/*
+ * 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.ignite.internal.util;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link IgniteSpinReadWriteLock}.
+ */
+@Timeout(20)
+class IgniteSpinReadWriteLockTest {
+ /** The lock under test. */
+ private final IgniteSpinReadWriteLock lock = new IgniteSpinReadWriteLock();
+
+ /** Executor service used to run tasks in threads different from the main test thread. */
+ private final ExecutorService executor = Executors.newCachedThreadPool();
+
+ /**
+ * Cleans up after a test.
+ *
+ */
+ @AfterEach
+ void cleanup() {
+ releaseReadLockHeldByCurrentThread();
+ releaseWriteLockHeldByCurrentThread();
+
+ IgniteUtils.shutdownAndAwaitTermination(executor, 3, TimeUnit.SECONDS);
+ }
+
+ /***/
+ private void releaseReadLockHeldByCurrentThread() {
+ while (true) {
+ try {
+ lock.readUnlock();
+ }
+ catch (IllegalMonitorStateException e) {
+ // released our read lock completely
+ break;
+ }
+ }
+ }
+
+ /***/
+ private void releaseWriteLockHeldByCurrentThread() {
+ while (lock.writeLockedByCurrentThread())
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void readLockDoesNotAllowWriteLockToBeAcquired() {
+ lock.readLock();
+
+ assertThatWriteLockAcquireAttemptBlocksForever();
+
+ lock.readUnlock();
+ }
+
+ /***/
+ private void assertThatWriteLockAcquireAttemptBlocksForever() {
+ Future<?> future = executor.submit(lock::writeLock);
+
+ assertThrows(TimeoutException.class, () -> future.get(500, TimeUnit.MILLISECONDS));
+ }
+
+ /**
+ *
+ */
+ @Test
+ void readLockDoesNotAllowWriteLockWithoutSleepsToBeAcquired() {
+ lock.readLock();
+
+ assertThatWriteLockAcquireAttemptWithoutSleepsBlocksForever();
+
+ lock.readUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void readLockDoesNotAllowWriteLockToBeAcquiredWithTimeout() throws Exception {
+ lock.readLock();
+
+ Boolean acquired = callWithTimeout(() -> lock.tryWriteLock(1, TimeUnit.MILLISECONDS));
+ assertThat(acquired, is(false));
+
+ lock.readUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void readLockAllowsReadLockToBeAcquired() throws Exception {
+ lock.readLock();
+
+ assertThatReadLockCanBeAcquired();
+ }
+
+ /***/
+ private void assertThatReadLockCanBeAcquired() throws InterruptedException, ExecutionException, TimeoutException {
+ runWithTimeout(lock::readLock);
+ }
+
+ /***/
+ private <T> T callWithTimeout(Callable<T> call) throws ExecutionException, InterruptedException, TimeoutException {
+ Future<T> future = executor.submit(call);
+ return getWithTimeout(future);
+ }
+
+ /***/
+ private void runWithTimeout(Runnable runnable) throws ExecutionException, InterruptedException, TimeoutException {
+ Future<?> future = executor.submit(runnable);
+ getWithTimeout(future);
+ }
+
+ /***/
+ private static <T> T getWithTimeout(Future<? extends T> future) throws ExecutionException,
+ InterruptedException, TimeoutException {
+ return future.get(10, TimeUnit.SECONDS);
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockDoesNotAllowReadLockToBeAcquired() {
+ lock.writeLock();
+
+ assertThatReadLockAcquireAttemptBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /***/
+ private void assertThatReadLockAcquireAttemptBlocksForever() {
+ Future<?> readLockAttemptFuture = executor.submit(lock::readLock);
+
+ assertThrows(TimeoutException.class, () -> readLockAttemptFuture.get(500, TimeUnit.MILLISECONDS));
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockDoesNotAllowWriteLockToBeAcquired() {
+ lock.writeLock();
+
+ assertThatWriteLockAcquireAttemptBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockAcquiredWithoutSleepsDoesNotAllowReadLockToBeAcquired() {
+ lock.writeLockBusy();
+
+ assertThatReadLockAcquireAttemptBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockAcquiredWithoutSleepsDoesNotAllowWriteLockToBeAcquired() {
+ lock.writeLockBusy();
+
+ assertThatWriteLockAcquireAttemptBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockDoesNotAllowWriteLockWithoutSleepsToBeAcquired() {
+ lock.writeLock();
+
+ assertThatWriteLockAcquireAttemptWithoutSleepsBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /***/
+ private void assertThatWriteLockAcquireAttemptWithoutSleepsBlocksForever() {
+ Future<?> future = executor.submit(lock::writeLockBusy);
+
+ assertThrows(TimeoutException.class, () -> future.get(500, TimeUnit.MILLISECONDS));
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockAcquiredWithoutSleepsDoesNotAllowWriteLockWithoutSleepsToBeAcquired() {
+ lock.writeLockBusy();
+
+ assertThatWriteLockAcquireAttemptWithoutSleepsBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /***/
+ @Test
+ void readUnlockReleasesTheLock() throws Exception {
+ lock.readLock();
+ lock.readUnlock();
+
+ runWithTimeout(lock::writeLock);
+ }
+
+ /***/
+ @Test
+ void writeUnlockReleasesTheLock() throws Exception {
+ lock.writeLock();
+ lock.writeUnlock();
+
+ assertThatReadLockCanBeAcquired();
+ }
+
+ /***/
+ @Test
+ void writeUnlockReleasesTheLockTakenWithoutSleeps() throws Exception {
+ lock.writeLockBusy();
+ lock.writeUnlock();
+
+ assertThatReadLockCanBeAcquired();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void testWriteLockReentry() {
+ lock.writeLock();
+
+ lock.writeLock();
+
+ assertTrue(lock.tryWriteLock());
+ }
+
+ /**
+ *
+ */
+ @Test
+ void testWriteLockReentryWithoutSleeps() {
+ lock.writeLockBusy();
+
+ lock.writeLockBusy();
+
+ assertTrue(lock.tryWriteLock());
+ }
+
+ /**
+ *
+ */
+ @Test
+ void testWriteLockReentryWithTryWriteLock() {
+ lock.tryWriteLock();
+
+ assertTrue(lock.tryWriteLock());
+ }
+
+ /**
+ *
+ */
+ @Test
+ void testWriteLockReentryWithTimeout() throws Exception {
+ lock.tryWriteLock(1, TimeUnit.MILLISECONDS);
+ lock.tryWriteLock(1, TimeUnit.MILLISECONDS);
+
+ assertTrue(lock.tryWriteLock());
+ }
+
+ /**
+ *
+ */
+ @Test
+ void testReadLockReentry() {
+ lock.readLock();
+
+ lock.readLock();
+
+ assertTrue(lock.tryReadLock());
+ }
+
+ /**
+ * @throws Exception If any error occurs.
+ */
+ @Test
+ void testReadLockReentryWhenConcurrentAttemptToAcquireWriteLockHappens() throws Exception {
+ lock.readLock();
+
+ Future<?> future = executor.submit(() -> {
+ assertFalse(lock.tryWriteLock());
+
+ lock.writeLock();
+ });
+
+ waitTillWriteLockAcquireAttemptIsInitiated();
+
+ lock.readLock();
+
+ assertTrue(lock.tryReadLock());
+
+ lock.readUnlock();
+ lock.readUnlock();
+ lock.readUnlock();
+
+ future.get(1, TimeUnit.SECONDS);
+ }
+
+ /***/
+ private void waitTillWriteLockAcquireAttemptIsInitiated() throws InterruptedException {
+ boolean sawAnAttempt = IgniteTestUtils.waitForCondition(
+ () -> lock.pendingWriteLocksCount() > 0, TimeUnit.SECONDS.toMillis(10));
+ assertTrue(sawAnAttempt, "Did not see any attempt to acquire write lock");
+ }
+
+ /**
+ *
+ */
+ @Test
+ void shouldAllowAcquireAndReleaseReadLockWhileHoldingWriteLock() {
+ lock.writeLock();
+
+ lock.readLock();
+ lock.readUnlock();
+
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void shouldAllowInterleavingHoldingReadAndWriteLocks() {
+ lock.writeLock();
+
+ lock.readLock();
+
+ lock.writeUnlock();
+
+ assertFalse(lock.tryWriteLock());
+
+ lock.readUnlock();
+
+ // Test that we can operate with write locks now.
+ lock.writeLock();
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void readLockReleasedLessTimesThanAcquiredShouldStillBeTaken() {
+ lock.readLock();
+ lock.readLock();
+ lock.readUnlock();
+
+ assertThatWriteLockAcquireAttemptBlocksForever();
+
+ lock.readUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockReleasedLessTimesThanAcquiredShouldStillBeTaken() {
+ lock.writeLock();
+ lock.writeLock();
+ lock.writeUnlock();
+
+ assertThatReadLockAcquireAttemptBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /**
+ *
+ */
+ @Test
+ void shouldThrowOnReadUnlockingWhenNotHoldingReadLock() {
+ assertThrows(IllegalMonitorStateException.class, lock::readUnlock);
+ }
+
+ /**
+ *
+ */
+ @Test
+ void shouldThrowOnWriteUnlockingWhenNotHoldingWriteLock() {
+ assertThrows(IllegalMonitorStateException.class, lock::writeUnlock);
+ }
+
+ /**
+ *
+ */
+ @Test
+ void readLockAcquiredWithTryReadLockDoesNotAllowWriteLockToBeAcquired() {
+ lock.tryReadLock();
+
+ assertThatWriteLockAcquireAttemptBlocksForever();
+
+ lock.readUnlock();
+ }
+
+ /**
+ */
+ @Test
+ void tryReadLockShouldReturnTrueWhenReadLockWasAcquiredSuccessfully() {
+ assertTrue(lock.tryReadLock());
+ }
+
+ /**
+ */
+ @Test
+ void tryReadLockShouldReturnFalseWhenReadLockCouldNotBeAcquired() throws Exception {
+ lock.writeLock();
+
+ Boolean acquired = callWithTimeout(lock::tryReadLock);
+
+ assertThat(acquired, is(false));
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockAcquiredWithTryWriteLockDoesNotAllowWriteLockToBeAcquired() {
+ lock.tryWriteLock();
+
+ assertThatReadLockAcquireAttemptBlocksForever();
+
+ lock.writeUnlock();
+ }
+
+ /**
+ */
+ @Test
+ void tryWriteLockShouldReturnTrueWhenWriteLockWasAcquiredSuccessfully() {
+ assertTrue(lock.tryWriteLock());
+ }
+
+ /**
+ */
+ @Test
+ void tryWriteLockShouldReturnFalseWhenWriteLockCouldNotBeAcquired() throws Exception {
+ lock.writeLock();
+
+ Boolean acquired = callWithTimeout(lock::tryWriteLock);
+
+ assertThat(acquired, is(false));
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockedByCurrentThreadShouldReturnTrueWhenLockedByCurrentThread() {
+ lock.writeLock();
+
+ assertTrue(lock.writeLockedByCurrentThread());
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockedByCurrentThreadShouldReturnFalseWhenNotLocked() {
+ assertFalse(lock.writeLockedByCurrentThread());
+ }
+
+ /**
+ *
+ */
+ @Test
+ void writeLockedByCurrentThreadShouldReturnFalseWhenLockedByAnotherThread() throws Exception {
+ lock.writeLock();
+
+ Boolean lockedByCaller = callWithTimeout(lock::writeLockedByCurrentThread);
+ assertThat(lockedByCaller, is(false));
+ }
+}