diff --git a/modules/accord b/modules/accord index de03d14e48..938ba19ada 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit de03d14e4821b669fecad5dcdb2de2dba3a9d926 +Subproject commit 938ba19adaf70bf9901fb2dc177dce313d7ee15c diff --git a/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java b/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java index 5285bd914e..10fbfd6621 100644 --- a/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java @@ -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); diff --git a/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java b/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java index c4a37ebf2e..1500517bcb 100644 --- a/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java @@ -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; } diff --git a/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java b/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java new file mode 100644 index 0000000000..c916c3239a --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java @@ -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 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 preparingToRun; + public final Gauge waitingToRun; + public final Gauge 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)); + } +} diff --git a/src/java/org/apache/cassandra/metrics/AccordReplicaMetrics.java b/src/java/org/apache/cassandra/metrics/AccordReplicaMetrics.java index 91973e4a95..8461789771 100644 --- a/src/java/org/apache/cassandra/metrics/AccordReplicaMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordReplicaMetrics.java @@ -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())); } } } diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index 756144c84d..fc50797005 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java @@ -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) diff --git a/src/java/org/apache/cassandra/metrics/LogLinearDecayingHistograms.java b/src/java/org/apache/cassandra/metrics/LogLinearDecayingHistograms.java new file mode 100644 index 0000000000..3d0279c0ca --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/LogLinearDecayingHistograms.java @@ -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 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; + } +} diff --git a/src/java/org/apache/cassandra/metrics/LogLinearHistogram.java b/src/java/org/apache/cassandra/metrics/LogLinearHistogram.java index 43420cf7f6..e6be4f9017 100644 --- a/src/java/org/apache/cassandra/metrics/LogLinearHistogram.java +++ b/src/java/org/apache/cassandra/metrics/LogLinearHistogram.java @@ -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,23 +261,19 @@ public class LogLinearHistogram public double getStdDev() { if (totalCount <= 1) - { return 0.0D; - } - else + + double mean = this.getMean(); + double sum = 0.0D; + + for(int i = 0; i < raw.length; ++i) { - double mean = this.getMean(); - double sum = 0.0D; - - for(int i = 0; i < raw.length; ++i) - { - long value = invertIndex(i); - double diff = value - mean; - sum += diff * diff * raw[i]; - } - - return Math.sqrt(sum / (totalCount - 1)); + long value = invertIndex(i); + double diff = value - mean; + sum += diff * diff * raw[i]; } + + return Math.sqrt(sum / (totalCount - 1)); } @Override @@ -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; diff --git a/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java b/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java new file mode 100644 index 0000000000..db91857b3c --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java @@ -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 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 shards = new ArrayList<>(); + final List 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 snapshot = Collections.emptyList(); + + public synchronized void refresh() + { + refresh(Clock.Global.nanoTime()); + } + + private synchronized List refresh(long now) + { + List 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 maybeRefresh() + { + long now = Clock.Global.nanoTime(); + if (snapshot.size() == histograms.size() && snapshotAt + REFRESH_RATE >= now) + return snapshot; + + return refresh(now); + } +} diff --git a/src/java/org/apache/cassandra/metrics/ShardedHistogram.java b/src/java/org/apache/cassandra/metrics/ShardedHistogram.java index 1677edad81..66defdac56 100644 --- a/src/java/org/apache/cassandra/metrics/ShardedHistogram.java +++ b/src/java/org/apache/cassandra/metrics/ShardedHistogram.java @@ -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(); diff --git a/src/java/org/apache/cassandra/metrics/ShardedLongGauges.java b/src/java/org/apache/cassandra/metrics/ShardedLongGauges.java new file mode 100644 index 0000000000..d2721a94c6 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/ShardedLongGauges.java @@ -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 +{ + private static final long REFRESH_RATE = TimeUnit.SECONDS.toNanos(15); + + public class ShardedLongGauge implements Gauge + { + final int gaugeIndex; + final ToLongFunction compute; + final LongBinaryOperator reduce; + + private ShardedLongGauge(int gaugeIndex, ToLongFunction compute, LongBinaryOperator reduce) + { + this.gaugeIndex = gaugeIndex; + this.compute = compute; + this.reduce = reduce; + } + + @Override + public Long getValue() + { + return maybeRefresh()[gaugeIndex]; + } + } + + static class LongGaugeShard + { + final Lock lock; + final T shard; + + LongGaugeShard(Lock lock, T shard) + { + this.lock = lock; + this.shard = shard; + } + + public void init(long[] init, List.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.ShardedLongGauge> gauges) + { + lock.lock(); + try + { + for (int i = 0 ; i < update.length ; ++i) + { + ShardedLongGauges.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> shards = new ArrayList<>(); + final List gauges = new ArrayList<>(); + + public synchronized void newShard(Lock guardedBy, S shard) + { + shards.add(new LongGaugeShard<>(guardedBy, shard)); + } + + public synchronized ShardedLongGauge newGauge(ToLongFunction 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); + } +} diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java index 0a1dd623f0..c17ccc879c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCache.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java @@ -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> 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 diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 5c1df8f136..99b763de21 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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, diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index 2fa8675563..c9731d101a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -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, 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 name, Agent agent); @@ -159,6 +167,13 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor waitingForQuiescence; private Queue 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 { executeDirectlyWithLock(cache::processNoEvictQueue); }, 1L, 1L, TimeUnit.SECONDS); @@ -1761,4 +1784,18 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor keysOrRanges, Predicate forEach) { Map commandsForKey = task.commandsForKey; diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index d2b87d25b1..2234ca00f4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java @@ -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 extends SubmittableTask implements Function< @Nullable Object2ObjectHashMap commands; @Nullable Object2ObjectHashMap commandsForKey; @Nullable Object2ObjectHashMap> loading; + LogLinearDecayingHistograms.Buffer histogramBuffer; // TODO (desired): collection supporting faster deletes but still fast poll (e.g. some ordered collection) @Nullable ArrayDeque> waitingToLoad; @Nullable RangeTxnScanner rangeScanner; @@ -299,12 +303,20 @@ public abstract class AccordTask 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); + } + commandStore.executor().elapsedWaitingToRun.increment(runningAt - waitingToRunAt, runningAt); + commandStore.executor().keys.increment(commandsForKey == null ? 0 : commandsForKey.size(), runningAt); } - else if (state.isExecuted()) + else if (state.isExecuted() && completedAt == 0) { completedAt = nanoTime(); } @@ -756,7 +768,17 @@ public abstract class AccordTask 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 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 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)); } } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializer.java index 34ab52fc2d..8c1bd7eca8 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializer.java @@ -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 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 { 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 } } + 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 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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java index 5c78547432..ff47950f7a 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java @@ -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> metrics = new HashMap<>(); for (int i = 0; i < SHARED_CLUSTER.size(); i++) { + SHARED_CLUSTER.get(i + 1).runOnInstance(() -> AccordExecutor.HISTOGRAMS.refresh()); Map 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())); diff --git a/test/unit/org/apache/cassandra/metrics/LogLinearDecayingHistogramTest.java b/test/unit/org/apache/cassandra/metrics/LogLinearDecayingHistogramTest.java new file mode 100644 index 0000000000..31ca6fbb54 --- /dev/null +++ b/test/unit/org/apache/cassandra/metrics/LogLinearDecayingHistogramTest.java @@ -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()); + } + } +} diff --git a/test/unit/org/apache/cassandra/metrics/LogLinearHistogramTest.java b/test/unit/org/apache/cassandra/metrics/LogLinearHistogramTest.java new file mode 100644 index 0000000000..c7954e2704 --- /dev/null +++ b/test/unit/org/apache/cassandra/metrics/LogLinearHistogramTest.java @@ -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 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 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 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 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()); + } + }); + } +} diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java index ac322403d1..f43ddb66e4 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java @@ -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);