natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve the algorithmic token allocation in case racks = RF 

Patch by Ekaterina Dimitrova, reviewed by brandonwilliams for CASSANDRA-15600
This commit is contained in:
Ekaterina Dimitrova 2020-02-27 18:47:33 -05:00 committed by Brandon Williams
parent 1ce58ac2ea
commit c9c022e07f
6 changed files with 175 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

42
src/java/org/apache/cassandra/dht/tokenallocator/NoReplicationTokenAllocator.java
View File

@ -114,24 +114,6 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
unitTokens.add(new Weighted<TokenInfo>(token.replicatedOwnership, token));
}
private Collection<Token> generateRandomTokens(UnitInfo<Unit> newUnit, int numTokens, Map<Unit, UnitInfo<Unit>> unitInfos)
{
Set<Token> 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<Token> addUnit(Unit newUnit, int numTokens)
{
assert !tokensInUnits.containsKey(newUnit);
@ -141,10 +123,10 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
Map<Unit, UnitInfo<Unit>> 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<Unit> head = createTokenInfos(unitInfos);
@ -172,7 +154,19 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
}
List<Token> 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<UnitInfo> unit : unitsToChange)
@ -193,7 +187,7 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
unit.value.ownership -= wt.weight;
}
double toTakeOver = unit.weight - targetAverage;
double toTakeOver = unit.weight - target;
// Split toTakeOver proportionally between the vnodes.
for (Weighted<TokenInfo> wt : tokens)
{
@ -230,6 +224,7 @@ public class NoReplicationTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
// 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<Unit> extends TokenAllocatorBase<Unit>
{
return 1;
}
public String toString()
{
return getClass().getSimpleName();
}
}

12
src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java
View File

@ -61,9 +61,12 @@ class ReplicationAwareTokenAllocator<Unit> extends TokenAllocatorBase<Unit>
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<Unit> extends TokenAllocatorBase<Unit>
return tokens;
}
Collection<Token> generateSplits(Unit newUnit, int numTokens)
{
Collection<Token> 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

28
src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java
View File

@ -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.

60
src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocatorBase.java
View File

@ -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<Unit> implements TokenAllocator<Unit>
{
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<Token, Unit> sortedTokens;
final ReplicationStrategy<Unit> strategy;
final IPartitioner partitioner;
@ -79,6 +86,59 @@ public abstract class TokenAllocatorBase<Unit> implements TokenAllocator<Unit>
return group;
}
Collection<Token> 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<Token> generateSplits(Unit newUnit, int numTokens, double minRatio, double maxRatio)
{
Random random = new Random(sortedTokens.size());
double potentialRatioGrowth = maxRatio - minRatio;
List<Token> 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.
*/

57
test/unit/org/apache/cassandra/dht/BootStrapperTest.java
View File

@ -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<Token> 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<InetAddress, Float> ownership = StorageService.instance.effectiveOwnership(ks);
boolean failed = false;
for (Map.Entry<InetAddress, Float> 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
{