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

Add AccordExecutorMetrics 

Also Introduce:
 - Sharded/LogLinearDecayingHistogram
Also Improve:
 - Do not take a reference to CFK unless relevant
Also Fix:
 - Sharded/LogLinearHistogram
 - ExecuteFlags serialization bug in ReadData

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-21017
This commit is contained in:
Benedict Elliott Smith 2025-11-11 10:54:04 +00:00
parent cf8050393c
commit 494c3c7bef
22 changed files with 1454 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit de03d14e4821b669fecad5dcdb2de2dba3a9d926
Subproject commit 938ba19adaf70bf9901fb2dc177dce313d7ee15c

4
src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java
View File

@ -44,7 +44,7 @@ public class AccordCacheMetrics
public AccordCacheGlobalMetrics()
{
DefaultNameFactory factory = new DefaultNameFactory("AccordCache");
DefaultNameFactory factory = new DefaultNameFactory(ACCORD_CACHE);
this.usedBytes = Metrics.gauge(factory.createMetricName("UsedBytes"), fromAccordService(sumExecutors(executor -> executor.cacheUnsafe().weightedSize()), 0L));
this.unreferencedBytes = Metrics.gauge(factory.createMetricName("UnreferencedBytes"), fromAccordService(sumExecutors(executor -> executor.cacheUnsafe().unreferencedBytes()), 0L));
}
@ -89,7 +89,7 @@ public class AccordCacheMetrics
public AccordCacheMetrics(String subTypeName)
{
DefaultNameFactory factory = new DefaultNameFactory("AccordCache", subTypeName);
DefaultNameFactory factory = new DefaultNameFactory(ACCORD_CACHE, subTypeName);
this.objectSize = Metrics.shardedHistogram(factory.createMetricName("EntrySize"));
this.hits = Metrics.gauge(factory.createMetricName("Hits"), hitRate::totalHits);
this.misses = Metrics.gauge(factory.createMetricName("Misses"), hitRate::totalMisses);

5
src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java
View File

@ -42,6 +42,7 @@ public class AccordCoordinatorMetrics
{
public final static AccordCoordinatorMetrics readMetrics = new AccordCoordinatorMetrics("ro");
public final static AccordCoordinatorMetrics writeMetrics = new AccordCoordinatorMetrics("rw");
public final static AccordCoordinatorMetrics syncPointMetrics = new AccordCoordinatorMetrics("rx");
public static final String ACCORD_COORDINATOR = "AccordCoordinator";
public static final String COORDINATOR_EPOCHS = "Epochs";
@ -192,7 +193,7 @@ public class AccordCoordinatorMetrics
{
throw new RuntimeException(e);
}
builder.append("]");
builder.append(']');
return builder.toString();
}
@ -206,6 +207,8 @@ public class AccordCoordinatorMetrics
return writeMetrics;
else if (txnId.isSomeRead())
return readMetrics;
else if (txnId.isSyncPoint())
return syncPointMetrics;
}
return null;
}

61
src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java Normal file
View File

@ -0,0 +1,61 @@
/*
* 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.metrics;
import java.util.concurrent.TimeUnit;
import com.codahale.metrics.Gauge;
import org.apache.cassandra.metrics.ShardedDecayingHistograms.ShardedDecayingHistogram;
import org.apache.cassandra.service.accord.AccordExecutor;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
import static org.apache.cassandra.service.accord.AccordExecutor.HISTOGRAMS;
public class AccordExecutorMetrics
{
public static final String ACCORD_EXECUTOR = "AccordExecutor";
public static final AccordExecutorMetrics INSTANCE = new AccordExecutorMetrics();
public final ShardedLongGauges<AccordExecutor> gauges = new ShardedLongGauges<>();
// latency
public final ShardedDecayingHistogram elapsedPreparingToRun = HISTOGRAMS.newHistogram(TimeUnit.SECONDS.toNanos(1L));
public final ShardedDecayingHistogram elapsedWaitingToRun = HISTOGRAMS.newHistogram(TimeUnit.SECONDS.toNanos(1L));
public final ShardedDecayingHistogram elapsedRunning = HISTOGRAMS.newHistogram(TimeUnit.SECONDS.toNanos(1L));
// number of keys involved
public final ShardedDecayingHistogram keys = HISTOGRAMS.newHistogram(1 << 12);
public final Gauge<Long> preparingToRun;
public final Gauge<Long> waitingToRun;
public final Gauge<Long> running;
public AccordExecutorMetrics()
{
DefaultNameFactory factory = new DefaultNameFactory(ACCORD_EXECUTOR);
Metrics.register(factory.createMetricName("ElapsedPreparingToRun"), elapsedPreparingToRun);
Metrics.register(factory.createMetricName("ElapsedWaitingToRun"), elapsedWaitingToRun);
Metrics.register(factory.createMetricName("ElapsedRunning"), elapsedRunning);
Metrics.register(factory.createMetricName("Keys"), keys);
preparingToRun = Metrics.register(factory.createMetricName("PreparingToRun"), gauges.newGauge(AccordExecutor::unsafePreparingToRunCount, Long::sum));
waitingToRun = Metrics.register(factory.createMetricName("WaitingToRun"), gauges.newGauge(AccordExecutor::unsafeWaitingToRunCount, Long::sum));
running = Metrics.register(factory.createMetricName("Running"), gauges.newGauge(AccordExecutor::unsafeRunningCount, Long::sum));
}
}

123
src/java/org/apache/cassandra/metrics/AccordReplicaMetrics.java
View File

@ -25,63 +25,85 @@ import accord.api.ReplicaEventListener;
import accord.local.Command;
import accord.local.SafeCommandStore;
import accord.primitives.PartialDeps;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import com.codahale.metrics.Counting;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.Timer;
import org.apache.cassandra.service.accord.api.AccordTimeService;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms.LogLinearDecayingHistogram;
import org.apache.cassandra.metrics.ShardedDecayingHistograms.DecayingHistogramsShard;
import org.apache.cassandra.metrics.ShardedDecayingHistograms.ShardedDecayingHistogram;
import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordSafeCommandStore;
import org.apache.cassandra.tracing.Tracing;
import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
import static org.apache.cassandra.service.accord.AccordExecutor.HISTOGRAMS;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
public class AccordReplicaMetrics
{
public final static AccordReplicaMetrics readMetrics = new AccordReplicaMetrics("ro");
public final static AccordReplicaMetrics writeMetrics = new AccordReplicaMetrics("rw");
public final static AccordReplicaMetrics syncPointMetrics = new AccordReplicaMetrics("rx");
public static final String ACCORD_REPLICA = "AccordReplica";
public static final String REPLICA_STABLE_LATENCY = "StableLatency";
public static final String REPLICA_PREAPPLY_LATENCY = "PreApplyLatency";
public static final String REPLICA_APPLY_LATENCY = "ApplyLatency";
public static final String REPLICA_APPLY_DURATION = "ApplyDuration";
public static final String REPLICA_DEPENDENCIES = "Dependencies";
static final class SubShard
{
final LogLinearDecayingHistogram stableLatency;
final LogLinearDecayingHistogram preapplyLatency;
final LogLinearDecayingHistogram applyLatency;
final LogLinearDecayingHistogram dependencies;
private SubShard(AccordReplicaMetrics metrics, DecayingHistogramsShard shard)
{
this.stableLatency = metrics.stableLatency.forShard(shard);
this.preapplyLatency = metrics.preapplyLatency.forShard(shard);
this.applyLatency = metrics.applyLatency.forShard(shard);
this.dependencies = metrics.dependencies.forShard(shard);
}
}
public static final class Shard
{
final SubShard reads, writes, syncPoints;
public Shard(DecayingHistogramsShard shard)
{
reads = new SubShard(readMetrics, shard);
writes = new SubShard(writeMetrics, shard);
syncPoints = new SubShard(syncPointMetrics, shard);
}
}
/**
* The time between start on the coordinator and commit on this replica.
*/
public final Timer stableLatency;
public final ShardedDecayingHistogram stableLatency = HISTOGRAMS.newHistogram(TimeUnit.SECONDS.toNanos(1L));
/**
* The time between start on the coordinator and arrival of the result on this replica.
*/
public final Timer preapplyLatency;
public final ShardedDecayingHistogram preapplyLatency = HISTOGRAMS.newHistogram(TimeUnit.SECONDS.toNanos(1L));
/**
* The time between start on the coordinator and application on this replica.
*/
public final Timer applyLatency;
/**
* TODO (expected): probably more interesting is latency from preapplied to apply;
* we already track local write latencies, whch this effectively duplicates (but including queueing latencies)
* Duration of applying changes.
*/
public final Timer applyDuration;
public final ShardedDecayingHistogram applyLatency = HISTOGRAMS.newHistogram(TimeUnit.SECONDS.toNanos(1L));
/**
* A histogram of the number of dependencies per transaction at this replica.
*/
public final Histogram dependencies;
public final ShardedDecayingHistogram dependencies = HISTOGRAMS.newHistogram(1 << 12);
private AccordReplicaMetrics(String scope)
{
DefaultNameFactory replica = new DefaultNameFactory(ACCORD_REPLICA, scope);
stableLatency = Metrics.timer(replica.createMetricName(REPLICA_STABLE_LATENCY));
preapplyLatency = Metrics.timer(replica.createMetricName(REPLICA_PREAPPLY_LATENCY));
applyLatency = Metrics.timer(replica.createMetricName(REPLICA_APPLY_LATENCY));
applyDuration = Metrics.timer(replica.createMetricName(REPLICA_APPLY_DURATION));
dependencies = Metrics.histogram(replica.createMetricName(REPLICA_DEPENDENCIES), true);
Metrics.register(replica.createMetricName(REPLICA_STABLE_LATENCY), stableLatency);
Metrics.register(replica.createMetricName(REPLICA_PREAPPLY_LATENCY), preapplyLatency);
Metrics.register(replica.createMetricName(REPLICA_APPLY_LATENCY), applyLatency);
Metrics.register(replica.createMetricName(REPLICA_DEPENDENCIES), dependencies);
}
@Override
@ -106,64 +128,79 @@ public class AccordReplicaMetrics
{
throw new RuntimeException(e);
}
builder.append("]");
builder.append(']');
return builder.toString();
}
public static class Listener implements ReplicaEventListener
{
private AccordReplicaMetrics forTransaction(TxnId txnId)
private SubShard forTransaction(SafeCommandStore safeStore, TxnId txnId)
{
if (txnId != null)
{
Shard shard = ((AccordCommandStore) safeStore.commandStore()).executor().replicaMetrics;
if (txnId.isWrite())
return writeMetrics;
return shard.writes;
else if (txnId.isSomeRead())
return readMetrics;
return shard.reads;
else if (txnId.isSyncPoint())
return shard.syncPoints;
}
return null;
}
private static long unixNanos()
{
return currentTimeMillis() * 1_000_000;
}
private static long elapsed(TxnId txnId)
{
return elapsed(unixNanos(), txnId);
}
private static long elapsed(long unixNanos, TxnId txnId)
{
return Math.max(0, unixNanos - (txnId.hlc() * 1000));
}
private static LogLinearDecayingHistograms.Buffer buffer(SafeCommandStore safeStore)
{
return ((AccordSafeCommandStore) safeStore).histogramBuffer();
}
@Override
public void onStable(SafeCommandStore safeStore, Command cmd)
{
Tracing.trace("Stable {} on {}", cmd.txnId(), safeStore.commandStore());
long now = AccordTimeService.nowMicros();
AccordReplicaMetrics metrics = forTransaction(cmd.txnId());
SubShard metrics = forTransaction(safeStore, cmd.txnId());
if (metrics != null)
{
long trxTimestamp = cmd.txnId().hlc();
metrics.stableLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS);
}
metrics.stableLatency.add(buffer(safeStore), elapsed(cmd.txnId()));
}
@Override
public void onPreApplied(SafeCommandStore safeStore, Command cmd)
{
Tracing.trace("Preapplied {} on {}", cmd.txnId(), safeStore.commandStore());
long now = AccordTimeService.nowMicros();
AccordReplicaMetrics metrics = forTransaction(cmd.txnId());
SubShard metrics = forTransaction(safeStore, cmd.txnId());
if (metrics != null)
{
Timestamp trxTimestamp = cmd.txnId();
metrics.preapplyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
long elapsed = elapsed(cmd.txnId());
metrics.preapplyLatency.add(buffer(safeStore), elapsed);
PartialDeps deps = cmd.partialDeps();
metrics.dependencies.update(deps != null ? deps.txnIdCount() : 0);
metrics.dependencies.add(buffer(safeStore), deps != null ? deps.txnIdCount() : 0);
}
}
@Override
public void onApplied(SafeCommandStore safeStore, Command cmd, long applyStartedAt)
public void onApplied(SafeCommandStore safeStore, Command cmd)
{
Tracing.trace("Applied {} on {}", cmd.txnId(), safeStore.commandStore());
long now = AccordTimeService.nowMicros();
AccordReplicaMetrics metrics = forTransaction(cmd.txnId());
SubShard metrics = forTransaction(safeStore, cmd.txnId());
if (metrics != null)
{
Timestamp trxTimestamp = cmd.txnId();
metrics.applyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
if (applyStartedAt > 0)
metrics.applyDuration.update(now - applyStartedAt, TimeUnit.MICROSECONDS);
long now = unixNanos();
metrics.applyLatency.add(buffer(safeStore), elapsed(now, cmd.txnId()));
}
}
}

1
src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
View File

@ -118,6 +118,7 @@ public class CassandraMetricsRegistry extends MetricRegistry
.add(AccordCoordinatorMetrics.ACCORD_COORDINATOR)
.add(AccordCacheMetrics.ACCORD_CACHE)
.add(AccordReplicaMetrics.ACCORD_REPLICA)
.add(AccordExecutorMetrics.ACCORD_EXECUTOR)
.add(AccordSystemMetrics.ACCORD_SYSTEM)
.add(BatchMetrics.TYPE_NAME)
.add(BufferPoolMetrics.TYPE_NAME)

283
src/java/org/apache/cassandra/metrics/LogLinearDecayingHistograms.java Normal file
View File

@ -0,0 +1,283 @@
/*
* 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.metrics;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import accord.utils.Invariants;
import org.apache.cassandra.metrics.LogLinearHistogram.LogLinearSnapshot;
import static org.apache.cassandra.metrics.DecayingEstimatedHistogramReservoir.MEAN_LIFETIME_IN_S;
import static org.apache.cassandra.metrics.LogLinearHistogram.MAX_INDEX;
import static org.apache.cassandra.metrics.LogLinearHistogram.index;
import static org.apache.cassandra.metrics.LogLinearHistogram.invertIndex;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* A simple single-threaded decaying histogram with log-linear bucket distribution.
* This has approximately the same accuracy as the lg 1.2 growth of EstimatedHistogram, but is simpler and faster.
*
* Logically very similar to DecayingEstimatedHistogramReservoir, only due to single-threaded behaviour this can be
* performed more simply and efficiently. We project forward single updates, and periodically rebase in place.
* We also share the costly arithmetic across a group of histograms that share mutual exclusivity.
*
* TODO (desired): improve performance and memory locality by using a small buffer for collecting updates with e.g. 4 bits per counter,
*/
public class LogLinearDecayingHistograms
{
public static class Buffer
{
static final long LARGE_VALUE = -1L >>> (1 + HISTOGRAM_INDEX_BITS);
private final LogLinearDecayingHistograms histograms;
private long[] buffer = new long[2];
private int bufferCount;
public Buffer(LogLinearDecayingHistograms histograms)
{
this.histograms = histograms;
}
private void add(long histogramIndex, long value)
{
Invariants.require(histogramIndex <= HISTOGRAM_INDEX_MASK);
Invariants.require(value >= 0);
if (value <= LARGE_VALUE) value <<= HISTOGRAM_INDEX_BITS;
else
{
value |= Long.MIN_VALUE;
value &= ~HISTOGRAM_INDEX_MASK;
}
value |= histogramIndex;
if (bufferCount == buffer.length)
buffer = Arrays.copyOf(buffer, bufferCount * 2);
buffer[bufferCount++] = value;
}
public void flush(long at)
{
for (int i = 0 ; i < bufferCount ; ++i)
{
long v = buffer[i];
int histogramIndex = (int) (v & HISTOGRAM_INDEX_MASK);
if (v < 0) v &= ~Long.MIN_VALUE & ~HISTOGRAM_INDEX_MASK;
else v >>>= HISTOGRAM_INDEX_BITS;
histograms.histograms.get(histogramIndex).increment(v, at);
}
bufferCount = 0;
}
public boolean isEmpty()
{
return bufferCount == 0;
}
}
public class LogLinearDecayingHistogram
{
final int histogramIndex;
private double[] buckets;
double totalCount;
private LogLinearDecayingHistogram(int histogramIndex, long expectedMaxValue)
{
this.histogramIndex = histogramIndex;
buckets = new double[LogLinearHistogram.bucketCount(expectedMaxValue)];
}
public void increment(long value)
{
increment(value, nanoTime());
}
public void increment(long value, long at)
{
int index = index(value);
if (bufferCount >= buffer.length)
{
Invariants.require(bufferCount == buffer.length);
flush();
Invariants.require(bufferCount == 0);
}
updateDecay(at);
long v = Double.doubleToRawLongBits(increment) & VALUE_MASK;
v |= histogramIndex | ((long)index << HISTOGRAM_INDEX_BITS);
buffer[bufferCount++] = v;
}
private double[] buckets(int withIndexAtLeast)
{
if (buckets.length <= withIndexAtLeast)
{
Invariants.require(withIndexAtLeast <= MAX_INDEX);
buckets = Arrays.copyOf(buckets, (withIndexAtLeast | 0x3) + 1);
}
return buckets;
}
public LogLinearSnapshot copyToSnapshot(long at)
{
LogLinearSnapshot snapshot = new LogLinearSnapshot(new long[buckets.length], 0);
updateSnapshot(snapshot, at);
return snapshot;
}
public void updateSnapshot(LogLinearSnapshot snapshot, long at)
{
decay(at);
if (snapshot.raw.length < buckets.length)
snapshot.raw = Arrays.copyOf(snapshot.raw, buckets.length);
long[] raw = snapshot.raw;
long total = 0;
for (int i = 0 ; i < buckets.length ; ++i)
{
long v = Math.round(buckets[i]);
raw[i] += v;
total += v;
}
snapshot.totalCount += total;
snapshot.cumulative = null;
}
public void add(Buffer buffer, long value)
{
Invariants.require(buffer.histograms == LogLinearDecayingHistograms.this);
buffer.add(histogramIndex, value);
}
}
private static final long ANTI_DECAY_REFRESH_RATE = TimeUnit.SECONDS.toNanos(1L);
private static final long DECAY_REFRESH_RATE = TimeUnit.HOURS.toNanos(1L);
private static final double DECAY_RATE = 1d/(TimeUnit.SECONDS.toNanos(1) * MEAN_LIFETIME_IN_S);
private static final int BUCKET_INDEX_BITS = 8;
private static final int HISTOGRAM_INDEX_BITS = 5;
private static final long HISTOGRAM_INDEX_MASK = (1L << HISTOGRAM_INDEX_BITS) - 1;
private static final long BUCKET_INDEX_MASK = (1L << BUCKET_INDEX_BITS) - 1;
private static final long VALUE_MASK = ~(HISTOGRAM_INDEX_MASK | (BUCKET_INDEX_MASK << HISTOGRAM_INDEX_BITS));
private static final int BUFFER_SIZE = 64;
static
{
Invariants.require((1 << BUCKET_INDEX_BITS) > MAX_INDEX);
}
private final long[] buffer = new long[BUFFER_SIZE];
private int bufferCount;
private long lastDecayedAt, lastAntiDecayedAt;
private double increment = 1d;
List<LogLinearDecayingHistogram> histograms = new ArrayList<>();
void updateDecay(long now)
{
long delta = now - lastAntiDecayedAt;
if (delta <= ANTI_DECAY_REFRESH_RATE)
return;
if (delta < DECAY_REFRESH_RATE) antiDecay(now);
else decay(now);
}
private void antiDecay(long now)
{
increment = Math.exp((now - lastDecayedAt) * DECAY_RATE);
}
private void flush()
{
for (int i = 0 ; i < bufferCount ; ++i)
{
long bits = buffer[i];
double increment = Double.longBitsToDouble(bits & VALUE_MASK);
int histogramIndex = (int) (bits & HISTOGRAM_INDEX_MASK);
int bucketIndex = (int)((bits >>> HISTOGRAM_INDEX_BITS) & BUCKET_INDEX_MASK);
LogLinearDecayingHistogram histogram = histograms.get(histogramIndex);
histogram.buckets(bucketIndex)[bucketIndex] += increment;
histogram.totalCount += increment;
}
bufferCount = 0;
}
private void decay(long now)
{
flush();
if (now <= lastDecayedAt)
return;
antiDecay(now);
double decay = 1d/increment;
increment = 1d;
for (LogLinearDecayingHistogram histogram : histograms)
{
if (histogram.totalCount <= 0d)
continue;
double[] buckets = histogram.buckets;
double total = 0d;
for (int i = 0 ; i < buckets.length ; ++i)
{
if (buckets[i] <= 0d)
continue;
double in = buckets[i];
double out = in * decay;
if (out < 0.5d)
out = 0d;
buckets[i] = out;
total += out;
}
histogram.totalCount = total;
}
lastDecayedAt = lastAntiDecayedAt = now;
}
public synchronized LogLinearDecayingHistogram allocate(long expectedMaxValue)
{
Invariants.require(histograms.size() <= HISTOGRAM_INDEX_MASK);
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistogram(histograms.size(), expectedMaxValue);
histograms.add(histogram);
return histogram;
}
public synchronized LogLinearDecayingHistogram get(int histogramIndex)
{
return histograms.get(histogramIndex);
}
public static double[] bucketsWithScale(long maxScale)
{
return bucketsWithLength(1 + index(maxScale));
}
public static double[] bucketsWithLength(int length)
{
Invariants.require(length <= MAX_INDEX + 1);
double[] buckets = new double[length];
for (int i = 0 ; i < length ; ++i)
buckets[i] = invertIndex(i);
return buckets;
}
}

98
src/java/org/apache/cassandra/metrics/LogLinearHistogram.java
View File

@ -28,8 +28,11 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import accord.utils.Invariants;
import accord.utils.SortedArrays;
import com.codahale.metrics.Snapshot;
import static accord.utils.SortedArrays.Search.CEIL;
/**
* A simple single-threaded histogram with log-linear buckets.
* This has approximately the same accuracy as the lg 1.2 growth of EstimatedHistogram, but is simpler and faster.
@ -40,13 +43,13 @@ import com.codahale.metrics.Snapshot;
*/
public class LogLinearHistogram
{
private static final int MAX_INDEX = 247;
public static final int MAX_INDEX = 247;
private long[] buckets;
long totalCount;
public LogLinearHistogram(long expectedMaxValue)
{
buckets = new long[buckets(expectedMaxValue)];
buckets = new long[bucketCount(expectedMaxValue)];
}
public void increment(long value)
@ -99,12 +102,12 @@ public class LogLinearHistogram
return buckets;
}
private static int buckets(long maxValue)
static int bucketCount(long maxValue)
{
return 1 + (index(maxValue) | 0x3);
}
private static int index(long value)
static int index(long value)
{
if (value < 4)
return (int) value;
@ -113,7 +116,7 @@ public class LogLinearHistogram
return (log + 1) * 4 + linear;
}
private static long invertIndex(int index)
static long invertIndex(int index)
{
if (index < 4)
return index;
@ -141,7 +144,7 @@ public class LogLinearHistogram
public static LogLinearSnapshot emptyForMax(long maxValue)
{
return new LogLinearSnapshot(buckets(maxValue));
return new LogLinearSnapshot(bucketCount(maxValue));
}
LogLinearSnapshot(int size)
@ -163,33 +166,37 @@ public class LogLinearHistogram
cumulative = new long[raw.length];
long sum = 0;
for (int i = 0 ; i < cumulative.length ; ++i)
cumulative[i] = sum += cumulative[i];
cumulative[i] = sum += raw[i];
}
return cumulative;
}
private long get(long tot)
private double get(long tot)
{
if (totalCount == 0)
if (totalCount == 0 || tot == 0)
return 0;
long[] cumulative = cumulative();
int i = Arrays.binarySearch(cumulative, tot);
if (i >= 0) while (i > 0 && cumulative[i-1] == tot) --i;
else i = Math.max(0, -2 - i);
long prevValue = invertIndex(i);
int i = SortedArrays.binarySearch(cumulative, 0, cumulative.length, tot, CEIL);
if (i >= 0)
return invertIndex(i);
i = Math.max(0, -2 - i);
long prevCount = cumulative[i];
long nextCount = cumulative[i + 1];
long prevValue = invertIndex(i);
long nextValue = invertIndex(i + 1);
long nextCount = i + 1 == cumulative.length ? totalCount : cumulative[i + 1];
long gap = tot - prevCount;
// should we use double arithmetic here to avoid overflow?
return prevValue + ((nextValue - prevValue) * gap) / (nextCount - prevCount);
double granularity = (nextValue - prevValue) / (double)(nextCount - prevCount);
double targetGap = tot - prevCount;
return prevValue + Math.round(targetGap * granularity);
}
@Override
public double getValue(double quantile)
{
return get(Math.round(totalCount * quantile));
return get(Math.max(1L, (long)Math.ceil(totalCount * quantile)));
}
@Override
@ -198,24 +205,48 @@ public class LogLinearHistogram
return Ints.saturatedCast(totalCount);
}
@Override
public long getMax()
{
return get(totalCount);
}
@Override
public double getMean()
{
if (totalCount <= 1)
if (totalCount == 0)
return 0.0D;
return get(totalCount / 2);
double sum = 0;
for (int i = 0; i < raw.length; i++)
{
if (raw[i] != 0)
sum += raw[i] * (double) invertIndex(i);
}
return sum / totalCount;
}
@Override
public long getMin()
{
return get(1);
if (totalCount == 0)
return 0;
long[] cumulative = cumulative();
int i = SortedArrays.binarySearch(cumulative, 0, cumulative.length, 1, CEIL);
if (i < 0)
{
i = Math.max(0, -2 - i);
if (cumulative[i] == 0)
++i;
}
return invertIndex(i);
}
@Override
public long getMax()
{
if (totalCount == 0)
return 0;
long[] cumulative = cumulative();
int i = SortedArrays.binarySearch(cumulative, 0, cumulative.length, totalCount, CEIL);
if (i < 0) i = -2 - i;
return invertIndex(i + 1);
}
/**
@ -230,11 +261,8 @@ public class LogLinearHistogram
public double getStdDev()
{
if (totalCount <= 1)
{
return 0.0D;
}
else
{
double mean = this.getMean();
double sum = 0.0D;
@ -247,7 +275,6 @@ public class LogLinearHistogram
return Math.sqrt(sum / (totalCount - 1));
}
}
@Override
public void dump(OutputStream output)
@ -274,13 +301,18 @@ public class LogLinearHistogram
return result;
}
public LogLinearSnapshot copyToSnapshot()
{
return new LogLinearSnapshot(buckets.clone(), totalCount);
}
public void updateSnapshot(LogLinearSnapshot snapshot)
{
if (snapshot.raw.length < buckets.length)
snapshot.raw = Arrays.copyOf(snapshot.raw, buckets.length);
long[] raw = snapshot.raw;
for (int i = 0 ; i < raw.length ; ++i)
for (int i = 0 ; i < buckets.length ; ++i)
raw[i] = raw[i] + buckets[i];
snapshot.totalCount += totalCount;
snapshot.cumulative = null;

159
src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java Normal file
View File

@ -0,0 +1,159 @@
/*
* 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.metrics;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import com.codahale.metrics.Snapshot;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms.LogLinearDecayingHistogram;
import org.apache.cassandra.metrics.LogLinearHistogram.LogLinearSnapshot;
import org.apache.cassandra.utils.Clock;
import static org.apache.cassandra.metrics.CassandraReservoir.BucketStrategy.log_linear;
public class ShardedDecayingHistograms
{
private static final long REFRESH_RATE = TimeUnit.SECONDS.toNanos(15);
public class ShardedDecayingHistogram extends OverrideHistogram
{
final int histogramIndex;
final long initialMaxValue;
private ShardedDecayingHistogram(int histogramIndex, long initialMaxValue)
{
this.histogramIndex = histogramIndex;
this.initialMaxValue = initialMaxValue;
}
@Override
public CassandraReservoir.BucketStrategy bucketStrategy()
{
return log_linear;
}
@Override
public long[] bucketStarts(int length)
{
return LogLinearHistogram.bucketsWithLength(length);
}
@Override
public synchronized long getCount()
{
return maybeRefresh().get(histogramIndex).totalCount;
}
@Override
public Snapshot getSnapshot()
{
return maybeRefresh().get(histogramIndex);
}
public LogLinearDecayingHistogram forShard(DecayingHistogramsShard shard)
{
return shard.histograms.get(histogramIndex);
}
}
public static class DecayingHistogramsShard
{
final Lock lock;
final LogLinearDecayingHistograms histograms;
DecayingHistogramsShard(Lock lock, LogLinearDecayingHistograms histograms)
{
this.lock = lock;
this.histograms = histograms;
}
void updateSnapshot(List<LogLinearSnapshot> update, long at)
{
lock.lock();
try
{
for (int i = 0 ; i < update.size() ; ++i)
histograms.get(i).updateSnapshot(update.get(i), at);
}
finally
{
lock.unlock();
}
}
public LogLinearDecayingHistograms unsafeGetInternal()
{
return histograms;
}
}
final List<DecayingHistogramsShard> shards = new ArrayList<>();
final List<ShardedDecayingHistogram> histograms = new ArrayList<>();
public synchronized DecayingHistogramsShard newShard(Lock guardedBy)
{
DecayingHistogramsShard shard = new DecayingHistogramsShard(guardedBy, new LogLinearDecayingHistograms());
for (int i = 0 ; i < histograms.size() ; ++i)
shard.histograms.allocate(histograms.get(i).initialMaxValue);
shards.add(shard);
return shard;
}
public synchronized ShardedDecayingHistogram newHistogram(long initialMaxValue)
{
ShardedDecayingHistogram histogram = new ShardedDecayingHistogram(histograms.size(), initialMaxValue);
for (int i = 0 ; i < shards.size() ; ++i)
shards.get(i).histograms.allocate(initialMaxValue);
histograms.add(histogram);
return histogram;
}
private long snapshotAt = Long.MIN_VALUE;
private List<LogLinearSnapshot> snapshot = Collections.emptyList();
public synchronized void refresh()
{
refresh(Clock.Global.nanoTime());
}
private synchronized List<LogLinearSnapshot> refresh(long now)
{
List<LogLinearSnapshot> snapshot = new ArrayList<>(histograms.size());
for (ShardedDecayingHistogram histogram : histograms)
snapshot.add(LogLinearSnapshot.emptyForMax(histogram.initialMaxValue));
for (DecayingHistogramsShard shard : shards)
shard.updateSnapshot(snapshot, now);
this.snapshot = snapshot;
this.snapshotAt = now;
return this.snapshot;
}
private synchronized List<LogLinearSnapshot> maybeRefresh()
{
long now = Clock.Global.nanoTime();
if (snapshot.size() == histograms.size() && snapshotAt + REFRESH_RATE >= now)
return snapshot;
return refresh(now);
}
}

13
src/java/org/apache/cassandra/metrics/ShardedHistogram.java
View File

@ -43,19 +43,6 @@ public class ShardedHistogram extends OverrideHistogram
this.histogram = histogram;
}
long total()
{
lock.lock();
try
{
return histogram.totalCount;
}
finally
{
lock.unlock();
}
}
public void updateSnapshot(LogLinearSnapshot snapshot)
{
lock.lock();

148
src/java/org/apache/cassandra/metrics/ShardedLongGauges.java Normal file
View File

@ -0,0 +1,148 @@
/*
* 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.metrics;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.function.LongBinaryOperator;
import java.util.function.ToLongFunction;
import accord.utils.Invariants;
import com.codahale.metrics.Gauge;
import org.apache.cassandra.utils.Clock;
public class ShardedLongGauges<S>
{
private static final long REFRESH_RATE = TimeUnit.SECONDS.toNanos(15);
public class ShardedLongGauge implements Gauge<Long>
{
final int gaugeIndex;
final ToLongFunction<S> compute;
final LongBinaryOperator reduce;
private ShardedLongGauge(int gaugeIndex, ToLongFunction<S> compute, LongBinaryOperator reduce)
{
this.gaugeIndex = gaugeIndex;
this.compute = compute;
this.reduce = reduce;
}
@Override
public Long getValue()
{
return maybeRefresh()[gaugeIndex];
}
}
static class LongGaugeShard<T>
{
final Lock lock;
final T shard;
LongGaugeShard(Lock lock, T shard)
{
this.lock = lock;
this.shard = shard;
}
public void init(long[] init, List<ShardedLongGauges<T>.ShardedLongGauge> gauges)
{
lock.lock();
try
{
for (int i = 0 ; i < init.length ; ++i)
init[i] = gauges.get(i).compute.applyAsLong(shard);
}
finally
{
lock.unlock();
}
}
public void update(long[] update, List<ShardedLongGauges<T>.ShardedLongGauge> gauges)
{
lock.lock();
try
{
for (int i = 0 ; i < update.length ; ++i)
{
ShardedLongGauges<T>.ShardedLongGauge gauge = gauges.get(i);
Invariants.require(gauge.gaugeIndex == i);
update[i] = gauge.reduce.applyAsLong(update[i], gauge.compute.applyAsLong(shard));
}
}
finally
{
lock.unlock();
}
}
}
final List<LongGaugeShard<S>> shards = new ArrayList<>();
final List<ShardedLongGauge> gauges = new ArrayList<>();
public synchronized void newShard(Lock guardedBy, S shard)
{
shards.add(new LongGaugeShard<>(guardedBy, shard));
}
public synchronized ShardedLongGauge newGauge(ToLongFunction<S> compute, LongBinaryOperator reduce)
{
ShardedLongGauge gauge = new ShardedLongGauge(gauges.size(), compute, reduce);
gauges.add(gauge);
return gauge;
}
private long snapshotAt = Long.MIN_VALUE;
private long[] snapshot = new long[0];
public synchronized void refresh()
{
refresh(Clock.Global.nanoTime());
}
private synchronized long[] refresh(long now)
{
if (gauges.isEmpty())
return new long[0];
long[] snapshot = new long[gauges.size()];
if (shards.isEmpty())
return snapshot;
shards.get(0).init(snapshot, gauges);
for (int i = 1; i < shards.size() ; ++i)
shards.get(i).update(snapshot, gauges);
this.snapshot = snapshot;
this.snapshotAt = now;
return this.snapshot;
}
private synchronized long[] maybeRefresh()
{
long now = Clock.Global.nanoTime();
if (snapshot.length == gauges.size() && snapshotAt + REFRESH_RATE >= now)
return snapshot;
return refresh(now);
}
}

35
src/java/org/apache/cassandra/service/accord/AccordCache.java
View File

@ -54,6 +54,7 @@ import accord.utils.UnhandledEnum;
import org.agrona.collections.Object2ObjectHashMap;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.exceptions.UnknownTableException;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.metrics.AccordCacheMetrics;
@ -62,11 +63,11 @@ import org.apache.cassandra.metrics.ShardedHitRate;
import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
import org.apache.cassandra.service.accord.events.CacheEvents;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTree;
import static accord.utils.Invariants.illegalState;
import static accord.utils.Invariants.require;
@ -219,6 +220,7 @@ public class AccordCache implements CacheSize
{
require(node.references() == 0);
// TODO (desired): special-case evict queue with 1 element as no point shrinking first
if (shrinkingOn && node.tryShrink())
{
IntrusiveLinkedList<AccordCacheEntry<?,?>> queue;
@ -1107,6 +1109,11 @@ public class AccordCache implements CacheSize
@Override
public Runnable save(AccordCommandStore commandStore, RoutingKey key, @Nullable CommandsForKey value, @Nullable Object serialized)
{
if (serialized != null)
{
ByteBuffer bb = (ByteBuffer)serialized;
serialized = bb.duplicate().position(prefixBytes(bb));
}
return commandStore.saveCommandsForKey(key, value, serialized);
}
@ -1125,19 +1132,35 @@ public class AccordCache implements CacheSize
@Override
public Object fullShrink(RoutingKey key, CommandsForKey value)
{
if (value.isEmpty())
if (value.isEmpty() || value.isLoadingPruned())
return null;
if (value.isLoadingPruned())
return value;
TxnId last = value.size() == 0 ? null : value.get(value.size() - 1);
TxnId minUndecided = value.minUndecided();
int lastSize = (int) CommandSerializers.txnId.serializedSize(last);
int minUndecidedSize = (int) CommandSerializers.txnId.serializedSize(minUndecided);
ByteBuffer result = Serialize.toBytesWithoutKey(lastSize + minUndecidedSize, value.maximalPrune());
int limit = result.limit();
result.limit(lastSize + minUndecidedSize);
CommandSerializers.txnId.serialize(last, result, ByteBufferAccessor.instance, 0);
CommandSerializers.txnId.serialize(minUndecided, result, ByteBufferAccessor.instance, lastSize);
result.limit(limit);
return result;
}
return Serialize.toBytesWithoutKey(value.maximalPrune());
private static int prefixBytes(ByteBuffer bb)
{
int prefix = (int) CommandSerializers.txnId.serializedSize(CommandSerializers.txnId.deserialize(bb, 0));
prefix += (int)CommandSerializers.txnId.serializedSize(CommandSerializers.txnId.deserialize(bb, prefix));
return prefix;
}
@Override
public CommandsForKey inflate(AccordCommandStore commandStore, RoutingKey key, Object shrunk)
{
return Serialize.fromBytes(key, (ByteBuffer)shrunk);
ByteBuffer bb = ((ByteBuffer)shrunk).duplicate();
bb.position(prefixBytes(bb));
return Serialize.fromBytes(key, bb, false);
}
@Override

2
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -67,6 +67,7 @@ import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResults.CountingResult;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
@ -162,6 +163,7 @@ public class AccordCommandStore extends CommandStore
volatile SafeRedundantBefore safeRedundantBefore;
private AccordSafeCommandStore current;
LogLinearDecayingHistograms.Buffer metricsBuffer;
public AccordCommandStore(int id,
NodeCommandStoreService node,

37
src/java/org/apache/cassandra/service/accord/AccordExecutor.java
View File

@ -69,6 +69,12 @@ import org.apache.cassandra.concurrent.DebuggableTask.DebuggableTaskRunner;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.metrics.AccordCacheMetrics;
import org.apache.cassandra.metrics.AccordExecutorMetrics;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms.LogLinearDecayingHistogram;
import org.apache.cassandra.metrics.ShardedDecayingHistograms;
import org.apache.cassandra.metrics.ShardedDecayingHistograms.DecayingHistogramsShard;
import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.SaveExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.UniqueSave;
@ -98,6 +104,8 @@ import static org.apache.cassandra.service.accord.AccordTask.State.WAITING_TO_RU
public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTask<?>, Boolean>, SaveExecutor, Shutdownable, AbstractAsyncExecutor
{
private static final Logger logger = LoggerFactory.getLogger(AccordExecutor.class);
public static final ShardedDecayingHistograms HISTOGRAMS = new ShardedDecayingHistograms();
public interface AccordExecutorFactory
{
AccordExecutor get(int executorId, Mode mode, int threads, IntFunction<String> name, Agent agent);
@ -159,6 +167,13 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
private List<Condition> waitingForQuiescence;
private Queue<WaitForCompletion> waitingForCompletion;
final LogLinearDecayingHistograms histograms;
final LogLinearDecayingHistogram elapsedPreparingToRun;
final LogLinearDecayingHistogram elapsedWaitingToRun;
final LogLinearDecayingHistogram elapsedRunning;
final LogLinearDecayingHistogram keys;
public final AccordReplicaMetrics.Shard replicaMetrics;
private static class WaitForCompletion
{
final int position;
@ -212,6 +227,14 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
registerJfrListener(executorId, commandsForKey, "CommandsForKey");
this.caches = new ExclusiveGlobalCaches(this, cache, commands, commandsForKey);
DecayingHistogramsShard histogramsShard = HISTOGRAMS.newShard(lock);
this.histograms = histogramsShard.unsafeGetInternal();
this.elapsedPreparingToRun = AccordExecutorMetrics.INSTANCE.elapsedPreparingToRun.forShard(histogramsShard);
this.elapsedWaitingToRun = AccordExecutorMetrics.INSTANCE.elapsedWaitingToRun.forShard(histogramsShard);
this.elapsedRunning = AccordExecutorMetrics.INSTANCE.elapsedRunning.forShard(histogramsShard);
this.keys = AccordExecutorMetrics.INSTANCE.keys.forShard(histogramsShard);
this.replicaMetrics = new AccordReplicaMetrics.Shard(histogramsShard);
ScheduledExecutors.scheduledFastTasks.scheduleAtFixedRate(() -> {
executeDirectlyWithLock(cache::processNoEvictQueue);
}, 1L, 1L, TimeUnit.SECONDS);
@ -1761,4 +1784,18 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
}
public int unsafePreparingToRunCount()
{
return waitingToLoad.size() + waitingToLoadRangeTxns.size() + scanningRanges.size() + loading.size();
}
public int unsafeWaitingToRunCount()
{
return waitingToRun.size();
}
public int unsafeRunningCount()
{
return running.size();
}
}

2
src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
View File

@ -37,8 +37,6 @@ import org.apache.cassandra.service.accord.serializers.Version;
import static accord.local.CommandStores.RangesForEpoch;
// TODO (required): test with large collection values, and perhaps split out some fields if they have a tendency to grow larger
// TODO (required): alert on metadata size
// TODO (required): versioning
public class AccordJournalValueSerializers
{
public interface FlyweightImage

14
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
View File

@ -40,6 +40,8 @@ import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms;
import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
import org.apache.cassandra.service.accord.AccordCommandStore.SafeRedundantBefore;
import org.apache.cassandra.service.paxos.PaxosState;
@ -207,6 +209,18 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
return commandStore.node();
}
public LogLinearDecayingHistograms.Buffer histogramBuffer()
{
if (task.histogramBuffer == null)
{
task.histogramBuffer = commandStore.metricsBuffer;
if (task.histogramBuffer == null)
task.histogramBuffer = commandStore.metricsBuffer = new LogLinearDecayingHistograms.Buffer(commandStore.executor().histograms);
Invariants.require(task.histogramBuffer.isEmpty());
}
return task.histogramBuffer;
}
private boolean visitForKey(Unseekables<?> keysOrRanges, Predicate<CommandsForKey> forEach)
{
Map<RoutingKey, AccordSafeCommandsForKey> commandsForKey = task.commandsForKey;

48
src/java/org/apache/cassandra/service/accord/AccordTask.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.service.accord;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
@ -59,12 +60,14 @@ import accord.utils.async.Cancellable;
import org.agrona.collections.Object2ObjectHashMap;
import org.agrona.collections.ObjectHashSet;
import org.apache.cassandra.concurrent.DebuggableTask;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
import org.apache.cassandra.service.accord.AccordCommandStore.Caches;
import org.apache.cassandra.service.accord.AccordExecutor.SubmittableTask;
import org.apache.cassandra.service.accord.AccordExecutor.TaskQueue;
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.Closeable;
@ -201,6 +204,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
@Nullable Object2ObjectHashMap<TxnId, AccordSafeCommand> commands;
@Nullable Object2ObjectHashMap<RoutingKey, AccordSafeCommandsForKey> commandsForKey;
@Nullable Object2ObjectHashMap<Object, AccordSafeState<?, ?>> loading;
LogLinearDecayingHistograms.Buffer histogramBuffer;
// TODO (desired): collection supporting faster deletes but still fast poll (e.g. some ordered collection)
@Nullable ArrayDeque<AccordCacheEntry<?, ?>> waitingToLoad;
@Nullable RangeTxnScanner rangeScanner;
@ -299,12 +303,20 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
Invariants.require(rangeScanner == null || rangeScanner.scanned);
Invariants.require(loading == null && waitingToLoad == null, "WAITING_TO_RUN => no loading or waiting; found %s", this, AccordTask::toDescription);
waitingToRunAt = nanoTime();
commandStore.executor().elapsedPreparingToRun.increment(waitingToRunAt - createdAt, runningAt);
}
else if (state == RUNNING)
{
runningAt = nanoTime();
if (waitingToRunAt == 0)
{
waitingToRunAt = runningAt;
commandStore.executor().elapsedPreparingToRun.increment(waitingToRunAt - createdAt, runningAt);
}
else if (state.isExecuted())
commandStore.executor().elapsedWaitingToRun.increment(runningAt - waitingToRunAt, runningAt);
commandStore.executor().keys.increment(commandsForKey == null ? 0 : commandsForKey.size(), runningAt);
}
else if (state.isExecuted() && completedAt == 0)
{
completedAt = nanoTime();
}
@ -756,7 +768,17 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
{
if (state == FAILING)
state(FAILED);
Invariants.expect(state.isExecuted());
releaseResources(commandStore.cachesExclusive());
if (runningAt != 0)
{
commandStore.executor().elapsedRunning.increment(completedAt - runningAt, completedAt);
}
if (histogramBuffer != null)
{
histogramBuffer.flush(completedAt);
histogramBuffer = null;
}
}
@Nullable
@ -979,13 +1001,6 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
default: throw new AssertionError("Unhandled Status: " + entry.status());
case WAITING_TO_LOAD:
case LOADING:
if (scanned)
// if we've finished scanning and not already taken a reference we shouldn't need to witness (unless modified)
return;
ensureLoading().put(entry.key(), commandsForKeyCache.acquire(entry));
if (entry.status() == Status.WAITING_TO_LOAD)
ensureWaitingToLoad().add(entry);
entry.loadingOrWaiting().add(AccordTask.this);
return;
case MODIFIED:
@ -995,6 +1010,23 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
case FAILED_TO_SAVE:
if (commandsForKey != null && commandsForKey.containsKey(entry.key()))
return;
Object v = entry.getOrShrunkExclusive();
if (v == null) return;
else if (v instanceof CommandsForKey)
{
if (!summaryLoader.isRelevant((CommandsForKey) v))
return;
}
else
{
TxnId last = CommandSerializers.txnId.deserialize((ByteBuffer) v);
int position = (int)CommandSerializers.txnId.serializedSize(last);
TxnId minUndecided = CommandSerializers.txnId.deserialize((ByteBuffer) v, position);
if (!summaryLoader.isRelevant(entry.key(), last, minUndecided))
return;
}
ensureCommandsForKey().putIfAbsent(entry.key(), commandsForKeyCache.acquire(entry));
}
}

19
src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializer.java
View File

@ -42,6 +42,7 @@ import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.VersionedSerializer;
@ -82,7 +83,7 @@ public class ReadDataSerializer implements IVersionedSerializer<ReadData>
boolean hasExecuteAtEpoch = (read.executeAt() != null ? read.executeAt() : read.txnId).epoch() != read.executeAtEpoch;
out.writeUnsignedVInt32(type.ordinal()
| (read.flags.bits() << 3)
| shiftedExecuteFlags(read.flags)
| (hasTxn ? HAS_TXN : 0)
| (hasExecuteAt ? HAS_EXECUTE_AT : 0)
| (hasExecuteAtEpoch ? HAS_EXECUTE_AT_EPOCH : 0)
@ -154,7 +155,7 @@ public class ReadDataSerializer implements IVersionedSerializer<ReadData>
{
int tmp = in.readUnsignedVInt32();
type = TYPES[tmp & 0x7];
flags = ExecuteFlags.get((tmp >>> 3) & 0x3);
flags = ExecuteFlags.get(unshiftedExecuteFlags(tmp));
hasTxn = (tmp & HAS_TXN) != 0;
hasExecuteAt = (tmp & HAS_EXECUTE_AT) != 0;
hasExecuteAtEpoch = (tmp & HAS_EXECUTE_AT_EPOCH) != 0;
@ -219,6 +220,17 @@ public class ReadDataSerializer implements IVersionedSerializer<ReadData>
}
}
private static int shiftedExecuteFlags(ExecuteFlags flags)
{
Invariants.require(flags.bits() <= 0x7);
return ((flags.bits() & 0x3) << 3) | ((flags.bits() & ~0x3) << 6);
}
private static int unshiftedExecuteFlags(int flags)
{
return ((flags & 0x18) >> 3) | ((flags & 0x100) >> 6);
}
@Override
public long serializedSize(ReadData read, Version version)
{
@ -228,12 +240,13 @@ public class ReadDataSerializer implements IVersionedSerializer<ReadData>
boolean hasExecuteAtEpoch = (read.executeAt() != null ? read.executeAt() : read.txnId).epoch() != read.executeAtEpoch;
long size = VIntCoding.computeUnsignedVIntSize(type.ordinal()
| (read.flags.bits() << 3)
| shiftedExecuteFlags(read.flags)
| (hasTxn ? HAS_TXN : 0)
| (hasExecuteAt ? HAS_EXECUTE_AT : 0)
| (hasExecuteAtEpoch ? HAS_EXECUTE_AT_EPOCH : 0))
+ CommandSerializers.txnId.serializedSize(read.txnId)
+ KeySerializers.participants.serializedSize(read.scope);
if (hasTxn)
size += CommandSerializers.partialTxn.serializedSize(read.partialTxn(), version);
if (hasExecuteAt)

4
test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java
View File

@ -49,6 +49,7 @@ import org.apache.cassandra.metrics.DefaultNameFactory;
import org.apache.cassandra.metrics.RatioGaugeSet;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.accord.AccordExecutor;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.exceptions.AccordReadPreemptedException;
import org.apache.cassandra.service.accord.exceptions.AccordWritePreemptedException;
@ -61,7 +62,6 @@ import static java.lang.String.format;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
public class AccordMetricsTest extends AccordTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordMetricsTest.class);
@ -332,7 +332,6 @@ public class AccordMetricsTest extends AccordTestBase
assertThat(metric.apply(AccordReplicaMetrics.REPLICA_STABLE_LATENCY)).isLessThanOrEqualTo(stable);
assertThat(metric.apply(AccordReplicaMetrics.REPLICA_PREAPPLY_LATENCY)).isEqualTo(executions);
assertThat(metric.apply(AccordReplicaMetrics.REPLICA_APPLY_LATENCY)).isEqualTo(applications);
assertThat(metric.apply(AccordReplicaMetrics.REPLICA_APPLY_DURATION)).isEqualTo(scope.equals("rw") ? applications : 0);
assertThat(metric.apply(AccordReplicaMetrics.REPLICA_DEPENDENCIES)).isEqualTo(executions);
// Verify that replica metrics are published to the appropriate virtual table:
@ -357,6 +356,7 @@ public class AccordMetricsTest extends AccordTestBase
Map<Integer, Map<String, Long>> metrics = new HashMap<>();
for (int i = 0; i < SHARED_CLUSTER.size(); i++)
{
SHARED_CLUSTER.get(i + 1).runOnInstance(() -> AccordExecutor.HISTOGRAMS.refresh());
Map<String, Long> map = SHARED_CLUSTER.get(i + 1).metrics().getCounters(name -> name.startsWith("org.apache.cassandra.metrics.Accord") || (name.startsWith("org.apache.cassandra.metrics.ClientRequest") && (name.endsWith("AccordRead") || name.endsWith("AccordWrite"))));
SHARED_CLUSTER.get(i + 1).metrics().getGauges(name -> name.startsWith("org.apache.cassandra.metrics.Accord"))
.forEach((key, value) -> map.put(key, ((Number)value).longValue()));

188
test/unit/org/apache/cassandra/metrics/LogLinearDecayingHistogramTest.java Normal file
View File

@ -0,0 +1,188 @@
/*
* 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.metrics;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Snapshot;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms.LogLinearDecayingHistogram;
import static org.junit.Assert.assertEquals;
public class LogLinearDecayingHistogramTest
{
public static final Logger logger = LoggerFactory.getLogger(LogLinearDecayingHistogramTest.class);
private static final double DOUBLE_ASSERT_DELTA = 0;
private static final long ONE_SECOND = TimeUnit.NANOSECONDS.toSeconds(1);
private static final long ONE_HOUR = TimeUnit.HOURS.toSeconds(1);
private static final long[] TIMES = new long[] { 0, ONE_SECOND - 1, ONE_SECOND, ONE_HOUR - 1, ONE_HOUR };
@Test
public void testMinMax()
{
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
histogram.increment(16, time);
Snapshot snapshot = histogram.copyToSnapshot(time + 1);
assertEquals(16, snapshot.getMin());
assertEquals(20, snapshot.getMax());
}
}
@Test
public void testMean()
{
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
for (int i = 0; i < 40; i++)
histogram.increment(0, time);
for (int i = 0; i < 20; i++)
histogram.increment(1, time);
for (int i = 0; i < 10; i++)
histogram.increment(2, time);
assertEquals(40/70d, histogram.copyToSnapshot(time + 1).getMean(), 0.1D);
}
}
@Test
public void testStdDev()
{
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
for (int i = 0; i < 20; i++)
histogram.increment(10, time);
for (int i = 0; i < 40; i++)
histogram.increment(20, time);
for (int i = 0; i < 20; i++)
histogram.increment(30, time);
Snapshot snapshot = histogram.copyToSnapshot(time + 1);
assertEquals(20.0D, snapshot.getMean(), 2.0D);
assertEquals(7.07D, snapshot.getStdDev(), 2.0D);
}
}
@Test
public void testPercentile()
{
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
// percentile of empty histogram is 0
assertEquals(0D, histogram.copyToSnapshot(0).getValue(0.99), DOUBLE_ASSERT_DELTA);
assertEquals(0D, histogram.copyToSnapshot(ONE_SECOND).getValue(0.99), DOUBLE_ASSERT_DELTA);
assertEquals(0D, histogram.copyToSnapshot(ONE_HOUR).getValue(0.99), DOUBLE_ASSERT_DELTA);
}
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
histogram.increment(1, time);
// percentile of a histogram with one element should be that element
assertEquals(1D, histogram.copyToSnapshot(time + 1).getValue(0.99), DOUBLE_ASSERT_DELTA);
}
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
histogram.increment(1, time);
histogram.increment(10, time);
assertEquals(10D, histogram.copyToSnapshot(time + 1).getValue(0.99), DOUBLE_ASSERT_DELTA);
}
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
histogram.increment(1, time);
histogram.increment(2, time);
histogram.increment(3, time);
histogram.increment(4, time);
histogram.increment(5, time);
Snapshot snapshot = histogram.copyToSnapshot(time + 1);
assertEquals(1, snapshot.getValue(0.00), DOUBLE_ASSERT_DELTA);
assertEquals(3, snapshot.getValue(0.50), DOUBLE_ASSERT_DELTA);
assertEquals(3, snapshot.getValue(0.60), DOUBLE_ASSERT_DELTA);
assertEquals(5, snapshot.getValue(1.00), DOUBLE_ASSERT_DELTA);
}
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
for (int i = 11; i <= 20; i++)
histogram.increment(i, time);
// Right now the histogram looks like:
// 10 12 14 16 20
// 1 2 2 4 1
// %: 10 30 50 90 100
Snapshot snapshot = histogram.copyToSnapshot(time + 1);
assertEquals(10, snapshot.getValue(0.01), DOUBLE_ASSERT_DELTA);
assertEquals(10, snapshot.getValue(0.10), DOUBLE_ASSERT_DELTA);
assertEquals(11, snapshot.getValue(0.11), DOUBLE_ASSERT_DELTA);
assertEquals(11, snapshot.getValue(0.20), DOUBLE_ASSERT_DELTA);
assertEquals(12, snapshot.getValue(0.21), DOUBLE_ASSERT_DELTA);
assertEquals(12, snapshot.getValue(0.30), DOUBLE_ASSERT_DELTA);
assertEquals(13, snapshot.getValue(0.31), DOUBLE_ASSERT_DELTA);
assertEquals(13, snapshot.getValue(0.40), DOUBLE_ASSERT_DELTA);
assertEquals(14, snapshot.getValue(0.41), DOUBLE_ASSERT_DELTA);
assertEquals(14, snapshot.getValue(0.50), DOUBLE_ASSERT_DELTA);
assertEquals(15, snapshot.getValue(0.51), DOUBLE_ASSERT_DELTA);
assertEquals(15, snapshot.getValue(0.70), DOUBLE_ASSERT_DELTA);
assertEquals(16, snapshot.getValue(0.71), DOUBLE_ASSERT_DELTA);
assertEquals(16, snapshot.getValue(0.90), DOUBLE_ASSERT_DELTA);
assertEquals(20, snapshot.getValue(0.91), DOUBLE_ASSERT_DELTA);
}
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
histogram.increment(0, time);
histogram.increment(0, time);
histogram.increment(1, time);
Snapshot snapshot = histogram.copyToSnapshot(time + 1);
assertEquals(0, snapshot.getValue(0.5), DOUBLE_ASSERT_DELTA);
assertEquals(1, snapshot.getValue(0.99), DOUBLE_ASSERT_DELTA);
}
}
@Test
public void testSize()
{
for (long time : TIMES)
{
LogLinearDecayingHistogram histogram = new LogLinearDecayingHistograms().allocate(1);
histogram.increment(42, time);
histogram.increment(42, time);
assertEquals(2, histogram.copyToSnapshot(time + 1).size());
}
}
}

313
test/unit/org/apache/cassandra/metrics/LogLinearHistogramTest.java Normal file
View File

@ -0,0 +1,313 @@
/*
* 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.metrics;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.RandomTestRunner;
import com.codahale.metrics.Snapshot;
import static org.junit.Assert.assertEquals;
public class LogLinearHistogramTest
{
public static final Logger logger = LoggerFactory.getLogger(LogLinearHistogramTest.class);
private static final double DOUBLE_ASSERT_DELTA = 0;
@Test
public void testMinMax()
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
histogram.increment(16);
Snapshot snapshot = histogram.copyToSnapshot();
assertEquals(16, snapshot.getMin());
assertEquals(20, snapshot.getMax());
}
@Test
public void testMean()
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
for (int i = 0; i < 40; i++)
histogram.increment(0);
for (int i = 0; i < 20; i++)
histogram.increment(1);
for (int i = 0; i < 10; i++)
histogram.increment(2);
assertEquals(40/70d, histogram.copyToSnapshot().getMean(), 0.1D);
}
@Test
public void testStdDev()
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
for (int i = 0; i < 20; i++)
histogram.increment(10);
for (int i = 0; i < 40; i++)
histogram.increment(20);
for (int i = 0; i < 20; i++)
histogram.increment(30);
Snapshot snapshot = histogram.copyToSnapshot();
assertEquals(20.0D, snapshot.getMean(), 2.0D);
assertEquals(7.07D, snapshot.getStdDev(), 2.0D);
}
@Test
public void testPercentile()
{
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
// percentile of empty histogram is 0
assertEquals(0D, histogram.copyToSnapshot().getValue(0.99), DOUBLE_ASSERT_DELTA);
histogram.increment(1);
// percentile of a histogram with one element should be that element
assertEquals(1D, histogram.copyToSnapshot().getValue(0.99), DOUBLE_ASSERT_DELTA);
histogram.increment(10);
assertEquals(10D, histogram.copyToSnapshot().getValue(0.99), DOUBLE_ASSERT_DELTA);
}
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
histogram.increment(1);
histogram.increment(2);
histogram.increment(3);
histogram.increment(4);
histogram.increment(5);
Snapshot snapshot = histogram.copyToSnapshot();
assertEquals(1, snapshot.getValue(0.00), DOUBLE_ASSERT_DELTA);
assertEquals(3, snapshot.getValue(0.50), DOUBLE_ASSERT_DELTA);
assertEquals(3, snapshot.getValue(0.60), DOUBLE_ASSERT_DELTA);
assertEquals(5, snapshot.getValue(1.00), DOUBLE_ASSERT_DELTA);
}
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
for (int i = 11; i <= 20; i++)
histogram.increment(i);
// Right now the histogram looks like:
// 10 12 14 16 20
// 1 2 2 4 1
// %: 10 30 50 90 100
Snapshot snapshot = histogram.copyToSnapshot();
assertEquals(10, snapshot.getValue(0.01), DOUBLE_ASSERT_DELTA);
assertEquals(10, snapshot.getValue(0.10), DOUBLE_ASSERT_DELTA);
assertEquals(11, snapshot.getValue(0.11), DOUBLE_ASSERT_DELTA);
assertEquals(11, snapshot.getValue(0.20), DOUBLE_ASSERT_DELTA);
assertEquals(12, snapshot.getValue(0.21), DOUBLE_ASSERT_DELTA);
assertEquals(12, snapshot.getValue(0.30), DOUBLE_ASSERT_DELTA);
assertEquals(13, snapshot.getValue(0.31), DOUBLE_ASSERT_DELTA);
assertEquals(13, snapshot.getValue(0.40), DOUBLE_ASSERT_DELTA);
assertEquals(14, snapshot.getValue(0.41), DOUBLE_ASSERT_DELTA);
assertEquals(14, snapshot.getValue(0.50), DOUBLE_ASSERT_DELTA);
assertEquals(15, snapshot.getValue(0.51), DOUBLE_ASSERT_DELTA);
assertEquals(15, snapshot.getValue(0.70), DOUBLE_ASSERT_DELTA);
assertEquals(16, snapshot.getValue(0.71), DOUBLE_ASSERT_DELTA);
assertEquals(16, snapshot.getValue(0.90), DOUBLE_ASSERT_DELTA);
assertEquals(20, snapshot.getValue(0.91), DOUBLE_ASSERT_DELTA);
}
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
histogram.increment(0);
histogram.increment(0);
histogram.increment(1);
Snapshot snapshot = histogram.copyToSnapshot();
assertEquals(0, snapshot.getValue(0.5), DOUBLE_ASSERT_DELTA);
assertEquals(1, snapshot.getValue(0.99), DOUBLE_ASSERT_DELTA);
}
}
@Test
public void testSize()
{
LogLinearHistogram histogram = new LogLinearHistogram(1);
histogram.increment(42);
histogram.increment(42);
assertEquals(2, histogram.copyToSnapshot().size());
}
@Test
public void testRandomizedMinMaxProperties()
{
RandomTestRunner.test().check(random -> {
for (int trial = 0; trial < 100; trial++)
{
int maxValue = 500;
LogLinearHistogram histogram = new LogLinearHistogram(maxValue);
List<Long> values = new ArrayList<>();
int numValues = 10 + random.nextInt(100);
for (int i = 0; i < numValues; i++)
{
long value = random.nextInt(maxValue);
histogram.increment(value);
values.add(value);
}
Snapshot snapshot = histogram.copyToSnapshot();
long expectedMin = Collections.min(values);
long expectedMax = Collections.max(values);
assertEquals("" + trial, LogLinearHistogram.invertIndex(LogLinearHistogram.index(expectedMin)),
snapshot.getMin());
assertEquals(LogLinearHistogram.invertIndex(1 + LogLinearHistogram.index(expectedMax)),
snapshot.getMax());
}
});
}
@Test
public void testRandomizedIncrementDecrementConsistency()
{
RandomTestRunner.test().check(random -> {
for (int trial = 0; trial < 50; trial++)
{
int maxValue = 500;
LogLinearHistogram histogram = new LogLinearHistogram(maxValue);
List<Long> values = new ArrayList<>();
// Add random values
int numValues = 20 + random.nextInt(80);
for (int i = 0; i < numValues; i++)
{
long value = random.nextInt(5000);
histogram.increment(value);
values.add(value);
}
Snapshot beforeSnapshot = histogram.copyToSnapshot();
// Remove and re-add some values
int numToModify = 1 + random.nextInt(Math.min(10, values.size()));
for (int i = 0; i < numToModify; i++)
{
int idx = random.nextInt(values.size());
long oldValue = values.get(idx);
long newValue = random.nextInt(5000);
histogram.decrement(oldValue);
histogram.increment(newValue);
values.set(idx, newValue);
}
Snapshot afterSnapshot = histogram.copyToSnapshot();
assertEquals(beforeSnapshot.size(), afterSnapshot.size());
long expectedMin = Collections.min(values);
long expectedMax = Collections.max(values);
assertEquals(LogLinearHistogram.invertIndex(LogLinearHistogram.index(expectedMin)),
afterSnapshot.getMin());
assertEquals(LogLinearHistogram.invertIndex(1 + LogLinearHistogram.index(expectedMax)),
afterSnapshot.getMax());
}
});
}
@Test
public void testRandomizedReplaceConsistency()
{
RandomTestRunner.test().check(random -> {
for (int trial = 0; trial < 100; trial++)
{
int maxValue = 500;
LogLinearHistogram histogram = new LogLinearHistogram(maxValue);
List<Long> values = new ArrayList<>();
int numValues = 20 + random.nextInt(80);
for (int i = 0; i < numValues; i++)
{
long value = random.nextInt(maxValue);
histogram.increment(value);
values.add(value);
}
Snapshot beforeSnapshot = histogram.copyToSnapshot();
int numToReplace = 1 + random.nextInt(Math.min(10, values.size()));
for (int i = 0; i < numToReplace; i++)
{
int idx = random.nextInt(values.size());
long oldValue = values.get(idx);
long newValue = random.nextInt(maxValue);
histogram.replace(oldValue, newValue);
values.set(idx, newValue);
}
Snapshot afterSnapshot = histogram.copyToSnapshot();
assertEquals(beforeSnapshot.size(), afterSnapshot.size());
long expectedMin = Collections.min(values);
long expectedMax = Collections.max(values);
assertEquals(LogLinearHistogram.invertIndex(LogLinearHistogram.index(expectedMin)),
afterSnapshot.getMin());
assertEquals(LogLinearHistogram.invertIndex(1 + LogLinearHistogram.index(expectedMax)),
afterSnapshot.getMax());
}
});
}
@Test
public void testRandomizedAutoGrowth()
{
RandomTestRunner.test().check(random -> {
for (int trial = 0; trial < 20; trial++)
{
int maxValue = 500;
LogLinearHistogram histogram = new LogLinearHistogram(maxValue);
List<Long> values = new ArrayList<>();
int numValues = 50;
for (int i = 0; i < numValues; i++)
{
long maxRange = (long) Math.pow(2, i / 5.0);
long value = random.nextInt((int) Math.min(maxRange, Integer.MAX_VALUE));
histogram.increment(value);
values.add(value);
}
Snapshot snapshot = histogram.copyToSnapshot();
assertEquals(numValues, snapshot.size());
long expectedMin = Collections.min(values);
long expectedMax = Collections.max(values);
assertEquals(LogLinearHistogram.invertIndex(LogLinearHistogram.index(expectedMin)),
snapshot.getMin());
assertEquals(LogLinearHistogram.invertIndex(1 + LogLinearHistogram.index(expectedMax)),
snapshot.getMax());
}
});
}
}

1
test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
View File

@ -69,7 +69,6 @@ import static org.apache.cassandra.service.accord.AccordTestUtils.txnId;
public class AccordCommandTest
{
static final AtomicLong clock = new AtomicLong(0);
private static final Node.Id ID1 = new Node.Id(1);
private static final Node.Id ID2 = new Node.Id(2);