diff --git a/.build/checkstyle.xml b/.build/checkstyle.xml index 7e91df304c..99faf7335e 100644 --- a/.build/checkstyle.xml +++ b/.build/checkstyle.xml @@ -58,15 +58,9 @@ - - - - - - - - - + + + diff --git a/src/java/org/apache/cassandra/concurrent/ExecutorLocals.java b/src/java/org/apache/cassandra/concurrent/ExecutorLocals.java index a5720dd37c..55c6ee2464 100644 --- a/src/java/org/apache/cassandra/concurrent/ExecutorLocals.java +++ b/src/java/org/apache/cassandra/concurrent/ExecutorLocals.java @@ -33,7 +33,7 @@ import io.netty.util.concurrent.FastThreadLocal; */ public class ExecutorLocals implements WithResources, Closeable { - private static final ExecutorLocals none = new ExecutorLocals(null, null); + private static final ExecutorLocals none = new ExecutorLocals(null, null, false); private static final FastThreadLocal locals = new FastThreadLocal() { @Override @@ -45,20 +45,23 @@ public class ExecutorLocals implements WithResources, Closeable public static class Impl { - protected static void set(TraceState traceState, ClientWarn.State clientWarnState) + @SuppressWarnings("resource") + protected static void set(TraceState traceState, ClientWarn.State clientWarnState, boolean eligibleForArtificialLatency) { - if (traceState == null && clientWarnState == null) locals.set(none); - else locals.set(new ExecutorLocals(traceState, clientWarnState)); + if (traceState == null && clientWarnState == null && !eligibleForArtificialLatency) locals.set(none); + else locals.set(new ExecutorLocals(traceState, clientWarnState, eligibleForArtificialLatency)); } } public final TraceState traceState; public final ClientWarn.State clientWarnState; + public final boolean eligibleForArtificialLatency; - protected ExecutorLocals(TraceState traceState, ClientWarn.State clientWarnState) + protected ExecutorLocals(TraceState traceState, ClientWarn.State clientWarnState, boolean eligibleForArtificialLatency) { this.traceState = traceState; this.clientWarnState = clientWarnState; + this.eligibleForArtificialLatency = eligibleForArtificialLatency; } /** @@ -82,7 +85,7 @@ public class ExecutorLocals implements WithResources, Closeable public static ExecutorLocals create(TraceState traceState) { ExecutorLocals current = locals.get(); - return current.traceState == traceState ? current : new ExecutorLocals(traceState, current.clientWarnState); + return current.traceState == traceState ? current : new ExecutorLocals(traceState, current.clientWarnState, current.eligibleForArtificialLatency); } public static void clear() diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index 9965a52e59..4cfaf562d8 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -195,6 +195,7 @@ public class AccordSpec * default transactional mode for tables created by this node when no transactional mode has been specified in the DDL */ public TransactionalMode default_transactional_mode = TransactionalMode.off; + public boolean ephemeralReadEnabled = true; public boolean state_cache_listener_jfr_enabled = false; diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index c72577e7ec..3ac0dc1f0b 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -62,6 +62,10 @@ public enum CassandraRelevantProperties ALLOW_UNSAFE_REPLACE("cassandra.allow_unsafe_replace"), ALLOW_UNSAFE_TRANSIENT_CHANGES("cassandra.allow_unsafe_transient_changes"), APPROXIMATE_TIME_PRECISION_MS("cassandra.approximate_time_precision_ms", "2"), + ARTIFICIAL_LATENCIES("cassandra.artificial_latencies"), + ARTIFICIAL_LATENCIES_UNSAFE("cassandra.artificial_latencies_unsafe"), + ARTIFICIAL_LATENCY_LIMIT("cassandra.artificial_latency_limit", "200ms"), + ARTIFICIAL_LATENCY_VERBS("cassandra.artificial_latency_verbs"), ASYNC_PROFILER_ENABLED("cassandra.async_profiler.enabled", "false"), ASYNC_PROFILER_UNSAFE_MODE("cassandra.async_profiler.unsafe_mode", "false"), /** 2 ** GENSALT_LOG2_ROUNDS rounds of hashing will be performed. */ diff --git a/src/java/org/apache/cassandra/cql3/QueryOptions.java b/src/java/org/apache/cassandra/cql3/QueryOptions.java index b7003388af..3244497914 100644 --- a/src/java/org/apache/cassandra/cql3/QueryOptions.java +++ b/src/java/org/apache/cassandra/cql3/QueryOptions.java @@ -116,7 +116,7 @@ public abstract class QueryOptions values, ByteArrayUtil.EMPTY_ARRAY_OF_BYTE_ARRAYS, skipMetadata, - new SpecificOptions(pageSize, pagingState, serialConsistency, timestamp, keyspace, nowInSeconds), + new SpecificOptions(pageSize, pagingState, serialConsistency, timestamp, keyspace, nowInSeconds, false), version); } @@ -264,6 +264,11 @@ public abstract class QueryOptions return nowInSeconds != UNSET_NOWINSEC ? nowInSeconds : state.getNowInSeconds(); } + public boolean isEligibleForArtificialLatency() + { + return getSpecificOptions().eligibleForArtificialLatency; + } + /** The keyspace that this query is bound to, or null if not relevant. */ public String getKeyspace() { return getSpecificOptions().keyspace; } @@ -631,7 +636,7 @@ public abstract class QueryOptions // Options that are likely to not be present in most queries static class SpecificOptions { - private static final SpecificOptions DEFAULT = new SpecificOptions(-1, null, null, Long.MIN_VALUE, null, UNSET_NOWINSEC); + private static final SpecificOptions DEFAULT = new SpecificOptions(-1, null, null, Long.MIN_VALUE, null, UNSET_NOWINSEC, false); private final int pageSize; private final PagingState state; @@ -639,13 +644,15 @@ public abstract class QueryOptions private final long timestamp; private final String keyspace; private final long nowInSeconds; + private final boolean eligibleForArtificialLatency; private SpecificOptions(int pageSize, PagingState state, ConsistencyLevel serialConsistency, long timestamp, String keyspace, - long nowInSeconds) + long nowInSeconds, + boolean eligibleForArtificialLatency) { this.pageSize = pageSize; this.state = state; @@ -653,11 +660,12 @@ public abstract class QueryOptions this.timestamp = timestamp; this.keyspace = keyspace; this.nowInSeconds = nowInSeconds; + this.eligibleForArtificialLatency = eligibleForArtificialLatency; } public SpecificOptions withNowInSec(long nowInSec) { - return new SpecificOptions(pageSize, state, serialConsistency, timestamp, keyspace, nowInSec); + return new SpecificOptions(pageSize, state, serialConsistency, timestamp, keyspace, nowInSec, eligibleForArtificialLatency); } } @@ -674,7 +682,9 @@ public abstract class QueryOptions TIMESTAMP, NAMES_FOR_VALUES, KEYSPACE, - NOW_IN_SECONDS; + NOW_IN_SECONDS, + ELIGIBLE_FOR_ARTIFICIAL_LATENCY, + ; private final int mask; @@ -755,7 +765,8 @@ public abstract class QueryOptions String keyspace = Flag.contains(flags, Flag.KEYSPACE) ? CBUtil.readString(body) : null; long nowInSeconds = Flag.contains(flags, Flag.NOW_IN_SECONDS) ? CassandraUInt.toLong(body.readInt()) : UNSET_NOWINSEC; - options = new SpecificOptions(pageSize, pagingState, serialConsistency, timestamp, keyspace, nowInSeconds); + boolean eligibleForArtificialLatency = Flag.contains(flags, Flag.ELIGIBLE_FOR_ARTIFICIAL_LATENCY); + options = new SpecificOptions(pageSize, pagingState, serialConsistency, timestamp, keyspace, nowInSeconds, eligibleForArtificialLatency); } DefaultQueryOptions opts = new DefaultQueryOptions(consistency, null, values, skipMetadata, options, version); diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index 2a8c30d4c9..086867933b 100644 --- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java +++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java @@ -45,7 +45,12 @@ public enum ConsistencyLevel SERIAL (8), LOCAL_SERIAL(9, true), LOCAL_ONE (10, true), - NODE_LOCAL (11, true); + NODE_LOCAL (11, true), + UNSAFE_DELAY_QUORUM(99, false), + UNSAFE_DELAY_SERIAL(100, false), + UNSAFE_DELAY_LOCAL_QUORUM(101, true), + UNSAFE_DELAY_LOCAL_SERIAL(102, true); + // Used by the binary protocol public final int code; @@ -143,12 +148,16 @@ public enum ConsistencyLevel return 2; case THREE: return 3; + case UNSAFE_DELAY_QUORUM: case QUORUM: + case UNSAFE_DELAY_SERIAL: case SERIAL: return quorumFor(replicationStrategy); case ALL: return replicationStrategy.getReplicationFactor().allReplicas; + case UNSAFE_DELAY_LOCAL_QUORUM: case LOCAL_QUORUM: + case UNSAFE_DELAY_LOCAL_SERIAL: case LOCAL_SERIAL: return localQuorumForOurDc(replicationStrategy); case EACH_QUORUM: @@ -178,13 +187,16 @@ public enum ConsistencyLevel { case ANY: break; - case LOCAL_ONE: case LOCAL_QUORUM: case LOCAL_SERIAL: + case LOCAL_ONE: + case UNSAFE_DELAY_LOCAL_QUORUM: case LOCAL_QUORUM: + case UNSAFE_DELAY_LOCAL_SERIAL: case LOCAL_SERIAL: // we will only count local replicas towards our response count, as these queries only care about local guarantees blockFor += pending.count(InOurDc.replicas()); break; case ONE: case TWO: case THREE: - case QUORUM: case EACH_QUORUM: - case SERIAL: + case UNSAFE_DELAY_QUORUM: case QUORUM: + case UNSAFE_DELAY_SERIAL: case SERIAL: + case EACH_QUORUM: case ALL: blockFor += pending.size(); } @@ -219,7 +231,9 @@ public enum ConsistencyLevel switch (this) { case SERIAL: + case UNSAFE_DELAY_SERIAL: case LOCAL_SERIAL: + case UNSAFE_DELAY_LOCAL_SERIAL: throw new InvalidRequestException("You must use conditional updates for serializable writes"); } } @@ -233,7 +247,9 @@ public enum ConsistencyLevel requireNetworkTopologyStrategy(replicationStrategy); break; case SERIAL: + case UNSAFE_DELAY_SERIAL: case LOCAL_SERIAL: + case UNSAFE_DELAY_LOCAL_SERIAL: throw new InvalidRequestException(this + " is not supported as conditional update commit consistency. Use ANY if you mean \"make sure it is accepted but I don't care how many replicas commit it for non-SERIAL reads\""); } } @@ -246,7 +262,16 @@ public enum ConsistencyLevel public boolean isSerialConsistency() { - return this == SERIAL || this == LOCAL_SERIAL; + switch (this) + { + case SERIAL: + case UNSAFE_DELAY_SERIAL: + case LOCAL_SERIAL: + case UNSAFE_DELAY_LOCAL_SERIAL: + return true; + default: + return false; + } } public void validateCounterForWrite(TableMetadata metadata) throws InvalidRequestException diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index d161fb315d..e88a6d1c27 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -925,12 +925,12 @@ public abstract class ReadCommand extends AbstractReadQuery */ public Message createMessage(boolean trackRepairedData, Dispatcher.RequestTime requestTime) { - List flags = new ArrayList<>(3); - flags.add(MessageFlag.CALL_BACK_ON_FAILURE); + int flags = MessageFlag.CALL_BACK_ON_FAILURE.addTo(0); + if (trackWarnings) - flags.add(MessageFlag.TRACK_WARNINGS); + flags = MessageFlag.TRACK_WARNINGS.addTo(flags); if (trackRepairedData) - flags.add(MessageFlag.TRACK_REPAIRED_DATA); + flags = MessageFlag.TRACK_REPAIRED_DATA.addTo(flags); return Message.outWithFlags(verb(), this, diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java index da52510bf3..61def76d1a 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java +++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java @@ -110,6 +110,7 @@ public class ReplicaPlans return true; case LOCAL_ONE: return countInOurDc(liveReplicas).hasAtleast(1, 1); + case UNSAFE_DELAY_LOCAL_QUORUM: case LOCAL_QUORUM: return countInOurDc(liveReplicas).hasAtleast(localQuorumForOurDc(replicationStrategy), 1); case EACH_QUORUM: @@ -157,6 +158,7 @@ public class ReplicaPlans throw UnavailableException.create(consistencyLevel, 1, blockForFullReplicas, localLive.allReplicas(), localLive.fullReplicas()); break; } + case UNSAFE_DELAY_LOCAL_QUORUM: case LOCAL_QUORUM: { Replicas.ReplicaCount localLive = countInOurDc(allLive); @@ -753,7 +755,7 @@ public class ReplicaPlans Replicas.temporaryAssertFull(liveAndDown.all()); // TODO CASSANDRA-14547 - if (consistencyForPaxos == ConsistencyLevel.LOCAL_SERIAL) + if (consistencyForPaxos.isDatacenterLocal()) { // TODO: we should cleanup our semantics here, as we're filtering ALL nodes to localDC which is unexpected for ReplicaPlan // Restrict natural and pending to node in the local DC only diff --git a/src/java/org/apache/cassandra/net/ArtificialLatency.java b/src/java/org/apache/cassandra/net/ArtificialLatency.java new file mode 100644 index 0000000000..e8c000d43e --- /dev/null +++ b/src/java/org/apache/cassandra/net/ArtificialLatency.java @@ -0,0 +1,397 @@ +/* + * 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.net; + +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumSet; +import java.util.PriorityQueue; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.concurrent.locks.LockSupport; +import java.util.function.ToLongFunction; +import java.util.stream.Collector; +import java.util.stream.Collectors; + +import org.agrona.collections.Object2LongHashMap; + +import org.apache.cassandra.concurrent.ExecutorLocals; +import org.apache.cassandra.concurrent.Interruptible; +import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.membership.Directory; +import org.apache.cassandra.tcm.membership.Location; +import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; + +import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; +import static org.apache.cassandra.concurrent.ExecutorFactory.SystemThreadTag.DAEMON; +import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.Interrupts.UNSYNCHRONIZED; +import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe.SAFE; +import static org.apache.cassandra.config.CassandraRelevantProperties.ARTIFICIAL_LATENCY_LIMIT; +import static org.apache.cassandra.net.MessagingService.instance; +import static org.apache.cassandra.utils.Clock.Global.nanoTime; + +/* + * Mechanism to delay the sending of messages to peers + */ +public class ArtificialLatency extends ExecutorLocals.Impl +{ + private static volatile Set artificialLatencyVerbs; + private static volatile boolean artificialLatencyOnlyPermittedConsistencyLevels = true; + private static volatile ToLongFunction artificialLatencyNanos; + private static String artificialLatencies; + + private static Sink running; + + static + { + setArtificialLatencyVerbs(CassandraRelevantProperties.ARTIFICIAL_LATENCY_VERBS.getString("")); + String latencies = CassandraRelevantProperties.ARTIFICIAL_LATENCIES.getString(); + String unsafeLatencies = CassandraRelevantProperties.ARTIFICIAL_LATENCIES_UNSAFE.getString(); + if (latencies != null) setArtificialLatencies(latencies); + else if (unsafeLatencies != null) unsafeSetArtificialLatencies(unsafeLatencies); + if (artificialLatencyNanos != null && !artificialLatencyVerbs.isEmpty()) + setEnabled(true); + } + + // ensure initialised + public static void touch() {} + + public static synchronized boolean isEnabled() + { + return running != null; + } + + public static synchronized void setEnabled(boolean enabled) + { + if (enabled) start(); + else stop(); + } + + public static synchronized void start() + { + if (running == null) + running = Sink.start(); + } + + public static synchronized void stop() + { + if (running != null) + { + running.stop(); + running = null; + } + } + + public static boolean isEligibleForArtificialLatency() + { + return ExecutorLocals.current().eligibleForArtificialLatency; + } + + public static void setEligibleForArtificialLatency(boolean eligibleForArtificialLatency) + { + ExecutorLocals current = ExecutorLocals.current(); + set(current.traceState, current.clientWarnState, eligibleForArtificialLatency); + } + + static class Sink implements OutboundSink.AsyncFilter, Interruptible.Task + { + static class Delayed implements Comparable + { + final Message message; + final InetAddressAndPort to; + final ConnectionType type; + final long deadline; + final OutboundSink.Sink sink; + + Delayed(Message message, InetAddressAndPort to, ConnectionType type, long deadline, OutboundSink.Sink sink) + { + this.message = message; + this.to = to; + this.type = type; + this.deadline = deadline; + this.sink = sink; + } + + @Override + public int compareTo(Delayed that) + { + return Long.compare(this.deadline, that.deadline); + } + } + + volatile boolean isShutdown; + + final ConcurrentLinkedQueue in = new ConcurrentLinkedQueue<>(); + // messages we have stashed in order to apply an artificial delay + // note that this queue is not ordered, so that if the artificial delay is modified + // it may not take effect until the difference between the two delays elapses + final PriorityQueue out = new PriorityQueue<>(); + final Interruptible executor = executorFactory().infiniteLoop("ArtificialLatency", this, SAFE, DAEMON, UNSYNCHRONIZED); + + volatile Thread waiting; + volatile long waitingUntil; + private static final AtomicLongFieldUpdater waitingUntilUpdater = AtomicLongFieldUpdater.newUpdater(Sink.class, "waitingUntil"); + + static Sink start() + { + Sink sink = new Sink(); + instance().outboundSink.add(sink); + return sink; + } + + void stop() + { + isShutdown = true; + artificialLatencyNanos = ignore -> 0; + instance().outboundSink.remove(this); + executor.shutdownNow(); + try + { + executor.awaitTermination(1, TimeUnit.DAYS); + } + catch (InterruptedException e) + { + throw new UncheckedInterruptedException(e); + } + } + + @Override + public void filter(Message message, InetAddressAndPort to, ConnectionType type, OutboundSink.Sink next) + { + if (artificialLatencyOnlyPermittedConsistencyLevels && !message.header.permitsArtificialLatency()) + { + next.accept(message, to, type); + return; + } + + if (!artificialLatencyVerbs.contains(message.verb())) + { + next.accept(message, to, type); + return; + } + + long addNanos = artificialLatencyNanos.applyAsLong(to); + if (addNanos <= 0) + { + next.accept(message, to, type); + return; + } + + long deadline = nanoTime() + addNanos; + Delayed delay = new Delayed(message, to, type, deadline, next); + in.add(delay); + + while (true) + { + long curWaitingUntil = waitingUntil; + if (deadline >= curWaitingUntil) + break; + + if (waitingUntilUpdater.compareAndSet(this, curWaitingUntil, Long.MIN_VALUE)) + { + Thread thread = waiting; + if (thread != null) + LockSupport.unpark(thread); + } + } + + if (isShutdown && in.remove(delay)) + next.accept(message, to, type); + } + + public void run(Interruptible.State state) throws InterruptedException + { + switch (state) + { + default: throw new IllegalStateException(); + case SHUTTING_DOWN: + { + drainIn(); + out.forEach(d -> instance().send(d.message, d.to, d.type)); + return; + } + case NORMAL: + { + waiting = Thread.currentThread(); + while (true) + { + long deadline; + while (true) + { + drainIn(); + deadline = out.isEmpty() ? Long.MAX_VALUE : out.peek().deadline; + if (waitingUntil == deadline) + break; + waitingUntil = deadline; + } + + long waitNanos = deadline - nanoTime(); + if (waitNanos <= 0) + break; + LockSupport.parkNanos(waitNanos); + } + } + case INTERRUPTED: + { + Delayed delayed; + long now = nanoTime(); + while (null != (delayed = out.peek()) && delayed.deadline <= now) + { + delayed.sink.accept(delayed.message, delayed.to, delayed.type); + out.poll(); + } + } + } + } + + private void drainIn() + { + for (Delayed delayed = in.poll(); delayed != null ; delayed = in.poll()) + out.add(delayed); + } + } + + public static String getArtificialLatencies() + { + return artificialLatencies; + } + + private static long parseNanos(String latency) + { + if (!latency.endsWith("ms")) + throw new IllegalArgumentException("Latency must be specified in terms of milliseconds (with 'ms' suffix)"); + + return TimeUnit.MILLISECONDS.toNanos(Long.parseLong(latency.substring(0, latency.length() - 2))); + } + + public static void setArtificialLatencies(String latencies) + { + setArtificialLatencies(latencies, parseNanos(ARTIFICIAL_LATENCY_LIMIT.getString())); + } + + public static void unsafeSetArtificialLatencies(String latencies) + { + setArtificialLatencies(latencies, Long.MAX_VALUE); + } + + private static synchronized void setArtificialLatencies(String latencies, long nanoLimit) + { + if (latencies.indexOf(',') < 0) + { + long nanos = parseNanos(latencies); + if (nanos >= nanoLimit) + throw new IllegalArgumentException("Artificial latency limit is " + nanoLimit + "ns; tried to set " + nanos + "ns"); + artificialLatencyNanos = ignore -> nanos; + } + else + { + String[] parse = latencies.split(","); + Object2LongHashMap dcLatencies = new Object2LongHashMap<>(-1L); + for (int i = 0 ; i < parse.length ; ++i) + { + String[] subparse = parse[i].split(":"); + String dc = subparse[0]; + long nanos = parseNanos(subparse[1]); + if (nanos >= nanoLimit) + throw new IllegalArgumentException("Artificial latency limit is " + nanoLimit + "ns; tried to set " + nanos + "ns"); + dcLatencies.put(dc, nanos); + } + artificialLatencyNanos = addr -> { + Directory directory = ClusterMetadata.current().directory; + NodeId nodeId = directory.peerId(addr); + if (nodeId == null) + return 0; + Location location = directory.location(nodeId); + if (location == null) + return 0; + return dcLatencies.getOrDefault(location.datacenter, 0L); + }; + } + artificialLatencies = latencies; + } + + public static String getArtificialLatencyVerbs() + { + return artificialLatencyVerbs.stream() + .map(Verb::toString) + .collect(Collectors.joining(",")); + } + + public static boolean getArtificialLatencyOnlyPermittedConsistencyLevels() + { + return artificialLatencyOnlyPermittedConsistencyLevels; + } + + public static void setArtificialLatencyVerbs(String commaDelimitedVerbs) + { + if (commaDelimitedVerbs.isEmpty()) + artificialLatencyVerbs = Collections.emptySet(); + else + artificialLatencyVerbs = Arrays.stream(commaDelimitedVerbs.split(",")) + .filter(s -> !s.isEmpty()) + .map(s -> { + try + { + return EnumSet.of(Verb.valueOf(s)); + } + catch (IllegalArgumentException iae) + { + try + { + return EnumSet.of(Verb.valueOf(s + "_REQ"), Verb.valueOf(s + "_RSP")); + } + catch (IllegalArgumentException ignore) {} + throw iae; + } + }) + .collect(Collector.of(() -> EnumSet.noneOf(Verb.class), Set::addAll, (left, right) -> { left.addAll(right); return left; })); + + } + + public static void setArtificialLatencyOnlyPermittedConsistencyLevels(boolean onlyPermitted) + { + artificialLatencyOnlyPermittedConsistencyLevels = onlyPermitted; + } + + public static String recommendedVerbs() + { + EnumSet verbs = EnumSet.noneOf(Verb.class); + verbs.add(Verb.MUTATION_REQ); + verbs.add(Verb.MUTATION_RSP); + verbs.add(Verb.READ_REQ); + verbs.add(Verb.READ_RSP); + verbs.add(Verb.RANGE_REQ); + verbs.add(Verb.RANGE_RSP); + verbs.add(Verb.READ_REPAIR_REQ); + verbs.add(Verb.READ_REPAIR_RSP); + verbs.add(Verb.HINT_REQ); + verbs.add(Verb.HINT_RSP); + for (Verb verb : Verb.values()) + { + if (verb.name().startsWith("ACCORD") || verb.name().startsWith("PAXOS")) + verbs.add(verb); + } + return verbs.stream().map(Verb::toString).collect(Collectors.joining(",")); + } +} diff --git a/src/java/org/apache/cassandra/net/Message.java b/src/java/org/apache/cassandra/net/Message.java index 98ecca9ef5..0143ad5a05 100644 --- a/src/java/org/apache/cassandra/net/Message.java +++ b/src/java/org/apache/cassandra/net/Message.java @@ -22,7 +22,6 @@ import java.nio.ByteBuffer; import java.util.Collections; import java.util.EnumMap; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -58,9 +57,11 @@ import static java.util.concurrent.TimeUnit.MINUTES; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static org.apache.cassandra.db.TypeSizes.sizeof; import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt; +import static org.apache.cassandra.net.MessageFlag.ARTIFICIAL_LATENCY; import static org.apache.cassandra.net.MessagingService.VERSION_40; import static org.apache.cassandra.net.MessagingService.VERSION_50; import static org.apache.cassandra.net.MessagingService.VERSION_60; +import static org.apache.cassandra.utils.Clock.Global.nanoTime; import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime; import static org.apache.cassandra.utils.vint.VIntCoding.computeUnsignedVIntSize; @@ -147,6 +148,12 @@ public class Message implements ResponseContext return header.expiresAtNanos; } + @Override + public boolean hasFlag(MessageFlag flag) + { + return header.hasFlag(flag); + } + /** For how long the message has lived. */ public long elapsedSinceCreated(TimeUnit units) { @@ -246,28 +253,31 @@ public class Message implements ResponseContext return outWithParam(nextId(), verb, 0, payload, flag.addTo(0), null, null); } - public static Message outWithFlags(Verb verb, T payload, MessageFlag flag1, MessageFlag flag2) + public static Message outWithFlag(Verb verb, T payload, Dispatcher.RequestTime requestTime, MessageFlag flag) { - assert !verb.isResponse(); - return outWithParam(nextId(), verb, 0, payload, flag2.addTo(flag1.addTo(0)), null, null); + return outWithFlags(verb, payload, requestTime, flag.addTo(0)); } - public static Message outWithFlags(Verb verb, T payload, Dispatcher.RequestTime requestTime, List flags) + public static Message outWithFlags(Verb verb, T payload, Dispatcher.RequestTime requestTime, int encodedFlags) { assert !verb.isResponse(); - int encodedFlags = 0; - for (MessageFlag flag : flags) - encodedFlags = flag.addTo(encodedFlags); + if (ArtificialLatency.isEligibleForArtificialLatency()) + encodedFlags = ARTIFICIAL_LATENCY.addTo(encodedFlags); - return new Message(new Header(nextId(), - epochSupplier.get(), - verb, - getBroadcastAddressAndPort(), - requestTime.startedAtNanos(), - requestTime.computeDeadline(verb.expiresAfterNanos()), - encodedFlags, - buildParams(null, null)), - payload); + return newWithFlags(nextId(), verb, payload, requestTime.startedAtNanos(), requestTime.computeDeadline(verb.expiresAfterNanos()), encodedFlags); + } + + private static Message newWithFlags(long id, Verb verb, T payload, long startedAtNanos, long expiresAtNanos, int encodedFlags) + { + return new Message<>(new Header(id, + epochSupplier.get(), + verb, + getBroadcastAddressAndPort(), + startedAtNanos, + expiresAtNanos, + encodedFlags, + buildParams(null, null)), + payload); } @VisibleForTesting @@ -294,6 +304,8 @@ public class Message implements ResponseContext long createdAtNanos = approxTime.now(); if (expiresAtNanos == 0) expiresAtNanos = verb.expiresAtNanos(createdAtNanos); + if (ArtificialLatency.isEligibleForArtificialLatency()) + flags = ARTIFICIAL_LATENCY.addTo(flags); return new Message<>(new Header(id, epochSupplier.get(), verb, from, createdAtNanos, expiresAtNanos, flags, buildParams(paramType, paramValue)), payload); } @@ -355,7 +367,8 @@ public class Message implements ResponseContext public static Message responseWith(T payload, ResponseContext respondTo) { - return outWithParam(respondTo.id(), respondTo.verb().responseVerb, respondTo.expiresAtNanos(), payload, null, null); + int encodedFlags = respondTo.hasFlag(ARTIFICIAL_LATENCY) ? ARTIFICIAL_LATENCY.addTo(0) : 0; + return newWithFlags(respondTo.id(), respondTo.verb().responseVerb, payload, nanoTime(), respondTo.expiresAtNanos(), encodedFlags); } /** Builds a response Message with no payload, and all the right fields inferred from request Message */ @@ -431,7 +444,7 @@ public class Message implements ResponseContext private static Map buildParams(ParamType type, Object value) { - Map params = NO_PARAMS; + EnumMap params = NO_PARAMS; if (Tracing.isTracing()) params = Tracing.instance.addTraceHeaders(new EnumMap<>(ParamType.class)); @@ -595,6 +608,11 @@ public class Message implements ResponseContext return !MessageFlag.NOT_FINAL.isIn(flags); } + boolean permitsArtificialLatency() + { + return ARTIFICIAL_LATENCY.isIn(flags); + } + @Nullable ForwardingInfo forwardTo() { @@ -665,7 +683,7 @@ public class Message implements ResponseContext private InetAddressAndPort from; private T payload; private int flags = 0; - private final Map params = new EnumMap<>(ParamType.class); + private final EnumMap params = new EnumMap<>(ParamType.class); private long createdAtNanos; private long expiresAtNanos; private long id; diff --git a/src/java/org/apache/cassandra/net/MessageFlag.java b/src/java/org/apache/cassandra/net/MessageFlag.java index 4c5762f979..da45918823 100644 --- a/src/java/org/apache/cassandra/net/MessageFlag.java +++ b/src/java/org/apache/cassandra/net/MessageFlag.java @@ -25,15 +25,16 @@ import static java.lang.Math.max; public enum MessageFlag { /** a failure response should be sent back in case of failure */ - CALL_BACK_ON_FAILURE (0), + CALL_BACK_ON_FAILURE(0), /** track repaired data - see CASSANDRA-14145 */ - TRACK_REPAIRED_DATA (1), + TRACK_REPAIRED_DATA (1), /** allow creating warnings or aborting queries based off query - see CASSANDRA-16850 */ - TRACK_WARNINGS(2), + TRACK_WARNINGS (2), /** whether this message should be sent on an URGENT channel despite its Verb default priority */ - URGENT(3), + URGENT (3), /** Allow a single callback to receive multiple responses until a final response is received **/ - NOT_FINAL(4) + NOT_FINAL (4), + ARTIFICIAL_LATENCY (5) ; private final int id; @@ -54,7 +55,7 @@ public enum MessageFlag /** * @return new flags value with this flag added */ - int addTo(int flags) + public int addTo(int flags) { return flags | (1 << id); } diff --git a/src/java/org/apache/cassandra/net/OutboundSink.java b/src/java/org/apache/cassandra/net/OutboundSink.java index 34c72dbc3a..3d19cd06d6 100644 --- a/src/java/org/apache/cassandra/net/OutboundSink.java +++ b/src/java/org/apache/cassandra/net/OutboundSink.java @@ -18,7 +18,7 @@ package org.apache.cassandra.net; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; -import java.util.function.BiPredicate; +import java.util.function.Predicate; import org.apache.cassandra.locator.InetAddressAndPort; @@ -38,22 +38,71 @@ public class OutboundSink void accept(Message message, InetAddressAndPort to, ConnectionType connectionType); } - private static class Filtered implements Sink + public interface Filter + { + public boolean test(Message message, InetAddressAndPort to, ConnectionType type); + } + + public interface AsyncFilter + { + void filter(Message message, InetAddressAndPort to, ConnectionType type, Sink next); + } + + private static abstract class AbstractFiltered implements Sink { - final BiPredicate, InetAddressAndPort> condition; final Sink next; - private Filtered(BiPredicate, InetAddressAndPort> condition, Sink next) + private AbstractFiltered(Sink next) { - this.condition = condition; this.next = next; } + abstract AbstractFiltered withNext(Sink next); + } + + private static class Filtered extends AbstractFiltered + { + final Filter condition; + + private Filtered(Filter condition, Sink next) + { + super(next); + this.condition = condition; + } + public void accept(Message message, InetAddressAndPort to, ConnectionType connectionType) { - if (condition.test(message, to)) + if (condition.test(message, to, connectionType)) next.accept(message, to, connectionType); } + + @Override + AbstractFiltered withNext(Sink next) + { + return new Filtered(condition, next); + } + } + + private static class AsyncFiltered extends AbstractFiltered + { + final AsyncFilter filter; + + private AsyncFiltered(AsyncFilter filter, Sink next) + { + super(next); + this.filter = filter; + } + + public void accept(Message message, InetAddressAndPort to, ConnectionType connectionType) + { + filter.filter(message, to, connectionType, next); + } + + @Override + AbstractFiltered withNext(Sink next) + { + return new AsyncFiltered(filter, next); + } } private volatile Sink sink; @@ -70,16 +119,26 @@ public class OutboundSink sink.accept(message, to, connectionType); } - public void add(BiPredicate, InetAddressAndPort> allow) + public void add(Filter allow) { sinkUpdater.updateAndGet(this, sink -> new Filtered(allow, sink)); } - public void remove(BiPredicate, InetAddressAndPort> allow) + public void remove(Filter allow) { sinkUpdater.updateAndGet(this, sink -> without(sink, allow)); } + public void add(AsyncFilter filter) + { + sinkUpdater.updateAndGet(this, sink -> new AsyncFiltered(filter, sink)); + } + + public void remove(AsyncFilter filter) + { + sinkUpdater.updateAndGet(this, sink -> without(sink, filter)); + } + public void clear() { sinkUpdater.updateAndGet(this, OutboundSink::clear); @@ -92,17 +151,28 @@ public class OutboundSink return sink; } - private static Sink without(Sink sink, BiPredicate, InetAddressAndPort> condition) + private static Sink without(Sink sink, Filter condition) { - if (!(sink instanceof Filtered)) - return sink; - - Filtered filtered = (Filtered) sink; - Sink next = without(filtered.next, condition); - return condition.equals(filtered.condition) ? next - : next == filtered.next - ? sink - : new Filtered(filtered.condition, next); + return without(sink, f -> f instanceof Filtered && condition.equals(((Filtered) f).condition)); } + private static Sink without(Sink sink, AsyncFilter filter) + { + return without(sink, f -> f instanceof AsyncFiltered && filter.equals(((AsyncFiltered) f).filter)); + } + + private static Sink without(Sink sink, Predicate remove) + { + if (!(sink instanceof AbstractFiltered)) + return sink; + + AbstractFiltered filtered = (AbstractFiltered) sink; + if (remove.test(filtered)) + return filtered.next; + + Sink next = without(filtered.next, remove); + if (next == filtered.next) + return filtered; + return filtered.withNext(next); + } } diff --git a/src/java/org/apache/cassandra/net/ResponseContext.java b/src/java/org/apache/cassandra/net/ResponseContext.java index c9b657a2ad..2b96c69e23 100644 --- a/src/java/org/apache/cassandra/net/ResponseContext.java +++ b/src/java/org/apache/cassandra/net/ResponseContext.java @@ -27,4 +27,5 @@ public interface ResponseContext extends ReplyContext InetAddressAndPort from(); Verb verb(); long expiresAtNanos(); + boolean hasFlag(MessageFlag flag); } diff --git a/src/java/org/apache/cassandra/service/ClientWarn.java b/src/java/org/apache/cassandra/service/ClientWarn.java index aff945c4d0..a6de392e65 100644 --- a/src/java/org/apache/cassandra/service/ClientWarn.java +++ b/src/java/org/apache/cassandra/service/ClientWarn.java @@ -40,7 +40,7 @@ public class ClientWarn extends ExecutorLocals.Impl public void set(State value) { ExecutorLocals current = ExecutorLocals.current(); - ExecutorLocals.Impl.set(current.traceState, value); + ExecutorLocals.Impl.set(current.traceState, value, current.eligibleForArtificialLatency); } public void warn(String text) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 5c1ebc9ed5..fefc6b089b 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -21,7 +21,6 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; @@ -129,6 +128,7 @@ import org.apache.cassandra.metrics.ClientRequestSizeMetrics; import org.apache.cassandra.metrics.DenylistMetrics; import org.apache.cassandra.metrics.ReadRepairMetrics; import org.apache.cassandra.metrics.StorageMetrics; +import org.apache.cassandra.net.ArtificialLatency; import org.apache.cassandra.net.ForwardingInfo; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessageFlag; @@ -759,7 +759,7 @@ public class StorageProxy implements StorageProxyMBean if (Iterables.size(missingMRC) > 0) { Tracing.trace("Repairing replicas that missed the most recent commit"); - sendCommit(mostRecent, missingMRC); + sendCommit(mostRecent, consistencyForPaxos, missingMRC); // TODO: provided commits don't invalid the prepare we just did above (which they don't), we could just wait // for all the missingMRC to acknowledge this commit and then move on with proposing our value. But that means // adding the ability to have commitPaxos block, which is exactly CASSANDRA-5442 will do. So once we have that @@ -782,7 +782,7 @@ public class StorageProxy implements StorageProxyMBean /** * Unlike commitPaxos, this does not wait for replies */ - private static void sendCommit(Commit commit, Iterable replicas) + private static void sendCommit(Commit commit, ConsistencyLevel consistencyForPaxos, Iterable replicas) { Message message = Message.out(PAXOS_COMMIT_REQ, commit); for (InetAddressAndPort target : replicas) @@ -796,7 +796,6 @@ public class StorageProxy implements StorageProxyMBean Message message = Message.out(PAXOS_PREPARE_REQ, toPrepare); boolean hasLocalRequest = false; - for (Replica replica: replicaPlan.contacts()) { if (replica.isSelf()) @@ -838,6 +837,7 @@ public class StorageProxy implements StorageProxyMBean { ProposeCallback callback = new ProposeCallback(replicaPlan.contacts().size(), replicaPlan.requiredParticipants(), !backoffIfPartial, replicaPlan.consistencyLevel(), requestTime); Message message = Message.out(PAXOS_PROPOSE_REQ, proposal); + for (Replica replica : replicaPlan.contacts()) { if (replica.isSelf()) @@ -1862,10 +1862,10 @@ public class StorageProxy implements StorageProxyMBean // belongs on a different server if (message == null) { - message = Message.outWithFlags(MUTATION_REQ, - mutation, - requestTime, - Collections.singletonList(MessageFlag.CALL_BACK_ON_FAILURE)); + message = Message.outWithFlag(MUTATION_REQ, + mutation, + requestTime, + MessageFlag.CALL_BACK_ON_FAILURE); } String dc = DatabaseDescriptor.getLocator().location(destination.endpoint()).datacenter; @@ -2356,7 +2356,7 @@ public class StorageProxy implements StorageProxyMBean try { - final ConsistencyLevel consistencyForReplayCommitsOrFetch = consistencyLevel == ConsistencyLevel.LOCAL_SERIAL + final ConsistencyLevel consistencyForReplayCommitsOrFetch = consistencyLevel.isDatacenterLocal() ? ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.QUORUM; @@ -3408,6 +3408,18 @@ public class StorageProxy implements StorageProxyMBean public Long getTruncateRpcTimeout() { return DatabaseDescriptor.getTruncateRpcTimeout(MILLISECONDS); } public void setTruncateRpcTimeout(Long timeoutInMillis) { DatabaseDescriptor.setTruncateRpcTimeout(timeoutInMillis); } + public boolean getArtificialLatencyEnabled() { return ArtificialLatency.isEnabled(); } + public void setArtificialLatencyEnabled(boolean enabled) { ArtificialLatency.setEnabled(enabled); } + + public String getArtificialLatencyVerbs() { return ArtificialLatency.getArtificialLatencyVerbs(); } + public void setArtificialLatencyVerbs(String commaDelimitedVerbs) { ArtificialLatency.setArtificialLatencyVerbs(commaDelimitedVerbs); } + + public String getArtificialLatencies() { return ArtificialLatency.getArtificialLatencies(); } + public void setArtificialLatencies(String latencies) { ArtificialLatency.setArtificialLatencies(latencies); } + + public boolean getAllowArtificialLatencyForAllConsistencyLevels() { return ArtificialLatency.getArtificialLatencyOnlyPermittedConsistencyLevels(); } + public void setAllowArtificialLatencyForAllConsistencyLevels(boolean onlyPermitted) { ArtificialLatency.setArtificialLatencyOnlyPermittedConsistencyLevels(onlyPermitted); } + public Long getNativeTransportMaxConcurrentConnections() { return DatabaseDescriptor.getNativeTransportMaxConcurrentConnections(); } public void setNativeTransportMaxConcurrentConnections(Long nativeTransportMaxConcurrentConnections) { DatabaseDescriptor.setNativeTransportMaxConcurrentConnections(nativeTransportMaxConcurrentConnections); } diff --git a/src/java/org/apache/cassandra/service/StorageProxyMBean.java b/src/java/org/apache/cassandra/service/StorageProxyMBean.java index 1c4887f1e1..eabe28ece7 100644 --- a/src/java/org/apache/cassandra/service/StorageProxyMBean.java +++ b/src/java/org/apache/cassandra/service/StorageProxyMBean.java @@ -52,6 +52,15 @@ public interface StorageProxyMBean public Long getTruncateRpcTimeout(); public void setTruncateRpcTimeout(Long timeoutInMillis); + public boolean getArtificialLatencyEnabled(); + public void setArtificialLatencyEnabled(boolean enabled); + public String getArtificialLatencyVerbs(); + public void setArtificialLatencyVerbs(String commaDelimitedVerbs); + public String getArtificialLatencies(); + public void setArtificialLatencies(String latencies); + public boolean getAllowArtificialLatencyForAllConsistencyLevels(); + public void setAllowArtificialLatencyForAllConsistencyLevels(boolean onlyPermitted); + public void setNativeTransportMaxConcurrentConnections(Long nativeTransportMaxConcurrentConnections); public Long getNativeTransportMaxConcurrentConnections(); diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java index 814aa74481..bbe2a2d25e 100644 --- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java +++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java @@ -361,7 +361,6 @@ public abstract class AbstractReadExecutor if (traceState != null) traceState.trace("speculating read retry on {}", extraReplica); logger.trace("speculating read retry on {}", extraReplica); - MessagingService.instance().sendWithCallback(retryCommand.createMessage(false, requestTime), extraReplica.endpoint(), handler); } } diff --git a/src/java/org/apache/cassandra/tracing/Tracing.java b/src/java/org/apache/cassandra/tracing/Tracing.java index f1c5b54b94..189fcee2e5 100644 --- a/src/java/org/apache/cassandra/tracing/Tracing.java +++ b/src/java/org/apache/cassandra/tracing/Tracing.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.net.InetAddress; import java.nio.ByteBuffer; import java.util.Collections; +import java.util.EnumMap; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -230,7 +231,7 @@ public abstract class Tracing extends ExecutorLocals.Impl public void set(TraceState tls) { ExecutorLocals current = ExecutorLocals.current(); - ExecutorLocals.Impl.set(tls, current.clientWarnState); + ExecutorLocals.Impl.set(tls, current.clientWarnState, current.eligibleForArtificialLatency); } public TraceState begin(final String request, final Map parameters) @@ -303,7 +304,7 @@ public abstract class Tracing extends ExecutorLocals.Impl } } - public Map addTraceHeaders(Map addToMutable) + public EnumMap addTraceHeaders(EnumMap addToMutable) { assert isTracing(); diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java index 29cc9246b0..0f6afe0fe1 100644 --- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java @@ -30,6 +30,7 @@ import org.apache.cassandra.cql3.QueryHandler; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.ResultSet; import org.apache.cassandra.exceptions.PreparedQueryNotFoundException; +import org.apache.cassandra.net.ArtificialLatency; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.tracing.Tracing; @@ -157,6 +158,9 @@ public class ExecuteMessage extends Message.Request if (traceRequest) traceQuery(state, prepared); + if (options.isEligibleForArtificialLatency()) + ArtificialLatency.setEligibleForArtificialLatency(true); + // Some custom QueryHandlers are interested by the bound names. We provide them this information // by wrapping the QueryOptions. QueryOptions queryOptions = QueryOptions.addColumnSpecifications(options, prepared.statement.getBindVariables()); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index c5498ccaf0..8a910028f9 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -121,6 +121,7 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.metrics.Sampler; import org.apache.cassandra.metrics.ThreadLocalMetrics; +import org.apache.cassandra.net.ArtificialLatency; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.NoPayload; @@ -374,7 +375,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance protected void registerMockMessaging(ICluster cluster) { - MessagingService.instance().outboundSink.add((message, to) -> { + MessagingService.instance().outboundSink.add((message, to, type) -> { if (!internodeMessagingStarted) { inInstancelogger.debug("Dropping outbound message {} to {} as internode messaging has not been started yet", @@ -405,7 +406,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance protected void registerOutboundFilter(ICluster cluster) { - MessagingService.instance().outboundSink.add((message, to) -> { + MessagingService.instance().outboundSink.add((message, to, type) -> { if (isShutdown()) return false; // TODO: Simulator needs this to trigger a failure int fromNum = config.num(); // since this instance is sending the message, from will always be this instance @@ -678,6 +679,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { partialStartup(cluster); } + ArtificialLatency.touch(); } catch (Throwable t) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java index 3fe95b5070..81f2aff281 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java @@ -32,14 +32,17 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicIntegerArray; +import java.util.function.IntSupplier; import com.google.common.util.concurrent.RateLimiter; +import org.agrona.collections.IntArrayList; +import org.apache.commons.math3.distribution.ZipfDistribution; +import org.apache.commons.math3.random.JDKRandomGenerator; import org.junit.BeforeClass; -import org.junit.Ignore; -import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,14 +57,15 @@ import org.apache.cassandra.distributed.api.IMessageFilters; import org.apache.cassandra.distributed.shared.DistributedTestBase; import org.apache.cassandra.net.Verb; import org.apache.cassandra.schema.Schema; -import org.apache.cassandra.service.accord.AccordExecutor; import org.apache.cassandra.service.accord.AccordKeyspace; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.utils.EstimatedHistogram; +import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static java.lang.System.currentTimeMillis; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS; public class AccordLoadTest extends AccordTestBase @@ -72,27 +76,236 @@ public class AccordLoadTest extends AccordTestBase public static void setUp() throws IOException { CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(MILLISECONDS.toSeconds(currentTimeMillis()))); -// AccordTestBase.setupCluster(builder -> builder, 3); - AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config - .with(Feature.NETWORK, Feature.GOSSIP) - .set("accord.shard_durability_target_splits", "8") - .set("accord.shard_durability_max_splits", "16") - .set("accord.shard_durability_cycle", "1m") - .set("accord.catchup_on_start_fail_latency", "2m") -// .set("accord.ephemeral_read_enabled", "true") - ), 3); + int nodeCount = 3; + AccordTestBase.setupCluster(builder -> builder.withDCs(nodeCount).withConfig(config -> { + config.with(Feature.NETWORK, Feature.GOSSIP) + .set("accord.shard_durability_target_splits", "8") + .set("accord.shard_durability_max_splits", "16") + .set("accord.shard_durability_cycle", "1m") + .set("accord.catchup_on_start_fail_latency", "2m"); + }), nodeCount); } - @Ignore - @Test - public void testLoad() throws Exception + public static class Settings + { + final int repairInterval; + final int compactionInterval; + final int journalFlushInterval; + final int cfkFlushInterval; + final int cfkCompactionPeriodSeconds; + final int dataFlushInterval; + final int restartInterval; + final int restartDecay; + final int batchSize; + final long batchPeriodNanos; + final int clientConcurrency; + final int clients; + final int clientRatePerSecond; + final int keysPerOperation; + final float readRatio; + final IntSupplier keySelector; + final boolean readBeforeWrite; + + Settings(SettingsBuilder builder) + { + this.repairInterval = builder.repairInterval; + this.compactionInterval = builder.compactionInterval; + this.journalFlushInterval = builder.journalFlushInterval; + this.cfkFlushInterval = builder.cfkFlushInterval; + this.cfkCompactionPeriodSeconds = builder.cfkCompactionPeriodSeconds; + this.dataFlushInterval = builder.dataFlushInterval; + this.restartInterval = builder.restartInterval; + this.restartDecay = builder.restartDecay; + this.batchSize = builder.batchSize; + this.batchPeriodNanos = builder.batchPeriodNanos; + this.clientConcurrency = builder.clientConcurrency; + this.clients = builder.clients; + this.clientRatePerSecond = builder.ratePerSecond; + this.keysPerOperation = builder.keysPerOperation; + this.readRatio = builder.readChance; + this.keySelector = builder.keySelector; + this.readBeforeWrite = builder.readBeforeWrite; + } + } + + // interval is measured in terms of *operations* unless otherwise specified + public static class SettingsBuilder + { + int repairInterval = Integer.MAX_VALUE; + int compactionInterval = Integer.MAX_VALUE; + int journalFlushInterval = Integer.MAX_VALUE; + int cfkFlushInterval = Integer.MAX_VALUE; + int cfkCompactionPeriodSeconds = Integer.MAX_VALUE; + int dataFlushInterval = Integer.MAX_VALUE; + int restartInterval = Integer.MAX_VALUE; + int restartDecay = 2; + int batchSize = 1000; + long batchPeriodNanos = SECONDS.toNanos(10); + int clientConcurrency = 50; + int clients = -1; + int ratePerSecond = 1000; + int keysPerOperation = 1; + float readChance = 0.5f; + IntSupplier keySelector; + boolean readBeforeWrite; + + public SettingsBuilder setRepairInterval(int repairInterval) + { + this.repairInterval = repairInterval; + return this; + } + + public SettingsBuilder setCompactionInterval(int compactionInterval) + { + this.compactionInterval = compactionInterval; + return this; + } + + public SettingsBuilder setJournalFlushInterval(int journalFlushInterval) + { + this.journalFlushInterval = journalFlushInterval; + return this; + } + + public SettingsBuilder setCfkFlushInterval(int cfkFlushInterval) + { + this.cfkFlushInterval = cfkFlushInterval; + return this; + } + + public SettingsBuilder setCfkCompactionPeriodSeconds(int cfkCompactionPeriodSeconds) + { + this.cfkCompactionPeriodSeconds = cfkCompactionPeriodSeconds; + return this; + } + + public SettingsBuilder setDataFlushInterval(int dataFlushInterval) + { + this.dataFlushInterval = dataFlushInterval; + return this; + } + + public SettingsBuilder setRestartInterval(int restartInterval) + { + this.restartInterval = restartInterval; + return this; + } + + public SettingsBuilder setRestartDecay(int restartDecay) + { + this.restartDecay = restartDecay; + return this; + } + + public SettingsBuilder setBatchSize(int batchSize) + { + this.batchSize = batchSize; + return this; + } + + public SettingsBuilder setBatchPeriodNanos(long batchPeriodNanos) + { + this.batchPeriodNanos = batchPeriodNanos; + return this; + } + + public SettingsBuilder setClientConcurrency(int clientConcurrency) + { + this.clientConcurrency = clientConcurrency; + return this; + } + + public SettingsBuilder setClients(int clients) + { + this.clients = clients; + return this; + } + + public SettingsBuilder setRatePerSecond(int ratePerSecond) + { + this.ratePerSecond = ratePerSecond; + return this; + } + + public SettingsBuilder setKeysPerOperation(int keysPerOperation) + { + this.keysPerOperation = keysPerOperation; + return this; + } + + public SettingsBuilder setReadChance(float readChance) + { + this.readChance = readChance; + return this; + } + + public SettingsBuilder setReadBeforeWrite(boolean readBeforeWrite) + { + this.readBeforeWrite = readBeforeWrite; + return this; + } + + public SettingsBuilder setKeySelector(IntSupplier keySelector) + { + this.keySelector = keySelector; + return this; + } + + public Settings build() + { + return new Settings(this); + } + } + + private static SettingsBuilder ycsbA(SettingsBuilder builder, int keyCount) + { + return builder.setKeySelector(ycsbZipfian(keyCount)) + .setReadChance(0.5f); + } + + private static SettingsBuilder ycsbB(SettingsBuilder builder, int keyCount) + { + return builder.setKeySelector(ycsbZipfian(keyCount)) + .setReadChance(0.95f); + } + + private static SettingsBuilder ycsbC(SettingsBuilder builder, int keyCount) + { + return builder.setKeySelector(ycsbZipfian(keyCount)) + .setReadChance(1.0f); + + } + + private static IntSupplier ycsbZipfian(int keyCount) + { + ZipfDistribution distribution = new ZipfDistribution(new JDKRandomGenerator(), keyCount, 0.99); + return distribution::sample; + } + + private static IntSupplier roundrobin(int keyCount) + { + AtomicInteger next = new AtomicInteger(); + return () -> { + int v = next.incrementAndGet(); + if (v < keyCount) + return v; + return next.updateAndGet(i -> i > keyCount ? 0 : i + 1); + }; + } + + private static IntSupplier uniform(int keyCount) + { + Random random = new Random(); + return () -> random.nextInt(keyCount); + } + + public void testLoad(final Settings settings) throws Exception { Cluster cluster = SHARED_CLUSTER; cluster.schemaChange("CREATE TABLE " + qualifiedAccordTableName + " (k int, v int, PRIMARY KEY(k)) WITH transactional_mode = 'full'"); try { - final ConcurrentHashMap verbs = new ConcurrentHashMap<>(); cluster.filters().outbound().messagesMatching(new IMessageFilters.Matcher() { @@ -104,129 +317,138 @@ public class AccordLoadTest extends AccordTestBase } }).drop(); - ICoordinator coordinator = cluster.coordinator(1); - final int repairInterval = Integer.MAX_VALUE; - final int compactionInterval = Integer.MAX_VALUE; -// final int flushInterval = 50_000; - final int journalFlushInterval = Integer.MAX_VALUE; - final int cfkFlushInterval = Integer.MAX_VALUE; - final int dataFlushInterval = Integer.MAX_VALUE; -// final int compactionPeriodSeconds = 0; -// int restartInterval = 30_000; - int restartInterval = Integer.MAX_VALUE; - final int restartDecay = 2; -// final int restartInterval = Integer.MAX_VALUE; - final int batchSizeLimit = 200; - final long batchTime = TimeUnit.SECONDS.toNanos(10); - final int concurrency = 200; - final int ratePerSecond = 500; -// final int keyCount = 10_000; - final int keyCount = 10_000; - final float readChance = 0.33f; - long nextRepairAt = repairInterval; - long nextCompactionAt = compactionInterval; - long nextJournalFlushAt = journalFlushInterval; - long nextDataFlushAt = dataFlushInterval; - long nextCfkFlushAt = cfkFlushInterval; - long nextRestartAt = restartInterval; + int clientCount = settings.clients < 0 ? cluster.size() : settings.clients; + long nextRepairAt = settings.repairInterval; + long nextCompactionAt = settings.compactionInterval; + long nextJournalFlushAt = settings.journalFlushInterval; + long nextDataFlushAt = settings.dataFlushInterval; + long nextCfkFlushAt = settings.cfkFlushInterval; + long nextRestartAt = settings.restartInterval; final ExecutorService restartExecutor = Executors.newSingleThreadExecutor(); + final ExecutorService clientExecutor = Executors.newFixedThreadPool(settings.clients); final BitSet initialised = new BitSet(); java.util.concurrent.Future restarting = null; cluster.get(1).nodetoolResult("cms", "reconfigure", "3").asserts().success(); -// cluster.forEach(i -> i.runOnInstance(() -> { -// if (compactionPeriodSeconds > 0) -// ((AccordService) AccordService.instance()).journal().compactor().updateCompactionPeriod(1, SECONDS); -// ((AccordSpec.JournalSpec)((AccordService) AccordService.instance()).journal().configuration()).segmentSize = 128 << 10; -// })); + if (settings.cfkCompactionPeriodSeconds > 0) + { + cluster.forEach(i -> i.runOnInstance(() -> { + ((AccordService) AccordService.instance()).journal().compactor().updateCompactionPeriod(settings.cfkCompactionPeriodSeconds, SECONDS); + })); + } + final AtomicBoolean stop = new AtomicBoolean(); Random random = new Random(); - final Semaphore inFlight = new Semaphore(concurrency); - final RateLimiter rateLimiter = RateLimiter.create(ratePerSecond); + Semaphore completed = new Semaphore(0); + AtomicIntegerArray coordinatorIndexes = new AtomicIntegerArray(clientCount); + final List> clients = new ArrayList<>(); + final EstimatedHistogram histogram = new EstimatedHistogram(200); + if (settings.clients >= cluster.size()) + throw new IllegalArgumentException("Cannot have more clients than nodes"); + if (settings.restartInterval < Integer.MAX_VALUE && settings.clients + 1 >= cluster.size()) + throw new IllegalArgumentException("If restarting, cannot have as many clients as nodes, as must reroute client requests during restart"); + + for (int client = 0 ; client < clientCount ; ++client) + { + final int clientIndex = client; + coordinatorIndexes.set(client, client + 1); + final RateLimiter rateLimiter = RateLimiter.create(settings.clientRatePerSecond); + final Semaphore inFlight = new Semaphore(0); + clients.add(clientExecutor.submit(() -> { + while (!stop.get()) + { + int coordinatorIdx = coordinatorIndexes.get(clientIndex); + ICoordinator coordinator = cluster.coordinator(coordinatorIdx); + try + { + rateLimiter.acquire(); + inFlight.acquire(); + long commandStart = System.nanoTime(); + IntArrayList keys = new IntArrayList(settings.keysPerOperation, -1); + for (int i = 0 ; i < settings.keysPerOperation ; ++i) + { + int k = settings.keySelector.getAsInt(); + if (!keys.containsInt(k)) + keys.add(k); + } + if (!keys.intStream().allMatch(initialised::get)) + { + coordinator.executeWithResult((success, fail) -> { + inFlight.release(); + completed.release(); + if (fail == null) + { + histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); + synchronized (initialised) + { + keys.forEachInt(initialised::set); + } + } + else + { + logger.error("{}", fail.getMessage()); + } + }, "UPDATE " + qualifiedAccordTableName + " SET v = 0 WHERE k IN ?", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, keys); + + } + else if (random.nextFloat() < settings.readRatio) + { + coordinator.executeWithResult((success, fail) -> { + inFlight.release(); + completed.release(); + if (fail == null) + histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); + }, "BEGIN TRANSACTION\n" + + "SELECT * FROM " + qualifiedAccordTableName + " WHERE k IN ?;\n" + + "COMMIT TRANSACTION;", ConsistencyLevel.SERIAL, keys + ); + } + else + { + coordinator.executeWithResult((success, fail) -> { + inFlight.release(); + completed.release(); + if (fail == null) + histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); + else + logger.error("{}", fail.getMessage()); + }, "BEGIN TRANSACTION\n" + + // "UPDATE " + qualifiedAccordTableName + " SET v = ? WHERE k = ?;\n" + + "UPDATE " + qualifiedAccordTableName + " SET v += ? WHERE k IN ?;\n" + + "COMMIT TRANSACTION;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, random.nextInt(100), keys); + } + } + catch (RejectedExecutionException e) + { + inFlight.release(); + } + catch (InterruptedException e) + { + throw new UncheckedInterruptedException(e); + } + } + })); + } + while (true) { - final EstimatedHistogram histogram = new EstimatedHistogram(200); long batchStart = System.nanoTime(); - long batchEnd = batchStart + batchTime; int batchSize = 0; - while (batchSize < batchSizeLimit) - { - inFlight.acquire(); - rateLimiter.acquire(); - try - { - long commandStart = System.nanoTime(); - int k1 = random.nextInt(keyCount); - int k2 = random.nextInt(keyCount); - if (random.nextFloat() < readChance) - { - coordinator.executeWithResult((success, fail) -> { - inFlight.release(); - if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); - }, "BEGIN TRANSACTION\n" + - "SELECT * FROM " + qualifiedAccordTableName + " WHERE k IN ?;\n" + - "COMMIT TRANSACTION;", ConsistencyLevel.SERIAL, - List.of(k1, k2) -// List.of(k1) - ); - } - else if (initialised.get(k1) && initialised.get(k2)) - { - coordinator.executeWithResult((success, fail) -> { - inFlight.release(); - if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); - }, "BEGIN TRANSACTION\n" + - "UPDATE " + qualifiedAccordTableName + " SET v += 1 WHERE k = ?;\n" + - "UPDATE " + qualifiedAccordTableName + " SET v += 1 WHERE k = ?;\n" + - "COMMIT TRANSACTION;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, k1, k2); - } - else - { - initialised.set(k1); - initialised.set(k2); - coordinator.executeWithResult((success, fail) -> { - inFlight.release(); - if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); - // else exceptions.add(fail); - }, "UPDATE " + qualifiedAccordTableName + " SET v = 0 WHERE k = ? IF NOT EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, k1); - coordinator.executeWithResult((success, fail) -> { - inFlight.release(); - if (fail == null) histogram.add(NANOSECONDS.toMicros(System.nanoTime() - commandStart)); - // else exceptions.add(fail); - }, "UPDATE " + qualifiedAccordTableName + " SET v = 0 WHERE k = ? IF NOT EXISTS;", ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM, k2); - } - } - catch (RejectedExecutionException e) - { - inFlight.release(); - while (true) - { - try - { - int index = 1 + random.nextInt(cluster.size()); - logger.info("Picking new coordinator ... {}", index); - coordinator = cluster.coordinator(index); - if (cluster.get(index).callOnInstance(() -> AccordService.started())) - break; - } - catch (Throwable t) { logger.info("Failed to select coordinator", t); } - } - } - batchSize++; - if (System.nanoTime() >= batchEnd) - break; - } + + if (completed.tryAcquire(settings.batchSize, settings.batchPeriodNanos, NANOSECONDS)) + batchSize = settings.batchSize; + batchSize += completed.drainPermits(); if ((nextRepairAt -= batchSize) <= 0) { - nextRepairAt += repairInterval; + nextRepairAt += settings.repairInterval; System.out.println("repairing..."); cluster.coordinator(1).instance().nodetool("repair", qualifiedAccordTableName); } if ((nextCompactionAt -= batchSize) <= 0) { - nextCompactionAt += compactionInterval; + nextCompactionAt += settings.compactionInterval; System.out.println("compacting accord..."); cluster.forEach(i -> { try { i.nodetool("compact", "system_accord.journal"); } @@ -236,7 +458,7 @@ public class AccordLoadTest extends AccordTestBase if ((nextJournalFlushAt -= batchSize) <= 0) { - nextJournalFlushAt += journalFlushInterval; + nextJournalFlushAt += settings.journalFlushInterval; System.out.println("flushing journal..."); cluster.forEach(i -> { try @@ -258,7 +480,7 @@ public class AccordLoadTest extends AccordTestBase if ((nextDataFlushAt -= batchSize) <= 0) { - nextDataFlushAt += dataFlushInterval; + nextDataFlushAt += settings.dataFlushInterval; System.out.println("flushing data..."); cluster.forEach(i -> { try @@ -276,7 +498,7 @@ public class AccordLoadTest extends AccordTestBase if ((nextCfkFlushAt -= batchSize) <= 0) { - nextCfkFlushAt += cfkFlushInterval; + nextCfkFlushAt += settings.cfkFlushInterval; System.out.println("flushing data..."); cluster.forEach(i -> { try @@ -300,8 +522,25 @@ public class AccordLoadTest extends AccordTestBase if (restarting != null) restarting.get(); - nextRestartAt += restartInterval; + nextRestartAt += settings.restartInterval; int nodeIdx = 1 + random.nextInt(cluster.size()); + out: for (int i = 0 ; i < coordinatorIndexes.length() ; ++i) + { + if (nodeIdx == coordinatorIndexes.get(i)) + { + cont: while (true) + { + int replaceIdx = 1 + random.nextInt(cluster.size()); + for (int j = 0 ; j < coordinatorIndexes.length() ; ++j) + { + if (coordinatorIndexes.get(j) == replaceIdx) + continue cont; + } + coordinatorIndexes.set(i, replaceIdx); + break out; + } + } + } restarting = restartExecutor.submit(() -> { System.out.printf("restarting node %d...\n", nodeIdx); try @@ -315,20 +554,12 @@ public class AccordLoadTest extends AccordTestBase throw new RuntimeException(e); } }); - if (nodeIdx == coordinator.instance().config().num()) - coordinator = cluster.coordinator((nodeIdx % cluster.size()) + 1); } } final Date date = new Date(); - System.out.printf("%tT rate: %.2f/s (%d total)\n", date, (((float)batchSizeLimit * 1000) / NANOSECONDS.toMillis(System.nanoTime() - batchStart)), batchSize); - System.out.printf("%tT percentiles: %d %d %d %d\n", date, histogram.percentile(.25)/1000, histogram.percentile(.5)/1000, histogram.percentile(.75)/1000, histogram.percentile(1)/1000); - cluster.forEach(() -> { - String waiting = ""; - for (AccordExecutor executor : AccordService.instance().executors()) - waiting += executor.unsafeWaitingToRunCount() + " "; - System.out.println(waiting); - }); + System.out.printf("%tT rate: %.2f/s (%d total)\n", date, (((float)settings.batchSize * 1000) / NANOSECONDS.toMillis(System.nanoTime() - batchStart)), batchSize); + System.out.printf("%tT percentiles: %d %d %d %d\n", date, histogram.percentile(.25)/1000, histogram.percentile(.5)/1000, histogram.percentile(.95)/1000, histogram.percentile(.99)/1000, histogram.percentile(1)/1000); class VerbCount { @@ -382,6 +613,6 @@ public class AccordLoadTest extends AccordTestBase AccordLoadTest.setUp(); AccordLoadTest test = new AccordLoadTest(); test.setup(); - test.testLoad(); + test.testLoad(ycsbA(new SettingsBuilder(), 1000).build()); } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java index bb01c624b6..f1c669efa7 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java @@ -46,6 +46,7 @@ import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.dht.Token; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ConnectionType; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.Verb; @@ -1155,7 +1156,7 @@ public class ClusterMetadataTestHelper final SettableFuture future = SettableFuture.create(); Set ignore = Sets.newHashSet(ignored); MessagingService.instance().outboundSink.clear(); - MessagingService.instance().outboundSink.add((Message message, InetAddressAndPort to) -> + MessagingService.instance().outboundSink.add((Message message, InetAddressAndPort to, ConnectionType type) -> { if (!ignore.contains(message.verb())) future.set(new MessageDelivery(message, to)); diff --git a/test/unit/org/apache/cassandra/concurrent/LocalAwareExecutorPlusTest.java b/test/unit/org/apache/cassandra/concurrent/LocalAwareExecutorPlusTest.java index f47046c759..6b100baf18 100644 --- a/test/unit/org/apache/cassandra/concurrent/LocalAwareExecutorPlusTest.java +++ b/test/unit/org/apache/cassandra/concurrent/LocalAwareExecutorPlusTest.java @@ -25,7 +25,7 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac public class LocalAwareExecutorPlusTest extends AbstractExecutorPlusTest { - final ExecutorLocals locals = new ExecutorLocals(null, null); + final ExecutorLocals locals = new ExecutorLocals(null, null, false); @Test public void testPooled() throws Throwable diff --git a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java index 5b42b98b1a..bb5e147e33 100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java @@ -91,7 +91,7 @@ public class ReadCommandVerbHandlerOutOfRangeTest MessagingService.instance().inboundSink.clear(); MessagingService.instance().outboundSink.clear(); - MessagingService.instance().outboundSink.add((message, to) -> false); + MessagingService.instance().outboundSink.add((message, to, type) -> false); MessagingService.instance().inboundSink.add((message) -> false); cfs = Keyspace.open(KEYSPACE_NONREPLICATED).getColumnFamilyStore(TABLE); diff --git a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java index c005011051..9fbe25df48 100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java @@ -92,7 +92,7 @@ public class ReadCommandVerbHandlerTest { MessagingService.instance().inboundSink.clear(); MessagingService.instance().outboundSink.clear(); - MessagingService.instance().outboundSink.add((message, to) -> false); + MessagingService.instance().outboundSink.add((message, to, type) -> false); MessagingService.instance().inboundSink.add((message) -> false); handler = new ReadCommandVerbHandler(); diff --git a/test/unit/org/apache/cassandra/db/compaction/AbstractPendingRepairTest.java b/test/unit/org/apache/cassandra/db/compaction/AbstractPendingRepairTest.java index c2ec223fbd..cc4e6b082b 100644 --- a/test/unit/org/apache/cassandra/db/compaction/AbstractPendingRepairTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/AbstractPendingRepairTest.java @@ -62,7 +62,7 @@ public class AbstractPendingRepairTest extends AbstractRepairTest LocalSessionAccessor.startup(); // cutoff messaging service - MessagingService.instance().outboundSink.add((message, to) -> false); + MessagingService.instance().outboundSink.add((message, to, type) -> false); MessagingService.instance().inboundSink.add((message) -> false); } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java index ac6c529fc5..db9788935f 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java @@ -505,7 +505,7 @@ public class CompactionIteratorTest extends CQLTester TableMetadata metadata = getCurrentColumnFamilyStore().metadata(); final HashMap> sentMessages = new HashMap<>(); - MessagingService.instance().outboundSink.add((message, to) -> { sentMessages.put(to, message); return false;}); + MessagingService.instance().outboundSink.add((message, to, type) -> { sentMessages.put(to, message); return false;}); // no duplicates sentMessages.clear(); diff --git a/test/unit/org/apache/cassandra/db/compression/CompressionDictionaryEventHandlerTest.java b/test/unit/org/apache/cassandra/db/compression/CompressionDictionaryEventHandlerTest.java index 143f18cab6..8ce15b6168 100644 --- a/test/unit/org/apache/cassandra/db/compression/CompressionDictionaryEventHandlerTest.java +++ b/test/unit/org/apache/cassandra/db/compression/CompressionDictionaryEventHandlerTest.java @@ -135,7 +135,7 @@ public class CompressionDictionaryEventHandlerTest AtomicReference capturedMessage = new AtomicReference<>(); // Capture outbound messages - MessagingService.instance().outboundSink.add((message, to) -> { + MessagingService.instance().outboundSink.add((message, to, type) -> { if (message.verb() == Verb.DICTIONARY_UPDATE_REQ) { capturedMessage.set((CompressionDictionaryUpdateMessage) message.payload); @@ -213,7 +213,7 @@ public class CompressionDictionaryEventHandlerTest public void testSendNotificationRobustness() { // Test that sending notifications doesn't throw even if messaging fails - MessagingService.instance().outboundSink.add((message, to) -> { + MessagingService.instance().outboundSink.add((message, to, type) -> { if (message.verb() == Verb.DICTIONARY_UPDATE_REQ) { throw new RuntimeException("Simulated messaging failure"); diff --git a/test/unit/org/apache/cassandra/db/transform/DuplicateRowCheckerTest.java b/test/unit/org/apache/cassandra/db/transform/DuplicateRowCheckerTest.java index ad812fc01e..15ac67eb89 100644 --- a/test/unit/org/apache/cassandra/db/transform/DuplicateRowCheckerTest.java +++ b/test/unit/org/apache/cassandra/db/transform/DuplicateRowCheckerTest.java @@ -71,7 +71,7 @@ public class DuplicateRowCheckerTest extends CQLTester public static void setupMessaging() { sentMessages = new HashMap<>(); - MessagingService.instance().outboundSink.add((message, to) -> { sentMessages.put(to, message); return false;}); + MessagingService.instance().outboundSink.add((message, to, type) -> { sentMessages.put(to, message); return false;}); } @Before diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index c8fdf82215..a9ff92a7bd 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -29,7 +29,6 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import java.util.function.BiPredicate; import javax.annotation.Nullable; @@ -78,8 +77,10 @@ import org.apache.cassandra.dht.Murmur3Partitioner.LongToken; import org.apache.cassandra.exceptions.ExceptionSerializer; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ConnectionType; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.net.OutboundSink; import org.apache.cassandra.net.Verb; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; @@ -1098,7 +1099,7 @@ public class AccordDebugKeyspaceTest extends CQLTester return (AccordService) AccordService.instance(); } - private static class AccordMsgFilter implements BiPredicate, InetAddressAndPort> + private static class AccordMsgFilter implements OutboundSink.Filter { volatile Condition preAccept = Condition.newOneTimeCondition(); volatile Condition commit = Condition.newOneTimeCondition(); @@ -1123,7 +1124,7 @@ public class AccordDebugKeyspaceTest extends CQLTester } @Override - public boolean test(Message msg, InetAddressAndPort to) + public boolean test(Message msg, InetAddressAndPort to, ConnectionType type) { if (!msg.verb().name().startsWith("ACCORD_")) return true; diff --git a/test/unit/org/apache/cassandra/net/MatcherResponse.java b/test/unit/org/apache/cassandra/net/MatcherResponse.java index 228272b3c8..36aa76b579 100644 --- a/test/unit/org/apache/cassandra/net/MatcherResponse.java +++ b/test/unit/org/apache/cassandra/net/MatcherResponse.java @@ -25,7 +25,6 @@ import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; -import java.util.function.BiPredicate; import java.util.function.Function; import com.google.common.collect.Multimap; @@ -45,7 +44,7 @@ public class MatcherResponse implements Closeable Multimaps.newListMultimap(new HashMap<>(), ArrayList::new); private final MockMessagingSpy spy = new MockMessagingSpy(); private final AtomicInteger limitCounter = new AtomicInteger(Integer.MAX_VALUE); - private BiPredicate, InetAddressAndPort> sink; + private OutboundSink.Filter sink; MatcherResponse(Matcher matcher) { @@ -160,9 +159,10 @@ public class MatcherResponse implements Closeable assert sink == null: "destroy() must be called first to register new response"; - sink = new BiPredicate, InetAddressAndPort>() + sink = new OutboundSink.Filter() { - public boolean test(Message message, InetAddressAndPort to) + @Override + public boolean test(Message message, InetAddressAndPort to, ConnectionType type) { // prevent outgoing message from being send in case matcher indicates a match // and instead send the mocked response diff --git a/test/unit/org/apache/cassandra/net/MessageTest.java b/test/unit/org/apache/cassandra/net/MessageTest.java index 1455cdf0c9..677199e4fc 100644 --- a/test/unit/org/apache/cassandra/net/MessageTest.java +++ b/test/unit/org/apache/cassandra/net/MessageTest.java @@ -42,6 +42,7 @@ import org.apache.cassandra.net.MessagingService.Version; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.tracing.Tracing.TraceType; +import org.apache.cassandra.transport.Dispatcher; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FreeRunningClock; import org.apache.cassandra.utils.TimeUUID; @@ -55,6 +56,7 @@ import static org.apache.cassandra.net.NoPayload.noPayload; import static org.apache.cassandra.net.ParamType.RESPOND_TO; import static org.apache.cassandra.net.ParamType.TRACE_SESSION; import static org.apache.cassandra.net.ParamType.TRACE_TYPE; +import static org.apache.cassandra.utils.Clock.Global.nanoTime; import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime; import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; import static org.junit.Assert.assertEquals; @@ -193,7 +195,7 @@ public class MessageTest { testCycle(Message.out(Verb._TEST_2, 42)); testCycle(Message.outWithFlag(Verb._TEST_2, 42, MessageFlag.CALL_BACK_ON_FAILURE)); - testCycle(Message.outWithFlags(Verb._TEST_2, 42, MessageFlag.CALL_BACK_ON_FAILURE, MessageFlag.TRACK_REPAIRED_DATA)); + testCycle(Message.outWithFlags(Verb._TEST_2, 42, new Dispatcher.RequestTime(nanoTime()), MessageFlag.CALL_BACK_ON_FAILURE.addTo(MessageFlag.TRACK_REPAIRED_DATA.addTo(0)))); testCycle(Message.outWithParam(1, Verb._TEST_2, 42, RESPOND_TO, FBUtilities.getBroadcastAddressAndPort())); } diff --git a/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java b/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java index 80442d514c..454e293a0f 100644 --- a/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java +++ b/test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java @@ -23,7 +23,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; -import java.util.function.BiPredicate; import org.junit.After; import org.junit.Assert; @@ -211,7 +210,7 @@ public class StartupClusterConnectivityCheckerTest } } - private static class Sink implements BiPredicate, InetAddressAndPort> + private static class Sink implements OutboundSink.Filter { private final boolean markAliveInGossip; private final boolean processConnectAck; @@ -227,7 +226,7 @@ public class StartupClusterConnectivityCheckerTest } @Override - public boolean test(Message message, InetAddressAndPort to) + public boolean test(Message message, InetAddressAndPort to, ConnectionType type) { ConnectionTypeRecorder recorder = seenConnectionRequests.computeIfAbsent(to, inetAddress -> new ConnectionTypeRecorder()); diff --git a/test/unit/org/apache/cassandra/repair/RepairJobTest.java b/test/unit/org/apache/cassandra/repair/RepairJobTest.java index 6fcb0416ba..008a018015 100644 --- a/test/unit/org/apache/cassandra/repair/RepairJobTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairJobTest.java @@ -880,7 +880,7 @@ public class RepairJobTest List> messageCapture) { MessagingService.instance().inboundSink.add(message -> message.verb().isResponse()); - MessagingService.instance().outboundSink.add((message, to) -> { + MessagingService.instance().outboundSink.add((message, to, type) -> { if (message == null || !(message.payload instanceof RepairMessage)) return false; diff --git a/test/unit/org/apache/cassandra/repair/ValidatorTest.java b/test/unit/org/apache/cassandra/repair/ValidatorTest.java index eaa60a0b7b..53b2435513 100644 --- a/test/unit/org/apache/cassandra/repair/ValidatorTest.java +++ b/test/unit/org/apache/cassandra/repair/ValidatorTest.java @@ -375,7 +375,7 @@ public class ValidatorTest private CompletableFuture registerOutgoingMessageSink() { final CompletableFuture future = new CompletableFuture<>(); - MessagingService.instance().outboundSink.add((message, to) -> future.complete(message)); + MessagingService.instance().outboundSink.add((message, to, type) -> future.complete(message)); return future; } }