You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/11/18 08:22:24 UTC
ignite git commit: 'Single get' operation optimization.
Repository: ignite
Updated Branches:
refs/heads/ignite-single-op-get 007ac2b1f -> c1e501670
'Single get' operation optimization.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c1e50167
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c1e50167
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c1e50167
Branch: refs/heads/ignite-single-op-get
Commit: c1e5016709e61a3218655cab98c6e4d9f0673906
Parents: 007ac2b
Author: sboikov <sb...@gridgain.com>
Authored: Wed Nov 18 09:02:36 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Nov 18 10:19:39 2015 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheAdapter.java | 6 +-
.../processors/cache/GridCacheMvccManager.java | 54 +++++---------
.../dht/GridPartitionedGetFuture.java | 9 ++-
.../dht/colocated/GridDhtColocatedCache.java | 30 ++++----
.../cache/GridCacheAbstractFullApiSelfTest.java | 75 ++++++++++++++++++++
.../GridCacheConcurrentTxMultiNodeTest.java | 15 ----
.../cache/GridCachePartitionedGetSelfTest.java | 3 +-
7 files changed, 121 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 32c1a2c..e3cd7dd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -1497,14 +1497,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
@Override public V applyx(IgniteInternalFuture<Map<K, V>> e) throws IgniteCheckedException {
Map<K, V> map = e.get();
+ assert map.isEmpty() || map.size() == 1 : map.size();
+
if (skipVals) {
- Boolean val = !map.isEmpty();
+ Boolean val = map.isEmpty() ? false : (Boolean)F.firstValue(map);
return (V)(val);
}
- assert map.isEmpty() || map.size() == 1 : map.size();
-
return map.get(key);
}
});
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
index c289725..558a4ee 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -105,14 +104,14 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
/** Active futures mapped by version ID. */
@GridToStringExclude
- private final ConcurrentMap<GridCacheVersion, Collection<GridCacheFuture<?>>> futs = newMap();
+ private final ConcurrentMap<GridCacheVersion, Collection<GridCacheFuture<?>>> verFuts = newMap();
/** Pending atomic futures. */
private final ConcurrentMap<GridCacheVersion, GridCacheAtomicFuture<?>> atomicFuts =
new ConcurrentHashMap8<>();
/** */
- private final ConcurrentMap<IgniteUuid, GridCacheFuture<?>> futs2 = new ConcurrentHashMap8<>();
+ private final ConcurrentMap<IgniteUuid, GridCacheFuture<?>> futs = new ConcurrentHashMap8<>();
/** Near to DHT version mapping. */
private final ConcurrentMap<GridCacheVersion, GridCacheVersion> near2dht = newMap();
@@ -144,7 +143,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
prev + ']');
if (owner != null && (owner.local() || owner.nearLocal())) {
- Collection<? extends GridCacheFuture> futCol = futs.get(owner.version());
+ Collection<? extends GridCacheFuture> futCol = verFuts.get(owner.version());
if (futCol != null) {
synchronized (futCol) {
@@ -268,13 +267,13 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
public Collection<GridCacheFuture<?>> activeFutures() {
ArrayList<GridCacheFuture<?>> col = new ArrayList<>();
- for (Collection<GridCacheFuture<?>> verFuts : futs.values()) {
- synchronized (verFuts) {
- col.addAll(verFuts);
+ for (Collection<GridCacheFuture<?>> futs : verFuts.values()) {
+ synchronized (futs) {
+ col.addAll(futs);
}
}
- col.addAll(futs2.values());
+ col.addAll(futs.values());
return col;
}
@@ -433,7 +432,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
* @param futId Future ID.
*/
public void addFuture(final GridCacheFuture<?> fut, final IgniteUuid futId) {
- GridCacheFuture<?> old = futs2.put(futId, fut);
+ GridCacheFuture<?> old = futs.put(futId, fut);
assert old == null : old;
}
@@ -456,7 +455,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
return true;
while (true) {
- Collection<GridCacheFuture<?>> old = futs.get(fut.version());
+ Collection<GridCacheFuture<?>> old = verFuts.get(fut.version());
if (old == null) {
Collection<GridCacheFuture<?>> col = new HashSet<GridCacheFuture<?>>(U.capacity(4), 0.75f) {
@@ -475,7 +474,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
}
};
- old = futs.putIfAbsent(fut.version(), col);
+ old = verFuts.putIfAbsent(fut.version(), col);
}
if (old != null) {
@@ -490,7 +489,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
// Future is being removed, so we force-remove here and try again.
if (empty) {
- if (futs.remove(fut.version(), old)) {
+ if (verFuts.remove(fut.version(), old)) {
if (log.isDebugEnabled())
log.debug("Removed future list from futures map for lock version: " + fut.version());
}
@@ -559,7 +558,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
* @param futId Future ID.
*/
public void removeFuture(IgniteUuid futId) {
- futs2.remove(futId);
+ futs.remove(futId);
}
/**
@@ -571,7 +570,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
if (!fut.trackable())
return true;
- Collection<GridCacheFuture<?>> cur = futs.get(fut.version());
+ Collection<GridCacheFuture<?>> cur = verFuts.get(fut.version());
if (cur == null)
return false;
@@ -591,7 +590,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
else if (log.isDebugEnabled())
log.debug("Attempted to remove a non-registered future (has it been already removed?): " + fut);
- if (empty && futs.remove(fut.version(), cur))
+ if (empty && verFuts.remove(fut.version(), cur))
if (log.isDebugEnabled())
log.debug("Removed future list from futures map for lock version: " + fut.version());
@@ -607,7 +606,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
*/
@SuppressWarnings({"unchecked"})
@Nullable public GridCacheFuture future(GridCacheVersion ver, IgniteUuid futId) {
- Collection<? extends GridCacheFuture> futs = this.futs.get(ver);
+ Collection<? extends GridCacheFuture> futs = this.verFuts.get(ver);
if (futs != null) {
synchronized (futs) {
@@ -633,26 +632,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
* @return Found future.
*/
@Nullable public GridCacheFuture future(IgniteUuid futId) {
- return futs2.get(futId);
- }
-
- /**
- * Gets all futures for given lock version, possibly empty collection.
- *
- * @param ver Version.
- * @return All futures for given lock version.
- */
- @SuppressWarnings("unchecked")
- public <T> Collection<? extends IgniteInternalFuture<T>> futures(GridCacheVersion ver) {
- Collection c = futs.get(ver);
-
- if (c == null)
- return Collections.<IgniteInternalFuture<T>>emptyList();
- else {
- synchronized (c) {
- return new ArrayList<>((Collection<IgniteInternalFuture<T>>)c);
- }
- }
+ return futs.get(futId);
}
/**
@@ -985,7 +965,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter {
X.println(">>> rmvLocksSize: " + rmvLocks.sizex());
X.println(">>> dhtLocCandsSize: " + dhtLocCands.size());
X.println(">>> lockedSize: " + locked.size());
- X.println(">>> futsSize: " + futs.size());
+ X.println(">>> futsSize: " + (verFuts.size() + futs.size()));
X.println(">>> near2dhtSize: " + near2dht.size());
X.println(">>> finishFutsSize: " + finishFuts.sizex());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 5a9b6c7..d5128ba 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -275,9 +275,11 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
add(new GridFinishedFuture<>(locVals));
if (hasRmtNodes) {
- trackable = true;
+ if (!trackable) {
+ trackable = true;
- cctx.mvcc().addFuture(this, futId);
+ cctx.mvcc().addFuture(this, futId);
+ }
}
// Create mini futures.
@@ -537,7 +539,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
if (needVer)
versionedResult(map, info.key(), info.value(), info.version());
- else
+ else {
cctx.addResult(map,
info.key(),
info.value(),
@@ -545,6 +547,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
keepCacheObjects,
deserializePortable,
false);
+ }
}
return map;
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index aebb634..a8b36e2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -211,6 +211,12 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
assert map.isEmpty() || map.size() == 1 : map.size();
+ if (skipVals) {
+ Boolean val = map.isEmpty() ? false : (Boolean)F.firstValue(map);
+
+ return (V)(val);
+ }
+
return (V)map.get(key);
}
});
@@ -225,18 +231,18 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
subjId = ctx.subjectIdPerCall(subjId, opCtx);
GridPartitionedSingleGetFuture fut = new GridPartitionedSingleGetFuture(ctx,
- ctx.toCacheKeyObject(key),
- topVer,
- opCtx == null || !opCtx.skipStore(),
- forcePrimary,
- subjId,
- taskName,
- deserializePortable,
- skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
- skipVals,
- canRemap,
- /*needVer*/false,
- /*keepCacheObjects*/false);
+ ctx.toCacheKeyObject(key),
+ topVer,
+ opCtx == null || !opCtx.skipStore(),
+ forcePrimary,
+ subjId,
+ taskName,
+ deserializePortable,
+ skipVals ? null : expiryPolicy(opCtx != null ? opCtx.expiry() : null),
+ skipVals,
+ canRemap,
+ /*needVer*/false,
+ /*keepCacheObjects*/false);
fut.init();
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 1d79e20..e8e86e9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -382,6 +382,81 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
/**
* @throws Exception If failed.
*/
+ public void testContainsKeyTx() throws Exception {
+ if (!txEnabled())
+ return;
+
+ IgniteCache<String, Integer> cache = jcache();
+
+ IgniteTransactions txs = ignite(0).transactions();
+
+ for (int i = 0; i < 10; i++) {
+ String key = String.valueOf(i);
+
+ try (Transaction tx = txs.txStart()) {
+ assertNull(key, cache.get(key));
+
+ assertFalse(cache.containsKey(key));
+
+ tx.commit();
+ }
+
+ try (Transaction tx = txs.txStart()) {
+ assertNull(key, cache.get(key));
+
+ cache.put(key, i);
+
+ assertTrue(cache.containsKey(key));
+
+ tx.commit();
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testContainsKeysTx() throws Exception {
+ if (!txEnabled())
+ return;
+
+ IgniteCache<String, Integer> cache = jcache();
+
+ IgniteTransactions txs = ignite(0).transactions();
+
+ Set<String> keys = new HashSet<>();
+
+ for (int i = 0; i < 10; i++) {
+ String key = String.valueOf(i);
+
+ keys.add(key);
+ }
+
+ try (Transaction tx = txs.txStart()) {
+ for (String key : keys)
+ assertNull(key, cache.get(key));
+
+ assertFalse(cache.containsKeys(keys));
+
+ tx.commit();
+ }
+
+ try (Transaction tx = txs.txStart()) {
+ for (String key : keys)
+ assertNull(key, cache.get(key));
+
+ for (String key : keys)
+ cache.put(key, 0);
+
+ assertTrue(cache.containsKeys(keys));
+
+ tx.commit();
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
public void testRemoveInExplicitLocks() throws Exception {
if (lockingEnabled()) {
IgniteCache<String, Integer> cache = jcache();
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
index 1ef77f2..45c8c2c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentTxMultiNodeTest.java
@@ -613,21 +613,6 @@ public class GridCacheConcurrentTxMultiNodeTest extends GridCommonAbstractTest {
X.println("Near entry [grid="+ g.name() + ", key=" + k + ", entry=" + nearEntry);
X.println("DHT entry [grid=" + g.name() + ", key=" + k + ", entry=" + dhtEntry);
-
- GridCacheMvccCandidate nearCand =
- nearEntry == null ? null : F.first(nearEntry.localCandidates());
-
- if (nearCand != null)
- X.println("Near futures: " +
- nearEntry.context().mvcc().futures(nearCand.version()));
-
- GridCacheMvccCandidate dhtCand =
- dhtEntry == null ? null : F.first(dhtEntry.localCandidates());
-
- if (dhtCand != null)
- X.println("Dht futures: " +
- dhtEntry.context().mvcc().futures(dhtCand.version()));
-
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c1e50167/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
index cbfc97b..2cb4a00 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePartitionedGetSelfTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
@@ -228,7 +229,7 @@ public class GridCachePartitionedGetSelfTest extends GridCommonAbstractTest {
@Override public void onMessage(UUID nodeId, Object msg) {
info("Received message from node [nodeId=" + nodeId + ", msg=" + msg + ']');
- if (msg instanceof GridNearGetRequest) {
+ if (msg instanceof GridNearSingleGetRequest) {
info("Setting flag: " + System.identityHashCode(received));
received.set(true);