diff --git a/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java b/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java index c37137b357..4555b17608 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java +++ b/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java @@ -21,7 +21,6 @@ package org.apache.cassandra.tcm.ownership; import java.io.IOException; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.Set; @@ -78,12 +77,12 @@ public class DataPlacement public DataPlacement combineReplicaGroups(DataPlacement other) { return new DataPlacement(PlacementForRange.builder() - .withReplicaGroups(reads.replicaGroups().values()) - .withReplicaGroups(other.reads.replicaGroups.values()) + .withReplicaGroups(reads.replicaGroups().endpoints) + .withReplicaGroups(other.reads.replicaGroups().endpoints) .build(), PlacementForRange.builder() - .withReplicaGroups(writes.replicaGroups().values()) - .withReplicaGroups(other.writes.replicaGroups.values()) + .withReplicaGroups(writes.replicaGroups().endpoints) + .withReplicaGroups(other.writes.replicaGroups().endpoints) .build()); } @@ -165,22 +164,11 @@ public class DataPlacement public String toString() { return "DataPlacement{" + - "reads=" + toString(reads.replicaGroups) + - ", writes=" + toString(writes.replicaGroups) + + "reads=" + reads + + ", writes=" + writes + '}'; } - public static String toString(Map predicted) - { - StringBuilder sb = new StringBuilder(); - for (Map.Entry e : predicted.entrySet()) - { - sb.append(e.getKey()).append("=").append(e.getValue()).append(",\n"); - } - - return sb.toString(); - } - @Override public boolean equals(Object o) { diff --git a/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java b/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java index 5d598a8191..2c7dc6746f 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java +++ b/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java @@ -150,12 +150,12 @@ public class DataPlacements extends ReplicationMap implements Met else { PlacementForRange.Builder reads = PlacementForRange.builder(placement.reads.replicaGroups().size()); - placement.reads.replicaGroups().forEach((range, endpoints) -> { + placement.reads.replicaGroups().endpoints.forEach((endpoints) -> { reads.withReplicaGroup(VersionedEndpoints.forRange(endpoints.lastModified(), endpoints.get().sorted(comparator))); }); PlacementForRange.Builder writes = PlacementForRange.builder(placement.writes.replicaGroups().size()); - placement.writes.replicaGroups().forEach((range, endpoints) -> { + placement.writes.replicaGroups().endpoints.forEach((endpoints) -> { writes.withReplicaGroup(VersionedEndpoints.forRange(endpoints.lastModified(), endpoints.get().sorted(comparator))); }); diff --git a/src/java/org/apache/cassandra/tcm/ownership/PlacementForRange.java b/src/java/org/apache/cassandra/tcm/ownership/PlacementForRange.java index 0169279d3e..db9ab3bbbe 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/PlacementForRange.java +++ b/src/java/org/apache/cassandra/tcm/ownership/PlacementForRange.java @@ -20,9 +20,12 @@ package org.apache.cassandra.tcm.ownership; import java.io.IOException; import java.util.*; +import java.util.function.BiConsumer; import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSortedMap; import com.google.common.collect.Maps; import org.apache.cassandra.dht.IPartitioner; @@ -41,6 +44,7 @@ import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.serialization.PartitionerAwareMetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; +import org.apache.cassandra.utils.AsymmetricOrdering; import static org.apache.cassandra.db.TypeSizes.sizeof; @@ -50,23 +54,34 @@ public class PlacementForRange public static final PlacementForRange EMPTY = PlacementForRange.builder().build(); - final SortedMap, VersionedEndpoints.ForRange> replicaGroups; + private final ReplicaGroups replicaGroups; public PlacementForRange(Map, VersionedEndpoints.ForRange> replicaGroups) { - this.replicaGroups = new TreeMap<>(replicaGroups); + ImmutableList.Builder> rangesBuilder = ImmutableList.builderWithExpectedSize(replicaGroups.size()); + ImmutableList.Builder endpointsBuilder = ImmutableList.builderWithExpectedSize(replicaGroups.size()); + Range prev = null; + for (Map.Entry, VersionedEndpoints.ForRange> entry : ImmutableSortedMap.copyOf(replicaGroups, Comparator.comparing(o -> o.left)).entrySet()) + { + if (prev != null && prev.right.compareTo(entry.getKey().left) > 0 ) + throw new IllegalArgumentException("Got overlapping ranges in replica groups: " + replicaGroups); + prev = entry.getKey(); + rangesBuilder.add(entry.getKey()); + endpointsBuilder.add(entry.getValue()); + } + this.replicaGroups = new ReplicaGroups(rangesBuilder.build(), endpointsBuilder.build()); } @VisibleForTesting - public Map, VersionedEndpoints.ForRange> replicaGroups() + public ReplicaGroups replicaGroups() { - return Collections.unmodifiableMap(replicaGroups); + return replicaGroups; } @VisibleForTesting public List> ranges() { - List> ranges = new ArrayList<>(replicaGroups.keySet()); + List> ranges = new ArrayList<>(this.replicaGroups.ranges); ranges.sort(Range::compareTo); return ranges; } @@ -76,7 +91,11 @@ public class PlacementForRange { // can't use range.isWrapAround() since range.unwrap() returns a wrapping range (right token is min value) assert range.right.compareTo(range.left) > 0 || range.right.equals(range.right.minValue()); - return replicaGroups.get(range); + // we're searching for an exact match to the input range here, can use standard binary search + int pos = Collections.binarySearch(replicaGroups.ranges, range, Comparator.comparing(o -> o.left)); + if (pos >= 0 && pos < replicaGroups.ranges.size() && replicaGroups.ranges.get(pos).equals(range)) + return replicaGroups.endpoints.get(pos); + return null; } /** @@ -86,37 +105,29 @@ public class PlacementForRange { EndpointsForRange.Builder builder = new EndpointsForRange.Builder(range); Epoch lastModified = Epoch.EMPTY; - - for (Map.Entry, VersionedEndpoints.ForRange> entry : replicaGroups.entrySet()) + // find a range containing the *right* token for the given range - Range is start exclusive so if we looked for the + // left one we could get the wrong range + int pos = ReplicaGroups.ordering.binarySearchAsymmetric(replicaGroups.ranges, range.right, AsymmetricOrdering.Op.CEIL); + if (pos >= 0 && pos < replicaGroups.ranges.size() && replicaGroups.ranges.get(pos).contains(range)) { - if (entry.getKey().contains(range)) - { - lastModified = Epoch.max(lastModified, entry.getValue().lastModified()); - builder.addAll(entry.getValue().get(), ReplicaCollection.Builder.Conflict.ALL); - } + VersionedEndpoints.ForRange eps = replicaGroups.endpoints.get(pos); + lastModified = eps.lastModified(); + builder.addAll(eps.get(), ReplicaCollection.Builder.Conflict.ALL); } - return VersionedEndpoints.forRange(lastModified, builder.build()); } public VersionedEndpoints.ForRange forRange(Token token) { - for (Map.Entry, VersionedEndpoints.ForRange> entry : replicaGroups.entrySet()) - { - if (entry.getKey().contains(token)) - return entry.getValue(); - } + int pos = ReplicaGroups.ordering.binarySearchAsymmetric(replicaGroups.ranges, token, AsymmetricOrdering.Op.CEIL); + if (pos >= 0 && pos < replicaGroups.endpoints.size()) + return replicaGroups.endpoints.get(pos); throw new IllegalStateException("Could not find range for token " + token + " in PlacementForRange: " + replicaGroups); } public VersionedEndpoints.ForToken forToken(Token token) { - for (Map.Entry, VersionedEndpoints.ForRange> entry : replicaGroups.entrySet()) - { - if (entry.getKey().contains(token)) - return entry.getValue().forToken(token); - } - throw new IllegalStateException("Could not find range for token " + token + " in PlacementForRange: " + replicaGroups); + return forRange(token).forToken(token); } public Delta difference(PlacementForRange next) @@ -130,8 +141,8 @@ public class PlacementForRange public RangesByEndpoint byEndpoint() { RangesByEndpoint.Builder builder = new RangesByEndpoint.Builder(); - for (Map.Entry, VersionedEndpoints.ForRange> oldPlacement : this.replicaGroups.entrySet()) - oldPlacement.getValue().byEndpoint().forEach(builder::put); + for (int i = 0; i < replicaGroups.size(); i++) + replicaGroups.endpoints.get(i).byEndpoint().forEach(builder::put); return builder.build(); } @@ -155,10 +166,10 @@ public class PlacementForRange public PlacementForRange withCappedLastModified(Epoch lastModified) { SortedMap, VersionedEndpoints.ForRange> copy = new TreeMap<>(); - for (Map.Entry, VersionedEndpoints.ForRange> entry : replicaGroups.entrySet()) + for (int i = 0; i < replicaGroups.size(); i++) { - Range range = entry.getKey(); - VersionedEndpoints.ForRange forRange = entry.getValue(); + Range range = replicaGroups.ranges.get(i); + VersionedEndpoints.ForRange forRange = replicaGroups.endpoints.get(i); if (forRange.lastModified().isAfter(lastModified)) forRange = forRange.withLastModified(lastModified); copy.put(range, forRange); @@ -184,7 +195,7 @@ public class PlacementForRange @VisibleForTesting public List toReplicaStringList() { - return replicaGroups.values() + return replicaGroups.endpoints .stream() .map(VersionedEndpoints.ForRange::get) .flatMap(AbstractReplicaCollection::stream) @@ -194,7 +205,7 @@ public class PlacementForRange public Builder unbuild() { - return new Builder(new HashMap<>(replicaGroups)); + return new Builder(replicaGroups.asMap()); } public static Builder builder() @@ -214,7 +225,7 @@ public class PlacementForRange return placement; Builder newPlacement = PlacementForRange.builder(); - List eprs = new ArrayList<>(placement.replicaGroups.values()); + List eprs = new ArrayList<>(placement.replicaGroups.endpoints); eprs.sort(Comparator.comparing(a -> a.range().left)); Token min = eprs.get(0).range().left; Token max = eprs.get(eprs.size() - 1).range().right; @@ -384,18 +395,18 @@ public class PlacementForRange { out.writeInt(t.replicaGroups.size()); - for (Map.Entry, VersionedEndpoints.ForRange> entry : t.replicaGroups.entrySet()) + for (int i = 0; i < t.replicaGroups.size(); i++) { - Range range = entry.getKey(); - VersionedEndpoints.ForRange efr = entry.getValue(); + Range range = t.replicaGroups.ranges.get(i); + VersionedEndpoints.ForRange efr = t.replicaGroups.endpoints.get(i); if (version.isAtLeast(Version.V2)) Epoch.serializer.serialize(efr.lastModified(), out, version); Token.metadataSerializer.serialize(range.left, out, partitioner, version); Token.metadataSerializer.serialize(range.right, out, partitioner, version); out.writeInt(efr.size()); - for (int i = 0; i < efr.size(); i++) + for (int efrIdx = 0; efrIdx < efr.size(); efrIdx++) { - Replica r = efr.get().get(i); + Replica r = efr.get().get(efrIdx); Token.metadataSerializer.serialize(r.range().left, out, partitioner, version); Token.metadataSerializer.serialize(r.range().right, out, partitioner, version); InetAddressAndPort.MetadataSerializer.serializer.serialize(r.endpoint(), out, version); @@ -444,19 +455,19 @@ public class PlacementForRange public long serializedSize(PlacementForRange t, IPartitioner partitioner, Version version) { long size = sizeof(t.replicaGroups.size()); - for (Map.Entry, VersionedEndpoints.ForRange> entry : t.replicaGroups.entrySet()) + for (int i = 0; i < t.replicaGroups.size(); i++) { - Range range = entry.getKey(); - VersionedEndpoints.ForRange efr = entry.getValue(); + Range range = t.replicaGroups.ranges.get(i); + VersionedEndpoints.ForRange efr = t.replicaGroups.endpoints.get(i); if (version.isAtLeast(Version.V2)) size += Epoch.serializer.serializedSize(efr.lastModified(), version); size += Token.metadataSerializer.serializedSize(range.left, partitioner, version); size += Token.metadataSerializer.serializedSize(range.right, partitioner, version); size += sizeof(efr.size()); - for (int i = 0; i < efr.size(); i++) + for (int efrIdx = 0; efrIdx < efr.size(); efrIdx++) { - Replica r = efr.get().get(i); + Replica r = efr.get().get(efrIdx); size += Token.metadataSerializer.serializedSize(r.range().left, partitioner, version); size += Token.metadataSerializer.serializedSize(r.range().right, partitioner, version); size += InetAddressAndPort.MetadataSerializer.serializer.serializedSize(r.endpoint(), version); @@ -481,4 +492,85 @@ public class PlacementForRange { return Objects.hash(replicaGroups); } + + public static class ReplicaGroups + { + public final ImmutableList> ranges; + public final ImmutableList endpoints; + + private static final AsymmetricOrdering, Token> ordering = new AsymmetricOrdering<>() + { + @Override + public int compare(Range left, Range right) + { + return left.compareTo(right); + } + + @Override + public int compareAsymmetric(Range range, Token token) + { + if (token.isMinimum() && !range.right.isMinimum()) + return -1; + if (range.left.compareTo(token) >= 0) + return 1; + if (!range.right.isMinimum() && range.right.compareTo(token) < 0) + return -1; + return 0; + } + }; + + public ReplicaGroups(ImmutableList> ranges, ImmutableList endpoints) + { + this.ranges = ranges; + this.endpoints = endpoints; + } + + public int size() + { + return ranges.size(); + } + + public boolean isEmpty() + { + return size() == 0; + } + + @VisibleForTesting + public Map, VersionedEndpoints.ForRange> asMap() + { + Map, VersionedEndpoints.ForRange> map = new HashMap<>(); + for (int i = 0; i < size(); i++) + map.put(ranges.get(i), endpoints.get(i)); + return map; + } + + public void forEach(BiConsumer, VersionedEndpoints.ForRange> consumer) + { + for (int i = 0; i < size(); i++) + consumer.accept(ranges.get(i), endpoints.get(i)); + } + + @Override + public String toString() + { + StringBuilder sb = new StringBuilder("ReplicaGroups{"); + forEach((range, eps) -> sb.append(range).append('=').append(eps).append(", ")); + return sb.append('}').toString(); + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (!(o instanceof ReplicaGroups)) return false; + ReplicaGroups entries = (ReplicaGroups) o; + return Objects.equals(ranges, entries.ranges) && Objects.equals(endpoints, entries.endpoints); + } + + @Override + public int hashCode() + { + return Objects.hash(ranges, endpoints); + } + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java index 8ece3dc7c8..1ec209e16c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java @@ -558,7 +558,7 @@ public class MetadataChangeSimulationTest extends CMSTestBase Set bouncing = new HashSet<>(); Set replicasFromBouncedReplicaSets = new HashSet<>(); outer: - for (VersionedEndpoints.ForRange placements : sut.service.metadata().placements.get(rf.asKeyspaceParams().replication).writes.replicaGroups().values()) + for (VersionedEndpoints.ForRange placements : sut.service.metadata().placements.get(rf.asKeyspaceParams().replication).writes.replicaGroups().endpoints) { List replicas = new ArrayList<>(metadata.directory.toNodeIds(placements.get().endpoints())); List bounceCandidates = new ArrayList<>(); @@ -702,10 +702,10 @@ public class MetadataChangeSimulationTest extends CMSTestBase public static void match(PlacementForRange actual, Map> predicted) throws Throwable { - Map, VersionedEndpoints.ForRange> actualGroups = actual.replicaGroups(); + Map, VersionedEndpoints.ForRange> actualGroups = actual.replicaGroups().asMap(); assert predicted.size() == actualGroups.size() : String.format("\nPredicted:\n%s(%d)" + - "\nActual:\n%s(%d)", toString(predicted), predicted.size(), toString(actual.replicaGroups()), actualGroups.size()); + "\nActual:\n%s(%d)", toString(predicted), predicted.size(), toString(actualGroups), actualGroups.size()); for (Map.Entry> entry : predicted.entrySet()) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/SimulatedOperation.java b/test/distributed/org/apache/cassandra/distributed/test/log/SimulatedOperation.java index 15ee938cf2..56a622c65c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/SimulatedOperation.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/SimulatedOperation.java @@ -156,7 +156,7 @@ public abstract class SimulatedOperation sutActions.next(); ClusterMetadata m2 = ClusterMetadata.current(); - Map, VersionedEndpoints.ForRange> after = m2.placements.get(simulatedState.rf.asKeyspaceParams().replication).reads.replicaGroups(); + Map, VersionedEndpoints.ForRange> after = m2.placements.get(simulatedState.rf.asKeyspaceParams().replication).reads.replicaGroups().asMap(); m1.placements.get(simulatedState.rf.asKeyspaceParams().replication).reads.replicaGroups().forEach((k, beforePlacements) -> { if (after.containsKey(k)) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/RangeVersioningTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/RangeVersioningTest.java index 47a80124e2..6c4bd5b583 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/RangeVersioningTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/RangeVersioningTest.java @@ -56,7 +56,7 @@ public class RangeVersioningTest extends FuzzTestBase Epoch smallestSeen = null; for (VersionedEndpoints.ForRange fr : metadata.placements .get(ReplicationParams.simple(i)) - .writes.replicaGroups().values()) + .writes.replicaGroups().endpoints) { if (smallestSeen == null || fr.lastModified().isBefore(smallestSeen)) smallestSeen = fr.lastModified(); diff --git a/test/unit/org/apache/cassandra/tcm/ownership/UniformRangePlacementTest.java b/test/unit/org/apache/cassandra/tcm/ownership/UniformRangePlacementTest.java index f046be7955..c7fe3f449b 100644 --- a/test/unit/org/apache/cassandra/tcm/ownership/UniformRangePlacementTest.java +++ b/test/unit/org/apache/cassandra/tcm/ownership/UniformRangePlacementTest.java @@ -235,9 +235,7 @@ public class UniformRangePlacementTest DataPlacement initialPlacement = builder.build(); DataPlacement split = initialPlacement.splitRangesForPlacement(tokens); - assertPlacement(split.writes, rg(-3, -9223372036854775808L, 1), - rg(-9223372036854775808L,-4611686018427387905L, 1), - rg(-4611686018427387905L, -3, 1)); + assertPlacement(split.writes, rg(-9223372036854775808L,-4611686018427387905L, 1), rg(-4611686018427387905L, -3, 1), rg(-3, -9223372036854775808L, 1)); } @Test @@ -254,7 +252,7 @@ public class UniformRangePlacementTest DataPlacement initialPlacement = builder.build(); DataPlacement split = initialPlacement.splitRangesForPlacement(tokens); - assertPlacement(split.writes, rg(3074457345618258602L,-9223372036854775808L, 1), rg(-9223372036854775808L, 3074457345618258602L, 1)); + assertPlacement(split.writes, rg(-9223372036854775808L, 3074457345618258602L, 1), rg(3074457345618258602L,-9223372036854775808L, 1)); } @Test @@ -268,7 +266,7 @@ public class UniformRangePlacementTest DataPlacement initialPlacement = builder.build(); List tokens = ImmutableList.of(token(Long.MIN_VALUE), token(0)); DataPlacement newPlacement = initialPlacement.splitRangesForPlacement(tokens); - assertEquals(2, newPlacement.writes.replicaGroups.values().size()); + assertEquals(2, newPlacement.writes.replicaGroups().size()); } private PlacementForRange initialPlacement() @@ -286,7 +284,7 @@ public class UniformRangePlacementTest private void assertPlacement(PlacementForRange placement, EndpointsForRange...expected) { - Collection replicaGroups = placement.replicaGroups.values().stream().map(v -> v.get()).collect(Collectors.toList()); + Collection replicaGroups = placement.replicaGroups().endpoints.stream().map(v -> v.get()).collect(Collectors.toList()); assertEquals(replicaGroups.size(), expected.length); int i = 0; boolean allMatch = true; diff --git a/test/unit/org/apache/cassandra/tcm/sequences/InProgressSequenceCancellationTest.java b/test/unit/org/apache/cassandra/tcm/sequences/InProgressSequenceCancellationTest.java index 44b04abcf2..dea913cfd1 100644 --- a/test/unit/org/apache/cassandra/tcm/sequences/InProgressSequenceCancellationTest.java +++ b/test/unit/org/apache/cassandra/tcm/sequences/InProgressSequenceCancellationTest.java @@ -323,7 +323,7 @@ public class InProgressSequenceCancellationTest DataPlacement otherPlacement = second.get(params); PlacementForRange r1 = placement.reads; PlacementForRange r2 = otherPlacement.reads; - assertEquals(r1.replicaGroups().keySet(), r2.replicaGroups().keySet()); + assertEquals(r1.replicaGroups().ranges, r2.replicaGroups().ranges); r1.replicaGroups().forEach((range, e1) -> { EndpointsForRange e2 = r2.forRange(range).get(); assertEquals(e1.size(),e2.size()); @@ -332,7 +332,7 @@ public class InProgressSequenceCancellationTest PlacementForRange w1 = placement.reads; PlacementForRange w2 = otherPlacement.reads; - assertEquals(w1.replicaGroups().keySet(), w2.replicaGroups().keySet()); + assertEquals(w1.replicaGroups().ranges, w2.replicaGroups().ranges); w1.replicaGroups().forEach((range, e1) -> { EndpointsForRange e2 = w2.forRange(range).get(); assertEquals(e1.size(),e2.size()); diff --git a/test/unit/org/apache/cassandra/tcm/sequences/SequencesUtils.java b/test/unit/org/apache/cassandra/tcm/sequences/SequencesUtils.java index 6c0580ea41..aa6cedddc2 100644 --- a/test/unit/org/apache/cassandra/tcm/sequences/SequencesUtils.java +++ b/test/unit/org/apache/cassandra/tcm/sequences/SequencesUtils.java @@ -67,7 +67,7 @@ public class SequencesUtils { LockedRanges.AffectedRangesBuilder affected = LockedRanges.AffectedRanges.builder(); placements.asMap().forEach((params, placement) -> { - placement.reads.replicaGroups().keySet().forEach((range) -> { + placement.reads.replicaGroups().ranges.forEach((range) -> { if (random.nextDouble() >= 0.6) affected.add(params, range); });