You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by as...@apache.org on 2021/07/15 12:52:58 UTC
[ignite-3] branch main updated: IGNITE-15057 Implemented
LockManager to handle concurrency control in transactions - Fixes #211.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov 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 fc7b39d IGNITE-15057 Implemented LockManager to handle concurrency control in transactions - Fixes #211.
fc7b39d is described below
commit fc7b39d61f963c6f7130e14a02be9c3bcf94f301
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Thu Jul 15 15:49:59 2021 +0300
IGNITE-15057 Implemented LockManager to handle concurrency control in transactions - Fixes #211.
Signed-off-by: Alexey Scherbakov <al...@gmail.com>
---
.../org/apache/ignite/raft/jraft/util/Utils.java | 2 +-
.../jraft/util/concurrent/AdjustableSemaphore.java | 1 -
modules/transactions/pom.xml | 7 +
...{LockOrderException.java => LockException.java} | 19 +-
.../org/apache/ignite/internal/tx/LockManager.java | 32 +-
.../org/apache/ignite/internal/tx/Timestamp.java | 85 ++++
.../tx/{LockOrderException.java => Waiter.java} | 15 +-
.../ignite/internal/tx/impl/HeapLockManager.java | 335 ++++++++++++++
.../internal/tx/AbstractLockManagerTest.java | 489 +++++++++++++++++++++
...ckManagerTest.java => HeapLockManagerTest.java} | 11 +-
10 files changed, 975 insertions(+), 21 deletions(-)
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Utils.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Utils.java
index 767e30a..ae6c7a1 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Utils.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Utils.java
@@ -125,7 +125,7 @@ public final class Utils {
}
/**
- * Run a task in thread pool,returns the future object.
+ * Run a task in thread pool,returns the future object. TODO asch confusion runInThread runInExecutor IGNITE-14382
*/
public static Future<?> runInThread(final ExecutorService executor, final Runnable runnable) {
return executor.submit(runnable);
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/concurrent/AdjustableSemaphore.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/concurrent/AdjustableSemaphore.java
index ec8ee1a..db41cae 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/concurrent/AdjustableSemaphore.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/concurrent/AdjustableSemaphore.java
@@ -111,7 +111,6 @@ public final class AdjustableSemaphore implements Serializable {
}
private static final class ResizeableSemaphore extends Semaphore {
-
private static final long serialVersionUID = 1204115455517785966L;
ResizeableSemaphore(int permits) {
diff --git a/modules/transactions/pom.xml b/modules/transactions/pom.xml
index 67fc4e6..d430c3a 100644
--- a/modules/transactions/pom.xml
+++ b/modules/transactions/pom.xml
@@ -49,6 +49,13 @@
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-core</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<!-- 3rd party dependencies -->
<dependency>
<groupId>org.jetbrains</groupId>
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockOrderException.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockException.java
similarity index 63%
copy from modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockOrderException.java
copy to modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockException.java
index 6ec2068..cb8388b 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockOrderException.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockException.java
@@ -17,10 +17,23 @@
package org.apache.ignite.internal.tx;
-import org.apache.ignite.lang.IgniteInternalException;
+import org.apache.ignite.lang.IgniteInternalCheckedException;
/**
- * This exception is thrown when lock cannot be acquired due to wrong locking order.
+ * This exception is thrown when a lock cannot be acquired due to conflict.
*/
-public class LockOrderException extends IgniteInternalException {
+public class LockException extends IgniteInternalCheckedException {
+ /**
+ * @param msg The message.
+ */
+ public LockException(String msg) {
+ super(msg);
+ }
+
+ /**
+ * @param waiter Conflicting waiter.
+ */
+ public LockException(Waiter waiter) {
+ super("Failed to acquire a lock due to a conflict with: " + waiter);
+ }
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockManager.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockManager.java
index 014a892..eea7f32 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockManager.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockManager.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.tx;
+import java.util.Collection;
import java.util.concurrent.CompletableFuture;
/**
@@ -28,27 +29,42 @@ public interface LockManager {
* @param key The key.
* @param timestamp The timestamp.
* @return The future.
- * @throws LockOrderException When a lock can't be taken due to wrong ordering.
+ * @throws LockException When a lock can't be taken due to possible deadlock.
*/
- public CompletableFuture<Void> tryAcquire(Object key, long timestamp) throws LockOrderException;
+ public CompletableFuture<Void> tryAcquire(Object key, Timestamp timestamp) throws LockException;
/**
* @param key The key.
- * @return {@code True} if the lock was released.
+ * @param timestamp The timestamp.
+ * @throws LockException If the unlock operation is invalid.
*/
- public boolean tryRelease(Object key);
+ public void tryRelease(Object key, Timestamp timestamp) throws LockException;
/**
* @param key The key.
* @param timestamp The timestamp.
* @return The future.
- * @throws LockOrderException When a lock can't be taken due to wrong ordering.
+ * @throws LockException When a lock can't be taken due to possible deadlock.
+ */
+ public CompletableFuture<Void> tryAcquireShared(Object key, Timestamp timestamp) throws LockException;
+
+ /**
+ * @param key The key.
+ * @param timestamp The timestamp.
+ * @throws LockException If the unlock operation is invalid.
*/
- public CompletableFuture<Void> tryAcquireShared(Object key, long timestamp) throws LockOrderException;
+ public void tryReleaseShared(Object key, Timestamp timestamp) throws LockException;
/**
* @param key The key.
- * @return {@code True} if the lock was released.
+ * @return The waiters queue.
+ */
+ public Collection<Timestamp> queue(Object key);
+
+ /**
+ * @param key The key.
+ * @param timestamp The timestamp.
+ * @return The waiter.
*/
- public boolean tryReleaseShared(Object key);
+ public Waiter waiter(Object key, Timestamp timestamp);
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/Timestamp.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/Timestamp.java
new file mode 100644
index 0000000..3c9c13b
--- /dev/null
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/Timestamp.java
@@ -0,0 +1,85 @@
+/*
+ * 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.tx;
+
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * The timestamp.
+ */
+public class Timestamp implements Comparable<Timestamp> {
+ /** */
+ private static long localTime;
+
+ /** */
+ private static long cntr;
+
+ /** */
+ private final long timestamp;
+
+ /**
+ * @param timestamp The timestamp.
+ */
+ Timestamp(long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ /**
+ * @param other Other version.
+ * @return Comparison result.
+ */
+ @Override public int compareTo(@NotNull Timestamp other) {
+ int ret = Long.compare(timestamp >> 16 << 16, other.timestamp >> 16 << 16);
+
+ return ret != 0 ? ret : Long.compare(timestamp << 48 >> 48, other.timestamp << 48 >> 48);
+ }
+
+ @Override public boolean equals(Object o) {
+ if (!(o instanceof Timestamp)) return false;
+
+ return compareTo((Timestamp) o) == 0;
+ }
+
+ @Override public int hashCode() {
+ return (int) (timestamp ^ (timestamp >>> 32));
+ }
+
+ /**
+ * TODO https://issues.apache.org/jira/browse/IGNITE-15129
+ * @return Next timestamp (monotonically increasing).
+ */
+ public static synchronized Timestamp nextVersion() {
+ long localTimeCpy = localTime;
+
+ // Truncate nanotime to 48 bits.
+ localTime = Math.max(localTimeCpy, System.nanoTime() >> 16 << 16);
+
+ if (localTimeCpy == localTime)
+ cntr++;
+ else
+ cntr = 0;
+
+ return new Timestamp(localTime | cntr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(Timestamp.class, this);
+ }
+}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockOrderException.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/Waiter.java
similarity index 79%
rename from modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockOrderException.java
rename to modules/transactions/src/main/java/org/apache/ignite/internal/tx/Waiter.java
index 6ec2068..081f97f 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/LockOrderException.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/Waiter.java
@@ -17,10 +17,17 @@
package org.apache.ignite.internal.tx;
-import org.apache.ignite.lang.IgniteInternalException;
-
/**
- * This exception is thrown when lock cannot be acquired due to wrong locking order.
+ * The waiter.
*/
-public class LockOrderException extends IgniteInternalException {
+public interface Waiter {
+ /**
+ * @return Associated timestamp.
+ */
+ Timestamp timestamp();
+
+ /**
+ * @return {@code True} if the waiter holds the lock.
+ */
+ boolean locked();
}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java
new file mode 100644
index 0000000..5ac6d1b
--- /dev/null
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java
@@ -0,0 +1,335 @@
+/*
+ * 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.tx.impl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.tostring.IgniteToStringExclude;
+import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.LockException;
+import org.apache.ignite.internal.tx.Timestamp;
+import org.apache.ignite.internal.tx.Waiter;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * A {@link LockManager} implementation which stores lock queues in the heap.
+ *
+ * Lock waiters are placed in the queue, ordered from oldest to yongest (highest Timestamp).
+ * When a new waiter is placed in the queue, it's validated against current lock owner: if where is an owner with a
+ * higher timestamp lock request is denied.
+ */
+public class HeapLockManager implements LockManager {
+ /** */
+ private ConcurrentHashMap<Object, LockState> locks = new ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> tryAcquire(Object key, Timestamp timestamp) throws LockException {
+ return lockState(key).tryAcquire(timestamp);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void tryRelease(Object key, Timestamp timestamp) throws LockException {
+ lockState(key).tryRelease(timestamp);
+ }
+
+ /** {@inheritDoc} */
+ @Override public CompletableFuture<Void> tryAcquireShared(Object key, Timestamp timestamp) throws LockException {
+ return lockState(key).tryAcquireShared(timestamp);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void tryReleaseShared(Object key, Timestamp timestamp) throws LockException {
+ lockState(key).tryReleaseShared(timestamp);
+ }
+
+ /**
+ * @param key The key.
+ * @return The lock state for the key.
+ */
+ private @NotNull LockState lockState(Object key) {
+ return locks.computeIfAbsent(key, k -> new LockState());
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<Timestamp> queue(Object key) {
+ return lockState(key).queue();
+ }
+
+ @Override public Waiter waiter(Object key, Timestamp timestamp) {
+ return lockState(key).waiter(timestamp);
+ }
+
+ /** A lock state. */
+ private static class LockState {
+ /** Waiters. */
+ private TreeMap<Timestamp, WaiterImpl> waiters = new TreeMap<>();
+
+ /**
+ * @param timestamp The timestamp.
+ * @return The future.
+ */
+ public CompletableFuture<Void> tryAcquire(Timestamp timestamp) throws LockException {
+ WaiterImpl w = new WaiterImpl(timestamp, false);
+
+ synchronized (waiters) {
+ waiters.put(timestamp, w);
+
+ // Check lock compatibility.
+ Map.Entry<Timestamp, WaiterImpl> nextEntry = waiters.higherEntry(timestamp);
+
+ // If we have a younger waiter in a locked state, when refuse to wait for lock.
+ if (nextEntry != null && nextEntry.getValue().locked()) {
+ waiters.remove(timestamp);
+
+ throw new LockException(nextEntry.getValue());
+ }
+
+ // Lock if oldest.
+ if (waiters.firstKey() == timestamp)
+ w.lock();
+ }
+
+ // Notify outside the monitor.
+ if (w.locked())
+ w.notifyLocked();
+
+ return w.fut;
+ }
+
+ /**
+ * @param timestamp The timestamp.
+ */
+ public void tryRelease(Timestamp timestamp) throws LockException {
+ Collection<WaiterImpl> locked = new ArrayList<>();
+
+ synchronized (waiters) {
+ Map.Entry<Timestamp, WaiterImpl> first = waiters.firstEntry();
+
+ if (first == null ||
+ !first.getKey().equals(timestamp) ||
+ !first.getValue().locked() ||
+ first.getValue().isForRead())
+ throw new LockException("Not exclusively locked by " + timestamp);
+
+ waiters.pollFirstEntry();
+
+ if (waiters.isEmpty())
+ return;
+
+ // Lock next waiter(s).
+ WaiterImpl waiter = waiters.firstEntry().getValue();
+
+ if (!waiter.isForRead()) {
+ waiter.lock();
+
+ locked.add(waiter);
+ }
+ else {
+ // Grant lock to all adjasent readers.
+ for (Map.Entry<Timestamp, WaiterImpl> entry : waiters.entrySet()) {
+ if (!entry.getValue().isForRead())
+ break;
+ else {
+ entry.getValue().lock();
+
+ locked.add(entry.getValue());
+ }
+ }
+ }
+ }
+
+ // Notify outside the monitor.
+ for (WaiterImpl waiter : locked) {
+ waiter.notifyLocked();
+ }
+ }
+
+ /**
+ * @param timestamp The timestamp.
+ * @return The future.
+ */
+ public CompletableFuture<Void> tryAcquireShared(Timestamp timestamp) throws LockException {
+ WaiterImpl waiter = new WaiterImpl(timestamp, true);
+
+ // Grant a lock to the oldest waiter.
+ synchronized (waiters) {
+ waiters.put(timestamp, waiter);
+
+ // Check lock compatibility.
+ Map.Entry<Timestamp, WaiterImpl> nextEntry = waiters.higherEntry(timestamp);
+
+ if (nextEntry != null) {
+ WaiterImpl nextWaiter = nextEntry.getValue();
+
+ if (nextWaiter.locked() && !nextWaiter.isForRead()) {
+ waiters.remove(timestamp);
+
+ throw new LockException(nextWaiter);
+ }
+ }
+
+ Map.Entry<Timestamp, WaiterImpl> prevEntry = waiters.lowerEntry(timestamp);
+
+ if (prevEntry == null || prevEntry.getValue().isForRead())
+ waiter.lock();
+ }
+
+ // Notify outside the monitor.
+ if (waiter.locked())
+ waiter.notifyLocked();
+
+ return waiter.fut;
+ }
+
+ public void tryReleaseShared(Timestamp timestamp) throws LockException {
+ WaiterImpl locked = null;
+
+ synchronized (waiters) {
+ WaiterImpl waiter = waiters.get(timestamp);
+
+ if (waiter == null || !waiter.locked() || !waiter.isForRead())
+ throw new LockException("Not shared locked by " + timestamp);
+
+ Map.Entry<Timestamp, WaiterImpl> nextEntry = waiters.higherEntry(timestamp);
+
+ waiters.remove(timestamp);
+
+ // Queue is empty.
+ if (nextEntry == null)
+ return;
+
+ // Lock next exclusive waiter.
+ WaiterImpl nextWaiter = nextEntry.getValue();
+
+ if (!nextWaiter.isForRead() && nextWaiter.timestamp().equals(waiters.firstEntry().getKey())) {
+ nextWaiter.lock();
+
+ locked = nextWaiter;
+ }
+ }
+
+ if (locked != null)
+ locked.notifyLocked();
+ }
+
+ /**
+ * @return Waiters queue.
+ */
+ public Collection<Timestamp> queue() {
+ synchronized (waiters) {
+ return new ArrayList<>(waiters.keySet());
+ }
+ }
+
+ /**
+ * @param timestamp The timestamp.
+ * @return The waiter.
+ */
+ public Waiter waiter(Timestamp timestamp) {
+ synchronized (waiters) {
+ return waiters.get(timestamp);
+ }
+ }
+ }
+
+ /**
+ * The waiter implementation.
+ */
+ private static class WaiterImpl implements Comparable<WaiterImpl>, Waiter {
+ /** Locked future. */
+ @IgniteToStringExclude
+ private final CompletableFuture<Void> fut;
+
+ /** Waiter timestamp. */
+ private final Timestamp timestamp;
+
+ /** {@code True} if a read request. */
+ private boolean forRead;
+
+ /** The state. */
+ private boolean locked = false;
+
+ /**
+ * @param timestamp The timestamp.
+ * @param forRead {@code True} to request a read lock.
+ */
+ WaiterImpl(Timestamp timestamp, boolean forRead) {
+ this.fut = new CompletableFuture<>();
+ this.timestamp = timestamp;
+ this.forRead = forRead;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int compareTo(@NotNull WaiterImpl o) {
+ return timestamp.compareTo(o.timestamp);
+ }
+
+ /**
+ * Notifies a future listeners.
+ */
+ private void notifyLocked() {
+ fut.complete(null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean locked() {
+ return this.locked;
+ }
+
+ /**
+ * Grant a lock.
+ */
+ private void lock() {
+ locked = true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Timestamp timestamp() {
+ return timestamp;
+ }
+
+ /**
+ * @return {@code True} if is locked for read.
+ */
+ public boolean isForRead() {
+ return forRead;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (!(o instanceof WaiterImpl)) return false;
+
+ return compareTo((WaiterImpl) o) == 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return timestamp.hashCode();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(WaiterImpl.class, this, "isDone", fut.isDone());
+ }
+ }
+}
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractLockManagerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractLockManagerTest.java
new file mode 100644
index 0000000..df756c0
--- /dev/null
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/AbstractLockManagerTest.java
@@ -0,0 +1,489 @@
+/*
+ * 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.tx;
+
+import java.util.Collection;
+import java.util.Random;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Tests a LockManager implementation.
+ */
+public abstract class AbstractLockManagerTest extends IgniteAbstractTest {
+ private LockManager lockManager;
+
+ @BeforeEach
+ public void before() {
+ lockManager = newInstance();
+ }
+
+ protected abstract LockManager newInstance();
+
+ @Test
+ public void testSingleKeyWrite() throws LockException {
+ Timestamp ts1 = Timestamp.nextVersion();
+
+ Object key = new String("test");
+
+ CompletableFuture<Void> fut0 = lockManager.tryAcquire(key, ts1);
+
+ assertTrue(fut0.isDone());
+
+ Collection<Timestamp> queue = lockManager.queue(key);
+
+ assertTrue(queue.size() == 1 && queue.iterator().next().equals(ts1));
+
+ Waiter waiter = lockManager.waiter(key, ts1);
+
+ assertTrue(waiter.locked());
+
+ lockManager.tryRelease(key, ts1);
+ }
+
+ @Test
+ public void testSingleKeyWriteLock() throws LockException {
+ Timestamp ts1 = Timestamp.nextVersion();
+
+ Object key = new String("test");
+
+ CompletableFuture<Void> fut0 = lockManager.tryAcquire(key, ts1);
+
+ assertTrue(fut0.isDone());
+
+ Timestamp ts2 = Timestamp.nextVersion();
+
+ assertTrue(ts1.compareTo(ts2) < 0);
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts2);
+
+ assertFalse(fut1.isDone());
+
+ assertTrue(lockManager.waiter(key, ts1).locked());
+ assertFalse(lockManager.waiter(key, ts2).locked());
+
+ lockManager.tryRelease(key, ts1);
+
+ assertTrue(fut1.isDone());
+
+ assertNull(lockManager.waiter(key, ts1));
+ assertTrue(lockManager.waiter(key, ts2).locked());
+
+ lockManager.tryRelease(key, ts2);
+
+ assertNull(lockManager.waiter(key, ts1));
+ assertNull(lockManager.waiter(key, ts2));
+ }
+
+ @Test
+ public void testSingleKeyReadWriteLock() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Timestamp ts2 = Timestamp.nextVersion();
+ Timestamp ts3 = Timestamp.nextVersion();
+ assertTrue(ts0.compareTo(ts1) < 0);
+ assertTrue(ts1.compareTo(ts2) < 0);
+ assertTrue(ts2.compareTo(ts3) < 0);
+ Object key = new String("test");
+
+ CompletableFuture<Void> fut0 = lockManager.tryAcquireShared(key, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut2 = lockManager.tryAcquireShared(key, ts2);
+ assertTrue(fut2.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquireShared(key, ts1);
+ assertTrue(fut1.isDone());
+
+ CompletableFuture<Void> fut3 = lockManager.tryAcquire(key, ts3);
+ assertFalse(fut3.isDone());
+
+ assertTrue(lockManager.waiter(key, ts0).locked());
+ assertTrue(lockManager.waiter(key, ts1).locked());
+ assertTrue(lockManager.waiter(key, ts2).locked());
+ assertFalse(lockManager.waiter(key, ts3).locked());
+
+ lockManager.tryReleaseShared(key, ts2);
+
+ assertTrue(lockManager.waiter(key, ts0).locked());
+ assertTrue(lockManager.waiter(key, ts1).locked());
+ assertNull(lockManager.waiter(key, ts2));
+ assertFalse(lockManager.waiter(key, ts3).locked());
+
+ lockManager.tryReleaseShared(key, ts0);
+
+ assertNull(lockManager.waiter(key, ts0));
+ assertTrue(lockManager.waiter(key, ts1).locked());
+ assertNull(lockManager.waiter(key, ts2));
+ assertFalse(lockManager.waiter(key, ts3).locked());
+
+ lockManager.tryReleaseShared(key, ts1);
+
+ assertNull(lockManager.waiter(key, ts0));
+ assertNull(lockManager.waiter(key, ts1));
+ assertNull(lockManager.waiter(key, ts2));
+ assertTrue(lockManager.waiter(key, ts3).locked());
+ }
+
+ @Test
+ public void testSingleKeyReadWriteConflict() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Object key = new String("test");
+
+ // Lock in order
+ CompletableFuture<Void> fut0 = lockManager.tryAcquireShared(key, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts1);
+ assertFalse(fut1.isDone());
+
+ lockManager.tryReleaseShared(key, ts0);
+ assertTrue(fut1.isDone());
+
+ lockManager.tryRelease(key, ts1);
+
+ assertTrue(lockManager.queue(key).isEmpty());
+
+ // Lock not in order
+ fut0 = lockManager.tryAcquireShared(key, ts1);
+ assertTrue(fut0.isDone());
+
+ try {
+ lockManager.tryAcquire(key, ts0);
+
+ fail();
+ }
+ catch (LockException e) {
+ // Expected.
+ }
+ }
+
+ @Test
+ public void testSingleKeyReadWriteConflict2() throws LockException {
+ Timestamp[] ts = generate(3);
+ Object key = new String("test");
+
+ // Lock in order
+ CompletableFuture<Void> fut0 = lockManager.tryAcquireShared(key, ts[1]);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts[2]);
+ assertFalse(fut1.isDone());
+
+ CompletableFuture<Void> fut2 = lockManager.tryAcquireShared(key, ts[0]);
+ assertTrue(fut2.isDone());
+
+ lockManager.tryReleaseShared(key, ts[1]);
+ lockManager.tryReleaseShared(key, ts[0]);
+
+ assertTrue(fut1.isDone());
+ }
+
+ @Test
+ public void testSingleKeyReadWriteConflict3() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Timestamp ts2 = Timestamp.nextVersion();
+ Object key = new String("test");
+
+ // Lock in order
+ CompletableFuture<Void> fut0 = lockManager.tryAcquireShared(key, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts2);
+ assertFalse(fut1.isDone());
+
+ CompletableFuture<Void> fut2 = lockManager.tryAcquireShared(key, ts1);
+ assertTrue(fut2.isDone());
+
+ assertFalse(lockManager.waiter(key, ts2).locked());
+
+ lockManager.tryReleaseShared(key, ts1);
+ lockManager.tryReleaseShared(key, ts0);
+
+ assertTrue(fut1.isDone());
+ }
+
+ @Test
+ public void testSingleKeyReadWriteConflict4() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Timestamp ts2 = Timestamp.nextVersion();
+ Timestamp ts3 = Timestamp.nextVersion();
+ Object key = new String("test");
+
+ CompletableFuture<Void> fut0 = lockManager.tryAcquireShared(key, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts1);
+ assertFalse(fut1.isDone());
+
+ CompletableFuture<Void> fut2 = lockManager.tryAcquire(key, ts3);
+ assertFalse(fut2.isDone());
+
+ CompletableFuture<Void> fut3 = lockManager.tryAcquire(key, ts2);
+ assertFalse(fut3.isDone());
+ }
+
+ @Test
+ public void testSingleKeyWriteWriteConflict() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Timestamp ts2 = Timestamp.nextVersion();
+ Object key = new String("test");
+
+ // Lock in order
+ CompletableFuture<Void> fut0 = lockManager.tryAcquire(key, ts1);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts2);
+ assertFalse(fut1.isDone());
+
+ try {
+ lockManager.tryAcquire(key, ts0);
+
+ fail();
+ }
+ catch (LockException e) {
+ // Expected.
+ }
+ }
+
+ @Test
+ public void testSingleKeyWriteWriteConflict2() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Timestamp ts2 = Timestamp.nextVersion();
+ Object key = new String("test");
+
+ // Lock in order
+ CompletableFuture<Void> fut0 = lockManager.tryAcquire(key, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key, ts2);
+ assertFalse(fut1.isDone());
+
+ CompletableFuture<Void> fut2 = lockManager.tryAcquire(key, ts1);
+ assertFalse(fut2.isDone());
+ }
+
+ @Test
+ public void testSingleKeyMultithreadedRead() throws InterruptedException {
+ LongAdder rLocks = new LongAdder();
+ LongAdder wLocks = new LongAdder();
+ LongAdder fLocks = new LongAdder();
+
+ doTestSingleKeyMultithreaded(5_000, rLocks, wLocks, fLocks, 0);
+
+ assertTrue(wLocks.sum() == 0);
+ assertTrue(fLocks.sum() == 0);
+ }
+
+ @Test
+ public void testSingleKeyMultithreadedWrite() throws InterruptedException {
+ LongAdder rLocks = new LongAdder();
+ LongAdder wLocks = new LongAdder();
+ LongAdder fLocks = new LongAdder();
+
+ doTestSingleKeyMultithreaded(5_000, rLocks, wLocks, fLocks, 1);
+
+ assertTrue(rLocks.sum() == 0);
+ }
+
+ @Test
+ public void testSingleKeyMultithreadedRandom() throws InterruptedException {
+ LongAdder rLocks = new LongAdder();
+ LongAdder wLocks = new LongAdder();
+ LongAdder fLocks = new LongAdder();
+
+ doTestSingleKeyMultithreaded(5_000, rLocks, wLocks, fLocks, 2);
+ }
+
+ @Test
+ public void testDeadlock() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Object key = new String("test");
+
+ CompletableFuture<Void> fut0 = lockManager.tryAcquireShared(key, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquireShared(key, ts1);
+ assertTrue(fut1.isDone());
+
+ try {
+ lockManager.tryAcquire(key, ts0);
+
+ fail();
+ }
+ catch (LockException e) {
+ // Expected.
+ }
+ }
+
+ @Test
+ public void testDeadlock2() throws LockException {
+ Timestamp ts0 = Timestamp.nextVersion();
+ Timestamp ts1 = Timestamp.nextVersion();
+ Object key1 = new String("test");
+ Object key2 = new String("test2");
+
+ CompletableFuture<Void> fut0 = lockManager.tryAcquire(key1, ts0);
+ assertTrue(fut0.isDone());
+
+ CompletableFuture<Void> fut1 = lockManager.tryAcquire(key2, ts1);
+ assertTrue(fut1.isDone());
+
+ try {
+ lockManager.tryAcquire(key2, ts0);
+
+ fail();
+ }
+ catch (LockException e) {
+ // Expected.
+ }
+
+ }
+
+ /**
+ * @param duration The duration.
+ * @param rLocks Read lock accumulator.
+ * @param wLocks Write lock accumulator.
+ * @param fLocks Failed lock accumulator.
+ * @param mode Mode: 0 - read only, 1 - write only, 2 - mixed random.
+ * @throws InterruptedException If interrupted while waiting.
+ */
+ private void doTestSingleKeyMultithreaded(
+ long duration,
+ LongAdder rLocks,
+ LongAdder wLocks,
+ LongAdder fLocks,
+ int mode
+ ) throws InterruptedException {
+ Object key = new String("test");
+
+ Thread[] threads = new Thread[Runtime.getRuntime().availableProcessors() * 2];;
+
+ CyclicBarrier startBar = new CyclicBarrier(threads.length, () -> log.info("Before test"));
+
+ AtomicBoolean stop = new AtomicBoolean();
+
+ Random r = new Random();
+
+ for (int i = 0; i < threads.length; i++) {
+ threads[i] = new Thread(new Runnable() {
+ @Override public void run() {
+ try {
+ startBar.await();
+ }
+ catch (Exception e) {
+ fail();
+ }
+
+ while (!stop.get()) {
+ Timestamp timestamp = Timestamp.nextVersion();
+
+ if (mode == 0 ? false : mode == 1 ? true : r.nextBoolean()) {
+ try {
+ CompletableFuture<Void> fut = lockManager.tryAcquire(key, timestamp);
+ try {
+ fut.get();
+ wLocks.increment();
+ }
+ catch (Exception e) {
+ fail("Expected normal execution");
+ }
+ }
+ catch (LockException e) {
+ fLocks.increment();
+ continue;
+ }
+
+ try {
+ lockManager.tryRelease(key, timestamp);
+ }
+ catch (LockException e) {
+ fail(e.getMessage());
+ }
+ }
+ else {
+ try {
+ CompletableFuture<Void> fut = lockManager.tryAcquireShared(key, timestamp);
+ try {
+ fut.get();
+ rLocks.increment();
+ }
+ catch (Exception e) {
+ fail("Expected normal execution");
+ }
+ }
+ catch (LockException e) {
+ fLocks.increment();
+ continue;
+ }
+
+ try {
+ lockManager.tryReleaseShared(key, timestamp);
+ }
+ catch (LockException e) {
+ fail(e.getMessage());
+ }
+ }
+ }
+ }
+ });
+
+ threads[i].setName("Worker" + i);
+ threads[i].start();
+ }
+
+ Thread.sleep(duration);
+
+ stop.set(true);
+
+ for (Thread thread : threads)
+ thread.join();
+
+ log.info("After test rLocks={} wLocks={} fLocks={}", rLocks.sum(), wLocks.sum(), fLocks.sum());
+
+ assertTrue(lockManager.queue(key).isEmpty());
+ }
+
+ private Timestamp[] generate(int num) {
+ Timestamp[] tmp = new Timestamp[num];
+
+ for (int i = 0; i < tmp.length; i++)
+ tmp[i] = Timestamp.nextVersion();
+
+ for (int i = 1; i < tmp.length; i++)
+ assertTrue(tmp[i - 1].compareTo(tmp[i]) < 0);
+
+ return tmp;
+ }
+}
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/LockManagerTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/HeapLockManagerTest.java
similarity index 78%
rename from modules/transactions/src/test/java/org/apache/ignite/internal/tx/LockManagerTest.java
rename to modules/transactions/src/test/java/org/apache/ignite/internal/tx/HeapLockManagerTest.java
index 4ab4c13..0e2be7d 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/LockManagerTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/HeapLockManagerTest.java
@@ -17,8 +17,11 @@
package org.apache.ignite.internal.tx;
-/**
- * Tests a LockManager implementation.
- */
-public class LockManagerTest {
+import org.apache.ignite.internal.tx.impl.HeapLockManager;
+
+/** */
+public class HeapLockManagerTest extends AbstractLockManagerTest {
+ @Override protected LockManager newInstance() {
+ return new HeapLockManager();
+ }
}