You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/09/26 15:23:10 UTC
[06/12] ignite git commit: IGNITE-9649 Reworked and improved logging
for critical paths of Ignite internal processes - Fixes #4830.
http://git-wip-us.apache.org/repos/asf/ignite/blob/9f546f87/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
new file mode 100644
index 0000000..9140322
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java
@@ -0,0 +1,1369 @@
+/*
+ * 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.distributed.dht.topology;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.events.DiscoveryEvent;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
+import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
+import org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionFullCountersMap;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.CachePartitionPartialCountersMap;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.mvcc.MvccCoordinator;
+import org.apache.ignite.internal.util.F0;
+import org.apache.ignite.internal.util.GridAtomicLong;
+import org.apache.ignite.internal.util.GridPartitionStateMap;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.EVICTED;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.MOVING;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
+
+/**
+ * Partition topology for node which does not have any local partitions.
+ */
+@GridToStringExclude
+public class GridClientPartitionTopology implements GridDhtPartitionTopology {
+ /** */
+ private static final GridDhtPartitionState[] MOVING_STATES = new GridDhtPartitionState[] {MOVING};
+
+ /** If true, then check consistency. */
+ private static final boolean CONSISTENCY_CHECK = false;
+
+ /** Flag to control amount of output for full map. */
+ private static final boolean FULL_MAP_DEBUG = false;
+
+ /** Cache shared context. */
+ private GridCacheSharedContext cctx;
+
+ /** Cache ID. */
+ private int grpId;
+
+ /** Logger. */
+ private final IgniteLogger log;
+
+ /** Node to partition map. */
+ private GridDhtPartitionFullMap node2part;
+
+ /** Partition to node map. */
+ private final Map<Integer, Set<UUID>> part2node = new HashMap<>();
+
+ /** */
+ private AffinityTopologyVersion lastExchangeVer;
+
+ /** */
+ private AffinityTopologyVersion topVer = AffinityTopologyVersion.NONE;
+
+ /** */
+ private volatile boolean stopping;
+
+ /** A future that will be completed when topology with version topVer will be ready to use. */
+ private volatile GridDhtTopologyFuture topReadyFut;
+
+ /** */
+ private final GridAtomicLong updateSeq = new GridAtomicLong(1);
+
+ /** Lock. */
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ /** Partition update counters. */
+ private CachePartitionFullCountersMap cntrMap;
+
+ /** */
+ private final Object similarAffKey;
+
+ /** */
+ private volatile DiscoCache discoCache;
+
+ /** */
+ private final int parts;
+
+ /** */
+ private volatile Map<Integer, Long> globalPartSizes;
+
+ /**
+ * @param cctx Context.
+ * @param discoCache Discovery data cache.
+ * @param grpId Group ID.
+ * @param parts Number of partitions in the group.
+ * @param similarAffKey Key to find caches with similar affinity.
+ */
+ public GridClientPartitionTopology(
+ GridCacheSharedContext<?, ?> cctx,
+ DiscoCache discoCache,
+ int grpId,
+ int parts,
+ Object similarAffKey
+ ) {
+ this.cctx = cctx;
+ this.discoCache = discoCache;
+ this.grpId = grpId;
+ this.similarAffKey = similarAffKey;
+ this.parts = parts;
+
+ topVer = AffinityTopologyVersion.NONE;
+
+ log = cctx.logger(getClass());
+
+ node2part = new GridDhtPartitionFullMap(cctx.localNode().id(),
+ cctx.localNode().order(),
+ updateSeq.get());
+
+ cntrMap = new CachePartitionFullCountersMap(parts);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int partitions() {
+ return parts;
+ }
+
+ /**
+ * @return Key to find caches with similar affinity.
+ */
+ @Nullable public Object similarAffinityKey() {
+ return similarAffKey;
+ }
+
+ /**
+ * @return Full map string representation.
+ */
+ @SuppressWarnings({"ConstantConditions"})
+ private String fullMapString() {
+ return node2part == null ? "null" : FULL_MAP_DEBUG ? node2part.toFullString() : node2part.toString();
+ }
+
+ /**
+ * @param map Map to get string for.
+ * @return Full map string representation.
+ */
+ @SuppressWarnings({"ConstantConditions"})
+ private String mapString(GridDhtPartitionMap map) {
+ return map == null ? "null" : FULL_MAP_DEBUG ? map.toFullString() : map.toString();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int groupId() {
+ return grpId;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings({"LockAcquiredButNotSafelyReleased"})
+ @Override public void readLock() {
+ lock.readLock().lock();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readUnlock() {
+ lock.readLock().unlock();
+ }
+
+ /** {@inheritDoc} */
+ @Override public MvccCoordinator mvccCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean holdsLock() {
+ return lock.isWriteLockedByCurrentThread() || lock.getReadHoldCount() > 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void updateTopologyVersion(
+ GridDhtTopologyFuture exchFut,
+ DiscoCache discoCache,
+ MvccCoordinator mvccCrd,
+ long updSeq,
+ boolean stopping
+ ) throws IgniteInterruptedCheckedException {
+ U.writeLock(lock);
+
+ try {
+ AffinityTopologyVersion exchTopVer = exchFut.initialVersion();
+
+ assert exchTopVer.compareTo(topVer) > 0 : "Invalid topology version [grp=" + grpId +
+ ", topVer=" + topVer +
+ ", exchVer=" + exchTopVer +
+ ", discoCacheVer=" + (this.discoCache != null ? this.discoCache.version() : "None") +
+ ", exchDiscoCacheVer=" + discoCache.version() + ']';
+
+ this.stopping = stopping;
+
+ topVer = exchTopVer;
+ this.discoCache = discoCache;
+
+ updateSeq.setIfGreater(updSeq);
+
+ topReadyFut = exchFut;
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public AffinityTopologyVersion readyTopologyVersion() {
+ lock.readLock().lock();
+
+ try {
+ assert topVer.topologyVersion() > 0;
+
+ return topVer;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public AffinityTopologyVersion lastTopologyChangeVersion() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDhtTopologyFuture topologyVersionFuture() {
+ assert topReadyFut != null;
+
+ return topReadyFut;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean stopping() {
+ return stopping;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean initPartitionsWhenAffinityReady(AffinityTopologyVersion affVer,
+ GridDhtPartitionsExchangeFuture exchFut) {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void beforeExchange(GridDhtPartitionsExchangeFuture exchFut,
+ boolean initParts,
+ boolean updateMoving)
+ throws IgniteCheckedException
+ {
+ ClusterNode loc = cctx.localNode();
+
+ U.writeLock(lock);
+
+ try {
+ if (stopping)
+ return;
+
+ discoCache = exchFut.events().discoveryCache();
+
+ beforeExchange0(loc, exchFut);
+
+ if (updateMoving) {
+ ExchangeDiscoveryEvents evts = exchFut.context().events();
+
+ GridAffinityAssignmentCache aff = cctx.affinity().affinity(grpId);
+
+ assert aff.lastVersion().equals(evts.topologyVersion());
+
+ createMovingPartitions(aff.readyAffinity(evts.topologyVersion()));
+ }
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * @param aff Affinity.
+ */
+ private void createMovingPartitions(AffinityAssignment aff) {
+ for (Map.Entry<UUID, GridDhtPartitionMap> e : node2part.entrySet()) {
+ GridDhtPartitionMap map = e.getValue();
+
+ addMoving(map, aff.backupPartitions(e.getKey()));
+ addMoving(map, aff.primaryPartitions(e.getKey()));
+ }
+ }
+
+ /**
+ * @param map Node partition state map.
+ * @param parts Partitions assigned to node.
+ */
+ private void addMoving(GridDhtPartitionMap map, Set<Integer> parts) {
+ if (F.isEmpty(parts))
+ return;
+
+ for (Integer p : parts) {
+ GridDhtPartitionState state = map.get(p);
+
+ if (state == null || state == EVICTED)
+ map.put(p, MOVING);
+ }
+ }
+
+ /**
+ * @param loc Local node.
+ * @param exchFut Exchange future.
+ */
+ private void beforeExchange0(ClusterNode loc, GridDhtPartitionsExchangeFuture exchFut) {
+ GridDhtPartitionExchangeId exchId = exchFut.exchangeId();
+
+ if (exchFut.context().events().hasServerLeft()) {
+ List<DiscoveryEvent> evts0 = exchFut.context().events().events();
+
+ for (int i = 0; i < evts0.size(); i++) {
+ DiscoveryEvent evt = evts0.get(i);
+
+ if (ExchangeDiscoveryEvents.serverLeftEvent(evt))
+ removeNode(evt.eventNode().id());
+ }
+ }
+
+ // In case if node joins, get topology at the time of joining node.
+ ClusterNode oldest = discoCache.oldestAliveServerNode();
+
+ assert oldest != null;
+
+ if (log.isDebugEnabled())
+ log.debug("Partition map beforeExchange [exchId=" + exchId + ", fullMap=" + fullMapString() + ']');
+
+ long updateSeq = this.updateSeq.incrementAndGet();
+
+ // If this is the oldest node.
+ if (oldest.id().equals(loc.id()) || exchFut.dynamicCacheGroupStarted(grpId)) {
+ if (node2part == null) {
+ node2part = new GridDhtPartitionFullMap(oldest.id(), oldest.order(), updateSeq);
+
+ if (log.isDebugEnabled())
+ log.debug("Created brand new full topology map on oldest node [exchId=" +
+ exchId + ", fullMap=" + fullMapString() + ']');
+ }
+ else if (!node2part.valid()) {
+ node2part = new GridDhtPartitionFullMap(oldest.id(), oldest.order(), updateSeq, node2part, false);
+
+ if (log.isDebugEnabled())
+ log.debug("Created new full topology map on oldest node [exchId=" + exchId + ", fullMap=" +
+ node2part + ']');
+ }
+ else if (!node2part.nodeId().equals(loc.id())) {
+ node2part = new GridDhtPartitionFullMap(oldest.id(), oldest.order(), updateSeq, node2part, false);
+
+ if (log.isDebugEnabled())
+ log.debug("Copied old map into new map on oldest node (previous oldest node left) [exchId=" +
+ exchId + ", fullMap=" + fullMapString() + ']');
+ }
+ }
+
+ consistencyCheck();
+
+ if (log.isDebugEnabled())
+ log.debug("Partition map after beforeExchange [exchId=" + exchId + ", fullMap=" +
+ fullMapString() + ']');
+ }
+
+ /** {@inheritDoc} */
+ @Override public void afterStateRestored(AffinityTopologyVersion topVer) {
+ // no-op
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean afterExchange(GridDhtPartitionsExchangeFuture exchFut) throws IgniteCheckedException {
+ AffinityTopologyVersion topVer = exchFut.topologyVersion();
+
+ lock.writeLock().lock();
+
+ try {
+ assert topVer.equals(exchFut.topologyVersion()) : "Invalid topology version [topVer=" +
+ topVer + ", exchId=" + exchFut.exchangeId() + ']';
+
+ if (log.isDebugEnabled())
+ log.debug("Partition map before afterExchange [exchId=" + exchFut.exchangeId() + ", fullMap=" +
+ fullMapString() + ']');
+
+ updateSeq.incrementAndGet();
+
+ consistencyCheck();
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridDhtLocalPartition localPartition(
+ int p,
+ AffinityTopologyVersion topVer,
+ boolean create
+ )
+ throws GridDhtInvalidPartitionException {
+ if (!create)
+ return null;
+
+ throw new GridDhtInvalidPartitionException(p, "Adding entry to evicted partition (often may be caused by " +
+ "inconsistent 'key.hashCode()' implementation) " +
+ "[part=" + p + ", topVer=" + topVer + ", this.topVer=" + this.topVer + ']');
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridDhtLocalPartition localPartition(int p, AffinityTopologyVersion topVer,
+ boolean create, boolean showRenting) throws GridDhtInvalidPartitionException {
+ return localPartition(p, topVer, create);
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDhtLocalPartition forceCreatePartition(int p) throws IgniteCheckedException {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDhtLocalPartition localPartition(int p) {
+ return localPartition(p, AffinityTopologyVersion.NONE, false);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void releasePartitions(int... parts) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<GridDhtLocalPartition> localPartitions() {
+ return Collections.emptyList();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<GridDhtLocalPartition> currentLocalPartitions() {
+ return Collections.emptyList();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onRemoved(GridDhtCacheEntry e) {
+ assert false : "Entry should not be removed from client topology: " + e;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDhtPartitionMap localPartitionMap() {
+ lock.readLock().lock();
+
+ try {
+ return new GridDhtPartitionMap(cctx.localNodeId(), updateSeq.get(), topVer,
+ GridPartitionStateMap.EMPTY, true);
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDhtPartitionState partitionState(UUID nodeId, int part) {
+ lock.readLock().lock();
+
+ try {
+ GridDhtPartitionMap partMap = node2part.get(nodeId);
+
+ if (partMap != null) {
+ GridDhtPartitionState state = partMap.get(part);
+
+ return state == null ? EVICTED : state;
+ }
+
+ return EVICTED;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public List<ClusterNode> nodes(int p,
+ AffinityAssignment affAssignment,
+ List<ClusterNode> affNodes) {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<ClusterNode> nodes(int p, AffinityTopologyVersion topVer) {
+ lock.readLock().lock();
+
+ try {
+ assert node2part != null && node2part.valid() : "Invalid node-to-partitions map [topVer=" + topVer +
+ ", node2part=" + node2part + ']';
+
+ List<ClusterNode> nodes = null;
+
+ Collection<UUID> nodeIds = part2node.get(p);
+
+ if (!F.isEmpty(nodeIds)) {
+ for (UUID nodeId : nodeIds) {
+ ClusterNode n = discoCache.node(nodeId);
+
+ if (n != null && (topVer.topologyVersion() < 0 || n.order() <= topVer.topologyVersion())) {
+ if (nodes == null)
+ nodes = new ArrayList<>(nodeIds.size());
+
+ nodes.add(n);
+ }
+ }
+ }
+
+ return nodes;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * @param p Partition.
+ * @param topVer Topology version ({@code -1} for all nodes).
+ * @param state Partition state.
+ * @param states Additional partition states.
+ * @return List of nodes for the partition.
+ */
+ private List<ClusterNode> nodes(int p, AffinityTopologyVersion topVer, GridDhtPartitionState state, GridDhtPartitionState... states) {
+ Collection<UUID> allIds = F.nodeIds(discoCache.cacheGroupAffinityNodes(grpId));
+
+ lock.readLock().lock();
+
+ try {
+ assert node2part != null && node2part.valid() : "Invalid node-to-partitions map [topVer=" + topVer +
+ ", allIds=" + allIds + ", node2part=" + node2part + ']';
+
+ Collection<UUID> nodeIds = part2node.get(p);
+
+ // Node IDs can be null if both, primary and backup, nodes disappear.
+ int size = nodeIds == null ? 0 : nodeIds.size();
+
+ if (size == 0)
+ return Collections.emptyList();
+
+ List<ClusterNode> nodes = new ArrayList<>(size);
+
+ for (UUID id : nodeIds) {
+ if (topVer.topologyVersion() > 0 && !F.contains(allIds, id))
+ continue;
+
+ if (hasState(p, id, state, states)) {
+ ClusterNode n = discoCache.node(id);
+
+ if (n != null && (topVer.topologyVersion() < 0 || n.order() <= topVer.topologyVersion()))
+ nodes.add(n);
+ }
+ }
+
+ return nodes;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<ClusterNode> owners(int p, AffinityTopologyVersion topVer) {
+ return nodes(p, topVer, OWNING, null);
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<ClusterNode> owners(int p) {
+ return owners(p, AffinityTopologyVersion.NONE);
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<List<ClusterNode>> allOwners() {
+ lock.readLock().lock();
+
+ try {
+ int parts = partitions();
+
+ List<List<ClusterNode>> res = new ArrayList<>(parts);
+
+ for (int i = 0; i < parts; i++)
+ res.add(new ArrayList<>());
+
+ List<ClusterNode> allNodes = discoCache.cacheGroupAffinityNodes(grpId);
+
+ for (int i = 0; i < allNodes.size(); i++) {
+ ClusterNode node = allNodes.get(i);
+
+ GridDhtPartitionMap nodeParts = node2part.get(node.id());
+
+ if (nodeParts != null) {
+ for (Map.Entry<Integer, GridDhtPartitionState> e : nodeParts.map().entrySet()) {
+ if (e.getValue() == OWNING) {
+ int part = e.getKey();
+
+ List<ClusterNode> owners = res.get(part);
+
+ owners.add(node);
+ }
+ }
+ }
+ }
+
+ return res;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<ClusterNode> moving(int p) {
+ return nodes(p, AffinityTopologyVersion.NONE, MOVING, null);
+ }
+
+ /**
+ * @param p Partition.
+ * @param topVer Topology version.
+ * @return List of nodes in state OWNING or MOVING.
+ */
+ private List<ClusterNode> ownersAndMoving(int p, AffinityTopologyVersion topVer) {
+ return nodes(p, topVer, OWNING, MOVING_STATES);
+ }
+
+ /** {@inheritDoc} */
+ @Override public long updateSequence() {
+ return updateSeq.get();
+ }
+
+ /**
+ * @return Last update sequence.
+ */
+ public long lastUpdateSequence() {
+ lock.writeLock().lock();
+
+ try {
+ return updateSeq.incrementAndGet();
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridDhtPartitionFullMap partitionMap(boolean onlyActive) {
+ lock.readLock().lock();
+
+ try {
+ if (stopping || node2part == null)
+ return null;
+
+ assert node2part.valid() : "Invalid node2part [node2part: " + node2part +
+ ", locNodeId=" + cctx.localNodeId() +
+ ", igniteInstanceName=" + cctx.igniteInstanceName() + ']';
+
+ GridDhtPartitionFullMap m = node2part;
+
+ return new GridDhtPartitionFullMap(m.nodeId(), m.nodeOrder(), m.updateSequence(), m, onlyActive);
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Checks should current partition map overwritten by new partition map
+ * Method returns true if topology version or update sequence of new map are greater than of current map.
+ *
+ * @param currentMap Current partition map.
+ * @param newMap New partition map.
+ * @return True if current partition map should be overwritten by new partition map, false in other case.
+ */
+ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridDhtPartitionMap newMap) {
+ return newMap != null &&
+ (newMap.topologyVersion().compareTo(currentMap.topologyVersion()) > 0 ||
+ newMap.topologyVersion().compareTo(currentMap.topologyVersion()) == 0 && newMap.updateSequence() > currentMap.updateSequence());
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings({"MismatchedQueryAndUpdateOfCollection"})
+ @Override public boolean update(
+ @Nullable AffinityTopologyVersion exchangeVer,
+ GridDhtPartitionFullMap partMap,
+ @Nullable CachePartitionFullCountersMap cntrMap,
+ Set<Integer> partsToReload,
+ @Nullable Map<Integer, Long> partSizes,
+ @Nullable AffinityTopologyVersion msgTopVer) {
+ if (log.isDebugEnabled())
+ log.debug("Updating full partition map [exchVer=" + exchangeVer + ", parts=" + fullMapString() + ']');
+
+ lock.writeLock().lock();
+
+ try {
+ if (exchangeVer != null && lastExchangeVer != null && lastExchangeVer.compareTo(exchangeVer) >= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Stale exchange id for full partition map update (will ignore) [lastExchId=" +
+ lastExchangeVer + ", exchVer=" + exchangeVer + ']');
+
+ return false;
+ }
+
+ if (msgTopVer != null && lastExchangeVer != null && lastExchangeVer.compareTo(msgTopVer) > 0) {
+ if (log.isDebugEnabled())
+ log.debug("Stale topology version for full partition map update message (will ignore) " +
+ "[lastExchId=" + lastExchangeVer + ", topVersion=" + msgTopVer + ']');
+
+ return false;
+ }
+
+ boolean fullMapUpdated = (node2part == null);
+
+ if (node2part != null) {
+ for (GridDhtPartitionMap part : node2part.values()) {
+ GridDhtPartitionMap newPart = partMap.get(part.nodeId());
+
+ if (shouldOverridePartitionMap(part, newPart)) {
+ fullMapUpdated = true;
+
+ if (log.isDebugEnabled())
+ log.debug("Overriding partition map in full update map [exchId=" + exchangeVer + ", curPart=" +
+ mapString(part) + ", newPart=" + mapString(newPart) + ']');
+ }
+ else {
+ // If for some nodes current partition has a newer map,
+ // then we keep the newer value.
+ partMap.put(part.nodeId(), part);
+ }
+ }
+
+ // Check that we have new nodes.
+ for (GridDhtPartitionMap part : partMap.values()) {
+ if (fullMapUpdated)
+ break;
+
+ fullMapUpdated = !node2part.containsKey(part.nodeId());
+ }
+
+ // Remove entry if node left.
+ for (Iterator<UUID> it = partMap.keySet().iterator(); it.hasNext(); ) {
+ UUID nodeId = it.next();
+
+ if (!cctx.discovery().alive(nodeId)) {
+ if (log.isDebugEnabled())
+ log.debug("Removing left node from full map update [nodeId=" + nodeId + ", partMap=" +
+ partMap + ']');
+
+ it.remove();
+ }
+ }
+ }
+
+ if (!fullMapUpdated) {
+ if (log.isDebugEnabled())
+ log.debug("No updates for full partition map (will ignore) [lastExch=" +
+ lastExchangeVer + ", exch=" + exchangeVer + ", curMap=" + node2part + ", newMap=" + partMap + ']');
+
+ return false;
+ }
+
+ if (exchangeVer != null)
+ lastExchangeVer = exchangeVer;
+
+ node2part = partMap;
+
+ updateSeq.incrementAndGet();
+
+ part2node.clear();
+
+ for (Map.Entry<UUID, GridDhtPartitionMap> e : node2part.entrySet()) {
+ for (Map.Entry<Integer, GridDhtPartitionState> e0 : e.getValue().entrySet()) {
+ if (e0.getValue() != MOVING && e0.getValue() != OWNING)
+ continue;
+
+ int p = e0.getKey();
+
+ Set<UUID> ids = part2node.get(p);
+
+ if (ids == null)
+ // Initialize HashSet to size 3 in anticipation that there won't be
+ // more than 3 nodes per partitions.
+ part2node.put(p, ids = U.newHashSet(3));
+
+ ids.add(e.getKey());
+ }
+ }
+
+ if (cntrMap != null)
+ this.cntrMap = new CachePartitionFullCountersMap(cntrMap);
+
+ if (partSizes != null)
+ this.globalPartSizes = partSizes;
+
+ consistencyCheck();
+
+ if (log.isDebugEnabled())
+ log.debug("Partition map after full update: " + fullMapString());
+
+ return false;
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void collectUpdateCounters(CachePartitionPartialCountersMap cntrMap) {
+ assert cntrMap != null;
+
+ lock.writeLock().lock();
+
+ try {
+ for (int i = 0; i < cntrMap.size(); i++) {
+ int pId = cntrMap.partitionAt(i);
+
+ long initialUpdateCntr = cntrMap.initialUpdateCounterAt(i);
+ long updateCntr = cntrMap.updateCounterAt(i);
+
+ if (this.cntrMap.updateCounter(pId) < updateCntr) {
+ this.cntrMap.initialUpdateCounter(pId, initialUpdateCntr);
+ this.cntrMap.updateCounter(pId, updateCntr);
+ }
+ }
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void applyUpdateCounters() {
+ // No-op on client topology.
+ }
+
+ /**
+ * Method checks is new partition map more stale than current partition map
+ * New partition map is stale if topology version or update sequence are less or equal than of current map
+ *
+ * @param currentMap Current partition map
+ * @param newMap New partition map
+ * @return True if new partition map is more stale than current partition map, false in other case
+ */
+ private boolean isStaleUpdate(GridDhtPartitionMap currentMap, GridDhtPartitionMap newMap) {
+ return currentMap != null && newMap.compareTo(currentMap) <= 0;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings({"MismatchedQueryAndUpdateOfCollection"})
+ @Override public boolean update(
+ @Nullable GridDhtPartitionExchangeId exchId,
+ GridDhtPartitionMap parts,
+ boolean force
+ ) {
+ if (log.isDebugEnabled())
+ log.debug("Updating single partition map [exchId=" + exchId + ", parts=" + mapString(parts) + ']');
+
+ if (!cctx.discovery().alive(parts.nodeId())) {
+ if (log.isDebugEnabled())
+ log.debug("Received partition update for non-existing node (will ignore) [exchId=" + exchId +
+ ", parts=" + parts + ']');
+
+ return false;
+ }
+
+ lock.writeLock().lock();
+
+ try {
+ if (stopping)
+ return false;
+
+ if (!force) {
+ if (lastExchangeVer != null && exchId != null && lastExchangeVer.compareTo(exchId.topologyVersion()) > 0) {
+ if (log.isDebugEnabled())
+ log.debug("Stale exchange id for single partition map update (will ignore) [lastExchVer=" +
+ lastExchangeVer + ", exchId=" + exchId + ']');
+
+ return false;
+ }
+ }
+
+ if (exchId != null)
+ lastExchangeVer = exchId.topologyVersion();
+
+ if (node2part == null)
+ // Create invalid partition map.
+ node2part = new GridDhtPartitionFullMap();
+
+ GridDhtPartitionMap cur = node2part.get(parts.nodeId());
+
+ if (force) {
+ if (cur != null && cur.topologyVersion().initialized())
+ parts.updateSequence(cur.updateSequence(), cur.topologyVersion());
+ }
+ else if (isStaleUpdate(cur, parts)) {
+ if (log.isDebugEnabled())
+ log.debug("Stale update for single partition map update (will ignore) [exchId=" + exchId +
+ ", curMap=" + cur + ", newMap=" + parts + ']');
+
+ return false;
+ }
+
+ long updateSeq = this.updateSeq.incrementAndGet();
+
+ node2part = new GridDhtPartitionFullMap(node2part, updateSeq);
+
+ boolean changed = false;
+
+ if (cur == null || !cur.equals(parts))
+ changed = true;
+
+ node2part.put(parts.nodeId(), parts);
+
+ // Add new mappings.
+ for (Map.Entry<Integer,GridDhtPartitionState> e : parts.entrySet()) {
+ int p = e.getKey();
+
+ Set<UUID> ids = part2node.get(p);
+
+ if (e.getValue() == MOVING || e.getValue() == OWNING) {
+ if (ids == null)
+ // Initialize HashSet to size 3 in anticipation that there won't be
+ // more than 3 nodes per partition.
+ part2node.put(p, ids = U.newHashSet(3));
+
+ changed |= ids.add(parts.nodeId());
+ }
+ else {
+ if (ids != null)
+ changed |= ids.remove(parts.nodeId());
+ }
+ }
+
+ // Remove obsolete mappings.
+ if (cur != null) {
+ for (Integer p : F.view(cur.keySet(), F0.notIn(parts.keySet()))) {
+ Set<UUID> ids = part2node.get(p);
+
+ if (ids != null)
+ changed |= ids.remove(parts.nodeId());
+ }
+ }
+
+ consistencyCheck();
+
+ if (log.isDebugEnabled())
+ log.debug("Partition map after single update: " + fullMapString());
+
+ return changed;
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment,
+ boolean updateRebalanceVer) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean detectLostPartitions(AffinityTopologyVersion affVer, DiscoveryEvent discoEvt) {
+ assert false : "detectLostPartitions should never be called on client topology";
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void resetLostPartitions(AffinityTopologyVersion affVer) {
+ assert false : "resetLostPartitions should never be called on client topology";
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<Integer> lostPartitions() {
+ assert false : "lostPartitions should never be called on client topology";
+
+ return Collections.emptyList();
+ }
+
+ /**
+ * Updates value for single partition.
+ *
+ * @param p Partition.
+ * @param nodeId Node ID.
+ * @param state State.
+ * @param updateSeq Update sequence.
+ */
+ private void updateLocal(int p, UUID nodeId, GridDhtPartitionState state, long updateSeq) {
+ assert lock.isWriteLockedByCurrentThread();
+ assert nodeId.equals(cctx.localNodeId());
+
+ // In case if node joins, get topology at the time of joining node.
+ ClusterNode oldest = discoCache.oldestAliveServerNode();
+
+ // If this node became the oldest node.
+ if (cctx.localNode().equals(oldest)) {
+ long seq = node2part.updateSequence();
+
+ if (seq != updateSeq) {
+ if (seq > updateSeq) {
+ if (this.updateSeq.get() < seq) {
+ // Update global counter if necessary.
+ boolean b = this.updateSeq.compareAndSet(this.updateSeq.get(), seq + 1);
+
+ assert b : "Invalid update sequence [updateSeq=" + updateSeq + ", seq=" + seq +
+ ", curUpdateSeq=" + this.updateSeq.get() + ", node2part=" + node2part.toFullString() + ']';
+
+ updateSeq = seq + 1;
+ }
+ else
+ updateSeq = seq;
+ }
+
+ node2part.updateSequence(updateSeq);
+ }
+ }
+
+ GridDhtPartitionMap map = node2part.get(nodeId);
+
+ if (map == null)
+ node2part.put(nodeId, map = new GridDhtPartitionMap(nodeId, updateSeq, topVer,
+ GridPartitionStateMap.EMPTY, false));
+
+ map.updateSequence(updateSeq, topVer);
+
+ map.put(p, state);
+
+ Set<UUID> ids = part2node.get(p);
+
+ if (ids == null)
+ part2node.put(p, ids = U.newHashSet(3));
+
+ ids.add(nodeId);
+ }
+
+ /**
+ * @param nodeId Node to remove.
+ */
+ private void removeNode(UUID nodeId) {
+ assert nodeId != null;
+ assert lock.writeLock().isHeldByCurrentThread();
+
+ ClusterNode loc = cctx.localNode();
+
+ if (node2part != null) {
+ if (!node2part.nodeId().equals(loc.id())) {
+ updateSeq.setIfGreater(node2part.updateSequence());
+
+ node2part = new GridDhtPartitionFullMap(loc.id(), loc.order(), updateSeq.incrementAndGet(),
+ node2part, false);
+ }
+ else
+ node2part = new GridDhtPartitionFullMap(node2part, node2part.updateSequence());
+
+ GridDhtPartitionMap parts = node2part.remove(nodeId);
+
+ if (parts != null) {
+ for (Integer p : parts.keySet()) {
+ Set<UUID> nodeIds = part2node.get(p);
+
+ if (nodeIds != null) {
+ nodeIds.remove(nodeId);
+
+ if (nodeIds.isEmpty())
+ part2node.remove(p);
+ }
+ }
+ }
+
+ consistencyCheck();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean own(GridDhtLocalPartition part) {
+ assert false : "Client topology should never own a partition: " + part;
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void ownMoving(AffinityTopologyVersion topVer) {
+ // No-op
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onEvicted(GridDhtLocalPartition part, boolean updateSeq) {
+ assert updateSeq || lock.isWriteLockedByCurrentThread();
+
+ lock.writeLock().lock();
+
+ try {
+ if (stopping)
+ return;
+
+ assert part.state() == EVICTED;
+
+ long seq = updateSeq ? this.updateSeq.incrementAndGet() : this.updateSeq.get();
+
+ updateLocal(part.id(), cctx.localNodeId(), part.state(), seq);
+
+ consistencyCheck();
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridDhtPartitionMap partitions(UUID nodeId) {
+ lock.readLock().lock();
+
+ try {
+ return node2part.get(nodeId);
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public Map<UUID, Set<Integer>> resetOwners(Map<Integer, Set<UUID>> ownersByUpdCounters, Set<Integer> haveHistory) {
+ Map<UUID, Set<Integer>> result = new HashMap<>();
+
+ lock.writeLock().lock();
+
+ try {
+ // Process remote partitions.
+ for (Map.Entry<Integer, Set<UUID>> entry : ownersByUpdCounters.entrySet()) {
+ int part = entry.getKey();
+ Set<UUID> newOwners = entry.getValue();
+
+ for (Map.Entry<UUID, GridDhtPartitionMap> remotes : node2part.entrySet()) {
+ UUID remoteNodeId = remotes.getKey();
+ GridDhtPartitionMap partMap = remotes.getValue();
+
+ GridDhtPartitionState state = partMap.get(part);
+
+ if (state == null || state != OWNING)
+ continue;
+
+ if (!newOwners.contains(remoteNodeId)) {
+ partMap.put(part, MOVING);
+
+ partMap.updateSequence(partMap.updateSequence() + 1, partMap.topologyVersion());
+
+ result.computeIfAbsent(remoteNodeId, n -> new HashSet<>());
+ result.get(remoteNodeId).add(part);
+ }
+ }
+ }
+
+ for (Map.Entry<UUID, Set<Integer>> entry : result.entrySet()) {
+ UUID nodeId = entry.getKey();
+ Set<Integer> partsToRebalance = entry.getValue();
+
+ if (!partsToRebalance.isEmpty()) {
+ Set<Integer> historical = partsToRebalance.stream()
+ .filter(haveHistory::contains)
+ .collect(Collectors.toSet());
+
+ // Filter out partitions having WAL history.
+ partsToRebalance.removeAll(historical);
+
+ U.warn(log, "Partitions have been scheduled for rebalancing due to outdated update counter "
+ + "[grpId=" + grpId
+ + ", nodeId=" + nodeId
+ + ", partsFull=" + S.compact(partsToRebalance)
+ + ", partsHistorical=" + S.compact(historical) + "]");
+ }
+ }
+
+ for (Map.Entry<Integer, Set<UUID>> entry : ownersByUpdCounters.entrySet())
+ part2node.put(entry.getKey(), entry.getValue());
+
+ updateSeq.incrementAndGet();
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ return result;
+ }
+
+ /** {@inheritDoc} */
+ @Override public CachePartitionFullCountersMap fullUpdateCounters() {
+ lock.readLock().lock();
+
+ try {
+ return new CachePartitionFullCountersMap(cntrMap);
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public CachePartitionPartialCountersMap localUpdateCounters(boolean skipZeros) {
+ return CachePartitionPartialCountersMap.EMPTY;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Map<Integer, Long> partitionSizes() {
+ return Collections.emptyMap();
+ }
+
+ /** {@inheritDoc} */
+ @Override @Nullable public Map<Integer, Long> globalPartSizes() {
+ lock.readLock().lock();
+
+ try {
+ if (globalPartSizes == null)
+ return Collections.emptyMap();
+
+ return Collections.unmodifiableMap(globalPartSizes);
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void globalPartSizes(@Nullable Map<Integer, Long> partSizes) {
+ lock.writeLock().lock();
+
+ try {
+ this.globalPartSizes = partSizes;
+ }
+ finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+
+ /** {@inheritDoc} */
+ @Override public boolean rebalanceFinished(AffinityTopologyVersion topVer) {
+ assert false : "Should not be called on non-affinity node";
+
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasMovingPartitions() {
+ lock.readLock().lock();
+
+ try {
+ assert node2part != null && node2part.valid() : "Invalid node2part [node2part: " + node2part +
+ ", locNodeId=" + cctx.localNodeId() +
+ ", igniteInstanceName=" + cctx.igniteInstanceName() + ']';
+
+ for (GridDhtPartitionMap map : node2part.values()) {
+ if (map.hasMovingPartitions())
+ return true;
+ }
+
+ return false;
+ }
+ finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void printMemoryStats(int threshold) {
+ X.println(">>> Cache partition topology stats [igniteInstanceName=" + cctx.igniteInstanceName() +
+ ", grpId=" + grpId + ']');
+ }
+
+ /**
+ * @param p Partition.
+ * @param nodeId Node ID.
+ * @param match State to match.
+ * @param matches Additional states.
+ * @return Filter for owners of this partition.
+ */
+ private boolean hasState(final int p, @Nullable UUID nodeId, final GridDhtPartitionState match,
+ final GridDhtPartitionState... matches) {
+ if (nodeId == null)
+ return false;
+
+ GridDhtPartitionMap parts = node2part.get(nodeId);
+
+ // Set can be null if node has been removed.
+ if (parts != null) {
+ GridDhtPartitionState state = parts.get(p);
+
+ if (state == match)
+ return true;
+
+ if (matches != null && matches.length > 0)
+ for (GridDhtPartitionState s : matches)
+ if (state == s)
+ return true;
+ }
+
+ return false;
+ }
+
+ /**
+ * Checks consistency after all operations.
+ */
+ private void consistencyCheck() {
+ if (CONSISTENCY_CHECK) {
+ assert lock.writeLock().isHeldByCurrentThread();
+
+ if (node2part == null)
+ return;
+
+ for (Map.Entry<UUID, GridDhtPartitionMap> e : node2part.entrySet()) {
+ for (Integer p : e.getValue().keySet()) {
+ Set<UUID> nodeIds = part2node.get(p);
+
+ assert nodeIds != null : "Failed consistency check [part=" + p + ", nodeId=" + e.getKey() + ']';
+ assert nodeIds.contains(e.getKey()) : "Failed consistency check [part=" + p + ", nodeId=" +
+ e.getKey() + ", nodeIds=" + nodeIds + ']';
+ }
+ }
+
+ for (Map.Entry<Integer, Set<UUID>> e : part2node.entrySet()) {
+ for (UUID nodeId : e.getValue()) {
+ GridDhtPartitionMap map = node2part.get(nodeId);
+
+ assert map != null : "Failed consistency check [part=" + e.getKey() + ", nodeId=" + nodeId + ']';
+ assert map.containsKey(e.getKey()) : "Failed consistency check [part=" + e.getKey() +
+ ", nodeId=" + nodeId + ']';
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/9f546f87/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtInvalidPartitionException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtInvalidPartitionException.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtInvalidPartitionException.java
new file mode 100644
index 0000000..6a2ca6b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtInvalidPartitionException.java
@@ -0,0 +1,51 @@
+/*
+ * 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.distributed.dht.topology;
+
+/**
+ * Exception thrown whenever entry is created for invalid partition.
+ */
+public class GridDhtInvalidPartitionException extends RuntimeException {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Partition. */
+ private final int part;
+
+ /**
+ * @param part Partition.
+ * @param msg Message.
+ */
+ public GridDhtInvalidPartitionException(int part, String msg) {
+ super(msg);
+
+ this.part = part;
+ }
+
+ /**
+ * @return Partition.
+ */
+ public int partition() {
+ return part;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return getClass() + " [part=" + part + ", msg=" + getMessage() + ']';
+ }
+}
\ No newline at end of file