From c000da13563907b99fe220a7c8bde3c1dec74ad5 Mon Sep 17 00:00:00 2001 From: Branimir Lambov Date: Wed, 26 Aug 2015 16:08:57 +0300 Subject: [PATCH] Improve NTS endpoints calculation patch by Branimir Lambov; reviewed by Aleksey Yeschenko for CASSANDRA-10200 --- CHANGES.txt | 1 + .../locator/NetworkTopologyStrategy.java | 157 +++++++------ .../cassandra/locator/TokenMetadata.java | 21 +- .../locator/NetworkTopologyStrategyTest.java | 213 +++++++++++++++++- 4 files changed, 317 insertions(+), 75 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 2710ed306c..77034ef4ed 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.2 + * Improve NTS endpoints calculation (CASSANDRA-10200) * Improve performance of the folderSize function (CASSANDRA-10677) * Add support for type casting in selection clause (CASSANDRA-10310) * Added graphing option to cassandra-stress (CASSANDRA-7918) diff --git a/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java b/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java index 307a07f072..9f74dcca60 100644 --- a/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java +++ b/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java @@ -28,6 +28,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.dht.Token; import org.apache.cassandra.locator.TokenMetadata.Topology; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; import com.google.common.collect.Multimap; @@ -48,14 +49,12 @@ import com.google.common.collect.Multimap; */ public class NetworkTopologyStrategy extends AbstractReplicationStrategy { - private final IEndpointSnitch snitch; private final Map datacenters; private static final Logger logger = LoggerFactory.getLogger(NetworkTopologyStrategy.class); public NetworkTopologyStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) throws ConfigurationException { super(keyspaceName, tokenMetadata, snitch, configOptions); - this.snitch = snitch; Map newDatacenters = new HashMap(); if (configOptions != null) @@ -75,17 +74,78 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy } /** - * calculate endpoints in one pass through the tokens by tracking our progress in each DC, rack etc. + * Endpoint adder applying the replication rules for a given DC. + */ + private static final class DatacenterEndpoints + { + /** List accepted endpoints get pushed into. */ + Set endpoints; + /** + * Racks encountered so far. Replicas are put into separate racks while possible. + * For efficiency the set is shared between the instances, using the location pair (dc, rack) to make sure + * clashing names aren't a problem. + */ + Set> racks; + + /** Number of replicas left to fill from this DC. */ + int rfLeft; + int acceptableRackRepeats; + + DatacenterEndpoints(int rf, int rackCount, int nodeCount, Set endpoints, Set> racks) + { + this.endpoints = endpoints; + this.racks = racks; + // If there aren't enough nodes in this DC to fill the RF, the number of nodes is the effective RF. + this.rfLeft = Math.min(rf, nodeCount); + // If there aren't enough racks in this DC to fill the RF, we'll still use at least one node from each rack, + // and the difference is to be filled by the first encountered nodes. + acceptableRackRepeats = rf - rackCount; + } + + /** + * Attempts to add an endpoint to the replicas for this datacenter, adding to the endpoints set if successful. + * Returns true if the endpoint was added, and this datacenter does not require further replicas. + */ + boolean addEndpointAndCheckIfDone(InetAddress ep, Pair location) + { + if (done()) + return false; + + if (racks.add(location)) + { + // New rack. + --rfLeft; + boolean added = endpoints.add(ep); + assert added; + return done(); + } + if (acceptableRackRepeats <= 0) + // There must be rfLeft distinct racks left, do not add any more rack repeats. + return false; + if (!endpoints.add(ep)) + // Cannot repeat a node. + return false; + // Added a node that is from an already met rack to match RF when there aren't enough racks. + --acceptableRackRepeats; + --rfLeft; + return done(); + } + + boolean done() + { + assert rfLeft >= 0; + return rfLeft == 0; + } + } + + /** + * calculate endpoints in one pass through the tokens by tracking our progress in each DC. */ - @SuppressWarnings("serial") public List calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata) { // we want to preserve insertion order so that the first added endpoint becomes primary Set replicas = new LinkedHashSet<>(); - // replicas we have found in each DC - Map> dcReplicas = new HashMap<>(datacenters.size()); - for (Map.Entry dc : datacenters.entrySet()) - dcReplicas.put(dc.getKey(), new HashSet(dc.getValue())); + Set> seenRacks = new HashSet<>(); Topology topology = tokenMetadata.getTopology(); // all endpoints in each DC, so we can check when we have exhausted all the members of a DC @@ -94,74 +154,45 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy Map> racks = topology.getDatacenterRacks(); assert !allEndpoints.isEmpty() && !racks.isEmpty() : "not aware of any cluster members"; - // tracks the racks we have already placed replicas in - Map> seenRacks = new HashMap<>(datacenters.size()); - for (Map.Entry dc : datacenters.entrySet()) - seenRacks.put(dc.getKey(), new HashSet()); + int dcsToFill = 0; + Map dcs = new HashMap<>(datacenters.size() * 2); - // tracks the endpoints that we skipped over while looking for unique racks - // when we relax the rack uniqueness we can append this to the current result so we don't have to wind back the iterator - Map> skippedDcEndpoints = new HashMap<>(datacenters.size()); - for (Map.Entry dc : datacenters.entrySet()) - skippedDcEndpoints.put(dc.getKey(), new LinkedHashSet()); + // Create a DatacenterEndpoints object for each non-empty DC. + for (Map.Entry en : datacenters.entrySet()) + { + String dc = en.getKey(); + int rf = en.getValue(); + int nodeCount = sizeOrZero(allEndpoints.get(dc)); + + if (rf <= 0 || nodeCount <= 0) + continue; + + DatacenterEndpoints dcEndpoints = new DatacenterEndpoints(rf, sizeOrZero(racks.get(dc)), nodeCount, replicas, seenRacks); + dcs.put(dc, dcEndpoints); + ++dcsToFill; + } Iterator tokenIter = TokenMetadata.ringIterator(tokenMetadata.sortedTokens(), searchToken, false); - while (tokenIter.hasNext() && !hasSufficientReplicas(dcReplicas, allEndpoints)) + while (dcsToFill > 0 && tokenIter.hasNext()) { Token next = tokenIter.next(); InetAddress ep = tokenMetadata.getEndpoint(next); - String dc = snitch.getDatacenter(ep); - // have we already found all replicas for this dc? - if (!datacenters.containsKey(dc) || hasSufficientReplicas(dc, dcReplicas, allEndpoints)) - continue; - // can we skip checking the rack? - if (seenRacks.get(dc).size() == racks.get(dc).keySet().size()) - { - dcReplicas.get(dc).add(ep); - replicas.add(ep); - } - else - { - String rack = snitch.getRack(ep); - // is this a new rack? - if (seenRacks.get(dc).contains(rack)) - { - skippedDcEndpoints.get(dc).add(ep); - } - else - { - dcReplicas.get(dc).add(ep); - replicas.add(ep); - seenRacks.get(dc).add(rack); - // if we've run out of distinct racks, add the hosts we skipped past already (up to RF) - if (seenRacks.get(dc).size() == racks.get(dc).keySet().size()) - { - Iterator skippedIt = skippedDcEndpoints.get(dc).iterator(); - while (skippedIt.hasNext() && !hasSufficientReplicas(dc, dcReplicas, allEndpoints)) - { - InetAddress nextSkipped = skippedIt.next(); - dcReplicas.get(dc).add(nextSkipped); - replicas.add(nextSkipped); - } - } - } - } + Pair location = topology.getLocation(ep); + DatacenterEndpoints dcEndpoints = dcs.get(location.left); + if (dcEndpoints != null && dcEndpoints.addEndpointAndCheckIfDone(ep, location)) + --dcsToFill; } - - return new ArrayList(replicas); + return new ArrayList<>(replicas); } - private boolean hasSufficientReplicas(String dc, Map> dcReplicas, Multimap allEndpoints) + private int sizeOrZero(Multimap collection) { - return dcReplicas.get(dc).size() >= Math.min(allEndpoints.get(dc).size(), getReplicationFactor(dc)); + return collection != null ? collection.asMap().size() : 0; } - private boolean hasSufficientReplicas(Map> dcReplicas, Multimap allEndpoints) + private int sizeOrZero(Collection collection) { - for (String dc : datacenters.keySet()) - if (!hasSufficientReplicas(dc, dcReplicas, allEndpoints)) - return false; - return true; + return collection != null ? collection.size() : 0; } public int getReplicationFactor() diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index e65b53e7c7..a3be9dedd3 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/java/org/apache/cassandra/locator/TokenMetadata.java @@ -828,20 +828,20 @@ public class TokenMetadata public Token getPredecessor(Token token) { - List tokens = sortedTokens(); + List tokens = sortedTokens(); int index = Collections.binarySearch(tokens, token); // assert index >= 0 : token + " not found in " + StringUtils.join(tokenToEndpointMap.keySet(), ", "); if (index < 0) index = -index-1; - return (Token) (index == 0 ? tokens.get(tokens.size() - 1) : tokens.get(index - 1)); + return index == 0 ? tokens.get(tokens.size() - 1) : tokens.get(index - 1); } public Token getSuccessor(Token token) { - List tokens = sortedTokens(); + List tokens = sortedTokens(); int index = Collections.binarySearch(tokens, token); // assert index >= 0 : token + " not found in " + StringUtils.join(tokenToEndpointMap.keySet(), ", "); if (index < 0) return (Token) tokens.get(-index-1); - return (Token) ((index == (tokens.size() - 1)) ? tokens.get(0) : tokens.get(index + 1)); + return (index == (tokens.size() - 1)) ? tokens.get(0) : tokens.get(index + 1); } /** @return a copy of the bootstrapping tokens map */ @@ -902,7 +902,7 @@ public class TokenMetadata } } - public static int firstTokenIndex(final ArrayList ring, Token start, boolean insertMin) + public static int firstTokenIndex(final ArrayList ring, Token start, boolean insertMin) { assert ring.size() > 0; // insert the minimum token (at index == -1) if we were asked to include it and it isn't a member of the ring @@ -930,7 +930,7 @@ public class TokenMetadata { if (ring.isEmpty()) return includeMin ? Iterators.singletonIterator(start.getPartitioner().getMinimumToken()) - : Iterators.emptyIterator(); + : Collections.emptyIterator(); final boolean insertMin = includeMin && !ring.get(0).isMinimum(); final int startIndex = firstTokenIndex(ring, start, insertMin); @@ -1279,5 +1279,14 @@ public class TokenMetadata { return dcRacks; } + + /** + * @return The DC and rack of the given endpoint. + */ + public Pair getLocation(InetAddress addr) + { + return currentLocations.get(addr); + } + } } diff --git a/test/unit/org/apache/cassandra/locator/NetworkTopologyStrategyTest.java b/test/unit/org/apache/cassandra/locator/NetworkTopologyStrategyTest.java index bbfdd3b208..3cba32827c 100644 --- a/test/unit/org/apache/cassandra/locator/NetworkTopologyStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/NetworkTopologyStrategyTest.java @@ -21,24 +21,26 @@ package org.apache.cassandra.locator; import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; +import java.util.stream.Collectors; import com.google.common.collect.HashMultimap; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.Multimap; + import org.junit.Assert; import org.junit.Test; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.dht.OrderPreservingPartitioner.StringToken; import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.locator.TokenMetadata.Topology; +import org.apache.cassandra.service.StorageService; public class NetworkTopologyStrategyTest { @@ -166,4 +168,203 @@ public class NetworkTopologyStrategyTest InetAddress add1 = InetAddress.getByAddress(bytes); metadata.updateNormalToken(token1, add1); } + + @Test + public void testCalculateEndpoints() throws UnknownHostException + { + final int NODES = 100; + final int VNODES = 64; + final int RUNS = 10; + StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance); + Map datacenters = ImmutableMap.of("rf1", 1, "rf3", 3, "rf5_1", 5, "rf5_2", 5, "rf5_3", 5); + List nodes = new ArrayList<>(NODES); + for (byte i=0; i datacenters, Random rand) + { + NetworkTopologyStrategy nts = new NetworkTopologyStrategy("ks", tokenMetadata, snitch, + datacenters.entrySet().stream(). + collect(Collectors.toMap(x -> x.getKey(), x -> Integer.toString(x.getValue())))); + for (int i=0; i<1000; ++i) + { + Token token = Murmur3Partitioner.instance.getRandomToken(rand); + List expected = calculateNaturalEndpoints(token, tokenMetadata, datacenters, snitch); + List actual = nts.calculateNaturalEndpoints(token, tokenMetadata); + if (endpointsDiffer(expected, actual)) + { + System.err.println("Endpoints mismatch for token " + token); + System.err.println(" expected: " + expected); + System.err.println(" actual : " + actual); + Assert.assertEquals("Endpoints for token " + token + " mismatch.", expected, actual); + } + } + } + + private boolean endpointsDiffer(List ep1, List ep2) + { + // Because the old algorithm does not put the nodes in the correct order in the case where more replicas + // are required than there are racks in a dc, we accept different order as long as the primary + // replica is the same. + if (ep1.equals(ep2)) + return false; + if (!ep1.get(0).equals(ep2.get(0))) + return true; + Set s1 = new HashSet<>(ep1); + Set s2 = new HashSet<>(ep2); + return !s1.equals(s2); + } + + IEndpointSnitch generateSnitch(Map datacenters, Collection nodes, Random rand) + { + final Map nodeToRack = new HashMap<>(); + final Map nodeToDC = new HashMap<>(); + Map> racksPerDC = new HashMap<>(); + datacenters.forEach((dc, rf) -> racksPerDC.put(dc, randomRacks(rf, rand))); + int rf = datacenters.values().stream().mapToInt(x -> x).sum(); + String[] dcs = new String[rf]; + int pos = 0; + for (Map.Entry dce : datacenters.entrySet()) + { + for (int i = 0; i < dce.getValue(); ++i) + dcs[pos++] = dce.getKey(); + } + + for (InetAddress node : nodes) + { + String dc = dcs[rand.nextInt(rf)]; + List racks = racksPerDC.get(dc); + String rack = racks.get(rand.nextInt(racks.size())); + nodeToRack.put(node, rack); + nodeToDC.put(node, dc); + } + + return new AbstractNetworkTopologySnitch() + { + public String getRack(InetAddress endpoint) + { + return nodeToRack.get(endpoint); + } + + public String getDatacenter(InetAddress endpoint) + { + return nodeToDC.get(endpoint); + } + }; + } + + private List randomRacks(int rf, Random rand) + { + int rc = rand.nextInt(rf * 3 - 1) + 1; + List racks = new ArrayList<>(rc); + for (int i=0; i calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata, Map datacenters, IEndpointSnitch snitch) + { + // we want to preserve insertion order so that the first added endpoint becomes primary + Set replicas = new LinkedHashSet<>(); + // replicas we have found in each DC + Map> dcReplicas = new HashMap<>(datacenters.size()); + for (Map.Entry dc : datacenters.entrySet()) + dcReplicas.put(dc.getKey(), new HashSet(dc.getValue())); + + Topology topology = tokenMetadata.getTopology(); + // all endpoints in each DC, so we can check when we have exhausted all the members of a DC + Multimap allEndpoints = topology.getDatacenterEndpoints(); + // all racks in a DC so we can check when we have exhausted all racks in a DC + Map> racks = topology.getDatacenterRacks(); + assert !allEndpoints.isEmpty() && !racks.isEmpty() : "not aware of any cluster members"; + + // tracks the racks we have already placed replicas in + Map> seenRacks = new HashMap<>(datacenters.size()); + for (Map.Entry dc : datacenters.entrySet()) + seenRacks.put(dc.getKey(), new HashSet()); + + // tracks the endpoints that we skipped over while looking for unique racks + // when we relax the rack uniqueness we can append this to the current result so we don't have to wind back the iterator + Map> skippedDcEndpoints = new HashMap<>(datacenters.size()); + for (Map.Entry dc : datacenters.entrySet()) + skippedDcEndpoints.put(dc.getKey(), new LinkedHashSet()); + + Iterator tokenIter = TokenMetadata.ringIterator(tokenMetadata.sortedTokens(), searchToken, false); + while (tokenIter.hasNext() && !hasSufficientReplicas(dcReplicas, allEndpoints, datacenters)) + { + Token next = tokenIter.next(); + InetAddress ep = tokenMetadata.getEndpoint(next); + String dc = snitch.getDatacenter(ep); + // have we already found all replicas for this dc? + if (!datacenters.containsKey(dc) || hasSufficientReplicas(dc, dcReplicas, allEndpoints, datacenters)) + continue; + // can we skip checking the rack? + if (seenRacks.get(dc).size() == racks.get(dc).keySet().size()) + { + dcReplicas.get(dc).add(ep); + replicas.add(ep); + } + else + { + String rack = snitch.getRack(ep); + // is this a new rack? + if (seenRacks.get(dc).contains(rack)) + { + skippedDcEndpoints.get(dc).add(ep); + } + else + { + dcReplicas.get(dc).add(ep); + replicas.add(ep); + seenRacks.get(dc).add(rack); + // if we've run out of distinct racks, add the hosts we skipped past already (up to RF) + if (seenRacks.get(dc).size() == racks.get(dc).keySet().size()) + { + Iterator skippedIt = skippedDcEndpoints.get(dc).iterator(); + while (skippedIt.hasNext() && !hasSufficientReplicas(dc, dcReplicas, allEndpoints, datacenters)) + { + InetAddress nextSkipped = skippedIt.next(); + dcReplicas.get(dc).add(nextSkipped); + replicas.add(nextSkipped); + } + } + } + } + } + + return new ArrayList(replicas); + } + + private static boolean hasSufficientReplicas(String dc, Map> dcReplicas, Multimap allEndpoints, Map datacenters) + { + return dcReplicas.get(dc).size() >= Math.min(allEndpoints.get(dc).size(), getReplicationFactor(dc, datacenters)); + } + + private static boolean hasSufficientReplicas(Map> dcReplicas, Multimap allEndpoints, Map datacenters) + { + for (String dc : datacenters.keySet()) + if (!hasSufficientReplicas(dc, dcReplicas, allEndpoints, datacenters)) + return false; + return true; + } + + public static int getReplicationFactor(String dc, Map datacenters) + { + Integer replicas = datacenters.get(dc); + return replicas == null ? 0 : replicas; + } }