ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [01/16] incubator-ignite git commit: # ignite-694
Date Thu, 09 Apr 2015 09:23:32 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-707 17c94b6b7 -> 2412422b2


# 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-707
Commit: db754aba1fcf9d762dba9b96e5333afde7efa0dd
Parents: 6553549
Author: sboikov <sboikov@gridgain.com>
Authored: Tue Apr 7 17:17:08 2015 +0300
Committer: sboikov <sboikov@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;
     }
 }


Mime
View raw message