natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Artificial Latency Injection 

patch by Benedict; reviewed by Alex Petrov and Ariel Weisberg for CASSANDRA-17024
This commit is contained in:
Benedict Elliott Smith 2021-01-19 18:23:27 +00:00 committed by Benedict Elliott Smith
parent 51a58ef22e
commit 8ad8e56378
35 changed files with 1035 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
.build/checkstyle.xml
View File

@ -58,15 +58,9 @@
</module>
<module name="SuppressWithNearbyCommentFilter">
<property name="commentFormat" value="checkstyle: permit this invocation"/>
<property name="idFormat" value="blockPathToFile"/>
<property name="influenceFormat" value="0"/>
</module>
<module name="SuppressWithNearbyCommentFilter">
<property name="commentFormat" value="checkstyle: permit this invocation"/>
<property name="idFormat" value="blockToCases"/>
<property name="influenceFormat" value="0"/>
<property name="commentFormat" value="checkstyle: permit this invocation"/>
<property name="idFormat" value="blockInstantNow|blockPathToFile|blockToCases"/>
<property name="influenceFormat" value="0"/>
</module>
<module name="RegexpSinglelineJava">

15
src/java/org/apache/cassandra/concurrent/ExecutorLocals.java
View File

@ -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<ExecutorLocals> locals = new FastThreadLocal<ExecutorLocals>()
{
@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()

1
src/java/org/apache/cassandra/config/AccordSpec.java
View File

@ -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;

4
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -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. */

23
src/java/org/apache/cassandra/cql3/QueryOptions.java
View File

@ -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);

35
src/java/org/apache/cassandra/db/ConsistencyLevel.java
View File

@ -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

8
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -925,12 +925,12 @@ public abstract class ReadCommand extends AbstractReadQuery
*/
public Message<ReadCommand> createMessage(boolean trackRepairedData, Dispatcher.RequestTime requestTime)
{
List<MessageFlag> 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,

4
src/java/org/apache/cassandra/locator/ReplicaPlans.java
View File

@ -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

397
src/java/org/apache/cassandra/net/ArtificialLatency.java Normal file
View File

@ -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<Verb> artificialLatencyVerbs;
private static volatile boolean artificialLatencyOnlyPermittedConsistencyLevels = true;
private static volatile ToLongFunction<InetAddressAndPort> 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<Delayed>
{
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<Delayed> 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<Delayed> out = new PriorityQueue<>();
final Interruptible executor = executorFactory().infiniteLoop("ArtificialLatency", this, SAFE, DAEMON, UNSYNCHRONIZED);
volatile Thread waiting;
volatile long waitingUntil;
private static final AtomicLongFieldUpdater<Sink> 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<String> 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<Verb> 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(","));
}
}

58
src/java/org/apache/cassandra/net/Message.java
View File

@ -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<T> 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<T> implements ResponseContext
return outWithParam(nextId(), verb, 0, payload, flag.addTo(0), null, null);
}
public static <T> Message<T> outWithFlags(Verb verb, T payload, MessageFlag flag1, MessageFlag flag2)
public static <T> Message<T> 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 <T> Message<T> outWithFlags(Verb verb, T payload, Dispatcher.RequestTime requestTime, List<MessageFlag> flags)
public static <T> Message<T> 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<T>(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 <T> Message<T> 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<T> 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<T> implements ResponseContext
public static <T> Message<T> 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<T> implements ResponseContext
private static Map<ParamType, Object> buildParams(ParamType type, Object value)
{
Map<ParamType, Object> params = NO_PARAMS;
EnumMap<ParamType, Object> params = NO_PARAMS;
if (Tracing.isTracing())
params = Tracing.instance.addTraceHeaders(new EnumMap<>(ParamType.class));
@ -595,6 +608,11 @@ public class Message<T> 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<T> implements ResponseContext
private InetAddressAndPort from;
private T payload;
private int flags = 0;
private final Map<ParamType, Object> params = new EnumMap<>(ParamType.class);
private final EnumMap<ParamType, Object> params = new EnumMap<>(ParamType.class);
private long createdAtNanos;
private long expiresAtNanos;
private long id;

13
src/java/org/apache/cassandra/net/MessageFlag.java
View File

@ -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);
}

106
src/java/org/apache/cassandra/net/OutboundSink.java
View File

@ -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<Message<?>, InetAddressAndPort> condition;
final Sink next;
private Filtered(BiPredicate<Message<?>, 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<Message<?>, InetAddressAndPort> allow)
public void add(Filter allow)
{
sinkUpdater.updateAndGet(this, sink -> new Filtered(allow, sink));
}
public void remove(BiPredicate<Message<?>, 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<Message<?>, 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<AbstractFiltered> 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);
}
}

1
src/java/org/apache/cassandra/net/ResponseContext.java
View File

@ -27,4 +27,5 @@ public interface ResponseContext extends ReplyContext
InetAddressAndPort from();
Verb verb();
long expiresAtNanos();
boolean hasFlag(MessageFlag flag);
}

2
src/java/org/apache/cassandra/service/ClientWarn.java
View File

@ -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)

30
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -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<InetAddressAndPort> replicas)
private static void sendCommit(Commit commit, ConsistencyLevel consistencyForPaxos, Iterable<InetAddressAndPort> replicas)
{
Message<Commit> message = Message.out(PAXOS_COMMIT_REQ, commit);
for (InetAddressAndPort target : replicas)
@ -796,7 +796,6 @@ public class StorageProxy implements StorageProxyMBean
Message<Commit> 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<Commit> 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); }

9
src/java/org/apache/cassandra/service/StorageProxyMBean.java
View File

@ -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();

1
src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
View File

@ -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);
}
}

5
src/java/org/apache/cassandra/tracing/Tracing.java
View File

@ -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<String, String> parameters)
@ -303,7 +304,7 @@ public abstract class Tracing extends ExecutorLocals.Impl
}
}
public Map<ParamType, Object> addTraceHeaders(Map<ParamType, Object> addToMutable)
public EnumMap<ParamType, Object> addTraceHeaders(EnumMap<ParamType, Object> addToMutable)
{
assert isTracing();

4
src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
View File

@ -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());

6
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -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)
{

501
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

@ -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<Verb, AtomicInteger> 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<java.util.concurrent.Future<?>> 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());
}
}

3
test/distributed/org/apache/cassandra/distributed/test/log/ClusterMetadataTestHelper.java
View File

@ -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<MessageDelivery> future = SettableFuture.create();
Set<Verb> 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));

2
test/unit/org/apache/cassandra/concurrent/LocalAwareExecutorPlusTest.java
View File

@ -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

2
test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerOutOfRangeTest.java
View File

@ -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);

2
test/unit/org/apache/cassandra/db/ReadCommandVerbHandlerTest.java
View File

@ -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();

2
test/unit/org/apache/cassandra/db/compaction/AbstractPendingRepairTest.java
View File

@ -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);
}

2
test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java
View File

@ -505,7 +505,7 @@ public class CompactionIteratorTest extends CQLTester
TableMetadata metadata = getCurrentColumnFamilyStore().metadata();
final HashMap<InetAddressAndPort, Message<?>> 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();

4
test/unit/org/apache/cassandra/db/compression/CompressionDictionaryEventHandlerTest.java
View File

@ -135,7 +135,7 @@ public class CompressionDictionaryEventHandlerTest
AtomicReference<CompressionDictionaryUpdateMessage> 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");

2
test/unit/org/apache/cassandra/db/transform/DuplicateRowCheckerTest.java
View File

@ -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

7
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -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<Message<?>, 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;

8
test/unit/org/apache/cassandra/net/MatcherResponse.java
View File

@ -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<Message<?>, 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<Message<?>, 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

4
test/unit/org/apache/cassandra/net/MessageTest.java
View File

@ -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()));
}

5
test/unit/org/apache/cassandra/net/StartupClusterConnectivityCheckerTest.java
View File

@ -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<Message<?>, 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());

2
test/unit/org/apache/cassandra/repair/RepairJobTest.java
View File

@ -880,7 +880,7 @@ public class RepairJobTest
List<Message<?>> 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;

2
test/unit/org/apache/cassandra/repair/ValidatorTest.java
View File

@ -375,7 +375,7 @@ public class ValidatorTest
private CompletableFuture<Message> registerOutgoingMessageSink()
{
final CompletableFuture<Message> future = new CompletableFuture<>();
MessagingService.instance().outboundSink.add((message, to) -> future.complete(message));
MessagingService.instance().outboundSink.add((message, to, type) -> future.complete(message));
return future;
}
}