You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vk...@apache.org on 2015/07/31 02:26:08 UTC
[1/5] incubator-ignite git commit: #ignite-1175: Add test for dht
local partition map.
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-104 5241e2747 -> 1e98372ed
#ignite-1175: Add test for dht local partition map.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/7ed4d15f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/7ed4d15f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/7ed4d15f
Branch: refs/heads/ignite-104
Commit: 7ed4d15f16c71e1683fd659865653a383d99259e
Parents: 5288b2d
Author: ivasilinets <iv...@gridgain.com>
Authored: Thu Jul 30 14:12:27 2015 +0300
Committer: ivasilinets <iv...@gridgain.com>
Committed: Thu Jul 30 14:12:27 2015 +0300
----------------------------------------------------------------------
...cheDhtLocalPartitionAfterRemoveSelfTest.java | 107 +++++++++++++++++++
1 file changed, 107 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ed4d15f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheDhtLocalPartitionAfterRemoveSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheDhtLocalPartitionAfterRemoveSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheDhtLocalPartitionAfterRemoveSelfTest.java
new file mode 100644
index 0000000..b04e41a
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheDhtLocalPartitionAfterRemoveSelfTest.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 org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.processors.cache.distributed.dht.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+/**
+ * Test for remove operation.
+ */
+public class CacheDhtLocalPartitionAfterRemoveSelfTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ ccfg.setEvictSynchronized(false);
+ ccfg.setNearConfiguration(null);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ startGrids(1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testMemoryUsage() throws Exception {
+ IgniteCache<TestKey, Integer> cache = grid(0).cache(null);
+
+ for (int i = 0; i < 1000; ++i)
+ cache.put(new TestKey("" + i), i);
+
+ for (int i = 0; i < 1000; ++i)
+ assert cache.getAndRemove(new TestKey("" + i)).equals(i);
+
+ assertEquals(0, cache.size());
+
+ int size = 0;
+
+ for (GridDhtLocalPartition p : dht(cache).topology().localPartitions()) {
+ int pSize = p.size();
+
+ size += pSize;
+ }
+
+ System.out.println("All size: " + size);
+ }
+
+ /**
+ * Test key.
+ */
+ private static class TestKey {
+ /** Key. */
+ private String key;
+
+ /**
+ * @param key Key.
+ */
+ public TestKey(String key) {
+ this.key = key;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return key.hashCode();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object obj) {
+ if (obj == null || !(obj instanceof TestKey))
+ return false;
+
+ return key.equals(((TestKey)obj).key);
+ }
+ }
+}
[2/5] incubator-ignite git commit: Merge branches 'ignite-104' and
'master' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into
ignite-104
Posted by vk...@apache.org.
Merge branches 'ignite-104' and 'master' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-104
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/dad46913
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/dad46913
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/dad46913
Branch: refs/heads/ignite-104
Commit: dad46913f38626f1335232549e5236ff21547812
Parents: 5241e27 7ed4d15
Author: Valentin Kulichenko <vk...@gridgain.com>
Authored: Thu Jul 30 11:53:26 2015 -0700
Committer: Valentin Kulichenko <vk...@gridgain.com>
Committed: Thu Jul 30 11:53:26 2015 -0700
----------------------------------------------------------------------
...cheDhtLocalPartitionAfterRemoveSelfTest.java | 107 +++++++++++++++++++
1 file changed, 107 insertions(+)
----------------------------------------------------------------------
[5/5] incubator-ignite git commit: IGNITE-104 - Ordered ATOMIC updates
Posted by vk...@apache.org.
IGNITE-104 - Ordered ATOMIC updates
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/1e98372e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/1e98372e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/1e98372e
Branch: refs/heads/ignite-104
Commit: 1e98372edca990804d1686521fd152c0549baa2c
Parents: f187add
Author: Valentin Kulichenko <vk...@gridgain.com>
Authored: Thu Jul 30 17:24:48 2015 -0700
Committer: Valentin Kulichenko <vk...@gridgain.com>
Committed: Thu Jul 30 17:24:48 2015 -0700
----------------------------------------------------------------------
.../internal/processors/cache/GridCacheMapEntry.java | 5 -----
.../distributed/dht/atomic/GridDhtAtomicCache.java | 9 +++------
.../cache/distributed/near/GridNearAtomicCache.java | 14 ++++----------
3 files changed, 7 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1e98372e/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 79d7989..5d2af9a 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
@@ -2161,9 +2161,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- if (!cctx.deferredDelete())
- markObsolete0(rmvVer, true);
-
res = hadVal;
}
@@ -2458,8 +2455,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/** {@inheritDoc} */
@Override public boolean markObsoleteVersion(GridCacheVersion ver) {
- assert cctx.deferredDelete();
-
boolean marked;
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1e98372e/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 19c89e7..3802b29 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
@@ -1177,7 +1177,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
}
catch (GridCacheEntryRemovedException e) {
- assert !ctx.deferredDelete() : "Entry should not become obsolete while holding lock.";
+ assert false : "Entry should not become obsolete while holding lock.";
e.printStackTrace();
}
@@ -1196,11 +1196,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
else {
GridDhtCacheEntry entry = e.get1();
- assert entry.obsolete();
-
- entry.onMarkedObsolete();
-
- removeEntry(entry);
+ if (entry.markObsoleteVersion(e.get2()))
+ removeEntry(entry);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1e98372e/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index 707facc..b3f4a1d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -245,11 +245,8 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
if (ctx.deferredDelete())
ctx.onDeferredDelete(entry, updRes.removeVersion());
else {
- assert entry.obsolete();
-
- entry.onMarkedObsolete();
-
- removeEntry(entry);
+ if (entry.markObsoleteVersion(updRes.removeVersion()))
+ removeEntry(entry);
}
}
@@ -352,11 +349,8 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K, V> {
if (ctx.deferredDelete())
ctx.onDeferredDelete(entry, updRes.removeVersion());
else {
- assert entry.obsolete();
-
- entry.onMarkedObsolete();
-
- removeEntry(entry);
+ if (entry.markObsoleteVersion(updRes.removeVersion()))
+ removeEntry(entry);
}
}
[4/5] incubator-ignite git commit: IGNITE-104 - Ordered ATOMIC updates
Posted by vk...@apache.org.
IGNITE-104 - Ordered ATOMIC updates
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/f187add2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/f187add2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/f187add2
Branch: refs/heads/ignite-104
Commit: f187add2d9e96fda1da4ea5a98940cd4663ee355
Parents: 2d16d99
Author: Valentin Kulichenko <vk...@gridgain.com>
Authored: Thu Jul 30 16:59:48 2015 -0700
Committer: Valentin Kulichenko <vk...@gridgain.com>
Committed: Thu Jul 30 16:59:48 2015 -0700
----------------------------------------------------------------------
.../distributed/dht/atomic/GridDhtAtomicCache.java | 5 +----
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 12 ------------
2 files changed, 1 insertion(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f187add2/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 a010baa..19c89e7 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
@@ -1182,9 +1182,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
e.printStackTrace();
}
finally {
- if (dhtFut != null && !remap)
- dhtFut.map();
-
if (locked != null)
unlockEntries(locked, req.topologyVersion());
@@ -1228,7 +1225,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
else {
// If there are backups, map backup update future.
if (dhtFut != null)
- dhtFut.onMapped();
+ dhtFut.map();
// Otherwise, complete the call.
else
completionCb.apply(req, res);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f187add2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index c05f4c6..eb1b42c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -90,9 +90,6 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
/** */
private boolean waitForExchange;
- /** */
- private boolean mapped;
-
/**
* @param cctx Cache context.
* @param completionCb Callback to invoke when future is completed.
@@ -385,15 +382,6 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
}
}
- mapped = true;
- }
-
- /**
- * On mapped callback.
- */
- public void onMapped() {
- assert mapped;
-
checkComplete();
// Send response right away if no ACKs from backup is required.
[3/5] incubator-ignite git commit: IGNITE-104 - Ordered ATOMIC updates
Posted by vk...@apache.org.
IGNITE-104 - Ordered ATOMIC updates
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/2d16d99f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/2d16d99f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/2d16d99f
Branch: refs/heads/ignite-104
Commit: 2d16d99f64fdfbff591124abcb4c5d42ac29d8bf
Parents: dad4691
Author: Valentin Kulichenko <vk...@gridgain.com>
Authored: Thu Jul 30 16:48:30 2015 -0700
Committer: Valentin Kulichenko <vk...@gridgain.com>
Committed: Thu Jul 30 16:48:30 2015 -0700
----------------------------------------------------------------------
.../ignite/codegen/MessageCodeGenerator.java | 2 +
.../org/apache/ignite/internal/GridTopic.java | 83 --------
.../managers/communication/GridIoManager.java | 199 ++++++++++++++++++-
.../managers/communication/GridIoMessage.java | 48 +++--
.../processors/cache/GridCacheIoManager.java | 99 +++++++--
.../processors/cache/GridCacheUtils.java | 11 -
.../dht/atomic/GridAtomicMappingKey.java | 86 ++++++++
.../dht/atomic/GridAtomicRequestTopic.java | 96 +++++++++
.../dht/atomic/GridDhtAtomicCache.java | 33 ++-
.../dht/atomic/GridDhtAtomicUpdateFuture.java | 89 ++-------
.../dht/atomic/GridNearAtomicUpdateFuture.java | 172 +++++++++++-----
.../dht/atomic/GridNearAtomicUpdateRequest.java | 63 ++++--
.../atomic/GridNearAtomicUpdateResponse.java | 36 +++-
.../preloader/GridDhtPartitionDemandPool.java | 15 +-
.../query/GridCacheDistributedQueryManager.java | 8 +-
.../resources/META-INF/classnames.properties | 2 +-
16 files changed, 730 insertions(+), 312 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index c560118..2510d65 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -142,6 +142,8 @@ public class MessageCodeGenerator {
MessageCodeGenerator gen = new MessageCodeGenerator(srcDir);
+// gen.generateAndWrite(GridIoMessage.class);
+
// gen.generateAndWrite(GridNearAtomicUpdateRequest.class);
// gen.generateAndWrite(GridNearAtomicUpdateResponse.class);
// gen.generateAndWrite(GridDhtAtomicUpdateRequest.class);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
index e9da40c..56aea1e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridTopic.java
@@ -183,15 +183,6 @@ public enum GridTopic {
}
/**
- * @param id1 ID1.
- * @param id2 ID2.
- * @return Grid message topic with specified IDs.
- */
- public Object topic(int id1, int id2) {
- return new T9(this, id1, id2);
- }
-
- /**
*
*/
private static class T1 implements Externalizable {
@@ -765,78 +756,4 @@ public enum GridTopic {
return S.toString(T8.class, this);
}
}
-
- /**
- */
- private static class T9 implements Externalizable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** */
- private GridTopic topic;
-
- /** */
- private int id1;
-
- /** */
- private int id2;
-
- /**
- * No-arg constructor needed for {@link Serializable}.
- */
- public T9() {
- // No-op.
- }
-
- /**
- * @param topic Topic.
- * @param id1 ID1.
- * @param id2 ID2.
- */
- private T9(GridTopic topic, int id1, int id2) {
- this.topic = topic;
- this.id1 = id1;
- this.id2 = id2;
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- int res = topic.ordinal();
-
- res += 31 * res + id1;
- res += 31 * res + id2;
-
- return res;
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object obj) {
- if (obj.getClass() == T9.class) {
- T9 that = (T9)obj;
-
- return topic == that.topic && id1 == that.id1 && id2 == that.id2;
- }
-
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public void writeExternal(ObjectOutput out) throws IOException {
- out.writeByte(topic.ordinal());
- out.writeInt(id1);
- out.writeInt(id2);
- }
-
- /** {@inheritDoc} */
- @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
- topic = fromOrdinal(in.readByte());
- id1 = in.readInt();
- id2 = in.readInt();
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(T9.class, this);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index c1fb79a..765ba65 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@ -100,6 +100,9 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
private final ConcurrentMap<Object, ConcurrentMap<UUID, GridCommunicationMessageSet>> msgSetMap =
new ConcurrentHashMap8<>();
+ /** */
+ private final ConcurrentMap<Object, SequentialMessageSet> seqMsgs = new ConcurrentHashMap8<>();
+
/** Local node ID. */
private final UUID locNodeId;
@@ -576,6 +579,8 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
{
if (msg.isOrdered())
processOrderedMessage(nodeId, msg, plc, msgC);
+ else if (msg.isSequential())
+ processSequentialMessage(nodeId, msg, plc, msgC);
else
processRegularMessage(nodeId, msg, plc, msgC);
@@ -591,6 +596,8 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
if (msg.isOrdered())
processOrderedMessage(nodeId, msg, plc, msgC);
+ else if (msg.isSequential())
+ processSequentialMessage(nodeId, msg, plc, msgC);
else
processRegularMessage(nodeId, msg, plc, msgC);
}
@@ -963,6 +970,78 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
}
/**
+ * @param nodeId Node ID.
+ * @param msg Message.
+ * @param plc Execution policy.
+ * @param msgC Closure to call when message processing finished.
+ */
+ private void processSequentialMessage(
+ final UUID nodeId,
+ final GridIoMessage msg,
+ byte plc,
+ final IgniteRunnable msgC
+ ) throws IgniteCheckedException {
+ final GridMessageListener lsnr = lsnrMap.get(msg.topic());
+
+ if (lsnr == null) {
+ if (log.isDebugEnabled())
+ log.debug("Ignoring message because listener is not found: " + msg);
+
+ if (msgC != null)
+ msgC.run();
+
+ return;
+ }
+
+ SequentialMessageSet msgSet = seqMsgs.get(msg.topic());
+
+ if (msgSet == null) {
+ SequentialMessageSet old = seqMsgs.putIfAbsent(msg.topic(), msgSet = new SequentialMessageSet());
+
+ if (old != null)
+ msgSet = old;
+ }
+
+ msgSet.add(nodeId, msg, msgC);
+
+ if (msgC == null) {
+ assert locNodeId.equals(nodeId);
+
+ msgSet.unwind(lsnr);
+ }
+ else {
+ assert !locNodeId.equals(nodeId);
+
+ final SequentialMessageSet msgSet0 = msgSet;
+
+ Runnable c = new Runnable() {
+ @Override public void run() {
+ try {
+ threadProcessingMessage(true);
+
+ msgSet0.unwind(lsnr);
+ }
+ finally {
+ threadProcessingMessage(false);
+ }
+ }
+ };
+
+ try {
+ pool(plc).execute(c);
+ }
+ catch (RejectedExecutionException e) {
+ U.error(log, "Failed to process sequential message due to execution rejection. " +
+ "Increase the upper bound on executor service provided by corresponding " +
+ "configuration property. Will attempt to process message in the listener " +
+ "thread instead [msgPlc=" + plc + ']', e);
+
+ c.run();
+ }
+ }
+ }
+
+ /**
* @param node Destination node.
* @param topic Topic to send the message to.
* @param topicOrd GridTopic enumeration ordinal.
@@ -980,6 +1059,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
Message msg,
byte plc,
boolean ordered,
+ boolean seq,
long timeout,
boolean skipOnTimeout
) throws IgniteCheckedException {
@@ -987,7 +1067,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
assert topic != null;
assert msg != null;
- GridIoMessage ioMsg = new GridIoMessage(plc, topic, topicOrd, msg, ordered, timeout, skipOnTimeout);
+ GridIoMessage ioMsg = new GridIoMessage(plc, topic, topicOrd, msg, ordered, seq, timeout, skipOnTimeout);
if (locNodeId.equals(node.id())) {
assert plc != P2P_POOL;
@@ -999,6 +1079,8 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
if (ordered)
processOrderedMessage(locNodeId, ioMsg, plc, null);
+ else if (seq)
+ processSequentialMessage(locNodeId, ioMsg, plc, null);
else
processRegularMessage0(ioMsg, locNodeId);
}
@@ -1050,7 +1132,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
if (node == null)
throw new IgniteCheckedException("Failed to send message to node (has node left grid?): " + nodeId);
- send(node, topic, topic.ordinal(), msg, plc, false, 0, false);
+ send(node, topic, topic.ordinal(), msg, plc, false, false, 0, false);
}
/**
@@ -1062,7 +1144,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
*/
public void send(ClusterNode node, Object topic, Message msg, byte plc)
throws IgniteCheckedException {
- send(node, topic, -1, msg, plc, false, 0, false);
+ send(node, topic, -1, msg, plc, false, false, 0, false);
}
/**
@@ -1074,7 +1156,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
*/
public void send(ClusterNode node, GridTopic topic, Message msg, byte plc)
throws IgniteCheckedException {
- send(node, topic, topic.ordinal(), msg, plc, false, 0, false);
+ send(node, topic, topic.ordinal(), msg, plc, false, false, 0, false);
}
/**
@@ -1096,7 +1178,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
) throws IgniteCheckedException {
assert timeout > 0 || skipOnTimeout;
- send(node, topic, (byte)-1, msg, plc, true, timeout, skipOnTimeout);
+ send(node, topic, (byte)-1, msg, plc, true, false, timeout, skipOnTimeout);
}
/**
@@ -1123,7 +1205,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
if (node == null)
throw new IgniteCheckedException("Failed to send message to node (has node left grid?): " + nodeId);
- send(node, topic, (byte)-1, msg, plc, true, timeout, skipOnTimeout);
+ send(node, topic, (byte)-1, msg, plc, true, false, timeout, skipOnTimeout);
}
/**
@@ -1146,7 +1228,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
throws IgniteCheckedException {
assert timeout > 0 || skipOnTimeout;
- send(nodes, topic, -1, msg, plc, true, timeout, skipOnTimeout);
+ send(nodes, topic, -1, msg, plc, true, false, timeout, skipOnTimeout);
}
/**
@@ -1162,7 +1244,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
Message msg,
byte plc
) throws IgniteCheckedException {
- send(nodes, topic, -1, msg, plc, false, 0, false);
+ send(nodes, topic, -1, msg, plc, false, false, 0, false);
}
/**
@@ -1178,7 +1260,48 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
Message msg,
byte plc
) throws IgniteCheckedException {
- send(nodes, topic, topic.ordinal(), msg, plc, false, 0, false);
+ send(nodes, topic, topic.ordinal(), msg, plc, false, false, 0, false);
+ }
+
+ /**
+ * Sends sequential message.
+ *
+ * @param nodeId Destination node ID.
+ * @param topic Topic.
+ * @param msg Message.
+ * @param plc Policy.
+ * @throws IgniteCheckedException In case of error.
+ */
+ public void sendSequentialMessage(
+ UUID nodeId,
+ Object topic,
+ Message msg,
+ byte plc
+ ) throws IgniteCheckedException {
+ ClusterNode node = ctx.discovery().node(nodeId);
+
+ if (node == null)
+ throw new IgniteCheckedException("Failed to send message to node (has node left grid?): " + nodeId);
+
+ sendSequentialMessage(node, topic, msg, plc);
+ }
+
+ /**
+ * Sends sequential message.
+ *
+ * @param node Destination node.
+ * @param topic Topic.
+ * @param msg Message.
+ * @param plc Policy.
+ * @throws IgniteCheckedException In case of error.
+ */
+ public void sendSequentialMessage(
+ ClusterNode node,
+ Object topic,
+ Message msg,
+ byte plc
+ ) throws IgniteCheckedException {
+ send(node, topic, -1, msg, plc, false, true, 0, false);
}
/**
@@ -1307,6 +1430,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
* @param msg Message to send.
* @param plc Type of processing.
* @param ordered Ordered flag.
+ * @param seq Sequential message flag.
* @param timeout Message timeout.
* @param skipOnTimeout Whether message can be skipped in timeout.
* @throws IgniteCheckedException Thrown in case of any errors.
@@ -1318,6 +1442,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
Message msg,
byte plc,
boolean ordered,
+ boolean seq,
long timeout,
boolean skipOnTimeout
) throws IgniteCheckedException {
@@ -1334,7 +1459,7 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
// messages to one node vs. many.
if (!nodes.isEmpty()) {
for (ClusterNode node : nodes)
- send(node, topic, topicOrd, msg, plc, ordered, timeout, skipOnTimeout);
+ send(node, topic, topicOrd, msg, plc, ordered, seq, timeout, skipOnTimeout);
}
else if (log.isDebugEnabled())
log.debug("Failed to send message to empty nodes collection [topic=" + topic + ", msg=" +
@@ -2216,4 +2341,58 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
return S.toString(DelayedMessage.class, this, super.toString());
}
}
+
+ /**
+ */
+ private static class SequentialMessageSet {
+ /** */
+ private final Queue<GridTuple3<UUID, GridIoMessage, IgniteRunnable>> queue = new ConcurrentLinkedDeque8<>();
+
+ /** */
+ private final AtomicBoolean reserve = new AtomicBoolean();
+
+ /**
+ * @param nodeId Node ID.
+ * @param msg Message.
+ * @param msgC Closure to call when message processing finished.
+ */
+ void add(UUID nodeId, GridIoMessage msg, IgniteRunnable msgC) {
+ queue.add(F.t(nodeId, msg, msgC));
+ }
+
+ /**
+ * @param lsnr Message listener.
+ */
+ void unwind(GridMessageListener lsnr) {
+ assert lsnr != null;
+
+ while (true) {
+ if (reserve.compareAndSet(false, true)) {
+ try {
+ GridTuple3<UUID, GridIoMessage, IgniteRunnable> t;
+
+ while ((t = queue.poll()) != null) {
+ try {
+ lsnr.onMessage(t.get1(), t.get2().message());
+ }
+ finally {
+ IgniteRunnable msgC = t.get3();
+
+ if (msgC != null)
+ msgC.run();
+ }
+ }
+ }
+ finally {
+ reserve.set(false);
+ }
+
+ if (queue.isEmpty())
+ return;
+ }
+ else
+ return;
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
index 6cf1ae5..d729f75 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessage.java
@@ -49,6 +49,9 @@ public class GridIoMessage implements Message {
/** Message ordered flag. */
private boolean ordered;
+ /** Sequential message flag. */
+ private boolean seq;
+
/** Message timeout. */
private long timeout;
@@ -72,6 +75,7 @@ public class GridIoMessage implements Message {
* @param topicOrd Topic ordinal value.
* @param msg Message.
* @param ordered Message ordered flag.
+ * @param seq Sequential message flag.
* @param timeout Timeout.
* @param skipOnTimeout Whether message can be skipped on timeout.
*/
@@ -81,18 +85,21 @@ public class GridIoMessage implements Message {
int topicOrd,
Message msg,
boolean ordered,
+ boolean seq,
long timeout,
boolean skipOnTimeout
) {
assert topic != null;
assert topicOrd <= Byte.MAX_VALUE;
assert msg != null;
+ assert !ordered || !seq; // Message can't be ordered and sequential at the same time.
this.plc = plc;
this.msg = msg;
this.topic = topic;
this.topicOrd = topicOrd;
this.ordered = ordered;
+ this.seq = seq;
this.timeout = timeout;
this.skipOnTimeout = skipOnTimeout;
}
@@ -167,6 +174,13 @@ public class GridIoMessage implements Message {
return ordered;
}
+ /**
+ * @return Sequential message flag.
+ */
+ boolean isSequential() {
+ return seq;
+ }
+
/** {@inheritDoc} */
@Override public boolean equals(Object obj) {
throw new AssertionError();
@@ -208,24 +222,30 @@ public class GridIoMessage implements Message {
writer.incrementState();
case 3:
- if (!writer.writeBoolean("skipOnTimeout", skipOnTimeout))
+ if (!writer.writeBoolean("seq", seq))
return false;
writer.incrementState();
case 4:
- if (!writer.writeLong("timeout", timeout))
+ if (!writer.writeBoolean("skipOnTimeout", skipOnTimeout))
return false;
writer.incrementState();
case 5:
- if (!writer.writeByteArray("topicBytes", topicBytes))
+ if (!writer.writeLong("timeout", timeout))
return false;
writer.incrementState();
case 6:
+ if (!writer.writeByteArray("topicBytes", topicBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 7:
if (!writer.writeInt("topicOrd", topicOrd))
return false;
@@ -261,19 +281,15 @@ public class GridIoMessage implements Message {
reader.incrementState();
case 2:
- byte plc0;
-
- plc0 = reader.readByte("plc");
+ plc = reader.readByte("plc");
if (!reader.isLastRead())
return false;
- plc = plc0;
-
reader.incrementState();
case 3:
- skipOnTimeout = reader.readBoolean("skipOnTimeout");
+ seq = reader.readBoolean("seq");
if (!reader.isLastRead())
return false;
@@ -281,7 +297,7 @@ public class GridIoMessage implements Message {
reader.incrementState();
case 4:
- timeout = reader.readLong("timeout");
+ skipOnTimeout = reader.readBoolean("skipOnTimeout");
if (!reader.isLastRead())
return false;
@@ -289,7 +305,7 @@ public class GridIoMessage implements Message {
reader.incrementState();
case 5:
- topicBytes = reader.readByteArray("topicBytes");
+ timeout = reader.readLong("timeout");
if (!reader.isLastRead())
return false;
@@ -297,6 +313,14 @@ public class GridIoMessage implements Message {
reader.incrementState();
case 6:
+ topicBytes = reader.readByteArray("topicBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 7:
topicOrd = reader.readInt("topicOrd");
if (!reader.isLastRead())
@@ -316,7 +340,7 @@ public class GridIoMessage implements Message {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 7;
+ return 8;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index 5858424..490a5d8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -62,8 +62,8 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
private ConcurrentMap<ListenerKey, IgniteBiInClosure<UUID, GridCacheMessage>>
clsHandlers = new ConcurrentHashMap8<>();
- /** Ordered handler registry. */
- private ConcurrentMap<Object, IgniteBiInClosure<UUID, ? extends GridCacheMessage>> orderedHandlers =
+ /** Per topic handler registry. */
+ private ConcurrentMap<Object, IgniteBiInClosure<UUID, ? extends GridCacheMessage>> perTopicHandlers =
new ConcurrentHashMap8<>();
/** Stopping flag. */
@@ -173,7 +173,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
@Override protected void onKernalStop0(boolean cancel) {
cctx.gridIO().removeMessageListener(TOPIC_CACHE);
- for (Object ordTopic : orderedHandlers.keySet())
+ for (Object ordTopic : perTopicHandlers.keySet())
cctx.gridIO().removeMessageListener(ordTopic);
boolean interrupted = false;
@@ -394,7 +394,8 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(
ctx.cacheId(),
nodeId,
- req.futureVersion());
+ req.futureVersion(),
+ req.partition());
res.error(req.classError());
@@ -813,6 +814,64 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
}
/**
+ * @param nodeId Destination node ID.
+ * @param topic Topic to send the message to.
+ * @param msg Message to send.
+ * @param plc IO policy.
+ * @throws IgniteCheckedException Thrown in case of any errors.
+ */
+ public void sendSequentialMessage(UUID nodeId, Object topic, GridCacheMessage msg, byte plc)
+ throws IgniteCheckedException {
+ ClusterNode n = cctx.discovery().node(nodeId);
+
+ if (n == null)
+ throw new ClusterTopologyCheckedException("Failed to send message because node left grid [nodeId=" +
+ nodeId + ", msg=" + msg + ']');
+
+ sendSequentialMessage(n, topic, msg, plc);
+ }
+
+ /**
+ * @param node Destination node.
+ * @param topic Topic to send the message to.
+ * @param msg Message to send.
+ * @param plc IO policy.
+ * @throws IgniteCheckedException Thrown in case of any errors.
+ */
+ public void sendSequentialMessage(ClusterNode node, Object topic, GridCacheMessage msg, byte plc)
+ throws IgniteCheckedException {
+ onSend(msg, node.id());
+
+ int cnt = 0;
+
+ while (cnt <= retryCnt) {
+ try {
+ cnt++;
+
+ cctx.gridIO().sendSequentialMessage(node, topic, msg, plc);
+
+ if (log.isDebugEnabled())
+ log.debug("Sent sequential cache message [topic=" + topic + ", msg=" + msg +
+ ", nodeId=" + node.id() + ']');
+
+ return;
+ }
+ catch (IgniteCheckedException e) {
+ if (cctx.discovery().node(node.id()) == null)
+ throw new ClusterTopologyCheckedException("Node left grid while sending sequential message [" +
+ "nodeId=" + node.id() + ", msg=" + msg + ']', e);
+
+ if (cnt == retryCnt)
+ throw e;
+ else if (log.isDebugEnabled())
+ log.debug("Failed to send message to node (will retry): " + node.id());
+ }
+
+ U.sleep(retryDelay);
+ }
+ }
+
+ /**
* @return ID that auto-grows based on local counter and counters received
* from other nodes.
*/
@@ -940,39 +999,39 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
}
/**
- * Adds ordered message handler.
+ * Adds per topic message handler.
*
* @param topic Topic.
* @param c Handler.
*/
@SuppressWarnings({"unchecked"})
- public void addOrderedHandler(Object topic, IgniteBiInClosure<UUID, ? extends GridCacheMessage> c) {
- if (orderedHandlers.putIfAbsent(topic, c) == null) {
- cctx.gridIO().addMessageListener(topic, new OrderedMessageListener(
+ public void addPerTopicHandler(Object topic, IgniteBiInClosure<UUID, ? extends GridCacheMessage> c) {
+ if (perTopicHandlers.putIfAbsent(topic, c) == null) {
+ cctx.gridIO().addMessageListener(topic, new PerTopicMessageListener(
(IgniteBiInClosure<UUID, GridCacheMessage>)c));
if (log != null && log.isDebugEnabled())
- log.debug("Registered ordered cache communication handler [topic=" + topic + ", handler=" + c + ']');
+ log.debug("Registered per topic cache communication handler [topic=" + topic + ", handler=" + c + ']');
}
else if (log != null)
- U.warn(log, "Failed to register ordered cache communication handler because it is already " +
+ U.warn(log, "Failed to register per topic cache communication handler because it is already " +
"registered for this topic [topic=" + topic + ", handler=" + c + ']');
}
/**
- * Removed ordered message handler.
+ * Removed per topic message handler.
*
* @param topic Topic.
*/
- public void removeOrderedHandler(Object topic) {
- if (orderedHandlers.remove(topic) != null) {
+ public void removePerTopicHandler(Object topic) {
+ if (perTopicHandlers.remove(topic) != null) {
cctx.gridIO().removeMessageListener(topic);
if (log != null && log.isDebugEnabled())
- log.debug("Unregistered ordered cache communication handler for topic:" + topic);
+ log.debug("Unregistered per topic cache communication handler for topic:" + topic);
}
else if (log != null)
- U.warn(log, "Failed to unregister ordered cache communication handler because it was not found " +
+ U.warn(log, "Failed to unregister per topic cache communication handler because it was not found " +
"for topic: " + topic);
}
@@ -1019,20 +1078,20 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
X.println(">>> ");
X.println(">>> Cache IO manager memory stats [grid=" + cctx.gridName() + ']');
X.println(">>> clsHandlersSize: " + clsHandlers.size());
- X.println(">>> orderedHandlersSize: " + orderedHandlers.size());
+ X.println(">>> perTopicHandlersSize: " + perTopicHandlers.size());
}
/**
- * Ordered message listener.
+ * Per topic message listener.
*/
- private class OrderedMessageListener implements GridMessageListener {
+ private class PerTopicMessageListener implements GridMessageListener {
/** */
private final IgniteBiInClosure<UUID, GridCacheMessage> c;
/**
* @param c Handler closure.
*/
- OrderedMessageListener(IgniteBiInClosure<UUID, GridCacheMessage> c) {
+ PerTopicMessageListener(IgniteBiInClosure<UUID, GridCacheMessage> c) {
this.c = c;
}
@@ -1040,7 +1099,7 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
@SuppressWarnings({"CatchGenericClass", "unchecked"})
@Override public void onMessage(final UUID nodeId, Object msg) {
if (log.isDebugEnabled())
- log.debug("Received cache ordered message [nodeId=" + nodeId + ", msg=" + msg + ']');
+ log.debug("Received per topic cache message [nodeId=" + nodeId + ", msg=" + msg + ']');
final GridCacheMessage cacheMsg = (GridCacheMessage)msg;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/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 d82acca..a313e3d 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
@@ -1733,15 +1733,4 @@ public class GridCacheUtils {
}
};
}
-
- /**
- * @param ctx Cache context.
- * @param part Partition.
- * @return Per-partition message topic.
- */
- public static Object partitionMessageTopic(GridCacheContext ctx, int part) {
- assert part >= 0;
-
- return TOPIC_CACHE.topic(ctx.cacheId(), part);
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicMappingKey.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicMappingKey.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicMappingKey.java
new file mode 100644
index 0000000..52e3c7f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicMappingKey.java
@@ -0,0 +1,86 @@
+/*
+ * 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.atomic;
+
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.util.*;
+
+/**
+ * Mapping Key.
+ */
+class GridAtomicMappingKey {
+ /** Node ID. */
+ private final UUID nodeId;
+
+ /** Partition. */
+ private final int part;
+
+ /**
+ * @param nodeId Node ID.
+ * @param part Partition.
+ */
+ GridAtomicMappingKey(UUID nodeId, int part) {
+ assert nodeId != null;
+ assert part >= -1 : part;
+
+ this.nodeId = nodeId;
+ this.part = part;
+ }
+
+ /**
+ * @return Node ID.
+ */
+ UUID nodeId() {
+ return nodeId;
+ }
+
+ /**
+ * @return Partition.
+ */
+ int partition() {
+ return part;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ GridAtomicMappingKey key = (GridAtomicMappingKey)o;
+
+ return nodeId.equals(key.nodeId) && part == key.part;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = nodeId.hashCode();
+
+ res = 31 * res + part;
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridAtomicMappingKey.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicRequestTopic.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicRequestTopic.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicRequestTopic.java
new file mode 100644
index 0000000..9feb409
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridAtomicRequestTopic.java
@@ -0,0 +1,96 @@
+/*
+ * 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.atomic;
+
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.io.*;
+
+/**
+ */
+class GridAtomicRequestTopic implements Externalizable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private int cacheId;
+
+ /** */
+ private int part;
+
+ /** */
+ private boolean near;
+
+ /**
+ * For {@link Externalizable}.
+ */
+ public GridAtomicRequestTopic() {
+ // No-op.
+ }
+
+ /**
+ * @param cacheId Cache ID.
+ * @param part Partition.
+ * @param near Near flag.
+ */
+ GridAtomicRequestTopic(int cacheId, int part, boolean near) {
+ this.cacheId = cacheId;
+ this.part = part;
+ this.near = near;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ GridAtomicRequestTopic topic = (GridAtomicRequestTopic)o;
+
+ return cacheId == topic.cacheId && part == topic.part && near == topic.near;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = cacheId;
+
+ res = 31 * res + part;
+ res = 31 * res + (near ? 1 : 0);
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeInt(cacheId);
+ out.writeInt(part);
+ out.writeBoolean(near);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ cacheId = in.readInt();
+ part = in.readInt();
+ near = in.readBoolean();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(GridAtomicRequestTopic.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/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 be35d00..a010baa 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
@@ -181,15 +181,19 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
});
- ctx.io().addHandler(ctx.cacheId(), GridNearAtomicUpdateRequest.class, new CI2<UUID, GridNearAtomicUpdateRequest>() {
- @Override public void apply(UUID nodeId, GridNearAtomicUpdateRequest req) {
- processNearAtomicUpdateRequest(nodeId, req);
- }
- });
-
if (ctx.config().isAtomicOrderedUpdates()) {
for (int part = 0; part < ctx.affinity().partitions(); part++) {
- ctx.io().addOrderedHandler(CU.partitionMessageTopic(ctx, part), new CI2<UUID, GridDhtAtomicUpdateRequest>() {
+ Object nearTopic = new GridAtomicRequestTopic(ctx.cacheId(), part, true);
+
+ ctx.io().addPerTopicHandler(nearTopic, new CI2<UUID, GridNearAtomicUpdateRequest>() {
+ @Override public void apply(UUID nodeId, GridNearAtomicUpdateRequest req) {
+ processNearAtomicUpdateRequest(nodeId, req);
+ }
+ });
+
+ Object dhtTopic = new GridAtomicRequestTopic(ctx.cacheId(), part, false);
+
+ ctx.io().addPerTopicHandler(dhtTopic, new CI2<UUID, GridDhtAtomicUpdateRequest>() {
@Override public void apply(UUID nodeId, GridDhtAtomicUpdateRequest req) {
processDhtAtomicUpdateRequest(nodeId, req);
}
@@ -197,6 +201,12 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
}
else {
+ ctx.io().addHandler(ctx.cacheId(), GridNearAtomicUpdateRequest.class, new CI2<UUID, GridNearAtomicUpdateRequest>() {
+ @Override public void apply(UUID nodeId, GridNearAtomicUpdateRequest req) {
+ processNearAtomicUpdateRequest(nodeId, req);
+ }
+ });
+
ctx.io().addHandler(ctx.cacheId(), GridDhtAtomicUpdateRequest.class, new CI2<UUID, GridDhtAtomicUpdateRequest>() {
@Override public void apply(UUID nodeId, GridDhtAtomicUpdateRequest req) {
processDhtAtomicUpdateRequest(nodeId, req);
@@ -238,8 +248,10 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
buf.finish();
if (ctx.config().isAtomicOrderedUpdates()) {
- for (int part = 0; part < ctx.affinity().partitions(); part++)
- ctx.io().removeOrderedHandler(CU.partitionMessageTopic(ctx, part));
+ for (int part = 0; part < ctx.affinity().partitions(); part++) {
+ ctx.io().removePerTopicHandler(new GridAtomicRequestTopic(ctx.cacheId(), part, true));
+ ctx.io().removePerTopicHandler(new GridAtomicRequestTopic(ctx.cacheId(), part, false));
+ }
}
}
@@ -1033,7 +1045,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
) {
GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(ctx.cacheId(),
nodeId,
- req.futureVersion());
+ req.futureVersion(),
+ req.partition());
List<KeyCacheObject> keys = req.keys();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index 93c20da..c05f4c6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -73,7 +73,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
/** Mappings. */
@GridToStringInclude
- private ConcurrentMap<MappingKey, GridDhtAtomicUpdateRequest> mappings = new ConcurrentHashMap8<>();
+ private ConcurrentMap<GridAtomicMappingKey, GridDhtAtomicUpdateRequest> mappings = new ConcurrentHashMap8<>();
/** Entries with readers. */
private Map<KeyCacheObject, GridDhtCacheEntry> nearReadersEntries;
@@ -142,8 +142,8 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
/** {@inheritDoc} */
@Override public Collection<? extends ClusterNode> nodes() {
- return F.view(F.viewReadOnly(mappings.keySet(), new C1<MappingKey, ClusterNode>() {
- @Override public ClusterNode apply(MappingKey mappingKey) {
+ return F.view(F.viewReadOnly(mappings.keySet(), new C1<GridAtomicMappingKey, ClusterNode>() {
+ @Override public ClusterNode apply(GridAtomicMappingKey mappingKey) {
return cctx.kernalContext().discovery().node(mappingKey.nodeId());
}
}), F.notNull());
@@ -154,15 +154,15 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
if (log.isDebugEnabled())
log.debug("Processing node leave event [fut=" + this + ", nodeId=" + nodeId + ']');
- Collection<MappingKey> mappingKeys = new ArrayList<>(mappings.size());
+ Collection<GridAtomicMappingKey> mappingKeys = new ArrayList<>(mappings.size());
- for (MappingKey mappingKey : mappings.keySet()) {
+ for (GridAtomicMappingKey mappingKey : mappings.keySet()) {
if (mappingKey.nodeId().equals(nodeId))
mappingKeys.add(mappingKey);
}
if (!mappingKeys.isEmpty()) {
- for (MappingKey mappingKey : mappingKeys)
+ for (GridAtomicMappingKey mappingKey : mappingKeys)
mappings.remove(mappingKey);
checkComplete();
@@ -234,7 +234,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
for (ClusterNode node : dhtNodes) {
UUID nodeId = node.id();
- MappingKey mappingKey = new MappingKey(nodeId, part);
+ GridAtomicMappingKey mappingKey = new GridAtomicMappingKey(nodeId, part);
if (!nodeId.equals(cctx.localNodeId())) {
GridDhtAtomicUpdateRequest updateReq = mappings.get(mappingKey);
@@ -287,7 +287,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
int part = cctx.config().isAtomicOrderedUpdates() ? entry.partition() : -1;
for (UUID nodeId : readers) {
- MappingKey mappingKey = new MappingKey(nodeId, part);
+ GridAtomicMappingKey mappingKey = new GridAtomicMappingKey(nodeId, part);
GridDhtAtomicUpdateRequest updateReq = mappings.get(mappingKey);
@@ -345,8 +345,8 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
*/
public void map() {
if (!mappings.isEmpty()) {
- for (Map.Entry<MappingKey, GridDhtAtomicUpdateRequest> e : mappings.entrySet()) {
- MappingKey mappingKey = e.getKey();
+ for (Map.Entry<GridAtomicMappingKey, GridDhtAtomicUpdateRequest> e : mappings.entrySet()) {
+ GridAtomicMappingKey mappingKey = e.getKey();
GridDhtAtomicUpdateRequest req = e.getValue();
UUID nodeId = mappingKey.nodeId();
@@ -359,7 +359,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
log.debug("Sending DHT atomic update request [nodeId=" + nodeId + ", req=" + req + ']');
if (part >= 0) {
- Object topic = CU.partitionMessageTopic(cctx, part);
+ Object topic = new GridAtomicRequestTopic(cctx.cacheId(), part, false);
cctx.io().sendOrderedMessage(nodeId, topic, req, cctx.ioPolicy(),
2 * cctx.gridConfig().getNetworkTimeout());
@@ -429,7 +429,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
}
}
- mappings.remove(new MappingKey(nodeId, updateRes.partition()));
+ mappings.remove(new GridAtomicMappingKey(nodeId, updateRes.partition()));
checkComplete();
}
@@ -445,7 +445,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
log.debug("Received deferred DHT atomic update future result [nodeId=" + nodeId + ']');
for (Integer part : res.partitions())
- mappings.remove(new MappingKey(nodeId, part));
+ mappings.remove(new GridAtomicMappingKey(nodeId, part));
checkComplete();
}
@@ -468,67 +468,4 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
return S.toString(GridDhtAtomicUpdateFuture.class, this);
}
- /**
- * Mapping Key.
- */
- private static class MappingKey {
- /** Node ID. */
- private final UUID nodeId;
-
- /** Partition. */
- private final int part;
-
- /**
- * @param nodeId Node ID.
- * @param part Partition.
- */
- MappingKey(UUID nodeId, int part) {
- assert nodeId != null;
- assert part >= -1 : part;
-
- this.nodeId = nodeId;
- this.part = part;
- }
-
- /**
- * @return Node ID.
- */
- UUID nodeId() {
- return nodeId;
- }
-
- /**
- * @return Partition.
- */
- int partition() {
- return part;
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
-
- if (o == null || getClass() != o.getClass())
- return false;
-
- MappingKey key = (MappingKey)o;
-
- return nodeId.equals(key.nodeId) && part == key.part;
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- int res = nodeId.hashCode();
-
- res = 31 * res + part;
-
- return res;
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(MappingKey.class, this);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/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 4c8a161..4642b1a 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
@@ -90,7 +90,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
/** Mappings. */
@GridToStringInclude
- private ConcurrentMap<UUID, GridNearAtomicUpdateRequest> mappings;
+ private ConcurrentMap<GridAtomicMappingKey, GridNearAtomicUpdateRequest> mappings;
/** Error. */
private volatile CachePartialUpdateCheckedException err;
@@ -246,7 +246,11 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
/** {@inheritDoc} */
@Override public Collection<? extends ClusterNode> nodes() {
- return F.view(F.viewReadOnly(mappings.keySet(), U.id2Node(cctx.kernalContext())), F.notNull());
+ return F.view(F.viewReadOnly(mappings.keySet(), new C1<GridAtomicMappingKey, ClusterNode>() {
+ @Override public ClusterNode apply(GridAtomicMappingKey mappingKey) {
+ return cctx.kernalContext().discovery().node(mappingKey.nodeId());
+ }
+ }), F.notNull());
}
/**
@@ -283,13 +287,24 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
return false;
}
- GridNearAtomicUpdateRequest req = mappings.get(nodeId);
+ Collection<GridAtomicMappingKey> mappingKeys = new ArrayList<>(mappings.size());
+ Collection<KeyCacheObject> failedKeys = new ArrayList<>();
+
+ for (Map.Entry<GridAtomicMappingKey, GridNearAtomicUpdateRequest> e : mappings.entrySet()) {
+ if (e.getKey().nodeId().equals(nodeId)) {
+ mappingKeys.add(e.getKey());
+
+ failedKeys.addAll(e.getValue().keys());
+ }
+ }
- if (req != null) {
- addFailedKeys(req.keys(), new ClusterTopologyCheckedException("Primary node left grid before response is " +
- "received: " + nodeId));
+ if (!mappingKeys.isEmpty()) {
+ if (!failedKeys.isEmpty())
+ addFailedKeys(failedKeys, new ClusterTopologyCheckedException("Primary node left grid before " +
+ "response is received: " + nodeId));
- mappings.remove(nodeId);
+ for (GridAtomicMappingKey key : mappingKeys)
+ mappings.remove(key);
checkComplete();
@@ -529,7 +544,9 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
}
}
else {
- GridNearAtomicUpdateRequest req = mappings.get(nodeId);
+ GridAtomicMappingKey mappingKey = new GridAtomicMappingKey(nodeId, res.partition());
+
+ GridNearAtomicUpdateRequest req = mappings.get(mappingKey);
if (req != null) { // req can be null if onResult is being processed concurrently with onNodeLeft.
updateNear(req, res);
@@ -547,7 +564,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
opRes = ret;
}
- mappings.remove(nodeId);
+ mappings.remove(mappingKey);
}
checkComplete();
@@ -763,7 +780,11 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
if (op != TRANSFORM)
val = cctx.toCacheObject(val);
- ClusterNode primary = cctx.affinity().primary(cacheKey, topVer);
+ int part = cctx.affinity().partition(cacheKey);
+ ClusterNode primary = cctx.affinity().primary(part, topVer);
+
+ if (!ccfg.isAtomicOrderedUpdates())
+ part = -1;
if (primary == null) {
onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache (all partition nodes " +
@@ -789,7 +810,8 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
subjId,
taskNameHash,
skipStore,
- cctx.kernalContext().clientNode());
+ cctx.kernalContext().clientNode(),
+ part);
req.addUpdateEntry(cacheKey,
val,
@@ -805,7 +827,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
}
// Optimize mapping for single key.
- mapSingle(primary.id(), req);
+ mapSingle(new GridAtomicMappingKey(primary.id(), part), req);
return;
}
@@ -825,13 +847,18 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
if (conflictRmvVals != null)
conflictRmvValsIt = conflictRmvVals.iterator();
- Map<UUID, GridNearAtomicUpdateRequest> pendingMappings = new HashMap<>(topNodes.size(), 1.0f);
+ Map<GridAtomicMappingKey, GridNearAtomicUpdateRequest> pendingMappings = new HashMap<>(topNodes.size(), 1.0f);
// Must do this in synchronized block because we need to atomically remove and add mapping.
// Otherwise checkComplete() may see empty intermediate state.
synchronized (this) {
- if (oldNodeId != null)
- removeMapping(oldNodeId);
+ if (oldNodeId != null) {
+ // TODO: IGNITE-104 - Try to avoid iteration.
+ for (Map.Entry<GridAtomicMappingKey, GridNearAtomicUpdateRequest> e : mappings.entrySet()) {
+ if (e.getKey().nodeId().equals(oldNodeId))
+ mappings.remove(e.getKey());
+ }
+ }
// For fastMap mode wait for all responses before remapping.
if (remap && fastMap && !mappings.isEmpty()) {
@@ -901,7 +928,10 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
if (op != TRANSFORM)
val = cctx.toCacheObject(val);
- Collection<ClusterNode> affNodes = mapKey(cacheKey, topVer, fastMap);
+ T2<Integer, Collection<ClusterNode>> t = mapKey(cacheKey, topVer, fastMap);
+
+ int part = ccfg.isAtomicOrderedUpdates() ? t.get1() : -1;
+ Collection<ClusterNode> affNodes = t.get2();
if (affNodes.isEmpty()) {
onDone(new ClusterTopologyServerNotFoundException("Failed to map keys for cache " +
@@ -922,7 +952,9 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
UUID nodeId = affNode.id();
- GridNearAtomicUpdateRequest mapped = pendingMappings.get(nodeId);
+ GridAtomicMappingKey mappingKey = new GridAtomicMappingKey(nodeId, part);
+
+ GridNearAtomicUpdateRequest mapped = pendingMappings.get(mappingKey);
if (mapped == null) {
mapped = new GridNearAtomicUpdateRequest(
@@ -942,11 +974,12 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
subjId,
taskNameHash,
skipStore,
- cctx.kernalContext().clientNode());
+ cctx.kernalContext().clientNode(),
+ part);
- pendingMappings.put(nodeId, mapped);
+ pendingMappings.put(mappingKey, mapped);
- GridNearAtomicUpdateRequest old = mappings.put(nodeId, mapped);
+ GridNearAtomicUpdateRequest old = mappings.put(mappingKey, mapped);
assert old == null || (old != null && remap) :
"Invalid mapping state [old=" + old + ", remap=" + remap + ']';
@@ -964,7 +997,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
}
if ((single == null || single) && pendingMappings.size() == 1) {
- Map.Entry<UUID, GridNearAtomicUpdateRequest> entry = F.first(pendingMappings.entrySet());
+ Map.Entry<GridAtomicMappingKey, GridNearAtomicUpdateRequest> entry = F.first(pendingMappings.entrySet());
single = true;
@@ -987,31 +1020,35 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
* @param fastMap Flag indicating whether mapping is performed for fast-circuit update.
* @return Collection of nodes to which key is mapped.
*/
- private Collection<ClusterNode> mapKey(
+ private T2<Integer, Collection<ClusterNode>> mapKey(
KeyCacheObject key,
AffinityTopologyVersion topVer,
boolean fastMap
) {
GridCacheAffinityManager affMgr = cctx.affinity();
+ int part = affMgr.partition(key);
+
// If we can send updates in parallel - do it.
- return fastMap ?
- cctx.topology().nodes(affMgr.partition(key), topVer) :
- Collections.singletonList(affMgr.primary(key, topVer));
+ Collection<ClusterNode> nodes = fastMap ?
+ cctx.topology().nodes(part, topVer) :
+ Collections.singletonList(affMgr.primary(part, topVer));
+
+ return new T2<>(part, nodes);
}
/**
* Maps future to single node.
*
- * @param nodeId Node ID.
+ * @param mappingKey Mapping key.
* @param req Request.
*/
- private void mapSingle(UUID nodeId, GridNearAtomicUpdateRequest req) {
- singleNodeId = nodeId;
+ private void mapSingle(GridAtomicMappingKey mappingKey, GridNearAtomicUpdateRequest req) {
+ singleNodeId = mappingKey.nodeId();
singleReq = req;
- if (cctx.localNodeId().equals(nodeId)) {
- cache.updateAllAsyncInternal(nodeId, req,
+ if (cctx.localNodeId().equals(mappingKey.nodeId())) {
+ cache.updateAllAsyncInternal(mappingKey.nodeId(), req,
new CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse>() {
@Override public void apply(GridNearAtomicUpdateRequest req,
GridNearAtomicUpdateResponse res) {
@@ -1026,7 +1063,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
if (log.isDebugEnabled())
log.debug("Sending near atomic update request [nodeId=" + req.nodeId() + ", req=" + req + ']');
- cctx.io().send(req.nodeId(), req, cctx.ioPolicy());
+ sendRequest(mappingKey, req);
if (syncMode == FULL_ASYNC && cctx.config().getAtomicWriteOrderMode() == PRIMARY)
onDone(new GridCacheReturn(cctx, true, null, true));
@@ -1042,34 +1079,37 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
*
* @param mappings Mappings to send.
*/
- private void doUpdate(Map<UUID, GridNearAtomicUpdateRequest> mappings) {
+ private void doUpdate(Map<GridAtomicMappingKey, GridNearAtomicUpdateRequest> mappings) {
UUID locNodeId = cctx.localNodeId();
- GridNearAtomicUpdateRequest locUpdate = null;
+ Collection<GridNearAtomicUpdateRequest> locUpdates = null;
// Send messages to remote nodes first, then run local update.
- for (GridNearAtomicUpdateRequest req : mappings.values()) {
+ for (Map.Entry<GridAtomicMappingKey, GridNearAtomicUpdateRequest> e : mappings.entrySet()) {
+ GridAtomicMappingKey mappingKey = e.getKey();
+ GridNearAtomicUpdateRequest req = e.getValue();
+
if (locNodeId.equals(req.nodeId())) {
- assert locUpdate == null : "Cannot have more than one local mapping [locUpdate=" + locUpdate +
- ", req=" + req + ']';
+ if (locUpdates == null)
+ locUpdates = new ArrayList<>(mappings.size());
- locUpdate = req;
+ locUpdates.add(req);
}
else {
try {
if (log.isDebugEnabled())
log.debug("Sending near atomic update request [nodeId=" + req.nodeId() + ", req=" + req + ']');
- cctx.io().send(req.nodeId(), req, cctx.ioPolicy());
+ sendRequest(mappingKey, req);
}
- catch (IgniteCheckedException e) {
- addFailedKeys(req.keys(), e);
+ catch (IgniteCheckedException ex) {
+ addFailedKeys(req.keys(), ex);
- removeMapping(req.nodeId());
+ removeMapping(mappingKey);
}
if (syncMode == PRIMARY_SYNC && !req.hasPrimary())
- removeMapping(req.nodeId());
+ removeMapping(mappingKey);
}
}
@@ -1077,28 +1117,52 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
// In FULL_ASYNC mode always return (null, true).
opRes = new GridCacheReturn(cctx, true, null, true);
- if (locUpdate != null) {
- cache.updateAllAsyncInternal(cctx.localNodeId(), locUpdate,
- new CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse>() {
- @Override public void apply(GridNearAtomicUpdateRequest req,
- GridNearAtomicUpdateResponse res) {
- assert res.futureVersion().equals(futVer) : futVer;
+ if (locUpdates != null) {
+ for (GridNearAtomicUpdateRequest locUpdate : locUpdates) {
+ cache.updateAllAsyncInternal(cctx.localNodeId(), locUpdate,
+ new CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse>() {
+ @Override public void apply(GridNearAtomicUpdateRequest req,
+ GridNearAtomicUpdateResponse res) {
+ assert res.futureVersion().equals(futVer) : futVer;
- onResult(res.nodeId(), res);
- }
- });
+ onResult(res.nodeId(), res);
+ }
+ });
+ }
}
checkComplete();
}
/**
+ * Sends request.
+ *
+ * @param mappingKey Mapping key.
+ * @param req Update request.
+ * @throws IgniteCheckedException In case of error.
+ */
+ private void sendRequest(GridAtomicMappingKey mappingKey, GridNearAtomicUpdateRequest req)
+ throws IgniteCheckedException {
+ if (mappingKey.partition() >= 0) {
+ Object topic = new GridAtomicRequestTopic(cctx.cacheId(), mappingKey.partition(), true);
+
+ cctx.io().sendOrderedMessage(mappingKey.nodeId(), topic, req, cctx.ioPolicy(),
+ 2 * cctx.gridConfig().getNetworkTimeout());
+ }
+ else {
+ assert mappingKey.partition() == -1;
+
+ cctx.io().send(req.nodeId(), req, cctx.ioPolicy());
+ }
+ }
+
+ /**
* Removes mapping from future mappings map.
*
- * @param nodeId Node ID to remove mapping for.
+ * @param mappingKey Mapping key.
*/
- private void removeMapping(UUID nodeId) {
- mappings.remove(nodeId);
+ private void removeMapping(GridAtomicMappingKey mappingKey) {
+ mappings.remove(mappingKey);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/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 86c5ab8..b3075c4 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
@@ -135,6 +135,9 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
/** */
private boolean clientReq;
+ /** Partition. */
+ private int part;
+
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -162,6 +165,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
* @param taskNameHash Task name hash code.
* @param skipStore Skip write-through to a persistent storage.
* @param clientReq Client node request flag.
+ * @param part Partition.
*/
public GridNearAtomicUpdateRequest(
int cacheId,
@@ -180,7 +184,8 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
@Nullable UUID subjId,
int taskNameHash,
boolean skipStore,
- boolean clientReq
+ boolean clientReq,
+ int part
) {
this.cacheId = cacheId;
this.nodeId = nodeId;
@@ -200,6 +205,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
this.taskNameHash = taskNameHash;
this.skipStore = skipStore;
this.clientReq = clientReq;
+ this.part = part;
keys = new ArrayList<>();
}
@@ -315,6 +321,13 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
}
/**
+ * @return Partition.
+ */
+ public int partition() {
+ return part;
+ }
+
+ /**
* @param key Key to add.
* @param val Optional update value.
* @param conflictTtl Conflict TTL (optional).
@@ -666,54 +679,60 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
writer.incrementState();
case 16:
- if (!writer.writeBoolean("retval", retval))
+ if (!writer.writeInt("part", part))
return false;
writer.incrementState();
case 17:
- if (!writer.writeBoolean("skipStore", skipStore))
+ if (!writer.writeBoolean("retval", retval))
return false;
writer.incrementState();
case 18:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeBoolean("skipStore", skipStore))
return false;
writer.incrementState();
case 19:
- if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 20:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
return false;
writer.incrementState();
case 21:
- if (!writer.writeBoolean("topLocked", topLocked))
+ if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
case 22:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeBoolean("topLocked", topLocked))
return false;
writer.incrementState();
case 23:
- if (!writer.writeMessage("updateVer", updateVer))
+ if (!writer.writeMessage("topVer", topVer))
return false;
writer.incrementState();
case 24:
+ if (!writer.writeMessage("updateVer", updateVer))
+ return false;
+
+ writer.incrementState();
+
+ case 25:
if (!writer.writeCollection("vals", vals, MessageCollectionItemType.MSG))
return false;
@@ -844,7 +863,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
case 16:
- retval = reader.readBoolean("retval");
+ part = reader.readInt("part");
if (!reader.isLastRead())
return false;
@@ -852,7 +871,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
case 17:
- skipStore = reader.readBoolean("skipStore");
+ retval = reader.readBoolean("retval");
if (!reader.isLastRead())
return false;
@@ -860,7 +879,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
case 18:
- subjId = reader.readUuid("subjId");
+ skipStore = reader.readBoolean("skipStore");
if (!reader.isLastRead())
return false;
@@ -868,6 +887,14 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
case 19:
+ subjId = reader.readUuid("subjId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 20:
byte syncModeOrd;
syncModeOrd = reader.readByte("syncMode");
@@ -879,7 +906,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
- case 20:
+ case 21:
taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
@@ -887,7 +914,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
- case 21:
+ case 22:
topLocked = reader.readBoolean("topLocked");
if (!reader.isLastRead())
@@ -895,7 +922,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
- case 22:
+ case 23:
topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
@@ -903,7 +930,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
- case 23:
+ case 24:
updateVer = reader.readMessage("updateVer");
if (!reader.isLastRead())
@@ -911,7 +938,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
reader.incrementState();
- case 24:
+ case 25:
vals = reader.readCollection("vals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -931,7 +958,7 @@ public class GridNearAtomicUpdateRequest extends GridCacheMessage implements Gri
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 25;
+ return 26;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
index 8e1bee2..e2d33d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java
@@ -92,6 +92,9 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
/** Near expire times. */
private GridLongList nearExpireTimes;
+ /** Partition. */
+ private int part;
+
/**
* Empty constructor required by {@link Externalizable}.
*/
@@ -103,11 +106,13 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
* @param cacheId Cache ID.
* @param nodeId Node ID this reply should be sent to.
* @param futVer Future version.
+ * @param part Partition.
*/
- public GridNearAtomicUpdateResponse(int cacheId, UUID nodeId, GridCacheVersion futVer) {
+ public GridNearAtomicUpdateResponse(int cacheId, UUID nodeId, GridCacheVersion futVer, int part) {
this.cacheId = cacheId;
this.nodeId = nodeId;
this.futVer = futVer;
+ this.part = part;
}
/** {@inheritDoc} */
@@ -138,7 +143,7 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
/**
* Sets update error.
- * @param err Exception.
+ * @param err
*/
public void error(IgniteCheckedException err){
this.err = err;
@@ -188,6 +193,13 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
}
/**
+ * @return Partition.
+ */
+ public int partition() {
+ return part;
+ }
+
+ /**
* Adds value to be put in near cache on originating node.
*
* @param keyIdx Key index.
@@ -485,12 +497,18 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
writer.incrementState();
case 12:
- if (!writer.writeCollection("remapKeys", remapKeys, MessageCollectionItemType.MSG))
+ if (!writer.writeInt("part", part))
return false;
writer.incrementState();
case 13:
+ if (!writer.writeCollection("remapKeys", remapKeys, MessageCollectionItemType.MSG))
+ return false;
+
+ writer.incrementState();
+
+ case 14:
if (!writer.writeMessage("ret", ret))
return false;
@@ -585,7 +603,7 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
reader.incrementState();
case 12:
- remapKeys = reader.readCollection("remapKeys", MessageCollectionItemType.MSG);
+ part = reader.readInt("part");
if (!reader.isLastRead())
return false;
@@ -593,6 +611,14 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
reader.incrementState();
case 13:
+ remapKeys = reader.readCollection("remapKeys", MessageCollectionItemType.MSG);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 14:
ret = reader.readMessage("ret");
if (!reader.isLastRead())
@@ -612,7 +638,7 @@ public class GridNearAtomicUpdateResponse extends GridCacheMessage implements Gr
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 14;
+ return 15;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemandPool.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemandPool.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemandPool.java
index a6e6c4d..37824eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemandPool.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemandPool.java
@@ -599,7 +599,7 @@ public class GridDhtPartitionDemandPool {
if (isCancelled() || topologyChanged())
return missed;
- cctx.io().addOrderedHandler(d.topic(), new CI2<UUID, GridDhtPartitionSupplyMessage>() {
+ cctx.io().addPerTopicHandler(d.topic(), new CI2<UUID, GridDhtPartitionSupplyMessage>() {
@Override public void apply(UUID nodeId, GridDhtPartitionSupplyMessage msg) {
addMessage(new SupplyMessage(nodeId, msg));
}
@@ -641,7 +641,7 @@ public class GridDhtPartitionDemandPool {
growTimeout(timeout);
// Ordered listener was removed if timeout expired.
- cctx.io().removeOrderedHandler(d.topic());
+ cctx.io().removePerTopicHandler(d.topic());
// Must create copy to be able to work with IO manager thread local caches.
d = new GridDhtPartitionDemandMessage(d, remaining);
@@ -650,13 +650,12 @@ public class GridDhtPartitionDemandPool {
d.topic(topic(++cntr));
// Create new ordered listener.
- cctx.io().addOrderedHandler(d.topic(),
- new CI2<UUID, GridDhtPartitionSupplyMessage>() {
- @Override public void apply(UUID nodeId,
- GridDhtPartitionSupplyMessage msg) {
+ cctx.io().addPerTopicHandler(d.topic(), new CI2<UUID, GridDhtPartitionSupplyMessage>() {
+ @Override public void apply(UUID nodeId, GridDhtPartitionSupplyMessage msg) {
addMessage(new SupplyMessage(nodeId, msg));
}
- });
+ }
+ );
// Resend message with larger timeout.
retry = true;
@@ -800,7 +799,7 @@ public class GridDhtPartitionDemandPool {
return missed;
}
finally {
- cctx.io().removeOrderedHandler(d.topic());
+ cctx.io().removePerTopicHandler(d.topic());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
index 316713f..a530a5d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java
@@ -556,11 +556,11 @@ public class GridCacheDistributedQueryManager<K, V> extends GridCacheQueryManage
final Object topic = topic(cctx.nodeId(), req.id());
- cctx.io().addOrderedHandler(topic, resHnd);
+ cctx.io().addPerTopicHandler(topic, resHnd);
fut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> fut) {
- cctx.io().removeOrderedHandler(topic);
+ cctx.io().removePerTopicHandler(topic);
}
});
@@ -665,11 +665,11 @@ public class GridCacheDistributedQueryManager<K, V> extends GridCacheQueryManage
final Object topic = topic(cctx.nodeId(), req.id());
- cctx.io().addOrderedHandler(topic, resHnd);
+ cctx.io().addPerTopicHandler(topic, resHnd);
fut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> fut) {
- cctx.io().removeOrderedHandler(topic);
+ cctx.io().removePerTopicHandler(topic);
}
});
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2d16d99f/modules/core/src/main/resources/META-INF/classnames.properties
----------------------------------------------------------------------
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index df4873a..3ec1d07 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -529,6 +529,7 @@ org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFu
org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture$MiniFuture$1
org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture$MiniFuture$2
org.apache.ignite.internal.processors.cache.distributed.dht.GridPartitionedGetFuture$MiniFuture$2$1
+org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridAtomicRequestTopic
org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache
org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache$10
org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache$11
@@ -806,7 +807,6 @@ org.apache.ignite.internal.processors.closure.GridClosureProcessor$T5
org.apache.ignite.internal.processors.closure.GridClosureProcessor$T6
org.apache.ignite.internal.processors.closure.GridClosureProcessor$T7
org.apache.ignite.internal.processors.closure.GridClosureProcessor$T8
-org.apache.ignite.internal.processors.closure.GridClosureProcessor$T9
org.apache.ignite.internal.processors.closure.GridClosureProcessor$TaskNoReduceAdapter
org.apache.ignite.internal.processors.closure.GridPeerDeployAwareTaskAdapter
org.apache.ignite.internal.processors.continuous.GridContinuousHandler