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

Merge branch 'cassandra-4.1' into cassandra-5.0

This commit is contained in:
Maxim Muzafarov 2024-10-02 15:28:44 +02:00
parent 6845359762 ad4ad5ef5a
commit 9dfcfaee65
4 changed files with 319 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,6 +1,7 @@
5.0.2
* Correct out-of-date metrics and configuration documentation for SAI (CASSANDRA-19898)
Merged from 4.1:
* Fix race condition in DecayingEstimatedHistogramReservoir during rescale (CASSANDRA-19365)
Merged from 4.0:
* Ensure thread-safety for CommitLogArchiver in CommitLog (CASSANDRA-19960)
* Fix text containing "/*" being interpreted as multiline comment in cqlsh (CASSANDRA-17667)

178
src/java/org/apache/cassandra/metrics/DecayingEstimatedHistogramReservoir.java
View File

@ -25,15 +25,15 @@ import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLongArray;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.ExponentiallyDecayingReservoir;
import com.codahale.metrics.Reservoir;
import com.codahale.metrics.Snapshot;
import org.apache.cassandra.utils.EstimatedHistogram;
@ -52,9 +52,8 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.DECAYING_E
* <p/>
* The histogram use forward decay [1] to make recent values more significant. The forward decay factor will be doubled
* every minute (half-life time set to 60 seconds) [2]. The forward decay landmark is reset every 30 minutes (or at
* first read/update after 30 minutes). During landmark reset, updates and reads in the reservoir will be blocked in a
* fashion similar to the one used in the metrics library [3]. The 30 minute rescale interval is used based on the
* assumption that in an extreme case we would have to collect a metric 1M times for a single bucket each second. By the
* first read/update after 30 minutes). The 30 minute rescale interval is used based on the assumption that in an
* extreme case we would have to collect a metric 1M times for a single bucket each second. By the
* end of the 30:th minute all collected values will roughly add up to 1.000.000 * 60 * pow(2, 30) which can be
* represented with 56 bits giving us some head room in a signed 64 bit long.
* <p/>
@ -82,6 +81,8 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.DECAYING_E
* <li>[2]: https://en.wikipedia.org/wiki/Half-life</li>
* <li>[3]: https://github.com/dropwizard/metrics/blob/v3.1.2/metrics-core/src/main/java/com/codahale/metrics/ExponentiallyDecayingReservoir.java</li>
* </ul>
*
* @see ExponentiallyDecayingReservoir
*/
public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoir
{
@ -149,19 +150,20 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
private final long[] bucketOffsets;
private final int distributionPrime;
private static final AtomicReferenceFieldUpdater<DecayingEstimatedHistogramReservoir, DecayingBuckets> decayingBucketsUpdater =
AtomicReferenceFieldUpdater.newUpdater(DecayingEstimatedHistogramReservoir.class, DecayingBuckets.class, "decayingBuckets");
// decayingBuckets and buckets are one element longer than bucketOffsets -- the last element is values greater than the last offset
private final AtomicLongArray decayingBuckets;
private volatile DecayingBuckets decayingBuckets;
private final AtomicLongArray buckets;
public static final long HALF_TIME_IN_S = 60L;
public static final double MEAN_LIFETIME_IN_S = HALF_TIME_IN_S / Math.log(2.0);
public static final long LANDMARK_RESET_INTERVAL_IN_NS = TimeUnit.MINUTES.toNanos(30L);
private final AtomicBoolean rescaling = new AtomicBoolean(false);
private volatile long decayLandmark;
// Wrapper around System.nanoTime() to simplify unit testing.
private final MonotonicClock clock;
/** Interval in minutes to reset the forward decay landmark for the decaying histograms. Default {@code 30 mins}. */
private final long landmarkResetIntervalInNs;
/**
* Construct a decaying histogram with default number of buckets and without considering zeroes.
@ -202,6 +204,16 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
@VisibleForTesting
DecayingEstimatedHistogramReservoir(boolean considerZeroes, int bucketCount, int stripes, MonotonicClock clock)
{
this(considerZeroes, bucketCount, stripes, clock, LANDMARK_RESET_INTERVAL_IN_NS);
}
@VisibleForTesting
public DecayingEstimatedHistogramReservoir(boolean considerZeroes,
int bucketCount,
int stripes,
MonotonicClock clock,
long landmarkResetIntervalInNs)
{
assert bucketCount <= MAX_BUCKET_COUNT : "bucket count cannot exceed: " + MAX_BUCKET_COUNT;
@ -222,10 +234,10 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
}
nStripes = stripes;
decayingBuckets = new AtomicLongArray((bucketOffsets.length + 1) * nStripes);
buckets = new AtomicLongArray((bucketOffsets.length + 1) * nStripes);
this.clock = clock;
decayLandmark = clock.now();
buckets = new AtomicLongArray((bucketOffsets.length + 1) * nStripes);
decayingBuckets = new DecayingBuckets(clock.now());
this.landmarkResetIntervalInNs = landmarkResetIntervalInNs;
int distributionPrime = 1;
for (int prime : DISTRIBUTION_PRIMES)
{
@ -246,11 +258,11 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
public void update(long value)
{
long now = clock.now();
rescaleIfNeeded(now);
DecayingBuckets rescaledDecayingBuckets = rescaleIfNeeded(now);
int index = findIndex(bucketOffsets, value);
updateBucket(decayingBuckets, index, Math.round(forwardDecayWeight(now)));
rescaledDecayingBuckets.update(index, now);
updateBucket(buckets, index, 1);
}
@ -288,11 +300,6 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
return value <= bucketOffsets[firstCandidate] ? firstCandidate : firstCandidate + 1;
}
private double forwardDecayWeight(long now)
{
return Math.exp(TimeUnit.NANOSECONDS.toSeconds(now - decayLandmark) / MEAN_LIFETIME_IN_S);
}
/**
* Returns the logical number of buckets where recorded values are stored. The actual number of physical buckets
* is size() * stripeCount()
@ -322,32 +329,34 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
@Override
public Snapshot getSnapshot()
{
rescaleIfNeeded();
return new EstimatedHistogramReservoirSnapshot(this);
}
@Override
public Snapshot getPercentileSnapshot()
{
rescaleIfNeeded();
return new DecayingBucketsOnlySnapshot(this);
}
private DecayingBuckets getDecayingBuckets()
{
return rescaleIfNeeded(clock.now());
}
/**
* @return true if this histogram has overflowed -- that is, a value larger than our largest bucket could bound was added
*/
@VisibleForTesting
boolean isOverflowed()
{
return bucketValue(bucketOffsets.length, true) > 0;
return bucketValue(bucketOffsets.length, getDecayingBuckets().decayBuckets) > 0;
}
private long bucketValue(int index, boolean withDecay)
private long bucketValue(int index, AtomicLongArray buckets)
{
long val = 0;
AtomicLongArray bs = withDecay ? decayingBuckets : buckets;
for (int stripe = 0; stripe < nStripes; stripe++)
val += bs.get(stripedIndex(index, stripe));
val += buckets.get(stripedIndex(index, stripe));
return val;
}
@ -355,65 +364,41 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
@VisibleForTesting
long stripedBucketValue(int i, boolean withDecay)
{
return withDecay ? decayingBuckets.get(i) : buckets.get(i);
return withDecay ? getDecayingBuckets().decayBuckets.get(i) : buckets.get(i);
}
private void rescaleIfNeeded()
private DecayingBuckets rescaleIfNeeded(long now)
{
rescaleIfNeeded(clock.now());
}
private void rescaleIfNeeded(long now)
{
if (needRescale(now))
DecayingBuckets buckets = decayingBuckets;
while (now - buckets.decayLandmark > landmarkResetIntervalInNs)
{
if (rescaling.compareAndSet(false, true))
{
try
{
rescale(now);
}
finally
{
decayLandmark = now;
rescaling.set(false);
}
}
}
}
double rescaleFactor = buckets.forwardDecayWeight(now);
DecayingBuckets newBuckets = new DecayingBuckets(now);
for (int i = 0; i < buckets.decayBuckets.length(); i++)
newBuckets.decayBuckets.set(i, Math.round(buckets.decayBuckets.get(i) / rescaleFactor));
private void rescale(long now)
{
// despite striping its safe to rescale each bucket individually
final double rescaleFactor = forwardDecayWeight(now);
for (int i = 0; i < decayingBuckets.length(); i++)
{
long newValue = Math.round(decayingBuckets.get(i) / rescaleFactor);
decayingBuckets.set(i, newValue);
boolean success = decayingBucketsUpdater.compareAndSet(this, buckets, newBuckets);
if (success)
return newBuckets;
buckets = decayingBuckets;
}
}
private boolean needRescale(long now)
{
return (now - decayLandmark) > LANDMARK_RESET_INTERVAL_IN_NS;
return buckets;
}
@VisibleForTesting
public void clear()
{
final int bucketCount = decayingBuckets.length();
for (int i = 0; i < bucketCount; i++)
{
decayingBuckets.set(i, 0L);
for (int i = 0; i < buckets.length(); i++)
buckets.set(i, 0L);
}
decayingBucketsUpdater.set(this, new DecayingBuckets(clock.now()));
}
/**
* Replaces current internal values with the given one from a Snapshot. This method is NOT thread safe, values
* added at the same time to this reservoir using methods such as update may lose their data
*/
public void rebase(EstimatedHistogramReservoirSnapshot snapshot)
private void rebase(EstimatedHistogramReservoirSnapshot snapshot)
{
// Check bucket count (a snapshot always has one stripe so the logical bucket count is used
if (size() != snapshot.decayingBuckets.length)
@ -430,19 +415,54 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
}
}
this.decayLandmark = snapshot.snapshotLandmark;
DecayingBuckets newDecayingBuckets = new DecayingBuckets(snapshot.snapshotLandmark);
for (int i = 0; i < size(); i++)
{
// set rebased values in the first stripe and clear out all other data
decayingBuckets.set(stripedIndex(i, 0), snapshot.decayingBuckets[i]);
newDecayingBuckets.decayBuckets.set(stripedIndex(i, 0), snapshot.decayingBuckets[i]);
buckets.set(stripedIndex(i, 0), snapshot.values[i]);
for (int stripe = 1; stripe < nStripes; stripe++)
{
decayingBuckets.set(stripedIndex(i, stripe), 0);
newDecayingBuckets.decayBuckets.set(stripedIndex(i, stripe), 0);
buckets.set(stripedIndex(i, stripe), 0);
}
}
decayingBucketsUpdater.set(this, newDecayingBuckets);
}
/**
* The DecayingBuckets class provides a facility to prevent destruction of the reservoir internal state from
* occurring. The root cause of CASSANDRA-19365 is lack of synchronization between udpates and rescaling.
* This class lets us retain lack of synchronization (for performance reasons) while changing the race condition
* effect from destructive to benign.
* <p>
* DecayingBuckets class encapsulates the decaying buckets and the decay landmark together. The decay landmark is
* immutable and so the values in the buckets are consistent with the landmark at any given time. In particular,
* every update is always given a weight that's consistent with weights given to other updates of the same buckets.
* <p>
* Additionally, the class allows creating snapshots of the reservoir without risking that the snapshot uses data
* from a half-rescaled reservoir.
*/
private class DecayingBuckets
{
private final long decayLandmark;
private final AtomicLongArray decayBuckets;
public DecayingBuckets(long decayLandmark)
{
this.decayLandmark = decayLandmark;
this.decayBuckets = new AtomicLongArray((bucketOffsets.length + 1) * nStripes);
}
public void update(int index, long now)
{
updateBucket(decayBuckets, index, forwardDecayWeight(now));
}
private long forwardDecayWeight(long now)
{
return Math.round(Math.exp(TimeUnit.NANOSECONDS.toSeconds(now - decayLandmark) / MEAN_LIFETIME_IN_S));
}
}
private static abstract class AbstractSnapshot extends Snapshot
@ -649,16 +669,19 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
super(reservoir);
int length = reservoir.size();
double rescaleFactor = reservoir.forwardDecayWeight(reservoir.clock.now());
this.values = new long[length];
this.snapshotLandmark = reservoir.decayLandmark;
DecayingBuckets decayingBucketsRef = reservoir.getDecayingBuckets();
this.snapshotLandmark = decayingBucketsRef.decayLandmark;
double rescaleFactor = decayingBucketsRef.forwardDecayWeight(reservoir.clock.now());
for (int i = 0; i < length; i++)
{
this.decayingBuckets[i] = Math.round(reservoir.bucketValue(i, true) / rescaleFactor);
this.values[i] = reservoir.bucketValue(i, false);
this.decayingBuckets[i] = Math.round(reservoir.bucketValue(i, decayingBucketsRef.decayBuckets) / rescaleFactor);
this.values[i] = reservoir.bucketValue(i, reservoir.buckets);
}
this.count = count();
this.reservoir = reservoir;
}
@ -777,11 +800,12 @@ public class DecayingEstimatedHistogramReservoir implements SnapshottingReservoi
super(reservoir);
int length = reservoir.size();
double rescaleFactor = reservoir.forwardDecayWeight(reservoir.clock.now());
DecayingBuckets decayingBucketsRef = reservoir.getDecayingBuckets();
double rescaleFactor = decayingBucketsRef.forwardDecayWeight(reservoir.clock.now());
for (int i = 0; i < length; i++)
{
this.decayingBuckets[i] = Math.round(reservoir.bucketValue(i, true) / rescaleFactor);
this.decayingBuckets[i] = Math.round(reservoir.bucketValue(i, decayingBucketsRef.decayBuckets) / rescaleFactor);
}
this.count = count();

97
test/microbench/org/apache/cassandra/test/microbench/DecayingEstimatedHistogramBench.java Normal file
View File

@ -0,0 +1,97 @@
/*
* 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.test.microbench;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir;
import org.apache.cassandra.utils.MonotonicClock;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.RunnerException;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.DEFAULT_BUCKET_COUNT;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.DEFAULT_STRIPE_COUNT;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.DEFAULT_ZERO_CONSIDERATION;
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 4, time = 2, timeUnit = TimeUnit.SECONDS)
@Threads(4)
@Fork(value = 3)
@State(Scope.Benchmark)
public class DecayingEstimatedHistogramBench
{
@Param({ "100000", "500000", "1000000" })
private long landmarkResetIntervalNs;
DecayingEstimatedHistogramReservoir reservoir;
@Setup
public void setup()
{
reservoir = new DecayingEstimatedHistogramReservoir(DEFAULT_ZERO_CONSIDERATION,
DEFAULT_BUCKET_COUNT,
DEFAULT_STRIPE_COUNT,
MonotonicClock.Global.approxTime,
landmarkResetIntervalNs);
}
@State(Scope.Thread)
public static class HistogramUpdateState
{
int update;
@Setup(Level.Invocation)
public void setup() throws Throwable
{
update = ThreadLocalRandom.current().nextInt(10, 1000);
}
}
@Benchmark
public void update(HistogramUpdateState state)
{
reservoir.update(state.update);
}
public static void main(String[] args) throws RunnerException
{
Options options = new OptionsBuilder()
.include(DecayingEstimatedHistogramBench.class.getSimpleName())
.build();
new Runner(options).run();
}
}

124
test/unit/org/apache/cassandra/metrics/DecayingEstimatedHistogramReservoirTest.java
View File

@ -18,22 +18,29 @@
package org.apache.cassandra.metrics;
import java.io.ByteArrayOutputStream;
import java.util.Arrays;
import java.util.Collection;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Function;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.runners.Enclosed;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Snapshot;
import org.apache.cassandra.utils.EstimatedHistogram;
@ -42,16 +49,21 @@ import org.apache.cassandra.utils.MonotonicClockTranslation;
import org.apache.cassandra.utils.Pair;
import org.quicktheories.core.Gen;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.LANDMARK_RESET_INTERVAL_IN_NS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.quicktheories.QuickTheory.qt;
import static org.quicktheories.generators.SourceDSL.*;
import static org.quicktheories.generators.SourceDSL.booleans;
import static org.quicktheories.generators.SourceDSL.integers;
import static org.quicktheories.generators.SourceDSL.longs;
@RunWith(Enclosed.class)
public class DecayingEstimatedHistogramReservoirTest
{
public static final Logger logger = LoggerFactory.getLogger(DecayingEstimatedHistogramReservoirTest.class);
public static class NonParameterizedTests
{
public static final int numExamples = 1000000;
@ -540,9 +552,9 @@ public class DecayingEstimatedHistogramReservoirTest
DecayingEstimatedHistogramReservoir histogram = new DecayingEstimatedHistogramReservoir(clock);
clock.addNanos(DecayingEstimatedHistogramReservoir.LANDMARK_RESET_INTERVAL_IN_NS - TimeUnit.SECONDS.toNanos(1L));
clock.addNanos(LANDMARK_RESET_INTERVAL_IN_NS - TimeUnit.SECONDS.toNanos(1L));
while (clock.now() < DecayingEstimatedHistogramReservoir.LANDMARK_RESET_INTERVAL_IN_NS + TimeUnit.SECONDS.toNanos(1L))
while (clock.now() < LANDMARK_RESET_INTERVAL_IN_NS + TimeUnit.SECONDS.toNanos(1L))
{
clock.addNanos(TimeUnit.MILLISECONDS.toNanos(900));
for (int i = 0; i < 1_000_000; i++)
@ -566,7 +578,7 @@ public class DecayingEstimatedHistogramReservoirTest
DecayingEstimatedHistogramReservoir histogram = new DecayingEstimatedHistogramReservoir(clock);
DecayingEstimatedHistogramReservoir another = new DecayingEstimatedHistogramReservoir(clock);
clock.addNanos(DecayingEstimatedHistogramReservoir.LANDMARK_RESET_INTERVAL_IN_NS - TimeUnit.SECONDS.toNanos(1L));
clock.addNanos(LANDMARK_RESET_INTERVAL_IN_NS - TimeUnit.SECONDS.toNanos(1L));
histogram.update(1000);
clock.addMillis(100);
@ -604,6 +616,110 @@ public class DecayingEstimatedHistogramReservoirTest
assertEquals(2, toSnapshot.apply(histogram).size());
}
/**
* This looks for invalid percentiles that are unchanged for too long to expose the CASSANDRA-19365 race
* condition between rescale and update. The idea is to update a histogram from multiple threads and observe
* if the reported p99 doesn't get stuck at a low value or p50 at a high value due to update with high weight
* being inserted after the buckets are rescaled.
* <p>
* The load has 95% of 42, and 5% of the time it's 1109. Despite that the histogram may be convinced for a long
* time that p99 is 42 or that p50 is 1109. The reason may be seen in the snapshot dump, where after rescale
* the bucket values may get very big due to the race condition and too big weight of the inserted samples.
* The values were picked to match bucket boundaries, but that's only for aesthetics.
* <p>
* In production the rescale happens every 30 minutes. In this test time we're pushing time to run faster,
* roughly 1000 times faster to hit the race condition in a reasonable time.
*/
@Test
public void testConcurrentUpdateAndRescale() throws InterruptedException
{
int UPDATE_THREADS = 60;
int maxTestDurationMillis = 30_000;
// how many times in a row the percentiles may be invalid before we fail the test
int tooManySuspiciousPercentilesThreshold = 5; // 5 translates to 500ms * 1000 speedup = 500s = 8m20s;
AtomicBoolean stop = new AtomicBoolean(false);
AtomicBoolean failed = new AtomicBoolean(false);
TestClock clock = new TestClock();
DecayingEstimatedHistogramReservoir histogram = new DecayingEstimatedHistogramReservoir(clock);
ExecutorService executors = Executors.newFixedThreadPool(2 + UPDATE_THREADS);
for (int i = 0; i < UPDATE_THREADS; i++)
{
executors.submit(() -> {
while (!stop.get() && !Thread.currentThread().isInterrupted())
{
// a mischievous usage pattern to quickly trigger the
// CASSANDRA-19365 race condition;
// the load has 95% of 42, and only 5% of the time it's 1109
// and yet, the histogram may be convinced for a long time that
// the p99 is 42 or that the p50 is 1109
for (int sampleIdx = 0; sampleIdx < 900; sampleIdx++)
histogram.update(42);
for (int sampleIdx = 0; sampleIdx < 50; sampleIdx++)
{
// add some noise so that low value samples do not race with the same likelyhood as the high value samples
Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(1, 10), MILLISECONDS);
histogram.update(1109);
}
}
});
}
// clock update thread
executors.submit(() -> {
while (!stop.get() && !Thread.currentThread().isInterrupted())
{
Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
// x1000 speedup so that we hit rescale interval every 30 minutes / 1000 = 1.8s
clock.addMillis(1000);
}
});
// percentiles check thread
executors.submit(() -> {
// how many times in a row p99 was suspiciously low or P50 suspiciously high
int consecutiveInvalidPercentiles = 0;
// how often to check the percentiles
int iterationDelayMillis = 100;
for (int i = 0; i < maxTestDurationMillis / iterationDelayMillis; i++)
{
Uninterruptibles.sleepUninterruptibly(iterationDelayMillis, MILLISECONDS);
Snapshot snapshot = toSnapshot.apply(histogram);
double p99 = snapshot.getValue(0.99);
double p50 = snapshot.getValue(0.50);
ByteArrayOutputStream output = new ByteArrayOutputStream();
snapshot.dump(output);
String decayingNonZeroBuckets = Arrays.stream(output.toString().split("\n"))
.filter(s -> !s.equals("0"))
.collect(Collectors.joining(","));
logger.info("\"clock={}, p50={}, p99={}, decaying non-zero buckets: {}",
clock.now() / 1_000_000, p50, p99, decayingNonZeroBuckets);
if (p99 < 100 || p50 > 900)
{
consecutiveInvalidPercentiles++;
logger.warn("p50 or p99 at suspicious level p50={}, p99={}", p50, p99);
if (consecutiveInvalidPercentiles > tooManySuspiciousPercentilesThreshold)
{
failed.set(true);
stop.set(true);
break;
}
}
else
{
consecutiveInvalidPercentiles = 0;
}
}
stop.set(true);
});
executors.shutdown();
boolean success = executors.awaitTermination(maxTestDurationMillis * 2, MILLISECONDS);
Assert.assertFalse("p50 too high or p99 too low for too long", failed.get());
Assert.assertTrue("Timeout exceeded the limit", success);
}
private void assertEstimatedQuantile(long expectedValue, double actualValue)
{
assertTrue("Expected at least [" + expectedValue + "] but actual is [" + actualValue + ']', actualValue >= expectedValue);