You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/04/27 10:09:26 UTC
[1/6] ignite git commit: IGNITE-2948 - Optimize usage of
GridCacheConcurrentMap
Repository: ignite
Updated Branches:
refs/heads/ignite-1786 7f0737d5a -> 307c21785
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSequenceApiSelfAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSequenceApiSelfAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSequenceApiSelfAbstractTest.java
index 8669863..d988b2c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSequenceApiSelfAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSequenceApiSelfAbstractTest.java
@@ -31,7 +31,6 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
-import org.apache.ignite.internal.processors.cache.GridCacheInternal;
import org.apache.ignite.internal.processors.cache.GridCacheUtils;
import org.apache.ignite.internal.processors.datastructures.GridCacheInternalKeyImpl;
import org.apache.ignite.internal.util.typedef.G;
@@ -317,25 +316,6 @@ public abstract class GridCacheSequenceApiSelfAbstractTest extends IgniteAtomics
/**
* @throws Exception If failed.
*/
- public void testEviction() throws Exception {
- String locSeqName = UUID.randomUUID().toString();
-
- IgniteAtomicSequence locSeq = grid().atomicSequence(locSeqName, 0, true);
-
- locSeq.addAndGet(153);
-
- GridCacheAdapter cache = ((IgniteKernal)grid()).internalCache(GridCacheUtils.ATOMICS_CACHE_NAME);
-
- assertNotNull(cache);
-
- cache.evictAll(cache.keySet());
-
- assert null != cache.get(new GridCacheInternalKeyImpl(locSeqName));
- }
-
- /**
- * @throws Exception If failed.
- */
public void testRemove() throws Exception {
String locSeqName = UUID.randomUUID().toString();
@@ -376,23 +356,6 @@ public abstract class GridCacheSequenceApiSelfAbstractTest extends IgniteAtomics
}
}, IllegalStateException.class, null);
- for (Object o : cache.keySet())
- assert !(o instanceof GridCacheInternal) : "Wrong keys [key=" + o + ']';
-
- for (Object o : cache.values())
- assert !(o instanceof GridCacheInternal) : "Wrong values [value=" + o + ']';
-
- for (Object o : cache.entrySet())
- assert !(o instanceof GridCacheInternal) : "Wrong entries [entry=" + o + ']';
-
- assert cache.keySet().isEmpty();
-
- assert cache.values().isEmpty();
-
- assert cache.entrySet().isEmpty();
-
- assert cache.size() == 0;
-
for (String seqName : seqNames)
assert null != cache.get(new GridCacheInternalKeyImpl(seqName));
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetAbstractSelfTest.java
index c63df40..9973b27 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetAbstractSelfTest.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -798,8 +799,8 @@ public abstract class GridCacheSetAbstractSelfTest extends IgniteCollectionAbstr
GridCacheContext cctx = GridTestUtils.getFieldValue(set0, "cctx");
for (int i = 0; i < gridCount(); i++) {
- Iterator<GridCacheEntryEx> entries =
- (grid(i)).context().cache().internalCache(cctx.name()).map().allEntries0().iterator();
+ Iterator<GridCacheMapEntry> entries =
+ (grid(i)).context().cache().internalCache(cctx.name()).map().entries().iterator();
while (entries.hasNext()) {
GridCacheEntryEx entry = entries.next();
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetFailoverAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetFailoverAbstractSelfTest.java
index ca57205..e6e09c7 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheSetFailoverAbstractSelfTest.java
@@ -29,8 +29,8 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteSet;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.internal.processors.datastructures.SetItemKey;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
@@ -174,8 +174,8 @@ public abstract class GridCacheSetFailoverAbstractSelfTest extends IgniteCollect
Set<IgniteUuid> setIds = new HashSet<>();
for (int i = 0; i < gridCount(); i++) {
- Iterator<GridCacheEntryEx> entries =
- ((IgniteKernal)grid(i)).context().cache().internalCache().map().allEntries0().iterator();
+ Iterator<GridCacheMapEntry> entries =
+ grid(i).context().cache().internalCache().map().entries().iterator();
while (entries.hasNext()) {
GridCacheEntryEx entry = entries.next();
@@ -228,4 +228,4 @@ public abstract class GridCacheSetFailoverAbstractSelfTest extends IgniteCollect
}
});
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/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 4a21765..e88c97b 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
@@ -27,6 +27,7 @@ import org.apache.ignite.IgniteAtomicLong;
import org.apache.ignite.IgniteAtomicReference;
import org.apache.ignite.IgniteAtomicSequence;
import org.apache.ignite.IgniteAtomicStamped;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteCountDownLatch;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteQueue;
@@ -349,7 +350,7 @@ public class IgniteDataStructureUniqueNameTest extends IgniteCollectionAbstractT
for (IgniteInternalFuture<Object> fut : futs) {
Object res = fut.get();
- if (res instanceof IgniteException)
+ if (res instanceof IgniteException || res instanceof IgniteCheckedException)
continue;
assertTrue("Unexpected object: " + res,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedQueueNoBackupsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedQueueNoBackupsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedQueueNoBackupsTest.java
index 880c638..f57e1c9 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedQueueNoBackupsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedQueueNoBackupsTest.java
@@ -26,7 +26,7 @@ import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.configuration.CollectionConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.testframework.GridTestUtils;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
@@ -79,8 +79,8 @@ public class IgnitePartitionedQueueNoBackupsTest extends GridCachePartitionedQue
for (int i = 0; i < gridCount(); i++) {
IgniteKernal grid = (IgniteKernal)grid(i);
- Iterator<GridCacheEntryEx> entries =
- grid.context().cache().internalCache(cctx.name()).map().allEntries0().iterator();
+ Iterator<GridCacheMapEntry> entries =
+ grid.context().cache().internalCache(cctx.name()).map().entries().iterator();
if (entries.hasNext()) {
if (setNodeId == null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSetNoBackupsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSetNoBackupsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSetNoBackupsSelfTest.java
index 2d23d2a..a73aa4a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSetNoBackupsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/partitioned/IgnitePartitionedSetNoBackupsSelfTest.java
@@ -23,7 +23,7 @@ import java.util.UUID;
import org.apache.ignite.configuration.CollectionConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.testframework.GridTestUtils;
/**
@@ -57,8 +57,8 @@ public class IgnitePartitionedSetNoBackupsSelfTest extends GridCachePartitionedS
for (int i = 0; i < gridCount(); i++) {
IgniteKernal grid = (IgniteKernal)grid(i);
- Iterator<GridCacheEntryEx> entries =
- grid.context().cache().internalCache(cctx.name()).map().allEntries0().iterator();
+ Iterator<GridCacheMapEntry> entries =
+ grid.context().cache().internalCache(cctx.name()).map().entries().iterator();
if (entries.hasNext()) {
if (setNodeId == null)
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtTestUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtTestUtils.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtTestUtils.java
index dd46e23..377a55f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtTestUtils.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCacheDhtTestUtils.java
@@ -85,7 +85,12 @@ public class GridCacheDhtTestUtils {
for (int i = 0; i < keyCnt; i++) {
KeyCacheObject cacheKey = ctx.toCacheKeyObject(i);
- cacheMap.putEntry(AffinityTopologyVersion.NONE, cacheKey, ctx.toCacheKeyObject("value" + i));
+ cacheMap.putEntryIfObsoleteOrAbsent(
+ AffinityTopologyVersion.NONE,
+ cacheKey,
+ ctx.toCacheKeyObject("value" + i),
+ false,
+ false);
dht.preloader().request(Collections.singleton(cacheKey), AffinityTopologyVersion.NONE);
@@ -224,4 +229,4 @@ public class GridCacheDhtTestUtils {
", entry=" + entry + ", node partitions=" + affParts + "]");
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOneNodeSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOneNodeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOneNodeSelfTest.java
index 4c92f90..d0fe936 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOneNodeSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOneNodeSelfTest.java
@@ -59,8 +59,8 @@ public class GridCacheNearOneNodeSelfTest extends GridCommonAbstractTest {
jcache().removeAll();
- assertEquals("DHT entries: " + dht().entries(), 0, dht().size());
- assertEquals("Near entries: " + near().entries(), 0, near().size());
+ assertEquals("DHT entries: " + dht().entrySet(), 0, dht().size());
+ assertEquals("Near entries: " + near().entrySet(), 0, near().size());
assertEquals(0, jcache().size());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
index e4d39d6..d67092f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsAbstractSelfTest.java
@@ -75,7 +75,7 @@ import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
-import java.util.Set;
+
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -1021,9 +1021,9 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest {
}
}
}, 10_000)) {
- Set<GridCacheEntryEx> set = dataCache.allEntries();
+ Iterable<? extends GridCacheEntryEx> entries = dataCache.allEntries();
- for (GridCacheEntryEx e: set) {
+ for (GridCacheEntryEx e: entries) {
X.println("deleted = " + e.deleted());
X.println("detached = " + e.detached());
X.println("info = " + e.info());
@@ -3238,9 +3238,9 @@ public abstract class IgfsAbstractSelfTest extends IgfsCommonAbstractTest {
private static void dumpCache(String cacheName, GridCacheAdapter<?,?> cache) {
X.println("=============================== " + cacheName + " cache dump: ");
- Set<GridCacheEntryEx> set = cache.entries();
+ Iterable<? extends GridCacheEntryEx> entries = cache.entries();
- for (GridCacheEntryEx e: set)
+ for (GridCacheEntryEx e: entries)
X.println("Lost " + cacheName + " entry = " + e);
}
[3/6] ignite git commit: IGNITE-2948 - Optimize usage of
GridCacheConcurrentMap
Posted by vo...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java
deleted file mode 100644
index 6d18b7d..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheKeySet.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import javax.cache.Cache;
-import org.apache.ignite.internal.util.GridSerializableSet;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.lang.IgnitePredicate;
-
-/**
- * Key set based on provided entries with all remove operations backed
- * by underlying cache.
- */
-public class GridCacheKeySet<K, V> extends GridSerializableSet<K> {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Cache context. */
- private final GridCacheContext<K, V> ctx;
-
- /** Filter. */
- private final IgnitePredicate<Cache.Entry<K, V>>[] filter;
-
- /** Base map. */
- private final Map<K, Cache.Entry<K, V>> map;
-
- /**
- * @param ctx Cache context.
- * @param c Entry collection.
- * @param filter Filter.
- */
- public GridCacheKeySet(GridCacheContext<K, V> ctx, Collection<? extends Cache.Entry<K, V>> c,
- IgnitePredicate<Cache.Entry<K, V>>[] filter) {
- map = new HashMap<>();
-
- assert ctx != null;
-
- this.ctx = ctx;
- this.filter = filter == null ? CU.<K, V>empty() : filter;
-
- for (Cache.Entry<K, V> e : c) {
- if (e != null)
- map.put(e.getKey(), e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public Iterator<K> iterator() {
- return new GridCacheIterator<>(ctx, map.values(), F.<K, V>cacheEntry2Key(), filter);
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- throw new UnsupportedOperationException();
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"SuspiciousMethodCalls"})
- @Override public boolean remove(Object o) {
- Cache.Entry<K, V> e = map.get(o);
-
- if (e == null || !F.isAll(e, filter))
- return false;
-
- map.remove(o);
-
- ctx.grid().cache(ctx.name()).remove(e.getKey());
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return F.size(map.values(), filter);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"SuspiciousMethodCalls"})
- @Override public boolean contains(Object o) {
- Cache.Entry<K, V> e = map.get(o);
-
- return e != null && F.isAll(e, filter);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 75d96d8..c9ff138 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -3192,7 +3192,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (rmv) {
onMarkedObsolete();
- cctx.cache().map().removeEntry(this);
+ cctx.cache().removeEntry(this);
}
}
}
@@ -4229,7 +4229,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (rmv) {
onMarkedObsolete();
- cctx.cache().map().removeEntry(this);
+ cctx.cache().removeEntry(this);
}
}
@@ -4287,18 +4287,32 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
flags |= IS_DELETED_MASK;
- cctx.decrementPublicSize(this);
+ decrementMapPublicSize();
}
else {
assert deletedUnlocked() : this;
flags &= ~IS_DELETED_MASK;
- cctx.incrementPublicSize(this);
+ incrementMapPublicSize();
}
}
/**
+ * Increments public size of map.
+ */
+ protected void incrementMapPublicSize() {
+ cctx.incrementPublicSize(this);
+ }
+
+ /**
+ * Decrements public size of map.
+ */
+ protected void decrementMapPublicSize() {
+ cctx.decrementPublicSize(this);
+ }
+
+ /**
* @return MVCC.
*/
@Nullable protected GridCacheMvcc mvccExtras() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 5214078..3d5052b 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -79,7 +79,6 @@ import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProce
import org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridNoStorageCacheMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearAtomicCache;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 84c5d73..a58c209 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -891,7 +891,7 @@ public class GridCacheProxyImpl<K, V> implements IgniteInternalCache<K, V>, Exte
}
/** {@inheritDoc} */
- @Override public Collection<V> values() {
+ @Override public Iterable<V> values() {
CacheOperationContext prev = gate.enter(opCtx);
try {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
index 98b1b59..e7010f5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
@@ -51,7 +51,6 @@ import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cache.store.CacheStoreSessionListener;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.configuration.FileSystemConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.internal.GridKernalContext;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridNoStorageCacheMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridNoStorageCacheMap.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridNoStorageCacheMap.java
new file mode 100644
index 0000000..2532882
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridNoStorageCacheMap.java
@@ -0,0 +1,107 @@
+/*
+ * 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;
+
+import java.util.Collections;
+import java.util.Set;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtOffHeapCacheEntry;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Empty cache map that will never store any entries.
+ */
+public class GridNoStorageCacheMap implements GridCacheConcurrentMap {
+ /** Context. */
+ private final GridCacheContext ctx;
+
+ /**
+ * @param ctx Cache context.
+ */
+ public GridNoStorageCacheMap(GridCacheContext ctx) {
+ this.ctx = ctx;
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridCacheMapEntry getEntry(KeyCacheObject key) {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridCacheMapEntry putEntryIfObsoleteOrAbsent(AffinityTopologyVersion topVer, KeyCacheObject key,
+ @Nullable CacheObject val, boolean create, boolean touch) {
+ if (create)
+ return ctx.useOffheapEntry() ?
+ new GridDhtOffHeapCacheEntry(ctx, topVer, key, key.hashCode(), val) :
+ new GridDhtCacheEntry(ctx, topVer, key, key.hashCode(), val);
+ else
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean removeEntry(GridCacheEntryEx entry) {
+ throw new AssertionError();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int publicSize() {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void incrementPublicSize(GridCacheEntryEx e) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void decrementPublicSize(GridCacheEntryEx e) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridCacheMapEntry randomEntry() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<KeyCacheObject> keySet(CacheEntryPredicate... filter) {
+ return Collections.emptySet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterable<GridCacheMapEntry> entries(CacheEntryPredicate... filter) {
+ return Collections.emptySet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterable<GridCacheMapEntry> allEntries(CacheEntryPredicate... filter) {
+ return Collections.emptySet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<GridCacheMapEntry> entrySet(CacheEntryPredicate... filter) {
+ return Collections.emptySet();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
index 5294f6a..3dc3471 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
@@ -911,7 +911,7 @@ public interface IgniteInternalCache<K, V> extends Iterable<Cache.Entry<K, V>> {
*
* @return Collection of cached values.
*/
- public Collection<V> values();
+ public Iterable<V> values();
/**
* Gets set of all entries cached on this node. You can remove
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObject.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObject.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObject.java
index c214ce3..ffb846c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObject.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObject.java
@@ -30,4 +30,15 @@ public interface KeyCacheObject extends CacheObject {
* @return {@code True} if internal cache key.
*/
public boolean internal();
+
+ /**
+ * @return Partition ID for this key or -1 if it is unknown.
+ */
+ public int partition();
+
+ /**
+ * Sets partition ID for this key.
+ * @param part Partition ID.
+ */
+ public void partition(int part);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
index e557c28..35e681c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.GridDirectTransient;
import org.jetbrains.annotations.Nullable;
/**
@@ -27,6 +28,10 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
/** */
private static final long serialVersionUID = 0L;
+ /** */
+ @GridDirectTransient
+ private int part = -1;
+
/**
*
*/
@@ -39,10 +44,30 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
* @param valBytes Value bytes.
*/
public KeyCacheObjectImpl(Object val, byte[] valBytes) {
+ this(val, valBytes, -1);
+ }
+
+ /**
+ * @param val Value.
+ * @param valBytes Value bytes.
+ * @param part Partition.
+ */
+ public KeyCacheObjectImpl(Object val, byte[] valBytes, int part) {
assert val != null;
this.val = val;
this.valBytes = valBytes;
+ this.part = part;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int partition() {
+ return part;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void partition(int part) {
+ this.part = part;
}
/** {@inheritDoc} */
@@ -96,6 +121,11 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
}
/** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
@Override public void prepareMarshal(CacheObjectContext ctx) throws IgniteCheckedException {
if (valBytes == null)
valBytes = ctx.kernalContext().cacheObjects().marshal(ctx, val);
@@ -119,4 +149,4 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
return val.equals(other.val);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
index 18e2d09..c043bc7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
@@ -773,20 +773,36 @@ public class CacheObjectBinaryProcessorImpl extends IgniteCacheObjectProcessorIm
/** {@inheritDoc} */
@Override public KeyCacheObject toCacheKeyObject(CacheObjectContext ctx, Object obj, boolean userObj) {
+ return toCacheKeyObject(ctx, obj, userObj, -1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public KeyCacheObject toCacheKeyObject(
+ CacheObjectContext ctx,
+ Object obj,
+ boolean userObj,
+ int partition
+ ) {
if (!((CacheObjectBinaryContext)ctx).binaryEnabled())
- return super.toCacheKeyObject(ctx, obj, userObj);
+ return super.toCacheKeyObject(ctx, obj, userObj, partition);
+
+ if (obj instanceof KeyCacheObject) {
+ ((KeyCacheObject)obj).partition(partition);
- if (obj instanceof KeyCacheObject)
return (KeyCacheObject)obj;
+ }
if (((CacheObjectBinaryContext)ctx).binaryEnabled()) {
obj = toBinary(obj);
- if (obj instanceof KeyCacheObject)
+ if (obj instanceof KeyCacheObject) {
+ ((KeyCacheObject)obj).partition(partition);
+
return (KeyCacheObject)obj;
+ }
}
- return toCacheKeyObject0(obj, userObj);
+ return toCacheKeyObject0(obj, userObj, partition);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
index 6e97ec5..03f6474 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
@@ -405,7 +405,7 @@ public abstract class GridDistributedCacheAdapter<K, V> extends GridCacheAdapter
try {
if (!locPart.isEmpty()) {
- for (GridDhtCacheEntry o : locPart.entries()) {
+ for (GridCacheEntryEx o : locPart.allEntries()) {
if (!o.obsoleteOrDeleted())
dataLdr.removeDataInternal(o.key());
}
@@ -428,7 +428,7 @@ public abstract class GridDistributedCacheAdapter<K, V> extends GridCacheAdapter
if (near != null) {
GridCacheVersion obsoleteVer = ctx.versions().next();
- for (GridCacheEntryEx e : near.map().allEntries0()) {
+ for (GridCacheEntryEx e : near.allEntries()) {
if (!e.valid(topVer) && e.markObsolete(obsoleteVer))
near.removeEntry(e);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
index 5d07b6f..20cd52c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
@@ -78,6 +78,10 @@ public class GridDistributedLockRequest extends GridDistributedBaseMessage {
@GridDirectCollection(KeyCacheObject.class)
private List<KeyCacheObject> keys;
+ /** Partition IDs of keys to lock. */
+ @GridDirectCollection(int.class)
+ protected List<Integer> partIds;
+
/** Array indicating whether value should be returned for a key. */
@GridToStringInclude
private boolean[] retVals;
@@ -232,7 +236,7 @@ public class GridDistributedLockRequest extends GridDistributedBaseMessage {
*
* @param skipStore Skip store flag.
*/
- private void skipStore(boolean skipStore){
+ private void skipStore(boolean skipStore) {
flags = skipStore ? (byte)(flags | SKIP_STORE_FLAG_MASK) : (byte)(flags & ~SKIP_STORE_FLAG_MASK);
}
@@ -284,11 +288,15 @@ public class GridDistributedLockRequest extends GridDistributedBaseMessage {
boolean retVal,
GridCacheContext ctx
) throws IgniteCheckedException {
- if (keys == null)
+ if (keys == null) {
keys = new ArrayList<>(keysCount());
+ partIds = new ArrayList<>(keysCount());
+ }
keys.add(key);
+ partIds.add(key.partition());
+
retVals[idx] = retVal;
idx++;
@@ -325,6 +333,13 @@ public class GridDistributedLockRequest extends GridDistributedBaseMessage {
GridCacheContext cctx = ctx.cacheContext(cacheId);
finishUnmarshalCacheObjects(keys, cctx, ldr);
+
+ if (partIds != null && !partIds.isEmpty()) {
+ assert partIds.size() == keys.size();
+
+ for (int i = 0; i < keys.size(); i++)
+ keys.get(i).partition(partIds.get(i));
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 2cf7276..2bb6bf3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -29,7 +29,6 @@ import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedUnlockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedUnlockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedUnlockRequest.java
index 213a0ff..7854ace 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedUnlockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedUnlockRequest.java
@@ -44,6 +44,10 @@ public class GridDistributedUnlockRequest extends GridDistributedBaseMessage {
@GridDirectCollection(KeyCacheObject.class)
private List<KeyCacheObject> keys;
+ /** Partition IDs. */
+ @GridDirectCollection(int.class)
+ protected List<Integer> partIds;
+
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -75,10 +79,13 @@ public class GridDistributedUnlockRequest extends GridDistributedBaseMessage {
* @throws IgniteCheckedException If failed.
*/
public void addKey(KeyCacheObject key, GridCacheContext ctx) throws IgniteCheckedException {
- if (keys == null)
+ if (keys == null) {
keys = new ArrayList<>(keysCount());
+ partIds = new ArrayList<>(keysCount());
+ }
keys.add(key);
+ partIds.add(key.partition());
}
/** {@inheritDoc}
@@ -94,6 +101,13 @@ public class GridDistributedUnlockRequest extends GridDistributedBaseMessage {
super.finishUnmarshal(ctx, ldr);
finishUnmarshalCacheObjects(keys, ctx.cacheContext(cacheId), ldr);
+
+ if (partIds != null && !partIds.isEmpty()) {
+ assert partIds.size() == keys.size();
+
+ for (int i = 0; i < keys.size(); i++)
+ keys.get(i).partition(partIds.get(i));
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedConcurrentMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedConcurrentMap.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedConcurrentMap.java
new file mode 100644
index 0000000..ff95de9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedConcurrentMap.java
@@ -0,0 +1,191 @@
+/*
+ * 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.distributed.dht;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMap;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.util.PartitionedReadOnlySet;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of GridCacheConcurrentMap that will delegate all method calls to corresponding local partition.
+ */
+public class GridCachePartitionedConcurrentMap implements GridCacheConcurrentMap {
+ /** Context. */
+ private final GridCacheContext ctx;
+
+ /**
+ * Constructor.
+ * @param ctx Context.
+ */
+ public GridCachePartitionedConcurrentMap(GridCacheContext ctx) {
+ this.ctx = ctx;
+ }
+
+ /**
+ * @param key Key.
+ * @param topVer Topology version.
+ * @param create Create flag.
+ * @return Local partition.
+ */
+ @Nullable private GridDhtLocalPartition localPartition(
+ KeyCacheObject key,
+ AffinityTopologyVersion topVer,
+ boolean create
+ ) {
+ int p = key.partition();
+
+ if (p == -1)
+ p = ctx.affinity().partition(key);
+
+ return ctx.topology().localPartition(p, topVer, create);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridCacheMapEntry getEntry(KeyCacheObject key) {
+ GridDhtLocalPartition part = localPartition(key, AffinityTopologyVersion.NONE, false);
+
+ if (part == null)
+ return null;
+
+ return part.getEntry(key);
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridCacheMapEntry putEntryIfObsoleteOrAbsent(AffinityTopologyVersion topVer, KeyCacheObject key,
+ @Nullable CacheObject val, boolean create, boolean touch) {
+ GridDhtLocalPartition part = localPartition(key, topVer, create);
+
+ if (part == null)
+ return null;
+
+ return part.putEntryIfObsoleteOrAbsent(topVer, key, val, create, touch);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ int size = 0;
+
+ for (GridDhtLocalPartition part : ctx.topology().localPartitions())
+ size += part.size();
+
+ return size;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int publicSize() {
+ int size = 0;
+
+ for (GridDhtLocalPartition part : ctx.topology().localPartitions())
+ size += part.publicSize();
+
+ return size;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void incrementPublicSize(GridCacheEntryEx e) {
+ localPartition(e.key(), AffinityTopologyVersion.NONE, true).incrementPublicSize(e);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void decrementPublicSize(GridCacheEntryEx e) {
+ localPartition(e.key(), AffinityTopologyVersion.NONE, true).decrementPublicSize(e);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean removeEntry(GridCacheEntryEx entry) {
+ GridDhtLocalPartition part = localPartition(entry.key(), AffinityTopologyVersion.NONE, false);
+
+ if (part == null)
+ return false;
+
+ return part.removeEntry(entry);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridCacheMapEntry randomEntry() {
+ return entries().iterator().next();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<KeyCacheObject> keySet(CacheEntryPredicate... filter) {
+ Collection<Set<KeyCacheObject>> sets = new ArrayList<>();
+
+ for (GridDhtLocalPartition partition : ctx.topology().localPartitions())
+ sets.add(partition.keySet(filter));
+
+ return new PartitionedReadOnlySet<>(sets);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterable<GridCacheMapEntry> entries(final CacheEntryPredicate... filter) {
+ return new Iterable<GridCacheMapEntry>() {
+ @Override public Iterator<GridCacheMapEntry> iterator() {
+ List<Iterator<GridCacheMapEntry>> iterators = new ArrayList<>();
+
+ for (GridDhtLocalPartition partition : ctx.topology().localPartitions())
+ iterators.add(partition.entries(filter).iterator());
+
+ return F.flatIterators(iterators);
+ }
+ };
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterable<GridCacheMapEntry> allEntries(final CacheEntryPredicate... filter) {
+ return new Iterable<GridCacheMapEntry>() {
+ @Override public Iterator<GridCacheMapEntry> iterator() {
+ List<Iterator<GridCacheMapEntry>> iterators = new ArrayList<>();
+
+ for (GridDhtLocalPartition partition : ctx.topology().localPartitions())
+ iterators.add(partition.allEntries(filter).iterator());
+
+ return F.flatIterators(iterators);
+ }
+ };
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<GridCacheMapEntry> entrySet(CacheEntryPredicate... filter) {
+ Collection<Set<GridCacheMapEntry>> sets = new ArrayList<>();
+
+ for (GridDhtLocalPartition partition : ctx.topology().localPartitions())
+ sets.add(partition.entrySet(filter));
+
+ return new PartitionedReadOnlySet<>(sets);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridCachePartitionedConcurrentMap.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index 3761d77..4635cad 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -357,13 +357,6 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Override public GridDhtLocalPartition onAdded(AffinityTopologyVersion topVer, GridDhtCacheEntry e) {
- assert false : "Entry should not be added to client topology: " + e;
-
- return null;
- }
-
- /** {@inheritDoc} */
@Override public void onRemoved(GridDhtCacheEntry e) {
assert false : "Entry should not be removed from client topology: " + e;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index faa980e..5ff674f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -158,9 +158,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @param ctx Context.
*/
protected GridDhtCacheAdapter(GridCacheContext<K, V> ctx) {
- super(ctx, ctx.config().getStartSize());
-
- top = new GridDhtPartitionTopologyImpl(ctx);
+ this(ctx, new GridCachePartitionedConcurrentMap(ctx));
}
/**
@@ -171,27 +169,13 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
*/
protected GridDhtCacheAdapter(GridCacheContext<K, V> ctx, GridCacheConcurrentMap map) {
super(ctx, map);
-
- top = new GridDhtPartitionTopologyImpl(ctx);
}
/** {@inheritDoc} */
@Override protected void init() {
- map.setEntryFactory(new GridCacheMapEntryFactory() {
- /** {@inheritDoc} */
- @Override public GridCacheMapEntry create(
- GridCacheContext ctx,
- AffinityTopologyVersion topVer,
- KeyCacheObject key,
- int hash,
- CacheObject val
- ) {
- if (ctx.useOffheapEntry())
- return new GridDhtOffHeapCacheEntry(ctx, topVer, key, hash, val);
+ super.init();
- return new GridDhtCacheEntry(ctx, topVer, key, hash, val);
- }
- });
+ top = new GridDhtPartitionTopologyImpl(ctx, entryFactory());
}
/** {@inheritDoc} */
@@ -252,6 +236,26 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
}
/**
+ * @return Cache map entry factory.
+ */
+ protected GridCacheMapEntryFactory entryFactory() {
+ return new GridCacheMapEntryFactory() {
+ @Override public GridCacheMapEntry create(
+ GridCacheContext ctx,
+ AffinityTopologyVersion topVer,
+ KeyCacheObject key,
+ int hash,
+ CacheObject val
+ ) {
+ if (ctx.useOffheapEntry())
+ return new GridDhtOffHeapCacheEntry(ctx, topVer, key, hash, val);
+
+ return new GridDhtCacheEntry(ctx, topVer, key, hash, val);
+ }
+ };
+ }
+
+ /**
* @return Near cache.
*/
public abstract GridNearCacheAdapter<K, V> near();
@@ -404,7 +408,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
*
* @throws GridDhtInvalidPartitionException If partition for the key is no longer valid.
*/
- @Override public GridCacheEntryEx entryEx(KeyCacheObject key, AffinityTopologyVersion topVer) throws GridDhtInvalidPartitionException {
+ @Override public GridCacheEntryEx entryEx(KeyCacheObject key,
+ AffinityTopologyVersion topVer) throws GridDhtInvalidPartitionException {
return super.entryEx(key, topVer);
}
@@ -423,7 +428,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @return DHT entry.
* @throws GridDhtInvalidPartitionException If partition for the key is no longer valid.
*/
- public GridDhtCacheEntry entryExx(KeyCacheObject key, AffinityTopologyVersion topVer) throws GridDhtInvalidPartitionException {
+ public GridDhtCacheEntry entryExx(KeyCacheObject key,
+ AffinityTopologyVersion topVer) throws GridDhtInvalidPartitionException {
return (GridDhtCacheEntry)entryEx(key, topVer);
}
@@ -640,7 +646,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean canRemap
- ) {
+ ) {
return getAllAsync0(keys,
readThrough,
/*don't check local tx. */false,
@@ -1084,7 +1090,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
final GridDhtLocalPartition part = ctx.topology().localPartition(partId,
ctx.discovery().topologyVersionEx(), false);
- Iterator<GridDhtCacheEntry> partIt = part == null ? null : part.entries().iterator();
+ Iterator<GridCacheMapEntry> partIt = part == null ? null : part.entries().iterator();
return new PartitionEntryIterator(partIt);
}
@@ -1152,7 +1158,8 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
}
/** {@inheritDoc} */
- @Override public List<GridCacheClearAllRunnable<K, V>> splitClearLocally(boolean srv, boolean near, boolean readers) {
+ @Override public List<GridCacheClearAllRunnable<K, V>> splitClearLocally(boolean srv, boolean near,
+ boolean readers) {
return ctx.affinityNode() ? super.splitClearLocally(srv, near, readers) :
Collections.<GridCacheClearAllRunnable<K, V>>emptyList();
}
@@ -1164,8 +1171,6 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
GridDhtLocalPartition part = topology().localPartition(entry.partition(), AffinityTopologyVersion.NONE,
false);
- // Do not remove entry on replica topology. Instead, add entry to removal queue.
- // It will be cleared eventually.
if (part != null) {
try {
part.onDeferredDelete(entry.key(), ver);
@@ -1211,16 +1216,16 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
assert primary || backup;
if (primary && backup)
- return iterator(map.entries0().iterator(), !ctx.keepBinary());
+ return iterator(entries().iterator(), !ctx.keepBinary());
else {
final AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion();
final Iterator<GridDhtLocalPartition> partIt = topology().currentLocalPartitions().iterator();
- Iterator<GridCacheEntryEx> it = new Iterator<GridCacheEntryEx>() {
- private GridCacheEntryEx next;
+ Iterator<GridCacheMapEntry> it = new Iterator<GridCacheMapEntry>() {
+ private GridCacheMapEntry next;
- private Iterator<GridDhtCacheEntry> curIt;
+ private Iterator<GridCacheMapEntry> curIt;
{
advance();
@@ -1230,11 +1235,11 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
return next != null;
}
- @Override public GridCacheEntryEx next() {
+ @Override public GridCacheMapEntry next() {
if (next == null)
throw new NoSuchElementException();
- GridCacheEntryEx e = next;
+ GridCacheMapEntry e = next;
advance();
@@ -1293,12 +1298,12 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
private Cache.Entry<K, V> last;
/** Partition iterator. */
- private final Iterator<GridDhtCacheEntry> partIt;
+ private final Iterator<GridCacheMapEntry> partIt;
/**
* @param partIt Partition iterator.
*/
- private PartitionEntryIterator(@Nullable Iterator<GridDhtCacheEntry> partIt) {
+ private PartitionEntryIterator(@Nullable Iterator<GridCacheMapEntry> partIt) {
this.partIt = partIt;
advance();
@@ -1335,9 +1340,9 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
private void advance() {
if (partIt != null) {
while (partIt.hasNext()) {
- GridDhtCacheEntry next = partIt.next();
+ GridCacheEntryEx next = partIt.next();
- if (next.isInternal() || !next.visitable(CU.empty0()))
+ if (next instanceof GridCacheMapEntry && (!((GridCacheMapEntry)next).visitable(CU.empty0())))
continue;
entry = next.wrapLazyValue();
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
index ab51bdb..95ef10b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
@@ -88,7 +88,11 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
super(ctx, key, hash, val);
// Record this entry with partition.
- locPart = ctx.dht().topology().onAdded(topVer, this);
+ int p = cctx.affinity().partition(key);
+
+ locPart = ctx.topology().localPartition(p, topVer, true);
+
+ assert locPart != null;
}
/** {@inheritDoc} */
@@ -182,8 +186,7 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
boolean reenter,
boolean tx,
boolean implicitSingle)
- throws GridCacheEntryRemovedException, GridDistributedLockCancelledException
- {
+ throws GridCacheEntryRemovedException, GridDistributedLockCancelledException {
assert serReadVer == null || serOrder != null;
assert !reenter || serOrder == null;
@@ -331,7 +334,8 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
* @throws GridCacheEntryRemovedException If entry has been removed.
*/
@SuppressWarnings({"NonPrivateFieldAccessedInSynchronizedContext"})
- @Nullable public synchronized IgniteBiTuple<GridCacheVersion, CacheObject> versionedValue(AffinityTopologyVersion topVer)
+ @Nullable public synchronized IgniteBiTuple<GridCacheVersion, CacheObject> versionedValue(
+ AffinityTopologyVersion topVer)
throws GridCacheEntryRemovedException {
if (isNew() || !valid(AffinityTopologyVersion.NONE) || deletedUnlocked())
return null;
@@ -599,7 +603,7 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
}
finally {
if (rmv)
- cctx.cache().removeIfObsolete(key); // Clear cache.
+ cctx.cache().removeEntry(this); // Clear cache.
}
}
@@ -713,6 +717,16 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
return S.toString(GridDhtCacheEntry.class, this, "super", super.toString());
}
+ /** {@inheritDoc} */
+ @Override protected void incrementMapPublicSize() {
+ locPart.incrementPublicSize(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void decrementMapPublicSize() {
+ locPart.decrementPublicSize(this);
+ }
+
/**
* Reader ID.
*/
@@ -793,7 +807,6 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
return txFut;
}
-
/** {@inheritDoc} */
@Override public boolean equals(Object o) {
if (this == o)
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
index 4fc1eaf..3ac4a41 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
@@ -17,12 +17,10 @@
package org.apache.ignite.internal.processors.cache.distributed.dht;
-import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -33,8 +31,15 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMap;
+import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMapImpl;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntryFactory;
import org.apache.ignite.internal.processors.cache.GridCacheSwapEntry;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader;
@@ -53,8 +58,7 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteUuid;
import org.jetbrains.annotations.NotNull;
-import org.jsr166.ConcurrentHashMap8;
-import org.jsr166.LongAdder8;
+import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE;
import static org.apache.ignite.events.EventType.EVT_CACHE_REBALANCE_OBJECT_UNLOADED;
@@ -66,7 +70,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDh
/**
* Key partition.
*/
-public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>, GridReservable {
+public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>, GridReservable, GridCacheConcurrentMap {
/** Maximum size for delete queue. */
public static final int MAX_DELETE_QUEUE_SIZE = Integer.getInteger(IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE,
200_000);
@@ -89,7 +93,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
private final GridFutureAdapter<?> rent;
/** Entries map. */
- private final ConcurrentMap<KeyCacheObject, GridDhtCacheEntry> map;
+ private final GridCacheConcurrentMap map;
/** Context. */
private final GridCacheContext cctx;
@@ -104,9 +108,6 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
/** Lock. */
private final ReentrantLock lock = new ReentrantLock();
- /** Public size counter. */
- private final LongAdder8 mapPubSize = new LongAdder8();
-
/** Remove queue. */
private final GridCircularBuffer<T2<KeyCacheObject, GridCacheVersion>> rmvQueue;
@@ -120,8 +121,8 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
* @param cctx Context.
* @param id Partition ID.
*/
- @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor")
- GridDhtLocalPartition(GridCacheContext cctx, int id) {
+ @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor") GridDhtLocalPartition(GridCacheContext cctx, int id,
+ GridCacheMapEntryFactory entryFactory) {
assert cctx != null;
this.id = id;
@@ -135,8 +136,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
}
};
- map = new ConcurrentHashMap8<>(cctx.config().getStartSize() /
- cctx.affinity().partitions());
+ map = new GridCacheConcurrentMapImpl(cctx, entryFactory, cctx.config().getStartSize() / cctx.affinity().partitions());
int delQueueSize = CU.isSystemCache(cctx.name()) ? 100 :
Math.max(MAX_DELETE_QUEUE_SIZE / cctx.affinity().partitions(), 20);
@@ -202,45 +202,30 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
}
/**
- * @return Entries belonging to partition.
- */
- public Collection<GridDhtCacheEntry> entries() {
- return map.values();
- }
-
- /**
* @return {@code True} if partition is empty.
*/
public boolean isEmpty() {
- return map.isEmpty();
+ return map.size() == 0;
}
- /**
- * @return Number of entries in this partition (constant-time method).
- */
- public int size() {
+ /** {@inheritDoc} */
+ @Override public int size() {
return map.size();
}
- /**
- * Increments public size of the map.
- */
- public void incrementPublicSize() {
- mapPubSize.increment();
+ /** {@inheritDoc} */
+ @Override public int publicSize() {
+ return map.publicSize();
}
- /**
- * Decrements public size of the map.
- */
- public void decrementPublicSize() {
- mapPubSize.decrement();
+ /** {@inheritDoc} */
+ @Override public void incrementPublicSize(GridCacheEntryEx e) {
+ map.incrementPublicSize(e);
}
- /**
- * @return Number of public (non-internal) entries in this partition.
- */
- public int publicSize() {
- return mapPubSize.intValue();
+ /** {@inheritDoc} */
+ @Override public void decrementPublicSize(GridCacheEntryEx e) {
+ map.decrementPublicSize(e);
}
/**
@@ -252,39 +237,57 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
return state == MOVING || state == OWNING || state == RENTING;
}
- /**
- * @param entry Entry to add.
- */
- void onAdded(GridDhtCacheEntry entry) {
- GridDhtPartitionState state = state();
+ /** {@inheritDoc} */
+ @Override @Nullable public GridCacheMapEntry getEntry(KeyCacheObject key) {
+ return map.getEntry(key);
+ }
- if (state == EVICTED)
- throw new GridDhtInvalidPartitionException(id, "Adding entry to invalid partition " +
- "(often may be caused by inconsistent 'key.hashCode()' implementation) [part=" + id + ']');
+ /** {@inheritDoc} */
+ @Override public boolean removeEntry(GridCacheEntryEx entry) {
+ return map.removeEntry(entry);
+ }
- map.put(entry.key(), entry);
+ /** {@inheritDoc} */
+ @Override public Iterable<GridCacheMapEntry> entries(
+ CacheEntryPredicate... filter) {
+ return map.entries(filter);
+ }
- if (!entry.isInternal()) {
- assert !entry.deleted() : entry;
+ /** {@inheritDoc} */
+ @Override public Iterable<GridCacheMapEntry> allEntries(CacheEntryPredicate... filter) {
+ return map.allEntries(filter);
+ }
- mapPubSize.increment();
- }
+ /** {@inheritDoc} */
+ @Override public Set<GridCacheMapEntry> entrySet(CacheEntryPredicate... filter) {
+ return map.entrySet(filter);
+ }
+
+ /** {@inheritDoc} */
+ @Override @Nullable public GridCacheMapEntry randomEntry() {
+ return map.randomEntry();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridCacheMapEntry putEntryIfObsoleteOrAbsent(
+ AffinityTopologyVersion topVer, KeyCacheObject key,
+ @Nullable CacheObject val, boolean create, boolean touch) {
+ return map.putEntryIfObsoleteOrAbsent(topVer, key, val, create, touch);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<KeyCacheObject> keySet(CacheEntryPredicate... filter) {
+ return map.keySet(filter);
}
/**
* @param entry Entry to remove.
*/
- @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
void onRemoved(GridDhtCacheEntry entry) {
assert entry.obsolete() : entry;
// Make sure to remove exactly this entry.
- synchronized (entry) {
- map.remove(entry.key(), entry);
-
- if (!entry.isInternal() && !entry.deleted())
- mapPubSize.decrement();
- }
+ map.removeEntry(entry);
// Attempt to evict.
tryEvict();
@@ -312,7 +315,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
/**
* Locks partition.
*/
- @SuppressWarnings( {"LockAcquiredButNotSafelyReleased"})
+ @SuppressWarnings({"LockAcquiredButNotSafelyReleased"})
public void lock() {
lock.lock();
}
@@ -338,11 +341,11 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
Map<KeyCacheObject, GridCacheVersion> evictHist0 = evictHist;
- if (evictHist0 != null ) {
+ if (evictHist0 != null) {
GridCacheVersion ver0 = evictHist0.get(key);
if (ver0 == null || ver0.isLess(ver)) {
- GridCacheVersion ver1 = evictHist0.put(key, ver);
+ GridCacheVersion ver1 = evictHist0.put(key, ver);
assert ver1 == ver0;
}
@@ -366,7 +369,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
Map<KeyCacheObject, GridCacheVersion> evictHist0 = evictHist;
- if (evictHist0 != null) {
+ if (evictHist0 != null) {
GridCacheVersion ver0 = evictHist0.get(key);
// Permit preloading if version in history
@@ -489,7 +492,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
int ord = (int)(reservations >> 32);
- if (map.isEmpty() && !GridQueryProcessor.isEnabled(cctx.config()) &&
+ if (isEmpty() && !GridQueryProcessor.isEnabled(cctx.config()) &&
ord == RENTING.ordinal() && (reservations & 0xFFFF) == 0 &&
casState(reservations, EVICTED)) {
if (log.isDebugEnabled())
@@ -538,7 +541,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
// Attempt to evict partition entries from cache.
clearAll();
- if (map.isEmpty() && casState(reservations, EVICTED)) {
+ if (isEmpty() && casState(reservations, EVICTED)) {
if (log.isDebugEnabled())
log.debug("Evicted partition: " + this);
@@ -655,7 +658,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
boolean rec = cctx.events().isRecordable(EVT_CACHE_REBALANCE_OBJECT_UNLOADED);
- Iterator<GridDhtCacheEntry> it = map.values().iterator();
+ Iterator<GridDhtCacheEntry> it = (Iterator)map.allEntries().iterator();
GridCloseableIterator<Map.Entry<byte[], GridCacheSwapEntry>> swapIt = null;
@@ -684,11 +687,9 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
cached = it.next();
if (cached.clearInternal(clearVer, swap, extras)) {
- map.remove(cached.key(), cached);
+ map.removeEntry(cached);
if (!cached.isInternal()) {
- mapPubSize.decrement();
-
if (rec) {
cctx.events().addEvent(cached.partition(),
cached.key(),
@@ -709,7 +710,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
}
}
catch (GridDhtInvalidPartitionException e) {
- assert map.isEmpty() && state() == EVICTED: "Invalid error [e=" + e + ", part=" + this + ']';
+ assert isEmpty() && state() == EVICTED : "Invalid error [e=" + e + ", part=" + this + ']';
assert swapEmpty() : "Invalid error when swap is not cleared [e=" + e + ", part=" + this + ']';
break; // Partition is already concurrently cleared and evicted.
@@ -789,7 +790,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
}
@Override public void remove() {
- map.remove(lastEntry.key(), lastEntry);
+ map.removeEntry(lastEntry);
}
};
}
@@ -811,7 +812,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
}
/** {@inheritDoc} */
- @SuppressWarnings( {"OverlyStrongTypeCast"})
+ @SuppressWarnings({"OverlyStrongTypeCast"})
@Override public boolean equals(Object obj) {
return obj instanceof GridDhtLocalPartition && (obj == this || ((GridDhtLocalPartition)obj).id() == id);
}
@@ -829,8 +830,7 @@ public class GridDhtLocalPartition implements Comparable<GridDhtLocalPartition>,
return S.toString(GridDhtLocalPartition.class, this,
"state", state(),
"reservations", reservations(),
- "empty", map.isEmpty(),
- "createTime", U.format(createTime),
- "mapPubSize", mapPubSize);
+ "empty", isEmpty(),
+ "createTime", U.format(createTime));
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
index 50167d8..95c6dfc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
@@ -419,6 +419,12 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
writer.incrementState();
+ case 30:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -515,6 +521,14 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
reader.incrementState();
+ case 30:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridDhtLockRequest.class);
@@ -527,7 +541,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 30;
+ return 31;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
index 7fba45d..2c483a1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
@@ -195,13 +195,6 @@ public interface GridDhtPartitionTopology {
public GridDhtPartitionFullMap partitionMap(boolean onlyActive);
/**
- * @param topVer Topology version.
- * @param e Entry added to cache.
- * @return Local partition.
- */
- public GridDhtLocalPartition onAdded(AffinityTopologyVersion topVer, GridDhtCacheEntry e);
-
- /**
* @param e Entry removed from cache.
*/
public void onRemoved(GridDhtCacheEntry e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index f0ce6d1..9f5fbfb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -28,7 +28,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.ConcurrentMap;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
@@ -37,6 +36,7 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityAssignment;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntryFactory;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap2;
@@ -50,7 +50,6 @@ 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.U;
import org.jetbrains.annotations.Nullable;
-import org.jsr166.ConcurrentHashMap8;
import static org.apache.ignite.events.EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST;
import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.EVICTED;
@@ -76,8 +75,7 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
private final IgniteLogger log;
/** */
- private final ConcurrentMap<Integer, GridDhtLocalPartition> locParts =
- new ConcurrentHashMap8<>();
+ private final GridDhtLocalPartition[] locParts;
/** Node to partition map. */
private GridDhtPartitionFullMap node2part;
@@ -103,6 +101,9 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** Lock. */
private final StripedCompositeReadWriteLock lock = new StripedCompositeReadWriteLock(16);
+ /** */
+ private final GridCacheMapEntryFactory entryFactory;
+
/** Partition update counter. */
private Map<Integer, Long> cntrMap = new HashMap<>();
@@ -112,12 +113,15 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/**
* @param cctx Context.
*/
- GridDhtPartitionTopologyImpl(GridCacheContext<?, ?> cctx) {
+ GridDhtPartitionTopologyImpl(GridCacheContext<?, ?> cctx, GridCacheMapEntryFactory entryFactory) {
assert cctx != null;
this.cctx = cctx;
+ this.entryFactory = entryFactory;
log = cctx.logger(getClass());
+
+ locParts = new GridDhtLocalPartition[cctx.config().getAffinity().partitions()];
}
/**
@@ -149,7 +153,7 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/**
* @return Full map string representation.
*/
- @SuppressWarnings( {"ConstantConditions"})
+ @SuppressWarnings({"ConstantConditions"})
private String fullMapString() {
return node2part == null ? "null" : FULL_MAP_DEBUG ? node2part.toFullString() : node2part.toString();
}
@@ -158,7 +162,7 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
* @param map Map to get string for.
* @return Full map string representation.
*/
- @SuppressWarnings( {"ConstantConditions"})
+ @SuppressWarnings({"ConstantConditions"})
private String mapString(GridDhtPartitionMap2 map) {
return map == null ? "null" : FULL_MAP_DEBUG ? map.toFullString() : map.toString();
}
@@ -172,34 +176,65 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
private boolean waitForRent() throws IgniteCheckedException {
boolean changed = false;
- // Synchronously wait for all renting partitions to complete.
- for (Iterator<GridDhtLocalPartition> it = locParts.values().iterator(); it.hasNext();) {
- GridDhtLocalPartition p = it.next();
+ GridDhtLocalPartition[] locPartsCopy = new GridDhtLocalPartition[locParts.length];
- GridDhtPartitionState state = p.state();
+ lock.readLock().lock();
+
+ try {
+ for (int i = 0; i < locParts.length; i++)
+ locPartsCopy[i] = locParts[i];
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+
+ GridDhtLocalPartition part;
+
+ for (int i = 0; i < locPartsCopy.length; i++) {
+ part = locPartsCopy[i];
+
+ if (part == null)
+ continue;
+
+ GridDhtPartitionState state = part.state();
if (state == RENTING || state == EVICTED) {
if (log.isDebugEnabled())
- log.debug("Waiting for renting partition: " + p);
+ log.debug("Waiting for renting partition: " + part);
// Wait for partition to empty out.
- p.rent(true).get();
+ part.rent(true).get();
if (log.isDebugEnabled())
- log.debug("Finished waiting for renting partition: " + p);
+ log.debug("Finished waiting for renting partition: " + part);
+ }
+ }
- // Remove evicted partition.
- it.remove();
+ // Remove evicted partition.
+ lock.writeLock().lock();
- changed = true;
+ try {
+ for (int i = 0; i < locParts.length; i++) {
+ part = locParts[i];
+
+ if (part == null)
+ continue;
+
+ if (part.state() == EVICTED) {
+ locParts[i] = null;
+ changed = true;
+ }
}
}
+ finally {
+ lock.writeLock().unlock();
+ }
return changed;
}
/** {@inheritDoc} */
- @SuppressWarnings( {"LockAcquiredButNotSafelyReleased"})
+ @SuppressWarnings({"LockAcquiredButNotSafelyReleased"})
@Override public void readLock() {
lock.readLock().lock();
}
@@ -267,7 +302,8 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Override public void initPartitions(GridDhtPartitionsExchangeFuture exchFut) throws IgniteInterruptedCheckedException {
+ @Override public void initPartitions(
+ GridDhtPartitionsExchangeFuture exchFut) throws IgniteInterruptedCheckedException {
U.writeLock(lock);
try {
@@ -300,12 +336,12 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
assert topVer.equals(exchFut.topologyVersion()) :
"Invalid topology [topVer=" + topVer +
- ", cache=" + cctx.name() +
- ", futVer=" + exchFut.topologyVersion() + ']';
+ ", cache=" + cctx.name() +
+ ", futVer=" + exchFut.topologyVersion() + ']';
assert cctx.affinity().affinityTopologyVersion().equals(exchFut.topologyVersion()) :
"Invalid affinity [topVer=" + cctx.affinity().affinityTopologyVersion() +
- ", cache=" + cctx.name()+
- ", futVer=" + exchFut.topologyVersion() + ']';
+ ", cache=" + cctx.name() +
+ ", futVer=" + exchFut.topologyVersion() + ']';
List<List<ClusterNode>> aff = cctx.affinity().assignments(exchFut.topologyVersion());
@@ -587,7 +623,8 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
/** {@inheritDoc} */
- @Nullable @Override public GridDhtLocalPartition localPartition(int p, AffinityTopologyVersion topVer, boolean create)
+ @Nullable @Override public GridDhtLocalPartition localPartition(int p, AffinityTopologyVersion topVer,
+ boolean create)
throws GridDhtInvalidPartitionException {
return localPartition(p, topVer, create, true);
}
@@ -597,21 +634,12 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
* @return Partition.
*/
private GridDhtLocalPartition createPartition(int p) {
- GridDhtLocalPartition loc = locParts.get(p);
-
- if (loc != null && loc.state() == EVICTED) {
- boolean rmv = locParts.remove(p, loc);
+ assert lock.isWriteLockedByCurrentThread();
- assert rmv;
-
- loc = null;
- }
+ GridDhtLocalPartition loc = locParts[p];
- if (loc == null) {
- GridDhtLocalPartition old = locParts.putIfAbsent(p, loc = new GridDhtLocalPartition(cctx, p));
-
- assert old == null : old;
- }
+ if (loc == null || loc.state() == EVICTED)
+ locParts[p] = loc = new GridDhtLocalPartition(cctx, p, entryFactory);
return loc;
}
@@ -627,54 +655,59 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
AffinityTopologyVersion topVer,
boolean create,
boolean updateSeq) {
- boolean belongs = create && cctx.affinity().localNode(p, topVer);
+ GridDhtLocalPartition loc;
- while (true) {
- GridDhtLocalPartition loc = locParts.get(p);
+ lock.readLock().lock();
- if (loc != null && loc.state() == EVICTED) {
- locParts.remove(p, loc);
+ try {
+ loc = locParts[p];
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+
+ if (loc != null && loc.state() != EVICTED)
+ return loc;
+
+ if (!create)
+ return null;
+
+ lock.writeLock().lock();
+
+ try {
+ loc = locParts[p];
+
+ boolean belongs = cctx.affinity().localNode(p, topVer);
- if (!create)
- return null;
+ if (loc != null && loc.state() == EVICTED) {
+ locParts[p] = loc = null;
if (!belongs)
throw new GridDhtInvalidPartitionException(p, "Adding entry to evicted partition " +
"(often may be caused by inconsistent 'key.hashCode()' implementation) " +
"[part=" + p + ", topVer=" + topVer + ", this.topVer=" + this.topVer + ']');
-
- continue;
}
- if (loc == null && create) {
+ if (loc == null) {
if (!belongs)
throw new GridDhtInvalidPartitionException(p, "Creating partition which does not belong to " +
"local node (often may be caused by inconsistent 'key.hashCode()' implementation) " +
"[part=" + p + ", topVer=" + topVer + ", this.topVer=" + this.topVer + ']');
- lock.writeLock().lock();
-
- try {
- GridDhtLocalPartition old = locParts.putIfAbsent(p,
- loc = new GridDhtLocalPartition(cctx, p));
+ locParts[p] = loc = new GridDhtLocalPartition(cctx, p, entryFactory);
- if (old != null)
- loc = old;
- else {
- if (updateSeq)
- this.updateSeq.incrementAndGet();
+ if (updateSeq)
+ this.updateSeq.incrementAndGet();
- if (log.isDebugEnabled())
- log.debug("Created local partition: " + loc);
- }
- }
- finally {
- lock.writeLock().unlock();
- }
+ if (log.isDebugEnabled())
+ log.debug("Created local partition: " + loc);
}
-
- return loc;
}
+ finally {
+ lock.writeLock().unlock();
+ }
+
+ return loc;
}
/** {@inheritDoc} */
@@ -682,8 +715,20 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
assert parts != null;
assert parts.length > 0;
+ GridDhtLocalPartition[] locPartsCopy = new GridDhtLocalPartition[parts.length];
+
+ lock.readLock().lock();
+
+ try {
+ for (int i = 0; i < parts.length; i++)
+ locPartsCopy[i] = locParts[parts[i]];
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+
for (int i = 0; i < parts.length; i++)
- locParts.get(parts[i]).release();
+ locPartsCopy[i].release();
}
/** {@inheritDoc} */
@@ -693,31 +738,28 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public List<GridDhtLocalPartition> localPartitions() {
- return new LinkedList<>(locParts.values());
- }
-
- /** {@inheritDoc} */
- @Override public Collection<GridDhtLocalPartition> currentLocalPartitions() {
- return locParts.values();
- }
-
- /** {@inheritDoc} */
- @Override public GridDhtLocalPartition onAdded(AffinityTopologyVersion topVer, GridDhtCacheEntry e) {
- /*
- * Make sure not to acquire any locks here as this method
- * may be called from sensitive synchronization blocks.
- * ===================================================
- */
+ LinkedList<GridDhtLocalPartition> list = new LinkedList<>();
- int p = cctx.affinity().partition(e.key());
+ lock.readLock().lock();
- GridDhtLocalPartition loc = localPartition(p, topVer, true);
+ try {
+ for (int i = 0; i < locParts.length; i++) {
+ GridDhtLocalPartition part = locParts[i];
- assert loc != null;
+ if (part != null)
+ list.add(part);
+ }
- loc.onAdded(e);
+ return list;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
- return loc;
+ /** {@inheritDoc} */
+ @Override public Collection<GridDhtLocalPartition> currentLocalPartitions() {
+ return localPartitions();
}
/** {@inheritDoc} */
@@ -736,11 +778,22 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public GridDhtPartitionMap2 localPartitionMap() {
+ Map<Integer, GridDhtPartitionState> map = new HashMap<>();
+
lock.readLock().lock();
try {
+ for (int i = 0; i < locParts.length; i++) {
+ GridDhtLocalPartition part = locParts[i];
+
+ if (part == null)
+ continue;
+
+ map.put(i, part.state());
+ }
+
return new GridDhtPartitionMap2(cctx.nodeId(), updateSeq.get(), topVer,
- F.viewReadOnly(locParts, CU.part2state()), true);
+ Collections.unmodifiableMap(map), true);
}
finally {
lock.readLock().unlock();
@@ -943,7 +996,12 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
this.cntrMap.put(e.getKey(), e.getValue());
}
- for (GridDhtLocalPartition part : locParts.values()) {
+ for (int i = 0; i < locParts.length; i++) {
+ GridDhtLocalPartition part = locParts[i];
+
+ if (part == null)
+ continue;
+
Long cntr = cntrMap.get(part.id());
if (cntr != null)
@@ -992,7 +1050,7 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
}
- for (Iterator<UUID> it = partMap.keySet().iterator(); it.hasNext();) {
+ for (Iterator<UUID> it = partMap.keySet().iterator(); it.hasNext(); ) {
UUID nodeId = it.next();
if (!cctx.discovery().alive(nodeId)) {
@@ -1078,7 +1136,12 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
this.cntrMap.put(e.getKey(), e.getValue());
}
- for (GridDhtLocalPartition part : locParts.values()) {
+ for (int i = 0; i < locParts.length; i++) {
+ GridDhtLocalPartition part = locParts[i];
+
+ if (part == null)
+ continue;
+
Long cntr = cntrMap.get(part.id());
if (cntr != null)
@@ -1178,12 +1241,15 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
UUID locId = cctx.nodeId();
- for (GridDhtLocalPartition part : locParts.values()) {
+ for (int p = 0; p < locParts.length; p++) {
+ GridDhtLocalPartition part = locParts[p];
+
+ if (part == null)
+ continue;
+
GridDhtPartitionState state = part.state();
if (state.active()) {
- int p = part.id();
-
List<ClusterNode> affNodes = aff.get(p);
if (!affNodes.contains(cctx.localNode())) {
@@ -1403,7 +1469,12 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
try {
Map<Integer, Long> res = new HashMap<>(cntrMap);
- for (GridDhtLocalPartition part : locParts.values()) {
+ for (int i = 0; i < locParts.length; i++) {
+ GridDhtLocalPartition part = locParts[i];
+
+ if (part == null)
+ continue;
+
Long cntr0 = res.get(part.id());
Long cntr1 = part.updateCounter();
@@ -1429,11 +1500,23 @@ class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
@Override public void printMemoryStats(int threshold) {
X.println(">>> Cache partition topology stats [grid=" + cctx.gridName() + ", cache=" + cctx.name() + ']');
- for (GridDhtLocalPartition part : locParts.values()) {
- int size = part.size();
+ lock.readLock().lock();
+
+ try {
+ for (int i = 0; i < locParts.length; i++) {
+ GridDhtLocalPartition part = locParts[i];
+
+ if (part == null)
+ continue;
+
+ int size = part.size();
- if (size >= threshold)
- X.println(">>> Local partition [part=" + part.id() + ", size=" + size + ']');
+ if (size >= threshold)
+ X.println(">>> Local partition [part=" + part.id() + ", size=" + size + ']');
+ }
+ }
+ finally {
+ lock.readLock().unlock();
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtUnlockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtUnlockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtUnlockRequest.java
index 38152a7..1d067da 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtUnlockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtUnlockRequest.java
@@ -120,6 +120,11 @@ public class GridDhtUnlockRequest extends GridDistributedUnlockRequest {
writer.incrementState();
+ case 9:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
}
return true;
@@ -144,6 +149,14 @@ public class GridDhtUnlockRequest extends GridDistributedUnlockRequest {
reader.incrementState();
+ case 9:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridDhtUnlockRequest.class);
@@ -156,6 +169,6 @@ public class GridDhtUnlockRequest extends GridDistributedUnlockRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 9;
+ return 10;
}
}
[4/6] ignite git commit: IGNITE-2948 - Optimize usage of
GridCacheConcurrentMap
Posted by vo...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMap.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMap.java
index 05ff71a..e733114 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMap.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMap.java
@@ -17,367 +17,14 @@
package org.apache.ignite.internal.processors.cache;
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.io.ObjectStreamException;
-import java.lang.reflect.Array;
-import java.util.AbstractCollection;
-import java.util.AbstractSet;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.Random;
import java.util.Set;
-import java.util.concurrent.locks.ReentrantLock;
-import javax.cache.Cache;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.util.F0;
-import org.apache.ignite.internal.util.lang.GridTriple;
-import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.P1;
-import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.A;
-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.lang.IgnitePredicate;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
-import org.jsr166.LongAdder8;
/**
- * Concurrent implementation of cache map.
+ * Concurrent cache map.
*/
-public class GridCacheConcurrentMap {
- /** Debug flag. */
- private static final boolean DEBUG = false;
-
- /** Random. */
- private static final Random RAND = new Random();
-
- /** The default load factor for this map. */
- private static final float DFLT_LOAD_FACTOR = 0.75f;
-
- /** The default concurrency level for this map. */
- private static final int DFLT_CONCUR_LEVEL = Runtime.getRuntime().availableProcessors() * 2;
-
- /**
- * The maximum capacity, used if a higher value is implicitly specified by either
- * of the constructors with arguments. Must be a power of two <= 1<<30 to ensure
- * that entries are indexable using integers.
- */
- private static final int MAX_CAP = 1 << 30;
-
- /** The maximum number of segments to allow. */
- private static final int MAX_SEGS = 1 << 16; // slightly conservative
-
- /**
- * Mask value for indexing into segments. The upper bits of a
- * key's hash code are used to choose the segment.
- */
- private final int segMask;
-
- /** Shift value for indexing within segments. */
- private final int segShift;
-
- /** The segments, each of which is a specialized hash table. */
- private final Segment[] segs;
-
- /** */
- private GridCacheMapEntryFactory factory;
-
- /** Cache context. */
- protected final GridCacheContext ctx;
-
- /** */
- private final LongAdder8 mapPubSize = new LongAdder8();
-
- /** */
- private final LongAdder8 mapSize = new LongAdder8();
-
- /** Filters cache internal entry. */
- private static final CacheEntryPredicate NON_INTERNAL =
- new CacheEntrySerializablePredicate(new CacheEntryPredicateAdapter() {
- @Override public boolean apply(GridCacheEntryEx entry) {
- return !entry.isInternal();
- }
- });
-
- /** Non-internal predicate array. */
- public static final CacheEntryPredicate[] NON_INTERNAL_ARR = new CacheEntryPredicate[] {NON_INTERNAL};
-
- /** Filters obsolete cache map entry. */
- private final IgnitePredicate<GridCacheMapEntry> obsolete =
- new P1<GridCacheMapEntry>() {
- @Override public boolean apply(GridCacheMapEntry entry) {
- return entry.obsolete();
- }
- };
-
- /**
- * Applies a supplemental hash function to a given hashCode, which
- * defends against poor quality hash functions. This is critical
- * because ConcurrentHashMap uses power-of-two length hash tables,
- * that otherwise encounter collisions for hashCodes that do not
- * differ in lower or upper bits.
- * <p>
- * This function has been taken from Java 8 ConcurrentHashMap with
- * slightly modifications.
- *
- * @param h Value to hash.
- * @return Hash value.
- */
- protected static int hash(int h) {
- return U.hash(h);
- }
-
- /**
- * Returns the segment that should be used for key with given hash
- *
- * @param hash The hash code for the key.
- * @return The segment.
- */
- private Segment segmentFor(int hash) {
- return segs[(hash >>> segShift) & segMask];
- }
-
- /**
- * Creates a new, empty map with the specified initial
- * capacity, load factor and concurrency level.
- *
- * @param ctx Cache context.
- * @param initCap the initial capacity. The implementation
- * performs internal sizing to accommodate this many elements.
- * @param factory Entry factory.
- * @param loadFactor the load factor threshold, used to control resizing.
- * Resizing may be performed when the average number of elements per
- * bin exceeds this threshold.
- * @param concurrencyLevel the estimated number of concurrently
- * updating threads. The implementation performs internal sizing
- * to try to accommodate this many threads.
- * @throws IllegalArgumentException if the initial capacity is
- * negative or the load factor or concurrencyLevel are
- * non-positive.
- */
- @SuppressWarnings({"unchecked"})
- protected GridCacheConcurrentMap(
- GridCacheContext ctx,
- int initCap,
- GridCacheMapEntryFactory factory,
- float loadFactor,
- int concurrencyLevel
- ) {
- this.ctx = ctx;
-
- if (!(loadFactor > 0) || initCap < 0 || concurrencyLevel <= 0)
- throw new IllegalArgumentException();
-
- if (concurrencyLevel > MAX_SEGS)
- concurrencyLevel = MAX_SEGS;
-
- // Find power-of-two sizes best matching arguments
- int sshift = 0;
- int ssize = 1;
-
- while (ssize < concurrencyLevel) {
- ++sshift;
- ssize <<= 1;
- }
-
- segShift = 32 - sshift;
- segMask = ssize - 1;
- segs = (Segment[])Array.newInstance(Segment.class, ssize);
-
- if (initCap > MAX_CAP)
- initCap = MAX_CAP;
-
- int c = initCap / ssize;
-
- if (c * ssize < initCap)
- ++c;
-
- int cap = 1;
-
- while (cap < c)
- cap <<= 1;
-
- if (cap < 16)
- cap = 16;
-
- for (int i = 0; i < segs.length; ++i)
- segs[i] = new Segment(cap, loadFactor);
-
- this.factory = factory;
- }
-
- /**
- * Creates a new, empty map with the specified initial capacity
- * and load factor and with the default concurrencyLevel (16).
- *
- * @param ctx Cache context.
- * @param initCap The implementation performs internal
- * sizing to accommodate this many elements.
- * @param factory Entries factory.
- * @throws IllegalArgumentException if the initial capacity of
- * elements is negative or the load factor is non-positive.
- */
- public GridCacheConcurrentMap(
- GridCacheContext ctx,
- int initCap,
- @Nullable GridCacheMapEntryFactory factory
- ) {
- this(ctx, initCap, factory, DFLT_LOAD_FACTOR, DFLT_CONCUR_LEVEL);
- }
-
- /**
- * Sets factory for entries.
- *
- * @param factory Entry factory.
- */
- public void setEntryFactory(GridCacheMapEntryFactory factory) {
- assert factory != null;
-
- this.factory = factory;
- }
-
- /**
- * @return Entries factory.
- */
- public GridCacheMapEntryFactory getEntryFactory() {
- return factory;
- }
-
- /**
- * @return Non-internal predicate.
- */
- private static CacheEntryPredicate[] nonInternal() {
- return NON_INTERNAL_ARR;
- }
-
- /**
- * @param filter Filter to add to non-internal-key filter.
- * @return Non-internal predicate.
- */
- private static CacheEntryPredicate[] nonInternal(
- CacheEntryPredicate[] filter) {
- return F.asArray(F0.and0(NON_INTERNAL_ARR, filter));
- }
-
- /**
- * @return {@code True} if this map is empty.
- */
- public boolean isEmpty() {
- return mapSize.sum() == 0;
- }
-
- /**
- * Returns the number of key-value mappings in this map.
- *
- * @return the number of key-value mappings in this map.
- */
- public int size() {
- return mapSize.intValue();
- }
-
- /**
- * @return Public size.
- */
- public int publicSize() {
- return mapPubSize.intValue();
- }
-
- /**
- * @param e Cache map entry.
- */
- public void decrementSize(GridCacheMapEntry e) {
- assert !e.isInternal();
- assert Thread.holdsLock(e);
- assert e.deletedUnlocked();
- assert ctx.deferredDelete();
-
- mapPubSize.decrement();
-
- segmentFor(e.hash()).decrementPublicSize();
- }
-
- /**
- * @param e Cache map entry.
- */
- public void incrementSize(GridCacheMapEntry e) {
- assert !e.isInternal();
- assert Thread.holdsLock(e);
- assert !e.deletedUnlocked();
- assert ctx.deferredDelete();
-
- mapPubSize.increment();
-
- segmentFor(e.hash()).incrementPublicSize();
- }
-
- /**
- * @param key Key.
- * @return {@code True} if map contains mapping for provided key.
- */
- public boolean containsKey(Object key) {
- int hash = hash(key.hashCode());
-
- return segmentFor(hash).containsKey(key, hash);
- }
-
- /**
- * Collection of all (possibly {@code null}) values.
- *
- * @param filter Filter.
- * @return a collection view of the values contained in this map.
- */
- public <K, V> Collection<V> allValues(CacheEntryPredicate[] filter) {
- return new Values<>(this, filter);
- }
-
- /**
- * @return Random entry out of hash map.
- */
- @Nullable public GridCacheMapEntry randomEntry() {
- while (true) {
- if (mapPubSize.sum() == 0)
- return null;
-
- // Desired and current indexes.
- int segIdx = RAND.nextInt(segs.length);
-
- Segment seg = null;
-
- for (int i = segIdx; i < segs.length + segIdx; i++) {
- Segment s = segs[i % segs.length];
-
- if (s.publicSize() > 0)
- seg = s;
- }
-
- if (seg == null)
- // It happened so that all public values had been removed from segments.
- return null;
-
- GridCacheMapEntry entry = seg.randomEntry();
-
- if (entry == null)
- continue;
-
- assert !(entry.key() instanceof GridCacheInternal);
-
- return entry;
- }
- }
-
+public interface GridCacheConcurrentMap {
/**
* Returns the entry associated with the specified key in the
* HashMap. Returns null if the HashMap contains no mapping
@@ -386,27 +33,7 @@ public class GridCacheConcurrentMap {
* @param key Key.
* @return Entry.
*/
- @Nullable public GridCacheMapEntry getEntry(Object key) {
- assert key != null;
-
- int hash = hash(key.hashCode());
-
- return segmentFor(hash).get(key, hash);
- }
-
- /**
- * @param topVer Topology version.
- * @param key Key.
- * @param val Value.
- * @return Cache entry for corresponding key-value pair.
- */
- public GridCacheMapEntry putEntry(AffinityTopologyVersion topVer, KeyCacheObject key, @Nullable CacheObject val) {
- assert key != null;
-
- int hash = hash(key.hashCode());
-
- return segmentFor(hash).put(key, hash, val, topVer);
- }
+ @Nullable public GridCacheMapEntry getEntry(KeyCacheObject key);
/**
* @param topVer Topology version.
@@ -416,1633 +43,72 @@ public class GridCacheConcurrentMap {
* @return Triple where the first element is current entry associated with the key,
* the second is created entry and the third is doomed (all may be null).
*/
- public GridTriple<GridCacheMapEntry> putEntryIfObsoleteOrAbsent(
+ @Nullable public GridCacheMapEntry putEntryIfObsoleteOrAbsent(
AffinityTopologyVersion topVer,
KeyCacheObject key,
@Nullable CacheObject val,
- boolean create)
- {
- assert key != null;
-
- int hash = hash(key.hashCode());
-
- return segmentFor(hash).putIfObsolete(key, hash, val, topVer, create);
- }
-
- /**
- * Copies all of the mappings from the specified map to this map
- * These mappings will replace any mappings that
- * this map had for any of the keys currently in the specified map.
- *
- * @param m mappings to be stored in this map.
- * @throws NullPointerException If the specified map is null.
- */
- public void putAll(Map<KeyCacheObject, CacheObject> m) {
- for (Map.Entry<KeyCacheObject, CacheObject> e : m.entrySet())
- putEntry(AffinityTopologyVersion.NONE, e.getKey(), e.getValue());
- }
+ boolean create,
+ boolean touch);
/**
* Removes passed in entry if it presents in the map.
*
- * @param e Entry to remove.
+ * @param entry Entry to remove.
* @return {@code True} if remove happened.
*/
- public boolean removeEntry(GridCacheEntryEx e) {
- assert e != null;
-
- KeyCacheObject key = e.key();
-
- int hash = hash(key.hashCode());
-
- return segmentFor(hash).remove(key, hash, same(e)) != null;
- }
+ public boolean removeEntry(GridCacheEntryEx entry);
/**
- * @param p Entry to check equality.
- * @return Predicate to filter the same (equal by ==) entry.
- */
- private IgnitePredicate<GridCacheMapEntry> same(final GridCacheEntryEx p) {
- return new P1<GridCacheMapEntry>() {
- @Override public boolean apply(GridCacheMapEntry e) {
- return e == p;
- }
- };
- }
-
- /**
- * Removes and returns the entry associated with the specified key
- * in the HashMap if entry is obsolete. Returns null if the HashMap
- * contains no mapping for this key.
+ * Returns the number of key-value mappings in this map.
*
- * @param key Key.
- * @return Removed entry, possibly {@code null}.
- */
- @SuppressWarnings( {"unchecked"})
- @Nullable public GridCacheMapEntry removeEntryIfObsolete(KeyCacheObject key) {
- assert key != null;
-
- int hash = hash(key.hashCode());
-
- return segmentFor(hash).remove(key, hash, obsolete);
- }
-
- /**
- * @param filter Filter.
- * @return Set of the mappings contained in this map.
+ * @return the number of key-value mappings in this map.
*/
- @SuppressWarnings({"unchecked"})
- public <K, V> Set<Cache.Entry<K, V>> entries(CacheEntryPredicate... filter) {
- return new EntrySet<>(this, filter);
- }
+ public int size();
/**
- * Returns entry set containing internal entries.
+ * Returns the number of publicly available key-value mappings in this map.
+ * It excludes entries that are marked as deleted.
*
- * @param filter Filter.
- * @return Set of the mappings contained in this map.
+ * @return the number of publicly available key-value mappings in this map.
*/
- @SuppressWarnings({"unchecked"})
- public <K, V> Set<Cache.Entry<K, V>> entriesx(CacheEntryPredicate... filter) {
- return new EntrySet<>(this, filter, true);
- }
+ public int publicSize();
/**
- * Internal entry set, excluding {@link GridCacheInternal} entries.
+ * Increments public size.
*
- * @return Set of the mappings contained in this map.
+ * @param e Entry that caused public size change.
*/
- public Set<GridCacheEntryEx> entries0() {
- return new Set0<>(this, GridCacheConcurrentMap.nonInternal());
- }
+ public void incrementPublicSize(GridCacheEntryEx e);
/**
- * Get striped entry iterator.
+ * Decrements public size.
*
- * @param id Expected modulo.
- * @param totalCnt Maximum modulo.
- * @return Striped entry iterator.
+ * @param e Entry that caused public size change.
*/
- public Iterator<GridCacheEntryEx> stripedEntryIterator(int id, int totalCnt) {
- return new Iterator0<>(this, false, GridCacheConcurrentMap.nonInternal(), id, totalCnt);
- }
+ public void decrementPublicSize(GridCacheEntryEx e);
- /**
- * Gets all internal entry set, including {@link GridCacheInternal} entries.
- *
- * @return All internal entry set, including {@link GridCacheInternal} entries.
- */
- public Set<GridCacheEntryEx> allEntries0() {
- return new Set0<>(this, CU.empty0());
- }
+ @Nullable public GridCacheMapEntry randomEntry();
/**
- * Key set.
- *
- * @param filter Filter.
- * @return Set of the keys contained in this map.
+ * @return Random entry out of hash map.
*/
- public <K, V> Set<K> keySet(CacheEntryPredicate... filter) {
- return new KeySet<>(this, filter, false);
- }
+ public Set<KeyCacheObject> keySet(CacheEntryPredicate... filter);
/**
- * Key set including internal keys.
- *
* @param filter Filter.
- * @return Set of the keys contained in this map.
+ * @return Iterable of the mappings contained in this map, excluding entries in unvisitable state.
*/
- public <K, V> Set<K> keySetx(CacheEntryPredicate... filter) {
- return new KeySet<>(this, filter, true);
- }
+ public Iterable<GridCacheMapEntry> entries(CacheEntryPredicate... filter);
/**
- * Collection of non-{@code null} values.
- *
* @param filter Filter.
- * @return Collection view of the values contained in this map.
- */
- public <K, V> Collection<V> values(CacheEntryPredicate... filter) {
- return allValues(filter);
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(GridCacheConcurrentMap.class, this, "size", mapSize, "pubSize", mapPubSize);
- }
-
- /**
- *
- */
- @SuppressWarnings({"LockAcquiredButNotSafelyReleased"})
- void printDebugInfo() {
- for (Segment s : segs)
- s.lock();
-
- try {
- X.println(">>> Cache map debug info: " + ctx.namexx());
-
- for (int i = 0; i < segs.length; i++) {
- Segment seg = segs[i];
-
- X.println(" Segment [idx=" + i + ", size=" + seg.size() + ']');
-
- HashEntry[] tab = seg.tbl;
-
- for (int j = 0; j < tab.length; j++)
- X.println(" Bucket [idx=" + j + ", bucket=" + tab[j] + ']');
- }
-
- checkConsistency();
- }
- finally {
- for (Segment s : segs)
- s.unlock();
- }
- }
-
- /**
- *
- */
- public void checkConsistency() {
- int size = 0;
- int pubSize = 0;
-
- IgniteLogger log = ctx.logger(GridCacheConcurrentMap.class);
-
- for (Segment s : segs) {
- assert s.isHeldByCurrentThread();
-
- HashEntry[] tab = s.tbl;
-
- for (HashEntry b : tab) {
- if (b != null) {
- HashEntry e = b;
-
- assert e != null;
-
- int cnt = 0;
- int pubCnt = 0;
-
- while (e != null) {
- cnt++;
-
- log.info("Cache map entry: " + e);
-
- if (!e.mapEntry.deleted()) {
- if (!(e.mapEntry.key instanceof GridCacheInternal))
- pubCnt++;
- }
-
- e = e.next;
- }
-
- size += cnt;
- pubSize += pubCnt;
- }
- }
- }
-
- assert size() == size : "Invalid size [expected=" + size() + ", actual=" + size + ']';
- assert publicSize() == pubSize : "Invalid public size [expected=" + publicSize() + ", actual=" + pubSize + ']';
- }
-
- /**
- * Segments are specialized versions of hash tables. This
- * subclasses from ReentrantLock opportunistically,
- * just to simplify some locking and avoid separate construction.
- */
- private class Segment extends ReentrantLock {
- /** */
- private static final long serialVersionUID = 0L;
-
- /*
- * Segments maintain a table of entry lists that are ALWAYS
- * kept in a consistent state, so can be read without locking.
- * Next fields of nodes are immutable (final). All list
- * additions are performed at the front of each bin. This
- * makes it easy to check changes, and also fast to traverse.
- * When nodes would otherwise be changed, new nodes are
- * created to replace them. This works well for hash tables
- * since the bin lists tend to be short. (The average length
- * is less than two for the default load factor threshold.)
- *
- * Read operations can thus proceed without locking, but rely
- * on selected uses of volatiles to ensure that completed
- * write operations performed by other threads are
- * noticed. For most purposes, the "count" field, tracking the
- * number of elements, serves as that volatile variable
- * ensuring visibility. This is convenient because this field
- * needs to be read in many read operations anyway:
- *
- * - All (unsynchronized) read operations must first read the
- * "count" field, and should not look at table entries if
- * it is 0.
- *
- * - All (synchronized) write operations should write to
- * the "count" field after structurally changing any bin.
- * The operations must not take any action that could even
- * momentarily cause a concurrent read operation to see
- * inconsistent data. This is made easier by the nature of
- * the read operations in Map. For example, no operation
- * can reveal that the table has grown but the threshold
- * has not yet been updated, so there are no atomicity
- * requirements for this with respect to reads.
- *
- * As a guide, all critical volatile reads and writes to the
- * count field are marked in code comments.
- */
-
- /**
- * The table is rehashed when its size exceeds this threshold.
- * (The value of this field is always <tt>(int)(capacity * loadFactor)</tt>.)
- */
- private int threshold;
-
- /** The number of public elements in this segment's region. */
- private final LongAdder8 pubSize = new LongAdder8();
-
- /**
- * The load factor for the hash table. Even though this value
- * is same for all segments, it is replicated to avoid needing
- * links to outer object.
- * @serial
- */
- private final float loadFactor;
-
- /** Entry table. */
- private volatile HashEntry[] tbl;
-
- /** The number of elements in this segment's region. */
- private volatile int size;
-
- /**
- * @param initCap Initial capacity.
- * @param lf Load factor.
- */
- @SuppressWarnings("unchecked")
- Segment(int initCap, float lf) {
- loadFactor = lf;
-
- tbl = new HashEntry[initCap];
-
- threshold = (int)(initCap * loadFactor);
- }
-
- /**
- * Returns properly casted first entry for given hash.
- *
- * @param hash Hash.
- * @return Entry for hash.
- */
- @Nullable HashEntry getFirst(HashEntry[] tbl, int hash) {
- return tbl[hash & (tbl.length - 1)];
- }
-
- /**
- * @param key Key.
- * @param hash Hash.
- * @return Value.
- */
- @Nullable GridCacheMapEntry get(Object key, int hash) {
- if (size != 0) {
- HashEntry e = getFirst(tbl, hash);
-
- while (e != null) {
- if (e.mapEntry.hash() == hash && key.equals(e.mapEntry.key()))
- return e.mapEntry;
-
- e = e.next;
- }
- }
-
- return null;
- }
-
- /**
- * @param key Key.
- * @param hash Hash.
- * @return {@code True} if segment contains value.
- */
- boolean containsKey(Object key, int hash) {
- if (size != 0) {
- HashEntry e = getFirst(tbl, hash);
-
- while (e != null) {
- if (e.mapEntry.hash() == hash && key.equals(e.mapEntry.key()))
- return true;
-
- e = e.next;
- }
- }
-
- return false;
- }
-
- /**
- * @param key Key.
- * @param hash Hash.
- * @param val Value.
- * @param topVer Topology version.
- * @return Associated value.
- */
- @SuppressWarnings({"unchecked"})
- GridCacheMapEntry put(KeyCacheObject key, int hash, @Nullable CacheObject val, AffinityTopologyVersion topVer) {
- lock();
-
- try {
- return put0(key, hash, val, topVer);
- }
- finally {
- unlock();
- }
- }
-
- /**
- * @param key Key.
- * @param hash Hash.
- * @param val Value.
- * @param topVer Topology version.
- * @return Associated value.
- */
- @SuppressWarnings({"unchecked", "SynchronizationOnLocalVariableOrMethodParameter"})
- private GridCacheMapEntry put0(KeyCacheObject key, int hash, CacheObject val, AffinityTopologyVersion topVer) {
- try {
- int c = size;
-
- if (c++ > threshold) // Ensure capacity.
- rehash();
-
- HashEntry[] tab = tbl;
-
- int idx = hash & (tab.length - 1);
-
- HashEntry bin = tab[idx];
-
- HashEntry e = bin;
-
- while (e != null && (e.mapEntry.hash() != hash || !key.equals(e.mapEntry.key())))
- e = e.next;
-
- GridCacheMapEntry retVal;
-
- if (e != null) {
- retVal = e.mapEntry;
-
- retVal.rawPut(val, 0);
- }
- else {
- HashEntry next = bin != null ? bin : null;
-
- GridCacheMapEntry newEntry = factory.create(ctx, topVer, key, hash, val);
-
- tab[idx] = new HashEntry(newEntry, next);
-
- retVal = newEntry;
-
- // Modify counters.
- if (!retVal.isInternal()) {
- mapPubSize.increment();
-
- pubSize.increment();
- }
-
- mapSize.increment();
-
- size = c;
- }
-
- return retVal;
- }
- finally {
- if (DEBUG)
- checkSegmentConsistency();
- }
- }
-
- /**
- * @param key Key.
- * @param hash Hash.
- * @param val Value.
- * @param topVer Topology version.
- * @param create Create flag.
- * @return Triple where the first element is current entry associated with the key,
- * the second is created entry and the third is doomed (all may be null).
- */
- @SuppressWarnings( {"unchecked"})
- GridTriple<GridCacheMapEntry> putIfObsolete(KeyCacheObject key,
- int hash,
- @Nullable CacheObject val,
- AffinityTopologyVersion topVer,
- boolean create
- ) {
- lock();
-
- try {
- HashEntry[] tab = tbl;
-
- int idx = hash & (tab.length - 1);
-
- HashEntry bin = tab[idx];
-
- GridCacheMapEntry cur = null;
- GridCacheMapEntry created = null;
- GridCacheMapEntry doomed = null;
-
- if (bin == null) {
- if (create)
- cur = created = put0(key, hash, val, topVer);
-
- return new GridTriple<>(cur, created, doomed);
- }
-
- HashEntry e = bin;
-
- while (e != null && (e.mapEntry.hash() != hash || !key.equals(e.mapEntry.key())))
- e = e.next;
-
- if (e != null) {
- if (e.mapEntry.obsolete()) {
- doomed = remove(key, hash, null);
-
- if (create)
- cur = created = put0(key, hash, val, topVer);
- }
- else
- cur = e.mapEntry;
- }
- else if (create)
- cur = created = put0(key, hash, val, topVer);
-
- return new GridTriple<>(cur, created, doomed);
- }
- finally {
- unlock();
- }
- }
-
- /**
- *
- */
- @SuppressWarnings("unchecked")
- void rehash() {
- HashEntry[] oldTbl = tbl;
-
- int oldCap = oldTbl.length;
-
- if (oldCap >= MAX_CAP)
- return;
-
- /*
- * Reclassify nodes in each list to new Map. Because we are
- * using power-of-two expansion, the elements from each bin
- * must either stay at same index, or move with a power of two
- * offset. We eliminate unnecessary node creation by catching
- * cases where old nodes can be reused because their next
- * fields won't change. Statistically, at the default
- * threshold, only about one-sixth of them need cloning when
- * a table doubles. The nodes they replace will be eligible for GC
- * as soon as they are no longer referenced by any
- * reader thread that may be in the midst of traversing table
- * right now.
- */
- HashEntry[] newTbl = new HashEntry[oldCap << 1];
-
- threshold = (int)(newTbl.length * loadFactor);
-
- int sizeMask = newTbl.length - 1;
-
- for (int i = 0; i < oldCap ; i++) {
- HashEntry e = oldTbl[i];
-
- if (e != null) {
- HashEntry next = e.next;
-
- int idx = e.mapEntry.hash() & sizeMask;
-
- if (next == null) // Single node on list
- newTbl[idx] = e;
- else { // Reuse consecutive sequence at same slot
- HashEntry lastRun = e;
-
- int lastIdx = idx;
-
- for (HashEntry last = next; last != null; last = last.next) {
- int k = last.mapEntry.hash() & sizeMask;
-
- if (k != lastIdx) {
- lastIdx = k;
- lastRun = last;
- }
- }
-
- newTbl[lastIdx] = lastRun;
-
- // Clone remaining nodes
- for (HashEntry p = e; p != lastRun; p = p.next) {
- int k = p.mapEntry.hash() & sizeMask;
-
- HashEntry n = newTbl[k];
-
- newTbl[k] = new HashEntry(p.mapEntry, n);
- }
- }
- }
- }
-
- tbl = newTbl;
-
- if (DEBUG)
- checkSegmentConsistency();
- }
-
- /**
- * Remove; match on key only if value null, else match both.
- *
- * @param key Key.
- * @param hash Hash.
- * @param filter Optional predicate.
- * @return Removed value.
- */
- @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
- @Nullable GridCacheMapEntry remove(Object key, int hash,
- @Nullable IgnitePredicate<GridCacheMapEntry> filter) {
- lock();
-
- try {
- int idx = hash & (tbl.length - 1);
-
- HashEntry bin = tbl[idx];
-
- if (bin == null)
- return null;
-
- HashEntry prev = null;
- HashEntry e = bin;
-
- while (e != null && (e.mapEntry.hash() != hash || !key.equals(e.mapEntry.key))) {
- prev = e;
-
- e = e.next;
- }
-
- if (e != null) {
- if (filter != null && !filter.apply(e.mapEntry))
- return null;
-
- if (prev == null)
- tbl[idx] = e.next;
- else
- prev.next = e.next;
-
- // Modify counters.
- synchronized (e) {
- if (!e.mapEntry.isInternal() && !e.mapEntry.deleted()) {
- mapPubSize.decrement();
-
- pubSize.decrement();
- }
- }
-
- mapSize.decrement();
-
- --size;
-
- return e.mapEntry;
- }
-
- return null;
- }
- finally {
- if (DEBUG)
- checkSegmentConsistency();
-
- unlock();
- }
- }
-
- /**
- * @return Entries count within segment.
- */
- int size() {
- return size;
- }
-
- /**
- * @return Public entries count within segment.
- */
- int publicSize() {
- return pubSize.intValue();
- }
-
- /**
- * Decrements segment public size.
- */
- void decrementPublicSize() {
- pubSize.decrement();
- }
-
- /**
- * Decrements segment public size.
- */
- void incrementPublicSize() {
- pubSize.increment();
- }
-
- /**
- * @return Random cache map entry from this segment.
- */
- @Nullable GridCacheMapEntry randomEntry() {
- if (size == 0)
- return null;
-
- HashEntry[] tbl = this.tbl;
-
- Collection<GridCacheMapEntry> entries = new ArrayList<>(3);
-
- int pubCnt = 0;
-
- int start = RAND.nextInt(tbl.length);
-
- outerLoop:
- {
- for (int i = start; i < start + tbl.length; i++) {
- HashEntry first = tbl[i & (tbl.length - 1)];
-
- if (first == null)
- continue;
-
- for (HashEntry e = first; e != null; e = e.next) {
- if (!e.mapEntry.isInternal())
- pubCnt++;
-
- entries.add(e.mapEntry);
-
- if (entries.size() == 3)
- break outerLoop;
-
- }
- }
- }
-
- if (entries.isEmpty())
- return null;
-
- if (pubCnt == 0)
- return null;
-
- // Desired and current indexes.
- int idx = RAND.nextInt(pubCnt);
-
- int i = 0;
-
- GridCacheMapEntry retVal = null;
-
- for (GridCacheMapEntry e : entries) {
- if (!(e.key instanceof GridCacheInternal)) {
- // In case desired entry was deleted, we return the closest one from left.
- retVal = e;
-
- if (idx == i++)
- break;
- }
- }
-
- return retVal;
- }
-
- /**
- *
- */
- void checkSegmentConsistency() {
- HashEntry[] tbl = this.tbl;
-
- int cnt = 0;
- int pubCnt = 0;
-
- for (HashEntry b : tbl) {
- if (b != null) {
- HashEntry e = b;
-
- assert e != null;
-
- while (e != null) {
- cnt++;
-
- if (!(e.mapEntry.key instanceof GridCacheInternal))
- pubCnt++;
-
- e = e.next;
- }
- }
- }
-
- assert cnt == size : "Entry count and header size mismatch [cnt=" + cnt + ", hdrSize=" +
- size + ", segment=" + this + ']';
- assert pubCnt == pubSize.intValue();
- }
- }
-
- /**
- * Iterator over {@link GridCacheEntryEx} elements.
- *
- * @param <K> Key type.
- * @param <V> Value type.
+ * @return Iterable of the mappings contained in this map, including entries in unvisitable state.
*/
- private static class Iterator0<K, V> implements Iterator<GridCacheEntryEx>, Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** */
- private int nextSegIdx;
-
- /** */
- private int nextTblIdx;
-
- /** */
- private HashEntry[] curTbl;
-
- /** */
- private HashEntry nextEntry;
-
- /** Next entry to return. */
- private HashEntry next;
-
- /** Next value. */
- private V nextVal;
-
- /** Current value. */
- private V curVal;
-
- /** */
- private boolean isVal;
-
- /** Current entry. */
- private HashEntry cur;
-
- /** Iterator filter. */
- private CacheEntryPredicate[] filter;
-
- /** Outer cache map. */
- private GridCacheConcurrentMap map;
-
- /** Cache context. */
- private GridCacheContext<K, V> ctx;
-
- /** Mod. */
- private int id;
-
- /** Mod count. */
- private int totalCnt;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public Iterator0() {
- // No-op.
- }
-
- /**
- * @param map Cache map.
- * @param isVal {@code True} if value iterator.
- * @param filter Entry filter.
- * @param id ID of the iterator.
- * @param totalCnt Total count of iterators.
- */
- @SuppressWarnings({"unchecked"})
- Iterator0(GridCacheConcurrentMap map, boolean isVal,
- CacheEntryPredicate[] filter, int id, int totalCnt) {
- this.filter = filter;
- this.isVal = isVal;
- this.id = id;
- this.totalCnt = totalCnt;
-
- this.map = map;
-
- ctx = map.ctx;
-
- nextSegIdx = map.segs.length - 1;
- nextTblIdx = -1;
-
- advance();
- }
-
- /**
- *
- */
- @SuppressWarnings({"unchecked"})
- private void advance() {
- if (nextEntry != null && advanceInBucket(nextEntry, true))
- return;
-
- while (nextTblIdx >= 0) {
- HashEntry bucket = curTbl[nextTblIdx--];
-
- if (bucket != null && advanceInBucket(bucket, false))
- return;
- }
-
- while (nextSegIdx >= 0) {
- int nextSegIdx0 = nextSegIdx--;
-
- GridCacheConcurrentMap.Segment seg = map.segs[nextSegIdx0];
-
- if (seg.size != 0 && (id == -1 || nextSegIdx0 % totalCnt == id)) {
- curTbl = seg.tbl;
-
- for (int j = curTbl.length - 1; j >= 0; --j) {
- HashEntry bucket = curTbl[j];
-
- if (bucket != null && advanceInBucket(bucket, false)) {
- nextTblIdx = j - 1;
-
- return;
- }
- }
- }
- }
- }
-
- /**
- * @param e Current next.
- * @param skipFirst {@code True} to skip check on first iteration.
- * @return {@code True} if advance succeeded.
- */
- @SuppressWarnings( {"unchecked"})
- private boolean advanceInBucket(@Nullable HashEntry e, boolean skipFirst) {
- if (e == null)
- return false;
-
- nextEntry = e;
-
- do {
- if (!skipFirst) {
- next = nextEntry;
-
- // Check if entry is visitable first before doing projection-aware peek.
- if (!next.mapEntry.visitable(filter))
- continue;
-
- if (isVal) {
- nextVal = next.mapEntry.<K, V>wrap().getValue();
-
- if (nextVal == null)
- continue;
- }
-
- return true;
- }
-
- // Perform checks in any case.
- skipFirst = false;
- }
- while ((nextEntry = nextEntry.next) != null);
-
- next = null;
- nextVal = null;
-
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public boolean hasNext() {
- return next != null && (!isVal || nextVal != null);
- }
-
- /**
- * @return Next value.
- */
- public V currentValue() {
- return curVal;
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public GridCacheEntryEx next() {
- HashEntry e = next;
- V v = nextVal;
-
- if (e == null)
- throw new NoSuchElementException();
-
- advance();
-
- cur = e;
- curVal = v;
-
- return cur.mapEntry;
- }
-
- /** {@inheritDoc} */
- @Override public void remove() {
- if (cur == null)
- throw new IllegalStateException();
-
- HashEntry e = cur;
-
- cur = null;
- curVal = null;
-
- try {
- ((IgniteKernal)ctx.grid()).getCache(ctx.name()).getAndRemove(e.mapEntry.key);
- }
- catch (IgniteCheckedException ex) {
- throw new IgniteException(ex);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(ctx);
- out.writeObject(filter);
- out.writeBoolean(isVal);
- out.writeInt(id);
- out.writeInt(totalCnt);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- ctx = (GridCacheContext<K, V>)in.readObject();
- filter = (CacheEntryPredicate[])in.readObject();
- isVal = in.readBoolean();
- id = in.readInt();
- totalCnt = in.readInt();
- }
-
- /**
- * Reconstructs object on unmarshalling.
- *
- * @return Reconstructed object.
- * @throws ObjectStreamException Thrown in case of unmarshalling error.
- */
- protected Object readResolve() throws ObjectStreamException {
- return new Iterator0<>(ctx.cache().map(), isVal, filter, id, totalCnt);
- }
- }
+ public Iterable<GridCacheMapEntry> allEntries(CacheEntryPredicate... filter);
/**
- * Entry set.
- */
- @SuppressWarnings("unchecked")
- private static class Set0<K, V> extends AbstractSet<GridCacheEntryEx> implements Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Filter. */
- private CacheEntryPredicate[] filter;
-
- /** Base map. */
- private GridCacheConcurrentMap map;
-
- /** Context. */
- private GridCacheContext<K, V> ctx;
-
- /** */
- private CacheOperationContext opCtxPerCall;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public Set0() {
- // No-op.
- }
-
- /**
- * @param map Base map.
- * @param filter Filter.
- */
- private Set0(GridCacheConcurrentMap map, CacheEntryPredicate[] filter) {
- assert map != null;
-
- this.map = map;
- this.filter = filter;
-
- ctx = map.ctx;
-
- opCtxPerCall = ctx.operationContextPerCall();
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Iterator<GridCacheEntryEx> iterator() {
- return new Iterator0<>(map, false, filter, -1, -1);
- }
-
- /**
- * @return Entry iterator.
- */
- Iterator<Cache.Entry<K, V>> entryIterator() {
- return new EntryIterator<>(map, filter, ctx, opCtxPerCall);
- }
-
- /**
- * @return Key iterator.
- */
- Iterator<K> keyIterator() {
- return new KeyIterator<>(map, opCtxPerCall != null && opCtxPerCall.isKeepBinary(), filter);
- }
-
- /**
- * @return Value iterator.
- */
- Iterator<V> valueIterator() {
- return new ValueIterator<>(map, filter, ctx);
- }
-
- /**
- * Checks for key containment.
- *
- * @param k Key to check.
- * @return {@code True} if key is in the map.
- */
- boolean containsKey(K k) {
- KeyCacheObject cacheKey = ctx.toCacheKeyObject(k);
-
- GridCacheEntryEx e = ctx.cache().peekEx(cacheKey);
-
- try {
- return e != null && !e.obsolete() &&
- (!e.deleted() || e.lockedByThread()) &&
- F.isAll(e, filter);
- }
- catch (GridCacheEntryRemovedException ignore) {
- return false;
- }
- }
-
- /**
- * @param v Checks if value is contained in
- * @return {@code True} if value is in the set.
- */
- boolean containsValue(V v) {
- A.notNull(v, "value");
-
- for (Iterator<V> it = valueIterator(); it.hasNext(); ) {
- V v0 = it.next();
-
- if (F.eq(v0, v))
- return true;
- }
-
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public boolean contains(Object o) {
- if (!(o instanceof GridCacheEntryEx))
- return false;
-
- GridCacheEntryEx e = (GridCacheEntryEx)o;
-
- GridCacheEntryEx cur = ctx.cache().peekEx(e.key());
-
- return cur != null && cur.equals(e);
- }
-
- /** {@inheritDoc} */
- @Override public boolean remove(Object o) {
- return o instanceof Cache.Entry && removeKey(((Map.Entry<K, V>)o).getKey());
- }
-
- /**
- * @param k Key to remove.
- * @return If key has been removed.
- */
- boolean removeKey(K k) {
- try {
- return ((IgniteKernal)ctx.grid()).getCache(ctx.name()).remove(k, CU.<K, V>empty());
- }
- catch (IgniteCheckedException e) {
- throw new IgniteException("Failed to remove cache entry for key: " + k, e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return F.isEmpty(filter) ? map.publicSize() : F.size(iterator());
- }
-
- /** {@inheritDoc} */
- @Override public boolean isEmpty() {
- return F.isEmpty(filter) ? map.publicSize() == 0 : !iterator().hasNext();
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- ctx.cache().clearLocallyAll(new KeySet<K, V>(map, filter, false), true, true, false);
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(ctx);
- out.writeObject(filter);
- }
-
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- ctx = (GridCacheContext<K, V>)in.readObject();
- filter = (CacheEntryPredicate[])in.readObject();
- }
-
- /**
- * Reconstructs object on unmarshalling.
- *
- * @return Reconstructed object.
- * @throws ObjectStreamException Thrown in case of unmarshalling error.
- */
- protected Object readResolve() throws ObjectStreamException {
- return new Set0<>(ctx.cache().map(), filter);
- }
- }
-
- /**
- * Iterator over hash table.
- * <p>
- * Note, class is static for {@link Externalizable}.
- */
- private static class EntryIterator<K, V> implements Iterator<Cache.Entry<K, V>>, Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Base iterator. */
- private Iterator0<K, V> it;
-
- /** */
- private GridCacheContext<K, V> ctx;
-
- /** */
- private CacheOperationContext opCtxPerCall;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public EntryIterator() {
- // No-op.
- }
-
- /**
- * @param map Cache map.
- * @param filter Entry filter.
- * @param ctx Cache context.
- * @param opCtxPerCall Operation context per call.
- */
- EntryIterator(
- GridCacheConcurrentMap map,
- CacheEntryPredicate[] filter,
- GridCacheContext<K, V> ctx,
- CacheOperationContext opCtxPerCall) {
- it = new Iterator0<>(map, false, filter, -1, -1);
-
- this.ctx = ctx;
- this.opCtxPerCall = opCtxPerCall;
- }
-
- /** {@inheritDoc} */
- @Override public boolean hasNext() {
- return it.hasNext();
- }
-
- /** {@inheritDoc} */
- @Override public Cache.Entry<K, V> next() {
- CacheOperationContext old = ctx.operationContextPerCall();
-
- ctx.operationContextPerCall(opCtxPerCall);
-
- try {
- return it.next().wrapLazyValue();
- }
- finally {
- ctx.operationContextPerCall(old);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void remove() {
- it.remove();
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(it);
- out.writeObject(ctx);
- out.writeObject(opCtxPerCall);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- it = (Iterator0<K, V>)in.readObject();
- ctx = (GridCacheContext<K, V>)in.readObject();
- opCtxPerCall = (CacheOperationContext)in.readObject();
- }
- }
-
- /**
- * Value iterator.
- * <p>
- * Note that class is static for {@link Externalizable}.
- */
- private static class ValueIterator<K, V> implements Iterator<V>, Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Hash table iterator. */
- private Iterator0<K, V> it;
-
- /** Context. */
- private GridCacheContext<K, V> ctx;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public ValueIterator() {
- // No-op.
- }
-
- /**
- * @param map Base map.
- * @param filter Value filter.
- * @param ctx Cache context.
- */
- private ValueIterator(
- GridCacheConcurrentMap map,
- CacheEntryPredicate[] filter,
- GridCacheContext ctx) {
- it = new Iterator0<>(map, true, filter, -1, -1);
-
- this.ctx = ctx;
- }
-
- /** {@inheritDoc} */
- @Override public boolean hasNext() {
- return it.hasNext();
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public V next() {
- it.next();
-
- // Cached value.
- return it.currentValue();
- }
-
- /** {@inheritDoc} */
- @Override public void remove() {
- it.remove();
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(it);
- out.writeObject(ctx);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- it = (Iterator0)in.readObject();
- ctx = (GridCacheContext<K, V>)in.readObject();
- }
- }
-
- /**
- * Key iterator.
- */
- private static class KeyIterator<K, V> implements Iterator<K>, Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Hash table iterator. */
- private Iterator0<K, V> it;
-
- /** Keep binary flag. */
- private boolean keepBinary;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public KeyIterator() {
- // No-op.
- }
-
- /**
- * @param map Cache map.
- * @param filter Filter.
- */
- private KeyIterator(GridCacheConcurrentMap map, boolean keepBinary, CacheEntryPredicate[] filter) {
- it = new Iterator0<>(map, false, filter, -1, -1);
- this.keepBinary = keepBinary;
- }
-
- /** {@inheritDoc} */
- @Override public boolean hasNext() {
- return it.hasNext();
- }
-
- /** {@inheritDoc} */
- @Override public K next() {
- return (K)it.ctx.cacheObjectContext().unwrapBinaryIfNeeded(it.next().key(), keepBinary, true);
- }
-
- /** {@inheritDoc} */
- @Override public void remove() {
- it.remove();
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(it);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- it = (Iterator0)in.readObject();
- }
- }
-
- /**
- * Key set.
- */
- private static class KeySet<K, V> extends AbstractSet<K> implements Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Base entry set. */
- private Set0<K, V> set;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public KeySet() {
- // No-op.
- }
-
- /**
- * @param map Base map.
- * @param filter Key filter.
- * @param internal Whether to allow internal keys.
- */
- private KeySet(GridCacheConcurrentMap map, CacheEntryPredicate[] filter, boolean internal) {
- assert map != null;
-
- set = new Set0<>(map, internal ? filter : nonInternal(filter));
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Iterator<K> iterator() {
- return set.keyIterator();
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return set.size();
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean contains(Object o) {
- return set.containsKey((K)o);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean remove(Object o) {
- return set.removeKey((K)o);
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- set.clear();
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(set);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- set = (Set0<K, V>)in.readObject();
- }
- }
-
- /**
- * Value set.
- * <p>
- * Note that the set is static for {@link Externalizable} support.
- */
- private static class Values<K, V> extends AbstractCollection<V> implements Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Base entry set. */
- private Set0<K, V> set;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public Values() {
- // No-op.
- }
-
- /**
- * @param map Base map.
- * @param filter Value filter.
- */
- private Values(GridCacheConcurrentMap map, CacheEntryPredicate[] filter) {
- assert map != null;
-
- set = new Set0<>(map, nonInternal(filter));
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Iterator<V> iterator() {
- return set.valueIterator();
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return set.size();
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean contains(Object o) {
- return set.containsValue((V)o);
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- set.clear();
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(set);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- set = (Set0<K, V>)in.readObject();
- }
- }
-
- /**
- * Entry set.
- */
- private static class EntrySet<K, V> extends AbstractSet<Cache.Entry<K, V>> implements Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Base entry set. */
- private Set0<K, V> set;
-
- /**
- * Empty constructor required for {@link Externalizable}.
- */
- public EntrySet() {
- // No-op.
- }
-
- /**
- * @param map Base map.
- * @param filter Key filter.
- */
- private EntrySet(GridCacheConcurrentMap map, CacheEntryPredicate[] filter) {
- this(map, filter, false);
- }
-
- /**
- * @param map Base map.
- * @param filter Key filter.
- * @param internal Whether to allow internal entries.
- */
- private EntrySet(GridCacheConcurrentMap map, CacheEntryPredicate[] filter,
- boolean internal) {
- assert map != null;
-
- set = new Set0<>(map, internal ? filter : nonInternal(filter));
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Iterator<Cache.Entry<K, V>> iterator() {
- return set.entryIterator();
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return set.size();
- }
-
- /** {@inheritDoc} */
- @Override public boolean isEmpty() {
- return set.isEmpty();
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean contains(Object o) {
- if (o instanceof CacheEntryImpl) {
- GridCacheEntryEx unwrapped = set.map.getEntry(((CacheEntryImpl)o).getKey());
-
- return unwrapped != null && set.contains(unwrapped);
- }
-
- return false;
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean remove(Object o) {
- return set.removeKey((K)o);
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- set.clear();
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeObject(set);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- set = (Set0<K, V>)in.readObject();
- }
- }
-
- /**
- *
+ * @param filter Filter.
+ * @return Set of the mappings contained in this map.
*/
- private static class HashEntry {
- /** */
- private final GridCacheMapEntry mapEntry;
-
- /** */
- @GridToStringExclude
- private volatile HashEntry next;
-
- /**
- * @param mapEntry Entry.
- * @param next Next.
- */
- private HashEntry(
- GridCacheMapEntry mapEntry,
- HashEntry next
- ) {
- this.mapEntry = mapEntry;
- this.next = next;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(HashEntry.class, this, super.toString());
- }
- }
+ public Set<GridCacheMapEntry> entrySet(CacheEntryPredicate... filter);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
new file mode 100644
index 0000000..48dae76
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
@@ -0,0 +1,344 @@
+/*
+ * 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;
+
+import java.util.AbstractSet;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+import org.jsr166.ConcurrentHashMap8;
+
+import static org.apache.ignite.events.EventType.EVT_CACHE_ENTRY_CREATED;
+import static org.apache.ignite.events.EventType.EVT_CACHE_ENTRY_DESTROYED;
+
+/**
+ * Implementation of concurrent cache map.
+ */
+public class GridCacheConcurrentMapImpl implements GridCacheConcurrentMap {
+ /** Default load factor. */
+ private static final float DFLT_LOAD_FACTOR = 0.75f;
+
+ /** Default concurrency level. */
+ private static final int DFLT_CONCUR_LEVEL = Runtime.getRuntime().availableProcessors() * 2;
+
+ /** Internal map. */
+ private final ConcurrentMap<KeyCacheObject, GridCacheMapEntry> map;
+
+ /** Map entry factory. */
+ private final GridCacheMapEntryFactory factory;
+
+ /** Cache context. */
+ private final GridCacheContext ctx;
+
+ /** Public size counter. */
+ private final AtomicInteger pubSize = new AtomicInteger();
+
+ /**
+ * Creates a new, empty map with the specified initial
+ * capacity.
+ *
+ * @param ctx Cache context.
+ * @param factory Entry factory.
+ * @param initialCapacity the initial capacity. The implementation
+ * performs internal sizing to accommodate this many elements.
+
+ * @throws IllegalArgumentException if the initial capacity is
+ * negative.
+ */
+ public GridCacheConcurrentMapImpl(GridCacheContext ctx, GridCacheMapEntryFactory factory, int initialCapacity) {
+ this(ctx, factory, initialCapacity, DFLT_LOAD_FACTOR, DFLT_CONCUR_LEVEL);
+ }
+
+ /**
+ * Creates a new, empty map with the specified initial
+ * capacity, load factor and concurrency level.
+ *
+ * @param ctx Cache context.
+ * @param factory Entry factory.
+ * @param initialCapacity the initial capacity. The implementation
+ * performs internal sizing to accommodate this many elements.
+ * @param loadFactor the load factor threshold, used to control resizing.
+ * Resizing may be performed when the average number of elements per
+ * bin exceeds this threshold.
+ * @param concurrencyLevel the estimated number of concurrently
+ * updating threads. The implementation performs internal sizing
+ * to try to accommodate this many threads.
+ * @throws IllegalArgumentException if the initial capacity is
+ * negative or the load factor or concurrencyLevel are
+ * non-positive.
+ */
+ public GridCacheConcurrentMapImpl(
+ GridCacheContext ctx,
+ GridCacheMapEntryFactory factory,
+ int initialCapacity,
+ float loadFactor,
+ int concurrencyLevel
+ ) {
+ this.ctx = ctx;
+ this.factory = factory;
+
+ map = new ConcurrentHashMap8<>(initialCapacity, loadFactor, concurrencyLevel);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridCacheMapEntry getEntry(KeyCacheObject key) {
+ return map.get(key);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridCacheMapEntry putEntryIfObsoleteOrAbsent(final AffinityTopologyVersion topVer,
+ KeyCacheObject key, @Nullable final CacheObject val, final boolean create, final boolean touch) {
+
+ GridCacheMapEntry cur = null;
+ GridCacheMapEntry created = null;
+ GridCacheMapEntry created0 = null;
+ GridCacheMapEntry doomed = null;
+
+ boolean done = false;
+
+ while (!done) {
+ GridCacheMapEntry entry = map.get(key);
+ created = null;
+ doomed = null;
+
+ if (entry == null) {
+ if (create) {
+ if (created0 == null)
+ created0 = factory.create(ctx, topVer, key, key.hashCode(), val);
+
+ cur = created = created0;
+
+ done = map.putIfAbsent(created.key(), created) == null;
+ }
+ else
+ done = true;
+ }
+ else {
+ if (entry.obsolete()) {
+ doomed = entry;
+
+ if (create) {
+ if (created0 == null)
+ created0 = factory.create(ctx, topVer, key, key.hashCode(), val);
+
+ cur = created = created0;
+
+ done = map.replace(entry.key(), doomed, created);
+ }
+ else
+ done = map.remove(entry.key(), doomed);
+ }
+ else {
+ cur = entry;
+
+ done = true;
+ }
+ }
+ }
+
+ int sizeChange = 0;
+
+ if (doomed != null) {
+ synchronized (doomed) {
+ if (!doomed.deleted())
+ sizeChange--;
+ }
+
+ if (ctx.events().isRecordable(EVT_CACHE_ENTRY_DESTROYED))
+ ctx.events().addEvent(doomed.partition(),
+ doomed.key(),
+ ctx.localNodeId(),
+ (IgniteUuid)null,
+ null,
+ EVT_CACHE_ENTRY_DESTROYED,
+ null,
+ false,
+ null,
+ false,
+ null,
+ null,
+ null,
+ true);
+ }
+
+ if (created != null) {
+ sizeChange++;
+
+ if (ctx.events().isRecordable(EVT_CACHE_ENTRY_CREATED))
+ ctx.events().addEvent(created.partition(),
+ created.key(),
+ ctx.localNodeId(),
+ (IgniteUuid)null,
+ null,
+ EVT_CACHE_ENTRY_CREATED,
+ null,
+ false,
+ null,
+ false,
+ null,
+ null,
+ null,
+ true);
+
+ if (touch)
+ ctx.evicts().touch(
+ cur,
+ topVer);
+ }
+
+ if (sizeChange != 0)
+ pubSize.addAndGet(sizeChange);
+
+ return cur;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean removeEntry(final GridCacheEntryEx entry) {
+ boolean removed = map.remove(entry.key(), entry);
+
+ if (removed) {
+ if (ctx.events().isRecordable(EVT_CACHE_ENTRY_DESTROYED))
+ // Event notification.
+ ctx.events().addEvent(entry.partition(), entry.key(), ctx.localNodeId(), (IgniteUuid)null, null,
+ EVT_CACHE_ENTRY_DESTROYED, null, false, null, false, null, null, null, false);
+
+ synchronized (entry) {
+ if (!entry.deleted())
+ decrementPublicSize(entry);
+ }
+ }
+
+ return removed;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ return map.size();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int publicSize() {
+ return pubSize.get();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void incrementPublicSize(GridCacheEntryEx e) {
+ pubSize.incrementAndGet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void decrementPublicSize(GridCacheEntryEx e) {
+ pubSize.decrementAndGet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<KeyCacheObject> keySet(final CacheEntryPredicate... filter) {
+ final IgnitePredicate<KeyCacheObject> p = new IgnitePredicate<KeyCacheObject>() {
+ @Override public boolean apply(KeyCacheObject key) {
+ GridCacheMapEntry entry = map.get(key);
+
+ return entry != null && entry.visitable(filter);
+ }
+ };
+
+ return new AbstractSet<KeyCacheObject>() {
+ @Override public Iterator<KeyCacheObject> iterator() {
+ return F.iterator0(map.keySet(), true, p);
+ }
+
+ @Override public int size() {
+ return F.size(iterator());
+ }
+
+ @Override public boolean contains(Object o) {
+ if (!(o instanceof KeyCacheObject))
+ return false;
+
+ return map.keySet().contains(o) && p.apply((KeyCacheObject)o);
+ }
+ };
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<GridCacheMapEntry> entries(final CacheEntryPredicate... filter) {
+ final IgnitePredicate<GridCacheMapEntry> p = new IgnitePredicate<GridCacheMapEntry>() {
+ @Override public boolean apply(GridCacheMapEntry entry) {
+ return entry.visitable(filter);
+ }
+ };
+
+ return F.viewReadOnly(map.values(), F.<GridCacheMapEntry>identity(), p);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<GridCacheMapEntry> allEntries(final CacheEntryPredicate... filter) {
+ final IgnitePredicate<GridCacheMapEntry> p = new IgnitePredicate<GridCacheMapEntry>() {
+ @Override public boolean apply(GridCacheMapEntry entry) {
+ return F.isAll(entry, filter);
+ }
+ };
+
+ return F.viewReadOnly(map.values(), F.<GridCacheMapEntry>identity(), p);
+ }
+
+ /** {@inheritDoc} */
+ @Deprecated @Nullable @Override public GridCacheMapEntry randomEntry() {
+ Iterator<GridCacheMapEntry> iterator = map.values().iterator();
+
+ if (iterator.hasNext())
+ return iterator.next();
+
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Set<GridCacheMapEntry> entrySet(final CacheEntryPredicate... filter) {
+ final IgnitePredicate<GridCacheMapEntry> p = new IgnitePredicate<GridCacheMapEntry>() {
+ @Override public boolean apply(GridCacheMapEntry entry) {
+ return entry.visitable(filter);
+ }
+ };
+
+ return new AbstractSet<GridCacheMapEntry>() {
+ @Override public Iterator<GridCacheMapEntry> iterator() {
+ return F.iterator0(map.values(), true, p);
+ }
+
+ @Override public int size() {
+ return F.size(iterator());
+ }
+
+ @Override public boolean contains(Object o) {
+ if (!(o instanceof GridCacheMapEntry))
+ return false;
+
+ GridCacheMapEntry entry = (GridCacheMapEntry)o;
+
+ return entry.equals(map.get(entry.key())) && p.apply(entry);
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index 88d6e04..7ad6c77 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -59,7 +59,6 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTransactionalCacheAdapter;
@@ -589,14 +588,7 @@ public class GridCacheContext<K, V> implements Externalizable {
assert e != null;
assert !e.isInternal() : e;
- cache.map().incrementSize(e);
-
- if (isDht() || isColocated() || isDhtAtomic()) {
- GridDhtLocalPartition part = topology().localPartition(e.partition(), AffinityTopologyVersion.NONE, false);
-
- if (part != null)
- part.incrementPublicSize();
- }
+ cache.incrementSize(e);
}
/**
@@ -607,14 +599,7 @@ public class GridCacheContext<K, V> implements Externalizable {
assert e != null;
assert !e.isInternal() : e;
- cache.map().decrementSize(e);
-
- if (isDht() || isColocated() || isDhtAtomic()) {
- GridDhtLocalPartition part = topology().localPartition(e.partition(), AffinityTopologyVersion.NONE, false);
-
- if (part != null)
- part.decrementPublicSize();
- }
+ cache.decrementSize(e);
}
/**
@@ -1097,8 +1082,8 @@ public class GridCacheContext<K, V> implements Externalizable {
for (CacheEntryPredicate p0 : p) {
if ((p0 instanceof CacheEntrySerializablePredicate) &&
- ((CacheEntrySerializablePredicate)p0).predicate() instanceof CacheEntryPredicateNoValue)
- return true;
+ ((CacheEntrySerializablePredicate)p0).predicate() instanceof CacheEntryPredicateNoValue)
+ return true;
}
return false;
@@ -1155,7 +1140,7 @@ public class GridCacheContext<K, V> implements Externalizable {
*/
@SuppressWarnings({"unchecked"})
public IgnitePredicate<Cache.Entry<K, V>>[] vararg(IgnitePredicate<Cache.Entry<K, V>> p) {
- return p == null ? CU.<K, V>empty() : new IgnitePredicate[]{p};
+ return p == null ? CU.<K, V>empty() : new IgnitePredicate[] {p};
}
/**
@@ -1320,8 +1305,6 @@ public class GridCacheContext<K, V> implements Externalizable {
return (opCtx != null && opCtx.skipStore());
}
-
-
/**
* @return {@code True} if need check near cache context.
*/
@@ -1804,9 +1787,20 @@ public class GridCacheContext<K, V> implements Externalizable {
* @return Cache key object.
*/
public KeyCacheObject toCacheKeyObject(Object obj) {
+ return toCacheKeyObject(obj, false);
+ }
+
+ /**
+ * @param obj Object.
+ * @return Cache key object.
+ */
+ public KeyCacheObject toCacheKeyObject(Object obj, boolean includePartition) {
assert validObjectForCache(obj) : obj;
- return cacheObjects().toCacheKeyObject(cacheObjCtx, obj, true);
+ if (includePartition)
+ return cacheObjects().toCacheKeyObject(cacheObjCtx, obj, true, affinity().partition(obj));
+ else
+ return cacheObjects().toCacheKeyObject(cacheObjCtx, obj, true);
}
/**
@@ -1838,8 +1832,7 @@ public class GridCacheContext<K, V> implements Externalizable {
* @throws IgniteCheckedException If failed.
*/
@Nullable public CacheObject unswapCacheObject(byte type, byte[] bytes, @Nullable IgniteUuid clsLdrId)
- throws IgniteCheckedException
- {
+ throws IgniteCheckedException {
if (ctx.config().isPeerClassLoadingEnabled() && type != CacheObject.TYPE_BYTE_ARR) {
ClassLoader ldr = clsLdrId != null ? deploy().getClassLoader(clsLdrId) : deploy().localLoader();
@@ -1973,15 +1966,24 @@ public class GridCacheContext<K, V> implements Externalizable {
/**
* @param keys Keys.
- * @return Co
+ * @return Read-only collection of KeyCacheObject instances.
*/
public Collection<KeyCacheObject> cacheKeysView(Collection<?> keys) {
+ return cacheKeysView(keys, false);
+ }
+
+ /**
+ * @param keys Keys.
+ * @param includePartition Include partition.
+ * @return Read-only collection of KeyCacheObject instances.
+ */
+ public Collection<KeyCacheObject> cacheKeysView(Collection<?> keys, final boolean includePartition) {
return F.viewReadOnly(keys, new C1<Object, KeyCacheObject>() {
@Override public KeyCacheObject apply(Object key) {
if (key == null)
throw new NullPointerException("Null key.");
- return toCacheKeyObject(key);
+ return toCacheKeyObject(key, includePartition);
}
});
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
index d54ecd6..8e66233 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheDeploymentManager.java
@@ -304,7 +304,7 @@ public class GridCacheDeploymentManager<K, V> extends GridCacheSharedManagerAdap
private void addEntries(ClassLoader ldr, Collection<KeyCacheObject> keys, GridCacheAdapter cache) {
GridCacheContext cacheCtx = cache.context();
- for (GridCacheEntryEx e : (Collection<GridCacheEntryEx>)cache.entries()) {
+ for (GridCacheEntryEx e : (Iterable<GridCacheEntryEx>)cache.entries()) {
boolean undeploy = cacheCtx.isNear() ?
undeploy(ldr, e, cacheCtx.near()) || undeploy(ldr, e, cacheCtx.near().dht()) :
undeploy(ldr, e, cacheCtx.cache());
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntrySet.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntrySet.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntrySet.java
deleted file mode 100644
index 1dbc564..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntrySet.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * 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;
-
-import java.util.AbstractSet;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Set;
-import javax.cache.Cache;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.lang.IgnitePredicate;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Entry set backed by cache itself.
- */
-public class GridCacheEntrySet<K, V> extends AbstractSet<Cache.Entry<K, V>> {
- /** Cache context. */
- private final GridCacheContext<K, V> ctx;
-
- /** Filter. */
- private final IgnitePredicate<Cache.Entry<K, V>>[] filter;
-
- /** Base set. */
- private final Set<Cache.Entry<K, V>> set;
-
- /**
- * @param ctx Cache context.
- * @param c Entry collection.
- * @param filter Filter.
- */
- public GridCacheEntrySet(GridCacheContext<K, V> ctx, Collection<? extends Cache.Entry<K, V>> c,
- @Nullable IgnitePredicate<Cache.Entry<K, V>>... filter) {
- set = new HashSet<>(c.size(), 1.0f);
-
- assert ctx != null;
-
- this.ctx = ctx;
- this.filter = filter;
-
- for (Cache.Entry<K, V> e : c) {
- if (e != null)
- set.add(e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public Iterator<Cache.Entry<K, V>> iterator() {
- return new GridCacheIterator<>(ctx, set, F.<Cache.Entry<K, V>>identity(), filter);
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- throw new UnsupportedOperationException("clear");
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean remove(Object o) {
- if (!(o instanceof CacheEntryImpl))
- return false;
-
- Cache.Entry<K, V> e = (Cache.Entry<K,V>)o;
-
- if (F.isAll(e, filter) && set.remove(e)) {
- try {
- ((IgniteKernal)ctx.grid()).getCache(ctx.name()).remove(e.getKey(), e.getValue());
- }
- catch (IgniteCheckedException ex) {
- throw new IgniteException(ex);
- }
-
- return true;
- }
-
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return F.size(set, filter);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"unchecked"})
- @Override public boolean contains(Object o) {
- if (!(o instanceof CacheEntryImpl))
- return false;
-
- Cache.Entry<K,V> e = (Cache.Entry<K, V>)o;
-
- return F.isAll(e, filter) && set.contains(e);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
index a21e18b..c102c58 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
@@ -1514,7 +1514,7 @@ public class GridCacheEvictionManager extends GridCacheManagerAdapter {
if (log.isDebugEnabled())
log.debug("Touching partition entries: " + part);
- touchOnTopologyChange(part.entries());
+ touchOnTopologyChange(part.allEntries());
}
}
}
[5/6] ignite git commit: IGNITE-2948 - Optimize usage of
GridCacheConcurrentMap
Posted by vo...@apache.org.
IGNITE-2948 - Optimize usage of GridCacheConcurrentMap
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3be3d16b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3be3d16b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3be3d16b
Branch: refs/heads/ignite-1786
Commit: 3be3d16b0899be7440a60d3eb5c476b789b34bfb
Parents: 00f8d65
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Tue Apr 26 15:07:28 2016 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Tue Apr 26 15:07:28 2016 +0300
----------------------------------------------------------------------
.../internal/benchmarks/model/IntValue.java | 19 +-
.../internal/binary/BinaryObjectImpl.java | 14 +
.../processors/cache/GridCacheAdapter.java | 448 ++--
.../cache/GridCacheAffinityManager.java | 10 +-
.../cache/GridCacheClearAllRunnable.java | 2 +-
.../cache/GridCacheConcurrentMap.java | 1996 +-----------------
.../cache/GridCacheConcurrentMapImpl.java | 344 +++
.../processors/cache/GridCacheContext.java | 56 +-
.../cache/GridCacheDeploymentManager.java | 2 +-
.../processors/cache/GridCacheEntrySet.java | 113 -
.../cache/GridCacheEvictionManager.java | 2 +-
.../processors/cache/GridCacheKeySet.java | 104 -
.../processors/cache/GridCacheMapEntry.java | 22 +-
.../processors/cache/GridCacheProcessor.java | 1 -
.../processors/cache/GridCacheProxyImpl.java | 2 +-
.../processors/cache/GridCacheUtils.java | 1 -
.../processors/cache/GridNoStorageCacheMap.java | 107 +
.../processors/cache/IgniteInternalCache.java | 2 +-
.../processors/cache/KeyCacheObject.java | 11 +
.../processors/cache/KeyCacheObjectImpl.java | 32 +-
.../binary/CacheObjectBinaryProcessorImpl.java | 24 +-
.../GridDistributedCacheAdapter.java | 4 +-
.../distributed/GridDistributedLockRequest.java | 19 +-
.../GridDistributedTxRemoteAdapter.java | 1 -
.../GridDistributedUnlockRequest.java | 16 +-
.../dht/GridCachePartitionedConcurrentMap.java | 191 ++
.../dht/GridClientPartitionTopology.java | 7 -
.../distributed/dht/GridDhtCacheAdapter.java | 77 +-
.../distributed/dht/GridDhtCacheEntry.java | 25 +-
.../distributed/dht/GridDhtLocalPartition.java | 152 +-
.../distributed/dht/GridDhtLockRequest.java | 16 +-
.../dht/GridDhtPartitionTopology.java | 7 -
.../dht/GridDhtPartitionTopologyImpl.java | 281 ++-
.../distributed/dht/GridDhtUnlockRequest.java | 15 +-
.../distributed/dht/GridNoStorageCacheMap.java | 122 --
.../dht/GridPartitionedGetFuture.java | 2 +-
.../dht/GridPartitionedSingleGetFuture.java | 2 +-
.../dht/atomic/GridDhtAtomicCache.java | 75 +-
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 22 +-
.../dht/atomic/GridNearAtomicUpdateFuture.java | 6 +-
.../dht/atomic/GridNearAtomicUpdateRequest.java | 31 +-
.../dht/colocated/GridDhtColocatedCache.java | 13 +-
.../dht/preloader/GridDhtPartitionDemander.java | 2 +-
.../dht/preloader/GridDhtPartitionSupplier.java | 8 +-
.../distributed/near/GridNearCacheAdapter.java | 94 +-
.../distributed/near/GridNearGetFuture.java | 2 +-
.../distributed/near/GridNearGetRequest.java | 45 +-
.../distributed/near/GridNearLockRequest.java | 16 +-
.../near/GridNearSingleGetRequest.java | 22 +-
.../distributed/near/GridNearUnlockRequest.java | 21 +-
.../processors/cache/local/GridLocalCache.java | 7 +-
.../local/atomic/GridLocalAtomicCache.java | 24 +-
.../cache/transactions/IgniteInternalTx.java | 1 -
.../cache/transactions/IgniteTxAdapter.java | 1 -
.../cache/transactions/IgniteTxEntry.java | 33 +-
.../cacheobject/IgniteCacheObjectProcessor.java | 10 +
.../IgniteCacheObjectProcessorImpl.java | 36 +-
.../GridCacheAtomicReferenceImpl.java | 2 +-
.../internal/util/PartitionedReadOnlySet.java | 71 +
.../util/StripedCompositeReadWriteLock.java | 10 +
.../ignite/internal/util/lang/GridFunc.java | 3 +-
.../ignite/internal/visor/cache/VisorCache.java | 6 +-
.../GridCachePreloadingEvictionsSelfTest.java | 4 +-
.../cache/GridCacheTtlManagerSelfTest.java | 2 +-
...idCacheValueConsistencyAbstractSelfTest.java | 3 +-
.../IgniteDynamicClientCacheStartSelfTest.java | 10 -
.../IgniteTxStoreExceptionAbstractSelfTest.java | 8 +-
.../GridCacheBinaryObjectsAbstractSelfTest.java | 2 +-
...actQueueFailoverDataConsistencySelfTest.java | 2 +-
.../GridCacheQueueCleanupSelfTest.java | 4 +-
.../GridCacheSequenceApiSelfAbstractTest.java | 37 -
.../GridCacheSetAbstractSelfTest.java | 5 +-
.../GridCacheSetFailoverAbstractSelfTest.java | 8 +-
.../IgniteDataStructureUniqueNameTest.java | 3 +-
.../IgnitePartitionedQueueNoBackupsTest.java | 6 +-
.../IgnitePartitionedSetNoBackupsSelfTest.java | 6 +-
.../distributed/dht/GridCacheDhtTestUtils.java | 9 +-
.../near/GridCacheNearOneNodeSelfTest.java | 4 +-
.../processors/igfs/IgfsAbstractSelfTest.java | 10 +-
79 files changed, 1958 insertions(+), 2975 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/model/IntValue.java
----------------------------------------------------------------------
diff --git a/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/model/IntValue.java b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/model/IntValue.java
index 0a97e36..45fbe79 100644
--- a/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/model/IntValue.java
+++ b/modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/model/IntValue.java
@@ -85,7 +85,24 @@ public class IntValue implements Externalizable, Binarylizable {
}
/** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ IntValue value = (IntValue)o;
+
+ return val == value.val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return val;
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return "Value [id=" + val + ']';
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
index 0997d6f..fa10de3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
@@ -72,6 +72,10 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
@GridDirectTransient
private boolean detachAllowed;
+ /** */
+ @GridDirectTransient
+ private int part = -1;
+
/**
* For {@link Externalizable}.
*/
@@ -94,6 +98,16 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
}
/** {@inheritDoc} */
+ @Override public int partition() {
+ return part;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void partition(int part) {
+ this.part = part;
+ }
+
+ /** {@inheritDoc} */
@Override public byte cacheObjectType() {
return TYPE_BINARY;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/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 d807e26..8c1a750 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
@@ -23,8 +23,8 @@ import java.io.InvalidObjectException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.ObjectStreamException;
+import java.util.AbstractSet;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -101,13 +101,11 @@ import org.apache.ignite.internal.processors.platform.cache.PlatformCacheEntryFi
import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
-import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridEmbeddedFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridClosureException;
-import org.apache.ignite.internal.util.lang.GridTriple;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.C2;
@@ -132,7 +130,6 @@ import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteProductVersion;
-import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.mxbean.CacheMetricsMXBean;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.apache.ignite.resources.IgniteInstanceResource;
@@ -146,8 +143,6 @@ import org.jsr166.LongAdder8;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_CACHE_KEY_VALIDATION_DISABLED;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_CACHE_RETRIES_COUNT;
-import static org.apache.ignite.events.EventType.EVT_CACHE_ENTRY_CREATED;
-import static org.apache.ignite.events.EventType.EVT_CACHE_ENTRY_DESTROYED;
import static org.apache.ignite.internal.GridClosureCallMode.BROADCAST;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_LOAD;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
@@ -304,7 +299,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
@SuppressWarnings("OverriddenMethodCallDuringObjectConstruction")
protected GridCacheAdapter(GridCacheContext<K, V> ctx, int startSize) {
- this(ctx, new GridCacheConcurrentMap(ctx, startSize, null));
+ this(ctx, null);
}
/**
@@ -312,7 +307,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param map Concurrent map.
*/
@SuppressWarnings({"OverriddenMethodCallDuringObjectConstruction", "deprecation"})
- protected GridCacheAdapter(final GridCacheContext<K, V> ctx, GridCacheConcurrentMap map) {
+ protected GridCacheAdapter(final GridCacheContext<K, V> ctx, @Nullable GridCacheConcurrentMap map) {
assert ctx != null;
this.ctx = ctx;
@@ -393,6 +388,22 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/**
+ * Increments map public size.
+ * @param e Map entry.
+ */
+ public void incrementSize(GridCacheMapEntry e) {
+ map.incrementPublicSize(e);
+ }
+
+ /**
+ * Decrements map public size.
+ * @param e Map entry.
+ */
+ public void decrementSize(GridCacheMapEntry e) {
+ map.decrementPublicSize(e);
+ }
+
+ /**
* @return Context.
*/
@Override public GridCacheContext<K, V> context() {
@@ -483,7 +494,6 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return new GridCacheProxyImpl<>((GridCacheContext<K1, V1>)ctx, (GridCacheAdapter<K1, V1>)this, opCtx);
}
-
/** {@inheritDoc} */
@Nullable @Override public ExpiryPolicy expiry() {
return null;
@@ -541,13 +551,25 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/**
+ * @return Entry factory.
+ */
+ protected abstract GridCacheMapEntryFactory entryFactory();
+
+ /**
* Starts this cache. Child classes should override this method
* to provide custom start-up behavior.
*
* @throws IgniteCheckedException If start failed.
*/
public void start() throws IgniteCheckedException {
- // No-op.
+ if (map == null) {
+ int initSize = ctx.config().getStartSize();
+
+ if (!isLocal())
+ initSize /= ctx.affinity().partitions();
+
+ map = new GridCacheConcurrentMapImpl(ctx, entryFactory(), initSize);
+ }
}
/**
@@ -687,7 +709,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
modes.backup = true;
if (modes.heap)
- its.add(iterator(map.entries0().iterator(), !ctx.keepBinary()));
+ its.add(iterator(map.entries().iterator(), !ctx.keepBinary()));
}
else if (modes.heap) {
if (modes.near && ctx.isNear())
@@ -946,74 +968,28 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
GridCacheMapEntry cur = map.getEntry(key);
if (cur == null || cur.obsolete()) {
- GridTriple<GridCacheMapEntry> t = map.putEntryIfObsoleteOrAbsent(
+ cur = map.putEntryIfObsoleteOrAbsent(
topVer,
key,
null,
- create);
-
- cur = t.get1();
-
- GridCacheEntryEx created = t.get2();
- GridCacheEntryEx doomed = t.get3();
-
- if (doomed != null && ctx.events().isRecordable(EVT_CACHE_ENTRY_DESTROYED))
- // Event notification.
- ctx.events().addEvent(doomed.partition(),
- doomed.key(),
- locNodeId,
- (IgniteUuid)null,
- null,
- EVT_CACHE_ENTRY_DESTROYED,
- null,
- false,
- null,
- false,
- null,
- null,
- null,
- true);
-
- if (created != null) {
- // Event notification.
- if (ctx.events().isRecordable(EVT_CACHE_ENTRY_CREATED))
- ctx.events().addEvent(created.partition(),
- created.key(),
- locNodeId,
- (IgniteUuid)null,
- null,
- EVT_CACHE_ENTRY_CREATED,
- null,
- false,
- null,
- false,
- null,
- null,
- null,
- true);
-
- if (touch)
- ctx.evicts().touch(
- cur,
- topVer);
- }
+ create, touch);
}
return cur;
}
/**
- * @return Set of internal cached entry representations, excluding {@link GridCacheInternal} keys.
+ * @return Set of internal cached entry representations.
*/
- public Set<GridCacheEntryEx> entries() {
- return map.entries0();
+ public Iterable<? extends GridCacheEntryEx> entries() {
+ return allEntries();
}
/**
- * @return Set of internal cached entry representations, including {@link GridCacheInternal} keys.
+ * @return Set of internal cached entry representations.
*/
- public Set<GridCacheEntryEx> allEntries() {
- return map.allEntries0();
+ public Iterable<? extends GridCacheEntryEx> allEntries() {
+ return map.entries();
}
/** {@inheritDoc} */
@@ -1022,8 +998,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
- @Override public Set<Cache.Entry<K, V>> entrySetx(CacheEntryPredicate... filter) {
- return map.entriesx(filter);
+ @Override public Set<Cache.Entry<K, V>> entrySetx(final CacheEntryPredicate... filter) {
+ return new EntrySet(map.entrySet(filter));
}
/** {@inheritDoc} */
@@ -1033,22 +1009,57 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/** {@inheritDoc} */
@Override public Set<K> keySet() {
- return map.keySet();
+ return new KeySet(map.entrySet());
}
/** {@inheritDoc} */
@Override public Set<K> keySetx() {
- return map.keySetx();
+ return keySet();
}
/** {@inheritDoc} */
@Override public Set<K> primaryKeySet() {
- return map.keySet(CU.cachePrimary(ctx.grid().affinity(ctx.name()), ctx.localNode()));
+ return new KeySet(map.entrySet(CU.cachePrimary(ctx.grid().affinity(ctx.name()), ctx.localNode())));
}
/** {@inheritDoc} */
- @Override public Collection<V> values() {
- return map.values();
+ @Override public Iterable<V> values() {
+ return values((CacheEntryPredicate[])null);
+ }
+
+ /**
+ * Collection of values cached on this node. You cannot modify this collection.
+ * <p>
+ * Iterator over this collection will not fail if collection was
+ * concurrently updated by another thread. This means that iterator may or
+ * may not return latest values depending on whether they were added before
+ * or after current iterator position.
+ * <p>
+ * NOTE: this operation is not distributed and returns only the values cached on this node.
+ *
+ * @param filter Filters.
+ * @return Collection of cached values.
+ */
+ public Iterable<V> values(final CacheEntryPredicate... filter) {
+ return new Iterable<V>() {
+ @Override public Iterator<V> iterator() {
+ return new Iterator<V>() {
+ private final Iterator<? extends GridCacheEntryEx> it = entries().iterator();
+
+ @Override public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ @Override public V next() {
+ return (V) it.next().wrap().getValue();
+ }
+
+ @Override public void remove() {
+ throw new UnsupportedOperationException("remove");
+ }
+ };
+ }
+ };
}
/**
@@ -1058,21 +1069,10 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
public void removeIfObsolete(KeyCacheObject key) {
assert key != null;
- GridCacheEntryEx entry = map.removeEntryIfObsolete(key);
+ GridCacheMapEntry entry = map.getEntry(key);
- if (entry != null) {
- assert entry.obsolete() : "Removed non-obsolete entry: " + entry;
-
- if (log.isDebugEnabled())
- log.debug("Removed entry from cache: " + entry);
-
- if (ctx.events().isRecordable(EVT_CACHE_ENTRY_DESTROYED))
- // Event notification.
- ctx.events().addEvent(entry.partition(), entry.key(), locNodeId, (IgniteUuid)null, null,
- EVT_CACHE_ENTRY_DESTROYED, null, false, null, false, null, null, null, false);
- }
- else if (log.isDebugEnabled())
- log.debug("Remove will not be done for key (obsolete entry got replaced or removed): " + key);
+ if (entry.obsolete())
+ removeEntry(entry);
}
/**
@@ -1088,7 +1088,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
int keySize = size();
int cnt = Math.min(keySize / CLEAR_ALL_SPLIT_THRESHOLD + (keySize % CLEAR_ALL_SPLIT_THRESHOLD != 0 ? 1 : 0),
- Runtime.getRuntime().availableProcessors());
+ Runtime.getRuntime().availableProcessors());
if (cnt == 0)
cnt = 1; // Still perform cleanup since there could be entries in swap.
@@ -1256,7 +1256,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param entry Removes entry from cache if currently mapped value is the same as passed.
*/
public void removeEntry(GridCacheEntryEx entry) {
- map.removeEntry(entry);
+ boolean removed = map.removeEntry(entry);
+
+ if (log.isDebugEnabled()) {
+ if (removed)
+ log.debug("Removed entry from cache: " + entry);
+ else
+ log.debug("Remove will not be done for key (entry got replaced or removed): " + entry.key());
+ }
}
/**
@@ -1399,7 +1406,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
T2<V, GridCacheVersion> t = (T2<V, GridCacheVersion>)get(key, !ctx.keepBinary(), true);
- CacheEntry<K, V> val = t != null ? new CacheEntryImplEx<>(key, t.get1(), t.get2()): null;
+ CacheEntry<K, V> val = t != null ? new CacheEntryImplEx<>(key, t.get1(), t.get2()) : null;
if (ctx.config().getInterceptor() != null) {
V val0 = (V)ctx.config().getInterceptor().onGet(key, t != null ? val.getValue() : null);
@@ -1424,7 +1431,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
IgniteInternalFuture<V> fut = getAsync(key, !ctx.keepBinary(), false);
if (ctx.config().getInterceptor() != null)
- fut = fut.chain(new CX1<IgniteInternalFuture<V>, V>() {
+ fut = fut.chain(new CX1<IgniteInternalFuture<V>, V>() {
@Override public V applyx(IgniteInternalFuture<V> f) throws IgniteCheckedException {
return (V)ctx.config().getInterceptor().onGet(key, f.get());
}
@@ -1451,20 +1458,20 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
IgniteInternalFuture<CacheEntry<K, V>> fr = fut.chain(
new CX1<IgniteInternalFuture<T2<V, GridCacheVersion>>, CacheEntry<K, V>>() {
- @Override public CacheEntry<K, V> applyx(IgniteInternalFuture<T2<V, GridCacheVersion>> f)
- throws IgniteCheckedException {
- T2<V, GridCacheVersion> t = f.get();
+ @Override public CacheEntry<K, V> applyx(IgniteInternalFuture<T2<V, GridCacheVersion>> f)
+ throws IgniteCheckedException {
+ T2<V, GridCacheVersion> t = f.get();
- CacheEntry<K, V> val = t != null ? new CacheEntryImplEx<>(key, t.get1(), t.get2()) : null;
- if (intercept) {
- V val0 = (V)ctx.config().getInterceptor().onGet(key, t != null ? val.getValue() : null);
+ CacheEntry<K, V> val = t != null ? new CacheEntryImplEx<>(key, t.get1(), t.get2()) : null;
+ if (intercept) {
+ V val0 = (V)ctx.config().getInterceptor().onGet(key, t != null ? val.getValue() : null);
- return new CacheEntryImplEx<>(key, val0, t != null ? t.get2() : null);
+ return new CacheEntryImplEx<>(key, val0, t != null ? t.get2() : null);
+ }
+ else
+ return val;
}
- else
- return val;
- }
- });
+ });
if (statsEnabled)
fut.listen(new UpdateGetTimeStatClosure<T2<V, GridCacheVersion>>(metrics0(), start));
@@ -1516,7 +1523,6 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return res;
}
-
/** {@inheritDoc} */
@Override public IgniteInternalFuture<Map<K, V>> getAllAsync(@Nullable final Collection<? extends K> keys) {
A.notNull(keys, "keys");
@@ -1541,7 +1547,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
/** {@inheritDoc} */
- @Override public IgniteInternalFuture<Collection<CacheEntry<K, V>>> getEntriesAsync(
+ @Override public IgniteInternalFuture<Collection<CacheEntry<K, V>>> getEntriesAsync(
@Nullable final Collection<? extends K> keys) {
A.notNull(keys, "keys");
@@ -1850,7 +1856,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final AffinityTopologyVersion topVer = tx == null ?
(canRemap ?
ctx.affinity().affinityTopologyVersion() : ctx.shared().exchange().readyAffinityVersion()) :
- tx.topologyVersion();
+ tx.topologyVersion();
int keysSize = keys.size();
@@ -2091,8 +2097,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @throws IgniteCheckedException If failed.
*/
@Nullable public V getAndPut(final K key, final V val, @Nullable final CacheEntryPredicate filter)
- throws IgniteCheckedException
- {
+ throws IgniteCheckedException {
boolean statsEnabled = ctx.config().isStatisticsEnabled();
long start = statsEnabled ? System.nanoTime() : 0L;
@@ -2351,7 +2356,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
IgniteInternalFuture<?> fut = asyncOp(new AsyncOp() {
@Override public IgniteInternalFuture op(IgniteTxLocalAdapter tx, AffinityTopologyVersion readyTopVer) {
Map<? extends K, EntryProcessor<K, V, Object>> invokeMap =
- Collections.singletonMap(key, (EntryProcessor<K, V, Object>) entryProcessor);
+ Collections.singletonMap(key, (EntryProcessor<K, V, Object>)entryProcessor);
return tx.invokeAsync(ctx, readyTopVer, invokeMap, args);
}
@@ -2578,7 +2583,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
}
});
- if(statsEnabled)
+ if (statsEnabled)
fut.listen(new UpdatePutTimeStatClosure<V>(metrics0(), start));
return fut;
@@ -2912,11 +2917,11 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
assert ctx.isLocal();
for (Iterator<KeyCacheObject> it = ctx.swap().offHeapKeyIterator(true, true, AffinityTopologyVersion.NONE);
- it.hasNext(); )
+ it.hasNext(); )
remove((K)it.next());
for (Iterator<KeyCacheObject> it = ctx.swap().swapKeyIterator(true, true, AffinityTopologyVersion.NONE);
- it.hasNext(); )
+ it.hasNext(); )
remove((K)it.next());
removeAll(keySet());
@@ -3436,7 +3441,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
if (ttl == CU.TTL_ZERO)
return;
- loadEntry(key, val, ver0, (IgniteBiPredicate<Object, Object>) p, topVer, replicate, ttl);
+ loadEntry(key, val, ver0, (IgniteBiPredicate<Object, Object>)p, topVer, replicate, ttl);
}
}, args);
}
@@ -3490,8 +3495,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/** {@inheritDoc} */
@Override public IgniteInternalFuture<?> localLoadCacheAsync(final IgniteBiPredicate<K, V> p,
- final Object[] args)
- {
+ final Object[] args) {
return ctx.closures().callLocalSafe(
ctx.projectSafe(new Callable<Object>() {
@Nullable @Override public Object call() throws IgniteCheckedException {
@@ -3596,8 +3600,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
public void localLoad(Collection<? extends K> keys,
@Nullable ExpiryPolicy plc)
- throws IgniteCheckedException
- {
+ throws IgniteCheckedException {
final boolean replicate = ctx.isDrEnabled();
final AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion();
@@ -4574,14 +4577,14 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @return Entry set.
*/
public Set<Cache.Entry<K, V>> entrySet(@Nullable CacheEntryPredicate... filter) {
- return map.entries(filter);
+ return entrySetx(filter);
}
/**
* @return Primary entry set.
*/
public Set<Cache.Entry<K, V>> primaryEntrySet() {
- return map.entries(CU.cachePrimary(ctx.grid().affinity(ctx.name()), ctx.localNode()));
+ return new EntrySet(map.entrySet(CU.cachePrimary(ctx.grid().affinity(ctx.name()), ctx.localNode())));
}
/**
@@ -4612,15 +4615,23 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
String taskName,
boolean deserializeBinary,
boolean needVer) throws IgniteCheckedException {
- return getAsync(key,
- !ctx.config().isReadFromBackup(),
- /*skip tx*/false,
- null,
- taskName,
- deserializeBinary,
- false,
- /*can remap*/true,
- needVer).get();
+ try {
+ return getAsync(key,
+ !ctx.config().isReadFromBackup(),
+ /*skip tx*/false,
+ null,
+ taskName,
+ deserializeBinary,
+ false,
+ /*can remap*/true,
+ needVer).get();
+ }
+ catch (IgniteException e) {
+ if (e.getCause(IgniteCheckedException.class) != null)
+ throw e.getCause(IgniteCheckedException.class);
+ else
+ throw e;
+ }
}
/**
@@ -4741,7 +4752,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param deserializeBinary Deserialize binary flag.
* @return Public API iterator.
*/
- protected Iterator<Cache.Entry<K, V>> iterator(final Iterator<GridCacheEntryEx> it,
+ protected Iterator<Cache.Entry<K, V>> iterator(final Iterator<? extends GridCacheEntryEx> it,
final boolean deserializeBinary) {
return new Iterator<Cache.Entry<K, V>>() {
{
@@ -4807,8 +4818,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
*/
@Nullable private Cache.Entry<K, V> toCacheEntry(GridCacheEntryEx entry,
boolean deserializeBinary)
- throws IgniteCheckedException, GridCacheEntryRemovedException
- {
+ throws IgniteCheckedException, GridCacheEntryRemovedException {
CacheObject val = entry.innerGet(
null,
null,
@@ -5497,7 +5507,8 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param loadArgs Arguments.
* @param plc Policy.
*/
- private LoadCacheJob(String cacheName, AffinityTopologyVersion topVer, IgniteBiPredicate<K, V> p, Object[] loadArgs,
+ private LoadCacheJob(String cacheName, AffinityTopologyVersion topVer, IgniteBiPredicate<K, V> p,
+ Object[] loadArgs,
ExpiryPolicy plc) {
super(cacheName, topVer);
@@ -5710,7 +5721,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/**
*
*/
- static class LoadKeysCallable<K, V> implements IgniteCallable<Void>, Externalizable{
+ static class LoadKeysCallable<K, V> implements IgniteCallable<Void>, Externalizable {
/** */
private static final long serialVersionUID = 0L;
@@ -5831,8 +5842,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/** {@inheritDoc} */
@Override public void applyx(KeyCacheObject key, Object val, GridCacheVersion ver)
- throws IgniteCheckedException
- {
+ throws IgniteCheckedException {
assert ver != null;
if (p != null && !p.apply(key.<K>value(ctx.cacheObjectContext(), false), (V)val))
@@ -5914,8 +5924,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
private LoadCacheClosure(String cacheName,
IgniteBiPredicate<K, V> p,
Object[] args,
- @Nullable ExpiryPolicy plc)
- {
+ @Nullable ExpiryPolicy plc) {
this.cacheName = cacheName;
this.p = p;
this.args = args;
@@ -6380,4 +6389,173 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
return null;
}
}
+
+ /**
+ * Iterator implementation for KeySet.
+ */
+ private final class KeySetIterator implements Iterator<K> {
+ /** Internal map entry iterator. */
+ private final Iterator<GridCacheMapEntry> internalIterator;
+
+ /** Keep binary flag. */
+ private final boolean keepBinary;
+
+ /** Current entry. */
+ private GridCacheMapEntry current;
+
+ /**
+ * Constructor.
+ * @param internalIterator Internal iterator.
+ * @param keepBinary Keep binary flag.
+ */
+ private KeySetIterator(Iterator<GridCacheMapEntry> internalIterator, boolean keepBinary) {
+ this.internalIterator = internalIterator;
+ this.keepBinary = keepBinary;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ return internalIterator.hasNext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public K next() {
+ current = internalIterator.next();
+
+ return (K)ctx.unwrapBinaryIfNeeded(current.key(), keepBinary, true);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void remove() {
+ if (current == null)
+ throw new IllegalStateException();
+
+ try {
+ GridCacheAdapter.this.getAndRemove((K)current.key());
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+
+ current = null;
+ }
+ }
+
+ /**
+ * A wrapper over internal map that provides set semantics and constant-time contains() check.
+ */
+ private final class KeySet extends AbstractSet<K> {
+ /** Internal entry set. */
+ private final Set<GridCacheMapEntry> internalSet;
+
+ /** Keep binary flag. */
+ private final boolean keepBinary;
+
+ /**
+ * Constructor
+ * @param internalSet Internal set.
+ */
+ private KeySet(Set<GridCacheMapEntry> internalSet) {
+ this.internalSet = internalSet;
+
+ CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+ keepBinary = opCtx != null && opCtx.isKeepBinary();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterator<K> iterator() {
+ return new KeySetIterator(internalSet.iterator(), keepBinary);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ return F.size(iterator());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean contains(Object o) {
+ GridCacheMapEntry entry = map.getEntry(ctx.toCacheKeyObject(o));
+
+ return entry != null && internalSet.contains(entry);
+ }
+ }
+
+ /**
+ * Iterator implementation for EntrySet.
+ */
+ private final class EntryIterator implements Iterator<Cache.Entry<K, V>> {
+
+ /** Internal iterator. */
+ private final Iterator<GridCacheMapEntry> internalIterator;
+
+ /** Current entry. */
+ private GridCacheMapEntry current;
+
+ /**
+ * Constructor.
+ * @param internalIterator Internal iterator.
+ */
+ private EntryIterator(Iterator<GridCacheMapEntry> internalIterator) {
+ this.internalIterator = internalIterator;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ return internalIterator.hasNext();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Cache.Entry<K, V> next() {
+ current = internalIterator.next();
+
+ return current.wrapLazyValue();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void remove() {
+ if (current == null)
+ throw new IllegalStateException();
+
+ try {
+ GridCacheAdapter.this.getAndRemove((K)current.wrapLazyValue().getKey());
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+
+ current = null;
+ }
+ }
+
+ /**
+ * A wrapper over internal map that provides set semantics and constant-time contains() check.
+ */
+ private final class EntrySet extends AbstractSet<Cache.Entry<K, V>> {
+
+ /** Internal set. */
+ private final Set<GridCacheMapEntry> internalSet;
+
+ /** Constructor. */
+ private EntrySet(Set<GridCacheMapEntry> internalSet) {
+ this.internalSet = internalSet;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterator<Cache.Entry<K, V>> iterator() {
+ return new EntryIterator(internalSet.iterator());
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ return F.size(iterator());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean contains(Object o) {
+ GridCacheMapEntry entry = map.getEntry(ctx.toCacheKeyObject(o));
+
+ return entry != null && internalSet.contains(entry);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index f1767e0..5e843dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -197,10 +197,18 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter {
public int partition(Object key) {
GridAffinityAssignmentCache aff0 = aff;
+ if (key instanceof KeyCacheObject && ((KeyCacheObject)key).partition() != -1)
+ return ((KeyCacheObject)key).partition();
+
if (aff0 == null)
throw new IgniteException(FAILED_TO_FIND_CACHE_ERR_MSG + cctx.name());
- return affFunction.partition(affinityKey(key));
+ int p = affFunction.partition(affinityKey(key));
+
+ if (key instanceof KeyCacheObject)
+ ((KeyCacheObject)key).partition(p);
+
+ return p;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheClearAllRunnable.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheClearAllRunnable.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheClearAllRunnable.java
index ffce82d..4f97e7b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheClearAllRunnable.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheClearAllRunnable.java
@@ -79,7 +79,7 @@ public class GridCacheClearAllRunnable<K, V> implements Runnable {
/** {@inheritDoc} */
@Override public void run() {
- Iterator<GridCacheEntryEx> iter = cache.map().stripedEntryIterator(id, totalCnt);
+ Iterator<? extends GridCacheEntryEx> iter = cache.entries().iterator();
while (iter.hasNext())
clearEntry(iter.next());
[6/6] ignite git commit: Merge branch 'master' into ignite-1786
Posted by vo...@apache.org.
Merge branch 'master' into ignite-1786
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/307c2178
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/307c2178
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/307c2178
Branch: refs/heads/ignite-1786
Commit: 307c217858afc95739a09aea2e6d32e5d4788477
Parents: 7f0737d 3be3d16
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Wed Apr 27 11:09:13 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Wed Apr 27 11:09:13 2016 +0300
----------------------------------------------------------------------
.../internal/benchmarks/model/IntValue.java | 19 +-
.../internal/binary/BinaryObjectImpl.java | 14 +
.../processors/cache/GridCacheAdapter.java | 448 ++--
.../cache/GridCacheAffinityManager.java | 10 +-
.../cache/GridCacheClearAllRunnable.java | 2 +-
.../cache/GridCacheConcurrentMap.java | 1996 +-----------------
.../cache/GridCacheConcurrentMapImpl.java | 344 +++
.../processors/cache/GridCacheContext.java | 56 +-
.../cache/GridCacheDeploymentManager.java | 2 +-
.../processors/cache/GridCacheEntrySet.java | 113 -
.../cache/GridCacheEvictionManager.java | 2 +-
.../processors/cache/GridCacheKeySet.java | 104 -
.../processors/cache/GridCacheMapEntry.java | 22 +-
.../processors/cache/GridCacheProcessor.java | 1 -
.../processors/cache/GridCacheProxyImpl.java | 2 +-
.../processors/cache/GridCacheUtils.java | 1 -
.../processors/cache/GridNoStorageCacheMap.java | 107 +
.../processors/cache/IgniteInternalCache.java | 2 +-
.../processors/cache/KeyCacheObject.java | 11 +
.../processors/cache/KeyCacheObjectImpl.java | 32 +-
.../binary/CacheObjectBinaryProcessorImpl.java | 24 +-
.../GridDistributedCacheAdapter.java | 4 +-
.../distributed/GridDistributedLockRequest.java | 19 +-
.../GridDistributedTxRemoteAdapter.java | 1 -
.../GridDistributedUnlockRequest.java | 16 +-
.../dht/GridCachePartitionedConcurrentMap.java | 191 ++
.../dht/GridClientPartitionTopology.java | 7 -
.../distributed/dht/GridDhtCacheAdapter.java | 77 +-
.../distributed/dht/GridDhtCacheEntry.java | 25 +-
.../distributed/dht/GridDhtLocalPartition.java | 152 +-
.../distributed/dht/GridDhtLockRequest.java | 16 +-
.../dht/GridDhtPartitionTopology.java | 7 -
.../dht/GridDhtPartitionTopologyImpl.java | 281 ++-
.../distributed/dht/GridDhtUnlockRequest.java | 15 +-
.../distributed/dht/GridNoStorageCacheMap.java | 122 --
.../dht/GridPartitionedGetFuture.java | 2 +-
.../dht/GridPartitionedSingleGetFuture.java | 2 +-
.../dht/atomic/GridDhtAtomicCache.java | 75 +-
.../dht/atomic/GridDhtAtomicUpdateRequest.java | 22 +-
.../dht/atomic/GridNearAtomicUpdateFuture.java | 6 +-
.../dht/atomic/GridNearAtomicUpdateRequest.java | 31 +-
.../dht/colocated/GridDhtColocatedCache.java | 13 +-
.../dht/preloader/GridDhtPartitionDemander.java | 2 +-
.../dht/preloader/GridDhtPartitionSupplier.java | 8 +-
.../distributed/near/GridNearCacheAdapter.java | 94 +-
.../distributed/near/GridNearGetFuture.java | 2 +-
.../distributed/near/GridNearGetRequest.java | 45 +-
.../distributed/near/GridNearLockRequest.java | 16 +-
.../near/GridNearSingleGetRequest.java | 22 +-
.../distributed/near/GridNearUnlockRequest.java | 21 +-
.../processors/cache/local/GridLocalCache.java | 7 +-
.../local/atomic/GridLocalAtomicCache.java | 24 +-
.../cache/transactions/IgniteInternalTx.java | 1 -
.../cache/transactions/IgniteTxAdapter.java | 1 -
.../cache/transactions/IgniteTxEntry.java | 33 +-
.../cacheobject/IgniteCacheObjectProcessor.java | 10 +
.../IgniteCacheObjectProcessorImpl.java | 36 +-
.../GridCacheAtomicReferenceImpl.java | 2 +-
.../internal/util/PartitionedReadOnlySet.java | 71 +
.../util/StripedCompositeReadWriteLock.java | 10 +
.../ignite/internal/util/lang/GridFunc.java | 3 +-
.../ignite/internal/visor/cache/VisorCache.java | 6 +-
.../GridCachePreloadingEvictionsSelfTest.java | 4 +-
.../cache/GridCacheTtlManagerSelfTest.java | 2 +-
...idCacheValueConsistencyAbstractSelfTest.java | 3 +-
.../IgniteDynamicClientCacheStartSelfTest.java | 10 -
.../IgniteTxStoreExceptionAbstractSelfTest.java | 8 +-
.../GridCacheBinaryObjectsAbstractSelfTest.java | 2 +-
...actQueueFailoverDataConsistencySelfTest.java | 2 +-
.../GridCacheQueueCleanupSelfTest.java | 4 +-
.../GridCacheSequenceApiSelfAbstractTest.java | 37 -
.../GridCacheSetAbstractSelfTest.java | 5 +-
.../GridCacheSetFailoverAbstractSelfTest.java | 8 +-
.../IgniteDataStructureUniqueNameTest.java | 3 +-
.../IgnitePartitionedQueueNoBackupsTest.java | 6 +-
.../IgnitePartitionedSetNoBackupsSelfTest.java | 6 +-
.../distributed/dht/GridCacheDhtTestUtils.java | 9 +-
.../near/GridCacheNearOneNodeSelfTest.java | 4 +-
.../processors/igfs/IgfsAbstractSelfTest.java | 10 +-
79 files changed, 1958 insertions(+), 2975 deletions(-)
----------------------------------------------------------------------
[2/6] ignite git commit: IGNITE-2948 - Optimize usage of
GridCacheConcurrentMap
Posted by vo...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridNoStorageCacheMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridNoStorageCacheMap.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridNoStorageCacheMap.java
deleted file mode 100644
index 274701f..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridNoStorageCacheMap.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.distributed.dht;
-
-import java.util.Map;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheObject;
-import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMap;
-import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
-import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
-import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.util.lang.GridTriple;
-import org.apache.ignite.internal.util.typedef.internal.S;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Empty cache map that will never store any entries.
- */
-public class GridNoStorageCacheMap extends GridCacheConcurrentMap {
- /** Empty triple. */
- private final GridTriple<GridCacheMapEntry> emptyTriple =
- new GridTriple<>(null, null, null);
-
- /**
- * @param ctx Cache context.
- */
- public GridNoStorageCacheMap(GridCacheContext ctx) {
- super(ctx, 0, null, 0.75f, 1);
- }
-
- /** {@inheritDoc} */
- @Override public boolean isEmpty() {
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return 0;
- }
-
- /** {@inheritDoc} */
- @Override public int publicSize() {
- return 0;
- }
-
- /** {@inheritDoc} */
- @Override public boolean containsKey(Object key) {
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public GridCacheMapEntry randomEntry() {
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public GridCacheMapEntry getEntry(Object key) {
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public GridCacheMapEntry putEntry(AffinityTopologyVersion topVer,
- KeyCacheObject key,
- @Nullable CacheObject val)
- {
- throw new AssertionError();
- }
-
- /** {@inheritDoc} */
- @Override public GridTriple<GridCacheMapEntry> putEntryIfObsoleteOrAbsent(
- AffinityTopologyVersion topVer,
- KeyCacheObject key,
- @Nullable CacheObject val,
- boolean create)
- {
- if (create) {
- GridCacheMapEntry entry = ctx.useOffheapEntry() ?
- new GridDhtOffHeapCacheEntry(ctx, topVer, key, hash(key.hashCode()), val) :
- new GridDhtCacheEntry(ctx, topVer, key, hash(key.hashCode()), val);
-
- return new GridTriple<>(entry, null, null);
- }
- else
- return emptyTriple;
- }
-
- /** {@inheritDoc} */
- @Override public void putAll(Map<KeyCacheObject, CacheObject> m) {
- throw new AssertionError();
- }
-
- /** {@inheritDoc} */
- @Override public boolean removeEntry(GridCacheEntryEx e) {
- throw new AssertionError();
- }
-
- /** {@inheritDoc} */
- @Override public GridCacheMapEntry removeEntryIfObsolete(KeyCacheObject key) {
- throw new AssertionError();
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(GridNoStorageCacheMap.class, this);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/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 e9873b7..9561ad8 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
@@ -490,7 +490,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
// Entry was not in memory or in swap, so we remove it from cache.
if (v == null) {
if (isNew && entry.markObsoleteIfEmpty(ver))
- cache.removeIfObsolete(key);
+ cache.removeEntry(entry);
}
else {
cctx.addResult(locVals,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 62c287a..fd59f48 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -414,7 +414,7 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im
// Entry was not in memory or in swap, so we remove it from cache.
if (v == null) {
if (isNew && entry.markObsoleteIfEmpty(ver))
- colocated.removeIfObsolete(key);
+ colocated.removeEntry(entry);
}
else {
if (!skipVals && cctx.config().isStatisticsEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index d28aaaa..ee909e6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -175,9 +175,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
/** {@inheritDoc} */
- @Override protected void init() {
- map.setEntryFactory(new GridCacheMapEntryFactory() {
- /** {@inheritDoc} */
+ @Override protected GridCacheMapEntryFactory entryFactory() {
+ return new GridCacheMapEntryFactory() {
@Override public GridCacheMapEntry create(
GridCacheContext ctx,
AffinityTopologyVersion topVer,
@@ -190,7 +189,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
return new GridDhtAtomicCacheEntry(ctx, topVer, key, hash, val);
}
- });
+ };
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void init() {
+ super.init();
updateReplyClos = new CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse>() {
@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
@@ -330,16 +334,24 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final boolean skipStore = opCtx != null && opCtx.skipStore();
- return getAsync0(ctx.toCacheKeyObject(key),
- !ctx.config().isReadFromBackup(),
- subjId,
- taskName,
- deserializeBinary,
- expiryPlc,
- false,
- skipStore,
- true,
- needVer).get();
+ try {
+ return getAsync0(ctx.toCacheKeyObject(key, true),
+ !ctx.config().isReadFromBackup(),
+ subjId,
+ taskName,
+ deserializeBinary,
+ expiryPlc,
+ false,
+ skipStore,
+ true,
+ needVer).get();
+ }
+ catch (IgniteException e) {
+ if (e.getCause(IgniteCheckedException.class) != null)
+ throw e.getCause(IgniteCheckedException.class);
+ else
+ throw e;
+ }
}
/** {@inheritDoc} */
@@ -369,7 +381,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
return asyncOp(new CO<IgniteInternalFuture<V>>() {
@Override public IgniteInternalFuture<V> apply() {
- return getAsync0(ctx.toCacheKeyObject(key),
+ return getAsync0(ctx.toCacheKeyObject(key, true),
forcePrimary,
subjId0,
taskName,
@@ -415,7 +427,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
return asyncOp(new CO<IgniteInternalFuture<Map<K, V>>>() {
@Override public IgniteInternalFuture<Map<K, V>> apply() {
- return getAllAsync0(ctx.cacheKeysView(keys),
+ return getAllAsync0(ctx.cacheKeysView(keys, true),
forcePrimary,
subjId0,
taskName,
@@ -808,13 +820,16 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
true,
TRANSFORM);
- return resFut.chain(new CX1<IgniteInternalFuture<Map<K, EntryProcessorResult<T>>>, Map<K, EntryProcessorResult<T>>>() {
- @Override public Map<K, EntryProcessorResult<T>> applyx(IgniteInternalFuture<Map<K, EntryProcessorResult<T>>> fut) throws IgniteCheckedException {
- Map<Object, EntryProcessorResult> resMap = (Map)fut.get();
+ return resFut.chain(
+ new CX1<IgniteInternalFuture<Map<K, EntryProcessorResult<T>>>, Map<K, EntryProcessorResult<T>>>() {
+ @Override public Map<K, EntryProcessorResult<T>> applyx(
+ IgniteInternalFuture<Map<K, EntryProcessorResult<T>>> fut
+ ) throws IgniteCheckedException {
+ Map<Object, EntryProcessorResult> resMap = (Map)fut.get();
- return ctx.unwrapInvokeResult(resMap, keepBinary);
- }
- });
+ return ctx.unwrapInvokeResult(resMap, keepBinary);
+ }
+ });
}
/** {@inheritDoc} */
@@ -1372,7 +1387,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
GridCacheVersion obsoleteVer = context().versions().next();
if (isNew && entry.markObsoleteIfEmpty(obsoleteVer))
- removeIfObsolete(key);
+ removeEntry(entry);
success = false;
}
@@ -1952,9 +1967,15 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
null,
req.keepBinary());
- Object val = ctx.config().getInterceptor().onBeforePut(new CacheLazyEntry(ctx, entry.key(),
- old, req.keepBinary()),
- updated.value(ctx.cacheObjectContext(), false));
+ Object val = ctx.config().getInterceptor().onBeforePut(
+ new CacheLazyEntry(
+ ctx,
+ entry.key(),
+ old,
+ req.keepBinary()),
+ updated.value(
+ ctx.cacheObjectContext(),
+ false));
if (val == null)
continue;
@@ -3329,7 +3350,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
respVers.add(ver);
- if (respVers.sizex() > DEFERRED_UPDATE_RESPONSE_BUFFER_SIZE && guard.compareAndSet(false, true))
+ if (respVers.sizex() > DEFERRED_UPDATE_RESPONSE_BUFFER_SIZE && guard.compareAndSet(false, true))
snd = true;
}
finally {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index b5e2835..6b050b1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -681,6 +681,13 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
if (nearEntryProcessors == null)
nearEntryProcessors = unmarshalCollection(nearEntryProcessorsBytes, ctx, ldr);
}
+
+ if (partIds != null && !partIds.isEmpty()) {
+ assert partIds.size() == keys.size();
+
+ for (int i = 0; i < keys.size(); i++)
+ keys.get(i).partition(partIds.get(i));
+ }
}
/** {@inheritDoc} */
@@ -835,6 +842,12 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
writer.incrementState();
+ case 25:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -1031,6 +1044,13 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
reader.incrementState();
+ case 25:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
}
return reader.afterMessageRead(GridDhtAtomicUpdateRequest.class);
@@ -1056,7 +1076,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements Grid
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 25;
+ return 26;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
index edebd8c..dfdf1a5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
@@ -854,7 +854,7 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
if (val == null && op != GridCacheOperation.DELETE)
continue;
- KeyCacheObject cacheKey = cctx.toCacheKeyObject(key);
+ KeyCacheObject cacheKey = cctx.toCacheKeyObject(key, true);
if (remapKeys != null && !remapKeys.contains(cacheKey))
continue;
@@ -969,12 +969,12 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
if (val == null && op != GridCacheOperation.DELETE)
throw new NullPointerException("Null value.");
- KeyCacheObject cacheKey = cctx.toCacheKeyObject(key);
+ KeyCacheObject cacheKey = cctx.toCacheKeyObject(key, true);
if (op != TRANSFORM)
val = cctx.toCacheObject(val);
- ClusterNode primary = cctx.affinity().primary(cacheKey, topVer);
+ ClusterNode primary = cctx.affinity().primary(cacheKey.partition(), topVer);
if (primary == null)
throw new ClusterTopologyServerNotFoundException("Failed to map keys for cache (all partition nodes " +
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
index 375c02f..e847c7c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateRequest.java
@@ -99,6 +99,10 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
@GridDirectCollection(CacheObject.class)
private List<CacheObject> vals;
+ /** Partitions of keys. */
+ @GridDirectCollection(int.class)
+ private List<Integer> partIds;
+
/** Entry processors. */
@GridDirectTransient
private List<EntryProcessor<Object, Object, Object>> entryProcessors;
@@ -246,6 +250,8 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
initSize = Math.min(maxEntryCnt, 10);
keys = new ArrayList<>(initSize);
+
+ partIds = new ArrayList<>(initSize);
}
/** {@inheritDoc} */
@@ -384,12 +390,13 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
if (op == TRANSFORM) {
assert val instanceof EntryProcessor : val;
- entryProcessor = (EntryProcessor<Object, Object, Object>) val;
+ entryProcessor = (EntryProcessor<Object, Object, Object>)val;
}
assert val != null || op == DELETE;
keys.add(key);
+ partIds.add(key.partition());
if (entryProcessor != null) {
if (entryProcessors == null)
@@ -652,6 +659,13 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
if (expiryPlcBytes != null && expiryPlc == null)
expiryPlc = ctx.marshaller().unmarshal(expiryPlcBytes, U.resolveClassLoader(ldr, ctx.gridConfig()));
+
+ if (partIds != null && !partIds.isEmpty()) {
+ assert partIds.size() == keys.size();
+
+ for (int i = 0; i < keys.size(); i++)
+ keys.get(i).partition(partIds.get(i));
+ }
}
/** {@inheritDoc} */
@@ -812,6 +826,11 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
writer.incrementState();
+ case 26:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
}
return true;
@@ -1020,6 +1039,14 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
+ case 26:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridNearAtomicUpdateRequest.class);
@@ -1048,7 +1075,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 26;
+ return 27;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/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 9c699fa..0ae434a 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
@@ -116,9 +116,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
}
/** {@inheritDoc} */
- @Override protected void init() {
- map.setEntryFactory(new GridCacheMapEntryFactory() {
- /** {@inheritDoc} */
+ @Override protected GridCacheMapEntryFactory entryFactory() {
+ return new GridCacheMapEntryFactory() {
@Override public GridCacheMapEntry create(
GridCacheContext ctx,
AffinityTopologyVersion topVer,
@@ -131,7 +130,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
return new GridDhtColocatedCacheEntry(ctx, topVer, key, hash, val);
}
- });
+ };
}
/** {@inheritDoc} */
@@ -245,8 +244,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
}
AffinityTopologyVersion topVer = tx == null ?
- (canRemap ? ctx.affinity().affinityTopologyVersion() : ctx.shared().exchange().readyAffinityVersion()) :
- tx.topologyVersion();
+ (canRemap ? ctx.affinity().affinityTopologyVersion() : ctx.shared().exchange().readyAffinityVersion()) :
+ tx.topologyVersion();
subjId = ctx.subjectIdPerCall(subjId, opCtx);
@@ -526,7 +525,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
GridCacheVersion obsoleteVer = context().versions().next();
if (isNew && entry.markObsoleteIfEmpty(obsoleteVer))
- removeIfObsolete(key);
+ removeEntry(entry);
success = false;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
index dc225cf..821d2e4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
@@ -675,7 +675,7 @@ public class GridDhtPartitionDemander {
"value, will ignore rebalance entries)");
if (cached.markObsoleteIfEmpty(null))
- cached.context().cache().removeIfObsolete(cached.key());
+ cached.context().cache().removeEntry(cached);
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
index d301ba9..b082c47 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionSupplier.java
@@ -32,8 +32,8 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfoCollectSwapListener;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.internal.processors.cache.GridCacheSwapEntry;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
import org.apache.ignite.internal.util.lang.GridCloseableIterator;
@@ -298,8 +298,8 @@ class GridDhtPartitionSupplier {
phase = SupplyContextPhase.ONHEAP;
if (phase == SupplyContextPhase.ONHEAP) {
- Iterator<GridDhtCacheEntry> entIt = sctx != null ?
- (Iterator<GridDhtCacheEntry>)sctx.entryIt : loc.entries().iterator();
+ Iterator<GridCacheMapEntry> entIt = sctx != null ?
+ (Iterator<GridCacheMapEntry>)sctx.entryIt : loc.allEntries().iterator();
while (entIt.hasNext()) {
if (!cctx.affinity().belongs(node, part, d.topologyVersion())) {
@@ -807,7 +807,7 @@ class GridDhtPartitionSupplier {
boolean partMissing = false;
- for (GridCacheEntryEx e : loc.entries()) {
+ for (GridCacheEntryEx e : loc.allEntries()) {
if (!cctx.affinity().belongs(node, part, d.topologyVersion())) {
// Demander no longer needs this partition, so we send '-1' partition and move on.
s.missed(part);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
index 4b876b4..7971173 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheAdapter.java
@@ -39,12 +39,10 @@ import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheEntryPredicateAdapter;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheClearAllRunnable;
-import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMap;
+import org.apache.ignite.internal.processors.cache.GridCacheConcurrentMapImpl;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
-import org.apache.ignite.internal.processors.cache.GridCacheEntrySet;
-import org.apache.ignite.internal.processors.cache.GridCacheKeySet;
import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.internal.processors.cache.GridCacheMapEntryFactory;
import org.apache.ignite.internal.processors.cache.GridCachePreloader;
@@ -54,16 +52,14 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.CacheGetFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
-import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.P1;
import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.jetbrains.annotations.NotNull;
@@ -94,9 +90,8 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
}
/** {@inheritDoc} */
- @Override protected void init() {
- map.setEntryFactory(new GridCacheMapEntryFactory() {
- /** {@inheritDoc} */
+ @Override protected GridCacheMapEntryFactory entryFactory() {
+ return new GridCacheMapEntryFactory() {
@Override public GridCacheMapEntry create(
GridCacheContext ctx,
AffinityTopologyVersion topVer,
@@ -111,7 +106,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
return new GridNearCacheEntry(ctx, key, hash, val);
}
- });
+ };
}
/**
@@ -121,10 +116,10 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
/** {@inheritDoc} */
@Override public void onReconnected() {
- map = new GridCacheConcurrentMap(
+ map = new GridCacheConcurrentMapImpl(
ctx,
- ctx.config().getNearConfiguration().getNearStartSize(),
- map.getEntryFactory());
+ entryFactory(),
+ ctx.config().getNearConfiguration().getNearStartSize());
}
/** {@inheritDoc} */
@@ -138,7 +133,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
}
/** {@inheritDoc} */
- @Override public GridCacheEntryEx entryEx(KeyCacheObject key, boolean touch) {
+ @Override public GridCacheMapEntry entryEx(KeyCacheObject key, boolean touch) {
GridNearCacheEntry entry = null;
while (true) {
@@ -157,7 +152,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
}
/** {@inheritDoc} */
- @Override public GridCacheEntryEx entryEx(KeyCacheObject key, AffinityTopologyVersion topVer) {
+ @Override public GridCacheMapEntry entryEx(KeyCacheObject key, AffinityTopologyVersion topVer) {
GridNearCacheEntry entry = null;
while (true) {
@@ -311,7 +306,7 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
/** {@inheritDoc} */
@Override public long sizeLong() {
- return nearEntries().size() + dht().sizeLong();
+ return nearEntries().size() + dht().size();
}
/** {@inheritDoc} */
@@ -339,15 +334,23 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
@Override public boolean apply(GridCacheEntryEx entry) {
GridNearCacheEntry nearEntry = (GridNearCacheEntry)entry;
- return nearEntry.valid(topVer);
+ return !nearEntry.deleted() && nearEntry.visitable(CU.empty0()) && nearEntry.valid(topVer);
}
});
}
/** {@inheritDoc} */
- @Override public Set<Cache.Entry<K, V>> entrySet(
- @Nullable CacheEntryPredicate... filter) {
- return new EntrySet(super.entrySet(filter), dht().entrySet(filter));
+ @Override public Set<Cache.Entry<K, V>> entrySet(@Nullable final CacheEntryPredicate... filter) {
+ CacheEntryPredicate p = new CacheEntryPredicateAdapter() {
+ @Override public boolean apply(GridCacheEntryEx ex) {
+ if (ex instanceof GridCacheMapEntry)
+ return ((GridCacheMapEntry)ex).visitable(filter);
+ else
+ return !ex.deleted() && F.isAll(ex, filter);
+ }
+ };
+
+ return new EntrySet(super.entrySet(p), dht().entrySet(p));
}
/** {@inheritDoc} */
@@ -355,49 +358,11 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
return dht().entrySet(part);
}
- /** {@inheritDoc} */
- @Override public Set<Cache.Entry<K, V>> primaryEntrySet() {
- final AffinityTopologyVersion topVer = ctx.affinity().affinityTopologyVersion();
-
- Collection<Cache.Entry<K, V>> entries =
- F.flatCollections(
- F.viewReadOnly(
- dht().topology().currentLocalPartitions(),
- new C1<GridDhtLocalPartition, Collection<Cache.Entry<K, V>>>() {
- @Override public Collection<Cache.Entry<K, V>> apply(GridDhtLocalPartition p) {
- Collection<GridDhtCacheEntry> entries0 = p.entries();
-
- return F.viewReadOnly(
- entries0,
- new C1<GridDhtCacheEntry, Cache.Entry<K, V>>() {
- @Override public Cache.Entry<K, V> apply(GridDhtCacheEntry e) {
- return e.wrapLazyValue();
- }
- },
- new P1<GridDhtCacheEntry>() {
- @Override public boolean apply(GridDhtCacheEntry e) {
- return !e.obsoleteOrDeleted();
- }
- });
- }
- },
- new P1<GridDhtLocalPartition>() {
- @Override public boolean apply(GridDhtLocalPartition p) {
- return p.primary(topVer);
- }
- }));
-
- return new GridCacheEntrySet<>(ctx, entries, null);
- }
-
- /** {@inheritDoc} */
- @Override public Set<K> keySet() {
- return new GridCacheKeySet<>(ctx, entrySet(), null);
- }
-
- /** {@inheritDoc} */
- @Override public Set<K> primaryKeySet() {
- return new GridCacheKeySet<>(ctx, primaryEntrySet(), null);
+ /**
+ * @return Keys for near cache only.
+ */
+ public Set<K> nearKeySet() {
+ return super.keySet();
}
/** {@inheritDoc} */
@@ -500,7 +465,8 @@ public abstract class GridNearCacheAdapter<K, V> extends GridDistributedCacheAda
}
/** {@inheritDoc} */
- @Override public List<GridCacheClearAllRunnable<K, V>> splitClearLocally(boolean srv, boolean near, boolean readers) {
+ @Override public List<GridCacheClearAllRunnable<K, V>> splitClearLocally(boolean srv, boolean near,
+ boolean readers) {
assert configuration().getNearConfiguration() != null;
if (ctx.affinityNode()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 1b30afb..7d29381 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -619,7 +619,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
// Entry was not in memory or in swap, so we remove it from cache.
if (v == null && isNew && dhtEntry.markObsoleteIfEmpty(ver))
- dht.removeIfObsolete(key);
+ dht.removeEntry(dhtEntry);
}
if (v != null) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
index 0c2451e..fa7f367 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
@@ -19,9 +19,11 @@ package org.apache.ignite.internal.processors.cache.distributed.near;
import java.io.Externalizable;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
@@ -69,7 +71,11 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
/** */
@GridDirectCollection(KeyCacheObject.class)
- private Collection<KeyCacheObject> keys;
+ private List<KeyCacheObject> keys;
+
+ /** Partition IDs. */
+ @GridDirectCollection(int.class)
+ private List<Integer> partIds;
/** */
@GridDirectCollection(boolean.class)
@@ -140,8 +146,17 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
this.futId = futId;
this.miniId = miniId;
this.ver = ver;
- this.keys = keys.keySet();
- this.flags = keys.values();
+
+ this.keys = new ArrayList<>(keys.size());
+ flags = new ArrayList<>(keys.size());
+ partIds = new ArrayList<>(keys.size());
+
+ for (Map.Entry<KeyCacheObject, Boolean> entry : keys.entrySet()) {
+ this.keys.add(entry.getKey());
+ flags.add(entry.getValue());
+ partIds.add(entry.getKey().partition());
+ }
+
this.readThrough = readThrough;
this.topVer = topVer;
this.subjId = subjId;
@@ -269,6 +284,13 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
while (keysIt.hasNext())
keyMap.put(keysIt.next(), flagsIt.next());
}
+
+ if (partIds != null && !partIds.isEmpty()) {
+ assert partIds.size() == keys.size();
+
+ for (int i = 0; i < keys.size(); i++)
+ keys.get(i).partition(partIds.get(i));
+ }
}
/** {@inheritDoc} */
@@ -363,6 +385,12 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
writer.incrementState();
+ case 15:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -475,6 +503,14 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
reader.incrementState();
+ case 15:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridNearGetRequest.class);
@@ -485,10 +521,9 @@ public class GridNearGetRequest extends GridCacheMessage implements GridCacheDep
return 49;
}
-
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 15;
+ return 16;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
index 2a6b0a8..2e8cd6e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
@@ -445,6 +445,12 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
writer.incrementState();
+ case 34:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -573,6 +579,14 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
reader.incrementState();
+ case 34:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridNearLockRequest.class);
@@ -585,7 +599,7 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 34;
+ return 35;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
index aa96720..62d37af 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
@@ -62,6 +62,9 @@ public class GridNearSingleGetRequest extends GridCacheMessage implements GridCa
/** */
private KeyCacheObject key;
+ /** Partition ID. */
+ private int partId = -1;
+
/** Flags. */
private byte flags;
@@ -118,6 +121,7 @@ public class GridNearSingleGetRequest extends GridCacheMessage implements GridCa
this.cacheId = cacheId;
this.futId = futId;
this.key = key;
+ this.partId = key.partition();
this.topVer = topVer;
this.subjId = subjId;
this.taskNameHash = taskNameHash;
@@ -233,6 +237,8 @@ public class GridNearSingleGetRequest extends GridCacheMessage implements GridCa
assert key != null;
+ key.partition(partId);
+
GridCacheContext cctx = ctx.cacheContext(cacheId);
key.finishUnmarshal(cctx.cacheObjectContext(), ldr);
@@ -305,6 +311,14 @@ public class GridNearSingleGetRequest extends GridCacheMessage implements GridCa
reader.incrementState();
+ case 10:
+ partId = reader.readInt("partId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridNearSingleGetRequest.class);
@@ -367,6 +381,12 @@ public class GridNearSingleGetRequest extends GridCacheMessage implements GridCa
writer.incrementState();
+ case 10:
+ if (!writer.writeInt("partId", partId))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -384,7 +404,7 @@ public class GridNearSingleGetRequest extends GridCacheMessage implements GridCa
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 10;
+ return 11;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearUnlockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearUnlockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearUnlockRequest.java
index 7652a4a..be78868 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearUnlockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearUnlockRequest.java
@@ -21,6 +21,7 @@ import java.io.Externalizable;
import java.nio.ByteBuffer;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedUnlockRequest;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
@@ -61,6 +62,14 @@ public class GridNearUnlockRequest extends GridDistributedUnlockRequest {
writer.onHeaderWritten();
}
+ switch (writer.state()) {
+ case 8:
+ if (!writer.writeCollection("partIds", partIds, MessageCollectionItemType.INT))
+ return false;
+
+ writer.incrementState();
+ }
+
return true;
}
@@ -74,6 +83,16 @@ public class GridNearUnlockRequest extends GridDistributedUnlockRequest {
if (!super.readFrom(buf, reader))
return false;
+ switch (reader.state()) {
+ case 8:
+ partIds = reader.readCollection("partIds", MessageCollectionItemType.INT);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+ }
+
return reader.afterMessageRead(GridNearUnlockRequest.class);
}
@@ -84,7 +103,7 @@ public class GridNearUnlockRequest extends GridDistributedUnlockRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 8;
+ return 9;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
index 4ce1f36..16a35d3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
@@ -80,9 +80,8 @@ public class GridLocalCache<K, V> extends GridCacheAdapter<K, V> {
}
/** {@inheritDoc} */
- @Override protected void init() {
- map.setEntryFactory(new GridCacheMapEntryFactory() {
- /** {@inheritDoc} */
+ @Override protected GridCacheMapEntryFactory entryFactory() {
+ return new GridCacheMapEntryFactory() {
@Override public GridCacheMapEntry create(
GridCacheContext ctx,
AffinityTopologyVersion topVer,
@@ -92,7 +91,7 @@ public class GridLocalCache<K, V> extends GridCacheAdapter<K, V> {
) {
return new GridLocalCacheEntry(ctx, key, hash, val);
}
- });
+ };
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 3e0e2c2..cb6152d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -45,19 +45,16 @@ import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheOperationContext;
import org.apache.ignite.internal.processors.cache.CachePartialUpdateCheckedException;
import org.apache.ignite.internal.processors.cache.CacheStorePartialUpdateException;
-import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
-import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
-import org.apache.ignite.internal.processors.cache.GridCacheMapEntryFactory;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridCachePreloader;
import org.apache.ignite.internal.processors.cache.GridCachePreloaderAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.IgniteCacheExpiryPolicy;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.local.GridLocalCacheEntry;
+import org.apache.ignite.internal.processors.cache.local.GridLocalCache;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.F0;
@@ -85,7 +82,7 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPD
/**
* Non-transactional local cache.
*/
-public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
+public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
/** */
private static final long serialVersionUID = 0L;
@@ -103,27 +100,12 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K, V> {
* @param ctx Cache context.
*/
public GridLocalAtomicCache(GridCacheContext<K, V> ctx) {
- super(ctx, ctx.config().getStartSize());
+ super(ctx);
preldr = new GridCachePreloaderAdapter(ctx);
}
/** {@inheritDoc} */
- @Override protected void init() {
- map.setEntryFactory(new GridCacheMapEntryFactory() {
- @Override public GridCacheMapEntry create(
- GridCacheContext ctx,
- AffinityTopologyVersion topVer,
- KeyCacheObject key,
- int hash,
- CacheObject val
- ) {
- return new GridLocalCacheEntry(ctx, key, hash, val);
- }
- });
- }
-
- /** {@inheritDoc} */
@Override public boolean isLocal() {
return true;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
index e08f9b0..87b8e07 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
@@ -25,7 +25,6 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index 17bed0b..b8447f0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -40,7 +40,6 @@ import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheEntryPredicate;
import org.apache.ignite.internal.processors.cache.CacheInvokeEntry;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
index e75ce91..4d8139f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
@@ -98,6 +98,9 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
/** Cache ID. */
private int cacheId;
+ /** Partition ID. */
+ private int partId = -1;
+
/** Transient tx key. */
@GridDirectTransient
private IgniteTxKey txKey;
@@ -251,11 +254,12 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
keepBinary(keepBinary);
key = entry.key();
+ partId = entry.key().partition();
cacheId = entry.context().cacheId();
}
- /**
+ /**
* This constructor is meant for local transactions.
*
* @param ctx Cache registry.
@@ -303,6 +307,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
addEntryProcessor(entryProcessor, invokeArgs);
key = entry.key();
+ partId = entry.key().partition();
cacheId = entry.context().cacheId();
}
@@ -336,6 +341,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
IgniteTxEntry cp = new IgniteTxEntry();
cp.key = key;
+ cp.partId = partId;
cp.cacheId = cacheId;
cp.ctx = ctx;
@@ -476,7 +482,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
*
* @param skipStore Skip store flag.
*/
- public void skipStore(boolean skipStore){
+ public void skipStore(boolean skipStore) {
setFlag(skipStore, SKIP_STORE_FLAG_MASK);
}
@@ -853,7 +859,8 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
* @param clsLdr Class loader.
* @throws IgniteCheckedException If un-marshalling failed.
*/
- public void unmarshal(GridCacheSharedContext<?, ?> ctx, boolean near, ClassLoader clsLdr) throws IgniteCheckedException {
+ public void unmarshal(GridCacheSharedContext<?, ?> ctx, boolean near,
+ ClassLoader clsLdr) throws IgniteCheckedException {
if (this.ctx == null) {
GridCacheContext<?, ?> cacheCtx = ctx.cacheContext(cacheId);
@@ -884,6 +891,8 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
key.finishUnmarshal(context().cacheObjectContext(), clsLdr);
+ key.partition(partId);
+
val.unmarshal(this.ctx, clsLdr);
if (expiryPlcBytes != null && expiryPlc == null)
@@ -933,7 +942,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
/**
* @param ver Entry version.
*/
- public void entryReadVersion(GridCacheVersion ver) {
+ public void entryReadVersion(GridCacheVersion ver) {
assert this.serReadVer == null;
assert ver != null;
@@ -1037,6 +1046,12 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
writer.incrementState();
+ case 12:
+ if (!writer.writeInt("partId", partId))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -1146,6 +1161,14 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
reader.incrementState();
+ case 12:
+ partId = reader.readInt("partId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(IgniteTxEntry.class);
@@ -1158,7 +1181,7 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 12;
+ return 13;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
index cadf1a9..686f308 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java
@@ -136,6 +136,16 @@ public interface IgniteCacheObjectProcessor extends GridProcessor {
/**
* @param ctx Cache context.
+ * @param obj Key value.
+ * @param userObj If {@code true} then given object is object provided by user and should be copied
+ * before stored in cache.
+ * @param partition ID of partition this key belongs to.
+ * @return Cache key object.
+ */
+ public KeyCacheObject toCacheKeyObject(CacheObjectContext ctx, Object obj, boolean userObj, int partition);
+
+ /**
+ * @param ctx Cache context.
* @param obj Object.
* @param userObj If {@code true} then given object is object provided by user and should be copied
* before stored in cache.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
index 9a70911..9b47e59 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
@@ -102,18 +102,24 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
/** {@inheritDoc} */
@Override public Object unmarshal(CacheObjectContext ctx, byte[] bytes, ClassLoader clsLdr)
- throws IgniteCheckedException
- {
+ throws IgniteCheckedException {
return ctx.kernalContext().cache().context().marshaller().unmarshal(bytes, U.resolveClassLoader(clsLdr,
ctx.kernalContext().config()));
}
/** {@inheritDoc} */
@Override @Nullable public KeyCacheObject toCacheKeyObject(CacheObjectContext ctx, Object obj, boolean userObj) {
- if (obj instanceof KeyCacheObject)
+ return toCacheKeyObject(ctx, obj, userObj, -1);
+ }
+
+ /** {@inheritDoc} */
+ @Override public KeyCacheObject toCacheKeyObject(CacheObjectContext ctx, Object obj, boolean userObj, int partition) {
+ if (obj instanceof KeyCacheObject) {
+ ((KeyCacheObject)obj).partition(partition);
return (KeyCacheObject)obj;
+ }
- return toCacheKeyObject0(obj, userObj);
+ return toCacheKeyObject0(obj, userObj, partition);
}
/**
@@ -123,17 +129,16 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
* @return Key cache object.
*/
@SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor")
- protected KeyCacheObject toCacheKeyObject0(Object obj, boolean userObj) {
+ protected KeyCacheObject toCacheKeyObject0(Object obj, boolean userObj, int partititon) {
if (!userObj)
- return new KeyCacheObjectImpl(obj, null);
+ return new KeyCacheObjectImpl(obj, null, partititon);
- return new UserKeyCacheObjectImpl(obj);
+ return new UserKeyCacheObjectImpl(obj, partititon);
}
/** {@inheritDoc} */
@Override public CacheObject toCacheObject(GridCacheContext ctx, long valPtr, boolean tmp)
- throws IgniteCheckedException
- {
+ throws IgniteCheckedException {
assert valPtr != 0;
int size = GridUnsafe.getInt(valPtr);
@@ -172,8 +177,7 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
/** {@inheritDoc} */
@Nullable @Override public CacheObject toCacheObject(CacheObjectContext ctx,
@Nullable Object obj,
- boolean userObj)
- {
+ boolean userObj) {
if (obj == null || obj instanceof CacheObject)
return (CacheObject)obj;
@@ -241,7 +245,6 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
return 0;
}
-
/** {@inheritDoc} */
@Override public Object unwrapTemporary(GridCacheContext ctx, Object obj) throws IgniteException {
return obj;
@@ -290,7 +293,14 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
* @param key Key.
*/
UserKeyCacheObjectImpl(Object key) {
- super(key, null);
+ this(key, -1);
+ }
+
+ /**
+ * @param key Key.
+ */
+ UserKeyCacheObjectImpl(Object key, int partition) {
+ super(key, null, partition);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheAtomicReferenceImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheAtomicReferenceImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheAtomicReferenceImpl.java
index b6cbf3b..448dd8b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheAtomicReferenceImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/GridCacheAtomicReferenceImpl.java
@@ -36,9 +36,9 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
+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;
-import static org.apache.ignite.internal.processors.cache.GridCacheUtils.retryTopologySafe;
/**
* Cache atomic reference implementation.
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/util/PartitionedReadOnlySet.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/PartitionedReadOnlySet.java b/modules/core/src/main/java/org/apache/ignite/internal/util/PartitionedReadOnlySet.java
new file mode 100644
index 0000000..5ab75c1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/PartitionedReadOnlySet.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.util;
+
+import java.util.AbstractSet;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Set;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ * Read-only wrapper over multiple sets.
+ */
+public class PartitionedReadOnlySet<T> extends AbstractSet<T> {
+ /** */
+ private final Collection<Set<T>> sets;
+
+ /**
+ * Constructor.
+ * @param sets Internal sets.
+ */
+ public PartitionedReadOnlySet(Collection<Set<T>> sets) {
+ this.sets = sets;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Iterator<T> iterator() {
+ Collection<Iterator<T>> iterators = new ArrayList<>(sets.size());
+
+ for (Set<T> set : sets)
+ iterators.add(set.iterator());
+
+ return F.flatIterators(iterators);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int size() {
+ int size = 0;
+
+ for (Set<T> set : sets)
+ size += set.size();
+
+ return size;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean contains(Object o) {
+ for (Set<T> set : sets)
+ if (set.contains(o))
+ return true;
+
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
index 96445d8..12940e6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/StripedCompositeReadWriteLock.java
@@ -86,6 +86,16 @@ public class StripedCompositeReadWriteLock implements ReadWriteLock {
}
/**
+ * Queries if the write lock is held by the current thread.
+ *
+ * @return {@code true} if the current thread holds the write lock and
+ * {@code false} otherwise
+ */
+ public boolean isWriteLockedByCurrentThread() {
+ return locks[locks.length - 1].isWriteLockedByCurrentThread();
+ }
+
+ /**
* Read lock.
*/
@SuppressWarnings("unused")
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
index 8234dc4..ab31625 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/lang/GridFunc.java
@@ -287,7 +287,6 @@ public class GridFunc {
}
};
-
/** */
private static final IgniteClosure CACHE_ENTRY_VAL_GET = new IgniteClosure() {
@SuppressWarnings({"unchecked"})
@@ -4665,4 +4664,4 @@ public class GridFunc {
public static IgnitePredicate<IgniteInternalFuture<?>> unfinishedFutures() {
return UNFINISHED_FUTURE;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCache.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCache.java
index b5151c7..1be7af8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/cache/VisorCache.java
@@ -233,9 +233,9 @@ public class VisorCache implements Serializable {
protected void estimateMemorySize(IgniteEx ignite, GridCacheAdapter ca, int sample) throws IgniteCheckedException {
int size = ca.size();
- Set<GridCacheEntryEx> set = ca.context().isNear()
- ? ((GridNearCacheAdapter)ca).dht().map().entries0()
- : ca.map().entries0();
+ Iterable<GridCacheEntryEx> set = ca.context().isNear()
+ ? ((GridNearCacheAdapter)ca).dht().entries()
+ : ca.entries();
long memSz = 0;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePreloadingEvictionsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePreloadingEvictionsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePreloadingEvictionsSelfTest.java
index ba43454..bcdab3e 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePreloadingEvictionsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCachePreloadingEvictionsSelfTest.java
@@ -108,7 +108,7 @@ public class GridCachePreloadingEvictionsSelfTest extends GridCommonAbstractTest
try {
final Ignite ignite1 = startGrid(1);
- IgniteCache<Integer, Object> cache1 = ignite1.cache(null);
+ final IgniteCache<Integer, Object> cache1 = ignite1.cache(null);
for (int i = 0; i < 5000; i++)
cache1.put(i, VALUE + i);
@@ -129,7 +129,7 @@ public class GridCachePreloadingEvictionsSelfTest extends GridCommonAbstractTest
info("Started evicting...");
for (int i = 0; i < 3000 && !done.get(); i++) {
- Cache.Entry<Integer, Object> entry = randomEntry(ignite1);
+ Cache.Entry<Integer, Object> entry = cache1.getEntry(i);
if (entry != null)
ignite1.cache(null).localEvict(Collections.<Object>singleton(entry.getKey()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManagerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManagerSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManagerSelfTest.java
index 5e746c3..7b4a5ac 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManagerSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManagerSelfTest.java
@@ -112,7 +112,7 @@ public class GridCacheTtlManagerSelfTest extends GridCommonAbstractTest {
assertNull(g.cache(null).get(key));
if (!g.internalCache().context().deferredDelete())
- assertNull(g.internalCache().map().getEntry(key));
+ assertNull(g.internalCache().map().getEntry(g.internalCache().context().toCacheKeyObject(key)));
}
});
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheValueConsistencyAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheValueConsistencyAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheValueConsistencyAbstractSelfTest.java
index dc6a503..aec448b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheValueConsistencyAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheValueConsistencyAbstractSelfTest.java
@@ -30,6 +30,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.util.typedef.F;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE;
import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.CLOCK;
@@ -432,7 +433,7 @@ public abstract class GridCacheValueConsistencyAbstractSelfTest extends GridCach
for (Object ignored : keys)
itSize++;
- int valsSize = cache.values().size();
+ int valsSize = F.size(cache.values().iterator());
info("cacheSize=" + cacheSize + ", keysSize=" + keySetSize + ", valsSize=" + valsSize +
", itSize=" + itSize + ']');
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
index b661b52..a88ad15 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
@@ -28,8 +28,6 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridNoStorageCacheMap;
-import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
@@ -313,14 +311,6 @@ public class IgniteDynamicClientCacheStartSelfTest extends GridCommonAbstractTes
assertEquals(near, cache.context().isNear());
- if (near)
- cache = ((GridNearCacheAdapter)cache).dht();
-
- if (srv)
- assertSame(GridCacheConcurrentMap.class, cache.map().getClass());
- else
- assertSame(GridNoStorageCacheMap.class, cache.map().getClass());
-
ClusterNode node = ((IgniteKernal)ignite).localNode();
for (Ignite ignite0 : Ignition.allGrids()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
index 9c8f7f8..b65b441 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxStoreExceptionAbstractSelfTest.java
@@ -369,7 +369,7 @@ public abstract class IgniteTxStoreExceptionAbstractSelfTest extends GridCacheAb
GridCacheAdapter cache = grid.internalCache(null);
- GridCacheMapEntry entry = cache.map().getEntry(key);
+ GridCacheMapEntry entry = cache.map().getEntry(cache.context().toCacheKeyObject(key));
log.info("Entry: " + entry);
@@ -378,11 +378,11 @@ public abstract class IgniteTxStoreExceptionAbstractSelfTest extends GridCacheAb
assertEquals("Unexpected entry for grid [idx=" + i + ", entry=" + entry + ']', putBefore,
entry.hasValue());
assertEquals("Unexpected entry for grid [idx=" + i + ", entry=" + entry + ']', putBefore ? 1 : null,
- entry.rawGetOrUnmarshal(false));
+ entry.rawGetOrUnmarshal(false).value(cache.ctx.cacheObjectContext(), false));
}
if (cache.isNear()) {
- entry = ((GridNearCacheAdapter)cache).dht().map().getEntry(key);
+ entry = ((GridNearCacheAdapter)cache).dht().map().getEntry(cache.context().toCacheKeyObject(key));
log.info("Dht entry: " + entry);
@@ -391,7 +391,7 @@ public abstract class IgniteTxStoreExceptionAbstractSelfTest extends GridCacheAb
assertEquals("Unexpected entry for grid [idx=" + i + ", entry=" + entry + ']', putBefore,
entry.hasValue());
assertEquals("Unexpected entry for grid [idx=" + i + ", entry=" + entry + ']', putBefore ? 1 : null,
- entry.rawGetOrUnmarshal(false));
+ entry.rawGetOrUnmarshal(false).value(cache.ctx.cacheObjectContext(), false));
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
index f776146..be20b68 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
@@ -145,7 +145,7 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
for (int i = 0; i < gridCount(); i++) {
GridCacheAdapter<Object, Object> c = ((IgniteKernal)grid(i)).internalCache();
- for (GridCacheEntryEx e : c.map().entries0()) {
+ for (GridCacheEntryEx e : c.map().entries()) {
Object key = e.key().value(c.context().cacheObjectContext(), false);
Object val = CU.value(e.rawGet(), c.context(), false);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractQueueFailoverDataConsistencySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractQueueFailoverDataConsistencySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractQueueFailoverDataConsistencySelfTest.java
index 2504105..45b4b9f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractQueueFailoverDataConsistencySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheAbstractQueueFailoverDataConsistencySelfTest.java
@@ -364,7 +364,7 @@ public abstract class GridCacheAbstractQueueFailoverDataConsistencySelfTest exte
GridCacheAffinityManager aff = cctx.affinity();
for (int i = 0; i < gridCount(); i++) {
- for (GridCacheEntryEx e : ((IgniteKernal)grid(i)).context().cache().internalCache(cctx.name()).map().allEntries0()) {
+ for (GridCacheEntryEx e : ((IgniteKernal)grid(i)).context().cache().internalCache(cctx.name()).allEntries()) {
if (aff.primary(grid(i).localNode(), e.key(), AffinityTopologyVersion.NONE)
&& e.key().value(cctx.cacheObjectContext(), false) instanceof GridCacheQueueHeaderKey)
return i;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3be3d16b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheQueueCleanupSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheQueueCleanupSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheQueueCleanupSelfTest.java
index d67e725..1eac282 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheQueueCleanupSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/GridCacheQueueCleanupSelfTest.java
@@ -31,7 +31,7 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.internal.processors.datastructures.GridCacheQueueHeaderKey;
import org.apache.ignite.internal.util.typedef.PAX;
import org.apache.ignite.testframework.GridTestUtils;
@@ -191,7 +191,7 @@ public class GridCacheQueueCleanupSelfTest extends IgniteCollectionAbstractTest
GridCacheAdapter<Object, Object> cache =
((IgniteKernal)grid(i)).context().cache().internalCache(queueCacheName);
- Iterator<GridCacheEntryEx> entries = cache.map().allEntries0().iterator();
+ Iterator<GridCacheMapEntry> entries = cache.map().entries().iterator();
while (entries.hasNext()) {
cnt++;