Update partition full map in parallel - Fixes
@@ -130,6 +130,7 @@ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYS
import static org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents.serverJoinEvent;
import static org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents.serverLeftEvent;
import static org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap.PARTIAL_COUNTERS_MAP_SINCE;
+import static org.apache.ignite.internal.util.IgniteUtils.doInParallel;
/**
* Future for exchanging partition maps.
@@ -3948,9 +3949,16 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
long time = System.currentTimeMillis();
- for (Map.Entry<Integer, GridDhtPartitionFullMap> entry : msg.partitions().entrySet()) {
- Integer grpId = entry.getKey();
+ int parallelismLvl = cctx.kernalContext().config().getSystemThreadPoolSize();
+
+ // Reserve at least 2 threads for system operations.
+ parallelismLvl = Math.max(1, parallelismLvl - 2);
+ try {
+ doInParallel(
+ parallelismLvl,
+ cctx.kernalContext().getSystemExecutorService(),
+ msg.partitions().keySet(), grpId -> {
CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
if (grp != null) {
@@ -3958,7 +3966,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
grp.topology().partitions());
grp.topology().update(resTopVer,
- entry.getValue(),
+ msg.partitions().get(grpId),
cntrMap,
msg.partsToReload(cctx.localNodeId(), grpId),
msg.partitionSizes(grpId),
@@ -3974,13 +3982,17 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
top.partitions());
top.update(resTopVer,
- entry.getValue(),
+ msg.partitions().get(grpId),
cntrMap,
Collections.emptySet(),
null,
null);
}
}
+ });
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
}
partitionsReceived = true;
