ContentionStrategy.java
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.paxos;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.codahale.metrics.Snapshot;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.NoSpamLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.DoubleSupplier;
import java.util.function.LongBinaryOperator;
import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import static java.lang.Double.parseDouble;
import static java.lang.Integer.parseInt;
import static java.lang.Math.*;
import static java.util.Arrays.stream;
import static java.util.concurrent.TimeUnit.*;
import static org.apache.cassandra.config.DatabaseDescriptor.*;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casReadMetrics;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casWriteMetrics;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Clock.waitUntil;
/**
* <p>A strategy for making back-off decisions for Paxos operations that fail to make progress because of other paxos operations.
* The strategy is defined by four factors: <ul>
* <li> {@link #min}
* <li> {@link #max}
* <li> {@link #minDelta}
* <li> {@link #waitRandomizer}
* </ul>
*
* <p>The first three represent time periods, and may be defined dynamically based on a simple calculation over: <ul>
* <li> {@code pX()} recent experienced latency distribution for successful operations,
* e.g. {@code p50(rw)} the maximum of read and write median latencies,
* {@code p999(r)} the 99.9th percentile of read latencies
* <li> {@code attempts} the number of failed attempts made by the operation so far
* <li> {@code constant} a user provided floating point constant
* </ul>
*
* <p>Their calculation may take any of these forms
* <li> constant {@code $constant$[mu]s}
* <li> dynamic constant {@code pX() * constant}
* <li> dynamic linear {@code pX() * constant * attempts}
* <li> dynamic exponential {@code pX() * constant ^ attempts}
*
* <p>Furthermore, the dynamic calculations can be bounded with a min/max, like so:
* {@code min[mu]s <= dynamic expr <= max[mu]s}
*
* e.g.
* <li> {@code 10ms <= p50(rw)*0.66}
* <li> {@code 10ms <= p95(rw)*1.8^attempts <= 100ms}
* <li> {@code 5ms <= p50(rw)*0.5}
*
* <p>These calculations are put together to construct a range from which we draw a random number.
* The period we wait for {@code X} will be drawn so that {@code min <= X < max}.
*
* <p>With the constraint that {@code max} must be {@code minDelta} greater than {@code min},
* but no greater than its expression-defined maximum. {@code max} will be increased up until
* this point, after which {@code min} will be decreased until this gap is imposed.
*
* <p>The {@link #waitRandomizer} property specifies the manner in which a random value is drawn from the range.
* It is defined using one of the following specifiers:
* <li> uniform
* <li> exp($power$) or exponential($power$)
* <li> qexp($power$) or qexponential($power$) or quantizedexponential($power$)
*
* The uniform specifier is self-explanatory, selecting all values in the range with equal probability.
* The exponential specifier draws values towards the end of the range with higher probability, raising
* a floating point number in the range [0..1.0) to the power provided, and translating the resulting value
* to a uniform value in the range.
* The quantized exponential specifier partitions the range into {@code attempts} buckets, then applies the pure
* exponential approach to draw values from [0..attempts), before drawing a uniform value from the corresponding bucket
*
* <p>Finally, there is also a {@link #traceAfterAttempts} property that permits initiating tracing of operations
* that experience a certain minimum number of failed paxos rounds due to contention. A setting of 0 or 1 will initiate
* a trace session after the first failed ballot.
*/
public class ContentionStrategy
{
private static final Logger logger = LoggerFactory.getLogger(ContentionStrategy.class);
private static final Pattern BOUND = Pattern.compile(
"(?<const>0|[0-9]+[mu]s)" +
"|((?<min>0|[0-9]+[mu]s) *<= *)?" +
"(p(?<perc>[0-9]+)\\((?<rw>r|w|rw|wr)\\)|(?<constbase>0|[0-9]+[mu]s))" +
"\\s*([*]\\s*(?<mod>[0-9.]+)?\\s*(?<modkind>[*^]\\s*attempts)?)?" +
"( *<= *(?<max>0|[0-9]+[mu]s))?");
private static final Pattern TIME = Pattern.compile(
"0|([0-9]+)ms|([0-9]+)us");
private static final Pattern RANDOMIZER = Pattern.compile(
"uniform|exp(onential)?[(](?<exp>[0-9.]+)[)]|q(uantized)?exp(onential)?[(](?<qexp>[0-9.]+)[)]");
private static final String DEFAULT_WAIT_RANDOMIZER = "qexp(1.5)"; // at least 0ms, and at least 66% of median latency
private static final String DEFAULT_MIN = "0 <= p50(rw)*0.66"; // at least 0ms, and at least 66% of median latency
private static final String DEFAULT_MAX = "10ms <= p95(rw)*1.8^attempts <= 100ms"; // p95 latency with exponential back-off at rate of 1.8^attempts
private static final String DEFAULT_MIN_DELTA = "5ms <= p50(rw)*0.5"; // at least 5ms, and at least 50% of median latency
private static volatile ContentionStrategy current;
// Factories can be useful for testing purposes, to supply custom implementations of selectors and modifiers.
final static LatencySelectorFactory selectors = new LatencySelectorFactory(){};
final static LatencyModifierFactory modifiers = new LatencyModifierFactory(){};
final static WaitRandomizerFactory randomizers = new WaitRandomizerFactory(){};
static
{
current = new ContentionStrategy(defaultWaitRandomizer(), defaultMinWait(), defaultMaxWait(), defaultMinDelta(), Integer.MAX_VALUE);
}
static interface LatencyModifierFactory
{
default LatencyModifier identity() { return (l, a) -> l; }
default LatencyModifier multiply(double constant) { return (l, a) -> saturatedCast(l * constant); }
default LatencyModifier multiplyByAttempts(double multiply) { return (l, a) -> saturatedCast(l * multiply * a); }
default LatencyModifier multiplyByAttemptsExp(double base) { return (l, a) -> saturatedCast(l * pow(base, a)); }
}
static interface LatencySupplier
{
abstract long get(double percentile);
}
static interface LatencySelector
{
abstract long select(LatencySupplier readLatencyHistogram, LatencySupplier writeLatencyHistogram);
}
static interface LatencySelectorFactory
{
default LatencySelector constant(long latency) { return (read, write) -> latency; }
default LatencySelector read(double percentile) { return (read, write) -> read.get(percentile); }
default LatencySelector write(double percentile) { return (read, write) -> write.get(percentile); }
default LatencySelector maxReadWrite(double percentile) { return (read, write) -> max(read.get(percentile), write.get(percentile)); }
}
static interface LatencyModifier
{
long modify(long latency, int attempts);
}
static interface WaitRandomizer
{
abstract long wait(long min, long max, int attempts);
}
static interface WaitRandomizerFactory
{
default LongBinaryOperator uniformLongSupplier() { return (min, max) -> ThreadLocalRandom.current().nextLong(min, max); } // DO NOT USE METHOD HANDLES (want to fetch afresh each time)
default DoubleSupplier uniformDoubleSupplier() { return () -> ThreadLocalRandom.current().nextDouble(); }
default WaitRandomizer uniform() { return new Uniform(uniformLongSupplier()); }
default WaitRandomizer exponential(double power) { return new Exponential(uniformLongSupplier(), uniformDoubleSupplier(), power); }
default WaitRandomizer quantizedExponential(double power) { return new QuantizedExponential(uniformLongSupplier(), uniformDoubleSupplier(), power); }
static class Uniform implements WaitRandomizer
{
final LongBinaryOperator uniformLong;
public Uniform(LongBinaryOperator uniformLong)
{
this.uniformLong = uniformLong;
}
@Override
public long wait(long min, long max, int attempts)
{
return uniformLong.applyAsLong(min, max);
}
}
static abstract class AbstractExponential implements WaitRandomizer
{
final LongBinaryOperator uniformLong;
final DoubleSupplier uniformDouble;
final double power;
public AbstractExponential(LongBinaryOperator uniformLong, DoubleSupplier uniformDouble, double power)
{
this.uniformLong = uniformLong;
this.uniformDouble = uniformDouble;
this.power = power;
}
}
static class Exponential extends AbstractExponential
{
public Exponential(LongBinaryOperator uniformLong, DoubleSupplier uniformDouble, double power)
{
super(uniformLong, uniformDouble, power);
}
@Override
public long wait(long min, long max, int attempts)
{
if (attempts == 1)
return uniformLong.applyAsLong(min, max);
double p = uniformDouble.getAsDouble();
long delta = max - min;
delta *= Math.pow(p, power);
return max - delta;
}
}
static class QuantizedExponential extends AbstractExponential
{
public QuantizedExponential(LongBinaryOperator uniformLong, DoubleSupplier uniformDouble, double power)
{
super(uniformLong, uniformDouble, power);
}
@Override
public long wait(long min, long max, int attempts)
{
long quanta = (max - min) / attempts;
if (attempts == 1 || quanta == 0)
return uniformLong.applyAsLong(min, max);
double p = uniformDouble.getAsDouble();
int base = (int) (attempts * Math.pow(p, power));
return max - ThreadLocalRandom.current().nextLong(quanta * base, quanta * (base + 1));
}
}
}
static class SnapshotAndTime
{
final long validUntil;
final Snapshot snapshot;
SnapshotAndTime(long validUntil, Snapshot snapshot)
{
this.validUntil = validUntil;
this.snapshot = snapshot;
}
}
static class TimeLimitedLatencySupplier extends AtomicReference<SnapshotAndTime> implements LatencySupplier
{
final Supplier<Snapshot> snapshotSupplier;
final long validForNanos;
TimeLimitedLatencySupplier(Supplier<Snapshot> snapshotSupplier, long time, TimeUnit units)
{
this.snapshotSupplier = snapshotSupplier;
this.validForNanos = units.toNanos(time);
}
private Snapshot getSnapshot()
{
long now = nanoTime();
SnapshotAndTime cur = get();
if (cur != null && cur.validUntil > now)
return cur.snapshot;
Snapshot newSnapshot = snapshotSupplier.get();
SnapshotAndTime next = new SnapshotAndTime(now + validForNanos, newSnapshot);
if (compareAndSet(cur, next))
return next.snapshot;
return accumulateAndGet(next, (a, b) -> a.validUntil > b.validUntil ? a : b).snapshot;
}
@Override
public long get(double percentile)
{
return (long)getSnapshot().getValue(percentile);
}
}
static class Bound
{
final long min, max, onFailure;
final LatencyModifier modifier;
final LatencySelector selector;
final LatencySupplier reads, writes;
Bound(long min, long max, long onFailure, LatencyModifier modifier, LatencySelector selector)
{
Preconditions.checkArgument(min<=max, "min (%s) must be less than or equal to max (%s)", min, max);
this.min = min;
this.max = max;
this.onFailure = onFailure;
this.modifier = modifier;
this.selector = selector;
this.reads = new TimeLimitedLatencySupplier(casReadMetrics.latency::getSnapshot, 10L, SECONDS);
this.writes = new TimeLimitedLatencySupplier(casWriteMetrics.latency::getSnapshot, 10L, SECONDS);
}
long get(int attempts)
{
try
{
long base = selector.select(reads, writes);
return max(min, min(max, modifier.modify(base, attempts)));
}
catch (Throwable t)
{
NoSpamLogger.getLogger(logger, 1L, MINUTES).info("", t);
return onFailure;
}
}
public String toString()
{
return "Bound{" +
"min=" + min +
", max=" + max +
", onFailure=" + onFailure +
", modifier=" + modifier +
", selector=" + selector +
'}';
}
}
final WaitRandomizer waitRandomizer;
final Bound min, max, minDelta;
final int traceAfterAttempts;
public ContentionStrategy(String waitRandomizer, String min, String max, String minDelta, int traceAfterAttempts)
{
this.waitRandomizer = parseWaitRandomizer(waitRandomizer);
this.min = parseBound(min, true);
this.max = parseBound(max, false);
this.minDelta = parseBound(minDelta, true);
this.traceAfterAttempts = traceAfterAttempts;
}
public enum Type
{
READ("Contended Paxos Read"), WRITE("Contended Paxos Write"), REPAIR("Contended Paxos Repair");
final String traceTitle;
final String lowercase;
Type(String traceTitle)
{
this.traceTitle = traceTitle;
this.lowercase = name().toLowerCase();
}
}
long computeWaitUntilForContention(int attempts, TableMetadata table, DecoratedKey partitionKey, ConsistencyLevel consistency, Type type)
{
if (attempts >= traceAfterAttempts && !Tracing.isTracing())
{
Tracing.instance.newSession(Tracing.TraceType.QUERY);
Tracing.instance.begin(type.traceTitle,
ImmutableMap.of(
"keyspace", table.keyspace,
"table", table.name,
"partitionKey", table.partitionKeyType.getString(partitionKey.getKey()),
"consistency", consistency.name(),
"kind", type.lowercase
));
logger.info("Tracing contended paxos {} for key {} on {}.{} with trace id {}",
type.lowercase,
ByteBufferUtil.bytesToHex(partitionKey.getKey()),
table.keyspace, table.name,
Tracing.instance.getSessionId());
}
long minWaitMicros = min.get(attempts);
long maxWaitMicros = max.get(attempts);
long minDeltaMicros = minDelta.get(attempts);
if (minWaitMicros + minDeltaMicros > maxWaitMicros)
{
maxWaitMicros = minWaitMicros + minDeltaMicros;
if (maxWaitMicros > this.max.max)
{
maxWaitMicros = this.max.max;
minWaitMicros = max(this.min.min, min(this.min.max, maxWaitMicros - minDeltaMicros));
}
}
long wait = waitRandomizer.wait(minWaitMicros, maxWaitMicros, attempts);
return nanoTime() + MICROSECONDS.toNanos(wait);
}
boolean doWaitForContention(long deadline, int attempts, TableMetadata table, DecoratedKey partitionKey, ConsistencyLevel consistency, Type type)
{
long until = computeWaitUntilForContention(attempts, table, partitionKey, consistency, type);
if (until >= deadline)
return false;
try
{
waitUntil(until);
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
return false;
}
return true;
}
static boolean waitForContention(long deadline, int attempts, TableMetadata table, DecoratedKey partitionKey, ConsistencyLevel consistency, Type type)
{
return current.doWaitForContention(deadline, attempts, table, partitionKey, consistency, type);
}
static long waitUntilForContention(int attempts, TableMetadata table, DecoratedKey partitionKey, ConsistencyLevel consistency, Type type)
{
return current.computeWaitUntilForContention(attempts, table, partitionKey, consistency, type);
}
static class ParsedStrategy
{
final String waitRandomizer, min, max, minDelta;
final ContentionStrategy strategy;
ParsedStrategy(String waitRandomizer, String min, String max, String minDelta, ContentionStrategy strategy)
{
this.waitRandomizer = waitRandomizer;
this.min = min;
this.max = max;
this.minDelta = minDelta;
this.strategy = strategy;
}
}
@VisibleForTesting
static ParsedStrategy parseStrategy(String spec)
{
String[] args = spec.split(",");
String waitRandomizer = find(args, "random");
String min = find(args, "min");
String max = find(args, "max");
String minDelta = find(args, "delta");
String trace = find(args, "trace");
if (waitRandomizer == null) waitRandomizer = defaultWaitRandomizer();
if (min == null) min = defaultMinWait();
if (max == null) max = defaultMaxWait();
if (minDelta == null) minDelta = defaultMinDelta();
int traceAfterAttempts = trace == null ? current.traceAfterAttempts: Integer.parseInt(trace);
ContentionStrategy strategy = new ContentionStrategy(waitRandomizer, min, max, minDelta, traceAfterAttempts);
return new ParsedStrategy(waitRandomizer, min, max, minDelta, strategy);
}
public static void setStrategy(String spec)
{
ParsedStrategy parsed = parseStrategy(spec);
current = parsed.strategy;
setPaxosContentionWaitRandomizer(parsed.waitRandomizer);
setPaxosContentionMinWait(parsed.min);
setPaxosContentionMaxWait(parsed.max);
setPaxosContentionMinDelta(parsed.minDelta);
}
public static String getStrategySpec()
{
return "min=" + defaultMinWait()
+ ",max=" + defaultMaxWait()
+ ",delta=" + defaultMinDelta()
+ ",random=" + defaultWaitRandomizer()
+ ",trace=" + current.traceAfterAttempts;
}
private static String find(String[] args, String param)
{
return stream(args).filter(s -> s.startsWith(param + '='))
.map(s -> s.substring(param.length() + 1))
.findFirst().orElse(null);
}
private static LatencySelector parseLatencySelector(Matcher m, LatencySelectorFactory selectors)
{
String perc = m.group("perc");
if (perc == null)
return selectors.constant(parseInMicros(m.group("constbase")));
double percentile = parseDouble("0." + perc);
String rw = m.group("rw");
if (rw.length() == 2)
return selectors.maxReadWrite(percentile);
else if ("r".equals(rw))
return selectors.read(percentile);
else
return selectors.write(percentile);
}
private static LatencyModifier parseLatencyModifier(Matcher m, LatencyModifierFactory modifiers)
{
String mod = m.group("mod");
if (mod == null)
return modifiers.identity();
double modifier = parseDouble(mod);
String modkind = m.group("modkind");
if (modkind == null)
return modifiers.multiply(modifier);
if (modkind.startsWith("*"))
return modifiers.multiplyByAttempts(modifier);
else if (modkind.startsWith("^"))
return modifiers.multiplyByAttemptsExp(modifier);
else
throw new IllegalArgumentException("Unrecognised attempt modifier: " + modkind);
}
static long saturatedCast(double v)
{
if (v > Long.MAX_VALUE)
return Long.MAX_VALUE;
return (long) v;
}
static WaitRandomizer parseWaitRandomizer(String input)
{
return parseWaitRandomizer(input, randomizers);
}
static WaitRandomizer parseWaitRandomizer(String input, WaitRandomizerFactory randomizers)
{
Matcher m = RANDOMIZER.matcher(input);
if (!m.matches())
throw new IllegalArgumentException(input + " does not match" + RANDOMIZER);
String exp;
exp = m.group("exp");
if (exp != null)
return randomizers.exponential(Double.parseDouble(exp));
exp = m.group("qexp");
if (exp != null)
return randomizers.quantizedExponential(Double.parseDouble(exp));
return randomizers.uniform();
}
static Bound parseBound(String input, boolean isMin)
{
return parseBound(input, isMin, selectors, modifiers);
}
@VisibleForTesting
static Bound parseBound(String input, boolean isMin, LatencySelectorFactory selectors, LatencyModifierFactory modifiers)
{
Matcher m = BOUND.matcher(input);
if (!m.matches())
throw new IllegalArgumentException(input + " does not match " + BOUND);
String maybeConst = m.group("const");
if (maybeConst != null)
{
long v = parseInMicros(maybeConst);
return new Bound(v, v, v, modifiers.identity(), selectors.constant(v));
}
long min = parseInMicros(m.group("min"), 0);
long max = parseInMicros(m.group("max"), maxQueryTimeoutMicros() / 2);
return new Bound(min, max, isMin ? min : max, parseLatencyModifier(m, modifiers), parseLatencySelector(m, selectors));
}
private static long parseInMicros(String input, long orElse)
{
if (input == null)
return orElse;
return parseInMicros(input);
}
private static long parseInMicros(String input)
{
Matcher m = TIME.matcher(input);
if (!m.matches())
throw new IllegalArgumentException(input + " does not match " + TIME);
String text;
if (null != (text = m.group(1)))
return parseInt(text) * 1000;
else if (null != (text = m.group(2)))
return parseInt(text);
else
return 0;
}
@VisibleForTesting
static String defaultWaitRandomizer()
{
return orElse(DatabaseDescriptor::getPaxosContentionWaitRandomizer, DEFAULT_WAIT_RANDOMIZER);
}
@VisibleForTesting
static String defaultMinWait()
{
return orElse(DatabaseDescriptor::getPaxosContentionMinWait, DEFAULT_MIN);
}
@VisibleForTesting
static String defaultMaxWait()
{
return orElse(DatabaseDescriptor::getPaxosContentionMaxWait, DEFAULT_MAX);
}
@VisibleForTesting
static String defaultMinDelta()
{
return orElse(DatabaseDescriptor::getPaxosContentionMinDelta, DEFAULT_MIN_DELTA);
}
@VisibleForTesting
static long maxQueryTimeoutMicros()
{
return max(max(getCasContentionTimeout(MICROSECONDS), getWriteRpcTimeout(MICROSECONDS)), getReadRpcTimeout(MICROSECONDS));
}
private static String orElse(Supplier<String> get, String orElse)
{
String result = get.get();
return result != null ? result : orElse;
}
}