/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.spi.communication.tcp;

import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.Channel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.StringJoiner;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteClientDisconnectedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.AddressResolver;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteFeatures;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.IgniteTooManyOpenFilesException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.managers.eventstorage.HighPriorityListener;
import org.apache.ignite.internal.processors.metric.impl.MetricUtils;
import org.apache.ignite.internal.processors.tracing.MTC;
import org.apache.ignite.internal.processors.tracing.NoopTracing;
import org.apache.ignite.internal.processors.tracing.SpanTags;
import org.apache.ignite.internal.processors.tracing.Tracing;
import org.apache.ignite.internal.util.GridConcurrentFactory;
import org.apache.ignite.internal.util.GridSpinReadWriteLock;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.internal.util.ipc.IpcEndpoint;
import org.apache.ignite.internal.util.ipc.IpcToNioAdapter;
import org.apache.ignite.internal.util.ipc.shmem.IpcOutOfSystemResourcesException;
import org.apache.ignite.internal.util.ipc.shmem.IpcSharedMemoryServerEndpoint;
import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
import org.apache.ignite.internal.util.nio.GridCommunicationClient;
import org.apache.ignite.internal.util.nio.GridConnectionBytesVerifyFilter;
import org.apache.ignite.internal.util.nio.GridDirectParser;
import org.apache.ignite.internal.util.nio.GridNioCodecFilter;
import org.apache.ignite.internal.util.nio.GridNioFilter;
import org.apache.ignite.internal.util.nio.GridNioMessageReaderFactory;
import org.apache.ignite.internal.util.nio.GridNioMessageTracker;
import org.apache.ignite.internal.util.nio.GridNioMessageWriterFactory;
import org.apache.ignite.internal.util.nio.GridNioRecoveryDescriptor;
import org.apache.ignite.internal.util.nio.GridNioServer;
import org.apache.ignite.internal.util.nio.GridNioServerListener;
import org.apache.ignite.internal.util.nio.GridNioServerListenerAdapter;
import org.apache.ignite.internal.util.nio.GridNioSession;
import org.apache.ignite.internal.util.nio.GridNioSessionMetaKey;
import org.apache.ignite.internal.util.nio.GridSelectorNioSessionImpl;
import org.apache.ignite.internal.util.nio.GridNioTracerFilter;
import org.apache.ignite.internal.util.nio.GridShmemCommunicationClient;
import org.apache.ignite.internal.util.nio.GridTcpNioCommunicationClient;
import org.apache.ignite.internal.util.nio.ssl.BlockingSslHandler;
import org.apache.ignite.internal.util.nio.ssl.GridNioSslFilter;
import org.apache.ignite.internal.util.nio.ssl.GridSslMeta;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CI2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageFormatter;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.resources.LoggerResource;
import org.apache.ignite.spi.ExponentialBackoffTimeoutStrategy;
import org.apache.ignite.spi.IgnitePortProtocol;
import org.apache.ignite.spi.IgniteSpiAdapter;
import org.apache.ignite.spi.IgniteSpiConfiguration;
import org.apache.ignite.spi.IgniteSpiConsistencyChecked;
import org.apache.ignite.spi.IgniteSpiContext;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.IgniteSpiMBeanAdapter;
import org.apache.ignite.spi.IgniteSpiMultipleInstancesSupport;
import org.apache.ignite.spi.IgniteSpiOperationTimeoutException;
import org.apache.ignite.spi.IgniteSpiOperationTimeoutHelper;
import org.apache.ignite.spi.IgniteSpiThread;
import org.apache.ignite.spi.IgniteSpiTimeoutObject;
import org.apache.ignite.spi.TimeoutStrategy;
import org.apache.ignite.spi.communication.CommunicationListener;
import org.apache.ignite.spi.communication.CommunicationSpi;
import org.apache.ignite.spi.communication.tcp.internal.CommunicationListenerEx;
import org.apache.ignite.spi.communication.tcp.internal.ConnectionKey;
import org.apache.ignite.spi.communication.tcp.internal.HandshakeException;
import org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture;
import org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationNodeConnectionCheckFuture;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage2;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.Nullable;

import static java.util.Collections.emptyList;
import static java.util.Objects.nonNull;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_TCP_COMM_SET_ATTR_HOST_NAMES;
import static org.apache.ignite.IgniteSystemProperties.getBoolean;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
import static org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesTable.traceName;
import static org.apache.ignite.internal.IgniteFeatures.CHANNEL_COMMUNICATION;
import static org.apache.ignite.internal.IgniteFeatures.nodeSupports;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.SSL_META;
import static org.apache.ignite.plugin.extensions.communication.Message.DIRECT_TYPE_SIZE;
import static org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture.SES_FUT_META;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.ALREADY_CONNECTED;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NEED_WAIT;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NODE_STOPPING;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.UNKNOWN_NODE;

/**
 * <tt>TcpCommunicationSpi</tt> is default communication SPI which uses
 * TCP/IP protocol and Java NIO to communicate with other nodes.
 * <p>
 * To enable communication with other nodes, this SPI adds {@link #ATTR_ADDRS}
 * and {@link #ATTR_PORT} local node attributes (see {@link ClusterNode#attributes()}.
 * <p>
 * At startup, this SPI tries to start listening to local port specified by
 * {@link #setLocalPort(int)} method. If local port is occupied, then SPI will
 * automatically increment the port number until it can successfully bind for
 * listening. {@link #setLocalPortRange(int)} configuration parameter controls
 * maximum number of ports that SPI will try before it fails. Port range comes
 * very handy when starting multiple grid nodes on the same machine or even
 * in the same VM. In this case all nodes can be brought up without a single
 * change in configuration.
 * <p>
 * This SPI caches connections to remote nodes so it does not have to reconnect every
 * time a message is sent. By default, idle connections are kept active for
 * {@link #DFLT_IDLE_CONN_TIMEOUT} period and then are closed. Use
 * {@link #setIdleConnectionTimeout(long)} configuration parameter to configure
 * you own idle connection timeout.
 * <h1 class="header">Failure Detection</h1>
 * Configuration defaults (see Configuration section below and
 * {@link IgniteConfiguration#getFailureDetectionTimeout()}) for details) are chosen to make possible for
 * communication SPI work reliably on most of hardware and virtual deployments, but this has made failure detection
 * time worse.
 * <p>
 * If it's needed to tune failure detection then it's highly recommended to do this using
 * {@link IgniteConfiguration#setFailureDetectionTimeout(long)}. This failure timeout automatically controls the
 * following parameters: {@link #getConnectTimeout()}, {@link #getMaxConnectTimeout()},
 * {@link #getReconnectCount()}. If any of those parameters is set explicitly, then the failure timeout setting will be
 * ignored.
 * <p>
 * If it's required to perform advanced settings of failure detection and
 * {@link IgniteConfiguration#getFailureDetectionTimeout()} is unsuitable then various {@code TcpCommunicationSpi}
 * configuration parameters may be used.
 * <h1 class="header">Configuration</h1>
 * <h2 class="header">Mandatory</h2>
 * This SPI has no mandatory configuration parameters.
 * <h2 class="header">Optional</h2>
 * The following configuration parameters are optional:
 * <ul>
 * <li>Address resolver (see {@link #setAddressResolver(AddressResolver)}</li>
 * <li>Node local IP address (see {@link #setLocalAddress(String)})</li>
 * <li>Node local port number (see {@link #setLocalPort(int)})</li>
 * <li>Local port range (see {@link #setLocalPortRange(int)}</li>
 * <li>Use paired connections (see {@link #setUsePairedConnections(boolean)}</li>
 * <li>Connections per node (see {@link #setConnectionsPerNode(int)})</li>
 * <li>Shared memory port (see {@link #setSharedMemoryPort(int)}</li>
 * <li>Idle connection timeout (see {@link #setIdleConnectionTimeout(long)})</li>
 * <li>Direct or heap buffer allocation (see {@link #setDirectBuffer(boolean)})</li>
 * <li>Direct or heap buffer allocation for sending (see {@link #setDirectSendBuffer(boolean)})</li>
 * <li>Count of selectors and selector threads for NIO server (see {@link #setSelectorsCount(int)})</li>
 * <li>Selector thread busy-loop iterations (see {@link #setSelectorSpins(long)}</li>
 * <li>{@code TCP_NODELAY} socket option for sockets (see {@link #setTcpNoDelay(boolean)})</li>
 * <li>Filter reachable addresses (see {@link #setFilterReachableAddresses(boolean)} </li>
 * <li>Message queue limit (see {@link #setMessageQueueLimit(int)})</li>
 * <li>Slow client queue limit (see {@link #setSlowClientQueueLimit(int)})</li>
 * <li>Connect timeout (see {@link #setConnectTimeout(long)})</li>
 * <li>Maximum connect timeout (see {@link #setMaxConnectTimeout(long)})</li>
 * <li>Reconnect attempts count (see {@link #setReconnectCount(int)})</li>
 * <li>Socket receive buffer size (see {@link #setSocketReceiveBuffer(int)})</li>
 * <li>Socket send buffer size (see {@link #setSocketSendBuffer(int)})</li>
 * <li>Socket write timeout (see {@link #setSocketWriteTimeout(long)})</li>
 * <li>Number of received messages after which acknowledgment is sent (see {@link #setAckSendThreshold(int)})</li>
 * <li>Maximum number of unacknowledged messages (see {@link #setUnacknowledgedMessagesBufferSize(int)})</li>
 * </ul>
 * <h2 class="header">Java Example</h2>
 * TcpCommunicationSpi is used by default and should be explicitly configured
 * only if some SPI configuration parameters need to be overridden.
 * <pre name="code" class="java">
 * TcpCommunicationSpi commSpi = new TcpCommunicationSpi();
 *
 * // Override local port.
 * commSpi.setLocalPort(4321);
 *
 * IgniteConfiguration cfg = new IgniteConfiguration();
 *
 * // Override default communication SPI.
 * cfg.setCommunicationSpi(commSpi);
 *
 * // Start grid.
 * Ignition.start(cfg);
 * </pre>
 * <h2 class="header">Spring Example</h2>
 * TcpCommunicationSpi can be configured from Spring XML configuration file:
 * <pre name="code" class="xml">
 * &lt;bean id="grid.custom.cfg" class="org.apache.ignite.configuration.IgniteConfiguration" singleton="true"&gt;
 *         ...
 *         &lt;property name="communicationSpi"&gt;
 *             &lt;bean class="org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi"&gt;
 *                 &lt;!-- Override local port. --&gt;
 *                 &lt;property name="localPort" value="4321"/&gt;
 *             &lt;/bean&gt;
 *         &lt;/property&gt;
 *         ...
 * &lt;/bean&gt;
 * </pre>
 * <p>
 * <img src="http://ignite.apache.org/images/spring-small.png">
 * <br>
 * For information about Spring framework visit <a href="http://www.springframework.org/">www.springframework.org</a>
 * @see CommunicationSpi
 */
@IgniteSpiMultipleInstancesSupport(true)
@IgniteSpiConsistencyChecked(optional = false)
public class TcpCommunicationSpi extends IgniteSpiAdapter implements CommunicationSpi<Message> {
    /** Time threshold to log too long connection establish. */
    private static final int CONNECTION_ESTABLISH_THRESHOLD_MS = 100;

    /** IPC error message. */
    public static final String OUT_OF_RESOURCES_TCP_MSG = "Failed to allocate shared memory segment " +
        "(switching to TCP, may be slower).";

    /** Node attribute that is mapped to node IP addresses (value is <tt>comm.tcp.addrs</tt>). */
    public static final String ATTR_ADDRS = "comm.tcp.addrs";

    /** Node attribute that is mapped to node host names (value is <tt>comm.tcp.host.names</tt>). */
    public static final String ATTR_HOST_NAMES = "comm.tcp.host.names";

    /** Node attribute that is mapped to node port number (value is <tt>comm.tcp.port</tt>). */
    public static final String ATTR_PORT = "comm.tcp.port";

    /** Node attribute that is mapped to node port number (value is <tt>comm.shmem.tcp.port</tt>). */
    public static final String ATTR_SHMEM_PORT = "comm.shmem.tcp.port";

    /** Node attribute that is mapped to node's external addresses (value is <tt>comm.tcp.ext-addrs</tt>). */
    public static final String ATTR_EXT_ADDRS = "comm.tcp.ext-addrs";

    /** */
    public static final String ATTR_PAIRED_CONN = "comm.tcp.pairedConnection";

    /** Default port which node sets listener to (value is <tt>47100</tt>). */
    public static final int DFLT_PORT = 47100;

    /** Default port which node sets listener for shared memory connections (value is <tt>48100</tt>). */
    public static final int DFLT_SHMEM_PORT = -1;

    /** Default idle connection timeout (value is <tt>10</tt>min). */
    public static final long DFLT_IDLE_CONN_TIMEOUT = 10 * 60_000;

    /** Default socket send and receive buffer size. */
    public static final int DFLT_SOCK_BUF_SIZE = 32 * 1024;

    /** Default connection timeout (value is <tt>5000</tt>ms). */
    public static final long DFLT_CONN_TIMEOUT = 5000;

    /** Default Maximum connection timeout (value is <tt>600,000</tt>ms). */
    public static final long DFLT_MAX_CONN_TIMEOUT = 10 * 60 * 1000;

    /** Default reconnect attempts count (value is <tt>10</tt>). */
    public static final int DFLT_RECONNECT_CNT = 10;

    /** Default message queue limit per connection (for incoming and outgoing . */
    public static final int DFLT_MSG_QUEUE_LIMIT = GridNioServer.DFLT_SEND_QUEUE_LIMIT;

    /**
     * Default count of selectors for TCP server equals to
     * {@code "Math.max(4, Runtime.getRuntime().availableProcessors() / 2)"}.
     */
    public static final int DFLT_SELECTORS_CNT = Math.max(4, Runtime.getRuntime().availableProcessors() / 2);

    /** Default initial connect/handshake timeout in case of failure detection enabled. */
    private static final int DFLT_INITIAL_TIMEOUT = 500;

    /** Default initial delay in case of target node is still out of topology. */
    private static final int DFLT_NEED_WAIT_DELAY = 200;

    /** Default delay between reconnects attempts in case of temporary network issues. */
    private static final int DFLT_RECONNECT_DELAY = 50;

    /**
     * Version when client is ready to wait to connect to server (could be needed when client tries to open connection
     * before it starts being visible for server)
     */
    private static final IgniteProductVersion VERSION_SINCE_CLIENT_COULD_WAIT_TO_CONNECT = IgniteProductVersion.fromString("2.1.4");

    /** Connection index meta for session. */
    public static final int CONN_IDX_META = GridNioSessionMetaKey.nextUniqueKey();

    /** Node consistent id meta for session. */
    public static final int CONSISTENT_ID_META = GridNioSessionMetaKey.nextUniqueKey();

    /** Message tracker meta for session. */
    private static final int TRACKER_META = GridNioSessionMetaKey.nextUniqueKey();

    /** Channel meta used for establishing channel connections. */
    private static final int CHANNEL_FUT_META = GridNioSessionMetaKey.nextUniqueKey();

    /**
     * Default local port range (value is <tt>100</tt>).
     * See {@link #setLocalPortRange(int)} for details.
     */
    public static final int DFLT_PORT_RANGE = 100;

    /** Default value for {@code TCP_NODELAY} socket option (value is <tt>true</tt>). */
    public static final boolean DFLT_TCP_NODELAY = true;

    /** Default value for {@code FILTER_REACHABLE_ADDRESSES} socket option (value is <tt>false</tt>). */
    public static final boolean DFLT_FILTER_REACHABLE_ADDRESSES = false;

    /** Default received messages threshold for sending ack. */
    public static final int DFLT_ACK_SND_THRESHOLD = 32;

    /** Default socket write timeout. */
    public static final long DFLT_SOCK_WRITE_TIMEOUT = 2000;

    /** Default connections per node. */
    public static final int DFLT_CONN_PER_NODE = 1;

    /** Maximum {@link GridNioSession} connections per node. */
    public static final int MAX_CONN_PER_NODE = 1024;

    /** No-op runnable. */
    private static final IgniteRunnable NOOP = () -> {};

    /** Node ID message type. */
    public static final short NODE_ID_MSG_TYPE = -1;

    /** Recovery last received ID message type. */
    public static final short RECOVERY_LAST_ID_MSG_TYPE = -2;

    /** Handshake message type. */
    public static final short HANDSHAKE_MSG_TYPE = -3;

    /** Handshake wait message type. */
    public static final short HANDSHAKE_WAIT_MSG_TYPE = -28;

    /** Communication metrics group name. */
    public static final String COMMUNICATION_METRICS_GROUP_NAME = MetricUtils.metricName("communication", "tcp");

    /** */
    public static final String SENT_MESSAGES_METRIC_NAME = "sentMessagesCount";

    /** */
    public static final String SENT_MESSAGES_METRIC_DESC = "Total number of messages sent by current node";

    /** */
    public static final String RECEIVED_MESSAGES_METRIC_NAME = "receivedMessagesCount";

    /** */
    public static final String RECEIVED_MESSAGES_METRIC_DESC = "Total number of messages received by current node";

    /** */
    public static final String SENT_MESSAGES_BY_TYPE_METRIC_NAME = "sentMessagesByType";

    /** */
    public static final String SENT_MESSAGES_BY_TYPE_METRIC_DESC =
        "Total number of messages with given type sent by current node";

    /** */
    public static final String RECEIVED_MESSAGES_BY_TYPE_METRIC_NAME = "receivedMessagesByType";

    /** */
    public static final String RECEIVED_MESSAGES_BY_TYPE_METRIC_DESC =
        "Total number of messages with given type received by current node";

    /** */
    public static final String SENT_MESSAGES_BY_NODE_CONSISTENT_ID_METRIC_NAME = "sentMessagesToNode";

    /** */
    public static final String SENT_MESSAGES_BY_NODE_CONSISTENT_ID_METRIC_DESC =
        "Total number of messages sent by current node to the given node";

    /** */
    public static final String RECEIVED_MESSAGES_BY_NODE_CONSISTENT_ID_METRIC_NAME = "receivedMessagesFromNode";

    /** */
    public static final String RECEIVED_MESSAGES_BY_NODE_CONSISTENT_ID_METRIC_DESC =
        "Total number of messages received by current node from the given node";

    /** */
    private ConnectGateway connectGate;

    /** */
    private ConnectionPolicy connPlc = new FirstConnectionPolicy();

    /** Channel connection index provider. */
    private ConnectionPolicy chConnPlc;

    /** */
    private boolean enableForcibleNodeKill = getBoolean(IgniteSystemProperties.IGNITE_ENABLE_FORCIBLE_NODE_KILL);

    /** */
    private boolean enableTroubleshootingLog = getBoolean(IgniteSystemProperties.IGNITE_TROUBLESHOOTING_LOGGER);

    /** Server listener. */
    private final GridNioServerListener<Message> srvLsnr =
        new GridNioServerListenerAdapter<Message>() {
            @Override public void onSessionWriteTimeout(GridNioSession ses) {
                LT.warn(log, "Communication SPI session write timed out (consider increasing " +
                    "'socketWriteTimeout' " + "configuration property) [remoteAddr=" + ses.remoteAddress() +
                    ", writeTimeout=" + sockWriteTimeout + ']');

                if (log.isDebugEnabled())
                    log.debug("Closing communication SPI session on write timeout [remoteAddr=" + ses.remoteAddress() +
                        ", writeTimeout=" + sockWriteTimeout + ']');

                ses.close();
            }

            @Override public void onConnected(GridNioSession ses) {
                if (ses.accepted()) {
                    if (log.isInfoEnabled())
                        log.info("Accepted incoming communication connection [locAddr=" + ses.localAddress() +
                            ", rmtAddr=" + ses.remoteAddress() + ']');

                    try {
                        boolean client = Boolean.TRUE.equals(ignite().configuration().isClientMode());

                        if (client || ctxInitLatch.getCount() == 0 || !isHandshakeWaitSupported()) {
                            if (log.isDebugEnabled())
                                log.debug("Sending local node ID to newly accepted session: " + ses);

                            ses.sendNoFuture(nodeIdMessage(), null);
                        }
                        else {
                            if (log.isDebugEnabled())
                                log.debug("Sending handshake wait message to newly accepted session: " + ses);

                            ses.sendNoFuture(new HandshakeWaitMessage(), null);
                        }
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Failed to send message: " + e, e);
                    }
                }
                else {
                    if (log.isInfoEnabled())
                        log.info("Established outgoing communication connection [locAddr=" + ses.localAddress() +
                            ", rmtAddr=" + ses.remoteAddress() + ']');
                }
            }

            @Override public void onDisconnected(GridNioSession ses, @Nullable Exception e) {
                ConnectionKey connId = ses.meta(CONN_IDX_META);

                if (connId != null) {
                    if (connId.dummy())
                        return;

                    UUID id = connId.nodeId();

                    GridCommunicationClient[] nodeClients = clients.get(id);

                    if (nodeClients != null) {
                        for (GridCommunicationClient client : nodeClients) {
                            if (client instanceof GridTcpNioCommunicationClient &&
                                ((GridTcpNioCommunicationClient)client).session() == ses) {
                                client.close();

                                removeNodeClient(id, client);
                            }
                        }
                    }

                    if (!stopping) {
                        GridNioRecoveryDescriptor outDesc = ses.outRecoveryDescriptor();

                        if (outDesc != null) {
                            if (outDesc.nodeAlive(getSpiContext().node(id))) {
                                if (!outDesc.messagesRequests().isEmpty()) {
                                    if (log.isDebugEnabled())
                                        log.debug("Session was closed but there are unacknowledged messages, " +
                                            "will try to reconnect [rmtNode=" + outDesc.node().id() + ']');

                                    DisconnectedSessionInfo disconnectData =
                                        new DisconnectedSessionInfo(outDesc, connId.connectionIndex());

                                    commWorker.addProcessDisconnectRequest(disconnectData);
                                }
                            }
                            else
                                outDesc.onNodeLeft();
                        }
                    }

                    CommunicationListener<Message> lsnr0 = lsnr;

                    if (lsnr0 != null)
                        lsnr0.onDisconnected(id);
                }
            }

            /**
             * @param ses Session.
             * @param msg Message.
             */
            private void onFirstMessage(final GridNioSession ses, Message msg) {
                UUID sndId;

                ConnectionKey connKey;

                if (msg instanceof NodeIdMessage) {
                    sndId = U.bytesToUuid(((NodeIdMessage)msg).nodeIdBytes(), 0);
                    connKey = new ConnectionKey(sndId, 0, -1);
                }
                else {
                    assert msg instanceof HandshakeMessage : msg;

                    HandshakeMessage msg0 = (HandshakeMessage)msg;

                    sndId = ((HandshakeMessage)msg).nodeId();
                    connKey = new ConnectionKey(sndId, msg0.connectionIndex(), msg0.connectCount());
                }

                if (log.isDebugEnabled())
                    log.debug("Remote node ID received: " + sndId);

                final ClusterNode rmtNode = getSpiContext().node(sndId);

                if (rmtNode == null) {
                    DiscoverySpi discoverySpi = ignite().configuration().getDiscoverySpi();

                    boolean unknownNode = true;

                    if (discoverySpi instanceof TcpDiscoverySpi) {
                        TcpDiscoverySpi tcpDiscoverySpi = (TcpDiscoverySpi)discoverySpi;

                        ClusterNode node0 = tcpDiscoverySpi.getNode0(sndId);

                        if (node0 != null) {
                            assert node0.isClient() : node0;

                            if (node0.version().compareTo(VERSION_SINCE_CLIENT_COULD_WAIT_TO_CONNECT) >= 0)
                                unknownNode = false;
                        }
                    }
                    else if (discoverySpi instanceof IgniteDiscoverySpi)
                        unknownNode = !((IgniteDiscoverySpi)discoverySpi).knownNode(sndId);

                    if (unknownNode) {
                        U.warn(log, "Close incoming connection, unknown node [nodeId=" + sndId + ", ses=" + ses + ']');

                        ses.send(new RecoveryLastReceivedMessage(UNKNOWN_NODE)).listen(new CI1<IgniteInternalFuture<?>>() {
                            @Override public void apply(IgniteInternalFuture<?> fut) {
                                ses.close();
                            }
                        });
                    }
                    else {
                        ses.send(new RecoveryLastReceivedMessage(NEED_WAIT)).listen(new CI1<IgniteInternalFuture<?>>() {
                            @Override public void apply(IgniteInternalFuture<?> fut) {
                                ses.close();
                            }
                        });
                    }

                    return;
                }

                ses.addMeta(CONSISTENT_ID_META, rmtNode.consistentId());

                final ConnectionKey old = ses.addMeta(CONN_IDX_META, connKey);

                assert old == null;

                ClusterNode locNode = getSpiContext().localNode();

                if (ses.remoteAddress() == null)
                    return;

                assert msg instanceof HandshakeMessage : msg;

                HandshakeMessage msg0 = (HandshakeMessage)msg;

                if (log.isDebugEnabled())
                    log.debug("Received handshake message [locNodeId=" + locNode.id() + ", rmtNodeId=" + sndId +
                        ", msg=" + msg0 + ']');

                if (isChannelConnIdx(msg0.connectionIndex()))
                    ses.send(new RecoveryLastReceivedMessage(0));
                else if (usePairedConnections(rmtNode)) {
                    final GridNioRecoveryDescriptor recoveryDesc = inRecoveryDescriptor(rmtNode, connKey);

                    ConnectClosureNew c = new ConnectClosureNew(ses, recoveryDesc, rmtNode);

                    boolean reserve = recoveryDesc.tryReserve(msg0.connectCount(), c);

                    if (reserve)
                        connectedNew(recoveryDesc, ses, true);
                    else {
                        if (c.failed) {
                            ses.send(new RecoveryLastReceivedMessage(ALREADY_CONNECTED));

                            closeStaleConnections(connKey);
                        }
                    }
                }
                else {
                    assert connKey.connectionIndex() >= 0 : connKey;

                    GridCommunicationClient[] curClients = clients.get(sndId);

                    GridCommunicationClient oldClient =
                        curClients != null && connKey.connectionIndex() < curClients.length ?
                            curClients[connKey.connectionIndex()] :
                            null;

                    boolean hasShmemClient = false;

                    if (oldClient != null) {
                        if (oldClient instanceof GridTcpNioCommunicationClient) {
                            if (log.isInfoEnabled())
                                log.info("Received incoming connection when already connected " +
                                    "to this node, rejecting [locNode=" + locNode.id() +
                                    ", rmtNode=" + sndId + ']');

                            ses.send(new RecoveryLastReceivedMessage(ALREADY_CONNECTED));

                            closeStaleConnections(connKey);

                            return;
                        }
                        else {
                            assert oldClient instanceof GridShmemCommunicationClient;

                            hasShmemClient = true;
                        }
                    }

                    GridFutureAdapter<GridCommunicationClient> fut = new GridFutureAdapter<>();

                    GridFutureAdapter<GridCommunicationClient> oldFut = clientFuts.putIfAbsent(connKey, fut);

                    final GridNioRecoveryDescriptor recoveryDesc = inRecoveryDescriptor(rmtNode, connKey);

                    if (oldFut == null) {
                        curClients = clients.get(sndId);

                        oldClient = curClients != null && connKey.connectionIndex() < curClients.length ?
                            curClients[connKey.connectionIndex()] : null;

                        if (oldClient != null) {
                            if (oldClient instanceof GridTcpNioCommunicationClient) {
                                assert oldClient.connectionIndex() == connKey.connectionIndex() : oldClient;

                                if (log.isInfoEnabled())
                                    log.info("Received incoming connection when already connected " +
                                        "to this node, rejecting [locNode=" + locNode.id() +
                                        ", rmtNode=" + sndId + ']');

                                ses.send(new RecoveryLastReceivedMessage(ALREADY_CONNECTED));

                                closeStaleConnections(connKey);

                                fut.onDone(oldClient);

                                return;
                            }
                            else {
                                assert oldClient instanceof GridShmemCommunicationClient;

                                hasShmemClient = true;
                            }
                        }

                        boolean reserved = recoveryDesc.tryReserve(msg0.connectCount(),
                            new ConnectClosure(ses, recoveryDesc, rmtNode, connKey, msg0, !hasShmemClient, fut));

                        if (log.isDebugEnabled())
                            log.debug("Received incoming connection from remote node " +
                                "[rmtNode=" + rmtNode.id() + ", reserved=" + reserved +
                                ", recovery=" + recoveryDesc + ']');

                        if (reserved) {
                            try {
                                GridTcpNioCommunicationClient client =
                                    connected(recoveryDesc, ses, rmtNode, msg0.received(), true, !hasShmemClient);

                                fut.onDone(client);
                            }
                            finally {
                                clientFuts.remove(connKey, fut);
                            }
                        }
                    }
                    else {
                        if (oldFut instanceof ConnectFuture && locNode.order() < rmtNode.order()) {
                            if (log.isInfoEnabled()) {
                                log.info("Received incoming connection from remote node while " +
                                    "connecting to this node, rejecting [locNode=" + locNode.id() +
                                    ", locNodeOrder=" + locNode.order() + ", rmtNode=" + rmtNode.id() +
                                    ", rmtNodeOrder=" + rmtNode.order() + ']');
                            }

                            ses.send(new RecoveryLastReceivedMessage(ALREADY_CONNECTED));
                        }
                        else {
                            // The code below causes a race condition between shmem and TCP (see IGNITE-1294)
                            boolean reserved = recoveryDesc.tryReserve(msg0.connectCount(),
                                new ConnectClosure(ses, recoveryDesc, rmtNode, connKey, msg0, !hasShmemClient, fut));

                            if (reserved)
                                connected(recoveryDesc, ses, rmtNode, msg0.received(), true, !hasShmemClient);
                        }
                    }
                }
            }

            /**
             * @param connKey Connection key.
             */
            private void closeStaleConnections(ConnectionKey connKey) {
                for (GridNioSession ses0 : nioSrvr.sessions()) {
                    ConnectionKey key0 = ses0.meta(CONN_IDX_META);

                    if (ses0.accepted() && key0 != null &&
                        key0.nodeId().equals(connKey.nodeId()) &&
                        key0.connectionIndex() == connKey.connectionIndex() &&
                        key0.connectCount() < connKey.connectCount())
                        ses0.close();
                }
            }

            @Override public void onMessageSent(GridNioSession ses, Message msg) {
                Object consistentId = ses.meta(CONSISTENT_ID_META);

                if (consistentId != null)
                    metricsLsnr.onMessageSent(msg, consistentId);
            }

            private void onChannelCreate(
                GridSelectorNioSessionImpl ses,
                ConnectionKey connKey,
                Message msg
            ) {
                cleanupLocalNodeRecoveryDescriptor(connKey);

                ses.send(msg)
                    .listen(sendFut -> {
                        if (sendFut.error() != null) {
                            U.error(log, "Fail to send channel creation response to the remote node. " +
                                "Session will be closed [nodeId=" + connKey.nodeId() +
                                ", idx=" + connKey.connectionIndex() + ']', sendFut.error());

                            ses.close();

                            return;
                        }

                        ses.closeSocketOnSessionClose(false);

                        // Close session and send response.
                        ses.close().listen(closeFut -> {
                            if (closeFut.error() != null) {
                                U.error(log, "Nio session has not been properly closed " +
                                        "[nodeId=" + connKey.nodeId() + ", idx=" + connKey.connectionIndex() + ']',
                                    closeFut.error());

                                U.closeQuiet(ses.key().channel());

                                return;
                            }

                            notifyChannelEvtListener(connKey.nodeId(), ses.key().channel(), msg);
                        });
                    });
            }

            @Override public void onMessage(final GridNioSession ses, Message msg) {
                MTC.span().addLog(() -> "Communication received");
                MTC.span().addTag(SpanTags.MESSAGE, () -> traceName(msg));

                ConnectionKey connKey = ses.meta(CONN_IDX_META);

                if (connKey == null) {
                    assert ses.accepted() : ses;

                    if (!connectGate.tryEnter()) {
                        if (log.isDebugEnabled())
                            log.debug("Close incoming connection, failed to enter gateway.");

                        ses.send(new RecoveryLastReceivedMessage(NODE_STOPPING)).listen(new CI1<IgniteInternalFuture<?>>() {
                            @Override public void apply(IgniteInternalFuture<?> fut) {
                                ses.close();
                            }
                        });

                        return;
                    }

                    try {
                        onFirstMessage(ses, msg);
                    }
                    finally {
                        connectGate.leave();
                    }
                }
                else {
                    Object consistentId = ses.meta(CONSISTENT_ID_META);

                    assert consistentId != null;

                    if (isChannelConnIdx(connKey.connectionIndex())) {
                        if (ses.meta(CHANNEL_FUT_META) == null)
                            onChannelCreate((GridSelectorNioSessionImpl)ses, connKey, msg);
                        else {
                            GridFutureAdapter<Channel> fut = ses.meta(CHANNEL_FUT_META);
                            GridSelectorNioSessionImpl ses0 = (GridSelectorNioSessionImpl)ses;

                            ses0.closeSocketOnSessionClose(false);

                            ses0.close().listen(f -> {
                                if (f.error() != null) {
                                    fut.onDone(f.error());

                                    return;
                                }

                                fut.onDone(ses0.key().channel());
                            });
                        }

                        return;
                    }

                    if (msg instanceof RecoveryLastReceivedMessage) {
                        metricsLsnr.onMessageReceived(msg, consistentId);

                        GridNioRecoveryDescriptor recovery = ses.outRecoveryDescriptor();

                        if (recovery != null) {
                            RecoveryLastReceivedMessage msg0 = (RecoveryLastReceivedMessage)msg;

                            if (log.isDebugEnabled()) {
                                log.debug("Received recovery acknowledgement [rmtNode=" + connKey.nodeId() +
                                    ", connIdx=" + connKey.connectionIndex() +
                                    ", rcvCnt=" + msg0.received() + ']');
                            }

                            recovery.ackReceived(msg0.received());
                        }

                        return;
                    }
                    else {
                        GridNioRecoveryDescriptor recovery = ses.inRecoveryDescriptor();

                        if (recovery != null) {
                            long rcvCnt = recovery.onReceived();

                            if (rcvCnt % ackSndThreshold == 0) {
                                if (log.isDebugEnabled()) {
                                    log.debug("Send recovery acknowledgement [rmtNode=" + connKey.nodeId() +
                                        ", connIdx=" + connKey.connectionIndex() +
                                        ", rcvCnt=" + rcvCnt + ']');
                                }

                                ses.systemMessage(new RecoveryLastReceivedMessage(rcvCnt));

                                recovery.lastAcknowledged(rcvCnt);
                            }
                        }
                        else if (connKey.dummy()) {
                            assert msg instanceof NodeIdMessage : msg;

                            TcpCommunicationNodeConnectionCheckFuture fut = ses.meta(SES_FUT_META);

                            assert fut != null : msg;

                            fut.onConnected(U.bytesToUuid(((NodeIdMessage)msg).nodeIdBytes(), 0));

                            nioSrvr.closeFromWorkerThread(ses);

                            return;
                        }
                    }

                    metricsLsnr.onMessageReceived(msg, consistentId);

                    IgniteRunnable c;

                    if (msgQueueLimit > 0) {
                        GridNioMessageTracker tracker = ses.meta(TRACKER_META);

                        if (tracker == null) {
                            GridNioMessageTracker old = ses.addMeta(TRACKER_META, tracker =
                                new GridNioMessageTracker(ses, msgQueueLimit));

                            assert old == null;
                        }

                        tracker.onMessageReceived();

                        c = tracker;
                    }
                    else
                        c = NOOP;

                    notifyListener(connKey.nodeId(), msg, c);
                }
            }

            /** {@inheritDoc} */
            @Override public void onFailure(FailureType failureType, Throwable failure) {
                if (ignite instanceof IgniteEx)
                    ((IgniteEx)ignite).context().failure().process(new FailureContext(failureType, failure));
            }

            /**
             * @param recovery Recovery descriptor.
             * @param ses Session.
             * @param node Node.
             * @param rcvCnt Number of received messages.
             * @param sndRes If {@code true} sends response for recovery handshake.
             * @param createClient If {@code true} creates NIO communication client.
             * @return Client.
             */
            private GridTcpNioCommunicationClient connected(
                GridNioRecoveryDescriptor recovery,
                GridNioSession ses,
                ClusterNode node,
                long rcvCnt,
                boolean sndRes,
                boolean createClient) {
                ConnectionKey connKey = ses.meta(CONN_IDX_META);

                assert connKey != null && connKey.connectionIndex() >= 0 : connKey;
                assert !usePairedConnections(node);

                recovery.onHandshake(rcvCnt);

                ses.inRecoveryDescriptor(recovery);
                ses.outRecoveryDescriptor(recovery);

                nioSrvr.resend(ses);

                try {
                    if (sndRes)
                        nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(recovery.received()));
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to send message: " + e, e);
                }

                recovery.onConnected();

                GridTcpNioCommunicationClient client = null;

                if (createClient) {
                    client = new GridTcpNioCommunicationClient(connKey.connectionIndex(), ses, log);

                    addNodeClient(node, connKey.connectionIndex(), client);
                }

                return client;
            }

            /**
             * @param recovery Recovery descriptor.
             * @param ses Session.
             * @param sndRes If {@code true} sends response for recovery handshake.
             */
            private void connectedNew(
                GridNioRecoveryDescriptor recovery,
                GridNioSession ses,
                boolean sndRes) {
                try {
                    ses.inRecoveryDescriptor(recovery);

                    if (sndRes)
                        nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(recovery.received()));

                    recovery.onConnected();
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to send message: " + e, e);
                }
            }

            /**
             *
             */
            class ConnectClosureNew implements IgniteInClosure<Boolean> {
                /** */
                private static final long serialVersionUID = 0L;

                /** */
                private final GridNioSession ses;

                /** */
                private final GridNioRecoveryDescriptor recoveryDesc;

                /** */
                private final ClusterNode rmtNode;

                /** */
                private boolean failed;

                /**
                 * @param ses Incoming session.
                 * @param recoveryDesc Recovery descriptor.
                 * @param rmtNode Remote node.
                 */
                ConnectClosureNew(GridNioSession ses,
                    GridNioRecoveryDescriptor recoveryDesc,
                    ClusterNode rmtNode) {
                    this.ses = ses;
                    this.recoveryDesc = recoveryDesc;
                    this.rmtNode = rmtNode;
                }

                /** {@inheritDoc} */
                @Override public void apply(Boolean success) {
                    try {
                        failed = !success;

                        if (success) {
                            IgniteInClosure<IgniteInternalFuture<?>> lsnr = new IgniteInClosure<IgniteInternalFuture<?>>() {
                                @Override public void apply(IgniteInternalFuture<?> msgFut) {
                                    try {
                                        msgFut.get();

                                        connectedNew(recoveryDesc, ses, false);
                                    }
                                    catch (IgniteCheckedException e) {
                                        if (log.isDebugEnabled())
                                            log.debug("Failed to send recovery handshake " +
                                                "[rmtNode=" + rmtNode.id() + ", err=" + e + ']');

                                        recoveryDesc.release();
                                    }
                                }
                            };

                            nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(recoveryDesc.received()), lsnr);
                        }
                        else
                            nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(ALREADY_CONNECTED));
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Failed to send message: " + e, e);
                    }
                }
            }

            /**
             *
             */
            @SuppressWarnings("PackageVisibleInnerClass")
            class ConnectClosure implements IgniteInClosure<Boolean> {
                /** */
                private static final long serialVersionUID = 0L;

                /** */
                private final GridNioSession ses;

                /** */
                private final GridNioRecoveryDescriptor recoveryDesc;

                /** */
                private final ClusterNode rmtNode;

                /** */
                private final HandshakeMessage msg;

                /** */
                private final GridFutureAdapter<GridCommunicationClient> fut;

                /** */
                private final boolean createClient;

                /** */
                private final ConnectionKey connKey;

                /**
                 * @param ses Incoming session.
                 * @param recoveryDesc Recovery descriptor.
                 * @param rmtNode Remote node.
                 * @param connKey Connection key.
                 * @param msg Handshake message.
                 * @param createClient If {@code true} creates NIO communication client..
                 * @param fut Connect future.
                 */
                ConnectClosure(GridNioSession ses,
                    GridNioRecoveryDescriptor recoveryDesc,
                    ClusterNode rmtNode,
                    ConnectionKey connKey,
                    HandshakeMessage msg,
                    boolean createClient,
                    GridFutureAdapter<GridCommunicationClient> fut) {
                    this.ses = ses;
                    this.recoveryDesc = recoveryDesc;
                    this.rmtNode = rmtNode;
                    this.connKey = connKey;
                    this.msg = msg;
                    this.createClient = createClient;
                    this.fut = fut;
                }

                /** {@inheritDoc} */
                @Override public void apply(Boolean success) {
                    if (success) {
                        try {
                            IgniteInClosure<IgniteInternalFuture<?>> lsnr = new IgniteInClosure<IgniteInternalFuture<?>>() {
                                @Override public void apply(IgniteInternalFuture<?> msgFut) {
                                    try {
                                        msgFut.get();

                                        GridTcpNioCommunicationClient client =
                                            connected(recoveryDesc, ses, rmtNode, msg.received(), false, createClient);

                                        fut.onDone(client);
                                    }
                                    catch (IgniteCheckedException e) {
                                        if (log.isDebugEnabled())
                                            log.debug("Failed to send recovery handshake " +
                                                "[rmtNode=" + rmtNode.id() + ", err=" + e + ']');

                                        recoveryDesc.release();

                                        fut.onDone();
                                    }
                                    finally {
                                        clientFuts.remove(connKey, fut);
                                    }
                                }
                            };

                            nioSrvr.sendSystem(ses, new RecoveryLastReceivedMessage(recoveryDesc.received()), lsnr);
                        }
                        catch (IgniteCheckedException e) {
                            U.error(log, "Failed to send message: " + e, e);
                        }
                    }
                    else {
                        try {
                            fut.onDone();
                        }
                        finally {
                            clientFuts.remove(connKey, fut);
                        }
                    }
                }
            }
        };

    /** Logger. */
    @LoggerResource
    private IgniteLogger log;

    /** Logger. */
    @LoggerResource(categoryName = "org.apache.ignite.internal.diagnostic")
    private IgniteLogger diagnosticLog;

    /** Local IP address. */
    private String locAddr;

    /** Complex variable that represents this node IP address. */
    private volatile InetAddress locHost;

    /** Local port which node uses. */
    private int locPort = DFLT_PORT;

    /** Local port range. */
    private int locPortRange = DFLT_PORT_RANGE;

    /** Local port which node uses to accept shared memory connections. */
    private int shmemPort = DFLT_SHMEM_PORT;

    /** Allocate direct buffer or heap buffer. */
    private boolean directBuf = true;

    /** Allocate direct buffer or heap buffer. */
    private boolean directSndBuf;

    /** Idle connection timeout. */
    private long idleConnTimeout = DFLT_IDLE_CONN_TIMEOUT;

    /** Connect timeout. */
    private long connTimeout = DFLT_CONN_TIMEOUT;

    /** Maximum connect timeout. */
    private long maxConnTimeout = DFLT_MAX_CONN_TIMEOUT;

    /** Reconnect attempts count. */
    private int reconCnt = DFLT_RECONNECT_CNT;

    /** Socket send buffer. */
    private int sockSndBuf = DFLT_SOCK_BUF_SIZE;

    /** Socket receive buffer. */
    private int sockRcvBuf = DFLT_SOCK_BUF_SIZE;

    /** Message queue limit. */
    private int msgQueueLimit = DFLT_MSG_QUEUE_LIMIT;

    /** Slow client queue limit. */
    private int slowClientQueueLimit;

    /** NIO server. */
    private GridNioServer<Message> nioSrvr;

    /** Shared memory server. */
    private IpcSharedMemoryServerEndpoint shmemSrv;

    /** */
    private boolean usePairedConnections;

    /** */
    private int connectionsPerNode = DFLT_CONN_PER_NODE;

    /** {@code TCP_NODELAY} option value for created sockets. */
    private boolean tcpNoDelay = DFLT_TCP_NODELAY;

    /** {@code FILTER_REACHABLE_ADDRESSES} option value for created sockets. */
    private boolean filterReachableAddresses = DFLT_FILTER_REACHABLE_ADDRESSES;

    /** Number of received messages after which acknowledgment is sent. */
    private int ackSndThreshold = DFLT_ACK_SND_THRESHOLD;

    /** Maximum number of unacknowledged messages. */
    private int unackedMsgsBufSize;

    /** Socket write timeout. */
    private long sockWriteTimeout = DFLT_SOCK_WRITE_TIMEOUT;

    /** Recovery and idle clients handler. */
    private CommunicationWorker commWorker;

    /** Shared memory accept worker. */
    private ShmemAcceptWorker shmemAcceptWorker;

    /** Shared memory workers. */
    private final Collection<ShmemWorker> shmemWorkers = new ConcurrentLinkedDeque<>();

    /** Clients. */
    private final ConcurrentMap<UUID, GridCommunicationClient[]> clients = GridConcurrentFactory.newMap();

    /** SPI listener. */
    private volatile CommunicationListener<Message> lsnr;

    /** Bound port. */
    private int boundTcpPort = -1;

    /** Bound port for shared memory server. */
    private int boundTcpShmemPort = -1;

    /** Count of selectors to use in TCP server. */
    private int selectorsCnt = DFLT_SELECTORS_CNT;

    /**
     * Defines how many non-blocking {@code selector.selectNow()} should be made before
     * falling into {@code selector.select(long)} in NIO server. Long value. Default is {@code 0}.
     * Can be set to {@code Long.MAX_VALUE} so selector threads will never block.
     */
    private long selectorSpins = IgniteSystemProperties.getLong("IGNITE_SELECTOR_SPINS", 0L);

    /** Address resolver. */
    private AddressResolver addrRslvr;

    /** Context initialization latch. */
    private final CountDownLatch ctxInitLatch = new CountDownLatch(1);

    /** Stopping flag (set to {@code true} when SPI gets stopping signal). */
    private volatile boolean stopping;

    /** Statistics. */
    private TcpCommunicationMetricsListener metricsLsnr;

    /** Client connect futures. */
    private final ConcurrentMap<ConnectionKey, GridFutureAdapter<GridCommunicationClient>> clientFuts =
        GridConcurrentFactory.newMap();

    /** */
    private final ConcurrentMap<ConnectionKey, GridNioRecoveryDescriptor> recoveryDescs = GridConcurrentFactory.newMap();

    /** */
    private final ConcurrentMap<ConnectionKey, GridNioRecoveryDescriptor> outRecDescs = GridConcurrentFactory.newMap();

    /** */
    private final ConcurrentMap<ConnectionKey, GridNioRecoveryDescriptor> inRecDescs = GridConcurrentFactory.newMap();

    /** */
    private final GridLocalEventListener discoLsnr = new DiscoveryListener();

    /** Tracing. */
    protected Tracing tracing;

    /**
     * @return {@code True} if ssl enabled.
     */
    private boolean isSslEnabled() {
        return ignite.configuration().getSslContextFactory() != null;
    }

    /**
     * Sets address resolver.
     *
     * @param addrRslvr Address resolver.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setAddressResolver(AddressResolver addrRslvr) {
        // Injection should not override value already set by Spring or user.
        if (this.addrRslvr == null)
            this.addrRslvr = addrRslvr;

        return this;
    }

    /**
     * See {@link #setAddressResolver(AddressResolver)}.
     *
     * @return Address resolver.
     */
    public AddressResolver getAddressResolver() {
        return addrRslvr;
    }

    /**
     * Injects resources.
     *
     * @param ignite Ignite.
     */
    @IgniteInstanceResource
    @Override protected void injectResources(Ignite ignite) {
        super.injectResources(ignite);

        if (ignite != null) {
            setAddressResolver(ignite.configuration().getAddressResolver());
            setLocalAddress(ignite.configuration().getLocalHost());
            tracing = ignite instanceof IgniteEx ? ((IgniteEx)ignite).context().tracing() : new NoopTracing();
        }
    }

    /**
     * Sets local host address for socket binding. Note that one node could have
     * additional addresses beside the loopback one. This configuration
     * parameter is optional.
     *
     * @param locAddr IP address. Default value is any available local
     *      IP address.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setLocalAddress(String locAddr) {
        // Injection should not override value already set by Spring or user.
        if (this.locAddr == null)
            this.locAddr = locAddr;

        return this;
    }

    /**
     * See {@link #setLocalAddress(String)}.
     *
     * @return Grid node IP address.
     */
    public String getLocalAddress() {
        return locAddr;
    }

    /**
     * Sets local port for socket binding.
     * <p>
     * If not provided, default value is {@link #DFLT_PORT}.
     *
     * @param locPort Port number.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setLocalPort(int locPort) {
        this.locPort = locPort;

        return this;
    }

    /**
     * See {@link #setLocalPort(int)}.
     *
     * @return Port number.
     */
    public int getLocalPort() {
        return locPort;
    }

    /**
     * Sets local port range for local host ports (value must greater than or equal to <tt>0</tt>).
     * If provided local port (see {@link #setLocalPort(int)}} is occupied,
     * implementation will try to increment the port number for as long as it is less than
     * initial value plus this range.
     * <p>
     * If port range value is <tt>0</tt>, then implementation will try bind only to the port provided by
     * {@link #setLocalPort(int)} method and fail if binding to this port did not succeed.
     * <p>
     * Local port range is very useful during development when more than one grid nodes need to run
     * on the same physical machine.
     * <p>
     * If not provided, default value is {@link #DFLT_PORT_RANGE}.
     *
     * @param locPortRange New local port range.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setLocalPortRange(int locPortRange) {
        this.locPortRange = locPortRange;

        return this;
    }

    /**
     * See {@link #setLocalPortRange(int)}.
     *
     * @return Local Port range.
     */
    public int getLocalPortRange() {
        return locPortRange;
    }

    /**
     * See {@link #setUsePairedConnections(boolean)}.
     *
     * @return {@code true} to use paired connections and {@code false} otherwise.
     */
    public boolean isUsePairedConnections() {
        return usePairedConnections;
    }

    /**
     * Set this to {@code true} if {@code TcpCommunicationSpi} should
     * maintain connection for outgoing and incoming messages separately.
     * In this case total number of connections between local and each remote node
     * is {@link #getConnectionsPerNode()} * 2.
     * <p>
     * Set this to {@code false} if each connection of {@link #getConnectionsPerNode()}
     * should be used for outgoing and incoming messages. In this case total number
     * of connections between local and each remote node is {@link #getConnectionsPerNode()}.
     * <p>
     * Default is {@code false}.
     *
     * @param usePairedConnections {@code true} to use paired connections and {@code false} otherwise.
     * @return {@code this} for chaining.
     * @see #getConnectionsPerNode()
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setUsePairedConnections(boolean usePairedConnections) {
        this.usePairedConnections = usePairedConnections;

        return this;
    }

    /**
     * Sets number of connections to each remote node. if {@link #isUsePairedConnections()}
     * is {@code true} then number of connections is doubled and half is used for incoming and
     * half for outgoing messages.
     *
     * @param maxConnectionsPerNode Number of connections per node.
     * @return {@code this} for chaining.
     * @see #isUsePairedConnections()
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setConnectionsPerNode(int maxConnectionsPerNode) {
        this.connectionsPerNode = maxConnectionsPerNode;

        return this;
    }

    /**
     * See {@link #setConnectionsPerNode(int)}.
     *
     * @return Number of connections per node.
     */
    public int getConnectionsPerNode() {
        return connectionsPerNode;
    }

    /**
     * Sets local port to accept shared memory connections.
     * <p>
     * If set to {@code -1} shared memory communication will be disabled.
     * <p>
     * If not provided, default value is {@link #DFLT_SHMEM_PORT}.
     *
     * @param shmemPort Port number.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSharedMemoryPort(int shmemPort) {
        this.shmemPort = shmemPort;

        return this;
    }

    /**
     * See {@link #setSharedMemoryPort(int)}.
     *
     * @return Port number.
     */
    public int getSharedMemoryPort() {
        return shmemPort;
    }

    /**
     * Sets maximum idle connection timeout upon which a connection
     * to client will be closed.
     * <p>
     * If not provided, default value is {@link #DFLT_IDLE_CONN_TIMEOUT}.
     *
     * @param idleConnTimeout Maximum idle connection time.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setIdleConnectionTimeout(long idleConnTimeout) {
        this.idleConnTimeout = idleConnTimeout;

        return this;
    }

    /**
     * See {@link #setIdleConnectionTimeout(long)}.
     *
     * @return Maximum idle connection time.
     */
    public long getIdleConnectionTimeout() {
        return idleConnTimeout;
    }

    /**
     * See {@link #setSocketWriteTimeout(long)}.
     *
     * @return Socket write timeout for TCP connections.
     */
    public long getSocketWriteTimeout() {
        return sockWriteTimeout;
    }

    /**
     * Sets socket write timeout for TCP connection. If message can not be written to
     * socket within this time then connection is closed and reconnect is attempted.
     * <p>
     * Default to {@link #DFLT_SOCK_WRITE_TIMEOUT}.
     *
     * @param sockWriteTimeout Socket write timeout for TCP connection.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSocketWriteTimeout(long sockWriteTimeout) {
        this.sockWriteTimeout = sockWriteTimeout;

        return this;
    }

    /**
     * See {@link #setAckSendThreshold(int)}.
     *
     * @return Number of received messages after which acknowledgment is sent.
     */
    public int getAckSendThreshold() {
        return ackSndThreshold;
    }

    /**
     * Sets number of received messages per connection to node after which acknowledgment message is sent.
     * <p>
     * Default to {@link #DFLT_ACK_SND_THRESHOLD}.
     *
     * @param ackSndThreshold Number of received messages after which acknowledgment is sent.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setAckSendThreshold(int ackSndThreshold) {
        this.ackSndThreshold = ackSndThreshold;

        return this;
    }

    /**
     * See {@link #setUnacknowledgedMessagesBufferSize(int)}.
     *
     * @return Maximum number of unacknowledged messages.
     */
    public int getUnacknowledgedMessagesBufferSize() {
        return unackedMsgsBufSize;
    }

    /**
     * Sets maximum number of stored unacknowledged messages per connection to node.
     * If number of unacknowledged messages exceeds this number then connection to node is
     * closed and reconnect is attempted.
     *
     * @param unackedMsgsBufSize Maximum number of unacknowledged messages.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setUnacknowledgedMessagesBufferSize(int unackedMsgsBufSize) {
        this.unackedMsgsBufSize = unackedMsgsBufSize;

        return this;
    }

    /**
     * Sets connect timeout used when establishing connection
     * with remote nodes.
     * <p>
     * {@code 0} is interpreted as infinite timeout.
     * <p>
     * If not provided, default value is {@link #DFLT_CONN_TIMEOUT}.
     * <p>
     * When this property is explicitly set {@link IgniteConfiguration#getFailureDetectionTimeout()} is ignored.
     *
     * @param connTimeout Connect timeout.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setConnectTimeout(long connTimeout) {
        this.connTimeout = connTimeout;

        failureDetectionTimeoutEnabled(false);

        return this;
    }

    /**
     * See {@link #setConnectTimeout(long)}.
     *
     * @return Connect timeout.
     */
    public long getConnectTimeout() {
        return connTimeout;
    }

    /**
     * Sets maximum connect timeout. If handshake is not established within connect timeout,
     * then SPI tries to repeat handshake procedure with increased connect timeout.
     * Connect timeout can grow till maximum timeout value,
     * if maximum timeout value is reached then the handshake is considered as failed.
     * <p>
     * {@code 0} is interpreted as infinite timeout.
     * <p>
     * If not provided, default value is {@link #DFLT_MAX_CONN_TIMEOUT}.
     * <p>
     * When this property is explicitly set {@link IgniteConfiguration#getFailureDetectionTimeout()} is ignored.
     *
     * @param maxConnTimeout Maximum connect timeout.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setMaxConnectTimeout(long maxConnTimeout) {
        this.maxConnTimeout = maxConnTimeout;

        failureDetectionTimeoutEnabled(false);

        return this;
    }

    /**
     * Gets maximum connect timeout.
     *
     * @return Maximum connect timeout.
     */
    public long getMaxConnectTimeout() {
        return maxConnTimeout;
    }

    /**
     * Sets maximum number of reconnect attempts used when establishing connection
     * with remote nodes.
     * <p>
     * If not provided, default value is {@link #DFLT_RECONNECT_CNT}.
     * <p>
     * When this property is explicitly set {@link IgniteConfiguration#getFailureDetectionTimeout()} is ignored.
     *
     * @param reconCnt Maximum number of reconnection attempts.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setReconnectCount(int reconCnt) {
        this.reconCnt = reconCnt;

        failureDetectionTimeoutEnabled(false);

        return this;
    }

    /**
     * Gets maximum number of reconnect attempts used when establishing connection
     * with remote nodes.
     *
     * @return Reconnects count.
     */
    public int getReconnectCount() {
        return reconCnt;
    }

    /**
     * Sets flag to allocate direct or heap buffer in SPI.
     * If value is {@code true}, then SPI will use {@link ByteBuffer#allocateDirect(int)} call.
     * Otherwise, SPI will use {@link ByteBuffer#allocate(int)} call.
     * <p>
     * If not provided, default value is {@code true}.
     *
     * @param directBuf Flag indicates to allocate direct or heap buffer in SPI.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setDirectBuffer(boolean directBuf) {
        this.directBuf = directBuf;

        return this;
    }

    /**
     * Gets flag that indicates whether direct or heap allocated buffer is used.
     *
     * @return Flag that indicates whether direct or heap allocated buffer is used.
     */
    public boolean isDirectBuffer() {
        return directBuf;
    }

    /**
     * Gets flag defining whether direct send buffer should be used.
     *
     * @return {@code True} if direct buffers should be used.
     */
    public boolean isDirectSendBuffer() {
        return directSndBuf;
    }

    /**
     * Sets whether to use direct buffer for sending.
     *
     * If not provided default is {@code false}.
     *
     * @param directSndBuf {@code True} to use direct buffers for send.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setDirectSendBuffer(boolean directSndBuf) {
        this.directSndBuf = directSndBuf;

        return this;
    }

    /**
     * Sets the count of selectors te be used in TCP server.
     * <p/>
     * If not provided, default value is {@link #DFLT_SELECTORS_CNT}.
     *
     * @param selectorsCnt Selectors count.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSelectorsCount(int selectorsCnt) {
        this.selectorsCnt = selectorsCnt;

        return this;
    }

    /**
     * See {@link #setSelectorsCount(int)}.
     *
     * @return Count of selectors in TCP server.
     */
    public int getSelectorsCount() {
        return selectorsCnt;
    }

    /**
     * See {@link #setSelectorSpins(long)}.
     *
     * @return Selector thread busy-loop iterations.
     */
    public long getSelectorSpins() {
        return selectorSpins;
    }

    /**
     * Defines how many non-blocking {@code selector.selectNow()} should be made before
     * falling into {@code selector.select(long)} in NIO server. Long value. Default is {@code 0}.
     * Can be set to {@code Long.MAX_VALUE} so selector threads will never block.
     *
     * @param selectorSpins Selector thread busy-loop iterations.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSelectorSpins(long selectorSpins) {
        this.selectorSpins = selectorSpins;

        return this;
    }

    /**
     * Sets value for {@code TCP_NODELAY} socket option. Each
     * socket will be opened using provided value.
     * <p>
     * Setting this option to {@code true} disables Nagle's algorithm
     * for socket decreasing latency and delivery time for small messages.
     * <p>
     * For systems that work under heavy network load it is advisable to
     * set this value to {@code false}.
     * <p>
     * If not provided, default value is {@link #DFLT_TCP_NODELAY}.
     *
     * @param tcpNoDelay {@code True} to disable TCP delay.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setTcpNoDelay(boolean tcpNoDelay) {
        this.tcpNoDelay = tcpNoDelay;

        return this;
    }

    /**
     * Gets value for {@code TCP_NODELAY} socket option.
     *
     * @return {@code True} if TCP delay is disabled.
     */
    public boolean isTcpNoDelay() {
        return tcpNoDelay;
    }

    /**
     * Gets value for {@code FILTER_REACHABLE_ADDRESSES} socket option.
     *
     * @return {@code True} if needed to filter reachable addresses.
     */
    public boolean isFilterReachableAddresses() {
        return filterReachableAddresses;
    }

    /**
     * Setting this option to {@code true} enables filter for reachable
     * addresses on creating tcp client.
     * <p>
     * Usually its advised to set this value to {@code false}.
     * <p>
     * If not provided, default value is {@link #DFLT_FILTER_REACHABLE_ADDRESSES}.
     *
     * @param filterReachableAddresses {@code True} to filter reachable addresses.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setFilterReachableAddresses(boolean filterReachableAddresses) {
        this.filterReachableAddresses = filterReachableAddresses;

        return this;
    }

    /**
     * Sets receive buffer size for sockets created or accepted by this SPI.
     * <p>
     * If not provided, default is {@link #DFLT_SOCK_BUF_SIZE}.
     *
     * @param sockRcvBuf Socket receive buffer size.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSocketReceiveBuffer(int sockRcvBuf) {
        this.sockRcvBuf = sockRcvBuf;

        return this;
    }

    /**
     * See {@link #setSocketReceiveBuffer(int)}.
     *
     * @return Socket receive buffer size.
     */
    public int getSocketReceiveBuffer() {
        return sockRcvBuf;
    }

    /**
     * Sets send buffer size for sockets created or accepted by this SPI.
     * <p>
     * If not provided, default is {@link #DFLT_SOCK_BUF_SIZE}.
     *
     * @param sockSndBuf Socket send buffer size.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSocketSendBuffer(int sockSndBuf) {
        this.sockSndBuf = sockSndBuf;

        return this;
    }

    /**
     * See {@link #setSocketSendBuffer(int)}.
     *
     * @return Socket send buffer size.
     */
    public int getSocketSendBuffer() {
        return sockSndBuf;
    }

    /**
     * Sets message queue limit for incoming and outgoing messages.
     * <p>
     * When set to positive number send queue is limited to the configured value.
     * {@code 0} disables the size limitations.
     * <p>
     * If not provided, default is {@link #DFLT_MSG_QUEUE_LIMIT}.
     *
     * @param msgQueueLimit Send queue size limit.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setMessageQueueLimit(int msgQueueLimit) {
        this.msgQueueLimit = msgQueueLimit;

        return this;
    }

    /**
     * Gets message queue limit for incoming and outgoing messages.
     *
     * @return Send queue size limit.
     */
    public int getMessageQueueLimit() {
        return msgQueueLimit;
    }

    /**
     * See {@link #setSlowClientQueueLimit(int)}.
     *
     * @return Slow client queue limit.
     */
    public int getSlowClientQueueLimit() {
        return slowClientQueueLimit;
    }

    /**
     * Sets slow client queue limit.
     * <p/>
     * When set to a positive number, communication SPI will monitor clients outbound message queue sizes and will drop
     * those clients whose queue exceeded this limit.
     * <p/>
     * This value should be set to less or equal value than {@link #getMessageQueueLimit()} which controls
     * message back-pressure for server nodes. The default value for this parameter is {@code 0}
     * which means {@code unlimited}.
     *
     * @param slowClientQueueLimit Slow client queue limit.
     * @return {@code this} for chaining.
     */
    @IgniteSpiConfiguration(optional = true)
    public TcpCommunicationSpi setSlowClientQueueLimit(int slowClientQueueLimit) {
        this.slowClientQueueLimit = slowClientQueueLimit;

        return this;
    }

    /** {@inheritDoc} */
    @Override public void setListener(CommunicationListener<Message> lsnr) {
        this.lsnr = lsnr;
    }

    /**
     * @return Listener.
     */
    public CommunicationListener getListener() {
        return lsnr;
    }

    /** {@inheritDoc} */
    @Override public int getSentMessagesCount() {
        // Listener could be not initialized yet, but discovery thread could try to aggregate metrics.
        if (metricsLsnr == null)
            return 0;

        return metricsLsnr.sentMessagesCount();
    }

    /** {@inheritDoc} */
    @Override public long getSentBytesCount() {
        // Listener could be not initialized yet, but discovery thread clould try to aggregate metrics.
        if (metricsLsnr == null)
            return 0;

        return metricsLsnr.sentBytesCount();
    }

    /** {@inheritDoc} */
    @Override public int getReceivedMessagesCount() {
        // Listener could be not initialized yet, but discovery thread could try to aggregate metrics.
        if (metricsLsnr == null)
            return 0;

        return metricsLsnr.receivedMessagesCount();
    }

    /** {@inheritDoc} */
    @Override public long getReceivedBytesCount() {
        // Listener could be not initialized yet, but discovery thread could try to aggregate metrics.
        if (metricsLsnr == null)
            return 0;

        return metricsLsnr.receivedBytesCount();
    }

    /**
     * Gets received messages counts (grouped by type).
     *
     * @return Map containing message types and respective counts.
     */
    public Map<String, Long> getReceivedMessagesByType() {
        return metricsLsnr.receivedMessagesByType();
    }

    /**
     * Gets received messages counts (grouped by node).
     *
     * @return Map containing sender nodes and respective counts.
     */
    public Map<UUID, Long> getReceivedMessagesByNode() {
        return metricsLsnr.receivedMessagesByNode();
    }

    /**
     * Gets sent messages counts (grouped by type).
     *
     * @return Map containing message types and respective counts.
     */
    public Map<String, Long> getSentMessagesByType() {
        return metricsLsnr.sentMessagesByType();
    }

    /**
     * Gets sent messages counts (grouped by node).
     *
     * @return Map containing receiver nodes and respective counts.
     */
    public Map<UUID, Long> getSentMessagesByNode() {
        return metricsLsnr.sentMessagesByNode();
    }

    /** {@inheritDoc} */
    @Override public int getOutboundMessagesQueueSize() {
        GridNioServer<Message> srv = nioSrvr;

        return srv != null ? srv.outboundMessagesQueueSize() : 0;
    }

    /** {@inheritDoc} */
    @Override public void resetMetrics() {
        metricsLsnr.resetMetrics();
    }

    /**
     * @param nodeId Target node ID.
     * @return Future.
     */
    public IgniteInternalFuture<String> dumpNodeStatistics(final UUID nodeId) {
        StringBuilder sb = new StringBuilder("Communication SPI statistics [rmtNode=").append(nodeId).append(']').append(U.nl());

        dumpInfo(sb, nodeId);

        GridNioServer<Message> nioSrvr = this.nioSrvr;

        if (nioSrvr != null) {
            sb.append("NIO sessions statistics:");

            IgnitePredicate<GridNioSession> p = new IgnitePredicate<GridNioSession>() {
                @Override public boolean apply(GridNioSession ses) {
                    ConnectionKey connId = ses.meta(CONN_IDX_META);

                    return connId != null && nodeId.equals(connId.nodeId());
                }
            };

            return nioSrvr.dumpStats(sb.toString(), p);
        }
        else {
            sb.append(U.nl()).append("GridNioServer is null.");

            return new GridFinishedFuture<>(sb.toString());
        }
    }

    /**
     * Dumps SPI per-connection stats to logs.
     */
    public void dumpStats() {
        final IgniteLogger log = this.diagnosticLog;

        if (log != null) {
            StringBuilder sb = new StringBuilder();

            dumpInfo(sb, null);

            U.warn(log, sb.toString());

            GridNioServer<Message> nioSrvr = this.nioSrvr;

            if (nioSrvr != null) {
                nioSrvr.dumpStats().listen(new CI1<IgniteInternalFuture<String>>() {
                    @Override public void apply(IgniteInternalFuture<String> fut) {
                        try {
                            U.warn(log, fut.get());
                        }
                        catch (Exception e) {
                            U.error(log, "Failed to dump NIO server statistics: " + e, e);
                        }
                    }
                });
            }
        }
    }

    /**
     * @param sb Message builder.
     * @param dstNodeId Target node ID.
     */
    private void dumpInfo(StringBuilder sb, UUID dstNodeId) {
        sb.append("Communication SPI recovery descriptors: ").append(U.nl());

        for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : recoveryDescs.entrySet()) {
            GridNioRecoveryDescriptor desc = entry.getValue();

            if (dstNodeId != null && !dstNodeId.equals(entry.getKey().nodeId()))
                continue;

            sb.append("    [key=").append(entry.getKey())
                .append(", msgsSent=").append(desc.sent())
                .append(", msgsAckedByRmt=").append(desc.acked())
                .append(", msgsRcvd=").append(desc.received())
                .append(", lastAcked=").append(desc.lastAcknowledged())
                .append(", reserveCnt=").append(desc.reserveCount())
                .append(", descIdHash=").append(System.identityHashCode(desc))
                .append(']').append(U.nl());
        }

        for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : outRecDescs.entrySet()) {
            GridNioRecoveryDescriptor desc = entry.getValue();

            if (dstNodeId != null && !dstNodeId.equals(entry.getKey().nodeId()))
                continue;

            sb.append("    [key=").append(entry.getKey())
                .append(", msgsSent=").append(desc.sent())
                .append(", msgsAckedByRmt=").append(desc.acked())
                .append(", reserveCnt=").append(desc.reserveCount())
                .append(", connected=").append(desc.connected())
                .append(", reserved=").append(desc.reserved())
                .append(", descIdHash=").append(System.identityHashCode(desc))
                .append(']').append(U.nl());
        }

        for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> entry : inRecDescs.entrySet()) {
            GridNioRecoveryDescriptor desc = entry.getValue();

            if (dstNodeId != null && !dstNodeId.equals(entry.getKey().nodeId()))
                continue;

            sb.append("    [key=").append(entry.getKey())
                .append(", msgsRcvd=").append(desc.received())
                .append(", lastAcked=").append(desc.lastAcknowledged())
                .append(", reserveCnt=").append(desc.reserveCount())
                .append(", connected=").append(desc.connected())
                .append(", reserved=").append(desc.reserved())
                .append(", handshakeIdx=").append(desc.handshakeIndex())
                .append(", descIdHash=").append(System.identityHashCode(desc))
                .append(']').append(U.nl());
        }

        sb.append("Communication SPI clients: ").append(U.nl());

        for (Map.Entry<UUID, GridCommunicationClient[]> entry : clients.entrySet()) {
            UUID clientNodeId = entry.getKey();

            if (dstNodeId != null && !dstNodeId.equals(clientNodeId))
                continue;

            GridCommunicationClient[] clients0 = entry.getValue();

            for (GridCommunicationClient client : clients0) {
                if (client != null) {
                    sb.append("    [node=").append(clientNodeId)
                        .append(", client=").append(client)
                        .append(']').append(U.nl());
                }
            }
        }
    }

    /** {@inheritDoc} */
    @Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException {
        initFailureDetectionTimeout();

        assertParameter(locPort > 1023, "locPort > 1023");
        assertParameter(locPort <= 0xffff, "locPort < 0xffff");
        assertParameter(locPortRange >= 0, "locPortRange >= 0");
        assertParameter(idleConnTimeout > 0, "idleConnTimeout > 0");
        assertParameter(sockRcvBuf >= 0, "sockRcvBuf >= 0");
        assertParameter(sockSndBuf >= 0, "sockSndBuf >= 0");
        assertParameter(msgQueueLimit >= 0, "msgQueueLimit >= 0");
        assertParameter(shmemPort > 0 || shmemPort == -1, "shmemPort > 0 || shmemPort == -1");
        assertParameter(selectorsCnt > 0, "selectorsCnt > 0");
        assertParameter(connectionsPerNode > 0, "connectionsPerNode > 0");
        assertParameter(connectionsPerNode <= MAX_CONN_PER_NODE, "connectionsPerNode <= 1024");

        if (!failureDetectionTimeoutEnabled()) {
            assertParameter(reconCnt > 0, "reconnectCnt > 0");
            assertParameter(connTimeout >= 0, "connTimeout >= 0");
            assertParameter(maxConnTimeout >= connTimeout, "maxConnTimeout >= connTimeout");
        }

        assertParameter(sockWriteTimeout >= 0, "sockWriteTimeout >= 0");
        assertParameter(ackSndThreshold > 0, "ackSndThreshold > 0");
        assertParameter(unackedMsgsBufSize >= 0, "unackedMsgsBufSize >= 0");

        if (unackedMsgsBufSize > 0) {
            assertParameter(unackedMsgsBufSize >= msgQueueLimit * 5,
                "Specified 'unackedMsgsBufSize' is too low, it should be at least 'msgQueueLimit * 5'.");

            assertParameter(unackedMsgsBufSize >= ackSndThreshold * 5,
                "Specified 'unackedMsgsBufSize' is too low, it should be at least 'ackSndThreshold * 5'.");
        }

        if (connectionsPerNode > 1)
            connPlc = new RoundRobinConnectionPolicy();
        else
            connPlc = new FirstConnectionPolicy();

        chConnPlc = new ConnectionPolicy() {
            /** Sequential connection index provider. */
            private final AtomicInteger chIdx = new AtomicInteger(MAX_CONN_PER_NODE + 1);

            @Override public int connectionIndex() {
                return chIdx.incrementAndGet();
            }
        };

        try {
            locHost = U.resolveLocalHost(locAddr);
        }
        catch (IOException e) {
            throw new IgniteSpiException("Failed to initialize local address: " + locAddr, e);
        }

        try {
            shmemSrv = resetShmemServer();
        }
        catch (IgniteCheckedException e) {
            U.warn(log, "Failed to start shared memory communication server.", e);
        }

        try {
            // This method potentially resets local port to the value
            // local node was bound to.
            nioSrvr = resetNioServer();
        }
        catch (IgniteCheckedException e) {
            throw new IgniteSpiException("Failed to initialize TCP server: " + locHost, e);
        }

        // Set local node attributes.
        try {
            IgniteBiTuple<Collection<String>, Collection<String>> addrs = U.resolveLocalAddresses(locHost);

            Collection<InetSocketAddress> extAddrs = addrRslvr == null ? null :
                U.resolveAddresses(addrRslvr, F.flat(Arrays.asList(addrs.get1(), addrs.get2())), boundTcpPort);

            Map<String, Object> res = new HashMap<>(5);

            boolean setEmptyHostNamesAttr = !getBoolean(IGNITE_TCP_COMM_SET_ATTR_HOST_NAMES, false) &&
                (!F.isEmpty(locAddr) && locHost.getHostAddress().equals(locAddr)) && !locHost.isAnyLocalAddress() &&
                !locHost.isLoopbackAddress();

            res.put(createSpiAttributeName(ATTR_ADDRS), addrs.get1());
            res.put(createSpiAttributeName(ATTR_HOST_NAMES), setEmptyHostNamesAttr ? emptyList() : addrs.get2());
            res.put(createSpiAttributeName(ATTR_PORT), boundTcpPort);
            res.put(createSpiAttributeName(ATTR_SHMEM_PORT), boundTcpShmemPort >= 0 ? boundTcpShmemPort : null);
            res.put(createSpiAttributeName(ATTR_EXT_ADDRS), extAddrs);
            res.put(createSpiAttributeName(ATTR_PAIRED_CONN), usePairedConnections);

            return res;
        }
        catch (IOException | IgniteCheckedException e) {
            throw new IgniteSpiException("Failed to resolve local host to addresses: " + locHost, e);
        }
    }

    /**
     * @return Bound TCP server port.
     */
    public int boundPort() {
        return boundTcpPort;
    }

    /** {@inheritDoc} */
    @Override public void spiStart(String igniteInstanceName) throws IgniteSpiException {
        assert locHost != null;

        // Start SPI start stopwatch.
        startStopwatch();

        // Ack parameters.
        if (log.isDebugEnabled()) {
            log.debug(configInfo("locAddr", locAddr));
            log.debug(configInfo("locPort", locPort));
            log.debug(configInfo("locPortRange", locPortRange));
            log.debug(configInfo("idleConnTimeout", idleConnTimeout));
            log.debug(configInfo("directBuf", directBuf));
            log.debug(configInfo("directSendBuf", directSndBuf));
            log.debug(configInfo("selectorsCnt", selectorsCnt));
            log.debug(configInfo("tcpNoDelay", tcpNoDelay));
            log.debug(configInfo("sockSndBuf", sockSndBuf));
            log.debug(configInfo("sockRcvBuf", sockRcvBuf));
            log.debug(configInfo("shmemPort", shmemPort));
            log.debug(configInfo("msgQueueLimit", msgQueueLimit));
            log.debug(configInfo("connectionsPerNode", connectionsPerNode));

            if (failureDetectionTimeoutEnabled()) {
                log.debug(configInfo("connTimeout", connTimeout));
                log.debug(configInfo("maxConnTimeout", maxConnTimeout));
                log.debug(configInfo("reconCnt", reconCnt));
            }
            else
                log.debug(configInfo("failureDetectionTimeout", failureDetectionTimeout()));

            log.debug(configInfo("sockWriteTimeout", sockWriteTimeout));
            log.debug(configInfo("ackSndThreshold", ackSndThreshold));
            log.debug(configInfo("unackedMsgsBufSize", unackedMsgsBufSize));
        }

        if (!tcpNoDelay)
            U.quietAndWarn(log, "'TCP_NO_DELAY' for communication is off, which should be used with caution " +
                "since may produce significant delays with some scenarios.");

        if (slowClientQueueLimit > 0 && msgQueueLimit > 0 && slowClientQueueLimit >= msgQueueLimit) {
            U.quietAndWarn(log, "Slow client queue limit is set to a value greater than or equal to message " +
                "queue limit (slow client queue limit will have no effect) [msgQueueLimit=" + msgQueueLimit +
                ", slowClientQueueLimit=" + slowClientQueueLimit + ']');
        }

        if (msgQueueLimit == 0)
            U.quietAndWarn(log, "Message queue limit is set to 0 which may lead to " +
                "potential OOMEs when running cache operations in FULL_ASYNC or PRIMARY_SYNC modes " +
                "due to message queues growth on sender and receiver sides.");

        registerMBean(igniteInstanceName, new TcpCommunicationSpiMBeanImpl(this), TcpCommunicationSpiMBean.class);

        connectGate = new ConnectGateway();

        if (shmemSrv != null) {
            shmemAcceptWorker = new ShmemAcceptWorker(shmemSrv);

            new IgniteThread(shmemAcceptWorker).start();
        }

        nioSrvr.start();

        commWorker = new CommunicationWorker(igniteInstanceName, log);

        new IgniteSpiThread(igniteInstanceName, commWorker.name(), log) {
            @Override protected void body() {
                commWorker.run();
            }
        }.start();

        // Ack start.
        if (log.isDebugEnabled())
            log.debug(startInfo());
    }

    /** {@inheritDoc} } */
    @Override public void onContextInitialized0(IgniteSpiContext spiCtx) throws IgniteSpiException {
        spiCtx.registerPort(boundTcpPort, IgnitePortProtocol.TCP);

        // SPI can start without shmem port.
        if (boundTcpShmemPort > 0)
            spiCtx.registerPort(boundTcpShmemPort, IgnitePortProtocol.TCP);

        spiCtx.addLocalEventListener(discoLsnr, EVT_NODE_LEFT, EVT_NODE_FAILED);

        ctxInitLatch.countDown();

        metricsLsnr = new TcpCommunicationMetricsListener(ignite, spiCtx);
    }

    /** {@inheritDoc} */
    @Override public IgniteSpiContext getSpiContext() {
        if (ctxInitLatch.getCount() > 0) {
            if (log.isDebugEnabled())
                log.debug("Waiting for context initialization.");

            try {
                U.await(ctxInitLatch);

                if (log.isDebugEnabled())
                    log.debug("Context has been initialized.");
            }
            catch (IgniteInterruptedCheckedException e) {
                U.warn(log, "Thread has been interrupted while waiting for SPI context initialization.", e);
            }
        }

        return super.getSpiContext();
    }

    /**
     * Recreates tpcSrvr socket instance.
     *
     * @return Server instance.
     * @throws IgniteCheckedException Thrown if it's not possible to create server.
     */
    private GridNioServer<Message> resetNioServer() throws IgniteCheckedException {
        if (boundTcpPort >= 0)
            throw new IgniteCheckedException("Tcp NIO server was already created on port " + boundTcpPort);

        IgniteCheckedException lastEx = null;

        // If configured TCP port is busy, find first available in range.
        int lastPort = locPortRange == 0 ? locPort : locPort + locPortRange - 1;

        for (int port = locPort; port <= lastPort; port++) {
            try {
                MessageFactory msgFactory = new MessageFactory() {
                    private MessageFactory impl;

                    @Nullable @Override public Message create(short type) {
                        if (impl == null)
                            impl = getSpiContext().messageFactory();

                        assert impl != null;

                        return impl.create(type);
                    }
                };

                GridNioMessageReaderFactory readerFactory = new GridNioMessageReaderFactory() {
                    private IgniteSpiContext context;

                    private MessageFormatter formatter;

                    @Override public MessageReader reader(GridNioSession ses, MessageFactory msgFactory)
                        throws IgniteCheckedException {
                        final IgniteSpiContext ctx = TcpCommunicationSpi.super.getSpiContext();

                        if (formatter == null || context != ctx) {
                            context = ctx;

                            formatter = context.messageFormatter();
                        }

                        assert formatter != null;

                        ConnectionKey key = ses.meta(CONN_IDX_META);

                        return key != null ? formatter.reader(key.nodeId(), msgFactory) : null;
                    }
                };

                GridNioMessageWriterFactory writerFactory = new GridNioMessageWriterFactory() {
                    private IgniteSpiContext context;

                    private MessageFormatter formatter;

                    @Override public MessageWriter writer(GridNioSession ses) throws IgniteCheckedException {
                        final IgniteSpiContext ctx = TcpCommunicationSpi.super.getSpiContext();

                        if (formatter == null || context != ctx) {
                            context = ctx;

                            formatter = context.messageFormatter();
                        }

                        assert formatter != null;

                        ConnectionKey key = ses.meta(CONN_IDX_META);

                        return key != null ? formatter.writer(key.nodeId()) : null;
                    }
                };

                GridDirectParser parser = new GridDirectParser(log.getLogger(GridDirectParser.class),
                    msgFactory,
                    readerFactory);

                IgnitePredicate<Message> skipRecoveryPred = new IgnitePredicate<Message>() {
                    @Override public boolean apply(Message msg) {
                        return msg instanceof RecoveryLastReceivedMessage;
                    }
                };

                boolean clientMode = Boolean.TRUE.equals(ignite.configuration().isClientMode());

                IgniteBiInClosure<GridNioSession, Integer> queueSizeMonitor =
                    !clientMode && slowClientQueueLimit > 0 ?
                        new CI2<GridNioSession, Integer>() {
                            @Override public void apply(GridNioSession ses, Integer qSize) {
                                checkClientQueueSize(ses, qSize);
                            }
                        } :
                        null;

                GridNioFilter[] filters;

                if (isSslEnabled()) {
                    GridNioSslFilter sslFilter =
                        new GridNioSslFilter(ignite.configuration().getSslContextFactory().create(),
                            true, ByteOrder.LITTLE_ENDIAN, log);

                    sslFilter.directMode(true);

                    sslFilter.wantClientAuth(true);
                    sslFilter.needClientAuth(true);

                    filters = new GridNioFilter[] {
                        new GridNioTracerFilter(log, tracing),
                        new GridNioCodecFilter(parser, log, true),
                        new GridConnectionBytesVerifyFilter(log),
                        sslFilter
                    };
                }
                else
                    filters = new GridNioFilter[] {
                        new GridNioTracerFilter(log, tracing),
                        new GridNioCodecFilter(parser, log, true),
                        new GridConnectionBytesVerifyFilter(log)
                    };

                GridNioServer.Builder<Message> builder = GridNioServer.<Message>builder()
                    .address(locHost)
                    .port(port)
                    .listener(srvLsnr)
                    .logger(log)
                    .selectorCount(selectorsCnt)
                    .igniteInstanceName(igniteInstanceName)
                    .serverName("tcp-comm")
                    .tcpNoDelay(tcpNoDelay)
                    .directBuffer(directBuf)
                    .byteOrder(ByteOrder.LITTLE_ENDIAN)
                    .socketSendBufferSize(sockSndBuf)
                    .socketReceiveBufferSize(sockRcvBuf)
                    .sendQueueLimit(msgQueueLimit)
                    .directMode(true)
                    .writeTimeout(sockWriteTimeout)
                    .selectorSpins(selectorSpins)
                    .filters(filters)
                    .writerFactory(writerFactory)
                    .skipRecoveryPredicate(skipRecoveryPred)
                    .messageQueueSizeListener(queueSizeMonitor)
                    .tracing(tracing)
                    .readWriteSelectorsAssign(usePairedConnections);

                if (ignite instanceof IgniteEx) {
                    IgniteEx igniteEx = (IgniteEx)ignite;

                    builder.workerListener(igniteEx.context().workersRegistry())
                        .metricRegistry(igniteEx.context().metric().registry(COMMUNICATION_METRICS_GROUP_NAME));
                }

                GridNioServer<Message> srvr = builder.build();

                boundTcpPort = port;

                // Ack Port the TCP server was bound to.
                if (log.isInfoEnabled()) {
                    log.info("Successfully bound communication NIO server to TCP port " +
                        "[port=" + boundTcpPort +
                        ", locHost=" + locHost +
                        ", selectorsCnt=" + selectorsCnt +
                        ", selectorSpins=" + srvr.selectorSpins() +
                        ", pairedConn=" + usePairedConnections + ']');
                }

                srvr.idleTimeout(idleConnTimeout);

                return srvr;
            }
            catch (IgniteCheckedException e) {
                if (X.hasCause(e, SSLException.class))
                    throw new IgniteSpiException("Failed to create SSL context. SSL factory: "
                        + ignite.configuration().getSslContextFactory() + '.', e);

                lastEx = e;

                if (log.isDebugEnabled())
                    log.debug("Failed to bind to local port (will try next port within range) [port=" + port +
                        ", locHost=" + locHost + ']');

                onException("Failed to bind to local port (will try next port within range) [port=" + port +
                    ", locHost=" + locHost + ']', e);
            }
        }

        // If free port wasn't found.
        throw new IgniteCheckedException("Failed to bind to any port within range [startPort=" + locPort +
            ", portRange=" + locPortRange + ", locHost=" + locHost + ']', lastEx);
    }

    /**
     * Creates new shared memory communication server.
     *
     * @return Server.
     * @throws IgniteCheckedException If failed.
     */
    @Nullable private IpcSharedMemoryServerEndpoint resetShmemServer() throws IgniteCheckedException {
        if (boundTcpShmemPort >= 0)
            throw new IgniteCheckedException("Shared memory server was already created on port " + boundTcpShmemPort);

        if (shmemPort == -1 || U.isWindows())
            return null;

        IgniteCheckedException lastEx = null;

        // If configured TCP port is busy, find first available in range.
        for (int port = shmemPort; port < shmemPort + locPortRange; port++) {
            try {
                IgniteConfiguration cfg = ignite.configuration();

                IpcSharedMemoryServerEndpoint srv =
                    new IpcSharedMemoryServerEndpoint(log, cfg.getNodeId(), igniteInstanceName, cfg.getWorkDirectory());

                srv.setPort(port);

                srv.omitOutOfResourcesWarning(true);

                srv.start();

                boundTcpShmemPort = port;

                // Ack Port the TCP server was bound to.
                if (log.isInfoEnabled())
                    log.info("Successfully bound shared memory communication to TCP port [port=" + boundTcpShmemPort +
                        ", locHost=" + locHost + ']');

                return srv;
            }
            catch (IgniteCheckedException e) {
                lastEx = e;

                if (log.isDebugEnabled())
                    log.debug("Failed to bind to local port (will try next port within range) [port=" + port +
                        ", locHost=" + locHost + ']');
            }
        }

        // If free port wasn't found.
        throw new IgniteCheckedException("Failed to bind shared memory communication to any port within range [startPort=" +
            locPort + ", portRange=" + locPortRange + ", locHost=" + locHost + ']', lastEx);
    }

    /** {@inheritDoc} */
    @Override public void spiStop() throws IgniteSpiException {
        assert stopping;

        unregisterMBean();

        // Stop TCP server.
        if (nioSrvr != null)
            nioSrvr.stop();

        U.cancel(commWorker);
        U.join(commWorker, log);

        U.cancel(shmemAcceptWorker);
        U.join(shmemAcceptWorker, log);

        U.cancel(shmemWorkers);
        U.join(shmemWorkers, log);

        shmemWorkers.clear();

        // Force closing on stop (safety).
        for (GridCommunicationClient[] clients0 : clients.values()) {
            for (GridCommunicationClient client : clients0) {
                if (client != null)
                    client.forceClose();
            }
        }

        // Clear resources.
        nioSrvr = null;
        commWorker = null;

        boundTcpPort = -1;

        // Ack stop.
        if (log.isDebugEnabled())
            log.debug(stopInfo());
    }

    /** {@inheritDoc} */
    @Override protected void onContextDestroyed0() {
        stopping = true;

        if (ctxInitLatch.getCount() > 0)
            // Safety.
            ctxInitLatch.countDown();

        if (connectGate != null)
            connectGate.stopped();

        getSpiContext().deregisterPorts();

        getSpiContext().removeLocalEventListener(discoLsnr);
    }

    /** {@inheritDoc} */
    @Override public void onClientDisconnected(IgniteFuture<?> reconnectFut) {
        connectGate.disconnected(reconnectFut);

        for (GridCommunicationClient[] clients0 : clients.values()) {
            for (GridCommunicationClient client : clients0) {
                if (client != null)
                    client.forceClose();
            }
        }

        IgniteClientDisconnectedCheckedException err = new IgniteClientDisconnectedCheckedException(reconnectFut,
            "Failed to connect client node disconnected.");

        for (GridFutureAdapter<GridCommunicationClient> clientFut : clientFuts.values())
            clientFut.onDone(err);

        recoveryDescs.clear();
        inRecDescs.clear();
        outRecDescs.clear();
    }

    /** {@inheritDoc} */
    @Override public void onClientReconnected(boolean clusterRestarted) {
        connectGate.reconnected();
    }

    /**
     * @param consistentId Consistent id of the node.
     * @param nodeId Left node ID.
     */
    void onNodeLeft(Object consistentId, UUID nodeId) {
        assert nodeId != null;

        metricsLsnr.onNodeLeft(consistentId);

        GridCommunicationClient[] clients0 = clients.remove(nodeId);

        if (clients0 != null) {
            for (GridCommunicationClient client : clients0) {
                if (client != null) {
                    if (log.isDebugEnabled())
                        log.debug("Forcing NIO client close since node has left [nodeId=" + nodeId +
                            ", client=" + client + ']');

                    client.forceClose();
                }
            }
        }
    }

    /** {@inheritDoc} */
    @Override protected void checkConfigurationConsistency0(IgniteSpiContext spiCtx, ClusterNode node, boolean starting)
        throws IgniteSpiException {
        // These attributes are set on node startup in any case, so we MUST receive them.
        checkAttributePresence(node, createSpiAttributeName(ATTR_ADDRS));
        checkAttributePresence(node, createSpiAttributeName(ATTR_HOST_NAMES));
        checkAttributePresence(node, createSpiAttributeName(ATTR_PORT));
    }

    /**
     * Checks that node has specified attribute and prints warning if it does not.
     *
     * @param node Node to check.
     * @param attrName Name of the attribute.
     */
    private void checkAttributePresence(ClusterNode node, String attrName) {
        if (node.attribute(attrName) == null)
            U.warn(log, "Remote node has inconsistent configuration (required attribute was not found) " +
                "[attrName=" + attrName + ", nodeId=" + node.id() +
                "spiCls=" + U.getSimpleName(TcpCommunicationSpi.class) + ']');
    }

    /** {@inheritDoc} */
    @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException {
        sendMessage0(node, msg, null);
    }

    /**
     * @param nodes Nodes to check connection with.
     * @return Result future (each bit in result BitSet contains connection status to corresponding node).
     */
    public IgniteFuture<BitSet> checkConnection(List<ClusterNode> nodes) {
        TcpCommunicationConnectionCheckFuture fut = new TcpCommunicationConnectionCheckFuture(
            this,
            log.getLogger(TcpCommunicationConnectionCheckFuture.class),
            nioSrvr,
            nodes);

        long timeout = failureDetectionTimeoutEnabled() ? failureDetectionTimeout() : connTimeout;

        if (log.isInfoEnabled())
            log.info("Start check connection process [nodeCnt=" + nodes.size() + ", timeout=" + timeout + ']');

        fut.init(timeout);

        return new IgniteFutureImpl<>(fut);
    }

    /**
     * Sends given message to destination node. Note that characteristics of the
     * exchange such as durability, guaranteed delivery or error notification is
     * dependant on SPI implementation.
     *
     * @param node Destination node.
     * @param msg Message to send.
     * @param ackC Ack closure.
     * @throws org.apache.ignite.spi.IgniteSpiException Thrown in case of any error during sending the message.
     *      Note that this is not guaranteed that failed communication will result
     *      in thrown exception as this is dependant on SPI implementation.
     */
    public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
        throws IgniteSpiException {
        sendMessage0(node, msg, ackC);
    }

    /**
     * @param node Destination node.
     * @param msg Message to send.
     * @param ackC Ack closure.
     * @throws org.apache.ignite.spi.IgniteSpiException Thrown in case of any error during sending the message.
     *      Note that this is not guaranteed that failed communication will result
     *      in thrown exception as this is dependant on SPI implementation.
     */
    private void sendMessage0(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
        throws IgniteSpiException {
        assert node != null;
        assert msg != null;

        if (log.isTraceEnabled())
            log.trace("Sending message with ack to node [node=" + node + ", msg=" + msg + ']');

        if (isLocalNodeDisconnected()) {
            throw new IgniteSpiException("Failed to send a message to remote node because local node has " +
                "been disconnected [rmtNodeId=" + node.id() + ']');
        }

        ClusterNode locNode = getLocalNode();

        if (locNode == null)
            throw new IgniteSpiException("Local node has not been started or fully initialized " +
                "[isStopping=" + getSpiContext().isStopping() + ']');

        if (node.id().equals(locNode.id()))
            notifyListener(node.id(), msg, NOOP);
        else {
            GridCommunicationClient client = null;

            int connIdx = connPlc.connectionIndex();

            try {
                boolean retry;

                do {
                    client = reserveClient(node, connIdx);

                    UUID nodeId = null;

                    if (!client.async())
                        nodeId = node.id();

                    retry = client.sendMessage(nodeId, msg, ackC);

                    client.release();

                    if (retry) {
                        removeNodeClient(node.id(), client);

                        ClusterNode node0 = getSpiContext().node(node.id());

                        if (node0 == null)
                            throw new ClusterTopologyCheckedException("Failed to send message to remote node " +
                                "(node has left the grid): " + node.id());
                    }

                    client = null;
                }
                while (retry);
            }
            catch (Throwable t) {
                if (stopping)
                    throw new IgniteSpiException("Node is stopping.", t);

                log.error("Failed to send message to remote node [node=" + node + ", msg=" + msg + ']', t);

                if (t instanceof Error)
                    throw (Error)t;

                if (t instanceof RuntimeException)
                    throw (RuntimeException)t;

                throw new IgniteSpiException("Failed to send message to remote node: " + node, t);
            }
            finally {
                if (client != null && removeNodeClient(node.id(), client))
                    client.forceClose();
            }
        }
    }

    /**
     * @return {@code True} if local node in disconnected state.
     */
    private boolean isLocalNodeDisconnected() {
        boolean disconnected = false;

        if (ignite instanceof IgniteKernal)
            disconnected = ((IgniteKernal)ignite).context().clientDisconnected();

        return disconnected;
    }

    /**
     * @param nodeId Node ID.
     * @param rmvClient Client to remove.
     * @return {@code True} if client was removed.
     */
    private boolean removeNodeClient(UUID nodeId, GridCommunicationClient rmvClient) {
        for (;;) {
            GridCommunicationClient[] curClients = clients.get(nodeId);

            if (curClients == null || rmvClient.connectionIndex() >= curClients.length || curClients[rmvClient.connectionIndex()] != rmvClient)
                return false;

            GridCommunicationClient[] newClients = Arrays.copyOf(curClients, curClients.length);

            newClients[rmvClient.connectionIndex()] = null;

            if (clients.replace(nodeId, curClients, newClients))
                return true;
        }
    }

    /**
     * @param node Node.
     * @param connIdx Connection index.
     * @param addClient Client to add.
     */
    private void addNodeClient(ClusterNode node, int connIdx, GridCommunicationClient addClient) {
        assert connectionsPerNode > 0 : connectionsPerNode;
        assert connIdx == addClient.connectionIndex() : addClient;

        if (connIdx >= connectionsPerNode) {
            assert !usePairedConnections(node);

            return;
        }

        for (;;) {
            GridCommunicationClient[] curClients = clients.get(node.id());

            assert curClients == null || curClients[connIdx] == null : "Client already created [node=" + node.id() +
                ", connIdx=" + connIdx +
                ", client=" + addClient +
                ", oldClient=" + curClients[connIdx] + ']';

            GridCommunicationClient[] newClients;

            if (curClients == null) {
                newClients = new GridCommunicationClient[connectionsPerNode];
                newClients[connIdx] = addClient;

                if (clients.putIfAbsent(node.id(), newClients) == null)
                    break;
            }
            else {
                newClients = Arrays.copyOf(curClients, curClients.length);
                newClients[connIdx] = addClient;

                if (clients.replace(node.id(), curClients, newClients))
                    break;
            }
        }
    }

    /**
     * Returns existing or just created client to node.
     *
     * @param node Node to which client should be open.
     * @param connIdx Connection index.
     * @return The existing or just created client.
     * @throws IgniteCheckedException Thrown if any exception occurs.
     */
    private GridCommunicationClient reserveClient(ClusterNode node, int connIdx) throws IgniteCheckedException {
        assert node != null;
        assert (connIdx >= 0 && connIdx < connectionsPerNode) || !usePairedConnections(node) : connIdx;

        UUID nodeId = node.id();

        while (true) {
            GridCommunicationClient[] curClients = clients.get(nodeId);

            GridCommunicationClient client = curClients != null && connIdx < curClients.length ?
                curClients[connIdx] : null;

            if (client == null) {
                if (stopping)
                    throw new IgniteSpiException("Node is stopping.");

                // Do not allow concurrent connects.
                GridFutureAdapter<GridCommunicationClient> fut = new ConnectFuture();

                ConnectionKey connKey = new ConnectionKey(nodeId, connIdx, -1);

                GridFutureAdapter<GridCommunicationClient> oldFut = clientFuts.putIfAbsent(connKey, fut);

                if (oldFut == null) {
                    try {
                        GridCommunicationClient[] curClients0 = clients.get(nodeId);

                        GridCommunicationClient client0 = curClients0 != null && connIdx < curClients0.length ?
                            curClients0[connIdx] : null;

                        if (client0 == null) {
                            client0 = createCommunicationClient(node, connIdx);

                            if (client0 != null) {
                                addNodeClient(node, connIdx, client0);

                                if (client0 instanceof GridTcpNioCommunicationClient) {
                                    GridTcpNioCommunicationClient tcpClient = ((GridTcpNioCommunicationClient)client0);

                                    if (tcpClient.session().closeTime() > 0 && removeNodeClient(nodeId, client0)) {
                                        if (log.isDebugEnabled())
                                            log.debug("Session was closed after client creation, will retry " +
                                                "[node=" + node + ", client=" + client0 + ']');

                                        client0 = null;
                                    }
                                }
                            }
                            else {
                                U.sleep(200);

                                if (getSpiContext().node(node.id()) == null)
                                    throw new ClusterTopologyCheckedException("Failed to send message " +
                                        "(node left topology): " + node);
                            }
                        }

                        fut.onDone(client0);
                    }
                    catch (Throwable e) {
                        fut.onDone(e);

                        if (e instanceof IgniteTooManyOpenFilesException)
                            throw e;

                        if (e instanceof Error)
                            throw (Error)e;
                    }
                    finally {
                        clientFuts.remove(connKey, fut);
                    }
                }
                else
                    fut = oldFut;

                client = fut.get();

                if (client == null) {
                    if (isLocalNodeDisconnected())
                        throw new IgniteCheckedException("Unable to create TCP client due to local node disconnecting.");
                    else
                        continue;
                }

                if (getSpiContext().node(nodeId) == null) {
                    if (removeNodeClient(nodeId, client))
                        client.forceClose();

                    throw new IgniteSpiException("Destination node is not in topology: " + node.id());
                }
            }

            assert connIdx == client.connectionIndex() : client;

            if (client.reserve())
                return client;
            else
                // Client has just been closed by idle worker. Help it and try again.
                removeNodeClient(nodeId, client);
        }
    }

    /**
     * @param node Node to create client for.
     * @param connIdx Connection index.
     * @return Client.
     * @throws IgniteCheckedException If failed.
     */
    @Nullable private GridCommunicationClient createCommunicationClient(ClusterNode node, int connIdx)
        throws IgniteCheckedException {
        assert node != null;

        Integer shmemPort = node.attribute(createSpiAttributeName(ATTR_SHMEM_PORT));

        ClusterNode locNode = getSpiContext().localNode();

        if (locNode == null)
            throw new IgniteCheckedException("Failed to create NIO client (local node is stopping)");

        if (log.isDebugEnabled())
            log.debug("Creating NIO client to node: " + node);

        // If remote node has shared memory server enabled and has the same set of MACs
        // then we are likely to run on the same host and shared memory communication could be tried.
        if (shmemPort != null && U.sameMacs(locNode, node)) {
            try {
                // https://issues.apache.org/jira/browse/IGNITE-11126 Rework failure detection logic.
                GridCommunicationClient client = createShmemClient(
                    node,
                    connIdx,
                    shmemPort);

                if (log.isDebugEnabled())
                    log.debug("Shmem client created: " + client);

                return client;
            }
            catch (IgniteCheckedException e) {
                if (e.hasCause(IpcOutOfSystemResourcesException.class))
                    // Has cause or is itself the IpcOutOfSystemResourcesException.
                    LT.warn(log, OUT_OF_RESOURCES_TCP_MSG);
                else if (getSpiContext().node(node.id()) != null)
                    LT.warn(log, e.getMessage());
                else if (log.isDebugEnabled())
                    log.debug("Failed to establish shared memory connection with local node (node has left): " +
                        node.id());
            }
        }

        final long start = System.currentTimeMillis();

        GridCommunicationClient client = createTcpClient(node, connIdx);

        final long time = System.currentTimeMillis() - start;

            if (time > CONNECTION_ESTABLISH_THRESHOLD_MS) {
                if (log.isInfoEnabled())
                    log.info("TCP client created [client=" + clientString(client, node) + ", duration=" + time + "ms]");
            }
            else if (log.isDebugEnabled())
                log.debug("TCP client created [client=" + clientString(client, node) + ", duration=" + time + "ms]");

        return client;
    }

    /**
     * Returns the string representation of client with protection from null client value. If the client if null,
     * string representation is built from cluster node.
     *
     * @param client communication client
     * @param node cluster node to which the client tried to establish a connection
     * @return string representation of client
     * @throws IgniteCheckedException if failed
     */
    private String clientString(GridCommunicationClient client, ClusterNode node) throws IgniteCheckedException {
        if (client == null) {
            assert node != null;

            StringJoiner joiner = new StringJoiner(", ");

            for (InetSocketAddress addr : nodeAddresses(node))
                joiner.add(addr.toString());

            return "null, node addrs=[" + joiner.toString() + "]";
        }
        else
            return client.toString();
    }

    /**
     * @param node Node.
     * @param port Port.
     * @param connIdx Connection index.
     * @return Client.
     * @throws IgniteCheckedException If failed.
     */
    @Nullable private GridCommunicationClient createShmemClient(ClusterNode node,
        int connIdx,
        Integer port) throws IgniteCheckedException {
        int attempt = 1;

        int connectAttempts = 1;

        long connTimeout0 = connTimeout;

        IgniteSpiOperationTimeoutHelper timeoutHelper = new IgniteSpiOperationTimeoutHelper(this,
            !node.isClient());

        while (true) {
            GridCommunicationClient client;

            try {
                client = new GridShmemCommunicationClient(
                    connIdx,
                    metricsLsnr.metricRegistry(),
                    port,
                    timeoutHelper.nextTimeoutChunk(connTimeout),
                    log,
                    getSpiContext().messageFormatter());
            }
            catch (IgniteCheckedException e) {
                if (timeoutHelper.checkFailureTimeoutReached(e))
                    throw e;

                // Reconnect for the second time, if connection is not established.
                if (connectAttempts < 2 && X.hasCause(e, ConnectException.class)) {
                    connectAttempts++;

                    continue;
                }

                throw e;
            }

            try {
                safeShmemHandshake(client, node.id(), timeoutHelper.nextTimeoutChunk(connTimeout0));
            }
            catch (IgniteSpiOperationTimeoutException e) {
                client.forceClose();

                if (failureDetectionTimeoutEnabled() && timeoutHelper.checkFailureTimeoutReached(e)) {
                    if (log.isDebugEnabled())
                        log.debug("Handshake timed out (failure threshold reached) [failureDetectionTimeout=" +
                            failureDetectionTimeout() + ", err=" + e.getMessage() + ", client=" + client + ']');

                    throw e;
                }

                assert !failureDetectionTimeoutEnabled();

                if (log.isDebugEnabled())
                    log.debug("Handshake timed out (will retry with increased timeout) [timeout=" + connTimeout0 +
                        ", err=" + e.getMessage() + ", client=" + client + ']');

                if (attempt == reconCnt || connTimeout0 > maxConnTimeout) {
                    if (log.isDebugEnabled())
                        log.debug("Handshake timedout (will stop attempts to perform the handshake) " +
                            "[timeout=" + connTimeout0 + ", maxConnTimeout=" + maxConnTimeout +
                            ", attempt=" + attempt + ", reconCnt=" + reconCnt +
                            ", err=" + e.getMessage() + ", client=" + client + ']');

                    throw e;
                }
                else {
                    attempt++;

                    connTimeout0 *= 2;

                    continue;
                }
            }
            catch (IgniteCheckedException | RuntimeException | Error e) {
                if (log.isDebugEnabled())
                    log.debug(
                        "Caught exception (will close client) [err=" + e.getMessage() + ", client=" + client + ']');

                client.forceClose();

                throw e;
            }

            return client;
        }
    }

    /**
     * Checks client message queue size and initiates client drop if message queue size exceeds the configured limit.
     *
     * @param ses Node communication session.
     * @param msgQueueSize Message queue size.
     */
    private void checkClientQueueSize(GridNioSession ses, int msgQueueSize) {
        if (slowClientQueueLimit > 0 && msgQueueSize > slowClientQueueLimit) {
            ConnectionKey id = ses.meta(CONN_IDX_META);

            if (id != null) {
                ClusterNode node = getSpiContext().node(id.nodeId());

                if (node != null && node.isClient()) {
                    String msg = "Client node outbound message queue size exceeded slowClientQueueLimit, " +
                        "the client will be dropped " +
                        "(consider changing 'slowClientQueueLimit' configuration property) " +
                        "[srvNode=" + getSpiContext().localNode().id() +
                        ", clientNode=" + node +
                        ", slowClientQueueLimit=" + slowClientQueueLimit + ']';

                    U.quietAndWarn(log, msg);

                    getSpiContext().failNode(id.nodeId(), msg);
                }
            }
        }
    }

    /**
     * @param node Node.
     * @return Node addresses.
     * @throws IgniteCheckedException If failed.
     */
    private Collection<InetSocketAddress> nodeAddresses(ClusterNode node) throws IgniteCheckedException {
        return nodeAddresses(node, filterReachableAddresses);
    }

    /**
     * @param node Node.
     * @param filterReachableAddresses Filter addresses flag.
     * @return Node addresses.
     * @throws IgniteCheckedException If node does not have addresses.
     */
    public Collection<InetSocketAddress> nodeAddresses(ClusterNode node, boolean filterReachableAddresses)
        throws IgniteCheckedException {
        Collection<String> rmtAddrs0 = node.attribute(createSpiAttributeName(ATTR_ADDRS));
        Collection<String> rmtHostNames0 = node.attribute(createSpiAttributeName(ATTR_HOST_NAMES));
        Integer boundPort = node.attribute(createSpiAttributeName(ATTR_PORT));
        Collection<InetSocketAddress> extAddrs = node.attribute(createSpiAttributeName(ATTR_EXT_ADDRS));

        boolean isRmtAddrsExist = (!F.isEmpty(rmtAddrs0) && boundPort != null);
        boolean isExtAddrsExist = !F.isEmpty(extAddrs);

        if (!isRmtAddrsExist && !isExtAddrsExist)
            throw new IgniteCheckedException("Failed to send message to the destination node. Node doesn't have any " +
                "TCP communication addresses or mapped external addresses. Check configuration and make sure " +
                "that you use the same communication SPI on all nodes. Remote node id: " + node.id());

        LinkedHashSet<InetSocketAddress> addrs;

        // Try to connect first on bound addresses.
        if (isRmtAddrsExist) {
            List<InetSocketAddress> addrs0 = new ArrayList<>(U.toSocketAddresses(rmtAddrs0, rmtHostNames0, boundPort));

            boolean sameHost = U.sameMacs(getSpiContext().localNode(), node);

            addrs0.sort(U.inetAddressesComparator(sameHost));

            addrs = new LinkedHashSet<>(addrs0);
        }
        else
            addrs = new LinkedHashSet<>();

        // Then on mapped external addresses.
        if (isExtAddrsExist)
            addrs.addAll(extAddrs);

        if (log.isDebugEnabled())
            log.debug("Addresses resolved from attributes [rmtNode=" + node.id() + ", addrs=" + addrs +
                ", isRmtAddrsExist=" + isRmtAddrsExist + ']');

        if (filterReachableAddresses) {
            Set<InetAddress> allInetAddrs = U.newHashSet(addrs.size());

            for (InetSocketAddress addr : addrs) {
                // Skip unresolved as addr.getAddress() can return null.
                if (!addr.isUnresolved())
                    allInetAddrs.add(addr.getAddress());
            }

            List<InetAddress> reachableInetAddrs = U.filterReachable(allInetAddrs);

            if (reachableInetAddrs.size() < allInetAddrs.size()) {
                LinkedHashSet<InetSocketAddress> addrs0 = U.newLinkedHashSet(addrs.size());

                List<InetSocketAddress> unreachableInetAddr = new ArrayList<>(allInetAddrs.size() - reachableInetAddrs.size());

                for (InetSocketAddress addr : addrs) {
                    if (reachableInetAddrs.contains(addr.getAddress()))
                        addrs0.add(addr);
                    else
                        unreachableInetAddr.add(addr);
                }

                addrs0.addAll(unreachableInetAddr);

                addrs = addrs0;
            }

            if (log.isDebugEnabled())
                log.debug("Addresses to connect for node [rmtNode=" + node.id() + ", addrs=" + addrs + ']');
        }

        return addrs;
    }

    /**
     * Establish TCP connection to remote node and returns client.
     *
     * @param node Remote node.
     * @param connIdx Connection index.
     * @return Client.
     * @throws IgniteCheckedException If failed.
     */
    protected GridCommunicationClient createTcpClient(ClusterNode node, int connIdx) throws IgniteCheckedException {
        GridNioSession ses = createNioSession(node, connIdx);

        return ses == null ?
            null : new GridTcpNioCommunicationClient(connIdx, ses, log);
    }

    /**
     * Returns the established TCP/IP connection between the current node and remote server. A handshake process of
     * negotiation between two communicating nodes will be performed before the {@link GridNioSession} created.
     * <p>
     *     The handshaking process contains of these steps:
     *
     *     <ol>
     *         <li>The local node opens a new {@link SocketChannel} in the <em>blocking</em> mode.</li>
     *         <li>The local node calls {@link SocketChannel#connect(SocketAddress)} to remote node.</li>
     *         <li>The remote GridNioAcceptWorker thread accepts new connection.</li>
     *         <li>The remote node sends back the {@link NodeIdMessage}.</li>
     *         <li>The local node reads NodeIdMessage from created channel.</li>
     *         <li>The local node sends the {@link HandshakeMessage2} to remote.</li>
     *         <li>The remote node processes {@link HandshakeMessage2} in {@link GridNioServerListener#onMessage(GridNioSession, Object)}.</li>
     *         <li>The remote node sends back the {@link RecoveryLastReceivedMessage}.</li>
     *     </ol>
     *
     *     The handshaking process ends.
     * </p>
     * <p>
     *     <em>Note.</em> The {@link HandshakeTimeoutObject} is created to control execution timeout during the
     *     whole handshaking process.
     * </p>
     *
     * @param node Remote node identifier to connect with.
     * @param connIdx Connection index based on configured {@link ConnectionPolicy}.
     * @return A {@link GridNioSession} connection representation.
     * @throws IgniteCheckedException If establish connection fails.
     */
    private GridNioSession createNioSession(ClusterNode node, int connIdx) throws IgniteCheckedException {
        Collection<InetSocketAddress> addrs = nodeAddresses(node);

        GridNioSession ses = null;
        IgniteCheckedException errs = null;

        long totalTimeout;

        if (failureDetectionTimeoutEnabled())
            totalTimeout = node.isClient() ? clientFailureDetectionTimeout() : failureDetectionTimeout();
        else {
            totalTimeout = ExponentialBackoffTimeoutStrategy.totalBackoffTimeout(
                connTimeout,
                maxConnTimeout,
                reconCnt
            );
        }

        for (InetSocketAddress addr : addrs) {
            if (addr.isUnresolved())
                continue;

            TimeoutStrategy connTimeoutStgy = new ExponentialBackoffTimeoutStrategy(
                totalTimeout,
                failureDetectionTimeoutEnabled() ? DFLT_INITIAL_TIMEOUT : connTimeout,
                maxConnTimeout
            );

            while (ses == null) { // Reconnection on handshake timeout.
                if (stopping)
                    throw new IgniteSpiException("Node is stopping.");

                if (isLocalNodeAddress(addr)) {
                    if (log.isDebugEnabled())
                        log.debug("Skipping local address [addr=" + addr +
                            ", locAddrs=" + node.attribute(createSpiAttributeName(ATTR_ADDRS)) +
                            ", node=" + node + ']');
                    break;
                }

                long timeout = 0;

                connectGate.enter();

                try {
                    if (getSpiContext().node(node.id()) == null)
                        throw new ClusterTopologyCheckedException("Failed to send message (node left topology): " + node);

                    SocketChannel ch = openSocketChannel();

                    ch.configureBlocking(true);

                    ch.socket().setTcpNoDelay(tcpNoDelay);
                    ch.socket().setKeepAlive(true);

                    if (sockRcvBuf > 0)
                        ch.socket().setReceiveBufferSize(sockRcvBuf);

                    if (sockSndBuf > 0)
                        ch.socket().setSendBufferSize(sockSndBuf);

                    ConnectionKey connKey = new ConnectionKey(node.id(), connIdx, -1);

                    GridNioRecoveryDescriptor recoveryDesc = outRecoveryDescriptor(node, connKey);

                    assert recoveryDesc != null :
                        "Recovery descriptor not found [connKey=" + connKey + ", rmtNode=" + node.id() + ']';

                    if (!recoveryDesc.reserve()) {
                        U.closeQuiet(ch);

                        // Ensure the session is closed.
                        GridNioSession sesFromRecovery = recoveryDesc.session();

                        if (sesFromRecovery != null) {
                            while (sesFromRecovery.closeTime() == 0)
                                sesFromRecovery.close();
                        }

                        return null;
                    }

                    long rcvCnt;

                    Map<Integer, Object> meta = new HashMap<>();

                    GridSslMeta sslMeta = null;

                    try {
                        timeout = connTimeoutStgy.nextTimeout();

                        ch.socket().connect(addr, (int) timeout);

                        if (getSpiContext().node(node.id()) == null)
                            throw new ClusterTopologyCheckedException("Failed to send message (node left topology): " + node);

                        if (isSslEnabled()) {
                            meta.put(SSL_META.ordinal(), sslMeta = new GridSslMeta());

                            SSLEngine sslEngine = ignite.configuration().getSslContextFactory().create().createSSLEngine();

                            sslEngine.setUseClientMode(true);

                            sslMeta.sslEngine(sslEngine);
                        }

                        ClusterNode locNode = getLocalNode();

                        if (locNode == null)
                            throw new IgniteCheckedException("Local node has not been started or " +
                                "fully initialized [isStopping=" + getSpiContext().isStopping() + ']');

                        timeout = connTimeoutStgy.nextTimeout(timeout);

                        rcvCnt = safeTcpHandshake(ch,
                            node.id(),
                            timeout,
                            sslMeta,
                            new HandshakeMessage2(locNode.id(),
                                recoveryDesc.incrementConnectCount(),
                                recoveryDesc.received(),
                                connIdx));

                        if (rcvCnt == ALREADY_CONNECTED)
                            return null;
                        else if (rcvCnt == NODE_STOPPING)
                            throw new ClusterTopologyCheckedException("Remote node started stop procedure: " + node.id());
                        else if (rcvCnt == UNKNOWN_NODE)
                            throw new ClusterTopologyCheckedException("Remote node does not observe current node " +
                                "in topology : " + node.id());
                        else if (rcvCnt == NEED_WAIT) {
                            //check that failure timeout will be reached after sleep(outOfTopDelay).
                            if (connTimeoutStgy.checkTimeout(DFLT_NEED_WAIT_DELAY)) {
                                U.warn(log, "Handshake NEED_WAIT timed out (will stop attempts to perform the handshake) " +
                                    "[node=" + node.id() +
                                    ", connTimeoutStgy=" + connTimeoutStgy +
                                    ", addr=" + addr +
                                    ", failureDetectionTimeoutEnabled=" + failureDetectionTimeoutEnabled() +
                                    ", timeout=" + timeout + ']');

                                throw new ClusterTopologyCheckedException("Failed to connect to node " +
                                    "(current or target node is out of topology on target node within timeout). " +
                                        "Make sure that each ComputeTask and cache Transaction has a timeout set " +
                                        "in order to prevent parties from waiting forever in case of network issues " +
                                        "[nodeId=" + node.id() + ", addrs=" + addrs + ']');
                            }
                            else {
                                if (log.isDebugEnabled())
                                    log.debug("NEED_WAIT received, handshake after delay [node = "
                                        + node + ", outOfTopologyDelay = " + DFLT_NEED_WAIT_DELAY + "ms]");

                                U.sleep(DFLT_NEED_WAIT_DELAY);

                                continue;
                            }
                        }
                        else if (rcvCnt < 0)
                            throw new IgniteCheckedException("Unsupported negative receivedCount [rcvCnt=" + rcvCnt +
                                ", senderNode=" + node + ']');

                        recoveryDesc.onHandshake(rcvCnt);

                        meta.put(CONSISTENT_ID_META, node.consistentId());
                        meta.put(CONN_IDX_META, connKey);
                        meta.put(GridNioServer.RECOVERY_DESC_META_KEY, recoveryDesc);

                        ses = nioSrvr.createSession(ch, meta, false, null).get();
                    }
                    finally {
                        if (ses == null) {
                            U.closeQuiet(ch);

                            if (recoveryDesc != null)
                                recoveryDesc.release();
                        }
                    }
                }
                catch (IgniteSpiOperationTimeoutException e) { // Handshake is timed out.
                    if (ses != null) {
                        ses.close();

                        ses = null;
                    }

                    onException("Handshake timed out (will retry with increased timeout) [connTimeoutStrategy=" + connTimeoutStgy +
                        ", addr=" + addr + ']', e);

                    if (log.isDebugEnabled())
                        log.debug("Handshake timed out (will retry with increased timeout) [connTimeoutStrategy=" + connTimeoutStgy +
                                ", addr=" + addr + ", err=" + e + ']'
                        );

                    if (connTimeoutStgy.checkTimeout()) {
                        U.warn(log, "Handshake timed out (will stop attempts to perform the handshake) " +
                            "[node=" + node.id() + ", connTimeoutStrategy=" + connTimeoutStgy +
                            ", err=" + e.getMessage() + ", addr=" + addr +
                            ", failureDetectionTimeoutEnabled=" + failureDetectionTimeoutEnabled() +
                            ", timeout=" + timeout + ']');

                        String msg = "Failed to connect to node (is node still alive?). " +
                            "Make sure that each ComputeTask and cache Transaction has a timeout set " +
                            "in order to prevent parties from waiting forever in case of network issues " +
                            "[nodeId=" + node.id() + ", addrs=" + addrs + ']';

                        if (errs == null)
                            errs = new IgniteCheckedException(msg, e);
                        else
                            errs.addSuppressed(new IgniteCheckedException(msg, e));

                        break;
                    }
                }
                catch (ClusterTopologyCheckedException e) {
                    throw e;
                }
                catch (Exception e) {
                    // Most probably IO error on socket connect or handshake.
                    if (ses != null) {
                        ses.close();

                        ses = null;
                    }

                    onException("Client creation failed [addr=" + addr + ", err=" + e + ']', e);

                    if (log.isDebugEnabled())
                        log.debug("Client creation failed [addr=" + addr + ", err=" + e + ']');

                    if (X.hasCause(e, "Too many open files", SocketException.class))
                        throw new IgniteTooManyOpenFilesException(e);

                    // check if timeout occured in case of unrecoverable exception
                    if (connTimeoutStgy.checkTimeout()) {
                        U.warn(log, "Connection timed out (will stop attempts to perform the connect) " +
                                "[node=" + node.id() + ", connTimeoutStgy=" + connTimeoutStgy +
                                ", failureDetectionTimeoutEnabled=" + failureDetectionTimeoutEnabled() +
                                ", timeout=" + timeout +
                                ", err=" + e.getMessage() + ", addr=" + addr + ']');

                        String msg = "Failed to connect to node (is node still alive?). " +
                                "Make sure that each ComputeTask and cache Transaction has a timeout set " +
                                "in order to prevent parties from waiting forever in case of network issues " +
                                "[nodeId=" + node.id() + ", addrs=" + addrs + ']';

                        if (errs == null)
                            errs = new IgniteCheckedException(msg, e);
                        else
                            errs.addSuppressed(new IgniteCheckedException(msg, e));

                        break;
                    }

                    if (isRecoverableException(e))
                        U.sleep(DFLT_RECONNECT_DELAY);
                    else {
                        String msg = "Failed to connect to node due to unrecoverable exception (is node still alive?). " +
                                "Make sure that each ComputeTask and cache Transaction has a timeout set " +
                                "in order to prevent parties from waiting forever in case of network issues " +
                                "[nodeId=" + node.id() + ", addrs=" + addrs + ", err= " + e + ']';

                        if (errs == null)
                            errs = new IgniteCheckedException(msg, e);
                        else
                            errs.addSuppressed(new IgniteCheckedException(msg, e));

                        break;
                    }
                }
                finally {
                    connectGate.leave();
                }

                CommunicationWorker commWorker0 = commWorker;

                if (commWorker0 != null && commWorker0.runner() == Thread.currentThread())
                    commWorker0.updateHeartbeat();
            }

            if (ses != null)
                break;
        }

        if (ses == null)
            processSessionCreationError(node, addrs, errs == null ? new IgniteCheckedException("No session found") : errs);

        return ses;
    }

    /**
     * Opens a socket channel.
     *
     * @return A new socket channel.
     * @throws IOException If an I/O error occurs.
     */
    protected SocketChannel openSocketChannel() throws IOException {
        return SocketChannel.open();
    }

    /**
     * Closing connections to node.
     * NOTE: It is recommended only for tests.
     *
     * @param nodeId Node for which to close connections.
     * @throws IgniteCheckedException If occurs.
     */
    void closeConnections(UUID nodeId) throws IgniteCheckedException {
        GridCommunicationClient[] clients = this.clients.remove(nodeId);

        if (nonNull(clients)) {
            for (GridCommunicationClient client : clients)
                client.forceClose();
        }

        for (ConnectionKey connKey : clientFuts.keySet()) {
            if (!nodeId.equals(connKey.nodeId()))
                continue;

            GridFutureAdapter<GridCommunicationClient> fut = clientFuts.remove(connKey);

            if (nonNull(fut))
                fut.get().forceClose();
        }
    }

    /**
     * Check is passed  socket address belong to current node. This method should return true only if the passed
     * in address represent an address which will result in a connection to the local node.
     *
     * @param addr address to check.
     * @return true if passed address belongs to local node, otherwise false.
     */
    private boolean isLocalNodeAddress(InetSocketAddress addr) {
        return addr.getPort() == boundTcpPort
            && (locHost.equals(addr.getAddress())
                || addr.getAddress().isAnyLocalAddress()
                || (locHost.isAnyLocalAddress() && U.isLocalAddress(addr.getAddress())));
    }

    /**
     * Process errors if TCP/IP {@link GridNioSession} creation to remote node hasn't been performed.
     *
     * @param node Remote node.
     * @param addrs Remote node addresses.
     * @param errs TCP client creation errors.
     * @throws IgniteCheckedException If failed.
     */
    protected void processSessionCreationError(
        ClusterNode node,
        Collection<InetSocketAddress> addrs,
        IgniteCheckedException errs
    ) throws IgniteCheckedException {
        assert errs != null;

        boolean commErrResolve = false;

        IgniteSpiContext ctx = getSpiContext();

        if (isRecoverableException(errs) && ctx.communicationFailureResolveSupported()) {
            commErrResolve = true;

            ctx.resolveCommunicationFailure(node, errs);
        }

        if (!commErrResolve && enableForcibleNodeKill) {
            if (ctx.node(node.id()) != null
                && node.isClient()
                && !getLocalNode().isClient()
                && isRecoverableException(errs)
            ) {
                // Only server can fail client for now, as in TcpDiscovery resolveCommunicationFailure() is not supported.
                String msg = "TcpCommunicationSpi failed to establish connection to node, node will be dropped from " +
                    "cluster [" + "rmtNode=" + node + ']';

                if (enableTroubleshootingLog)
                    U.error(log, msg, errs);
                else
                    U.warn(log, msg);

                ctx.failNode(node.id(), "TcpCommunicationSpi failed to establish connection to node [" +
                    "rmtNode=" + node +
                    ", errs=" + errs +
                    ", connectErrs=" + X.getSuppressedList(errs) + ']');
            }
        }

        throw errs;
    }

    /**
     * @param errs Error.
     * @return {@code True} if error was caused by some connection IO error or IgniteCheckedException due to timeout.
     */
    private boolean isRecoverableException(Exception errs) {
        return X.hasCause(
            errs,
            IOException.class,
            HandshakeException.class,
            IgniteSpiOperationTimeoutException.class
        );
    }

    /** */
    private IgniteSpiOperationTimeoutException handshakeTimeoutException() {
        return new IgniteSpiOperationTimeoutException("Failed to perform handshake due to timeout " +
            "(consider increasing 'connectionTimeout' configuration property).");
    }

    /**
     * Performs handshake in timeout-safe way.
     *
     * @param client Client.
     * @param rmtNodeId Remote node.
     * @param timeout Timeout for handshake.
     * @throws IgniteCheckedException If handshake failed or wasn't completed withing timeout.
     */
    @SuppressWarnings("ThrowFromFinallyBlock")
    private void safeShmemHandshake(
        GridCommunicationClient client,
        UUID rmtNodeId,
        long timeout
    ) throws IgniteCheckedException {
        HandshakeTimeoutObject<GridCommunicationClient> obj = new HandshakeTimeoutObject<>(client,
            U.currentTimeMillis() + timeout);

        addTimeoutObject(obj);

        try {
            client.doHandshake(new HandshakeClosure(rmtNodeId));
        }
        finally {
            if (obj.cancel())
                removeTimeoutObject(obj);
            else
                throw handshakeTimeoutException();
        }
    }

    /**
     * Performs handshake in timeout-safe way.
     *
     * @param ch Socket channel.
     * @param rmtNodeId Remote node.
     * @param timeout Timeout for handshake.
     * @param sslMeta Session meta.
     * @param msg {@link HandshakeMessage} or {@link HandshakeMessage2} to send.
     * @return Handshake response.
     * @throws IgniteCheckedException If handshake failed or wasn't completed withing timeout.
     */
    @SuppressWarnings("ThrowFromFinallyBlock")
    private long safeTcpHandshake(
        SocketChannel ch,
        UUID rmtNodeId,
        long timeout,
        GridSslMeta sslMeta,
        HandshakeMessage msg
    ) throws IgniteCheckedException {
        HandshakeTimeoutObject obj = new HandshakeTimeoutObject<>(ch, U.currentTimeMillis() + timeout);

        addTimeoutObject(obj);

        long rcvCnt;

        try {
            BlockingSslHandler sslHnd = null;

            ByteBuffer buf;

            // Step 1. Get remote node response with the remote nodeId value.
            if (isSslEnabled()) {
                assert sslMeta != null;

                sslHnd = new BlockingSslHandler(sslMeta.sslEngine(), ch, directBuf, ByteOrder.LITTLE_ENDIAN, log);

                if (!sslHnd.handshake())
                    throw new HandshakeException("SSL handshake is not completed.");

                ByteBuffer handBuff = sslHnd.applicationBuffer();

                if (handBuff.remaining() >= DIRECT_TYPE_SIZE) {
                    short msgType = makeMessageType(handBuff.get(0), handBuff.get(1));

                    if (msgType == HANDSHAKE_WAIT_MSG_TYPE)
                        return NEED_WAIT;
                }

                if (handBuff.remaining() < NodeIdMessage.MESSAGE_FULL_SIZE) {
                    buf = ByteBuffer.allocate(1000);

                    int read = ch.read(buf);

                    if (read == -1)
                        throw new HandshakeException("Failed to read remote node ID (connection closed).");

                    buf.flip();

                    buf = sslHnd.decode(buf);

                    if (handBuff.remaining() >= DIRECT_TYPE_SIZE) {
                        short msgType = makeMessageType(handBuff.get(0), handBuff.get(1));

                        if (msgType == HANDSHAKE_WAIT_MSG_TYPE)
                            return NEED_WAIT;
                    }
                }
                else
                    buf = handBuff;
            }
            else {
                buf = ByteBuffer.allocate(NodeIdMessage.MESSAGE_FULL_SIZE);

                for (int i = 0; i < NodeIdMessage.MESSAGE_FULL_SIZE; ) {
                    int read = ch.read(buf);

                    if (read == -1)
                        throw new HandshakeException("Failed to read remote node ID (connection closed).");

                    if (read >= DIRECT_TYPE_SIZE) {
                        short msgType = makeMessageType(buf.get(0), buf.get(1));

                        if (msgType == HANDSHAKE_WAIT_MSG_TYPE)
                            return NEED_WAIT;
                    }

                    i += read;
                }
            }

            UUID rmtNodeId0 = U.bytesToUuid(buf.array(), Message.DIRECT_TYPE_SIZE);

            if (!rmtNodeId.equals(rmtNodeId0))
                throw new HandshakeException("Remote node ID is not as expected [expected=" + rmtNodeId +
                    ", rcvd=" + rmtNodeId0 + ']');
            else if (log.isDebugEnabled())
                log.debug("Received remote node ID: " + rmtNodeId0);

            if (isSslEnabled()) {
                assert sslHnd != null;

                U.writeFully(ch, sslHnd.encrypt(ByteBuffer.wrap(U.IGNITE_HEADER)));
            }
            else
                U.writeFully(ch, ByteBuffer.wrap(U.IGNITE_HEADER));

            // Step 2. Prepare Handshake message to send to the remote node.
            if (log.isDebugEnabled())
                log.debug("Writing handshake message [rmtNode=" + rmtNodeId + ", msg=" + msg + ']');

            buf = ByteBuffer.allocate(msg.getMessageSize());

            buf.order(ByteOrder.LITTLE_ENDIAN);

            boolean written = msg.writeTo(buf, null);

            assert written;

            buf.flip();

            if (isSslEnabled()) {
                assert sslHnd != null;

                U.writeFully(ch, sslHnd.encrypt(buf));
            }
            else
                U.writeFully(ch, buf);

            if (log.isDebugEnabled())
                log.debug("Waiting for handshake [rmtNode=" + rmtNodeId + ']');

            // Step 3. Waiting for response from the remote node with their receive count message.
            if (isSslEnabled()) {
                assert sslHnd != null;

                buf = ByteBuffer.allocate(1000);
                buf.order(ByteOrder.LITTLE_ENDIAN);

                ByteBuffer decode = ByteBuffer.allocate(2 * buf.capacity());
                decode.order(ByteOrder.LITTLE_ENDIAN);

                for (int i = 0; i < RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE; ) {
                    int read = ch.read(buf);

                    if (read == -1)
                        throw new HandshakeException("Failed to read remote node recovery handshake " +
                            "(connection closed).");

                    buf.flip();

                    ByteBuffer decode0 = sslHnd.decode(buf);

                    i += decode0.remaining();

                    decode = appendAndResizeIfNeeded(decode, decode0);

                    buf.clear();
                }

                decode.flip();

                rcvCnt = decode.getLong(Message.DIRECT_TYPE_SIZE);

                if (decode.limit() > RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE) {
                    decode.position(RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE);

                    sslMeta.decodedBuffer(decode);
                }

                ByteBuffer inBuf = sslHnd.inputBuffer();

                if (inBuf.position() > 0)
                    sslMeta.encodedBuffer(inBuf);
            }
            else {
                buf = ByteBuffer.allocate(RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE);

                buf.order(ByteOrder.LITTLE_ENDIAN);

                for (int i = 0; i < RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE; ) {
                    int read = ch.read(buf);

                    if (read == -1)
                        throw new HandshakeException("Failed to read remote node recovery handshake " +
                            "(connection closed).");

                    i += read;
                }

                rcvCnt = buf.getLong(Message.DIRECT_TYPE_SIZE);
            }

            if (log.isDebugEnabled())
                log.debug("Received handshake message [rmtNode=" + rmtNodeId + ", rcvCnt=" + rcvCnt + ']');

            if (rcvCnt == -1) {
                if (log.isDebugEnabled())
                    log.debug("Connection rejected, will retry client creation [rmtNode=" + rmtNodeId + ']');
            }
        }
        catch (IOException e) {
            if (log.isDebugEnabled())
                log.debug("Failed to read from channel: " + e);

            throw new IgniteCheckedException("Failed to read from channel.", e);
        }
        finally {
            if (obj.cancel())
                removeTimeoutObject(obj);
            else
                throw handshakeTimeoutException();
        }

        return rcvCnt;
    }

    /**
     * @param sndId Sender ID.
     * @param msg Communication message.
     * @param msgC Closure to call when message processing finished.
     */
    protected void notifyListener(UUID sndId, Message msg, IgniteRunnable msgC) {
        CommunicationListener<Message> lsnr = this.lsnr;

        MTC.span().addLog(() -> "Communication listeners notified");

        if (lsnr != null)
            // Notify listener of a new message.
            lsnr.onMessage(sndId, msg, msgC);
        else if (log.isDebugEnabled())
            log.debug("Received communication message without any registered listeners (will ignore, " +
                "is node stopping?) [senderNodeId=" + sndId + ", msg=" + msg + ']');
    }

    /**
     * @param nodeId The remote node id.
     * @param channel The configured channel to notify listeners with.
     * @param initMsg Channel initialization message with additional channel params.
     */
    private void notifyChannelEvtListener(UUID nodeId, Channel channel, Message initMsg) {
        if (log.isDebugEnabled())
            log.debug("Notify appropriate listeners due to a new channel opened: " + channel);

        CommunicationListener<Message> lsnr0 = lsnr;

        if (lsnr0 instanceof CommunicationListenerEx)
            ((CommunicationListenerEx<Message>)lsnr0).onChannelOpened(nodeId, initMsg, channel);
    }

    /**
     * @param target Target buffer to append to.
     * @param src Source buffer to get data.
     * @return Original or expanded buffer.
     */
    private ByteBuffer appendAndResizeIfNeeded(ByteBuffer target, ByteBuffer src) {
        if (target.remaining() < src.remaining()) {
            int newSize = Math.max(target.capacity() * 2, target.capacity() + src.remaining());

            ByteBuffer tmp = ByteBuffer.allocate(newSize);

            tmp.order(target.order());

            target.flip();

            tmp.put(target);

            target = tmp;
        }

        target.put(src);

        return target;
    }

    /**
     * Stops service threads to simulate node failure.
     *
     * FOR TEST PURPOSES ONLY!!!
     */
    public void simulateNodeFailure() {
        if (nioSrvr != null)
            nioSrvr.stop();

        if (commWorker != null)
            U.interrupt(commWorker.runner());

        U.join(commWorker, log);

        for (GridCommunicationClient[] clients0 : clients.values()) {
            for (GridCommunicationClient client : clients0) {
                if (client != null)
                    client.forceClose();
            }
        }
    }

    /**
     * @param node Node.
     * @param key Connection key.
     * @return Recovery descriptor for outgoing connection.
     */
    private GridNioRecoveryDescriptor outRecoveryDescriptor(ClusterNode node, ConnectionKey key) {
        if (usePairedConnections(node))
            return recoveryDescriptor(outRecDescs, true, node, key);
        else
            return recoveryDescriptor(recoveryDescs, false, node, key);
    }

    /**
     * @param node Node.
     * @param key Connection key.
     * @return Recovery descriptor for incoming connection.
     */
    private GridNioRecoveryDescriptor inRecoveryDescriptor(ClusterNode node, ConnectionKey key) {
        if (usePairedConnections(node))
            return recoveryDescriptor(inRecDescs, true, node, key);
        else
            return recoveryDescriptor(recoveryDescs, false, node, key);
    }

    /**
     * @param node Node.
     * @return {@code True} if can use in/out connection pair for communication.
     */
    private boolean usePairedConnections(ClusterNode node) {
        if (usePairedConnections) {
            Boolean attr = node.attribute(createSpiAttributeName(ATTR_PAIRED_CONN));

            return attr != null && attr;
        }

        return false;
    }

    /**
     * @param key The connection key to cleanup descriptors on local node.
     */
    private void cleanupLocalNodeRecoveryDescriptor(ConnectionKey key) {
        ClusterNode node = getLocalNode();

        if (usePairedConnections(node)) {
            inRecDescs.remove(key);
            outRecDescs.remove(key);
        }
        else
            recoveryDescs.remove(key);
    }

    /**
     * @param recoveryDescs Descriptors map.
     * @param pairedConnections {@code True} if in/out connections pair is used for communication with node.
     * @param node Node.
     * @param key Connection key.
     * @return Recovery receive data for given node.
     */
    private GridNioRecoveryDescriptor recoveryDescriptor(
        ConcurrentMap<ConnectionKey, GridNioRecoveryDescriptor> recoveryDescs,
        boolean pairedConnections,
        ClusterNode node,
        ConnectionKey key) {
        GridNioRecoveryDescriptor recovery = recoveryDescs.get(key);

        if (recovery == null) {
            if (log.isDebugEnabled())
                log.debug("Missing recovery descriptor for the node (will create a new one) " +
                    "[locNodeId=" + getLocalNode().id() +
                    ", key=" + key + ", rmtNode=" + node + ']');

            int maxSize = Math.max(msgQueueLimit, ackSndThreshold);

            int queueLimit = unackedMsgsBufSize != 0 ? unackedMsgsBufSize : (maxSize * 128);

            GridNioRecoveryDescriptor old = recoveryDescs.putIfAbsent(key,
                recovery = new GridNioRecoveryDescriptor(pairedConnections, queueLimit, node, log));

            if (old != null) {
                recovery = old;

                if (log.isDebugEnabled())
                    log.debug("Will use existing recovery descriptor: " + recovery);
            }
            else {
                if (log.isDebugEnabled())
                    log.debug("Initialized recovery descriptor [desc=" + recovery + ", maxSize=" + maxSize +
                        ", queueLimit=" + queueLimit + ']');
            }
        }

        return recovery;
    }

    /**
     * @param msg Error message.
     * @param e Exception.
     */
    private void onException(String msg, Exception e) {
        getExceptionRegistry().onException(msg, e);
    }

    /**
     * @return Node ID message.
     */
    private NodeIdMessage nodeIdMessage() {
        final UUID locNodeId = (ignite instanceof IgniteEx) ? ((IgniteEx)ignite).context().localNodeId() :
            safeLocalNodeId();

        return new NodeIdMessage(locNodeId);
    }

    /**
     * @return Local node ID.
     */
    private UUID safeLocalNodeId() {
        ClusterNode locNode = getLocalNode();

        UUID id;

        if (locNode == null) {
            U.warn(log, "Local node is not started or fully initialized [isStopping=" +
                getSpiContext().isStopping() + ']');

            id = new UUID(0, 0);
        }
        else
            id = locNode.id();

        return id;
    }

    /** {@inheritDoc} */
    @Override public TcpCommunicationSpi setName(String name) {
        super.setName(name);

        return this;
    }

    /**
     * Checks whether remote nodes support {@link HandshakeWaitMessage}.
     *
     * @return {@code True} if remote nodes support {@link HandshakeWaitMessage}.
     */
    private boolean isHandshakeWaitSupported() {
        DiscoverySpi discoSpi = ignite().configuration().getDiscoverySpi();

        if (discoSpi instanceof IgniteDiscoverySpi)
            return ((IgniteDiscoverySpi)discoSpi).allNodesSupport(IgniteFeatures.TCP_COMMUNICATION_SPI_HANDSHAKE_WAIT_MESSAGE);
        else {
            Collection<ClusterNode> nodes = discoSpi.getRemoteNodes();

            return IgniteFeatures.allNodesSupports(nodes, IgniteFeatures.TCP_COMMUNICATION_SPI_HANDSHAKE_WAIT_MESSAGE);
        }
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return S.toString(TcpCommunicationSpi.class, this);
    }

    /**
     * This worker takes responsibility to shut the server down when stopping,
     * No other thread shall stop passed server.
     */
    private class ShmemAcceptWorker extends GridWorker {
        /** */
        private final IpcSharedMemoryServerEndpoint srv;

        /**
         * @param srv Server.
         */
        ShmemAcceptWorker(IpcSharedMemoryServerEndpoint srv) {
            super(igniteInstanceName, "shmem-communication-acceptor", TcpCommunicationSpi.this.log);

            this.srv = srv;
        }

        /** {@inheritDoc} */
        @Override protected void body() throws InterruptedException {
            try {
                while (!Thread.interrupted()) {
                    ShmemWorker e = new ShmemWorker(srv.accept());

                    shmemWorkers.add(e);

                    new IgniteThread(e).start();
                }
            }
            catch (IgniteCheckedException e) {
                if (!isCancelled())
                    U.error(log, "Shmem server failed.", e);
            }
            finally {
                srv.close();
            }
        }

        /** {@inheritDoc} */
        @Override public void cancel() {
            super.cancel();

            srv.close();
        }
    }

    /**
     * Write message type to output stream.
     *
     * @param os Output stream.
     * @param type Message type.
     * @throws IOException On error.
     */
    private static void writeMessageType(OutputStream os, short type) throws IOException {
        os.write((byte)(type & 0xFF));
        os.write((byte)((type >> 8) & 0xFF));
    }

    /**
     * Write message type to byte buffer.
     *
     * @param buf Byte buffer.
     * @param type Message type.
     */
    public static void writeMessageType(ByteBuffer buf, short type) {
        buf.put((byte)(type & 0xFF));
        buf.put((byte)((type >> 8) & 0xFF));
    }

    /**
     * Concatenates the two parameter bytes to form a message type value.
     *
     * @param b0 The first byte.
     * @param b1 The second byte.
     */
    public static short makeMessageType(byte b0, byte b1) {
        return (short)((b1 & 0xFF) << 8 | b0 & 0xFF);
    }

    /**
     * @param ignite Ignite.
     */
    private static WorkersRegistry getWorkersRegistry(Ignite ignite) {
        return ignite instanceof IgniteEx ? ((IgniteEx)ignite).context().workersRegistry() : null;
    }

    /**
     * @param remote Destination cluster node to communicate with.
     * @param initMsg Configuration channel attributes wrapped into the message.
     * @return The future, which will be finished on channel ready.
     * @throws IgniteSpiException If fails.
     */
    public IgniteInternalFuture<Channel> openChannel(
        ClusterNode remote,
        Message initMsg
    ) throws IgniteSpiException {
        assert !remote.isLocal() : remote;
        assert initMsg != null;
        assert chConnPlc != null;
        assert nodeSupports(remote, CHANNEL_COMMUNICATION) : "Node doesn't support direct connection over socket channel " +
                "[nodeId=" + remote.id() + ']';

        ConnectionKey key = new ConnectionKey(remote.id(), chConnPlc.connectionIndex());

        GridFutureAdapter<Channel> chFut = new GridFutureAdapter<>();

        connectGate.enter();

        try {
            GridNioSession ses = createNioSession(remote, key.connectionIndex());

            assert ses != null : "Session must be established [remoteId=" + remote.id() + ", key=" + key + ']';

            cleanupLocalNodeRecoveryDescriptor(key);
            ses.addMeta(CHANNEL_FUT_META, chFut);

            // Send configuration message over the created session.
            ses.send(initMsg)
                .listen(f -> {
                    if (f.error() != null) {
                        GridFutureAdapter<Channel> rq = ses.meta(CHANNEL_FUT_META);

                        assert rq != null;

                        rq.onDone(f.error());

                        ses.close();

                        return;
                    }

                    addTimeoutObject(new IgniteSpiTimeoutObject() {
                        @Override public IgniteUuid id() {
                            return IgniteUuid.randomUuid();
                        }

                        @Override public long endTime() {
                            return U.currentTimeMillis() + connTimeout;
                        }

                        @Override public void onTimeout() {
                            // Close session if request not complete yet.
                            GridFutureAdapter<Channel> rq = ses.meta(CHANNEL_FUT_META);

                            assert rq != null;

                            if (rq.onDone(handshakeTimeoutException()))
                                ses.close();
                        }
                    });
                });

            return chFut;
        }
        catch (IgniteCheckedException e) {
            throw new IgniteSpiException("Unable to create new channel connection to the remote node: " + remote, e);
        }
        finally {
            connectGate.leave();
        }
    }

    /**
     * @param connIdx Connection index to check.
     * @return {@code true} if connection index is related to the channel create request\response.
     */
    private boolean isChannelConnIdx(int connIdx) {
        return connIdx > MAX_CONN_PER_NODE;
    }

    /**
     *
     */
    private class DiscoveryListener implements GridLocalEventListener, HighPriorityListener {
        /** {@inheritDoc} */
        @Override public void onEvent(Event evt) {
            assert evt instanceof DiscoveryEvent : evt;
            assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED;

            ClusterNode node = ((DiscoveryEvent)evt).eventNode();

            onNodeLeft(node.consistentId(), node.id());
        }

        /** {@inheritDoc} */
        @Override public int order() {
            return 0;
        }
    }

    /**
     *
     */
    private class ShmemWorker extends GridWorker {
        /** */
        private final IpcEndpoint endpoint;

        /**
         * @param endpoint Endpoint.
         */
        private ShmemWorker(IpcEndpoint endpoint) {
            super(igniteInstanceName, "shmem-worker", TcpCommunicationSpi.this.log);

            this.endpoint = endpoint;
        }

        /** {@inheritDoc} */
        @Override protected void body() throws InterruptedException {
            try {
                MessageFactory msgFactory = new MessageFactory() {
                    private MessageFactory impl;

                    @Nullable @Override public Message create(short type) {
                        if (impl == null)
                            impl = getSpiContext().messageFactory();

                        assert impl != null;

                        return impl.create(type);
                    }
                };

                GridNioMessageWriterFactory writerFactory = new GridNioMessageWriterFactory() {
                    private MessageFormatter formatter;

                    @Override public MessageWriter writer(GridNioSession ses) throws IgniteCheckedException {
                        if (formatter == null)
                            formatter = getSpiContext().messageFormatter();

                        assert formatter != null;

                        ConnectionKey connKey = ses.meta(CONN_IDX_META);

                        return connKey != null ? formatter.writer(connKey.nodeId()) : null;
                    }
                };

                GridNioMessageReaderFactory readerFactory = new GridNioMessageReaderFactory() {
                    private MessageFormatter formatter;

                    @Override public MessageReader reader(GridNioSession ses, MessageFactory msgFactory)
                        throws IgniteCheckedException {
                        if (formatter == null)
                            formatter = getSpiContext().messageFormatter();

                        assert formatter != null;

                        ConnectionKey connKey = ses.meta(CONN_IDX_META);

                        return connKey != null ? formatter.reader(connKey.nodeId(), msgFactory) : null;
                    }
                };

                IpcToNioAdapter<Message> adapter = new IpcToNioAdapter<>(
                    metricsLsnr.metricRegistry(),
                    log,
                    endpoint,
                    srvLsnr,
                    writerFactory,
                    new GridNioTracerFilter(log, tracing),
                    new GridNioCodecFilter(
                        new GridDirectParser(log.getLogger(GridDirectParser.class), msgFactory, readerFactory),
                        log,
                        true),
                    new GridConnectionBytesVerifyFilter(log)
                );

                adapter.serve();
            }
            finally {
                shmemWorkers.remove(this);

                endpoint.close();
            }
        }

        /** {@inheritDoc} */
        @Override public void cancel() {
            super.cancel();

            endpoint.close();
        }

        /** {@inheritDoc} */
        @Override protected void cleanup() {
            super.cleanup();

            endpoint.close();
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(ShmemWorker.class, this);
        }
    }

    /**
     *
     */
    private class CommunicationWorker extends GridWorker {
        /** */
        private final BlockingQueue<DisconnectedSessionInfo> q = new LinkedBlockingQueue<>();

        /**
         * @param igniteInstanceName Ignite instance name.
         * @param log Logger.
         */
        private CommunicationWorker(String igniteInstanceName, IgniteLogger log) {
            super(igniteInstanceName, "tcp-comm-worker", log, getWorkersRegistry(ignite));
        }

        /** */
        @Override protected void body() throws InterruptedException {
            if (log.isDebugEnabled())
                log.debug("Tcp communication worker has been started.");

            Throwable err = null;

            try {
                while (!isCancelled()) {
                    DisconnectedSessionInfo disconnectData;

                    blockingSectionBegin();

                    try {
                        disconnectData = q.poll(idleConnTimeout, TimeUnit.MILLISECONDS);
                    }
                    finally {
                        blockingSectionEnd();
                    }

                    if (disconnectData != null)
                        processDisconnect(disconnectData);
                    else
                        processIdle();

                    onIdle();
                }
            }
            catch (Throwable t) {
                if (!(t instanceof InterruptedException))
                    err = t;

                throw t;
            }
            finally {
                if (ignite instanceof IgniteEx) {
                    if (err == null && !stopping)
                        err = new IllegalStateException("Thread  " + getName() + " is terminated unexpectedly.");

                    if (err instanceof OutOfMemoryError)
                        ((IgniteEx)ignite).context().failure().process(new FailureContext(CRITICAL_ERROR, err));
                    else if (err != null)
                        ((IgniteEx)ignite).context().failure().process(
                            new FailureContext(SYSTEM_WORKER_TERMINATION, err));
                }
            }
        }

        /**
         *
         */
        private void processIdle() {
            cleanupRecovery();

            for (Map.Entry<UUID, GridCommunicationClient[]> e : clients.entrySet()) {
                UUID nodeId = e.getKey();

                for (GridCommunicationClient client : e.getValue()) {
                    if (client == null)
                        continue;

                    ClusterNode node = getSpiContext().node(nodeId);

                    if (node == null) {
                        if (log.isDebugEnabled())
                            log.debug("Forcing close of non-existent node connection: " + nodeId);

                        client.forceClose();

                        removeNodeClient(nodeId, client);

                        continue;
                    }

                    GridNioRecoveryDescriptor recovery = null;

                    if (!usePairedConnections(node) && client instanceof GridTcpNioCommunicationClient) {
                        recovery = recoveryDescs.get(new ConnectionKey(
                            node.id(), client.connectionIndex(), -1)
                        );

                        if (recovery != null && recovery.lastAcknowledged() != recovery.received()) {
                            RecoveryLastReceivedMessage msg = new RecoveryLastReceivedMessage(recovery.received());

                            if (log.isDebugEnabled())
                                log.debug("Send recovery acknowledgement on timeout [rmtNode=" + nodeId +
                                    ", rcvCnt=" + msg.received() + ']');

                            try {
                                nioSrvr.sendSystem(((GridTcpNioCommunicationClient)client).session(), msg);

                                recovery.lastAcknowledged(msg.received());
                            }
                            catch (IgniteCheckedException err) {
                                U.error(log, "Failed to send message: " + err, err);
                            }

                            continue;
                        }
                    }

                    long idleTime = client.getIdleTime();

                    if (idleTime >= idleConnTimeout) {
                        if (recovery == null && usePairedConnections(node))
                            recovery = outRecDescs.get(new ConnectionKey(
                                node.id(), client.connectionIndex(), -1)
                            );

                        if (recovery != null &&
                            recovery.nodeAlive(getSpiContext().node(nodeId)) &&
                            !recovery.messagesRequests().isEmpty()) {
                            if (log.isDebugEnabled())
                                log.debug("Node connection is idle, but there are unacknowledged messages, " +
                                    "will wait: " + nodeId);

                            continue;
                        }

                        if (log.isDebugEnabled())
                            log.debug("Closing idle node connection: " + nodeId);

                        if (client.close() || client.closed())
                            removeNodeClient(nodeId, client);
                    }
                }
            }

            for (GridNioSession ses : nioSrvr.sessions()) {
                GridNioRecoveryDescriptor recovery = ses.inRecoveryDescriptor();

                if (recovery != null && usePairedConnections(recovery.node())) {
                    assert ses.accepted() : ses;

                    sendAckOnTimeout(recovery, ses);
                }
            }
        }

        /**
         * @param recovery Recovery descriptor.
         * @param ses Session.
         */
        private void sendAckOnTimeout(GridNioRecoveryDescriptor recovery, GridNioSession ses) {
            if (recovery != null && recovery.lastAcknowledged() != recovery.received()) {
                RecoveryLastReceivedMessage msg = new RecoveryLastReceivedMessage(recovery.received());

                if (log.isDebugEnabled()) {
                    log.debug("Send recovery acknowledgement on timeout [rmtNode=" + recovery.node().id() +
                        ", rcvCnt=" + msg.received() +
                        ", lastAcked=" + recovery.lastAcknowledged() + ']');
                }

                try {
                    nioSrvr.sendSystem(ses, msg);

                    recovery.lastAcknowledged(msg.received());
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to send message: " + e, e);
                }
            }
        }

        /**
         *
         */
        private void cleanupRecovery() {
            cleanupRecovery(recoveryDescs);
            cleanupRecovery(inRecDescs);
            cleanupRecovery(outRecDescs);
        }

        /**
         * @param recoveryDescs Recovery descriptors to cleanup.
         */
        private void cleanupRecovery(ConcurrentMap<ConnectionKey, GridNioRecoveryDescriptor> recoveryDescs) {
            Set<ConnectionKey> left = null;

            for (Map.Entry<ConnectionKey, GridNioRecoveryDescriptor> e : recoveryDescs.entrySet()) {
                if (left != null && left.contains(e.getKey()))
                    continue;

                GridNioRecoveryDescriptor recoveryDesc = e.getValue();

                if (!recoveryDesc.nodeAlive(getSpiContext().node(e.getKey().nodeId()))) {
                    if (left == null)
                        left = new HashSet<>();

                    left.add(e.getKey());
                }
            }

            if (left != null) {
                assert !left.isEmpty();

                for (ConnectionKey id : left) {
                    GridNioRecoveryDescriptor recoveryDesc = recoveryDescs.get(id);

                    if (recoveryDesc != null && recoveryDesc.onNodeLeft())
                        recoveryDescs.remove(id, recoveryDesc);
                }
            }
        }

        /**
         * @param sesInfo Disconnected session information.
         */
        private void processDisconnect(DisconnectedSessionInfo sesInfo) {
            GridNioRecoveryDescriptor recoveryDesc = sesInfo.recoveryDesc;

            ClusterNode node = recoveryDesc.node();

            if (!recoveryDesc.nodeAlive(getSpiContext().node(node.id())))
                return;

            try {
                if (log.isDebugEnabled())
                    log.debug("Recovery reconnect [rmtNode=" + recoveryDesc.node().id() + ']');

                GridCommunicationClient client = reserveClient(node, sesInfo.connIdx);

                client.release();
            }
            catch (ClusterTopologyCheckedException e) {
                if (log.isDebugEnabled())
                    log.debug("Recovery reconnect failed, node stopping [rmtNode=" + recoveryDesc.node().id() + ']');
            }
            catch (IgniteTooManyOpenFilesException e) {
                onException(e.getMessage(), e);

                throw e;
            }
            catch (IgniteCheckedException | IgniteException e) {
                try {
                    if (recoveryDesc.nodeAlive(getSpiContext().node(node.id())) && getSpiContext().pingNode(node.id())) {
                        if (log.isDebugEnabled())
                            log.debug("Recovery reconnect failed, will retry " +
                                "[rmtNode=" + recoveryDesc.node().id() + ", err=" + e + ']');

                        addProcessDisconnectRequest(sesInfo);
                    }
                    else {
                        if (log.isDebugEnabled())
                            log.debug("Recovery reconnect failed, " +
                                "node left [rmtNode=" + recoveryDesc.node().id() + ", err=" + e + ']');

                        onException("Recovery reconnect failed, node left [rmtNode=" + recoveryDesc.node().id() + "]",
                            e);
                    }
                }
                catch (IgniteClientDisconnectedException ignored) {
                    if (log.isDebugEnabled())
                        log.debug("Failed to ping node, client disconnected.");
                }
            }
        }

        /**
         * @param sesInfo Disconnected session information.
         */
        void addProcessDisconnectRequest(DisconnectedSessionInfo sesInfo) {
            boolean add = q.add(sesInfo);

            assert add;
        }
    }

    /**
     *
     */
    private static class ConnectFuture extends GridFutureAdapter<GridCommunicationClient> {
        // No-op.
    }

    /**
     *
     */
    private static class HandshakeTimeoutObject<T> implements IgniteSpiTimeoutObject {
        /** */
        private final IgniteUuid id = IgniteUuid.randomUuid();

        /** */
        private final T obj;

        /** */
        private final long endTime;

        /** */
        private final AtomicBoolean done = new AtomicBoolean();

        /**
         * @param obj Client.
         * @param endTime End time.
         */
        private HandshakeTimeoutObject(T obj, long endTime) {
            assert obj != null;
            assert obj instanceof GridCommunicationClient || obj instanceof SelectableChannel;
            assert endTime > 0;

            this.obj = obj;
            this.endTime = endTime;
        }

        /**
         * @return {@code True} if object has not yet been timed out.
         */
        boolean cancel() {
            return done.compareAndSet(false, true);
        }

        /** {@inheritDoc} */
        @Override public void onTimeout() {
            if (done.compareAndSet(false, true)) {
                // Close socket - timeout occurred.
                if (obj instanceof GridCommunicationClient)
                    ((GridCommunicationClient)obj).forceClose();
                else
                    U.closeQuiet((AutoCloseable)obj);
            }
        }

        /** {@inheritDoc} */
        @Override public long endTime() {
            return endTime;
        }

        /** {@inheritDoc} */
        @Override public IgniteUuid id() {
            return id;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(HandshakeTimeoutObject.class, this);
        }
    }

    /**
     *
     */
    private class HandshakeClosure extends IgniteInClosure2X<InputStream, OutputStream> {
        /** */
        private static final long serialVersionUID = 0L;

        /** */
        private final UUID rmtNodeId;

        /**
         * @param rmtNodeId Remote node ID.
         */
        private HandshakeClosure(UUID rmtNodeId) {
            this.rmtNodeId = rmtNodeId;
        }

        /** {@inheritDoc} */
        @Override public void applyx(InputStream in, OutputStream out) throws IgniteCheckedException {
            try {
                // Handshake.
                byte[] b = new byte[NodeIdMessage.MESSAGE_FULL_SIZE];

                int n = 0;

                while (n < NodeIdMessage.MESSAGE_FULL_SIZE) {
                    int cnt = in.read(b, n, NodeIdMessage.MESSAGE_FULL_SIZE - n);

                    if (cnt < 0)
                        throw new IgniteCheckedException("Failed to get remote node ID (end of stream reached)");

                    n += cnt;
                }

                // First 4 bytes are for length.
                UUID id = U.bytesToUuid(b, Message.DIRECT_TYPE_SIZE);

                if (!rmtNodeId.equals(id))
                    throw new IgniteCheckedException("Remote node ID is not as expected [expected=" + rmtNodeId +
                        ", rcvd=" + id + ']');
                else if (log.isDebugEnabled())
                    log.debug("Received remote node ID: " + id);
            }
            catch (SocketTimeoutException e) {
                throw new IgniteCheckedException("Failed to perform handshake due to timeout (consider increasing " +
                    "'connectionTimeout' configuration property).", e);
            }
            catch (IOException e) {
                throw new IgniteCheckedException("Failed to perform handshake.", e);
            }

            try {
                ClusterNode localNode = getLocalNode();

                if (localNode == null)
                    throw new IgniteSpiException("Local node has not been started or fully initialized " +
                        "[isStopping=" + getSpiContext().isStopping() + ']');

                UUID id = localNode.id();

                NodeIdMessage msg = new NodeIdMessage(id);

                out.write(U.IGNITE_HEADER);
                writeMessageType(out, NODE_ID_MSG_TYPE);
                out.write(msg.nodeIdBytes());

                out.flush();

                if (log.isDebugEnabled())
                    log.debug("Sent local node ID [locNodeId=" + id + ", rmtNodeId=" + rmtNodeId + ']');
            }
            catch (IOException e) {
                throw new IgniteCheckedException("Failed to perform handshake.", e);
            }
        }
    }

    /**
     *
     */
    private static class ConnectGateway {
        /** */
        private GridSpinReadWriteLock lock = new GridSpinReadWriteLock();

        /** */
        private IgniteException err;

        /**
         *
         */
        void enter() {
            lock.readLock();

            if (err != null) {
                lock.readUnlock();

                throw err;
            }
        }

        /**
         * @return {@code True} if entered gateway.
         */
        boolean tryEnter() {
            lock.readLock();

            boolean res = err == null;

            if (!res)
                lock.readUnlock();

            return res;
        }

        /**
         *
         */
        void leave() {
            lock.readUnlock();
        }

        /**
         * @param reconnectFut Reconnect future.
         */
        void disconnected(IgniteFuture<?> reconnectFut) {
            lock.writeLock();

            err = new IgniteClientDisconnectedException(reconnectFut, "Failed to connect, client node disconnected.");

            lock.writeUnlock();
        }

        /**
         *
         */
        void reconnected() {
            lock.writeLock();

            try {
                if (err instanceof IgniteClientDisconnectedException)
                    err = null;
            }
            finally {
                lock.writeUnlock();
            }
        }

        /**
         *
         */
        void stopped() {
            lock.readLock();

            err = new IgniteException("Failed to connect, node stopped.");

            lock.readUnlock();
        }
    }

    /**
     *
     */
    private static class DisconnectedSessionInfo {
        /** */
        private final GridNioRecoveryDescriptor recoveryDesc;

        /** */
        private int connIdx;

        /**
         * @param recoveryDesc Recovery descriptor.
         * @param connIdx Connection index.
         */
        DisconnectedSessionInfo(@Nullable GridNioRecoveryDescriptor recoveryDesc, int connIdx) {
            this.recoveryDesc = recoveryDesc;
            this.connIdx = connIdx;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(DisconnectedSessionInfo.class, this);
        }
    }

    /**
     *
     */
    interface ConnectionPolicy {
        /**
         * @return Thread connection index.
         */
        int connectionIndex();
    }

    /** */
    private static class FirstConnectionPolicy implements ConnectionPolicy {
        /** {@inheritDoc} */
        @Override public int connectionIndex() {
            return 0;
        }
    }

    /** */
    private class RoundRobinConnectionPolicy implements ConnectionPolicy {
        /** {@inheritDoc} */
        @Override public int connectionIndex() {
            return (int)(U.safeAbs(Thread.currentThread().getId()) % connectionsPerNode);
        }
    }

    /**
     * MBean implementation for TcpCommunicationSpi.
     */
    private class TcpCommunicationSpiMBeanImpl extends IgniteSpiMBeanAdapter implements TcpCommunicationSpiMBean {
        /** {@inheritDoc} */
        TcpCommunicationSpiMBeanImpl(IgniteSpiAdapter spiAdapter) {
            super(spiAdapter);
        }

        /** {@inheritDoc} */
        @Override public String getLocalAddress() {
            return TcpCommunicationSpi.this.getLocalAddress();
        }

        /** {@inheritDoc} */
        @Override public int getLocalPort() {
            return TcpCommunicationSpi.this.getLocalPort();
        }

        /** {@inheritDoc} */
        @Override public int getLocalPortRange() {
            return TcpCommunicationSpi.this.getLocalPortRange();
        }

        /** {@inheritDoc} */
        @Override public boolean isUsePairedConnections() {
            return TcpCommunicationSpi.this.isUsePairedConnections();
        }

        /** {@inheritDoc} */
        @Override public int getConnectionsPerNode() {
            return TcpCommunicationSpi.this.getConnectionsPerNode();
        }

        /** {@inheritDoc} */
        @Override public int getSharedMemoryPort() {
            return TcpCommunicationSpi.this.getSharedMemoryPort();
        }

        /** {@inheritDoc} */
        @Override public long getIdleConnectionTimeout() {
            return TcpCommunicationSpi.this.getIdleConnectionTimeout();
        }

        /** {@inheritDoc} */
        @Override public long getSocketWriteTimeout() {
            return TcpCommunicationSpi.this.getSocketWriteTimeout();
        }

        /** {@inheritDoc} */
        @Override public int getAckSendThreshold() {
            return TcpCommunicationSpi.this.getAckSendThreshold();
        }

        /** {@inheritDoc} */
        @Override public int getUnacknowledgedMessagesBufferSize() {
            return TcpCommunicationSpi.this.getUnacknowledgedMessagesBufferSize();
        }

        /** {@inheritDoc} */
        @Override public long getConnectTimeout() {
            return TcpCommunicationSpi.this.getConnectTimeout();
        }

        /** {@inheritDoc} */
        @Override public long getMaxConnectTimeout() {
            return TcpCommunicationSpi.this.getMaxConnectTimeout();
        }

        /** {@inheritDoc} */
        @Override public int getReconnectCount() {
            return TcpCommunicationSpi.this.getReconnectCount();
        }

        /** {@inheritDoc} */
        @Override public boolean isDirectBuffer() {
            return TcpCommunicationSpi.this.isDirectBuffer();
        }

        /** {@inheritDoc} */
        @Override public boolean isDirectSendBuffer() {
            return TcpCommunicationSpi.this.isDirectSendBuffer();
        }

        /** {@inheritDoc} */
        @Override public int getSelectorsCount() {
            return TcpCommunicationSpi.this.getSelectorsCount();
        }

        /** {@inheritDoc} */
        @Override public long getSelectorSpins() {
            return TcpCommunicationSpi.this.getSelectorSpins();
        }

        /** {@inheritDoc} */
        @Override public boolean isTcpNoDelay() {
            return TcpCommunicationSpi.this.isTcpNoDelay();
        }

        /** {@inheritDoc} */
        @Override public int getSocketReceiveBuffer() {
            return TcpCommunicationSpi.this.getSocketReceiveBuffer();
        }

        /** {@inheritDoc} */
        @Override public int getSocketSendBuffer() {
            return TcpCommunicationSpi.this.getSocketSendBuffer();
        }

        /** {@inheritDoc} */
        @Override public int getMessageQueueLimit() {
            return TcpCommunicationSpi.this.getMessageQueueLimit();
        }

        /** {@inheritDoc} */
        @Override public int getSlowClientQueueLimit() {
            return TcpCommunicationSpi.this.getSlowClientQueueLimit();
        }

        /** {@inheritDoc} */
        @Override public void dumpStats() {
            TcpCommunicationSpi.this.dumpStats();
        }

        /** {@inheritDoc} */
        @Override public int getSentMessagesCount() {
            return TcpCommunicationSpi.this.getSentMessagesCount();
        }

        /** {@inheritDoc} */
        @Override public long getSentBytesCount() {
            return TcpCommunicationSpi.this.getSentBytesCount();
        }

        /** {@inheritDoc} */
        @Override public int getReceivedMessagesCount() {
            return TcpCommunicationSpi.this.getReceivedMessagesCount();
        }

        /** {@inheritDoc} */
        @Override public long getReceivedBytesCount() {
            return TcpCommunicationSpi.this.getReceivedBytesCount();
        }

        /** {@inheritDoc} */
        @Override public Map<String, Long> getReceivedMessagesByType() {
            return TcpCommunicationSpi.this.getReceivedMessagesByType();
        }

        /** {@inheritDoc} */
        @Override public Map<UUID, Long> getReceivedMessagesByNode() {
            return TcpCommunicationSpi.this.getReceivedMessagesByNode();
        }

        /** {@inheritDoc} */
        @Override public Map<String, Long> getSentMessagesByType() {
            return TcpCommunicationSpi.this.getSentMessagesByType();
        }

        /** {@inheritDoc} */
        @Override public Map<UUID, Long> getSentMessagesByNode() {
            return TcpCommunicationSpi.this.getSentMessagesByNode();
        }

        /** {@inheritDoc} */
        @Override public int getOutboundMessagesQueueSize() {
            return TcpCommunicationSpi.this.getOutboundMessagesQueueSize();
        }
    }
}