This is an automated email from the ASF dual-hosted git repository.
brandonwilliams pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new c9c022e Improve the algorithmic token allocation in case racks = RF
c9c022e is described below
commit c9c022e07fd4fc6afc59cd1149fd18bc84d3fb39
Author: Ekaterina Dimitrova <[email protected]>
AuthorDate: Thu Feb 27 18:47:33 2020 -0500
Improve the algorithmic token allocation in case racks = RF
Patch by Ekaterina Dimitrova, reviewed by brandonwilliams for
CASSANDRA-15600
---
CHANGES.txt | 1 +
.../NoReplicationTokenAllocator.java | 44 ++++++++--------
.../ReplicationAwareTokenAllocator.java | 12 ++++-
.../dht/tokenallocator/TokenAllocation.java | 28 +++++++++-
.../dht/tokenallocator/TokenAllocatorBase.java | 60 ++++++++++++++++++++++
.../org/apache/cassandra/dht/BootStrapperTest.java | 57 +++++++++++++++++++-
6 files changed, 176 insertions(+), 26 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 9cd6040..5858c19 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 f6a1592..0ac8951 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<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();
+ }
}
diff --git
a/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java
b/src/java/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocator.java
index 36fc8c2..eb498ce 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<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
diff --git
a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java
b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java
index bfa281e..bd6d980 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 f59bfd4..3d7e6b9 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<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.
*/
diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
index c0b6d5c..c5cce58 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<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
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
