ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yzhda...@apache.org
Subject [39/51] [abbrv] incubator-ignite git commit: Merge branch 'sprint-1' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-24
Date Tue, 27 Jan 2015 13:46:03 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/a4d5dc63/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
index 0000000,255432c..538fa93
mode 000000,100644..100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java
@@@ -1,0 -1,2205 +1,2016 @@@
+ /*
+  * 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.internal.managers.communication;
+ 
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.events.*;
+ import org.apache.ignite.internal.*;
+ import org.apache.ignite.internal.util.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.marshaller.*;
+ import org.apache.ignite.spi.*;
+ import org.apache.ignite.spi.communication.*;
+ import org.apache.ignite.internal.managers.*;
+ import org.apache.ignite.internal.managers.deployment.*;
+ import org.apache.ignite.internal.managers.eventstorage.*;
+ import org.apache.ignite.internal.processors.timeout.*;
+ import org.apache.ignite.internal.util.direct.*;
+ import org.apache.ignite.internal.util.tostring.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.apache.ignite.internal.util.worker.*;
+ import org.jdk8.backport.*;
+ import org.jetbrains.annotations.*;
+ 
+ import java.io.*;
+ import java.util.*;
+ import java.util.Map.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+ import java.util.concurrent.locks.*;
+ 
+ import static org.apache.ignite.events.IgniteEventType.*;
+ import static org.apache.ignite.internal.GridTopic.*;
+ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
+ import static org.apache.ignite.internal.util.nio.GridNioBackPressureControl.*;
+ import static org.jdk8.backport.ConcurrentLinkedHashMap.QueuePolicy.*;
+ 
+ /**
+  * Grid communication manager.
+  */
+ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializable>> {
+     /** Max closed topics to store. */
+     public static final int MAX_CLOSED_TOPICS = 10240;
+ 
 -    /** Ordered messages comparator. */
 -    private static final Comparator<IgniteBiTuple<GridIoMessage, Long>> MSG_CMP =
 -        new Comparator<IgniteBiTuple<GridIoMessage, Long>>() {
 -            @Override public int compare(IgniteBiTuple<GridIoMessage, Long> t1, IgniteBiTuple<GridIoMessage, Long> t2) {
 -                return t1.get1().messageId() < t2.get1().messageId() ? -1 :
 -                    t1.get1().messageId() == t2.get1().messageId() ? 0 : 1;
 -            }
 -        };
 -
+     /** Listeners by topic. */
+     private final ConcurrentMap<Object, GridMessageListener> lsnrMap = new ConcurrentHashMap8<>();
+ 
+     /** Disconnect listeners. */
+     private final Collection<GridDisconnectListener> disconnectLsnrs = new ConcurrentLinkedQueue<>();
+ 
+     /** Public pool. */
+     private ExecutorService pubPool;
+ 
+     /** Internal P2P pool. */
+     private ExecutorService p2pPool;
+ 
+     /** Internal system pool. */
+     private ExecutorService sysPool;
+ 
+     /** Internal management pool. */
+     private ExecutorService mgmtPool;
+ 
+     /** Affinity assignment executor service. */
+     private ExecutorService affPool;
+ 
+     /** Utility cache pool. */
+     private ExecutorService utilityCachePool;
+ 
+     /** Discovery listener. */
+     private GridLocalEventListener discoLsnr;
+ 
+     /** */
+     private final ConcurrentMap<Object, ConcurrentMap<UUID, GridCommunicationMessageSet>> msgSetMap =
+         new ConcurrentHashMap8<>();
+ 
 -    /** Messages ID generator (per topic). */
 -    private final ConcurrentMap<Object, ConcurrentMap<UUID, AtomicLong>> msgIdMap =
 -        new ConcurrentHashMap8<>();
 -
+     /** Local node ID. */
+     private final UUID locNodeId;
+ 
+     /** Discovery delay. */
+     private final long discoDelay;
+ 
+     /** Cache for messages that were received prior to discovery. */
+     private final ConcurrentMap<UUID, ConcurrentLinkedDeque8<DelayedMessage>> waitMap =
+         new ConcurrentHashMap8<>();
+ 
+     /** Communication message listener. */
+     private CommunicationListener<Serializable> commLsnr;
+ 
+     /** Grid marshaller. */
+     private final IgniteMarshaller marsh;
+ 
+     /** Busy lock. */
+     private final GridSpinReadWriteLock busyLock = new GridSpinReadWriteLock();
+ 
+     /** Lock to sync maps access. */
+     private final ReadWriteLock lock = new ReentrantReadWriteLock();
+ 
+     /** Message cache. */
+     private ThreadLocal<IgniteBiTuple<Object, byte[]>> cacheMsg =
+         new GridThreadLocal<IgniteBiTuple<Object, byte[]>>() {
+             @Nullable @Override protected IgniteBiTuple<Object, byte[]> initialValue() {
+                 return null;
+             }
+         };
+ 
+     /** Fully started flag. When set to true, can send and receive messages. */
+     private volatile boolean started;
+ 
+     /** Closed topics. */
+     private final GridBoundedConcurrentLinkedHashSet<Object> closedTopics =
+         new GridBoundedConcurrentLinkedHashSet<>(MAX_CLOSED_TOPICS, MAX_CLOSED_TOPICS, 0.75f, 256,
+             PER_SEGMENT_Q_OPTIMIZED_RMV);
+ 
+     /** Workers count. */
+     private final LongAdder workersCnt = new LongAdder();
+ 
+     /**
+      * @param ctx Grid kernal context.
+      */
+     @SuppressWarnings("deprecation")
+     public GridIoManager(GridKernalContext ctx) {
+         super(ctx, ctx.config().getCommunicationSpi());
+ 
+         locNodeId = ctx.localNodeId();
+ 
+         discoDelay = ctx.config().getDiscoveryStartupDelay();
+ 
+         marsh = ctx.config().getMarshaller();
+     }
+ 
+     /**
+      * Resets metrics for this manager.
+      */
+     public void resetMetrics() {
+         getSpi().resetMetrics();
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("deprecation")
+     @Override public void start() throws IgniteCheckedException {
+         assertParameter(discoDelay > 0, "discoveryStartupDelay > 0");
+ 
+         startSpi();
+ 
+         pubPool = ctx.config().getExecutorService();
+         p2pPool = ctx.config().getPeerClassLoadingExecutorService();
+         sysPool = ctx.config().getSystemExecutorService();
+         mgmtPool = ctx.config().getManagementExecutorService();
+         utilityCachePool = ctx.utilityCachePool();
+         affPool = Executors.newFixedThreadPool(1);
+ 
+         getSpi().setListener(commLsnr = new CommunicationListener<Serializable>() {
+             @Override public void onMessage(UUID nodeId, Serializable msg, IgniteRunnable msgC) {
+                 try {
+                     onMessage0(nodeId, (GridIoMessage)msg, msgC);
+                 }
+                 catch (ClassCastException ignored) {
+                     U.error(log, "Communication manager received message of unknown type (will ignore): " +
+                         msg.getClass().getName() + ". Most likely GridCommunicationSpi is being used directly, " +
+                         "which is illegal - make sure to send messages only via GridProjection API.");
+                 }
+             }
+ 
+             @Override public void onDisconnected(UUID nodeId) {
+                 for (GridDisconnectListener lsnr : disconnectLsnrs)
+                     lsnr.onNodeDisconnected(nodeId);
+             }
+         });
+ 
+         if (log.isDebugEnabled())
+             log.debug(startInfo());
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings({"deprecation", "SynchronizationOnLocalVariableOrMethodParameter"})
+     @Override public void onKernalStart0() throws IgniteCheckedException {
+         discoLsnr = new GridLocalEventListener() {
+             @SuppressWarnings({"TooBroadScope", "fallthrough"})
+             @Override public void onEvent(IgniteEvent evt) {
+                 assert evt instanceof IgniteDiscoveryEvent : "Invalid event: " + evt;
+ 
+                 IgniteDiscoveryEvent discoEvt = (IgniteDiscoveryEvent)evt;
+ 
+                 UUID nodeId = discoEvt.eventNode().id();
+ 
+                 switch (evt.type()) {
+                     case EVT_NODE_JOINED:
 -                        ConcurrentLinkedDeque8<DelayedMessage> delayedMsgs = null;
 -
 -                        lock.writeLock().lock();
 -
 -                        try {
 -                            if (started)
 -                                delayedMsgs = waitMap.remove(nodeId);
 -                        }
 -                        finally {
 -                            lock.writeLock().unlock();
 -                        }
 -
 -                        if (log.isDebugEnabled())
 -                            log.debug("Processing messages from discovery startup delay list " +
 -                                "(sender node joined topology): " + delayedMsgs);
 -
 -                        // After write lock released.
 -                        if (delayedMsgs != null)
 -                            for (DelayedMessage msg : delayedMsgs)
 -                                commLsnr.onMessage(msg.nodeId(), msg.message(), msg.callback());
++                        assert waitMap.get(nodeId) == null; // We can't receive messages from undiscovered nodes.
+ 
+                         break;
+ 
+                     case EVT_NODE_LEFT:
+                     case EVT_NODE_FAILED:
+                         for (Map.Entry<Object, ConcurrentMap<UUID, GridCommunicationMessageSet>> e :
+                             msgSetMap.entrySet()) {
+                             ConcurrentMap<UUID, GridCommunicationMessageSet> map = e.getValue();
+ 
+                             GridCommunicationMessageSet set;
+ 
+                             boolean empty;
+ 
+                             synchronized (map) {
+                                 set = map.remove(nodeId);
+ 
+                                 empty = map.isEmpty();
+                             }
+ 
+                             if (set != null) {
+                                 if (log.isDebugEnabled())
+                                     log.debug("Removed message set due to node leaving grid: " + set);
+ 
+                                 // Unregister timeout listener.
+                                 ctx.timeout().removeTimeoutObject(set);
+ 
+                                 // Node may still send stale messages for this topic
+                                 // even after discovery notification is done.
+                                 closedTopics.add(set.topic());
+                             }
+ 
+                             if (empty)
+                                 msgSetMap.remove(e.getKey(), map);
+                         }
+ 
+                         // Clean up delayed and ordered messages (need exclusive lock).
+                         lock.writeLock().lock();
+ 
+                         try {
+                             ConcurrentLinkedDeque8<DelayedMessage> waitList = waitMap.remove(nodeId);
+ 
+                             if (log.isDebugEnabled())
+                                 log.debug("Removed messages from discovery startup delay list " +
+                                     "(sender node left topology): " + waitList);
+                         }
+                         finally {
+                             lock.writeLock().unlock();
+                         }
+ 
+                         break;
+ 
+                     default:
+                         assert false : "Unexpected event: " + evt;
+                 }
+             }
+         };
+ 
+         ctx.event().addLocalEventListener(discoLsnr, EVT_NODE_JOINED, EVT_NODE_LEFT, EVT_NODE_FAILED);
+ 
+         // Make sure that there are no stale messages due to window between communication
+         // manager start and kernal start.
+         // 1. Process wait list.
+         Collection<Collection<DelayedMessage>> delayedMsgs = new ArrayList<>();
+ 
+         lock.writeLock().lock();
+ 
+         try {
+             started = true;
+ 
+             for (Entry<UUID, ConcurrentLinkedDeque8<DelayedMessage>> e : waitMap.entrySet()) {
+                 if (ctx.discovery().node(e.getKey()) != null) {
+                     ConcurrentLinkedDeque8<DelayedMessage> waitList = waitMap.remove(e.getKey());
+ 
+                     if (log.isDebugEnabled())
+                         log.debug("Processing messages from discovery startup delay list: " + waitList);
+ 
+                     if (waitList != null)
+                         delayedMsgs.add(waitList);
+                 }
+             }
+         }
+         finally {
+             lock.writeLock().unlock();
+         }
+ 
+         // After write lock released.
+         if (!delayedMsgs.isEmpty()) {
+             for (Collection<DelayedMessage> col : delayedMsgs)
+                 for (DelayedMessage msg : col)
+                     commLsnr.onMessage(msg.nodeId(), msg.message(), msg.callback());
+         }
+ 
+         // 2. Process messages sets.
+         for (Map.Entry<Object, ConcurrentMap<UUID, GridCommunicationMessageSet>> e : msgSetMap.entrySet()) {
+             ConcurrentMap<UUID, GridCommunicationMessageSet> map = e.getValue();
+ 
+             for (GridCommunicationMessageSet set : map.values()) {
+                 if (ctx.discovery().node(set.nodeId()) == null) {
+                     // All map modifications should be synced for consistency.
+                     boolean rmv;
+ 
+                     synchronized (map) {
+                         rmv = map.remove(set.nodeId(), set);
+                     }
+ 
+                     if (rmv) {
+                         if (log.isDebugEnabled())
+                             log.debug("Removed message set due to node leaving grid: " + set);
+ 
+                         // Unregister timeout listener.
+                         ctx.timeout().removeTimeoutObject(set);
+                     }
+ 
+                 }
+             }
+ 
+             boolean rmv;
+ 
+             synchronized (map) {
+                 rmv = map.isEmpty();
+             }
+ 
+             if (rmv) {
+                 msgSetMap.remove(e.getKey(), map);
+ 
+                 // Node may still send stale messages for this topic
+                 // even after discovery notification is done.
+                 closedTopics.add(e.getKey());
+             }
+         }
+     }
+ 
+     /** {@inheritDoc} */
+     @SuppressWarnings("BusyWait")
+     @Override public void onKernalStop0(boolean cancel) {
+         // No more communication messages.
+         getSpi().setListener(null);
+ 
+         busyLock.writeLock();
+ 
+         U.shutdownNow(getClass(), affPool, log);
+ 
+         boolean interrupted = false;
+ 
+         while (workersCnt.sum() != 0) {
+             try {
+                 Thread.sleep(200);
+             }
+             catch (InterruptedException ignored) {
+                 interrupted = true;
+             }
+         }
+ 
+         if (interrupted)
+             Thread.currentThread().interrupt();
+ 
+         GridEventStorageManager evtMgr = ctx.event();
+ 
+         if (evtMgr != null && discoLsnr != null)
+             evtMgr.removeLocalEventListener(discoLsnr);
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void stop(boolean cancel) throws IgniteCheckedException {
+         stopSpi();
+ 
+         // Clear cache.
+         cacheMsg.set(null);
+ 
+         if (log.isDebugEnabled())
+             log.debug(stopInfo());
+     }
+ 
+     /**
+      * @param nodeId Node ID.
+      * @param msg Message bytes.
+      * @param msgC Closure to call when message processing finished.
+      */
+     @SuppressWarnings("fallthrough")
+     private void onMessage0(UUID nodeId, GridIoMessage msg, IgniteRunnable msgC) {
+         assert nodeId != null;
+         assert msg != null;
+ 
+         if (!busyLock.tryReadLock()) {
+             if (log.isDebugEnabled())
+                 log.debug("Received communication message while stopping grid.");
+ 
+             return;
+         }
+ 
+         try {
+             // Check discovery.
+             ClusterNode node = ctx.discovery().node(nodeId);
+ 
+             if (node == null) {
+                 if (log.isDebugEnabled())
+                     log.debug("Ignoring message from dead node [senderId=" + nodeId + ", msg=" + msg + ']');
+ 
+                 return; // We can't receive messages from non-discovered ones.
+             }
+ 
+             if (msg.topic() == null) {
+                 int topicOrd = msg.topicOrdinal();
+ 
+                 msg.topic(topicOrd >= 0 ? GridTopic.fromOrdinal(topicOrd) : marsh.unmarshal(msg.topicBytes(), null));
+             }
+ 
+             if (!started) {
+                 lock.readLock().lock();
+ 
+                 try {
+                     if (!started) { // Sets to true in write lock, so double checking.
+                         // Received message before valid context is set to manager.
+                         if (log.isDebugEnabled())
+                             log.debug("Adding message to waiting list [senderId=" + nodeId +
+                                 ", msg=" + msg + ']');
+ 
+                         ConcurrentLinkedDeque8<DelayedMessage> list =
+                             F.addIfAbsent(waitMap, nodeId, F.<DelayedMessage>newDeque());
+ 
+                         assert list != null;
+ 
+                         list.add(new DelayedMessage(nodeId, msg, msgC));
+ 
+                         return;
+                     }
+                 }
+                 finally {
+                     lock.readLock().unlock();
+                 }
+             }
+ 
+             // If message is P2P, then process in P2P service.
+             // This is done to avoid extra waiting and potential deadlocks
+             // as thread pool may not have any available threads to give.
+             GridIoPolicy plc = msg.policy();
+ 
+             switch (plc) {
+                 case P2P_POOL: {
 -                    processP2PMessage(node, msg, msgC);
++                    processP2PMessage(nodeId, msg, msgC);
+ 
+                     break;
+                 }
+ 
+                 case PUBLIC_POOL:
+                 case SYSTEM_POOL:
+                 case MANAGEMENT_POOL:
+                 case AFFINITY_POOL:
+                 case UTILITY_CACHE_POOL: {
+                     if (msg.isOrdered())
 -                        processOrderedMessage(node, msg, plc, msgC);
++                        processOrderedMessage(nodeId, msg, plc, msgC);
+                     else
 -                        processRegularMessage(node, msg, plc, msgC);
++                        processRegularMessage(nodeId, msg, plc, msgC);
+ 
+                     break;
+                 }
+             }
+         }
+         catch (IgniteCheckedException e) {
+             U.error(log, "Failed to process message (will ignore): " + msg, e);
+         }
+         finally {
+             busyLock.readUnlock();
+         }
+     }
+ 
+     /**
+      * Gets execution pool for policy.
+      *
+      * @param plc Policy.
+      * @return Execution pool.
+      */
+     private Executor pool(GridIoPolicy plc) {
+         switch (plc) {
+             case P2P_POOL:
+                 return p2pPool;
+             case SYSTEM_POOL:
+                 return sysPool;
+             case PUBLIC_POOL:
+                 return pubPool;
+             case MANAGEMENT_POOL:
+                 return mgmtPool;
+             case AFFINITY_POOL:
+                 return affPool;
+             case UTILITY_CACHE_POOL:
+                 assert utilityCachePool != null : "Utility cache pool is not configured.";
+ 
+                 return utilityCachePool;
+ 
+             default: {
+                 assert false : "Invalid communication policy: " + plc;
+ 
+                 // Never reached.
+                 return null;
+             }
+         }
+     }
+ 
+     /**
 -     * @param msg Message bytes.
 -     * @return Policy.
 -     */
 -    private GridIoPolicy policy(byte[] msg) {
 -        GridIoPolicy plc = GridIoPolicy.fromOrdinal(msg[0]);
 -
 -        if (plc == null)
 -            throw new IllegalStateException("Failed to parse message policy: " + Arrays.toString(msg));
 -
 -        return plc;
 -    }
 -
 -    /**
 -     * @param msg Message bytes.
 -     * @return {@code True} if ordered.
 -     */
 -    private boolean ordered(byte[] msg) {
 -        return msg[1] == 1;
 -    }
 -
 -    /**
 -     * @param node Node.
++     * @param nodeId Node ID.
+      * @param msg Message.
+      * @param msgC Closure to call when message processing finished.
+      */
 -    @SuppressWarnings("deprecation")
 -    private void processP2PMessage(final ClusterNode node, final GridIoMessage msg, final IgniteRunnable msgC) {
++    private void processP2PMessage(
++        final UUID nodeId,
++        final GridIoMessage msg,
++        final IgniteRunnable msgC
++    ) {
+         workersCnt.increment();
+ 
+         Runnable c = new GridWorker(ctx.gridName(), "msg-worker", log) {
+             @Override protected void body() {
+                 try {
+                     threadProcessingMessage(true);
+ 
+                     GridMessageListener lsnr = lsnrMap.get(msg.topic());
+ 
+                     if (lsnr == null)
+                         return;
+ 
+                     Object obj = msg.message();
+ 
+                     assert obj != null;
+ 
 -                    lsnr.onMessage(node.id(), obj);
++                    lsnr.onMessage(nodeId, obj);
+                 }
+                 finally {
+                     threadProcessingMessage(false);
+ 
+                     workersCnt.decrement();
+ 
+                     msgC.run();
+                 }
+             }
+         };
+ 
+         try {
+             p2pPool.execute(c);
+         }
+         catch (RejectedExecutionException e) {
+             U.error(log, "Failed to process P2P message due to execution rejection. Increase the upper bound " +
+                 "on 'ExecutorService' provided by 'GridConfiguration.getPeerClassLoadingExecutorService()'. " +
+                 "Will attempt to process message in the listener thread instead.", e);
+ 
+             c.run();
+         }
+     }
+ 
+     /**
 -     * @param node Node.
++     * @param nodeId Node ID.
+      * @param msg Message.
+      * @param plc Execution policy.
+      * @param msgC Closure to call when message processing finished.
+      */
 -    private void processRegularMessage(final ClusterNode node, final GridIoMessage msg, GridIoPolicy plc,
 -        final IgniteRunnable msgC) {
++    private void processRegularMessage(
++        final UUID nodeId,
++        final GridIoMessage msg,
++        GridIoPolicy plc,
++        final IgniteRunnable msgC
++    ) {
+         workersCnt.increment();
+ 
+         Runnable c = new GridWorker(ctx.gridName(), "msg-worker", log) {
+             @Override protected void body() {
+                 try {
+                     threadProcessingMessage(true);
+ 
 -                    processRegularMessage0(msg, node.id());
++                    processRegularMessage0(msg, nodeId);
+                 }
+                 finally {
+                     threadProcessingMessage(false);
+ 
+                     workersCnt.decrement();
+ 
+                     msgC.run();
+                 }
+             }
+         };
+ 
+         try {
+             pool(plc).execute(c);
+         }
+         catch (RejectedExecutionException e) {
+             U.error(log, "Failed to process regular message due to execution rejection. Increase the upper bound " +
+                 "on 'ExecutorService' provided by 'GridConfiguration.getExecutorService()'. " +
+                 "Will attempt to process message in the listener thread instead.", e);
+ 
+             c.run();
+         }
+     }
+ 
+     /**
+      * @param msg Message.
+      * @param nodeId Node ID.
+      */
+     @SuppressWarnings("deprecation")
+     private void processRegularMessage0(GridIoMessage msg, UUID nodeId) {
+         GridMessageListener lsnr = lsnrMap.get(msg.topic());
+ 
+         if (lsnr == null)
+             return;
+ 
+         Object obj = msg.message();
+ 
+         assert obj != null;
+ 
+         lsnr.onMessage(nodeId, obj);
+     }
+ 
+     /**
 -     * @param node Node.
++     * @param nodeId Node ID.
+      * @param msg Ordered message.
+      * @param plc Execution policy.
 -     * @param msgC Closure to call when message processing finished.
++     * @param msgC Closure to call when message processing finished ({@code null} for sync processing).
+      */
+     @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
 -    private void processOrderedMessage(final ClusterNode node, final GridIoMessage msg, final GridIoPolicy plc,
 -        final IgniteRunnable msgC) {
++    private void processOrderedMessage(
++        final UUID nodeId,
++        final GridIoMessage msg,
++        final GridIoPolicy plc,
++        @Nullable final IgniteRunnable msgC
++    ) {
+         assert msg != null;
+ 
 -        workersCnt.increment();
 -
 -        Runnable c = new GridWorker(ctx.gridName(), "msg-worker", log) {
 -            @Override protected void body() {
 -                try {
 -                    threadProcessingMessage(true);
 -
 -                    processOrderedMessage0(msg, plc, node.id());
 -                }
 -                finally {
 -                    threadProcessingMessage(false);
 -
 -                    workersCnt.decrement();
 -
 -                    msgC.run();
 -                }
 -            }
 -        };
 -
 -        try {
 -            pool(plc).execute(c);
 -        }
 -        catch (RejectedExecutionException e) {
 -            U.error(log, "Failed to process ordered message due to execution rejection. " +
 -                "Increase the upper bound on executor service provided by corresponding " +
 -                "configuration property. Will attempt to process message in the listener " +
 -                "thread instead [msgPlc=" + plc + ']', e);
 -
 -            c.run();
 -        }
 -    }
 -
 -    /**
 -     * @param msg Message.
 -     * @param plc Policy.
 -     * @param nodeId Node ID.
 -     */
 -    @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
 -    private void processOrderedMessage0(GridIoMessage msg, GridIoPolicy plc, UUID nodeId) {
+         long timeout = msg.timeout();
+         boolean skipOnTimeout = msg.skipOnTimeout();
+ 
+         boolean isNew = false;
+ 
+         ConcurrentMap<UUID, GridCommunicationMessageSet> map;
+ 
+         GridCommunicationMessageSet set = null;
+ 
+         while (true) {
+             map = msgSetMap.get(msg.topic());
+ 
+             if (map == null) {
+                 set = new GridCommunicationMessageSet(plc, msg.topic(), nodeId, timeout, skipOnTimeout, msg);
+ 
+                 map = new ConcurrentHashMap0<>();
+ 
+                 map.put(nodeId, set);
+ 
+                 ConcurrentMap<UUID, GridCommunicationMessageSet> old = msgSetMap.putIfAbsent(
+                     msg.topic(), map);
+ 
+                 if (old != null)
+                     map = old;
+                 else {
+                     isNew = true;
+ 
+                     // Put succeeded.
+                     break;
+                 }
+             }
+ 
+             boolean rmv = false;
+ 
+             synchronized (map) {
+                 if (map.isEmpty())
+                     rmv = true;
+                 else {
+                     set = map.get(nodeId);
+ 
+                     if (set == null) {
+                         GridCommunicationMessageSet old = map.putIfAbsent(nodeId,
+                             set = new GridCommunicationMessageSet(plc, msg.topic(),
+                                 nodeId, timeout, skipOnTimeout, msg));
+ 
+                         assert old == null;
+ 
+                         isNew = true;
+ 
+                         // Put succeeded.
+                         break;
+                     }
+                 }
+             }
+ 
+             if (rmv)
+                 msgSetMap.remove(msg.topic(), map);
+             else {
+                 assert set != null;
+                 assert !isNew;
+ 
+                 set.add(msg);
+ 
+                 break;
+             }
+         }
+ 
 -        if (ctx.discovery().node(nodeId) == null) {
++        if (isNew && ctx.discovery().node(nodeId) == null) {
+             if (log.isDebugEnabled())
+                 log.debug("Message is ignored as sender has left the grid: " + msg);
+ 
+             assert map != null;
+ 
+             boolean rmv;
+ 
+             synchronized (map) {
+                 map.remove(nodeId);
+ 
+                 rmv = map.isEmpty();
+             }
+ 
+             if (rmv)
+                 msgSetMap.remove(msg.topic(), map);
+ 
+             return;
+         }
+ 
+         if (isNew && set.endTime() != Long.MAX_VALUE)
+             ctx.timeout().addTimeoutObject(set);
+ 
 -        GridMessageListener lsnr = lsnrMap.get(msg.topic());
++        if (set.reserved()) {
++            // Set is reserved which means that it is currently processed by worker thread.
++            msgC.run();
+ 
 -        if (lsnr != null)
 -            unwindMessageSet(set, lsnr, false);
 -        else if (closedTopics.contains(msg.topic())) {
 -            if (log.isDebugEnabled())
 -                log.debug("Message is ignored as it came for the closed topic: " + msg);
++            return;
++        }
+ 
 -            assert map != null;
++        final GridMessageListener lsnr = lsnrMap.get(msg.topic());
++
++        if (lsnr == null) {
++            if (closedTopics.contains(msg.topic())) {
++                if (log.isDebugEnabled())
++                    log.debug("Message is ignored as it came for the closed topic: " + msg);
++
++                assert map != null;
++
++                msgSetMap.remove(msg.topic(), map);
++            }
++            else if (log.isDebugEnabled()) {
++                // Note that we simply keep messages if listener is not
++                // registered yet, until one will be registered.
++                log.debug("Received message for unknown listener (messages will be kept until a " +
++                    "listener is registered): " + msg);
++            }
++
++            // Mark the message as processed.
++            if (msgC != null)
++                msgC.run();
+ 
 -            msgSetMap.remove(msg.topic(), map);
++            return;
+         }
 -        else if (log.isDebugEnabled()) {
 -            // Note that we simply keep messages if listener is not
 -            // registered yet, until one will be registered.
 -            log.debug("Received message for unknown listener (messages will be kept until a " +
 -                "listener is registered): " + msg);
++
++        if (msgC == null) {
++            // Message from local node can be processed in sync manner.
++            assert locNodeId.equals(nodeId);
++
++            unwindMessageSet(set, lsnr);
++
++            return;
++        }
++
++        // Set is not reserved and new worker should be submitted.
++        workersCnt.increment();
++
++        final GridCommunicationMessageSet msgSet0 = set;
++
++        Runnable c = new GridWorker(ctx.gridName(), "msg-worker", log) {
++            @Override protected void body() {
++                try {
++                    threadProcessingMessage(true);
++
++                    unwindMessageSet(msgSet0, lsnr);
++                }
++                finally {
++                    threadProcessingMessage(false);
++
++                    workersCnt.decrement();
++
++                    msgC.run();
++                }
++            }
++        };
++
++        try {
++            pool(plc).execute(c);
++        }
++        catch (RejectedExecutionException e) {
++            U.error(log, "Failed to process ordered message due to execution rejection. " +
++                "Increase the upper bound on executor service provided by corresponding " +
++                "configuration property. Will attempt to process message in the listener " +
++                "thread instead [msgPlc=" + plc + ']', e);
++
++            c.run();
+         }
+     }
+ 
+     /**
+      * @param msgSet Message set to unwind.
+      * @param lsnr Listener to notify.
 -     * @param force Whether to force unwind and drop missing
 -     *      ordered messages that are not received yet.
+      */
 -    @SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter", "deprecation"})
 -    private void unwindMessageSet(GridCommunicationMessageSet msgSet, GridMessageListener lsnr, boolean force) {
++    private void unwindMessageSet(GridCommunicationMessageSet msgSet, GridMessageListener lsnr) {
+         // Loop until message set is empty or
+         // another thread owns the reservation.
+         while (true) {
+             if (msgSet.reserve()) {
+                 try {
 -                    Collection<GridIoMessage> orderedMsgs = msgSet.unwind(force);
 -
 -                    if (!orderedMsgs.isEmpty()) {
 -                        for (GridIoMessage msg : orderedMsgs) {
 -                            Object obj = msg.message();
 -
 -                            assert obj != null;
 -
 -                            lsnr.onMessage(msgSet.nodeId(), obj);
 -                        }
 -                    }
 -                    else if (log.isDebugEnabled())
 -                        log.debug("No messages were unwound: " + msgSet);
++                    msgSet.unwind(lsnr);
+                 }
+                 finally {
+                     msgSet.release();
+                 }
+ 
+                 // Check outside of reservation block.
+                 if (!msgSet.changed()) {
+                     if (log.isDebugEnabled())
+                         log.debug("Message set has not been changed: " + msgSet);
+ 
+                     break;
+                 }
+             }
+             else {
+                 if (log.isDebugEnabled())
+                     log.debug("Another thread owns reservation: " + msgSet);
+ 
+                 return;
+             }
+         }
+     }
+ 
+     /**
+      * @param node Destination node.
+      * @param topic Topic to send the message to.
+      * @param topicOrd GridTopic enumeration ordinal.
+      * @param msg Message to send.
+      * @param plc Type of processing.
 -     * @param msgId Message ID.
++     * @param ordered Ordered flag.
+      * @param timeout Timeout.
+      * @param skipOnTimeout Whether message can be skipped on timeout.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
 -    private void send(ClusterNode node, Object topic, int topicOrd, GridTcpCommunicationMessageAdapter msg,
 -        GridIoPolicy plc, long msgId, long timeout, boolean skipOnTimeout) throws IgniteCheckedException {
++    private void send(
++        ClusterNode node,
++        Object topic,
++        int topicOrd,
++        GridTcpCommunicationMessageAdapter msg,
++        GridIoPolicy plc,
++        boolean ordered,
++        long timeout,
++        boolean skipOnTimeout
++    ) throws IgniteCheckedException {
+         assert node != null;
+         assert topic != null;
+         assert msg != null;
+         assert plc != null;
+ 
 -        GridIoMessage ioMsg = new GridIoMessage(plc, topic, topicOrd, msg, msgId, timeout, skipOnTimeout);
++        GridIoMessage ioMsg = new GridIoMessage(plc, topic, topicOrd, msg, ordered, timeout, skipOnTimeout);
+ 
+         if (locNodeId.equals(node.id())) {
+             assert plc != P2P_POOL;
+ 
+             CommunicationListener commLsnr = this.commLsnr;
+ 
+             if (commLsnr == null)
+                 throw new IgniteCheckedException("Trying to send message when grid is not fully started.");
+ 
 -            if (msgId > 0)
 -                processOrderedMessage0(ioMsg, plc, locNodeId);
++            if (ordered)
++                processOrderedMessage(locNodeId, ioMsg, plc, null);
+             else
+                 processRegularMessage0(ioMsg, locNodeId);
+         }
+         else {
+             if (topicOrd < 0)
+                 ioMsg.topicBytes(marsh.marshal(topic));
+ 
+             try {
+                 getSpi().sendMessage(node, ioMsg);
+             }
+             catch (IgniteSpiException e) {
+                 throw new IgniteCheckedException("Failed to send message (node may have left the grid or " +
+                     "TCP connection cannot be established due to firewall issues) " +
+                     "[node=" + node + ", topic=" + topic +
+                     ", msg=" + msg + ", policy=" + plc + ']', e);
+             }
+         }
+     }
+ 
+     /**
+      * @param nodeId Id of destination node.
+      * @param topic Topic to send the message to.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
+     public void send(UUID nodeId, Object topic, GridTcpCommunicationMessageAdapter msg, GridIoPolicy plc)
+         throws IgniteCheckedException {
+         ClusterNode node = ctx.discovery().node(nodeId);
+ 
+         if (node == null)
+             throw new IgniteCheckedException("Failed to send message to node (has node left grid?): " + nodeId);
+ 
+         send(node, topic, msg, plc);
+     }
+ 
+     /**
+      * @param nodeId Id of destination node.
+      * @param topic Topic to send the message to.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
+     @SuppressWarnings("TypeMayBeWeakened")
+     public void send(UUID nodeId, GridTopic topic, GridTcpCommunicationMessageAdapter msg, GridIoPolicy plc)
+         throws IgniteCheckedException {
+         ClusterNode node = ctx.discovery().node(nodeId);
+ 
+         if (node == null)
+             throw new IgniteCheckedException("Failed to send message to node (has node left grid?): " + nodeId);
+ 
 -        send(node, topic, topic.ordinal(), msg, plc, -1, 0, false);
++        send(node, topic, topic.ordinal(), msg, plc, false, 0, false);
+     }
+ 
+     /**
+      * @param node Destination node.
+      * @param topic Topic to send the message to.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
+     public void send(ClusterNode node, Object topic, GridTcpCommunicationMessageAdapter msg, GridIoPolicy plc)
+         throws IgniteCheckedException {
 -        send(node, topic, -1, msg, plc, -1, 0, false);
++        send(node, topic, -1, msg, plc, false, 0, false);
+     }
+ 
+     /**
+      * @param node Destination node.
+      * @param topic Topic to send the message to.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
+     public void send(ClusterNode node, GridTopic topic, GridTcpCommunicationMessageAdapter msg, GridIoPolicy plc)
+         throws IgniteCheckedException {
 -        send(node, topic, topic.ordinal(), msg, plc, -1, 0, false);
 -    }
 -
 -    /**
 -     * @param topic Message topic.
 -     * @param nodeId Node ID.
 -     * @return Next ordered message ID.
 -     */
 -    public long nextMessageId(Object topic, UUID nodeId) {
 -        ConcurrentMap<UUID, AtomicLong> map = msgIdMap.get(topic);
 -
 -        if (map == null) {
 -            ConcurrentMap<UUID, AtomicLong> lastMap = msgIdMap.putIfAbsent(topic,
 -                map = new ConcurrentHashMap8<>());
 -
 -            if (lastMap != null)
 -                map = lastMap;
 -        }
 -
 -        AtomicLong msgId = map.get(nodeId);
 -
 -        if (msgId == null) {
 -            AtomicLong lastMsgId = map.putIfAbsent(nodeId, msgId = new AtomicLong(0));
 -
 -            if (lastMsgId != null)
 -                msgId = lastMsgId;
 -        }
 -
 -        long id = msgId.incrementAndGet();
 -
 -        if (log.isDebugEnabled())
 -            log.debug("Got next message ID [topic=" + topic + ", nodeId=" + nodeId + ", id=" + id + ']');
 -
 -        return id;
 -    }
 -
 -    /**
 -     * @param topic Message topic.
 -     */
 -    public void removeMessageId(Object topic) {
 -        if (log.isDebugEnabled())
 -            log.debug("Remove message ID for topic: " + topic);
 -
 -        msgIdMap.remove(topic);
++        send(node, topic, topic.ordinal(), msg, plc, false, 0, false);
+     }
+ 
+     /**
+      * @param node Destination node.
+      * @param topic Topic to send the message to.
 -     * @param msgId Ordered message ID.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @param timeout Timeout to keep a message on receiving queue.
+      * @param skipOnTimeout Whether message can be skipped on timeout.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
 -    public void sendOrderedMessage(ClusterNode node, Object topic, long msgId, GridTcpCommunicationMessageAdapter msg,
 -        GridIoPolicy plc, long timeout, boolean skipOnTimeout) throws IgniteCheckedException {
++    public void sendOrderedMessage(
++        ClusterNode node,
++        Object topic,
++        GridTcpCommunicationMessageAdapter msg,
++        GridIoPolicy plc,
++        long timeout,
++        boolean skipOnTimeout
++    ) throws IgniteCheckedException {
+         assert timeout > 0 || skipOnTimeout;
+ 
 -        send(node, topic, (byte)-1, msg, plc, msgId, timeout, skipOnTimeout);
 -    }
 -
 -    /**
 -     * @param nodeId Destination node.
 -     * @param topic Topic to send the message to.
 -     * @param msgId Ordered message ID.
 -     * @param msg Message to send.
 -     * @param plc Type of processing.
 -     * @param timeout Timeout to keep a message on receiving queue.
 -     * @param skipOnTimeout Whether message can be skipped on timeout.
 -     * @throws IgniteCheckedException Thrown in case of any errors.
 -     */
 -    public void sendOrderedMessage(UUID nodeId, Object topic, long msgId, GridTcpCommunicationMessageAdapter msg,
 -        GridIoPolicy plc, long timeout, boolean skipOnTimeout) throws IgniteCheckedException {
 -        assert timeout > 0 || skipOnTimeout;
 -
 -        ClusterNode node = ctx.discovery().node(nodeId);
 -
 -        if (node == null)
 -            throw new IgniteCheckedException("Failed to send message to node (has node left grid?): " + nodeId);
 -
 -        send(node, topic, (byte)-1, msg, plc, msgId, timeout, skipOnTimeout);
++        send(node, topic, (byte)-1, msg, plc, true, timeout, skipOnTimeout);
+     }
+ 
+     /**
+      * @param nodes Destination nodes.
+      * @param topic Topic to send the message to.
 -     * @param msgId Ordered message ID.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @param timeout Timeout to keep a message on receiving queue.
+      * @param skipOnTimeout Whether message can be skipped on timeout.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
 -    public void sendOrderedMessage(Collection<? extends ClusterNode> nodes, Object topic, long msgId,
 -        GridTcpCommunicationMessageAdapter msg, GridIoPolicy plc, long timeout, boolean skipOnTimeout)
++    public void sendOrderedMessage(
++        Collection<? extends ClusterNode> nodes,
++        Object topic,
++        GridTcpCommunicationMessageAdapter msg,
++        GridIoPolicy plc,
++        long timeout,
++        boolean skipOnTimeout
++    )
+         throws IgniteCheckedException {
+         assert timeout > 0 || skipOnTimeout;
+ 
 -        send(nodes, topic, -1, msg, plc, msgId, timeout, skipOnTimeout);
++        send(nodes, topic, -1, msg, plc, true, timeout, skipOnTimeout);
+     }
+ 
+     /**
+      * @param nodes Destination nodes.
+      * @param topic Topic to send the message to.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
 -    public void send(Collection<? extends ClusterNode> nodes, Object topic, GridTcpCommunicationMessageAdapter msg,
 -        GridIoPolicy plc) throws IgniteCheckedException {
 -        send(nodes, topic, -1, msg, plc, -1, 0, false);
++    public void send(
++        Collection<? extends ClusterNode> nodes,
++        Object topic,
++        GridTcpCommunicationMessageAdapter msg,
++        GridIoPolicy plc
++    ) throws IgniteCheckedException {
++        send(nodes, topic, -1, msg, plc, false, 0, false);
+     }
+ 
+     /**
+      * @param nodes Destination nodes.
+      * @param topic Topic to send the message to.
+      * @param msg Message to send.
+      * @param plc Type of processing.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
 -    public void send(Collection<? extends ClusterNode> nodes, GridTopic topic, GridTcpCommunicationMessageAdapter msg,
 -        GridIoPolicy plc) throws IgniteCheckedException {
 -        send(nodes, topic, topic.ordinal(), msg, plc, -1, 0, false);
++    public void send(
++        Collection<? extends ClusterNode> nodes,
++        GridTopic topic,
++        GridTcpCommunicationMessageAdapter msg,
++        GridIoPolicy plc
++    ) throws IgniteCheckedException {
++        send(nodes, topic, topic.ordinal(), msg, plc, false, 0, false);
+     }
+ 
+     /**
+      * Sends a peer deployable user message.
+      *
+      * @param nodes Destination nodes.
+      * @param msg Message to send.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
+     public void sendUserMessage(Collection<? extends ClusterNode> nodes, Object msg) throws IgniteCheckedException {
+         sendUserMessage(nodes, msg, null, false, 0);
+     }
+ 
+     /**
+      * Sends a peer deployable user message.
+      *
+      * @param nodes Destination nodes.
+      * @param msg Message to send.
+      * @param topic Message topic to use.
+      * @param ordered Is message ordered?
+      * @param timeout Message timeout in milliseconds for ordered messages.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
+     @SuppressWarnings("ConstantConditions")
+     public void sendUserMessage(Collection<? extends ClusterNode> nodes, Object msg,
+         @Nullable Object topic, boolean ordered, long timeout) throws IgniteCheckedException {
+         boolean loc = nodes.size() == 1 && F.first(nodes).id().equals(locNodeId);
+ 
+         byte[] serMsg = null;
+         byte[] serTopic = null;
+ 
+         if (!loc) {
+             serMsg = marsh.marshal(msg);
+ 
+             if (topic != null)
+                 serTopic = marsh.marshal(topic);
+         }
+ 
+         GridDeployment dep = null;
+ 
+         String depClsName = null;
+ 
+         if (ctx.config().isPeerClassLoadingEnabled()) {
+             Class<?> cls0 = U.detectClass(msg);
+ 
+             if (U.isJdk(cls0) && topic != null)
+                 cls0 = U.detectClass(topic);
+ 
+             dep = ctx.deploy().deploy(cls0, U.detectClassLoader(cls0));
+ 
+             if (dep == null)
+                 throw new IgniteDeploymentException("Failed to deploy user message: " + msg);
+ 
+             depClsName = cls0.getName();
+         }
+ 
+         GridTcpCommunicationMessageAdapter ioMsg = new GridIoUserMessage(
+             msg,
+             serMsg,
+             depClsName,
+             topic,
+             serTopic,
+             dep != null ? dep.classLoaderId() : null,
+             dep != null ? dep.deployMode() : null,
+             dep != null ? dep.userVersion() : null,
+             dep != null ? dep.participants() : null);
+ 
 -        if (ordered) {
 -            long msgId = nextMessageId(TOPIC_COMM_USER, locNodeId);
 -
 -            sendOrderedMessage(nodes, TOPIC_COMM_USER, msgId, ioMsg, PUBLIC_POOL, timeout, true);
 -        }
++        if (ordered)
++            sendOrderedMessage(nodes, TOPIC_COMM_USER, ioMsg, PUBLIC_POOL, timeout, true);
+         else if (loc)
+             send(F.first(nodes), TOPIC_COMM_USER, ioMsg, PUBLIC_POOL);
+         else {
+             ClusterNode locNode = F.find(nodes, null, F.localNode(locNodeId));
+ 
+             Collection<? extends ClusterNode> rmtNodes = F.view(nodes, F.remoteNodes(locNodeId));
+ 
+             if (locNode != null)
+                 send(locNode, TOPIC_COMM_USER, ioMsg, PUBLIC_POOL);
+ 
+             if (!rmtNodes.isEmpty())
+                 send(rmtNodes, TOPIC_COMM_USER, ioMsg, PUBLIC_POOL);
+         }
+     }
+ 
+     /**
+      * @param topic Topic to subscribe to.
+      * @param p Message predicate.
+      */
+     public void addUserMessageListener(@Nullable final Object topic, @Nullable final IgniteBiPredicate<UUID, ?> p) {
+         if (p != null) {
+             try {
+                 addMessageListener(TOPIC_COMM_USER,
+                     new GridUserMessageListener(topic, (IgniteBiPredicate<UUID, Object>)p));
+             }
+             catch (IgniteCheckedException e) {
+                 throw new IgniteException(e);
+             }
+         }
+     }
+ 
+     /**
+      * @param topic Topic to unsubscribe from.
+      * @param p Message predicate.
+      */
+     public void removeUserMessageListener(@Nullable Object topic, IgniteBiPredicate<UUID, ?> p) {
+         try {
+             removeMessageListener(TOPIC_COMM_USER,
+                 new GridUserMessageListener(topic, (IgniteBiPredicate<UUID, Object>)p));
+         }
+         catch (IgniteCheckedException e) {
+             throw new IgniteException(e);
+         }
+     }
+ 
+     /**
+      * @param nodes Destination nodes.
+      * @param topic Topic to send the message to.
+      * @param topicOrd Topic ordinal value.
+      * @param msg Message to send.
+      * @param plc Type of processing.
 -     * @param msgId Message ID (for ordered messages) or -1 (for unordered messages).
++     * @param ordered Ordered flag.
+      * @param timeout Message timeout.
+      * @param skipOnTimeout Whether message can be skipped in timeout.
+      * @throws IgniteCheckedException Thrown in case of any errors.
+      */
 -    private void send(Collection<? extends ClusterNode> nodes, Object topic, int topicOrd,
 -        GridTcpCommunicationMessageAdapter msg, GridIoPolicy plc, long msgId, long timeout, boolean skipOnTimeout)
++    private void send(
++        Collection<? extends ClusterNode> nodes,
++        Object topic,
++        int topicOrd,
++        GridTcpCommunicationMessageAdapter msg,
++        GridIoPolicy plc,
++        boolean ordered,
++        long timeout,
++        boolean skipOnTimeout
++    )
+         throws IgniteCheckedException {
+         assert nodes != null;
+         assert topic != null;
+         assert msg != null;
+         assert plc != null;
+ 
 -        if (msgId < 0)
++        if (!ordered)
+             assert F.find(nodes, null, F.localNode(locNodeId)) == null :
+                 "Internal GridGain code should never call the method with local node in a node list.";
+ 
+         try {
+             // Small optimization, as communication SPIs may have lighter implementation for sending
+             // messages to one node vs. many.
+             if (!nodes.isEmpty()) {
+                 boolean first = true;
+ 
+                 for (ClusterNode node : nodes) {
+                     GridTcpCommunicationMessageAdapter msg0 = first ? msg : msg.clone();
+ 
+                     first = false;
+ 
 -                    send(node, topic, topicOrd, msg0, plc, msgId, timeout, skipOnTimeout);
++                    send(node, topic, topicOrd, msg0, plc, ordered, timeout, skipOnTimeout);
+                 }
+             }
+             else if (log.isDebugEnabled())
+                 log.debug("Failed to send message to empty nodes collection [topic=" + topic + ", msg=" +
+                     msg + ", policy=" + plc + ']');
+         }
+         catch (IgniteSpiException e) {
+             throw new IgniteCheckedException("Failed to send message (nodes may have left the grid or " +
+                 "TCP connection cannot be established due to firewall issues) " +
+                 "[nodes=" + nodes + ", topic=" + topic +
+                 ", msg=" + msg + ", policy=" + plc + ']', e);
+         }
+     }
+ 
+     /**
+      * @param topic Listener's topic.
+      * @param lsnr Listener to add.
+      */
+     @SuppressWarnings({"TypeMayBeWeakened", "deprecation"})
+     public void addMessageListener(GridTopic topic, GridMessageListener lsnr) {
+         addMessageListener((Object)topic, lsnr);
+     }
+ 
+     /**
+      * @param lsnr Listener to add.
+      */
+     public void addDisconnectListener(GridDisconnectListener lsnr) {
+         disconnectLsnrs.add(lsnr);
+     }
+ 
+     /**
+      * @param topic Listener's topic.
+      * @param lsnr Listener to add.
+      */
+     @SuppressWarnings({"deprecation", "SynchronizationOnLocalVariableOrMethodParameter"})
+     public void addMessageListener(Object topic, final GridMessageListener lsnr) {
+         assert lsnr != null;
+         assert topic != null;
+ 
+         // Make sure that new topic is not in the list of closed topics.
+         closedTopics.remove(topic);
+ 
+         GridMessageListener lsnrs;
+ 
+         for (;;) {
+             lsnrs = lsnrMap.putIfAbsent(topic, lsnr);
+ 
+             if (lsnrs == null) {
+                 lsnrs = lsnr;
+ 
+                 break;
+             }
+ 
+             assert lsnrs != null;
+ 
+             if (!(lsnrs instanceof ArrayListener)) { // We are putting the second listener, creating array.
+                 GridMessageListener arrLsnr = new ArrayListener(lsnrs, lsnr);
+ 
+                 if (lsnrMap.replace(topic, lsnrs, arrLsnr)) {
+                     lsnrs = arrLsnr;
+ 
+                     break;
+                 }
+             }
+             else {
+                 if (((ArrayListener)lsnrs).add(lsnr))
+                     break;
+ 
+                 // Add operation failed because array is already empty and is about to be removed, helping and retrying.
+                 lsnrMap.remove(topic, lsnrs);
+             }
+         }
+ 
+         Map<UUID, GridCommunicationMessageSet> map = msgSetMap.get(topic);
+ 
+         Collection<GridCommunicationMessageSet> msgSets = map != null ? map.values() : null;
+ 
+         if (msgSets != null) {
+             final GridMessageListener lsnrs0 = lsnrs;
+ 
+             boolean success = true;
+ 
+             try {
+                 for (final GridCommunicationMessageSet msgSet : msgSets) {
+                     success = false;
+ 
+                     workersCnt.increment();
+ 
+                     pool(msgSet.policy()).execute(new GridWorker(ctx.gridName(), "msg-worker", log) {
+                         @Override protected void body() {
+                             try {
 -                                unwindMessageSet(msgSet, lsnrs0, false);
++                                unwindMessageSet(msgSet, lsnrs0);
+                             }
+                             finally {
+                                 workersCnt.decrement();
+                             }
+                         }
+                     });
+ 
+                     success = true;
+                 }
+             }
+             catch (RejectedExecutionException e) {
+                 U.error(log, "Failed to process delayed message due to execution rejection. Increase the upper bound " +
+                     "on executor service provided in 'GridConfiguration.getExecutorService()'). Will attempt to " +
+                     "process message in the listener thread instead.", e);
+ 
+                 for (GridCommunicationMessageSet msgSet : msgSets)
 -                    unwindMessageSet(msgSet, lsnr, false);
++                    unwindMessageSet(msgSet, lsnr);
+             }
+             finally {
+                 // Decrement for last runnable submission of which failed.
+                 if (!success)
+                     workersCnt.decrement();
+             }
+         }
+     }
+ 
+     /**
+      * @param topic Message topic.
+      * @return Whether or not listener was indeed removed.
+      */
+     public boolean removeMessageListener(GridTopic topic) {
+         return removeMessageListener((Object)topic);
+     }
+ 
+     /**
+      * @param topic Message topic.
+      * @return Whether or not listener was indeed removed.
+      */
+     public boolean removeMessageListener(Object topic) {
+         return removeMessageListener(topic, null);
+     }
+ 
+     /**
+      * @param topic Listener's topic.
+      * @param lsnr Listener to remove.
+      * @return Whether or not the lsnr was removed.
+      */
+     @SuppressWarnings("deprecation")
+     public boolean removeMessageListener(GridTopic topic, @Nullable GridMessageListener lsnr) {
+         return removeMessageListener((Object)topic, lsnr);
+     }
+ 
+     /**
+      * @param topic Listener's topic.
+      * @param lsnr Listener to remove.
+      * @return Whether or not the lsnr was removed.
+      */
+     @SuppressWarnings({"deprecation", "SynchronizationOnLocalVariableOrMethodParameter"})
+     public boolean removeMessageListener(Object topic, @Nullable final GridMessageListener lsnr) {
+         assert topic != null;
+ 
+         boolean rmv = true;
+ 
+         Collection<GridCommunicationMessageSet> msgSets = null;
+ 
+         // If listener is null, then remove all listeners.
+         if (lsnr == null) {
+             closedTopics.add(topic);
+ 
+             rmv = lsnrMap.remove(topic) != null;
+ 
+             Map<UUID, GridCommunicationMessageSet> map = msgSetMap.remove(topic);
+ 
+             if (map != null)
+                 msgSets = map.values();
+         }
+         else {
+             for (;;) {
+                 GridMessageListener lsnrs = lsnrMap.get(topic);
+ 
+                 // If removing listener before subscription happened.
+                 if (lsnrs == null) {
+                     closedTopics.add(topic);
+ 
+                     Map<UUID, GridCommunicationMessageSet> map = msgSetMap.remove(topic);
+ 
+                     if (map != null)
+                         msgSets = map.values();
+ 
+                     rmv = false;
+ 
+                     break;
+                 }
+                 else {
+                     boolean empty = false;
+ 
+                     if (!(lsnrs instanceof ArrayListener)) {
+                         if (lsnrs.equals(lsnr)) {
+                             if (!lsnrMap.remove(topic, lsnrs))
+                                 continue; // Retry because it can be packed to array listener.
+ 
+                             empty = true;
+                         }
+                         else
+                             rmv = false;
+                     }
+                     else {
+                         ArrayListener arrLsnr = (ArrayListener)lsnrs;
+ 
+                         if (arrLsnr.remove(lsnr))
+                             empty = arrLsnr.isEmpty();
+                         else
+                             // Listener was not found.
+                             rmv = false;
+ 
+                         if (empty)
+                             lsnrMap.remove(topic, lsnrs);
+                     }
+ 
+                     // If removing last subscribed listener.
+                     if (empty) {
+                         closedTopics.add(topic);
+ 
+                         Map<UUID, GridCommunicationMessageSet> map = msgSetMap.remove(topic);
+ 
+                         if (map != null)
+                             msgSets = map.values();
+                     }
+ 
+                     break;
+                 }
+             }
+         }
+ 
+         if (msgSets != null)
+             for (GridCommunicationMessageSet msgSet : msgSets)
+                 ctx.timeout().removeTimeoutObject(msgSet);
+ 
+         if (rmv && log.isDebugEnabled())
+             log.debug("Removed message listener [topic=" + topic + ", lsnr=" + lsnr + ']');
+ 
+         return rmv;
+     }
+ 
+     /**
+      * Gets sent messages count.
+      *
+      * @return Sent messages count.
+      */
+     public int getSentMessagesCount() {
+         return getSpi().getSentMessagesCount();
+     }
+ 
+     /**
+      * Gets sent bytes count.
+      *
+      * @return Sent bytes count.
+      */
+     public long getSentBytesCount() {
+         return getSpi().getSentBytesCount();
+     }
+ 
+     /**
+      * Gets received messages count.
+      *
+      * @return Received messages count.
+      */
+     public int getReceivedMessagesCount() {
+         return getSpi().getReceivedMessagesCount();
+     }
+ 
+     /**
+      * Gets received bytes count.
+      *
+      * @return Received bytes count.
+      */
+     public long getReceivedBytesCount() {
+         return getSpi().getReceivedBytesCount();
+     }
+ 
+     /**
+      * Gets outbound messages queue size.
+      *
+      * @return Outbound messages queue size.
+      */
+     public int getOutboundMessagesQueueSize() {
+         return getSpi().getOutboundMessagesQueueSize();
+     }
+ 
+     /** {@inheritDoc} */
+     @Override public void printMemoryStats() {
+         X.println(">>>");
+         X.println(">>> IO manager memory stats [grid=" + ctx.gridName() + ']');
+         X.println(">>>  lsnrMapSize: " + lsnrMap.size());
+         X.println(">>>  msgSetMapSize: " + msgSetMap.size());
 -        X.println(">>>  msgIdMapSize: " + msgIdMap.size());
+         X.println(">>>  closedTopicsSize: " + closedTopics.sizex());
+         X.println(">>>  discoWaitMapSize: " + waitMap.size());
+     }
+ 
+     /**
+      * Linked chain of listeners.
+      */
+     private static class ArrayListener implements GridMessageListener {
+         /** */
+         private volatile GridMessageListener[] arr;
+ 
+         /**
+          * @param arr Array of listeners.
+          */
+         ArrayListener(GridMessageListener... arr) {
+             this.arr = arr;
+         }
+ 
+         /**
+          * Passes message to the whole chain.
+          *
+          * @param nodeId Node ID.
+          * @param msg Message.
+          */
+         @Override public void onMessage(UUID nodeId, Object msg) {
+             GridMessageListener[] arr0 = arr;
+ 
+             if (arr0 == null)
+                 return;
+ 
+             for (GridMessageListener l : arr0)
+                 l.onMessage(nodeId, msg);
+         }
+ 
+         /**
+          * @return {@code true} If this instance is empty.
+          */
+         boolean isEmpty() {
+             return arr == null;
+         }
+ 
+         /**
+          * @param l Listener.
+          * @return {@code true} If listener was removed.
+          */
+         synchronized boolean remove(GridMessageListener l) {
+             GridMessageListener[] arr0 = arr;
+ 
+             if (arr0 == null)
+                 return false;
+ 
+             if (arr0.length == 1) {
+                 if (!arr0[0].equals(l))
+                     return false;
+ 
+                 arr = null;
+ 
+                 return true;
+             }
+ 
+             for (int i = 0; i < arr0.length; i++) {
+                 if (arr0[i].equals(l)) {
+                     int newLen = arr0.length - 1;
+ 
+                     if (i == newLen) // Remove last.
+                         arr = Arrays.copyOf(arr0, newLen);
+                     else {
+                         GridMessageListener[] arr1 = new GridMessageListener[newLen];
+ 
+                         if (i != 0) // Not remove first.
+                             System.arraycopy(arr0, 0, arr1, 0, i);
+ 
+                         System.arraycopy(arr0, i + 1, arr1, i, newLen - i);
+ 
+                         arr = arr1;
+                     }
+ 
+                     return true;
+                 }
+             }
+ 
+             return false;
+         }
+ 
+         /**
+          * @param l Listener.
+          * @return {@code true} if listener was added. Add can fail if this instance is empty and is about to be removed
+          *         from map.
+          */
+         synchronized boolean add(GridMessageListener l) {
+             GridMessageListener[] arr0 = arr;
+ 
+             if (arr0 == null)
+                 return false;
+ 
+             int oldLen = arr0.length;
+ 
+             arr0 = Arrays.copyOf(arr0, oldLen + 1);
+ 
+             arr0[oldLen] = l;
+ 
+             arr = arr0;
+ 
+             return true;
+         }
+     }
+ 
+     /**
+      * This class represents a message listener wrapper that knows about peer deployment.
+      */
+     private class GridUserMessageListener implements GridMessageListener {
+         /** Predicate listeners. */
+         private final IgniteBiPredicate<UUID, Object> predLsnr;
+ 
+         /** User message topic. */
+         private final Object topic;
+ 
+         /**
+          * @param topic User topic.
+          * @param predLsnr Predicate listener.
+          * @throws IgniteCheckedException If failed to inject resources to predicates.
+          */
+         GridUserMessageListener(@Nullable Object topic, @Nullable IgniteBiPredicate<UUID, Object> predLsnr)
+             throws IgniteCheckedException {
+             this.topic = topic;
+             this.predLsnr = predLsnr;
+ 
+             if (predLsnr != null)
+                 ctx.resource().injectGeneric(predLsnr);
+         }
+ 
+         /** {@inheritDoc} */
+         @SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter", "ConstantConditions",
+             "OverlyStrongTypeCast"})
+         @Override public void onMessage(UUID nodeId, Object msg) {
+             if (!(msg instanceof GridIoUserMessage)) {
+                 U.error(log, "Received unknown message (potentially fatal problem): " + msg);
+ 
+                 return;
+             }
+ 
+             GridIoUserMessage ioMsg = (GridIoUserMessage)msg;
+ 
+             ClusterNode node = ctx.discovery().node(nodeId);
+ 
+             if (node == null) {
+                 U.warn(log, "Failed to resolve sender node (did the node left grid?): " + nodeId);
+ 
+                 return;
+             }
+ 
+             Object msgBody = ioMsg.body();
+ 
+             assert msgBody != null || ioMsg.bodyBytes() != null;
+ 
+             try {
+                 byte[] msgTopicBytes = ioMsg.topicBytes();
+ 
+                 Object msgTopic = ioMsg.topic();
+ 
+                 GridDeployment dep = ioMsg.deployment();
+ 
+                 if (dep == null && ctx.config().isPeerClassLoadingEnabled() &&
+                     ioMsg.deploymentClassName() != null) {
+                     dep = ctx.deploy().getGlobalDeployment(
+                         ioMsg.deploymentMode(),
+                         ioMsg.deploymentClassName(),
+                         ioMsg.deploymentClassName(),
+                         ioMsg.userVersion(),
+                         nodeId,
+                         ioMsg.classLoaderId(),
+                         ioMsg.loaderParticipants(),
+                         null);
+ 
+                     if (dep == null)
+                         throw new IgniteDeploymentException(
+                             "Failed to obtain deployment information for user message. " +
+                             "If you are using custom message or topic class, try implementing " +
+                             "GridPeerDeployAware interface. [msg=" + ioMsg + ']');
+ 
+                     ioMsg.deployment(dep); // Cache deployment.
+                 }
+ 
+                 // Unmarshall message topic if needed.
+                 if (msgTopic == null && msgTopicBytes != null) {
+                     msgTopic = marsh.unmarshal(msgTopicBytes, dep != null ? dep.classLoader() : null);
+ 
+                     ioMsg.topic(msgTopic); // Save topic to avoid future unmarshallings.
+                 }
+ 
+                 if (!F.eq(topic, msgTopic))
+                     return;
+ 
+                 if (msgBody == null) {
+                     msgBody = marsh.unmarshal(ioMsg.bodyBytes(), dep != null ? dep.classLoader() : null);
+ 
+                     ioMsg.body(msgBody); // Save body to avoid future unmarshallings.
+                 }
+ 
+                 // Resource injection.
+                 if (dep != null)
+                     ctx.resource().inject(dep, dep.deployedClass(ioMsg.deploymentClassName()), msgBody);
+             }
+             catch (IgniteCheckedException e) {
+                 U.error(log, "Failed to unmarshal user message [node=" + nodeId + ", message=" +
+                     msg + ']', e);
+             }
+ 
+             if (msgBody != null) {
+                 if (predLsnr != null) {
+                     if (!predLsnr.apply(nodeId, msgBody))
+                         removeMessageListener(TOPIC_COMM_USER, this);
+                 }
+             }
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public boolean equals(Object o) {
+             if (this == o)
+                 return true;
+ 
+             if (o == null || getClass() != o.getClass())
+                 return false;
+ 
+             GridUserMessageListener l = (GridUserMessageListener)o;
+ 
+             return F.eq(predLsnr, l.predLsnr) && F.eq(topic, l.topic);
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public int hashCode() {
+             int res = predLsnr != null ? predLsnr.hashCode() : 0;
+ 
+             res = 31 * res + (topic != null ? topic.hashCode() : 0);
+ 
+             return res;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(GridUserMessageListener.class, this);
+         }
+     }
+ 
+     /**
+      * Ordered communication message set.
+      */
+     private class GridCommunicationMessageSet implements GridTimeoutObject {
+         /** */
+         private final UUID nodeId;
+ 
+         /** */
+         private long endTime;
+ 
+         /** */
+         private final IgniteUuid timeoutId;
+ 
+         /** */
++        @GridToStringInclude
+         private final Object topic;
+ 
+         /** */
+         private final GridIoPolicy plc;
+ 
+         /** */
+         @GridToStringInclude
 -        private final List<IgniteBiTuple<GridIoMessage, Long>> msgs = new ArrayList<>();
 -
 -        /** */
 -        private long nextMsgId = 1;
++        private final Queue<IgniteBiTuple<GridIoMessage, Long>> msgs = new ConcurrentLinkedDeque<>();
+ 
+         /** */
+         private final AtomicBoolean reserved = new AtomicBoolean();
+ 
+         /** */
+         private final long timeout;
+ 
+         /** */
+         private final boolean skipOnTimeout;
+ 
+         /** */
+         private long lastTs;
+ 
 -        /** */
 -        private volatile boolean changed;
 -
+         /**
+          * @param plc Communication policy.
+          * @param topic Communication topic.
+          * @param nodeId Node ID.
+          * @param timeout Timeout.
+          * @param skipOnTimeout Whether message can be skipped on timeout.
+          * @param msg Message to add immediately.
+          */
 -        GridCommunicationMessageSet(GridIoPolicy plc, Object topic, UUID nodeId, long timeout, boolean skipOnTimeout,
 -            GridIoMessage msg) {
++        GridCommunicationMessageSet(
++            GridIoPolicy plc,
++            Object topic,
++            UUID nodeId,
++            long timeout,
++            boolean skipOnTimeout,
++            GridIoMessage msg
++        ) {
+             assert nodeId != null;
+             assert topic != null;
+             assert plc != null;
+             assert msg != null;
+ 
+             this.plc = plc;
+             this.nodeId = nodeId;
+             this.topic = topic;
+             this.timeout = timeout == 0 ? ctx.config().getNetworkTimeout() : timeout;
+             this.skipOnTimeout = skipOnTimeout;
+ 
+             endTime = endTime(timeout);
+ 
+             timeoutId = IgniteUuid.randomUuid();
+ 
+             lastTs = U.currentTimeMillis();
+ 
+             msgs.add(F.t(msg, lastTs));
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public IgniteUuid timeoutId() {
+             return timeoutId;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public long endTime() {
+             return endTime;
+         }
+ 
+         /** {@inheritDoc} */
+         @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
+         @Override public void onTimeout() {
+             GridMessageListener lsnr = lsnrMap.get(topic);
+ 
+             if (lsnr != null) {
+                 long delta = 0;
+ 
+                 if (skipOnTimeout) {
+                     while (true) {
+                         delta = 0;
+ 
+                         boolean unwind = false;
+ 
+                         synchronized (this) {
+                             if (!msgs.isEmpty()) {
+                                 delta = U.currentTimeMillis() - lastTs;
+ 
+                                 if (delta >= timeout)
+                                     unwind = true;
+                             }
+                         }
+ 
+                         if (unwind)
 -                            unwindMessageSet(this, lsnr, true);
++                            unwindMessageSet(this, lsnr);
+                         else
+                             break;
+                     }
+                 }
+ 
+                 // Someone is still listening to messages, so delay set removal.
+                 endTime = endTime(timeout - delta);
+ 
+                 ctx.timeout().addTimeoutObject(this);
+ 
+                 return;
+             }
+ 
+             if (log.isDebugEnabled())
+                 log.debug("Removing message set due to timeout: " + this);
+ 
+             ConcurrentMap<UUID, GridCommunicationMessageSet> map = msgSetMap.get(topic);
+ 
+             if (map != null) {
+                 boolean rmv;
+ 
+                 synchronized (map) {
+                     rmv = map.remove(nodeId, this) && map.isEmpty();
+                 }
+ 
+                 if (rmv)
+                     msgSetMap.remove(topic, map);
+             }
+         }
+ 
+         /**
+          * @return ID of node that sent the messages in the set.
+          */
+         UUID nodeId() {
+             return nodeId;
+         }
+ 
+         /**
+          * @return Communication policy.
+          */
+         GridIoPolicy policy() {
+             return plc;
+         }
+ 
+         /**
+          * @return Message topic.
+          */
+         Object topic() {
+             return topic;
+         }
+ 
+         /**
+          * @return {@code True} if successful.
+          */
+         boolean reserve() {
+             return reserved.compareAndSet(false, true);
+         }
+ 
+         /**
++         * @return {@code True} if set is reserved.
++         */
++        boolean reserved() {
++            return reserved.get();
++        }
++
++        /**
+          * Releases reservation.
+          */
+         void release() {
+             assert reserved.get() : "Message set was not reserved: " + this;
+ 
+             reserved.set(false);
+         }
+ 
+         /**
 -         * @param force Whether to force unwind and drop missing
 -         *      ordered messages that are not received yet.
 -         * @return Session request.
++         * @param lsnr Listener to notify.
+          */
 -        synchronized Collection<GridIoMessage> unwind(boolean force) {
++        void unwind(GridMessageListener lsnr) {
+             assert reserved.get();
+ 
 -            changed = false;
 -
 -            if (msgs.isEmpty())
 -                return Collections.emptyList();
 -
 -            if (msgs.size() == 1) {
 -                IgniteBiTuple<GridIoMessage, Long> t = msgs.get(0);
 -
 -                GridIoMessage msg = t.get1();
 -
 -                if (force || msg.messageId() == nextMsgId) {
 -                    if (msg.messageId() != nextMsgId) {
 -                        for (long skipped = nextMsgId; skipped < msg.messageId(); skipped++) {
 -                            U.warn(log, "Skipped ordered message due to timeout, consider increasing " +
 -                                "networkTimeout configuration property [topic=" + topic + ", msgId=" +
 -                                skipped + ", timeout=" + timeout + ']');
 -                        }
 -                    }
 -
 -                    nextMsgId = msg.messageId() + 1;
 -
 -                    lastTs = t.get2();
 -
 -                    msgs.clear();
 -
 -                    return Collections.singleton(msg);
 -                }
 -
 -                return Collections.emptyList();
 -            }
 -
 -            // Sort before unwinding.
 -            Collections.sort(msgs, MSG_CMP);
 -
 -            Collection<GridIoMessage> orderedMsgs = new LinkedList<>();
 -
 -            for (Iterator<IgniteBiTuple<GridIoMessage, Long>> iter = msgs.iterator(); iter.hasNext();) {
 -                IgniteBiTuple<GridIoMessage, Long> t = iter.next();
 -
 -                GridIoMessage msg = t.get1();
 -
 -                if (force || msg.messageId() == nextMsgId) {
 -                    if (msg.messageId() != nextMsgId) {
 -                        for (long skipped = nextMsgId; skipped < msg.messageId(); skipped++) {
 -                            U.warn(log, "Skipped ordered message due to timeout, consider increasing " +
 -                                "networkTimeout configuration property [topic=" + topic + ", msgId=" +
 -                                skipped + ", timeout=" + timeout + ']');
 -                        }
 -                    }
 -
 -                    force = false;
 -
 -                    orderedMsgs.add(msg);
 -
 -                    nextMsgId = msg.messageId() + 1;
 -
 -                    lastTs = t.get2();
 -
 -                    iter.remove();
 -                }
 -                else
 -                    break;
 -            }
 -
 -            return orderedMsgs;
++            for (IgniteBiTuple<GridIoMessage, Long> t = msgs.poll(); t != null; t = msgs.poll())
++                lsnr.onMessage(nodeId, t.get1().message());
+         }
+ 
+         /**
+          * @param msg Message to add.
+          */
 -        synchronized void add(GridIoMessage msg) {
 -            if (msg.messageId() >= nextMsgId) {
 -                msgs.add(F.t(msg, U.currentTimeMillis()));
 -
 -                changed = true;
 -            }
 -            else {
 -                U.warn(log, "Received previously skipped ordered message (will be dropped) [topic=" + topic +
 -                    ", msgId=" + msg.messageId() + ", timeout=" + timeout + ']');
 -            }
++        void add(GridIoMessage msg) {
++            msgs.add(F.t(msg, U.currentTimeMillis()));
+         }
+ 
+         /**
+          * @return {@code True} if set has messages to unwind.
+          */
+         boolean changed() {
 -            return changed;
++            return !msgs.isEmpty();
+         }
+ 
+         /**
+          * Calculates end time with overflow check.
+          *
+          * @param timeout Timeout in milliseconds.
+          * @return End time in milliseconds.
+          */
+         private long endTime(long timeout) {
+             long endTime = U.currentTimeMillis() + timeout;
+ 
+             // Account for overflow.
+             if (endTime < 0)
+                 endTime = Long.MAX_VALUE;
+ 
+             return endTime;
+         }
+ 
+         /** {@inheritDoc} */
 -        @Override public synchronized String toString() {
++        @Override public String toString() {
+             return S.toString(GridCommunicationMessageSet.class, this);
+         }
+     }
+ 
+     /**
+      *
+      */
+     private static class ConcurrentHashMap0<K, V> extends ConcurrentHashMap8<K, V> {
+         /** */
+         private static final long serialVersionUID = 0L;
+ 
+         /** */
+         private int hash;
+ 
+         /**
+          * @param o Object to be compared for equality with this map.
+          * @return {@code True} only for {@code this}.
+          */
+         @Override public boolean equals(Object o) {
+             return o == this;
+         }
+ 
+         /**
+          * @return Identity hash code.
+          */
+         @Override public int hashCode() {
+             if (hash == 0) {
+                 int hash0 = System.identityHashCode(this);
+ 
+                 hash = hash0 != 0 ? hash0 : -1;
+             }
+ 
+             return hash;
+         }
+     }
+ 
+     /**
+      *
+      */
 -    private static class ConcurrentHashSet0<E> extends GridConcurrentHashSet<E> {
 -        /** */
 -        private static final long serialVersionUID = 0L;
 -
 -        /** */
 -        private int hash;
 -
 -        /**
 -         *
 -         */
 -        private ConcurrentHashSet0() {
 -            super(1, 1, 1);
 -        }
 -
 -        /**
 -         * @param o Object to be compared for equality with this map.
 -         * @return {@code True} only for {@code this}.
 -         */
 -        @Override public boolean equals(Object o) {
 -            return o == this;
 -        }
 -
 -        /**
 -         * @return Identity hash code.
 -         */
 -        @Override public int hashCode() {
 -            if (hash == 0) {
 -                int hash0 = System.identityHashCode(this);
 -
 -                hash = hash0 != 0 ? hash0 : -1;
 -            }
 -
 -            return hash;
 -        }
 -    }
 -
 -    /**
 -     *
 -     */
+     private static class DelayedMessage {
+         /** */
+         private final UUID nodeId;
+ 
+         /** */
+         private final GridIoMessage msg;
+ 
+         /** */
+         private final IgniteRunnable msgC;
+ 
 -        /** */
 -        private final long rcvTime = U.currentTimeMillis();
 -
+         /**
+          * @param nodeId Node ID.
+          * @param msg Message.
+          * @param msgC Callback.
+          */
+         private DelayedMessage(UUID nodeId, GridIoMessage msg, IgniteRunnable msgC) {
+             this.nodeId = nodeId;
+             this.msg = msg;
+             this.msgC = msgC;
+         }
+ 
+         /**
 -         * @return Receive time.
 -         */
 -        public long receiveTime() {
 -            return rcvTime;
 -        }
 -
 -        /**
+          * @return Message char.
+          */
+         public IgniteRunnable callback() {
+             return msgC;
+         }
+ 
+         /**
+          * @return Message.
+          */
+         public GridIoMessage message() {
+             return msg;
+         }
+ 
+         /**
+          * @return Node id.
+          */
+         public UUID nodeId() {
+             return nodeId;
+         }
+ 
+         /** {@inheritDoc} */
+         @Override public String toString() {
+             return S.toString(DelayedMessage.class, this, super.toString());
+         }
+     }
+ }


Mime
View raw message