You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/04/24 15:04:09 UTC
[46/50] [abbrv] ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-2.0' into ignite-3054
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 305af3d,34ee414..425560e
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@@ -127,8 -131,8 +132,11 @@@ class ClientImpl extends TcpDiscoveryIm
private static final Object SPI_RECONNECT_FAILED = "SPI_RECONNECT_FAILED";
/** */
+ private static final Object SPI_RECONNECT = "SPI_RECONNECT";
+
++ /** */
+ private static final int DFLT_BYTE_ARR_STREAM_SIZE = 32 * 1024;
+
/** Remote nodes. */
private final ConcurrentMap<UUID, TcpDiscoveryNode> rmtNodes = new ConcurrentHashMap8<>();
@@@ -637,12 -647,9 +654,12 @@@
msg.client(true);
- spi.writeToSocket(sock, msg, timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
+ if (async)
+ writeToSocketWithLength(msg, sock, timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
+ else
+ spi.writeToSocket(sock, msg, timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
- spi.stats.onMessageSent(msg, U.currentTimeMillis() - tstamp, 0);
+ spi.stats.onMessageSent(msg, U.currentTimeMillis() - tstamp);
if (log.isDebugEnabled())
log.debug("Message has been sent to address [msg=" + msg + ", addr=" + addr +
@@@ -1023,11 -1058,8 +1095,14 @@@
private TcpDiscoveryAbstractMessage unackedMsg;
/** */
+ private boolean writeLen;
+
+ /** */
+ private volatile long lastMsgReceived;
+
++ /** */
+ private CountDownLatch forceLeaveLatch;
+
/**
*
*/
@@@ -1050,11 -1082,33 +1125,34 @@@
}
/**
+ * Sends {@link TcpDiscoveryNodeLeftMessage} and closes socket.
+ *
+ * @throws InterruptedException If interrupted.
+ */
+ private void forceLeave() throws InterruptedException {
+ CountDownLatch forceLeaveLatch;
+
+ synchronized (mux) {
+ // If writer was stopped.
+ if (sock == null)
+ return;
+
+ this.forceLeaveLatch = forceLeaveLatch = new CountDownLatch(1);
+
+ unackedMsg = null;
+
+ mux.notifyAll();
+ }
+
+ forceLeaveLatch.await();
+ }
+
+ /**
* @param sock Socket.
* @param clientAck {@code True} is server supports client message acknowlede.
+ * @param writeLen Add message length if {@code true}.
*/
- private void setSocket(Socket sock, boolean clientAck) {
+ private void setSocket(Socket sock, boolean clientAck, boolean writeLen) {
synchronized (mux) {
this.sock = sock;
@@@ -1092,16 -1144,9 +1190,16 @@@
}
}
+ /**
+ * Updates timeout when message received.
+ */
+ void messageReceived() {
+ lastMsgReceived = U.currentTimeMillis();
+ }
+
/** {@inheritDoc} */
@Override protected void body() throws InterruptedException {
- TcpDiscoveryAbstractMessage msg;
+ TcpDiscoveryAbstractMessage msg = null;
while (!Thread.currentThread().isInterrupted()) {
Socket sock;
@@@ -1586,7 -1772,15 +1831,15 @@@
joinCnt++;
- final JoinResult joinRes = joinTopology(false, spi.joinTimeout);
- T2<SocketStream, Boolean> joinRes;
++ final JoinResult joinRes;
+ try {
+ joinRes = joinTopology(false, spi.joinTimeout);
+ }
+ catch (IgniteSpiException e) {
+ joinError(e);
+
+ return;
+ }
if (joinRes == null) {
if (join)
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 758b576,47c13e1..0aa6625
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@@ -195,13 -169,6 +195,13 @@@ class ServerImpl extends TcpDiscoveryIm
/** */
private static final int ENSURED_MSG_HIST_SIZE = getInteger(IGNITE_DISCOVERY_CLIENT_RECONNECT_HISTORY_SIZE, 512);
++ /** ClientNioMessageWorker in GridNioSession. */
++ private static final int NIO_WORKER_META = GridNioSessionMetaKey.nextUniqueKey();
++
+ /** */
+ private static final IgniteProductVersion CUSTOM_MSG_ALLOW_JOINING_FOR_VERIFIED_SINCE =
+ IgniteProductVersion.fromString("1.5.0");
+
- /** ClientNioMessageWorker in GridNioSession. */
- private static final int NIO_WORKER_META = GridNioSessionMetaKey.nextUniqueKey();
-
/** */
private IgniteThreadPoolExecutor utilityPool;
@@@ -355,19 -324,9 +363,20 @@@
msgWorker = new RingMessageWorker();
msgWorker.start();
- tcpSrvr = new TcpServer();
+ if (tcpSrvr == null)
+ tcpSrvr = new TcpServer();
+ clientNioSrv = createClientNioServer(gridName);
+ clientNioSrv.start();
+
+ nioClientProcessingPool = new IgniteThreadPoolExecutor(
+ "disco-client-nio-msg-processor",
+ gridName,
+ spi.getClientNioThreads(),
+ spi.getClientNioThreads() * 2,
+ 60_000L,
+ new LinkedBlockingQueue<Runnable>());
+
spi.initLocalNode(tcpSrvr.port, true);
locNode = spi.locNode;
@@@ -4833,22 -4683,17 +4831,24 @@@
}
synchronized (mux) {
- failedNodes.remove(node);
+ failedNodes.remove(failedNode);
- leavingNodes.remove(node);
+ leavingNodes.remove(failedNode);
- failedNodesMsgSent.remove(node.id());
+ failedNodesMsgSent.remove(failedNode.id());
- final ClientMessageProcessor worker = clientMsgWorkers.remove(node.id());
+ if (!msg.force()) { // ClientMessageWorker will stop after sending force fail message.
- ClientMessageWorker worker = clientMsgWorkers.remove(failedNode.id());
++ final ClientMessageProcessor worker = clientMsgWorkers.remove(node.id());
+
- if (worker != null) {
- if (worker instanceof ClientMessageWorker) {
- final ClientMessageWorker wrk = (ClientMessageWorker)worker;
++ if (worker != null) {
++ if (worker instanceof ClientMessageWorker) {
++ final ClientMessageWorker wrk = (ClientMessageWorker)worker;
- wrk.interrupt();
- if (worker != null)
- worker.interrupt();
++ wrk.interrupt();
++ }
++ else if (worker instanceof ClientNioMessageWorker)
++ ((ClientNioMessageWorker)worker).nonblockingStop();
+ }
- else if (worker instanceof ClientNioMessageWorker)
- ((ClientNioMessageWorker)worker).nonblockingStop();
}
}
@@@ -6441,950 -5998,42 +6438,954 @@@
return;
}
- catch (IOException e) {
+
+ // Handshake.
+ TcpDiscoveryHandshakeRequest req = (TcpDiscoveryHandshakeRequest)msg;
+
+ UUID nodeId = req.creatorNodeId();
+
+ this.nodeId = nodeId;
+
+ TcpDiscoveryHandshakeResponse res =
+ new TcpDiscoveryHandshakeResponse(locNodeId, locNode.internalOrder());
+
+ boolean asyncMode = false;
+
+ if (req.client()) {
+ res.clientAck(true);
+
+ // If client proposes async mode accept it.
+ if (req.asyncMode()) {
+ res.asyncMode(true);
+
+ asyncMode = true;
+ }
+ }
+
+ // It can happen if a remote node is stopped and it has a loopback address in the list of addresses,
+ // the local node sends a handshake request message on the loopback address, so we get here.
+ if (locNodeId.equals(nodeId)) {
+ assert !req.client();
+
if (log.isDebugEnabled())
- U.error(log, "Caught exception on message read [sock=" + sock + ", locNodeId=" + locNodeId +
- ", rmtNodeId=" + nodeId + ']', e);
+ log.debug("Handshake request from local node: " + req);
- if (isInterrupted() || sock.isClosed())
- return;
+ return;
+ }
+
+ if (req.client()) {
+ ClientMessageProcessor clientProc = asyncMode
+ ? new ClientNioMessageWorker(nodeId, sock)
+ : new ClientMessageWorker(sock, nodeId);
+
+ while (true) {
+ ClientMessageProcessor old = clientMsgWorkers.putIfAbsent(nodeId, clientProc);
+
+ if (old == null)
+ break;
+
+ if (old instanceof ClientMessageWorker) {
+ ClientMessageWorker oldWrk = (ClientMessageWorker)old;
+
+ if (oldWrk.isInterrupted()) {
+ clientMsgWorkers.remove(nodeId, old);
+
+ continue;
+ }
+
+ oldWrk.join(500);
+
+ old = clientMsgWorkers.putIfAbsent(nodeId, clientProc);
+
+ if (old == null)
+ break;
+
+ if (log.isDebugEnabled())
+ log.debug("Already have client message worker, closing connection " +
+ "[locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId +
+ ", workerSock=" + oldWrk.sock +
+ ", sock=" + sock + ']');
+
+ return;
+ }
+ else if (old instanceof ClientNioMessageWorker) {
+ final ClientNioMessageWorker nioOldWrk = (ClientNioMessageWorker)old;
+
+ if (nioOldWrk.state() == WorkerState.STOPPED)
+ clientMsgWorkers.remove(nodeId, nioOldWrk);
+ else {
+ if (nioOldWrk.waitForState(500, WorkerState.STOPPED)) {
+ old = clientMsgWorkers.putIfAbsent(nodeId, clientProc);
+
+ if (old == null)
+ break;
+ }
+
+ if (log.isDebugEnabled())
+ log.debug("Already have client message worker, closing connection " +
+ "[locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId +
+ ", workerSock=" + nioOldWrk.sock +
+ ", sock=" + sock + ']');
+
+ return;
+ }
+ }
+ }
+
+ if (log.isDebugEnabled())
+ log.debug("Created client message worker [locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId + ", sock=" + sock + ']');
+
+ assert clientProc == clientMsgWorkers.get(nodeId);
+
+ clientMsgWrk = clientProc;
+ }
+
+ if (clientMsgWrk instanceof ClientNioMessageWorker) {
+ final ClientNioMessageWorker nioWrk = (ClientNioMessageWorker)clientMsgWrk;
+
+ nioWrk.start();
+
+ nioWrk.sendMessage(res, null);
+
+ nioClient = true;
+
+ return;
+ }
+
+ spi.writeToSocket(sock, res, spi.failureDetectionTimeoutEnabled() ?
+ spi.failureDetectionTimeout() : spi.getSocketTimeout());
+
+ if (log.isDebugEnabled())
+ log.debug("Initialized connection with remote node [nodeId=" + nodeId +
+ ", client=" + req.client() + ']');
+
+ if (debugMode) {
+ debugLog(msg, "Initialized connection with remote node [nodeId=" + nodeId +
+ ", client=" + req.client() + ']');
+ }
+ }
+ catch (IOException e) {
+ if (log.isDebugEnabled())
+ U.error(log, "Caught exception on handshake [err=" + e +", sock=" + sock + ']', e);
+
+ if (X.hasCause(e, SSLException.class) && spi.isSslEnabled() && !spi.isNodeStopping0())
+ LT.warn(log, "Failed to initialize connection " +
+ "(missing SSL configuration on remote node?) " +
+ "[rmtAddr=" + sock.getInetAddress() + ']', true);
+ else if ((X.hasCause(e, ObjectStreamException.class) || !sock.isClosed())
+ && !spi.isNodeStopping0()) {
+ if (U.isMacInvalidArgumentError(e))
+ LT.error(log, e, "Failed to initialize connection [sock=" + sock + "]\n\t" +
+ U.MAC_INVALID_ARG_MSG);
+ else {
+ U.error(
+ log,
+ "Failed to initialize connection (this can happen due to short time " +
+ "network problems and can be ignored if does not affect node discovery) " +
+ "[sock=" + sock + ']',
+ e);
+ }
+ }
+
+ onException("Caught exception on handshake [err=" + e + ", sock=" + sock + ']', e);
+
+ return;
+ }
+ catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ U.error(log, "Caught exception on handshake [err=" + e +", sock=" + sock + ']', e);
+
+ onException("Caught exception on handshake [err=" + e +", sock=" + sock + ']', e);
+
+ if (e.hasCause(SocketTimeoutException.class))
+ LT.warn(log, "Socket operation timed out on handshake " +
+ "(consider increasing 'networkTimeout' configuration property) " +
+ "[netTimeout=" + spi.netTimeout + ']');
+
+ else if (e.hasCause(ClassNotFoundException.class))
+ LT.warn(log, "Failed to read message due to ClassNotFoundException " +
+ "(make sure same versions of all classes are available on all nodes) " +
+ "[rmtAddr=" + sock.getRemoteSocketAddress() +
+ ", err=" + X.cause(e, ClassNotFoundException.class).getMessage() + ']');
+
+ // Always report marshalling problems.
+ else if (e.hasCause(ObjectStreamException.class) ||
+ (!sock.isClosed() && !e.hasCause(IOException.class)))
+ LT.error(log, e, "Failed to initialize connection [sock=" + sock + ']');
+
+ return;
+ }
+
+ long sockTimeout = spi.failureDetectionTimeoutEnabled() ? spi.failureDetectionTimeout() :
+ spi.getSocketTimeout();
+
+ final ClientMessageWorker clientMsgWrk0 = clientMsgWrk == null ? null
+ : (ClientMessageWorker)clientMsgWrk;
+
+ while (!isInterrupted()) {
+ try {
+ TcpDiscoveryAbstractMessage msg = U.unmarshal(spi.marshaller(), in,
+ U.resolveClassLoader(spi.ignite().configuration()));
+
+ msg.senderNodeId(nodeId);
+
+ DebugLogger debugLog = messageLogger(msg);
+
+ if (debugLog.isDebugEnabled())
+ debugLog.debug("Message has been received: " + msg);
+
+ spi.stats.onMessageReceived(msg);
+
+ if (debugMode && recordable(msg))
+ debugLog(msg, "Message has been received: " + msg);
+
+ if (msg instanceof TcpDiscoveryConnectionCheckMessage) {
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ continue;
+ }
+ else if (msg instanceof TcpDiscoveryJoinRequestMessage) {
+ TcpDiscoveryJoinRequestMessage req = (TcpDiscoveryJoinRequestMessage)msg;
+
+ if (!req.responded()) {
+ boolean ok = processJoinRequestMessage(req, clientMsgWrk0);
+
+ if (clientMsgWrk0 != null && ok)
+ continue;
+ else
+ // Direct join request - no need to handle this socket anymore.
+ break;
+ }
+ }
+ else if (msg instanceof TcpDiscoveryClientReconnectMessage) {
+ if (clientMsgWrk0 != null) {
+ TcpDiscoverySpiState state = spiStateCopy();
+
+ if (state == CONNECTED) {
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ if (clientMsgWrk0.getState() == State.NEW)
+ clientMsgWrk0.start();
+
+ msgWorker.addMessage(msg);
+
+ continue;
+ }
+ else {
+ spi.writeToSocket(msg, sock, RES_CONTINUE_JOIN, sockTimeout);
+
+ break;
+ }
+ }
+ }
+ else if (msg instanceof TcpDiscoveryDuplicateIdMessage) {
+ // Send receipt back.
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ boolean ignored = false;
+
+ TcpDiscoverySpiState state = null;
+
+ synchronized (mux) {
+ if (spiState == CONNECTING) {
+ joinRes.set(msg);
+
+ spiState = DUPLICATE_ID;
+
+ mux.notifyAll();
+ }
+ else {
+ ignored = true;
+
+ state = spiState;
+ }
+ }
+
+ if (ignored && log.isDebugEnabled())
+ log.debug("Duplicate ID message has been ignored [msg=" + msg +
+ ", spiState=" + state + ']');
+
+ continue;
+ }
+ else if (msg instanceof TcpDiscoveryAuthFailedMessage) {
+ // Send receipt back.
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ boolean ignored = false;
+
+ TcpDiscoverySpiState state = null;
+
+ synchronized (mux) {
+ if (spiState == CONNECTING) {
+ joinRes.set(msg);
+
+ spiState = AUTH_FAILED;
+
+ mux.notifyAll();
+ }
+ else {
+ ignored = true;
+
+ state = spiState;
+ }
+ }
+
+ if (ignored && log.isDebugEnabled())
+ log.debug("Auth failed message has been ignored [msg=" + msg +
+ ", spiState=" + state + ']');
+
+ continue;
+ }
+ else if (msg instanceof TcpDiscoveryCheckFailedMessage) {
+ // Send receipt back.
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ boolean ignored = false;
+
+ TcpDiscoverySpiState state = null;
+
+ synchronized (mux) {
+ if (spiState == CONNECTING) {
+ joinRes.set(msg);
+
+ spiState = CHECK_FAILED;
+
+ mux.notifyAll();
+ }
+ else {
+ ignored = true;
+
+ state = spiState;
+ }
+ }
+
+ if (ignored && log.isDebugEnabled())
+ log.debug("Check failed message has been ignored [msg=" + msg +
+ ", spiState=" + state + ']');
+
+ continue;
+ }
+ else if (msg instanceof TcpDiscoveryLoopbackProblemMessage) {
+ // Send receipt back.
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ boolean ignored = false;
+
+ TcpDiscoverySpiState state = null;
+
+ synchronized (mux) {
+ if (spiState == CONNECTING) {
+ joinRes.set(msg);
+
+ spiState = LOOPBACK_PROBLEM;
+
+ mux.notifyAll();
+ }
+ else {
+ ignored = true;
+
+ state = spiState;
+ }
+ }
+
+ if (ignored && log.isDebugEnabled())
+ log.debug("Loopback problem message has been ignored [msg=" + msg +
+ ", spiState=" + state + ']');
+
+ continue;
+ }
+ if (msg instanceof TcpDiscoveryPingResponse) {
+ assert msg.client() : msg;
+
+ ClientMessageProcessor clientWorker = clientMsgWorkers.get(msg.creatorNodeId());
+
+ if (clientWorker != null)
+ clientWorker.pingResult(true);
+
+ continue;
+ }
+
+ TcpDiscoveryClientHeartbeatMessage heartbeatMsg = null;
+
+ if (msg instanceof TcpDiscoveryClientHeartbeatMessage)
+ heartbeatMsg = (TcpDiscoveryClientHeartbeatMessage)msg;
+ else
+ msgWorker.addMessage(msg);
+
+ // Send receipt back.
+ if (clientMsgWrk != null) {
+ TcpDiscoveryClientAckResponse ack = new TcpDiscoveryClientAckResponse(locNodeId, msg.id());
+
+ ack.verify(locNodeId);
+
+ clientMsgWrk.addMessage(ack);
+ }
+ else
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ if (heartbeatMsg != null)
+ processClientHeartbeatMessage(heartbeatMsg);
+ }
+ catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ U.error(log, "Caught exception on message read [sock=" + sock +
+ ", locNodeId=" + locNodeId + ", rmtNodeId=" + nodeId + ']', e);
+
+ onException("Caught exception on message read [sock=" + sock +
+ ", locNodeId=" + locNodeId + ", rmtNodeId=" + nodeId + ']', e);
+
+ if (isInterrupted() || sock.isClosed())
+ return;
+
+ if (e.hasCause(ClassNotFoundException.class))
+ LT.warn(log, "Failed to read message due to ClassNotFoundException " +
+ "(make sure same versions of all classes are available on all nodes) " +
+ "[rmtNodeId=" + nodeId +
+ ", err=" + X.cause(e, ClassNotFoundException.class).getMessage() + ']');
+
+ // Always report marshalling errors.
+ boolean err = e.hasCause(ObjectStreamException.class) ||
+ (nodeAlive(nodeId) && spiStateCopy() == CONNECTED && !X.hasCause(e, IOException.class));
+
+ if (err)
+ LT.error(log, e, "Failed to read message [sock=" + sock + ", locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId + ']');
+
+ return;
+ }
+ catch (IOException e) {
+ if (log.isDebugEnabled())
+ U.error(log, "Caught exception on message read [sock=" + sock + ", locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId + ']', e);
+
+ if (isInterrupted() || sock.isClosed())
+ return;
+
+ // Always report marshalling errors (although it is strange here).
+ boolean err = X.hasCause(e, ObjectStreamException.class) ||
+ (nodeAlive(nodeId) && spiStateCopy() == CONNECTED);
+
+ if (err)
+ LT.error(log, e, "Failed to send receipt on message [sock=" + sock +
+ ", locNodeId=" + locNodeId + ", rmtNodeId=" + nodeId + ']');
+
+ onException("Caught exception on message read [sock=" + sock + ", locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId + ']', e);
+
+ return;
+ }
+ }
+ }
+ finally {
+ if (!nioClient) {
+ if (clientMsgWrk != null) {
+ if (log.isDebugEnabled())
+ log.debug("Client connection failed [sock=" + sock + ", locNodeId=" + locNodeId +
+ ", rmtNodeId=" + nodeId + ']');
+
+ clientMsgWorkers.remove(nodeId, clientMsgWrk);
+
+ U.interrupt((ClientMessageWorker)clientMsgWrk);
+ }
+
+ U.closeQuiet(sock);
+ }
+ }
+ }
+
+ /**
+ * Processes client heartbeat message.
+ *
+ * @param msg Heartbeat message.
+ */
+ private void processClientHeartbeatMessage(TcpDiscoveryClientHeartbeatMessage msg) {
+ assert msg.client();
+
+ ClientMessageProcessor wrk = clientMsgWorkers.get(msg.creatorNodeId());
+
+ if (wrk != null)
+ wrk.metrics(msg.metrics());
+ else if (log.isDebugEnabled())
+ log.debug("Received heartbeat message from unknown client node: " + msg);
+ }
+
+ /**
+ * @param msg Join request message.
+ * @param clientMsgWrk Client message worker to start.
+ * @return Whether connection was successful.
+ * @throws IOException If IO failed.
+ */
+ @SuppressWarnings({"IfMayBeConditional"})
+ private boolean processJoinRequestMessage(TcpDiscoveryJoinRequestMessage msg,
+ @Nullable ClientMessageWorker clientMsgWrk) throws IOException, IgniteCheckedException {
+ assert msg != null;
+ assert !msg.responded();
+
+ TcpDiscoverySpiState state = spiStateCopy();
+
+ long sockTimeout = spi.failureDetectionTimeoutEnabled() ? spi.failureDetectionTimeout() :
+ spi.getSocketTimeout();
+
+ if (state == CONNECTED) {
+ spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
+
+ if (log.isDebugEnabled())
+ log.debug("Responded to join request message [msg=" + msg + ", res=" + RES_OK + ']');
+
+ msg.responded(true);
+
+ if (clientMsgWrk != null && clientMsgWrk.getState() == State.NEW) {
+ clientMsgWrk.clientVersion(U.productVersion(msg.node()));
+
+ clientMsgWrk.start();
+ }
+
+ msgWorker.addMessage(msg);
+
+ return true;
+ }
+ else {
+ spi.stats.onMessageProcessingStarted(msg);
+
+ Integer res;
+
+ SocketAddress rmtAddr = sock.getRemoteSocketAddress();
+
+ if (state == CONNECTING) {
+ if (noResAddrs.contains(rmtAddr) ||
+ getLocalNodeId().compareTo(msg.creatorNodeId()) < 0)
+ // Remote node node has not responded to join request or loses UUID race.
+ res = RES_WAIT;
+ else
+ // Remote node responded to join request and wins UUID race.
+ res = RES_CONTINUE_JOIN;
+ }
+ else
+ // Local node is stopping. Remote node should try next one.
+ res = RES_CONTINUE_JOIN;
+
+ spi.writeToSocket(msg, sock, res, sockTimeout);
+
+ if (log.isDebugEnabled())
+ log.debug("Responded to join request message [msg=" + msg + ", res=" + res + ']');
+
+ fromAddrs.addAll(msg.node().socketAddresses());
+
+ spi.stats.onMessageProcessingFinished(msg);
+
+ return false;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void interrupt() {
+ super.interrupt();
+
+ U.closeQuiet(sock);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void cleanup() {
+ super.cleanup();
+
+ if (!nioClient)
+ U.closeQuiet(sock);
+
+ synchronized (mux) {
+ readers.remove(this);
+ }
+
+ spi.stats.onSocketReaderRemoved();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "Socket reader [id=" + getId() + ", name=" + getName() + ", nodeId=" + nodeId + ']';
+ }
+ }
+
+ /**
+ * SPI Statistics printer.
+ */
+ private class StatisticsPrinter extends IgniteSpiThread {
+ /**
+ * Constructor.
+ */
+ StatisticsPrinter() {
+ super(spi.ignite().name(), "tcp-disco-stats-printer", log);
+
+ assert spi.statsPrintFreq > 0;
+
+ assert log.isInfoEnabled();
+
+ setPriority(spi.threadPri);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings({"BusyWait"})
+ @Override protected void body() throws InterruptedException {
+ if (log.isDebugEnabled())
+ log.debug("Statistics printer has been started.");
+
+ while (!isInterrupted()) {
+ Thread.sleep(spi.statsPrintFreq);
+
+ printStatistics();
+ }
+ }
+ }
+
+ /**
+ * Provides communication with client node.
+ */
+ private interface ClientMessageProcessor {
+ /**
+ * @param msg Message.
+ */
+ void addMessage(TcpDiscoveryAbstractMessage msg);
+
+ /**
+ * @param msg Message.
+ * @param msgBytes Marshalled message.
+ */
+ void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[] msgBytes);
+
+ /**
+ * @param timeoutHelper Timeout helper.
+ * @return Success flag.
+ * @throws InterruptedException
+ */
+ boolean ping(IgniteSpiOperationTimeoutHelper timeoutHelper) throws InterruptedException;
+
+ /**
+ * @param res Ping result.
+ */
+ void pingResult(boolean res);
+
+ /**
+ * @return Client ID.
+ */
+ UUID clientNodeId();
+
+ /**
+ * @return Cluster metrics.
+ */
+ ClusterMetrics metrics();
+
+ /**
+ * @param metrics Cluster metrics.
+ */
+ void metrics(ClusterMetrics metrics);
+ }
+
+ /**
+ *
+ */
+ private class ClientMessageWorker extends MessageWorkerAdapter<T2<TcpDiscoveryAbstractMessage, byte[]>>
+ implements ClientMessageProcessor {
+ /** Node ID. */
+ private final UUID clientNodeId;
+
+ /** Socket. */
+ private final Socket sock;
+
+ /** Current client metrics. */
+ private volatile ClusterMetrics metrics;
+
+ /** */
+ private IgniteProductVersion clientVer;
+
+ /** */
+ private volatile ClientMessagePinger pinger;
+
+ /**
+ * @param sock Socket.
+ * @param clientNodeId Node ID.
+ */
+ protected ClientMessageWorker(Socket sock, UUID clientNodeId) throws IOException {
+ super("tcp-disco-client-message-worker", 2000);
+
+ this.sock = sock;
+ this.clientNodeId = clientNodeId;
+ }
+
+ /**
+ * @param clientVer Client version.
+ */
+ void clientVersion(IgniteProductVersion clientVer) {
+ this.clientVer = clientVer;
+ }
+
+ /**
+ * @return Current client metrics.
+ */
+ @Override public ClusterMetrics metrics() {
+ return metrics;
+ }
+
+ /**
+ * @param metrics New current client metrics.
+ */
+ @Override public void metrics(ClusterMetrics metrics) {
+ this.metrics = metrics;
+ }
+
+ /**
+ * @param msg Message.
+ */
+ @Override public void addMessage(TcpDiscoveryAbstractMessage msg) {
+ addMessage(msg, null);
+ }
+
+ /**
+ * @param msg Message.
+ * @param msgBytes Optional message bytes.
+ */
+ @Override public void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[] msgBytes) {
+ T2 t = new T2<>(msg, msgBytes);
+
+ if (msg.highPriority())
+ queue.addFirst(t);
+ else
+ queue.add(t);
+
+ DebugLogger log = messageLogger(msg);
+
+ if (log.isDebugEnabled())
+ log.debug("Message has been added to client queue: " + msg);
+ }
+
+ /** {@inheritDoc} */
+ @Override public UUID clientNodeId() {
+ return clientNodeId;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void processMessage(T2<TcpDiscoveryAbstractMessage, byte[]> msgT) {
+ boolean success = false;
+
+ TcpDiscoveryAbstractMessage msg = msgT.get1();
+
+ try {
+ assert msg.verified() : msg;
+
+ byte[] msgBytes = msgT.get2();
+
+ if (msgBytes == null)
+ msgBytes = U.marshal(spi.marshaller(), msg);
+
+ DebugLogger msgLog = messageLogger(msg);
+
+ if (msg instanceof TcpDiscoveryClientAckResponse) {
+ if (clientVer == null) {
+ ClusterNode node = spi.getNode(clientNodeId);
+
+ if (node != null)
+ clientVer = IgniteUtils.productVersion(node);
+ else if (msgLog.isDebugEnabled())
+ msgLog.debug("Skip sending message ack to client, fail to get client node " +
+ "[sock=" + sock + ", locNodeId=" + getLocalNodeId() +
+ ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+ }
+
- if (clientVer != null &&
- clientVer.compareTo(TcpDiscoveryClientAckResponse.CLIENT_ACK_SINCE_VERSION) >= 0) {
++ if (clientVer != null) {
+ if (msgLog.isDebugEnabled())
+ msgLog.debug("Sending message ack to client [sock=" + sock + ", locNodeId="
+ + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+
+ spi.writeToSocket(sock, msg, msgBytes, spi.failureDetectionTimeoutEnabled() ?
+ spi.failureDetectionTimeout() : spi.getSocketTimeout());
+ }
+ }
+ else {
+ if (msgLog.isDebugEnabled())
+ msgLog.debug("Redirecting message to client [sock=" + sock + ", locNodeId="
+ + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+
+ assert topologyInitialized(msg) : msg;
+
+ spi.writeToSocket(sock, msg, msgBytes, spi.failureDetectionTimeoutEnabled() ?
+ spi.failureDetectionTimeout() : spi.getSocketTimeout());
+ }
+
- success = true;
++ boolean clientFailed = msg instanceof TcpDiscoveryNodeFailedMessage &&
++ ((TcpDiscoveryNodeFailedMessage)msg).failedNodeId().equals(clientNodeId);
++
++ assert !clientFailed || msg.force() : msg;
++
++ success = !clientFailed;
+ }
+ catch (IgniteCheckedException | IOException e) {
+ if (log.isDebugEnabled())
+ U.error(log, "Client connection failed [sock=" + sock + ", locNodeId="
+ + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']', e);
+
+ onException("Client connection failed [sock=" + sock + ", locNodeId="
+ + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']', e);
+ }
+ finally {
+ if (!success) {
+ clientMsgWorkers.remove(clientNodeId, this);
+
+ U.interrupt(this);
+
+ U.closeQuiet(sock);
+ }
+ }
+ }
+
+ /**
+ * @param msg Message.
+ * @return {@code True} if topology initialized.
+ */
+ private boolean topologyInitialized(TcpDiscoveryAbstractMessage msg) {
+ if (msg instanceof TcpDiscoveryNodeAddedMessage) {
+ TcpDiscoveryNodeAddedMessage addedMsg = (TcpDiscoveryNodeAddedMessage)msg;
+
+ if (clientNodeId.equals(addedMsg.node().id()))
+ return addedMsg.topology() != null;
+ }
+
+ return true;
+ }
+
+ /**
+ * @param res Ping result.
+ */
+ @Override public void pingResult(boolean res) {
+ pinger().pingResult(res);
+ }
+
+ /**
+ * @param timeoutHelper Timeout controller.
+ * @return Ping result.
+ * @throws InterruptedException If interrupted.
+ */
+ @Override public boolean ping(IgniteSpiOperationTimeoutHelper timeoutHelper) throws InterruptedException {
+ return pinger().ping(timeoutHelper);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void cleanup() {
+ super.cleanup();
+
+ pingResult(false);
+
+ U.closeQuiet(sock);
+ }
+
+ /**
+ * @return Pinger.
+ */
+ private ClientMessagePinger pinger() {
+ if (pinger == null) {
+ synchronized (this) {
+ if (pinger == null)
+ pinger = new ClientMessagePinger(this);
+ }
+ }
+
+ return pinger;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(ClientMessageWorker.class, this);
+ }
+ }
+
+ /**
+ * Base class for message workers.
+ */
+ protected abstract class MessageWorkerAdapter<T> extends IgniteSpiThread {
+ /** Message queue. */
+ protected final BlockingDeque<T> queue = new LinkedBlockingDeque<>();
+
+ /** Backed interrupted flag. */
+ private volatile boolean interrupted;
+
+ /** Polling timeout. */
+ private final long pollingTimeout;
+
+ /**
+ * @param name Thread name.
+ * @param pollingTimeout Messages polling timeout.
+ */
+ MessageWorkerAdapter(String name, long pollingTimeout) {
+ super(spi.ignite().name(), name, log);
+
+ this.pollingTimeout = pollingTimeout;
+
+ setPriority(spi.threadPri);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void body() throws InterruptedException {
+ if (log.isDebugEnabled())
+ log.debug("Message worker started [locNodeId=" + getConfiguredNodeId() + ']');
+
+ while (!isInterrupted()) {
+ T msg = queue.poll(pollingTimeout, TimeUnit.MILLISECONDS);
+
+ if (msg == null)
+ noMessageLoop();
+ else
+ processMessage(msg);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void interrupt() {
+ interrupted = true;
+
+ super.interrupt();
+ }
- // Always report marshalling errors (although it is strange here).
- boolean err = X.hasCause(e, ObjectStreamException.class) ||
- (nodeAlive(nodeId) && spiStateCopy() == CONNECTED);
+ /** {@inheritDoc} */
+ @Override public boolean isInterrupted() {
+ return interrupted || super.isInterrupted();
+ }
- if (err)
- LT.error(log, e, "Failed to send receipt on message [sock=" + sock +
- ", locNodeId=" + locNodeId + ", rmtNodeId=" + nodeId + ']');
+ /**
+ * @return Current queue size.
+ */
+ int queueSize() {
+ return queue.size();
+ }
- onException("Caught exception on message read [sock=" + sock + ", locNodeId=" + locNodeId +
- ", rmtNodeId=" + nodeId + ']', e);
+ /**
+ * @param msg Message.
+ */
+ protected abstract void processMessage(T msg);
- return;
- }
- }
- }
- finally {
- if (clientMsgWrk != null) {
- if (log.isDebugEnabled())
- log.debug("Client connection failed [sock=" + sock + ", locNodeId=" + locNodeId +
- ", rmtNodeId=" + nodeId + ']');
+ /**
+ * Called when there is no message to process giving ability to perform other activity.
+ */
+ protected void noMessageLoop() {
+ // No-op.
+ }
+ }
- clientMsgWorkers.remove(nodeId, clientMsgWrk);
+ /**
+ *
+ */
+ private static class GridPingFutureAdapter<R> extends GridFutureAdapter<R> {
+ /** */
+ private static final long serialVersionUID = 0L;
- U.interrupt(clientMsgWrk);
- }
+ /** Socket. */
+ private volatile Socket sock;
- U.closeQuiet(sock);
- }
+ /**
+ * Returns socket associated with this ping future.
+ *
+ * @return Socket or {@code null} if no socket associated.
+ */
+ public Socket sock() {
+ return sock;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index aa0c8f4,25804c7..d05953a
--- 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
@@@ -52,10 -51,7 +52,9 @@@ import org.apache.ignite.IgniteLogger
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.AddressResolver;
import org.apache.ignite.configuration.IgniteConfiguration;
- import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.nio.GridNioFuture;
+import org.apache.ignite.internal.util.nio.GridNioSession;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
index 9a989c4,e982b2f..1cb6a0d
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
@@@ -49,8 -49,8 +49,11 @@@ public abstract class TcpDiscoveryAbstr
protected static final int CLIENT_ACK_FLAG_POS = 4;
/** */
+ protected static final int ASYNC_MODE_POS = 5;
+
++ /** */
+ protected static final int FORCE_FAIL_FLAG_POS = 8;
+
/** Sender of the message (transient). */
private transient UUID sndNodeId;
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCache150ClientsTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteCacheSslStartStopSelfTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 78752b9,524fdcf..bddd369
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@@ -154,12 -150,9 +156,12 @@@ public class TcpClientDiscoverySpiSelfT
/** */
private boolean reconnectDisabled;
+ /** */
+ private static ExecutorService pool;
+
/** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(gridName);
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
TcpDiscoverySpi disco = getDiscoverySpi();
http://git-wip-us.apache.org/repos/asf/ignite/blob/87205f97/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
----------------------------------------------------------------------