diff --git a/modules/accord b/modules/accord index 5ffe3d504b..d99ad84cc4 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 5ffe3d504bb5aa1ff1c2b96d817791e40f7ced0f +Subproject commit d99ad84cc49a96299a9ae55183e38ee6f1aa3f47 diff --git a/src/java/org/apache/cassandra/locator/AbstractNetworkTopologySnitch.java b/src/java/org/apache/cassandra/locator/AbstractNetworkTopologySnitch.java index b6901e2766..ff0842fdf5 100644 --- a/src/java/org/apache/cassandra/locator/AbstractNetworkTopologySnitch.java +++ b/src/java/org/apache/cassandra/locator/AbstractNetworkTopologySnitch.java @@ -17,6 +17,10 @@ */ package org.apache.cassandra.locator; +import java.util.Comparator; + +import org.apache.cassandra.utils.Sortable; + /** * An endpoint snitch tells Cassandra information about network topology that it can use to route * requests more efficiently. @@ -31,4 +35,16 @@ public abstract class AbstractNetworkTopologySnitch extends AbstractEndpointSnit { return proximity.compareEndpoints(address, r1, r2); } + + @Override + public boolean supportCompareByEndpoint() + { + return proximity.supportCompareByEndpoint(); + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + return proximity.endpointComparator(address, addresses); + } } diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 1d3810613f..003347ef29 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -42,6 +42,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MBeanWrapper; +import org.apache.cassandra.utils.Sortable; import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORE_DYNAMIC_SNITCH_SEVERITY; @@ -170,10 +171,13 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers. // TODO: avoid copy replicas = delegate.sortedByProximity(address, replicas); - HashMap scores = this.scores; // Make sure the score don't change in the middle of the loop below - // (which wouldn't really matter here but its cleaner that way). - ArrayList subsnitchOrderedScores = new ArrayList<>(replicas.size()); - for (Replica replica : replicas) + return shouldSortByScore(scores, replicas) ? sortedByProximityWithScore(address, replicas) : replicas; + } + + private > boolean shouldSortByScore(HashMap scores, C sortedReplicas) + { + ArrayList subsnitchOrderedScores = new ArrayList<>(sortedReplicas.size()); + for (Endpoint replica : sortedReplicas) { Double score = scores.get(replica.endpoint()); if (score == null) @@ -193,12 +197,10 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers. for (Double subsnitchScore : subsnitchOrderedScores) { if (subsnitchScore > (sortedScoreIterator.next() * badnessThreshold)) - { - return sortedByProximityWithScore(address, replicas); - } + return true; } - return replicas; + return false; } private static double defaultStore(InetAddressAndPort target) @@ -208,6 +210,11 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers. // Compare endpoints given an immutable snapshot of the scores private int compareEndpoints(InetAddressAndPort target, Replica a1, Replica a2, Map scores) + { + return compareEndpoints(a1, a2, scores, (a, b) -> delegate.compareEndpoints(target, a, b)); + } + + private int compareEndpoints(T a1, T a2, Map scores, Comparator subCompare) { Double scored1 = scores.get(a1.endpoint()); Double scored2 = scores.get(a2.endpoint()); @@ -223,7 +230,7 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers. } if (scored1.equals(scored2)) - return delegate.compareEndpoints(target, a1, a2); + return subCompare.compare(a1, a2); if (scored1 < scored2) return -1; else @@ -409,4 +416,26 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers. } return maxScore; } + + @Override + public boolean supportCompareByEndpoint() + { + return delegate.supportCompareByEndpoint(); + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + if (!delegate.supportCompareByEndpoint()) + throw new UnsupportedOperationException(); + assert address.equals(FBUtilities.getBroadcastAddressAndPort()); // we only know about ourself + Comparator compare = delegate.endpointComparator(address, addresses); + if (addresses.size() < 2) + return compare; + HashMap scores = this.scores; + Comparator compareWithScore = (r1, r2) -> compareEndpoints(r1, r2, scores, compare); + return dynamicBadnessThreshold == 0 || shouldSortByScore(scores, addresses.sorted(compare)) ? + compareWithScore : + compare; + } } diff --git a/src/java/org/apache/cassandra/locator/Endpoint.java b/src/java/org/apache/cassandra/locator/Endpoint.java new file mode 100644 index 0000000000..5a44bfd61c --- /dev/null +++ b/src/java/org/apache/cassandra/locator/Endpoint.java @@ -0,0 +1,24 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +public interface Endpoint +{ + InetAddressAndPort endpoint(); +} diff --git a/src/java/org/apache/cassandra/locator/IEndpointSnitch.java b/src/java/org/apache/cassandra/locator/IEndpointSnitch.java index 4d50336810..ef275ec12f 100644 --- a/src/java/org/apache/cassandra/locator/IEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/IEndpointSnitch.java @@ -18,9 +18,11 @@ package org.apache.cassandra.locator; import java.net.InetSocketAddress; +import java.util.Comparator; import java.util.Set; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Sortable; /** * This interface helps determine location of node in the datacenter relative to another node. @@ -101,5 +103,14 @@ public interface IEndpointSnitch { return false; } -} + default boolean supportCompareByEndpoint() + { + return false; + } + + default > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + throw new UnsupportedOperationException(); + } +} diff --git a/src/java/org/apache/cassandra/locator/NetworkTopologyProximity.java b/src/java/org/apache/cassandra/locator/NetworkTopologyProximity.java index eddcb36303..f7f30dafb7 100644 --- a/src/java/org/apache/cassandra/locator/NetworkTopologyProximity.java +++ b/src/java/org/apache/cassandra/locator/NetworkTopologyProximity.java @@ -20,10 +20,18 @@ package org.apache.cassandra.locator; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.tcm.membership.Location; +import org.apache.cassandra.utils.Sortable; + +import java.util.Comparator; public class NetworkTopologyProximity extends BaseProximity { public int compareEndpoints(InetAddressAndPort address, Replica r1, Replica r2) + { + return compareByEndpoints(address, r1, r2); + } + + public int compareByEndpoints(InetAddressAndPort address, Endpoint r1, Endpoint r2) { InetAddressAndPort a1 = r1.endpoint(); InetAddressAndPort a2 = r2.endpoint(); @@ -48,4 +56,18 @@ public class NetworkTopologyProximity extends BaseProximity return 1; return 0; } + + @Override + public boolean supportCompareByEndpoint() + { + return true; + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + if (!supportCompareByEndpoint()) + throw new UnsupportedOperationException(); + return (a, b) -> compareByEndpoints(address, a, b); + } } diff --git a/src/java/org/apache/cassandra/locator/NoOpProximity.java b/src/java/org/apache/cassandra/locator/NoOpProximity.java index 342f12e619..69f5250d70 100644 --- a/src/java/org/apache/cassandra/locator/NoOpProximity.java +++ b/src/java/org/apache/cassandra/locator/NoOpProximity.java @@ -18,6 +18,10 @@ package org.apache.cassandra.locator; +import org.apache.cassandra.utils.Sortable; + +import java.util.Comparator; + public class NoOpProximity extends BaseProximity { @Override @@ -34,4 +38,23 @@ public class NoOpProximity extends BaseProximity // Collections.sort is guaranteed to be stable) return 0; } + + @Override + public boolean supportCompareByEndpoint() + { + return true; + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + return this::compareByEndpoint; + } + + private int compareByEndpoint(Endpoint a, Endpoint b) + { + // Making all endpoints equal ensures we won't change the original ordering (since + // Collections.sort is guaranteed to be stable) + return 0; + } } diff --git a/src/java/org/apache/cassandra/locator/NodeProximity.java b/src/java/org/apache/cassandra/locator/NodeProximity.java index cbb7158aaf..af7c17eaac 100644 --- a/src/java/org/apache/cassandra/locator/NodeProximity.java +++ b/src/java/org/apache/cassandra/locator/NodeProximity.java @@ -18,6 +18,10 @@ package org.apache.cassandra.locator; +import org.apache.cassandra.utils.Sortable; + +import java.util.Comparator; + public interface NodeProximity { /** @@ -35,4 +39,14 @@ public interface NodeProximity * to be faster than 2 sequential queries, one against l1 followed by one against l2. */ public boolean isWorthMergingForRangeQuery(ReplicaCollection merged, ReplicaCollection l1, ReplicaCollection l2); + + default boolean supportCompareByEndpoint() + { + return false; + } + + default > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + throw new UnsupportedOperationException(); + } } diff --git a/src/java/org/apache/cassandra/locator/Replica.java b/src/java/org/apache/cassandra/locator/Replica.java index b1f68b2101..41a16a459b 100644 --- a/src/java/org/apache/cassandra/locator/Replica.java +++ b/src/java/org/apache/cassandra/locator/Replica.java @@ -51,7 +51,7 @@ import static org.apache.cassandra.dht.AbstractBounds.tokenSerializer; * and such and what the result is WRT to transientness. Definitely avoid creating fake Replicas with misinformation * about endpoints, ranges, or transientness. */ -public final class Replica implements Comparable +public final class Replica implements Comparable, Endpoint { public static final IPartitionerDependentSerializer serializer = new Serializer(); @@ -105,6 +105,7 @@ public final class Replica implements Comparable return (full ? "Full" : "Transient") + '(' + endpoint() + ',' + range + ')'; } + @Override public final InetAddressAndPort endpoint() { return endpoint; diff --git a/src/java/org/apache/cassandra/locator/ReplicaCollection.java b/src/java/org/apache/cassandra/locator/ReplicaCollection.java index b679b506b0..f1dac0042b 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaCollection.java +++ b/src/java/org/apache/cassandra/locator/ReplicaCollection.java @@ -24,11 +24,13 @@ import java.util.Set; import java.util.function.Predicate; import java.util.stream.Stream; +import org.apache.cassandra.utils.Sortable; + /** * A collection like class for Replica objects. Represents both a well defined order on the contained Replica objects, * and efficient methods for accessing the contained Replicas, directly and as a projection onto their endpoints and ranges. */ -public interface ReplicaCollection> extends Iterable +public interface ReplicaCollection> extends Sortable { /** * @return a Set of the endpoints of the contained Replicas. diff --git a/src/java/org/apache/cassandra/locator/SimpleSnitch.java b/src/java/org/apache/cassandra/locator/SimpleSnitch.java index e06316fa26..953f2e5c89 100644 --- a/src/java/org/apache/cassandra/locator/SimpleSnitch.java +++ b/src/java/org/apache/cassandra/locator/SimpleSnitch.java @@ -17,6 +17,10 @@ */ package org.apache.cassandra.locator; +import java.util.Comparator; + +import org.apache.cassandra.utils.Sortable; + /** * A simple endpoint snitch implementation that treats Strategy order as proximity, * allowing non-read-repaired reads to prefer a single endpoint, which improves @@ -58,4 +62,16 @@ public class SimpleSnitch extends AbstractEndpointSnitch { return sorter.isWorthMergingForRangeQuery(merged, l1, l2); } + + @Override + public boolean supportCompareByEndpoint() + { + return sorter.supportCompareByEndpoint(); + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + return sorter.endpointComparator(address, addresses); + } } diff --git a/src/java/org/apache/cassandra/locator/SnitchAdapter.java b/src/java/org/apache/cassandra/locator/SnitchAdapter.java index 1a32dd6497..b90e6fed16 100644 --- a/src/java/org/apache/cassandra/locator/SnitchAdapter.java +++ b/src/java/org/apache/cassandra/locator/SnitchAdapter.java @@ -18,12 +18,14 @@ package org.apache.cassandra.locator; +import java.util.Comparator; import java.util.HashSet; import java.util.Set; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.membership.Location; +import org.apache.cassandra.utils.Sortable; public class SnitchAdapter implements InitialLocationProvider, NodeProximity, NodeAddressConfig { @@ -81,4 +83,16 @@ public class SnitchAdapter implements InitialLocationProvider, NodeProximity, No { return snitch.preferLocalConnections(); } + + @Override + public boolean supportCompareByEndpoint() + { + return snitch.supportCompareByEndpoint(); + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + return snitch.endpointComparator(address, addresses); + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 23554c7feb..a323edaf17 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -613,7 +613,7 @@ public class AccordJournal implements Shutdownable static { // make noise early if we forget to update our version mappings - Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_50); + Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_50, "Expected current version to be %d but given %d", MessagingService.VERSION_50, MessagingService.current_version); } private static int msVersion(int version) diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 2a4ed5a2a7..5de4daa233 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -71,6 +71,8 @@ import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification; import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.api.AccordRoutingKey.KeyspaceSplitter; import org.apache.cassandra.service.accord.api.AccordScheduler; +import org.apache.cassandra.service.accord.api.AccordTopologySorter; +import org.apache.cassandra.service.accord.api.CompositeTopologySorter; import org.apache.cassandra.service.accord.exceptions.ReadPreemptedException; import org.apache.cassandra.service.accord.exceptions.WritePreemptedException; import org.apache.cassandra.service.accord.txn.TxnData; @@ -138,7 +140,17 @@ public class AccordService implements IAccordService, Shutdownable } @Override - public void startup() {} + public void startup() + { + try + { + AccordTopologySorter.checkSnitchSupported(DatabaseDescriptor.getNodeProximity()); + } + catch (Throwable t) + { + logger.warn("Current snitch is not compatable with Accord, make sure to fix the snitch before enabling Accord; {}", t.toString()); + } + } @Override public void shutdownAndWait(long timeout, TimeUnit unit) { } @@ -241,7 +253,8 @@ public class AccordService implements IAccordService, Shutdownable agent, new DefaultRandom(), scheduler, - SizeOfIntersectionSorter.SUPPLIER, + CompositeTopologySorter.create(SizeOfIntersectionSorter.SUPPLIER, + new AccordTopologySorter.Supplier(configService, DatabaseDescriptor.getNodeProximity())), SimpleProgressLog::new, AccordCommandStores.factory(journal), configuration); diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java b/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java index d385cf5d31..88a0d18b7f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java @@ -35,6 +35,7 @@ import accord.topology.Topology; import accord.utils.Invariants; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.EndpointsForRange; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.schema.DistributedSchema; import org.apache.cassandra.schema.KeyspaceMetadata; @@ -47,7 +48,6 @@ import org.apache.cassandra.tcm.membership.Directory; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.ownership.DataPlacement; import org.apache.cassandra.tcm.ownership.DataPlacements; -import org.apache.cassandra.tcm.ownership.VersionedEndpoints; public class AccordTopologyUtils { @@ -56,7 +56,7 @@ public class AccordTopologyUtils return new Node.Id(nodeId.id()); } - private static Shard createShard(TokenRange range, Directory directory, VersionedEndpoints.ForRange reads, VersionedEndpoints.ForRange writes) + private static Shard createShard(TokenRange range, Directory directory, EndpointsForRange reads, EndpointsForRange writes) { Function endpointMapper = e -> { NodeId tcmId = directory.peerId(e); @@ -106,8 +106,10 @@ public class AccordTopologyUtils List shards = new ArrayList<>(ranges.size()); for (Range range : ranges) { - VersionedEndpoints.ForRange reads = placement.reads.forRange(range); - VersionedEndpoints.ForRange writes = placement.reads.forRange(range); + // TODO (consider, low priority): flesh out how Accord and Transient Replicas work together + // Accord needs to be able to read the full data from a single replica, but with transient ones they may only have a hash. + EndpointsForRange reads = placement.reads.forRange(range).get().filter(r -> r.isFull()); + EndpointsForRange writes = placement.writes.forRange(range).get().filter(r -> r.isFull()); // TCM doesn't create wrap around ranges Invariants.checkArgument(!range.isWrapAround() || range.right.equals(range.right.minValue()), diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordTopologySorter.java b/src/java/org/apache/cassandra/service/accord/api/AccordTopologySorter.java new file mode 100644 index 0000000000..4f9eff850b --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/api/AccordTopologySorter.java @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service.accord.api; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.Set; + +import accord.api.TopologySorter; +import accord.local.Node; +import accord.topology.ShardSelection; +import accord.topology.Topologies; +import accord.topology.Topology; +import org.apache.cassandra.locator.*; +import org.apache.cassandra.service.accord.AccordEndpointMapper; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Sortable; + +public class AccordTopologySorter implements TopologySorter +{ + public static class Supplier implements TopologySorter.Supplier + { + private final AccordEndpointMapper mapper; + private final NodeProximity proximity; + + public Supplier(AccordEndpointMapper mapper, NodeProximity proximity) + { + checkSnitchSupported(proximity); + this.mapper = mapper; + this.proximity = proximity; + } + + @Override + public TopologySorter get(Topology topologies) + { + return create(topologies.nodes()); + } + + @Override + public TopologySorter get(Topologies topologies) + { + return create(topologies.nodes()); + } + + private AccordTopologySorter create(Set nodes) + { + SortableEndpoints endpoints = SortableEndpoints.from(nodes, mapper); + Comparator comparator = proximity.endpointComparator(FBUtilities.getBroadcastAddressAndPort(), endpoints); + return new AccordTopologySorter(mapper, comparator); + } + } + + private final AccordEndpointMapper mapper; + + private final Comparator comparator; + private AccordTopologySorter(AccordEndpointMapper mapper, Comparator comparator) + { + this.mapper = mapper; + this.comparator = comparator; + } + + public static void checkSnitchSupported(NodeProximity proximity) + { + if (!proximity.supportCompareByEndpoint()) + { + if (proximity instanceof DynamicEndpointSnitch) + proximity = ((DynamicEndpointSnitch) proximity).delegate; + throw new IllegalArgumentException("Unsupported snitch " + proximity.getClass() + "; supportCompareByEndpoint returned false"); + } + } + + @Override + public int compare(Node.Id node1, Node.Id node2, ShardSelection shards) + { + return comparator.compare(() -> mapper.mappedEndpoint(node1), () -> mapper.mappedEndpoint(node2)); + } + + private static class EndpointTuple implements Endpoint + { + final InetAddressAndPort endpoint; + + private EndpointTuple(InetAddressAndPort endpoint) + { + this.endpoint = endpoint; + } + + @Override + public InetAddressAndPort endpoint() + { + return endpoint; + } + } + + private static class SortableEndpoints extends ArrayList implements Sortable + { + public SortableEndpoints(int initialCapacity) + { + super(initialCapacity); + } + + public SortableEndpoints sorted(Comparator comparator) + { + sort(comparator); + return this; + } + + static SortableEndpoints from(Set nodes, AccordEndpointMapper mapper) + { + SortableEndpoints result = new SortableEndpoints(nodes.size()); + nodes.forEach(id -> result.add(new EndpointTuple(mapper.mappedEndpoint(id)))); + return result; + } + } +} diff --git a/src/java/org/apache/cassandra/service/accord/api/CompositeTopologySorter.java b/src/java/org/apache/cassandra/service/accord/api/CompositeTopologySorter.java new file mode 100644 index 0000000000..3886cde12d --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/api/CompositeTopologySorter.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service.accord.api; + +import accord.api.TopologySorter; +import accord.local.Node; +import accord.topology.ShardSelection; +import accord.topology.Topologies; +import accord.topology.Topology; + +public class CompositeTopologySorter implements TopologySorter +{ + public static class Supplier implements TopologySorter.Supplier + { + private final TopologySorter.Supplier[] delegates; + + private Supplier(TopologySorter.Supplier[] delegates) + { + this.delegates = delegates; + } + + @Override + public TopologySorter get(Topology topologies) + { + TopologySorter[] sorters = new TopologySorter[delegates.length]; + for (int i = 0; i < sorters.length; i++) + sorters[i] = delegates[i].get(topologies); + return new CompositeTopologySorter(sorters); + } + + @Override + public TopologySorter get(Topologies topologies) + { + TopologySorter[] sorters = new TopologySorter[delegates.length]; + for (int i = 0; i < sorters.length; i++) + sorters[i] = delegates[i].get(topologies); + return new CompositeTopologySorter(sorters); + } + } + + private final TopologySorter[] delegates; + + private CompositeTopologySorter(TopologySorter[] delegates) + { + this.delegates = delegates; + } + + public static TopologySorter.Supplier create(TopologySorter.Supplier... delegates) + { + switch (delegates.length) + { + case 0: throw new IllegalArgumentException("Can not create an empty sorter"); + case 1: return delegates[0]; + default: return new CompositeTopologySorter.Supplier(delegates); + } + } + + @Override + public int compare(Node.Id node1, Node.Id node2, ShardSelection shards) + { + for (int i = 0; i < delegates.length; i++) + { + int rc = delegates[i].compare(node1, node2, shards); + if (rc != 0) return rc; + } + return 0; + } +} diff --git a/src/java/org/apache/cassandra/utils/Sortable.java b/src/java/org/apache/cassandra/utils/Sortable.java new file mode 100644 index 0000000000..145967cb24 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/Sortable.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.utils; + +import java.util.Comparator; + +public interface Sortable> extends Iterable +{ + int size(); + + S sorted(Comparator comparator); +} diff --git a/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java index 7f7cab1110..78e04454fa 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java @@ -63,7 +63,7 @@ class AccordClusterSimulation extends ClusterSimulation impleme AccordClusterSimulation(RandomSource random, long seed, int uniqueNum, Builder builder) throws IOException { super(random, seed, uniqueNum, builder, - config -> {}, + config -> config.set("storage_compatibility_mode", "NONE"), (simulated, schedulers, cluster, options) -> { int[] primaryKeys = primaryKeys(seed, builder.primaryKeyCount()); KindOfSequence.Period jitter = RandomSource.Choices.uniform(KindOfSequence.values()).choose(random) diff --git a/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosClusterSimulation.java index 03d7e61e7f..a0c6682211 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosClusterSimulation.java @@ -79,6 +79,7 @@ class PaxosClusterSimulation extends ClusterSimulation implemen .set("paxos_cache_size", (builder.stateCache != null ? builder.stateCache : random.uniformFloat() < 0.5) ? null : "0MiB") .set("paxos_state_purging", "repaired") .set("paxos_on_linearizability_violations", "log") + .set("storage_compatibility_mode", "NONE") , (simulated, schedulers, cluster, options) -> { int[] primaryKeys = primaryKeys(seed, builder.primaryKeyCount()); diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedSnitch.java b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedSnitch.java index 55fe73c301..7692e0ae5c 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedSnitch.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedSnitch.java @@ -28,16 +28,49 @@ import java.util.stream.IntStream; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.IInstanceConfig; -import org.apache.cassandra.locator.IEndpointSnitch; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.locator.Replica; -import org.apache.cassandra.locator.ReplicaCollection; +import org.apache.cassandra.locator.*; import org.apache.cassandra.simulator.cluster.NodeLookup; +import org.apache.cassandra.utils.Sortable; public class SimulatedSnitch extends NodeLookup { + private static class SimulatedProximity implements NodeProximity + { + @Override + public > C sortedByProximity(InetAddressAndPort address, C addresses) + { + return addresses.sorted(Comparator.comparingInt(SimulatedSnitch::asInt)); + } + + @Override + public int compareEndpoints(InetAddressAndPort target, Replica r1, Replica r2) + { + return Comparator.comparingInt(SimulatedSnitch::asInt).compare(r1, r2); + } + + @Override + public boolean isWorthMergingForRangeQuery(ReplicaCollection merged, ReplicaCollection l1, ReplicaCollection l2) + { + return false; + } + + @Override + public boolean supportCompareByEndpoint() + { + return true; + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + return Comparator.comparingInt(SimulatedSnitch::asInt); + } + } + public static class Instance implements IEndpointSnitch { + private final NodeProximity proximity = new SimulatedProximity(); + private static volatile Function LOOKUP_DC; public String getRack(InetAddressAndPort endpoint) @@ -52,12 +85,12 @@ public class SimulatedSnitch extends NodeLookup public > C sortedByProximity(InetAddressAndPort address, C addresses) { - return addresses.sorted(Comparator.comparingInt(SimulatedSnitch::asInt)); + return proximity.sortedByProximity(address, addresses); } public int compareEndpoints(InetAddressAndPort target, Replica r1, Replica r2) { - return Comparator.comparingInt(SimulatedSnitch::asInt).compare(r1, r2); + return proximity.compareEndpoints(target, r1, r2); } public void gossiperStarting() @@ -66,7 +99,7 @@ public class SimulatedSnitch extends NodeLookup public boolean isWorthMergingForRangeQuery(ReplicaCollection merged, ReplicaCollection l1, ReplicaCollection l2) { - return false; + return proximity.isWorthMergingForRangeQuery(merged, l1, l2); } public static void setup(Function lookupDc) @@ -127,7 +160,7 @@ public class SimulatedSnitch extends NodeLookup return Arrays.asList(nameOfDcs); } - private static int asInt(Replica address) + private static int asInt(Endpoint address) { byte[] bytes = address.endpoint().addressBytes; return bytes[0] | (bytes[1] << 8) | (bytes[2] << 16) | (bytes[3] << 24); diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java index 79b9efaa89..130053658e 100644 --- a/test/unit/org/apache/cassandra/ServerTestUtils.java +++ b/test/unit/org/apache/cassandra/ServerTestUtils.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.net.UnknownHostException; import java.util.Arrays; import java.util.Collections; +import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -44,6 +45,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.format.big.BigTableReader; import org.apache.cassandra.io.sstable.indexsummary.IndexSummarySupport; import org.apache.cassandra.io.util.File; +import org.apache.cassandra.locator.Endpoint; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.locator.BaseProximity; @@ -70,6 +72,7 @@ import org.apache.cassandra.tcm.transformations.Register; import org.apache.cassandra.tcm.transformations.UnsafeJoin; import org.apache.cassandra.tcm.transformations.cms.Initialize; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Sortable; import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION; @@ -108,6 +111,18 @@ public final class ServerTestUtils { return 0; } + + @Override + public boolean supportCompareByEndpoint() + { + return true; + } + + @Override + public > Comparator endpointComparator(InetAddressAndPort address, C addresses) + { + return (a, b) -> 0; + } }); } diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index a3f9d2325f..9ab3dab6bf 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -272,6 +272,7 @@ public class DatabaseDescriptorRefTest "org.apache.cassandra.io.util.PathUtils$IOToLongFunction", "org.apache.cassandra.io.util.RebufferingInputStream", "org.apache.cassandra.io.util.SpinningDiskOptimizationStrategy", + "org.apache.cassandra.locator.Endpoint", "org.apache.cassandra.locator.IEndpointSnitch", "org.apache.cassandra.locator.InetAddressAndPort", "org.apache.cassandra.locator.Locator", diff --git a/test/unit/org/apache/cassandra/locator/NodeProximityEndpointCompareTest.java b/test/unit/org/apache/cassandra/locator/NodeProximityEndpointCompareTest.java new file mode 100644 index 0000000000..4721046978 --- /dev/null +++ b/test/unit/org/apache/cassandra/locator/NodeProximityEndpointCompareTest.java @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Modifier; + +import org.assertj.core.api.Assertions; +import org.junit.Test; +import org.reflections.Reflections; +import org.reflections.scanners.Scanners; +import org.reflections.util.ConfigurationBuilder; + +import org.apache.cassandra.config.DatabaseDescriptor; + +public class NodeProximityEndpointCompareTest +{ + static + { + DatabaseDescriptor.clientInitialization(); + } + + @Test + public void allSupportEndpoint() throws InvocationTargetException, InstantiationException, IllegalAccessException + { + Reflections reflections = new Reflections(new ConfigurationBuilder() + .forPackage("org.apache.cassandra") + .setScanners(Scanners.SubTypes) + .setExpandSuperTypes(true)); + + for (Class klass : reflections.getSubTypesOf(NodeProximity.class)) + { + if (Modifier.isAbstract(klass.getModifiers()) + || Modifier.isPrivate(klass.getModifiers()) // private can not be created normally, so these are scoped to tests and can be ignored + || klass.isAnonymousClass()) + continue; + Constructor declaredConstructor; + try + { + declaredConstructor = klass.getDeclaredConstructor(); + } + catch (NoSuchMethodException e) + { + // DynamicEndpointSnitch or test snitch... we can not create this normally + continue; + } + if (Modifier.isPrivate(declaredConstructor.getModifiers())) + continue; + NodeProximity proximity = declaredConstructor.newInstance(); + Assertions.assertThat(proximity.supportCompareByEndpoint()) + .describedAs("Snitch %s does not support compare by endpoint!", proximity.getClass()) + .isTrue(); + } + } +} \ No newline at end of file