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 2015/10/23 18:10:45 UTC
[4/6] ignite git commit: Adds test coverage; Simplifies example;
Adds test coverage; Simplifies example;
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/8c6852d0
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/8c6852d0
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/8c6852d0
Branch: refs/heads/ignite-638
Commit: 8c6852d07481c3943584f676b48b8c938342e640
Parents: 77f9deb
Author: Vladisav Jelisavcic <vl...@gmail.com>
Authored: Thu Oct 15 01:17:02 2015 +0200
Committer: vladisav <vl...@gmail.com>
Committed: Thu Oct 15 01:17:02 2015 +0200
----------------------------------------------------------------------
.../datastructures/IgniteSemaphoreExample.java | 78 +----
.../datastructures/DataStructuresProcessor.java | 26 +-
.../datastructures/GridCacheSemaphoreImpl.java | 12 +-
.../datastructures/GridCacheSemaphoreState.java | 28 +-
.../IgniteClientReconnectAtomicsTest.java | 44 ++-
.../IgniteClientDataStructuresAbstractTest.java | 59 +++-
.../IgniteDataStructureUniqueNameTest.java | 14 +-
.../IgniteSemaphoreAbstractSelfTest.java | 328 +++++++++++++++++++
.../local/IgniteLocalSemaphoreSelfTest.java | 98 ++++++
.../IgnitePartitionedSemaphoreSelfTest.java | 34 ++
.../IgniteReplicatedSemaphoreSelfTest.java | 33 ++
.../cache/GridCacheDataStructuresLoadTest.java | 283 +++++++++-------
12 files changed, 812 insertions(+), 225 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteSemaphoreExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteSemaphoreExample.java b/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteSemaphoreExample.java
index 5849f5f..ece0ffc 100644
--- a/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteSemaphoreExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/datastructures/IgniteSemaphoreExample.java
@@ -1,10 +1,7 @@
package org.apache.ignite.examples.datastructures;
-import java.util.LinkedList;
-import java.util.Queue;
import java.util.UUID;
import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.Ignition;
import org.apache.ignite.examples.ExampleNodeStartup;
@@ -43,31 +40,16 @@ public class IgniteSemaphoreExample {
// Make name of semaphore.
final String semaphoreName = UUID.randomUUID().toString();
- // Make name of mutex
- final String mutexName = UUID.randomUUID().toString();
-
- // Make shared resource
- final String resourceName = UUID.randomUUID().toString();
-
- // Get cache view where the resource will be held
- IgniteCache<String, Queue<String>> cache = ignite.getOrCreateCache(CACHE_NAME);
-
- // Put the resource queue the cache
- cache.put(resourceName, new LinkedList<>());
-
// Initialize semaphore.
IgniteSemaphore semaphore = ignite.semaphore(semaphoreName, 0, false, true);
- // Initialize mutex.
- IgniteSemaphore mutex = ignite.semaphore(mutexName, 1, false, true);
-
// Start consumers on all cluster nodes.
for (int i = 0; i < NUM_CONSUMERS; i++)
- ignite.compute().withAsync().run(new Consumer(mutexName, semaphoreName, resourceName));
+ ignite.compute().withAsync().run(new Consumer(semaphoreName));
// Start producers on all cluster nodes.
for (int i = 0; i < NUM_PRODUCERS; i++)
- ignite.compute().withAsync().run(new Producer(mutexName, semaphoreName, resourceName));
+ ignite.compute().withAsync().run(new Producer(semaphoreName));
System.out.println("Master node is waiting for all other nodes to finish...");
@@ -88,21 +70,11 @@ public class IgniteSemaphoreExample {
/** Semaphore name. */
protected final String semaphoreName;
- /** Mutex name. */
- protected final String mutexName;
-
- /** Resource name. */
- protected final String resourceName;
-
/**
- * @param mutexName Mutex name.
* @param semaphoreName Semaphore name.
- * @param resourceName Resource name.
*/
- SemaphoreExampleClosure(String mutexName, String semaphoreName, String resourceName) {
+ SemaphoreExampleClosure(String semaphoreName) {
this.semaphoreName = semaphoreName;
- this.mutexName = mutexName;
- this.resourceName = resourceName;
}
}
@@ -112,33 +84,18 @@ public class IgniteSemaphoreExample {
private static class Producer extends SemaphoreExampleClosure {
/**
- * @param mutexName Mutex name.
* @param semaphoreName Semaphore name.
- * @param resourceName Resource name.
*/
- public Producer(String mutexName, String semaphoreName, String resourceName) {
- super(mutexName, semaphoreName, resourceName);
+ public Producer(String semaphoreName) {
+ super(semaphoreName);
}
/** {@inheritDoc} */
@Override public void run() {
IgniteSemaphore semaphore = Ignition.ignite().semaphore(semaphoreName, 0, true, true);
- IgniteSemaphore mutex = Ignition.ignite().semaphore(mutexName, 0, true, true);
for (int i = 0; i < ITEM_COUNT; i++) {
- // Mutex is used to access shared resource.
- mutex.acquire();
-
- Queue<String> queue = (Queue<String>)Ignition.ignite().cache(CACHE_NAME).get(resourceName);
-
- queue.add(Ignition.ignite().cluster().localNode().id().toString());
-
- Ignition.ignite().cache(CACHE_NAME).put(resourceName, queue);
-
- System.out.println("Producer [nodeId=" + Ignition.ignite().cluster().localNode().id() + "] produced data. Available: " + semaphore.availablePermits());
-
- // Mutex is released for others to access the resource.
- mutex.release();
+ System.out.println("Producer [nodeId=" + Ignition.ignite().cluster().localNode().id() + "]. Available: " + semaphore.availablePermits());
// Signals others that shared resource is available.
semaphore.release();
@@ -160,36 +117,21 @@ public class IgniteSemaphoreExample {
private static class Consumer extends SemaphoreExampleClosure {
/**
- * @param mutexName Mutex name.
* @param semaphoreName Semaphore name.
- * @param resourceName Resource name.
*/
- public Consumer(String mutexName, String semaphoreName, String resourceName) {
- super(mutexName, semaphoreName, resourceName);
+ public Consumer(String semaphoreName) {
+ super(semaphoreName);
}
/** {@inheritDoc} */
@Override public void run() {
IgniteSemaphore semaphore = Ignition.ignite().semaphore(semaphoreName, 0, true, true);
- IgniteSemaphore mutex = Ignition.ignite().semaphore(mutexName, 0, true, true);
for (int i = 0; i < ITEM_COUNT; i++) {
- // Block if queue is empty.
+ // Block if no permits are available.
semaphore.acquire();
- // Mutex is used to access shared resource.
- mutex.acquire();
-
- Queue<String> queue = (Queue<String>)Ignition.ignite().cache(CACHE_NAME).get(resourceName);
-
- String data = queue.remove();
-
- Ignition.ignite().cache(CACHE_NAME).put(resourceName, queue);
-
- System.out.println("Consumer [nodeId=" + Ignition.ignite().cluster().localNode().id() + "] consumed data generated by producer [nodeId=" + data + "]");
-
- // Signals others that shared resource is available.
- mutex.release();
+ System.out.println("Consumer [nodeId=" + Ignition.ignite().cluster().localNode().id() + "]. Available: " + semaphore.availablePermits());
}
System.out.println("Consumer [nodeId=" + Ignition.ignite().cluster().localNode().id() + "] finished. ");
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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 220dec2..b97b24f 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
@@ -1195,19 +1195,12 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
* {@code create} is false.
* @throws IgniteCheckedException If operation failed.
*/
- public IgniteSemaphore semaphore(final String name,
- final int cnt,
- final boolean fair,
- final boolean create)
- throws IgniteCheckedException
- {
+ public IgniteSemaphore semaphore(final String name, final int cnt, final boolean fair, final boolean create)
+ throws IgniteCheckedException {
A.notNull(name, "name");
awaitInitialization();
- if (create)
- A.ensure(cnt >= 0, "count can not be negative");
-
checkAtomicsConfiguration();
startQuery();
@@ -1234,12 +1227,12 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
return null;
if (val == null) {
- val = new GridCacheSemaphoreState(cnt, 0);
+ val = new GridCacheSemaphoreState(cnt, 0, fair);
dsView.put(key, val);
}
- semaphore = new GridCacheSemaphoreImpl(name, val.getCnt(),
+ semaphore = new GridCacheSemaphoreImpl(name, val.getCount(),
fair,
key,
semaphoreView,
@@ -1278,19 +1271,17 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
awaitInitialization();
removeDataStructure(new IgniteOutClosureX<Void>() {
- @Override
- public Void applyx() throws IgniteCheckedException {
+ @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.
- GridCacheSemaphoreState val =
- cast(dsView.get(key), GridCacheSemaphoreState.class);
+ GridCacheSemaphoreState val = cast(dsView.get(key), GridCacheSemaphoreState.class);
if (val != null) {
- if (val.getCnt() < 0) {
+ if (val.getCount() < 0) {
throw new IgniteCheckedException("Failed to remove semaphore " +
"with blocked threads. ");
}
@@ -1318,8 +1309,7 @@ public final class DataStructuresProcessor extends GridProcessorAdapter {
* @throws IgniteCheckedException If removing failed or class of object is different to expected class.
*/
private <R> boolean removeInternal(final GridCacheInternal key, final Class<R> cls) throws IgniteCheckedException {
- return CU.outTx(
- new Callable<Boolean>() {
+ return CU.outTx(new Callable<Boolean>() {
@Override public Boolean call() throws Exception {
try (IgniteInternalTx tx = CU.txStartInternal(dsCacheCtx, dsView, PESSIMISTIC, REPEATABLE_READ)) {
// Check correctness type of removable object.
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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 d2a966f..78d923a 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
@@ -261,7 +261,7 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
if (val == null)
throw new IgniteCheckedException("Failed to find semaphore with given name: " + name);
- boolean retVal = val.getCnt() == expVal;
+ boolean retVal = val.getCount() == expVal;
if (retVal) {
/* If current thread is queued, than this call is the call that is going to be unblocked. */
@@ -274,7 +274,7 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
sync.threadMap.remove(Thread.currentThread());
}
- val.setCnt(newVal);
+ val.setCount(newVal);
semaphoreView.put(key, val);
@@ -396,7 +396,7 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
return null;
}
- final int count = val.getCnt();
+ final int count = val.getCount();
tx.commit();
@@ -448,7 +448,7 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
return;
// Update permission count.
- sync.setPermits(val.getCnt());
+ sync.setPermits(val.getCount());
// Update waiters count.
sync.setWaiters(val.getWaiters());
@@ -463,7 +463,7 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
}
/** {@inheritDoc} */
- @Override public void acquire() throws IgniteException {
+ @Override public void acquire() throws IgniteInterruptedException {
acquire(1);
}
@@ -523,7 +523,7 @@ public final class GridCacheSemaphoreImpl implements GridCacheSemaphoreEx, Exter
if (val == null)
throw new IgniteException("Failed to find semaphore with given name: " + name);
- int count = val.getCnt();
+ int count = val.getCount();
tx.rollback();
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreState.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreState.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreState.java
index e25649e..1109b53 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreState.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheSemaphoreState.java
@@ -17,7 +17,7 @@ public class GridCacheSemaphoreState implements GridCacheInternal, Externalizabl
/**
* Permission count.
*/
- private int cnt;
+ private int count;
/**
* Waiter id.
@@ -32,10 +32,10 @@ public class GridCacheSemaphoreState implements GridCacheInternal, Externalizabl
/**
* Constructor.
*
- * @param cnt Number of permissions.
+ * @param count Number of permissions.
*/
- public GridCacheSemaphoreState(int cnt, int waiters) {
- this.cnt = cnt;
+ public GridCacheSemaphoreState(int count, int waiters) {
+ this.count = count;
this.waiters = waiters;
this.fair = false;
}
@@ -43,10 +43,10 @@ public class GridCacheSemaphoreState implements GridCacheInternal, Externalizabl
/**
* Constructor.
*
- * @param cnt Number of permissions.
+ * @param count Number of permissions.
*/
- public GridCacheSemaphoreState(int cnt, int waiters, boolean fair) {
- this.cnt = cnt;
+ public GridCacheSemaphoreState(int count, int waiters, boolean fair) {
+ this.count = count;
this.waiters = waiters;
this.fair = fair;
}
@@ -59,17 +59,17 @@ public class GridCacheSemaphoreState implements GridCacheInternal, Externalizabl
}
/**
- * @param cnt New count.
+ * @param count New count.
*/
- public void setCnt(int cnt) {
- this.cnt = cnt;
+ public void setCount(int count) {
+ this.count = count;
}
/**
* @return Current count.
*/
- public int getCnt() {
- return cnt;
+ public int getCount() {
+ return count;
}
public int getWaiters() {
@@ -95,7 +95,7 @@ public class GridCacheSemaphoreState implements GridCacheInternal, Externalizabl
* {@inheritDoc}
*/
@Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeInt(cnt);
+ out.writeInt(count);
out.writeInt(waiters);
out.writeBoolean(fair);
}
@@ -104,7 +104,7 @@ public class GridCacheSemaphoreState implements GridCacheInternal, Externalizabl
* {@inheritDoc}
*/
@Override public void readExternal(ObjectInput in) throws IOException {
- cnt = in.readInt();
+ count = in.readInt();
waiters = in.readInt();
fair = in.readBoolean();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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 55dbb57..c46b5c8 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
@@ -26,6 +26,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.IgniteSemaphore;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse;
import org.apache.ignite.testframework.GridTestUtils;
@@ -675,4 +676,45 @@ public class IgniteClientReconnectAtomicsTest extends IgniteClientReconnectAbstr
assertTrue(srvLatch.await(1000));
assertTrue(clientLatch.await(1000));
}
-}
\ No newline at end of file
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSemaphoreReconnect() throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ Ignite srv = clientRouter(client);
+
+ IgniteSemaphore clientSemaphore = client.semaphore("semaphore1", 3, false, true);
+
+ assertEquals(3, clientSemaphore.availablePermits());
+
+ final IgniteSemaphore srvSemaphore = srv.semaphore("semaphore1", 3, false, false);
+
+ assertEquals(3, srvSemaphore.availablePermits());
+
+ reconnectClientNode(client, srv, new Runnable() {
+ @Override public void run() {
+ srvSemaphore.acquire();
+ }
+ });
+
+ assertEquals(2, srvSemaphore.availablePermits());
+ assertEquals(2, clientSemaphore.availablePermits());
+
+ srvSemaphore.acquire();
+
+ assertEquals(1, srvSemaphore.availablePermits());
+ assertEquals(1, clientSemaphore.availablePermits());
+
+ clientSemaphore.acquire();
+
+ assertEquals(0, srvSemaphore.availablePermits());
+ assertEquals(0, clientSemaphore.availablePermits());
+
+ assertFalse(srvSemaphore.tryAcquire());
+ assertFalse(srvSemaphore.tryAcquire());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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 989f75f..bf6dcda 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
@@ -24,6 +24,7 @@ import org.apache.ignite.IgniteAtomicLong;
import org.apache.ignite.IgniteAtomicSequence;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteSet;
import org.apache.ignite.configuration.CollectionConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -267,6 +268,62 @@ public abstract class IgniteClientDataStructuresAbstractTest extends GridCommonA
/**
* @throws Exception If failed.
*/
+ public void testSemaphore() throws Exception {
+ Ignite clientNode = clientIgnite();
+ Ignite srvNode = serverNode();
+
+ testSemaphore(clientNode, srvNode);
+ testSemaphore(srvNode, clientNode);
+ }
+
+ /**
+ * @param creator Creator node.
+ * @param other Other node.
+ * @throws Exception If failed.
+ */
+ private void testSemaphore(Ignite creator, final Ignite other) throws Exception {
+ assertNull(creator.semaphore("semaphore1", 1, true, false));
+ assertNull(other.semaphore("semaphore1", 1, true, false));
+
+ try (IgniteSemaphore semaphore = creator.semaphore("semaphore1", -1, true, true)) {
+ assertNotNull(semaphore);
+
+ assertEquals(-1, semaphore.availablePermits());
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ U.sleep(1000);
+
+ IgniteSemaphore semaphore0 = other.semaphore("semaphore1", -1, true, false);
+
+ assertEquals(-1, semaphore0.availablePermits());
+
+ log.info("Release semaphore.");
+
+ semaphore0.release(2);
+
+ return null;
+ }
+ });
+
+ log.info("Acquire semaphore.");
+
+ assertTrue(semaphore.tryAcquire(1, 5000, TimeUnit.MILLISECONDS));
+
+ log.info("Finished wait.");
+
+ fut.get();
+
+ assertEquals(0, semaphore.availablePermits());
+ }
+
+ assertNull(creator.semaphore("semaphore1", 1, true, false));
+ assertNull(other.semaphore("semaphore1", 1, true, false));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testQueue() throws Exception {
Ignite clientNode = clientIgnite();
Ignite srvNode = serverNode();
@@ -343,4 +400,4 @@ public abstract class IgniteClientDataStructuresAbstractTest extends GridCommonA
return ignite;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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 f5305a2..4a21765 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
@@ -30,6 +30,7 @@ import org.apache.ignite.IgniteAtomicStamped;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteSet;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMemoryMode;
@@ -239,7 +240,7 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
private void testUniqueName(final boolean singleGrid) throws Exception {
final String name = IgniteUuid.randomUuid().toString();
- final int DS_TYPES = 7;
+ final int DS_TYPES = 8;
final int THREADS = DS_TYPES * 3;
@@ -314,6 +315,12 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
break;
+ case 7:
+ log.info("Create atomic semaphore, grid: " + ignite.name());
+
+ res = ignite.semaphore(name, 0, false, true);
+
+ break;
default:
fail();
@@ -352,7 +359,8 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
res instanceof IgniteAtomicStamped ||
res instanceof IgniteCountDownLatch ||
res instanceof IgniteQueue ||
- res instanceof IgniteSet);
+ res instanceof IgniteSet ||
+ res instanceof IgniteSemaphore);
log.info("Data structure created: " + dataStructure);
@@ -371,4 +379,4 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
dataStructure.close();
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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
new file mode 100644
index 0000000..977a414
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteSemaphoreAbstractSelfTest.java
@@ -0,0 +1,328 @@
+/*
+ * 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.Random;
+import java.util.concurrent.Callable;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.IgniteSemaphore;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteKernal;
+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.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;
+
+/**
+ * Cache semaphore self test.
+ */
+public abstract class IgniteSemaphoreAbstractSelfTest 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 testSemaphore() throws Exception {
+ checkSemaphore();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void checkSemaphore() throws Exception {
+ // Test API.
+
+ checkAcquire();
+
+ checkRelease();
+
+ checkFair();
+
+ // Test main functionality.
+ IgniteSemaphore semaphore1 = grid(0).semaphore("semaphore", -2, true, true);
+
+ assertEquals(-2, semaphore1.availablePermits());
+
+ 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 semaphore in multiple threads on each node.
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(
+ new Callable<Object>() {
+ @Nullable @Override public Object call() throws Exception {
+ IgniteSemaphore semaphore = ignite.semaphore("semaphore", -2, true, true);
+
+ assert semaphore != null && semaphore.availablePermits() == -2;
+
+ log.info("Thread is going to wait on semaphore: " + Thread.currentThread().getName());
+
+ assert semaphore.tryAcquire(1, 1, MINUTES);
+
+ log.info("Thread is again runnable: " + Thread.currentThread().getName());
+
+ semaphore.release();
+
+ return null;
+ }
+ },
+ 5,
+ "test-thread"
+ );
+
+ fut.get();
+
+ return null;
+ }
+ });
+
+ IgniteFuture<Object> fut = comp.future();
+
+ Thread.sleep(3000);
+
+ semaphore1.release(2);
+
+ assert semaphore1.availablePermits() == 0;
+
+ semaphore1.release(1);
+
+ // Ensure there are no hangs.
+ fut.get();
+
+ // Test operations on removed semaphore.
+ semaphore1.close();
+
+ checkRemovedSemaphore(semaphore1);
+ }
+
+ /**
+ * @param semaphore Semaphore.
+ * @throws Exception If failed.
+ */
+ protected void checkRemovedSemaphore(final IgniteSemaphore semaphore) throws Exception {
+ assert GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ return semaphore.removed();
+ }
+ }, 5000);
+
+ assert semaphore.removed();
+ }
+
+ /**
+ * @throws Exception Exception.
+ */
+ private void checkFair() throws Exception {
+ // Checks only if semaphore is initialized properly
+ IgniteSemaphore semaphore = createSemaphore("rmv", 5, true);
+
+ assert semaphore.isFair();
+
+ removeSemaphore("rmv");
+
+ IgniteSemaphore semaphore1 = createSemaphore("rmv1", 5, false);
+
+ assert !semaphore1.isFair();
+
+ removeSemaphore("rmv1");
+ }
+
+ /**
+ * @throws Exception Exception.
+ */
+ private void checkAcquire() throws Exception {
+ // Check only 'false' cases here. Successful await is tested over the grid.
+ IgniteSemaphore semaphore = createSemaphore("acquire", 5, false);
+
+ assert !semaphore.tryAcquire(10);
+ assert !semaphore.tryAcquire(10, 10, MICROSECONDS);
+
+ removeSemaphore("acquire");
+ }
+
+ /**
+ * @throws Exception Exception.
+ */
+ private void checkRelease() throws Exception {
+ IgniteSemaphore semaphore = createSemaphore("release", 5, false);
+
+ semaphore.release();
+ assert semaphore.availablePermits() == 6;
+
+ semaphore.release(2);
+ assert semaphore.availablePermits() == 8;
+
+ assert semaphore.drainPermits() == 8;
+ assert semaphore.availablePermits() == 0;
+
+ removeSemaphore("release");
+
+ checkRemovedSemaphore(semaphore);
+
+ IgniteSemaphore semaphore2 = createSemaphore("release2", -5, false);
+
+ semaphore2.release();
+
+ assert semaphore2.availablePermits() == -4;
+
+ semaphore2.release(2);
+
+ assert semaphore2.availablePermits() == -2;
+
+ assert semaphore2.drainPermits() == -2;
+
+ assert semaphore2.availablePermits() == 0;
+
+ removeSemaphore("release2");
+
+ checkRemovedSemaphore(semaphore2);
+ }
+
+ /**
+ * @param semaphoreName Semaphore name.
+ * @param numPermissions Initial number of permissions.
+ * @param fair Fairness flag.
+ * @return New semaphore.
+ * @throws Exception If failed.
+ */
+ private IgniteSemaphore createSemaphore(String semaphoreName, int numPermissions, boolean fair)
+ throws Exception {
+ IgniteSemaphore semaphore = grid(RND.nextInt(NODES_CNT)).semaphore(semaphoreName, numPermissions, fair, true);
+
+ // Test initialization.
+ assert semaphoreName.equals(semaphore.name());
+ assert semaphore.availablePermits() == numPermissions;
+ assert semaphore.getQueueLength() == 0;
+ assert semaphore.isFair() == fair;
+
+ return semaphore;
+ }
+
+ /**
+ * @param semaphoreName Semaphore name.
+ * @throws Exception If failed.
+ */
+ private void removeSemaphore(String semaphoreName)
+ throws Exception {
+ IgniteSemaphore semaphore = grid(RND.nextInt(NODES_CNT)).semaphore(semaphoreName, 10, false, true);
+
+ assert semaphore != null;
+
+ if (semaphore.availablePermits() < 0)
+ semaphore.release(-semaphore.availablePermits());
+
+ // Remove semaphore on random node.
+ IgniteSemaphore semaphore0 = grid(RND.nextInt(NODES_CNT)).semaphore(semaphoreName, 0, false, true);
+
+ assertNotNull(semaphore0);
+
+ semaphore0.close();
+
+ // Ensure semaphore is removed on all nodes.
+ for (Ignite g : G.allGrids())
+ assertNull(((IgniteKernal)g).context().dataStructures().semaphore(semaphoreName, 10, true, false));
+
+ checkRemovedSemaphore(semaphore);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSemaphoreMultinode1() throws Exception {
+ if (gridCount() == 1)
+ return;
+
+ IgniteSemaphore semaphore = grid(0).semaphore("s1", 0, true, 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 {
+ IgniteSemaphore semaphore = ignite.semaphore("s1", 0, true, false);
+
+ assertNotNull(semaphore);
+
+ boolean wait = semaphore.tryAcquire(30_000, MILLISECONDS);
+
+ assertTrue(wait);
+
+ return null;
+ }
+ }));
+ }
+
+ for (int i = 0; i < 10; i++)
+ semaphore.release();
+
+ for (IgniteInternalFuture<?> fut : futs)
+ fut.get(30_000);
+ }
+
+ /** {@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/8c6852d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalSemaphoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalSemaphoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalSemaphoreSelfTest.java
new file mode 100644
index 0000000..a516fc1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/local/IgniteLocalSemaphoreSelfTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.IgniteSemaphore;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.datastructures.IgniteSemaphoreAbstractSelfTest;
+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 IgniteLocalSemaphoreSelfTest extends IgniteSemaphoreAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode atomicsCacheMode() {
+ return LOCAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void testSemaphore() throws Exception {
+ // Test main functionality.
+ IgniteSemaphore semaphore = grid(0).semaphore("semaphore", -2, false, true);
+
+ assertNotNull(semaphore);
+
+ assertEquals(-2, semaphore.availablePermits());
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(
+ new Callable<Object>() {
+ @Nullable @Override public Object call() throws Exception {
+ IgniteSemaphore semaphore = grid(0).semaphore("semaphore", -2, false, true);
+
+ assert semaphore != null && semaphore.availablePermits() == -2;
+
+ info("Thread is going to wait on semaphore: " + Thread.currentThread().getName());
+
+ assert semaphore.tryAcquire(1, 1, MINUTES);
+
+ info("Thread is again runnable: " + Thread.currentThread().getName());
+
+ semaphore.release();
+
+ return null;
+ }
+ },
+ THREADS_CNT,
+ "test-thread"
+ );
+
+ Thread.sleep(3000);
+
+ assert semaphore.availablePermits() == -2;
+
+ semaphore.release(2);
+
+ assert semaphore.availablePermits() == 0;
+
+ semaphore.release();
+
+ // Ensure there are no hangs.
+ fut.get();
+
+ // Test operations on removed latch.
+ IgniteSemaphore semaphore0 = grid(0).semaphore("semaphore", 0, false, false);
+
+ assertNotNull(semaphore0);
+
+ semaphore0.close();
+
+ checkRemovedSemaphore(semaphore0);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSemaphoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSemaphoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSemaphoreSelfTest.java
new file mode 100644
index 0000000..059e80a
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSemaphoreSelfTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.IgniteSemaphoreAbstractSelfTest;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+
+/**
+ *
+ */
+public class IgnitePartitionedSemaphoreSelfTest extends IgniteSemaphoreAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode atomicsCacheMode() {
+ return PARTITIONED;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedSemaphoreSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedSemaphoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedSemaphoreSelfTest.java
new file mode 100644
index 0000000..f58754f
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/replicated/IgniteReplicatedSemaphoreSelfTest.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.IgniteSemaphoreAbstractSelfTest;
+
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+
+/**
+ *
+ */
+public class IgniteReplicatedSemaphoreSelfTest extends IgniteSemaphoreAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode atomicsCacheMode() {
+ return REPLICATED;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8c6852d0/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 2fa8940..d4ca9a5 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
@@ -28,6 +28,7 @@ import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteQueue;
+import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.Ignition;
import org.apache.ignite.configuration.CollectionConfiguration;
@@ -60,6 +61,9 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
/** Count down latch name. */
private static final String TEST_LATCH_NAME = "test-latch";
+ /** Semaphore name. */
+ private static final String TEST_SEMAPHORE_NAME = "test-semaphore";
+
/** */
private static final CollectionConfiguration colCfg = new CollectionConfiguration();
@@ -69,6 +73,9 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
/** Count down latch initial count. */
private static final int LATCH_INIT_CNT = 1000;
+ /** Semaphore initial count. */
+ private static final int SEMAPHORE_INIT_CNT = 1000;
+
/** */
private static final boolean LONG = false;
@@ -88,6 +95,9 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
private static final boolean LATCH = true;
/** */
+ private static final boolean SEMAPHORE = true;
+
+ /** */
private GridCacheDataStructuresLoadTest() {
// No-op
}
@@ -95,210 +105,247 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
/** Atomic long write closure. */
private final CIX1<Ignite> longWriteClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicLong al = ignite.atomicLong(TEST_LONG_NAME, 0, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicLong al = ignite.atomicLong(TEST_LONG_NAME, 0, true);
- for (int i = 0; i < operationsPerTx; i++) {
- al.addAndGet(RAND.nextInt(MAX_INT));
+ for (int i = 0; i < operationsPerTx; i++) {
+ al.addAndGet(RAND.nextInt(MAX_INT));
- long cnt = writes.incrementAndGet();
+ long cnt = writes.incrementAndGet();
- if (cnt % WRITE_LOG_MOD == 0)
- info("Performed " + cnt + " writes.");
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
}
- }
- };
+ };
/** Atomic long read closure. */
private final CIX1<Ignite> longReadClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicLong al = ignite.atomicLong(TEST_LONG_NAME, 0, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicLong al = ignite.atomicLong(TEST_LONG_NAME, 0, true);
- for (int i = 0; i < operationsPerTx; i++) {
- al.get();
+ for (int i = 0; i < operationsPerTx; i++) {
+ al.get();
- long cnt = reads.incrementAndGet();
+ long cnt = reads.incrementAndGet();
- if (cnt % READ_LOG_MOD == 0)
- info("Performed " + cnt + " reads.");
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
}
- }
- };
+ };
/** Atomic reference write closure. */
private final CIX1<Ignite> refWriteClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicReference<Integer> ar = ignite.atomicReference(TEST_REF_NAME,
- null, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicReference<Integer> ar = ignite.atomicReference(TEST_REF_NAME,
+ null, true);
- for (int i = 0; i < operationsPerTx; i++) {
- ar.set(RAND.nextInt(MAX_INT));
+ for (int i = 0; i < operationsPerTx; i++) {
+ ar.set(RAND.nextInt(MAX_INT));
- long cnt = writes.incrementAndGet();
+ long cnt = writes.incrementAndGet();
- if (cnt % WRITE_LOG_MOD == 0)
- info("Performed " + cnt + " writes.");
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
}
- }
- };
+ };
/** Atomic reference read closure. */
private final CIX1<Ignite> refReadClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicReference<Integer> ar = ignite.atomicReference(TEST_REF_NAME, null,
- true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicReference<Integer> ar = ignite.atomicReference(TEST_REF_NAME, null,
+ true);
- for (int i = 0; i < operationsPerTx; i++) {
- ar.get();
+ for (int i = 0; i < operationsPerTx; i++) {
+ ar.get();
- long cnt = reads.incrementAndGet();
+ long cnt = reads.incrementAndGet();
- if (cnt % READ_LOG_MOD == 0)
- info("Performed " + cnt + " reads.");
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
}
- }
- };
+ };
/** Atomic sequence write closure. */
private final CIX1<Ignite> seqWriteClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicSequence as = ignite.atomicSequence(TEST_SEQ_NAME, 0, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicSequence as = ignite.atomicSequence(TEST_SEQ_NAME, 0, true);
- for (int i = 0; i < operationsPerTx; i++) {
- as.addAndGet(RAND.nextInt(MAX_INT) + 1);
+ for (int i = 0; i < operationsPerTx; i++) {
+ as.addAndGet(RAND.nextInt(MAX_INT) + 1);
- long cnt = writes.incrementAndGet();
+ long cnt = writes.incrementAndGet();
- if (cnt % WRITE_LOG_MOD == 0)
- info("Performed " + cnt + " writes.");
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
}
- }
- };
+ };
/** Atomic sequence read closure. */
private final CIX1<Ignite> seqReadClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicSequence as = ignite.atomicSequence(TEST_SEQ_NAME, 0, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicSequence as = ignite.atomicSequence(TEST_SEQ_NAME, 0, true);
- for (int i = 0; i < operationsPerTx; i++) {
- as.get();
+ for (int i = 0; i < operationsPerTx; i++) {
+ as.get();
- long cnt = reads.incrementAndGet();
+ long cnt = reads.incrementAndGet();
- if (cnt % READ_LOG_MOD == 0)
- info("Performed " + cnt + " reads.");
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
}
- }
- };
+ };
/** Atomic stamped write closure. */
private final CIX1<Ignite> stampWriteClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicStamped<Integer, Integer> as = ignite.atomicStamped(TEST_STAMP_NAME,
- 0, 0, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicStamped<Integer, Integer> as = ignite.atomicStamped(TEST_STAMP_NAME,
+ 0, 0, true);
- for (int i = 0; i < operationsPerTx; i++) {
- as.set(RAND.nextInt(MAX_INT), RAND.nextInt(MAX_INT));
+ for (int i = 0; i < operationsPerTx; i++) {
+ as.set(RAND.nextInt(MAX_INT), RAND.nextInt(MAX_INT));
- long cnt = writes.incrementAndGet();
+ long cnt = writes.incrementAndGet();
- if (cnt % WRITE_LOG_MOD == 0)
- info("Performed " + cnt + " writes.");
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
}
- }
- };
+ };
/** Atomic stamped read closure. */
private final CIX1<Ignite> stampReadClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteAtomicStamped<Integer, Integer> as = ignite.atomicStamped(TEST_STAMP_NAME,
- 0, 0, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteAtomicStamped<Integer, Integer> as = ignite.atomicStamped(TEST_STAMP_NAME,
+ 0, 0, true);
- for (int i = 0; i < operationsPerTx; i++) {
- as.get();
+ for (int i = 0; i < operationsPerTx; i++) {
+ as.get();
- long cnt = reads.incrementAndGet();
+ long cnt = reads.incrementAndGet();
- if (cnt % READ_LOG_MOD == 0)
- info("Performed " + cnt + " reads.");
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
}
- }
- };
+ };
/** Queue write closure. */
private final CIX1<Ignite> queueWriteClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteQueue<Integer> q = ignite.queue(TEST_QUEUE_NAME, 0, colCfg);
+ @Override public void applyx(Ignite ignite) {
+ IgniteQueue<Integer> q = ignite.queue(TEST_QUEUE_NAME, 0, colCfg);
- for (int i = 0; i < operationsPerTx; i++) {
- q.put(RAND.nextInt(MAX_INT));
+ for (int i = 0; i < operationsPerTx; i++) {
+ q.put(RAND.nextInt(MAX_INT));
- long cnt = writes.incrementAndGet();
+ long cnt = writes.incrementAndGet();
- if (cnt % WRITE_LOG_MOD == 0)
- info("Performed " + cnt + " writes.");
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
}
- }
- };
+ };
/** Queue read closure. */
private final CIX1<Ignite> queueReadClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteQueue<Integer> q = ignite.queue(TEST_QUEUE_NAME, 0, colCfg);
+ @Override public void applyx(Ignite ignite) {
+ IgniteQueue<Integer> q = ignite.queue(TEST_QUEUE_NAME, 0, colCfg);
- for (int i = 0; i < operationsPerTx; i++) {
- q.peek();
+ for (int i = 0; i < operationsPerTx; i++) {
+ q.peek();
- long cnt = reads.incrementAndGet();
+ long cnt = reads.incrementAndGet();
- if (cnt % READ_LOG_MOD == 0)
- info("Performed " + cnt + " reads.");
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
}
- }
- };
+ };
/** Count down latch write closure. */
private final CIX1<Ignite> latchWriteClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteCountDownLatch l = ignite.countDownLatch(TEST_LATCH_NAME, LATCH_INIT_CNT, true, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteCountDownLatch l = ignite.countDownLatch(TEST_LATCH_NAME, LATCH_INIT_CNT, true, true);
- for (int i = 0; i < operationsPerTx; i++) {
- l.countDown();
+ for (int i = 0; i < operationsPerTx; i++) {
+ l.countDown();
- long cnt = writes.incrementAndGet();
+ long cnt = writes.incrementAndGet();
- if (cnt % WRITE_LOG_MOD == 0)
- info("Performed " + cnt + " writes.");
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
}
- }
- };
+ };
/** Count down latch read closure. */
private final CIX1<Ignite> latchReadClos =
new CIX1<Ignite>() {
- @Override public void applyx(Ignite ignite) {
- IgniteCountDownLatch l = ignite.countDownLatch(TEST_LATCH_NAME, LATCH_INIT_CNT, true, true);
+ @Override public void applyx(Ignite ignite) {
+ IgniteCountDownLatch l = ignite.countDownLatch(TEST_LATCH_NAME, LATCH_INIT_CNT, true, true);
- for (int i = 0; i < operationsPerTx; i++) {
- l.count();
+ for (int i = 0; i < operationsPerTx; i++) {
+ l.count();
- long cnt = reads.incrementAndGet();
+ long cnt = reads.incrementAndGet();
- if (cnt % READ_LOG_MOD == 0)
- info("Performed " + cnt + " reads.");
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
}
- }
- };
+ };
+
+ /** Semaphore write closure. */
+ private final CIX1<Ignite> semaphoreWriteClos =
+ new CIX1<Ignite>() {
+ @Override public void applyx(Ignite ignite) {
+ IgniteSemaphore s = ignite.semaphore(TEST_SEMAPHORE_NAME, SEMAPHORE_INIT_CNT, false, true);
+
+ for (int i = 0; i < operationsPerTx; i++) {
+ if ((i % 2) == 0)
+ s.release();
+ else
+ s.acquire();
+
+ long cnt = writes.incrementAndGet();
+
+ if (cnt % WRITE_LOG_MOD == 0)
+ info("Performed " + cnt + " writes.");
+ }
+ }
+ };
+
+ /** Semaphore read closure. */
+ private final CIX1<Ignite> semaphoreReadClos =
+ new CIX1<Ignite>() {
+ @Override public void applyx(Ignite ignite) {
+ IgniteSemaphore s = ignite.semaphore(TEST_SEMAPHORE_NAME, SEMAPHORE_INIT_CNT, false, true);
+
+ for (int i = 0; i < operationsPerTx; i++) {
+ s.availablePermits();
+
+ long cnt = reads.incrementAndGet();
+
+ if (cnt % READ_LOG_MOD == 0)
+ info("Performed " + cnt + " reads.");
+ }
+ }
+ };
/**
* @param args Arguments.
@@ -362,6 +409,14 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
test.loadTestIgnite(test.latchWriteClos, test.latchReadClos);
}
+
+ System.gc();
+
+ if (SEMAPHORE) {
+ info("Testing semaphore...");
+
+ test.loadTestIgnite(test.semaphoreWriteClos, test.semaphoreReadClos);
+ }
}
}
@@ -407,7 +462,7 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
@Nullable @Override public Object call() throws Exception {
long start = System.currentTimeMillis();
- while(!done.get()) {
+ while (!done.get()) {
if (tx) {
try (Transaction tx = ignite.transactions().txStart()) {
readClos.apply(ignite);
@@ -447,4 +502,4 @@ public final class GridCacheDataStructuresLoadTest extends GridCacheAbstractLoad
throw new RuntimeException(e);
}
}
-}
\ No newline at end of file
+}