# ignite-537
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/b8870305 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/b8870305 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/b8870305 Branch: refs/heads/ignite-721 Commit: b8870305c74ea5ed489a98de73086e5bb67ff9a3 Parents: c1c1cf9 Author: sboikov <sboi...@gridgain.com> Authored: Fri Apr 10 13:16:28 2015 +0300 Committer: sboikov <sboi...@gridgain.com> Committed: Fri Apr 10 14:48:44 2015 +0300 ---------------------------------------------------------------------- .../continuous/GridContinuousProcessor.java | 25 +- .../spi/discovery/DiscoverySpiDataExchange.java | 9 +- .../discovery/tcp/TcpClientDiscoverySpi.java | 11 +- .../spi/discovery/tcp/TcpDiscoverySpi.java | 482 ++++++++++++------- .../discovery/tcp/TcpDiscoverySpiAdapter.java | 27 ++ .../TcpDiscoveryCustomEventMessage.java | 14 +- .../messages/TcpDiscoveryGetClassResponse.java | 3 + .../continuous/GridEventConsumeSelfTest.java | 2 +- .../discovery/AbstractDiscoverySelfTest.java | 4 +- .../tcp/TcpDiscoverySpiStartStopSelfTest.java | 4 +- .../junits/spi/GridSpiAbstractTest.java | 4 +- 11 files changed, 357 insertions(+), 228 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java index 99b9c0c..54da8e6 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java @@ -30,7 +30,6 @@ import org.apache.ignite.internal.processors.cache.*; import org.apache.ignite.internal.processors.timeout.*; import org.apache.ignite.internal.util.*; import org.apache.ignite.internal.util.future.*; -import org.apache.ignite.internal.util.lang.*; import org.apache.ignite.internal.util.tostring.*; import org.apache.ignite.internal.util.typedef.*; import org.apache.ignite.internal.util.typedef.internal.*; @@ -1560,7 +1559,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter { /** * Discovery data. */ - private static class DiscoveryData implements Externalizable, GridPeerDeployAware { + private static class DiscoveryData implements Externalizable { /** */ private static final long serialVersionUID = 0L; @@ -1578,16 +1577,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter { // No-op. } - @Override - public Class<?> deployClass() { - return U.peerDeployAware0(items).deployClass(); - } - - @Override - public ClassLoader classLoader() { - return U.peerDeployAware0(items).classLoader(); - } - /** * @param nodeId Node ID. */ @@ -1627,7 +1616,7 @@ public class GridContinuousProcessor extends GridProcessorAdapter { /** * Discovery data item. */ - private static class DiscoveryDataItem implements Externalizable, GridPeerDeployAware { + private static class DiscoveryDataItem implements Externalizable { /** */ private static final long serialVersionUID = 0L; @@ -1677,16 +1666,6 @@ public class GridContinuousProcessor extends GridProcessorAdapter { this.interval = interval; } - @Override - public Class<?> deployClass() { - return prjPred.getClass(); - } - - @Override - public ClassLoader classLoader() { - return prjPred.getClass().getClassLoader(); - } - /** {@inheritDoc} */ @Override public void writeExternal(ObjectOutput out) throws IOException { U.writeUuid(out, routineId); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java index 1f64c87..29f6b5f 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiDataExchange.java @@ -38,11 +38,8 @@ public interface DiscoverySpiDataExchange { /** * Notifies discovery manager about data received from remote node. * - * @param joiningNodeId Joining node ID. - * @param nodeId Remote node ID for which data is provided. - * @param data Collection of marshalled discovery data objects from different components. + * @param joiningNodeId Remote node ID. + * @param data Collection of discovery data objects from different components. */ - public void onExchange(UUID joiningNodeId, - UUID nodeId, - Map<Integer, Object> data); + public void onExchange(UUID joiningNodeId, UUID nodeId, Map<Integer, Object> data); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java index f613f4a..4a448f2 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java @@ -896,11 +896,8 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp Map<UUID, Map<Integer, byte[]>> dataMap = msg.oldNodesDiscoveryData(); if (dataMap != null) { - for (Map.Entry<UUID, Map<Integer, byte[]>> entry : dataMap.entrySet()) { -// exchange.onExchange(newNodeId, -// entry.getKey(), -// entry.getValue()); - } + for (Map.Entry<UUID, Map<Integer, byte[]>> entry : dataMap.entrySet()) + onExchange(newNodeId, entry.getKey(), entry.getValue(), null); } locNode.setAttributes(node.attributes()); @@ -922,8 +919,8 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp Map<Integer, byte[]> data = msg.newNodeDiscoveryData(); -// if (data != null) -// exchange.onExchange(newNodeId, newNodeId, data); + if (data != null) + onExchange(newNodeId, newNodeId, data, null); } } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/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 3fa6fd1..bc0fec5 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 @@ -50,6 +50,7 @@ import java.net.*; import java.text.*; import java.util.*; import java.util.concurrent.*; +import java.util.concurrent.locks.*; import static org.apache.ignite.events.EventType.*; import static org.apache.ignite.internal.IgniteNodeAttributes.*; @@ -287,6 +288,15 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized") private ConcurrentLinkedDeque<String> debugLog; + /** */ + private DeploymentClassLoadersCleaner p2pLdrCleaner; + + /** Class loaders for event data unmarshalling. */ + private ConcurrentMap<UUID, DiscoveryDeploymentClassLoader> p2pLdrs = new ConcurrentHashMap<>(); + + /** Class loader for local data unmarshalling. */ + private ClassLoader locLdr = U.gridClassLoader(); + /** {@inheritDoc} */ @IgniteInstanceResource @Override public void injectResources(Ignite ignite) { @@ -764,6 +774,11 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov ipFinderCleaner.start(); } + if (ignite.configuration().isPeerClassLoadingEnabled()) { + p2pLdrCleaner = new DeploymentClassLoadersCleaner(); + p2pLdrCleaner.start(); + } + if (log.isDebugEnabled() && !restart) log.debug(startInfo()); @@ -988,6 +1003,9 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov U.interrupt(statsPrinter); U.join(statsPrinter, log); + U.interrupt(p2pLdrCleaner); + U.join(p2pLdrCleaner, log); + if (ipFinder != null) ipFinder.close(); @@ -1246,7 +1264,16 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov /** {@inheritDoc} */ @Override public void sendCustomEvent(Serializable evt) { - msgWorker.addMessage(new TcpDiscoveryCustomEventMessage(getLocalNodeId(), evt)); + byte[] msgBytes; + + try { + msgBytes = marsh.marshal(evt); + } + catch (IgniteCheckedException e) { + throw new IgniteSpiException("Failed to marshal custom event: " + evt, e); + } + + msgWorker.addMessage(new TcpDiscoveryCustomEventMessage(getLocalNodeId(), msgBytes)); } /** @@ -1377,87 +1404,6 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov log.debug("Discovery SPI has been connected to topology with order: " + locNode.internalOrder()); } - /** */ - private GridConcurrentHashSet<ClassLoader> exchangeLdrs = new GridConcurrentHashSet(); - - /** - * @param nodeId Node ID> - * @return Marshalled exchange data. - * @throws IgniteSpiException If failed. - */ - protected Map<Integer, byte[]> collectExchangeData(UUID nodeId) throws IgniteSpiException { - boolean p2pEnaled = ignite.configuration().isPeerClassLoadingEnabled(); - - Map<Integer, Object> data = exchange.collect(nodeId); - - Map<Integer, byte[]> data0 = U.newHashMap(data.size()); - - for (Map.Entry<Integer, Object> entry : data.entrySet()) { - try { - byte[] bytes = marsh.marshal(entry.getValue()); - - data0.put(entry.getKey(), bytes); - - if (p2pEnaled) { - ClassLoader ldr; - - if (entry.getValue() instanceof GridPeerDeployAware) - ldr = ((GridPeerDeployAware)entry.getValue()).classLoader(); - else - ldr = entry.getValue().getClass().getClassLoader(); - - log.info("Add loader: " + entry.getValue() + " " + ldr); - - if (ldr != null) { - if (exchangeLdrs.add(ldr)) - log.info("Added loader: " + ldr); - } - } - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to marshal discovery data " + - "[comp=" + entry.getKey() + ", data=" + entry.getValue() + ']', e); - - throw new IgniteSpiException("Failed to marshal discovery data.", e); - } - } - - return data0; - } - - /** - * @param joiningNode Joining node. - * @param nodeId Remote node ID for which data is provided. - * @param data Collection of marshalled discovery data objects from different components. - */ - protected void onExchange(TcpDiscoveryNode joiningNode, - UUID nodeId, - Map<Integer, byte[]> data) { - TcpDiscoveryNode node; - - if (joiningNode.id().equals(nodeId)) - node = joiningNode; - else - node = ring.node(nodeId); - - ClassLoader clsLdr = exchangeClassLoader(node); - - Map<Integer, Object> data0 = U.newHashMap(data.size()); - - for (Map.Entry<Integer, byte[]> entry : data.entrySet()) { - try { - Object compData = marsh.unmarshal(entry.getValue(), clsLdr); - - data0.put(entry.getKey(), compData); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to unmarshal discovery data for component: " + entry.getKey(), e); - } - } - - exchange.onExchange(joiningNode.id(), nodeId, data0); - } - /** * Tries to send join request message to a random node presenting in topology. * Address is provided by {@link TcpDiscoveryIpFinder} and message is @@ -2275,14 +2221,13 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov /** * @param req Get class request. + * @return Get class response. */ private TcpDiscoveryGetClassResponse processGetClassRequest(TcpDiscoveryGetClassRequest req) { assert !F.isEmpty(req.className()) : req; String rsrc = U.classNameToResourceName(req.className()); - log.info("Get class request: " + req.className() + " " + rsrc); - InputStream in = locLdr.getResourceAsStream(rsrc); byte[] clsBytes = null; @@ -2559,6 +2504,41 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov } /** + * + */ + private class DeploymentClassLoadersCleaner extends IgniteSpiThread { + /** + * Constructor. + */ + private DeploymentClassLoadersCleaner() { + super(gridName, "tcp-disco-p2p-ldr-cleaner", log); + + setPriority(threadPri); + } + + /** {@inheritDoc} */ + @SuppressWarnings("BusyWait") + @Override protected void body() throws InterruptedException { + if (log.isDebugEnabled()) + log.debug("Deployment class loader cleaner has been started."); + + while (!isInterrupted()) { + Thread.sleep(5000); + + for (DiscoveryDeploymentClassLoader ldr : p2pLdrs.values()) { + if (ring.node(ldr.nodeId()) == null) { + ldr.onNodeLeft(); + + p2pLdrs.remove(ldr.nodeId(), ldr); + } + else + ldr.closeConnectionIfNotUsed(); + } + } + } + } + + /** * Pending messages container. */ private static class PendingMessages { @@ -3489,10 +3469,10 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov private void processClientReconnectMessage(TcpDiscoveryClientReconnectMessage msg) { UUID locNodeId = getLocalNodeId(); - boolean isLocalNodeRouter = locNodeId.equals(msg.routerNodeId()); + boolean isLocNodeRouter = locNodeId.equals(msg.routerNodeId()); if (!msg.verified()) { - assert isLocalNodeRouter; + assert isLocNodeRouter; msg.verify(locNodeId); } @@ -3530,7 +3510,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov else if (log.isDebugEnabled()) log.debug("Reconnecting client node is already failed [nodeId=" + nodeId + ']'); - if (isLocalNodeRouter) { + if (isLocNodeRouter) { ClientMessageWorker wrk = clientMsgWorkers.get(nodeId); if (wrk != null) @@ -3676,7 +3656,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov Map<Integer, byte[]> data = msg.newNodeDiscoveryData(); if (data != null) - onExchange(node, node.id(), data); + onExchange(node.id(), node.id(), data, exchangeClassLoader(node, node.id())); msg.addDiscoveryData(locNodeId, collectExchangeData(node.id())); } @@ -3747,8 +3727,12 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov // Notify outside of synchronized block. if (dataMap != null) { - for (Map.Entry<UUID, Map<Integer, byte[]>> entry : dataMap.entrySet()) - onExchange(node, entry.getKey(), entry.getValue()); + for (Map.Entry<UUID, Map<Integer, byte[]>> entry : dataMap.entrySet()) { + onExchange(node.id(), + entry.getKey(), + entry.getValue(), + exchangeClassLoader(node, entry.getKey())); + } } } @@ -4524,13 +4508,27 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov Collection<ClusterNode> snapshot = hist.get(msg.topologyVersion()); - if (lsnr != null && (spiState == CONNECTED || spiState == DISCONNECTING)) + if (lsnr != null && (spiState == CONNECTED || spiState == DISCONNECTING)) { + assert msg.messageBytes() != null; + + TcpDiscoveryNode node = ring.node(msg.creatorNodeId()); + + Serializable msgObj; + + try { + msgObj = marsh.unmarshal(msg.messageBytes(), customMessageClassLoader(node)); + } + catch (IgniteCheckedException e) { + throw new IgniteSpiException("Failed to unmarshal discovery custom message.", e); + } + lsnr.onDiscovery(DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT, msg.topologyVersion(), - ring.node(msg.creatorNodeId()), + node, snapshot, hist, - msg.message()); + msgObj); + } } if (ring.hasRemoteNodes()) @@ -5303,177 +5301,311 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov } } - /** */ - private ConcurrentMap<UUID, DiscoveryDeploymentClassLoader> ldrs = new ConcurrentHashMap<>(); - - /** */ - private LocalDeploymentClassLoader locLdr = new LocalDeploymentClassLoader(null); - /** - * + * @param node Node created event. + * @return Class loader for custom event unmarshalling. */ - class LocalDeploymentClassLoader extends ClassLoader { - public LocalDeploymentClassLoader(ClassLoader parent) { - super(parent); - } - - @Override - public InputStream getResourceAsStream(String name) { - log.info("Local getResourceAsStream: " + name); - - for (ClassLoader ldr : exchangeLdrs) { - InputStream in = ldr.getResourceAsStream(name); + @Nullable protected ClassLoader customMessageClassLoader(TcpDiscoveryNode node) { + assert ignite != null; - if (in != null) - return in; - } + if (!ignite.configuration().isPeerClassLoadingEnabled()) + return null; - return super.getResourceAsStream(name); - } + if (node.id().equals(getLocalNodeId())) + return locLdr; - @Override public Class<?> loadClass(String name) throws ClassNotFoundException { - log.info("Local loadClass: " + name); + DiscoveryDeploymentClassLoader ldr = p2pLdrs.get(node.id()); - for (ClassLoader ldr : exchangeLdrs) { - try { - return ldr.loadClass(name); - } - catch (ClassNotFoundException ignore) { - } - } + if (ldr == null) + ldr = F.addIfAbsent(p2pLdrs, node.id(), new DiscoveryDeploymentClassLoader(node)); - return super.loadClass(name); - } + return ldr; } /** - * @param nodeId Node ID. + * @param joiningNode Joining node. + * @param nodeId Remote node provided data. * @return Class loader for exchange data unmarshalling. */ - protected ClassLoader exchangeClassLoader(TcpDiscoveryNode node) { + @Nullable protected ClassLoader exchangeClassLoader(TcpDiscoveryNode joiningNode, UUID nodeId) { + assert joiningNode != null; assert ignite != null; - UUID nodeId = node.id(); - if (!ignite.configuration().isPeerClassLoadingEnabled()) return null; if (nodeId.equals(getLocalNodeId())) return locLdr; - DiscoveryDeploymentClassLoader ldr = ldrs.get(nodeId); + TcpDiscoveryNode node; + + if (joiningNode.id().equals(nodeId)) + node = joiningNode; + else { + node = ring.node(nodeId); + + if (node == null) { + if (log.isDebugEnabled()) + log.debug("Node provided exchange data left, will use local class loader " + + "for exchange data [nodeId=" + nodeId + ']'); + + return locLdr; + } + } + + if (node.isClient()) // Do not support loading from client nodes. + return locLdr; + + DiscoveryDeploymentClassLoader ldr = p2pLdrs.get(nodeId); if (ldr == null) - ldr = F.addIfAbsent(ldrs, nodeId, new DiscoveryDeploymentClassLoader(node)); + ldr = F.addIfAbsent(p2pLdrs, nodeId, new DiscoveryDeploymentClassLoader(node)); return ldr; } /** + * @param nodeId Node ID. + * @return Marshalled exchange data. + * @throws IgniteSpiException If failed. + */ + private Map<Integer, byte[]> collectExchangeData(UUID nodeId) throws IgniteSpiException { + Map<Integer, Object> data = exchange.collect(nodeId); + + Map<Integer, byte[]> data0 = U.newHashMap(data.size()); + + for (Map.Entry<Integer, Object> entry : data.entrySet()) { + try { + byte[] bytes = marsh.marshal(entry.getValue()); + + data0.put(entry.getKey(), bytes); + } + catch (IgniteCheckedException e) { + U.error(log, "Failed to marshal discovery data " + + "[comp=" + entry.getKey() + ", data=" + entry.getValue() + ']', e); + + throw new IgniteSpiException("Failed to marshal discovery data.", e); + } + } + + return data0; + } + + /** * */ private class DiscoveryDeploymentClassLoader extends ClassLoader { /** */ - private final TcpDiscoveryNode node; + private final UUID nodeId; + + /** */ + private volatile TcpDiscoveryNode node; /** */ private Socket sock; + /** */ + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + /** * @param node Node. */ public DiscoveryDeploymentClassLoader(TcpDiscoveryNode node) { + assert !node.isClient(); assert !node.id().equals(getLocalNodeId()); this.node = node; + + nodeId = node.id(); } - @Override protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException { - // log.info("P2p load class: " + name); + /** + * @return Target node ID. + */ + UUID nodeId() { + return nodeId; + } - return super.loadClass(name, resolve); + /** + * Node left callback. + */ + void onNodeLeft() { + lock.writeLock().lock(); + + try { + if (sock != null) { + if (log.isDebugEnabled()) + log.debug("Closing deployment class loader connection on node left [node=" + nodeId + ']'); + + U.closeQuiet(sock); + + sock = null; + } + + node = null; + } + finally { + lock.writeLock().unlock(); + } + } + + /** + * Closes connection if there is no class loading in progress. + */ + void closeConnectionIfNotUsed() { + if (lock.writeLock().tryLock()) { + try { + if (sock != null) { + if (log.isDebugEnabled()) + log.debug("Closing idle deployment class loader connection [node=" + nodeId + ']'); + + U.closeQuiet(sock); + + sock = null; + } + } + finally { + lock.writeLock().unlock(); + } + } } /** {@inheritDoc} */ @Override protected Class<?> findClass(String name) throws ClassNotFoundException { - log.info("P2p find class: " + name); + if (node == null) + throw new ClassNotFoundException("Failed to load class, peer node left " + + "[cls=" +name + ", node=" + nodeId + ']'); - TcpDiscoveryGetClassResponse res = requestClass(name); + lock.readLock().lock(); - if (res.error() != null) - throw new ClassNotFoundException(res.error()); + try { + TcpDiscoveryGetClassResponse res = requestClass(name); - log.info("P2p loaded: " + name); + if (res == null) + throw new ClassNotFoundException("Failed to load class, can not connect to peer node " + + "[cls=" + name + ", node=" + nodeId + ']'); - assert res.classBytes() != null; + if (res.error() != null) + throw new ClassNotFoundException(res.error()); - Class<?> cls = defineClass(name, res.classBytes(), 0, res.classBytes().length); + assert res.classBytes() != null; - return cls; + return defineClass(name, res.classBytes(), 0, res.classBytes().length); + } + finally { + lock.readLock().unlock(); + } } /** * @param name Class name. - * @return Class response. - * @throws ClassNotFoundException If request failed. + * @return Class response or {@code null} if failed to connect. */ - private TcpDiscoveryGetClassResponse requestClass(String name) throws ClassNotFoundException { - sock = connect(); + @Nullable private synchronized TcpDiscoveryGetClassResponse requestClass(String name) { + TcpDiscoveryGetClassRequest msg = new TcpDiscoveryGetClassRequest(getLocalNodeId(), name); - if (sock == null) - throw new ClassNotFoundException("Failed to load class, can not connect to peer node " + - "[cls=" + name + ", node=" + nodeId + ']'); + for (int i = 0; i < reconCnt; i++) { + if (sock == null) { + TcpDiscoveryNode node0 = node; - try { - writeToSocket(sock, new TcpDiscoveryGetClassRequest(getLocalNodeId(), name)); + if (node0 == null) + return null; // Node left. - TcpDiscoveryGetClassResponse res = readMessage(sock, null, netTimeout); + sock = connect(node0); - return res; - } - catch (IOException | IgniteCheckedException e) { - e.printStackTrace(); + if (sock == null) + break; + } + + try { + return request(sock, msg); + } + catch (IOException | IgniteCheckedException e) { + U.closeQuiet(sock); - throw new ClassNotFoundException("Failed to load class: " + name, e); + sock = null; + } } + + node = null; // Consider node failed. + + p2pLdrs.remove(nodeId, this); + + return null; } - private Socket connect() { - if (sock == null) { - for (InetSocketAddress addr : getNodeAddresses(node, U.sameMacs(locNode, node))) { - sock = connect(addr); + /** + * @param sock Socket. + * @param msg Message. + * @return Response. + * @throws IOException If request failed. + * @throws IgniteCheckedException If request failed. + */ + private TcpDiscoveryGetClassResponse request(Socket sock, TcpDiscoveryGetClassRequest msg) + throws IOException, IgniteCheckedException + { + long tstamp = U.currentTimeMillis(); - if (sock != null) - break; - } + writeToSocket(sock, msg); + + stats.onMessageSent(msg, U.currentTimeMillis() - tstamp); + + TcpDiscoveryGetClassResponse res = readMessage(sock, null, netTimeout); + + stats.onMessageReceived(res); + + return res; + } + + /** + * @param node Node. + * @return Socket or {@code null} if failed to connect. + */ + private Socket connect(TcpDiscoveryNode node) { + Socket sock = null; + + for (InetSocketAddress addr : getNodeAddresses(node, U.sameMacs(locNode, node))) { + sock = connect(addr); + + if (sock != null) + break; } return sock; } + /** + * @param addr Address. + * @return Socket or {@code null} if failed to connect. + */ private Socket connect(InetSocketAddress addr) { - UUID locNodeId = getLocalNodeId(); - - Socket sock = null; + TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest(getLocalNodeId()); for (int i = 0; i < reconCnt; i++) { + Socket sock = null; + + long tstamp = U.currentTimeMillis(); + try { sock = openSocket(addr); - writeToSocket(sock, new TcpDiscoveryHandshakeRequest(locNodeId)); + writeToSocket(sock, req); TcpDiscoveryHandshakeResponse res = readMessage(sock, null, netTimeout); - break; + if (!res.creatorNodeId().equals(nodeId)) + return null; + + stats.onClientSocketInitialized(U.currentTimeMillis() - tstamp); + + return sock; } catch (IOException | IgniteCheckedException e) { - e.printStackTrace(); - U.closeQuiet(sock); } } - return sock; + return null; } } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java index 9cde198..db75051 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java @@ -744,6 +744,33 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements Discov } /** + * @param joiningNodeID Joining node ID. + * @param nodeId Remote node ID for which data is provided. + * @param data Collection of marshalled discovery data objects from different components. + * @param clsLdr Class loader for discovery data unmarshalling. + */ + protected void onExchange(UUID joiningNodeID, + UUID nodeId, + Map<Integer, byte[]> data, + ClassLoader clsLdr) + { + Map<Integer, Object> data0 = U.newHashMap(data.size()); + + for (Map.Entry<Integer, byte[]> entry : data.entrySet()) { + try { + Object compData = marsh.unmarshal(entry.getValue(), clsLdr); + + data0.put(entry.getKey(), compData); + } + catch (IgniteCheckedException e) { + U.error(log, "Failed to unmarshal discovery data for component: " + entry.getKey(), e); + } + } + + exchange.onExchange(joiningNodeID, nodeId, data0); + } + + /** * Handles sockets timeouts. */ protected class SocketTimeoutWorker extends IgniteSpiThread { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java index 234efaa..6f78953 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java @@ -17,7 +17,6 @@ package org.apache.ignite.spi.discovery.tcp.messages; -import java.io.*; import java.util.*; /** @@ -29,21 +28,22 @@ public class TcpDiscoveryCustomEventMessage extends TcpDiscoveryAbstractMessage private static final long serialVersionUID = 0L; /** */ - private final Serializable msg; + private final byte[] msgBytes; /** * @param creatorNodeId Creator node id. + * @param msgBytes Serialized message. */ - public TcpDiscoveryCustomEventMessage(UUID creatorNodeId, Serializable msg) { + public TcpDiscoveryCustomEventMessage(UUID creatorNodeId, byte[] msgBytes) { super(creatorNodeId); - this.msg = msg; + this.msgBytes = msgBytes; } /** - * @return Message. + * @return Serialized message. */ - public Serializable message() { - return msg; + public byte[] messageBytes() { + return msgBytes; } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryGetClassResponse.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryGetClassResponse.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryGetClassResponse.java index b1afe1c..d377aee 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryGetClassResponse.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryGetClassResponse.java @@ -26,6 +26,9 @@ import java.util.*; */ public class TcpDiscoveryGetClassResponse extends TcpDiscoveryAbstractMessage { /** */ + private static final long serialVersionUID = 0L; + + /** */ private String errMsg; /** */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java index 4bd2901..4273c9d 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java @@ -763,7 +763,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements * @throws Exception If failed. */ // TODO: GG-6730 - public void testNodeJoinWithP2P() throws Exception { + public void _testNodeJoinWithP2P() throws Exception { final Collection<UUID> nodeIds = new HashSet<>(); final AtomicInteger cnt = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(GRID_CNT + 1); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java index 91d3514..234e3cd 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java @@ -383,9 +383,7 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri return new HashMap<>(); } - @Override public void onExchange(UUID joiningNodeId, - UUID nodeId, - Map<Integer, Object> data) { + @Override public void onExchange(UUID joiningNodeId, UUID nodeId, Map<Integer, Object> data) { // No-op. } }); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java index 058d908..b0e22b4 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiStartStopSelfTest.java @@ -48,9 +48,7 @@ public class TcpDiscoverySpiStartStopSelfTest extends GridSpiStartStopAbstractTe return null; } - @Override public void onExchange(UUID joiningNodeId, - UUID nodeId, - Map<Integer, Object> data) { + @Override public void onExchange(UUID joiningNodeId, UUID nodeId, Map<Integer, Object> data) { // No-op. } }; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b8870305/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java index 82f3d6c..7898c3d 100644 --- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/spi/GridSpiAbstractTest.java @@ -219,9 +219,7 @@ public abstract class GridSpiAbstractTest<T extends IgniteSpi> extends GridAbstr return new HashMap<>(); } - @Override public void onExchange(UUID joiningNodeId, - UUID nodeId, - Map<Integer, Object> data) { + @Override public void onExchange(UUID joiningNodeId, UUID nodeId, Map<Integer, Object> data) { } });
