ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject ignite git commit: Merge remote-tracking branch 'apache/ignite-5578' into ignite-5872-5578
Date Thu, 10 Aug 2017 14:10:18 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-5872-5578 [created] 30339d35c


Merge remote-tracking branch 'apache/ignite-5578' into ignite-5872-5578


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

Branch: refs/heads/ignite-5872-5578
Commit: 30339d35c9bba365db0369ca8d892efb3a2b3a96
Parents: 855ece3 bc9a416
Author: Alexey Goncharuk <alexey.goncharuk@gmail.com>
Authored: Thu Aug 10 17:09:53 2017 +0300
Committer: Alexey Goncharuk <alexey.goncharuk@gmail.com>
Committed: Thu Aug 10 17:09:53 2017 +0300

----------------------------------------------------------------------
 .../apache/ignite/IgniteSystemProperties.java   |    3 +
 .../java/org/apache/ignite/TestDebugLog.java    |  219 ++
 .../internal/IgniteDiagnosticMessage.java       |    2 +-
 .../communication/GridIoMessageFactory.java     |    9 +-
 .../internal/managers/discovery/DiscoCache.java |   79 +-
 .../discovery/GridDiscoveryManager.java         |   28 +-
 .../affinity/AffinityTopologyVersion.java       |    7 +
 .../affinity/GridAffinityAssignmentCache.java   |   42 +
 .../affinity/GridAffinityProcessor.java         |    8 +-
 .../cache/CacheAffinitySharedManager.java       |  688 ++++--
 .../processors/cache/CacheGroupContext.java     |   18 +-
 .../cache/CachePartitionExchangeWorkerTask.java |    5 +-
 .../ClientCacheChangeDummyDiscoveryMessage.java |    5 +
 .../cache/ClientCacheUpdateTimeout.java         |    5 +
 .../processors/cache/ClusterCachesInfo.java     |   22 +-
 .../processors/cache/ExchangeContext.java       |  131 ++
 .../cache/ExchangeDiscoveryEvents.java          |  262 +++
 .../processors/cache/GridCacheAdapter.java      |    8 +-
 .../processors/cache/GridCacheContext.java      |    2 +-
 .../processors/cache/GridCacheIoManager.java    |   57 +-
 .../processors/cache/GridCacheMapEntry.java     |    6 +-
 .../GridCachePartitionExchangeManager.java      |  413 +++-
 .../processors/cache/GridCacheProcessor.java    |   14 +-
 .../dht/ClientCacheDhtTopologyFuture.java       |   12 +-
 .../dht/GridClientPartitionTopology.java        |  132 +-
 .../distributed/dht/GridDhtCacheAdapter.java    |   18 +-
 .../distributed/dht/GridDhtLocalPartition.java  |    4 +-
 .../dht/GridDhtPartitionTopology.java           |   37 +-
 .../dht/GridDhtPartitionTopologyImpl.java       |  492 +++--
 .../dht/GridDhtPartitionsReservation.java       |    2 +-
 .../distributed/dht/GridDhtTopologyFuture.java  |   36 +-
 .../dht/GridDhtTransactionalCacheAdapter.java   |   95 +-
 .../dht/GridPartitionedGetFuture.java           |    4 +-
 .../dht/GridPartitionedSingleGetFuture.java     |    4 +-
 .../GridDhtAtomicAbstractUpdateFuture.java      |    2 +-
 .../dht/atomic/GridDhtAtomicCache.java          |   31 +-
 .../GridNearAtomicSingleUpdateFuture.java       |    1 -
 .../colocated/GridDhtColocatedLockFuture.java   |    2 +-
 .../preloader/CacheGroupAffinityMessage.java    |  274 +++
 .../preloader/ForceRebalanceExchangeTask.java   |    5 +
 .../preloader/GridDhtPartitionExchangeId.java   |   11 +
 .../dht/preloader/GridDhtPartitionMap.java      |    2 +-
 .../dht/preloader/GridDhtPartitionSupplier.java |    2 +-
 .../GridDhtPartitionsAbstractMessage.java       |   37 +-
 .../GridDhtPartitionsExchangeFuture.java        | 1974 +++++++++++++-----
 .../preloader/GridDhtPartitionsFullMessage.java |  161 +-
 .../GridDhtPartitionsSingleMessage.java         |   78 +-
 .../GridDhtPartitionsSingleRequest.java         |   47 +-
 .../dht/preloader/GridDhtPreloader.java         |   34 +-
 .../IgniteDhtPartitionCountersMap.java          |    7 +
 .../dht/preloader/InitNewCoordinatorFuture.java |  307 +++
 .../RebalanceReassignExchangeTask.java          |    5 +
 .../distributed/near/GridNearCacheAdapter.java  |    2 +-
 .../distributed/near/GridNearGetFuture.java     |    4 +-
 .../distributed/near/GridNearLockFuture.java    |    2 +-
 ...arOptimisticSerializableTxPrepareFuture.java |    1 +
 .../near/GridNearOptimisticTxPrepareFuture.java |    1 +
 .../GridNearPessimisticTxPrepareFuture.java     |    1 +
 .../near/GridNearTxPrepareRequest.java          |   14 +
 .../GridCacheDatabaseSharedManager.java         |    7 +-
 .../cache/query/GridCacheQueryAdapter.java      |    4 +-
 .../cache/transactions/IgniteTxAdapter.java     |    2 +-
 .../cache/transactions/IgniteTxHandler.java     |  184 +-
 .../closure/GridClosureProcessor.java           |   36 +-
 .../cluster/GridClusterStateProcessor.java      |    2 +-
 .../datastreamer/DataStreamProcessor.java       |   57 +-
 .../datastreamer/DataStreamerImpl.java          |   65 +-
 .../datastreamer/PlatformDataStreamer.java      |    3 +-
 .../query/schema/SchemaExchangeWorkerTask.java  |    5 +
 .../SchemaNodeLeaveExchangeWorkerTask.java      |    5 +
 .../processors/task/GridTaskWorker.java         |    8 +-
 .../org/apache/ignite/thread/IgniteThread.java  |    9 +
 .../internal/TestDelayingCommunicationSpi.java  |   63 +
 ...CacheExchangeMessageDuplicatedStateTest.java |    9 +-
 .../IgniteClientCacheStartFailoverTest.java     |    4 +-
 .../IgniteClusterActivateDeactivateTest.java    |    4 +-
 .../cache/IgniteDynamicCacheStartSelfTest.java  |   26 +-
 ...niteTopologyValidatorGridSplitCacheTest.java |    6 +-
 ...AffinityCoordinatorDynamicStartStopTest.java |    2 +-
 ...eAbstractDataStructuresFailoverSelfTest.java |    7 +-
 .../distributed/CacheExchangeMergeTest.java     | 1530 ++++++++++++++
 .../CacheLateAffinityAssignmentTest.java        |  598 ++++--
 ...CacheLoadingConcurrentGridStartSelfTest.java |   11 +
 .../CacheLockReleaseNodeLeaveTest.java          |   13 +-
 .../distributed/CachePartitionStateTest.java    |   18 +-
 ...ncurrentGridStartSelfTestAllowOverwrite.java |   33 +
 ...niteCacheClientNodeChangingTopologyTest.java |    5 +-
 ...teCacheClientNodePartitionsExchangeTest.java |   52 +-
 ...ePrimaryNodeFailureRecoveryAbstractTest.java |    4 +-
 ...eAtomicInvalidPartitionHandlingSelfTest.java |   36 +-
 .../IgniteChangeGlobalStateTest.java            |   11 +-
 .../join/JoinInActiveNodeToActiveCluster.java   |    4 +-
 .../junits/common/GridCommonAbstractTest.java   |   22 +-
 .../testsuites/IgniteCacheTestSuite2.java       |    7 +-
 .../testsuites/IgniteCacheTestSuite6.java       |    3 +
 .../cache/WaitMapExchangeFinishCallable.java    |    4 +-
 96 files changed, 7264 insertions(+), 1469 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index 9762586,3d25084..a597227
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@@ -439,9 -457,9 +457,9 @@@ public class CacheAffinitySharedManager
                              GridClientPartitionTopology clientTop = cctx.exchange().clearClientTopology(grp.groupId());
  
                              if (clientTop != null) {
-                                 grp.topology().update(topVer,
+                                 grp.topology().update(grpHolder.affinity().lastVersion(),
                                      clientTop.partitionMap(true),
 -                                    clientTop.updateCounters(false),
 +                                    clientTop.fullUpdateCounters(),
                                      Collections.<Integer>emptySet(),
                                      null);
                              }

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 27fda14,48909d4..70f1dfa
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@@ -668,21 -723,23 +723,21 @@@ public class GridCachePartitionExchange
          if (top != null)
              return top;
  
-         CacheGroupDescriptor grpDesc = cctx.cache().cacheGroupDescriptors().get(grpId);
 -        Object affKey = null;
 -
+         CacheGroupDescriptor grpDesc = cctx.affinity().cacheGroups().get(grpId);
  
-         assert grpDesc != null: "Failed for exchange: " + exchFut;
 -        if (grpDesc != null) {
 -            CacheConfiguration<?, ?> ccfg = grpDesc.config();
++        assert grpDesc != null;
  
 -            AffinityFunction aff = ccfg.getAffinity();
 +        CacheConfiguration<?, ?> ccfg = grpDesc.config();
  
 -            affKey = cctx.kernalContext().affinity().similaryAffinityKey(aff,
 -                ccfg.getNodeFilter(),
 -                ccfg.getBackups(),
 -                aff.partitions());
 -        }
 +        AffinityFunction aff = ccfg.getAffinity();
 +
 +        Object affKey = cctx.kernalContext().affinity().similaryAffinityKey(aff,
 +            ccfg.getNodeFilter(),
 +            ccfg.getBackups(),
 +            aff.partitions());
  
          GridClientPartitionTopology old = clientTops.putIfAbsent(grpId,
-             top = new GridClientPartitionTopology(cctx, grpId, exchFut, aff.partitions(),
affKey));
 -            top = new GridClientPartitionTopology(cctx, grpId, affKey));
++            top = new GridClientPartitionTopology(cctx, grpId, aff.partitions(), affKey));
  
          return old != null ? old : top;
      }
@@@ -1074,11 -1118,9 +1116,10 @@@
       * @param id Exchange ID.
       */
      private void sendLocalPartitions(ClusterNode node, @Nullable GridDhtPartitionExchangeId
id) {
-         GridDhtPartitionsSingleMessage m = createPartitionsSingleMessage(node,
-             id,
+         GridDhtPartitionsSingleMessage m = createPartitionsSingleMessage(id,
              cctx.kernalContext().clientNode(),
 -            false);
 +            false,
 +            null);
  
          if (log.isDebugEnabled())
              log.debug("Sending local partitions [nodeId=" + node.id() + ", msg=" + m + ']');
@@@ -1103,13 -1144,10 +1143,12 @@@
       * @param sndCounters {@code True} if need send partition update counters.
       * @return Message.
       */
 -    public GridDhtPartitionsSingleMessage createPartitionsSingleMessage(@Nullable GridDhtPartitionExchangeId
exchangeId,
 +    public GridDhtPartitionsSingleMessage createPartitionsSingleMessage(
-         ClusterNode targetNode,
 +        @Nullable GridDhtPartitionExchangeId exchangeId,
          boolean clientOnlyExchange,
 -        boolean sndCounters)
 -    {
 +        boolean sndCounters,
 +        ExchangeActions exchActions
 +    ) {
          GridDhtPartitionsSingleMessage m = new GridDhtPartitionsSingleMessage(exchangeId,
              clientOnlyExchange,
              cctx.versions().last(),

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index f844e19,745e7d7..77792c7
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@@ -36,9 -36,9 +36,11 @@@ import org.apache.ignite.internal.Ignit
  import org.apache.ignite.internal.managers.discovery.DiscoCache;
  import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
  import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+ import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
+ import org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents;
  import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionFullCountersMap;
 +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
  import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
  import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
  import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
@@@ -116,15 -117,11 +118,13 @@@ public class GridClientPartitionTopolog
      /**
       * @param cctx Context.
       * @param grpId Group ID.
-      * @param exchFut Exchange ID.
 +     * @param parts Number of partitions in the group.
       * @param similarAffKey Key to find caches with similar affinity.
       */
      public GridClientPartitionTopology(
-         GridCacheSharedContext cctx,
+         GridCacheSharedContext<?, ?> cctx,
          int grpId,
-         GridDhtPartitionsExchangeFuture exchFut,
 +        int parts,
          Object similarAffKey
      ) {
          this.cctx = cctx;
@@@ -137,16 -132,9 +135,11 @@@
  
          log = cctx.logger(getClass());
  
-         lock.writeLock().lock();
+         node2part = new GridDhtPartitionFullMap(cctx.localNode().id(),
+             cctx.localNode().order(),
+             updateSeq.get());
 +
-         try {
-             cntrMap = new CachePartitionFullCountersMap(parts);
- 
-             beforeExchange0(cctx.localNode(), exchFut);
-         }
-         finally {
-             lock.writeLock().unlock();
-         }
++        cntrMap = new CachePartitionFullCountersMap(parts);
      }
  
      /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
index 4608977,0dea5e4..22205ea
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
@@@ -255,9 -263,9 +263,9 @@@ public interface GridDhtPartitionTopolo
       * @return {@code True} if local state was changed.
       */
      public boolean update(
-         @Nullable AffinityTopologyVersion exchangeVer,
+         @Nullable AffinityTopologyVersion exchangeResVer,
          GridDhtPartitionFullMap partMap,
 -        @Nullable Map<Integer, T2<Long, Long>> cntrMap,
 +        @Nullable CachePartitionFullCountersMap cntrMap,
          Set<Integer> partsToReload,
          @Nullable AffinityTopologyVersion msgTopVer);
  
@@@ -267,19 -276,13 +276,20 @@@
       * @return {@code True} if local state was changed.
       */
      public boolean update(@Nullable GridDhtPartitionExchangeId exchId,
-         GridDhtPartitionMap parts);
+         GridDhtPartitionMap parts,
+         boolean force);
  
      /**
 +     * Collects update counters collected during exchange. Called on coordinator.
 +     *
       * @param cntrMap Counters map.
       */
 -    public void applyUpdateCounters(Map<Integer, T2<Long, Long>> cntrMap);
 +    public void collectUpdateCounters(CachePartitionPartialCountersMap cntrMap);
 +
 +    /**
 +     * Applies update counters collected during exchange on coordinator. Called on coordinator.
 +     */
 +    public void applyUpdateCounters();
  
      /**
       * Checks if there is at least one owner for each partition in the cache topology.

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index 84ae9af,72ab8c8..23e28b0
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@@ -41,9 -41,8 +41,10 @@@ import org.apache.ignite.internal.manag
  import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
  import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
  import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+ import org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents;
  import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionFullCountersMap;
 +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
  import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
  import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
  import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
@@@ -1252,8 -1312,15 +1311,8 @@@ public class GridDhtPartitionTopologyIm
                      if (state == OWNING) {
                          GridDhtLocalPartition locPart = locParts.get(p);
  
-                         assert locPart != null;
+                         assert locPart != null : grp.cacheOrGroupName();
  
 -                        if (incomeCntrMap != null) {
 -                            T2<Long, Long> cntr = incomeCntrMap.get(p);
 -
 -                            if (cntr != null && cntr.get2() > locPart.updateCounter())
 -                                locPart.updateCounter(cntr.get2());
 -                        }
 -
                          if (locPart.state() == MOVING) {
                              boolean success = locPart.own();
  

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsAbstractMessage.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsAbstractMessage.java
index cae3ce2,95c1a4f..84cc792
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsAbstractMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsAbstractMessage.java
@@@ -90,6 -104,19 +102,13 @@@ public abstract class GridDhtPartitions
      }
  
      /**
+      * @param exchId Exchange ID.
+      */
+     public void exchangeId(GridDhtPartitionExchangeId exchId) {
+         this.exchId = exchId;
+     }
+ 
+     /**
 -     * @param grpId Cache group ID.
 -     * @return Parition update counters.
 -     */
 -    public abstract Map<Integer, T2<Long, Long>> partitionUpdateCounters(int
grpId);
 -
 -    /**
       * @return Last used version among all nodes.
       */
      @Nullable public GridCacheVersion lastVersion() {

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 8530a23,07f36af..861ab38
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@@ -622,9 -762,9 +762,9 @@@ public class GridDhtPartitionsExchangeF
                  boolean updateTop = exchId.topologyVersion().equals(grp.localStartVersion());
  
                  if (updateTop && clientTop != null) {
-                     top.update(topologyVersion(),
+                     top.update(null,
                          clientTop.partitionMap(true),
 -                        clientTop.updateCounters(false),
 +                        clientTop.fullUpdateCounters(),
                          Collections.<Integer>emptySet(),
                          null);
                  }
@@@ -1129,17 -1228,15 +1228,16 @@@
                  true);
          }
          else {
-             msg = cctx.exchange().createPartitionsSingleMessage(node,
-                 exchangeId(),
+             msg = cctx.exchange().createPartitionsSingleMessage(exchangeId(),
                  false,
 -                true);
 +                true,
 +                exchActions);
-         }
  
-         Map<Integer, Map<Integer, Long>> partHistReserved0 = partHistReserved;
+             Map<Integer, Map<Integer, Long>> partHistReserved0 = partHistReserved;
  
-         if (partHistReserved0 != null)
-             msg.partitionHistoryCounters(partHistReserved0);
+             if (partHistReserved0 != null)
+                 msg.partitionHistoryCounters(partHistReserved0);
+         }
  
          if (stateChangeExchange() && changeGlobalStateE != null)
              msg.setError(changeGlobalStateE);
@@@ -1720,34 -2155,102 +2158,107 @@@
                  }
              }
  
-             for (GridDhtPartitionsAbstractMessage msg : msgs.values()) {
-                 if (msg instanceof GridDhtPartitionsSingleMessage) {
-                     GridDhtPartitionsSingleMessage msg0 = (GridDhtPartitionsSingleMessage)msg;
+             if (exchCtx.mergeExchanges()) {
+                 log.info("Coordinator received all messages, try merge [ver=" + initialVersion()
+ ']');
+ 
+                 boolean finish = cctx.exchange().mergeExchangesOnCoordinator(this);
+ 
+                 if (!finish)
+                     return;
+             }
+ 
+             finishExchangeOnCoordinator(sndResNodes);
+         }
+         catch (IgniteCheckedException e) {
+             if (reconnectOnError(e))
+                 onDone(new IgniteNeedReconnectException(cctx.localNode(), e));
+             else
+                 onDone(e);
+         }
+     }
+ 
+     /**
+      * @param sndResNodes Additional nodes to send finish message to.
+      */
+     private void finishExchangeOnCoordinator(@Nullable Collection<ClusterNode> sndResNodes)
{
+         try {
+             AffinityTopologyVersion resTopVer = exchCtx.events().topologyVersion();
  
-                     for (Map.Entry<Integer, GridDhtPartitionMap> entry : msg0.partitions().entrySet())
{
-                         Integer grpId = entry.getKey();
-                         CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
+             log.info("finishExchangeOnCoordinator [topVer=" + initialVersion() +
+                 ", resVer=" + resTopVer + ']');
  
-                         GridDhtPartitionTopology top = grp != null ? grp.topology() :
-                             cctx.exchange().clientTopology(grpId, this);
+             Map<Integer, CacheGroupAffinityMessage> idealAffDiff = null;
  
-                         CachePartitionPartialCountersMap cntrs = msg0.partitionUpdateCounters(grpId);
+             if (exchCtx.mergeExchanges()) {
+                 synchronized (mux) {
+                     if (mergedJoinExchMsgs != null) {
+                         for (Map.Entry<UUID, GridDhtPartitionsSingleMessage> e : mergedJoinExchMsgs.entrySet())
{
+                             msgs.put(e.getKey(), e.getValue());
  
-                         if (cntrs != null)
-                             top.collectUpdateCounters(cntrs);
+                             updatePartitionSingleMap(e.getKey(), e.getValue());
+                         }
                      }
                  }
+ 
+                 assert exchCtx.events().hasServerJoin() || exchCtx.events().hasServerLeft();
+ 
+                 exchCtx.events().processEvents(this);
+ 
+                 if (exchCtx.events().hasServerLeft())
+                     idealAffDiff = cctx.affinity().onServerLeftWithExchangeMergeProtocol(this);
+                 else
+                     cctx.affinity().onServerJoinWithExchangeMergeProtocol(this, true);
+ 
+                 for (CacheGroupDescriptor desc : cctx.affinity().cacheGroups().values())
{
+                     if (desc.config().getCacheMode() == CacheMode.LOCAL)
+                         continue;
+ 
+                     CacheGroupContext grp = cctx.cache().cacheGroup(desc.groupId());
+ 
+                     GridDhtPartitionTopology top = grp != null ? grp.topology() :
+                         cctx.exchange().clientTopology(desc.groupId());
+ 
+                     top.beforeExchange(this, true, true);
+                 }
+             }
+ 
+             Map<Integer, CacheGroupAffinityMessage> joinedNodeAff = null;
+ 
+             for (Map.Entry<UUID, GridDhtPartitionsSingleMessage> e : msgs.entrySet())
{
+                 GridDhtPartitionsSingleMessage msg = e.getValue();
+ 
+                 // Apply update counters after all single messages are received.
+                 for (Map.Entry<Integer, GridDhtPartitionMap> entry : msg.partitions().entrySet())
{
+                     Integer grpId = entry.getKey();
+ 
+                     CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
+ 
+                     GridDhtPartitionTopology top = grp != null ? grp.topology() :
+                         cctx.exchange().clientTopology(grpId);
+ 
 -                    Map<Integer, T2<Long, Long>> cntrs = msg.partitionUpdateCounters(grpId);
++                        CachePartitionPartialCountersMap cntrs = msg.partitionUpdateCounters(grpId);
+ 
+                     if (cntrs != null)
 -                        top.applyUpdateCounters(cntrs);
++                        top.collectUpdateCounters(cntrs);
+                 }
+ 
+                 Collection<Integer> affReq = msg.cacheGroupsAffinityRequest();
+ 
+                 if (affReq != null) {
+                     joinedNodeAff = CacheGroupAffinityMessage.createAffinityMessages(cctx,
+                         resTopVer,
+                         affReq,
+                         joinedNodeAff);
+                 }
              }
  
 +            for (CacheGroupContext grpCtx : cctx.cache().cacheGroups()) {
 +                if (!grpCtx.isLocal())
 +                    grpCtx.topology().applyUpdateCounters();
 +            }
 +
-             if (discoEvt.type() == EVT_NODE_JOINED)
-                 assignPartitionsStates();
-             else if (discoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
-                 assert discoEvt instanceof DiscoveryCustomEvent;
+             if (firstDiscoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
+                 assert firstDiscoEvt instanceof DiscoveryCustomEvent;
  
                  if (activateCluster())
                      assignPartitionsStates();

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
index 5f415e2,a164e85..2bb19cd
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsFullMessage.java
@@@ -32,6 -32,8 +32,7 @@@ import org.apache.ignite.internal.proce
  import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
  import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
  import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+ import org.apache.ignite.internal.util.typedef.F;
 -import org.apache.ignite.internal.util.typedef.T2;
  import org.apache.ignite.internal.util.typedef.internal.S;
  import org.apache.ignite.internal.util.typedef.internal.U;
  import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
index 3c11fa7,bc7d314..44815ca
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleMessage.java
@@@ -17,12 -17,14 +17,14 @@@
  
  package org.apache.ignite.internal.processors.cache.distributed.dht.preloader;
  
+ import java.util.Collection;
 -import java.util.Map;
 -import java.util.HashMap;
 +import java.io.Externalizable;
  import java.nio.ByteBuffer;
  import java.util.Collections;
 -import java.io.Externalizable;
 +import java.util.HashMap;
 +import java.util.Map;
  import org.apache.ignite.IgniteCheckedException;
+ import org.apache.ignite.internal.GridDirectCollection;
  import org.apache.ignite.internal.GridDirectMap;
  import org.apache.ignite.internal.GridDirectTransient;
  import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsSingleRequest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteDhtPartitionCountersMap.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteDhtPartitionCountersMap.java
index 75cfb37,dc2fbf8..e7954d9
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteDhtPartitionCountersMap.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/IgniteDhtPartitionCountersMap.java
@@@ -30,9 -32,16 +30,16 @@@ public class IgniteDhtPartitionCounters
      private static final long serialVersionUID = 0L;
  
      /** */
 -    private Map<Integer, Map<Integer, T2<Long, Long>>> map;
 +    private Map<Integer, CachePartitionFullCountersMap> map;
  
      /**
+      * @return {@code True} if map is empty.
+      */
+     public synchronized boolean empty() {
+         return map == null || map.isEmpty();
+     }
+ 
+     /**
       * @param cacheId Cache ID.
       * @param cntrMap Counters map.
       */

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/ignite/blob/30339d35/modules/core/src/main/java/org/apache/ignite/thread/IgniteThread.java
----------------------------------------------------------------------


Mime
View raw message