You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2018/10/03 13:38:44 UTC
[2/2] cassandra git commit: ReplicaCollection follow-up
ReplicaCollection follow-up
patch by Benedict; reviewed by Alex Petrov and Ariel Weisberg for CASSANDRA-14726
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e645b917
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e645b917
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e645b917
Branch: refs/heads/trunk
Commit: e645b9172c5d50fc2af407de724e46121edfe109
Parents: 467068d
Author: Benedict Elliott Smith <be...@apple.com>
Authored: Fri Sep 7 19:28:16 2018 +0100
Committer: Benedict Elliott Smith <be...@apple.com>
Committed: Wed Oct 3 14:38:22 2018 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/db/DiskBoundaryManager.java | 8 +-
.../org/apache/cassandra/db/ReadCommand.java | 15 +-
.../db/compaction/CompactionManager.java | 28 +-
.../db/streaming/CassandraStreamManager.java | 2 +-
.../org/apache/cassandra/dht/RangeStreamer.java | 10 +-
.../locator/AbstractReplicaCollection.java | 394 ++++++++++++++++---
.../locator/AbstractReplicationStrategy.java | 10 +-
.../org/apache/cassandra/locator/Endpoints.java | 47 +--
.../cassandra/locator/EndpointsByRange.java | 16 +-
.../cassandra/locator/EndpointsByReplica.java | 18 +-
.../cassandra/locator/EndpointsForRange.java | 93 ++---
.../cassandra/locator/EndpointsForToken.java | 87 ++--
.../locator/NetworkTopologyStrategy.java | 10 +-
.../locator/OldNetworkTopologyStrategy.java | 4 +-
.../cassandra/locator/PendingRangeMaps.java | 38 +-
.../cassandra/locator/RangesAtEndpoint.java | 163 +++-----
.../cassandra/locator/RangesByEndpoint.java | 14 +-
.../cassandra/locator/ReplicaCollection.java | 70 ++--
.../apache/cassandra/locator/ReplicaLayout.java | 1 +
.../cassandra/locator/ReplicaMultimap.java | 26 +-
.../apache/cassandra/locator/ReplicaPlans.java | 13 +-
.../cassandra/locator/SimpleStrategy.java | 12 +-
.../apache/cassandra/locator/TokenMetadata.java | 8 +-
.../cassandra/service/RangeRelocator.java | 4 +-
.../cassandra/service/StorageService.java | 12 +-
.../service/reads/AbstractReadExecutor.java | 14 +-
.../db/compaction/AntiCompactionTest.java | 4 +-
.../dht/RangeFetchMapCalculatorTest.java | 55 ++-
.../locator/ReplicaCollectionTest.java | 207 +++++++---
.../org/apache/cassandra/service/MoveTest.java | 5 +-
.../cassandra/service/MoveTransientTest.java | 32 +-
.../cassandra/service/StorageServiceTest.java | 4 +-
.../service/reads/DataResolverTest.java | 5 +-
34 files changed, 883 insertions(+), 547 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ae321e5..25c2728 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0
+ * ReplicaCollection follow-up (CASSANDRA-14726)
* Transient node receives full data requests (CASSANDRA-14762)
* Enable snapshot artifacts publish (CASSANDRA-12704)
* Introduce RangesAtEndpoint.unwrap to simplify StreamSession.addTransferRanges (CASSANDRA-14770)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/DiskBoundaryManager.java b/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
index 0961a42..69aabfd 100644
--- a/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
+++ b/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
@@ -123,19 +123,19 @@ public class DiskBoundaryManager
*
* The final entry in the returned list will always be the partitioner maximum tokens upper key bound
*/
- private static List<PartitionPosition> getDiskBoundaries(RangesAtEndpoint ranges, IPartitioner partitioner, Directories.DataDirectory[] dataDirectories)
+ private static List<PartitionPosition> getDiskBoundaries(RangesAtEndpoint replicas, IPartitioner partitioner, Directories.DataDirectory[] dataDirectories)
{
assert partitioner.splitter().isPresent();
Splitter splitter = partitioner.splitter().get();
boolean dontSplitRanges = DatabaseDescriptor.getNumTokens() > 1;
- List<Splitter.WeightedRange> weightedRanges = new ArrayList<>(ranges.size());
+ List<Splitter.WeightedRange> weightedRanges = new ArrayList<>(replicas.size());
// note that Range.sort unwraps any wraparound ranges, so we need to sort them here
- for (Range<Token> r : Range.sort(ranges.fullRanges()))
+ for (Range<Token> r : Range.sort(replicas.onlyFull().ranges()))
weightedRanges.add(new Splitter.WeightedRange(1.0, r));
- for (Range<Token> r : Range.sort(ranges.transientRanges()))
+ for (Range<Token> r : Range.sort(replicas.onlyTransient().ranges()))
weightedRanges.add(new Splitter.WeightedRange(0.1, r));
weightedRanges.sort(Comparator.comparing(Splitter.WeightedRange::left));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/db/ReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index ada4ae6..cbef7ea 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -66,6 +66,9 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.HashingUtils;
+import static com.google.common.collect.Iterables.any;
+import static com.google.common.collect.Iterables.filter;
+
/**
* General interface for storage-engine read commands (common to both range and
* single partition commands).
@@ -326,10 +329,10 @@ public abstract class ReadCommand extends AbstractReadQuery
/**
* Returns a copy of this command with acceptsTransient set to true.
*/
- public ReadCommand copyAsTransientQuery(ReplicaCollection<?> replicas)
+ public ReadCommand copyAsTransientQuery(Iterable<Replica> replicas)
{
- if (Iterables.any(replicas, Replica::isFull))
- throw new IllegalArgumentException("Can't make a transient request on full replicas: " + replicas.filter(Replica::isFull));
+ if (any(replicas, Replica::isFull))
+ throw new IllegalArgumentException("Can't make a transient request on full replicas: " + Iterables.toString(filter(replicas, Replica::isFull)));
return copyAsTransientQuery();
}
@@ -348,10 +351,10 @@ public abstract class ReadCommand extends AbstractReadQuery
/**
* Returns a copy of this command with isDigestQuery set to true.
*/
- public ReadCommand copyAsDigestQuery(ReplicaCollection<?> replicas)
+ public ReadCommand copyAsDigestQuery(Iterable<Replica> replicas)
{
- if (Iterables.any(replicas, Replica::isTransient))
- throw new IllegalArgumentException("Can't make a digest request on a transient replica " + replicas.filter(Replica::isTransient));
+ if (any(replicas, Replica::isTransient))
+ throw new IllegalArgumentException("Can't make a digest request on a transient replica " + Iterables.toString(filter(replicas, Replica::isTransient)));
return copyAsDigestQuery();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 39f3db5..b8b4993 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -522,8 +522,8 @@ public class CompactionManager implements CompactionManagerMBean
// if local ranges is empty, it means no data should remain
final RangesAtEndpoint replicas = StorageService.instance.getLocalReplicas(keyspace.getName());
final Set<Range<Token>> allRanges = replicas.ranges();
- final Set<Range<Token>> transientRanges = replicas.filter(Replica::isTransient).ranges();
- final Set<Range<Token>> fullRanges = replicas.filter(Replica::isFull).ranges();
+ final Set<Range<Token>> transientRanges = replicas.onlyTransient().ranges();
+ final Set<Range<Token>> fullRanges = replicas.onlyFull().ranges();
final boolean hasIndexes = cfStore.indexManager.hasIndexes();
return parallelAllSSTableOperation(cfStore, new OneSSTableOperation()
@@ -722,14 +722,14 @@ public class CompactionManager implements CompactionManagerMBean
* Caller must reference the validatedForRepair sstables (via ParentRepairSession.getActiveRepairedSSTableRefs(..)).
*
* @param cfs
- * @param ranges token ranges to be repaired
+ * @param replicas token ranges to be repaired
* @param validatedForRepair SSTables containing the repaired ranges. Should be referenced before passing them.
* @param sessionID the repair session we're anti-compacting for
* @throws InterruptedException
* @throws IOException
*/
public void performAnticompaction(ColumnFamilyStore cfs,
- RangesAtEndpoint ranges,
+ RangesAtEndpoint replicas,
Refs<SSTableReader> validatedForRepair,
LifecycleTransaction txn,
UUID sessionID) throws IOException
@@ -738,21 +738,21 @@ public class CompactionManager implements CompactionManagerMBean
{
ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(sessionID);
Preconditions.checkArgument(!prs.isPreview(), "Cannot anticompact for previews");
- Preconditions.checkArgument(!ranges.isEmpty(), "No ranges to anti-compact");
+ Preconditions.checkArgument(!replicas.isEmpty(), "No ranges to anti-compact");
if (logger.isInfoEnabled())
logger.info("{} Starting anticompaction for {}.{} on {}/{} sstables", PreviewKind.NONE.logPrefix(sessionID), cfs.keyspace.getName(), cfs.getTableName(), validatedForRepair.size(), cfs.getLiveSSTables().size());
if (logger.isTraceEnabled())
- logger.trace("{} Starting anticompaction for ranges {}", PreviewKind.NONE.logPrefix(sessionID), ranges);
+ logger.trace("{} Starting anticompaction for ranges {}", PreviewKind.NONE.logPrefix(sessionID), replicas);
Set<SSTableReader> sstables = new HashSet<>(validatedForRepair);
- validateSSTableBoundsForAnticompaction(sessionID, sstables, ranges);
- mutateFullyContainedSSTables(cfs, validatedForRepair, sstables.iterator(), ranges.fullRanges(), txn, sessionID, false);
- mutateFullyContainedSSTables(cfs, validatedForRepair, sstables.iterator(), ranges.transientRanges(), txn, sessionID, true);
+ validateSSTableBoundsForAnticompaction(sessionID, sstables, replicas);
+ mutateFullyContainedSSTables(cfs, validatedForRepair, sstables.iterator(), replicas.onlyFull().ranges(), txn, sessionID, false);
+ mutateFullyContainedSSTables(cfs, validatedForRepair, sstables.iterator(), replicas.onlyTransient().ranges(), txn, sessionID, true);
assert txn.originals().equals(sstables);
if (!sstables.isEmpty())
- doAntiCompaction(cfs, ranges, txn, sessionID);
+ doAntiCompaction(cfs, replicas, txn, sessionID);
txn.finish();
}
finally
@@ -963,8 +963,8 @@ public class CompactionManager implements CompactionManagerMBean
Keyspace keyspace = cfs.keyspace;
final RangesAtEndpoint replicas = StorageService.instance.getLocalReplicas(keyspace.getName());
final Set<Range<Token>> allRanges = replicas.ranges();
- final Set<Range<Token>> transientRanges = replicas.filter(Replica::isTransient).ranges();
- final Set<Range<Token>> fullRanges = replicas.filter(Replica::isFull).ranges();
+ final Set<Range<Token>> transientRanges = replicas.onlyTransient().ranges();
+ final Set<Range<Token>> fullRanges = replicas.onlyFull().ranges();
boolean hasIndexes = cfs.indexManager.hasIndexes();
SSTableReader sstable = lookupSSTable(cfs, entry.getValue());
@@ -1512,8 +1512,8 @@ public class CompactionManager implements CompactionManagerMBean
transWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, UNREPAIRED_SSTABLE, pendingRepair, true, sstableAsSet, txn));
unrepairedWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, UNREPAIRED_SSTABLE, NO_PENDING_REPAIR, false, sstableAsSet, txn));
- Predicate<Token> fullChecker = !ranges.fullRanges().isEmpty() ? new Range.OrderedRangeContainmentChecker(ranges.fullRanges()) : t -> false;
- Predicate<Token> transChecker = !ranges.transientRanges().isEmpty() ? new Range.OrderedRangeContainmentChecker(ranges.transientRanges()) : t -> false;
+ Predicate<Token> fullChecker = !ranges.onlyFull().isEmpty() ? new Range.OrderedRangeContainmentChecker(ranges.onlyFull().ranges()) : t -> false;
+ Predicate<Token> transChecker = !ranges.onlyTransient().isEmpty() ? new Range.OrderedRangeContainmentChecker(ranges.onlyTransient().ranges()) : t -> false;
while (ci.hasNext())
{
try (UnfilteredRowIterator partition = ci.next())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
index 6c2631c..b88a5d6 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamManager.java
@@ -142,7 +142,7 @@ public class CassandraStreamManager implements TableStreamManager
}).refs);
- List<Range<Token>> normalizedFullRanges = Range.normalize(replicas.filter(Replica::isFull).ranges());
+ List<Range<Token>> normalizedFullRanges = Range.normalize(replicas.onlyFull().ranges());
List<Range<Token>> normalizedAllRanges = Range.normalize(replicas.ranges());
//Create outgoing file streams for ranges possibly skipping repaired ranges in sstables
List<OutgoingStream> streams = new ArrayList<>(refs.size());
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/dht/RangeStreamer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/RangeStreamer.java b/src/java/org/apache/cassandra/dht/RangeStreamer.java
index b50a4e2..75e2530 100644
--- a/src/java/org/apache/cassandra/dht/RangeStreamer.java
+++ b/src/java/org/apache/cassandra/dht/RangeStreamer.java
@@ -39,7 +39,7 @@ import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.EndpointsByRange;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.RangesAtEndpoint;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -418,7 +418,7 @@ public class RangeStreamer
endpoints -> snitchGetSortedListByProximity.apply(localAddress, endpoints);
//This list of replicas is just candidates. With strict consistency it's going to be a narrow list.
- EndpointsByReplica.Mutable rangesToFetchWithPreferredEndpoints = new EndpointsByReplica.Mutable();
+ EndpointsByReplica.Builder rangesToFetchWithPreferredEndpoints = new EndpointsByReplica.Builder();
for (Replica toFetch : fetchRanges)
{
//Replica that is sufficient to provide the data we need
@@ -539,7 +539,7 @@ public class RangeStreamer
}
}
}
- return rangesToFetchWithPreferredEndpoints.asImmutableView();
+ return rangesToFetchWithPreferredEndpoints.build();
}
/**
@@ -573,14 +573,14 @@ public class RangeStreamer
//the surface area to test and introduce bugs.
//In the future it's possible we could run it twice once for full ranges with only full replicas
//and once with transient ranges and all replicas. Then merge the result.
- EndpointsByRange.Mutable unwrapped = new EndpointsByRange.Mutable();
+ EndpointsByRange.Builder unwrapped = new EndpointsByRange.Builder();
for (Map.Entry<Replica, Replica> entry : rangesWithSources.flattenEntries())
{
Replicas.temporaryAssertFull(entry.getValue());
unwrapped.put(entry.getKey().range(), entry.getValue());
}
- EndpointsByRange unwrappedView = unwrapped.asImmutableView();
+ EndpointsByRange unwrappedView = unwrapped.build();
RangeFetchMapCalculator calculator = new RangeFetchMapCalculator(unwrappedView, sourceFilters, keyspace);
Multimap<InetAddressAndPort, Range<Token>> rangeFetchMapMap = calculator.getRangeFetchMap();
logger.info("Output from RangeFetchMapCalculator for keyspace {}", keyspace);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java b/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
index 94ff991..374afc6 100644
--- a/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
+++ b/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java
@@ -18,12 +18,17 @@
package org.apache.cassandra.locator;
+import com.carrotsearch.hppc.ObjectIntOpenHashMap;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
-import java.util.ArrayList;
+import java.util.AbstractMap;
+import java.util.AbstractSet;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
-import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.function.BiConsumer;
@@ -41,14 +46,14 @@ import java.util.stream.Stream;
*/
public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollection<C>> implements ReplicaCollection<C>
{
- protected static final List<Replica> EMPTY_LIST = new ArrayList<>(); // since immutable, can safely return this to avoid megamorphic callsites
+ protected static final ReplicaList EMPTY_LIST = new ReplicaList(); // since immutable, can safely return this to avoid megamorphic callsites
- public static <C extends ReplicaCollection<C>, B extends Builder<C, ?, B>> Collector<Replica, B, C> collector(Set<Collector.Characteristics> characteristics, Supplier<B> supplier)
+ public static <C extends ReplicaCollection<C>, B extends Builder<C>> Collector<Replica, B, C> collector(Set<Collector.Characteristics> characteristics, Supplier<B> supplier)
{
return new Collector<Replica, B, C>()
{
private final BiConsumer<B, Replica> accumulator = Builder::add;
- private final BinaryOperator<B> combiner = (a, b) -> { a.addAll(b.mutable); return a; };
+ private final BinaryOperator<B> combiner = (a, b) -> { a.addAll(b); return a; };
private final Function<B, C> finisher = Builder::build;
public Supplier<B> supplier() { return supplier; }
public BiConsumer<B, Replica> accumulator() { return accumulator; }
@@ -58,45 +63,336 @@ public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollect
};
}
- protected final List<Replica> list;
- protected final boolean isSnapshot;
- protected AbstractReplicaCollection(List<Replica> list, boolean isSnapshot)
+ /**
+ * A simple list with no comodification checks and immutability by default (only append permitted, and only one initial copy)
+ * this permits us to reduce the amount of garbage generated, by not wrapping iterators or unnecessarily copying
+ * and reduces the amount of indirection necessary, as well as ensuring monomorphic callsites
+ */
+ protected static class ReplicaList implements Iterable<Replica>
{
- this.list = list;
- this.isSnapshot = isSnapshot;
+ private static final Replica[] EMPTY = new Replica[0];
+ Replica[] contents;
+ int begin, size;
+
+ public ReplicaList() { this(0); }
+ public ReplicaList(int capacity) { contents = capacity == 0 ? EMPTY : new Replica[capacity]; }
+ public ReplicaList(Replica[] contents, int begin, int size) { this.contents = contents; this.begin = begin; this.size = size; }
+
+ public boolean isSubList(ReplicaList subList)
+ {
+ return subList.contents == contents;
+ }
+
+ public Replica get(int index)
+ {
+ if (index > size)
+ throw new IndexOutOfBoundsException();
+ return contents[begin + index];
+ }
+
+ public void add(Replica replica)
+ {
+ // can only add to full array - if we have sliced it, we must be a snapshot
+ if (begin != 0)
+ throw new IllegalStateException();
+
+ if (size == contents.length)
+ {
+ int newSize;
+ if (size < 3) newSize = 3;
+ else if (size < 9) newSize = 9;
+ else newSize = size * 2;
+ contents = Arrays.copyOf(contents, newSize);
+ }
+ contents[size++] = replica;
+ }
+
+ public int size()
+ {
+ return size;
+ }
+
+ public boolean isEmpty()
+ {
+ return size == 0;
+ }
+
+ public ReplicaList subList(int begin, int end)
+ {
+ if (end > size || begin > end) throw new IndexOutOfBoundsException();
+ return new ReplicaList(contents, this.begin + begin, end - begin);
+ }
+
+ public ReplicaList sorted(Comparator<Replica> comparator)
+ {
+ Replica[] copy = Arrays.copyOfRange(contents, begin, begin + size);
+ Arrays.sort(copy, comparator);
+ return new ReplicaList(copy, 0, copy.length);
+ }
+
+ public Stream<Replica> stream()
+ {
+ return Arrays.stream(contents, begin, begin + size);
+ }
+
+ // we implement our own iterator, because it is trivial to do so, and in monomorphic call sites
+ // will compile down to almost optimal indexed for loop
+ @Override
+ public Iterator<Replica> iterator()
+ {
+ return new Iterator<Replica>()
+ {
+ final int end = begin + size;
+ int i = begin;
+ @Override
+ public boolean hasNext()
+ {
+ return i < end;
+ }
+
+ @Override
+ public Replica next()
+ {
+ if (!hasNext()) throw new IllegalStateException();
+ return contents[i++];
+ }
+ };
+ }
+
+ // we implement our own iterator, because it is trivial to do so, and in monomorphic call sites
+ // will compile down to almost optimal indexed for loop
+ public <K> Iterator<K> transformIterator(Function<Replica, K> function)
+ {
+ return new Iterator<K>()
+ {
+ final int end = begin + size;
+ int i = begin;
+ @Override
+ public boolean hasNext()
+ {
+ return i < end;
+ }
+
+ @Override
+ public K next()
+ {
+ return function.apply(contents[i++]);
+ }
+ };
+ }
+
+ // we implement our own iterator, because it is trivial to do so, and in monomorphic call sites
+ // will compile down to almost optimal indexed for loop
+ // in this case, especially, it is impactful versus Iterables.limit(Iterables.filter())
+ private Iterator<Replica> filterIterator(Predicate<Replica> predicate, int limit)
+ {
+ return new Iterator<Replica>()
+ {
+ final int end = begin + size;
+ int next = begin;
+ int count = 0;
+ { updateNext(); }
+ void updateNext()
+ {
+ if (count == limit) next = end;
+ while (next < end && !predicate.test(contents[next]))
+ ++next;
+ ++count;
+ }
+ @Override
+ public boolean hasNext()
+ {
+ return next < end;
+ }
+
+ @Override
+ public Replica next()
+ {
+ if (!hasNext()) throw new IllegalStateException();
+ Replica result = contents[next++];
+ updateNext();
+ return result;
+ }
+ };
+ }
+
+ @VisibleForTesting
+ public boolean equals(Object to)
+ {
+ if (to == null || to.getClass() != ReplicaList.class)
+ return false;
+ ReplicaList that = (ReplicaList) to;
+ if (this.size != that.size) return false;
+ return Iterables.elementsEqual(this, that);
+ }
}
- // if subList == null, should return self (or a clone thereof)
- protected abstract C snapshot(List<Replica> subList);
- protected abstract C self();
/**
- * construct a new Mutable of our own type, so that we can concatenate
- * TODO: this isn't terribly pretty, but we need sometimes to select / merge two Endpoints of unknown type;
+ * A simple map that ensures the underlying list's iteration order is maintained, and can be shared with
+ * subLists (either produced via subList, or via filter that naturally produced a subList).
+ * This permits us to reduce the amount of garbage generated, by not unnecessarily copying,
+ * reduces the amount of indirection necessary, as well as ensuring monomorphic callsites.
+ * The underlying map is also more efficient, particularly for such small collections as we typically produce.
*/
- public abstract Mutable<C> newMutable(int initialCapacity);
+ protected static class ReplicaMap<K> extends AbstractMap<K, Replica>
+ {
+ private final Function<Replica, K> toKey;
+ private final ReplicaList list;
+ // we maintain a map of key -> index in our list; this lets us share with subLists (or between Builder and snapshots)
+ // since we only need to corroborate that the list index we find is within the bounds of our list
+ // (if not, it's a shared map, and the key only occurs in one of our ancestors)
+ private final ObjectIntOpenHashMap<K> map;
+ private Set<K> keySet;
+ private Set<Entry<K, Replica>> entrySet;
+
+ abstract class AbstractImmutableSet<T> extends AbstractSet<T>
+ {
+ @Override
+ public boolean removeAll(Collection<?> c) { throw new UnsupportedOperationException(); }
+ @Override
+ public boolean remove(Object o) { throw new UnsupportedOperationException(); }
+ @Override
+ public int size() { return list.size(); }
+ }
+
+ class KeySet extends AbstractImmutableSet<K>
+ {
+ @Override
+ public boolean contains(Object o) { return containsKey(o); }
+ @Override
+ public Iterator<K> iterator() { return list.transformIterator(toKey); }
+ }
+
+ class EntrySet extends AbstractImmutableSet<Entry<K, Replica>>
+ {
+ @Override
+ public boolean contains(Object o)
+ {
+ Preconditions.checkNotNull(o);
+ if (!(o instanceof Entry<?, ?>)) return false;
+ return Objects.equals(get(((Entry) o).getKey()), ((Entry) o).getValue());
+ }
+
+ @Override
+ public Iterator<Entry<K, Replica>> iterator()
+ {
+ return list.transformIterator(r -> new SimpleImmutableEntry<>(toKey.apply(r), r));
+ }
+ }
+
+ public ReplicaMap(ReplicaList list, Function<Replica, K> toKey)
+ {
+ // 8*0.65 => RF=5; 16*0.65 ==> RF=10
+ // use list capacity if empty, otherwise use actual list size
+ this.toKey = toKey;
+ this.map = new ObjectIntOpenHashMap<>(list.size == 0 ? list.contents.length : list.size, 0.65f);
+ this.list = list;
+ for (int i = list.begin ; i < list.begin + list.size ; ++i)
+ {
+ boolean inserted = internalPutIfAbsent(list.contents[i], i);
+ assert inserted;
+ }
+ }
+
+ public ReplicaMap(ReplicaList list, Function<Replica, K> toKey, ObjectIntOpenHashMap<K> map)
+ {
+ this.toKey = toKey;
+ this.list = list;
+ this.map = map;
+ }
+
+ // to be used only by subclasses of AbstractReplicaCollection
+ boolean internalPutIfAbsent(Replica replica, int index)
+ {
+ K key = toKey.apply(replica);
+ int otherIndex = map.put(key, index + 1);
+ if (otherIndex == 0)
+ return true;
+ map.put(key, otherIndex);
+ return false;
+ }
+
+ @Override
+ public boolean containsKey(Object key)
+ {
+ Preconditions.checkNotNull(key);
+ return get((K)key) != null;
+ }
+
+ public Replica get(Object key)
+ {
+ Preconditions.checkNotNull(key);
+ int index = map.get((K)key) - 1;
+ // since this map can be shared between sublists (or snapshots of mutables)
+ // we have to first corroborate that the index we've found is actually within our list's bounds
+ if (index < list.begin || index >= list.begin + list.size)
+ return null;
+ return list.contents[index];
+ }
+
+ @Override
+ public Replica remove(Object key)
+ {
+ throw new UnsupportedOperationException();
+ }
- public C snapshot()
+ @Override
+ public Set<K> keySet()
+ {
+ Set<K> ks = keySet;
+ if (ks == null)
+ keySet = ks = new KeySet();
+ return ks;
+ }
+
+ @Override
+ public Set<Entry<K, Replica>> entrySet()
+ {
+ Set<Entry<K, Replica>> es = entrySet;
+ if (es == null)
+ entrySet = es = new EntrySet();
+ return es;
+ }
+
+ public int size()
+ {
+ return list.size();
+ }
+
+ ReplicaMap<K> forSubList(ReplicaList subList)
+ {
+ assert subList.contents == list.contents;
+ return new ReplicaMap<>(subList, toKey, map);
+ }
+ }
+
+ protected final ReplicaList list;
+ AbstractReplicaCollection(ReplicaList list)
{
- return isSnapshot ? self()
- : snapshot(list.isEmpty() ? EMPTY_LIST
- : new ArrayList<>(list));
+ this.list = list;
}
+ /**
+ * construct a new Builder of our own type, so that we can concatenate
+ * TODO: this isn't terribly pretty, but we need sometimes to select / merge two Endpoints of unknown type;
+ */
+ public abstract Builder<C> newBuilder(int initialCapacity);
+
+ // return a new "sub-collection" with some sub-selection of the contents of this collection
+ abstract C snapshot(ReplicaList newList);
+ // return this object, if it is an immutable snapshot, otherwise returns a copy with these properties
+ public abstract C snapshot();
+
/** see {@link ReplicaCollection#subList(int, int)}*/
public final C subList(int start, int end)
{
- List<Replica> subList;
- if (isSnapshot)
- {
- if (start == 0 && end == size()) return self();
- else if (start == end) subList = EMPTY_LIST;
- else subList = list.subList(start, end);
- }
- else
- {
- if (start == end) subList = EMPTY_LIST;
- else subList = new ArrayList<>(list.subList(start, end)); // TODO: we could take a subList here, but comodification checks stop us
- }
+ if (start == 0 && end == size())
+ return snapshot();
+
+ ReplicaList subList;
+ if (start == end) subList = EMPTY_LIST;
+ else subList = list.subList(start, end);
+
return snapshot(subList);
}
@@ -122,7 +418,7 @@ public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollect
if (isEmpty())
return snapshot();
- List<Replica> copy = null;
+ ReplicaList copy = null;
int beginRun = -1, endRun = -1;
int i = 0;
for (; i < list.size() ; ++i)
@@ -136,7 +432,7 @@ public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollect
beginRun = i;
else if (endRun > 0)
{
- copy = new ArrayList<>(Math.min(limit, (list.size() - i) + (endRun - beginRun)));
+ copy = new ReplicaList(Math.min(limit, (list.size() - i) + (endRun - beginRun)));
for (int j = beginRun ; j < endRun ; ++j)
copy.add(list.get(j));
copy.add(list.get(i));
@@ -160,12 +456,22 @@ public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollect
return snapshot(copy);
}
+ /** see {@link ReplicaCollection#filterLazily(Predicate)}*/
+ public final Iterable<Replica> filterLazily(Predicate<Replica> predicate)
+ {
+ return filterLazily(predicate, Integer.MAX_VALUE);
+ }
+
+ /** see {@link ReplicaCollection#filterLazily(Predicate,int)}*/
+ public final Iterable<Replica> filterLazily(Predicate<Replica> predicate, int limit)
+ {
+ return () -> list.filterIterator(predicate, limit);
+ }
+
/** see {@link ReplicaCollection#sorted(Comparator)}*/
- public final C sorted(Comparator<Replica> comparator)
+ public final C sorted(Comparator<Replica> comparator)
{
- List<Replica> copy = new ArrayList<>(list);
- copy.sort(comparator);
- return snapshot(copy);
+ return snapshot(list.sorted(comparator));
}
public final Replica get(int i)
@@ -224,19 +530,19 @@ public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollect
@Override
public final String toString()
{
- return list.toString();
+ return Iterables.toString(list);
}
- static <C extends AbstractReplicaCollection<C>> C concat(C replicas, C extraReplicas, Mutable.Conflict ignoreConflicts)
+ static <C extends AbstractReplicaCollection<C>> C concat(C replicas, C extraReplicas, Builder.Conflict ignoreConflicts)
{
if (extraReplicas.isEmpty())
return replicas;
if (replicas.isEmpty())
return extraReplicas;
- Mutable<C> mutable = replicas.newMutable(replicas.size() + extraReplicas.size());
- mutable.addAll(replicas, Mutable.Conflict.NONE);
- mutable.addAll(extraReplicas, ignoreConflicts);
- return mutable.asSnapshot();
+ Builder<C> builder = replicas.newBuilder(replicas.size() + extraReplicas.size());
+ builder.addAll(replicas, Builder.Conflict.NONE);
+ builder.addAll(extraReplicas, ignoreConflicts);
+ return builder.build();
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
index bad736f..818e20e 100644
--- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
+++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
@@ -23,7 +23,7 @@ import java.util.*;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -223,7 +223,7 @@ public abstract class AbstractReplicationStrategy
*/
public RangesByEndpoint getAddressReplicas(TokenMetadata metadata)
{
- RangesByEndpoint.Mutable map = new RangesByEndpoint.Mutable();
+ RangesByEndpoint.Builder map = new RangesByEndpoint.Builder();
for (Token token : metadata.sortedTokens())
{
@@ -236,7 +236,7 @@ public abstract class AbstractReplicationStrategy
}
}
- return map.asImmutableView();
+ return map.build();
}
public RangesAtEndpoint getAddressReplicas(TokenMetadata metadata, InetAddressAndPort endpoint)
@@ -260,7 +260,7 @@ public abstract class AbstractReplicationStrategy
public EndpointsByRange getRangeAddresses(TokenMetadata metadata)
{
- EndpointsByRange.Mutable map = new EndpointsByRange.Mutable();
+ EndpointsByRange.Builder map = new EndpointsByRange.Builder();
for (Token token : metadata.sortedTokens())
{
@@ -273,7 +273,7 @@ public abstract class AbstractReplicationStrategy
}
}
- return map.asImmutableView();
+ return map.build();
}
public RangesByEndpoint getAddressReplicas()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/Endpoints.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/Endpoints.java b/src/java/org/apache/cassandra/locator/Endpoints.java
index 28e578c..ee42e36 100644
--- a/src/java/org/apache/cassandra/locator/Endpoints.java
+++ b/src/java/org/apache/cassandra/locator/Endpoints.java
@@ -18,13 +18,10 @@
package org.apache.cassandra.locator;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.apache.cassandra.utils.FBUtilities;
import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -36,13 +33,16 @@ import java.util.Set;
*/
public abstract class Endpoints<E extends Endpoints<E>> extends AbstractReplicaCollection<E>
{
- static final Map<InetAddressAndPort, Replica> EMPTY_MAP = Collections.unmodifiableMap(new LinkedHashMap<>());
+ static ReplicaMap<InetAddressAndPort> endpointMap(ReplicaList list) { return new ReplicaMap<>(list, Replica::endpoint); }
+ static final ReplicaMap<InetAddressAndPort> EMPTY_MAP = endpointMap(EMPTY_LIST);
- volatile Map<InetAddressAndPort, Replica> byEndpoint;
+ // volatile not needed, as has only final members,
+ // besides (transitively) those that cache objects that themselves have only final members
+ ReplicaMap<InetAddressAndPort> byEndpoint;
- Endpoints(List<Replica> list, boolean isSnapshot, Map<InetAddressAndPort, Replica> byEndpoint)
+ Endpoints(ReplicaList list, ReplicaMap<InetAddressAndPort> byEndpoint)
{
- super(list, isSnapshot);
+ super(list);
this.byEndpoint = byEndpoint;
}
@@ -54,9 +54,9 @@ public abstract class Endpoints<E extends Endpoints<E>> extends AbstractReplicaC
public Map<InetAddressAndPort, Replica> byEndpoint()
{
- Map<InetAddressAndPort, Replica> map = byEndpoint;
+ ReplicaMap<InetAddressAndPort> map = byEndpoint;
if (map == null)
- byEndpoint = map = buildByEndpoint(list);
+ byEndpoint = map = endpointMap(list);
return map;
}
@@ -75,19 +75,6 @@ public abstract class Endpoints<E extends Endpoints<E>> extends AbstractReplicaC
replica);
}
- private static Map<InetAddressAndPort, Replica> buildByEndpoint(List<Replica> list)
- {
- // TODO: implement a delegating map that uses our superclass' list, and is immutable
- Map<InetAddressAndPort, Replica> byEndpoint = new LinkedHashMap<>(list.size());
- for (Replica replica : list)
- {
- Replica prev = byEndpoint.put(replica.endpoint(), replica);
- assert prev == null : "duplicate endpoint in EndpointsForRange: " + prev + " and " + replica;
- }
-
- return Collections.unmodifiableMap(byEndpoint);
- }
-
public E withoutSelf()
{
InetAddressAndPort self = FBUtilities.getBroadcastAddressAndPort();
@@ -121,7 +108,7 @@ public abstract class Endpoints<E extends Endpoints<E>> extends AbstractReplicaC
*/
public E select(Iterable<InetAddressAndPort> endpoints, boolean ignoreMissing)
{
- ReplicaCollection.Mutable<E> copy = newMutable(
+ Builder<E> copy = newBuilder(
endpoints instanceof Collection<?>
? ((Collection<InetAddressAndPort>) endpoints).size()
: size()
@@ -136,9 +123,9 @@ public abstract class Endpoints<E extends Endpoints<E>> extends AbstractReplicaC
throw new IllegalArgumentException(endpoint + " is not present in " + this);
continue;
}
- copy.add(select, ReplicaCollection.Mutable.Conflict.DUPLICATE);
+ copy.add(select, Builder.Conflict.DUPLICATE);
}
- return copy.asSnapshot();
+ return copy.build();
}
/**
@@ -157,10 +144,10 @@ public abstract class Endpoints<E extends Endpoints<E>> extends AbstractReplicaC
public static <E extends Endpoints<E>> E append(E replicas, Replica extraReplica)
{
- Mutable<E> mutable = replicas.newMutable(replicas.size() + 1);
- mutable.addAll(replicas);
- mutable.add(extraReplica, Conflict.NONE);
- return mutable.asSnapshot();
+ Builder<E> builder = replicas.newBuilder(replicas.size() + 1);
+ builder.addAll(replicas);
+ builder.add(extraReplica, Conflict.NONE);
+ return builder.build();
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/EndpointsByRange.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/EndpointsByRange.java b/src/java/org/apache/cassandra/locator/EndpointsByRange.java
index cdc8a68..2d1cde6 100644
--- a/src/java/org/apache/cassandra/locator/EndpointsByRange.java
+++ b/src/java/org/apache/cassandra/locator/EndpointsByRange.java
@@ -19,12 +19,14 @@
package org.apache.cassandra.locator;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
public class EndpointsByRange extends ReplicaMultimap<Range<Token>, EndpointsForRange>
@@ -40,12 +42,12 @@ public class EndpointsByRange extends ReplicaMultimap<Range<Token>, EndpointsFor
return map.getOrDefault(range, EndpointsForRange.empty(range));
}
- public static class Mutable extends ReplicaMultimap.Mutable<Range<Token>, EndpointsForRange.Mutable>
+ public static class Builder extends ReplicaMultimap.Builder<Range<Token>, EndpointsForRange.Builder>
{
@Override
- protected EndpointsForRange.Mutable newMutable(Range<Token> range)
+ protected EndpointsForRange.Builder newBuilder(Range<Token> range)
{
- return new EndpointsForRange.Mutable(range);
+ return new EndpointsForRange.Builder(range);
}
// TODO: consider all ignoreDuplicates cases
@@ -54,9 +56,11 @@ public class EndpointsByRange extends ReplicaMultimap<Range<Token>, EndpointsFor
get(range).addAll(replicas, ignoreConflicts);
}
- public EndpointsByRange asImmutableView()
+ public EndpointsByRange build()
{
- return new EndpointsByRange(Collections.unmodifiableMap(Maps.transformValues(map, EndpointsForRange.Mutable::asImmutableView)));
+ return new EndpointsByRange(
+ ImmutableMap.copyOf(
+ Maps.transformValues(this.map, EndpointsForRange.Builder::build)));
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/EndpointsByReplica.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/EndpointsByReplica.java b/src/java/org/apache/cassandra/locator/EndpointsByReplica.java
index ceea2d1..72d8751 100644
--- a/src/java/org/apache/cassandra/locator/EndpointsByReplica.java
+++ b/src/java/org/apache/cassandra/locator/EndpointsByReplica.java
@@ -19,10 +19,12 @@
package org.apache.cassandra.locator;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
public class EndpointsByReplica extends ReplicaMultimap<Replica, EndpointsForRange>
@@ -38,23 +40,25 @@ public class EndpointsByReplica extends ReplicaMultimap<Replica, EndpointsForRan
return map.getOrDefault(range, EndpointsForRange.empty(range.range()));
}
- public static class Mutable extends ReplicaMultimap.Mutable<Replica, EndpointsForRange.Mutable>
+ public static class Builder extends ReplicaMultimap.Builder<Replica, EndpointsForRange.Builder>
{
@Override
- protected EndpointsForRange.Mutable newMutable(Replica replica)
+ protected EndpointsForRange.Builder newBuilder(Replica replica)
{
- return new EndpointsForRange.Mutable(replica.range());
+ return new EndpointsForRange.Builder(replica.range());
}
// TODO: consider all ignoreDuplicates cases
public void putAll(Replica range, EndpointsForRange replicas, Conflict ignoreConflicts)
{
- map.computeIfAbsent(range, r -> newMutable(r)).addAll(replicas, ignoreConflicts);
+ map.computeIfAbsent(range, r -> newBuilder(r)).addAll(replicas, ignoreConflicts);
}
- public EndpointsByReplica asImmutableView()
+ public EndpointsByReplica build()
{
- return new EndpointsByReplica(Collections.unmodifiableMap(Maps.transformValues(map, EndpointsForRange.Mutable::asImmutableView)));
+ return new EndpointsByReplica(
+ ImmutableMap.copyOf(
+ Maps.transformValues(this.map, EndpointsForRange.Builder::build)));
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/EndpointsForRange.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/EndpointsForRange.java b/src/java/org/apache/cassandra/locator/EndpointsForRange.java
index f812951..7039df0 100644
--- a/src/java/org/apache/cassandra/locator/EndpointsForRange.java
+++ b/src/java/org/apache/cassandra/locator/EndpointsForRange.java
@@ -22,13 +22,8 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
import static com.google.common.collect.Iterables.all;
@@ -40,13 +35,9 @@ import static com.google.common.collect.Iterables.all;
public class EndpointsForRange extends Endpoints<EndpointsForRange>
{
private final Range<Token> range;
- private EndpointsForRange(Range<Token> range, List<Replica> list, boolean isSnapshot)
+ private EndpointsForRange(Range<Token> range, ReplicaList list, ReplicaMap<InetAddressAndPort> byEndpoint)
{
- this(range, list, isSnapshot, null);
- }
- private EndpointsForRange(Range<Token> range, List<Replica> list, boolean isSnapshot, Map<InetAddressAndPort, Replica> byEndpoint)
- {
- super(list, isSnapshot, byEndpoint);
+ super(list, byEndpoint);
this.range = range;
assert range != null;
}
@@ -57,95 +48,77 @@ public class EndpointsForRange extends Endpoints<EndpointsForRange>
}
@Override
- public Mutable newMutable(int initialCapacity)
+ public Builder newBuilder(int initialCapacity)
{
- return new Mutable(range, initialCapacity);
+ return new Builder(range, initialCapacity);
}
public EndpointsForToken forToken(Token token)
{
if (!range.contains(token))
throw new IllegalArgumentException(token + " is not contained within " + range);
- return new EndpointsForToken(token, list, isSnapshot, byEndpoint);
+ return new EndpointsForToken(token, list, byEndpoint);
}
@Override
- public EndpointsForRange self()
+ public EndpointsForRange snapshot()
{
return this;
}
@Override
- protected EndpointsForRange snapshot(List<Replica> snapshot)
+ EndpointsForRange snapshot(ReplicaList newList)
{
- if (snapshot.isEmpty()) return empty(range);
- return new EndpointsForRange(range, snapshot, true);
+ if (newList.isEmpty()) return empty(range);
+ ReplicaMap<InetAddressAndPort> byEndpoint = null;
+ if (this.byEndpoint != null && list.isSubList(newList))
+ byEndpoint = this.byEndpoint.forSubList(newList);
+ return new EndpointsForRange(range, newList, byEndpoint);
}
- public static class Mutable extends EndpointsForRange implements ReplicaCollection.Mutable<EndpointsForRange>
+ public static class Builder extends EndpointsForRange implements ReplicaCollection.Builder<EndpointsForRange>
{
- boolean hasSnapshot;
- public Mutable(Range<Token> range) { this(range, 0); }
- public Mutable(Range<Token> range, int capacity) { super(range, new ArrayList<>(capacity), false, new LinkedHashMap<>(capacity)); }
+ boolean built;
+ public Builder(Range<Token> range) { this(range, 0); }
+ public Builder(Range<Token> range, int capacity) { this(range, new ReplicaList(capacity)); }
+ private Builder(Range<Token> range, ReplicaList list) { super(range, list, endpointMap(list)); }
- public void add(Replica replica, Conflict ignoreConflict)
+ public EndpointsForRange.Builder add(Replica replica, Conflict ignoreConflict)
{
- if (hasSnapshot) throw new IllegalStateException();
+ if (built) throw new IllegalStateException();
Preconditions.checkNotNull(replica);
if (!replica.range().contains(super.range))
throw new IllegalArgumentException("Replica " + replica + " does not contain " + super.range);
- Replica prev = super.byEndpoint.put(replica.endpoint(), replica);
- if (prev != null)
+ if (!super.byEndpoint.internalPutIfAbsent(replica, list.size()))
{
- super.byEndpoint.put(replica.endpoint(), prev); // restore prev
switch (ignoreConflict)
{
case DUPLICATE:
- if (prev.equals(replica))
+ if (byEndpoint().get(replica.endpoint()).equals(replica))
break;
case NONE:
- throw new IllegalArgumentException("Conflicting replica added (expected unique endpoints): " + replica + "; existing: " + prev);
+ throw new IllegalArgumentException("Conflicting replica added (expected unique endpoints): "
+ + replica + "; existing: " + byEndpoint().get(replica.endpoint()));
case ALL:
}
- return;
+ return this;
}
list.add(replica);
+ return this;
}
@Override
- public Map<InetAddressAndPort, Replica> byEndpoint()
- {
- // our internal map is modifiable, but it is unsafe to modify the map externally
- // it would be possible to implement a safe modifiable map, but it is probably not valuable
- return Collections.unmodifiableMap(super.byEndpoint());
- }
-
- private EndpointsForRange get(boolean isSnapshot)
+ public EndpointsForRange snapshot()
{
- return new EndpointsForRange(super.range, super.list, isSnapshot, Collections.unmodifiableMap(super.byEndpoint));
+ return snapshot(list.subList(0, list.size()));
}
- public EndpointsForRange asImmutableView()
- {
- return get(false);
- }
-
- public EndpointsForRange asSnapshot()
- {
- hasSnapshot = true;
- return get(true);
- }
- }
-
- public static class Builder extends ReplicaCollection.Builder<EndpointsForRange, Mutable, EndpointsForRange.Builder>
- {
- public Builder(Range<Token> range) { this(range, 0); }
- public Builder(Range<Token> range, int capacity) { super (new Mutable(range, capacity)); }
- public boolean containsEndpoint(InetAddressAndPort endpoint)
+ public EndpointsForRange build()
{
- return mutable.asImmutableView().byEndpoint.containsKey(endpoint);
+ built = true;
+ return new EndpointsForRange(super.range, super.list, super.byEndpoint);
}
}
@@ -160,16 +133,16 @@ public class EndpointsForRange extends Endpoints<EndpointsForRange>
public static EndpointsForRange empty(Range<Token> range)
{
- return new EndpointsForRange(range, EMPTY_LIST, true, EMPTY_MAP);
+ return new EndpointsForRange(range, EMPTY_LIST, EMPTY_MAP);
}
public static EndpointsForRange of(Replica replica)
{
// we only use ArrayList or ArrayList.SubList, to ensure callsites are bimorphic
- ArrayList<Replica> one = new ArrayList<>(1);
+ ReplicaList one = new ReplicaList(1);
one.add(replica);
// we can safely use singletonMap, as we only otherwise use LinkedHashMap
- return new EndpointsForRange(replica.range(), one, true, Collections.unmodifiableMap(Collections.singletonMap(replica.endpoint(), replica)));
+ return new EndpointsForRange(replica.range(), one, endpointMap(one));
}
public static EndpointsForRange of(Replica ... replicas)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/EndpointsForToken.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/EndpointsForToken.java b/src/java/org/apache/cassandra/locator/EndpointsForToken.java
index 3446dc9..c709988 100644
--- a/src/java/org/apache/cassandra/locator/EndpointsForToken.java
+++ b/src/java/org/apache/cassandra/locator/EndpointsForToken.java
@@ -21,13 +21,8 @@ package org.apache.cassandra.locator;
import com.google.common.base.Preconditions;
import org.apache.cassandra.dht.Token;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
/**
* A ReplicaCollection where all Replica are required to cover a range that fully contains the token() defined in the builder().
@@ -37,14 +32,10 @@ import java.util.Map;
public class EndpointsForToken extends Endpoints<EndpointsForToken>
{
private final Token token;
- private EndpointsForToken(Token token, List<Replica> list, boolean isSnapshot)
- {
- this(token, list, isSnapshot, null);
- }
- EndpointsForToken(Token token, List<Replica> list, boolean isSnapshot, Map<InetAddressAndPort, Replica> byEndpoint)
+ EndpointsForToken(Token token, ReplicaList list, ReplicaMap<InetAddressAndPort> byEndpoint)
{
- super(list, isSnapshot, byEndpoint);
+ super(list, byEndpoint);
this.token = token;
assert token != null;
}
@@ -55,85 +46,71 @@ public class EndpointsForToken extends Endpoints<EndpointsForToken>
}
@Override
- public Mutable newMutable(int initialCapacity)
+ public Builder newBuilder(int initialCapacity)
{
- return new Mutable(token, initialCapacity);
+ return new Builder(token, initialCapacity);
}
@Override
- public EndpointsForToken self()
+ public EndpointsForToken snapshot()
{
return this;
}
@Override
- protected EndpointsForToken snapshot(List<Replica> subList)
+ protected EndpointsForToken snapshot(ReplicaList newList)
{
- if (subList.isEmpty()) return empty(token);
- return new EndpointsForToken(token, subList, true);
+ if (newList.isEmpty()) return empty(token);
+ ReplicaMap<InetAddressAndPort> byEndpoint = null;
+ if (this.byEndpoint != null && list.isSubList(newList))
+ byEndpoint = this.byEndpoint.forSubList(newList);
+ return new EndpointsForToken(token, newList, byEndpoint);
}
- public static class Mutable extends EndpointsForToken implements ReplicaCollection.Mutable<EndpointsForToken>
+ public static class Builder extends EndpointsForToken implements ReplicaCollection.Builder<EndpointsForToken>
{
- boolean hasSnapshot;
- public Mutable(Token token) { this(token, 0); }
- public Mutable(Token token, int capacity) { super(token, new ArrayList<>(capacity), false, new LinkedHashMap<>(capacity)); }
+ boolean built;
+ public Builder(Token token) { this(token, 0); }
+ public Builder(Token token, int capacity) { this(token, new ReplicaList(capacity)); }
+ private Builder(Token token, ReplicaList list) { super(token, list, endpointMap(list)); }
- public void add(Replica replica, Conflict ignoreConflict)
+ public EndpointsForToken.Builder add(Replica replica, Conflict ignoreConflict)
{
- if (hasSnapshot) throw new IllegalStateException();
+ if (built) throw new IllegalStateException();
Preconditions.checkNotNull(replica);
if (!replica.range().contains(super.token))
throw new IllegalArgumentException("Replica " + replica + " does not contain " + super.token);
- Replica prev = super.byEndpoint.put(replica.endpoint(), replica);
- if (prev != null)
+ if (!super.byEndpoint.internalPutIfAbsent(replica, list.size()))
{
- super.byEndpoint.put(replica.endpoint(), prev); // restore prev
switch (ignoreConflict)
{
case DUPLICATE:
- if (prev.equals(replica))
+ if (byEndpoint().get(replica.endpoint()).equals(replica))
break;
case NONE:
- throw new IllegalArgumentException("Conflicting replica added (expected unique endpoints): " + replica + "; existing: " + prev);
+ throw new IllegalArgumentException("Conflicting replica added (expected unique endpoints): "
+ + replica + "; existing: " + byEndpoint().get(replica.endpoint()));
case ALL:
}
- return;
+ return this;
}
list.add(replica);
+ return this;
}
@Override
- public Map<InetAddressAndPort, Replica> byEndpoint()
+ public EndpointsForToken snapshot()
{
- // our internal map is modifiable, but it is unsafe to modify the map externally
- // it would be possible to implement a safe modifiable map, but it is probably not valuable
- return Collections.unmodifiableMap(super.byEndpoint());
+ return snapshot(list.subList(0, list.size()));
}
- private EndpointsForToken get(boolean isSnapshot)
+ public EndpointsForToken build()
{
- return new EndpointsForToken(super.token, super.list, isSnapshot, Collections.unmodifiableMap(super.byEndpoint));
+ built = true;
+ return new EndpointsForToken(super.token, super.list, super.byEndpoint);
}
-
- public EndpointsForToken asImmutableView()
- {
- return get(false);
- }
-
- public EndpointsForToken asSnapshot()
- {
- hasSnapshot = true;
- return get(true);
- }
- }
-
- public static class Builder extends ReplicaCollection.Builder<EndpointsForToken, Mutable, EndpointsForToken.Builder>
- {
- public Builder(Token token) { this(token, 0); }
- public Builder(Token token, int capacity) { super (new Mutable(token, capacity)); }
}
public static Builder builder(Token token)
@@ -147,16 +124,16 @@ public class EndpointsForToken extends Endpoints<EndpointsForToken>
public static EndpointsForToken empty(Token token)
{
- return new EndpointsForToken(token, EMPTY_LIST, true, EMPTY_MAP);
+ return new EndpointsForToken(token, EMPTY_LIST, EMPTY_MAP);
}
public static EndpointsForToken of(Token token, Replica replica)
{
// we only use ArrayList or ArrayList.SubList, to ensure callsites are bimorphic
- ArrayList<Replica> one = new ArrayList<>(1);
+ ReplicaList one = new ReplicaList(1);
one.add(replica);
// we can safely use singletonMap, as we only otherwise use LinkedHashMap
- return new EndpointsForToken(token, one, true, Collections.unmodifiableMap(Collections.singletonMap(replica.endpoint(), replica)));
+ return new EndpointsForToken(token, one, endpointMap(one));
}
public static EndpointsForToken of(Token token, Replica ... replicas)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java b/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
index c63f4f3..713a75e 100644
--- a/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
+++ b/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
@@ -20,7 +20,7 @@ package org.apache.cassandra.locator;
import java.util.*;
import java.util.Map.Entry;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,7 +87,7 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
private static final class DatacenterEndpoints
{
/** List accepted endpoints get pushed into. */
- EndpointsForRange.Mutable replicas;
+ EndpointsForRange.Builder replicas;
/**
* Racks encountered so far. Replicas are put into separate racks while possible.
@@ -101,7 +101,7 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
int acceptableRackRepeats;
int transients;
- DatacenterEndpoints(ReplicationFactor rf, int rackCount, int nodeCount, EndpointsForRange.Mutable replicas, Set<Pair<String, String>> racks)
+ DatacenterEndpoints(ReplicationFactor rf, int rackCount, int nodeCount, EndpointsForRange.Builder replicas, Set<Pair<String, String>> racks)
{
this.replicas = replicas;
this.racks = racks;
@@ -168,7 +168,7 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
Token replicaStart = tokenMetadata.getPredecessor(replicaEnd);
Range<Token> replicatedRange = new Range<>(replicaStart, replicaEnd);
- EndpointsForRange.Mutable builder = new EndpointsForRange.Mutable(replicatedRange);
+ EndpointsForRange.Builder builder = new EndpointsForRange.Builder(replicatedRange);
Set<Pair<String, String>> seenRacks = new HashSet<>();
Topology topology = tokenMetadata.getTopology();
@@ -206,7 +206,7 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
if (dcEndpoints != null && dcEndpoints.addEndpointAndCheckIfDone(ep, location, replicatedRange))
--dcsToFill;
}
- return builder.asImmutableView();
+ return builder.build();
}
private int sizeOrZero(Multimap<?, ?> collection)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/OldNetworkTopologyStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/OldNetworkTopologyStrategy.java b/src/java/org/apache/cassandra/locator/OldNetworkTopologyStrategy.java
index 449c51e..37152b4 100644
--- a/src/java/org/apache/cassandra/locator/OldNetworkTopologyStrategy.java
+++ b/src/java/org/apache/cassandra/locator/OldNetworkTopologyStrategy.java
@@ -54,7 +54,7 @@ public class OldNetworkTopologyStrategy extends AbstractReplicationStrategy
Token previousToken = metadata.getPredecessor(primaryToken);
Range<Token> tokenRange = new Range<>(previousToken, primaryToken);
- EndpointsForRange.Builder replicas = EndpointsForRange.builder(tokenRange, rf.allReplicas);
+ EndpointsForRange.Builder replicas = new EndpointsForRange.Builder(tokenRange, rf.allReplicas);
assert !rf.hasTransientReplicas() : "support transient replicas";
replicas.add(new Replica(metadata.getEndpoint(primaryToken), previousToken, primaryToken, true));
@@ -98,7 +98,7 @@ public class OldNetworkTopologyStrategy extends AbstractReplicationStrategy
{
Token t = iter.next();
Replica replica = new Replica(metadata.getEndpoint(t), previousToken, primaryToken, true);
- if (!replicas.containsEndpoint(replica.endpoint()))
+ if (!replicas.endpoints().contains(replica.endpoint()))
replicas.add(replica);
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/PendingRangeMaps.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/PendingRangeMaps.java b/src/java/org/apache/cassandra/locator/PendingRangeMaps.java
index b8b7bc6..f9e3f66 100644
--- a/src/java/org/apache/cassandra/locator/PendingRangeMaps.java
+++ b/src/java/org/apache/cassandra/locator/PendingRangeMaps.java
@@ -23,11 +23,11 @@ package org.apache.cassandra.locator;
import com.google.common.collect.Iterators;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import java.util.*;
-public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, EndpointsForRange.Mutable>>
+public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, EndpointsForRange.Builder>>
{
/**
* We have for NavigableMap to be able to search for ranges containing a token efficiently.
@@ -35,7 +35,7 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
* First two are for non-wrap-around ranges, and the last two are for wrap-around ranges.
*/
// ascendingMap will sort the ranges by the ascending order of right token
- private final NavigableMap<Range<Token>, EndpointsForRange.Mutable> ascendingMap;
+ private final NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingMap;
/**
* sorting end ascending, if ends are same, sorting begin descending, so that token (end, end) will
@@ -50,7 +50,7 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
};
// ascendingMap will sort the ranges by the descending order of left token
- private final NavigableMap<Range<Token>, EndpointsForRange.Mutable> descendingMap;
+ private final NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingMap;
/**
* sorting begin descending, if begins are same, sorting end descending, so that token (begin, begin) will
@@ -66,7 +66,7 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
};
// these two maps are for warp around ranges.
- private final NavigableMap<Range<Token>, EndpointsForRange.Mutable> ascendingMapForWrapAround;
+ private final NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingMapForWrapAround;
/**
* for wrap around range (begin, end], which begin > end.
@@ -82,7 +82,7 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
return o1.left.compareTo(o2.left);
};
- private final NavigableMap<Range<Token>, EndpointsForRange.Mutable> descendingMapForWrapAround;
+ private final NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingMapForWrapAround;
/**
* for wrap around ranges, which begin > end.
@@ -106,13 +106,13 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
static final void addToMap(Range<Token> range,
Replica replica,
- NavigableMap<Range<Token>, EndpointsForRange.Mutable> ascendingMap,
- NavigableMap<Range<Token>, EndpointsForRange.Mutable> descendingMap)
+ NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingMap,
+ NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingMap)
{
- EndpointsForRange.Mutable replicas = ascendingMap.get(range);
+ EndpointsForRange.Builder replicas = ascendingMap.get(range);
if (replicas == null)
{
- replicas = new EndpointsForRange.Mutable(range,1);
+ replicas = new EndpointsForRange.Builder(range,1);
ascendingMap.put(range, replicas);
descendingMap.put(range, replicas);
}
@@ -132,13 +132,13 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
}
static final void addIntersections(EndpointsForToken.Builder replicasToAdd,
- NavigableMap<Range<Token>, EndpointsForRange.Mutable> smallerMap,
- NavigableMap<Range<Token>, EndpointsForRange.Mutable> biggerMap)
+ NavigableMap<Range<Token>, EndpointsForRange.Builder> smallerMap,
+ NavigableMap<Range<Token>, EndpointsForRange.Builder> biggerMap)
{
// find the intersection of two sets
for (Range<Token> range : smallerMap.keySet())
{
- EndpointsForRange.Mutable replicas = biggerMap.get(range);
+ EndpointsForRange.Builder replicas = biggerMap.get(range);
if (replicas != null)
{
replicasToAdd.addAll(replicas);
@@ -153,8 +153,8 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
Range<Token> searchRange = new Range<>(token, token);
// search for non-wrap-around maps
- NavigableMap<Range<Token>, EndpointsForRange.Mutable> ascendingTailMap = ascendingMap.tailMap(searchRange, true);
- NavigableMap<Range<Token>, EndpointsForRange.Mutable> descendingTailMap = descendingMap.tailMap(searchRange, false);
+ NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingTailMap = ascendingMap.tailMap(searchRange, true);
+ NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingTailMap = descendingMap.tailMap(searchRange, false);
// add intersections of two maps
if (ascendingTailMap.size() < descendingTailMap.size())
@@ -171,11 +171,11 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
descendingTailMap = descendingMapForWrapAround.tailMap(searchRange, false);
// add them since they are all necessary.
- for (Map.Entry<Range<Token>, EndpointsForRange.Mutable> entry : ascendingTailMap.entrySet())
+ for (Map.Entry<Range<Token>, EndpointsForRange.Builder> entry : ascendingTailMap.entrySet())
{
replicas.addAll(entry.getValue());
}
- for (Map.Entry<Range<Token>, EndpointsForRange.Mutable> entry : descendingTailMap.entrySet())
+ for (Map.Entry<Range<Token>, EndpointsForRange.Builder> entry : descendingTailMap.entrySet())
{
replicas.addAll(entry.getValue());
}
@@ -187,7 +187,7 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
{
StringBuilder sb = new StringBuilder();
- for (Map.Entry<Range<Token>, EndpointsForRange.Mutable> entry : this)
+ for (Map.Entry<Range<Token>, EndpointsForRange.Builder> entry : this)
{
Range<Token> range = entry.getKey();
@@ -202,7 +202,7 @@ public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, Endpoi
}
@Override
- public Iterator<Map.Entry<Range<Token>, EndpointsForRange.Mutable>> iterator()
+ public Iterator<Map.Entry<Range<Token>, EndpointsForRange.Builder>> iterator()
{
return Iterators.concat(ascendingMap.entrySet().iterator(), ascendingMapForWrapAround.entrySet().iterator());
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e645b917/src/java/org/apache/cassandra/locator/RangesAtEndpoint.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/RangesAtEndpoint.java b/src/java/org/apache/cassandra/locator/RangesAtEndpoint.java
index 8319d92..33ddffd 100644
--- a/src/java/org/apache/cassandra/locator/RangesAtEndpoint.java
+++ b/src/java/org/apache/cassandra/locator/RangesAtEndpoint.java
@@ -25,20 +25,17 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import java.net.UnknownHostException;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collector;
-import java.util.stream.Collectors;
import static com.google.common.collect.Iterables.all;
-import static org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict.*;
+import static org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict.*;
/**
* A ReplicaCollection for Ranges occurring at an endpoint. All Replica will be for the same endpoint,
@@ -46,20 +43,20 @@ import static org.apache.cassandra.locator.ReplicaCollection.Mutable.Conflict.*;
*/
public class RangesAtEndpoint extends AbstractReplicaCollection<RangesAtEndpoint>
{
- private static final Map<Range<Token>, Replica> EMPTY_MAP = Collections.unmodifiableMap(new LinkedHashMap<>());
+ private static ReplicaMap<Range<Token>> rangeMap(ReplicaList list) { return new ReplicaMap<>(list, Replica::range); }
+ private static final ReplicaMap<Range<Token>> EMPTY_MAP = rangeMap(EMPTY_LIST);
private final InetAddressAndPort endpoint;
- private volatile Map<Range<Token>, Replica> byRange;
- private volatile RangesAtEndpoint fullRanges;
- private volatile RangesAtEndpoint transRanges;
- private RangesAtEndpoint(InetAddressAndPort endpoint, List<Replica> list, boolean isSnapshot)
- {
- this(endpoint, list, isSnapshot, null);
- }
- private RangesAtEndpoint(InetAddressAndPort endpoint, List<Replica> list, boolean isSnapshot, Map<Range<Token>, Replica> byRange)
+ // volatile not needed, as all of these caching collections have final members,
+ // besides (transitively) those that cache objects that themselves have only final members
+ private ReplicaMap<Range<Token>> byRange;
+ private RangesAtEndpoint onlyFull;
+ private RangesAtEndpoint onlyTransient;
+
+ private RangesAtEndpoint(InetAddressAndPort endpoint, ReplicaList list, ReplicaMap<Range<Token>> byRange)
{
- super(list, isSnapshot);
+ super(list);
this.endpoint = endpoint;
this.byRange = byRange;
assert endpoint != null;
@@ -79,36 +76,47 @@ public class RangesAtEndpoint extends AbstractReplicaCollection<RangesAtEndpoint
);
}
+ /**
+ * @return a set of all unique Ranges
+ * This method is threadsafe, though it is not synchronised
+ */
public Set<Range<Token>> ranges()
{
return byRange().keySet();
}
+ /**
+ * @return a map of all Ranges, to their owning Replica instance
+ * This method is threadsafe, though it is not synchronised
+ */
public Map<Range<Token>, Replica> byRange()
{
- Map<Range<Token>, Replica> map = byRange;
+ ReplicaMap<Range<Token>> map = byRange;
if (map == null)
- byRange = map = buildByRange(list);
+ byRange = map = rangeMap(list);
return map;
}
@Override
- protected RangesAtEndpoint snapshot(List<Replica> subList)
+ protected RangesAtEndpoint snapshot(ReplicaList newList)
{
- if (subList.isEmpty()) return empty(endpoint);
- return new RangesAtEndpoint(endpoint, subList, true);
+ if (newList.isEmpty()) return empty(endpoint);
+ ReplicaMap<Range<Token>> byRange = null;
+ if (this.byRange != null && list.isSubList(newList))
+ byRange = this.byRange.forSubList(newList);
+ return new RangesAtEndpoint(endpoint, newList, byRange);
}
@Override
- public RangesAtEndpoint self()
+ public RangesAtEndpoint snapshot()
{
return this;
}
@Override
- public ReplicaCollection.Mutable<RangesAtEndpoint> newMutable(int initialCapacity)
+ public ReplicaCollection.Builder<RangesAtEndpoint> newBuilder(int initialCapacity)
{
- return new Mutable(endpoint, initialCapacity);
+ return new Builder(endpoint, initialCapacity);
}
@Override
@@ -120,30 +128,20 @@ public class RangesAtEndpoint extends AbstractReplicaCollection<RangesAtEndpoint
replica);
}
- public RangesAtEndpoint full()
- {
- RangesAtEndpoint coll = fullRanges;
- if (fullRanges == null)
- fullRanges = coll = filter(Replica::isFull);
- return coll;
- }
-
- public RangesAtEndpoint trans()
- {
- RangesAtEndpoint coll = transRanges;
- if (transRanges == null)
- transRanges = coll = filter(Replica::isTransient);
- return coll;
- }
-
- public Collection<Range<Token>> fullRanges()
+ public RangesAtEndpoint onlyFull()
{
- return full().ranges();
+ RangesAtEndpoint result = onlyFull;
+ if (onlyFull == null)
+ onlyFull = result = filter(Replica::isFull);
+ return result;
}
- public Collection<Range<Token>> transientRanges()
+ public RangesAtEndpoint onlyTransient()
{
- return trans().ranges();
+ RangesAtEndpoint result = onlyTransient;
+ if (onlyTransient == null)
+ onlyTransient = result = filter(Replica::isTransient);
+ return result;
}
public boolean contains(Range<Token> range, boolean isFull)
@@ -152,19 +150,6 @@ public class RangesAtEndpoint extends AbstractReplicaCollection<RangesAtEndpoint
return replica != null && replica.isFull() == isFull;
}
- private static Map<Range<Token>, Replica> buildByRange(List<Replica> list)
- {
- // TODO: implement a delegating map that uses our superclass' list, and is immutable
- Map<Range<Token>, Replica> byRange = new LinkedHashMap<>(list.size());
- for (Replica replica : list)
- {
- Replica prev = byRange.put(replica.range(), replica);
- assert prev == null : "duplicate range in RangesAtEndpoint: " + prev + " and " + replica;
- }
-
- return Collections.unmodifiableMap(byRange);
- }
-
/**
* @return if there are no wrap around ranges contained in this RangesAtEndpoint, return self;
* otherwise, return a RangesAtEndpoint covering the same logical portions of the ring, but with those ranges unwrapped
@@ -201,89 +186,71 @@ public class RangesAtEndpoint extends AbstractReplicaCollection<RangesAtEndpoint
return collector(ImmutableSet.of(), () -> new Builder(endpoint));
}
- public static class Mutable extends RangesAtEndpoint implements ReplicaCollection.Mutable<RangesAtEndpoint>
+ public static class Builder extends RangesAtEndpoint implements ReplicaCollection.Builder<RangesAtEndpoint>
{
- boolean hasSnapshot;
- public Mutable(InetAddressAndPort endpoint) { this(endpoint, 0); }
- public Mutable(InetAddressAndPort endpoint, int capacity) { super(endpoint, new ArrayList<>(capacity), false, new LinkedHashMap<>(capacity)); }
+ boolean built;
+ public Builder(InetAddressAndPort endpoint) { this(endpoint, 0); }
+ public Builder(InetAddressAndPort endpoint, int capacity) { this(endpoint, new ReplicaList(capacity)); }
+ private Builder(InetAddressAndPort endpoint, ReplicaList list) { super(endpoint, list, rangeMap(list)); }
- public void add(Replica replica, Conflict ignoreConflict)
+ public RangesAtEndpoint.Builder add(Replica replica, Conflict ignoreConflict)
{
- if (hasSnapshot) throw new IllegalStateException();
+ if (built) throw new IllegalStateException();
Preconditions.checkNotNull(replica);
if (!Objects.equals(super.endpoint, replica.endpoint()))
throw new IllegalArgumentException("Replica " + replica + " has incorrect endpoint (expected " + super.endpoint + ")");
- Replica prev = super.byRange.put(replica.range(), replica);
- if (prev != null)
+ if (!super.byRange.internalPutIfAbsent(replica, list.size()))
{
- super.byRange.put(replica.range(), prev); // restore prev
switch (ignoreConflict)
{
case DUPLICATE:
- if (prev.equals(replica))
+ if (byRange().get(replica.range()).equals(replica))
break;
case NONE:
- throw new IllegalArgumentException("Conflicting replica added (expected unique ranges): " + replica + "; existing: " + prev);
+ throw new IllegalArgumentException("Conflicting replica added (expected unique ranges): "
+ + replica + "; existing: " + byRange().get(replica.range()));
case ALL:
}
- return;
+ return this;
}
list.add(replica);
+ return this;
}
@Override
- public Map<Range<Token>, Replica> byRange()
+ public RangesAtEndpoint snapshot()
{
- // our internal map is modifiable, but it is unsafe to modify the map externally
- // it would be possible to implement a safe modifiable map, but it is probably not valuable
- return Collections.unmodifiableMap(super.byRange());
+ return snapshot(list.subList(0, list.size()));
}
- public RangesAtEndpoint get(boolean isSnapshot)
+ public RangesAtEndpoint build()
{
- return new RangesAtEndpoint(super.endpoint, super.list, isSnapshot, Collections.unmodifiableMap(super.byRange));
- }
-
- public RangesAtEndpoint asImmutableView()
- {
- return get(false);
- }
-
- public RangesAtEndpoint asSnapshot()
- {
- hasSnapshot = true;
- return get(true);
+ built = true;
+ return new RangesAtEndpoint(super.endpoint, super.list, super.byRange);
}
}
- public static class Builder extends ReplicaCollection.Builder<RangesAtEndpoint, Mutable, RangesAtEndpoint.Builder>
+ public static Builder builder(InetAddressAndPort endpoint)
{
- public Builder(InetAddressAndPort endpoint) { this(endpoint, 0); }
- public Builder(InetAddressAndPort endpoint, int capacity) { super(new Mutable(endpoint, capacity)); }
+ return new Builder(endpoint);
}
-
- public static RangesAtEndpoint.Builder builder(InetAddressAndPort endpoint)
- {
- return new RangesAtEndpoint.Builder(endpoint);
- }
-
- public static RangesAtEndpoint.Builder builder(InetAddressAndPort endpoint, int capacity)
+ public static Builder builder(InetAddressAndPort endpoint, int capacity)
{
- return new RangesAtEndpoint.Builder(endpoint, capacity);
+ return new Builder(endpoint, capacity);
}
public static RangesAtEndpoint empty(InetAddressAndPort endpoint)
{
- return new RangesAtEndpoint(endpoint, EMPTY_LIST, true, EMPTY_MAP);
+ return new RangesAtEndpoint(endpoint, EMPTY_LIST, EMPTY_MAP);
}
public static RangesAtEndpoint of(Replica replica)
{
- ArrayList<Replica> one = new ArrayList<>(1);
+ ReplicaList one = new ReplicaList(1);
one.add(replica);
- return new RangesAtEndpoint(replica.endpoint(), one, true, Collections.unmodifiableMap(Collections.singletonMap(replica.range(), replica)));
+ return new RangesAtEndpoint(replica.endpoint(), one, rangeMap(one));
}
public static RangesAtEndpoint of(Replica ... replicas)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org