ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [09/50] incubator-ignite git commit: ignite-389 Partition scan query fallback test
Date Thu, 11 Jun 2015 10:31:25 GMT
ignite-389 Partition scan query fallback test


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

Branch: refs/heads/ignite-929
Commit: 29dc7221c12db1e39a17de4471a8c5ebed4b8709
Parents: 5d6bb53
Author: agura <agura@gridgain.com>
Authored: Fri May 29 16:28:34 2015 +0300
Committer: agura <agura@gridgain.com>
Committed: Fri May 29 16:28:34 2015 +0300

----------------------------------------------------------------------
 ...CacheScanPartitionQueryFallbackSelfTest.java | 335 ++++++++++++++-----
 1 file changed, 259 insertions(+), 76 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/29dc7221/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
index 31336e6..dfa7296 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheScanPartitionQueryFallbackSelfTest.java
@@ -21,8 +21,10 @@ import org.apache.ignite.*;
 import org.apache.ignite.cache.*;
 import org.apache.ignite.cluster.*;
 import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.managers.communication.*;
 import org.apache.ignite.internal.processors.affinity.*;
+import org.apache.ignite.internal.processors.cache.distributed.dht.*;
 import org.apache.ignite.internal.processors.cache.query.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.lang.*;
@@ -32,15 +34,17 @@ import org.apache.ignite.spi.communication.tcp.*;
 import org.apache.ignite.testframework.junits.common.*;
 
 import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
 
 /**
  * Tests partition scan query fallback.
  */
 public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractTest {
     /** Grid count. */
-    private static final int GRID_CNT = 5;
+    private static final int GRID_CNT = 3;
 
-    /** Kys count. */
+    /** Keys count. */
     private static final int KEYS_CNT = 5000;
 
     /** Backups. */
@@ -49,20 +53,29 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
     /** Cache mode. */
     private CacheMode cacheMode;
 
-    /** Fallback. */
-    private boolean fallback;
+    /** Client mode. */
+    private volatile boolean clientMode;
 
-    /** Primary node id. */
-    private static volatile UUID expNodeId;
+    /** Expected first node ID. */
+    private static UUID expNodeId;
 
-    /** Fail node id. */
-    private static volatile UUID failNodeId;
+    /** Expected fallback node ID. */
+    private static UUID expFallbackNodeId;
+
+    /** Communication SPI factory. */
+    private CommunicationSpiFactory commSpiFactory;
+
+    /** Latch. */
+    private static CountDownLatch latch;
+
+    /** Test entries. */
+    private Map<Integer, Map<Integer, Integer>> entries = new HashMap<>();
 
     /** {@inheritDoc} */
     @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
         IgniteConfiguration cfg = super.getConfiguration(gridName);
 
-        cfg.setCommunicationSpi(new TestCommunicationSpi());
+        cfg.setCommunicationSpi(commSpiFactory.create());
 
         CacheConfiguration ccfg = defaultCacheConfiguration();
         ccfg.setCacheMode(cacheMode);
@@ -72,142 +85,312 @@ public class CacheScanPartitionQueryFallbackSelfTest extends GridCommonAbstractT
 
         cfg.setCacheConfiguration(ccfg);
 
+        cfg.setClientMode(clientMode);
+
         return cfg;
     }
 
     /**
+     * Scan should perform on the local node.
+     *
      * @throws Exception If failed.
      */
-    public void testPrimary() throws Exception {
+    public void testScanLocal() throws Exception {
         cacheMode = CacheMode.PARTITIONED;
         backups = 0;
-        failNodeId = null;
-        fallback = false;
+        commSpiFactory = new TestLocalCommunicationSpiFactory();
 
-        doTestScanPartition();
+        try {
+            Ignite ignite = startGrids(GRID_CNT);
+
+            IgniteCacheProxy<Integer, Integer> cache = fillCache(ignite);
+
+            int part = anyLocalPartition(cache.context());
+
+            CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null,
part, false);
+
+            doTestScanQuery(qry);
+        }
+        finally {
+            stopAllGrids();
+        }
     }
 
     /**
+     * Scan should perform on the remote node.
+     *
      * @throws Exception If failed.
      */
-    public void testFallbackToBackup() throws Exception {
+    public void testScanRemote() throws Exception {
         cacheMode = CacheMode.PARTITIONED;
-        backups = 1;
-        failNodeId = null;
-        fallback = true;
+        backups = 0;
+        commSpiFactory = new TestRemoteCommunicationSpiFactory();
 
-        doTestScanPartition();
+        try {
+            Ignite ignite = startGrids(GRID_CNT);
+
+            IgniteCacheProxy<Integer, Integer> cache = fillCache(ignite);
+
+            IgniteBiTuple<Integer, UUID> tup = remotePartition(cache.context());
+
+            int part = tup.get1();
+
+            expNodeId = tup.get2();
+
+            CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null,
part, false);
+
+            doTestScanQuery(qry);
+        }
+        finally {
+            stopAllGrids();
+        }
     }
 
     /**
+     * Scan should try first remote node and fallbacks to second remote node.
+     *
      * @throws Exception If failed.
      */
-    protected void doTestScanPartition() throws Exception {
-        try {
-            Ignite ignite = startGrids(GRID_CNT);
+    public void testScanFallback() throws Exception {
+        cacheMode = CacheMode.PARTITIONED;
+        backups = 1;
+        commSpiFactory = new TestFallbackCommunicationSpiFactory();
 
-            IgniteCacheProxy<Integer, Integer> cache =
-                (IgniteCacheProxy<Integer, Integer>)ignite.<Integer, Integer>cache(null);
+        final Set<Integer> candidates = new TreeSet<>();
 
-            Map<Integer, Map<Integer, Integer>> entries = new HashMap<>();
+        final AtomicBoolean test = new AtomicBoolean(false);
 
-            for (int i = 0; i < KEYS_CNT; i++) {
-                cache.put(i, i);
+        for(int j = 0; j < 2; j++) {
+            clientMode = true;
 
-                int part = cache.context().affinity().partition(i);
+            latch = new CountDownLatch(1);
 
-                Map<Integer, Integer> partEntries = entries.get(part);
+            try {
+                final Ignite ignite0 = startGrid(0);
 
-                if (partEntries == null)
-                    entries.put(part, partEntries = new HashMap<>());
+                clientMode = false;
 
-                partEntries.put(i, i);
-            }
+                final IgniteEx ignite1 = startGrid(1);
+                final IgniteEx ignite2 = startGrid(2);
+                startGrid(3);
 
-            IgniteBiTuple<Integer, UUID> tup = remotePartition(cache.context(), true);
+                if (test.get()) {
+                    expNodeId = ignite1.localNode().id();
+                    expFallbackNodeId = ignite2.localNode().id();
+                }
 
-            int part = tup.get1();
+                final IgniteCacheProxy<Integer, Integer> cache = fillCache(ignite0);
 
-            if (fallback)
-                failNodeId = tup.get2();
-            else
-                expNodeId = tup.get2();
+                if (!test.get()) {
+                    candidates.addAll(localPartitions(ignite1));
+                    candidates.retainAll(localPartitions(ignite2));
+                }
 
-            if (fallback)
-                expNodeId = remoteBackup(part, cache.context());
+                Runnable run = new Runnable() {
+                    @Override public void run() {
+                        try {
+                            startGrid(4);
+                            startGrid(5);
 
-            CacheQuery<Map.Entry<Integer, Integer>> qry = cache.context().queries().createScanQuery(null,
part, false);
+                            awaitPartitionMapExchange();
+
+                            if (!test.get()) {
+                                Set<Integer> parts = localPartitions(ignite1);
+                                candidates.removeAll(parts);
+                            }
 
-            CacheQueryFuture<Map.Entry<Integer, Integer>> fut = qry.execute();
+                            latch.countDown();
+                        }
+                        catch (Exception e) {
+                            e.printStackTrace();
+                        }
 
-            Collection<Map.Entry<Integer, Integer>> expEntries = fut.get();
+                    }
+                };
 
-            for (Map.Entry<Integer, Integer> e : expEntries) {
-                Map<Integer, Integer> map = entries.get(part);
+                int part;
+                CacheQuery<Map.Entry<Integer, Integer>> qry = null;
 
-                if(map == null)
-                    assertTrue(expEntries.isEmpty());
+                if (test.get()) {
+                    part = F.first(candidates);
+
+                    qry = cache.context().queries().createScanQuery(null, part, false);
+                }
+
+                new Thread(run).start();
+
+                if (test.get())
+                    doTestScanQuery(qry);
                 else
-                    assertEquals(map.get(e.getKey()), e.getValue());
+                    latch.await();
+            }
+            finally {
+                test.set(true);
+
+                stopAllGrids();
             }
         }
-        finally {
-            stopAllGrids();
+    }
+
+    /**
+     * @param ignite Ignite.
+     */
+    protected IgniteCacheProxy<Integer, Integer> fillCache(Ignite ignite) {
+        IgniteCacheProxy<Integer, Integer> cache =
+            (IgniteCacheProxy<Integer, Integer>)ignite.<Integer, Integer>cache(null);
+
+        for (int i = 0; i < KEYS_CNT; i++) {
+            cache.put(i, i);
+
+            int part = cache.context().affinity().partition(i);
+
+            Map<Integer, Integer> partEntries = entries.get(part);
+
+            if (partEntries == null)
+                entries.put(part, partEntries = new HashMap<>());
+
+            partEntries.put(i, i);
         }
+
+        return cache;
+    }
+
+    /**
+     * @param qry Query.
+     */
+    protected void doTestScanQuery(
+        CacheQuery<Map.Entry<Integer, Integer>> qry) throws IgniteCheckedException
{
+        CacheQueryFuture<Map.Entry<Integer, Integer>> fut = qry.execute();
+
+        Collection<Map.Entry<Integer, Integer>> expEntries = fut.get();
+
+        for (Map.Entry<Integer, Integer> e : expEntries) {
+            Map<Integer, Integer> map = entries.get(((GridCacheQueryAdapter)qry).partition());
+
+            if (map == null)
+                assertTrue(expEntries.isEmpty());
+            else
+                assertEquals(map.get(e.getKey()), e.getValue());
+        }
+    }
+
+    /**
+     * @param cctx Cctx.
+     */
+    private static int anyLocalPartition(GridCacheContext<?, ?> cctx) {
+        return F.first(cctx.topology().localPartitions()).id();
     }
 
     /**
      * @param cctx Cctx.
-     * @param primary Primary.
      */
-    private IgniteBiTuple<Integer, UUID> remotePartition(final GridCacheContext cctx,
boolean primary) {
+    private IgniteBiTuple<Integer, UUID> remotePartition(final GridCacheContext cctx)
{
         ClusterNode node = F.first(cctx.kernalContext().grid().cluster().forRemotes().nodes());
 
         GridCacheAffinityManager affMgr = cctx.affinity();
 
         AffinityTopologyVersion topVer = affMgr.affinityTopologyVersion();
 
-        Set<Integer> parts = primary ?
-            affMgr.primaryPartitions(node.id(), topVer) : affMgr.backupPartitions(node.id(),
topVer);
+        Set<Integer> parts = affMgr.primaryPartitions(node.id(), topVer);
 
         return new IgniteBiTuple<>(F.first(parts), node.id());
     }
 
     /**
-     * @param part Partition.
-     * @param cctx Cctx.
+     * @param ignite Ignite.
      */
-    private UUID remoteBackup(int part, final GridCacheContext cctx) {
-        final UUID locUuid = cctx.localNodeId();
+    private Set<Integer> localPartitions(Ignite ignite) {
+        GridCacheContext cctx = ((IgniteCacheProxy)ignite.cache(null)).context();
+
+        Collection<GridDhtLocalPartition> owningParts = F.view(cctx.topology().localPartitions(),
+            new IgnitePredicate<GridDhtLocalPartition>() {
+                @Override public boolean apply(GridDhtLocalPartition part) {
+                    return part.state() == GridDhtPartitionState.OWNING;
+                }
+            });
+
+        return new HashSet<>(F.transform(owningParts, new IgniteClosure<GridDhtLocalPartition,
Integer>() {
+            @Override public Integer apply(GridDhtLocalPartition part) {
+                return part.id();
+            }
+        }));
+    }
 
-        GridCacheAffinityManager affMgr = cctx.affinity();
+    /**
+     * Factory for tests specific communication SPI.
+     */
+    private interface CommunicationSpiFactory {
+        /**
+         * Creates communication SPI instance.
+         */
+        TcpCommunicationSpi create();
+    }
 
-        AffinityTopologyVersion topVer = affMgr.affinityTopologyVersion();
+    /**
+     *
+     */
+    private static class TestLocalCommunicationSpiFactory implements CommunicationSpiFactory
{
+        /** {@inheritDoc} */
+        @Override public TcpCommunicationSpi create() {
+            return new TcpCommunicationSpi() {
+                @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException
{
+                    Object origMsg = ((GridIoMessage)msg).message();
 
-        return F.first(F.view(affMgr.backups(part, topVer), new IgnitePredicate<ClusterNode>()
{
-            @Override public boolean apply(ClusterNode node) {
-                return !node.id().equals(locUuid);
-            }
-        })).id();
+                    if (origMsg instanceof GridCacheQueryRequest)
+                        fail(); //should use local node
+
+                    super.sendMessage(node, msg);
+                }
+            };
+        }
     }
 
     /**
      *
      */
-    private static class TestCommunicationSpi extends TcpCommunicationSpi {
+    private static class TestRemoteCommunicationSpiFactory implements CommunicationSpiFactory
{
         /** {@inheritDoc} */
-        @Override public void sendMessage(ClusterNode node, Message msg)
-            throws IgniteSpiException {
-            Object origMsg = ((GridIoMessage)msg).message();
+        @Override public TcpCommunicationSpi create() {
+            return new TcpCommunicationSpi() {
+                @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException
{
+                    Object origMsg = ((GridIoMessage)msg).message();
 
-            if (origMsg instanceof GridCacheQueryRequest) {
-                if (node.id().equals(failNodeId))
-                    throw new IgniteSpiException("");
-                else
-                    assertEquals(expNodeId, node.id());
-            }
+                    if (origMsg instanceof GridCacheQueryRequest)
+                        assertEquals(expNodeId, node.id());
+
+                    super.sendMessage(node, msg);
+                }
+            };
+        }
+    }
 
-            super.sendMessage(node, msg);
+    /**
+     *
+     */
+    private static class TestFallbackCommunicationSpiFactory implements CommunicationSpiFactory
{
+        /** {@inheritDoc} */
+        @Override public TcpCommunicationSpi create() {
+            return new TcpCommunicationSpi() {
+                @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException
{
+                    Object origMsg = ((GridIoMessage)msg).message();
+
+                    if (origMsg instanceof GridCacheQueryRequest) {
+                        if (latch.getCount() > 0)
+                            assertEquals(expNodeId, node.id());
+                        else
+                            assertEquals(expFallbackNodeId, node.id());
+
+                        try {
+                            latch.await();
+                        }
+                        catch (InterruptedException e) {
+                            throw new IgniteSpiException(e);
+                        }
+                    }
+
+                    super.sendMessage(node, msg);
+                }
+            };
         }
     }
 }


Mime
View raw message