You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by yz...@apache.org on 2016/04/27 17:31:41 UTC
[1/3] ignite git commit: IGNITE-642 Implement IgniteReentrantLock
data structure
Repository: ignite
Updated Branches:
refs/heads/master 222637733 -> f9bd9a338
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteLockAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteLockAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteLockAbstractSelfTest.java
new file mode 100644
index 0000000..8fb9049
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteLockAbstractSelfTest.java
@@ -0,0 +1,1629 @@
+/*
+ * 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.processors.cache.datastructures;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteCondition;
+import org.apache.ignite.IgniteCountDownLatch;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteInterruptedException;
+import org.apache.ignite.IgniteLock;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.PA;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Rule;
+import org.junit.rules.ExpectedException;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static org.apache.ignite.cache.CacheMode.LOCAL;
+
+/**
+ * Cache reentrant lock self test.
+ */
+public abstract class IgniteLockAbstractSelfTest extends IgniteAtomicsAbstractTest
+ implements Externalizable {
+ /** */
+ private static final int NODES_CNT = 4;
+
+ /** */
+ protected static final int THREADS_CNT = 5;
+
+ /** */
+ private static final Random RND = new Random();
+
+ /** */
+ @Rule
+ public final ExpectedException exception = ExpectedException.none();
+
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return NODES_CNT;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLock() throws Exception {
+ checkReentrantLock(false);
+
+ checkReentrantLock(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testFailover() throws Exception {
+ if (atomicsCacheMode() == LOCAL)
+ return;
+
+ checkFailover(true, false);
+
+ checkFailover(false, false);
+
+ checkFailover(true, true);
+
+ checkFailover(false, true);
+ }
+
+ /**
+ * @param failoverSafe Failover safe flag.
+ * @throws Exception
+ */
+ private void checkFailover(final boolean failoverSafe, final boolean fair) throws Exception {
+ IgniteEx g = startGrid(NODES_CNT + 1);
+
+ // For vars locality.
+ {
+ // Ensure not exists.
+ assert g.reentrantLock("lock", failoverSafe, fair, false) == null;
+
+ IgniteLock lock = g.reentrantLock("lock", failoverSafe, fair, true);
+
+ lock.lock();
+
+ assert lock.tryLock();
+
+ assertEquals(2, lock.getHoldCount());
+ }
+
+ Ignite g0 = grid(0);
+
+ final IgniteLock lock0 = g0.reentrantLock("lock", false, fair, false);
+
+ assert !lock0.tryLock();
+
+ assertEquals(0, lock0.getHoldCount());
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ try {
+ lock0.lock();
+
+ info("Acquired in separate thread.");
+
+ // Lock is acquired silently only in failoverSafe mode.
+ assertTrue(failoverSafe);
+
+ lock0.unlock();
+
+ info("Released lock in separate thread.");
+ }
+ catch (IgniteException e) {
+ if (!failoverSafe)
+ info("Ignored expected exception: " + e);
+ else
+ throw e;
+ }
+ return null;
+ }
+ },
+ 1);
+
+ Thread.sleep(100);
+
+ g.close();
+
+ fut.get(500);
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkReentrantLock(final boolean fair) throws Exception {
+ // Test API.
+ checkLock(fair);
+
+ checkFailoverSafe(fair);
+
+ // Test main functionality.
+ IgniteLock lock1 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertFalse(lock1.isLocked());
+
+ lock1.lock();
+
+ IgniteCompute comp = grid(0).compute().withAsync();
+
+ comp.call(new IgniteCallable<Object>() {
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ @LoggerResource
+ private IgniteLogger log;
+
+ @Nullable @Override public Object call() throws Exception {
+ // Test reentrant lock in multiple threads on each node.
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(
+ new Callable<Object>() {
+ @Nullable @Override public Object call() throws Exception {
+ IgniteLock lock = ignite.reentrantLock("lock", true, fair, true);
+
+ assert lock != null;
+
+ log.info("Thread is going to wait on reentrant lock: " + Thread.currentThread().getName());
+
+ assert lock.tryLock(1, MINUTES);
+
+ log.info("Thread is again runnable: " + Thread.currentThread().getName());
+
+ lock.unlock();
+
+ return null;
+ }
+ },
+ 5,
+ "test-thread"
+ );
+
+ fut.get();
+
+ return null;
+ }
+ });
+
+ IgniteFuture<Object> fut = comp.future();
+
+ Thread.sleep(3000);
+
+ assert lock1.isHeldByCurrentThread();
+
+ assert lock1.getHoldCount() == 1;
+
+ lock1.lock();
+
+ assert lock1.isHeldByCurrentThread();
+
+ assert lock1.getHoldCount() == 2;
+
+ lock1.unlock();
+
+ lock1.unlock();
+
+ // Ensure there are no hangs.
+ fut.get();
+
+ // Test operations on removed lock.
+ lock1.close();
+
+ checkRemovedReentrantLock(lock1);
+ }
+
+ /**
+ * @param lock IgniteLock.
+ * @throws Exception If failed.
+ */
+ protected void checkRemovedReentrantLock(final IgniteLock lock) throws Exception {
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return lock.removed();
+ }
+ }, 5000);
+
+ assert lock.removed();
+ }
+
+ /**
+ * This method only checks if parameter of new reentrant lock is initialized properly.
+ * For tests considering failure recovery see @GridCachePartitionedNodeFailureSelfTest.
+ *
+ * @throws Exception Exception.
+ */
+ private void checkFailoverSafe(final boolean fair) throws Exception {
+ // Checks only if reentrant lock is initialized properly
+ IgniteLock lock = createReentrantLock("rmv", true, fair);
+
+ assert lock.isFailoverSafe();
+
+ removeReentrantLock("rmv", fair);
+
+ IgniteLock lock1 = createReentrantLock("rmv1", false, fair);
+
+ assert !lock1.isFailoverSafe();
+
+ removeReentrantLock("rmv1", fair);
+ }
+
+ /**
+ * @throws Exception Exception.
+ */
+ private void checkLock(final boolean fair) throws Exception {
+ // Check only 'false' cases here. Successful lock is tested over the grid.
+ final IgniteLock lock = createReentrantLock("acquire", false, fair);
+
+ lock.lock();
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ assertNotNull(lock);
+
+ assert !lock.tryLock();
+
+ assert !lock.tryLock(10, MICROSECONDS);
+
+ return null;
+ }
+ });
+
+ fut.get();
+
+ lock.unlock();
+
+ removeReentrantLock("acquire", fair);
+ }
+
+ /**
+ * @param lockName Reentrant lock name.
+ * @param failoverSafe FailoverSafe flag.
+ * @param fair Fairness flag.
+ * @return New distributed reentrant lock.
+ * @throws Exception If failed.
+ */
+ private IgniteLock createReentrantLock(String lockName, boolean failoverSafe, boolean fair)
+ throws Exception {
+ IgniteLock lock = grid(RND.nextInt(NODES_CNT)).reentrantLock(lockName, failoverSafe, fair, true);
+
+ // Test initialization.
+ assert lockName.equals(lock.name());
+ assert !lock.isLocked();
+ assert lock.isFailoverSafe() == failoverSafe;
+ assert lock.isFair() == fair;
+
+ return lock;
+ }
+
+ /**
+ * @param lockName Reentrant lock name.
+ * @throws Exception If failed.
+ */
+ private void removeReentrantLock(String lockName, final boolean fair)
+ throws Exception {
+ IgniteLock lock = grid(RND.nextInt(NODES_CNT)).reentrantLock(lockName, false, fair, true);
+
+ assert lock != null;
+
+ // Remove lock on random node.
+ IgniteLock lock0 = grid(RND.nextInt(NODES_CNT)).reentrantLock(lockName, false, fair, true);
+
+ assertNotNull(lock0);
+
+ lock0.close();
+
+ // Ensure reentrant lock is removed on all nodes.
+ for (Ignite g : G.allGrids())
+ assertNull(((IgniteKernal)g).context().dataStructures().reentrantLock(lockName, false, fair, false));
+
+ checkRemovedReentrantLock(lock);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockSerialization() throws Exception {
+ final IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ info("Lock created: " + lock);
+
+ lock.isFailoverSafe();
+ lock.isFair();
+
+ grid(ThreadLocalRandom.current().nextInt(G.allGrids().size())).compute().broadcast(new IgniteCallable<Object>() {
+ @Nullable @Override public Object call() throws Exception {
+ Thread.sleep(1000);
+
+ lock.lock();
+
+ try {
+ info("Inside lock: " + lock.getHoldCount());
+ }
+ finally {
+ lock.unlock();
+ }
+
+ return null;
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testInitialization() throws Exception {
+ // Test #name() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertEquals("lock", lock.name());
+
+ lock.close();
+ }
+
+ // Test #isFailoverSafe() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ info("Lock created: " + lock);
+
+ assertTrue(lock.isFailoverSafe());
+
+ lock.close();
+ }
+
+ // Test #isFair() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertTrue(lock.isFair());
+
+ lock.close();
+ }
+
+ // Test #isBroken() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertFalse(lock.isBroken());
+
+ lock.close();
+ }
+
+ // Test #getOrCreateCondition(String ) method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertNotNull(lock.getOrCreateCondition("condition"));
+
+ lock.close();
+ }
+
+ // Test #getHoldCount() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertEquals(0, lock.getHoldCount());
+
+ lock.close();
+ }
+
+ // Test #isHeldByCurrentThread() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertFalse(lock.isHeldByCurrentThread());
+
+ lock.close();
+ }
+
+ // Test #isLocked() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertFalse(lock.isLocked());
+
+ lock.close();
+ }
+
+ // Test #hasQueuedThreads() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertFalse(lock.hasQueuedThreads());
+
+ lock.close();
+ }
+
+ // Test #hasQueuedThread(Thread ) method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertFalse(lock.hasQueuedThread(Thread.currentThread()));
+
+ lock.close();
+ }
+
+ // Test #hasWaiters(IgniteCondition ) method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ try {
+ IgniteCondition cond = grid(0).reentrantLock("lock2", true, true, true).getOrCreateCondition("cond");
+
+ lock.hasWaiters(cond);
+
+ fail("Condition not associated with this lock passed as argument.");
+ }
+ catch (IllegalArgumentException e) {
+ info("IllegalArgumentException thrown as it should be.");
+ }
+
+ try {
+ IgniteCondition cond = lock.getOrCreateCondition("condition");
+
+ lock.hasWaiters(cond);
+
+ fail("This method should throw exception when lock is not held.");
+ }
+ catch (IllegalMonitorStateException e) {
+ info("IllegalMonitorStateException thrown as it should be.");
+ }
+
+ lock.close();
+ }
+
+ // Test #getWaitQueueLength(IgniteCondition ) method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ try {
+ IgniteCondition cond = grid(0).reentrantLock("lock2", true, true, true).getOrCreateCondition("cond");
+
+ lock.getWaitQueueLength(cond);
+
+ fail("Condition not associated with this lock passed as argument.");
+ }
+ catch (IllegalArgumentException e) {
+ info("IllegalArgumentException thrown as it should be.");
+ }
+
+ try {
+ IgniteCondition cond = lock.getOrCreateCondition("condition");
+
+ lock.getWaitQueueLength(cond);
+
+ fail("This method should throw exception when lock is not held.");
+ }
+ catch (IllegalMonitorStateException e) {
+ info("IllegalMonitorStateException thrown as it should be.");
+ }
+
+ lock.close();
+ }
+
+ // Test #lock() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ lock.lock();
+
+ lock.unlock();
+
+ lock.close();
+ }
+
+ // Test #lockInterruptibly() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ lock.lockInterruptibly();
+
+ lock.unlock();
+
+ lock.close();
+ }
+
+ // Test #tryLock() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ boolean success = lock.tryLock();
+
+ assertTrue(success);
+
+ lock.unlock();
+
+ lock.close();
+ }
+
+ // Test #tryLock(long, TimeUnit) method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ boolean success = lock.tryLock(1, MILLISECONDS);
+
+ assertTrue(success);
+
+ lock.unlock();
+
+ lock.close();
+ }
+
+ // Test #unlock() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ try {
+ lock.unlock();
+
+ fail("This method should throw exception when lock is not held.");
+ }
+ catch (IllegalMonitorStateException e) {
+ info("IllegalMonitorStateException thrown as it should be.");
+ }
+
+ lock.close();
+ }
+
+ // Test #removed() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ assertFalse(lock.removed());
+
+ lock.close();
+
+ assertTrue(lock.removed());
+ }
+
+ // Test #close() method.
+ {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, true, true);
+
+ lock.close();
+
+ assertTrue(lock.removed());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockMultinode1() throws Exception {
+ testReentrantLockMultinode1(false);
+
+ testReentrantLockMultinode1(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testReentrantLockMultinode1(final boolean fair) throws Exception {
+ if (gridCount() == 1)
+ return;
+
+ IgniteLock lock = grid(0).reentrantLock("s1", true, fair, true);
+
+ List<IgniteInternalFuture<?>> futs = new ArrayList<>();
+
+ for (int i = 0; i < gridCount(); i++) {
+ final Ignite ignite = grid(i);
+
+ futs.add(GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ IgniteLock lock = ignite.reentrantLock("s1", true, fair, false);
+
+ assertNotNull(lock);
+
+ IgniteCondition cond1 = lock.getOrCreateCondition("c1");
+
+ IgniteCondition cond2 = lock.getOrCreateCondition("c2");
+
+ try {
+ boolean wait = lock.tryLock(30_000, MILLISECONDS);
+
+ assertTrue(wait);
+
+ cond2.signal();
+
+ cond1.await();
+ }
+ finally {
+ lock.unlock();
+ }
+
+ return null;
+ }
+ }));
+ }
+
+ boolean done = false;
+
+ while(!done) {
+ done = true;
+
+ for (IgniteInternalFuture<?> fut : futs){
+ if(!fut.isDone())
+ done = false;
+ }
+
+ try{
+ lock.lock();
+
+ lock.getOrCreateCondition("c1").signal();
+
+ lock.getOrCreateCondition("c2").await(10,MILLISECONDS);
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+
+ for (IgniteInternalFuture<?> fut : futs)
+ fut.get(30_000);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockInterruptibly() throws Exception {
+ testLockInterruptibly(false);
+
+ testLockInterruptibly(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testLockInterruptibly(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 2;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ lock0.lock();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ boolean isInterrupted = false;
+
+ try {
+ lock0.lockInterruptibly();
+ }
+ catch (IgniteInterruptedException e) {
+ assertFalse(Thread.currentThread().isInterrupted());
+
+ isInterrupted = true;
+ }
+ finally {
+ // Assert that thread was interrupted.
+ assertTrue(isInterrupted);
+
+ // Assert that locked is still owned by main thread.
+ assertTrue(lock0.isLocked());
+
+ // Assert that this thread doesn't own the lock.
+ assertFalse(lock0.isHeldByCurrentThread());
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ // Wait for all threads to attempt to acquire lock.
+ while (startedThreads.size() != totalThreads) {
+ Thread.sleep(1000);
+ }
+
+ for (Thread t : startedThreads)
+ t.interrupt();
+
+ fut.get();
+
+ lock0.unlock();
+
+ assertFalse(lock0.isLocked());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockInterruptiblyMultinode() throws Exception {
+ testLockInterruptiblyMultinode(false);
+
+ testLockInterruptiblyMultinode(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testLockInterruptiblyMultinode(final boolean fair) throws Exception {
+ if (gridCount() == 1)
+ return;
+
+ // Initialize reentrant lock.
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ lock0.lock();
+
+ // Number of threads, one per node.
+ final int threadCount = gridCount();
+
+ final AtomicLong threadCounter = new AtomicLong(0);
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ final int localNodeId = (int)threadCounter.getAndIncrement();
+
+ final Ignite grid = grid(localNodeId);
+
+ IgniteClosure<Ignite, Void> closure = new IgniteClosure<Ignite, Void>() {
+ @Override public Void apply(Ignite ignite) {
+ final IgniteLock l = ignite.reentrantLock("lock", true, true, true);
+
+ final AtomicReference<Thread> thread = new AtomicReference<>();
+
+ final AtomicBoolean done = new AtomicBoolean(false);
+
+ final AtomicBoolean exceptionThrown = new AtomicBoolean(false);
+
+ final IgniteCountDownLatch latch = ignite.countDownLatch("latch", threadCount, false, true);
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try{
+ thread.set(Thread.currentThread());
+
+ l.lockInterruptibly();
+ }
+ catch(IgniteInterruptedException e){
+ exceptionThrown.set(true);
+ }
+ finally {
+ done.set(true);
+ }
+
+ return null;
+ }
+ });
+
+ // Wait until l.lock() has been called.
+ while(!l.hasQueuedThreads()){
+ // No-op.
+ }
+
+ latch.countDown();
+
+ latch.await();
+
+ thread.get().interrupt();
+
+ while(!done.get()){
+ // No-op.
+ }
+
+ try {
+ fut.get();
+ }
+ catch (IgniteCheckedException e) {
+ fail(e.getMessage());
+
+ throw new RuntimeException(e);
+ }
+
+ assertTrue(exceptionThrown.get());
+
+ return null;
+ }
+ };
+
+ closure.apply(grid);
+
+ return null;
+ }
+ }, threadCount);
+
+ fut.get();
+
+ lock0.unlock();
+
+ info("Checking if interrupted threads are removed from global waiting queue...");
+
+ // Check if interrupted threads are removed from global waiting queue.
+ boolean locked = lock0.tryLock(1000, MILLISECONDS);
+
+ info("Interrupted threads successfully removed from global waiting queue. ");
+
+ assertTrue(locked);
+
+ lock0.unlock();
+
+ assertFalse(lock0.isLocked());
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLock() throws Exception {
+ testLock(false);
+
+ testLock(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testLock(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 2;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ lock0.lock();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ boolean isInterrupted = false;
+
+ try {
+ lock0.lock();
+ }
+ catch (IgniteInterruptedException e) {
+ isInterrupted = true;
+
+ fail("Lock() method is uninterruptible.");
+ }
+ finally {
+ // Assert that thread was not interrupted.
+ assertFalse(isInterrupted);
+
+ // Assert that interrupted flag is set and clear it in order to call unlock().
+ assertTrue(Thread.interrupted());
+
+ // Assert that lock is still owned by this thread.
+ assertTrue(lock0.isLocked());
+
+ // Assert that this thread does own the lock.
+ assertTrue(lock0.isHeldByCurrentThread());
+
+ // Release lock.
+ lock0.unlock();
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ // Wait for all threads to attempt to acquire lock.
+ while (startedThreads.size() != totalThreads) {
+ Thread.sleep(500);
+ }
+
+ for (Thread t : startedThreads)
+ t.interrupt();
+
+ lock0.unlock();
+
+ fut.get();
+
+ assertFalse(lock0.isLocked());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTryLock() throws Exception {
+ testTryLock(false);
+
+ testTryLock(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testTryLock(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 2;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ lock0.lock();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ boolean isInterrupted = false;
+
+ boolean locked = false;
+
+ try {
+ locked = lock0.tryLock();
+ }
+ catch (IgniteInterruptedException e) {
+ isInterrupted = true;
+
+ fail("tryLock() method is uninterruptible.");
+ }
+ finally {
+ // Assert that thread was not interrupted.
+ assertFalse(isInterrupted);
+
+ // Assert that lock is locked.
+ assertTrue(lock0.isLocked());
+
+ // Assert that this thread does own the lock.
+ assertEquals(locked, lock0.isHeldByCurrentThread());
+
+ // Release lock.
+ if (locked)
+ lock0.unlock();
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ // Wait for all threads to attempt to acquire lock.
+ while (startedThreads.size() != totalThreads) {
+ Thread.sleep(500);
+ }
+
+ for (Thread t : startedThreads)
+ t.interrupt();
+
+ fut.get();
+
+ lock0.unlock();
+
+ assertFalse(lock0.isLocked());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTryLockTimed() throws Exception {
+ testTryLockTimed(false);
+
+ testTryLockTimed(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testTryLockTimed(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 2;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ lock0.lock();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ boolean isInterrupted = false;
+
+ boolean locked = false;
+
+ try {
+ locked = lock0.tryLock(100, TimeUnit.MILLISECONDS);
+ }
+ catch (IgniteInterruptedException e) {
+ isInterrupted = true;
+ }
+ finally {
+ // Assert that thread was not interrupted.
+ assertFalse(isInterrupted);
+
+ // Assert that tryLock returned false.
+ assertFalse(locked);
+
+ // Assert that lock is still owned by main thread.
+ assertTrue(lock0.isLocked());
+
+ // Assert that this thread doesn't own the lock.
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ // Release lock.
+ if (locked)
+ lock0.unlock();
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ fut.get();
+
+ lock0.unlock();
+
+ assertFalse(lock0.isLocked());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testConditionAwaitUninterruptibly() throws Exception {
+ testConditionAwaitUninterruptibly(false);
+
+ testConditionAwaitUninterruptibly(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testConditionAwaitUninterruptibly(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 2;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ boolean isInterrupted = false;
+
+ lock0.lock();
+
+ IgniteCondition cond = lock0.getOrCreateCondition("cond");
+
+ try {
+ cond.awaitUninterruptibly();
+ }
+ catch (IgniteInterruptedException e) {
+ isInterrupted = true;
+ }
+ finally {
+ // Assert that thread was not interrupted.
+ assertFalse(isInterrupted);
+
+ // Assert that lock is still locked.
+ assertTrue(lock0.isLocked());
+
+ // Assert that this thread does own the lock.
+ assertTrue(lock0.isHeldByCurrentThread());
+
+ // Clear interrupt flag.
+ assertTrue(Thread.interrupted());
+
+ // Release lock.
+ if (lock0.isHeldByCurrentThread())
+ lock0.unlock();
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ // Wait for all threads to attempt to acquire lock.
+ while (startedThreads.size() != totalThreads) {
+ Thread.sleep(500);
+ }
+
+ lock0.lock();
+
+ for (Thread t : startedThreads) {
+ t.interrupt();
+
+ lock0.getOrCreateCondition("cond").signal();
+ }
+
+ lock0.unlock();
+
+ fut.get();
+
+ assertFalse(lock0.isLocked());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testConditionInterruptAwait() throws Exception {
+ testConditionInterruptAwait(false);
+
+ testConditionInterruptAwait(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testConditionInterruptAwait(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 2;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ boolean isInterrupted = false;
+
+ lock0.lock();
+
+ IgniteCondition cond = lock0.getOrCreateCondition("cond");
+
+ try {
+ cond.await();
+ }
+ catch (IgniteInterruptedException e) {
+ isInterrupted = true;
+ }
+ finally {
+ // Assert that thread was interrupted.
+ assertTrue(isInterrupted);
+
+ // Assert that lock is still locked.
+ assertTrue(lock0.isLocked());
+
+ // Assert that this thread does own the lock.
+ assertTrue(lock0.isHeldByCurrentThread());
+
+ // Release lock.
+ if (lock0.isHeldByCurrentThread())
+ lock0.unlock();
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ // Wait for all threads to attempt to acquire lock.
+ while (startedThreads.size() != totalThreads) {
+ Thread.sleep(500);
+ }
+
+ for (Thread t : startedThreads)
+ t.interrupt();
+
+ fut.get();
+
+ assertFalse(lock0.isLocked());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testHasQueuedThreads() throws Exception {
+ testHasQueuedThreads(false);
+
+ testHasQueuedThreads(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testHasQueuedThreads(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 5;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ final Set<Thread> finishedThreads = new GridConcurrentHashSet<>();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ startedThreads.add(Thread.currentThread());
+
+ lock0.lock();
+
+ // Wait until every thread tries to lock.
+ do {
+ Thread.sleep(1000);
+ }
+ while (startedThreads.size() != totalThreads);
+
+ try {
+ info("Acquired in separate thread. ");
+
+ assertTrue(lock0.isHeldByCurrentThread());
+
+ assertFalse(lock0.hasQueuedThread(Thread.currentThread()));
+
+ finishedThreads.add(Thread.currentThread());
+
+ if (startedThreads.size() != finishedThreads.size()) {
+ assertTrue(lock0.hasQueuedThreads());
+ }
+
+ for (Thread t : startedThreads) {
+ assertTrue(lock0.hasQueuedThread(t) != finishedThreads.contains(t));
+ }
+ }
+ finally {
+ lock0.unlock();
+
+ assertFalse(lock0.isHeldByCurrentThread());
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ fut.get();
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testHasConditionQueuedThreads() throws Exception {
+ testHasConditionQueuedThreads(false);
+
+ testHasConditionQueuedThreads(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void testHasConditionQueuedThreads(final boolean fair) throws Exception {
+ final IgniteLock lock0 = grid(0).reentrantLock("lock", true, fair, true);
+
+ assertEquals(0, lock0.getHoldCount());
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ final int totalThreads = 5;
+
+ final Set<Thread> startedThreads = new GridConcurrentHashSet<>();
+
+ final Set<Thread> finishedThreads = new GridConcurrentHashSet<>();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ assertFalse(lock0.isHeldByCurrentThread());
+
+ IgniteCondition cond = lock0.getOrCreateCondition("cond");
+
+ lock0.lock();
+
+ startedThreads.add(Thread.currentThread());
+
+ // Wait until every thread tries to lock.
+ do {
+ cond.await();
+
+ Thread.sleep(1000);
+ }
+ while (startedThreads.size() != totalThreads);
+
+ try {
+ info("Acquired in separate thread. Number of threads waiting on condition: "
+ + lock0.getWaitQueueLength(cond));
+
+ assertTrue(lock0.isHeldByCurrentThread());
+
+ assertFalse(lock0.hasQueuedThread(Thread.currentThread()));
+
+ finishedThreads.add(Thread.currentThread());
+
+ if (startedThreads.size() != finishedThreads.size()) {
+ assertTrue(lock0.hasWaiters(cond));
+ }
+
+ for (Thread t : startedThreads) {
+ if (!finishedThreads.contains(t))
+ assertTrue(lock0.hasWaiters(cond));
+ }
+
+ assertTrue(lock0.getWaitQueueLength(cond) == (startedThreads.size() - finishedThreads.size()));
+ }
+ finally {
+ cond.signal();
+
+ lock0.unlock();
+
+ assertFalse(lock0.isHeldByCurrentThread());
+ }
+
+ return null;
+ }
+ }, totalThreads);
+
+ IgniteCondition cond = lock0.getOrCreateCondition("cond");
+
+ lock0.lock();
+
+ try {
+ // Wait until all threads are waiting on condition.
+ while (lock0.getWaitQueueLength(cond) != totalThreads) {
+ lock0.unlock();
+
+ Thread.sleep(1000);
+
+ lock0.lock();
+ }
+
+ // Signal once to get things started.
+ cond.signal();
+ }
+ finally {
+ lock0.unlock();
+ }
+
+ fut.get();
+
+ assertFalse(lock0.hasQueuedThreads());
+
+ for (Thread t : startedThreads)
+ assertFalse(lock0.hasQueuedThread(t));
+
+ lock0.close();
+ }
+
+ /**
+ * Tests if lock is evenly acquired among nodes when fair flag is set on.
+ * Since exact ordering of lock acquisitions cannot be guaranteed because it also depends
+ * on the OS thread scheduling, certain deviation from uniform distribution is tolerated.
+ * @throws Exception If failed.
+ */
+ public void testFairness() throws Exception {
+ if (gridCount() == 1)
+ return;
+
+ // Total number of ops.
+ final long opsCount = 10000;
+
+ // Allowed deviation from uniform distribution.
+ final double tolerance = 0.05;
+
+ // Shared counter.
+ final String OPS_COUNTER = "ops_counter";
+
+ // Number of threads, one per node.
+ final int threadCount = gridCount();
+
+ final AtomicLong threadCounter = new AtomicLong(0);
+
+ Ignite ignite = startGrid(gridCount());
+
+ // Initialize reentrant lock.
+ IgniteLock l = ignite.reentrantLock("lock", true, true, true);
+
+ // Initialize OPS_COUNTER.
+ ignite.getOrCreateCache(OPS_COUNTER).put(OPS_COUNTER, (long)0);
+
+ final Map<Integer, Long> counts = new ConcurrentHashMap<>();
+
+ IgniteInternalFuture<?> fut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ final int localNodeId = (int)threadCounter.getAndIncrement();
+
+ final Ignite grid = grid(localNodeId);
+
+ IgniteClosure<Ignite, Long> closure = new IgniteClosure<Ignite, Long>() {
+ @Override public Long apply(Ignite ignite) {
+ IgniteLock l = ignite.reentrantLock("lock", true, true, true);
+
+ long localCount = 0;
+
+ IgniteCountDownLatch latch = ignite.countDownLatch("latch", threadCount, false, true);
+
+ latch.countDown();
+
+ latch.await();
+
+ while(true){
+ l.lock();
+
+ try {
+ long opsCounter = (long) ignite.getOrCreateCache(OPS_COUNTER).get(OPS_COUNTER);
+
+ if(opsCounter == opsCount)
+ break;
+
+ ignite.getOrCreateCache(OPS_COUNTER).put(OPS_COUNTER, ++opsCounter);
+
+ localCount++;
+
+ if(localCount > 1000){
+ assertTrue(localCount < (1 + tolerance) * opsCounter / threadCount);
+
+ assertTrue(localCount > (1 - tolerance) * opsCounter / threadCount);
+ }
+
+ if(localCount % 100 == 0) {
+ info("Node [id=" +ignite.cluster().localNode().id() + "] acquired " +
+ localCount + " times. " + "Total ops count: " +
+ opsCounter + "/" + opsCount +"]");
+ }
+ }
+ finally {
+ l.unlock();
+ }
+ }
+
+ return localCount;
+ }
+ };
+
+ long localCount = closure.apply(grid);
+
+ counts.put(localNodeId, localCount);
+
+ return null;
+ }
+ }, threadCount);
+
+ fut.get();
+
+ long totalSum = 0;
+
+ for(int i=0; i<gridCount(); i++){
+
+ totalSum += counts.get(i);
+
+ info("Node " + grid(i).localNode().id() + " acquired the lock " + counts.get(i) + " times. ");
+ }
+
+ assertEquals(totalSum, opsCount);
+
+ l.close();
+
+ ignite.close();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ // No-op.
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteSemaphoreAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteSemaphoreAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteSemaphoreAbstractSelfTest.java
index e60aed3..200e276 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteSemaphoreAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteSemaphoreAbstractSelfTest.java
@@ -77,6 +77,7 @@ public abstract class IgniteSemaphoreAbstractSelfTest extends IgniteAtomicsAbstr
*/
public void testSemaphore() throws Exception {
checkSemaphore();
+ checkSemaphoreSerialization();
}
/**
@@ -231,6 +232,36 @@ public abstract class IgniteSemaphoreAbstractSelfTest extends IgniteAtomicsAbstr
}
/**
+ * @throws Exception If failed.
+ */
+ private void checkSemaphoreSerialization() throws Exception {
+ final IgniteSemaphore sem = grid(0).semaphore("semaphore", -gridCount() + 1, true, true);
+
+ assertEquals(-gridCount() + 1, sem.availablePermits());
+
+ grid(0).compute().broadcast(new IgniteCallable<Object>() {
+ @Nullable @Override public Object call() throws Exception {
+ sem.release();
+
+ return null;
+ }
+ });
+
+ assert sem.availablePermits() == 1;
+
+ sem.acquire();
+
+ assert sem.availablePermits() == 0;
+
+ sem.release();
+
+ // Test operations on removed semaphore.
+ sem.close();
+
+ checkRemovedSemaphore(sem);
+ }
+
+ /**
* @param semaphore Semaphore.
* @throws Exception If failed.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalLockSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalLockSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalLockSelfTest.java
new file mode 100644
index 0000000..7e1a11c
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalLockSelfTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.datastructures.local;
+
+import java.util.concurrent.Callable;
+import org.apache.ignite.IgniteLock;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.datastructures.IgniteLockAbstractSelfTest;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.Nullable;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static org.apache.ignite.cache.CacheMode.LOCAL;
+
+/**
+ *
+ */
+public class IgniteLocalLockSelfTest extends IgniteLockAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode atomicsCacheMode() {
+ return LOCAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void testReentrantLock() throws Exception {
+ // Test main functionality.
+ IgniteLock lock = grid(0).reentrantLock("lock", true, false, true);
+
+ assertNotNull(lock);
+
+ assertEquals(0, lock.getHoldCount());
+
+ lock.lock();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(
+ new Callable<Object>() {
+ @Nullable @Override public Object call() throws Exception {
+ IgniteLock lock = grid(0).reentrantLock("lock", true, false, true);
+
+ assert lock != null;
+
+ info("Thread is going to wait on lock: " + Thread.currentThread().getName());
+
+ assert lock.tryLock(1, MINUTES);
+
+ info("Thread is again runnable: " + Thread.currentThread().getName());
+
+ lock.unlock();
+
+ return null;
+ }
+ },
+ THREADS_CNT,
+ "test-thread"
+ );
+
+ Thread.sleep(3000);
+
+ assert lock.isLocked();
+
+ assert lock.getHoldCount() == 1;
+
+ lock.lock();
+
+ assert lock.isLocked();
+
+ assert lock.getHoldCount() == 2;
+
+ lock.unlock();
+
+ assert lock.isLocked();
+
+ assert lock.getHoldCount() == 1;
+
+ lock.unlock();
+
+ // Ensure there are no hangs.
+ fut.get();
+
+ // Test operations on removed lock.
+ IgniteLock lock0 = grid(0).reentrantLock("lock", true, false, false);
+
+ assertNotNull(lock0);
+
+ lock0.close();
+
+ checkRemovedReentrantLock(lock0);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedLockSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedLockSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedLockSelfTest.java
new file mode 100644
index 0000000..787f1e3
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedLockSelfTest.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.datastructures.partitioned;
+
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.internal.processors.cache.datastructures.IgniteLockAbstractSelfTest;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+
+/**
+ *
+ */
+public class IgnitePartitionedLockSelfTest extends IgniteLockAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode atomicsCacheMode() {
+ return PARTITIONED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedLockSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedLockSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedLockSelfTest.java
new file mode 100644
index 0000000..00bb0fa
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedLockSelfTest.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.datastructures.replicated;
+
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.internal.processors.cache.datastructures.IgniteLockAbstractSelfTest;
+
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+
+/**
+ *
+ */
+public class IgniteReplicatedLockSelfTest extends IgniteLockAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode atomicsCacheMode() {
+ return REPLICATED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheLockAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheLockAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheLockAbstractTest.java
index ae591bd..43acc41 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheLockAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheLockAbstractTest.java
@@ -573,4 +573,4 @@ public abstract class GridCacheLockAbstractTest extends GridCommonAbstractTest {
}
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/loadtests/cache/GridCacheDataStructuresLoadTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/cache/GridCacheDataStructuresLoadTest.java b/modules/core/src/test/java/org/apache/ignite/loadtests/cache/GridCacheDataStructuresLoadTest.java
index d4ca9a5..18537c5 100644
--- a/modules/core/src/test/java/org/apache/ignite/loadtests/cache/GridCacheDataStructuresLoadTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/loadtests/cache/GridCacheDataStructuresLoadTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.IgniteAtomicStamped;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCountDownLatch;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteQueue;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteSystemProperties;
@@ -64,6 +65,9 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
/** Semaphore name. */
private static final String TEST_SEMAPHORE_NAME = "test-semaphore";
+ /** Reentrant lock name. */
+ private static final String TEST_REENTRANT_LOCK_NAME = "test-reentrant-lock";
+
/** */
private static final CollectionConfiguration colCfg = new CollectionConfiguration();
@@ -98,6 +102,9 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
private static final boolean SEMAPHORE = true;
/** */
+ private static final boolean REENTRANTLOCK = true;
+
+ /** */
private GridCacheDataStructuresLoadTest() {
// No-op
}
@@ -347,6 +354,44 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
}
};
+
+ /** Reentrant lock read closure. */
+ private final CIX1<Ignite> reentrantLockReadClos =
+ new CIX1<Ignite>() {
+ @Override public void applyx(Ignite ignite) {
+ IgniteLock r = ignite.reentrantLock(TEST_REENTRANT_LOCK_NAME, true, false, true);
+
+ for (int i = 0; i < operationsPerTx; i++) {
+ r.isLocked();
+
+ long cnt = reads.incrementAndGet();
+
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
+ }
+ };
+
+ /** Reentrant lock write closure. */
+ private final CIX1<Ignite> reentrantLockWriteClos =
+ new CIX1<Ignite>() {
+ @Override public void applyx(Ignite ignite) {
+ IgniteLock r = ignite.reentrantLock(TEST_REENTRANT_LOCK_NAME, true, false, true);
+
+ for (int i = 0; i < operationsPerTx; i++) {
+ if ((i % 2) == 0)
+ r.lock();
+ else
+ r.unlock();
+
+ long cnt = writes.incrementAndGet();
+
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
+ }
+ };
+
/**
* @param args Arguments.
* @throws IgniteCheckedException In case of error.
@@ -417,6 +462,14 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
test.loadTestIgnite(test.semaphoreWriteClos, test.semaphoreReadClos);
}
+
+ System.gc();
+
+ if (REENTRANTLOCK) {
+ info("Testing reentrant lock...");
+
+ test.loadTestIgnite(test.reentrantLockWriteClos, test.reentrantLockReadClos);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/testframework/junits/IgniteMock.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/IgniteMock.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/IgniteMock.java
index c49c730..c9859fc 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/IgniteMock.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/IgniteMock.java
@@ -30,6 +30,7 @@ import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCluster;
import org.apache.ignite.IgniteCompute;
import org.apache.ignite.IgniteCountDownLatch;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteEvents;
@@ -367,6 +368,15 @@ public class IgniteMock implements Ignite {
}
/** {@inheritDoc} */
+ @Nullable @Override public IgniteLock reentrantLock(String name,
+ boolean failoverSafe,
+ boolean fair,
+ boolean create)
+ {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public <T> IgniteQueue<T> queue(String name,
int cap,
CollectionConfiguration cfg)
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteProcessProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteProcessProxy.java
index a2e0d5a..2598bc5 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteProcessProxy.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteProcessProxy.java
@@ -42,6 +42,7 @@ import org.apache.ignite.IgniteIllegalStateException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteMessaging;
import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteScheduler;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteServices;
@@ -545,6 +546,12 @@ public class IgniteProcessProxy implements IgniteEx {
}
/** {@inheritDoc} */
+ @Override public IgniteLock reentrantLock(String name, boolean failoverSafe,
+ boolean fair, boolean create) throws IgniteException {
+ throw new UnsupportedOperationException("Operation isn't supported yet.");
+ }
+
+ /** {@inheritDoc} */
@Override public <T> IgniteQueue<T> queue(String name, int cap,
@Nullable CollectionConfiguration cfg) throws IgniteException {
throw new UnsupportedOperationException("Operation isn't supported yet.");
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
index e663a99..59a18e5 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheDataStructuresSelfTestSuite.java
@@ -33,6 +33,7 @@ import org.apache.ignite.internal.processors.cache.datastructures.local.GridCach
import org.apache.ignite.internal.processors.cache.datastructures.local.GridCacheLocalSetSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.local.IgniteLocalAtomicLongApiSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.local.IgniteLocalCountDownLatchSelfTest;
+import org.apache.ignite.internal.processors.cache.datastructures.local.IgniteLocalLockSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.local.IgniteLocalSemaphoreSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.GridCachePartitionedAtomicOffheapQueueApiSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.GridCachePartitionedAtomicOffheapQueueCreateMultiNodeSelfTest;
@@ -66,6 +67,7 @@ import org.apache.ignite.internal.processors.cache.datastructures.partitioned.Gr
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.GridCachePartitionedSetSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.IgnitePartitionedAtomicLongApiSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.IgnitePartitionedCountDownLatchSelfTest;
+import org.apache.ignite.internal.processors.cache.datastructures.partitioned.IgnitePartitionedLockSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.IgnitePartitionedQueueNoBackupsTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.IgnitePartitionedSemaphoreSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.partitioned.IgnitePartitionedSetNoBackupsSelfTest;
@@ -80,6 +82,7 @@ import org.apache.ignite.internal.processors.cache.datastructures.replicated.Gri
import org.apache.ignite.internal.processors.cache.datastructures.replicated.GridCacheReplicatedSetSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.replicated.IgniteReplicatedAtomicLongApiSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.replicated.IgniteReplicatedCountDownLatchSelfTest;
+import org.apache.ignite.internal.processors.cache.datastructures.replicated.IgniteReplicatedLockSelfTest;
import org.apache.ignite.internal.processors.cache.datastructures.replicated.IgniteReplicatedSemaphoreSelfTest;
import org.apache.ignite.internal.processors.cache.distributed.replicated.IgniteCacheAtomicReplicatedNodeRestartSelfTest;
@@ -107,6 +110,7 @@ public class IgniteCacheDataStructuresSelfTestSuite extends TestSuite {
suite.addTest(new TestSuite(GridCacheLocalAtomicQueueApiSelfTest.class));
suite.addTest(new TestSuite(IgniteLocalCountDownLatchSelfTest.class));
suite.addTest(new TestSuite(IgniteLocalSemaphoreSelfTest.class));
+ suite.addTest(new TestSuite(IgniteLocalLockSelfTest.class));
suite.addTest(new TestSuite(GridCacheReplicatedSequenceApiSelfTest.class));
suite.addTest(new TestSuite(GridCacheReplicatedSequenceMultiNodeSelfTest.class));
@@ -117,6 +121,7 @@ public class IgniteCacheDataStructuresSelfTestSuite extends TestSuite {
suite.addTest(new TestSuite(GridCacheReplicatedDataStructuresFailoverSelfTest.class));
suite.addTest(new TestSuite(IgniteReplicatedCountDownLatchSelfTest.class));
suite.addTest(new TestSuite(IgniteReplicatedSemaphoreSelfTest.class));
+ suite.addTest(new TestSuite(IgniteReplicatedLockSelfTest.class));
suite.addTest(new TestSuite(IgniteCacheAtomicReplicatedNodeRestartSelfTest.class));
suite.addTest(new TestSuite(GridCachePartitionedSequenceApiSelfTest.class));
@@ -139,6 +144,7 @@ public class IgniteCacheDataStructuresSelfTestSuite extends TestSuite {
suite.addTest(new TestSuite(IgnitePartitionedCountDownLatchSelfTest.class));
suite.addTest(new TestSuite(IgniteDataStructureWithJobTest.class));
suite.addTest(new TestSuite(IgnitePartitionedSemaphoreSelfTest.class));
+ suite.addTest(new TestSuite(IgnitePartitionedLockSelfTest.class));
suite.addTest(new TestSuite(GridCachePartitionedSetFailoverSelfTest.class));
suite.addTest(new TestSuite(GridCachePartitionedOffheapSetFailoverSelfTest.class));
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/spring/src/main/java/org/apache/ignite/IgniteSpringBean.java
----------------------------------------------------------------------
diff --git a/modules/spring/src/main/java/org/apache/ignite/IgniteSpringBean.java b/modules/spring/src/main/java/org/apache/ignite/IgniteSpringBean.java
index 92d8c1d..03c7b0e 100644
--- a/modules/spring/src/main/java/org/apache/ignite/IgniteSpringBean.java
+++ b/modules/spring/src/main/java/org/apache/ignite/IgniteSpringBean.java
@@ -425,6 +425,17 @@ public class IgniteSpringBean implements Ignite, DisposableBean, InitializingBea
}
/** {@inheritDoc} */
+ @Nullable @Override public IgniteLock reentrantLock(String name,
+ boolean failoverSafe,
+ boolean fair,
+ boolean create)
+ {
+ assert g != null;
+
+ return g.reentrantLock(name, failoverSafe, create, fair);
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public <T> IgniteQueue<T> queue(String name,
int cap,
CollectionConfiguration cfg)
[2/3] ignite git commit: IGNITE-642 Implement IgniteReentrantLock
data structure
Posted by yz...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockImpl.java
new file mode 100644
index 0000000..3ab7289
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockImpl.java
@@ -0,0 +1,1538 @@
+/*
+ * 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.processors.datastructures;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.InvalidObjectException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.io.ObjectStreamException;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.AbstractQueuedSynchronizer;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteCondition;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteInterruptedException;
+import org.apache.ignite.IgniteLock;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.IgnitionEx;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.transactions.TransactionRollbackException;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.processors.cache.GridCacheUtils.retryTopologySafe;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ * Cache reentrant lock implementation based on AbstractQueuedSynchronizer.
+ */
+public final class GridCacheLockImpl implements GridCacheLockEx, Externalizable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Deserialization stash. */
+ private static final ThreadLocal<String> stash = new ThreadLocal<>();
+
+ /** Logger. */
+ private IgniteLogger log;
+
+ /** Reentrant lock name. */
+ private String name;
+
+ /** Removed flag. */
+ private volatile boolean rmvd;
+
+ /** Reentrant lock key. */
+ private GridCacheInternalKey key;
+
+ /** Reentrant lock projection. */
+ private IgniteInternalCache<GridCacheInternalKey, GridCacheLockState> lockView;
+
+ /** Cache context. */
+ private final GridCacheContext ctx;
+
+ /** Initialization guard. */
+ private final AtomicBoolean initGuard = new AtomicBoolean();
+
+ /** Initialization latch. */
+ private final CountDownLatch initLatch = new CountDownLatch(1);
+
+ /** Lock that provides non-overlapping processing of updates. */
+ private Lock updateLock = new ReentrantLock();
+
+ /** Internal synchronization object. */
+ private Sync sync;
+
+ /** Flag indicating that every operation on this lock should be interrupted. */
+ private volatile boolean interruptAll;
+
+ /**
+ * Empty constructor required by {@link Externalizable}.
+ */
+ public GridCacheLockImpl() {
+ // This instance should never be used directly.
+ ctx = null;
+ }
+
+ /**
+ * Synchronization implementation for reentrant lock using AbstractQueuedSynchronizer.
+ */
+ @SuppressWarnings({"CallToThreadYield", "CallToSignalInsteadOfSignalAll"})
+ private class Sync extends AbstractQueuedSynchronizer {
+ /** */
+ private static final long serialVersionUID = 1192457210091910933L;
+
+ /** */
+ private static final long LOCK_FREE = 0;
+
+ /** Map containing condition objects. */
+ private Map<String, ConditionObject> conditionMap;
+
+ /** List of condition signal calls on this node. */
+ private Map<String, Integer> outgoingSignals;
+
+ /** Last condition waited on. */
+ @Nullable
+ private volatile String lastCondition;
+
+ /** True if any node owning the lock had failed. */
+ private volatile boolean isBroken;
+
+ /** UUID of the node that currently owns the lock. */
+ private volatile UUID currentOwnerNode;
+
+ /** ID of the thread that currently owns the lock. */
+ private volatile long currentOwnerThreadId;
+
+ /** UUID of this node. */
+ private final UUID thisNode;
+
+ /** FailoverSafe flag. */
+ private final boolean failoverSafe;
+
+ /** Fairness flag. */
+ private final boolean fair;
+
+ /** Threads that are waiting on this lock. */
+ private Set<Long> waitingThreads;
+
+ /**
+ * @param state State.
+ */
+ protected Sync(GridCacheLockState state) {
+ setState(state.get());
+
+ thisNode = ctx.localNodeId();
+
+ currentOwnerNode = state.getId();
+
+ currentOwnerThreadId = state.getThreadId();
+
+ conditionMap = new HashMap<>();
+
+ outgoingSignals = new HashMap<>();
+
+ failoverSafe = state.isFailoverSafe();
+
+ fair = state.isFair();
+
+ waitingThreads = new ConcurrentSkipListSet<>();
+ }
+
+ /**
+ *
+ */
+ protected void addOutgoingSignal(String condition) {
+ int cnt = 0;
+
+ if (outgoingSignals.containsKey(condition)) {
+ cnt = outgoingSignals.get(condition);
+
+ // SignalAll has already been called.
+ if (cnt == 0)
+ return;
+ }
+
+ outgoingSignals.put(condition, cnt + 1);
+ }
+
+ protected void addOutgoingSignalAll(String condition) {
+ outgoingSignals.put(condition, 0);
+ }
+
+ /**
+ * Process any condition await calls on this node.
+ */
+ private String processAwait() {
+ if (lastCondition == null)
+ return null;
+
+ String ret = lastCondition;
+
+ lastCondition = null;
+
+ return ret;
+ }
+
+ /** */
+ private Map<String, Integer> processSignal() {
+ Map<String,Integer> ret = new HashMap<>(outgoingSignals);
+
+ outgoingSignals.clear();
+
+ return ret;
+ }
+
+ /** Interrupt every thread on this node waiting on this lock. */
+ private synchronized void interruptAll() {
+ // First release all threads waiting on associated condition queues.
+ if (!conditionMap.isEmpty()) {
+ // Temporarily obtain ownership of the lock,
+ // in order to signal all conditions.
+ UUID tempUUID = getOwnerNode();
+
+ long tempThreadID = currentOwnerThreadId;
+
+ setCurrentOwnerNode(thisNode);
+
+ currentOwnerThreadId = Thread.currentThread().getId();
+
+ for (Condition c : conditionMap.values())
+ c.signalAll();
+
+ // Restore owner node and owner thread.
+ setCurrentOwnerNode(tempUUID);
+
+ currentOwnerThreadId = tempThreadID;
+ }
+
+ // Interrupt any future call to acquire/release on this sync object.
+ interruptAll = true;
+
+ // Interrupt any ongoing transactions.
+ for (Thread t: getQueuedThreads())
+ t.interrupt();
+ }
+
+ /** Check if lock is in correct state (i.e. not broken in non-failoversafe mode),
+ * if not throw {@linkplain IgniteInterruptedException} */
+ private void validate(final boolean throwInterrupt) {
+ // Interrupted flag shouldn't be always cleared
+ // (e.g. lock() method doesn't throw exception and doesn't clear interrupted)
+ // but should be cleared if this method is called after lock breakage or node stop.
+ // If interruptAll is set, exception is thrown anyway.
+ boolean interrupted = Thread.currentThread().isInterrupted();
+
+ // Clear interrupt flag.
+ if (throwInterrupt || interruptAll)
+ Thread.interrupted();
+
+ if (interruptAll)
+ throw new IgniteException("Lock broken (possible reason: node stopped" +
+ " or node owning lock failed while in non-failoversafe mode).");
+
+ // Global queue should be synchronized only if interrupted exception should be thrown.
+ if (fair && (throwInterrupt && interrupted) && !interruptAll) {
+ synchronizeQueue(true, Thread.currentThread());
+
+ throw new IgniteInterruptedException("Lock is interrupted.");
+ }
+ }
+
+ /**
+ * Sets the number of permits currently acquired on this lock. This method should only be used in {@linkplain
+ * GridCacheLockImpl#onUpdate(GridCacheLockState)}.
+ *
+ * @param permits Number of permits acquired at this reentrant lock.
+ */
+ final synchronized void setPermits(int permits) {
+ setState(permits);
+ }
+
+ /**
+ * Gets the number of permissions currently acquired at this lock.
+ *
+ * @return Number of permits acquired at this reentrant lock.
+ */
+ final int getPermits() {
+ return getState();
+ }
+
+ /**
+ * Sets the UUID of the node that currently owns this lock. This method should only be used in {@linkplain
+ * GridCacheLockImpl#onUpdate(GridCacheLockState)}.
+ *
+ * @param ownerNode UUID of the node owning this lock.
+ */
+ final synchronized void setCurrentOwnerNode(UUID ownerNode) {
+ currentOwnerNode = ownerNode;
+ }
+
+ /**
+ * Gets the UUID of the node that currently owns the lock.
+ *
+ * @return UUID of the node that currently owns the lock.
+ */
+ final UUID getOwnerNode() {
+ return currentOwnerNode;
+ }
+
+ /**
+ * Checks if latest call to acquire/release was called on this node.
+ * Should only be called from update method.
+ *
+ * @param newOwnerID ID of the node that is about to acquire this lock (or null).
+ * @return true if acquire/release that triggered last update came from this node.
+ */
+ protected boolean isLockedLocally(UUID newOwnerID) {
+ return thisNode.equals(getOwnerNode()) || thisNode.equals(newOwnerID);
+ }
+
+ protected void setCurrentOwnerThread(long newOwnerThreadId) {
+ currentOwnerThreadId = newOwnerThreadId;
+ }
+
+ /**
+ * Returns true if node that owned the locked failed before call to unlock.
+ *
+ * @return true if any node failed while owning the lock.
+ */
+ protected boolean isBroken() {
+ return isBroken;
+ }
+
+ /** */
+ protected void setBroken(boolean isBroken) {
+ this.isBroken = isBroken;
+ }
+
+ /** */
+ protected synchronized boolean hasPredecessor(LinkedList<UUID> nodes) {
+ if (!fair)
+ return false;
+
+ for (Iterator<UUID> it = nodes.iterator(); it.hasNext(); ) {
+ UUID node = it.next();
+
+ if (ctx.discovery().node(node) == null) {
+ it.remove();
+
+ continue;
+ }
+
+ return !node.equals(thisNode);
+ }
+
+ return false;
+ }
+
+ /**
+ * Performs tryLock.
+ * @param acquires Number of permits to acquire.
+ * @param fair Fairness parameter.
+ * @return {@code True} if succeeded, false otherwise.
+ */
+ final boolean tryAcquire(final int acquires, final boolean fair) {
+ // If broken in non-failoversafe mode, exit immediately.
+ if (interruptAll)
+ return true;
+
+ final Thread current = Thread.currentThread();
+
+ boolean failed = false;
+
+ int c = getState();
+
+ // Wait for lock to reach stable state.
+ while (c != 0) {
+ UUID currentOwner = currentOwnerNode;
+
+ if (currentOwner != null) {
+ failed = ctx.discovery().node(currentOwner) == null;
+
+ break;
+ }
+
+ c = getState();
+ }
+
+ // Check if lock is released or current owner failed.
+ if (c == 0 || failed) {
+ if (compareAndSetGlobalState(0, acquires, current, fair)) {
+
+ // Not used for synchronization (we use ThreadID), but updated anyway.
+ setExclusiveOwnerThread(current);
+
+ while (!isHeldExclusively() && !interruptAll)
+ Thread.yield();
+
+ return true;
+ }
+ }
+ else if (isHeldExclusively()) {
+ int nextc = c + acquires;
+
+ if (nextc < 0) // overflow
+ throw new Error("Maximum lock count exceeded.");
+
+ setState(nextc);
+
+ return true;
+ }
+
+ if (fair && !isQueued(current))
+ synchronizeQueue(false, current);
+
+ return false;
+ }
+
+ /**
+ * Performs lock.
+ */
+ final void lock() {
+ acquire(1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected final boolean tryAcquire(int acquires) {
+ return tryAcquire(acquires, fair);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected final boolean tryRelease(int releases) {
+ // This method is called with release==0 only when trying to wake through update,
+ // to check if some other node released the lock.
+ if (releases == 0)
+ return true;
+
+ // If broken in non-failoversafe mode, exit immediately.
+ if (interruptAll)
+ return true;
+
+ int c = getState() - releases;
+
+ if (!isHeldExclusively()) {
+ log.error("Lock.unlock() is called in illegal state [callerNodeId=" + thisNode + ", ownerNodeId="
+ + currentOwnerNode + ", callerThreadId=" + Thread.currentThread().getId() + ", ownerThreadId="
+ + currentOwnerThreadId + ", lockState=" + getState() + "]");
+
+ throw new IllegalMonitorStateException();
+ }
+
+ boolean free = false;
+
+ if (c == 0) {
+ free = true;
+
+ setGlobalState(0, processAwait(), processSignal());
+
+ while (isHeldExclusively() && !interruptAll)
+ Thread.yield();
+ }
+ else
+ setState(c);
+
+ return free;
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected final boolean isHeldExclusively() {
+ // While we must in general read state before owner,
+ // we don't need to do so to check if current thread is owner
+
+ return currentOwnerThreadId == Thread.currentThread().getId() && thisNode.equals(currentOwnerNode);
+ }
+
+ /**
+ * @param name Condition name.
+ * @return Condition object.
+ */
+ final synchronized IgniteCondition newCondition(String name) {
+ if (conditionMap.containsKey(name))
+ return new IgniteConditionObject(name, conditionMap.get(name));
+
+ ConditionObject cond = new ConditionObject();
+
+ conditionMap.put(name, cond);
+
+ return new IgniteConditionObject(name, cond);
+ }
+
+ // Methods relayed from outer class
+
+ final int getHoldCount() {
+ return isHeldExclusively() ? getState() : 0;
+ }
+
+ final boolean isLocked() throws IgniteCheckedException {
+ return getState() != 0 || lockView.get(key).get() != 0;
+ }
+
+ /**
+ * This method is used for synchronizing the reentrant lock state across all nodes.
+ */
+ protected boolean compareAndSetGlobalState(final int expVal, final int newVal,
+ final Thread newThread, final boolean bargingProhibited) {
+ try {
+ return CU.outTx(
+ retryTopologySafe(new Callable<Boolean>() {
+ @Override public Boolean call() throws Exception {
+ try (IgniteInternalTx tx = CU.txStartInternal(ctx, lockView, PESSIMISTIC, REPEATABLE_READ)) {
+
+ GridCacheLockState val = lockView.get(key);
+
+ if (val == null)
+ throw new IgniteCheckedException("Failed to find reentrant lock with given name: " + name);
+
+ final long newThreadID = newThread.getId();
+
+ LinkedList<UUID> nodes = val.getNodes();
+
+ // Barging is prohibited in fair mode unless tryLock() is called.
+ if (!(bargingProhibited && hasPredecessor(nodes))) {
+ if (val.get() == expVal || ctx.discovery().node(val.getId()) == null) {
+ val.set(newVal);
+
+ val.setId(thisNode);
+
+ val.setThreadId(newThreadID);
+
+ val.setSignals(null);
+
+ // This node is already in queue, except in cases where this is the only node
+ // or this is a call to tryLock(), in which case barging is ok.
+ // Queue is only updated if this is fair lock.
+ if (val.isFair() && (nodes.isEmpty() || !bargingProhibited))
+ nodes.addFirst(thisNode);
+
+ val.setNodes(nodes);
+
+ val.setChanged(true);
+
+ lockView.put(key, val);
+
+ tx.commit();
+
+ return true;
+ }
+ }
+
+ return false;
+ }
+ catch (Exception e) {
+ if (interruptAll) {
+ log.info("Node is stopped (or lock is broken in non-failover safe mode)," +
+ " aborting transaction.");
+
+ // Return immediately, exception will be thrown later.
+ return true;
+ }
+ else {
+ if (Thread.currentThread().isInterrupted()) {
+ log.info("Thread is interrupted while attempting to acquire lock.");
+
+ // Delegate the decision to throw InterruptedException to the AQS.
+ sync.release(0);
+
+ return false;
+ }
+
+ U.error(log, "Failed to compare and set: " + this, e);
+ }
+
+ throw e;
+ }
+ }
+ }),
+ ctx
+ );
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /**
+ * This method is used for synchronizing the number of acquire attempts on this lock across all nodes.
+ *
+ * @param cancelled true if acquire attempt is cancelled, false if acquire attempt should be registered.
+ */
+ protected boolean synchronizeQueue(final boolean cancelled, final Thread thread) {
+ final AtomicBoolean interrupted = new AtomicBoolean(false);
+
+ try {
+ return CU.outTx(
+ retryTopologySafe(new Callable<Boolean>() {
+ @Override public Boolean call() throws Exception {
+ try (IgniteInternalTx tx = CU.txStartInternal(ctx, lockView, PESSIMISTIC, REPEATABLE_READ)) {
+ GridCacheLockState val = lockView.get(key);
+
+ if (val == null)
+ throw new IgniteCheckedException("Failed to find reentrant lock with given name: " + name);
+
+ LinkedList<UUID> nodes = val.getNodes();
+
+ if (!cancelled) {
+ nodes.add(thisNode);
+
+ val.setChanged(false);
+
+ lockView.put(key, val);
+
+ tx.commit();
+
+ // Keep track of all threads that are queued in global queue.
+ // We deliberately don't use #sync.isQueued(), because AQS
+ // cancel threads immediately after throwing interrupted exception.
+ sync.waitingThreads.add(thread.getId());
+
+ return true;
+ }
+ else {
+ if (sync.waitingThreads.contains(thread.getId())) {
+ // Update other nodes if this is the first node in queue.
+ val.setChanged(nodes.lastIndexOf(thisNode) == 0);
+
+ nodes.removeLastOccurrence(thisNode);
+
+ lockView.put(key, val);
+
+ tx.commit();
+
+ sync.waitingThreads.remove(thread.getId());
+
+ return true;
+ }
+ }
+
+ return false;
+ }
+ catch (Exception e) {
+ if (interruptAll) {
+ log.info("Node is stopped (or lock is broken in non-failover safe mode)," +
+ " aborting transaction.");
+
+ // Abort this attempt to synchronize queue and start another one,
+ // that will return immediately.
+ sync.release(0);
+
+ return false;
+ }
+ else {
+ // If thread got interrupted, abort this attempt to synchronize queue,
+ // clear interrupt flag and try again, and let the AQS decide
+ // whether to throw an exception or ignore it.
+ if (Thread.interrupted() || X.hasCause(e, InterruptedException.class)) {
+ interrupted.set(true);
+
+ throw new TransactionRollbackException("Thread got interrupted " +
+ "while synchronizing the global queue, retrying. ");
+ }
+
+ U.error(log, "Failed to synchronize global lock queue: " + this, e);
+ }
+
+ throw e;
+ }
+ }
+ }),
+ ctx
+ );
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ finally {
+ // Restore interrupt flag and let AQS decide what to do with it.
+ if (interrupted.get())
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Sets the global state across all nodes after releasing the reentrant lock.
+ *
+ * @param newVal New state.
+ * @param lastCondition Id of the condition await is called.
+ * @param outgoingSignals Map containing signal calls on this node since the last acquisition of the lock.
+ */
+ protected boolean setGlobalState(final int newVal, @Nullable final String lastCondition, final Map<String, Integer> outgoingSignals) {
+ try {
+ return CU.outTx(
+ retryTopologySafe(new Callable<Boolean>() {
+ @Override public Boolean call() throws Exception {
+ try (IgniteInternalTx tx = CU.txStartInternal(ctx, lockView, PESSIMISTIC, REPEATABLE_READ)) {
+ GridCacheLockState val = lockView.get(key);
+
+ if (val == null)
+ throw new IgniteCheckedException("Failed to find reentrant lock with given name: " + name);
+
+ val.set(newVal);
+
+ if (newVal == 0) {
+ val.setId(null);
+
+ val.setThreadId(LOCK_FREE);
+ }
+
+ val.setChanged(true);
+
+ // If this lock is fair, remove this node from queue.
+ if (val.isFair() && newVal == 0) {
+ UUID removedNode = val.getNodes().removeFirst();
+
+ assert(thisNode.equals(removedNode));
+ }
+
+ // Get global condition queue.
+ Map<String, LinkedList<UUID>> condMap = val.getConditionMap();
+
+ // Create map containing signals from this node.
+ Map<UUID, LinkedList<String>> signalMap = new HashMap<>();
+
+ // Put any signal calls on this node to global state.
+ if (!outgoingSignals.isEmpty()) {
+ for (String condition : outgoingSignals.keySet()) {
+ int cnt = outgoingSignals.get(condition);
+
+ // Get queue for this condition.
+ List<UUID> list = condMap.get(condition);
+
+ if (list != null && !list.isEmpty()) {
+ // Check if signalAll was called.
+ if (cnt == 0) {
+ cnt = list.size();
+ }
+
+ // Remove from global condition queue.
+ for (int i = 0; i < cnt; i++) {
+ if (list.isEmpty())
+ break;
+
+ UUID uuid = list.remove(0);
+
+ // Skip if node to be released is not alive anymore.
+ if (ctx.discovery().node(uuid) == null) {
+ cnt++;
+
+ continue;
+ }
+
+ LinkedList<String> queue = signalMap.get(uuid);
+
+ if (queue == null) {
+ queue = new LinkedList<>();
+
+ signalMap.put(uuid, queue);
+ }
+
+ queue.add(condition);
+ }
+ }
+ }
+ }
+
+ val.setSignals(signalMap);
+
+ // Check if this release is called after condition.await() call;
+ // If true, add this node to the global waiting queue.
+ if (lastCondition != null) {
+ LinkedList<UUID> queue;
+
+ //noinspection IfMayBeConditional
+ if (!condMap.containsKey(lastCondition))
+ // New condition object.
+ queue = new LinkedList<>();
+ else
+ // Existing condition object.
+ queue = condMap.get(lastCondition);
+
+ queue.add(thisNode);
+
+ condMap.put(lastCondition, queue);
+ }
+
+ val.setConditionMap(condMap);
+
+ lockView.put(key, val);
+
+ tx.commit();
+
+ return true;
+ }
+ catch (Exception e) {
+ if (interruptAll) {
+ log.info("Node is stopped (or lock is broken in non-failover safe mode)," +
+ " aborting transaction.");
+
+ return true;
+ }
+ else
+ U.error(log, "Failed to release: " + this, e);
+
+ throw e;
+ }
+ }
+ }),
+ ctx
+ );
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ protected synchronized boolean checkIncomingSignals(GridCacheLockState state) {
+ if (state.getSignals() == null)
+ return false;
+
+ LinkedList<String> signals = state.getSignals().get(thisNode);
+
+ if (signals == null || signals.isEmpty())
+ return false;
+
+ UUID tempUUID = getOwnerNode();
+
+ Thread tempThread = getExclusiveOwnerThread();
+
+ long tempThreadID = currentOwnerThreadId;
+
+ // Temporarily allow current thread to signal condition object.
+ // This is safe to do because:
+ // 1. if release was called on this node,
+ // it was called from currently active thread;
+ // 2. if release came from a thread on any other node,
+ // all threads on this node are already blocked.
+ setCurrentOwnerNode(thisNode);
+
+ setExclusiveOwnerThread(Thread.currentThread());
+
+ currentOwnerThreadId = Thread.currentThread().getId();
+
+ for (String signal: signals)
+ conditionMap.get(signal).signal();
+
+ // Restore owner node and owner thread.
+ setCurrentOwnerNode(tempUUID);
+
+ setExclusiveOwnerThread(tempThread);
+
+ currentOwnerThreadId = tempThreadID;
+
+ return true;
+ }
+
+ /**
+ * Condition implementation for {@linkplain IgniteLock}.
+ *
+ **/
+ private class IgniteConditionObject implements IgniteCondition {
+ /** */
+ private final String name;
+
+ /** */
+ private final AbstractQueuedSynchronizer.ConditionObject object;
+
+ /**
+ * @param name Condition name.
+ * @param object Condition object.
+ */
+ protected IgniteConditionObject(String name, ConditionObject object) {
+ this.name = name;
+
+ this.object = object;
+ }
+
+ /**
+ * Name of this condition.
+ *
+ * @return name Name of this condition object.
+ */
+ @Override public String name() {
+ return name;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void await() throws IgniteInterruptedException {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ lastCondition = name;
+
+ object.await();
+
+ sync.validate(true);
+ }
+ catch (InterruptedException e) {
+ throw new IgniteInterruptedException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void awaitUninterruptibly() {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ lastCondition = name;
+
+ object.awaitUninterruptibly();
+
+ sync.validate(false);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public long awaitNanos(long nanosTimeout) throws IgniteInterruptedException {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ lastCondition = name;
+
+ long result = object.awaitNanos(nanosTimeout);
+
+ sync.validate(true);
+
+ return result;
+ }
+ catch (InterruptedException e) {
+ throw new IgniteInterruptedException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean await(long time, TimeUnit unit) throws IgniteInterruptedException {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ lastCondition = name;
+
+ boolean result = object.await(time, unit);
+
+ sync.validate(true);
+
+ return result;
+ }
+ catch (InterruptedException e) {
+ throw new IgniteInterruptedException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean awaitUntil(Date deadline) throws IgniteInterruptedException {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ lastCondition = name;
+
+ boolean result = object.awaitUntil(deadline);
+
+ sync.validate(true);
+
+ return result;
+ }
+ catch (InterruptedException e) {
+ throw new IgniteInterruptedException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void signal() {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ validate(false);
+
+ addOutgoingSignal(name);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void signalAll() {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ if (!isHeldExclusively())
+ throw new IllegalMonitorStateException();
+
+ sync.validate(false);
+
+ addOutgoingSignalAll(name);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+ }
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param name Reentrant lock name.
+ * @param key Reentrant lock key.
+ * @param lockView Reentrant lock projection.
+ * @param ctx Cache context.
+ */
+ @SuppressWarnings("unchecked")
+ public GridCacheLockImpl(String name,
+ GridCacheInternalKey key,
+ IgniteInternalCache<GridCacheInternalKey, GridCacheLockState> lockView,
+ GridCacheContext ctx) {
+ assert name != null;
+ assert key != null;
+ assert ctx != null;
+ assert lockView != null;
+
+ this.name = name;
+ this.key = key;
+ this.lockView = lockView;
+ this.ctx = ctx;
+
+ log = ctx.logger(getClass());
+ }
+
+ /**
+ * @throws IgniteCheckedException If operation failed.
+ */
+ private void initializeReentrantLock() throws IgniteCheckedException {
+ if (initGuard.compareAndSet(false, true)) {
+ try {
+ sync = CU.outTx(
+ retryTopologySafe(new Callable<Sync>() {
+ @Override public Sync call() throws Exception {
+ try (IgniteInternalTx tx = CU.txStartInternal(ctx, lockView, PESSIMISTIC, REPEATABLE_READ)) {
+ GridCacheLockState val = lockView.get(key);
+
+ if (val == null) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to find reentrant lock with given name: " + name);
+
+ return null;
+ }
+
+ tx.rollback();
+
+ return new Sync(val);
+ }
+ }
+ }),
+ ctx
+ );
+
+ if (log.isDebugEnabled())
+ log.debug("Initialized internal sync structure: " + sync);
+ }
+ finally {
+ initLatch.countDown();
+ }
+ }
+ else {
+ U.await(initLatch);
+
+ if (sync == null)
+ throw new IgniteCheckedException("Internal reentrant lock has not been properly initialized.");
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onUpdate(GridCacheLockState val) {
+ // Called only on initialization, so it's safe to ignore update.
+ if (sync == null)
+ return;
+
+ updateLock.lock();
+
+ try {
+ // If this update is a result of unsuccessful acquire in fair mode, no local update should be done.
+ if (!val.isChanged())
+ return;
+
+ // Check if update came from this node.
+ boolean local = sync.isLockedLocally(val.getId());
+
+ // Process any incoming signals.
+ boolean incomingSignals = sync.checkIncomingSignals(val);
+
+ // Update permission count.
+ sync.setPermits(val.get());
+
+ // Update owner's node id.
+ sync.setCurrentOwnerNode(val.getId());
+
+ // Update owner's thread id.
+ sync.setCurrentOwnerThread(val.getThreadId());
+
+ // Check if any threads waiting on this node need to be notified.
+ if ((incomingSignals || sync.getPermits() == 0) && !local) {
+ // Try to notify any waiting threads.
+ sync.release(0);
+ }
+
+ } finally{
+ updateLock.unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onNodeRemoved(UUID nodeId) {
+ updateLock.lock();
+
+ try {
+ if (nodeId.equals(sync.getOwnerNode())) {
+ sync.setBroken(true);
+
+ if (!sync.failoverSafe) {
+ sync.interruptAll();
+ }
+ }
+
+ // Try to notify any waiting threads.
+ sync.release(0);
+ }
+ finally {
+ updateLock.unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onStop() {
+ if (sync == null) {
+ interruptAll = true;
+
+ return;
+ }
+
+ sync.setBroken(true);
+
+ sync.interruptAll();
+
+ // Try to notify any waiting threads.
+ sync.release(0);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String name() {
+ return name;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void lock() {
+ ctx.kernalContext().gateway().readLock();
+
+ try{
+ initializeReentrantLock();
+
+ sync.lock();
+
+ sync.validate(false);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void lockInterruptibly() throws IgniteInterruptedException {
+ ctx.kernalContext().gateway().readLock();
+
+ try {
+ initializeReentrantLock();
+
+ sync.acquireInterruptibly(1);
+
+ sync.validate(true);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ catch (InterruptedException e) {
+ if (sync.fair)
+ sync.synchronizeQueue(true, Thread.currentThread());
+
+ throw new IgniteInterruptedException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean tryLock() {
+ ctx.kernalContext().gateway().readLock();
+
+ try{
+ initializeReentrantLock();
+
+ boolean result = sync.tryAcquire(1, false);
+
+ sync.validate(false);
+
+ return result;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean tryLock(long timeout, TimeUnit unit) throws IgniteInterruptedException {
+ ctx.kernalContext().gateway().readLock();
+
+ try{
+ initializeReentrantLock();
+
+ boolean result = sync.tryAcquireNanos(1, unit.toNanos(timeout));
+
+ sync.validate(true);
+
+ return result;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ catch (InterruptedException e) {
+ if (sync.fair)
+ sync.synchronizeQueue(true, Thread.currentThread());
+
+ throw new IgniteInterruptedException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void unlock() {
+ ctx.kernalContext().gateway().readLock();
+
+ try{
+ initializeReentrantLock();
+
+ // Validate before release.
+ sync.validate(false);
+
+ sync.release(1);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ @NotNull @Override public Condition newCondition() {
+ throw new UnsupportedOperationException("IgniteLock does not allow creation of nameless conditions. ");
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteCondition getOrCreateCondition(String name) {
+ ctx.kernalContext().gateway().readLock();
+
+ try{
+ initializeReentrantLock();
+
+ IgniteCondition result = sync.newCondition(name);
+
+ sync.validate(false);
+
+ return result;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ finally {
+ ctx.kernalContext().gateway().readUnlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getHoldCount() {
+ try{
+ initializeReentrantLock();
+
+ return sync.getHoldCount();
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isHeldByCurrentThread() {
+ try{
+ initializeReentrantLock();
+
+ return sync.isHeldExclusively();
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isLocked() {
+ try{
+ initializeReentrantLock();
+
+ return sync.isLocked();
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasQueuedThreads() {
+ try{
+ initializeReentrantLock();
+
+ return sync.hasQueuedThreads();
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasQueuedThread(Thread thread) {
+ try{
+ initializeReentrantLock();
+
+ return sync.isQueued(thread);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasWaiters(IgniteCondition condition) {
+ try{
+ initializeReentrantLock();
+
+ AbstractQueuedSynchronizer.ConditionObject c = sync.conditionMap.get(condition.name());
+
+ if (c == null)
+ throw new IllegalArgumentException();
+
+ return sync.hasWaiters(c);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getWaitQueueLength(IgniteCondition condition) {
+ try{
+ initializeReentrantLock();
+
+ AbstractQueuedSynchronizer.ConditionObject c = sync.conditionMap.get(condition.name());
+
+ if (c == null)
+ throw new IllegalArgumentException();
+
+ return sync.getWaitQueueLength(c);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ @Override public boolean isFailoverSafe() {
+ try{
+ initializeReentrantLock();
+
+ return sync.failoverSafe;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ @Override public boolean isFair() {
+ try{
+ initializeReentrantLock();
+
+ return sync.fair;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isBroken() {
+ try{
+ initializeReentrantLock();
+
+ return sync.isBroken();
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridCacheInternalKey key() {
+ return key;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean removed() {
+ return rmvd;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onRemoved() {
+ return rmvd = true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void needCheckNotRemoved() {
+
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeUTF(name);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ stash.set(in.readUTF());
+ }
+
+ /**
+ * Reconstructs object on unmarshalling.
+ *
+ * @return Reconstructed object.
+ * @throws ObjectStreamException Thrown in case of unmarshalling error.
+ */
+ private Object readResolve() throws ObjectStreamException {
+ String name = stash.get();
+
+ assert name != null;
+
+ try {
+ IgniteLock lock = IgnitionEx.localIgnite().context().dataStructures().reentrantLock(
+ name,
+ false,
+ false,
+ false);
+
+ if (lock == null)
+ throw new IllegalStateException("Lock was not found on deserialization: " + name);
+
+ return lock;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.withCause(new InvalidObjectException(e.getMessage()), e);
+ }
+ finally {
+ stash.remove();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ if (!rmvd) {
+ try {
+ boolean force = sync != null && (sync.isBroken() && !sync.failoverSafe);
+
+ ctx.kernalContext().dataStructures().removeReentrantLock(name, force);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridCacheLockImpl.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockState.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockState.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockState.java
new file mode 100644
index 0000000..3feb9bf
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockState.java
@@ -0,0 +1,353 @@
+/*
+ * 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.processors.datastructures;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.GridCacheInternal;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ * Grid cache reentrant lock state.
+ */
+public final class GridCacheLockState implements GridCacheInternal, Externalizable, Cloneable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Count. */
+ private int cnt;
+
+ /** Owner thread local ID. */
+ private long threadId;
+
+ /** Owner node ID. */
+ private UUID id;
+
+ /** FailoverSafe flag. */
+ private boolean failoverSafe;
+
+ /** Map containing state for each condition object associated with this lock. */
+ @GridToStringInclude
+ private Map<String, LinkedList<UUID>> conditionMap;
+
+ /** Map containing unprocessed signals for condition objects that are associated with this lock. */
+ @GridToStringInclude
+ private Map<UUID, LinkedList<String>> signals;
+
+ /** Flag indicating lock fairness. */
+ private boolean fair;
+
+ /** Queue containing nodes that are waiting to acquire this lock, used to ensure fairness. */
+ @GridToStringInclude
+ private LinkedList<UUID> nodes;
+
+ /**
+ * Flag indicating that global state changed.
+ * Used in fair mode to ensure that only successful acquires and releases trigger update.
+ */
+ private boolean changed;
+
+ /**
+ * Constructor.
+ *
+ * @param cnt Initial count.
+ * @param id UUID of owning node.
+ * @param threadID ID of the current thread.
+ * @param failoverSafe true if created in failoverSafe mode.
+ * @param fair true if created in fair mode.
+ */
+ public GridCacheLockState(int cnt, UUID id, long threadID, boolean failoverSafe, boolean fair) {
+ assert cnt >= 0;
+
+ this.id = id;
+
+ this.threadId = threadID;
+
+ conditionMap = new HashMap();
+
+ signals = null;
+
+ nodes = new LinkedList<UUID>();
+
+ this.fair = fair;
+
+ this.failoverSafe = failoverSafe;
+ }
+
+ /**
+ * Empty constructor required for {@link Externalizable}.
+ */
+ public GridCacheLockState() {
+ // No-op.
+ }
+
+ /**
+ * @param cnt New count.
+ */
+ public void set(int cnt) {
+ this.cnt = cnt;
+ }
+
+ /**
+ * @return Current count.
+ */
+ public int get() {
+ return cnt;
+ }
+
+ /**
+ * @return Current owner thread ID.
+ */
+ public long getThreadId() {
+ return threadId;
+ }
+
+ /**
+ * @param threadId New thread owner ID.
+ */
+ public void setThreadId(long threadId) {
+ this.threadId = threadId;
+ }
+
+ /**
+ * @return Current owner node ID.
+ */
+ public UUID getId() {
+ return id;
+ }
+
+ /**
+ * @return New owner node ID.
+ */
+ public void setId(UUID id) {
+ this.id = id;
+ }
+
+ /**
+ * @return Failover safe flag.
+ */
+ public boolean isFailoverSafe() {
+ return failoverSafe;
+ }
+
+ /**
+ * @return Condition count.
+ */
+ public int condtionCount(){
+ return conditionMap.size();
+ }
+
+ /**
+ * @return Condition map.
+ */
+ public Map<String, LinkedList<UUID>> getConditionMap() {
+ return conditionMap;
+ }
+
+ /**
+ * @param conditionMap Condition map.
+ */
+ public void setConditionMap(Map<String, LinkedList<UUID>> conditionMap) {
+ this.conditionMap = conditionMap;
+ }
+
+ /**
+ * @return Signals.
+ */
+ public Map<UUID, LinkedList<String>> getSignals() {
+ return signals;
+ }
+
+ /**
+ * @param signals Signals.
+ */
+ public void setSignals(Map<UUID, LinkedList<String>> signals) {
+ this.signals = signals;
+ }
+
+ /**
+ * @return Nodes.
+ */
+ public LinkedList<UUID> getNodes() {
+ return nodes;
+ }
+
+ /**
+ * @param nodes Nodes.
+ */
+ public void setNodes(LinkedList<UUID> nodes) {
+ this.nodes = nodes;
+ }
+
+ /**
+ * @return Fair flag.
+ */
+ public boolean isFair() {
+ return fair;
+ }
+
+ /**
+ * @return Changed flag.
+ */
+ public boolean isChanged() {
+ return changed;
+ }
+
+ /**
+ * @param changed Changed flag.
+ */
+ public void setChanged(boolean changed) {
+ this.changed = changed;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object clone() throws CloneNotSupportedException {
+ return super.clone();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeInt(cnt);
+ out.writeLong(threadId);
+ U.writeUuid(out, id);
+
+ out.writeBoolean(failoverSafe);
+
+ out.writeBoolean(fair);
+
+ out.writeBoolean(changed);
+
+ out.writeBoolean(conditionMap != null);
+
+ if (conditionMap != null) {
+ out.writeInt(conditionMap.size());
+
+ for (Map.Entry<String, LinkedList<UUID>> e : conditionMap.entrySet()) {
+ U.writeString(out, e.getKey());
+
+ out.writeInt(e.getValue().size());
+
+ for (UUID uuid:e.getValue())
+ U.writeUuid(out, uuid);
+ }
+ }
+
+ out.writeBoolean(signals != null);
+
+ if (signals != null) {
+ out.writeInt(signals.size());
+
+ for (Map.Entry<UUID, LinkedList<String>> e : signals.entrySet()) {
+ U.writeUuid(out, e.getKey());
+
+ out.writeInt(e.getValue().size());
+
+ for (String condition:e.getValue())
+ U.writeString(out, condition);
+ }
+ }
+
+ out.writeBoolean(nodes != null);
+
+ if (nodes != null) {
+ out.writeInt(nodes.size());
+
+ for (UUID uuid: nodes)
+ U.writeUuid(out, uuid);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException {
+ cnt = in.readInt();
+ threadId = in.readLong();
+ id = U.readUuid(in);
+
+ failoverSafe = in.readBoolean();
+
+ fair = in.readBoolean();
+
+ changed = in.readBoolean();
+
+ if (in.readBoolean()) {
+ int size = in.readInt();
+
+ conditionMap = U.newLinkedHashMap(size);
+
+ for (int i = 0; i < size; i++) {
+ String key = U.readString(in);
+
+ int size1 = in.readInt();
+
+ LinkedList<UUID> list = new LinkedList();
+
+ for (int j = 0; j < size1; j++)
+ list.add(U.readUuid(in));
+
+ conditionMap.put(key, list);
+ }
+ }
+
+ if (in.readBoolean()) {
+ assert (conditionMap != null);
+
+ int size = in.readInt();
+
+ signals = U.newLinkedHashMap(size);
+
+ for (int i = 0; i < size; i++) {
+ UUID node = U.readUuid(in);
+
+ int size1 = in.readInt();
+
+ LinkedList<String> list = new LinkedList();
+
+ for (int j = 0; j < size1; j++)
+ list.add(U.readString(in));
+
+ signals.put(node, list);
+ }
+ }
+ else
+ signals = null;
+
+ if (in.readBoolean()) {
+ int size = in.readInt();
+
+ nodes = new LinkedList();
+
+ for (int i = 0; i < size; i++)
+ nodes.add(U.readUuid(in));
+ }
+ else
+ nodes = null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridCacheLockState.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreImpl.java
index c365f9d..8f196be 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreImpl.java
@@ -19,8 +19,10 @@ package org.apache.ignite.internal.processors.datastructures;
import java.io.Externalizable;
import java.io.IOException;
+import java.io.InvalidObjectException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
+import java.io.ObjectStreamException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
@@ -32,7 +34,9 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgnitionEx;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
@@ -885,6 +889,35 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
t.set2(in.readUTF());
}
+ /**
+ * Reconstructs object on unmarshalling.
+ *
+ * @return Reconstructed object.
+ * @throws ObjectStreamException Thrown in case of unmarshalling error.
+ */
+ private Object readResolve() throws ObjectStreamException {
+ try {
+ IgniteBiTuple<GridKernalContext, String> t = stash.get();
+
+ IgniteSemaphore sem = IgnitionEx.localIgnite().context().dataStructures().semaphore(
+ t.get2(),
+ 0,
+ false,
+ false);
+
+ if (sem == null)
+ throw new IllegalStateException("Semaphore was not found on deserialization: " + t.get2());
+
+ return sem;
+ }
+ catch (IgniteCheckedException e) {
+ throw U.withCause(new InvalidObjectException(e.getMessage()), e);
+ }
+ finally {
+ stash.remove();
+ }
+ }
+
/** {@inheritDoc} */
@Override public void close() {
if (!rmvd) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index 034c314..9ae74a7 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -1037,6 +1037,8 @@ org.apache.ignite.internal.processors.datastructures.GridCacheQueueItemKey
org.apache.ignite.internal.processors.datastructures.GridCacheQueueProxy
org.apache.ignite.internal.processors.datastructures.GridCacheSemaphoreImpl
org.apache.ignite.internal.processors.datastructures.GridCacheSemaphoreState
+org.apache.ignite.internal.processors.datastructures.GridCacheLockImpl
+org.apache.ignite.internal.processors.datastructures.GridCacheLockState
org.apache.ignite.internal.processors.datastructures.GridCacheSetHeader
org.apache.ignite.internal.processors.datastructures.GridCacheSetHeaderKey
org.apache.ignite.internal.processors.datastructures.GridCacheSetImpl$CollocatedItemKey
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
index 4653ce9..4622fff 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAtomicsTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.IgniteAtomicSequence;
import org.apache.ignite.IgniteAtomicStamped;
import org.apache.ignite.IgniteClientDisconnectedException;
import org.apache.ignite.IgniteCountDownLatch;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
import org.apache.ignite.testframework.GridTestUtils;
@@ -774,4 +775,61 @@ public class IgniteClientReconnectAtomicsTest extends IgniteClientReconnectAbstr
assertFalse(srvSemaphore.tryAcquire());
assertFalse(srvSemaphore.tryAcquire());
}
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockReconnect() throws Exception {
+ testReentrantLockReconnect(false);
+
+ testReentrantLockReconnect(true);
+ }
+
+ private void testReentrantLockReconnect(final boolean fair) throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ Ignite srv = clientRouter(client);
+
+ IgniteLock clientLock = client.reentrantLock("lock1", true, fair, true);
+
+ assertEquals(false, clientLock.isLocked());
+
+ final IgniteLock srvLock = srv.reentrantLock("lock1", true, fair, true);
+
+ assertEquals(false, srvLock.isLocked());
+
+ reconnectClientNode(client, srv, new Runnable() {
+ @Override public void run() {
+ srvLock.lock();
+ }
+ });
+
+ assertTrue(srvLock.isLocked());
+ assertTrue(clientLock.isLocked());
+
+ assertEquals(1, srvLock.getHoldCount());
+
+ srvLock.lock();
+
+ assertTrue(srvLock.isLocked());
+ assertTrue(clientLock.isLocked());
+
+ assertEquals(2, srvLock.getHoldCount());
+
+ srvLock.unlock();
+
+ assertTrue(srvLock.isLocked());
+ assertTrue(clientLock.isLocked());
+
+ assertEquals(1, srvLock.getHoldCount());
+
+ srvLock.unlock();
+
+ assertFalse(srvLock.isLocked());
+ assertFalse(clientLock.isLocked());
+
+ assertEquals(0, srvLock.getHoldCount());
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractDataStructuresFailoverSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractDataStructuresFailoverSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractDataStructuresFailoverSelfTest.java
index 74023e9..6ba65ab 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractDataStructuresFailoverSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractDataStructuresFailoverSelfTest.java
@@ -33,9 +33,11 @@ import org.apache.ignite.IgniteAtomicLong;
import org.apache.ignite.IgniteAtomicReference;
import org.apache.ignite.IgniteAtomicSequence;
import org.apache.ignite.IgniteAtomicStamped;
+import org.apache.ignite.IgniteCompute;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteQueue;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.cache.CacheMode;
@@ -75,10 +77,10 @@ public abstract class GridCacheAbstractDataStructuresFailoverSelfTest extends Ig
private static final String TRANSACTIONAL_CACHE_NAME = "tx_cache";
/** */
- private static final int TOP_CHANGE_CNT = 5;
+ private static final int TOP_CHANGE_CNT = 2;
/** */
- private static final int TOP_CHANGE_THREAD_CNT = 3;
+ private static final int TOP_CHANGE_THREAD_CNT = 2;
/** */
private boolean client;
@@ -586,6 +588,208 @@ public abstract class GridCacheAbstractDataStructuresFailoverSelfTest extends Ig
/**
* @throws Exception If failed.
*/
+ public void testReentrantLockFailsWhenServersLeft() throws Exception {
+ testReentrantLockFailsWhenServersLeft(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testFairReentrantLockFailsWhenServersLeft() throws Exception {
+ testReentrantLockFailsWhenServersLeft(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockFailsWhenServersLeft(final boolean fair) throws Exception {
+ client = true;
+
+ Ignite client = startGrid(gridCount());
+
+ Ignite server = grid(0);
+
+ // Initialize lock.
+ IgniteLock srvLock = server.reentrantLock("lock", true, fair, true);
+
+ IgniteSemaphore semaphore = server.semaphore("sync", 0, true, true);
+
+ IgniteCompute compute = client.compute().withAsync();
+
+ compute.apply(new IgniteClosure<Ignite, Object>() {
+ @Override public Object apply(Ignite ignite) {
+ final IgniteLock l = ignite.reentrantLock("lock", true, fair, true);
+
+ l.lock();
+
+ assertTrue(l.isHeldByCurrentThread());
+
+ l.unlock();
+
+ assertFalse(l.isHeldByCurrentThread());
+
+ // Signal the server to go down.
+ ignite.semaphore("sync", 0, true, true).release();
+
+ boolean isExceptionThrown = false;
+
+ try {
+ // Wait for the server to go down.
+ Thread.sleep(1000);
+
+ l.lock();
+
+ fail("Exception must be thrown.");
+ }
+ catch (InterruptedException e) {
+ fail("Interrupted exception not expected here.");
+ }
+ catch (IgniteException e) {
+ isExceptionThrown = true;
+ }
+ finally {
+ assertTrue(isExceptionThrown);
+
+ assertFalse(l.isHeldByCurrentThread());
+ }
+ return null;
+ }
+ }, client);
+
+ // Wait for the lock on client to be acquired then released.
+ semaphore.acquire();
+
+ for (int i = 0; i < gridCount(); i++)
+ stopGrid(i);
+
+ compute.future().get();
+
+ client.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockConstantTopologyChangeFailoverSafe() throws Exception {
+ doTestReentrantLock(new ConstantTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockConstantMultipleTopologyChangeFailoverSafe() throws Exception {
+ doTestReentrantLock(multipleTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockConstantTopologyChangeNonFailoverSafe() throws Exception {
+ doTestReentrantLock(new ConstantTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), false, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReentrantLockConstantMultipleTopologyChangeNonFailoverSafe() throws Exception {
+ doTestReentrantLock(multipleTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), false, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testFairReentrantLockConstantTopologyChangeFailoverSafe() throws Exception {
+ doTestReentrantLock(new ConstantTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testFairReentrantLockConstantMultipleTopologyChangeFailoverSafe() throws Exception {
+ doTestReentrantLock(multipleTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), true, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testFairReentrantLockConstantTopologyChangeNonFailoverSafe() throws Exception {
+ doTestReentrantLock(new ConstantTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), false, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testFairReentrantLockConstantMultipleTopologyChangeNonFailoverSafe() throws Exception {
+ doTestReentrantLock(multipleTopologyChangeWorker(TOP_CHANGE_THREAD_CNT), false, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void doTestReentrantLock(ConstantTopologyChangeWorker topWorker, final boolean failoverSafe, final boolean fair) throws Exception {
+ try (IgniteLock lock = grid(0).reentrantLock(STRUCTURE_NAME, failoverSafe, fair, true)) {
+ IgniteInternalFuture<?> fut = topWorker.startChangingTopology(new IgniteClosure<Ignite, Object>() {
+ @Override public Object apply(Ignite ignite) {
+ final IgniteLock l = ignite.reentrantLock(STRUCTURE_NAME, failoverSafe, fair, false);
+
+ final AtomicBoolean done = new AtomicBoolean(false);
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try{
+ l.lock();
+ }
+ finally {
+ done.set(true);
+ }
+
+ return null;
+ }
+ });
+
+ // Wait until l.lock() has been called.
+ while(!l.hasQueuedThreads() && !done.get()){
+ // No-op.
+ }
+
+ return null;
+ }
+ });
+
+ while (!fut.isDone()) {
+ while (true) {
+ try {
+ lock.lock();
+ }
+ catch (IgniteException e) {
+ // Exception may happen in non-failoversafe mode.
+ if (failoverSafe)
+ throw e;
+ }
+ finally {
+ // Broken lock cannot be used in non-failoversafe mode.
+ if(!lock.isBroken() || failoverSafe) {
+ assertTrue(lock.isHeldByCurrentThread());
+
+ lock.unlock();
+
+ assertFalse(lock.isHeldByCurrentThread());
+ }
+ break;
+ }
+ }
+ }
+
+ fut.get();
+
+ for (Ignite g : G.allGrids())
+ assertFalse(g.reentrantLock(STRUCTURE_NAME, failoverSafe, fair, false).isHeldByCurrentThread());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testCountDownLatchConstantTopologyChange() throws Exception {
doTestCountDownLatch(new ConstantTopologyChangeWorker(TOP_CHANGE_THREAD_CNT));
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteClientDataStructuresAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteClientDataStructuresAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteClientDataStructuresAbstractTest.java
index 34e7080..5929d42 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteClientDataStructuresAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteClientDataStructuresAbstractTest.java
@@ -18,11 +18,13 @@
package org.apache.ignite.internal.processors.cache.datastructures;
import java.util.concurrent.Callable;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteAtomicLong;
import org.apache.ignite.IgniteAtomicSequence;
import org.apache.ignite.IgniteCountDownLatch;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteQueue;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteSet;
@@ -327,6 +329,74 @@ public abstract class IgniteClientDataStructuresAbstractTest extends GridCommonA
/**
* @throws Exception If failed.
*/
+ public void testReentrantLock() throws Exception {
+ Ignite clientNode = clientIgnite();
+ Ignite srvNode = serverNode();
+
+ testReentrantLock(clientNode, srvNode);
+ testReentrantLock(srvNode, clientNode);
+ }
+
+ /**
+ * @param creator Creator node.
+ * @param other Other node.
+ * @throws Exception If failed.
+ */
+ private void testReentrantLock(Ignite creator, final Ignite other) throws Exception {
+ assertNull(creator.reentrantLock("lock1", true, false, false));
+ assertNull(other.reentrantLock("lock1", true, false, false));
+
+ try (IgniteLock lock = creator.reentrantLock("lock1", true, false, true)) {
+ assertNotNull(lock);
+
+ assertFalse(lock.isLocked());
+
+ final Semaphore semaphore = new Semaphore(0);
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ IgniteLock lock0 = other.reentrantLock("lock1", true, false, false);
+
+ lock0.lock();
+
+ assertTrue(lock0.isLocked());
+
+ semaphore.release();
+
+ U.sleep(1000);
+
+ log.info("Release reentrant lock.");
+
+ lock0.unlock();
+
+ return null;
+ }
+ });
+
+ semaphore.acquire();
+
+ log.info("Try acquire lock.");
+
+ assertTrue(lock.tryLock(5000, TimeUnit.MILLISECONDS));
+
+ log.info("Finished wait.");
+
+ fut.get();
+
+ assertTrue(lock.isLocked());
+
+ lock.unlock();
+
+ assertFalse(lock.isLocked());
+ }
+
+ assertNull(creator.reentrantLock("lock1", true, false, false));
+ assertNull(other.reentrantLock("lock1", true, false, false));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testQueue() throws Exception {
Ignite clientNode = clientIgnite();
Ignite srvNode = serverNode();
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteDataStructureUniqueNameTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteDataStructureUniqueNameTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteDataStructureUniqueNameTest.java
index e88c97b..a8e0095 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteDataStructureUniqueNameTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteDataStructureUniqueNameTest.java
@@ -31,6 +31,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteSet;
import org.apache.ignite.cache.CacheAtomicityMode;
@@ -241,7 +242,7 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
private void testUniqueName(final boolean singleGrid) throws Exception {
final String name = IgniteUuid.randomUuid().toString();
- final int DS_TYPES = 8;
+ final int DS_TYPES = 9;
final int THREADS = DS_TYPES * 3;
@@ -322,6 +323,13 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
res = ignite.semaphore(name, 0, false, true);
break;
+
+ case 8:
+ log.info("Create atomic reentrant lock, grid: " + ignite.name());
+
+ res = ignite.reentrantLock(name, true, true, true);
+
+ break;
default:
fail();
@@ -361,7 +369,8 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
res instanceof IgniteCountDownLatch ||
res instanceof IgniteQueue ||
res instanceof IgniteSet ||
- res instanceof IgniteSemaphore);
+ res instanceof IgniteSemaphore ||
+ res instanceof IgniteLock);
log.info("Data structure created: " + dataStructure);
[3/3] ignite git commit: IGNITE-642 Implement IgniteReentrantLock
data structure
Posted by yz...@apache.org.
IGNITE-642 Implement IgniteReentrantLock data structure
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f9bd9a33
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f9bd9a33
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f9bd9a33
Branch: refs/heads/master
Commit: f9bd9a3383c1a914970a1d0a391b345cc584430d
Parents: 2226377
Author: Vladisav Jelisavcic <vl...@gmail.com>
Authored: Wed Apr 27 18:31:26 2016 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Wed Apr 27 18:31:26 2016 +0300
----------------------------------------------------------------------
.../datastructures/IgniteLockExample.java | 293 ++++
.../ignite/examples/CacheExamplesSelfTest.java | 8 +
.../src/main/java/org/apache/ignite/Ignite.java | 17 +
.../java/org/apache/ignite/IgniteCondition.java | 338 ++++
.../main/java/org/apache/ignite/IgniteLock.java | 489 ++++++
.../ignite/internal/GridKernalContextImpl.java | 4 +-
.../apache/ignite/internal/IgniteKernal.java | 21 +
.../datastructures/DataStructuresProcessor.java | 158 +-
.../datastructures/GridCacheLockEx.java | 52 +
.../datastructures/GridCacheLockImpl.java | 1538 +++++++++++++++++
.../datastructures/GridCacheLockState.java | 353 ++++
.../datastructures/GridCacheSemaphoreImpl.java | 33 +
.../resources/META-INF/classnames.properties | 2 +
.../IgniteClientReconnectAtomicsTest.java | 58 +
...eAbstractDataStructuresFailoverSelfTest.java | 208 ++-
.../IgniteClientDataStructuresAbstractTest.java | 70 +
.../IgniteDataStructureUniqueNameTest.java | 13 +-
.../IgniteLockAbstractSelfTest.java | 1629 ++++++++++++++++++
.../IgniteSemaphoreAbstractSelfTest.java | 31 +
.../local/IgniteLocalLockSelfTest.java | 110 ++
.../IgnitePartitionedLockSelfTest.java | 33 +
.../IgniteReplicatedLockSelfTest.java | 33 +
.../distributed/GridCacheLockAbstractTest.java | 2 +-
.../cache/GridCacheDataStructuresLoadTest.java | 53 +
.../ignite/testframework/junits/IgniteMock.java | 10 +
.../junits/multijvm/IgniteProcessProxy.java | 7 +
.../IgniteCacheDataStructuresSelfTestSuite.java | 6 +
.../org/apache/ignite/IgniteSpringBean.java | 11 +
28 files changed, 5570 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteLockExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteLockExample.java b/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteLockExample.java
new file mode 100644
index 0000000..1f84787
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteLockExample.java
@@ -0,0 +1,293 @@
+/*
+ * 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.examples.datastructures;
+
+import java.util.UUID;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCondition;
+import org.apache.ignite.IgniteLock;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.examples.ExampleNodeStartup;
+import org.apache.ignite.lang.IgniteRunnable;
+
+/**
+ * This example demonstrates cache based reentrant lock.
+ * <p>
+ * Remote nodes should always be started with special configuration
+ * file which enables P2P class loading: {@code 'ignite.{sh|bat} examples/config/example-ignite.xml'}.
+ * <p>
+ * Alternatively you can run {@link ExampleNodeStartup} in another JVM which will start node with {@code
+ * examples/config/example-ignite.xml} configuration.
+ */
+public class IgniteLockExample {
+ /** Number of items for each producer/consumer to produce/consume. */
+ private static final int OPS_COUNT = 100;
+
+ /** Number of producers. */
+ private static final int NUM_PRODUCERS = 5;
+
+ /** Number of consumers. */
+ private static final int NUM_CONSUMERS = 5;
+
+ /** Cache name. */
+ private static final String CACHE_NAME = "cache";
+
+ /** Name of the global resource. */
+ private static final String QUEUE_ID = "queue";
+
+ /** Name of the synchronization variable. */
+ private static final String SYNC_NAME = "done";
+
+ /** Name of the condition object. */
+ private static final String NOT_FULL = "notFull";
+
+ /** Name of the condition object. */
+ private static final String NOT_EMPTY = "notEmpty";
+
+ /**
+ * Executes example.
+ *
+ * @param args Command line arguments, none required.
+ */
+ public static void main(String[] args) {
+ try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
+ System.out.println();
+ System.out.println(">>> Cache atomic reentrant lock example started.");
+
+ // Make name of reentrant lock.
+ final String reentrantLockName = UUID.randomUUID().toString();
+
+ // Initialize lock.
+ IgniteLock lock = ignite.reentrantLock(reentrantLockName, true, false, true);
+
+ // Init distributed cache.
+ IgniteCache<String, Integer> cache = ignite.getOrCreateCache(CACHE_NAME);
+
+ // Init shared variable.
+ cache.put(QUEUE_ID, 0);
+
+ // Shared variable indicating number of jobs left to be completed.
+ cache.put(SYNC_NAME, NUM_PRODUCERS + NUM_CONSUMERS);
+
+ // Start consumers on all cluster nodes.
+ for (int i = 0; i < NUM_CONSUMERS; i++)
+ ignite.compute().withAsync().run(new Consumer(reentrantLockName));
+
+ // Start producers on all cluster nodes.
+ for (int i = 0; i < NUM_PRODUCERS; i++)
+ ignite.compute().withAsync().run(new Producer(reentrantLockName));
+
+ System.out.println("Master node is waiting for all other nodes to finish...");
+
+ // Wait for everyone to finish.
+ try {
+ lock.lock();
+
+ IgniteCondition notDone = lock.getOrCreateCondition(SYNC_NAME);
+
+ int count = cache.get(SYNC_NAME);
+
+ while(count > 0) {
+ notDone.await();
+
+ count = cache.get(SYNC_NAME);
+ }
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+
+ System.out.flush();
+ System.out.println();
+ System.out.println("Finished reentrant lock example...");
+ System.out.println("Check all nodes for output (this node is also part of the cluster).");
+ }
+
+ /**
+ * Closure which simply acquires reentrant lock.
+ */
+ private abstract static class ReentrantLockExampleClosure implements IgniteRunnable {
+ /** Semaphore name. */
+ protected final String reentrantLockName;
+
+ /**
+ * @param reentrantLockName Reentrant lock name.
+ */
+ ReentrantLockExampleClosure(String reentrantLockName) {
+ this.reentrantLockName = reentrantLockName;
+ }
+ }
+
+ /**
+ * Closure which simulates producer.
+ */
+ private static class Producer extends ReentrantLockExampleClosure {
+ /**
+ * @param reentrantLockName Reentrant lock name.
+ */
+ public Producer(String reentrantLockName) {
+ super(reentrantLockName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void run() {
+ System.out.println("Producer started. ");
+
+ IgniteLock lock = Ignition.ignite().reentrantLock(reentrantLockName, true, false, true);
+
+ // Condition to wait on when queue is full.
+ IgniteCondition notFull = lock.getOrCreateCondition(NOT_FULL);
+
+ // Signaled to wait on when queue is empty.
+ IgniteCondition notEmpty = lock.getOrCreateCondition(NOT_EMPTY);
+
+ // Signaled when job is done.
+ IgniteCondition done = lock.getOrCreateCondition(SYNC_NAME);
+
+ IgniteCache<String, Integer> cache = Ignition.ignite().cache(CACHE_NAME);
+
+ for (int i = 0; i < OPS_COUNT; i++) {
+ try {
+ lock.lock();
+
+ int val = cache.get(QUEUE_ID);
+
+ while(val >= 100){
+ System.out.println("Queue is full. Producer [nodeId=" + Ignition.ignite().cluster().localNode().id() +
+ " paused.");
+
+ notFull.await();
+
+ val = cache.get(QUEUE_ID);
+ }
+
+ val++;
+
+ System.out.println("Producer [nodeId=" + Ignition.ignite().cluster().localNode().id() +
+ ", available=" + val + ']');
+
+ cache.put(QUEUE_ID, val);
+
+ notEmpty.signalAll();
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+
+ System.out.println("Producer finished [nodeId=" + Ignition.ignite().cluster().localNode().id() + ']');
+
+ try {
+ lock.lock();
+
+ int count = cache.get(SYNC_NAME);
+
+ count--;
+
+ cache.put(SYNC_NAME, count);
+
+ // Signals the master thread.
+ done.signal();
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ }
+
+ /**
+ * Closure which simulates consumer.
+ */
+ private static class Consumer extends ReentrantLockExampleClosure {
+ /**
+ * @param reentrantLockName ReentrantLock name.
+ */
+ public Consumer(String reentrantLockName) {
+ super(reentrantLockName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void run() {
+ System.out.println("Consumer started. ");
+
+ Ignite g = Ignition.ignite();
+
+ IgniteLock lock = g.reentrantLock(reentrantLockName, true, false, true);
+
+ // Condition to wait on when queue is full.
+ IgniteCondition notFull = lock.getOrCreateCondition(NOT_FULL);
+
+ // Signaled to wait on when queue is empty.
+ IgniteCondition notEmpty = lock.getOrCreateCondition(NOT_EMPTY);
+
+ // Signaled when job is done.
+ IgniteCondition done = lock.getOrCreateCondition(SYNC_NAME);
+
+ IgniteCache<String, Integer> cache = g.cache(CACHE_NAME);
+
+ for (int i = 0; i < OPS_COUNT; i++) {
+ try {
+ lock.lock();
+
+ int val = cache.get(QUEUE_ID);
+
+ while (val <= 0) {
+ System.out.println("Queue is empty. Consumer [nodeId=" +
+ Ignition.ignite().cluster().localNode().id() + " paused.");
+
+ notEmpty.await();
+
+ val = cache.get(QUEUE_ID);
+ }
+
+ val--;
+
+ System.out.println("Consumer [nodeId=" + Ignition.ignite().cluster().localNode().id() +
+ ", available=" + val + ']');
+
+ cache.put(QUEUE_ID, val);
+
+ notFull.signalAll();
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+
+ System.out.println("Consumer finished [nodeId=" + Ignition.ignite().cluster().localNode().id() + ']');
+
+ try {
+ lock.lock();
+
+ int count = cache.get(SYNC_NAME);
+
+ count--;
+
+ cache.put(SYNC_NAME, count);
+
+ // Signals the master thread.
+ done.signal();
+ }
+ finally {
+ lock.unlock();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/examples/src/test/java/org/apache/ignite/examples/CacheExamplesSelfTest.java
----------------------------------------------------------------------
diff --git a/examples/src/test/java/org/apache/ignite/examples/CacheExamplesSelfTest.java b/examples/src/test/java/org/apache/ignite/examples/CacheExamplesSelfTest.java
index 541291b..43b05b5 100644
--- a/examples/src/test/java/org/apache/ignite/examples/CacheExamplesSelfTest.java
+++ b/examples/src/test/java/org/apache/ignite/examples/CacheExamplesSelfTest.java
@@ -33,6 +33,7 @@ import org.apache.ignite.examples.datastructures.IgniteAtomicSequenceExample;
import org.apache.ignite.examples.datastructures.IgniteAtomicStampedExample;
import org.apache.ignite.examples.datastructures.IgniteCountDownLatchExample;
import org.apache.ignite.examples.datastructures.IgniteQueueExample;
+import org.apache.ignite.examples.datastructures.IgniteLockExample;
import org.apache.ignite.examples.datastructures.IgniteSemaphoreExample;
import org.apache.ignite.examples.datastructures.IgniteSetExample;
import org.apache.ignite.testframework.junits.common.GridAbstractExamplesTest;
@@ -100,6 +101,13 @@ public class CacheExamplesSelfTest extends GridAbstractExamplesTest {
/**
* @throws Exception If failed.
*/
+ public void testCacheLockExample() throws Exception {
+ IgniteLockExample.main(EMPTY_ARGS);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testCacheQueueExample() throws Exception {
IgniteQueueExample.main(EMPTY_ARGS);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/Ignite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/Ignite.java b/modules/core/src/main/java/org/apache/ignite/Ignite.java
index 5703744..b62672e 100644
--- a/modules/core/src/main/java/org/apache/ignite/Ignite.java
+++ b/modules/core/src/main/java/org/apache/ignite/Ignite.java
@@ -466,6 +466,23 @@ public interface Ignite extends AutoCloseable {
throws IgniteException;
/**
+ * Gets or creates reentrant lock. If reentrant lock is not found in cache and {@code create} flag
+ * is {@code true}, it is created using provided name.
+ *
+ * @param name Name of the lock.
+ * @param failoverSafe {@code True} to create failover safe lock which means that
+ * if any node leaves topology, all locks already acquired by that node are silently released
+ * and become available for other nodes to acquire. If flag is {@code false} then
+ * all threads on other nodes waiting to acquire lock are interrupted.
+ * @param fair If {@code True}, fair lock will be created.
+ * @param create Boolean flag indicating whether data structure should be created if does not exist.
+ * @return ReentrantLock for the given name.
+ * @throws IgniteException If reentrant lock could not be fetched or created.
+ */
+ public IgniteLock reentrantLock(String name, boolean failoverSafe, boolean fair, boolean create)
+ throws IgniteException;
+
+ /**
* Will get a named queue from cache and create one if it has not been created yet and {@code cfg} is not
* {@code null}.
* If queue is present already, queue properties will not be changed. Use
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/IgniteCondition.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteCondition.java b/modules/core/src/main/java/org/apache/ignite/IgniteCondition.java
new file mode 100644
index 0000000..020f23a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteCondition.java
@@ -0,0 +1,338 @@
+/*
+ * 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;
+
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+
+/**
+ * This interface provides a rich API for working with condition objects
+ * associated with distributed reentrant locks.
+ * <p>
+ * <h1 class="header">Functionality</h1>
+ * IgniteCondition provides functionality similar to {@code java.util.concurrent.locks.Condition}.
+ */
+public interface IgniteCondition extends Condition {
+ /**
+ * Name of ignite condition.
+ *
+ * @return Name of ignite condition.
+ */
+ public String name();
+
+ /**
+ * Causes the current thread to wait until it is signalled or
+ * {@linkplain Thread#interrupt interrupted}.
+ *
+ * <p>The lock associated with this {@code IgniteCondition} is atomically
+ * released and the current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until <em>one</em> of six things happens:
+ * <ul>
+ * <li>Some other thread (on any node) invokes the {@link #signal} method for this
+ * {@code Condition} and the current thread happens to be chosen as the
+ * thread to be awakened; or
+ * <li>Some other thread (on any node) invokes the {@link #signalAll} method for this
+ * {@code Condition}; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts} the
+ * current thread, and interruption of thread suspension is supported; or
+ * <li>Some other node in grid fails, and lock is created in non-failoverSafe mode; or
+ * <li>Local node is stopped; or
+ * <li>A "<em>spurious wakeup</em>" occurs.
+ * </ul>
+ *
+ * <p>If lock is not broken (because of failure of lock owner node)
+ * in non-failoverSafe mode and local node is alive,
+ * before this method can return the current thread must
+ * re-acquire the lock associated with this condition. In all other cases
+ * when the thread returns it is <em>guaranteed</em> to hold this lock.
+ *
+ * <p>If the current thread:
+ * <ul>
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting
+ * and interruption of thread suspension is supported,
+ * </ul>
+ * then {@link IgniteInterruptedException} is thrown and the current thread's
+ * interrupted status is cleared. It is not specified, in the first
+ * case, whether or not the test for interruption occurs before the lock
+ * is released.
+ *
+ * <p><b>Implementation Considerations</b>
+ *
+ * <p>The current thread is assumed to hold the lock associated with this
+ * {@code Condition} when this method is called. If not, an {@link IllegalMonitorStateException}
+ * will be thrown.
+ *
+ * @throws IgniteInterruptedException if the current thread is interrupted
+ * @throws IgniteException if the node stopped, or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override void await() throws IgniteInterruptedException, IgniteException;
+
+ /**
+ * Causes the current thread to wait until it is signalled.
+ *
+ * <p>The lock associated with this condition is atomically
+ * released and the current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until <em>one</em> of five things happens:
+ * <ul>
+ * <li>Some other thread invokes the {@link #signal} method for this
+ * {@code Condition} and the current thread happens to be chosen as the
+ * thread to be awakened; or
+ * <li>Some other thread invokes the {@link #signalAll} method for this
+ * {@code Condition}; or
+ * <li>Some other node in grid fails, and lock is created in non-failoverSafe mode; or
+ * <li>Local node is stopped; or
+ * <li>A "<em>spurious wakeup</em>" occurs.
+ * </ul>
+ *
+ * <p>If lock is not broken (because of failure of lock owner node)
+ * in non-failoverSafe mode and local node is alive,
+ * before this method can return the current thread must
+ * re-acquire the lock associated with this condition. In all other cases,
+ * when the thread returns it is <em>guaranteed</em> to hold this lock.
+ *
+ * <p>If the current thread's interrupted status is set when it enters
+ * this method, or it is {@linkplain Thread#interrupt interrupted}
+ * while waiting, it will continue to wait until signalled. When it finally
+ * returns from this method its interrupted status will still
+ * be set.
+ *
+ * <p><b>Implementation Considerations</b>
+ *
+ * <p>The current thread is assumed to hold the lock associated with this
+ * {@code Condition} when this method is called. If not, an {@link IllegalMonitorStateException}
+ * will be thrown.
+ *
+ * @throws IgniteException if the node stopped, or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override void awaitUninterruptibly() throws IgniteException;
+
+ /**
+ * Causes the current thread to wait until it is signalled or interrupted,
+ * or the specified waiting time elapses.
+ *
+ * <p>The lock associated with this condition is atomically
+ * released and the current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until <em>one</em> of seven things happens:
+ * <ul>
+ * <li>Some other thread invokes the {@link #signal} method for this
+ * {@code Condition} and the current thread happens to be chosen as the
+ * thread to be awakened; or
+ * <li>Some other thread invokes the {@link #signalAll} method for this
+ * {@code Condition}; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts} the
+ * current thread, and interruption of thread suspension is supported; or
+ * <li>The specified waiting time elapses; or
+ * <li>Some other node in grid fails, and lock is created in non-failoverSafe mode; or
+ * <li>Local node is stopped; or
+ * <li>A "<em>spurious wakeup</em>" occurs.
+ * </ul>
+ *
+ * <p>If lock is not broken (because of failure of lock owner node)
+ * in non-failoverSafe mode and local node is alive,
+ * before this method can return the current thread must
+ * re-acquire the lock associated with this condition. When the
+ * thread returns it is <em>guaranteed</em> to hold this lock.
+ *
+ * <p>If the current thread:
+ * <ul>
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting
+ * and interruption of thread suspension is supported,
+ * </ul>
+ * then {@link IgniteInterruptedException} is thrown and the current thread's
+ * interrupted status is cleared. It is not specified, in the first
+ * case, whether or not the test for interruption occurs before the lock
+ * is released.
+ *
+ * <p>The method returns an estimate of the number of nanoseconds
+ * remaining to wait given the supplied {@code nanosTimeout}
+ * value upon return, or a value less than or equal to zero if it
+ * timed out. This value can be used to determine whether and how
+ * long to re-wait in cases where the wait returns but an awaited
+ * condition still does not hold. Typical uses of this method take
+ * the following form:
+ *
+ * <pre> {@code
+ * boolean aMethod(long timeout, TimeUnit unit) {
+ * long nanos = unit.toNanos(timeout);
+ * lock.lock();
+ * try {
+ * while (!conditionBeingWaitedFor()) {
+ * if (nanos <= 0L)
+ * return false;
+ * nanos = theCondition.awaitNanos(nanos);
+ * }
+ * // ...
+ * } finally {
+ * lock.unlock();
+ * }
+ * }}</pre>
+ *
+ * <p>Design note: This method requires a nanosecond argument so
+ * as to avoid truncation errors in reporting remaining times.
+ * Such precision loss would make it difficult for programmers to
+ * ensure that total waiting times are not systematically shorter
+ * than specified when re-waits occur.
+ *
+ * <p><b>Implementation Considerations</b>
+ *
+ * <p>The current thread is assumed to hold the lock associated with this
+ * {@code Condition} when this method is called. If not, an {@link IllegalMonitorStateException}
+ * will be thrown.
+ *
+ * @param nanosTimeout the maximum time to wait, in nanoseconds
+ * @return an estimate of the {@code nanosTimeout} value minus
+ * the time spent waiting upon return from this method.
+ * A positive value may be used as the argument to a
+ * subsequent call to this method to finish waiting out
+ * the desired time. A value less than or equal to zero
+ * indicates that no time remains.
+ * @throws IgniteInterruptedException if the current thread is interrupted
+ * @throws IgniteException if the node stopped, or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override long awaitNanos(long nanosTimeout) throws IgniteInterruptedException, IgniteException;
+
+ /**
+ * Causes the current thread to wait until it is signalled or interrupted,
+ * or the specified waiting time elapses. This method is behaviorally
+ * equivalent to:
+ * <pre> {@code awaitNanos(unit.toNanos(time)) > 0}</pre>
+ *
+ * @param time the maximum time to wait
+ * @param unit the time unit of the {@code time} argument
+ * @return {@code false} if the waiting time detectably elapsed
+ * before return from the method, else {@code true}
+ * @throws IgniteInterruptedException if the current thread is interrupted
+ * @throws IgniteException if the node stopped, or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override boolean await(long time, TimeUnit unit) throws IgniteInterruptedException, IgniteException;
+
+ /**
+ * Causes the current thread to wait until it is signalled or interrupted,
+ * or the specified deadline elapses.
+ *
+ * <p>The lock associated with this condition is atomically
+ * released and the current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until <em>one</em> of seven things happens:
+ * <ul>
+ * <li>Some other thread invokes the {@link #signal} method for this
+ * {@code Condition} and the current thread happens to be chosen as the
+ * thread to be awakened; or
+ * <li>Some other thread invokes the {@link #signalAll} method for this
+ * {@code Condition}; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts} the
+ * current thread, and interruption of thread suspension is supported; or
+ * <li>Some other node in grid fails, and lock is created in non-failoverSafe mode; or
+ * <li>Local node is stopped; or
+ * <li>The specified deadline elapses; or
+ * <li>A "<em>spurious wakeup</em>" occurs.
+ * </ul>
+ *
+ * <p>If lock is not broken (because of failure of lock owner node)
+ * in non-failoverSafe mode and local node is alive,
+ * before this method can return the current thread must
+ * re-acquire the lock associated with this condition. When the
+ * thread returns it is <em>guaranteed</em> to hold this lock.
+ *
+ * <p>If the current thread:
+ * <ul>
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting
+ * and interruption of thread suspension is supported,
+ * </ul>
+ * then {@link IgniteInterruptedException} is thrown and the current thread's
+ * interrupted status is cleared. It is not specified, in the first
+ * case, whether or not the test for interruption occurs before the lock
+ * is released.
+ *
+ * <p>The return value indicates whether the deadline has elapsed,
+ * which can be used as follows:
+ * <pre> {@code
+ * boolean aMethod(Date deadline) {
+ * boolean stillWaiting = true;
+ * lock.lock();
+ * try {
+ * while (!conditionBeingWaitedFor()) {
+ * if (!stillWaiting)
+ * return false;
+ * stillWaiting = theCondition.awaitUntil(deadline);
+ * }
+ * // ...
+ * } finally {
+ * lock.unlock();
+ * }
+ * }}</pre>
+ *
+ * <p><b>Implementation Considerations</b>
+ *
+ * <p>The current thread is assumed to hold the lock associated with this
+ * {@code Condition} when this method is called. If not, an {@link IllegalMonitorStateException}
+ * will be thrown.
+ *
+ * @param deadline the absolute time to wait until
+ * @return {@code false} if the deadline has elapsed upon return, else
+ * {@code true}
+ * @throws IgniteInterruptedException if the current thread is interrupted
+ * (and interruption of thread suspension is supported)
+ * @throws IgniteException if the node stopped, or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override boolean awaitUntil(Date deadline) throws IgniteInterruptedException, IgniteException;
+
+ /**
+ * Wakes up one waiting thread.
+ *
+ * <p>If any threads are waiting on this condition then one
+ * is selected for waking up. That thread must then re-acquire the
+ * lock before returning from {@code await}.
+ *
+ * <p><b>Implementation Considerations</b>
+ *
+ * <p>The current thread is assumed to hold the lock associated with this
+ * {@code Condition} when this method is called. If not, an {@link IllegalMonitorStateException}
+ * will be thrown.
+ *
+ * @throws IgniteException if node is stopped or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override void signal() throws IgniteException;
+
+ /**
+ * Wakes up all waiting threads.
+ *
+ * <p>If any threads are waiting on this condition then they are
+ * all woken up. Each thread must re-acquire the lock before it can
+ * return from {@code await}.
+ *
+ * <p><b>Implementation Considerations</b>
+ *
+ * <p>The current thread is assumed to hold the lock associated with this
+ * {@code Condition} when this method is called. If not, an {@link IllegalMonitorStateException}
+ * will be thrown.
+ *
+ * @throws IgniteException if node is stopped or
+ * node owning the lock failed in non-failoversafe mode
+ */
+ @Override void signalAll() throws IgniteException;
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/IgniteLock.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteLock.java b/modules/core/src/main/java/org/apache/ignite/IgniteLock.java
new file mode 100644
index 0000000..2b3ad3d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteLock.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;
+
+import java.io.Closeable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+
+/**
+ * This interface provides a rich API for working with distributed reentrant locks.
+ * <p>
+ * <h1 class="header">Functionality</h1>
+ * Distributed reentrant lock provides functionality similar to {@code java.util.concurrent.ReentrantLock}.
+ * <h1 class="header">Creating Distributed ReentrantLock</h1>
+ * Instance of cache reentrant lock can be created by calling the following method:
+ * {@link Ignite#reentrantLock(String, boolean, boolean, boolean)}.
+ * <h1 class="header">Protection from failover</h1>
+ * Ignite lock can automatically recover from node failure.
+ * <ul>
+ * <li>If failoverSafe flag is set to true upon creation,
+ * in case a node owning the lock fails, lock will be automatically released and become available for threads on other
+ * nodes to acquire. No exception will be thrown.
+ * <li>If failoverSafe flag is set to false upon creation,
+ * in case a node owning the lock fails, {@code IgniteException} will be thrown on every other node attempting to
+ * perform any operation on this lock. No automatic recovery will be attempted,
+ * and lock will be marked as broken (i.e. unusable), which can be checked using the method #isBroken().
+ * Broken lock cannot be reused again.
+ * </ul>
+ *
+ * <h1 class="header">Implementation issues</h1>
+ * Ignite lock comes in two flavours: fair and non-fair. Non-fair lock assumes no ordering should be imposed
+ * on acquiring threads; in case of contention, threads from all nodes compete for the lock once the lock is released.
+ * In most cases this is the desired behaviour. However, in some cases, using the non-fair lock can lead to uneven load
+ * distribution among nodes.
+ * Fair lock solves this issue by imposing strict FIFO ordering policy at a cost of an additional transaction.
+ * This ordering does not guarantee fairness of thread scheduling (similar to {@code java.util.concurrent.ReentrantLock}).
+ * Thus, one of many threads on any node using a fair lock may obtain it multiple times in succession while other
+ * active threads are not progressing and not currently holding the lock. Also note that the untimed tryLock method
+ * does not honor the fairness setting. It will succeed if the lock is available even if other threads are waiting.
+ *
+ * </p>
+ * As a rule of thumb, whenever there is a reasonable time window between successive calls to release and acquire
+ * the lock, non-fair lock should be preferred:
+ *
+ * <pre> {@code
+ * while(someCondition){
+ * // do anything
+ * lock.lock();
+ * try{
+ * // ...
+ * }
+ * finally {
+ * lock.unlock();
+ * }
+ * }
+ * }</pre>
+ *
+ * If successive calls to release/acquire are following immediately,
+ * e.g.
+ *
+ * <pre> {@code
+ * while(someCondition){
+ * lock.lock();
+ * try {
+ * // do something
+ * }
+ * finally {
+ * lock.unlock();
+ * }
+ * }
+ * }</pre>
+ *
+ * using the fair lock is reasonable in order to allow even distribution of load among nodes
+ * (although overall throughput may be lower due to increased overhead).
+ *
+ */
+public interface IgniteLock extends Lock, Closeable {
+ /**
+ * Name of atomic reentrant lock.
+ *
+ * @return Name of atomic reentrant lock.
+ */
+ public String name();
+
+ /**
+ * Acquires the distributed reentrant lock.
+ *
+ * <p>Acquires the lock if it is not held by another thread and returns
+ * immediately, setting the lock hold count to one.
+ *
+ * <p>If the current thread already holds this lock then the hold count
+ * is incremented by one and the method returns immediately.
+ *
+ * <p>If the lock is held by another thread then the
+ * current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until one of four things happens:
+ *
+ * <ul>
+ *
+ * <li>The lock is acquired by the current thread; or
+ *
+ * <li>Lock is broken (any node failed while owning this lock), and lock is created in
+ * non-failoverSafe mode.
+ *
+ * <li>Local node is stopped.
+ *
+ * @throws IgniteException if the node is stopped or broken in non-failoverSafe mode
+ */
+ void lock() throws IgniteException;
+
+ /**
+ * Acquires the lock unless the current thread is
+ * {@linkplain Thread#interrupt interrupted}.
+ *
+ * <p>Acquires the lock if it is not held by another thread and returns
+ * immediately, setting the lock hold count to one.
+ *
+ * <p>If the current thread already holds this lock then the hold count
+ * is incremented by one and the method returns immediately.
+ *
+ * <p>If the lock is held by another thread then the
+ * current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until one of four things happens:
+ *
+ * <ul>
+ *
+ * <li>The lock is acquired by the current thread; or
+ *
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts} the
+ * current thread.
+ *
+ * <li>Lock is broken (any node failed while owning this lock), and lock is created in
+ * non-failoverSafe mode.
+ *
+ * <li>Local node is stopped.
+ *
+ * </ul>
+ *
+ * <p>If the lock is acquired by the current thread then the lock hold
+ * count is set to one.
+ *
+ * <p>If the current thread:
+ *
+ * <ul>
+ *
+ * <li>has its interrupted status set on entry to this method; or
+ *
+ * <li>is {@linkplain Thread#interrupt interrupted} while acquiring
+ * the lock; or
+ *
+ * then {@link IgniteInterruptedException} is thrown and the current thread's
+ * interrupted status is cleared.
+ *
+ * </ul>
+ *
+ * <p>{@link IgniteException} is thrown in case:
+ *
+ * <ul>
+ *
+ * <li>the lock is broken before or during the attempt to acquire this lock; or
+ *
+ * <li>local node is stopped,
+ *
+ * </ul>
+ *
+ * <p>In this implementation, as this method is an explicit
+ * interruption point, preference is given to responding to the
+ * interrupt over normal or reentrant acquisition of the lock.
+ *
+ * @throws IgniteInterruptedException if the current thread is interrupted
+ * @throws IgniteException if the lock is broken in non-failoverSafe mode (any node failed while owning this lock),
+ * or local node is stopped
+ */
+ @Override public void lockInterruptibly() throws IgniteInterruptedException, IgniteException;
+
+ /**
+ * Acquires the lock only if it is free at the time of invocation.
+ *
+ * <p>Acquires the lock if it is available and returns immediately
+ * with the value {@code true}.
+ * If the lock is not available then this method will return
+ * immediately with the value {@code false}.
+ *
+ * <p>A typical usage idiom for this method would be:
+ * <pre> {@code
+ * Lock lock = ...;
+ * if (lock.tryLock()) {
+ * try {
+ * // manipulate protected state
+ * } finally {
+ * lock.unlock();
+ * }
+ * } else {
+ * // perform alternative actions
+ * }}</pre>
+ *
+ * This usage ensures that the lock is unlocked if it was acquired, and
+ * doesn't try to unlock if the lock was not acquired.
+ *
+ * If node is stopped, or any node failed while owning the lock in non-failoverSafe mode,
+ * then {@link IgniteException} is thrown.
+ *
+ * @return {@code true} if the lock was acquired and
+ * {@code false} otherwise
+ *
+ * @throws IgniteException if node is stopped, or lock is already broken in non-failover safe mode
+ */
+ @Override public boolean tryLock() throws IgniteException;
+
+ /**
+ * Acquires the lock if it is not held by another thread within the given
+ * waiting time and the current thread has not been
+ * {@linkplain Thread#interrupt interrupted}.
+ *
+ * <p>Acquires the lock if it is not held by another thread and returns
+ * immediately with the value {@code true}, setting the lock hold count
+ * to one.
+ *
+ * <p>If the current thread
+ * already holds this lock then the hold count is incremented by one and
+ * the method returns {@code true}.
+ *
+ * <p>If the lock is held by another thread then the
+ * current thread becomes disabled for thread scheduling
+ * purposes and lies dormant until one of five things happens:
+ *
+ * <ul>
+ *
+ * <li>The lock is acquired by the current thread; or
+ *
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
+ *
+ * <li>Lock is broken (any node failed while owning this lock), and lock is created in
+ * non-failoverSafe mode.
+ *
+ * <li>Local node is stopped.
+ *
+ * <li>The specified waiting time elapses
+ *
+ * </ul>
+ *
+ * <p>If the lock is acquired then the value {@code true} is returned and
+ * the lock hold count is set to one.
+ *
+ * <p>If the current thread:
+ *
+ * <ul>
+ *
+ * <li>has its interrupted status set on entry to this method; or
+ *
+ * <li>is {@linkplain Thread#interrupt interrupted} while
+ * acquiring the lock; or
+ *
+ * </ul>
+ * then {@link IgniteInterruptedException} is thrown and the current thread's
+ * interrupted status is cleared.
+ *
+ * <p>{@link IgniteException} is thrown in case:
+ *
+ * <ul>
+ *
+ * <li>the lock is broken before or during the attempt to acquire this lock; or
+ *
+ * <li>local node is stopped,
+ *
+ * </ul>
+ *
+ * <p>If the specified waiting time elapses then the value {@code false}
+ * is returned. If the time is less than or equal to zero, the method
+ * will not wait at all.
+ *
+ * <p>In this implementation, as this method is an explicit
+ * interruption point, preference is given to responding to the
+ * interrupt over normal or reentrant acquisition of the lock, and
+ * over reporting the elapse of the waiting time.
+ *
+ * @param timeout the time to wait for the lock
+ * @param unit the time unit of the timeout argument
+ * @return {@code true} if the lock was free and was acquired by the
+ * current thread, or the lock was already held by the current
+ * thread; and {@code false} if the waiting time elapsed before
+ * the lock could be acquired
+ * @throws IgniteInterruptedException if the current thread is interrupted
+ * @throws IgniteException if node is stopped, or lock is already broken in non-failover safe mode
+ * @throws NullPointerException if the time unit is null
+ */
+ @Override public boolean tryLock(long timeout, TimeUnit unit) throws IgniteInterruptedException, IgniteException;
+
+ /**
+ * Releases the lock.
+ *
+ * If lock is not owned by current thread, then an {@link
+ * IllegalMonitorStateException} is thrown.
+ * If lock is already broken prior to invocation of this method, and
+ * lock is created in non-failover safe mode, then {@link IgniteException} is thrown.
+ *
+ * @throws IllegalMonitorStateException if not owned by current thread
+ * @throws IgniteException if node is stopped, or lock is already broken in non-failover safe mode
+ */
+ void unlock() throws IgniteInterruptedException;
+
+ /**
+ * Returns a {@link Condition} instance for use with this
+ * {@link IgniteLock} instance.
+ *
+ * <ul>
+ *
+ * <li>If this lock is not held when any of the {@link Condition}
+ * {@linkplain Condition#await() waiting} or {@linkplain
+ * Condition#signal signalling} methods are called, then an {@link
+ * IllegalMonitorStateException} is thrown.
+ *
+ * <li>When the condition {@linkplain Condition#await() waiting}
+ * methods are called the lock is released and, before they
+ * return, the lock is reacquired and the lock hold count restored
+ * to what it was when the method was called.
+ *
+ * <li>If a thread is {@linkplain Thread#interrupt interrupted}
+ * while waiting then the wait will terminate, an {@link
+ * IgniteInterruptedException} will be thrown, and the thread's
+ * interrupted status will be cleared.
+ *
+ * <li> Waiting threads are signalled in FIFO order.
+ *
+ * </ul>
+ *
+ * @param name Name of the distributed condition object
+ *
+ * @return the Condition object
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public IgniteCondition getOrCreateCondition(String name) throws IgniteException;
+
+ /**
+ * This method is not supported in IgniteLock,
+ * Any invocation of this method will result in {@linkplain UnsupportedOperationException}.
+ * Correct way to obtain Condition object is through method {@linkplain IgniteLock#getOrCreateCondition(String)}
+ *
+ */
+ @Override public Condition newCondition();
+
+ /**
+ * Queries the number of holds on this lock by the current thread.
+ *
+ * @return the number of holds on this lock by the current thread,
+ * or zero if this lock is not held by the current thread
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public int getHoldCount() throws IgniteException;
+
+ /**
+ * Queries if this lock is held by the current thread.
+ *
+ * @return {@code true} if current thread holds this lock and
+ * {@code false} otherwise
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean isHeldByCurrentThread() throws IgniteException;
+
+ /**
+ * Queries if this lock is held by any thread on any node. This method is
+ * designed for use in monitoring of the system state,
+ * not for synchronization control.
+ *
+ * @return {@code true} if any thread on this or any other node holds this lock and
+ * {@code false} otherwise
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean isLocked() throws IgniteException;
+
+ /**
+ * Queries whether any threads on this node are waiting to acquire this lock. Note that
+ * because cancellations may occur at any time, a {@code true}
+ * return does not guarantee that any other thread will ever
+ * acquire this lock. This method is designed primarily for use in
+ * monitoring of the system state.
+ *
+ * @return {@code true} if there may be other threads on this node waiting to
+ * acquire the lock
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean hasQueuedThreads() throws IgniteException;
+
+ /**
+ * Queries whether the given thread is waiting to acquire this
+ * lock. Note that because cancellations may occur at any time, a
+ * {@code true} return does not guarantee that this thread
+ * will ever acquire this lock. This method is designed primarily for use
+ * in monitoring of the system state.
+ *
+ * @param thread the thread
+ * @return {@code true} if the given thread is queued waiting for this lock
+ * @throws NullPointerException if the thread is null
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean hasQueuedThread(Thread thread) throws IgniteException;
+
+ /**
+ * Queries whether any threads on this node are waiting on the given condition
+ * associated with this lock. Note that because timeouts and
+ * interrupts may occur at any time, a {@code true} return does
+ * not guarantee that a future {@code signal} will awaken any
+ * threads. This method is designed primarily for use in
+ * monitoring of the system state.
+ *
+ * @param condition the condition
+ * @return {@code true} if there are any waiting threads on this node
+ * @throws IllegalMonitorStateException if this lock is not held
+ * @throws IllegalArgumentException if the given condition is
+ * not associated with this lock
+ * @throws NullPointerException if the condition is null
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean hasWaiters(IgniteCondition condition) throws IgniteException;
+
+ /**
+ * Returns an estimate of the number of threads on this node that are waiting on the
+ * given condition associated with this lock. Note that because
+ * timeouts and interrupts may occur at any time, the estimate
+ * serves only as an upper bound on the actual number of waiters.
+ * This method is designed for use in monitoring of the system
+ * state, not for synchronization control.
+ *
+ * @param condition the condition
+ * @return the estimated number of waiting threads on this node
+ * @throws IllegalMonitorStateException if this lock is not held
+ * @throws IllegalArgumentException if the given condition is
+ * not associated with this lock
+ * @throws NullPointerException if the condition is null
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public int getWaitQueueLength(IgniteCondition condition) throws IgniteException;
+
+ /**
+ * Returns {@code true} if this lock is safe to use after node failure.
+ * If not, IgniteInterruptedException is thrown on every other node after node failure.
+ *
+ * @return {@code true} if this reentrant lock has failoverSafe set true
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean isFailoverSafe();
+
+ /**
+ * Returns {@code true} if this lock is fair. Fairness flag can only be set on lock creation.
+ *
+ * @return {@code true} if this reentrant lock has fairness flag set true.
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean isFair();
+
+ /**
+ * Returns true if any node that owned the locked failed before releasing the lock.
+ *
+ * @return true if any node failed while owning the lock since the lock on this node was initialized.
+ * @throws IgniteException if the lock is not initialized or already removed
+ */
+ public boolean isBroken() throws IgniteException;
+
+ /**
+ * Gets status of reentrant lock.
+ *
+ * @return {@code true} if reentrant lock was removed from cache, {@code false} in other case.
+ */
+ public boolean removed();
+
+ /**
+ * Removes reentrant lock.
+ *
+ * @throws IgniteException If operation failed.
+ */
+ @Override public void close();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 79d67df..1ff4543 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -876,7 +876,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
/** {@inheritDoc} */
@Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- stash.set(U.readString(in));
+ U.readString(in); // Read for compatibility only. See #readResolve().
}
/**
@@ -887,7 +887,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
*/
protected Object readResolve() throws ObjectStreamException {
try {
- return IgnitionEx.gridx(stash.get()).context();
+ return IgnitionEx.localIgnite().context();
}
catch (IllegalStateException e) {
throw U.withCause(new InvalidObjectException(e.getMessage()), e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 5094415..0f180b2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -64,6 +64,7 @@ import org.apache.ignite.IgniteFileSystem;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteMessaging;
import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteScheduler;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteServices;
@@ -3036,6 +3037,26 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
}
/** {@inheritDoc} */
+ @Nullable @Override public IgniteLock reentrantLock(
+ String name,
+ boolean failoverSafe,
+ boolean fair,
+ boolean create
+ ) {
+ guard();
+
+ try {
+ return ctx.dataStructures().reentrantLock(name, failoverSafe, fair, create);
+ }
+ catch (IgniteCheckedException e) {
+ throw U.convertException(e);
+ }
+ finally {
+ unguard();
+ }
+ }
+
+ /** {@inheritDoc} */
@Nullable @Override public <T> IgniteQueue<T> queue(String name,
int cap,
CollectionConfiguration cfg)
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
index 10aa71e..1cad22f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
@@ -43,6 +43,7 @@ import org.apache.ignite.IgniteAtomicStamped;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteQueue;
import org.apache.ignite.IgniteSemaphore;
@@ -95,6 +96,7 @@ import static org.apache.ignite.internal.processors.datastructures.DataStructure
import static org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor.DataStructureType.COUNT_DOWN_LATCH;
import static org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor.DataStructureType.QUEUE;
import static org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor.DataStructureType.SEMAPHORE;
+import static org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor.DataStructureType.REENTRANT_LOCK;
import static org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor.DataStructureType.SET;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
@@ -135,6 +137,9 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
/** Cache contains only {@code GridCacheSemaphoreState}. */
private IgniteInternalCache<GridCacheInternalKey, GridCacheSemaphoreState> semView;
+ /** Cache contains only {@code GridCacheLockState}. */
+ private IgniteInternalCache<GridCacheInternalKey, GridCacheLockState> reentrantLockView;
+
/** Cache contains only {@code GridCacheAtomicReferenceValue}. */
private IgniteInternalCache<GridCacheInternalKey, GridCacheAtomicReferenceValue> atomicRefView;
@@ -177,7 +182,7 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
ctx.event().addLocalEventListener(
new GridLocalEventListener() {
@Override public void onEvent(final Event evt) {
- // This may require cache operation to exectue,
+ // This may require cache operation to execute,
// therefore cannot use event notification thread.
ctx.closure().callLocalSafe(
new Callable<Object>() {
@@ -189,6 +194,8 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
for (GridCacheRemovable ds : dsMap.values()) {
if (ds instanceof GridCacheSemaphoreEx)
((GridCacheSemaphoreEx)ds).onNodeRemoved(leftNodeId);
+ else if (ds instanceof GridCacheLockEx)
+ ((GridCacheLockEx)ds).onNodeRemoved(leftNodeId);
}
return null;
@@ -224,6 +231,8 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
semView = atomicsCache;
+ reentrantLockView = atomicsCache;
+
atomicLongView = atomicsCache;
atomicRefView = atomicsCache;
@@ -262,6 +271,9 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
for (GridCacheRemovable ds : dsMap.values()) {
if (ds instanceof GridCacheSemaphoreEx)
((GridCacheSemaphoreEx)ds).stop();
+
+ if (ds instanceof GridCacheLockEx)
+ ((GridCacheLockEx)ds).onStop();
}
if (initLatch.getCount() > 0) {
@@ -1332,6 +1344,124 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
}
/**
+ * Gets or creates reentrant lock. If reentrant lock is not found in cache,
+ * it is created using provided name, failover mode, and fairness mode parameters.
+ *
+ * @param name Name of the reentrant lock.
+ * @param failoverSafe Flag indicating behaviour in case of failure.
+ * @param fair Flag indicating fairness policy of this lock.
+ * @param create If {@code true} reentrant lock will be created in case it is not in cache.
+ * @return ReentrantLock for the given name or {@code null} if it is not found and
+ * {@code create} is false.
+ * @throws IgniteCheckedException If operation failed.
+ */
+ public IgniteLock reentrantLock(final String name, final boolean failoverSafe, final boolean fair, final boolean create)
+ throws IgniteCheckedException {
+ A.notNull(name, "name");
+
+ awaitInitialization();
+
+ checkAtomicsConfiguration();
+
+ startQuery();
+
+ return getAtomic(new IgniteOutClosureX<IgniteLock>() {
+ @Override public IgniteLock applyx() throws IgniteCheckedException {
+ GridCacheInternalKey key = new GridCacheInternalKeyImpl(name);
+
+ dsCacheCtx.gate().enter();
+
+ try (IgniteInternalTx tx = CU.txStartInternal(dsCacheCtx, dsView, PESSIMISTIC, REPEATABLE_READ)) {
+ GridCacheLockState val = cast(dsView.get(key), GridCacheLockState.class);
+
+ // Check that reentrant lock hasn't been created in other thread yet.
+ GridCacheLockEx reentrantLock = cast(dsMap.get(key), GridCacheLockEx.class);
+
+ if (reentrantLock != null) {
+ assert val != null;
+
+ return reentrantLock;
+ }
+
+ if (val == null && !create)
+ return null;
+
+ if (val == null) {
+ val = new GridCacheLockState(0, dsCacheCtx.nodeId(), 0, failoverSafe, fair);
+
+ dsView.put(key, val);
+ }
+
+ GridCacheLockEx reentrantLock0 = new GridCacheLockImpl(
+ name,
+ key,
+ reentrantLockView,
+ dsCacheCtx);
+
+ dsMap.put(key, reentrantLock0);
+
+ tx.commit();
+
+ return reentrantLock0;
+ }
+ catch (Error | Exception e) {
+ dsMap.remove(key);
+
+ U.error(log, "Failed to create reentrant lock: " + name, e);
+
+ throw e;
+ }
+ finally {
+ dsCacheCtx.gate().leave();
+ }
+ }
+ }, new DataStructureInfo(name, REENTRANT_LOCK, null), create, GridCacheLockEx.class);
+ }
+
+ /**
+ * Removes reentrant lock from cache.
+ *
+ * @param name Name of the reentrant lock.
+ * @param broken Flag indicating the reentrant lock is broken and should be removed unconditionally.
+ * @throws IgniteCheckedException If operation failed.
+ */
+ public void removeReentrantLock(final String name, final boolean broken) throws IgniteCheckedException {
+ assert name != null;
+ assert dsCacheCtx != null;
+
+ awaitInitialization();
+
+ removeDataStructure(new IgniteOutClosureX<Void>() {
+ @Override public Void applyx() throws IgniteCheckedException {
+ GridCacheInternal key = new GridCacheInternalKeyImpl(name);
+
+ dsCacheCtx.gate().enter();
+
+ try (IgniteInternalTx tx = CU.txStartInternal(dsCacheCtx, dsView, PESSIMISTIC, REPEATABLE_READ)) {
+ // Check correctness type of removable object.
+ GridCacheLockState val = cast(dsView.get(key), GridCacheLockState.class);
+
+ if (val != null) {
+ if (val.get() > 0 && !broken)
+ throw new IgniteCheckedException("Failed to remove reentrant lock with blocked threads. ");
+
+ dsView.remove(key);
+
+ tx.commit();
+ }
+ else
+ tx.setRollbackOnly();
+
+ return null;
+ }
+ finally {
+ dsCacheCtx.gate().leave();
+ }
+ }
+ }, name, REENTRANT_LOCK, null);
+ }
+
+ /**
* Remove internal entry by key from cache.
*
* @param key Internal entry key.
@@ -1379,7 +1509,8 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
@Override public boolean evaluate(CacheEntryEvent<?, ?> evt) throws CacheEntryListenerException {
if (evt.getEventType() == EventType.CREATED || evt.getEventType() == EventType.UPDATED)
return evt.getValue() instanceof GridCacheCountDownLatchValue ||
- evt.getValue() instanceof GridCacheSemaphoreState;
+ evt.getValue() instanceof GridCacheSemaphoreState ||
+ evt.getValue() instanceof GridCacheLockState;
else {
assert evt.getEventType() == EventType.REMOVED : evt;
@@ -1476,7 +1607,25 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
", actual=" + sem.getClass() + ", value=" + sem + ']');
}
}
+ else if (val0 instanceof GridCacheLockState) {
+ GridCacheInternalKey key = evt.getKey();
+
+ // Notify reentrant lock on changes.
+ final GridCacheRemovable reentrantLock = dsMap.get(key);
+
+ GridCacheLockState val = (GridCacheLockState)val0;
+
+ if (reentrantLock instanceof GridCacheLockEx) {
+ final GridCacheLockEx lock0 = (GridCacheLockEx)reentrantLock;
+ lock0.onUpdate(val);
+ }
+ else if (reentrantLock != null) {
+ U.error(log, "Failed to cast object " +
+ "[expected=" + IgniteLock.class.getSimpleName() +
+ ", actual=" + reentrantLock.getClass() + ", value=" + reentrantLock + ']');
+ }
+ }
}
else {
assert evt.getEventType() == EventType.REMOVED : evt;
@@ -1694,7 +1843,10 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
SET(IgniteSet.class.getSimpleName()),
/** */
- SEMAPHORE(IgniteSemaphore.class.getSimpleName());
+ SEMAPHORE(IgniteSemaphore.class.getSimpleName()),
+
+ /** */
+ REENTRANT_LOCK(IgniteLock.class.getSimpleName());
/** */
private static final DataStructureType[] VALS = values();
http://git-wip-us.apache.org/repos/asf/ignite/blob/f9bd9a33/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockEx.java
new file mode 100644
index 0000000..0887345
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheLockEx.java
@@ -0,0 +1,52 @@
+/*
+ * 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.processors.datastructures;
+
+import java.util.UUID;
+import org.apache.ignite.IgniteLock;
+
+/**
+ * Grid cache reentrant lock ({@code 'Ex'} stands for external).
+ */
+public interface GridCacheLockEx extends IgniteLock, GridCacheRemovable {
+ /**
+ * Get current reentrant lock latch key.
+ *
+ * @return Lock key.
+ */
+ public GridCacheInternalKey key();
+
+ /**
+ * Callback to notify reentrant lock on changes.
+ *
+ * @param state New reentrant lock state.
+ */
+ public void onUpdate(GridCacheLockState state);
+
+ /**
+ * Callback to notify semaphore on topology changes.
+ *
+ * @param nodeId Id of the node that left the grid.
+ */
+ public void onNodeRemoved(UUID nodeId);
+
+ /**
+ * Callback to notify local reentrant lock instance on node stop.
+ */
+ public void onStop();
+}