ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject incubator-ignite git commit: # ignite-23 avoid exchange for client node join/leave
Date Mon, 18 May 2015 14:50:24 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-23 [created] a2a6f31fe


# ignite-23 avoid exchange for client node join/leave


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/a2a6f31f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/a2a6f31f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/a2a6f31f

Branch: refs/heads/ignite-23
Commit: a2a6f31fe4207dbb3cd2146c3f22a5c99d83fc5b
Parents: 5c8591c
Author: sboikov <sboikov@gridgain.com>
Authored: Mon May 18 17:50:06 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Mon May 18 17:50:06 2015 +0300

----------------------------------------------------------------------
 .../affinity/GridAffinityAssignmentCache.java   |  27 +++
 .../cache/GridCacheAffinityManager.java         |  12 ++
 .../GridDhtPartitionsExchangeFuture.java        | 100 +++++++---
 .../IgniteCacheClientNodeExchangeTest.java      | 184 +++++++++++++++++++
 .../testsuites/IgniteCacheTestSuite2.java       |   2 +
 5 files changed, 300 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a2a6f31f/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
index eccd9f9..18ac65a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
@@ -103,6 +103,32 @@ public class GridAffinityAssignmentCache {
     }
 
     /**
+     * @param node Node.
+     * @param topVer Topology version.
+     */
+    public void clientNodeTopologyChange(ClusterNode node, AffinityTopologyVersion topVer)
{
+        GridAffinityAssignment assignment = head.get();
+
+        assert assignment.primaryPartitions(node.id()).isEmpty() : node;
+        assert assignment.backupPartitions(node.id()).isEmpty() : node;
+
+        GridAffinityAssignment assignmentCpy = new GridAffinityAssignment(topVer, assignment.assignment());
+
+        affCache.put(topVer, assignmentCpy);
+        head.set(assignmentCpy);
+
+        for (Map.Entry<AffinityTopologyVersion, AffinityReadyFuture> entry : readyFuts.entrySet())
{
+            if (entry.getKey().compareTo(topVer) <= 0) {
+                if (log.isDebugEnabled())
+                    log.debug("Completing topology ready future (use previous affinity) "
+
+                        "[locNodeId=" + ctx.localNodeId() + ", futVer=" + entry.getKey()
+ ", topVer=" + topVer + ']');
+
+                entry.getValue().onDone(topVer);
+            }
+        }
+    }
+
+    /**
      * Initializes affinity with given topology version and assignment. The assignment is
calculated on remote nodes
      * and brought to local node on partition map exchange.
      *
@@ -422,6 +448,7 @@ public class GridAffinityAssignmentCache {
 
         /**
          *
+         * @param reqTopVer Required topology version.
          */
         private AffinityReadyFuture(AffinityTopologyVersion reqTopVer) {
             this.reqTopVer = reqTopVer;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a2a6f31f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index fe7efd5..6541e9a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -140,6 +140,7 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter
{
      *
      * @param topVer Topology version to calculate affinity for.
      * @param discoEvt Discovery event that causes this topology change.
+     * @return Affinity assignments.
      */
     public List<List<ClusterNode>> calculateAffinity(AffinityTopologyVersion
topVer, DiscoveryEvent discoEvt) {
         assert !cctx.isLocal();
@@ -148,6 +149,17 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter
{
     }
 
     /**
+     *
+     * @param node Node.
+     * @param topVer Topology version.
+     */
+    public void clientNodeTopologyChange(ClusterNode node, AffinityTopologyVersion topVer)
{
+        assert !cctx.isLocal();
+
+        aff.clientNodeTopologyChange(node, topVer);
+    }
+
+    /**
      * @return Partition count.
      */
     public int partitions() {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a2a6f31f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 599e391..f4dcf3b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -44,6 +44,7 @@ import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 import java.util.concurrent.locks.*;
 
+import static org.apache.ignite.events.EventType.*;
 import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
 
 /**
@@ -146,6 +147,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
     /** Dynamic cache change requests. */
     private Collection<DynamicCacheChangeRequest> reqs;
 
+    /** Cache validation results. */
     private volatile Map<Integer, Boolean> cacheValidRes;
 
     /**
@@ -200,6 +202,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
      * @param cctx Cache context.
      * @param busyLock Busy lock.
      * @param exchId Exchange ID.
+     * @param reqs Cache change requests.
      */
     public GridDhtPartitionsExchangeFuture(
         GridCacheSharedContext cctx,
@@ -322,7 +325,8 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
     }
 
     /**
-     * Rechecks topology.
+     * @param cacheCtx Cache context.
+     * @throws IgniteCheckedException If failed.
      */
     private void initTopology(GridCacheContext cacheCtx) throws IgniteCheckedException {
         if (stopping(cacheCtx.cacheId()))
@@ -471,6 +475,62 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
                 // will return corresponding nodes.
                 U.await(evtLatch);
 
+                if (!dummy && !forcePreload && F.isEmpty(reqs)) { // If exchange
initiated by node join or leave.
+                    assert discoEvt != null;
+
+                    int type = discoEvt.type();
+
+                    assert type == EVT_NODE_JOINED || type == EVT_NODE_LEFT || type == EVT_NODE_FAILED
: discoEvt;
+
+                    ClusterNode node = discoEvt.eventNode();
+
+                    if (!node.isLocal()) {
+                        boolean affNode = false;
+
+                        for (GridCacheContext cacheCtx : cctx.cacheContexts()) {
+                            if (cacheCtx.isLocal())
+                                continue;
+
+                            if (CU.affinityNode(node, cacheCtx.config().getNodeFilter()))
{
+                                affNode = true;
+
+                                break;
+                            }
+                        }
+
+                        if (!affNode) {
+                            for (GridCacheContext cacheCtx : cctx.cacheContexts()) {
+                                if (cacheCtx.isLocal())
+                                    continue;
+
+                                cacheCtx.affinity().clientNodeTopologyChange(node, exchId.topologyVersion());
+
+                                GridDhtPartitionTopology top = cacheCtx.topology();
+
+                                GridDhtPartitionMap parts = top.partitions(node.id());
+
+                                assert parts == null || parts.size() == 0 : parts;
+
+                                top.updateTopologyVersion(exchId, this, -1, stopping(cacheCtx.cacheId()));
+                            }
+
+                            if (!exchId.isLeft()) {
+                                rmtNodes = new ConcurrentLinkedQueue<>(F.asList(node));
+
+                                rmtIds = F.asList(node.id());
+                            }
+
+                            ready.set(true);
+
+                            initFut.onDone(true);
+
+                            onDone(exchId.topologyVersion());
+
+                            return;
+                        }
+                    }
+                }
+
                 startCaches();
 
                 for (GridCacheContext cacheCtx : cctx.cacheContexts()) {
@@ -484,8 +544,8 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
 
                 List<String> cachesWithoutNodes = null;
 
-                for (String name : cctx.cache().cacheNames()) {
-                    if (exchId.isLeft()) {
+                if (exchId.isLeft()) {
+                    for (String name : cctx.cache().cacheNames()) {
                         if (cctx.discovery().cacheAffinityNodes(name, topologyVersion()).isEmpty())
{
                             if (cachesWithoutNodes == null)
                                 cachesWithoutNodes = new ArrayList<>();
@@ -521,7 +581,8 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
                 }
 
                 if (cachesWithoutNodes != null) {
-                    StringBuilder sb = new StringBuilder("All server nodes for the following
caches have left the cluster: ");
+                    StringBuilder sb =
+                        new StringBuilder("All server nodes for the following caches have
left the cluster: ");
 
                     for (int i = 0; i < cachesWithoutNodes.size(); i++) {
                         String cache = cachesWithoutNodes.get(i);
@@ -666,36 +727,25 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
             if (log.isDebugEnabled())
                 log.debug("Initialized future: " + this);
 
-            if (canSkipExchange())
-                onDone(exchId.topologyVersion());
+            // If this node is not oldest.
+            if (!oldestNode.get().id().equals(cctx.localNodeId()))
+                sendPartitions();
             else {
-                // If this node is not oldest.
-                if (!oldestNode.get().id().equals(cctx.localNodeId()))
-                    sendPartitions();
-                else {
-                    boolean allReceived = allReceived();
-
-                    if (allReceived && replied.compareAndSet(false, true)) {
-                        if (spreadPartitions())
-                            onDone(exchId.topologyVersion());
-                    }
-                }
+                boolean allReceived = allReceived();
 
-                scheduleRecheck();
+                if (allReceived && replied.compareAndSet(false, true)) {
+                    if (spreadPartitions())
+                        onDone(exchId.topologyVersion());
+                }
             }
+
+            scheduleRecheck();
         }
         else
             assert false : "Skipped init future: " + this;
     }
 
     /**
-     * @return {@code True} if no distributed exchange is needed.
-     */
-    private boolean canSkipExchange() {
-        return false; // TODO ignite-23;
-    }
-
-    /**
      *
      */
     private void dumpPendingObjects() {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a2a6f31f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeExchangeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeExchangeTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeExchangeTest.java
new file mode 100644
index 0000000..66db3c6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeExchangeTest.java
@@ -0,0 +1,184 @@
+/*
+ * 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.cluster.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.managers.communication.*;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.*;
+import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.communication.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.eclipse.jetty.util.*;
+
+import java.util.*;
+
+/**
+ *
+ */
+public class IgniteCacheClientNodeExchangeTest extends GridCommonAbstractTest {
+    /** */
+    protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /** */
+    private boolean client;
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+        cfg.setClientMode(client);
+
+        CacheConfiguration ccfg = new CacheConfiguration();
+
+        cfg.setCacheConfiguration(ccfg);
+
+        cfg.setCommunicationSpi(new TestCommunicationSpi());
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        stopAllGrids();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testNoPartitionExchangeForClient() throws Exception {
+        Ignite ignite0 = startGrid(0);
+
+        TestCommunicationSpi spi0 = (TestCommunicationSpi)ignite0.configuration().getCommunicationSpi();
+
+        Ignite ignite1 = startGrid(1);
+
+        TestCommunicationSpi spi1 = (TestCommunicationSpi)ignite1.configuration().getCommunicationSpi();
+
+        assertEquals(0, spi0.partitionsSingleMessages().size());
+        assertEquals(1, spi0.partitionsFullMessages().size());
+
+        assertEquals(1, spi1.partitionsSingleMessages().size());
+        assertEquals(0, spi1.partitionsFullMessages().size());
+
+        spi0.reset();
+        spi1.reset();
+
+        client = true;
+
+        for (int i = 0; i < 3; i++) {
+            log.info("Start client node: " + i);
+
+            Ignite ignite2 = startGrid(2);
+
+            TestCommunicationSpi spi2 = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi();
+
+            assertEquals(0, spi0.partitionsSingleMessages().size());
+            assertEquals(1, spi0.partitionsFullMessages().size());
+
+            assertEquals(0, spi1.partitionsSingleMessages().size());
+            assertEquals(0, spi1.partitionsFullMessages().size());
+
+            assertEquals(1, spi2.partitionsSingleMessages().size());
+            assertEquals(0, spi2.partitionsFullMessages().size());
+
+            spi0.reset();
+            spi1.reset();
+            spi2.reset();
+
+            log.info("Stop client node.");
+
+            ignite2.close();
+
+            assertEquals(0, spi0.partitionsSingleMessages().size());
+            assertEquals(0, spi0.partitionsFullMessages().size());
+
+            assertEquals(0, spi1.partitionsSingleMessages().size());
+            assertEquals(0, spi1.partitionsFullMessages().size());
+        }
+    }
+
+    /**
+     * Test communication SPI.
+     */
+    private static class TestCommunicationSpi extends TcpCommunicationSpi {
+        /** */
+        private ConcurrentHashSet<GridDhtPartitionsSingleMessage> partSingleMsgs =
new ConcurrentHashSet<>();
+
+        /** */
+        private ConcurrentHashSet<GridDhtPartitionsFullMessage> partFullMsgs = new
ConcurrentHashSet<>();
+
+        /** */
+        @LoggerResource
+        private IgniteLogger log;
+
+        /** {@inheritDoc} */
+        @Override public void sendMessage(ClusterNode node, Message msg) {
+            super.sendMessage(node, msg);
+
+            Object msg0 = ((GridIoMessage)msg).message();
+
+            if (msg0 instanceof GridDhtPartitionsSingleMessage) {
+                if (((GridDhtPartitionsSingleMessage)msg0).exchangeId() != null) {
+                    log.info("Partitions message: " + msg0.getClass().getSimpleName());
+
+                    partSingleMsgs.add((GridDhtPartitionsSingleMessage) msg0);
+                }
+            }
+            else if (msg0 instanceof GridDhtPartitionsFullMessage) {
+                if (((GridDhtPartitionsFullMessage)msg0).exchangeId() != null) {
+                    log.info("Partitions message: " + msg0.getClass().getSimpleName());
+
+                    partFullMsgs.add((GridDhtPartitionsFullMessage) msg0);
+                }
+            }
+        }
+
+        /**
+         *
+         */
+        void reset() {
+            partSingleMsgs.clear();
+            partFullMsgs.clear();
+        }
+
+        /**
+         * @return Sent partitions single messages.
+         */
+        Collection<GridDhtPartitionsSingleMessage> partitionsSingleMessages() {
+            return partSingleMsgs;
+        }
+
+        /**
+         * @return Sent partitions full messages.
+         */
+        Collection<GridDhtPartitionsFullMessage> partitionsFullMessages() {
+            return partFullMsgs;
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a2a6f31f/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
index 5738778..6031dcb 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
@@ -138,6 +138,8 @@ public class IgniteCacheTestSuite2 extends TestSuite {
         suite.addTest(new TestSuite(GridCacheNearPrimarySyncSelfTest.class));
         suite.addTest(new TestSuite(GridCacheColocatedPrimarySyncSelfTest.class));
 
+        suite.addTest(new TestSuite(IgniteCacheClientNodeExchangeTest.class));
+
         return suite;
     }
 }


Mime
View raw message