diff --git a/CHANGES.txt b/CHANGES.txt index 9cd6040369..5858c198c3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0-alpha4 + * Improve the algorithmic token allocation in case racks = RF (CASSANDRA-15600) * Include finalized pending sstables in preview repair (CASSANDRA-15553) * Reverted to the original behavior of CLUSTERING ORDER on CREATE TABLE (CASSANDRA-15271) * Correct inaccurate logging message (CASSANDRA-15549) diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java b/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java index f6a159208b..0ac8951765 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java @@ -114,24 +114,6 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase unitTokens.add(new Weighted(token.replicatedOwnership, token)); } - private Collection generateRandomTokens(UnitInfo newUnit, int numTokens, Map> unitInfos) - { - Set tokens = new HashSet<>(numTokens); - while (tokens.size() < numTokens) - { - Token token = partitioner.getRandomToken(); - if (!sortedTokens.containsKey(token)) - { - tokens.add(token); - sortedTokens.put(token, newUnit.unit); - } - } - unitInfos.put(newUnit.unit, newUnit); - createTokenInfos(unitInfos); - TokenAllocatorDiagnostics.randomTokensGenerated(this, numTokens, sortedUnits, sortedTokens, newUnit.unit, tokens); - return tokens; - } - public Collection addUnit(Unit newUnit, int numTokens) { assert !tokensInUnits.containsKey(newUnit); @@ -141,10 +123,10 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase Map> unitInfos = createUnitInfos(groups); if (unitInfos.isEmpty()) - return generateRandomTokens(newUnitInfo, numTokens, unitInfos); + return generateSplits(newUnit, numTokens); if (numTokens > sortedTokens.size()) - return generateRandomTokens(newUnitInfo, numTokens, unitInfos); + return generateSplits(newUnit, numTokens); TokenInfo head = createTokenInfos(unitInfos); @@ -172,7 +154,19 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase } List newTokens = Lists.newArrayListWithCapacity(numTokens); + // Generate different size nodes, at most at 2/(numTokens*2+1) difference, + // but tighten the spread as the number of nodes grows (since it increases the time until we need to use nodes + // we have just split). + double sizeCorrection = Math.min(1.0, (numTokens + 1.0) / (unitInfos.size() + 1.0)); + double spread = targetAverage * sizeCorrection * 2.0 / (2 * numTokens + 1); + // The biggest target is assigned to the biggest existing node. This should result in better balance in + // the amount of data that needs to be streamed from the different sources to the new node. + double target = targetAverage + spread / 2; + + // This step intentionally divides by the count (rather than count - 1) because we also need to count the new + // node. This leaves the last position in the spread (i.e. the smallest size, least data to stream) for it. + double step = spread / unitsToChange.size(); int nr = 0; // calculate the tokens for (Weighted unit : unitsToChange) @@ -193,7 +187,7 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase unit.value.ownership -= wt.weight; } - double toTakeOver = unit.weight - targetAverage; + double toTakeOver = unit.weight - target; // Split toTakeOver proportionally between the vnodes. for (Weighted wt : tokens) { @@ -230,6 +224,7 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase // adjust the weight for current unit sortedUnits.add(new Weighted<>(unit.value.ownership, unit.value)); + target -= step; ++nr; } sortedUnits.add(new Weighted<>(newUnitInfo.ownership, newUnitInfo)); @@ -267,4 +262,9 @@ public class NoReplicationTokenAllocator extends TokenAllocatorBase { return 1; } + + public String toString() + { + return getClass().getSimpleName(); + } } diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java b/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java index 36fc8c22e0..eb498ce55f 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java @@ -61,9 +61,12 @@ class ReplicationAwareTokenAllocator extends TokenAllocatorBase if (unitCount() < replicas) // Allocation does not matter; everything replicates everywhere. + //However, at this point it is + // important to start the cluster/datacenter with suitably varied token range sizes so that the algorithm + // can maintain good balance for any number of nodes. return generateRandomTokens(newUnit, numTokens); if (numTokens > sortedTokens.size()) - // Some of the heuristics below can't deal with this case. Use random for now, later allocations can fix any problems this may cause. + // Some of the heuristics below can't deal with this very unlikely case. Use splits for now, later allocations can fix any problems this may cause. return generateRandomTokens(newUnit, numTokens); // ============= construct our initial token ring state ============= @@ -154,6 +157,13 @@ class ReplicationAwareTokenAllocator extends TokenAllocatorBase return tokens; } + Collection generateSplits(Unit newUnit, int numTokens) + { + Collection tokens = super.generateSplits(newUnit, numTokens); + unitToTokens.putAll(newUnit, tokens); + return tokens; + } + /** * Construct the token ring as a CircularList of TokenInfo, * and populate the ownership of the UnitInfo's provided diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java index bfa281e0c2..bd6d9802db 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java @@ -255,7 +255,7 @@ public class TokenAllocation ? topology.getDatacenterRacks().get(dc).asMap().size() : 1; - if (racks >= replicas) + if (racks > replicas) { return new StrategyAdapter() { @@ -278,6 +278,32 @@ public class TokenAllocation } }; } + else if (racks == replicas) + { + // When the number of racks is the same as the replication factor, everything must replicate exactly once + // in each rack. This is the same as having independent rings from each rack. + final String rack = snitch.getRack(endpoint); + return new StrategyAdapter() + { + @Override + public int replicas() + { + return 1; + } + + @Override + public Object getGroup(InetAddressAndPort unit) + { + return unit; + } + + @Override + public boolean inAllocationRing(InetAddressAndPort other) + { + return dc.equals(snitch.getDatacenter(other)) && rack.equals(snitch.getRack(other)); + } + }; + } else if (racks == 1) { // One rack, each node treated as separate. diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorBase.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorBase.java index f59bfd4696..3d7e6b9656 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorBase.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorBase.java @@ -18,9 +18,13 @@ package org.apache.cassandra.dht.tokenallocator; +import java.util.Collection; +import java.util.List; import java.util.Map; import java.util.NavigableMap; +import java.util.Random; +import com.google.common.collect.Lists; import com.google.common.collect.Maps; import org.apache.cassandra.dht.IPartitioner; @@ -28,6 +32,9 @@ import org.apache.cassandra.dht.Token; public abstract class TokenAllocatorBase implements TokenAllocator { + static final double MIN_INITIAL_SPLITS_RATIO = 1.0 - 1.0 / Math.sqrt(5.0); + static final double MAX_INITIAL_SPLITS_RATIO = MIN_INITIAL_SPLITS_RATIO + 0.075; + final NavigableMap sortedTokens; final ReplicationStrategy strategy; final IPartitioner partitioner; @@ -79,6 +86,59 @@ public abstract class TokenAllocatorBase implements TokenAllocator return group; } + Collection generateSplits(Unit newUnit, int numTokens) + { + return generateSplits(newUnit, numTokens, MIN_INITIAL_SPLITS_RATIO, MAX_INITIAL_SPLITS_RATIO); + } + /** + * Selects tokens by repeatedly splitting the largest range in the ring at the given ratio. + * + * This is used to choose tokens for the first nodes in the ring where the algorithm cannot be applied (e.g. when + * number of nodes < RF). It generates a reasonably chaotic initial token split, after which the algorithm behaves + * well for an unbounded number of nodes. + */ + Collection generateSplits(Unit newUnit, int numTokens, double minRatio, double maxRatio) + { + Random random = new Random(sortedTokens.size()); + + double potentialRatioGrowth = maxRatio - minRatio; + + List tokens = Lists.newArrayListWithExpectedSize(numTokens); + + if (sortedTokens.isEmpty()) + { + // Select a random start token. This has no effect on distribution, only on where the local ring is "centered". + // Using a random start decreases the chances of clash with the tokens of other datacenters in the ring. + Token t = partitioner.getRandomToken(); + tokens.add(t); + sortedTokens.put(t, newUnit); + } + + while (tokens.size() < numTokens) + { + // split max span using given ratio + Token prev = sortedTokens.lastKey(); + double maxsz = 0; + Token t1 = null; + Token t2 = null; + for (Token curr : sortedTokens.keySet()) + { + double sz = prev.size(curr); + if (sz > maxsz) + { + maxsz = sz; + t1 = prev; t2 = curr; + } + prev = curr; + } + assert t1 != null; + Token t = partitioner.split(t1, t2, minRatio + potentialRatioGrowth * random.nextDouble()); + tokens.add(t); + sortedTokens.put(t, newUnit); + } + return tokens; + } + /** * Unique group object that one or more UnitInfo objects link to. */ diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java index c0b6d5cb70..c5cce58a9b 100644 --- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java +++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java @@ -20,10 +20,14 @@ package org.apache.cassandra.dht; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.fail; +import org.junit.Assert; +import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Collection; import java.util.List; +import java.util.Map; +import java.util.Random; import java.util.UUID; import com.google.common.base.Predicate; @@ -147,6 +151,8 @@ public class BootStrapperTest generateFakeEndpoints(tmd, numOldNodes, numVNodes, "0", "0"); } + Random rand = new Random(1); + private void generateFakeEndpoints(TokenMetadata tmd, int numOldNodes, int numVNodes, String dc, String rack) throws UnknownHostException { IPartitioner p = tmd.partitioner; @@ -157,7 +163,7 @@ public class BootStrapperTest InetAddressAndPort addr = InetAddressAndPort.getByName("127." + dc + "." + rack + "." + (i + 1)); List tokens = Lists.newArrayListWithCapacity(numVNodes); for (int j = 0; j < numVNodes; ++j) - tokens.add(p.getRandomToken()); + tokens.add(p.getRandomToken(rand)); tmd.updateNormalTokens(tokens, addr); } @@ -270,7 +276,54 @@ public class BootStrapperTest } } - + @Test + public void testAllocateTokensRfEqRacks() throws UnknownHostException + { + IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch(); + try + { + DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch()); + int vn = 8; + int replicas = 3; + int rackCount = replicas; + String ks = "BootStrapperTestNTSKeyspaceRfEqRacks"; + String dc = "1"; + + TokenMetadata metadata = StorageService.instance.getTokenMetadata(); + metadata.clearUnsafe(); + metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.1.0.99")); + metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.15.0.99")); + + SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1")); + int base = 5; + for (int i = 0; i < rackCount; ++i) + generateFakeEndpoints(metadata, base << i, vn, dc, Integer.toString(i)); // unbalanced racks + + int cnt = 5; + for (int i = 0; i < cnt; ++i) + allocateTokensForNode(vn, ks, metadata, InetAddressAndPort.getByName("127." + dc + ".0." + (99 + i))); + + double target = 1.0 / (base + cnt); + double permittedOver = 1.0 / (2 * vn + 1) + 0.01; + + Map ownership = StorageService.instance.effectiveOwnership(ks); + boolean failed = false; + for (Map.Entry o : ownership.entrySet()) + { + int rack = o.getKey().getAddress()[2]; + if (rack != 0) + continue; + + System.out.format("Node %s owns %f ratio to optimal %.2f\n", o.getKey(), o.getValue(), o.getValue() / target); + if (o.getValue()/target > 1 + permittedOver) + failed = true; + } + Assert.assertFalse(String.format("One of the nodes in the rack has over %.2f%% overutilization.", permittedOver * 100), failed); + } finally { + DatabaseDescriptor.setEndpointSnitch(oldSnitch); + } + } + @Test public void testAllocateTokensMultipleKeyspaces() throws UnknownHostException {