ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [15/52] [abbrv] incubator-ignite git commit: # Renaming
Date Fri, 05 Dec 2014 14:07:16 GMT
# Renaming


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

Branch: refs/heads/master
Commit: 4f31df32c1efa257555c2ae9a154d8e4fa943864
Parents: 717180b
Author: sboikov <sboikov@gridgain.com>
Authored: Fri Dec 5 16:50:11 2014 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Fri Dec 5 16:50:11 2014 +0300

----------------------------------------------------------------------
 .../store/CacheNodeWithStoreStartup.java        |   2 +-
 .../org/gridgain/grid/kernal/GridGainEx.java    |   2 +-
 .../discovery/tcp/TcpClientDiscoverySpi.java    |   4 +-
 .../grid/spi/discovery/tcp/TcpDiscoverySpi.java |  12 +-
 .../discovery/tcp/TcpDiscoverySpiAdapter.java   |   3 +-
 .../jdbc/GridTcpDiscoveryJdbcIpFinder.java      | 361 ---------
 .../ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java | 361 +++++++++
 .../GridTcpDiscoveryMulticastIpFinder.java      | 752 -------------------
 .../TcpDiscoveryMulticastIpFinder.java          | 752 +++++++++++++++++++
 .../GridTcpDiscoverySharedFsIpFinder.java       | 258 -------
 .../sharedfs/TcpDiscoverySharedFsIpFinder.java  | 258 +++++++
 .../test/config/loaders/grid-cfg-2-grids.xml    |   4 +-
 .../core/src/test/config/loaders/grid-cfg.xml   |   4 +-
 .../discovery/tcp/GridTcpDiscoverySelfTest.java |   4 +-
 .../GridTcpDiscoveryJdbcIpFinderSelfTest.java   |   6 +-
 ...idTcpDiscoveryMulticastIpFinderSelfTest.java |  16 +-
 ...ridTcpDiscoverySharedFsIpFinderSelfTest.java |   6 +-
 .../testframework/junits/GridAbstractTest.java  |   2 +-
 .../junits/spi/GridSpiAbstractTest.java         |   6 +-
 19 files changed, 1405 insertions(+), 1408 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/examples/src/main/java/org/gridgain/examples/datagrid/store/CacheNodeWithStoreStartup.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/gridgain/examples/datagrid/store/CacheNodeWithStoreStartup.java b/examples/src/main/java/org/gridgain/examples/datagrid/store/CacheNodeWithStoreStartup.java
index 2bf0cfd..9035376 100644
--- a/examples/src/main/java/org/gridgain/examples/datagrid/store/CacheNodeWithStoreStartup.java
+++ b/examples/src/main/java/org/gridgain/examples/datagrid/store/CacheNodeWithStoreStartup.java
@@ -50,7 +50,7 @@ public class CacheNodeWithStoreStartup {
         // Discovery SPI.
         TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
 
-        TcpDiscoveryVmIpFinder ipFinder = new GridTcpDiscoveryMulticastIpFinder();
+        TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryMulticastIpFinder();
 
         ipFinder.setAddresses(Arrays.asList("127.0.0.1:47500..47509"));
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java b/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java
index 178d323..5469e86 100644
--- a/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java
+++ b/modules/core/src/main/java/org/gridgain/grid/kernal/GridGainEx.java
@@ -1676,7 +1676,7 @@ public class GridGainEx {
                 TcpDiscoverySpi tcpDisco = (TcpDiscoverySpi)discoSpi;
 
                 if (tcpDisco.getIpFinder() == null)
-                    tcpDisco.setIpFinder(new GridTcpDiscoveryMulticastIpFinder());
+                    tcpDisco.setIpFinder(new TcpDiscoveryMulticastIpFinder());
             }
 
             if (evtSpi == null)

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpClientDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpClientDiscoverySpi.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpClientDiscoverySpi.java
index ff50969..bb3a2ca 100644
--- a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpClientDiscoverySpi.java
+++ b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpClientDiscoverySpi.java
@@ -228,8 +228,8 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
             throw new IgniteSpiException("Unknown local address: " + locAddr, e);
         }
 
-        if (ipFinder instanceof GridTcpDiscoveryMulticastIpFinder) {
-            GridTcpDiscoveryMulticastIpFinder mcastIpFinder = ((GridTcpDiscoveryMulticastIpFinder)ipFinder);
+        if (ipFinder instanceof TcpDiscoveryMulticastIpFinder) {
+            TcpDiscoveryMulticastIpFinder mcastIpFinder = ((TcpDiscoveryMulticastIpFinder)ipFinder);
 
             if (mcastIpFinder.getLocalAddress() == null)
                 mcastIpFinder.setLocalAddress(locAddr);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpi.java
index 7a1b964..acd2599 100644
--- a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -22,9 +22,7 @@ import org.gridgain.grid.kernal.managers.security.*;
 import org.gridgain.grid.security.*;
 import org.gridgain.grid.spi.discovery.*;
 import org.gridgain.grid.spi.discovery.tcp.internal.*;
-import org.gridgain.grid.spi.discovery.tcp.ipfinder.jdbc.*;
 import org.gridgain.grid.spi.discovery.tcp.ipfinder.multicast.*;
-import org.gridgain.grid.spi.discovery.tcp.ipfinder.sharedfs.*;
 import org.gridgain.grid.spi.discovery.tcp.messages.*;
 import org.gridgain.grid.util.*;
 import org.gridgain.grid.util.future.*;
@@ -68,11 +66,11 @@ import static org.gridgain.grid.spi.discovery.tcp.messages.GridTcpDiscoveryStatu
  * (see {@link #setIpFinder(org.gridgain.grid.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder)}).
  * See the following IP finder implementations for details on configuration:
  * <ul>
- * <li>{@link GridTcpDiscoverySharedFsIpFinder}</li>
+ * <li>{@link org.gridgain.grid.spi.discovery.tcp.ipfinder.sharedfs.TcpDiscoverySharedFsIpFinder}</li>
  * <li>{@gglink org.gridgain.grid.spi.discovery.tcp.ipfinder.s3.GridTcpDiscoveryS3IpFinder}</li>
- * <li>{@link GridTcpDiscoveryJdbcIpFinder}</li>
+ * <li>{@link org.gridgain.grid.spi.discovery.tcp.ipfinder.jdbc.TcpDiscoveryJdbcIpFinder}</li>
  * <li>{@link org.gridgain.grid.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder}</li>
- * <li>{@link GridTcpDiscoveryMulticastIpFinder} - default</li>
+ * <li>{@link org.gridgain.grid.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder} - default</li>
  * </ul>
  * </li>
  * </ul>
@@ -853,8 +851,8 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
 
         registerMBean(gridName, this, TcpDiscoverySpiMBean.class);
 
-        if (ipFinder instanceof GridTcpDiscoveryMulticastIpFinder) {
-            GridTcpDiscoveryMulticastIpFinder mcastIpFinder = ((GridTcpDiscoveryMulticastIpFinder)ipFinder);
+        if (ipFinder instanceof TcpDiscoveryMulticastIpFinder) {
+            TcpDiscoveryMulticastIpFinder mcastIpFinder = ((TcpDiscoveryMulticastIpFinder)ipFinder);
 
             if (mcastIpFinder.getLocalAddress() == null)
                 mcastIpFinder.setLocalAddress(locAddr);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpiAdapter.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
index 03b422a..47ee030 100644
--- a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
+++ b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
@@ -20,7 +20,6 @@ import org.gridgain.grid.*;
 import org.gridgain.grid.spi.discovery.*;
 import org.gridgain.grid.spi.discovery.tcp.internal.*;
 import org.gridgain.grid.spi.discovery.tcp.ipfinder.*;
-import org.gridgain.grid.spi.discovery.tcp.ipfinder.multicast.*;
 import org.gridgain.grid.spi.discovery.tcp.messages.*;
 import org.gridgain.grid.util.*;
 import org.gridgain.grid.util.io.*;
@@ -183,7 +182,7 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements Discov
     /**
      * Sets IP finder for IP addresses sharing and storing.
      * <p>
-     * If not provided {@link GridTcpDiscoveryMulticastIpFinder} will be used by default.
+     * If not provided {@link org.gridgain.grid.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder} will be used by default.
      *
      * @param ipFinder IP finder.
      */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/GridTcpDiscoveryJdbcIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/GridTcpDiscoveryJdbcIpFinder.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/GridTcpDiscoveryJdbcIpFinder.java
deleted file mode 100644
index 0e1a15b..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/GridTcpDiscoveryJdbcIpFinder.java
+++ /dev/null
@@ -1,361 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.spi.discovery.tcp.ipfinder.jdbc;
-
-import org.apache.ignite.*;
-import org.apache.ignite.resources.*;
-import org.apache.ignite.spi.*;
-import org.gridgain.grid.*;
-import org.gridgain.grid.spi.discovery.tcp.ipfinder.*;
-import org.gridgain.grid.util.typedef.*;
-import org.gridgain.grid.util.typedef.internal.*;
-import org.gridgain.grid.util.tostring.*;
-
-import javax.sql.*;
-import java.net.*;
-import java.sql.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-import static java.sql.Connection.*;
-
-/**
- * JDBC-based IP finder.
- * <h1 class="header">Configuration</h1>
- * <h2 class="header">Mandatory</h2>
- * <ul>
- *     <li>Data source (see {@link #setDataSource(DataSource)}).</li>
- * </ul>
- * <h2 class="header">Optional</h2>
- * The following configuration parameters are optional:
- * <ul>
- *     <li>Flag indicating whether DB schema should be initialized by GridGain (default behaviour) or
- *         was explicitly created by user (see {@link #setInitSchema(boolean)})</li>
- * </ul>
- * <p>
- * The database will contain 1 table which will hold IP addresses.
- */
-public class GridTcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
-    /** Query to get addresses. */
-    public static final String GET_ADDRS_QRY = "select hostname, port from tbl_addrs";
-
-    /** Query to register address. */
-    public static final String REG_ADDR_QRY = "insert into tbl_addrs values (?, ?)";
-
-    /** Query to unregister address. */
-    public static final String UNREG_ADDR_QRY = "delete from tbl_addrs where hostname = ? and port = ?";
-
-    /** Query to create addresses table. */
-    public static final String CREATE_ADDRS_TABLE_QRY =
-        "create table if not exists tbl_addrs (" +
-        "hostname VARCHAR(1024), " +
-        "port INT)";
-
-    /** Query to check database validity. */
-    public static final String CHK_QRY = "select count(*) from tbl_addrs";
-
-    /** Grid logger. */
-    @IgniteLoggerResource
-    private IgniteLogger log;
-
-    /** Data source. */
-    private DataSource dataSrc;
-
-    /** Flag for schema initialization. */
-    private boolean initSchema = true;
-
-    /** Init guard. */
-    @GridToStringExclude
-    private final AtomicBoolean initGuard = new AtomicBoolean();
-
-    /** Init latch. */
-    @GridToStringExclude
-    private final CountDownLatch initLatch = new CountDownLatch(1);
-
-    /**
-     * Constructor.
-     */
-    public GridTcpDiscoveryJdbcIpFinder() {
-        setShared(true);
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<InetSocketAddress> getRegisteredAddresses() throws IgniteSpiException {
-        init();
-
-        Connection conn = null;
-
-        PreparedStatement stmt = null;
-
-        ResultSet rs = null;
-
-        try {
-            conn = dataSrc.getConnection();
-
-            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
-
-            stmt = conn.prepareStatement(GET_ADDRS_QRY);
-
-            rs = stmt.executeQuery();
-
-            Collection<InetSocketAddress> addrs = new LinkedList<>();
-
-            while (rs.next())
-                addrs.add(new InetSocketAddress(rs.getString(1), rs.getInt(2)));
-
-            return addrs;
-        }
-        catch (SQLException e) {
-            throw new IgniteSpiException("Failed to get registered addresses version.", e);
-        }
-        finally {
-            U.closeQuiet(rs);
-            U.closeQuiet(stmt);
-            U.closeQuiet(conn);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void registerAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
-        assert !F.isEmpty(addrs);
-
-        init();
-
-        Connection conn = null;
-
-        PreparedStatement stmtUnreg = null;
-
-        PreparedStatement stmtReg = null;
-
-        boolean committed = false;
-
-        try {
-            conn = dataSrc.getConnection();
-
-            conn.setAutoCommit(false);
-
-            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
-
-            stmtUnreg = conn.prepareStatement(UNREG_ADDR_QRY);
-            stmtReg = conn.prepareStatement(REG_ADDR_QRY);
-
-            for (InetSocketAddress addr : addrs) {
-                stmtUnreg.setString(1, addr.getAddress().getHostAddress());
-                stmtUnreg.setInt(2, addr.getPort());
-
-                stmtUnreg.addBatch();
-
-                stmtReg.setString(1, addr.getAddress().getHostAddress());
-                stmtReg.setInt(2, addr.getPort());
-
-                stmtReg.addBatch();
-            }
-
-            stmtUnreg.executeBatch();
-            stmtUnreg.close();
-
-            stmtReg.executeBatch();
-            stmtReg.close();
-
-            conn.commit();
-
-            committed = true;
-        }
-        catch (SQLException e) {
-            U.rollbackConnectionQuiet(conn);
-
-            throw new IgniteSpiException("Failed to register addresses: " + addrs, e);
-        }
-        finally {
-            if (!committed)
-                U.rollbackConnectionQuiet(conn);
-
-            U.closeQuiet(stmtUnreg);
-            U.closeQuiet(stmtReg);
-            U.closeQuiet(conn);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void unregisterAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
-        assert !F.isEmpty(addrs);
-
-        init();
-
-        Connection conn = null;
-
-        PreparedStatement stmt = null;
-
-        boolean committed = false;
-
-        try {
-            conn = dataSrc.getConnection();
-
-            conn.setAutoCommit(false);
-
-            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
-
-            stmt = conn.prepareStatement(UNREG_ADDR_QRY);
-
-            for (InetSocketAddress addr : addrs) {
-                stmt.setString(1, addr.getAddress().getHostAddress());
-                stmt.setInt(2, addr.getPort());
-
-                stmt.addBatch();
-            }
-
-            stmt.executeBatch();
-            conn.commit();
-
-            committed = true;
-        }
-        catch (SQLException e) {
-            U.rollbackConnectionQuiet(conn);
-
-            throw new IgniteSpiException("Failed to unregister addresses: " + addrs, e);
-        }
-        finally {
-            if (!committed)
-                U.rollbackConnectionQuiet(conn);
-
-            U.closeQuiet(stmt);
-            U.closeQuiet(conn);
-        }
-    }
-
-    /**
-     * Sets data source.
-     * <p>
-     * Data source should be fully configured and ready-to-use.
-     *
-     * @param dataSrc Data source.
-     */
-    @IgniteSpiConfiguration(optional = false)
-    public void setDataSource(DataSource dataSrc) {
-        this.dataSrc = dataSrc;
-    }
-
-    /**
-     * Flag indicating whether DB schema should be initialized by GridGain (default behaviour) or
-     * was explicitly created by user.
-     *
-     * @param initSchema {@code True} if DB schema should be initialized by GridGain (default behaviour),
-     *      {code @false} if schema was explicitly created by user.
-     */
-    @IgniteSpiConfiguration(optional = true)
-    public void setInitSchema(boolean initSchema) {
-        this.initSchema = initSchema;
-    }
-
-    /**
-     * Checks configuration validity.
-     *
-     * @throws org.apache.ignite.spi.IgniteSpiException If any error occurs.
-     */
-    private void init() throws IgniteSpiException {
-        if (initGuard.compareAndSet(false, true)) {
-            if (dataSrc == null)
-                throw new IgniteSpiException("Data source is null (you must configure it via setDataSource(..)" +
-                    " configuration property)");
-
-            if (!initSchema) {
-                initLatch.countDown();
-
-                checkSchema();
-
-                return;
-            }
-
-            Connection conn = null;
-
-            Statement stmt = null;
-
-            boolean committed = false;
-
-            try {
-                conn = dataSrc.getConnection();
-
-                conn.setAutoCommit(false);
-
-                conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
-
-                // Create tbl_addrs.
-                stmt = conn.createStatement();
-
-                stmt.executeUpdate(CREATE_ADDRS_TABLE_QRY);
-
-                conn.commit();
-
-                committed = true;
-
-                if (log.isDebugEnabled())
-                    log.debug("DB schema has been initialized.");
-            }
-            catch (SQLException e) {
-                U.rollbackConnectionQuiet(conn);
-
-                throw new IgniteSpiException("Failed to initialize DB schema.", e);
-            }
-            finally {
-                if (!committed)
-                    U.rollbackConnectionQuiet(conn);
-
-                U.closeQuiet(stmt);
-                U.closeQuiet(conn);
-
-                initLatch.countDown();
-            }
-        }
-        else
-            checkSchema();
-    }
-
-    /**
-     * Checks correctness of existing DB schema.
-     *
-     * @throws org.apache.ignite.spi.IgniteSpiException If schema wasn't properly initialized.
-     */
-    private void checkSchema() throws IgniteSpiException {
-        try {
-            U.await(initLatch);
-        }
-        catch (GridInterruptedException e) {
-            throw new IgniteSpiException("Thread has been interrupted.", e);
-        }
-
-        Connection conn = null;
-
-        Statement stmt = null;
-
-        try {
-            conn = dataSrc.getConnection();
-
-            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
-
-            // Check if tbl_addrs exists and database initialized properly.
-            stmt = conn.createStatement();
-
-            stmt.execute(CHK_QRY);
-        }
-        catch (SQLException e) {
-            throw new IgniteSpiException("IP finder has not been properly initialized.", e);
-        }
-        finally {
-            U.closeQuiet(stmt);
-            U.closeQuiet(conn);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridTcpDiscoveryJdbcIpFinder.class, this);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
new file mode 100644
index 0000000..4b5c00b
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/jdbc/TcpDiscoveryJdbcIpFinder.java
@@ -0,0 +1,361 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.spi.discovery.tcp.ipfinder.jdbc;
+
+import org.apache.ignite.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.gridgain.grid.*;
+import org.gridgain.grid.spi.discovery.tcp.ipfinder.*;
+import org.gridgain.grid.util.typedef.*;
+import org.gridgain.grid.util.typedef.internal.*;
+import org.gridgain.grid.util.tostring.*;
+
+import javax.sql.*;
+import java.net.*;
+import java.sql.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import static java.sql.Connection.*;
+
+/**
+ * JDBC-based IP finder.
+ * <h1 class="header">Configuration</h1>
+ * <h2 class="header">Mandatory</h2>
+ * <ul>
+ *     <li>Data source (see {@link #setDataSource(DataSource)}).</li>
+ * </ul>
+ * <h2 class="header">Optional</h2>
+ * The following configuration parameters are optional:
+ * <ul>
+ *     <li>Flag indicating whether DB schema should be initialized by GridGain (default behaviour) or
+ *         was explicitly created by user (see {@link #setInitSchema(boolean)})</li>
+ * </ul>
+ * <p>
+ * The database will contain 1 table which will hold IP addresses.
+ */
+public class TcpDiscoveryJdbcIpFinder extends TcpDiscoveryIpFinderAdapter {
+    /** Query to get addresses. */
+    public static final String GET_ADDRS_QRY = "select hostname, port from tbl_addrs";
+
+    /** Query to register address. */
+    public static final String REG_ADDR_QRY = "insert into tbl_addrs values (?, ?)";
+
+    /** Query to unregister address. */
+    public static final String UNREG_ADDR_QRY = "delete from tbl_addrs where hostname = ? and port = ?";
+
+    /** Query to create addresses table. */
+    public static final String CREATE_ADDRS_TABLE_QRY =
+        "create table if not exists tbl_addrs (" +
+        "hostname VARCHAR(1024), " +
+        "port INT)";
+
+    /** Query to check database validity. */
+    public static final String CHK_QRY = "select count(*) from tbl_addrs";
+
+    /** Grid logger. */
+    @IgniteLoggerResource
+    private IgniteLogger log;
+
+    /** Data source. */
+    private DataSource dataSrc;
+
+    /** Flag for schema initialization. */
+    private boolean initSchema = true;
+
+    /** Init guard. */
+    @GridToStringExclude
+    private final AtomicBoolean initGuard = new AtomicBoolean();
+
+    /** Init latch. */
+    @GridToStringExclude
+    private final CountDownLatch initLatch = new CountDownLatch(1);
+
+    /**
+     * Constructor.
+     */
+    public TcpDiscoveryJdbcIpFinder() {
+        setShared(true);
+    }
+
+    /** {@inheritDoc} */
+    @Override public Collection<InetSocketAddress> getRegisteredAddresses() throws IgniteSpiException {
+        init();
+
+        Connection conn = null;
+
+        PreparedStatement stmt = null;
+
+        ResultSet rs = null;
+
+        try {
+            conn = dataSrc.getConnection();
+
+            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
+
+            stmt = conn.prepareStatement(GET_ADDRS_QRY);
+
+            rs = stmt.executeQuery();
+
+            Collection<InetSocketAddress> addrs = new LinkedList<>();
+
+            while (rs.next())
+                addrs.add(new InetSocketAddress(rs.getString(1), rs.getInt(2)));
+
+            return addrs;
+        }
+        catch (SQLException e) {
+            throw new IgniteSpiException("Failed to get registered addresses version.", e);
+        }
+        finally {
+            U.closeQuiet(rs);
+            U.closeQuiet(stmt);
+            U.closeQuiet(conn);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void registerAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
+        assert !F.isEmpty(addrs);
+
+        init();
+
+        Connection conn = null;
+
+        PreparedStatement stmtUnreg = null;
+
+        PreparedStatement stmtReg = null;
+
+        boolean committed = false;
+
+        try {
+            conn = dataSrc.getConnection();
+
+            conn.setAutoCommit(false);
+
+            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
+
+            stmtUnreg = conn.prepareStatement(UNREG_ADDR_QRY);
+            stmtReg = conn.prepareStatement(REG_ADDR_QRY);
+
+            for (InetSocketAddress addr : addrs) {
+                stmtUnreg.setString(1, addr.getAddress().getHostAddress());
+                stmtUnreg.setInt(2, addr.getPort());
+
+                stmtUnreg.addBatch();
+
+                stmtReg.setString(1, addr.getAddress().getHostAddress());
+                stmtReg.setInt(2, addr.getPort());
+
+                stmtReg.addBatch();
+            }
+
+            stmtUnreg.executeBatch();
+            stmtUnreg.close();
+
+            stmtReg.executeBatch();
+            stmtReg.close();
+
+            conn.commit();
+
+            committed = true;
+        }
+        catch (SQLException e) {
+            U.rollbackConnectionQuiet(conn);
+
+            throw new IgniteSpiException("Failed to register addresses: " + addrs, e);
+        }
+        finally {
+            if (!committed)
+                U.rollbackConnectionQuiet(conn);
+
+            U.closeQuiet(stmtUnreg);
+            U.closeQuiet(stmtReg);
+            U.closeQuiet(conn);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void unregisterAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
+        assert !F.isEmpty(addrs);
+
+        init();
+
+        Connection conn = null;
+
+        PreparedStatement stmt = null;
+
+        boolean committed = false;
+
+        try {
+            conn = dataSrc.getConnection();
+
+            conn.setAutoCommit(false);
+
+            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
+
+            stmt = conn.prepareStatement(UNREG_ADDR_QRY);
+
+            for (InetSocketAddress addr : addrs) {
+                stmt.setString(1, addr.getAddress().getHostAddress());
+                stmt.setInt(2, addr.getPort());
+
+                stmt.addBatch();
+            }
+
+            stmt.executeBatch();
+            conn.commit();
+
+            committed = true;
+        }
+        catch (SQLException e) {
+            U.rollbackConnectionQuiet(conn);
+
+            throw new IgniteSpiException("Failed to unregister addresses: " + addrs, e);
+        }
+        finally {
+            if (!committed)
+                U.rollbackConnectionQuiet(conn);
+
+            U.closeQuiet(stmt);
+            U.closeQuiet(conn);
+        }
+    }
+
+    /**
+     * Sets data source.
+     * <p>
+     * Data source should be fully configured and ready-to-use.
+     *
+     * @param dataSrc Data source.
+     */
+    @IgniteSpiConfiguration(optional = false)
+    public void setDataSource(DataSource dataSrc) {
+        this.dataSrc = dataSrc;
+    }
+
+    /**
+     * Flag indicating whether DB schema should be initialized by GridGain (default behaviour) or
+     * was explicitly created by user.
+     *
+     * @param initSchema {@code True} if DB schema should be initialized by GridGain (default behaviour),
+     *      {code @false} if schema was explicitly created by user.
+     */
+    @IgniteSpiConfiguration(optional = true)
+    public void setInitSchema(boolean initSchema) {
+        this.initSchema = initSchema;
+    }
+
+    /**
+     * Checks configuration validity.
+     *
+     * @throws org.apache.ignite.spi.IgniteSpiException If any error occurs.
+     */
+    private void init() throws IgniteSpiException {
+        if (initGuard.compareAndSet(false, true)) {
+            if (dataSrc == null)
+                throw new IgniteSpiException("Data source is null (you must configure it via setDataSource(..)" +
+                    " configuration property)");
+
+            if (!initSchema) {
+                initLatch.countDown();
+
+                checkSchema();
+
+                return;
+            }
+
+            Connection conn = null;
+
+            Statement stmt = null;
+
+            boolean committed = false;
+
+            try {
+                conn = dataSrc.getConnection();
+
+                conn.setAutoCommit(false);
+
+                conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
+
+                // Create tbl_addrs.
+                stmt = conn.createStatement();
+
+                stmt.executeUpdate(CREATE_ADDRS_TABLE_QRY);
+
+                conn.commit();
+
+                committed = true;
+
+                if (log.isDebugEnabled())
+                    log.debug("DB schema has been initialized.");
+            }
+            catch (SQLException e) {
+                U.rollbackConnectionQuiet(conn);
+
+                throw new IgniteSpiException("Failed to initialize DB schema.", e);
+            }
+            finally {
+                if (!committed)
+                    U.rollbackConnectionQuiet(conn);
+
+                U.closeQuiet(stmt);
+                U.closeQuiet(conn);
+
+                initLatch.countDown();
+            }
+        }
+        else
+            checkSchema();
+    }
+
+    /**
+     * Checks correctness of existing DB schema.
+     *
+     * @throws org.apache.ignite.spi.IgniteSpiException If schema wasn't properly initialized.
+     */
+    private void checkSchema() throws IgniteSpiException {
+        try {
+            U.await(initLatch);
+        }
+        catch (GridInterruptedException e) {
+            throw new IgniteSpiException("Thread has been interrupted.", e);
+        }
+
+        Connection conn = null;
+
+        Statement stmt = null;
+
+        try {
+            conn = dataSrc.getConnection();
+
+            conn.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
+
+            // Check if tbl_addrs exists and database initialized properly.
+            stmt = conn.createStatement();
+
+            stmt.execute(CHK_QRY);
+        }
+        catch (SQLException e) {
+            throw new IgniteSpiException("IP finder has not been properly initialized.", e);
+        }
+        finally {
+            U.closeQuiet(stmt);
+            U.closeQuiet(conn);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(TcpDiscoveryJdbcIpFinder.class, this);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/GridTcpDiscoveryMulticastIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/GridTcpDiscoveryMulticastIpFinder.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/GridTcpDiscoveryMulticastIpFinder.java
deleted file mode 100644
index 1c4a5f9..0000000
--- a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/GridTcpDiscoveryMulticastIpFinder.java
+++ /dev/null
@@ -1,752 +0,0 @@
-/* @java.file.header */
-
-/*  _________        _____ __________________        _____
- *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
- *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
- *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
- *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
- */
-
-package org.gridgain.grid.spi.discovery.tcp.ipfinder.multicast;
-
-import org.apache.ignite.*;
-import org.apache.ignite.marshaller.*;
-import org.apache.ignite.marshaller.jdk.*;
-import org.apache.ignite.resources.*;
-import org.apache.ignite.spi.*;
-import org.gridgain.grid.*;
-import org.gridgain.grid.spi.discovery.tcp.ipfinder.vm.*;
-import org.gridgain.grid.util.typedef.*;
-import org.gridgain.grid.util.typedef.internal.*;
-import org.gridgain.grid.util.tostring.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-
-import static org.apache.ignite.IgniteSystemProperties.*;
-import static org.apache.ignite.spi.IgnitePortProtocol.*;
-
-/**
- * Multicast-based IP finder.
- * <p>
- * When TCP discovery starts this finder sends multicast request and waits
- * for some time when others nodes reply to this request with messages containing
- * their addresses (time IP finder waits for response and number of attempts to
- * re-send multicast request in case if no replies are received can be configured,
- * see {@link #setResponseWaitTime(int)} and {@link #setAddressRequestAttempts(int)}).
- * <p>
- * In addition to address received via multicast this finder can work with pre-configured
- * list of addresses specified via {@link #setAddresses(Collection)} method.
- * <h1 class="header">Configuration</h1>
- * <h2 class="header">Mandatory</h2>
- * There are no mandatory configuration parameters.
- * <h2 class="header">Optional</h2>
- * <ul>
- *      <li>Multicast IP address (see {@link #setMulticastGroup(String)}).</li>
- *      <li>Multicast port number (see {@link #setMulticastPort(int)}).</li>
- *      <li>Address response wait time (see {@link #setResponseWaitTime(int)}).</li>
- *      <li>Address request attempts (see {@link #setAddressRequestAttempts(int)}).</li>
- *      <li>Pre-configured addresses (see {@link #setAddresses(Collection)})</li>
- *      <li>Local address (see {@link #setLocalAddress(String)})</li>
- * </ul>
- */
-public class GridTcpDiscoveryMulticastIpFinder extends TcpDiscoveryVmIpFinder {
-    /** Default multicast IP address (value is {@code 228.1.2.4}). */
-    public static final String DFLT_MCAST_GROUP = "228.1.2.4";
-
-    /** Default multicast port number (value is {@code 47400}). */
-    public static final int DFLT_MCAST_PORT = 47400;
-
-    /** Default time IP finder waits for reply to multicast address request (value is {@code 500}). */
-    public static final int DFLT_RES_WAIT_TIME = 500;
-
-    /** Default number of attempts to send multicast address request (value is {@code 2}). */
-    public static final int DFLT_ADDR_REQ_ATTEMPTS = 2;
-
-    /** Address request message data. */
-    private static final byte[] MSG_ADDR_REQ_DATA = U.GG_HEADER;
-
-    /** */
-    private static final IgniteMarshaller marsh = new IgniteJdkMarshaller();
-
-    /** Grid logger. */
-    @IgniteLoggerResource
-    private IgniteLogger log;
-
-    /** Grid name. */
-    @IgniteNameResource
-    @GridToStringExclude
-    private String gridName;
-
-    /** Multicast IP address as string. */
-    private String mcastGrp = DFLT_MCAST_GROUP;
-
-    /** Multicast port number. */
-    private int mcastPort = DFLT_MCAST_PORT;
-
-    /** Time IP finder waits for reply to multicast address request. */
-    private int resWaitTime = DFLT_RES_WAIT_TIME;
-
-    /** Number of attempts to send multicast address request. */
-    private int addrReqAttempts = DFLT_ADDR_REQ_ATTEMPTS;
-
-    /** Local address */
-    private String locAddr;
-
-    /** */
-    @GridToStringExclude
-    private Collection<AddressSender> addrSnds;
-
-    /**
-     * Constructs new IP finder.
-     */
-    public GridTcpDiscoveryMulticastIpFinder() {
-        setShared(true);
-    }
-
-    /**
-     * Sets IP address of multicast group.
-     * <p>
-     * If not provided, default value is {@link #DFLT_MCAST_GROUP}.
-     *
-     * @param mcastGrp Multicast IP address.
-     */
-    @IgniteSpiConfiguration(optional = true)
-    public void setMulticastGroup(String mcastGrp) {
-        this.mcastGrp = mcastGrp;
-    }
-
-    /**
-     * Gets IP address of multicast group.
-     *
-     * @return Multicast IP address.
-     */
-    public String getMulticastGroup() {
-        return mcastGrp;
-    }
-
-    /**
-     * Sets port number which multicast messages are sent to.
-     * <p>
-     * If not provided, default value is {@link #DFLT_MCAST_PORT}.
-     *
-     * @param mcastPort Multicast port number.
-     */
-    @IgniteSpiConfiguration(optional = true)
-    public void setMulticastPort(int mcastPort) {
-        this.mcastPort = mcastPort;
-    }
-
-    /**
-     * Gets port number which multicast messages are sent to.
-     *
-     * @return Port number.
-     */
-    public int getMulticastPort() {
-        return mcastPort;
-    }
-
-    /**
-     * Sets time in milliseconds IP finder waits for reply to
-     * multicast address request.
-     * <p>
-     * If not provided, default value is {@link #DFLT_RES_WAIT_TIME}.
-     *
-     * @param resWaitTime Time IP finder waits for reply to multicast address request.
-     */
-    @IgniteSpiConfiguration(optional = true)
-    public void setResponseWaitTime(int resWaitTime) {
-        this.resWaitTime = resWaitTime;
-    }
-
-    /**
-     * Gets time in milliseconds IP finder waits for reply to
-     * multicast address request.
-     *
-     * @return Time IP finder waits for reply to multicast address request.
-     */
-    public int getResponseWaitTime() {
-        return resWaitTime;
-    }
-
-    /**
-     * Sets number of attempts to send multicast address request. IP finder re-sends
-     * request only in case if no reply for previous request is received.
-     * <p>
-     * If not provided, default value is {@link #DFLT_ADDR_REQ_ATTEMPTS}.
-     *
-     * @param addrReqAttempts Number of attempts to send multicast address request.
-     */
-    @IgniteSpiConfiguration(optional = true)
-    public void setAddressRequestAttempts(int addrReqAttempts) {
-        this.addrReqAttempts = addrReqAttempts;
-    }
-
-    /**
-     * Gets number of attempts to send multicast address request. IP finder re-sends
-     * request only in case if no reply for previous request is received.
-     *
-     * @return Number of attempts to send multicast address request.
-     */
-    public int getAddressRequestAttempts() {
-        return addrReqAttempts;
-    }
-
-    /**
-     * Sets local host address used by this IP finder. If provided address is non-loopback then multicast
-     * socket is bound to this interface. If local address is not set or is any local address then IP finder
-     * creates multicast sockets for all found non-loopback addresses.
-     * <p>
-     * If not provided then this property is initialized by the local address set in {@link org.gridgain.grid.spi.discovery.tcp.TcpDiscoverySpi}
-     * configuration.
-     *
-     * @param locAddr Local host address.
-     * @see org.gridgain.grid.spi.discovery.tcp.TcpDiscoverySpi#setLocalAddress(String)
-     */
-    @IgniteSpiConfiguration(optional = true)
-    public void setLocalAddress(String locAddr) {
-        this.locAddr = locAddr;
-    }
-
-    /**
-     * Gets local address that multicast IP finder uses.
-     *
-     * @return Local address.
-     */
-    public String getLocalAddress() {
-        return locAddr;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void initializeLocalAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
-        // If GRIDGAIN_OVERRIDE_MCAST_GRP system property is set, use its value to override multicast group from
-        // configuration. Used for testing purposes.
-        String overrideMcastGrp = System.getProperty(GG_OVERRIDE_MCAST_GRP);
-
-        if (overrideMcastGrp != null)
-            mcastGrp = overrideMcastGrp;
-
-        if (F.isEmpty(mcastGrp))
-            throw new IgniteSpiException("Multicast IP address is not specified.");
-
-        if (mcastPort < 0 || mcastPort > 65535)
-            throw new IgniteSpiException("Invalid multicast port: " + mcastPort);
-
-        if (resWaitTime <= 0)
-            throw new IgniteSpiException("Invalid wait time, value greater than zero is expected: " + resWaitTime);
-
-        if (addrReqAttempts <= 0)
-            throw new IgniteSpiException("Invalid number of address request attempts, " +
-                "value greater than zero is expected: " + addrReqAttempts);
-
-        if (F.isEmpty(getRegisteredAddresses()))
-            U.warn(log, "GridTcpDiscoveryMulticastIpFinder has no pre-configured addresses " +
-                "(it is recommended in production to specify at least one address in " +
-                "GridTcpDiscoveryMulticastIpFinder.getAddresses() configuration property)");
-
-        InetAddress mcastAddr;
-
-        try {
-            mcastAddr = InetAddress.getByName(mcastGrp);
-        }
-        catch (UnknownHostException e) {
-            throw new IgniteSpiException("Unknown multicast group: " + mcastGrp, e);
-        }
-
-        if (!mcastAddr.isMulticastAddress())
-            throw new IgniteSpiException("Invalid multicast group address: " + mcastAddr);
-
-        Collection<String> locAddrs;
-
-        try {
-            locAddrs = U.resolveLocalAddresses(U.resolveLocalHost(locAddr)).get1();
-        }
-        catch (IOException | GridException e) {
-            throw new IgniteSpiException("Failed to resolve local addresses [locAddr=" + locAddr + ']', e);
-        }
-
-        assert locAddrs != null;
-
-        addrSnds = new ArrayList<>(locAddrs.size());
-
-        Collection<InetAddress> reqItfs = new ArrayList<>(locAddrs.size()); // Interfaces used to send requests.
-
-        for (String locAddr : locAddrs) {
-            InetAddress addr;
-
-            try {
-                addr = InetAddress.getByName(locAddr);
-            }
-            catch (UnknownHostException e) {
-                if (log.isDebugEnabled())
-                    log.debug("Failed to resolve local address [locAddr=" + locAddr + ", err=" + e + ']');
-
-                continue;
-            }
-
-            if (!addr.isLoopbackAddress()) {
-                try {
-                    addrSnds.add(new AddressSender(mcastAddr, addr, addrs));
-
-                    reqItfs.add(addr);
-                }
-                catch (IOException e) {
-                    if (log.isDebugEnabled())
-                        log.debug("Failed to create multicast socket [mcastAddr=" + mcastAddr +
-                            ", mcastGrp=" + mcastGrp + ", mcastPort=" + mcastPort + ", locAddr=" + addr +
-                            ", err=" + e + ']');
-                }
-            }
-        }
-
-        if (addrSnds.isEmpty()) {
-            try {
-                // Create non-bound socket if local host is loopback or failed to create sockets explicitly
-                // bound to interfaces.
-                addrSnds.add(new AddressSender(mcastAddr, null, addrs));
-            }
-            catch (IOException e) {
-                throw new IgniteSpiException("Failed to create multicast socket [mcastAddr=" + mcastAddr +
-                    ", mcastGrp=" + mcastGrp + ", mcastPort=" + mcastPort + ']', e);
-            }
-        }
-
-        for (AddressSender addrSnd :addrSnds)
-            addrSnd.start();
-
-        Collection<InetSocketAddress> ret;
-
-        if (reqItfs.size() > 1) {
-            ret = new HashSet<>();
-
-            Collection<AddressReceiver> rcvrs = new ArrayList<>();
-
-            for (InetAddress itf : reqItfs) {
-                AddressReceiver rcvr = new AddressReceiver(mcastAddr, itf);
-
-                rcvr.start();
-
-                rcvrs.add(rcvr);
-            }
-
-            for (AddressReceiver rcvr : rcvrs) {
-                try {
-                    rcvr.join();
-
-                    ret.addAll(rcvr.addresses());
-                }
-                catch (InterruptedException ignore) {
-                    U.warn(log, "Got interrupted while receiving address request.");
-
-                    Thread.currentThread().interrupt();
-
-                    break;
-                }
-            }
-        }
-        else
-            ret = requestAddresses(mcastAddr, F.first(reqItfs));
-
-        if (!ret.isEmpty())
-            registerAddresses(ret);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void onSpiContextInitialized(IgniteSpiContext spiCtx) throws IgniteSpiException {
-        super.onSpiContextInitialized(spiCtx);
-
-        spiCtx.registerPort(mcastPort, UDP);
-    }
-
-    /**
-     * Sends multicast address request message and waits for reply. Response wait time and number
-     * of request attempts are configured as properties {@link #setResponseWaitTime} and
-     * {@link #setAddressRequestAttempts}.
-     *
-     * @param mcastAddr Multicast address where to send request.
-     * @param sockItf Optional interface multicast socket should be bound to.
-     * @return Collection of received addresses.
-     */
-    private Collection<InetSocketAddress> requestAddresses(InetAddress mcastAddr, @Nullable InetAddress sockItf) {
-        Collection<InetSocketAddress> rmtAddrs = new HashSet<>();
-
-        try {
-            DatagramPacket reqPckt = new DatagramPacket(MSG_ADDR_REQ_DATA, MSG_ADDR_REQ_DATA.length,
-                mcastAddr, mcastPort);
-
-            byte[] resData = new byte[AddressResponse.MAX_DATA_LENGTH];
-
-            DatagramPacket resPckt = new DatagramPacket(resData, resData.length);
-
-            boolean sndError = false;
-
-            for (int i = 0; i < addrReqAttempts; i++) {
-                MulticastSocket sock = null;
-
-                try {
-                    sock = new MulticastSocket(0);
-
-                    // Use 'false' to enable support for more than one node on the same machine.
-                    sock.setLoopbackMode(false);
-
-                    if (sockItf != null)
-                        sock.setInterface(sockItf);
-
-                    sock.setSoTimeout(resWaitTime);
-
-                    reqPckt.setData(MSG_ADDR_REQ_DATA);
-
-                    try {
-                        sock.send(reqPckt);
-                    }
-                    catch (IOException e) {
-                        if (!handleNetworkError(e))
-                            break;
-
-                        if (i < addrReqAttempts - 1) {
-                            if (log.isDebugEnabled())
-                                log.debug("Failed to send multicast address request (will retry in 500 ms): " + e);
-
-                            U.sleep(500);
-                        }
-                        else {
-                            if (log.isDebugEnabled())
-                                log.debug("Failed to send multicast address request: " + e);
-                        }
-
-                        sndError = true;
-
-                        continue;
-                    }
-
-                    long rcvEnd = U.currentTimeMillis() + resWaitTime;
-
-                    try {
-                        while (U.currentTimeMillis() < rcvEnd) { // Try to receive multiple responses.
-                            sock.receive(resPckt);
-
-                            byte[] data = resPckt.getData();
-
-                            if (!U.bytesEqual(U.GG_HEADER, 0, data, 0, U.GG_HEADER.length)) {
-                                U.error(log, "Failed to verify message header.");
-
-                                continue;
-                            }
-
-                            AddressResponse addrRes;
-
-                            try {
-                                addrRes = new AddressResponse(data);
-                            }
-                            catch (GridException e) {
-                                LT.warn(log, e, "Failed to deserialize multicast response.");
-
-                                continue;
-                            }
-
-                            rmtAddrs.addAll(addrRes.addresses());
-                        }
-                    }
-                    catch (SocketTimeoutException ignored) {
-                        if (log.isDebugEnabled()) // DatagramSocket.receive timeout has expired.
-                            log.debug("Address receive timeout.");
-                    }
-                }
-                catch (IOException e) {
-                    U.error(log, "Failed to request nodes addresses.", e);
-                }
-                finally {
-                    U.close(sock);
-                }
-
-                if (!rmtAddrs.isEmpty())
-                    break;
-
-                if (i < addrReqAttempts - 1) // Wait some time before re-sending address request.
-                    U.sleep(200);
-            }
-
-            if (log.isDebugEnabled())
-                log.debug("Received nodes addresses: " + rmtAddrs);
-
-            if (rmtAddrs.isEmpty() && sndError)
-                U.quietAndWarn(log, "Failed to send multicast message (is multicast enabled on this node?).");
-
-            return rmtAddrs;
-        }
-        catch (GridInterruptedException ignored) {
-            U.warn(log, "Got interrupted while sending address request.");
-
-            Thread.currentThread().interrupt();
-
-            return rmtAddrs;
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void close() {
-        for (AddressSender addrSnd : addrSnds)
-            U.interrupt(addrSnd);
-
-        for (AddressSender addrSnd : addrSnds)
-            U.join(addrSnd, log);
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridTcpDiscoveryMulticastIpFinder.class, this, "super", super.toString());
-    }
-
-    /**
-     * @param e Network error to handle.
-     * @return {@code True} if this error is recoverable and the operation can be retried.
-     */
-    private boolean handleNetworkError(IOException e) {
-        if ("Network is unreachable".equals(e.getMessage()) && U.isMacOs()) {
-            U.warn(log, "Multicast does not work on Mac OS JVM loopback address (configure external IP address " +
-                "for 'localHost' configuration property)");
-
-            return false;
-        }
-
-        return true;
-    }
-
-    /**
-     * Response to multicast address request.
-     */
-    private static class AddressResponse {
-        /** Maximum supported multicast message. */
-        public static final int MAX_DATA_LENGTH = 64 * 1024;
-
-        /** */
-        private byte[] data;
-
-        /** */
-        private Collection<InetSocketAddress> addrs;
-
-        /**
-         * @param addrs Addresses discovery SPI binds to.
-         * @throws GridException If marshalling failed.
-         */
-        private AddressResponse(Collection<InetSocketAddress> addrs) throws GridException {
-            this.addrs = addrs;
-
-            byte[] addrsData = marsh.marshal(addrs);
-            data = new byte[U.GG_HEADER.length + addrsData.length];
-
-            if (data.length > MAX_DATA_LENGTH)
-                throw new GridException("Too long data packet [size=" + data.length + ", max=" + MAX_DATA_LENGTH + "]");
-
-            System.arraycopy(U.GG_HEADER, 0, data, 0, U.GG_HEADER.length);
-            System.arraycopy(addrsData, 0, data, 4, addrsData.length);
-        }
-
-        /**
-         * @param data Message data.
-         * @throws GridException If unmarshalling failed.
-         */
-        private AddressResponse(byte[] data) throws GridException {
-            assert U.bytesEqual(U.GG_HEADER, 0, data, 0, U.GG_HEADER.length);
-
-            this.data = data;
-
-            addrs = marsh.unmarshal(Arrays.copyOfRange(data, U.GG_HEADER.length, data.length), null);
-        }
-
-        /**
-         * @return Message data.
-         */
-        byte[] data() {
-            return data;
-        }
-
-        /**
-         * @return IP address discovery SPI binds to.
-         */
-        public Collection<InetSocketAddress> addresses() {
-            return addrs;
-        }
-    }
-
-    /**
-     * Thread sends multicast address request message and waits for reply.
-     */
-    private class AddressReceiver extends IgniteSpiThread {
-        /** */
-        private final InetAddress mcastAddr;
-
-        /** */
-        private final InetAddress sockAddr;
-
-        /** */
-        private Collection<InetSocketAddress> addrs;
-
-        /**
-         * @param mcastAddr Multicast address where to send request.
-         * @param sockAddr Optional address multicast socket should be bound to.
-         */
-        private AddressReceiver(InetAddress mcastAddr, InetAddress sockAddr) {
-            super(gridName, "tcp-disco-multicast-addr-rcvr", log);
-            this.mcastAddr = mcastAddr;
-            this.sockAddr = sockAddr;
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void body() throws InterruptedException {
-            addrs = requestAddresses(mcastAddr, sockAddr);
-        }
-
-        /**
-         * @return Received addresses.
-         */
-        Collection<InetSocketAddress> addresses() {
-            return addrs;
-        }
-    }
-
-    /**
-     * Thread listening for multicast address requests and sending response
-     * containing socket address this node's discovery SPI listens to.
-     */
-    private class AddressSender extends IgniteSpiThread {
-        /** */
-        @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
-        private MulticastSocket sock;
-
-        /** */
-        private final InetAddress mcastGrp;
-
-        /** */
-        private final Collection<InetSocketAddress> addrs;
-
-        /** */
-        private final InetAddress sockItf;
-
-        /**
-         * @param mcastGrp Multicast address.
-         * @param sockItf Optional interface multicast socket should be bound to.
-         * @param addrs Local node addresses.
-         * @throws IOException If fails to create multicast socket.
-         */
-        private AddressSender(InetAddress mcastGrp, @Nullable InetAddress sockItf, Collection<InetSocketAddress> addrs)
-            throws IOException {
-            super(gridName, "tcp-disco-multicast-addr-sender", log);
-            this.mcastGrp = mcastGrp;
-            this.addrs = addrs;
-            this.sockItf = sockItf;
-
-            sock = createSocket();
-        }
-
-        /**
-         * Creates multicast socket and joins multicast group.
-         *
-         * @throws IOException If fails to create socket or join multicast group.
-         * @return Multicast socket.
-         */
-        private MulticastSocket createSocket() throws IOException {
-            MulticastSocket sock = new MulticastSocket(mcastPort);
-
-            sock.setLoopbackMode(false); // Use 'false' to enable support for more than one node on the same machine.
-
-            if (sockItf != null)
-                sock.setInterface(sockItf);
-
-            if (sock.getLoopbackMode())
-                U.warn(log, "Loopback mode is disabled which prevents nodes on the same machine from discovering " +
-                    "each other.");
-
-            sock.joinGroup(mcastGrp);
-
-            return sock;
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void body() throws InterruptedException {
-            AddressResponse res;
-
-            try {
-                res = new AddressResponse(addrs);
-            }
-            catch (GridException e) {
-                U.error(log, "Failed to prepare multicast message.", e);
-
-                return;
-            }
-
-            byte[] reqData = new byte[MSG_ADDR_REQ_DATA.length];
-
-            DatagramPacket pckt = new DatagramPacket(reqData, reqData.length);
-
-            while (!isInterrupted()) {
-                try {
-                    MulticastSocket sock;
-
-                    synchronized (this) {
-                        if (isInterrupted())
-                            return;
-
-                        sock = this.sock;
-
-                        if (sock == null)
-                            sock = createSocket();
-                    }
-
-                    sock.receive(pckt);
-
-                    if (!U.bytesEqual(U.GG_HEADER, 0, reqData, 0, U.GG_HEADER.length)) {
-                        U.error(log, "Failed to verify message header.");
-
-                        continue;
-                    }
-
-                    try {
-                        sock.send(new DatagramPacket(res.data(), res.data().length, pckt.getAddress(), pckt.getPort()));
-                    }
-                    catch (IOException e) {
-                        if (e.getMessage().contains("Operation not permitted")) {
-                            if (log.isDebugEnabled())
-                                log.debug("Got 'operation not permitted' error, ignoring: " + e);
-                        }
-                        else
-                            throw e;
-                    }
-                }
-                catch (IOException e) {
-                    if (!isInterrupted()) {
-                        U.error(log, "Failed to send/receive address message (will try to reconnect).", e);
-
-                        synchronized (this) {
-                            U.close(sock);
-
-                            sock = null;
-                        }
-                    }
-                }
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override public void interrupt() {
-            super.interrupt();
-
-            synchronized (this) {
-                U.close(sock);
-
-                sock = null;
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void cleanup() {
-            synchronized (this) {
-                U.close(sock);
-
-                sock = null;
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/4f31df32/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
new file mode 100644
index 0000000..acc20a4
--- /dev/null
+++ b/modules/core/src/main/java/org/gridgain/grid/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
@@ -0,0 +1,752 @@
+/* @java.file.header */
+
+/*  _________        _____ __________________        _____
+ *  __  ____/___________(_)______  /__  ____/______ ____(_)_______
+ *  _  / __  __  ___/__  / _  __  / _  / __  _  __ `/__  / __  __ \
+ *  / /_/ /  _  /    _  /  / /_/ /  / /_/ /  / /_/ / _  /  _  / / /
+ *  \____/   /_/     /_/   \_,__/   \____/   \__,_/  /_/   /_/ /_/
+ */
+
+package org.gridgain.grid.spi.discovery.tcp.ipfinder.multicast;
+
+import org.apache.ignite.*;
+import org.apache.ignite.marshaller.*;
+import org.apache.ignite.marshaller.jdk.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.gridgain.grid.*;
+import org.gridgain.grid.spi.discovery.tcp.ipfinder.vm.*;
+import org.gridgain.grid.util.typedef.*;
+import org.gridgain.grid.util.typedef.internal.*;
+import org.gridgain.grid.util.tostring.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+import static org.apache.ignite.IgniteSystemProperties.*;
+import static org.apache.ignite.spi.IgnitePortProtocol.*;
+
+/**
+ * Multicast-based IP finder.
+ * <p>
+ * When TCP discovery starts this finder sends multicast request and waits
+ * for some time when others nodes reply to this request with messages containing
+ * their addresses (time IP finder waits for response and number of attempts to
+ * re-send multicast request in case if no replies are received can be configured,
+ * see {@link #setResponseWaitTime(int)} and {@link #setAddressRequestAttempts(int)}).
+ * <p>
+ * In addition to address received via multicast this finder can work with pre-configured
+ * list of addresses specified via {@link #setAddresses(Collection)} method.
+ * <h1 class="header">Configuration</h1>
+ * <h2 class="header">Mandatory</h2>
+ * There are no mandatory configuration parameters.
+ * <h2 class="header">Optional</h2>
+ * <ul>
+ *      <li>Multicast IP address (see {@link #setMulticastGroup(String)}).</li>
+ *      <li>Multicast port number (see {@link #setMulticastPort(int)}).</li>
+ *      <li>Address response wait time (see {@link #setResponseWaitTime(int)}).</li>
+ *      <li>Address request attempts (see {@link #setAddressRequestAttempts(int)}).</li>
+ *      <li>Pre-configured addresses (see {@link #setAddresses(Collection)})</li>
+ *      <li>Local address (see {@link #setLocalAddress(String)})</li>
+ * </ul>
+ */
+public class TcpDiscoveryMulticastIpFinder extends TcpDiscoveryVmIpFinder {
+    /** Default multicast IP address (value is {@code 228.1.2.4}). */
+    public static final String DFLT_MCAST_GROUP = "228.1.2.4";
+
+    /** Default multicast port number (value is {@code 47400}). */
+    public static final int DFLT_MCAST_PORT = 47400;
+
+    /** Default time IP finder waits for reply to multicast address request (value is {@code 500}). */
+    public static final int DFLT_RES_WAIT_TIME = 500;
+
+    /** Default number of attempts to send multicast address request (value is {@code 2}). */
+    public static final int DFLT_ADDR_REQ_ATTEMPTS = 2;
+
+    /** Address request message data. */
+    private static final byte[] MSG_ADDR_REQ_DATA = U.GG_HEADER;
+
+    /** */
+    private static final IgniteMarshaller marsh = new IgniteJdkMarshaller();
+
+    /** Grid logger. */
+    @IgniteLoggerResource
+    private IgniteLogger log;
+
+    /** Grid name. */
+    @IgniteNameResource
+    @GridToStringExclude
+    private String gridName;
+
+    /** Multicast IP address as string. */
+    private String mcastGrp = DFLT_MCAST_GROUP;
+
+    /** Multicast port number. */
+    private int mcastPort = DFLT_MCAST_PORT;
+
+    /** Time IP finder waits for reply to multicast address request. */
+    private int resWaitTime = DFLT_RES_WAIT_TIME;
+
+    /** Number of attempts to send multicast address request. */
+    private int addrReqAttempts = DFLT_ADDR_REQ_ATTEMPTS;
+
+    /** Local address */
+    private String locAddr;
+
+    /** */
+    @GridToStringExclude
+    private Collection<AddressSender> addrSnds;
+
+    /**
+     * Constructs new IP finder.
+     */
+    public TcpDiscoveryMulticastIpFinder() {
+        setShared(true);
+    }
+
+    /**
+     * Sets IP address of multicast group.
+     * <p>
+     * If not provided, default value is {@link #DFLT_MCAST_GROUP}.
+     *
+     * @param mcastGrp Multicast IP address.
+     */
+    @IgniteSpiConfiguration(optional = true)
+    public void setMulticastGroup(String mcastGrp) {
+        this.mcastGrp = mcastGrp;
+    }
+
+    /**
+     * Gets IP address of multicast group.
+     *
+     * @return Multicast IP address.
+     */
+    public String getMulticastGroup() {
+        return mcastGrp;
+    }
+
+    /**
+     * Sets port number which multicast messages are sent to.
+     * <p>
+     * If not provided, default value is {@link #DFLT_MCAST_PORT}.
+     *
+     * @param mcastPort Multicast port number.
+     */
+    @IgniteSpiConfiguration(optional = true)
+    public void setMulticastPort(int mcastPort) {
+        this.mcastPort = mcastPort;
+    }
+
+    /**
+     * Gets port number which multicast messages are sent to.
+     *
+     * @return Port number.
+     */
+    public int getMulticastPort() {
+        return mcastPort;
+    }
+
+    /**
+     * Sets time in milliseconds IP finder waits for reply to
+     * multicast address request.
+     * <p>
+     * If not provided, default value is {@link #DFLT_RES_WAIT_TIME}.
+     *
+     * @param resWaitTime Time IP finder waits for reply to multicast address request.
+     */
+    @IgniteSpiConfiguration(optional = true)
+    public void setResponseWaitTime(int resWaitTime) {
+        this.resWaitTime = resWaitTime;
+    }
+
+    /**
+     * Gets time in milliseconds IP finder waits for reply to
+     * multicast address request.
+     *
+     * @return Time IP finder waits for reply to multicast address request.
+     */
+    public int getResponseWaitTime() {
+        return resWaitTime;
+    }
+
+    /**
+     * Sets number of attempts to send multicast address request. IP finder re-sends
+     * request only in case if no reply for previous request is received.
+     * <p>
+     * If not provided, default value is {@link #DFLT_ADDR_REQ_ATTEMPTS}.
+     *
+     * @param addrReqAttempts Number of attempts to send multicast address request.
+     */
+    @IgniteSpiConfiguration(optional = true)
+    public void setAddressRequestAttempts(int addrReqAttempts) {
+        this.addrReqAttempts = addrReqAttempts;
+    }
+
+    /**
+     * Gets number of attempts to send multicast address request. IP finder re-sends
+     * request only in case if no reply for previous request is received.
+     *
+     * @return Number of attempts to send multicast address request.
+     */
+    public int getAddressRequestAttempts() {
+        return addrReqAttempts;
+    }
+
+    /**
+     * Sets local host address used by this IP finder. If provided address is non-loopback then multicast
+     * socket is bound to this interface. If local address is not set or is any local address then IP finder
+     * creates multicast sockets for all found non-loopback addresses.
+     * <p>
+     * If not provided then this property is initialized by the local address set in {@link org.gridgain.grid.spi.discovery.tcp.TcpDiscoverySpi}
+     * configuration.
+     *
+     * @param locAddr Local host address.
+     * @see org.gridgain.grid.spi.discovery.tcp.TcpDiscoverySpi#setLocalAddress(String)
+     */
+    @IgniteSpiConfiguration(optional = true)
+    public void setLocalAddress(String locAddr) {
+        this.locAddr = locAddr;
+    }
+
+    /**
+     * Gets local address that multicast IP finder uses.
+     *
+     * @return Local address.
+     */
+    public String getLocalAddress() {
+        return locAddr;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void initializeLocalAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
+        // If GRIDGAIN_OVERRIDE_MCAST_GRP system property is set, use its value to override multicast group from
+        // configuration. Used for testing purposes.
+        String overrideMcastGrp = System.getProperty(GG_OVERRIDE_MCAST_GRP);
+
+        if (overrideMcastGrp != null)
+            mcastGrp = overrideMcastGrp;
+
+        if (F.isEmpty(mcastGrp))
+            throw new IgniteSpiException("Multicast IP address is not specified.");
+
+        if (mcastPort < 0 || mcastPort > 65535)
+            throw new IgniteSpiException("Invalid multicast port: " + mcastPort);
+
+        if (resWaitTime <= 0)
+            throw new IgniteSpiException("Invalid wait time, value greater than zero is expected: " + resWaitTime);
+
+        if (addrReqAttempts <= 0)
+            throw new IgniteSpiException("Invalid number of address request attempts, " +
+                "value greater than zero is expected: " + addrReqAttempts);
+
+        if (F.isEmpty(getRegisteredAddresses()))
+            U.warn(log, "GridTcpDiscoveryMulticastIpFinder has no pre-configured addresses " +
+                "(it is recommended in production to specify at least one address in " +
+                "GridTcpDiscoveryMulticastIpFinder.getAddresses() configuration property)");
+
+        InetAddress mcastAddr;
+
+        try {
+            mcastAddr = InetAddress.getByName(mcastGrp);
+        }
+        catch (UnknownHostException e) {
+            throw new IgniteSpiException("Unknown multicast group: " + mcastGrp, e);
+        }
+
+        if (!mcastAddr.isMulticastAddress())
+            throw new IgniteSpiException("Invalid multicast group address: " + mcastAddr);
+
+        Collection<String> locAddrs;
+
+        try {
+            locAddrs = U.resolveLocalAddresses(U.resolveLocalHost(locAddr)).get1();
+        }
+        catch (IOException | GridException e) {
+            throw new IgniteSpiException("Failed to resolve local addresses [locAddr=" + locAddr + ']', e);
+        }
+
+        assert locAddrs != null;
+
+        addrSnds = new ArrayList<>(locAddrs.size());
+
+        Collection<InetAddress> reqItfs = new ArrayList<>(locAddrs.size()); // Interfaces used to send requests.
+
+        for (String locAddr : locAddrs) {
+            InetAddress addr;
+
+            try {
+                addr = InetAddress.getByName(locAddr);
+            }
+            catch (UnknownHostException e) {
+                if (log.isDebugEnabled())
+                    log.debug("Failed to resolve local address [locAddr=" + locAddr + ", err=" + e + ']');
+
+                continue;
+            }
+
+            if (!addr.isLoopbackAddress()) {
+                try {
+                    addrSnds.add(new AddressSender(mcastAddr, addr, addrs));
+
+                    reqItfs.add(addr);
+                }
+                catch (IOException e) {
+                    if (log.isDebugEnabled())
+                        log.debug("Failed to create multicast socket [mcastAddr=" + mcastAddr +
+                            ", mcastGrp=" + mcastGrp + ", mcastPort=" + mcastPort + ", locAddr=" + addr +
+                            ", err=" + e + ']');
+                }
+            }
+        }
+
+        if (addrSnds.isEmpty()) {
+            try {
+                // Create non-bound socket if local host is loopback or failed to create sockets explicitly
+                // bound to interfaces.
+                addrSnds.add(new AddressSender(mcastAddr, null, addrs));
+            }
+            catch (IOException e) {
+                throw new IgniteSpiException("Failed to create multicast socket [mcastAddr=" + mcastAddr +
+                    ", mcastGrp=" + mcastGrp + ", mcastPort=" + mcastPort + ']', e);
+            }
+        }
+
+        for (AddressSender addrSnd :addrSnds)
+            addrSnd.start();
+
+        Collection<InetSocketAddress> ret;
+
+        if (reqItfs.size() > 1) {
+            ret = new HashSet<>();
+
+            Collection<AddressReceiver> rcvrs = new ArrayList<>();
+
+            for (InetAddress itf : reqItfs) {
+                AddressReceiver rcvr = new AddressReceiver(mcastAddr, itf);
+
+                rcvr.start();
+
+                rcvrs.add(rcvr);
+            }
+
+            for (AddressReceiver rcvr : rcvrs) {
+                try {
+                    rcvr.join();
+
+                    ret.addAll(rcvr.addresses());
+                }
+                catch (InterruptedException ignore) {
+                    U.warn(log, "Got interrupted while receiving address request.");
+
+                    Thread.currentThread().interrupt();
+
+                    break;
+                }
+            }
+        }
+        else
+            ret = requestAddresses(mcastAddr, F.first(reqItfs));
+
+        if (!ret.isEmpty())
+            registerAddresses(ret);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void onSpiContextInitialized(IgniteSpiContext spiCtx) throws IgniteSpiException {
+        super.onSpiContextInitialized(spiCtx);
+
+        spiCtx.registerPort(mcastPort, UDP);
+    }
+
+    /**
+     * Sends multicast address request message and waits for reply. Response wait time and number
+     * of request attempts are configured as properties {@link #setResponseWaitTime} and
+     * {@link #setAddressRequestAttempts}.
+     *
+     * @param mcastAddr Multicast address where to send request.
+     * @param sockItf Optional interface multicast socket should be bound to.
+     * @return Collection of received addresses.
+     */
+    private Collection<InetSocketAddress> requestAddresses(InetAddress mcastAddr, @Nullable InetAddress sockItf) {
+        Collection<InetSocketAddress> rmtAddrs = new HashSet<>();
+
+        try {
+            DatagramPacket reqPckt = new DatagramPacket(MSG_ADDR_REQ_DATA, MSG_ADDR_REQ_DATA.length,
+                mcastAddr, mcastPort);
+
+            byte[] resData = new byte[AddressResponse.MAX_DATA_LENGTH];
+
+            DatagramPacket resPckt = new DatagramPacket(resData, resData.length);
+
+            boolean sndError = false;
+
+            for (int i = 0; i < addrReqAttempts; i++) {
+                MulticastSocket sock = null;
+
+                try {
+                    sock = new MulticastSocket(0);
+
+                    // Use 'false' to enable support for more than one node on the same machine.
+                    sock.setLoopbackMode(false);
+
+                    if (sockItf != null)
+                        sock.setInterface(sockItf);
+
+                    sock.setSoTimeout(resWaitTime);
+
+                    reqPckt.setData(MSG_ADDR_REQ_DATA);
+
+                    try {
+                        sock.send(reqPckt);
+                    }
+                    catch (IOException e) {
+                        if (!handleNetworkError(e))
+                            break;
+
+                        if (i < addrReqAttempts - 1) {
+                            if (log.isDebugEnabled())
+                                log.debug("Failed to send multicast address request (will retry in 500 ms): " + e);
+
+                            U.sleep(500);
+                        }
+                        else {
+                            if (log.isDebugEnabled())
+                                log.debug("Failed to send multicast address request: " + e);
+                        }
+
+                        sndError = true;
+
+                        continue;
+                    }
+
+                    long rcvEnd = U.currentTimeMillis() + resWaitTime;
+
+                    try {
+                        while (U.currentTimeMillis() < rcvEnd) { // Try to receive multiple responses.
+                            sock.receive(resPckt);
+
+                            byte[] data = resPckt.getData();
+
+                            if (!U.bytesEqual(U.GG_HEADER, 0, data, 0, U.GG_HEADER.length)) {
+                                U.error(log, "Failed to verify message header.");
+
+                                continue;
+                            }
+
+                            AddressResponse addrRes;
+
+                            try {
+                                addrRes = new AddressResponse(data);
+                            }
+                            catch (GridException e) {
+                                LT.warn(log, e, "Failed to deserialize multicast response.");
+
+                                continue;
+                            }
+
+                            rmtAddrs.addAll(addrRes.addresses());
+                        }
+                    }
+                    catch (SocketTimeoutException ignored) {
+                        if (log.isDebugEnabled()) // DatagramSocket.receive timeout has expired.
+                            log.debug("Address receive timeout.");
+                    }
+                }
+                catch (IOException e) {
+                    U.error(log, "Failed to request nodes addresses.", e);
+                }
+                finally {
+                    U.close(sock);
+                }
+
+                if (!rmtAddrs.isEmpty())
+                    break;
+
+                if (i < addrReqAttempts - 1) // Wait some time before re-sending address request.
+                    U.sleep(200);
+            }
+
+            if (log.isDebugEnabled())
+                log.debug("Received nodes addresses: " + rmtAddrs);
+
+            if (rmtAddrs.isEmpty() && sndError)
+                U.quietAndWarn(log, "Failed to send multicast message (is multicast enabled on this node?).");
+
+            return rmtAddrs;
+        }
+        catch (GridInterruptedException ignored) {
+            U.warn(log, "Got interrupted while sending address request.");
+
+            Thread.currentThread().interrupt();
+
+            return rmtAddrs;
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void close() {
+        for (AddressSender addrSnd : addrSnds)
+            U.interrupt(addrSnd);
+
+        for (AddressSender addrSnd : addrSnds)
+            U.join(addrSnd, log);
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(TcpDiscoveryMulticastIpFinder.class, this, "super", super.toString());
+    }
+
+    /**
+     * @param e Network error to handle.
+     * @return {@code True} if this error is recoverable and the operation can be retried.
+     */
+    private boolean handleNetworkError(IOException e) {
+        if ("Network is unreachable".equals(e.getMessage()) && U.isMacOs()) {
+            U.warn(log, "Multicast does not work on Mac OS JVM loopback address (configure external IP address " +
+                "for 'localHost' configuration property)");
+
+            return false;
+        }
+
+        return true;
+    }
+
+    /**
+     * Response to multicast address request.
+     */
+    private static class AddressResponse {
+        /** Maximum supported multicast message. */
+        public static final int MAX_DATA_LENGTH = 64 * 1024;
+
+        /** */
+        private byte[] data;
+
+        /** */
+        private Collection<InetSocketAddress> addrs;
+
+        /**
+         * @param addrs Addresses discovery SPI binds to.
+         * @throws GridException If marshalling failed.
+         */
+        private AddressResponse(Collection<InetSocketAddress> addrs) throws GridException {
+            this.addrs = addrs;
+
+            byte[] addrsData = marsh.marshal(addrs);
+            data = new byte[U.GG_HEADER.length + addrsData.length];
+
+            if (data.length > MAX_DATA_LENGTH)
+                throw new GridException("Too long data packet [size=" + data.length + ", max=" + MAX_DATA_LENGTH + "]");
+
+            System.arraycopy(U.GG_HEADER, 0, data, 0, U.GG_HEADER.length);
+            System.arraycopy(addrsData, 0, data, 4, addrsData.length);
+        }
+
+        /**
+         * @param data Message data.
+         * @throws GridException If unmarshalling failed.
+         */
+        private AddressResponse(byte[] data) throws GridException {
+            assert U.bytesEqual(U.GG_HEADER, 0, data, 0, U.GG_HEADER.length);
+
+            this.data = data;
+
+            addrs = marsh.unmarshal(Arrays.copyOfRange(data, U.GG_HEADER.length, data.length), null);
+        }
+
+        /**
+         * @return Message data.
+         */
+        byte[] data() {
+            return data;
+        }
+
+        /**
+         * @return IP address discovery SPI binds to.
+         */
+        public Collection<InetSocketAddress> addresses() {
+            return addrs;
+        }
+    }
+
+    /**
+     * Thread sends multicast address request message and waits for reply.
+     */
+    private class AddressReceiver extends IgniteSpiThread {
+        /** */
+        private final InetAddress mcastAddr;
+
+        /** */
+        private final InetAddress sockAddr;
+
+        /** */
+        private Collection<InetSocketAddress> addrs;
+
+        /**
+         * @param mcastAddr Multicast address where to send request.
+         * @param sockAddr Optional address multicast socket should be bound to.
+         */
+        private AddressReceiver(InetAddress mcastAddr, InetAddress sockAddr) {
+            super(gridName, "tcp-disco-multicast-addr-rcvr", log);
+            this.mcastAddr = mcastAddr;
+            this.sockAddr = sockAddr;
+        }
+
+        /** {@inheritDoc} */
+        @Override protected void body() throws InterruptedException {
+            addrs = requestAddresses(mcastAddr, sockAddr);
+        }
+
+        /**
+         * @return Received addresses.
+         */
+        Collection<InetSocketAddress> addresses() {
+            return addrs;
+        }
+    }
+
+    /**
+     * Thread listening for multicast address requests and sending response
+     * containing socket address this node's discovery SPI listens to.
+     */
+    private class AddressSender extends IgniteSpiThread {
+        /** */
+        @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
+        private MulticastSocket sock;
+
+        /** */
+        private final InetAddress mcastGrp;
+
+        /** */
+        private final Collection<InetSocketAddress> addrs;
+
+        /** */
+        private final InetAddress sockItf;
+
+        /**
+         * @param mcastGrp Multicast address.
+         * @param sockItf Optional interface multicast socket should be bound to.
+         * @param addrs Local node addresses.
+         * @throws IOException If fails to create multicast socket.
+         */
+        private AddressSender(InetAddress mcastGrp, @Nullable InetAddress sockItf, Collection<InetSocketAddress> addrs)
+            throws IOException {
+            super(gridName, "tcp-disco-multicast-addr-sender", log);
+            this.mcastGrp = mcastGrp;
+            this.addrs = addrs;
+            this.sockItf = sockItf;
+
+            sock = createSocket();
+        }
+
+        /**
+         * Creates multicast socket and joins multicast group.
+         *
+         * @throws IOException If fails to create socket or join multicast group.
+         * @return Multicast socket.
+         */
+        private MulticastSocket createSocket() throws IOException {
+            MulticastSocket sock = new MulticastSocket(mcastPort);
+
+            sock.setLoopbackMode(false); // Use 'false' to enable support for more than one node on the same machine.
+
+            if (sockItf != null)
+                sock.setInterface(sockItf);
+
+            if (sock.getLoopbackMode())
+                U.warn(log, "Loopback mode is disabled which prevents nodes on the same machine from discovering " +
+                    "each other.");
+
+            sock.joinGroup(mcastGrp);
+
+            return sock;
+        }
+
+        /** {@inheritDoc} */
+        @Override protected void body() throws InterruptedException {
+            AddressResponse res;
+
+            try {
+                res = new AddressResponse(addrs);
+            }
+            catch (GridException e) {
+                U.error(log, "Failed to prepare multicast message.", e);
+
+                return;
+            }
+
+            byte[] reqData = new byte[MSG_ADDR_REQ_DATA.length];
+
+            DatagramPacket pckt = new DatagramPacket(reqData, reqData.length);
+
+            while (!isInterrupted()) {
+                try {
+                    MulticastSocket sock;
+
+                    synchronized (this) {
+                        if (isInterrupted())
+                            return;
+
+                        sock = this.sock;
+
+                        if (sock == null)
+                            sock = createSocket();
+                    }
+
+                    sock.receive(pckt);
+
+                    if (!U.bytesEqual(U.GG_HEADER, 0, reqData, 0, U.GG_HEADER.length)) {
+                        U.error(log, "Failed to verify message header.");
+
+                        continue;
+                    }
+
+                    try {
+                        sock.send(new DatagramPacket(res.data(), res.data().length, pckt.getAddress(), pckt.getPort()));
+                    }
+                    catch (IOException e) {
+                        if (e.getMessage().contains("Operation not permitted")) {
+                            if (log.isDebugEnabled())
+                                log.debug("Got 'operation not permitted' error, ignoring: " + e);
+                        }
+                        else
+                            throw e;
+                    }
+                }
+                catch (IOException e) {
+                    if (!isInterrupted()) {
+                        U.error(log, "Failed to send/receive address message (will try to reconnect).", e);
+
+                        synchronized (this) {
+                            U.close(sock);
+
+                            sock = null;
+                        }
+                    }
+                }
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public void interrupt() {
+            super.interrupt();
+
+            synchronized (this) {
+                U.close(sock);
+
+                sock = null;
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override protected void cleanup() {
+            synchronized (this) {
+                U.close(sock);
+
+                sock = null;
+            }
+        }
+    }
+}


Mime
View raw message