You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/04/09 16:59:23 UTC
[07/50] [abbrv] incubator-ignite git commit: # ignite-694
# ignite-694
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/db754aba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/db754aba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/db754aba
Branch: refs/heads/ignite-673
Commit: db754aba1fcf9d762dba9b96e5333afde7efa0dd
Parents: 6553549
Author: sboikov <sb...@gridgain.com>
Authored: Tue Apr 7 17:17:08 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue Apr 7 17:17:08 2015 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheIoManager.java | 77 ++++++------
.../GridCachePartitionExchangeManager.java | 121 ++++++++++++++++++-
.../IgniteCacheAtomicNodeJoinTest.java | 44 +++++++
.../IgniteCacheNodeJoinAbstractTest.java | 109 +++++++++++++++++
.../distributed/IgniteCacheTxNodeJoinTest.java | 38 ++++++
.../IgniteCacheFailoverTestSuite.java | 4 +
6 files changed, 356 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/db754aba/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 7bd6726..8805853 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
@@ -23,6 +23,7 @@ import org.apache.ignite.internal.*;
import org.apache.ignite.internal.cluster.*;
import org.apache.ignite.internal.managers.communication.*;
import org.apache.ignite.internal.managers.deployment.*;
+import org.apache.ignite.internal.processors.affinity.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.lang.*;
import org.apache.ignite.internal.util.typedef.*;
@@ -76,7 +77,6 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
/** Message listener. */
private GridMessageListener lsnr = new GridMessageListener() {
- @SuppressWarnings("unchecked")
@Override public void onMessage(final UUID nodeId, Object msg) {
if (log.isDebugEnabled())
log.debug("Received unordered cache communication message [nodeId=" + nodeId +
@@ -84,44 +84,20 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
final GridCacheMessage cacheMsg = (GridCacheMessage)msg;
- int msgIdx = cacheMsg.lookupIndex();
+ AffinityTopologyVersion locAffVer = cctx.exchange().topologyVersion();
+ AffinityTopologyVersion rmtAffVer = cacheMsg.topologyVersion();
- IgniteBiInClosure<UUID, GridCacheMessage> c = null;
-
- if (msgIdx >= 0) {
- IgniteBiInClosure[] cacheClsHandlers = idxClsHandlers.get(cacheMsg.cacheId());
-
- if (cacheClsHandlers != null)
- c = cacheClsHandlers[msgIdx];
- }
-
- if (c == null)
- c = clsHandlers.get(new ListenerKey(cacheMsg.cacheId(), cacheMsg.getClass()));
-
- if (c == null) {
- if (log.isDebugEnabled())
- log.debug("Received message without registered handler (will ignore) [msg=" + msg +
- ", nodeId=" + nodeId + ']');
-
- return;
- }
-
- long locTopVer = cctx.discovery().topologyVersion();
- long rmtTopVer = cacheMsg.topologyVersion().topologyVersion();
-
- if (locTopVer < rmtTopVer) {
+ if (locAffVer.compareTo(rmtAffVer) < 0) {
if (log.isDebugEnabled())
log.debug("Received message has higher topology version [msg=" + msg +
- ", locTopVer=" + locTopVer + ", rmtTopVer=" + rmtTopVer + ']');
-
- IgniteInternalFuture<Long> topFut = cctx.discovery().topologyFuture(rmtTopVer);
+ ", locTopVer=" + locAffVer + ", rmtTopVer=" + rmtAffVer + ']');
- if (!topFut.isDone()) {
- final IgniteBiInClosure<UUID, GridCacheMessage> c0 = c;
+ IgniteInternalFuture<?> topFut = cctx.exchange().affinityReadyFuture(rmtAffVer);
- topFut.listen(new CI1<IgniteInternalFuture<Long>>() {
- @Override public void apply(IgniteInternalFuture<Long> t) {
- onMessage0(nodeId, cacheMsg, c0);
+ if (topFut != null && !topFut.isDone()) {
+ topFut.listen(new CI1<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> t) {
+ handleMessage(nodeId, cacheMsg);
}
});
@@ -129,10 +105,41 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
}
}
- onMessage0(nodeId, cacheMsg, c);
+ handleMessage(nodeId, cacheMsg);
}
};
+ /**
+ * @param nodeId Sender node ID.
+ * @param cacheMsg Message.
+ */
+ @SuppressWarnings("unchecked")
+ private void handleMessage(UUID nodeId, GridCacheMessage cacheMsg) {
+ int msgIdx = cacheMsg.lookupIndex();
+
+ IgniteBiInClosure<UUID, GridCacheMessage> c = null;
+
+ if (msgIdx >= 0) {
+ IgniteBiInClosure[] cacheClsHandlers = idxClsHandlers.get(cacheMsg.cacheId());
+
+ if (cacheClsHandlers != null)
+ c = cacheClsHandlers[msgIdx];
+ }
+
+ if (c == null)
+ c = clsHandlers.get(new ListenerKey(cacheMsg.cacheId(), cacheMsg.getClass()));
+
+ if (c == null) {
+ if (log.isDebugEnabled())
+ log.debug("Received message without registered handler (will ignore) [msg=" + cacheMsg +
+ ", nodeId=" + nodeId + ']');
+
+ return;
+ }
+
+ onMessage0(nodeId, cacheMsg, c);
+ }
+
/** {@inheritDoc} */
@Override public void start0() throws IgniteCheckedException {
retryDelay = cctx.gridConfig().getNetworkSendRetryDelay();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/db754aba/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index d2d3531..e8e3ea1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -90,6 +90,14 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
/** */
private volatile GridDhtPartitionsExchangeFuture lastInitializedFut;
+ /** */
+ private final ConcurrentMap<AffinityTopologyVersion, AffinityReadyFuture> readyFuts = new ConcurrentHashMap8<>();
+
+ /** */
+ private final AtomicReference<AffinityTopologyVersion> readyTopVer =
+ new AtomicReference<>(AffinityTopologyVersion.NONE);
+
+
/**
* Partition map futures.
* This set also contains already completed exchange futures to address race conditions when coordinator
@@ -311,6 +319,9 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
for (GridDhtPartitionsExchangeFuture f : exchFuts.values())
f.onDone(new IgniteInterruptedCheckedException("Grid is stopping: " + cctx.gridName()));
+ for (AffinityReadyFuture f : readyFuts.values())
+ f.onDone(new IgniteInterruptedCheckedException("Grid is stopping: " + cctx.gridName()));
+
U.cancel(exchWorker);
if (log.isDebugEnabled())
@@ -372,7 +383,10 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
* @return Topology version.
*/
public AffinityTopologyVersion topologyVersion() {
- return lastInitializedFut.exchangeId().topologyVersion();
+ GridDhtPartitionsExchangeFuture lastInitializedFut0 = lastInitializedFut;
+
+ return lastInitializedFut0 != null
+ ? lastInitializedFut0.exchangeId().topologyVersion() : AffinityTopologyVersion.NONE;
}
/**
@@ -383,6 +397,49 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
}
/**
+ * @param ver Topology version.
+ * @return Future or {@code null} is future is already completed.
+ */
+ @Nullable IgniteInternalFuture<?> affinityReadyFuture(AffinityTopologyVersion ver) {
+ GridDhtPartitionsExchangeFuture lastInitializedFut0 = lastInitializedFut;
+
+ if (lastInitializedFut0 != null && lastInitializedFut0.topologyVersion().compareTo(ver) >= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Return lastInitializedFut for topology ready future " +
+ "[ver=" + ver + ", fut=" + lastInitializedFut0 + ']');
+
+ return lastInitializedFut0;
+ }
+
+ AffinityTopologyVersion topVer = readyTopVer.get();
+
+ if (topVer.compareTo(ver) >= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Return finished future for topology ready future [ver=" + ver + ", topVer=" + topVer + ']');
+
+ return null;
+ }
+
+ GridFutureAdapter<AffinityTopologyVersion> fut = F.addIfAbsent(readyFuts, ver,
+ new AffinityReadyFuture(ver));
+
+ if (log.isDebugEnabled())
+ log.debug("Created topology ready future [ver=" + ver + ", fut=" + fut + ']');
+
+ topVer = readyTopVer.get();
+
+ if (topVer.compareTo(ver) >= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Completing created topology ready future " +
+ "[ver=" + topVer + ", topVer=" + topVer + ", fut=" + fut + ']');
+
+ fut.onDone(topVer);
+ }
+
+ return fut;
+ }
+
+ /**
* @return {@code true} if entered to busy state.
*/
private boolean enterBusy() {
@@ -642,8 +699,38 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
/**
* @param exchFut Exchange.
+ * @param err Error.
*/
- public void onExchangeDone(GridDhtPartitionsExchangeFuture exchFut, Throwable err) {
+ public void onExchangeDone(GridDhtPartitionsExchangeFuture exchFut, @Nullable Throwable err) {
+ if (err == null) {
+ AffinityTopologyVersion topVer = exchFut.topologyVersion();
+
+ if (log.isDebugEnabled())
+ log.debug("Exchange done [topVer=" + topVer + ", fut=" + exchFut + ']');
+
+ while (true) {
+ AffinityTopologyVersion readyVer = readyTopVer.get();
+
+ if (readyVer.compareTo(topVer) >= 0)
+ break;
+
+ if (readyTopVer.compareAndSet(readyVer, topVer))
+ break;
+ }
+
+ for (Map.Entry<AffinityTopologyVersion, AffinityReadyFuture> entry : readyFuts.entrySet()) {
+ if (entry.getKey().compareTo(topVer) <= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Completing created topology ready future " +
+ "[ver=" + topVer + ", fut=" + entry.getValue() + ']');
+
+ entry.getValue().onDone(topVer);
+ }
+ }
+ }
+ else if (log.isDebugEnabled())
+ log.debug("Exchange done with error [fut=" + exchFut + ", err=" + err + ']');
+
ExchangeFutureSet exchFuts0 = exchFuts;
if (exchFuts0 != null) {
@@ -1130,4 +1217,34 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
*/
protected abstract void onMessage(ClusterNode node, M msg);
}
+
+ /**
+ * Affinity ready future.
+ */
+ private class AffinityReadyFuture extends GridFutureAdapter<AffinityTopologyVersion> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private AffinityTopologyVersion topVer;
+
+ /**
+ * @param topVer Topology version.
+ */
+ private AffinityReadyFuture(AffinityTopologyVersion topVer) {
+ this.topVer = topVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onDone(AffinityTopologyVersion res, @Nullable Throwable err) {
+ assert res != null || err != null;
+
+ boolean done = super.onDone(res, err);
+
+ if (done)
+ readyFuts.remove(topVer, this);
+
+ return done;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/db754aba/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeJoinTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeJoinTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeJoinTest.java
new file mode 100644
index 0000000..c0818bd
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheAtomicNodeJoinTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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;
+
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.*;
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheAtomicNodeJoinTest extends IgniteCacheNodeJoinAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicWriteOrderMode atomicWriteOrderMode() {
+ return PRIMARY;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return ATOMIC;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected NearCacheConfiguration nearConfiguration() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/db754aba/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheNodeJoinAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheNodeJoinAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheNodeJoinAbstractTest.java
new file mode 100644
index 0000000..bed83ec
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheNodeJoinAbstractTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.testframework.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import static org.apache.ignite.cache.CacheMode.*;
+
+/**
+ *
+ */
+public abstract class IgniteCacheNodeJoinAbstractTest extends IgniteCacheAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int gridCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception {
+ CacheConfiguration cfg = super.cacheConfiguration(gridName);
+
+ cfg.setReadFromBackup(false); // Force remote 'get'.
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGet() throws Exception {
+ final IgniteCache<Integer, Integer> cache = jcache(0);
+
+ final int KEYS = 1000;
+
+ Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < KEYS; i++)
+ map.put(i, i);
+
+ for (int i = 0; i < 10; i++) {
+ log.info("Iteration: " + i);
+
+ cache.putAll(map);
+
+ final IgniteInternalFuture fut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ startGrid(1);
+
+ return null;
+ }
+ });
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ GridTestUtils.runMultiThreaded(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ while (!stop.get() && !fut.isDone()) {
+ for (int key = 0; key < KEYS; key++) {
+ assertNotNull(cache.get(key));
+
+ if (key % 100 == 0 && fut.isDone())
+ break;
+ }
+ }
+
+ return null;
+ }
+ }, 10, "test-get");
+
+ try {
+ fut.get(60_000);
+ }
+ finally {
+ stop.set(true);
+ }
+
+ stopGrid(1);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/db754aba/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheTxNodeJoinTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheTxNodeJoinTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheTxNodeJoinTest.java
new file mode 100644
index 0000000..d250cd6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheTxNodeJoinTest.java
@@ -0,0 +1,38 @@
+/*
+ * 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;
+
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheTxNodeJoinTest extends IgniteCacheNodeJoinAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return TRANSACTIONAL;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected NearCacheConfiguration nearConfiguration() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/db754aba/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFailoverTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFailoverTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFailoverTestSuite.java
index 9556235..da79c55 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFailoverTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFailoverTestSuite.java
@@ -19,6 +19,7 @@ package org.apache.ignite.testsuites;
import junit.framework.*;
import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.distributed.*;
import org.apache.ignite.internal.processors.cache.distributed.dht.*;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.*;
import org.apache.ignite.internal.processors.cache.distributed.near.*;
@@ -56,6 +57,9 @@ public class IgniteCacheFailoverTestSuite extends TestSuite {
//suite.addTestSuite(GridCacheColocatedFailoverSelfTest.class); TODO IGNITE-631.
//suite.addTestSuite(GridCacheReplicatedFailoverSelfTest.class); TODO IGNITE-631.
+ suite.addTestSuite(IgniteCacheAtomicNodeJoinTest.class);
+ suite.addTestSuite(IgniteCacheTxNodeJoinTest.class);
+
return suite;
}
}