tools/stress/src/org/apache/cassandra/stress/report/StressMetrics.java - cassandra - Git at Google

 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 static java.util.concurrent.TimeUnit.NANOSECONDS;

 import java.io.FileNotFoundException;
 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.ResultLogger;
 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 implements MeasurementSink
 {
     private final List<Consumer> consumers = new ArrayList<>();
     private final ResultLogger output;
     private final Thread thread;
     private final Uncertainty rowRateUncertainty = new Uncertainty();
     private final CountDownLatch stopped = new CountDownLatch(1);
     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 = 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(ResultLogger output, final long logIntervalMillis, StressSettings settings)
     {
         this.output = output;
         if(settings.log.hdrFile != null)
         {
             try
             {
                 histogramWriter = new HistogramLogWriter(settings.log.hdrFile);
                 histogramWriter.outputComment("Logging op latencies for Cassandra Stress");
                 histogramWriter.outputLogFormatVersion();
                 final long roundedEpoch = epochMs - (epochMs%1000);
                 histogramWriter.outputBaseTime(roundedEpoch);
                 histogramWriter.setBaseTime(roundedEpoch);
                 histogramWriter.outputStartTime(roundedEpoch);
                 histogramWriter.outputLegend();
             }
             catch (FileNotFoundException e)
             {
                 throw new IllegalArgumentException(e);
             }
         }
         else
         {
             histogramWriter = null;
         }
         Callable<JmxCollector.GcStats> gcStatsCollector;
         totalGcStats = new JmxCollector.GcStats(0);
         try
         {
             gcStatsCollector = new JmxCollector(settings.node.resolveAllPermitted(settings), settings.port.jmxPort);
         }
         catch (Throwable t)
         {
             if (settings.log.level == Level.VERBOSE)
             {
                 t.printStackTrace();
             }
             System.err.println("Failed to connect over JMX; not collecting these stats");
             gcStatsCollector = () -> totalGcStats;
         }
         this.gcStatsCollector = gcStatsCollector;
         this.totalCurrentInterval = new TimingInterval(settings.rate.isFixed);
         this.totalSummaryInterval = new TimingInterval(settings.rate.isFixed);
         printHeader("", output);
         thread = new Thread(() -> {
             reportingLoop(logIntervalMillis);
         });
         thread.setName("StressMetrics");
     }
     public void start()
     {
         thread.start();
     }

     public void waitUntilConverges(double targetUncertainty, int minMeasurements, int maxMeasurements) throws InterruptedException
     {
         rowRateUncertainty.await(targetUncertainty, minMeasurements, maxMeasurements);
     }

     public void cancel()
     {
         cancelled = true;
         stop = true;
         thread.interrupt();
         rowRateUncertainty.wakeAll();
     }

     public void stop() throws InterruptedException
     {
         stop = true;
         thread.interrupt();
         stopped.await();
     }


     private void reportingLoop(final long logIntervalMillis)
     {
         // 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)
     {

         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 = opTypeToCurrentTimingInterval.size() > 1;

             for (Map.Entry<String, TimingInterval> type : opTypeToCurrentTimingInterval.entrySet())
             {
                 final String opName = type.getKey();
                 final TimingInterval opInterval = type.getValue();
                 if (logPerOpSummaryLine)
                 {
                     printRow("", opName, opInterval, opTypeToSummaryTimingInterval.get(opName), gcStats, rowRateUncertainty, output);
                 }
                 logHistograms(opName, opInterval);
                 opInterval.reset();
             }

             printRow("", "total", totalCurrentInterval, totalSummaryInterval, gcStats, rowRateUncertainty, output);
             totalCurrentInterval.reset();
         }
     }

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


     private void logHistograms(String opName, TimingInterval opInterval)
     {
         if (histogramWriter == null)
             return;
         final long startNs = opInterval.startNanos();
         final long endNs = opInterval.endNanos();

         logHistogram(opName + "-st", startNs, endNs, opInterval.serviceTime());
         logHistogram(opName + "-rt", startNs, endNs, opInterval.responseTime());
         logHistogram(opName + "-wt", startNs, endNs, opInterval.waitTime());
     }

     private void logHistogram(String opName, final long startNs, final long endNs, final Histogram histogram)
     {
         if (histogram.getTotalCount() != 0)
         {
             histogram.setTag(opName);
             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);
         }
     }


     // PRINT FORMATTING

     public static final String HEADFORMAT = "%-10s%10s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%7s,%9s,%7s,%7s,%8s,%8s,%8s,%8s";
     public static final String ROWFORMAT =  "%-10s%10d,%8.0f,%8.0f,%8.0f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%7.1f,%9.5f,%7d,%7.0f,%8.0f,%8.0f,%8.0f,%8.0f";
     public static final String[] HEADMETRICS = new String[]{"type", "total ops","op/s","pk/s","row/s","mean","med",".95",".99",".999","max","time","stderr", "errors", "gc: #", "max ms", "sum ms", "sdv ms", "mb"};
     public static final String HEAD = String.format(HEADFORMAT, (Object[]) HEADMETRICS);

     private static void printHeader(String prefix, ResultLogger output)
     {
         output.println(prefix + HEAD);
     }

     private static void printRow(String prefix, String type, TimingInterval interval, TimingInterval total,
                                  JmxCollector.GcStats gcStats, Uncertainty opRateUncertainty, ResultLogger output)
     {
         output.println(prefix + String.format(ROWFORMAT,
                 type + ",",
                 total.operationCount(),
                 interval.opRate(),
                 interval.partitionRate(),
                 interval.rowRate(),
                 interval.meanLatencyMs(),
                 interval.medianLatencyMs(),
                 interval.latencyAtPercentileMs(95.0),
                 interval.latencyAtPercentileMs(99.0),
                 interval.latencyAtPercentileMs(99.9),
                 interval.maxLatencyMs(),
                 total.runTimeMs() / 1000f,
                 opRateUncertainty.getUncertainty(),
                 interval.errorCount,
                 gcStats.count,
                 gcStats.maxms,
                 gcStats.summs,
                 gcStats.sdvms,
                 gcStats.bytes / (1 << 20)
         ));
     }

     public void summarise()
     {
         output.println("\n");
         output.println("Results:");

         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()));
         output.println(String.format("Latency mean              : %6.1f ms %s", history.meanLatencyMs(), opHistory.meanLatencies()));
         output.println(String.format("Latency median            : %6.1f ms %s", history.medianLatencyMs(), opHistory.medianLatencies()));
         output.println(String.format("Latency 95th percentile   : %6.1f ms %s", history.latencyAtPercentileMs(95.0), opHistory.latenciesAtPercentile(95.0)));
         output.println(String.format("Latency 99th percentile   : %6.1f ms %s", history.latencyAtPercentileMs(99.0), opHistory.latenciesAtPercentile(99.0)));
         output.println(String.format("Latency 99.9th percentile : %6.1f ms %s", history.latencyAtPercentileMs(99.9), opHistory.latenciesAtPercentile(99.9)));
         output.println(String.format("Latency max               : %6.1f ms %s", history.maxLatencyMs(), opHistory.maxLatencies()));
         output.println(String.format("Total partitions          : %,10d %s",   history.partitionCount, opHistory.partitionCounts()));
         output.println(String.format("Total errors              : %,10d %s",   history.errorCount, opHistory.errorCounts()));
         output.println(String.format("Total GC count            : %,1.0f", totalGcStats.count));
         output.println(String.format("Total GC memory           : %s", FBUtilities.prettyPrintMemory((long)totalGcStats.bytes, true)));
         output.println(String.format("Total GC time             : %,6.1f seconds", totalGcStats.summs / 1000));
         output.println(String.format("Avg GC time               : %,6.1f ms", totalGcStats.summs / totalGcStats.count));
         output.println(String.format("StdDev GC time            : %,6.1f ms", totalGcStats.sdvms));
         output.println("Total operation time      : " + DurationFormatUtils.formatDuration(
                 history.runTimeMs(), "HH:mm:ss", true));
         output.println(""); // Newline is important here to separate the aggregates section from the END or the next stress iteration
     }

     public static void summarise(List<String> ids, List<StressMetrics> summarise, ResultLogger out)
     {
         int idLen = 0;
         for (String id : ids)
             idLen = Math.max(id.length(), idLen);
         String formatstr = "%" + idLen + "s, ";
         printHeader(String.format(formatstr, "id"), out);
         for (int i = 0 ; i < ids.size() ; i++)
         {
             for (Map.Entry<String, TimingInterval> type : summarise.get(i).opTypeToSummaryTimingInterval.entrySet())
             {
                 printRow(String.format(formatstr, ids.get(i)),
                          type.getKey(),
                          type.getValue(),
                          type.getValue(),
                          summarise.get(i).totalGcStats,
                          summarise.get(i).rowRateUncertainty,
                          out);
             }
             TimingInterval hist = summarise.get(i).totalSummaryInterval;
             printRow(String.format(formatstr, ids.get(i)),
                     "total",
                     hist,
                     hist,
                     summarise.get(i).totalGcStats,
                     summarise.get(i).rowRateUncertainty,
                     out
             );
         }
     }

     public boolean wasCancelled()
     {
         return cancelled;
     }

     public void add(Consumer consumer)
     {
         consumers.add(consumer);
     }

     public double opRate()
     {
         return totalSummaryInterval.opRate();
     }
 }
	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 static java.util.concurrent.TimeUnit.NANOSECONDS;

	import java.io.FileNotFoundException;
	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.ResultLogger;
	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 implements MeasurementSink
	{
	private final List<Consumer> consumers = new ArrayList<>();
	private final ResultLogger output;
	private final Thread thread;
	private final Uncertainty rowRateUncertainty = new Uncertainty();
	private final CountDownLatch stopped = new CountDownLatch(1);
	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 = 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(ResultLogger output, final long logIntervalMillis, StressSettings settings)
	{
	this.output = output;
	if(settings.log.hdrFile != null)
	{
	try
	{
	histogramWriter = new HistogramLogWriter(settings.log.hdrFile);
	histogramWriter.outputComment("Logging op latencies for Cassandra Stress");
	histogramWriter.outputLogFormatVersion();
	final long roundedEpoch = epochMs - (epochMs%1000);
	histogramWriter.outputBaseTime(roundedEpoch);
	histogramWriter.setBaseTime(roundedEpoch);
	histogramWriter.outputStartTime(roundedEpoch);
	histogramWriter.outputLegend();
	}
	catch (FileNotFoundException e)
	{
	throw new IllegalArgumentException(e);
	}
	}
	else
	{
	histogramWriter = null;
	}
	Callable<JmxCollector.GcStats> gcStatsCollector;
	totalGcStats = new JmxCollector.GcStats(0);
	try
	{
	gcStatsCollector = new JmxCollector(settings.node.resolveAllPermitted(settings), settings.port.jmxPort);
	}
	catch (Throwable t)
	{
	if (settings.log.level == Level.VERBOSE)
	{
	t.printStackTrace();
	}
	System.err.println("Failed to connect over JMX; not collecting these stats");
	gcStatsCollector = () -> totalGcStats;
	}
	this.gcStatsCollector = gcStatsCollector;
	this.totalCurrentInterval = new TimingInterval(settings.rate.isFixed);
	this.totalSummaryInterval = new TimingInterval(settings.rate.isFixed);
	printHeader("", output);
	thread = new Thread(() -> {
	reportingLoop(logIntervalMillis);
	});
	thread.setName("StressMetrics");
	}
	public void start()
	{
	thread.start();
	}

	public void waitUntilConverges(double targetUncertainty, int minMeasurements, int maxMeasurements) throws InterruptedException
	{
	rowRateUncertainty.await(targetUncertainty, minMeasurements, maxMeasurements);
	}

	public void cancel()
	{
	cancelled = true;
	stop = true;
	thread.interrupt();
	rowRateUncertainty.wakeAll();
	}

	public void stop() throws InterruptedException
	{
	stop = true;
	thread.interrupt();
	stopped.await();
	}


	private void reportingLoop(final long logIntervalMillis)
	{
	// 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)
	{

	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 = opTypeToCurrentTimingInterval.size() > 1;

	for (Map.Entry<String, TimingInterval> type : opTypeToCurrentTimingInterval.entrySet())
	{
	final String opName = type.getKey();
	final TimingInterval opInterval = type.getValue();
	if (logPerOpSummaryLine)
	{
	printRow("", opName, opInterval, opTypeToSummaryTimingInterval.get(opName), gcStats, rowRateUncertainty, output);
	}
	logHistograms(opName, opInterval);
	opInterval.reset();
	}

	printRow("", "total", totalCurrentInterval, totalSummaryInterval, gcStats, rowRateUncertainty, output);
	totalCurrentInterval.reset();
	}
	}

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


	private void logHistograms(String opName, TimingInterval opInterval)
	{
	if (histogramWriter == null)
	return;
	final long startNs = opInterval.startNanos();
	final long endNs = opInterval.endNanos();

	logHistogram(opName + "-st", startNs, endNs, opInterval.serviceTime());
	logHistogram(opName + "-rt", startNs, endNs, opInterval.responseTime());
	logHistogram(opName + "-wt", startNs, endNs, opInterval.waitTime());
	}

	private void logHistogram(String opName, final long startNs, final long endNs, final Histogram histogram)
	{
	if (histogram.getTotalCount() != 0)
	{
	histogram.setTag(opName);
	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);
	}
	}


	// PRINT FORMATTING

	public static final String HEADFORMAT = "%-10s%10s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%8s,%7s,%9s,%7s,%7s,%8s,%8s,%8s,%8s";
	public static final String ROWFORMAT = "%-10s%10d,%8.0f,%8.0f,%8.0f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%8.1f,%7.1f,%9.5f,%7d,%7.0f,%8.0f,%8.0f,%8.0f,%8.0f";
	public static final String[] HEADMETRICS = new String[]{"type", "total ops","op/s","pk/s","row/s","mean","med",".95",".99",".999","max","time","stderr", "errors", "gc: #", "max ms", "sum ms", "sdv ms", "mb"};
	public static final String HEAD = String.format(HEADFORMAT, (Object[]) HEADMETRICS);

	private static void printHeader(String prefix, ResultLogger output)
	{
	output.println(prefix + HEAD);
	}

	private static void printRow(String prefix, String type, TimingInterval interval, TimingInterval total,
	JmxCollector.GcStats gcStats, Uncertainty opRateUncertainty, ResultLogger output)
	{
	output.println(prefix + String.format(ROWFORMAT,
	type + ",",
	total.operationCount(),
	interval.opRate(),
	interval.partitionRate(),
	interval.rowRate(),
	interval.meanLatencyMs(),
	interval.medianLatencyMs(),
	interval.latencyAtPercentileMs(95.0),
	interval.latencyAtPercentileMs(99.0),
	interval.latencyAtPercentileMs(99.9),
	interval.maxLatencyMs(),
	total.runTimeMs() / 1000f,
	opRateUncertainty.getUncertainty(),
	interval.errorCount,
	gcStats.count,
	gcStats.maxms,
	gcStats.summs,
	gcStats.sdvms,
	gcStats.bytes / (1 << 20)
	));
	}

	public void summarise()
	{
	output.println("\n");
	output.println("Results:");

	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()));
	output.println(String.format("Latency mean : %6.1f ms %s", history.meanLatencyMs(), opHistory.meanLatencies()));
	output.println(String.format("Latency median : %6.1f ms %s", history.medianLatencyMs(), opHistory.medianLatencies()));
	output.println(String.format("Latency 95th percentile : %6.1f ms %s", history.latencyAtPercentileMs(95.0), opHistory.latenciesAtPercentile(95.0)));
	output.println(String.format("Latency 99th percentile : %6.1f ms %s", history.latencyAtPercentileMs(99.0), opHistory.latenciesAtPercentile(99.0)));
	output.println(String.format("Latency 99.9th percentile : %6.1f ms %s", history.latencyAtPercentileMs(99.9), opHistory.latenciesAtPercentile(99.9)));
	output.println(String.format("Latency max : %6.1f ms %s", history.maxLatencyMs(), opHistory.maxLatencies()));
	output.println(String.format("Total partitions : %,10d %s", history.partitionCount, opHistory.partitionCounts()));
	output.println(String.format("Total errors : %,10d %s", history.errorCount, opHistory.errorCounts()));
	output.println(String.format("Total GC count : %,1.0f", totalGcStats.count));
	output.println(String.format("Total GC memory : %s", FBUtilities.prettyPrintMemory((long)totalGcStats.bytes, true)));
	output.println(String.format("Total GC time : %,6.1f seconds", totalGcStats.summs / 1000));
	output.println(String.format("Avg GC time : %,6.1f ms", totalGcStats.summs / totalGcStats.count));
	output.println(String.format("StdDev GC time : %,6.1f ms", totalGcStats.sdvms));
	output.println("Total operation time : " + DurationFormatUtils.formatDuration(
	history.runTimeMs(), "HH:mm:ss", true));
	output.println(""); // Newline is important here to separate the aggregates section from the END or the next stress iteration
	}

	public static void summarise(List<String> ids, List<StressMetrics> summarise, ResultLogger out)
	{
	int idLen = 0;
	for (String id : ids)
	idLen = Math.max(id.length(), idLen);
	String formatstr = "%" + idLen + "s, ";
	printHeader(String.format(formatstr, "id"), out);
	for (int i = 0 ; i < ids.size() ; i++)
	{
	for (Map.Entry<String, TimingInterval> type : summarise.get(i).opTypeToSummaryTimingInterval.entrySet())
	{
	printRow(String.format(formatstr, ids.get(i)),
	type.getKey(),
	type.getValue(),
	type.getValue(),
	summarise.get(i).totalGcStats,
	summarise.get(i).rowRateUncertainty,
	out);
	}
	TimingInterval hist = summarise.get(i).totalSummaryInterval;
	printRow(String.format(formatstr, ids.get(i)),
	"total",
	hist,
	hist,
	summarise.get(i).totalGcStats,
	summarise.get(i).rowRateUncertainty,
	out
	);
	}
	}

	public boolean wasCancelled()
	{
	return cancelled;
	}

	public void add(Consumer consumer)
	{
	consumers.add(consumer);
	}

	public double opRate()
	{
	return totalSummaryInterval.opRate();
	}
	}