diff --git a/modules/accord b/modules/accord index 8ccce74581..6bae51f6a4 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 8ccce745818cf80c7cff82c3554e4a88e9e540db +Subproject commit 6bae51f6a4bd560a82840fa0809cd1d630696cc0 diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 0c4eb44496..09100da8d3 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -5505,7 +5505,7 @@ public class DatabaseDescriptor public static boolean getAccordTransactionsEnabled() { - return conf == null ? false : conf.accord.enabled; + return conf != null && conf.accord.enabled; } public static void setAccordTransactionsEnabled(boolean b) diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index 2a96bcc16d..caff467cd7 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -132,7 +132,7 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement, public static final String WRITE_TXN_EMPTY_WITH_IGNORED_READS = "Write txn produced no mutation, and its reads do not return to the caller; ignoring..."; public static final String WRITE_TXN_EMPTY_WITH_NO_READS = "Write txn produced no mutation, and had no reads; ignoring..."; - private static NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(LoggerFactory.getLogger(TransactionStatement.class), 1, TimeUnit.MINUTES); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(LoggerFactory.getLogger(TransactionStatement.class), 1, TimeUnit.MINUTES); static class NamedSelect { diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index de989d7f6e..c3f08de148 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -2186,7 +2186,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace Map> startLookup = null; for (ActiveEpoch epoch : snapshot) { - Topology topology = epoch.global(); + Topology topology = epoch.all(); for (Shard shard : topology.shards()) { Range range = shard.range; diff --git a/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java b/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java index 21f1297584..3df852af9d 100644 --- a/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java @@ -90,7 +90,7 @@ public class AccordCacheMetrics public AccordCacheMetrics(String subTypeName) { DefaultNameFactory factory = new DefaultNameFactory(ACCORD_CACHE, subTypeName); - this.objectSize = Metrics.shardedHistogram(factory.createMetricName("EntrySize")); + this.objectSize = Metrics.shardedHistogram(factory.createMetricName("EntrySize"), false); this.hits = Metrics.gauge(factory.createMetricName("Hits"), hitRate::totalHits); this.misses = Metrics.gauge(factory.createMetricName("Misses"), hitRate::totalMisses); this.requests = Metrics.gauge(factory.createMetricName("Requests"), hitRate::totalRequests); diff --git a/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java b/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java index 1500517bcb..12c736614f 100644 --- a/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java @@ -52,6 +52,7 @@ public class AccordCoordinatorMetrics public static final String COORDINATOR_PREACCEPT_LATENCY = "PreAcceptLatency"; public static final String COORDINATOR_EXECUTE_LATENCY = "ExecuteLatency"; public static final String COORDINATOR_APPLY_LATENCY = "ApplyLatency"; + public static final String EPHEMERAL = "Ephemeral"; public static final String FAST_PATHS = "FastPaths"; public static final String MEDIUM_PATHS = "MediumPaths"; public static final String SLOW_PATHS = "SlowPaths"; @@ -98,6 +99,11 @@ public class AccordCoordinatorMetrics */ public final Histogram tables; + /** + * The number of ephemeral transactions executed on this coordinator. + */ + public final Meter ephemeral; + /** * The number of fast path transactions executed on this coordinator. */ @@ -159,6 +165,7 @@ public class AccordCoordinatorMetrics keys = Metrics.histogram(coordinator.createMetricName(COORDINATOR_KEYS), true); tables = Metrics.histogram(coordinator.createMetricName(COORDINATOR_TABLES), true); + ephemeral = Metrics.meter(coordinator.createMetricName(EPHEMERAL)); fastPaths = Metrics.meter(coordinator.createMetricName(FAST_PATHS)); mediumPaths = Metrics.meter(coordinator.createMetricName(MEDIUM_PATHS)); slowPaths = Metrics.meter(coordinator.createMetricName(SLOW_PATHS)); @@ -168,7 +175,7 @@ public class AccordCoordinatorMetrics invalidations = Metrics.meter(coordinator.createMetricName(INVALIDATIONS)); recoveryDelay = Metrics.timer(coordinator.createMetricName(RECOVERY_DELAY)); recoveryDuration = Metrics.timer(coordinator.createMetricName(RECOVERY_TIME)); - fastPathToTotal = new RatioGaugeSet(fastPaths, RatioGaugeSet.sum(fastPaths, mediumPaths, slowPaths), coordinator, FAST_PATH_TO_TOTAL + ".%s"); + fastPathToTotal = new RatioGaugeSet(fastPaths, RatioGaugeSet.sum(ephemeral, fastPaths, mediumPaths, slowPaths), coordinator, FAST_PATH_TO_TOTAL + ".%s"); } @Override @@ -238,6 +245,7 @@ public class AccordCoordinatorMetrics { switch (path) { + case EPHEMERAL: metrics.ephemeral.mark(); break; case FAST: metrics.fastPaths.mark(); break; case MEDIUM: metrics.mediumPaths.mark(); break; case SLOW: metrics.slowPaths.mark(); break; diff --git a/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java b/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java index c916c3239a..94d8e43ab5 100644 --- a/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordExecutorMetrics.java @@ -58,4 +58,8 @@ public class AccordExecutorMetrics waitingToRun = Metrics.register(factory.createMetricName("WaitingToRun"), gauges.newGauge(AccordExecutor::unsafeWaitingToRunCount, Long::sum)); running = Metrics.register(factory.createMetricName("Running"), gauges.newGauge(AccordExecutor::unsafeRunningCount, Long::sum)); } + + public static void touch() + { + } } diff --git a/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java b/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java index 313ed92bb2..1ef552588f 100644 --- a/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java @@ -21,12 +21,14 @@ package org.apache.cassandra.metrics; import java.lang.reflect.Field; import java.util.concurrent.TimeUnit; +import accord.api.SystemEventListener; import accord.impl.progresslog.DefaultProgressLog; import accord.local.MaxDecidedRX; import accord.local.RedundantBefore; import accord.primitives.TxnId; import accord.topology.TopologyManager; import accord.utils.Invariants; +import com.codahale.metrics.Counter; import com.codahale.metrics.Counting; import com.codahale.metrics.Gauge; import org.apache.cassandra.metrics.LogLinearHistogram.LogLinearSnapshot; @@ -44,7 +46,7 @@ import static org.apache.cassandra.metrics.AccordMetricUtils.fromDurabilityServi import static org.apache.cassandra.metrics.AccordMetricUtils.fromTopologyManager; import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; -public class AccordSystemMetrics +public class AccordSystemMetrics implements SystemEventListener { public final static AccordSystemMetrics metrics = new AccordSystemMetrics(); private static final long REFRESH_RATE = TimeUnit.SECONDS.toNanos(30); @@ -52,6 +54,10 @@ public class AccordSystemMetrics public static final String ACCORD_SYSTEM = "AccordSystem"; public static final String MIN_EPOCH = "MinEpoch"; public static final String MAX_EPOCH = "MaxEpoch"; + public static final String MAX_PENDING_EPOCH = "MaxPendingEpoch"; + public static final String ERRORS = "Errors"; + public static final String EPOCH_WAITS = "EpochWaits"; + public static final String EPOCH_TIMEOUTS = "EpochTimeouts"; public static final String PROGRESS_LOG_ACTIVE = "ProgressLogActive"; public static final String PROGRESS_LOG_SIZE = "ProgressLogSize"; public static final String PROGRESS_LOG_AGE = "ProgressLogAge"; @@ -66,6 +72,10 @@ public class AccordSystemMetrics public final Gauge minEpoch; public final Gauge maxEpoch; + public final Gauge maxPendingEpoch; + public final Counter errors; + public final Counter epochWaits; + public final Counter epochTimeouts; public final Gauge progressLogActive; public final Gauge durabilityQueueActive; public final Gauge durabilityQueuePending; @@ -142,17 +152,33 @@ public class AccordSystemMetrics DefaultNameFactory factory = new DefaultNameFactory(ACCORD_SYSTEM); minEpoch = Metrics.gauge(factory.createMetricName(MIN_EPOCH), fromTopologyManager(TopologyManager::minEpoch)); maxEpoch = Metrics.gauge(factory.createMetricName(MAX_EPOCH), fromTopologyManager(TopologyManager::epoch)); + maxPendingEpoch = Metrics.gauge(factory.createMetricName(MAX_PENDING_EPOCH), fromTopologyManager(TopologyManager::pendingEpoch)); + errors = Metrics.counter(factory.createMetricName(ERRORS)); + epochTimeouts = Metrics.counter(factory.createMetricName(EPOCH_TIMEOUTS)); + epochWaits = Metrics.counter(factory.createMetricName(EPOCH_WAITS)); durabilityQueueActive = Metrics.gauge(factory.createMetricName(DURABILITY_QUEUE_ACTIVE), fromDurabilityService(durability -> (long)durability.queue().activeCount())); durabilityQueuePending = Metrics.gauge(factory.createMetricName(DURABILITY_QUEUE_PENDING), fromDurabilityService(durability -> (long)durability.queue().pendingCount())); progressLogActive = Metrics.gauge(factory.createMetricName(PROGRESS_LOG_ACTIVE), fromDurabilityService(durability -> (long)durability.queue().activeCount())); - progressLogSize = Metrics.onDemandHistogram(factory.createMetricName(PROGRESS_LOG_SIZE), () -> maybeRefreshHistograms().progressLogSize); - progressLogAge = Metrics.onDemandHistogram(factory.createMetricName(PROGRESS_LOG_AGE), () -> maybeRefreshHistograms().progressLogAge); - syncPointAgreedLag = Metrics.onDemandHistogram(factory.createMetricName(SYNCPOINT_AGREED_LAG), () -> maybeRefreshHistograms().syncPointAgreedLag); - locallyAppliedLag = Metrics.onDemandHistogram(factory.createMetricName(LOCALLY_APPLIED_LAG), () -> maybeRefreshHistograms().locallyAppliedLag); - locallyDurableLag = Metrics.onDemandHistogram(factory.createMetricName(LOCALLY_DURABLE_LAG), () -> maybeRefreshHistograms().locallyDurableLag); - quorumAppliedLag = Metrics.onDemandHistogram(factory.createMetricName(QUORUM_APPLIED_LAG), () -> maybeRefreshHistograms().quorumAppliedLag); - shardAppliedLag = Metrics.onDemandHistogram(factory.createMetricName(SHARD_APPLIED_LAG), () -> maybeRefreshHistograms().shardAppliedLag); - gcLag = Metrics.onDemandHistogram(factory.createMetricName(GC_LAG), () -> maybeRefreshHistograms().gcLag); + progressLogSize = Metrics.onDemandHistogram(factory.createMetricName(PROGRESS_LOG_SIZE), () -> maybeRefreshHistograms().progressLogSize, false); + progressLogAge = Metrics.onDemandHistogram(factory.createMetricName(PROGRESS_LOG_AGE), () -> maybeRefreshHistograms().progressLogAge, false); + syncPointAgreedLag = Metrics.onDemandHistogram(factory.createMetricName(SYNCPOINT_AGREED_LAG), () -> maybeRefreshHistograms().syncPointAgreedLag, false); + locallyAppliedLag = Metrics.onDemandHistogram(factory.createMetricName(LOCALLY_APPLIED_LAG), () -> maybeRefreshHistograms().locallyAppliedLag, false); + locallyDurableLag = Metrics.onDemandHistogram(factory.createMetricName(LOCALLY_DURABLE_LAG), () -> maybeRefreshHistograms().locallyDurableLag, false); + quorumAppliedLag = Metrics.onDemandHistogram(factory.createMetricName(QUORUM_APPLIED_LAG), () -> maybeRefreshHistograms().quorumAppliedLag, false); + shardAppliedLag = Metrics.onDemandHistogram(factory.createMetricName(SHARD_APPLIED_LAG), () -> maybeRefreshHistograms().shardAppliedLag, false); + gcLag = Metrics.onDemandHistogram(factory.createMetricName(GC_LAG), () -> maybeRefreshHistograms().gcLag, false); + } + + @Override + public void onWaitingForEpoch(long epoch) + { + epochWaits.inc(); + } + + @Override + public void onTimeoutForEpoch(long epoch, int count) + { + epochTimeouts.inc(count); } private synchronized Snapshot maybeRefreshHistograms() @@ -190,6 +216,8 @@ public class AccordSystemMetrics for (int i = 0 ; i < redundantBefore.size() ; ++i) { RedundantBefore.Bounds bounds = redundantBefore.valueAt(i); + if (bounds == null) + continue; builder.locallyAppliedLag.increment(ageSeconds(nowSeconds, bounds.maxBound(LOCALLY_APPLIED))); builder.locallyDurableLag.increment(ageSeconds(nowSeconds, bounds.maxBoundBoth(LOCALLY_DURABLE_TO_DATA_STORE, LOCALLY_DURABLE_TO_COMMAND_STORE))); builder.quorumAppliedLag.increment(ageSeconds(nowSeconds, bounds.maxBound(QUORUM_APPLIED))); diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index fc50797005..c11fd5d018 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java @@ -48,7 +48,6 @@ import com.codahale.metrics.Metered; import com.codahale.metrics.Metric; import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.MetricSet; -import com.codahale.metrics.Timer; import org.apache.cassandra.db.virtual.CollectionVirtualTableAdapter; import org.apache.cassandra.db.virtual.VirtualTable; import org.apache.cassandra.db.virtual.model.CounterMetricRow; @@ -177,8 +176,8 @@ public class CassandraMetricsRegistry extends MetricRegistry return Double.toString(((OverrideHistogram) metric).getSnapshot().getMedian()); else if (metric instanceof Meter) return Long.toString(((Meter) metric).getCount()); - else if (metric instanceof Timer) - return Long.toString(((Timer) metric).getCount()); + else if (metric instanceof com.codahale.metrics.Timer) + return Long.toString(((com.codahale.metrics.Timer) metric).getCount()); else throw new IllegalStateException("Unknown metric type: " + metric.getClass().getName()); } @@ -357,14 +356,14 @@ public class CassandraMetricsRegistry extends MetricRegistry return histogram; } - public ShardedHistogram shardedHistogram(MetricName name) + public ShardedHistogram shardedHistogram(MetricName name, boolean isCumulative) { - return register(name, new ShardedHistogram()); + return register(name, new ShardedHistogram(isCumulative)); } - public OnDemandHistogram onDemandHistogram(MetricName name, Supplier snapshot) + public OnDemandHistogram onDemandHistogram(MetricName name, Supplier snapshot, boolean isCumulative) { - return register(name, new OnDemandHistogram(snapshot)); + return register(name, new OnDemandHistogram(snapshot, isCumulative)); } public > T gauge(MetricName name, T gauge) @@ -379,7 +378,7 @@ public class CassandraMetricsRegistry extends MetricRegistry return gaugeLoc; } - public Timer timer(MetricName name) + public SnapshottingTimer timer(MetricName name) { return timer(name, DEFAULT_TIMER_UNIT); } @@ -773,17 +772,22 @@ public class CassandraMetricsRegistry extends MetricRegistry } /** - * Returns a histogram describing the values recorded since the last time this method was called. + * If the Histogram has cumulative data, returns a histogram describing the values recorded since the last time this method was called. * - * ex. If the counts are [0, 1, 2, 1] at the time the first caller arrives, but change to [1, 2, 3, 2] by the + * ex. If the counts are [0, 1, 2, 1] at the time the first caller arrives, but change to [1, 2, 3, 2] by the * time a second caller arrives, the second caller will receive [1, 1, 1, 1]. * + * If the Histogram does not have cumulative data, simply returns the current snapshot. + * * @return a histogram whose bucket offsets are assumed to be in nanoseconds */ @Override public synchronized long[] getRecentValues() { - long[] now = metric.getSnapshot().getValues(); + long[] now = values(); + if (!metric.isCumulative()) + return now; + long[] delta = delta(now, last); last = now; return delta; @@ -968,6 +972,12 @@ public class CassandraMetricsRegistry extends MetricRegistry long[] getRecentValues(); String getDurationUnit(); + + String bucketsId(); + + long[] rawBuckets(int count); + + long[] rawValues(); } static class JmxTimer extends JmxMeter implements JmxTimerMBean @@ -1049,7 +1059,10 @@ public class CassandraMetricsRegistry extends MetricRegistry @Override public long[] values() { - return metric.getSnapshot().getValues(); + long[] values = metric.getSnapshot().getValues(); + if (metric.bucketStrategy() == CassandraReservoir.BucketStrategy.log_linear) + values = metric.bucketStrategy().translateTo(CassandraReservoir.BucketStrategy.exp_12_nozero, values); + return values; } /** @@ -1063,7 +1076,10 @@ public class CassandraMetricsRegistry extends MetricRegistry @Override public synchronized long[] getRecentValues() { - long[] now = metric.getSnapshot().getValues(); + long[] now = values(); + if (!metric.isCumulative()) + return now; + long[] delta = delta(now, last); last = now; return delta; @@ -1074,6 +1090,24 @@ public class CassandraMetricsRegistry extends MetricRegistry { return durationUnit; } + + @Override + public String bucketsId() + { + return metric.bucketStrategy().name(); + } + + @Override + public long[] rawBuckets(int count) + { + return metric.bucketStarts(count); + } + + @Override + public long[] rawValues() + { + return metric.getSnapshot().getValues(); + } } /** diff --git a/src/java/org/apache/cassandra/metrics/LatencyMetrics.java b/src/java/org/apache/cassandra/metrics/LatencyMetrics.java index e4abd40bbf..dda9844752 100644 --- a/src/java/org/apache/cassandra/metrics/LatencyMetrics.java +++ b/src/java/org/apache/cassandra/metrics/LatencyMetrics.java @@ -24,9 +24,7 @@ import java.util.concurrent.TimeUnit; import com.google.common.collect.Lists; import com.codahale.metrics.Counter; -import com.codahale.metrics.Reservoir; import com.codahale.metrics.Snapshot; -import com.codahale.metrics.Timer; import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; @@ -169,12 +167,11 @@ public class LatencyMetrics Metrics.remove(factory.createMetricName(namePrefix + "TotalLatency")); } - public class LatencyMetricsTimer extends Timer + public class LatencyMetricsTimer extends OverrideTimer implements org.apache.cassandra.metrics.Timer { - long releasedLatencyCount = 0; - public LatencyMetricsTimer(Reservoir reservoir) + public LatencyMetricsTimer(CassandraReservoir reservoir) { super(reservoir); } diff --git a/src/java/org/apache/cassandra/metrics/OnDemandHistogram.java b/src/java/org/apache/cassandra/metrics/OnDemandHistogram.java index 4ad890e3c0..f00966824b 100644 --- a/src/java/org/apache/cassandra/metrics/OnDemandHistogram.java +++ b/src/java/org/apache/cassandra/metrics/OnDemandHistogram.java @@ -28,9 +28,11 @@ import static org.apache.cassandra.metrics.CassandraReservoir.BucketStrategy.log public class OnDemandHistogram extends OverrideHistogram { final Supplier snapshot; - protected OnDemandHistogram(Supplier snapshot) + final boolean isCumulative; + protected OnDemandHistogram(Supplier snapshot, boolean isCumulative) { this.snapshot = snapshot; + this.isCumulative = isCumulative; } @Override @@ -56,4 +58,10 @@ public class OnDemandHistogram extends OverrideHistogram { return LogLinearHistogram.bucketsWithLength(length); } + + @Override + public boolean isCumulative() + { + return isCumulative; + } } diff --git a/src/java/org/apache/cassandra/metrics/OverrideHistogram.java b/src/java/org/apache/cassandra/metrics/OverrideHistogram.java index 2553d5e01f..e45480247d 100644 --- a/src/java/org/apache/cassandra/metrics/OverrideHistogram.java +++ b/src/java/org/apache/cassandra/metrics/OverrideHistogram.java @@ -21,7 +21,7 @@ package org.apache.cassandra.metrics; import com.codahale.metrics.Snapshot; import org.agrona.UnsafeAccess; -public abstract class OverrideHistogram extends com.codahale.metrics.Histogram +public class OverrideHistogram extends com.codahale.metrics.Histogram { private static final CassandraReservoir NO_RESERVOIR = new CassandraReservoir() { @Override public Snapshot getPercentileSnapshot() { return null; } @@ -61,4 +61,9 @@ public abstract class OverrideHistogram extends com.codahale.metrics.Histogram { return reservoir.buckets(length); } + + public boolean isCumulative() + { + return true; + } } diff --git a/src/java/org/apache/cassandra/metrics/OverrideMeter.java b/src/java/org/apache/cassandra/metrics/OverrideMeter.java new file mode 100644 index 0000000000..4daae7e1b7 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/OverrideMeter.java @@ -0,0 +1,29 @@ +/* + * 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 com.codahale.metrics.Clock; + +public class OverrideMeter extends com.codahale.metrics.Meter implements Meter +{ + public OverrideMeter(Clock clock) + { + super(clock); + } +} diff --git a/src/java/org/apache/cassandra/metrics/OverrideTimer.java b/src/java/org/apache/cassandra/metrics/OverrideTimer.java new file mode 100644 index 0000000000..e2fe1e45fd --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/OverrideTimer.java @@ -0,0 +1,239 @@ +/* + * 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.time.Duration; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +import com.codahale.metrics.ExponentiallyDecayingReservoir; +import com.codahale.metrics.Reservoir; +import com.codahale.metrics.Snapshot; + +/** + * An alternative to Dropwizard Timer which implements the same kind of API. + * it has more efficent latency histogram implementation and consumes less memory. + * + * NOTE: Dropwizard Timer is a concrete class and there is no an interface for Dropwizard Timer logic, + * so we have to create an alternative hierarchy. + */ +public class OverrideTimer extends com.codahale.metrics.Timer implements Timer +{ + protected final Meter meter; + protected final OverrideHistogram histogram; + // usually we need precise clocks for timing, so we do not replace it with an approximate version + protected final MetricClock clock; + + /** + * Creates a new {@link Timer} using an {@link ExponentiallyDecayingReservoir} and the default + * {@link MetricClock}. + */ + public OverrideTimer() + { + this(new DecayingEstimatedHistogramReservoir()); + } + + /** + * Creates a new {@link Timer} that uses the given {@link Reservoir}. + * + * @param reservoir the {@link Reservoir} implementation the timer should use + */ + public OverrideTimer(CassandraReservoir reservoir) + { + this(reservoir, MetricClock.defaultClock()); + } + + /** + * Creates a new {@link Timer} that uses the given {@link Reservoir} and {@link MetricClock}. + * + * @param reservoir the {@link Reservoir} implementation the timer should use + * @param clock the {@link MetricClock} implementation the timer should use + */ + public OverrideTimer(CassandraReservoir reservoir, MetricClock clock) + { + // the precise clock is intentionally not propagated to ThreadLocalMeter + // we do not need a precise and more expensive time within the meter + this(new OverrideMeter(clock), new OverrideHistogram(reservoir), clock); + } + + public OverrideTimer(Meter meter, OverrideHistogram histogram, MetricClock clock) + { + // original Codahale meter and histogram are set to null to reduce memory footprint + super(null, null, clock); + this.meter = meter; + this.histogram = histogram; + this.clock = clock; + } + + /** + * Adds a recorded duration. + * + * @param duration the length of the duration + * @param unit the scale unit of {@code duration} + */ + @Override + public void update(long duration, TimeUnit unit) + { + update(unit.toNanos(duration)); + } + + /** + * Adds a recorded duration. + * + * @param duration the {@link Duration} to add to the timer. Negative or zero value are ignored. + */ + @Override + public void update(Duration duration) + { + update(duration.toNanos()); + } + + /** + * Times and records the duration of event. + * + * @param event a {@link Callable} whose {@link Callable#call()} method implements a process + * whose duration should be timed + * @param the type of the value returned by {@code event} + * @return the value returned by {@code event} + * @throws Exception if {@code event} throws an {@link Exception} + */ + @Override + public T time(Callable event) throws Exception + { + final long startTime = clock.getTick(); + try + { + return event.call(); + } + finally + { + update(clock.getTick() - startTime); + } + } + + /** + * Times and records the duration of event. Should not throw exceptions, for that use the + * {@link #time(Callable)} method. + * + * @param event a {@link Supplier} whose {@link Supplier#get()} method implements a process + * whose duration should be timed + * @param the type of the value returned by {@code event} + * @return the value returned by {@code event} + */ + @Override + public T timeSupplier(Supplier event) + { + final long startTime = clock.getTick(); + try + { + return event.get(); + } + finally + { + update(clock.getTick() - startTime); + } + } + + /** + * Times and records the duration of event. + * + * @param event a {@link Runnable} whose {@link Runnable#run()} method implements a process + * whose duration should be timed + */ + @Override + public void time(Runnable event) + { + final long startTime = clock.getTick(); + try + { + event.run(); + } + finally + { + update(clock.getTick() - startTime); + } + } + + @Override + public Timer.Context startTime() + { + return new Timer.Context(this, clock); + } + + @Override + public long getCount() + { + return histogram.getCount(); + } + + @Override + public double getFifteenMinuteRate() + { + return meter.getFifteenMinuteRate(); + } + + @Override + public double getFiveMinuteRate() + { + return meter.getFiveMinuteRate(); + } + + @Override + public double getMeanRate() + { + return meter.getMeanRate(); + } + + @Override + public double getOneMinuteRate() + { + return meter.getOneMinuteRate(); + } + + @Override + public Snapshot getSnapshot() + { + return histogram.getSnapshot(); + } + + private void update(long duration) + { + if (duration >= 0) + { + histogram.update(duration); + meter.mark(); + } + } + + public CassandraReservoir.BucketStrategy bucketStrategy() + { + return histogram.bucketStrategy(); + } + + public long[] bucketStarts(int length) + { + return histogram.reservoir.buckets(length); + } + + public boolean isCumulative() + { + return true; + } +} diff --git a/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java b/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java index db91857b3c..92620fb034 100644 --- a/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java +++ b/src/java/org/apache/cassandra/metrics/ShardedDecayingHistograms.java @@ -74,6 +74,12 @@ public class ShardedDecayingHistograms { return shard.histograms.get(histogramIndex); } + + @Override + public boolean isCumulative() + { + return false; + } } public static class DecayingHistogramsShard diff --git a/src/java/org/apache/cassandra/metrics/ShardedHistogram.java b/src/java/org/apache/cassandra/metrics/ShardedHistogram.java index 66defdac56..3ac77a9717 100644 --- a/src/java/org/apache/cassandra/metrics/ShardedHistogram.java +++ b/src/java/org/apache/cassandra/metrics/ShardedHistogram.java @@ -58,11 +58,12 @@ public class ShardedHistogram extends OverrideHistogram } final CopyOnWriteArrayList shards = new CopyOnWriteArrayList<>(); + final boolean isCumulative; final long initialMaxValue; - public ShardedHistogram() + public ShardedHistogram(boolean isCumulative) { - this(1 << 16); + this(isCumulative, 1 << 16); } @Override @@ -77,8 +78,9 @@ public class ShardedHistogram extends OverrideHistogram return LogLinearHistogram.bucketsWithLength(length); } - public ShardedHistogram(long initialMaxValue) + public ShardedHistogram(boolean isCumulative, long initialMaxValue) { + this.isCumulative = isCumulative; this.initialMaxValue = initialMaxValue; } @@ -130,4 +132,10 @@ public class ShardedHistogram extends OverrideHistogram { return maybeRefresh(); } + + @Override + public boolean isCumulative() + { + return isCumulative; + } } diff --git a/src/java/org/apache/cassandra/metrics/ThreadLocalTimer.java b/src/java/org/apache/cassandra/metrics/ThreadLocalTimer.java index 771c68b04f..c653a1dc60 100644 --- a/src/java/org/apache/cassandra/metrics/ThreadLocalTimer.java +++ b/src/java/org/apache/cassandra/metrics/ThreadLocalTimer.java @@ -18,14 +18,8 @@ package org.apache.cassandra.metrics; -import java.time.Duration; -import java.util.concurrent.Callable; -import java.util.concurrent.TimeUnit; -import java.util.function.Supplier; - import com.codahale.metrics.ExponentiallyDecayingReservoir; import com.codahale.metrics.Reservoir; -import com.codahale.metrics.Snapshot; /** * An alternative to Dropwizard Timer which implements the same kind of API. @@ -34,13 +28,8 @@ import com.codahale.metrics.Snapshot; * NOTE: Dropwizard Timer is a concrete class and there is no an interface for Dropwizard Timer logic, * so we have to create an alternative hierarchy. */ -public class ThreadLocalTimer extends com.codahale.metrics.Timer implements Timer +public class ThreadLocalTimer extends OverrideTimer { - private final Meter meter; - private final ThreadLocalHistogram histogram; - // usually we need precise clocks for timing, so we do not replace it with an approximate version - private final MetricClock clock; - /** * Creates a new {@link Timer} using an {@link ExponentiallyDecayingReservoir} and the default * {@link MetricClock}. @@ -76,149 +65,6 @@ public class ThreadLocalTimer extends com.codahale.metrics.Timer implements Time public ThreadLocalTimer(Meter meter, ThreadLocalHistogram histogram, MetricClock clock) { // original Codahale meter and histogram are set to null to reduce memory footprint - super(null, null, clock); - this.meter = meter; - this.histogram = histogram; - this.clock = clock; - } - - /** - * Adds a recorded duration. - * - * @param duration the length of the duration - * @param unit the scale unit of {@code duration} - */ - @Override - public void update(long duration, TimeUnit unit) - { - update(unit.toNanos(duration)); - } - - /** - * Adds a recorded duration. - * - * @param duration the {@link Duration} to add to the timer. Negative or zero value are ignored. - */ - @Override - public void update(Duration duration) - { - update(duration.toNanos()); - } - - /** - * Times and records the duration of event. - * - * @param event a {@link Callable} whose {@link Callable#call()} method implements a process - * whose duration should be timed - * @param the type of the value returned by {@code event} - * @return the value returned by {@code event} - * @throws Exception if {@code event} throws an {@link Exception} - */ - @Override - public T time(Callable event) throws Exception - { - final long startTime = clock.getTick(); - try - { - return event.call(); - } - finally - { - update(clock.getTick() - startTime); - } - } - - /** - * Times and records the duration of event. Should not throw exceptions, for that use the - * {@link #time(Callable)} method. - * - * @param event a {@link Supplier} whose {@link Supplier#get()} method implements a process - * whose duration should be timed - * @param the type of the value returned by {@code event} - * @return the value returned by {@code event} - */ - @Override - public T timeSupplier(Supplier event) - { - final long startTime = clock.getTick(); - try - { - return event.get(); - } - finally - { - update(clock.getTick() - startTime); - } - } - - /** - * Times and records the duration of event. - * - * @param event a {@link Runnable} whose {@link Runnable#run()} method implements a process - * whose duration should be timed - */ - @Override - public void time(Runnable event) - { - final long startTime = clock.getTick(); - try - { - event.run(); - } - finally - { - update(clock.getTick() - startTime); - } - } - - @Override - public Timer.Context startTime() - { - return new Timer.Context(this, clock); - } - - @Override - public long getCount() - { - return histogram.getCount(); - } - - @Override - public double getFifteenMinuteRate() - { - return meter.getFifteenMinuteRate(); - } - - @Override - public double getFiveMinuteRate() - { - return meter.getFiveMinuteRate(); - } - - @Override - public double getMeanRate() - { - return meter.getMeanRate(); - } - - @Override - public double getOneMinuteRate() - { - return meter.getOneMinuteRate(); - } - - @Override - public Snapshot getSnapshot() - { - return histogram.getSnapshot(); - } - - private void update(long duration) - { - if (duration >= 0) - { - histogram.update(duration); - meter.mark(); - } + super(meter, histogram, clock); } } diff --git a/src/java/org/apache/cassandra/metrics/Timer.java b/src/java/org/apache/cassandra/metrics/Timer.java index ed764c03c9..84369be485 100644 --- a/src/java/org/apache/cassandra/metrics/Timer.java +++ b/src/java/org/apache/cassandra/metrics/Timer.java @@ -25,6 +25,7 @@ import java.util.function.Supplier; import com.codahale.metrics.Metered; import com.codahale.metrics.Sampling; +import org.apache.cassandra.metrics.CassandraReservoir.BucketStrategy; /** * An interface which mimics {@link com.codahale.metrics.Timer} API and allows alternative implementations @@ -70,6 +71,10 @@ public interface Timer extends Metered, Sampling } } + BucketStrategy bucketStrategy(); + long[] bucketStarts(int length); + boolean isCumulative(); + void update(long duration, TimeUnit unit); void update(Duration duration); diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index d88314c320..2e34080f99 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -320,7 +320,7 @@ public class AccordCommandStore extends CommandStore RedundantBefore.QuickBounds bounds = unsafeGetRedundantBefore().get(key); if (bounds == null) return cfk; // TODO (required): I don't think this should be possible? but we hit it on some test - return cfk.withRedundantBeforeAtLeast(bounds.gcBefore, false); + return cfk.withGcBeforeAtLeast(bounds.gcBefore, false); } boolean validateCommandsForKey(RoutableKey key, CommandsForKey evicting) diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index 0de343feb7..783299dc4f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -303,7 +303,7 @@ public class AccordKeyspace return null; // TODO (desired): consider whether better to not compact any validation failures, since we expect is already overwritten - CommandsForKey updated = current.withRedundantBeforeAtLeast(redundantBefore.gcBefore(), false); + CommandsForKey updated = current.withGcBeforeAtLeast(redundantBefore.gcBefore(), false); if (current == updated) return row; diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index e7e5762fb1..e9874fae0a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -46,6 +46,7 @@ import accord.primitives.Txn; import org.apache.cassandra.config.AccordSpec; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.SystemKeyspace.BootstrapState; +import org.apache.cassandra.metrics.AccordExecutorMetrics; import org.apache.cassandra.metrics.AccordReplicaMetrics; import org.apache.cassandra.metrics.AccordSystemMetrics; import org.apache.cassandra.service.accord.api.AccordViolationHandler; @@ -161,51 +162,79 @@ public class AccordService implements IAccordService, Shutdownable // Listener is initialized before Accord is initialized public static MetadataChangeListener instance = new MetadataChangeListener(); - private MetadataChangeListener() {} - - private final AtomicReference collector = new AtomicReference<>(new PreInitStateCollector()); - - @Override - public void notifyPreCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + interface Sink { - collector.get().notifyPreCommit(prev, next, fromSnapshot); - } - - @Override - public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) - { - collector.get().notifyPostCommit(prev, next, fromSnapshot); - } - - @VisibleForTesting - public void resetForTesting(ClusterMetadata metadata) - { - PreInitStateCollector stateCollector = new PreInitStateCollector(); - stateCollector.items.add(metadata); - collector.set(stateCollector); + Sink update(ClusterMetadata next); } /** * Collects TCM events from startup util full Accord initialization to avoid races with TCM and creating gaps between * epochs restored from journal and reported by TCM. **/ - - static class PreInitStateCollector implements ChangeListener + static final class Collector implements Sink { - private final List items = new ArrayList<>(4); + static final Collector EMPTY = new Collector(new ClusterMetadata[0]); + final ClusterMetadata[] saved; + + Collector(ClusterMetadata[] saved) + { + this.saved = saved; + } @Override - public synchronized void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + public Sink update(ClusterMetadata next) { - logger.debug("Saving epoch {} to deliver after startup", next.epoch); - items.add(next); + ClusterMetadata[] newSaved = Arrays.copyOf(saved, saved.length + 1); + newSaved[saved.length] = next; + return new Collector(newSaved); } - public synchronized List getItems() + public List getItems() { - return new ArrayList<>(items); + return Arrays.asList(saved); } } + + private final AtomicReference sink = new AtomicReference<>(Collector.EMPTY); + + private MetadataChangeListener() {} + + @Override + public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + { + while (true) + { + Sink curSink = sink.get(); + Sink nextSink = curSink.update(next); + if (nextSink == curSink || sink.compareAndSet(curSink, nextSink)) + return; + } + } + + Collector replaceSinkIfEmpty(Sink newSink) + { + while (true) + { + Sink curSink = sink.get(); + Invariants.require(curSink instanceof Collector); + if (curSink == Collector.EMPTY) + { + if (sink.compareAndSet(curSink, newSink)) + return null; + } + else + { + if (sink.compareAndSet(curSink, Collector.EMPTY)) + return (Collector) curSink; + } + } + } + + @VisibleForTesting + public void unsafeResetForTesting(ClusterMetadata metadata) + { + sink.set(Collector.EMPTY.update(metadata)); + } } private static final Logger logger = LoggerFactory.getLogger(AccordService.class); @@ -325,6 +354,7 @@ public class AccordService implements IAccordService, Shutdownable AccordReplicaMetrics.touch(); AccordSystemMetrics.touch(); + AccordExecutorMetrics.touch(); AccordViolationHandler.setup(); return as; } @@ -422,6 +452,7 @@ public class AccordService implements IAccordService, Shutdownable journal.start(node); node.load(); + ClusterMetadata metadata = ClusterMetadata.current(); endpointMapper.updateMapping(metadata); @@ -575,24 +606,28 @@ public class AccordService implements IAccordService, Shutdownable } // Subscribe to TCM events, and collect any we may have missed to report now - ChangeListener prevListener = MetadataChangeListener.instance.collector.getAndSet(new ChangeListener() + MetadataChangeListener.Sink sink = new MetadataChangeListener.Sink() { @Override - public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) + public MetadataChangeListener.Sink update(ClusterMetadata next) { if (state != State.SHUTDOWN) maybeReportMetadata(next); + return this; } - }); + }; - Invariants.require((prevListener instanceof MetadataChangeListener.PreInitStateCollector), - "Listener should have been initialized with Accord pre-init state collector, but was " + prevListener.getClass()); - - MetadataChangeListener.PreInitStateCollector preinit = (MetadataChangeListener.PreInitStateCollector) prevListener; - for (ClusterMetadata item : preinit.getItems()) + while (true) { - if (item.epoch.getEpoch() > highestKnown) - maybeReportMetadata(item); + MetadataChangeListener.Collector collector = MetadataChangeListener.instance.replaceSinkIfEmpty(sink); + if (collector == null) + break; + + for (ClusterMetadata item : collector.getItems()) + { + if (item.epoch.getEpoch() > highestKnown) + maybeReportMetadata(item); + } } } catch (InterruptedException e) diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 636535e576..67825bad6b 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -67,6 +67,7 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.exceptions.RequestTimeoutException; import org.apache.cassandra.metrics.AccordReplicaMetrics; +import org.apache.cassandra.metrics.AccordSystemMetrics; import org.apache.cassandra.net.ResponseContext; import org.apache.cassandra.service.RetryStrategy; import org.apache.cassandra.service.accord.AccordService; @@ -197,7 +198,11 @@ public class AccordAgent implements Agent, OwnershipEventListener public static void handleException(Throwable t) { - if (t instanceof RequestTimeoutException || t instanceof CancellationException || t instanceof TimeoutException || t instanceof Timeout) + if (t instanceof RequestTimeoutException) + return; + + AccordSystemMetrics.metrics.errors.inc(); + if (t instanceof CancellationException || t instanceof TimeoutException || t instanceof Timeout) return; JVMStabilityInspector.uncaughtException(Thread.currentThread(), t); } diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java index 2c5f405205..c98ba63a4a 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java @@ -41,7 +41,6 @@ import accord.primitives.Txn; import accord.primitives.TxnId; import accord.primitives.Writes; import accord.topology.Topologies; -import accord.topology.Topologies.SelectNodeOwnership; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.service.accord.AccordEndpointMapper; import org.apache.cassandra.service.accord.txn.AccordUpdate; @@ -90,12 +89,12 @@ public class AccordInteropAdapter extends TxnAdapter } @Override - public void persist(Node node, SequentialAsyncExecutor executor, Topologies any, Route require, Route sendTo, SelectNodeOwnership selectSendTo, FullRoute route, Ballot ballot, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, boolean informDurableOnDone, BiConsumer callback) + public void persist(Node node, SequentialAsyncExecutor executor, Topologies any, Route require, Route sendTo, FullRoute route, Ballot ballot, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, boolean informDurableOnDone, BiConsumer callback) { - if (applyKind == Minimal && doInteropPersist(node, executor, any, require, sendTo, selectSendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, informDurableOnDone, callback)) + if (applyKind == Minimal && doInteropPersist(node, executor, any, require, sendTo, ballot, txnId, txn, executeAt, deps, writes, result, route, informDurableOnDone, callback)) return; - super.persist(node, executor, any, require, sendTo, selectSendTo, route, ballot, flags, txnId, txn, executeAt, deps, writes, result, informDurableOnDone, callback); + super.persist(node, executor, any, require, sendTo, route, ballot, flags, txnId, txn, executeAt, deps, writes, result, informDurableOnDone, callback); } private boolean doInteropExecute(Node node, SequentialAsyncExecutor executor, FullRoute route, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, BiConsumer callback) @@ -121,7 +120,7 @@ public class AccordInteropAdapter extends TxnAdapter return true; } - private boolean doInteropPersist(Node node, SequentialAsyncExecutor executor, Topologies any, Route require, Route sendTo, SelectNodeOwnership selectSendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute, boolean informDurableOnDone, BiConsumer callback) + private boolean doInteropPersist(Node node, SequentialAsyncExecutor executor, Topologies any, Route require, Route sendTo, Ballot ballot, TxnId txnId, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute fullRoute, boolean informDurableOnDone, BiConsumer callback) { Update update = txn.update(); ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null; @@ -132,7 +131,7 @@ public class AccordInteropAdapter extends TxnAdapter AccordInteropPersist persist; try { - Topologies all = execution(node, any, sendTo, selectSendTo, fullRoute, txnId, executeAt); + Topologies all = execution(node, any, sendTo, fullRoute, txnId, executeAt); persist = new AccordInteropPersist(node, executor, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, CoordinationFlags.none(), informDurableOnDone, Minimal, callback); } catch (Throwable t) diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java index cbc6182674..6872ed79e1 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -95,7 +95,7 @@ import org.apache.cassandra.transport.Dispatcher; import static accord.coordinate.CoordinationAdapter.Factory.Kind.Standard; import static accord.primitives.Txn.Kind.Write; -import static accord.topology.Topologies.SelectNodeOwnership.SHARE; +import static accord.topology.SelectShards.LIVE; import static accord.utils.Invariants.illegalState; import static accord.utils.Invariants.requireArgument; import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadMetrics; @@ -159,9 +159,9 @@ public class AccordInteropExecution implements ReadCoordinator // TODO (required): compare this to latest logic in Accord, make sure it makes sense ActiveEpochs epochs = node.topology().active(); - this.executes = epochs.forEpoch(route, executeAt.epoch(), SHARE); + this.executes = epochs.forEpoch(route, executeAt.epoch(), LIVE); this.allTopologies = txnId.epoch() != executeAt.epoch() - ? epochs.preciseEpochs(route, txnId.epoch(), executeAt.epoch(), SHARE) + ? epochs.preciseEpochs(route, txnId.epoch(), executeAt.epoch(), LIVE) : executes; this.executeTopology = executes.getEpoch(executeAt.epoch()); this.coordinateTopology = allTopologies.getEpoch(txnId.epoch()); diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java index 0e8f681a48..232db188db 100644 --- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java +++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java @@ -926,7 +926,7 @@ public abstract class LocalLog implements Closeable addListener(new MetadataSnapshotListener()); addListener(new ClientNotificationListener()); addListener(new UpgradeMigrationListener()); - if (DatabaseDescriptor.getAccord().enabled) + if (DatabaseDescriptor.getAccordTransactionsEnabled()) addListener(AccordService.MetadataChangeListener.instance); } diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index a362938aba..d6479e1813 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -717,6 +717,9 @@ public class ClusterUtils public static Epoch maxEpoch(ICluster cluster, int... nodes) { + if (nodes == null || nodes.length == 0) + return maxEpoch(cluster); + Epoch max = null; for (int id : nodes) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index 2577d02a7a..bdee5765ac 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -47,6 +47,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.primitives.Unseekables; +import accord.topology.SelectShards; import accord.topology.Topologies; import accord.topology.TopologyException; import org.apache.cassandra.config.Config.PaxosVariant; @@ -711,7 +712,7 @@ public abstract class AccordCQLTestBase extends AccordTestBase Unseekables routables = AccordTestUtils.createTxn(sb.toString()).keys().toParticipants(); long epoch = AccordService.instance().topology().epoch(); Topologies topology; - try { topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch); } + try { topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch, SelectShards.ALL); } catch (TopologyException e) { throw new RuntimeException(e); } // we don't detect out-of-bounds read/write yet, so use this to validate we reach different shards Assertions.assertThat(topology.totalShards()).isEqualTo(2); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java index 8443526e9f..01632f1cd3 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java @@ -110,7 +110,7 @@ public class AccordSimpleFastPathTest extends TestBaseImpl long epoch = cm.epoch.getEpoch(); TopologyManager tm = AccordService.instance().topology(); - Topology topology = tm.active().getKnown(epoch).global(); + Topology topology = tm.active().getKnown(epoch).all(); Assert.assertFalse(topology.shards().isEmpty()); topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.nodes.without(shard.notInFastPath))); return cm.epoch.getEpoch(); diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java index e4e6de559f..ce927926d9 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java @@ -124,7 +124,7 @@ public class AccordVirtualTablesTest extends CQLTester // range is no longer "added" so doesn't show up as synced! long e2 = 2; - ActiveEpoch a2 = active(topology(e2, T1), EpochReady.done(e2), a1.global()); + ActiveEpoch a2 = active(topology(e2, T1), EpochReady.done(e2), a1.all()); tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a2, a1 }, -1)); assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS), row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE)); diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java index aa37658fc6..865d1b799e 100644 --- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java @@ -518,7 +518,7 @@ public class RouteIndexTest extends CQLTester AccordService startAccord() { ClusterMetadata metadata = ClusterMetadata.current(); - AccordService.MetadataChangeListener.instance.resetForTesting(metadata); + AccordService.MetadataChangeListener.instance.unsafeResetForTesting(metadata); NodeId tcmNodeId = metadata.myNodeId(); AccordService.unsafeSetNewAccordService(null); AccordService.localStartup(tcmNodeId); diff --git a/test/unit/org/apache/cassandra/metrics/JmxVirtualTableMetricsTest.java b/test/unit/org/apache/cassandra/metrics/JmxVirtualTableMetricsTest.java index 9f79d656c1..45348f2775 100644 --- a/test/unit/org/apache/cassandra/metrics/JmxVirtualTableMetricsTest.java +++ b/test/unit/org/apache/cassandra/metrics/JmxVirtualTableMetricsTest.java @@ -68,7 +68,7 @@ public class JmxVirtualTableMetricsTest extends CQLTester metricToNameMap.put(MetricType.METER, registry.meter("meter")); metricToNameMap.put(MetricType.COUNTER, registry.counter("counter")); metricToNameMap.put(MetricType.HISTOGRAM, registry.histogram("histogram", () -> new ClearableHistogram(new DecayingEstimatedHistogramReservoir(true)))); - metricToNameMap.put(MetricType.TIMER, registry.timer("timer")); + metricToNameMap.put(MetricType.TIMER, registry.timer("timer", () -> new SnapshottingTimer(new DecayingEstimatedHistogramReservoir()))); metricToNameMap.put(MetricType.GAUGE, registry.gauge("gauge", () -> gaugeValue::get)); CassandraMetricsRegistry.metricGroups.forEach(group -> { diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java index b94237befc..aee4fabb44 100644 --- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java +++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java @@ -392,7 +392,7 @@ public class EpochSyncTest .isFalse(); // validate topology manager - Ranges ranges = tm.active().getKnown(epoch).global().ranges().mergeTouching(); + Ranges ranges = tm.active().getKnown(epoch).all().ranges().mergeTouching(); Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching(); Assertions.assertThat(actual) .describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.without(actual)) @@ -404,7 +404,7 @@ public class EpochSyncTest continue; Assertions.assertThat(tm.active().hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue(); - Topology topology = tm.active().getKnown(epoch).global(); + Topology topology = tm.active().getKnown(epoch).all(); Ranges ranges = topology.ranges().mergeTouching(); Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching(); // TopologyManager defines syncComplete for an epoch as (epoch - 1).syncComplete. This means that an epoch has reached quorum, but will still miss ranges as previous epochs have not