ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [2/4] incubator-ignite git commit: ignite-752: added new tests, fixed bugs
Date Wed, 22 Jul 2015 08:43:58 GMT
ignite-752: added new tests, fixed bugs


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

Branch: refs/heads/ignite-752
Commit: f2a2dcf6baac19cef0cbb61592decd2135c6646e
Parents: 6d5e7d3
Author: Denis Magda <dmagda@gridgain.com>
Authored: Wed Jul 22 10:48:56 2015 +0300
Committer: Denis Magda <dmagda@gridgain.com>
Committed: Wed Jul 22 10:48:56 2015 +0300

----------------------------------------------------------------------
 .../ignite/spi/discovery/tcp/ServerImpl.java    |  19 ++-
 .../spi/discovery/tcp/TcpDiscoverySpi.java      |   3 +-
 ...entDiscoverySpiFailureThresholdSelfTest.java | 120 ++++++++++++++++++-
 .../tcp/TcpClientDiscoverySpiSelfTest.java      |  22 ++--
 4 files changed, 152 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f2a2dcf6/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 877d53c..56472aa 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -501,6 +501,8 @@ class ServerImpl extends TcpDiscoveryImpl {
 
                 int reconCnt = 0;
 
+                boolean openedSock = false;
+
                 while (true) {
                     try {
                         if (addr.isUnresolved())
@@ -510,6 +512,8 @@ class ServerImpl extends TcpDiscoveryImpl {
 
                         sock = spi.openSocket(addr, timeoutHelper);
 
+                        openedSock = true;
+
                         spi.writeToSocket(sock, new TcpDiscoveryPingRequest(locNodeId, clientNodeId),
                             timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
 
@@ -537,9 +541,14 @@ class ServerImpl extends TcpDiscoveryImpl {
 
                         errs.add(e);
 
+                        reconCnt++;
+
+                        if (!openedSock && reconCnt == 2)
+                            break;
+
                         if (timeoutHelper.checkThresholdReached(e))
                             break;
-                        else if (!spi.failureDetectionThresholdEnabled() && ++reconCnt
== spi.getReconnectCount())
+                        else if (!spi.failureDetectionThresholdEnabled() && reconCnt
== spi.getReconnectCount())
                             break;
                     }
                     finally {
@@ -1846,6 +1855,9 @@ class ServerImpl extends TcpDiscoveryImpl {
 
         /** {@inheritDoc} */
         @Override protected void noMessageLoop() {
+            if (locNode == null)
+                return;
+
             checkConnection();
 
             sendHeartbeatMessage();
@@ -1919,11 +1931,12 @@ class ServerImpl extends TcpDiscoveryImpl {
                         if (debugMode)
                             debugLog("No next node in topology.");
 
-                        /*if (ring.hasRemoteNodes()) {
+                        if (ring.hasRemoteNodes() && !(msg instanceof TcpDiscoveryConnectionCheckMessage)
&&
+                            !(msg instanceof TcpDiscoveryStatusCheckMessage && msg.creatorNodeId().equals(locNodeId)))
{
                             msg.senderNodeId(locNodeId);
 
                             addMessage(msg);
-                        }*/
+                        }
 
                         break;
                     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f2a2dcf6/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 588ff98..3821a0a 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -1369,7 +1369,8 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi,
T
      * @throws IOException If IO failed or read timed out.
      * @throws IgniteCheckedException If unmarshalling failed.
      */
-    protected <T> T readMessage(Socket sock, @Nullable InputStream in, long timeout)
throws IOException, IgniteCheckedException {
+    protected <T> T readMessage(Socket sock, @Nullable InputStream in, long timeout)
throws IOException,
+        IgniteCheckedException {
         assert sock != null;
 
         int oldTimeout = sock.getSoTimeout();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f2a2dcf6/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureThresholdSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureThresholdSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureThresholdSelfTest.java
index 8e80047..939286d 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureThresholdSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureThresholdSelfTest.java
@@ -17,7 +17,14 @@
 
 package org.apache.ignite.spi.discovery.tcp;
 
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
 import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.spi.discovery.tcp.messages.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.net.*;
 
 /**
  * Client-based discovery SPI test with failure detection threshold enabled.
@@ -26,6 +33,9 @@ public class TcpClientDiscoverySpiFailureThresholdSelfTest extends TcpClientDisc
     /** */
     private final static int FAILURE_AWAIT_TIME = 7_000;
 
+    /** */
+    private static boolean useTestSpi;
+
     /** {@inheritDoc} */
     @Override protected boolean useFailureDetectionThreshold() {
         return true;
@@ -33,7 +43,7 @@ public class TcpClientDiscoverySpiFailureThresholdSelfTest extends TcpClientDisc
 
     /** {@inheritDoc} */
     @Override protected long failureDetectionThreshold() {
-        return 10_000;
+        return useTestSpi ? 5000 : 10_000;
     }
 
     /** {@inheritDoc} */
@@ -41,6 +51,18 @@ public class TcpClientDiscoverySpiFailureThresholdSelfTest extends TcpClientDisc
         return failureDetectionThreshold() + FAILURE_AWAIT_TIME;
     }
 
+    /** {@inheritDoc} */
+    @Override protected TcpDiscoverySpi getDiscoverySpi() {
+        return useTestSpi ? new TestTcpDiscoverySpi() : super.getDiscoverySpi();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        return cfg;
+    }
+
     /**
      * @throws Exception in case of error.
      */
@@ -60,4 +82,100 @@ public class TcpClientDiscoverySpiFailureThresholdSelfTest extends TcpClientDisc
         assertEquals(failureDetectionThreshold(),
             ((TcpDiscoverySpi)(G.ignite("client-0").configuration().getDiscoverySpi())).failureDetectionThreshold());
     }
+
+    /**
+     * @throws Exception in case of error.
+     */
+    public void testFailureThresholdWorkability() throws Exception {
+        useTestSpi = true;
+
+        TestTcpDiscoverySpi firstSpi = null;
+        TestTcpDiscoverySpi secondSpi = null;
+
+        try {
+            startServerNodes(2);
+
+            checkNodes(2, 0);
+
+            firstSpi = (TestTcpDiscoverySpi)(G.ignite("server-0").configuration().getDiscoverySpi());
+            secondSpi = (TestTcpDiscoverySpi)(G.ignite("server-1").configuration().getDiscoverySpi());
+
+            assert firstSpi.err == null;
+
+            secondSpi.readDelay = failureDetectionThreshold() + 5000;
+
+            assertFalse(firstSpi.pingNode(secondSpi.getLocalNodeId()));
+
+            Thread.sleep(failureDetectionThreshold());
+
+            assertTrue(firstSpi.err != null && X.hasCause(firstSpi.err, SocketTimeoutException.class));
+
+            firstSpi.reset();
+            secondSpi.reset();
+
+            assertTrue(firstSpi.pingNode(secondSpi.getLocalNodeId()));
+
+            assertTrue(firstSpi.err == null);
+        }
+        finally {
+            useTestSpi = false;
+
+            if (firstSpi != null)
+                firstSpi.reset();
+
+            if (secondSpi != null)
+                secondSpi.reset();
+        }
+    }
+
+    /**
+     *
+     */
+    private static class TestTcpDiscoverySpi extends TcpDiscoverySpi {
+        /** */
+        private long readDelay;
+
+        /** */
+        private Exception err;
+
+        /** {@inheritDoc} */
+        @Override protected <T> T readMessage(Socket sock, @Nullable InputStream in,
long timeout)
+            throws IOException, IgniteCheckedException {
+
+            if (readDelay < failureDetectionThreshold()) {
+                try {
+                    T msg = super.readMessage(sock, in, timeout);
+
+                    return msg;
+                }
+                catch (Exception e) {
+                    err = e;
+
+                    throw e;
+                }
+            }
+            else {
+                T msg = super.readMessage(sock, in, timeout);
+
+                if (msg instanceof TcpDiscoveryPingRequest) {
+                    try {
+                        Thread.sleep(2000);
+                    } catch (InterruptedException e) {
+                        // Ignore
+                    }
+                    throw new SocketTimeoutException("Forced timeout");
+                }
+
+                return msg;
+            }
+        }
+
+        /**
+         * Resets testing state.
+         */
+        private void reset() {
+            readDelay = 0;
+            err = null;
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f2a2dcf6/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 9df7bd9..a67b5cf 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@ -118,7 +118,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest
{
     @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
         IgniteConfiguration cfg = super.getConfiguration(gridName);
 
-        TestTcpDiscoverySpi disco = new TestTcpDiscoverySpi();
+        TcpDiscoverySpi disco = getDiscoverySpi();
 
         disco.setMaxMissedClientHeartbeats(maxMissedClientHbs);
 
@@ -167,7 +167,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest
{
 
         disco.setClientReconnectDisabled(reconnectDisabled);
 
-        disco.afterWrite(afterWrite);
+        if (disco instanceof TestTcpDiscoverySpi)
+            ((TestTcpDiscoverySpi)disco).afterWrite(afterWrite);
 
         cfg.setDiscoverySpi(disco);
 
@@ -177,6 +178,14 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest
{
         return cfg;
     }
 
+    /**
+     * Returns TCP Discovery SPI instance to use in a test.
+     * @return TCP Discovery SPI.
+     */
+    protected TcpDiscoverySpi getDiscoverySpi() {
+        return new TestTcpDiscoverySpi();
+    }
+
     /** {@inheritDoc} */
     @Override protected void beforeTest() throws Exception {
         Collection<InetSocketAddress> addrs = IP_FINDER.getRegisteredAddresses();
@@ -411,12 +420,12 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest
{
 
         final CountDownLatch latch = new CountDownLatch(1);
 
-        ((TcpDiscoverySpi)srv1.configuration().getDiscoverySpi()).addIncomeConnectionListener(new
IgniteInClosure<Socket>() {
+        ((TcpDiscoverySpi)srv1.configuration().getDiscoverySpi()).addIncomeConnectionListener(new
IgniteInClosure
+            <Socket>() {
             @Override public void apply(Socket sock) {
                 try {
                     latch.await();
-                }
-                catch (InterruptedException e) {
+                } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
             }
@@ -778,8 +787,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest
{
             @Override public void apply(TcpDiscoveryAbstractMessage msg) {
                 try {
                     Thread.sleep(1000000);
-                }
-                catch (InterruptedException ignored) {
+                } catch (InterruptedException ignored) {
                     Thread.interrupted();
                 }
             }


Mime
View raw message