Fix Cassandra Stress reporting thread model and precision

Patch by Nitsan Wakart; reviewed by tjake for CASSANDRA-12585

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)

NOTICE.txt

@@ -86,3 +86,6 @@ Copyright (c) 2000-2011 INRIA, France Telecom
 
 HdrHistogram
 http://hdrhistogram.org
+
+JCTools
+http://jctools.github.io/JCTools/

build.xml

@@ -452,6 +452,7 @@
           <dependency groupId="com.github.rholder" artifactId="snowball-stemmer" version="1.3.0.581.1" />
           <dependency groupId="com.googlecode.concurrent-trees" artifactId="concurrent-trees" version="2.4.0" />
 	  <dependency groupId="com.github.ben-manes.caffeine" artifactId="caffeine" version="2.2.6" />
+	  <dependency groupId="org.jctools" artifactId="jctools-core" version="1.2.1"/>
         </dependencyManagement>
         <developer id="alakshman" name="Avinash Lakshman"/>
         <developer id="aleksey" name="Aleksey Yeschenko"/>
@@ -618,6 +619,7 @@
         <dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>
         <dependency groupId="org.caffinitas.ohc" artifactId="ohc-core"/>
 	<dependency groupId="com.github.ben-manes.caffeine" artifactId="caffeine" />
+	<dependency groupId="org.jctools" artifactId="jctools-core"/>
       </artifact:pom>
       <artifact:pom id="thrift-pom"
                     artifactId="cassandra-thrift"

lib/licenses/jctools-core-1.2.1.txt

@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed 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.

tools/stress/src/org/apache/cassandra/stress/Operation.java

@@ -20,11 +20,11 @@ package org.apache.cassandra.stress;
 
 import java.io.IOException;
 
+import org.apache.cassandra.stress.report.Timer;
 import org.apache.cassandra.stress.settings.SettingsLog;
 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.InvalidRequestException;
 import org.apache.cassandra.transport.SimpleClient;
 
@@ -138,11 +138,6 @@ public abstract class Operation
             System.err.println(message);
     }
 
-    public void close()
-    {
-        timer.close();
-    }
-
     public void intendedStartNs(long intendedTime)
     {
         timer.intendedTimeNs(intendedTime);

tools/stress/src/org/apache/cassandra/stress/StressAction.java

@@ -22,6 +22,7 @@ import java.io.OutputStream;
 import java.io.PrintStream;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Queue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -29,12 +30,15 @@ import java.util.concurrent.locks.LockSupport;
 
 import org.apache.cassandra.stress.operations.OpDistribution;
 import org.apache.cassandra.stress.operations.OpDistributionFactory;
+import org.apache.cassandra.stress.report.StressMetrics;
 import org.apache.cassandra.stress.settings.ConnectionAPI;
 import org.apache.cassandra.stress.settings.SettingsCommand;
 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.transport.SimpleClient;
+import org.jctools.queues.SpscArrayQueue;
+import org.jctools.queues.SpscUnboundedArrayQueue;
 
 import com.google.common.util.concurrent.Uninterruptibles;
 
@@ -183,8 +187,8 @@ public class StressAction implements Runnable
         double improvement = 0;
         for (int i = results.size() - count ; i < results.size() ; i++)
         {
-            double prev = results.get(i - 1).getTiming().getHistory().opRate();
+            double prev = results.get(i - 1).opRate();
-            double cur = results.get(i).getTiming().getHistory().opRate();
+            double cur = results.get(i).opRate();
             improvement += (cur - prev) / prev;
         }
         return improvement / count;
@@ -219,7 +223,7 @@ public class StressAction implements Runnable
         final Consumer[] consumers = new Consumer[threadCount];
         for (int i = 0; i < threadCount; i++)
         {
-            consumers[i] = new Consumer(operations.get(metrics.getTiming(), isWarmup),
+            consumers[i] = new Consumer(operations, isWarmup,
                                         done, start, releaseConsumers, workManager, metrics, rateLimiter);
         }
 
@@ -238,7 +242,9 @@ public class StressAction implements Runnable
         }
         // start counting from NOW!
         if(rateLimiter != null)
+        {
             rateLimiter.start();
+        }
         // release the hounds!!!
         releaseConsumers.countDown();
 
@@ -353,18 +359,28 @@ public class StressAction implements Runnable
             return op;
         }
 
-        void close()
-        {
-            operations.closeTimers();
-        }
-
         void abort()
         {
             workManager.stop();
         }
     }
-
+    public static class OpMeasurement
-    private class Consumer extends Thread
+    {
+        public String opType;
+        public long intended,started,ended,rowCnt,partitionCnt;
+        public boolean err;
+        @Override
+        public String toString()
+        {
+            return "OpMeasurement [opType=" + opType + ", intended=" + intended + ", started=" + started + ", ended="
+                    + ended + ", rowCnt=" + rowCnt + ", partitionCnt=" + partitionCnt + ", err=" + err + "]";
+        }
+    }
+    public interface MeasurementSink
+    {
+        void record(String opType,long intended, long started, long ended, long rowCnt, long partitionCnt, boolean err);
+    }
+    public class Consumer extends Thread implements MeasurementSink
     {
         private final StreamOfOperations opStream;
         private final StressMetrics metrics;
@@ -372,8 +388,10 @@ public class StressAction implements Runnable
         private final CountDownLatch done;
         private final CountDownLatch start;
         private final CountDownLatch releaseConsumers;
-
+        public final Queue<OpMeasurement> measurementsRecycling;
-        public Consumer(OpDistribution operations,
+        public final Queue<OpMeasurement> 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<OpMeasurement>(8*1024);
+            this.measurementsReporting =  new SpscUnboundedArrayQueue<OpMeasurement>(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);
+        }
     }
 }

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;

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;

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();
-    }
 }

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();
 }

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<OpDistributionFactory> each();
 }

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
 {

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<Operation, Double> op : operations.getPmf())
-        {
-            op.getFirst().close();
-        }
-    }
 }

tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java

@@ -21,17 +21,19 @@ package org.apache.cassandra.stress.operations;
  */
 
 
-import java.util.*;
+import java.util.ArrayList;
-
+import java.util.Collections;
-import org.apache.cassandra.stress.generate.*;
+import java.util.List;
-import org.apache.cassandra.stress.util.Timing;
+import java.util.Map;
-import org.apache.commons.math3.distribution.EnumeratedDistribution;
-import org.apache.commons.math3.util.Pair;
 
 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<T> implements OpDistributionFactory
 {
@@ -47,13 +49,13 @@ public abstract class SampledOpDistributionFactory<T> implements OpDistributionF
     protected abstract List<? extends Operation> 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<Pair<Operation, Double>> operations = new ArrayList<>();
         for (Map.Entry<T, Double> ratio : ratios.entrySet())
         {
-            List<? extends Operation> ops = get(timing.newTimer(ratio.getKey().toString()),
+            List<? extends Operation> 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<T> implements OpDistributionF
         {
             out.add(new OpDistributionFactory()
             {
-                public OpDistribution get(Timing timing, boolean isWarmup)
+                public OpDistribution get(boolean isWarmup, MeasurementSink sink)
                 {
-                    List<? extends Operation> ops = SampledOpDistributionFactory.this.get(timing.newTimer(ratio.getKey().toString()),
+                    List<? extends Operation> ops = SampledOpDistributionFactory.this.get(new Timer(ratio.getKey().toString(), sink),
                                                                                           newGenerator(),
                                                                                           ratio.getKey(),
                                                                                           isWarmup);

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<Integer>
 {

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<Integer>
 {

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<Integer>
 {

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;

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<ByteBuffer[][]>
 {

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;
 

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;

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;
 

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;

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;
 

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
 {

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;
 

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
 {

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
 {

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;

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.JmxCollector.GcStats;
-import org.apache.cassandra.stress.util.TimingInterval;
-import org.apache.cassandra.stress.util.TimingIntervals;
 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<Consumer> 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<JmxCollector.GcStats> 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<String, TimingInterval> opTypeToCurrentTimingInterval = new TreeMap<>();
+    private final Map<String, TimingInterval> opTypeToSummaryTimingInterval = new TreeMap<>();
+    private final Queue<OpMeasurement> 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);
+                final long roundedEpoch = epochMs - (epochMs%1000);
-                histogramWriter.setBaseTime(epochMs);
+                histogramWriter.outputBaseTime(roundedEpoch);
-                histogramWriter.outputStartTime(epochMs);
+                histogramWriter.setBaseTime(roundedEpoch);
+                histogramWriter.outputStartTime(roundedEpoch);
                 histogramWriter.outputLegend();
             }
             catch (FileNotFoundException e)
@@ -92,9 +109,8 @@ public class StressMetrics
         }
         catch (Throwable t)
         {
-            switch (settings.log.level)
+            if (settings.log.level == Level.VERBOSE)
             {
-                case VERBOSE:
                 t.printStackTrace();
             }
             System.err.println("Failed to connect over JMX; not collecting these stats");
@@ -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()
+        thread = new Thread(() -> {
-        {
+            reportingLoop(logIntervalMillis);
-            @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.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<JmxCollector.GcStats> result = timing.snap(gcStatsCollector);
+        // align report timing to the nearest second
-        totalGcStats = JmxCollector.GcStats.aggregate(Arrays.asList(totalGcStats, result.extra));
+        final long currentTimeMs = System.currentTimeMillis();
-        TimingInterval current = result.intervals.combine();
+        final long startTimeMs = currentTimeMs - (currentTimeMs % 1000);
-        TimingInterval history = timing.getHistory().combine();
+        // reporting interval starts rounded to the second
-        rowRateUncertainty.update(current.adjustedRowRate());
+        long reportingStartNs = (System.nanoTime() - TimeUnit.MILLISECONDS.toNanos(currentTimeMs - startTimeMs));
-        if (current.operationCount() != 0)
+        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<String, TimingInterval> type : result.intervals.intervals().entrySet())
+            for (Map.Entry<String, TimingInterval> 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<leftoversSize;i++)
+        {
+            OpMeasurement last = leftovers.poll();
+            if (last.ended <= intervalEnd)
+            {
+                record(last.opType, last.intended, last.started, last.ended, last.rowCnt, last.partitionCnt, last.err);
+                // round robin-ish redistribution of leftovers
+                consumers.get(i%consumers.size()).measurementsRecycling.offer(last);
+            }
+            else
+            {
+                // no record for you! wait one interval!
+                leftovers.offer(last);
+            }
+        }
+        // record interval collected measurements
+        for (Consumer c: consumers) {
+            Queue<OpMeasurement> in = c.measurementsReporting;
+            Queue<OpMeasurement> 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<String, TimingInterval> 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));
+            final long relativeStartNs = startNs - epochNs;
-            histogram.setEndTimeStamp(epochMs + NANOSECONDS.toMillis(endNs - 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();
+        TimingIntervals opHistory = new TimingIntervals(opTypeToSummaryTimingInterval);
-        TimingInterval history = opHistory.combine();
+        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<String, TimingInterval> type : summarise.get(i).timing.getHistory().intervals().entrySet())
+            for (Map.Entry<String, TimingInterval> 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();
+    }
 }

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();
+    }
+}

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 responseTime = new Histogram(3);
-    private final Histogram serviceTime;
+    private final Histogram serviceTime = new Histogram(3);
-    private final Histogram waitTime;
+    private final Histogram waitTime = new Histogram(3);
 
     public static final long[] EMPTY_SAMPLE = new long[0];
     // nanos
-    private final long startNs;
+    private long startNs = Long.MAX_VALUE;
-    private final long endNs;
+    private long endNs = Long.MIN_VALUE;
-    public final long pauseStart;
 
     // discrete
-    public final long partitionCount;
+    public long partitionCount;
-    public final long rowCount;
+    public long rowCount;
-    public final long errorCount;
+    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" +
+        return String.format("Start: %d end: %d maxLatency: %d pCount: %d rcount: %d opCount: %d errors: %d",
-                             " pCount: %d rcount: %d opCount: %d errors: %d",
+                             startNs, endNs, getLatencyHistogram().getMaxValue(),
-                             startNs, endNs, getLatencyHistogram().getMaxValue(), pauseStart,
                              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<TimingInterval> 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);
+        }
+    }
  }
 

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<String, TimingInterval> intervals;
-    TimingIntervals(Iterable<String> opTypes)
-    {
-        long now = System.nanoTime();
-        intervals = new TreeMap<>();
-        for (String opType : opTypes)
-            intervals.put(opType, new TimingInterval(now));
-    }
 
-    TimingIntervals(Map<String, TimingInterval> intervals)
+    public TimingIntervals(Map<String, TimingInterval> intervals)
     {
         this.intervals = intervals;
     }
 
-    public TimingIntervals merge(TimingIntervals with, long start)
-    {
-        assert intervals.size() == with.intervals.size();
-        TreeMap<String, TimingInterval> 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<String, TimingInterval> entry : intervals.entrySet())

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));
             }
 

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

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

tools/stress/src/org/apache/cassandra/stress/util/Timer.java

@@ -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();
-    }
-}

tools/stress/src/org/apache/cassandra/stress/util/Timing.java

@@ -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<String, List<Timer>> 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<E>
-    {
-        public final E extra;
-        public final TimingIntervals intervals;
-        public TimingResult(E extra, TimingIntervals intervals)
-        {
-            this.extra = extra;
-            this.intervals = intervals;
-        }
-    }
-
-    public <E> TimingResult<E> snap(Callable<E> call) throws InterruptedException
-    {
-        // Count up total # of timers
-        int timerCount = 0;
-        for (List<Timer> timersForOperation : timers.values())
-        {
-            timerCount += timersForOperation.size();
-        }
-        final CountDownLatch ready = new CountDownLatch(timerCount);
-
-        // request reports
-        for (List <Timer> 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<String, TimingInterval> intervals = new TreeMap<>();
-        for (Map.Entry<String, List<Timer>> entry : timers.entrySet())
-        {
-            List<TimingInterval> 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<Timer>());
-
-        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;
-    }
-}