diff --git a/CHANGES.txt b/CHANGES.txt index 2e8fa4e26e..54ffcf1bcb 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.10 + * Fix Cassandra Stress reporting thread model and precision (CASSANDRA-12585) * Add JMH benchmarks.jar (CASSANDRA-12586) * Add row offset support to SASI (CASSANDRA-11990) * Cleanup uses of AlterTableStatementColumn (CASSANDRA-12567) diff --git a/NOTICE.txt b/NOTICE.txt index 1c552fc972..dcdf03dc41 100644 --- a/NOTICE.txt +++ b/NOTICE.txt @@ -86,3 +86,6 @@ Copyright (c) 2000-2011 INRIA, France Telecom HdrHistogram http://hdrhistogram.org + +JCTools +http://jctools.github.io/JCTools/ diff --git a/build.xml b/build.xml index 858030a928..6954e04b58 100644 --- a/build.xml +++ b/build.xml @@ -452,6 +452,7 @@ + @@ -618,6 +619,7 @@ + measurementsRecycling; + public final Queue measurementsReporting; + public Consumer(OpDistributionFactory operations, + boolean isWarmup, CountDownLatch done, CountDownLatch start, CountDownLatch releaseConsumers, @@ -381,13 +399,18 @@ public class StressAction implements Runnable StressMetrics metrics, UniformRateLimiter rateLimiter) { + OpDistribution opDistribution = operations.get(isWarmup, this); this.done = done; this.start = start; this.releaseConsumers = releaseConsumers; this.metrics = metrics; - this.opStream = new StreamOfOperations(operations, rateLimiter, workManager); + this.opStream = new StreamOfOperations(opDistribution, rateLimiter, workManager); + this.measurementsRecycling = new SpscArrayQueue(8*1024); + this.measurementsReporting = new SpscUnboundedArrayQueue(2048); + metrics.add(this); } + public void run() { try @@ -464,8 +487,24 @@ public class StressAction implements Runnable finally { done.countDown(); - opStream.close(); } } + + @Override + public void record(String opType, long intended, long started, long ended, long rowCnt, long partitionCnt, boolean err) + { + OpMeasurement opMeasurement = measurementsRecycling.poll(); + if(opMeasurement == null) { + opMeasurement = new OpMeasurement(); + } + opMeasurement.opType = opType; + opMeasurement.intended = intended; + opMeasurement.started = started; + opMeasurement.ended = ended; + opMeasurement.rowCnt = rowCnt; + opMeasurement.partitionCnt = partitionCnt; + opMeasurement.err = err; + measurementsReporting.offer(opMeasurement); + } } } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressGraph.java b/tools/stress/src/org/apache/cassandra/stress/StressGraph.java index 3b383fad16..196964c30d 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressGraph.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressGraph.java @@ -35,7 +35,7 @@ import java.util.regex.Pattern; import com.google.common.io.ByteStreams; import org.apache.commons.lang3.StringUtils; - +import org.apache.cassandra.stress.report.StressMetrics; import org.apache.cassandra.stress.settings.StressSettings; import org.json.simple.JSONArray; import org.json.simple.JSONObject; diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 1343cf12dd..b45462f93a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java @@ -50,11 +50,11 @@ import org.apache.cassandra.stress.operations.userdefined.TokenRangeQuery; import org.apache.cassandra.stress.operations.userdefined.SchemaInsert; import org.apache.cassandra.stress.operations.userdefined.SchemaQuery; import org.apache.cassandra.stress.operations.userdefined.ValidatingSchemaQuery; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.*; import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.stress.util.MultiPrintStream; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.Compression; import org.apache.cassandra.thrift.ThriftConversion; import org.apache.thrift.TException; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java index 9a3522c8b8..aa5ddfdf12 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java @@ -36,9 +36,4 @@ public class FixedOpDistribution implements OpDistribution { return operation; } - - public void closeTimers() - { - operation.close(); - } } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java index 33a0c93b66..bef8954b8c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java @@ -25,8 +25,5 @@ import org.apache.cassandra.stress.Operation; public interface OpDistribution { - Operation next(); - - public void closeTimers(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistributionFactory.java b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistributionFactory.java index 14e6dfbcfd..2477c36b96 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistributionFactory.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistributionFactory.java @@ -21,11 +21,11 @@ package org.apache.cassandra.stress.operations; */ -import org.apache.cassandra.stress.util.Timing; +import org.apache.cassandra.stress.StressAction.MeasurementSink; public interface OpDistributionFactory { - public OpDistribution get(Timing timing, boolean isWarmup); + public OpDistribution get(boolean isWarmup, MeasurementSink sink); public String desc(); Iterable each(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/PartitionOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/PartitionOperation.java index 93290fca7e..bad0a94e0c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/PartitionOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/PartitionOperation.java @@ -29,9 +29,9 @@ import org.apache.cassandra.stress.generate.PartitionIterator; import org.apache.cassandra.stress.generate.RatioDistribution; import org.apache.cassandra.stress.generate.Seed; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.OptionRatioDistribution; import org.apache.cassandra.stress.settings.StressSettings; -import org.apache.cassandra.stress.util.Timer; public abstract class PartitionOperation extends Operation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java index fc0229e97a..6d7f9e4205 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java @@ -51,12 +51,4 @@ public class SampledOpDistribution implements OpDistribution remaining--; return cur; } - - public void closeTimers() - { - for (Pair op : operations.getPmf()) - { - op.getFirst().close(); - } - } } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java index 0b206f9953..180003919f 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java @@ -21,17 +21,19 @@ package org.apache.cassandra.stress.operations; */ -import java.util.*; - -import org.apache.cassandra.stress.generate.*; -import org.apache.cassandra.stress.util.Timing; -import org.apache.commons.math3.distribution.EnumeratedDistribution; -import org.apache.commons.math3.util.Pair; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; import org.apache.cassandra.stress.Operation; +import org.apache.cassandra.stress.StressAction.MeasurementSink; import org.apache.cassandra.stress.generate.DistributionFactory; +import org.apache.cassandra.stress.generate.DistributionFixed; import org.apache.cassandra.stress.generate.PartitionGenerator; -import org.apache.cassandra.stress.util.Timer; +import org.apache.cassandra.stress.report.Timer; +import org.apache.commons.math3.distribution.EnumeratedDistribution; +import org.apache.commons.math3.util.Pair; public abstract class SampledOpDistributionFactory implements OpDistributionFactory { @@ -47,13 +49,13 @@ public abstract class SampledOpDistributionFactory implements OpDistributionF protected abstract List get(Timer timer, PartitionGenerator generator, T key, boolean isWarmup); protected abstract PartitionGenerator newGenerator(); - public OpDistribution get(Timing timing, boolean isWarmup) + public OpDistribution get(boolean isWarmup, MeasurementSink sink) { PartitionGenerator generator = newGenerator(); List> operations = new ArrayList<>(); for (Map.Entry ratio : ratios.entrySet()) { - List ops = get(timing.newTimer(ratio.getKey().toString()), + List ops = get(new Timer(ratio.getKey().toString(), sink), generator, ratio.getKey(), isWarmup); for (Operation op : ops) operations.add(new Pair<>(op, ratio.getValue() / ops.size())); @@ -76,9 +78,9 @@ public abstract class SampledOpDistributionFactory implements OpDistributionF { out.add(new OpDistributionFactory() { - public OpDistribution get(Timing timing, boolean isWarmup) + public OpDistribution get(boolean isWarmup, MeasurementSink sink) { - List ops = SampledOpDistributionFactory.this.get(timing.newTimer(ratio.getKey().toString()), + List ops = SampledOpDistributionFactory.this.get(new Timer(ratio.getKey().toString(), sink), newGenerator(), ratio.getKey(), isWarmup); diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java index bb8135cb69..2874b4e411 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java @@ -29,9 +29,9 @@ import org.apache.cassandra.stress.generate.Distribution; import org.apache.cassandra.stress.generate.DistributionFactory; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; -import org.apache.cassandra.stress.util.Timer; public class CqlCounterAdder extends CqlOperation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java index d91ab37458..eb908b0138 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java @@ -27,9 +27,9 @@ import java.util.List; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; -import org.apache.cassandra.stress.util.Timer; public class CqlCounterGetter extends CqlOperation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java index fdf50070c2..3fda6c2c9d 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java @@ -27,9 +27,9 @@ import java.util.List; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; -import org.apache.cassandra.stress.util.Timer; public class CqlInserter extends CqlOperation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java index 097c1a054f..647ba87e59 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java @@ -31,12 +31,12 @@ import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.ConnectionStyle; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.Compression; import org.apache.cassandra.thrift.CqlResult; import org.apache.cassandra.thrift.CqlRow; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java index 0ee17e93e6..61c6553612 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java @@ -28,9 +28,9 @@ import java.util.List; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; -import org.apache.cassandra.stress.util.Timer; public class CqlReader extends CqlOperation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java index 1f9a2c81b7..db355047ba 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java @@ -26,10 +26,10 @@ import java.util.concurrent.ThreadLocalRandom; import org.apache.cassandra.stress.Operation; import org.apache.cassandra.stress.generate.*; import org.apache.cassandra.stress.operations.PartitionOperation; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.CqlVersion; import org.apache.cassandra.stress.settings.StressSettings; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.SlicePredicate; import org.apache.cassandra.thrift.SliceRange; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java index be34a0723a..42f8bc9059 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java @@ -28,10 +28,10 @@ import org.apache.cassandra.stress.generate.Distribution; import org.apache.cassandra.stress.generate.DistributionFactory; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.ColumnOrSuperColumn; import org.apache.cassandra.thrift.CounterColumn; import org.apache.cassandra.thrift.Mutation; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java index ca81fe9b11..4bec3b2bfa 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java @@ -23,10 +23,10 @@ import java.util.List; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.ColumnParent; import org.apache.cassandra.thrift.SlicePredicate; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java index 1827c06b98..ecaa140e25 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java @@ -26,10 +26,10 @@ import java.util.Map; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.Column; import org.apache.cassandra.thrift.ColumnOrSuperColumn; import org.apache.cassandra.thrift.Mutation; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java index d77dc6a81d..4d530b9266 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java @@ -23,10 +23,10 @@ import java.util.List; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.ColumnOrSuperColumn; import org.apache.cassandra.thrift.ColumnParent; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java index 9eebce21b3..96b3392983 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java @@ -43,10 +43,10 @@ import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.io.sstable.CQLSSTableWriter; import org.apache.cassandra.stress.WorkManager; import org.apache.cassandra.stress.generate.*; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; public class SchemaInsert extends SchemaStatement { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java index 0d8e756efc..2764704052 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java @@ -33,10 +33,10 @@ import com.datastax.driver.core.PreparedStatement; import com.datastax.driver.core.ResultSet; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.stress.generate.*; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.CqlResult; import org.apache.cassandra.thrift.ThriftConversion; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java index c83787b378..ca1f5fa679 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java @@ -32,9 +32,9 @@ import com.datastax.driver.core.PreparedStatement; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.stress.generate.Row; import org.apache.cassandra.stress.operations.PartitionOperation; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JavaDriverClient; -import org.apache.cassandra.stress.util.Timer; public abstract class SchemaStatement extends PartitionOperation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/TokenRangeQuery.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/TokenRangeQuery.java index 7a0f02d8f4..f561f6101e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/TokenRangeQuery.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/TokenRangeQuery.java @@ -39,10 +39,10 @@ import org.apache.cassandra.stress.Operation; import org.apache.cassandra.stress.StressYaml; import org.apache.cassandra.stress.WorkManager; import org.apache.cassandra.stress.generate.TokenRangeIterator; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; public class TokenRangeQuery extends Operation { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/ValidatingSchemaQuery.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/ValidatingSchemaQuery.java index 4547a37bae..a731b99b51 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/ValidatingSchemaQuery.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/ValidatingSchemaQuery.java @@ -33,10 +33,10 @@ import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.stress.generate.*; import org.apache.cassandra.stress.generate.Row; import org.apache.cassandra.stress.operations.PartitionOperation; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.stress.util.ThriftClient; -import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.Compression; import org.apache.cassandra.thrift.CqlResult; import org.apache.cassandra.thrift.CqlRow; diff --git a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java b/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java similarity index 57% rename from tools/stress/src/org/apache/cassandra/stress/StressMetrics.java rename to tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java index 86e9a7af4b..a5965471a2 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java +++ b/tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java @@ -1,4 +1,4 @@ -package org.apache.cassandra.stress; +package org.apache.cassandra.stress.report; /* * * Licensed to the Apache Software Foundation (ASF) under one @@ -25,41 +25,57 @@ import static java.util.concurrent.TimeUnit.NANOSECONDS; import java.io.FileNotFoundException; import java.io.PrintStream; +import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.Map.Entry; +import java.util.Queue; +import java.util.TreeMap; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; import org.HdrHistogram.Histogram; import org.HdrHistogram.HistogramLogWriter; +import org.apache.cassandra.stress.StressAction.Consumer; +import org.apache.cassandra.stress.StressAction.MeasurementSink; +import org.apache.cassandra.stress.StressAction.OpMeasurement; +import org.apache.cassandra.stress.settings.SettingsLog.Level; import org.apache.cassandra.stress.settings.StressSettings; import org.apache.cassandra.stress.util.JmxCollector; -import org.apache.cassandra.stress.util.Timing; -import org.apache.cassandra.stress.util.TimingInterval; -import org.apache.cassandra.stress.util.TimingIntervals; +import org.apache.cassandra.stress.util.JmxCollector.GcStats; import org.apache.cassandra.stress.util.Uncertainty; import org.apache.cassandra.utils.FBUtilities; import org.apache.commons.lang3.time.DurationFormatUtils; -public class StressMetrics +public class StressMetrics implements MeasurementSink { - + private final List consumers = new ArrayList<>(); private final PrintStream output; private final Thread thread; private final Uncertainty rowRateUncertainty = new Uncertainty(); private final CountDownLatch stopped = new CountDownLatch(1); - private final Timing timing; private final Callable gcStatsCollector; private final HistogramLogWriter histogramWriter; private final long epochNs = System.nanoTime(); private final long epochMs = System.currentTimeMillis(); - private volatile JmxCollector.GcStats totalGcStats; + private volatile JmxCollector.GcStats totalGcStats = new GcStats(0); private volatile boolean stop = false; private volatile boolean cancelled = false; + + // collected data for intervals and summary + private final Map opTypeToCurrentTimingInterval = new TreeMap<>(); + private final Map opTypeToSummaryTimingInterval = new TreeMap<>(); + private final Queue leftovers = new ArrayDeque<>(); + private final TimingInterval totalCurrentInterval; + private final TimingInterval totalSummaryInterval; + public StressMetrics(PrintStream output, final long logIntervalMillis, StressSettings settings) { this.output = output; @@ -70,9 +86,10 @@ public class StressMetrics histogramWriter = new HistogramLogWriter(settings.log.hdrFile); histogramWriter.outputComment("Logging op latencies for Cassandra Stress"); histogramWriter.outputLogFormatVersion(); - histogramWriter.outputBaseTime(epochMs); - histogramWriter.setBaseTime(epochMs); - histogramWriter.outputStartTime(epochMs); + final long roundedEpoch = epochMs - (epochMs%1000); + histogramWriter.outputBaseTime(roundedEpoch); + histogramWriter.setBaseTime(roundedEpoch); + histogramWriter.outputStartTime(roundedEpoch); histogramWriter.outputLegend(); } catch (FileNotFoundException e) @@ -92,10 +109,9 @@ public class StressMetrics } catch (Throwable t) { - switch (settings.log.level) + if (settings.log.level == Level.VERBOSE) { - case VERBOSE: - t.printStackTrace(); + t.printStackTrace(); } System.err.println("Failed to connect over JMX; not collecting these stats"); gcStatsCollector = new Callable() @@ -107,56 +123,14 @@ public class StressMetrics }; } this.gcStatsCollector = gcStatsCollector; - this.timing = new Timing(settings.rate.isFixed); - + this.totalCurrentInterval = new TimingInterval(settings.rate.isFixed); + this.totalSummaryInterval = new TimingInterval(settings.rate.isFixed); printHeader("", output); - thread = new Thread(new Runnable() - { - @Override - public void run() - { - timing.start(); - try { - - while (!stop) - { - try - { - long sleepNanos = timing.getHistory().endNanos() - System.nanoTime(); - long sleep = (sleepNanos / 1000000) + logIntervalMillis; - - if (sleep < logIntervalMillis >>> 3) - // if had a major hiccup, sleep full interval - Thread.sleep(logIntervalMillis); - else - Thread.sleep(sleep); - - update(); - } catch (InterruptedException e) - { - break; - } - } - - update(); - } - catch (InterruptedException e) - {} - catch (Exception e) - { - cancel(); - e.printStackTrace(StressMetrics.this.output); - } - finally - { - rowRateUncertainty.wakeAll(); - stopped.countDown(); - } - } + thread = new Thread(() -> { + reportingLoop(logIntervalMillis); }); thread.setName("StressMetrics"); } - public void start() { thread.start(); @@ -182,33 +156,163 @@ public class StressMetrics stopped.await(); } - private void update() throws InterruptedException + + private void reportingLoop(final long logIntervalMillis) { - Timing.TimingResult result = timing.snap(gcStatsCollector); - totalGcStats = JmxCollector.GcStats.aggregate(Arrays.asList(totalGcStats, result.extra)); - TimingInterval current = result.intervals.combine(); - TimingInterval history = timing.getHistory().combine(); - rowRateUncertainty.update(current.adjustedRowRate()); - if (current.operationCount() != 0) + // align report timing to the nearest second + final long currentTimeMs = System.currentTimeMillis(); + final long startTimeMs = currentTimeMs - (currentTimeMs % 1000); + // reporting interval starts rounded to the second + long reportingStartNs = (System.nanoTime() - TimeUnit.MILLISECONDS.toNanos(currentTimeMs - startTimeMs)); + final long parkIntervalNs = TimeUnit.MILLISECONDS.toNanos(logIntervalMillis); + try + { + while (!stop) + { + final long wakupTarget = reportingStartNs + parkIntervalNs; + sleepUntil(wakupTarget); + if (stop) + { + break; + } + recordInterval(wakupTarget, parkIntervalNs); + reportingStartNs += parkIntervalNs; + } + + final long end = System.nanoTime(); + recordInterval(end, end - reportingStartNs); + } + catch (Exception e) + { + e.printStackTrace(); + cancel(); + } + finally + { + rowRateUncertainty.wakeAll(); + stopped.countDown(); + } + } + + + private void sleepUntil(final long until) + { + long parkFor; + while (!stop && + (parkFor = until - System.nanoTime()) > 0) + { + LockSupport.parkNanos(parkFor); + } + } + + @Override + public void record(String opType, long intended, long started, long ended, long rowCnt, long partitionCnt, boolean err) + { + TimingInterval current = opTypeToCurrentTimingInterval.computeIfAbsent(opType, k -> new TimingInterval(totalCurrentInterval.isFixed)); + record(current, intended, started, ended, rowCnt, partitionCnt, err); + } + + private void record(TimingInterval t, long intended, long started, long ended, long rowCnt, long partitionCnt, boolean err) + { + t.rowCount += rowCnt; + t.partitionCount += partitionCnt; + if (err) + t.errorCount++; + if (intended != 0) { + t.responseTime().recordValue(ended-intended); + t.waitTime().recordValue(started-intended); + } + final long sTime = ended-started; + t.serviceTime().recordValue(sTime); + } + + private void recordInterval(long intervalEnd, long parkIntervalNs) throws InterruptedException + { + + drainConsumerMeasurements(intervalEnd, parkIntervalNs); + + GcStats gcStats = null; + try + { + gcStats = gcStatsCollector.call(); + } + catch (Exception e) + { + gcStats = new GcStats(0); + } + totalGcStats = JmxCollector.GcStats.aggregate(Arrays.asList(totalGcStats, gcStats)); + + rowRateUncertainty.update(totalCurrentInterval.adjustedRowRate()); + if (totalCurrentInterval.operationCount() != 0) { // if there's a single operation we only print the total - final boolean logPerOpSummaryLine = result.intervals.intervals().size() > 1; + final boolean logPerOpSummaryLine = opTypeToCurrentTimingInterval.size() > 1; - for (Map.Entry type : result.intervals.intervals().entrySet()) + for (Map.Entry type : opTypeToCurrentTimingInterval.entrySet()) { final String opName = type.getKey(); final TimingInterval opInterval = type.getValue(); if (logPerOpSummaryLine) { - printRow("", opName, opInterval, timing.getHistory().get(type.getKey()), result.extra, rowRateUncertainty, output); + printRow("", opName, opInterval, opTypeToSummaryTimingInterval.get(opName), gcStats, rowRateUncertainty, output); } logHistograms(opName, opInterval); + opInterval.reset(); } - printRow("", "total", current, history, result.extra, rowRateUncertainty, output); + printRow("", "total", totalCurrentInterval, totalSummaryInterval, gcStats, rowRateUncertainty, output); + totalCurrentInterval.reset(); } - if (timing.done()) - stop = true; + } + + private void drainConsumerMeasurements(long intervalEnd, long parkIntervalNs) + { + // record leftover measurements if any + int leftoversSize = leftovers.size(); + for (int i=0;i in = c.measurementsReporting; + Queue out = c.measurementsRecycling; + OpMeasurement last; + while ((last = in.poll()) != null) + { + if (last.ended > intervalEnd) + { + // measurements for any given consumer are ordered, we stop when we stop. + leftovers.add(last); + break; + } + record(last.opType, last.intended, last.started, last.ended, last.rowCnt, last.partitionCnt, last.err); + out.offer(last); + } + } + // set timestamps and summarize + for (Entry currPerOp : opTypeToCurrentTimingInterval.entrySet()) { + currPerOp.getValue().endNanos(intervalEnd); + currPerOp.getValue().startNanos(intervalEnd-parkIntervalNs); + TimingInterval summaryPerOp = opTypeToSummaryTimingInterval.computeIfAbsent(currPerOp.getKey(), k -> new TimingInterval(totalCurrentInterval.isFixed)); + summaryPerOp.add(currPerOp.getValue()); + totalCurrentInterval.add(currPerOp.getValue()); + } + totalCurrentInterval.endNanos(intervalEnd); + totalCurrentInterval.startNanos(intervalEnd-parkIntervalNs); + + totalSummaryInterval.add(totalCurrentInterval); } @@ -229,8 +333,12 @@ public class StressMetrics if (histogram.getTotalCount() != 0) { histogram.setTag(opName); - histogram.setStartTimeStamp(epochMs + NANOSECONDS.toMillis(startNs - epochNs)); - histogram.setEndTimeStamp(epochMs + NANOSECONDS.toMillis(endNs - epochNs)); + final long relativeStartNs = startNs - epochNs; + final long startMs = (long) (1000 *((epochMs + NANOSECONDS.toMillis(relativeStartNs))/1000.0)); + histogram.setStartTimeStamp(startMs); + final long relativeEndNs = endNs - epochNs; + final long endMs = (long) (1000 *((epochMs + NANOSECONDS.toMillis(relativeEndNs))/1000.0)); + histogram.setEndTimeStamp(endMs); histogramWriter.outputIntervalHistogram(histogram); } } @@ -278,8 +386,8 @@ public class StressMetrics output.println("\n"); output.println("Results:"); - TimingIntervals opHistory = timing.getHistory(); - TimingInterval history = opHistory.combine(); + TimingIntervals opHistory = new TimingIntervals(opTypeToSummaryTimingInterval); + TimingInterval history = this.totalSummaryInterval; output.println(String.format("Op rate : %,8.0f op/s %s", history.opRate(), opHistory.opRates())); output.println(String.format("Partition rate : %,8.0f pk/s %s", history.partitionRate(), opHistory.partitionRates())); output.println(String.format("Row rate : %,8.0f row/s %s", history.rowRate(), opHistory.rowRates())); @@ -310,7 +418,7 @@ public class StressMetrics printHeader(String.format(formatstr, "id"), out); for (int i = 0 ; i < ids.size() ; i++) { - for (Map.Entry type : summarise.get(i).timing.getHistory().intervals().entrySet()) + for (Map.Entry type : summarise.get(i).opTypeToSummaryTimingInterval.entrySet()) { printRow(String.format(formatstr, ids.get(i)), type.getKey(), @@ -320,7 +428,7 @@ public class StressMetrics summarise.get(i).rowRateUncertainty, out); } - TimingInterval hist = summarise.get(i).timing.getHistory().combine(); + TimingInterval hist = summarise.get(i).totalSummaryInterval; printRow(String.format(formatstr, ids.get(i)), "total", hist, @@ -332,14 +440,18 @@ public class StressMetrics } } - public Timing getTiming() - { - return timing; - } - public boolean wasCancelled() { return cancelled; } + public void add(Consumer consumer) + { + consumers.add(consumer); + } + + public double opRate() + { + return totalSummaryInterval.opRate(); + } } diff --git a/tools/stress/src/org/apache/cassandra/stress/report/Timer.java b/tools/stress/src/org/apache/cassandra/stress/report/Timer.java new file mode 100644 index 0000000000..b3df52f985 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/report/Timer.java @@ -0,0 +1,63 @@ +package org.apache.cassandra.stress.report; +/* + * + * 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. + * + */ + + +import org.apache.cassandra.stress.StressAction.MeasurementSink; + +// a timer - this timer must be used by a single thread, and co-ordinates with other timers by +public final class Timer +{ + private final String opType; + private final MeasurementSink sink; + + // event timing info + private long intendedTimeNs; + private long startTimeNs; + + public Timer(String opType, MeasurementSink sink) + { + this.opType = opType; + this.sink = sink; + } + + + public void stop(long partitionCount, long rowCount, boolean error) + { + sink.record(opType, intendedTimeNs, startTimeNs, System.nanoTime(), rowCount, partitionCount, error); + resetTimes(); + } + + private void resetTimes() + { + intendedTimeNs = startTimeNs = 0; + } + + public void intendedTimeNs(long v) + { + intendedTimeNs = v; + } + + public void start() + { + startTimeNs = System.nanoTime(); + } +} \ No newline at end of file diff --git a/tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java b/tools/stress/src/org/apache/cassandra/stress/report/TimingInterval.java similarity index 64% rename from tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java rename to tools/stress/src/org/apache/cassandra/stress/report/TimingInterval.java index bb9587f49a..4d124a29f5 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java +++ b/tools/stress/src/org/apache/cassandra/stress/report/TimingInterval.java @@ -1,4 +1,4 @@ -package org.apache.cassandra.stress.util; +package org.apache.cassandra.stress.report; /* * * Licensed to the Apache Software Foundation (ASF) under one @@ -26,93 +26,32 @@ import org.HdrHistogram.Histogram; // used for both single timer results and merged timer results public final class TimingInterval { - private final Histogram responseTime; - private final Histogram serviceTime; - private final Histogram waitTime; + private final Histogram responseTime = new Histogram(3); + private final Histogram serviceTime = new Histogram(3); + private final Histogram waitTime = new Histogram(3); public static final long[] EMPTY_SAMPLE = new long[0]; // nanos - private final long startNs; - private final long endNs; - public final long pauseStart; + private long startNs = Long.MAX_VALUE; + private long endNs = Long.MIN_VALUE; // discrete - public final long partitionCount; - public final long rowCount; - public final long errorCount; + public long partitionCount; + public long rowCount; + public long errorCount; public final boolean isFixed; + public TimingInterval(boolean isFixed){ + this.isFixed = isFixed; + } public String toString() { - return String.format("Start: %d end: %d maxLatency: %d pauseStart: %d" + - " pCount: %d rcount: %d opCount: %d errors: %d", - startNs, endNs, getLatencyHistogram().getMaxValue(), pauseStart, + return String.format("Start: %d end: %d maxLatency: %d pCount: %d rcount: %d opCount: %d errors: %d", + startNs, endNs, getLatencyHistogram().getMaxValue(), partitionCount, rowCount, getLatencyHistogram().getTotalCount(), errorCount); } - TimingInterval(long time) - { - startNs = endNs = time; - partitionCount = rowCount = errorCount = 0; - pauseStart = 0; - responseTime = new Histogram(3); - serviceTime = new Histogram(3); - waitTime = new Histogram(3); - isFixed = false; - } - - TimingInterval(long start, long end, long maxPauseStart, long partitionCount, - long rowCount, long errorCount, Histogram r, Histogram s, Histogram w, boolean isFixed) - { - this.startNs = start; - this.endNs = Math.max(end, start); - this.partitionCount = partitionCount; - this.rowCount = rowCount; - this.errorCount = errorCount; - this.pauseStart = maxPauseStart; - this.responseTime = r; - this.serviceTime = s; - this.waitTime = w; - this.isFixed = isFixed; - - } - - // merge multiple timer intervals together - static TimingInterval merge(Iterable intervals, long start) - { - long partitionCount = 0, rowCount = 0, errorCount = 0; - long end = 0; - long pauseStart = 0; - Histogram responseTime = new Histogram(3); - Histogram serviceTime = new Histogram(3); - Histogram waitTime = new Histogram(3); - boolean isFixed = false; - for (TimingInterval interval : intervals) - { - if (interval != null) - { - end = Math.max(end, interval.endNs); - partitionCount += interval.partitionCount; - rowCount += interval.rowCount; - errorCount += interval.errorCount; - - if (interval.getLatencyHistogram().getMaxValue() > serviceTime.getMaxValue()) - { - pauseStart = interval.pauseStart; - } - responseTime.add(interval.responseTime); - serviceTime.add(interval.serviceTime); - waitTime.add(interval.waitTime); - isFixed |= interval.isFixed; - } - } - - - return new TimingInterval(start, end, pauseStart, partitionCount, rowCount, - errorCount, responseTime, serviceTime, waitTime, isFixed); - - } public double opRate() { @@ -230,5 +169,66 @@ public final class TimingInterval { return getLatencyHistogram().getTotalCount(); } + + + public void startNanos(long started) + { + this.startNs = started; + } + public void endNanos(long ended) + { + this.endNs = ended; + } + + + public void reset() + { + this.endNs = Long.MIN_VALUE; + this.startNs = Long.MAX_VALUE; + this.errorCount = 0; + this.rowCount = 0; + this.partitionCount = 0; + if(this.responseTime.getTotalCount() != 0) + { + this.responseTime.reset(); + } + if(this.serviceTime.getTotalCount() != 0) + { + this.serviceTime.reset(); + } + if(this.waitTime.getTotalCount() != 0) + { + this.waitTime.reset(); + } + } + + public void add(TimingInterval value) + { + if(this.startNs > value.startNs) + { + this.startNs = value.startNs; + } + if(this.endNs < value.endNs) + { + this.endNs = value.endNs; + } + + this.errorCount += value.errorCount; + this.rowCount += value.rowCount; + this.partitionCount += value.partitionCount; + + if (value.responseTime.getTotalCount() != 0) + { + this.responseTime.add(value.responseTime); + } + if (value.serviceTime.getTotalCount() != 0) + { + this.serviceTime.add(value.serviceTime); + } + if (value.waitTime.getTotalCount() != 0) + { + this.waitTime.add(value.waitTime); + } + } } diff --git a/tools/stress/src/org/apache/cassandra/stress/util/TimingIntervals.java b/tools/stress/src/org/apache/cassandra/stress/report/TimingIntervals.java similarity index 72% rename from tools/stress/src/org/apache/cassandra/stress/util/TimingIntervals.java rename to tools/stress/src/org/apache/cassandra/stress/report/TimingIntervals.java index 05860065d1..747a42accc 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/TimingIntervals.java +++ b/tools/stress/src/org/apache/cassandra/stress/report/TimingIntervals.java @@ -1,4 +1,4 @@ -package org.apache.cassandra.stress.util; +package org.apache.cassandra.stress.report; import java.util.Arrays; import java.util.Map; @@ -7,46 +7,17 @@ import java.util.TreeMap; public class TimingIntervals { final Map intervals; - TimingIntervals(Iterable opTypes) - { - long now = System.nanoTime(); - intervals = new TreeMap<>(); - for (String opType : opTypes) - intervals.put(opType, new TimingInterval(now)); - } - TimingIntervals(Map intervals) + public TimingIntervals(Map intervals) { this.intervals = intervals; } - public TimingIntervals merge(TimingIntervals with, long start) - { - assert intervals.size() == with.intervals.size(); - TreeMap ret = new TreeMap<>(); - - for (String opType : intervals.keySet()) - { - assert with.intervals.containsKey(opType); - ret.put(opType, TimingInterval.merge(Arrays.asList(intervals.get(opType), with.intervals.get(opType)), start)); - } - - return new TimingIntervals(ret); - } - public TimingInterval get(String opType) { return intervals.get(opType); } - public TimingInterval combine() - { - long start = Long.MAX_VALUE; - for (TimingInterval ti : intervals.values()) - start = Math.min(start, ti.startNanos()); - - return TimingInterval.merge(intervals.values(), start); - } public String str(TimingInterval.TimingParameter value, String unit) { @@ -55,6 +26,11 @@ public class TimingIntervals public String str(TimingInterval.TimingParameter value, double rank, String unit) { + if (intervals.size() == 0) + { + return "[]"; + } + StringBuilder sb = new StringBuilder("["); for (Map.Entry entry : intervals.entrySet()) diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java index c4257191ec..b7f87e95a6 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java @@ -26,6 +26,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import org.apache.cassandra.stress.StressAction.MeasurementSink; import org.apache.cassandra.stress.generate.DistributionFactory; import org.apache.cassandra.stress.generate.PartitionGenerator; import org.apache.cassandra.stress.generate.SeedManager; @@ -37,8 +38,8 @@ import org.apache.cassandra.stress.operations.FixedOpDistribution; import org.apache.cassandra.stress.operations.OpDistribution; import org.apache.cassandra.stress.operations.OpDistributionFactory; import org.apache.cassandra.stress.operations.predefined.PredefinedOperation; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.util.MultiPrintStream; -import org.apache.cassandra.stress.util.Timing; // Settings unique to the mixed command type public class SettingsCommandPreDefined extends SettingsCommand @@ -53,9 +54,11 @@ public class SettingsCommandPreDefined extends SettingsCommand final SeedManager seeds = new SeedManager(settings); return new OpDistributionFactory() { - public OpDistribution get(Timing timing, boolean isWarmup) + public OpDistribution get(boolean isWarmup, MeasurementSink sink) { - return new FixedOpDistribution(PredefinedOperation.operation(type, timing.newTimer(type.toString()), + final Timer timer1 = new Timer(type.toString(), sink); + final Timer timer = timer1; + return new FixedOpDistribution(PredefinedOperation.operation(type, timer, newGenerator(settings), seeds, settings, add)); } diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java index 0e361baf60..9c58c5b7a8 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java @@ -30,8 +30,8 @@ import org.apache.cassandra.stress.generate.SeedManager; import org.apache.cassandra.stress.operations.OpDistributionFactory; import org.apache.cassandra.stress.operations.SampledOpDistributionFactory; import org.apache.cassandra.stress.operations.predefined.PredefinedOperation; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.util.MultiPrintStream; -import org.apache.cassandra.stress.util.Timer; // Settings unique to the mixed command type public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java index 7f30688e30..66e6df3046 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java @@ -36,8 +36,8 @@ import org.apache.cassandra.stress.generate.SeedManager; import org.apache.cassandra.stress.generate.TokenRangeIterator; import org.apache.cassandra.stress.operations.OpDistributionFactory; import org.apache.cassandra.stress.operations.SampledOpDistributionFactory; +import org.apache.cassandra.stress.report.Timer; import org.apache.cassandra.stress.util.MultiPrintStream; -import org.apache.cassandra.stress.util.Timer; // Settings unique to the mixed command type public class SettingsCommandUser extends SettingsCommand diff --git a/tools/stress/src/org/apache/cassandra/stress/util/Timer.java b/tools/stress/src/org/apache/cassandra/stress/util/Timer.java deleted file mode 100644 index bb19bb6149..0000000000 --- a/tools/stress/src/org/apache/cassandra/stress/util/Timer.java +++ /dev/null @@ -1,167 +0,0 @@ -package org.apache.cassandra.stress.util; -/* - * - * 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. - * - */ - - -import java.util.concurrent.CountDownLatch; - -import org.HdrHistogram.Histogram; - -// a timer - this timer must be used by a single thread, and co-ordinates with other timers by -public final class Timer -{ - private Histogram responseTime = new Histogram(3); - private Histogram serviceTime = new Histogram(3); - private Histogram waitTime = new Histogram(3); - - // event timing info - private long intendedTimeNs; - private long startTimeNs; - private long endTimeNs; - - - // aggregate info - private long errorCount; - private long partitionCount; - private long rowCount; - private long max; - private long maxStart; - private long upToDateAsOf; - private long lastSnap = System.nanoTime(); - - // communication with summary/logging thread - private volatile CountDownLatch reportRequest; - volatile TimingInterval report; - private volatile TimingInterval finalReport; - private final boolean isFixed; - - public Timer(boolean isFixed) - { - this.isFixed = isFixed; - } - - public boolean running() - { - return finalReport == null; - } - - public void stop(long partitionCount, long rowCount, boolean error) - { - endTimeNs = System.nanoTime(); - maybeReport(); - long now = System.nanoTime(); - if (intendedTimeNs != 0) - { - long rTime = endTimeNs - intendedTimeNs; - responseTime.recordValue(rTime); - long wTime = startTimeNs - intendedTimeNs; - waitTime.recordValue(wTime); - } - - long sTime = endTimeNs - startTimeNs; - serviceTime.recordValue(sTime); - - if (sTime > max) - { - maxStart = startTimeNs; - max = sTime; - } - this.partitionCount += partitionCount; - this.rowCount += rowCount; - if (error) - this.errorCount++; - upToDateAsOf = now; - resetTimes(); - } - - private void resetTimes() - { - intendedTimeNs = startTimeNs = endTimeNs = 0; - } - - private TimingInterval buildReport() - { - final TimingInterval report = new TimingInterval(lastSnap, upToDateAsOf, maxStart, partitionCount, - rowCount, errorCount, responseTime, serviceTime, waitTime, isFixed); - // reset counters - partitionCount = 0; - rowCount = 0; - max = 0; - errorCount = 0; - lastSnap = upToDateAsOf; - responseTime = new Histogram(3); - serviceTime = new Histogram(3); - waitTime = new Histogram(3); - - return report; - } - - // checks to see if a report has been requested, and if so produces the report, signals and clears the request - private void maybeReport() - { - if (reportRequest != null) - { - synchronized (this) - { - report = buildReport(); - reportRequest.countDown(); - reportRequest = null; - } - } - } - - // checks to see if the timer is dead; if not requests a report, and otherwise fulfills the request itself - synchronized void requestReport(CountDownLatch signal) - { - if (finalReport != null) - { - report = finalReport; - finalReport = new TimingInterval(0); - signal.countDown(); - } - else - reportRequest = signal; - } - - // closes the timer; if a request is outstanding, it furnishes the request, otherwise it populates finalReport - public synchronized void close() - { - if (reportRequest == null) - finalReport = buildReport(); - else - { - finalReport = new TimingInterval(0); - report = buildReport(); - reportRequest.countDown(); - reportRequest = null; - } - } - - public void intendedTimeNs(long v) - { - intendedTimeNs = v; - } - - public void start() - { - startTimeNs = System.nanoTime(); - } -} \ No newline at end of file diff --git a/tools/stress/src/org/apache/cassandra/stress/util/Timing.java b/tools/stress/src/org/apache/cassandra/stress/util/Timing.java deleted file mode 100644 index a304db78e2..0000000000 --- a/tools/stress/src/org/apache/cassandra/stress/util/Timing.java +++ /dev/null @@ -1,147 +0,0 @@ -package org.apache.cassandra.stress.util; -/* - * - * 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. - * - */ - - -import java.util.*; -import java.util.concurrent.Callable; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -// relatively simple timing class for getting a uniform sample of latencies, and saving other metrics -// ensures accuracy of timing by having single threaded timers that are check-pointed by the snapping thread, -// which waits for them to report back. They report back the data up to the last event prior to the check-point. -// if the threads are blocked/paused this may mean a period of time longer than the checkpoint elapses, but that all -// metrics calculated over the interval are accurate -public class Timing -{ - // concurrency: this should be ok as the consumers are created serially by StressAction.run / warmup - // Probably the CopyOnWriteArrayList could be changed to an ordinary list as well. - private final Map> timers = new TreeMap<>(); - private volatile TimingIntervals history; - private boolean done; - private boolean isFixed; - - public Timing(boolean isFixed) - { - this.isFixed = isFixed; - } - - // TIMING - - public static class TimingResult - { - public final E extra; - public final TimingIntervals intervals; - public TimingResult(E extra, TimingIntervals intervals) - { - this.extra = extra; - this.intervals = intervals; - } - } - - public TimingResult snap(Callable call) throws InterruptedException - { - // Count up total # of timers - int timerCount = 0; - for (List timersForOperation : timers.values()) - { - timerCount += timersForOperation.size(); - } - final CountDownLatch ready = new CountDownLatch(timerCount); - - // request reports - for (List timersForOperation : timers.values()) - { - for(Timer timer : timersForOperation) - { - timer.requestReport(ready); - } - } - - E extra; - try - { - extra = call.call(); - } - catch (Exception e) - { - if (e instanceof InterruptedException) - throw (InterruptedException) e; - throw new RuntimeException(e); - } - - // TODO fail gracefully after timeout if a thread is stuck - if (!ready.await(5L, TimeUnit.MINUTES)) - { - throw new RuntimeException("Timed out waiting for a timer thread - seems one got stuck. Check GC/Heap size"); - } - - boolean done = true; - - // reports have been filled in by timer threadCount, so merge - Map intervals = new TreeMap<>(); - for (Map.Entry> entry : timers.entrySet()) - { - List operationIntervals = new ArrayList<>(); - for (Timer timer : entry.getValue()) - { - operationIntervals.add(timer.report); - done &= !timer.running(); - } - - intervals.put(entry.getKey(), TimingInterval.merge(operationIntervals, - history.get(entry.getKey()).endNanos())); - } - - TimingIntervals result = new TimingIntervals(intervals); - this.done = done; - history = history.merge(result, history.startNanos()); - return new TimingResult<>(extra, result); - } - - // build a new timer and add it to the set of running timers. - public Timer newTimer(String opType) - { - final Timer timer = new Timer(isFixed); - - if (!timers.containsKey(opType)) - timers.put(opType, new ArrayList()); - - timers.get(opType).add(timer); - return timer; - } - - public void start() - { - history = new TimingIntervals(timers.keySet()); - } - - public boolean done() - { - return done; - } - - public TimingIntervals getHistory() - { - return history; - } -}