| /* |
| * 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.processors.cache; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.stream.Collectors; |
| import javax.cache.CacheException; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.IgniteSystemProperties; |
| import org.apache.ignite.cache.affinity.AffinityFunction; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.DataRegionConfiguration; |
| import org.apache.ignite.configuration.DataStorageConfiguration; |
| import org.apache.ignite.configuration.NearCacheConfiguration; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException; |
| import org.apache.ignite.internal.events.DiscoveryCustomEvent; |
| import org.apache.ignite.internal.managers.discovery.DiscoCache; |
| import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage; |
| import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener; |
| 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.distributed.dht.ClientCacheDhtTopologyFuture; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentResponse; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAssignmentFetchFuture; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CacheGroupAffinityMessage; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridClientPartitionTopology; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology; |
| import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState; |
| import org.apache.ignite.internal.util.GridConcurrentHashSet; |
| import org.apache.ignite.internal.processors.tracing.Span; |
| import org.apache.ignite.internal.util.GridLongList; |
| import org.apache.ignite.internal.util.GridPartitionStateMap; |
| import org.apache.ignite.internal.util.future.GridCompoundFuture; |
| import org.apache.ignite.internal.util.future.GridFinishedFuture; |
| import org.apache.ignite.internal.util.future.GridFutureAdapter; |
| import org.apache.ignite.internal.util.lang.GridPlainRunnable; |
| import org.apache.ignite.internal.util.lang.IgniteInClosureX; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.internal.CU; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteClosure; |
| import org.apache.ignite.lang.IgniteFuture; |
| import org.apache.ignite.lang.IgniteInClosure; |
| import org.apache.ignite.lang.IgniteUuid; |
| import org.jetbrains.annotations.Nullable; |
| |
| import static org.apache.ignite.cache.CacheMode.LOCAL; |
| import static org.apache.ignite.cache.CacheMode.PARTITIONED; |
| import static org.apache.ignite.cache.CacheRebalanceMode.NONE; |
| import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_JOINED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; |
| import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING; |
| import static org.apache.ignite.internal.processors.tracing.SpanType.AFFINITY_CALCULATION; |
| |
| /** |
| * |
| */ |
| @SuppressWarnings("ForLoopReplaceableByForEach") |
| public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdapter<K, V> { |
| /** */ |
| private final long clientCacheMsgTimeout = |
| IgniteSystemProperties.getLong(IgniteSystemProperties.IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT, 10_000); |
| |
| /** */ |
| private static final IgniteClosure<ClusterNode, UUID> NODE_TO_ID = new IgniteClosure<ClusterNode, UUID>() { |
| @Override public UUID apply(ClusterNode node) { |
| return node.id(); |
| } |
| }; |
| |
| /** */ |
| private static final IgniteClosure<ClusterNode, Long> NODE_TO_ORDER = new IgniteClosure<ClusterNode, Long>() { |
| @Override public Long apply(ClusterNode node) { |
| return node.order(); |
| } |
| }; |
| |
| /** Affinity information for all started caches (initialized on coordinator). */ |
| private ConcurrentMap<Integer, CacheGroupHolder> grpHolders = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private CacheMemoryOverheadValidator validator = new CacheMemoryOverheadValidator(); |
| |
| /** Topology version which requires affinity re-calculation (set from discovery thread). */ |
| private AffinityTopologyVersion lastAffVer; |
| |
| /** Registered caches (updated from exchange thread). */ |
| private CachesRegistry cachesRegistry; |
| |
| /** */ |
| private WaitRebalanceInfo waitInfo; |
| |
| /** */ |
| private final Object mux = new Object(); |
| |
| /** Pending affinity assignment futures. */ |
| private final ConcurrentMap<Long, GridDhtAssignmentFetchFuture> pendingAssignmentFetchFuts = |
| new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final ThreadLocal<ClientCacheChangeDiscoveryMessage> clientCacheChanges = new ThreadLocal<>(); |
| |
| /** Discovery listener. */ |
| private final GridLocalEventListener discoLsnr = new GridLocalEventListener() { |
| @Override public void onEvent(Event evt) { |
| DiscoveryEvent e = (DiscoveryEvent)evt; |
| |
| assert e.type() == EVT_NODE_LEFT || e.type() == EVT_NODE_FAILED; |
| |
| ClusterNode n = e.eventNode(); |
| |
| for (GridDhtAssignmentFetchFuture fut : pendingAssignmentFetchFuts.values()) |
| fut.onNodeLeft(n.id()); |
| } |
| }; |
| |
| /** {@inheritDoc} */ |
| @Override protected void start0() throws IgniteCheckedException { |
| super.start0(); |
| |
| cctx.kernalContext().event().addLocalEventListener(discoLsnr, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| |
| cachesRegistry = new CachesRegistry(cctx); |
| } |
| |
| /** |
| * Callback invoked from discovery thread when discovery message is received. |
| * |
| * @param type Event type. |
| * @param customMsg Custom message instance. |
| * @param node Event node. |
| * @param topVer Topology version. |
| * @param state Cluster state. |
| */ |
| void onDiscoveryEvent(int type, |
| @Nullable DiscoveryCustomMessage customMsg, |
| ClusterNode node, |
| AffinityTopologyVersion topVer, |
| DiscoveryDataClusterState state) { |
| if (type == EVT_NODE_JOINED && node.isLocal()) |
| lastAffVer = null; |
| |
| if ((state.transition() || !state.active()) && |
| !DiscoveryCustomEvent.requiresCentralizedAffinityAssignment(customMsg)) |
| return; |
| |
| if ((!node.isClient() && (type == EVT_NODE_FAILED || type == EVT_NODE_JOINED || type == EVT_NODE_LEFT)) || |
| DiscoveryCustomEvent.requiresCentralizedAffinityAssignment(customMsg)) { |
| synchronized (mux) { |
| assert lastAffVer == null || topVer.compareTo(lastAffVer) > 0 : |
| "lastAffVer=" + lastAffVer + ", topVer=" + topVer + ", customMsg=" + customMsg; |
| |
| lastAffVer = topVer; |
| } |
| } |
| } |
| |
| /** |
| * Must be called from exchange thread. |
| */ |
| public IgniteInternalFuture<?> initCachesOnLocalJoin( |
| Map<Integer, CacheGroupDescriptor> grpDescs, |
| Map<String, DynamicCacheDescriptor> cacheDescs |
| ) { |
| return cachesRegistry.init(grpDescs, cacheDescs); |
| } |
| |
| /** |
| * Callback invoked from discovery thread when discovery custom message is received. |
| * |
| * @param msg Customer message. |
| * @return {@code True} if minor topology version should be increased. |
| */ |
| boolean onCustomEvent(CacheAffinityChangeMessage msg) { |
| if (msg.exchangeId() != null) { |
| if (log.isDebugEnabled()) { |
| log.debug("Ignore affinity change message [lastAffVer=" + lastAffVer + |
| ", msgExchId=" + msg.exchangeId() + |
| ", msgVer=" + msg.topologyVersion() + ']'); |
| } |
| |
| return false; |
| } |
| |
| // Skip message if affinity was already recalculated. |
| boolean exchangeNeeded = lastAffVer == null || lastAffVer.equals(msg.topologyVersion()); |
| |
| msg.exchangeNeeded(exchangeNeeded); |
| |
| if (exchangeNeeded) { |
| if (log.isDebugEnabled()) { |
| log.debug("Need process affinity change message [lastAffVer=" + lastAffVer + |
| ", msgExchId=" + msg.exchangeId() + |
| ", msgVer=" + msg.topologyVersion() + ']'); |
| } |
| } |
| else { |
| if (log.isDebugEnabled()) { |
| log.debug("Ignore affinity change message [lastAffVer=" + lastAffVer + |
| ", msgExchId=" + msg.exchangeId() + |
| ", msgVer=" + msg.topologyVersion() + ']'); |
| } |
| } |
| |
| return exchangeNeeded; |
| } |
| |
| /** |
| * @param topVer Expected topology version. |
| */ |
| private void onCacheGroupStopped(AffinityTopologyVersion topVer) { |
| CacheAffinityChangeMessage msg = null; |
| |
| synchronized (mux) { |
| if (waitInfo == null || !waitInfo.topVer.equals(topVer)) |
| return; |
| |
| if (waitInfo.waitGrps.isEmpty()) { |
| msg = affinityChangeMessage(waitInfo); |
| |
| waitInfo = null; |
| } |
| } |
| |
| try { |
| if (msg != null) |
| cctx.discovery().sendCustomEvent(msg); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send affinity change message.", e); |
| } |
| } |
| |
| /** |
| * @param top Topology. |
| * @param checkGrpId Group ID. |
| */ |
| void checkRebalanceState(GridDhtPartitionTopology top, Integer checkGrpId) { |
| CacheAffinityChangeMessage msg = null; |
| |
| synchronized (mux) { |
| if (waitInfo == null || !waitInfo.topVer.equals(lastAffVer)) |
| return; |
| |
| Set<Integer> partWait = waitInfo.waitGrps.get(checkGrpId); |
| |
| boolean rebalanced = true; |
| |
| if (partWait != null) { |
| CacheGroupHolder grpHolder = grpHolders.get(checkGrpId); |
| |
| if (grpHolder != null) { |
| for (Iterator<Integer> it = partWait.iterator(); it.hasNext(); ) { |
| Integer part = it.next(); |
| |
| List<ClusterNode> owners = top.owners(part, waitInfo.topVer); |
| List<ClusterNode> ideal = waitInfo.assignments.get(checkGrpId).get(part); |
| |
| if (!owners.containsAll(ideal)) { |
| rebalanced = false; |
| |
| break; |
| } |
| else |
| it.remove(); |
| } |
| } |
| |
| if (rebalanced) { |
| waitInfo.waitGrps.remove(checkGrpId); |
| |
| if (waitInfo.waitGrps.isEmpty()) { |
| msg = affinityChangeMessage(waitInfo); |
| |
| waitInfo = null; |
| } |
| } |
| } |
| |
| try { |
| if (msg != null) |
| cctx.discovery().sendCustomEvent(msg); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send affinity change message.", e); |
| } |
| } |
| } |
| |
| /** |
| * @return Group IDs. |
| */ |
| public Set<Integer> waitGroups() { |
| synchronized (mux) { |
| if (waitInfo == null || !waitInfo.topVer.equals(lastAffVer)) |
| return Collections.emptySet(); |
| |
| return new HashSet<>(waitInfo.waitGrps.keySet()); |
| } |
| } |
| |
| /** |
| * @return {@code true} if rebalance expected. |
| */ |
| public boolean rebalanceRequired() { |
| synchronized (mux) { |
| return waitInfo != null; |
| } |
| } |
| |
| /** |
| * Adds group partition to wait list. |
| * <p> |
| * Late affinity switch will be triggered as soons as wait list becomes empty. |
| * |
| * @param grpId Group id. |
| * @param part Part. |
| * @param topVer Topology version. |
| * @param assignment Ideal assignment. |
| */ |
| public void addToWaitGroup(int grpId, int part, AffinityTopologyVersion topVer, List<ClusterNode> assignment) { |
| synchronized (mux) { |
| if (waitInfo == null) |
| waitInfo = new WaitRebalanceInfo(topVer); |
| |
| waitInfo.add(grpId, part, assignment); |
| } |
| } |
| |
| /** |
| * @param waitInfo Cache rebalance information. |
| * @return Message. |
| */ |
| @Nullable private CacheAffinityChangeMessage affinityChangeMessage(WaitRebalanceInfo waitInfo) { |
| if (waitInfo.assignments.isEmpty()) // Possible if all awaited caches were destroyed. |
| return null; |
| |
| return new CacheAffinityChangeMessage(waitInfo.topVer, waitInfo.deploymentIds); |
| } |
| |
| /** |
| * @param grp Cache group. |
| */ |
| void onCacheGroupCreated(CacheGroupContext grp) { |
| // no-op |
| } |
| |
| /** |
| * @param reqId Request ID. |
| * @param startReqs Client cache start request. |
| * @return Descriptors for caches to start. |
| */ |
| @Nullable private List<DynamicCacheDescriptor> clientCachesToStart( |
| UUID reqId, |
| Map<String, DynamicCacheChangeRequest> startReqs |
| ) { |
| List<DynamicCacheDescriptor> startDescs = new ArrayList<>(startReqs.size()); |
| |
| for (DynamicCacheChangeRequest startReq : startReqs.values()) { |
| DynamicCacheDescriptor desc = cachesRegistry.cache(CU.cacheId(startReq.cacheName())); |
| |
| if (desc == null) { |
| CacheException err = new CacheException("Failed to start client cache " + |
| "(a cache with the given name is not started): " + startReq.cacheName()); |
| |
| cctx.cache().completeClientCacheChangeFuture(reqId, err); |
| |
| return null; |
| } |
| |
| if (cctx.cacheContext(desc.cacheId()) != null) |
| continue; |
| |
| startDescs.add(desc); |
| } |
| |
| return startDescs; |
| } |
| |
| /** |
| * @param crd Coordinator flag. |
| * @param msg Change request. |
| * @param topVer Current topology version. |
| * @param discoCache Discovery data cache. |
| * @return Map of started caches (cache ID to near enabled flag). |
| */ |
| @Nullable private Map<Integer, Boolean> processClientCacheStartRequests( |
| boolean crd, |
| ClientCacheChangeDummyDiscoveryMessage msg, |
| AffinityTopologyVersion topVer, |
| DiscoCache discoCache |
| ) { |
| Map<String, DynamicCacheChangeRequest> startReqs = msg.startRequests(); |
| |
| List<DynamicCacheDescriptor> startDescs = clientCachesToStart(msg.requestId(), startReqs); |
| |
| if (startDescs == null || startDescs.isEmpty()) { |
| cctx.cache().completeClientCacheChangeFuture(msg.requestId(), null); |
| |
| return null; |
| } |
| |
| Map<Integer, GridDhtAssignmentFetchFuture> fetchFuts = U.newHashMap(startDescs.size()); |
| |
| Map<Integer, Boolean> startedInfos = U.newHashMap(startDescs.size()); |
| |
| List<StartCacheInfo> startCacheInfos = startDescs.stream() |
| .map(desc -> { |
| DynamicCacheChangeRequest changeReq = startReqs.get(desc.cacheName()); |
| |
| startedInfos.put(desc.cacheId(), changeReq.nearCacheConfiguration() != null); |
| |
| return new StartCacheInfo( |
| desc.cacheConfiguration(), |
| desc, |
| changeReq.nearCacheConfiguration(), |
| topVer, |
| changeReq.disabledAfterStart(), |
| true |
| ); |
| }).collect(Collectors.toList()); |
| |
| Set<String> startedCaches = startCacheInfos.stream() |
| .map(info -> info.getCacheDescriptor().cacheName()) |
| .collect(Collectors.toSet()); |
| |
| try { |
| cctx.cache().prepareStartCaches(startCacheInfos); |
| } |
| catch (IgniteCheckedException e) { |
| cctx.cache().closeCaches(startedCaches, false); |
| |
| cctx.cache().completeClientCacheChangeFuture(msg.requestId(), e); |
| |
| return null; |
| } |
| |
| Set<CacheGroupDescriptor> groupDescs = startDescs.stream() |
| .map(DynamicCacheDescriptor::groupDescriptor) |
| .collect(Collectors.toSet()); |
| |
| for (CacheGroupDescriptor grpDesc : groupDescs) { |
| try { |
| CacheGroupContext grp = cctx.cache().cacheGroup(grpDesc.groupId()); |
| |
| assert grp != null : grpDesc.groupId(); |
| assert !grp.affinityNode() || grp.isLocal() : grp.cacheOrGroupName(); |
| |
| // Skip for local caches. |
| if (grp.isLocal()) |
| continue; |
| |
| CacheGroupHolder grpHolder = grpHolders.get(grp.groupId()); |
| |
| assert !crd || (grpHolder != null && grpHolder.affinity().idealAssignmentRaw() != null); |
| |
| if (grpHolder == null) |
| grpHolder = getOrCreateGroupHolder(topVer, grpDesc); |
| |
| // If current node is not client and current node have no aff holder. |
| if (grpHolder.nonAffNode() && !cctx.localNode().isClient()) { |
| GridDhtPartitionsExchangeFuture excFut = context().exchange().lastFinishedFuture(); |
| |
| grp.topology().updateTopologyVersion(excFut, discoCache, -1, false); |
| |
| // Exchange free cache creation, just replacing client topology with dht. |
| // Topology should be initialized before the use. |
| grp.topology().beforeExchange(excFut, true, false); |
| |
| grpHolder = new CacheGroupAffNodeHolder(grp, grpHolder.affinity()); |
| |
| grpHolders.put(grp.groupId(), grpHolder); |
| |
| GridClientPartitionTopology clientTop = cctx.exchange().clearClientTopology(grp.groupId()); |
| |
| if (clientTop != null) { |
| grp.topology().update( |
| grpHolder.affinity().lastVersion(), |
| clientTop.partitionMap(true), |
| clientTop.fullUpdateCounters(), |
| Collections.<Integer>emptySet(), |
| null, |
| null, |
| null, |
| clientTop.lostPartitions()); |
| } |
| |
| assert grpHolder.affinity().lastVersion().equals(grp.affinity().lastVersion()); |
| } |
| else if (!crd && !fetchFuts.containsKey(grp.groupId())) { |
| boolean topVerLessOrNotInitialized = !grp.topology().initialized() || |
| grp.topology().readyTopologyVersion().compareTo(topVer) < 0; |
| |
| if (grp.affinity().lastVersion().compareTo(topVer) < 0 || topVerLessOrNotInitialized) { |
| GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx, |
| grp.groupId(), |
| topVer, |
| discoCache); |
| |
| fetchFut.init(true); |
| |
| fetchFuts.put(grp.groupId(), fetchFut); |
| } |
| } |
| } |
| catch (IgniteCheckedException e) { |
| cctx.cache().closeCaches(startedCaches, false); |
| |
| cctx.cache().completeClientCacheChangeFuture(msg.requestId(), e); |
| |
| return null; |
| } |
| } |
| |
| for (GridDhtAssignmentFetchFuture fetchFut : fetchFuts.values()) { |
| try { |
| CacheGroupContext grp = cctx.cache().cacheGroup(fetchFut.groupId()); |
| |
| assert grp != null; |
| |
| GridDhtAffinityAssignmentResponse res = fetchAffinity(topVer, |
| null, |
| discoCache, |
| grp.affinity(), |
| fetchFut); |
| |
| GridDhtPartitionFullMap partMap; |
| ClientCacheDhtTopologyFuture topFut; |
| |
| if (res != null) { |
| partMap = res.partitionMap(); |
| |
| assert partMap != null : res; |
| |
| topFut = new ClientCacheDhtTopologyFuture(topVer); |
| } |
| else { |
| partMap = new GridDhtPartitionFullMap(cctx.localNodeId(), cctx.localNode().order(), 1); |
| |
| topFut = new ClientCacheDhtTopologyFuture(topVer, |
| new ClusterTopologyServerNotFoundException("All server nodes left grid.")); |
| } |
| |
| grp.topology().updateTopologyVersion(topFut, |
| discoCache, |
| -1, |
| false); |
| |
| grp.topology().update(topVer, partMap, null, Collections.emptySet(), null, null, null, null); |
| |
| topFut.validate(grp, discoCache.allNodes()); |
| } |
| catch (IgniteCheckedException e) { |
| cctx.cache().closeCaches(startedCaches, false); |
| |
| cctx.cache().completeClientCacheChangeFuture(msg.requestId(), e); |
| |
| return null; |
| } |
| } |
| |
| for (DynamicCacheDescriptor desc : startDescs) { |
| if (desc.cacheConfiguration().getCacheMode() != LOCAL) { |
| CacheGroupContext grp = cctx.cache().cacheGroup(desc.groupId()); |
| |
| assert grp != null; |
| |
| grp.topology().onExchangeDone(null, grp.affinity().cachedAffinity(topVer), true); |
| } |
| } |
| |
| cctx.cache().initCacheProxies(topVer, null); |
| |
| startReqs.keySet().forEach(req -> cctx.cache().completeProxyInitialize(req)); |
| |
| cctx.cache().completeClientCacheChangeFuture(msg.requestId(), null); |
| |
| return startedInfos; |
| } |
| |
| /** |
| * @param msg Change request. |
| * @param topVer Current topology version. |
| * @param crd Coordinator flag. |
| * @return Closed caches IDs. |
| */ |
| private Set<Integer> processCacheCloseRequests( |
| ClientCacheChangeDummyDiscoveryMessage msg, |
| boolean crd, |
| AffinityTopologyVersion topVer |
| ) { |
| Set<String> cachesToClose = msg.cachesToClose(); |
| |
| Set<Integer> closed = cctx.cache().closeCaches(cachesToClose, true); |
| |
| for (CacheGroupHolder hld : grpHolders.values()) { |
| if (!hld.nonAffNode() && cctx.cache().cacheGroup(hld.groupId()) == null) { |
| int grpId = hld.groupId(); |
| |
| // All client cache groups were stopped, need create 'client' CacheGroupHolder. |
| CacheGroupHolder grpHolder = grpHolders.remove(grpId); |
| |
| assert grpHolder != null && !grpHolder.nonAffNode() : grpHolder; |
| |
| try { |
| grpHolder = createHolder( |
| cctx, |
| cachesRegistry.group(grpId), |
| topVer, |
| grpHolder.affinity() |
| ); |
| |
| grpHolders.put(grpId, grpHolder); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to initialize cache: " + e, e); |
| } |
| } |
| } |
| |
| cctx.cache().completeClientCacheChangeFuture(msg.requestId(), null); |
| |
| return closed; |
| } |
| |
| /** |
| * Process non affinity node cache start/close requests, called from exchange thread. |
| * |
| * @param msg Change request. |
| */ |
| void processClientCachesRequests(ClientCacheChangeDummyDiscoveryMessage msg) { |
| // Get ready exchange version. |
| AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion(); |
| |
| DiscoCache discoCache = cctx.discovery().discoCache(topVer); |
| |
| ClusterNode node = discoCache.oldestAliveServerNode(); |
| |
| // Resolve coordinator for specific version. |
| boolean crd = node != null && node.isLocal(); |
| |
| Map<Integer, Boolean> startedCaches = null; |
| Set<Integer> closedCaches = null; |
| |
| // Check and start caches via dummy message. |
| if (msg.startRequests() != null) |
| startedCaches = processClientCacheStartRequests(crd, msg, topVer, discoCache); |
| |
| // Check and close caches via dummy message. |
| if (msg.cachesToClose() != null) |
| closedCaches = processCacheCloseRequests(msg, crd, topVer); |
| |
| // Shedule change message. |
| if (startedCaches != null || closedCaches != null) |
| scheduleClientChangeMessage(startedCaches, closedCaches); |
| } |
| |
| /** |
| * Sends discovery message about started/closed client caches, called from exchange thread. |
| * |
| * @param timeoutObj Timeout object initiated send. |
| */ |
| void sendClientCacheChangesMessage(ClientCacheUpdateTimeout timeoutObj) { |
| ClientCacheChangeDiscoveryMessage msg = clientCacheChanges.get(); |
| |
| // Timeout object was changed if one more client cache changed during timeout, |
| // another timeoutObj was scheduled. |
| if (msg != null && msg.updateTimeoutObject() == timeoutObj) { |
| assert !msg.empty() : msg; |
| |
| clientCacheChanges.remove(); |
| |
| msg.checkCachesExist(cachesRegistry.allCaches().keySet()); |
| |
| try { |
| if (!msg.empty()) |
| cctx.discovery().sendCustomEvent(msg); |
| } |
| catch (IgniteCheckedException e) { |
| U.error(log, "Failed to send discovery event: " + e, e); |
| } |
| } |
| } |
| |
| /** |
| * @param startedCaches Started caches. |
| * @param closedCaches Closed caches. |
| */ |
| private void scheduleClientChangeMessage(Map<Integer, Boolean> startedCaches, Set<Integer> closedCaches) { |
| ClientCacheChangeDiscoveryMessage msg = clientCacheChanges.get(); |
| |
| if (msg == null) { |
| msg = new ClientCacheChangeDiscoveryMessage(startedCaches, closedCaches); |
| |
| clientCacheChanges.set(msg); |
| } |
| else { |
| msg.merge(startedCaches, closedCaches); |
| |
| if (msg.empty()) { |
| cctx.time().removeTimeoutObject(msg.updateTimeoutObject()); |
| |
| clientCacheChanges.remove(); |
| |
| return; |
| } |
| } |
| |
| if (msg.updateTimeoutObject() != null) |
| cctx.time().removeTimeoutObject(msg.updateTimeoutObject()); |
| |
| long timeout = clientCacheMsgTimeout; |
| |
| if (timeout <= 0) |
| timeout = 10_000; |
| |
| ClientCacheUpdateTimeout timeoutObj = new ClientCacheUpdateTimeout(cctx, timeout); |
| |
| msg.updateTimeoutObject(timeoutObj); |
| |
| cctx.time().addTimeoutObject(timeoutObj); |
| } |
| |
| /** |
| * @param fut Exchange future. |
| * @param exchActions Exchange actions. |
| */ |
| public void onCustomMessageNoAffinityChange( |
| GridDhtPartitionsExchangeFuture fut, |
| @Nullable final ExchangeActions exchActions |
| ) { |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| forAllCacheGroups(new IgniteInClosureX<GridAffinityAssignmentCache>() { |
| @Override public void applyx(GridAffinityAssignmentCache aff) { |
| if (exchActions != null && exchActions.cacheGroupStopping(aff.groupId())) |
| return; |
| |
| aff.clientEventTopologyChange(evts.lastEvent(), evts.topologyVersion()); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| } |
| }); |
| } |
| |
| /** |
| * @param cctx Stopped cache context. |
| */ |
| public void stopCacheOnReconnect(GridCacheContext cctx) { |
| cachesRegistry.unregisterCache(cctx.cacheId()); |
| } |
| |
| /** |
| * @param grpCtx Stopped cache group context. |
| */ |
| public void stopCacheGroupOnReconnect(CacheGroupContext grpCtx) { |
| cachesRegistry.unregisterGroup(grpCtx.groupId()); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onDisconnected(IgniteFuture<?> reconnectFut) { |
| Iterator<Integer> it = grpHolders.keySet().iterator(); |
| |
| while (it.hasNext()) { |
| int grpId = it.next(); |
| |
| it.remove(); |
| |
| cctx.io().removeHandler(true, grpId, GridDhtAffinityAssignmentResponse.class); |
| } |
| |
| assert grpHolders.isEmpty(); |
| |
| super.onDisconnected(reconnectFut); |
| } |
| |
| /** |
| * Called during the rollback of the exchange partitions procedure in order to stop the given cache even if it's not |
| * fully initialized (e.g. failed on cache init stage). |
| * |
| * @param fut Exchange future. |
| * @param crd Coordinator flag. |
| * @param exchActions Cache change requests. |
| */ |
| public void forceCloseCaches( |
| GridDhtPartitionsExchangeFuture fut, |
| boolean crd, |
| final ExchangeActions exchActions |
| ) { |
| assert exchActions != null && !exchActions.empty() && exchActions.cacheStartRequests().isEmpty() : exchActions; |
| |
| IgniteInternalFuture<?> res = cachesRegistry.update(exchActions); |
| |
| assert res.isDone() : "There should be no caches to start: " + exchActions; |
| |
| processCacheStopRequests(fut, crd, exchActions, true); |
| |
| cctx.cache().forceCloseCaches(exchActions); |
| } |
| |
| /** |
| * Called on exchange initiated for cache start/stop request. |
| * |
| * @param fut Exchange future. |
| * @param crd Coordinator flag. |
| * @param exchActions Cache change requests. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public IgniteInternalFuture<?> onCacheChangeRequest( |
| GridDhtPartitionsExchangeFuture fut, |
| boolean crd, |
| final ExchangeActions exchActions |
| ) throws IgniteCheckedException { |
| assert exchActions != null && !exchActions.empty() : exchActions; |
| |
| IgniteInternalFuture<?> res = cachesRegistry.update(exchActions); |
| |
| // Affinity did not change for existing caches. |
| onCustomMessageNoAffinityChange(fut, exchActions); |
| |
| fut.timeBag().finishGlobalStage("Update caches registry"); |
| |
| processCacheStartRequests(fut, crd, exchActions); |
| |
| Set<Integer> stoppedGrps = processCacheStopRequests(fut, crd, exchActions, false); |
| |
| if (stoppedGrps != null) { |
| AffinityTopologyVersion notifyTopVer = null; |
| |
| synchronized (mux) { |
| if (waitInfo != null) { |
| for (Integer grpId : stoppedGrps) { |
| boolean rmv = waitInfo.waitGrps.remove(grpId) != null; |
| |
| if (rmv) { |
| notifyTopVer = waitInfo.topVer; |
| |
| waitInfo.assignments.remove(grpId); |
| } |
| } |
| } |
| } |
| |
| if (notifyTopVer != null) { |
| final AffinityTopologyVersion topVer = notifyTopVer; |
| |
| cctx.kernalContext().closure().runLocalSafe(new GridPlainRunnable() { |
| @Override public void run() { |
| onCacheGroupStopped(topVer); |
| } |
| }); |
| } |
| } |
| |
| ClientCacheChangeDiscoveryMessage msg = clientCacheChanges.get(); |
| |
| if (msg != null) { |
| msg.checkCachesExist(cachesRegistry.allCaches().keySet()); |
| |
| if (msg.empty()) |
| clientCacheChanges.remove(); |
| } |
| |
| return res; |
| } |
| |
| /** |
| * Process cache start requests. |
| * |
| * @param fut Exchange future. |
| * @param crd Coordinator flag. |
| * @param exchActions Cache change requests. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private void processCacheStartRequests( |
| GridDhtPartitionsExchangeFuture fut, |
| boolean crd, |
| final ExchangeActions exchActions |
| ) throws IgniteCheckedException { |
| assert exchActions != null && !exchActions.empty() : exchActions; |
| |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| Map<StartCacheInfo, DynamicCacheChangeRequest> startCacheInfos = new LinkedHashMap<>(); |
| |
| for (ExchangeActions.CacheActionData action : exchActions.cacheStartRequests()) { |
| DynamicCacheDescriptor cacheDesc = action.descriptor(); |
| |
| DynamicCacheChangeRequest req = action.request(); |
| |
| boolean startCache; |
| |
| NearCacheConfiguration nearCfg = null; |
| |
| if (req.locallyConfigured() || (cctx.localNodeId().equals(req.initiatingNodeId()) && !exchActions.activate())) { |
| startCache = true; |
| |
| nearCfg = req.nearCacheConfiguration(); |
| } |
| else { |
| // Cache should not be started |
| assert cctx.cacheContext(cacheDesc.cacheId()) == null |
| : "Starting cache has not null context: " + cacheDesc.cacheName(); |
| |
| IgniteCacheProxyImpl cacheProxy = cctx.cache().jcacheProxy(req.cacheName(), false); |
| |
| // If it has proxy then try to start it |
| if (cacheProxy != null) { |
| // Cache should be in restarting mode |
| assert cacheProxy.isRestarting() |
| : "Cache has non restarting proxy " + cacheProxy; |
| |
| startCache = true; |
| } |
| else { |
| startCache = CU.affinityNode(cctx.localNode(), |
| cacheDesc.groupDescriptor().config().getNodeFilter()); |
| } |
| } |
| |
| if (startCache) { |
| startCacheInfos.put( |
| new StartCacheInfo( |
| req.startCacheConfiguration(), |
| cacheDesc, |
| nearCfg, |
| evts.topologyVersion(), |
| req.disabledAfterStart() |
| ), |
| req |
| ); |
| } |
| else |
| cctx.kernalContext().query().initQueryStructuresForNotStartedCache(cacheDesc); |
| } |
| |
| Map<StartCacheInfo, IgniteCheckedException> failedCaches = cctx.cache().prepareStartCachesIfPossible(startCacheInfos.keySet()); |
| |
| for (Map.Entry<StartCacheInfo, IgniteCheckedException> entry : failedCaches.entrySet()) { |
| if (cctx.localNode().isClient()) { |
| U.error(log, "Failed to initialize cache. Will try to rollback cache start routine. " + |
| "[cacheName=" + entry.getKey().getStartedConfiguration().getName() + ']', entry.getValue()); |
| |
| cctx.cache().closeCaches(Collections.singleton(entry.getKey().getStartedConfiguration().getName()), false); |
| |
| cctx.cache().completeCacheStartFuture(startCacheInfos.get(entry.getKey()), false, entry.getValue()); |
| } |
| else |
| throw entry.getValue(); |
| } |
| |
| Set<StartCacheInfo> failedCacheInfos = failedCaches.keySet(); |
| |
| List<StartCacheInfo> cacheInfos = startCacheInfos.keySet().stream() |
| .filter(failedCacheInfos::contains) |
| .collect(Collectors.toList()); |
| |
| for (StartCacheInfo info : cacheInfos) { |
| if (fut.cacheAddedOnExchange(info.getCacheDescriptor().cacheId(), info.getCacheDescriptor().receivedFrom())) { |
| if (fut.events().discoveryCache().cacheGroupAffinityNodes(info.getCacheDescriptor().groupId()).isEmpty()) |
| U.quietAndWarn(log, "No server nodes found for cache client: " + info.getCacheDescriptor().cacheName()); |
| } |
| } |
| |
| fut.timeBag().finishGlobalStage("Start caches"); |
| |
| initAffinityOnCacheGroupsStart(fut, exchActions, crd); |
| |
| fut.timeBag().finishGlobalStage("Affinity initialization on cache group start"); |
| } |
| |
| /** |
| * Initializes affinity for started cache groups received during {@code fut}. |
| * |
| * @param fut Exchange future. |
| * @param exchangeActions Exchange actions. |
| * @param crd {@code True} if local node is coordinator. |
| */ |
| private void initAffinityOnCacheGroupsStart( |
| GridDhtPartitionsExchangeFuture fut, |
| ExchangeActions exchangeActions, |
| boolean crd |
| ) throws IgniteCheckedException { |
| List<CacheGroupDescriptor> startedGroups = exchangeActions.cacheStartRequests().stream() |
| .map(action -> action.descriptor().groupDescriptor()) |
| .distinct() |
| .collect(Collectors.toList()); |
| |
| U.doInParallel( |
| cctx.kernalContext().getSystemExecutorService(), |
| startedGroups, |
| grpDesc -> { |
| initStartedGroup(fut, grpDesc, crd); |
| |
| fut.timeBag().finishLocalStage("Affinity initialization on cache group start " + |
| "[grp=" + grpDesc.cacheOrGroupName() + "]"); |
| |
| validator.validateCacheGroup(grpDesc); |
| |
| return null; |
| } |
| ); |
| } |
| |
| /** |
| * Process cache stop requests. |
| * |
| * @param fut Exchange future. |
| * @param crd Coordinator flag. |
| * @param exchActions Cache change requests. |
| * @param forceClose Force close flag. |
| * @return Set of cache groups to be stopped. |
| */ |
| private Set<Integer> processCacheStopRequests( |
| GridDhtPartitionsExchangeFuture fut, |
| boolean crd, |
| final ExchangeActions exchActions, |
| boolean forceClose |
| ) { |
| assert exchActions != null && !exchActions.empty() : exchActions; |
| |
| for (ExchangeActions.CacheActionData action : exchActions.cacheStopRequests()) |
| cctx.cache().blockGateway(action.request().cacheName(), true, action.request().restart()); |
| |
| for (ExchangeActions.CacheGroupActionData action : exchActions.cacheGroupsToStop()) |
| cctx.exchange().clearClientTopology(action.descriptor().groupId()); |
| |
| Set<Integer> stoppedGrps = null; |
| |
| for (ExchangeActions.CacheGroupActionData data : exchActions.cacheGroupsToStop()) { |
| if (data.descriptor().config().getCacheMode() != LOCAL) { |
| CacheGroupHolder cacheGrp = grpHolders.remove(data.descriptor().groupId()); |
| |
| assert !crd || (cacheGrp != null || forceClose) : data.descriptor(); |
| |
| if (cacheGrp != null) { |
| if (stoppedGrps == null) |
| stoppedGrps = new HashSet<>(); |
| |
| stoppedGrps.add(cacheGrp.groupId()); |
| |
| cctx.io().removeHandler(true, cacheGrp.groupId(), GridDhtAffinityAssignmentResponse.class); |
| } |
| } |
| } |
| |
| return stoppedGrps; |
| } |
| |
| /** |
| * |
| */ |
| public void clearGroupHoldersAndRegistry() { |
| grpHolders.clear(); |
| |
| cachesRegistry.unregisterAll(); |
| } |
| |
| /** |
| * Called when received {@link CacheAffinityChangeMessage} which should complete exchange. |
| * |
| * @param exchFut Exchange future. |
| * @param msg Affinity change message. |
| */ |
| public void onExchangeChangeAffinityMessage( |
| GridDhtPartitionsExchangeFuture exchFut, |
| CacheAffinityChangeMessage msg |
| ) { |
| if (log.isDebugEnabled()) { |
| log.debug("Process exchange affinity change message [exchVer=" + exchFut.initialVersion() + |
| ", msg=" + msg + ']'); |
| } |
| |
| assert exchFut.exchangeId().equals(msg.exchangeId()) : msg; |
| |
| final AffinityTopologyVersion topVer = exchFut.initialVersion(); |
| |
| final Map<Integer, Map<Integer, List<UUID>>> assignment = msg.assignmentChange(); |
| |
| assert assignment != null; |
| |
| forAllCacheGroups(new IgniteInClosureX<GridAffinityAssignmentCache>() { |
| @Override public void applyx(GridAffinityAssignmentCache aff) { |
| List<List<ClusterNode>> idealAssignment = aff.idealAssignmentRaw(); |
| |
| assert idealAssignment != null; |
| |
| Map<Integer, List<UUID>> cacheAssignment = assignment.get(aff.groupId()); |
| |
| List<List<ClusterNode>> newAssignment; |
| |
| if (cacheAssignment != null) { |
| newAssignment = new ArrayList<>(idealAssignment); |
| |
| for (Map.Entry<Integer, List<UUID>> e : cacheAssignment.entrySet()) |
| newAssignment.set(e.getKey(), toNodes(topVer, e.getValue())); |
| } |
| else |
| newAssignment = idealAssignment; |
| |
| aff.initialize(topVer, newAssignment); |
| |
| exchFut.timeBag().finishLocalStage("Affinity recalculate by change affinity message " + |
| "[grp=" + aff.cacheOrGroupName() + "]"); |
| } |
| }); |
| } |
| |
| /** |
| * Called on exchange initiated by {@link CacheAffinityChangeMessage} which sent after rebalance finished. |
| * |
| * @param exchFut Exchange future. |
| * @param msg Message. |
| */ |
| public void onChangeAffinityMessage( |
| final GridDhtPartitionsExchangeFuture exchFut, |
| final CacheAffinityChangeMessage msg |
| ) { |
| assert msg.topologyVersion() != null && msg.exchangeId() == null : msg; |
| assert msg.partitionsMessage() == null : msg; |
| assert msg.assignmentChange() == null : msg; |
| |
| final AffinityTopologyVersion topVer = exchFut.initialVersion(); |
| |
| if (log.isDebugEnabled()) { |
| log.debug("Process affinity change message [exchVer=" + topVer + |
| ", msgVer=" + msg.topologyVersion() + ']'); |
| } |
| |
| final Map<Integer, IgniteUuid> deploymentIds = msg.cacheDeploymentIds(); |
| |
| forAllCacheGroups(new IgniteInClosureX<GridAffinityAssignmentCache>() { |
| @Override public void applyx(GridAffinityAssignmentCache aff) { |
| AffinityTopologyVersion affTopVer = aff.lastVersion(); |
| |
| assert affTopVer.topologyVersion() > 0 : affTopVer; |
| |
| CacheGroupDescriptor desc = cachesRegistry.group(aff.groupId()); |
| |
| assert desc != null : aff.cacheOrGroupName(); |
| |
| IgniteUuid deploymentId = desc.deploymentId(); |
| |
| if (!deploymentId.equals(deploymentIds.get(aff.groupId()))) { |
| aff.clientEventTopologyChange(exchFut.firstEvent(), topVer); |
| |
| return; |
| } |
| |
| if (!aff.partitionPrimariesDifferentToIdeal(affTopVer).isEmpty()) |
| aff.initialize(topVer, aff.idealAssignmentRaw()); |
| else { |
| if (!aff.assignments(aff.lastVersion()).equals(aff.idealAssignmentRaw())) |
| // This should never happen on Late Affinity Assignment switch and must trigger Failure Handler. |
| throw new AssertionError("Not an ideal distribution duplication attempt on LAA " + |
| "[grp=" + aff.cacheOrGroupName() + ", lastAffinity=" + aff.lastVersion() + |
| ", cacheAffinity=" + aff.cachedVersions() + "]"); |
| |
| aff.clientEventTopologyChange(exchFut.firstEvent(), topVer); |
| } |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| |
| exchFut.timeBag().finishLocalStage("Affinity change by custom message " + |
| "[grp=" + aff.cacheOrGroupName() + "]"); |
| } |
| }); |
| } |
| |
| /** |
| * Called on exchange initiated by client node join/fail. |
| * |
| * @param fut Exchange future. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public void onClientEvent(final GridDhtPartitionsExchangeFuture fut) throws IgniteCheckedException { |
| boolean locJoin = fut.firstEvent().eventNode().isLocal(); |
| |
| if (!locJoin) { |
| forAllCacheGroups(new IgniteInClosureX<GridAffinityAssignmentCache>() { |
| @Override public void applyx(GridAffinityAssignmentCache aff) throws IgniteCheckedException { |
| AffinityTopologyVersion topVer = fut.initialVersion(); |
| |
| aff.clientEventTopologyChange(fut.firstEvent(), topVer); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| } |
| }); |
| } |
| else |
| fetchAffinityOnJoin(fut); |
| } |
| |
| /** |
| * @param fut Future to add. |
| */ |
| public void addDhtAssignmentFetchFuture(GridDhtAssignmentFetchFuture fut) { |
| GridDhtAssignmentFetchFuture old = pendingAssignmentFetchFuts.putIfAbsent(fut.id(), fut); |
| |
| assert old == null : "More than one thread is trying to fetch partition assignments [fut=" + fut + |
| ", allFuts=" + pendingAssignmentFetchFuts + ']'; |
| } |
| |
| /** |
| * @param fut Future to remove. |
| */ |
| public void removeDhtAssignmentFetchFuture(GridDhtAssignmentFetchFuture fut) { |
| boolean rmv = pendingAssignmentFetchFuts.remove(fut.id(), fut); |
| |
| assert rmv : "Failed to remove assignment fetch future: " + fut.id(); |
| } |
| |
| /** |
| * @param nodeId Node ID. |
| * @param res Response. |
| */ |
| private void processAffinityAssignmentResponse(UUID nodeId, GridDhtAffinityAssignmentResponse res) { |
| if (log.isDebugEnabled()) |
| log.debug("Processing affinity assignment response [node=" + nodeId + ", res=" + res + ']'); |
| |
| GridDhtAssignmentFetchFuture fut = pendingAssignmentFetchFuts.get(res.futureId()); |
| |
| if (fut != null) |
| fut.onResponse(nodeId, res); |
| } |
| |
| /** |
| * @param c Cache closure. |
| */ |
| private void forAllRegisteredCacheGroups(IgniteInClosureX<CacheGroupDescriptor> c) { |
| Collection<CacheGroupDescriptor> affinityCaches = cachesRegistry.allGroups().values().stream() |
| .filter(desc -> desc.config().getCacheMode() != LOCAL) |
| .collect(Collectors.toList()); |
| |
| try { |
| U.doInParallel(cctx.kernalContext().getSystemExecutorService(), affinityCaches, t -> { |
| c.applyx(t); |
| |
| return null; |
| }); |
| } |
| catch (IgniteCheckedException e) { |
| throw new IgniteException("Failed to execute affinity operation on cache groups", e); |
| } |
| } |
| |
| /** |
| * @param c Closure. |
| */ |
| private void forAllCacheGroups(IgniteInClosureX<GridAffinityAssignmentCache> c) { |
| Collection<GridAffinityAssignmentCache> affinityCaches = grpHolders.values().stream() |
| .map(CacheGroupHolder::affinity) |
| .collect(Collectors.toList()); |
| |
| try { |
| U.doInParallel(cctx.kernalContext().getSystemExecutorService(), affinityCaches, t -> { |
| c.applyx(t); |
| |
| return null; |
| }); |
| } |
| catch (IgniteCheckedException e) { |
| throw new IgniteException("Failed to execute affinity operation on cache groups", e); |
| } |
| } |
| |
| /** |
| * @param fut Exchange future. |
| * @param grpDesc Cache group descriptor. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private void initStartedGroup(GridDhtPartitionsExchangeFuture fut, final CacheGroupDescriptor grpDesc, boolean crd) |
| throws IgniteCheckedException { |
| assert grpDesc != null && grpDesc.groupId() != 0 : grpDesc; |
| |
| if (grpDesc.config().getCacheMode() == LOCAL) |
| return; |
| |
| int grpId = grpDesc.groupId(); |
| |
| CacheGroupHolder grpHolder = grpHolders.get(grpId); |
| |
| CacheGroupContext grp = cctx.kernalContext().cache().cacheGroup(grpId); |
| |
| if (grpHolder != null && grpHolder.nonAffNode() && grp != null) { |
| assert grpHolder.affinity().idealAssignmentRaw() != null; |
| |
| grpHolder = new CacheGroupAffNodeHolder(grp, grpHolder.affinity()); |
| |
| grpHolders.put(grpId, grpHolder); |
| } |
| else if (grpHolder == null) { |
| grpHolder = getOrCreateGroupHolder(fut.initialVersion(), grpDesc); |
| |
| calculateAndInit(fut.events(), grpHolder.affinity(), fut.initialVersion()); |
| } |
| else if (!crd && grp != null && grp.localStartVersion().equals(fut.initialVersion())) |
| initAffinity(cachesRegistry.group(grp.groupId()), grp.affinity(), fut); |
| } |
| |
| /** |
| * Initialized affinity for cache received from node joining on this exchange. |
| * |
| * @param crd Coordinator flag. |
| * @param fut Exchange future. |
| * @param descs Cache descriptors. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public IgniteInternalFuture<?> initStartedCaches( |
| boolean crd, |
| final GridDhtPartitionsExchangeFuture fut, |
| Collection<DynamicCacheDescriptor> descs |
| ) throws IgniteCheckedException { |
| IgniteInternalFuture<?> res = cachesRegistry.addUnregistered(descs); |
| |
| if (fut.context().mergeExchanges()) |
| return res; |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| CacheGroupHolder cache = getOrCreateGroupHolder(fut.initialVersion(), desc); |
| |
| if (cache.affinity().lastVersion().equals(AffinityTopologyVersion.NONE)) { |
| initAffinity(desc, cache.affinity(), fut); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| |
| fut.timeBag().finishLocalStage("Affinity initialization (new cache) " + |
| "[grp=" + desc.cacheOrGroupName() + ", crd=" + crd + "]"); |
| } |
| } |
| }); |
| |
| return res; |
| } |
| |
| /** |
| * @param desc Cache group descriptor. |
| * @param aff Affinity. |
| * @param fut Exchange future. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private void initAffinity(CacheGroupDescriptor desc, |
| GridAffinityAssignmentCache aff, |
| GridDhtPartitionsExchangeFuture fut) |
| throws IgniteCheckedException { |
| assert desc != null : aff.cacheOrGroupName(); |
| |
| ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| if (canCalculateAffinity(desc, aff, fut)) |
| calculateAndInit(evts, aff, evts.topologyVersion()); |
| else { |
| GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx, |
| desc.groupId(), |
| evts.topologyVersion(), |
| evts.discoveryCache()); |
| |
| fetchFut.init(false); |
| |
| fetchAffinity(evts.topologyVersion(), |
| evts, |
| evts.discoveryCache(), |
| aff, |
| fetchFut); |
| } |
| } |
| |
| /** |
| * @param desc Cache group descriptor. |
| * @param aff Affinity. |
| * @param fut Exchange future. |
| * @return {@code True} if local node can calculate affinity on it's own for this partition map exchange. |
| */ |
| private boolean canCalculateAffinity(CacheGroupDescriptor desc, |
| GridAffinityAssignmentCache aff, |
| GridDhtPartitionsExchangeFuture fut) { |
| assert desc != null : aff.cacheOrGroupName(); |
| |
| // Do not request affinity from remote nodes if affinity function is not centralized. |
| if (!aff.centralizedAffinityFunction()) |
| return true; |
| |
| // If local node did not initiate exchange or local node is the only cache node in grid. |
| Collection<ClusterNode> affNodes = fut.events().discoveryCache().cacheGroupAffinityNodes(aff.groupId()); |
| |
| return fut.cacheGroupAddedOnExchange(aff.groupId(), desc.receivedFrom()) || |
| !fut.exchangeId().nodeId().equals(cctx.localNodeId()) || |
| (affNodes.isEmpty() || (affNodes.size() == 1 && affNodes.contains(cctx.localNode()))); |
| } |
| |
| /** |
| * @param grpId Cache group ID. |
| * @return Affinity assignments. |
| */ |
| public GridAffinityAssignmentCache affinity(Integer grpId) { |
| CacheGroupHolder grpHolder = grpHolders.get(grpId); |
| |
| assert grpHolder != null : debugGroupName(grpId); |
| |
| return grpHolder.affinity(); |
| } |
| |
| /** |
| * Applies affinity diff from the received full message. |
| * |
| * @param fut Current exchange future. |
| * @param idealAffDiff Map [Cache group id - Affinity distribution] which contains difference with ideal affinity. |
| */ |
| public void applyAffinityFromFullMessage( |
| final GridDhtPartitionsExchangeFuture fut, |
| final Map<Integer, CacheGroupAffinityMessage> idealAffDiff |
| ) { |
| // Please do not use following pattern of code (nodesByOrder, affCache). NEVER. |
| final Map<Long, ClusterNode> nodesByOrder = new ConcurrentHashMap<>(); |
| |
| forAllCacheGroups(new IgniteInClosureX<GridAffinityAssignmentCache>() { |
| @Override public void applyx(GridAffinityAssignmentCache aff) { |
| ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| List<List<ClusterNode>> idealAssignment = aff.calculate(evts.topologyVersion(), evts, evts.discoveryCache()).assignment(); |
| |
| CacheGroupAffinityMessage affMsg = idealAffDiff != null ? idealAffDiff.get(aff.groupId()) : null; |
| |
| List<List<ClusterNode>> newAssignment; |
| |
| if (affMsg != null) { |
| Map<Integer, GridLongList> diff = affMsg.assignmentsDiff(); |
| |
| assert !F.isEmpty(diff); |
| |
| newAssignment = new ArrayList<>(idealAssignment); |
| |
| for (Map.Entry<Integer, GridLongList> e : diff.entrySet()) { |
| GridLongList assign = e.getValue(); |
| |
| newAssignment.set(e.getKey(), CacheGroupAffinityMessage.toNodes(assign, |
| nodesByOrder, |
| evts.discoveryCache())); |
| } |
| } |
| else |
| newAssignment = idealAssignment; |
| |
| aff.initialize(evts.topologyVersion(), newAssignment); |
| |
| fut.timeBag().finishLocalStage("Affinity applying from full message " + |
| "[grp=" + aff.cacheOrGroupName() + "]"); |
| } |
| }); |
| } |
| |
| /** |
| * @param fut Current exchange future. |
| * @param receivedAff Map [Cache group id - Affinity distribution] received from coordinator to apply. |
| * @param resTopVer Result topology version. |
| * @return Set of cache groups with no affinity localed in given {@code receivedAff}. |
| */ |
| public Set<Integer> onLocalJoin( |
| final GridDhtPartitionsExchangeFuture fut, |
| final Map<Integer, CacheGroupAffinityMessage> receivedAff, |
| final AffinityTopologyVersion resTopVer |
| ) { |
| final Set<Integer> affReq = fut.context().groupsAffinityRequestOnJoin(); |
| |
| final Map<Long, ClusterNode> nodesByOrder = new ConcurrentHashMap<>(); |
| |
| // Such cache group may exist if cache is already destroyed on server nodes |
| // and coordinator have no affinity for that group. |
| final Set<Integer> noAffinityGroups = new GridConcurrentHashSet<>(); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| CacheGroupHolder holder = getOrCreateGroupHolder(fut.initialVersion(), desc); |
| |
| GridAffinityAssignmentCache aff = holder.affinity(); |
| |
| CacheGroupContext grp = cctx.cache().cacheGroup(holder.groupId()); |
| |
| if (affReq != null && affReq.contains(aff.groupId())) { |
| assert resTopVer.compareTo(aff.lastVersion()) >= 0 : aff.lastVersion(); |
| |
| CacheGroupAffinityMessage affMsg = receivedAff.get(aff.groupId()); |
| |
| if (affMsg == null) { |
| noAffinityGroups.add(aff.groupId()); |
| |
| // Use ideal affinity to resume cache initialize process. |
| calculateAndInit(evts, aff, evts.topologyVersion()); |
| |
| return; |
| } |
| |
| List<List<ClusterNode>> assignments = affMsg.createAssignments(nodesByOrder, evts.discoveryCache()); |
| |
| assert resTopVer.equals(evts.topologyVersion()) : "resTopVer=" + resTopVer + |
| ", evts.topVer=" + evts.topologyVersion(); |
| |
| List<List<ClusterNode>> idealAssign = |
| affMsg.createIdealAssignments(nodesByOrder, evts.discoveryCache()); |
| |
| if (idealAssign != null) |
| aff.idealAssignment(evts.topologyVersion(), idealAssign); |
| else { |
| assert !aff.centralizedAffinityFunction() : aff; |
| |
| // Calculate ideal assignments. |
| aff.calculate(evts.topologyVersion(), evts, evts.discoveryCache()); |
| } |
| |
| aff.initialize(evts.topologyVersion(), assignments); |
| } |
| else if (grp != null && fut.cacheGroupAddedOnExchange(aff.groupId(), grp.receivedFrom())) |
| calculateAndInit(evts, aff, evts.topologyVersion()); |
| |
| if (grp != null) |
| grp.topology().initPartitionsWhenAffinityReady(resTopVer, fut); |
| |
| fut.timeBag().finishLocalStage("Affinity initialization (local join) " + |
| "[grp=" + aff.cacheOrGroupName() + "]"); |
| } |
| }); |
| |
| return noAffinityGroups; |
| } |
| |
| /** |
| * @param fut Current exchange future. |
| * @param crd Coordinator flag. |
| */ |
| public void onServerJoinWithExchangeMergeProtocol(GridDhtPartitionsExchangeFuture fut, boolean crd) { |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| assert fut.context().mergeExchanges(); |
| assert evts.hasServerJoin() && !evts.hasServerLeft(); |
| |
| initAffinityOnNodeJoin(fut, crd); |
| } |
| |
| /** |
| * @param fut Current exchange future. |
| * @return Computed difference with ideal affinity. |
| */ |
| public Map<Integer, CacheGroupAffinityMessage> onServerLeftWithExchangeMergeProtocol( |
| final GridDhtPartitionsExchangeFuture fut |
| ) { |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| assert fut.context().mergeExchanges(); |
| assert evts.hasServerLeft(); |
| |
| return onReassignmentEnforced(fut); |
| } |
| |
| /** |
| * Called on exchange initiated by baseline server node leave on fully-rebalanced topology. |
| * |
| * @param fut Exchange future. |
| */ |
| public void onExchangeFreeSwitch(final GridDhtPartitionsExchangeFuture fut) { |
| assert (fut.events().hasServerLeft() && !fut.firstEvent().eventNode().isClient()) : fut.firstEvent(); |
| assert !fut.context().mergeExchanges(); |
| |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| AffinityTopologyVersion topVer = evts.topologyVersion(); |
| |
| CacheGroupHolder cache = getOrCreateGroupHolder(topVer, desc); |
| |
| calculateAndInit(evts, cache.affinity(), topVer); // Fully rebalanced. Initializing ideal assignment. |
| |
| fut.timeBag().finishLocalStage( |
| "Affinity initialization (exchange-free switch on fully-rebalanced topology) " + |
| "[grp=" + desc.cacheOrGroupName() + "]"); |
| } |
| }); |
| |
| } |
| |
| /** |
| * Selects current alive owners for some partition as affinity distribution. |
| * |
| * @param aliveNodes Alive cluster nodes. |
| * @param curOwners Current affinity owners for some partition. |
| * |
| * @return List of current alive affinity owners. |
| * {@code null} if affinity owners should be inherited from ideal assignment as is. |
| */ |
| private @Nullable List<ClusterNode> selectCurrentAliveOwners( |
| Set<ClusterNode> aliveNodes, |
| List<ClusterNode> curOwners |
| ) { |
| List<ClusterNode> aliveCurOwners = curOwners.stream().filter(aliveNodes::contains).collect(Collectors.toList()); |
| |
| return !aliveCurOwners.isEmpty() ? aliveCurOwners : null; |
| } |
| |
| /** |
| * Calculates affinity on coordinator for custom event types that require centralized assignment. |
| * |
| * @param fut Current exchange future. |
| * @return Computed difference with ideal affinity. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public Map<Integer, CacheGroupAffinityMessage> onCustomEventWithEnforcedAffinityReassignment( |
| final GridDhtPartitionsExchangeFuture fut) throws IgniteCheckedException { |
| assert DiscoveryCustomEvent.requiresCentralizedAffinityAssignment(fut.firstEvent()); |
| |
| Map<Integer, CacheGroupAffinityMessage> result = onReassignmentEnforced(fut); |
| |
| return result; |
| } |
| |
| /** |
| * Calculates new affinity assignment on coordinator and creates affinity diff messages for other nodes. |
| * |
| * @param fut Current exchange future. |
| * @return Computed difference with ideal affinity. |
| */ |
| public Map<Integer, CacheGroupAffinityMessage> onReassignmentEnforced( |
| final GridDhtPartitionsExchangeFuture fut) { |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| AffinityTopologyVersion topVer = evts.topologyVersion(); |
| |
| CacheGroupHolder grpHolder = getOrCreateGroupHolder(topVer, desc); |
| |
| // Already calculated. |
| if (grpHolder.affinity().lastVersion().equals(topVer)) |
| return; |
| |
| List<List<ClusterNode>> assign = grpHolder.affinity().calculate(topVer, evts, evts.discoveryCache()).assignment(); |
| |
| if (!grpHolder.rebalanceEnabled || fut.cacheGroupAddedOnExchange(desc.groupId(), desc.receivedFrom())) |
| grpHolder.affinity().initialize(topVer, assign); |
| |
| fut.timeBag().finishLocalStage("Affinity initialization (enforced) " + |
| "[grp=" + desc.cacheOrGroupName() + "]"); |
| } |
| }); |
| |
| Map<Integer, Map<Integer, List<Long>>> diff = initAffinityBasedOnPartitionsAvailability(evts.topologyVersion(), |
| fut, |
| NODE_TO_ORDER, |
| true); |
| |
| return CacheGroupAffinityMessage.createAffinityDiffMessages(diff); |
| } |
| |
| /** |
| * Called on exchange initiated by server node join. |
| * |
| * @param fut Exchange future. |
| * @param crd Coordinator flag. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public void onServerJoin(final GridDhtPartitionsExchangeFuture fut, boolean crd) |
| throws IgniteCheckedException { |
| assert !fut.firstEvent().eventNode().isClient(); |
| |
| boolean locJoin = fut.firstEvent().eventNode().isLocal(); |
| |
| if (locJoin) { |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| AffinityTopologyVersion topVer = fut.initialVersion(); |
| |
| CacheGroupHolder grpHolder = getOrCreateGroupHolder(topVer, desc); |
| |
| if (crd) { |
| calculateAndInit(fut.events(), grpHolder.affinity(), topVer); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| |
| fut.timeBag().finishLocalStage("First node affinity initialization (node join) " + |
| "[grp=" + desc.cacheOrGroupName() + "]"); |
| } |
| } |
| }); |
| |
| if (!crd) { |
| fetchAffinityOnJoin(fut); |
| |
| fut.timeBag().finishLocalStage("Affinity fetch"); |
| } |
| } |
| else |
| initAffinityOnNodeJoin(fut, crd); |
| } |
| |
| /** |
| * @param fut Exchange future |
| * @param crd Coordinator flag. |
| */ |
| public void onBaselineTopologyChanged(final GridDhtPartitionsExchangeFuture fut, boolean crd) { |
| assert !fut.firstEvent().eventNode().isClient(); |
| |
| initAffinityOnNodeJoin(fut, crd); |
| } |
| |
| /** |
| * @param grpIds Cache group IDs. |
| * @return Cache names. |
| */ |
| private String groupNames(Collection<Integer> grpIds) { |
| StringBuilder names = new StringBuilder(); |
| |
| for (Integer grpId : grpIds) { |
| String name = cachesRegistry.group(grpId).cacheOrGroupName(); |
| |
| if (names.length() != 0) |
| names.append(", "); |
| |
| names.append(name); |
| } |
| |
| return names.toString(); |
| } |
| |
| /** |
| * @param grpId Group ID. |
| * @return Group name for debug purpose. |
| */ |
| private String debugGroupName(int grpId) { |
| CacheGroupDescriptor desc = cachesRegistry.group(grpId); |
| |
| if (desc != null) |
| return desc.cacheOrGroupName(); |
| else |
| return "Unknown group: " + grpId; |
| } |
| |
| /** |
| * @param evts Discovery events. |
| * @param aff Affinity. |
| * @param topVer Topology version. |
| */ |
| private void calculateAndInit(ExchangeDiscoveryEvents evts, |
| GridAffinityAssignmentCache aff, |
| AffinityTopologyVersion topVer) |
| { |
| List<List<ClusterNode>> assignment = aff.calculate(topVer, evts, evts.discoveryCache()).assignment(); |
| |
| aff.initialize(topVer, assignment); |
| } |
| |
| /** |
| * @param fut Exchange future. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private void fetchAffinityOnJoin(GridDhtPartitionsExchangeFuture fut) throws IgniteCheckedException { |
| AffinityTopologyVersion topVer = fut.initialVersion(); |
| |
| List<GridDhtAssignmentFetchFuture> fetchFuts = Collections.synchronizedList(new ArrayList<>()); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| CacheGroupHolder holder = getOrCreateGroupHolder(topVer, desc); |
| |
| if (fut.cacheGroupAddedOnExchange(desc.groupId(), desc.receivedFrom())) { |
| // In case if merge is allowed do not calculate affinity since it can change on exchange end. |
| if (!fut.context().mergeExchanges()) |
| calculateAndInit(fut.events(), holder.affinity(), topVer); |
| } |
| else { |
| if (fut.context().fetchAffinityOnJoin()) { |
| GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx, |
| desc.groupId(), |
| topVer, |
| fut.events().discoveryCache()); |
| |
| fetchFut.init(false); |
| |
| fetchFuts.add(fetchFut); |
| } |
| else { |
| if (!fut.events().discoveryCache().serverNodes().isEmpty()) |
| fut.context().addGroupAffinityRequestOnJoin(desc.groupId()); |
| else |
| calculateAndInit(fut.events(), holder.affinity(), topVer); |
| } |
| } |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| } |
| }); |
| |
| for (int i = 0; i < fetchFuts.size(); i++) { |
| GridDhtAssignmentFetchFuture fetchFut = fetchFuts.get(i); |
| |
| int grpId = fetchFut.groupId(); |
| |
| fetchAffinity(topVer, |
| fut.events(), |
| fut.events().discoveryCache(), |
| groupAffinity(grpId), |
| fetchFut); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| } |
| } |
| |
| /** |
| * @param topVer Topology version. |
| * @param events Discovery events. |
| * @param discoCache Discovery data cache. |
| * @param affCache Affinity. |
| * @param fetchFut Affinity fetch future. |
| * @return Affinity assignment response. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private GridDhtAffinityAssignmentResponse fetchAffinity( |
| AffinityTopologyVersion topVer, |
| @Nullable ExchangeDiscoveryEvents events, |
| DiscoCache discoCache, |
| GridAffinityAssignmentCache affCache, |
| GridDhtAssignmentFetchFuture fetchFut |
| ) throws IgniteCheckedException { |
| assert affCache != null; |
| |
| GridDhtAffinityAssignmentResponse res = fetchFut.get(); |
| |
| if (res == null) { |
| List<List<ClusterNode>> aff = affCache.calculate(topVer, events, discoCache).assignment(); |
| |
| affCache.initialize(topVer, aff); |
| } |
| else { |
| List<List<ClusterNode>> idealAff = res.idealAffinityAssignment(discoCache); |
| |
| if (idealAff != null) |
| affCache.idealAssignment(topVer, idealAff); |
| else { |
| assert !affCache.centralizedAffinityFunction(); |
| |
| affCache.calculate(topVer, events, discoCache); |
| } |
| |
| List<List<ClusterNode>> aff = res.affinityAssignment(discoCache); |
| |
| assert aff != null : res; |
| |
| affCache.initialize(topVer, aff); |
| } |
| |
| return res; |
| } |
| |
| /** |
| * Called on exchange initiated by server node leave or custom event with centralized affinity assignment. |
| * |
| * @param fut Exchange future. |
| * @param crd Coordinator flag. |
| * @return {@code True} if affinity should be assigned by coordinator. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public boolean onCentralizedAffinityChange(final GridDhtPartitionsExchangeFuture fut, |
| boolean crd) throws IgniteCheckedException { |
| assert (fut.events().hasServerLeft() && !fut.firstEvent().eventNode().isClient()) || |
| DiscoveryCustomEvent.requiresCentralizedAffinityAssignment(fut.firstEvent()) : fut.firstEvent(); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| CacheGroupHolder cache = getOrCreateGroupHolder(fut.initialVersion(), desc); |
| |
| cache.aff.calculate(fut.initialVersion(), fut.events(), fut.events().discoveryCache()); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| |
| fut.timeBag().finishLocalStage("Affinity centralized initialization (crd) " + |
| "[grp=" + desc.cacheOrGroupName() + ", crd=" + crd + "]"); |
| |
| validator.validateCacheGroup(desc); |
| } |
| }); |
| |
| synchronized (mux) { |
| waitInfo = null; |
| } |
| |
| return true; |
| } |
| |
| /** |
| * @param fut Exchange future. |
| * @param newAff {@code True} if there are no older nodes with affinity info available. |
| * @return Future completed when caches initialization is done. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public IgniteInternalFuture<?> initCoordinatorCaches( |
| final GridDhtPartitionsExchangeFuture fut, |
| final boolean newAff |
| ) throws IgniteCheckedException { |
| boolean locJoin = fut.firstEvent().eventNode().isLocal(); |
| |
| if (!locJoin) |
| return null; |
| |
| final List<IgniteInternalFuture<AffinityTopologyVersion>> futs = Collections.synchronizedList(new ArrayList<>()); |
| |
| final AffinityTopologyVersion topVer = fut.initialVersion(); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| CacheGroupHolder grpHolder = getOrCreateGroupHolder(topVer, desc); |
| |
| if (grpHolder.affinity().idealAssignmentRaw() != null) |
| return; |
| |
| // Need initialize holders and affinity if this node became coordinator during this exchange. |
| int grpId = desc.groupId(); |
| |
| CacheGroupContext grp = cctx.cache().cacheGroup(grpId); |
| |
| if (grp == null) { |
| grpHolder = createHolder(cctx, desc, topVer, null); |
| |
| final GridAffinityAssignmentCache aff = grpHolder.affinity(); |
| |
| if (newAff) { |
| if (!aff.lastVersion().equals(topVer)) |
| calculateAndInit(fut.events(), aff, topVer); |
| |
| grpHolder.topology(fut.context().events().discoveryCache()).beforeExchange(fut, true, false); |
| } |
| else { |
| List<GridDhtPartitionsExchangeFuture> exchFuts = cctx.exchange().exchangeFutures(); |
| |
| int idx = exchFuts.indexOf(fut); |
| |
| assert idx >= 0 && idx < exchFuts.size() - 1 : "Invalid exchange futures state [cur=" + idx + |
| ", total=" + exchFuts.size() + ']'; |
| |
| GridDhtPartitionsExchangeFuture futureToFetchAffinity = null; |
| |
| for (int i = idx + 1; i < exchFuts.size(); i++) { |
| GridDhtPartitionsExchangeFuture prev = exchFuts.get(i); |
| |
| assert prev.isDone() && prev.topologyVersion().compareTo(topVer) < 0; |
| |
| if (prev.isMerged()) |
| continue; |
| |
| futureToFetchAffinity = prev; |
| |
| break; |
| } |
| |
| if (futureToFetchAffinity == null) |
| throw new IgniteCheckedException("Failed to find completed exchange future to fetch affinity."); |
| |
| if (log.isDebugEnabled()) { |
| log.debug("Need initialize affinity on coordinator [" + |
| "cacheGrp=" + desc.cacheOrGroupName() + |
| "prevAff=" + futureToFetchAffinity.topologyVersion() + ']'); |
| } |
| |
| GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture( |
| cctx, |
| desc.groupId(), |
| futureToFetchAffinity.topologyVersion(), |
| futureToFetchAffinity.events().discoveryCache() |
| ); |
| |
| fetchFut.init(false); |
| |
| final GridFutureAdapter<AffinityTopologyVersion> affFut = new GridFutureAdapter<>(); |
| |
| final GridDhtPartitionsExchangeFuture futureToFetchAffinity0 = futureToFetchAffinity; |
| |
| fetchFut.listen(new IgniteInClosureX<IgniteInternalFuture<GridDhtAffinityAssignmentResponse>>() { |
| @Override public void applyx(IgniteInternalFuture<GridDhtAffinityAssignmentResponse> fetchFut) |
| throws IgniteCheckedException { |
| fetchAffinity( |
| futureToFetchAffinity0.topologyVersion(), |
| futureToFetchAffinity0.events(), |
| futureToFetchAffinity0.events().discoveryCache(), |
| aff, |
| (GridDhtAssignmentFetchFuture)fetchFut |
| ); |
| |
| aff.calculate(topVer, fut.events(), fut.events().discoveryCache()); |
| |
| affFut.onDone(topVer); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| } |
| }); |
| |
| futs.add(affFut); |
| } |
| } |
| else { |
| grpHolder = new CacheGroupAffNodeHolder(grp); |
| |
| if (newAff) { |
| GridAffinityAssignmentCache aff = grpHolder.affinity(); |
| |
| if (!aff.lastVersion().equals(topVer)) |
| calculateAndInit(fut.events(), aff, topVer); |
| |
| grpHolder.topology(fut.context().events().discoveryCache()).beforeExchange(fut, true, false); |
| } |
| } |
| |
| grpHolders.put(grpHolder.groupId(), grpHolder); |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| |
| fut.timeBag().finishLocalStage("Coordinator affinity cache init " + |
| "[grp=" + desc.cacheOrGroupName() + "]"); |
| } |
| }); |
| |
| if (!futs.isEmpty()) { |
| GridCompoundFuture<AffinityTopologyVersion, ?> affFut = new GridCompoundFuture<>(); |
| |
| for (IgniteInternalFuture<AffinityTopologyVersion> f : futs) |
| affFut.add(f); |
| |
| affFut.markInitialized(); |
| |
| return affFut; |
| } |
| |
| return null; |
| } |
| |
| /** |
| * @param topVer Topology version. |
| * @param desc Cache descriptor. |
| * @return Cache holder. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private CacheGroupHolder getOrCreateGroupHolder(AffinityTopologyVersion topVer, CacheGroupDescriptor desc) |
| throws IgniteCheckedException { |
| CacheGroupHolder cacheGrp = grpHolders.get(desc.groupId()); |
| |
| if (cacheGrp != null) |
| return cacheGrp; |
| |
| return createGroupHolder(topVer, desc, cctx.cache().cacheGroup(desc.groupId()) != null); |
| } |
| |
| /** |
| * @param topVer Topology version. |
| * @param desc Cache descriptor. |
| * @param affNode Affinity node flag. |
| * @return Cache holder. |
| * @throws IgniteCheckedException If failed. |
| */ |
| private CacheGroupHolder createGroupHolder( |
| AffinityTopologyVersion topVer, |
| CacheGroupDescriptor desc, |
| boolean affNode |
| ) throws IgniteCheckedException { |
| assert topVer != null; |
| assert desc != null; |
| |
| CacheGroupContext grp = cctx.cache().cacheGroup(desc.groupId()); |
| |
| cctx.io().addCacheGroupHandler(desc.groupId(), GridDhtAffinityAssignmentResponse.class, |
| this::processAffinityAssignmentResponse); |
| |
| assert (affNode && grp != null) || (!affNode && grp == null); |
| |
| CacheGroupHolder cacheGrp = affNode ? |
| new CacheGroupAffNodeHolder(grp) : |
| createHolder(cctx, desc, topVer, null); |
| |
| CacheGroupHolder old = grpHolders.put(desc.groupId(), cacheGrp); |
| |
| assert old == null : old; |
| |
| return cacheGrp; |
| } |
| |
| /** |
| * @param fut Current exchange future. |
| * @param crd Coordinator flag. |
| */ |
| private void initAffinityOnNodeJoin(final GridDhtPartitionsExchangeFuture fut, boolean crd) { |
| final ExchangeDiscoveryEvents evts = fut.context().events(); |
| |
| final WaitRebalanceInfo waitRebalanceInfo = new WaitRebalanceInfo(evts.lastServerEventVersion()); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| CacheGroupHolder grpHolder = getOrCreateGroupHolder(evts.topologyVersion(), desc); |
| |
| CacheGroupHolder cache = getOrCreateGroupHolder(evts.topologyVersion(), desc); |
| |
| // Already calculated. |
| if (cache.affinity().lastVersion().equals(evts.topologyVersion())) |
| return; |
| |
| Span affCalcSpan = cctx.kernalContext().tracing().create(AFFINITY_CALCULATION, fut.span()) |
| .addTag("cache.group", desc.cacheOrGroupName()); |
| |
| boolean latePrimary = cache.rebalanceEnabled; |
| |
| boolean grpAdded = evts.nodeJoined(desc.receivedFrom()); |
| |
| initAffinityOnNodeJoin(evts, |
| grpAdded, |
| grpHolder, |
| crd ? waitRebalanceInfo : null, |
| latePrimary); |
| |
| if (crd && grpAdded) { |
| AffinityAssignment aff = grpHolder.aff.cachedAffinity(grpHolder.aff.lastVersion()); |
| |
| assert evts.topologyVersion().equals(aff.topologyVersion()) : "Unexpected version [" + |
| "grp=" + grpHolder.aff.cacheOrGroupName() + |
| ", evts=" + evts.topologyVersion() + |
| ", aff=" + grpHolder.aff.lastVersion() + ']'; |
| |
| Map<UUID, GridDhtPartitionMap> map = affinityFullMap(aff); |
| |
| for (GridDhtPartitionMap map0 : map.values()) |
| grpHolder.topology(fut.context().events().discoveryCache()).update(fut.exchangeId(), map0, true); |
| } |
| |
| cctx.exchange().exchangerUpdateHeartbeat(); |
| |
| affCalcSpan.end(); |
| |
| fut.timeBag().finishLocalStage("Affinity initialization (node join) " + |
| "[grp=" + desc.cacheOrGroupName() + ", crd=" + crd + "]"); |
| } |
| }); |
| |
| if (crd) { |
| if (log.isDebugEnabled()) { |
| log.debug("Computed new affinity after node join [topVer=" + evts.lastServerEventVersion() + |
| ", waitGrps=" + groupNames(waitRebalanceInfo.waitGrps.keySet()) + ']'); |
| } |
| } |
| |
| synchronized (mux) { |
| waitInfo = !waitRebalanceInfo.empty() ? waitRebalanceInfo : null; |
| } |
| } |
| |
| /** |
| * @param aff Affinity assignment. |
| */ |
| private Map<UUID, GridDhtPartitionMap> affinityFullMap(AffinityAssignment aff) { |
| Map<UUID, GridDhtPartitionMap> map = new HashMap<>(); |
| |
| for (int p = 0; p < aff.assignment().size(); p++) { |
| Collection<UUID> ids = aff.getIds(p); |
| |
| for (UUID nodeId : ids) { |
| GridDhtPartitionMap partMap = map.get(nodeId); |
| |
| if (partMap == null) { |
| partMap = new GridDhtPartitionMap(nodeId, |
| 1L, |
| aff.topologyVersion(), |
| new GridPartitionStateMap(), |
| false); |
| |
| map.put(nodeId, partMap); |
| } |
| |
| partMap.put(p, OWNING); |
| } |
| } |
| |
| return map; |
| } |
| |
| /** |
| * @param evts Discovery events processed during exchange. |
| * @param addedOnExchnage {@code True} if cache group was added during this exchange. |
| * @param grpHolder Group holder. |
| * @param rebalanceInfo Rebalance information on coordinator or null on other nodes. |
| * @param latePrimary If {@code true} delays primary assignment if it is not owner. |
| */ |
| private void initAffinityOnNodeJoin( |
| ExchangeDiscoveryEvents evts, |
| boolean addedOnExchnage, |
| CacheGroupHolder grpHolder, |
| @Nullable WaitRebalanceInfo rebalanceInfo, |
| boolean latePrimary |
| ) { |
| GridAffinityAssignmentCache aff = grpHolder.affinity(); |
| |
| if (addedOnExchnage) { |
| if (!aff.lastVersion().equals(evts.topologyVersion())) |
| calculateAndInit(evts, aff, evts.topologyVersion()); |
| |
| return; |
| } |
| |
| AffinityTopologyVersion affTopVer = aff.lastVersion(); |
| |
| assert affTopVer.topologyVersion() > 0 : "Affinity is not initialized [grp=" + aff.cacheOrGroupName() + |
| ", topVer=" + affTopVer + ", node=" + cctx.localNodeId() + ']'; |
| |
| List<List<ClusterNode>> curAff = aff.assignments(affTopVer); |
| |
| assert aff.idealAssignment() != null : "Previous assignment is not available."; |
| |
| List<List<ClusterNode>> idealAssignment = aff.calculate(evts.topologyVersion(), evts, evts.discoveryCache()).assignment(); |
| List<List<ClusterNode>> newAssignment = null; |
| |
| if (latePrimary) { |
| for (int p = 0; p < idealAssignment.size(); p++) { |
| List<ClusterNode> newNodes = idealAssignment.get(p); |
| List<ClusterNode> curNodes = curAff.get(p); |
| |
| ClusterNode curPrimary = !curNodes.isEmpty() ? curNodes.get(0) : null; |
| ClusterNode newPrimary = !newNodes.isEmpty() ? newNodes.get(0) : null; |
| |
| if (curPrimary != null && newPrimary != null && !curPrimary.equals(newPrimary)) { |
| assert cctx.discovery().node(evts.topologyVersion(), curPrimary.id()) != null : curPrimary; |
| |
| List<ClusterNode> nodes0 = latePrimaryAssignment(aff, |
| p, |
| curPrimary, |
| newNodes, |
| rebalanceInfo); |
| |
| if (newAssignment == null) |
| newAssignment = new ArrayList<>(idealAssignment); |
| |
| newAssignment.set(p, nodes0); |
| } |
| |
| GridDhtPartitionTopology top = grpHolder.topology(evts.discoveryCache()); |
| |
| if (rebalanceInfo != null) { |
| List<ClusterNode> owners = top.owners(p, evts.topologyVersion()); |
| |
| // If current owners are empty no supplier can exist. |
| // A group with lost partitions never gets rebalanced so should not be added to waitInfo. |
| if (!owners.isEmpty() && !owners.containsAll(idealAssignment.get(p)) && |
| !top.lostPartitions().contains(p)) |
| rebalanceInfo.add(aff.groupId(), p, newNodes); |
| } |
| } |
| } |
| |
| if (newAssignment == null) |
| newAssignment = idealAssignment; |
| |
| aff.initialize(evts.topologyVersion(), newAssignment); |
| } |
| |
| /** |
| * @param aff Cache. |
| * @param part Partition. |
| * @param curPrimary Current primary. |
| * @param newNodes New ideal assignment. |
| * @param rebalance Rabalance information holder. |
| * @return Assignment. |
| */ |
| private List<ClusterNode> latePrimaryAssignment( |
| GridAffinityAssignmentCache aff, |
| int part, |
| ClusterNode curPrimary, |
| List<ClusterNode> newNodes, |
| @Nullable WaitRebalanceInfo rebalance |
| ) { |
| assert curPrimary != null; |
| assert !F.isEmpty(newNodes); |
| assert !curPrimary.equals(newNodes.get(0)); |
| |
| List<ClusterNode> nodes0 = new ArrayList<>(newNodes.size() + 1); |
| |
| nodes0.add(curPrimary); |
| |
| for (int i = 0; i < newNodes.size(); i++) { |
| ClusterNode node = newNodes.get(i); |
| |
| if (!node.equals(curPrimary)) |
| nodes0.add(node); |
| } |
| |
| if (rebalance != null) |
| rebalance.add(aff.groupId(), part, newNodes); |
| |
| return nodes0; |
| } |
| |
| /** |
| * @param fut Exchange future. |
| * @return Affinity assignment. |
| * @throws IgniteCheckedException If failed. |
| */ |
| public IgniteInternalFuture<Map<Integer, Map<Integer, List<UUID>>>> initAffinityOnNodeLeft( |
| final GridDhtPartitionsExchangeFuture fut) throws IgniteCheckedException { |
| assert !fut.context().mergeExchanges(); |
| |
| IgniteInternalFuture<?> initFut = initCoordinatorCaches(fut, false); |
| |
| if (initFut != null && !initFut.isDone()) { |
| final GridFutureAdapter<Map<Integer, Map<Integer, List<UUID>>>> resFut = new GridFutureAdapter<>(); |
| |
| initFut.listen(new IgniteInClosure<IgniteInternalFuture<?>>() { |
| @Override public void apply(IgniteInternalFuture<?> initFut) { |
| try { |
| resFut.onDone(initAffinityBasedOnPartitionsAvailability(fut.initialVersion(), fut, NODE_TO_ID, false)); |
| } |
| catch (Exception e) { |
| resFut.onDone(e); |
| } |
| } |
| }); |
| |
| return resFut; |
| } |
| else |
| return new GridFinishedFuture<>(initAffinityBasedOnPartitionsAvailability(fut.initialVersion(), fut, NODE_TO_ID, false)); |
| } |
| |
| /** |
| * Initializes current affinity assignment based on partitions availability. Nodes that have most recent data will |
| * be considered affinity nodes. |
| * |
| * @param topVer Topology version. |
| * @param fut Exchange future. |
| * @param c Closure converting affinity diff. |
| * @param initAff {@code True} if need initialize affinity. |
| * @return Affinity assignment for each of registered cache group. |
| */ |
| private <T> Map<Integer, Map<Integer, List<T>>> initAffinityBasedOnPartitionsAvailability( |
| final AffinityTopologyVersion topVer, |
| final GridDhtPartitionsExchangeFuture fut, |
| final IgniteClosure<ClusterNode, T> c, |
| final boolean initAff |
| ) { |
| final boolean enforcedCentralizedAssignment = |
| DiscoveryCustomEvent.requiresCentralizedAffinityAssignment(fut.firstEvent()); |
| |
| final WaitRebalanceInfo waitRebalanceInfo = enforcedCentralizedAssignment ? |
| new WaitRebalanceInfo(fut.exchangeId().topologyVersion()) : |
| new WaitRebalanceInfo(fut.context().events().lastServerEventVersion()); |
| |
| final Collection<ClusterNode> aliveNodes = fut.context().events().discoveryCache().serverNodes(); |
| |
| final Map<Integer, Map<Integer, List<T>>> assignment = new ConcurrentHashMap<>(); |
| |
| forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() { |
| @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException { |
| CacheGroupHolder grpHolder = getOrCreateGroupHolder(topVer, desc); |
| |
| if (!grpHolder.rebalanceEnabled || |
| (fut.cacheGroupAddedOnExchange(desc.groupId(), desc.receivedFrom()) && !enforcedCentralizedAssignment)) |
| return; |
| |
| AffinityTopologyVersion affTopVer = grpHolder.affinity().lastVersion(); |
| |
| assert (affTopVer.topologyVersion() > 0 && !affTopVer.equals(topVer)) || enforcedCentralizedAssignment : |
| "Invalid affinity version [last=" + affTopVer + ", futVer=" + topVer + ", grp=" + desc.cacheOrGroupName() + ']'; |
| |
| List<List<ClusterNode>> curAssignment = grpHolder.affinity().assignments(affTopVer); |
| List<List<ClusterNode>> newAssignment = grpHolder.affinity().idealAssignmentRaw(); |
| |
| assert newAssignment != null; |
| |
| List<List<ClusterNode>> newAssignment0 = initAff ? new ArrayList<>(newAssignment) : null; |
| |
| GridDhtPartitionTopology top = grpHolder.topology(fut.context().events().discoveryCache()); |
| |
| Map<Integer, List<T>> cacheAssignment = null; |
| |
| for (int p = 0; p < newAssignment.size(); p++) { |
| List<ClusterNode> newNodes = newAssignment.get(p); |
| List<ClusterNode> curNodes = curAssignment.get(p); |
| |
| assert aliveNodes.containsAll(newNodes) : "Invalid new assignment [grp=" + grpHolder.aff.cacheOrGroupName() + |
| ", nodes=" + newNodes + |
| ", topVer=" + fut.context().events().discoveryCache().version() + |
| ", evts=" + fut.context().events().events() + "]"; |
| |
| ClusterNode curPrimary = !curNodes.isEmpty() ? curNodes.get(0) : null; |
| ClusterNode newPrimary = !newNodes.isEmpty() ? newNodes.get(0) : null; |
| |
| List<ClusterNode> newNodes0 = null; |
| |
| assert newPrimary == null || aliveNodes.contains(newPrimary) : "Invalid new primary [" + |
| "grp=" + desc.cacheOrGroupName() + |
| ", node=" + newPrimary + |
| ", topVer=" + topVer + ']'; |
| |
| List<ClusterNode> owners = top.owners(p, topVer); |
| |
| // It is essential that curPrimary node has partition in OWNING state. |
| if (!owners.isEmpty() && !owners.contains(curPrimary)) |
| curPrimary = owners.get(0); |
| |
| // If new assignment is empty preserve current ownership for alive nodes. |
| if (curPrimary != null && newPrimary == null) { |
| newNodes0 = new ArrayList<>(curNodes.size()); |
| |
| for (ClusterNode node : curNodes) { |
| if (aliveNodes.contains(node)) |
| newNodes0.add(node); |
| } |
| } |
| else if (curPrimary != null && !curPrimary.equals(newPrimary)) { |
| GridDhtPartitionState state = top.partitionState(newPrimary.id(), p); |
| |
| if (aliveNodes.contains(curPrimary)) { |
| if (state != OWNING) { |
| newNodes0 = latePrimaryAssignment(grpHolder.affinity(), |
| p, |
| curPrimary, |
| newNodes, |
| waitRebalanceInfo); |
| } |
| } |
| else { |
| if (state != OWNING) { |
| for (int i = 1; i < curNodes.size(); i++) { |
| ClusterNode curNode = curNodes.get(i); |
| |
| if (top.partitionState(curNode.id(), p) == OWNING && |
| aliveNodes.contains(curNode)) { |
| newNodes0 = latePrimaryAssignment(grpHolder.affinity(), |
| p, |
| curNode, |
| newNodes, |
| waitRebalanceInfo); |
| |
| break; |
| } |
| } |
| |
| if (newNodes0 == null) { |
| for (ClusterNode owner : owners) { |
| if (aliveNodes.contains(owner)) { |
| newNodes0 = latePrimaryAssignment(grpHolder.affinity(), |
| p, |
| owner, |
| newNodes, |
| waitRebalanceInfo); |
| |
| break; |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| // This will happen if no primary is changed but some backups still need to be rebalanced. |
| if (!owners.isEmpty() && !owners.containsAll(newNodes) && !top.lostPartitions().contains(p)) |
| waitRebalanceInfo.add(grpHolder.groupId(), p, newNodes); |
| |
| if (newNodes0 != null) { |
| assert aliveNodes.containsAll(newNodes0) : "Invalid late assignment [grp=" + grpHolder.aff.cacheOrGroupName() + |
| ", nodes=" + newNodes + |
| ", topVer=" + fut.context().events().discoveryCache().version() + |
| ", evts=" + fut.context().events().events() + "]"; |
| |
| if (newAssignment0 != null) |
| newAssignment0.set(p, newNodes0); |
| |
| if (cacheAssignment == null) |
| cacheAssignment = new HashMap<>(); |
| |
| List<T> n = new ArrayList<>(newNodes0.size()); |
| |
| for (int i = 0; i < newNodes0.size(); i++) |
| n.add(c.apply(newNodes0.get(i))); |
| |
| cacheAssignment.put(p, n); |
| } |
| } |
| |
| if (cacheAssignment != null) |
| assignment.put(grpHolder.groupId(), cacheAssignment); |
| |
| if (initAff) |
| grpHolder.affinity().initialize(topVer, newAssignment0); |
| |
| fut.timeBag().finishLocalStage("Affinity recalculation (partitions availability) " + |
| "[grp=" + desc.cacheOrGroupName() + "]"); |
| } |
| }); |
| |
| if (log.isDebugEnabled()) { |
| log.debug("Computed new affinity after node left [topVer=" + topVer + |
| ", waitGrps=" + groupNames(waitRebalanceInfo.waitGrps.keySet()) + ']'); |
| } |
| |
| synchronized (mux) { |
| waitInfo = !waitRebalanceInfo.empty() ? waitRebalanceInfo : null; |
| } |
| |
| return assignment; |
| } |
| |
| /** |
| * @return All registered cache groups. |
| */ |
| public Map<Integer, CacheGroupDescriptor> cacheGroups() { |
| return cachesRegistry.allGroups(); |
| } |
| |
| /** |
| * @return All registered cache groups. |
| */ |
| public Map<Integer, DynamicCacheDescriptor> caches() { |
| return cachesRegistry.allCaches(); |
| } |
| |
| /** |
| * @param grpId Cache group ID |
| * @return Cache affinity cache. |
| */ |
| @Nullable public GridAffinityAssignmentCache groupAffinity(int grpId) { |
| CacheGroupHolder grpHolder = grpHolders.get(grpId); |
| |
| return grpHolder != null ? grpHolder.affinity() : null; |
| } |
| |
| /** |
| * |
| */ |
| public void dumpDebugInfo() { |
| if (!pendingAssignmentFetchFuts.isEmpty()) { |
| U.warn(log, "Pending assignment fetch futures:"); |
| |
| for (GridDhtAssignmentFetchFuture fut : pendingAssignmentFetchFuts.values()) |
| U.warn(log, ">>> " + fut); |
| } |
| } |
| |
| /** |
| * @param nodes Nodes. |
| * @return IDs. |
| */ |
| private static List<UUID> toIds0(List<ClusterNode> nodes) { |
| List<UUID> partIds = new ArrayList<>(nodes.size()); |
| |
| for (int i = 0; i < nodes.size(); i++) |
| partIds.add(nodes.get(i).id()); |
| |
| return partIds; |
| } |
| |
| /** |
| * @param topVer Topology version. |
| * @param ids IDs. |
| * @return Nodes. |
| */ |
| private List<ClusterNode> toNodes(AffinityTopologyVersion topVer, List<UUID> ids) { |
| List<ClusterNode> nodes = new ArrayList<>(ids.size()); |
| |
| for (int i = 0; i < ids.size(); i++) { |
| UUID id = ids.get(i); |
| |
| ClusterNode node = cctx.discovery().node(topVer, id); |
| |
| assert node != null : "Failed to get node [id=" + id + |
| ", topVer=" + topVer + |
| ", locNode=" + cctx.localNode() + |
| ", allNodes=" + cctx.discovery().nodes(topVer) + ']'; |
| |
| nodes.add(node); |
| } |
| |
| return nodes; |
| } |
| |
| /** |
| * @return Primary nodes for local backups. |
| */ |
| public Set<ClusterNode> idealPrimaryNodesForLocalBackups() { |
| Set<ClusterNode> res = new GridConcurrentHashSet<>(); |
| |
| ClusterNode loc = cctx.localNode(); |
| |
| forAllCacheGroups(new IgniteInClosureX<GridAffinityAssignmentCache>() { |
| @Override public void applyx(GridAffinityAssignmentCache aff) { |
| CacheGroupDescriptor desc = cachesRegistry.group(aff.groupId()); |
| |
| if (desc.config().getCacheMode() == PARTITIONED) { |
| List<List<ClusterNode>> assignment = aff.idealAssignmentRaw(); |
| |
| HashSet<ClusterNode> primaries = new HashSet<>(); |
| |
| for (List<ClusterNode> nodes : assignment) { |
| if (nodes.indexOf(loc) > 0) |
| primaries.add(nodes.get(0)); |
| } |
| |
| res.addAll(primaries); |
| } |
| } |
| }); |
| |
| return res; |
| } |
| |
| /** |
| * |
| */ |
| abstract class CacheGroupHolder { |
| /** */ |
| private final GridAffinityAssignmentCache aff; |
| |
| /** */ |
| private final boolean rebalanceEnabled; |
| |
| /** |
| * @param rebalanceEnabled Cache rebalance flag. |
| * @param aff Affinity cache. |
| * @param initAff Existing affinity cache. |
| */ |
| CacheGroupHolder(boolean rebalanceEnabled, |
| GridAffinityAssignmentCache aff, |
| @Nullable GridAffinityAssignmentCache initAff) { |
| this.aff = aff; |
| |
| if (initAff != null) |
| aff.init(initAff); |
| |
| this.rebalanceEnabled = rebalanceEnabled; |
| } |
| |
| /** |
| * @return Client holder flag. |
| */ |
| abstract boolean nonAffNode(); |
| |
| /** |
| * @return Group ID. |
| */ |
| int groupId() { |
| return aff.groupId(); |
| } |
| |
| /** |
| * @return Partitions number. |
| */ |
| int partitions() { |
| return aff.partitions(); |
| } |
| |
| /** |
| * @param discoCache Discovery data cache. |
| * @return Cache topology. |
| */ |
| abstract GridDhtPartitionTopology topology(DiscoCache discoCache); |
| |
| /** |
| * @return Affinity. |
| */ |
| GridAffinityAssignmentCache affinity() { |
| return aff; |
| } |
| } |
| |
| /** |
| * Created cache is started on coordinator. |
| */ |
| private class CacheGroupAffNodeHolder extends CacheGroupHolder { |
| /** */ |
| private final CacheGroupContext grp; |
| |
| /** |
| * @param grp Cache group. |
| */ |
| CacheGroupAffNodeHolder(CacheGroupContext grp) { |
| this(grp, null); |
| } |
| |
| /** |
| * @param grp Cache group. |
| * @param initAff Current affinity. |
| */ |
| CacheGroupAffNodeHolder(CacheGroupContext grp, @Nullable GridAffinityAssignmentCache initAff) { |
| super(grp.rebalanceEnabled(), grp.affinity(), initAff); |
| |
| assert !grp.isLocal() : grp; |
| |
| this.grp = grp; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean nonAffNode() { |
| return false; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public GridDhtPartitionTopology topology(DiscoCache discoCache) { |
| return grp.topology(); |
| } |
| } |
| |
| /** |
| * Created if cache is not started on coordinator. |
| */ |
| private class CacheGroupNoAffOrFilteredHolder extends CacheGroupHolder { |
| /** */ |
| private final GridCacheSharedContext cctx; |
| |
| /** |
| * @param rebalanceEnabled Rebalance flag. |
| * @param cctx Context. |
| * @param aff Affinity. |
| * @param initAff Current affinity. |
| */ |
| CacheGroupNoAffOrFilteredHolder( |
| boolean rebalanceEnabled, |
| GridCacheSharedContext cctx, |
| GridAffinityAssignmentCache aff, |
| @Nullable GridAffinityAssignmentCache initAff |
| ) { |
| super(rebalanceEnabled, aff, initAff); |
| |
| this.cctx = cctx; |
| } |
| |
| /** |
| * @param cctx Context. |
| * @param grpDesc Cache group descriptor. |
| * @param topVer Current exchange version. |
| * @return Cache holder. |
| * @throws IgniteCheckedException If failed. |
| */ |
| CacheGroupNoAffOrFilteredHolder create( |
| GridCacheSharedContext cctx, |
| CacheGroupDescriptor grpDesc, |
| AffinityTopologyVersion topVer |
| ) throws IgniteCheckedException { |
| return create(cctx, grpDesc, topVer, null); |
| } |
| |
| /** |
| * @param cctx Context. |
| * @param grpDesc Cache group descriptor. |
| * @param topVer Current exchange version. |
| * @param initAff Current affinity. |
| * @return Cache holder. |
| * @throws IgniteCheckedException If failed. |
| */ |
| CacheGroupNoAffOrFilteredHolder create( |
| GridCacheSharedContext cctx, |
| CacheGroupDescriptor grpDesc, |
| AffinityTopologyVersion topVer, |
| @Nullable GridAffinityAssignmentCache initAff |
| ) throws IgniteCheckedException { |
| assert grpDesc != null; |
| assert !cctx.kernalContext().clientNode() || !CU.affinityNode(cctx.localNode(), grpDesc.config().getNodeFilter()); |
| |
| CacheConfiguration<?, ?> ccfg = grpDesc.config(); |
| |
| assert ccfg != null : grpDesc; |
| assert ccfg.getCacheMode() != LOCAL : ccfg.getName(); |
| |
| assert !cctx.discovery().cacheGroupAffinityNodes(grpDesc.groupId(), |
| topVer).contains(cctx.localNode()) : grpDesc.cacheOrGroupName(); |
| |
| AffinityFunction affFunc = cctx.cache().clone(ccfg.getAffinity()); |
| |
| cctx.kernalContext().resource().injectGeneric(affFunc); |
| cctx.kernalContext().resource().injectCacheName(affFunc, ccfg.getName()); |
| |
| U.startLifecycleAware(F.asList(affFunc)); |
| |
| GridAffinityAssignmentCache aff = new GridAffinityAssignmentCache(cctx.kernalContext(), |
| grpDesc.cacheOrGroupName(), |
| grpDesc.groupId(), |
| affFunc, |
| ccfg.getNodeFilter(), |
| ccfg.getBackups(), |
| ccfg.getCacheMode() == LOCAL |
| ); |
| |
| return new CacheGroupNoAffOrFilteredHolder(ccfg.getRebalanceMode() != NONE, cctx, aff, initAff); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public boolean nonAffNode() { |
| return true; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public GridDhtPartitionTopology topology(DiscoCache discoCache) { |
| return cctx.exchange().clientTopology(groupId(), discoCache); |
| } |
| } |
| |
| private CacheGroupNoAffOrFilteredHolder createHolder( |
| GridCacheSharedContext cctx, |
| CacheGroupDescriptor grpDesc, |
| AffinityTopologyVersion topVer, |
| @Nullable GridAffinityAssignmentCache initAff |
| ) throws IgniteCheckedException { |
| assert grpDesc != null; |
| assert !cctx.kernalContext().clientNode() || !CU.affinityNode(cctx.localNode(), grpDesc.config().getNodeFilter()); |
| |
| CacheConfiguration<?, ?> ccfg = grpDesc.config(); |
| |
| assert ccfg != null : grpDesc; |
| assert ccfg.getCacheMode() != LOCAL : ccfg.getName(); |
| |
| assert !cctx.discovery().cacheGroupAffinityNodes(grpDesc.groupId(), |
| topVer).contains(cctx.localNode()) : grpDesc.cacheOrGroupName(); |
| |
| AffinityFunction affFunc = cctx.cache().clone(ccfg.getAffinity()); |
| |
| cctx.kernalContext().resource().injectGeneric(affFunc); |
| cctx.kernalContext().resource().injectCacheName(affFunc, ccfg.getName()); |
| |
| U.startLifecycleAware(F.asList(affFunc)); |
| |
| GridAffinityAssignmentCache aff = new GridAffinityAssignmentCache(cctx.kernalContext(), |
| grpDesc.cacheOrGroupName(), |
| grpDesc.groupId(), |
| affFunc, |
| ccfg.getNodeFilter(), |
| ccfg.getBackups(), |
| ccfg.getCacheMode() == LOCAL |
| ); |
| |
| return new CacheGroupNoAffOrFilteredHolder(ccfg.getRebalanceMode() != NONE, cctx, aff, initAff); |
| } |
| |
| /** |
| * Tracks rebalance state on coordinator. |
| * After all partitions are rebalanced the current affinity is switched to ideal. |
| */ |
| class WaitRebalanceInfo { |
| /** */ |
| private final AffinityTopologyVersion topVer; |
| |
| /** */ |
| private final Map<Integer, Set<Integer>> waitGrps = new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final Map<Integer /** Group id. */, Map<Integer /** Partition id. */, List<ClusterNode>>> assignments = |
| new ConcurrentHashMap<>(); |
| |
| /** */ |
| private final Map<Integer, IgniteUuid> deploymentIds = new ConcurrentHashMap<>(); |
| |
| /** |
| * @param topVer Topology version. |
| */ |
| WaitRebalanceInfo(AffinityTopologyVersion topVer) { |
| this.topVer = topVer; |
| } |
| |
| /** |
| * @return {@code True} if there are partitions waiting for rebalancing. |
| */ |
| boolean empty() { |
| boolean isEmpty = waitGrps.isEmpty(); |
| |
| if (!isEmpty) { |
| assert waitGrps.size() == assignments.size(); |
| |
| return false; |
| } |
| |
| return isEmpty; |
| } |
| |
| /** |
| * Adds a partition to wait set. |
| * |
| * @param grpId Group ID. |
| * @param part Partition. |
| * @param assignment New assignment. |
| */ |
| void add(Integer grpId, Integer part, List<ClusterNode> assignment) { |
| deploymentIds.putIfAbsent(grpId, cachesRegistry.group(grpId).deploymentId()); |
| |
| waitGrps.computeIfAbsent(grpId, k -> new HashSet<>()).add(part); |
| |
| assignments.computeIfAbsent(grpId, k -> new HashMap<>()).put(part, assignment); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public String toString() { |
| return "WaitRebalanceInfo [topVer=" + topVer + ", grps=" + waitGrps + ']'; |
| } |
| } |
| |
| /** |
| * Validator for memory overhead of persistent caches. |
| * |
| * Persistent cache requires some overhead in dataregion memory, e.g. a metapage per partition created by the cache. |
| * If this overhead reaches some limit (hardcoded to 15% for now) it may cause critical errors on node during |
| * checkpoint. |
| * |
| * Validator is intended to analyze cache group configuration and print warning to log to inform user about |
| * found problem. |
| */ |
| class CacheMemoryOverheadValidator { |
| /** */ |
| private static final double MEMORY_OVERHEAD_THRESHOLD = 0.15; |
| |
| /** |
| * Validates cache group configuration and prints warning if it violates 15% overhead limit. |
| * |
| * @param grpDesc Descriptor of cache group to validate. |
| */ |
| void validateCacheGroup(CacheGroupDescriptor grpDesc) { |
| DataStorageConfiguration dsCfg = cctx.gridConfig().getDataStorageConfiguration(); |
| CacheConfiguration<?, ?> grpCfg = grpDesc.config(); |
| |
| if (!CU.isPersistentCache(grpCfg, dsCfg) || CU.isSystemCache(grpDesc.cacheOrGroupName())) |
| return; |
| |
| CacheGroupHolder grpHolder = grpHolders.get(grpDesc.groupId()); |
| |
| if (grpHolder != null) { |
| int partsNum = 0; |
| UUID locNodeId = cctx.localNodeId(); |
| |
| List<List<ClusterNode>> assignment = grpHolder.aff.idealAssignment().assignment(); |
| |
| for (List<ClusterNode> nodes : assignment) { |
| if (nodes.stream().anyMatch(n -> n.id().equals(locNodeId))) |
| partsNum++; |
| } |
| |
| if (partsNum == 0) |
| return; |
| |
| DataRegionConfiguration drCfg = findDataRegion(dsCfg, grpCfg.getDataRegionName()); |
| |
| if (drCfg == null) |
| return; |
| |
| if ((1.0 * partsNum * dsCfg.getPageSize()) / drCfg.getMaxSize() > MEMORY_OVERHEAD_THRESHOLD) |
| log.warning(buildWarningMessage(grpDesc, drCfg, dsCfg.getPageSize(), partsNum)); |
| } |
| } |
| |
| /** |
| * Builds explanatory warning message. |
| * |
| * @param grpDesc Configuration of cache group violating memory overhead threshold. |
| * @param drCfg Configuration of data region configuration with not sufficient memory. |
| */ |
| private String buildWarningMessage(CacheGroupDescriptor grpDesc, |
| DataRegionConfiguration drCfg, |
| int pageSize, |
| int partsNum |
| ) { |
| String res = "Cache group '%s'" + |
| " brings high overhead for its metainformation in data region '%s'." + |
| " Metainformation required for its partitions (%d partitions, %d bytes per partition, %d MBs total)" + |
| " will consume more than 15%% of data region memory (%d MBs)." + |
| " It may lead to critical errors on the node and cluster instability." + |
| " Please reduce number of partitions, add more memory to the data region" + |
| " or add more server nodes for this cache group."; |
| |
| return String.format( |
| res, |
| grpDesc.cacheOrGroupName(), |
| drCfg.getName(), |
| partsNum, |
| pageSize, |
| U.sizeInMegabytes(partsNum * pageSize), |
| U.sizeInMegabytes(drCfg.getMaxSize()) |
| ); |
| } |
| |
| /** |
| * Finds data region by name. |
| * |
| * @param dsCfg Data storage configuration. |
| * @param drName Data region name. |
| * |
| * @return Found data region. |
| */ |
| @Nullable private DataRegionConfiguration findDataRegion(DataStorageConfiguration dsCfg, String drName) { |
| if (dsCfg.getDataRegionConfigurations() == null || drName == null) |
| return dsCfg.getDefaultDataRegionConfiguration(); |
| |
| if (dsCfg.getDefaultDataRegionConfiguration().getName().equals(drName)) |
| return dsCfg.getDefaultDataRegionConfiguration(); |
| |
| Optional<DataRegionConfiguration> cfgOpt = Arrays.stream(dsCfg.getDataRegionConfigurations()) |
| .filter(drCfg -> drCfg.getName().equals(drName)) |
| .findFirst(); |
| |
| return cfgOpt.isPresent() ? cfgOpt.get() : null; |
| } |
| } |
| } |
