INT-3641: JMX: Lazy Stats
JIRA: https://jira.spring.io/browse/INT-3641 Instead of maintaining a moving average on each event, store the events for offline analysis. Retain 5*window samples - this means that the earlies retained sample contributes just 0.5% to the sum. E.g. with a window of 10, the earliest sample is 0.9**50 (0.005). Also, defer the conversion from nanoseconds to milliseconds to the retrieval side. Experimentation shows this increases perfomance by approximately 2x. Sending 1B messages to nullChannel. With Proxy: 1.2M/sec Afer proxy removed: 2.4M/sec With this change: 5.3M/sec INT-3641: Polishing - PR Comments JavaDocs polishing
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
49c231e547
commit
3a114ec83a
@@ -44,16 +44,16 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics
|
||||
public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
|
||||
|
||||
private final ExponentialMovingAverage sendDuration = new ExponentialMovingAverage(
|
||||
DEFAULT_MOVING_AVERAGE_WINDOW);
|
||||
DEFAULT_MOVING_AVERAGE_WINDOW, 1000000.);
|
||||
|
||||
private final ExponentialMovingAverageRate sendErrorRate = new ExponentialMovingAverageRate(
|
||||
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
|
||||
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true);
|
||||
|
||||
private final ExponentialMovingAverageRatio sendSuccessRatio = new ExponentialMovingAverageRatio(
|
||||
ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
|
||||
ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true);
|
||||
|
||||
private final ExponentialMovingAverageRate sendRate = new ExponentialMovingAverageRate(
|
||||
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
|
||||
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW, true);
|
||||
|
||||
private final AtomicLong sendCount = new AtomicLong();
|
||||
|
||||
@@ -83,10 +83,10 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics
|
||||
logger.trace("Recording send on channel(" + this.name + ")");
|
||||
}
|
||||
|
||||
double start = 0;
|
||||
long start = 0;
|
||||
if (isFullStatsEnabled()) {
|
||||
start = System.nanoTime() / 1000000.;
|
||||
this.sendRate.increment();
|
||||
start = System.nanoTime();
|
||||
this.sendRate.increment(start);
|
||||
}
|
||||
this.sendCount.incrementAndGet();
|
||||
return new DefaultChannelMetricsContext(start);
|
||||
@@ -95,14 +95,15 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics
|
||||
@Override
|
||||
public void afterSend(MetricsContext context, boolean result) {
|
||||
if (result && isFullStatsEnabled()) {
|
||||
double now = System.nanoTime() / 1000000.;
|
||||
long now = System.nanoTime();
|
||||
this.sendSuccessRatio.success(now);
|
||||
this.sendDuration.append(now - ((DefaultChannelMetricsContext) context).start);
|
||||
}
|
||||
else {
|
||||
if (isFullStatsEnabled()) {
|
||||
this.sendSuccessRatio.failure(System.nanoTime() / 1000000.);
|
||||
this.sendErrorRate.increment();
|
||||
long now = System.nanoTime();
|
||||
this.sendSuccessRatio.failure(now);
|
||||
this.sendErrorRate.increment(now);
|
||||
}
|
||||
this.sendErrorCount.incrementAndGet();
|
||||
}
|
||||
@@ -241,9 +242,9 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics
|
||||
|
||||
private static class DefaultChannelMetricsContext implements MetricsContext {
|
||||
|
||||
private final double start;
|
||||
private final long start;
|
||||
|
||||
public DefaultChannelMetricsContext(double start) {
|
||||
public DefaultChannelMetricsContext(long start) {
|
||||
this.start = start;
|
||||
}
|
||||
|
||||
|
||||
@@ -39,7 +39,8 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics
|
||||
|
||||
private final AtomicLong errorCount = new AtomicLong();
|
||||
|
||||
private final ExponentialMovingAverage duration = new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW);
|
||||
private final ExponentialMovingAverage duration = new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW,
|
||||
1000000.);
|
||||
|
||||
public DefaultMessageHandlerMetrics() {
|
||||
super(null);
|
||||
@@ -55,9 +56,9 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("messageHandler(" + this.name + ") message(" + message + ") :");
|
||||
}
|
||||
double start = 0;
|
||||
long start = 0;
|
||||
if (isFullStatsEnabled()) {
|
||||
start = System.nanoTime() / 1000000.;
|
||||
start = System.nanoTime();
|
||||
}
|
||||
this.handleCount.incrementAndGet();
|
||||
this.activeCount.incrementAndGet();
|
||||
@@ -68,7 +69,7 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics
|
||||
public void afterHandle(MetricsContext context, boolean success) {
|
||||
this.activeCount.decrementAndGet();
|
||||
if (isFullStatsEnabled() && success) {
|
||||
this.duration.append(System.nanoTime() / 1000000. - ((DefaultHandlerMetricsContext) context).start);
|
||||
this.duration.append(System.nanoTime() - ((DefaultHandlerMetricsContext) context).start);
|
||||
}
|
||||
else if (!success) {
|
||||
this.errorCount.incrementAndGet();
|
||||
@@ -142,9 +143,9 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics
|
||||
|
||||
private static class DefaultHandlerMetricsContext implements MetricsContext {
|
||||
|
||||
private final double start;
|
||||
private final long start;
|
||||
|
||||
public DefaultHandlerMetricsContext(double start) {
|
||||
public DefaultHandlerMetricsContext(long start) {
|
||||
this.start = start;
|
||||
}
|
||||
|
||||
|
||||
@@ -13,74 +13,124 @@
|
||||
|
||||
package org.springframework.integration.support.management;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Cumulative statistics for a series of real numbers with higher weight given to recent data but without storing any
|
||||
* history. Clients call {@link #append(double)} every time there is a new measurement, and then can collect summary
|
||||
* Cumulative statistics for a series of real numbers with higher weight given to recent data.
|
||||
* Clients call {@link #append(double)} every time there is a new measurement, and then can collect summary
|
||||
* statistics from the convenience getters (e.g. {@link #getStatistics()}). Older values are given exponentially smaller
|
||||
* weight, with a decay factor determined by a "window" size chosen by the caller. The result is a good approximation to
|
||||
* the statistics of the series but with more weight given to recent measurements, so if the statistics change over time
|
||||
* those trends can be approximately reflected.
|
||||
* those trends can be approximately reflected. For performance reasons, the calculation is performed on retrieval,
|
||||
* {@code window * 5} samples are retained meaning that the earliest retained value contributes just 0.5% to the
|
||||
* sum.
|
||||
*
|
||||
* @author Dave Syer
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
public class ExponentialMovingAverage {
|
||||
|
||||
private volatile long count;
|
||||
|
||||
private volatile double weight;
|
||||
|
||||
private volatile double sum;
|
||||
|
||||
private volatile double sumSquares;
|
||||
|
||||
private volatile double min;
|
||||
private volatile double min = Double.MAX_VALUE;
|
||||
|
||||
private volatile double max;
|
||||
|
||||
private final double decay;
|
||||
private final List<Double> samples = new LinkedList<Double>();
|
||||
|
||||
private final int retention;
|
||||
|
||||
private final int window;
|
||||
|
||||
private final double factor;
|
||||
|
||||
|
||||
/**
|
||||
* Create a moving average accumulator with decay lapse window provided. Measurements older than this will have
|
||||
* smaller weight than <code>1/e</code>.
|
||||
*
|
||||
* @param window the exponential lapse window (number of measurements)
|
||||
*/
|
||||
public ExponentialMovingAverage(int window) {
|
||||
this.decay = 1 - 1. / window;
|
||||
this(window, 1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a moving average accumulator with decay lapse window provided. Measurements older than this will have
|
||||
* smaller weight than <code>1/e</code>.
|
||||
* @param window the exponential lapse window (number of measurements)
|
||||
* @param factor a factor by which raw values are reduced during analysis; e.g. to analyze in ms and
|
||||
* raw values are ns, set the factor to 1000000.0.
|
||||
* @since 4.2
|
||||
*/
|
||||
public ExponentialMovingAverage(int window, double factor) {
|
||||
this.window = window;
|
||||
this.retention = window * 5;// last retained value contributes just 0.5% to the sum
|
||||
this.factor = factor;
|
||||
}
|
||||
|
||||
public synchronized void reset() {
|
||||
weight = 0;
|
||||
sum = 0;
|
||||
sumSquares = 0;
|
||||
count = 0;
|
||||
min = 0;
|
||||
min = Double.MAX_VALUE;
|
||||
max = 0;
|
||||
samples.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a new measurement to the series.
|
||||
*
|
||||
* @param value the measurement to append
|
||||
*/
|
||||
public synchronized void append(double value) {
|
||||
if (value > max || count == 0) {
|
||||
max = value;
|
||||
if (this.samples.size() == this.retention) {
|
||||
samples.remove(0);
|
||||
}
|
||||
if (value < min || count == 0) {
|
||||
min = value;
|
||||
}
|
||||
sum = decay * sum + value;
|
||||
sumSquares = decay * sumSquares + value * value;
|
||||
weight = decay * weight + 1;
|
||||
samples.add(value);
|
||||
count++;//NOSONAR - false positive, we're synchronized
|
||||
}
|
||||
|
||||
private Statistics calc() {
|
||||
List<Double> copy;
|
||||
long count;
|
||||
synchronized (this) {
|
||||
copy = new ArrayList<Double>(this.samples);
|
||||
count = this.count;
|
||||
}
|
||||
double sum = 0;
|
||||
double decay = 1 - 1. / this.window;
|
||||
double sumSquares = 0;
|
||||
double weight = 0;
|
||||
double min = this.min;
|
||||
double max = this.max;
|
||||
for (Double value : copy) {
|
||||
value /= this.factor;
|
||||
if (value > max) {
|
||||
max = value;
|
||||
}
|
||||
if (value < min) {
|
||||
min = value;
|
||||
}
|
||||
sum = decay * sum + value;
|
||||
sumSquares = decay * sumSquares + value * value;
|
||||
weight = decay * weight + 1;
|
||||
}
|
||||
synchronized (this) {
|
||||
if (max > this.max) {
|
||||
this.max = max;
|
||||
}
|
||||
if (min < this.min) {
|
||||
this.min = min;
|
||||
}
|
||||
}
|
||||
double mean = weight > 0 ? sum / weight : 0.;
|
||||
double var = weight > 0 ? sumSquares / weight - mean * mean : 0.;
|
||||
double standardDeviation = var > 0 ? Math.sqrt(var) : 0;
|
||||
return new Statistics(count, min == Double.MAX_VALUE ? 0 : min, max, mean, standardDeviation);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the number of measurements recorded
|
||||
*/
|
||||
@@ -99,37 +149,35 @@ public class ExponentialMovingAverage {
|
||||
* @return the mean value
|
||||
*/
|
||||
public double getMean() {
|
||||
return weight > 0 ? sum / weight : 0.;
|
||||
return calc().getMean();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the approximate standard deviation
|
||||
*/
|
||||
public double getStandardDeviation() {
|
||||
double mean = getMean();
|
||||
double var = weight > 0 ? sumSquares / weight - mean * mean : 0.;
|
||||
return var > 0 ? Math.sqrt(var) : 0;
|
||||
return calc().getStandardDeviation();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the maximum value recorded (not weighted)
|
||||
*/
|
||||
public double getMax() {
|
||||
return max;
|
||||
return calc().getMax();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the minimum value recorded (not weighted)
|
||||
*/
|
||||
public double getMin() {
|
||||
return min;
|
||||
return calc().getMin();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return summary statistics (count, mean, standard deviation etc.)
|
||||
*/
|
||||
public Statistics getStatistics() {
|
||||
return new Statistics(count, min, max, getMean(), getStandardDeviation());
|
||||
return calc();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -13,10 +13,14 @@
|
||||
|
||||
package org.springframework.integration.support.management;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Cumulative statistics for an event rate with higher weight given to recent data but without storing any history.
|
||||
* Cumulative statistics for an event rate with higher weight given to recent data.
|
||||
* Clients call {@link #increment()} when a new event occurs, and then use convenience methods (e.g. {@link #getMean()})
|
||||
* to retrieve estimates of the rate of event arrivals and the statistics of the series. Older values are given
|
||||
* exponentially smaller weight, with a decay factor determined by a duration chosen by the client. The rate measurement
|
||||
@@ -27,29 +31,36 @@ package org.springframework.integration.support.management;
|
||||
* <li>per measurement according to the lapse window supplied: <code>weight = exp(-i/L)</code> where <code>L</code> is
|
||||
* the lapse window and <code>i</code> is the sequence number of the measurement.</li>
|
||||
* </ul>
|
||||
*
|
||||
* For performance reasons, the calculation is performed on retrieval,
|
||||
* {@code window * 5} samples are retained meaning that the earliest retained value contributes just 0.5% to the
|
||||
* sum.
|
||||
* @author Dave Syer
|
||||
* @author Gary Russell
|
||||
*
|
||||
*/
|
||||
public class ExponentialMovingAverageRate {
|
||||
|
||||
private final ExponentialMovingAverage rates;
|
||||
|
||||
private volatile double weight;
|
||||
|
||||
private volatile double sum;
|
||||
|
||||
private volatile double min;
|
||||
private volatile double min = Double.MAX_VALUE;
|
||||
|
||||
private volatile double max;
|
||||
|
||||
private volatile double t0 = System.nanoTime() / 1000000.;
|
||||
private volatile double t0;
|
||||
|
||||
private volatile long count;
|
||||
|
||||
private final double lapse;
|
||||
|
||||
private final double period;
|
||||
|
||||
private final List<Long> times = new LinkedList<Long>();
|
||||
|
||||
private final int retention;
|
||||
|
||||
private final int window;
|
||||
|
||||
private final double factor;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* @param period the period to base the rate measurement (in seconds)
|
||||
@@ -57,45 +68,107 @@ public class ExponentialMovingAverageRate {
|
||||
* @param window the exponential lapse window (number of measurements)
|
||||
*/
|
||||
public ExponentialMovingAverageRate(double period, double lapsePeriod, int window) {
|
||||
rates = new ExponentialMovingAverage(window);
|
||||
this(period, lapsePeriod, window, false);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param period the period to base the rate measurement (in seconds)
|
||||
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
|
||||
* @param window the exponential lapse window (number of measurements)
|
||||
* @param millis when true, analyze the data as milliseconds instead of the native nanoseconds
|
||||
* @since 4.2
|
||||
*/
|
||||
public ExponentialMovingAverageRate(double period, double lapsePeriod, int window, boolean millis) {
|
||||
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to milliseconds
|
||||
this.period = period * 1000; // convert to milliseconds
|
||||
this.window = window;
|
||||
this.retention = window * 5;
|
||||
this.factor = millis ? 1000000 : 1;
|
||||
this.t0 = System.nanoTime() / this.factor;
|
||||
}
|
||||
|
||||
|
||||
public synchronized void reset() {
|
||||
min = 0;
|
||||
max = 0;
|
||||
weight = 0;
|
||||
sum = 0;
|
||||
t0 = System.nanoTime() / 1000000.;
|
||||
rates.reset();
|
||||
this.min = Double.MAX_VALUE;
|
||||
this.max = 0;
|
||||
this.count = 0;
|
||||
this.times.clear();
|
||||
t0 = System.nanoTime() / this.factor;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a new event to the series.
|
||||
*/
|
||||
public synchronized void increment() {
|
||||
double t = System.nanoTime() / 1000000.;
|
||||
double value = t > t0 ? (t - t0) / period : 0;
|
||||
if (value > max || getCount() == 0) {
|
||||
max = value;
|
||||
increment(System.nanoTime());
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a new event to the series at time t.
|
||||
* @param t a new event to the series in milliseconds.
|
||||
*/
|
||||
public synchronized void increment(long t) {
|
||||
if (this.times.size() == this.retention) {
|
||||
this.times.remove(0);
|
||||
}
|
||||
if (value < min || getCount() == 0) {
|
||||
min = value;
|
||||
this.times.add(t);
|
||||
this.count++;//NOSONAR - false positive, we're synchronized
|
||||
}
|
||||
|
||||
private Statistics calc() {
|
||||
List<Long> copy;
|
||||
long count;
|
||||
synchronized (this) {
|
||||
copy = new ArrayList<Long>(this.times);
|
||||
count = this.count;
|
||||
}
|
||||
double alpha = Math.exp((t0 - t) * lapse);
|
||||
t0 = t;
|
||||
sum = alpha * sum + value;
|
||||
weight = alpha * weight + 1;
|
||||
rates.append(sum > 0 ? weight / sum : 0);
|
||||
ExponentialMovingAverage rates = new ExponentialMovingAverage(window);
|
||||
double t0 = 0;
|
||||
double sum = 0;
|
||||
double weight = 0;
|
||||
double min = this.min;
|
||||
double max = this.max;
|
||||
int size = copy.size();
|
||||
for (Long time : copy) {
|
||||
double t = time / this.factor;
|
||||
if (size == 1) {
|
||||
t0 = this.t0;
|
||||
}
|
||||
else if (t0 == 0) {
|
||||
t0 = t;
|
||||
continue;
|
||||
}
|
||||
double delta = t - t0;
|
||||
double value = delta > 0 ? delta / period : 0;
|
||||
if (value > max) {
|
||||
max = value;
|
||||
}
|
||||
if (value < min) {
|
||||
min = value;
|
||||
}
|
||||
double alpha = Math.exp(-delta * lapse);
|
||||
t0 = t;
|
||||
sum = alpha * sum + value;
|
||||
weight = alpha * weight + 1;
|
||||
rates.append(sum > 0 ? weight / sum : 0);
|
||||
}
|
||||
synchronized (this) {
|
||||
if (max > this.max) {
|
||||
this.max = max;
|
||||
}
|
||||
if (min < this.min) {
|
||||
this.min = min;
|
||||
}
|
||||
}
|
||||
return new Statistics(count, min < Double.MAX_VALUE ? min : 0, max, rates.getMean(),
|
||||
rates.getStandardDeviation());
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the number of measurements recorded
|
||||
*/
|
||||
public int getCount() {
|
||||
return rates.getCount();
|
||||
return (int) this.count;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -103,40 +176,55 @@ public class ExponentialMovingAverageRate {
|
||||
* @since 3.0
|
||||
*/
|
||||
public long getCountLong() {
|
||||
return rates.getCountLong();
|
||||
return this.count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in seconds since the last measurement
|
||||
*/
|
||||
public double getTimeSinceLastMeasurement() {
|
||||
return System.nanoTime() / 1000000. - t0;
|
||||
double t0 = lastTime();
|
||||
return (System.nanoTime() / this.factor - t0);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the mean value
|
||||
*/
|
||||
public double getMean() {
|
||||
long count = rates.getCountLong();
|
||||
long count = this.count;
|
||||
count = count > this.retention ? this.retention : count;
|
||||
if (count == 0) {
|
||||
return 0;
|
||||
}
|
||||
double t = System.nanoTime() / 1000000.;
|
||||
double t0 = lastTime();
|
||||
double t = System.nanoTime() / this.factor;
|
||||
double value = t > t0 ? (t - t0) / period : 0;
|
||||
return count / (count / rates.getMean() + value);
|
||||
return count / (count / calc().getMean() + value);
|
||||
}
|
||||
|
||||
private double lastTime() {
|
||||
if (this.times.size() > 0) {
|
||||
synchronized (this) {
|
||||
return this.times.get(this.times.size() - 1) / this.factor;
|
||||
}
|
||||
}
|
||||
else {
|
||||
return this.t0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the approximate standard deviation
|
||||
*/
|
||||
public double getStandardDeviation() {
|
||||
return rates.getStandardDeviation();
|
||||
return calc().getStandardDeviation();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the maximum value recorded (not weighted)
|
||||
*/
|
||||
public double getMax() {
|
||||
double min = calc().getMin();
|
||||
return min > 0 ? 1 / min : 0;
|
||||
}
|
||||
|
||||
@@ -144,6 +232,7 @@ public class ExponentialMovingAverageRate {
|
||||
* @return the minimum value recorded (not weighted)
|
||||
*/
|
||||
public double getMin() {
|
||||
double max = calc().getMax();
|
||||
return max > 0 ? 1 / max : 0;
|
||||
}
|
||||
|
||||
@@ -151,7 +240,7 @@ public class ExponentialMovingAverageRate {
|
||||
* @return summary statistics (count, mean, standard deviation etc.)
|
||||
*/
|
||||
public Statistics getStatistics() {
|
||||
return new Statistics(getCount(), min, max, getMean(), getStandardDeviation());
|
||||
return calc();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -13,10 +13,15 @@
|
||||
|
||||
package org.springframework.integration.support.management;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Cumulative statistics for success ratio with higher weight given to recent data but without storing any history.
|
||||
* Cumulative statistics for success ratio with higher weight given to recent data.
|
||||
* Clients call {@link #success()} or {@link #failure()} when an event occurs, and the ratio of success to total events
|
||||
* is accumulated. Older values are given exponentially smaller weight, with a decay factor determined by a duration
|
||||
* chosen by the client. The rate measurement weights decay in two dimensions:
|
||||
@@ -26,31 +31,55 @@ package org.springframework.integration.support.management;
|
||||
* <li>per measurement according to the lapse window supplied: <code>weight = exp(-i/L)</code> where <code>L</code> is
|
||||
* the lapse window and <code>i</code> is the sequence number of the measurement.</li>
|
||||
* </ul>
|
||||
*
|
||||
* For performance reasons, the calculation is performed on retrieval,
|
||||
* {@code window * 5} samples are retained meaning that the earliest retained value contributes just 0.5% to the
|
||||
* sum.
|
||||
* @author Dave Syer
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
public class ExponentialMovingAverageRatio {
|
||||
|
||||
private volatile double weight;
|
||||
private volatile double t0;
|
||||
|
||||
private volatile double sum;
|
||||
private volatile long count;
|
||||
|
||||
private volatile double t0 = System.nanoTime() / 1000000.;
|
||||
private volatile double min = Double.MAX_VALUE;
|
||||
|
||||
private volatile double max;
|
||||
|
||||
private final double lapse;
|
||||
|
||||
private final ExponentialMovingAverage cumulative;
|
||||
private final List<Long> times = new LinkedList<Long>();
|
||||
|
||||
private final List<Integer> values = new LinkedList<Integer>();
|
||||
|
||||
private final int retention;
|
||||
|
||||
private final int window;
|
||||
|
||||
private final double factor;
|
||||
|
||||
/**
|
||||
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
|
||||
* @param window the exponential lapse window (number of measurements)
|
||||
*/
|
||||
public ExponentialMovingAverageRatio(double lapsePeriod, int window) {
|
||||
this.cumulative = new ExponentialMovingAverage(window);
|
||||
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
|
||||
this(lapsePeriod, window, false);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
|
||||
* @param window the exponential lapse window (number of measurements)
|
||||
* @param millis when true, analyze the data as milliseconds instead of the native nanoseconds
|
||||
* @since 4.2
|
||||
*/
|
||||
public ExponentialMovingAverageRatio(double lapsePeriod, int window, boolean millis) {
|
||||
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to milliseconds
|
||||
this.window = window;
|
||||
this.retention = window * 5;
|
||||
this.factor = millis ? 1000000 : 1;
|
||||
this.t0 = System.nanoTime() / factor;
|
||||
}
|
||||
|
||||
|
||||
@@ -58,14 +87,14 @@ public class ExponentialMovingAverageRatio {
|
||||
* Add a new event with successful outcome.
|
||||
*/
|
||||
public void success() {
|
||||
append(1, System.nanoTime() / 1000000.);
|
||||
append(1, System.nanoTime());
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a new event with successful outcome at time t.
|
||||
* @param t the time in milliseconds.
|
||||
*/
|
||||
public void success(double t) {
|
||||
public void success(long t) {
|
||||
append(1, t);
|
||||
}
|
||||
|
||||
@@ -73,92 +102,162 @@ public class ExponentialMovingAverageRatio {
|
||||
* Add a new event with failed outcome.
|
||||
*/
|
||||
public void failure() {
|
||||
append(0, System.nanoTime() / 1000000.);
|
||||
append(0, System.nanoTime());
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a new event with failed outcome at time t.
|
||||
* @param t a new event with failed outcome in milliseconds.
|
||||
*/
|
||||
public void failure(double t) {
|
||||
public void failure(long t) {
|
||||
append(0, t);
|
||||
}
|
||||
|
||||
public synchronized void reset() {
|
||||
weight = 0;
|
||||
sum = 0;
|
||||
t0 = System.nanoTime() / 1000000.;
|
||||
cumulative.reset();
|
||||
t0 = System.nanoTime() / this.factor;
|
||||
this.times.clear();
|
||||
this.values.clear();
|
||||
this.count = 0;
|
||||
this.max = 0;
|
||||
this.min = Double.MAX_VALUE;
|
||||
}
|
||||
|
||||
private synchronized void append(int value, double t) {
|
||||
double alpha = Math.exp((t0 - t) * lapse);
|
||||
t0 = t;
|
||||
sum = alpha * sum + value;
|
||||
weight = alpha * weight + 1;
|
||||
cumulative.append(sum / weight);
|
||||
private synchronized void append(int value, long t) {
|
||||
if (this.times.size() == this.retention) {
|
||||
this.times.remove(0);
|
||||
this.values.remove(0);
|
||||
}
|
||||
this.times.add(t);
|
||||
this.values.add(value);
|
||||
this.count++;//NOSONAR - false positive, we're synchronized
|
||||
}
|
||||
|
||||
private Statistics calc() {
|
||||
List<Long> copyTimes;
|
||||
List<Integer> copyValues;
|
||||
long count;
|
||||
synchronized (this) {
|
||||
copyTimes = new ArrayList<Long>(this.times);
|
||||
copyValues = new ArrayList<Integer>(this.values);
|
||||
count = this.count;
|
||||
}
|
||||
ExponentialMovingAverage cumulative = new ExponentialMovingAverage(window);
|
||||
double t0 = 0;
|
||||
double sum = 0;
|
||||
double weight = 0;
|
||||
double min = this.min;
|
||||
double max = this.max;
|
||||
int size = copyTimes.size();
|
||||
Iterator<Integer> values = copyValues.iterator();
|
||||
for (Long time : copyTimes) {
|
||||
double t = time / this.factor;
|
||||
if (size == 1) {
|
||||
t0 = this.t0;
|
||||
}
|
||||
else if (t0 == 0) {
|
||||
t0 = t;
|
||||
values.next();
|
||||
continue;
|
||||
}
|
||||
double alpha = Math.exp((t0 - t) * this.lapse);
|
||||
t0 = t;
|
||||
sum = alpha * sum + values.next();
|
||||
weight = alpha * weight + 1;
|
||||
double value = sum / weight;
|
||||
if (value > max) {
|
||||
max = value;
|
||||
}
|
||||
if (value < min) {
|
||||
min = value;
|
||||
}
|
||||
cumulative.append(value);
|
||||
}
|
||||
synchronized (this) {
|
||||
if (max > this.max) {
|
||||
this.max = max;
|
||||
}
|
||||
if (min < this.min) {
|
||||
this.min = min;
|
||||
}
|
||||
}
|
||||
return new Statistics(count, min < Double.MAX_VALUE ? min : 0, max, cumulative.getMean(),
|
||||
cumulative.getStandardDeviation());
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the number of measurements recorded
|
||||
*/
|
||||
public int getCount() {
|
||||
return cumulative.getCount();
|
||||
return (int) this.count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the number of measurements recorded
|
||||
*/
|
||||
public long getCountLong() {
|
||||
return cumulative.getCountLong();
|
||||
return this.count;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in seconds since the last measurement
|
||||
*/
|
||||
public double getTimeSinceLastMeasurement() {
|
||||
return (System.nanoTime() / 1000000. - t0) / 1000.;
|
||||
double delta = System.nanoTime() - lastTime();
|
||||
return delta / 1000. / this.factor;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the mean success rate
|
||||
*/
|
||||
public double getMean() {
|
||||
long count = cumulative.getCountLong();
|
||||
if (count == 0) {
|
||||
if (this.count == 0) {
|
||||
// Optimistic to start: success rate is 100%
|
||||
return 1;
|
||||
}
|
||||
double t = System.nanoTime() / 1000000.;
|
||||
double alpha = Math.exp((t0 - t) * lapse);
|
||||
return alpha * cumulative.getMean() + 1 - alpha;
|
||||
Statistics statistics = calc();
|
||||
double t = System.nanoTime() / this.factor;
|
||||
double mean = statistics.getMean();
|
||||
double alpha = Math.exp((lastTime() / this.factor - t) * this.lapse);
|
||||
return alpha * mean + 1 - alpha;
|
||||
}
|
||||
|
||||
private double lastTime() {
|
||||
if (this.times.size() > 0) {
|
||||
synchronized (this) {
|
||||
return this.times.get(this.times.size() - 1);
|
||||
}
|
||||
}
|
||||
else {
|
||||
return this.t0 * this.factor;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the approximate standard deviation of the success rate measurements
|
||||
*/
|
||||
public double getStandardDeviation() {
|
||||
return cumulative.getStandardDeviation();
|
||||
return calc().getStandardDeviation();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the maximum value recorded of the exponential weighted average (per measurement) success rate
|
||||
*/
|
||||
public double getMax() {
|
||||
return cumulative.getMax();
|
||||
return calc().getMax();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the minimum value recorded of the exponential weighted average (per measurement) success rate
|
||||
*/
|
||||
public double getMin() {
|
||||
return cumulative.getMin();
|
||||
return calc().getMin();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return summary statistics (count, mean, standard deviation etc.)
|
||||
*/
|
||||
public Statistics getStatistics() {
|
||||
return new Statistics(getCount(), getMin(), getMax(), getMean(), getStandardDeviation());
|
||||
return calc();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -25,7 +25,7 @@ import java.lang.annotation.Target;
|
||||
import org.springframework.jmx.export.annotation.ManagedResource;
|
||||
|
||||
/**
|
||||
* Clone of {@link ManagedResource} limiting beans thus annoated so that they
|
||||
* Clone of {@link ManagedResource} limiting beans thus annotated so that they
|
||||
* will only be exported by the {@code IntegrationMBeanExporter}.
|
||||
*
|
||||
* @author Gary Russell
|
||||
|
||||
@@ -12,8 +12,10 @@
|
||||
*/
|
||||
package org.springframework.integration.support.management;
|
||||
|
||||
import static org.hamcrest.Matchers.lessThan;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
@@ -21,7 +23,6 @@ import org.apache.commons.logging.LogFactory;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.util.StopWatch;
|
||||
|
||||
/**
|
||||
@@ -33,14 +34,7 @@ public class ExponentialMovingAverageRateTests {
|
||||
|
||||
private final static Log logger = LogFactory.getLog(ExponentialMovingAverageRateTests.class);
|
||||
|
||||
private final ExponentialMovingAverageRate history = new ExponentialMovingAverageRate(1., 10., 10);
|
||||
|
||||
@Test
|
||||
public void testWindow() {
|
||||
ExponentialMovingAverageRate rate = new ExponentialMovingAverageRate(1., 10., 20);
|
||||
double decay = TestUtils.getPropertyValue(rate, "rates.decay", Double.class);
|
||||
assertEquals(95, (int) (decay * 100.));
|
||||
}
|
||||
private final ExponentialMovingAverageRate history = new ExponentialMovingAverageRate(1., 10., 10, true);
|
||||
|
||||
@Test
|
||||
public void testGetCount() {
|
||||
@@ -86,7 +80,7 @@ public class ExponentialMovingAverageRateTests {
|
||||
Thread.sleep(20L);
|
||||
elapsed = System.currentTimeMillis() - t0;
|
||||
if (elapsed < 80L) {
|
||||
assertTrue(history.getMean() < before);
|
||||
assertThat(history.getMean(), lessThan(before));
|
||||
}
|
||||
else {
|
||||
logger.warn("Test took too long to verify mean");
|
||||
@@ -127,7 +121,7 @@ public class ExponentialMovingAverageRateTests {
|
||||
assertEquals(0, history.getMax(), 0.01);
|
||||
}
|
||||
|
||||
@Test
|
||||
@Test @Ignore // tolerance needed is too dependent on hardware
|
||||
public void testRate() {
|
||||
ExponentialMovingAverageRate rate = new ExponentialMovingAverageRate(1, 60, 10);
|
||||
int count = 1000000;
|
||||
@@ -138,7 +132,15 @@ public class ExponentialMovingAverageRateTests {
|
||||
}
|
||||
watch.stop();
|
||||
double calculatedRate = count / (double) watch.getTotalTimeMillis() * 1000;
|
||||
assertEquals(calculatedRate, rate.getMean(), 2000000);
|
||||
assertEquals(calculatedRate, rate.getMean(), 4000000);
|
||||
}
|
||||
|
||||
@Test @Ignore
|
||||
public void testPerf() {
|
||||
ExponentialMovingAverageRate rate = new ExponentialMovingAverageRate(1, 60, 10);
|
||||
for (int i = 0; i < 1000000; i++) {
|
||||
rate.increment();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -15,10 +15,13 @@
|
||||
*/
|
||||
package org.springframework.integration.support.management;
|
||||
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.hamcrest.Matchers.not;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
@@ -28,7 +31,7 @@ import org.junit.Test;
|
||||
public class ExponentialMovingAverageRatioTests {
|
||||
|
||||
private final ExponentialMovingAverageRatio history = new ExponentialMovingAverageRatio(
|
||||
0.5, 10);
|
||||
0.5, 10, true);
|
||||
|
||||
@Test
|
||||
public void testGetCount() {
|
||||
@@ -41,7 +44,7 @@ public class ExponentialMovingAverageRatioTests {
|
||||
public void testGetTimeSinceLastMeasurement() throws Exception {
|
||||
history.success();
|
||||
Thread.sleep(20L);
|
||||
assertTrue(history.getTimeSinceLastMeasurement() > 0);
|
||||
assertThat(history.getTimeSinceLastMeasurement(), Matchers.greaterThan(0.));
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -79,6 +82,7 @@ public class ExponentialMovingAverageRatioTests {
|
||||
@Test
|
||||
public void testGetMeanFailuresHighRate() throws Exception {
|
||||
assertEquals(1, history.getMean(), 0.01);
|
||||
history.success();// need an extra now that we can't determine the time between the first and previous
|
||||
history.success();
|
||||
assertEquals(average(1), history.getMean(), 0.01);
|
||||
history.failure();
|
||||
@@ -90,6 +94,7 @@ public class ExponentialMovingAverageRatioTests {
|
||||
@Test
|
||||
public void testGetMeanFailuresLowRate() throws Exception {
|
||||
assertEquals(1, history.getMean(), 0.01);
|
||||
history.failure();// need an extra now that we can't determine the time between the first and previous
|
||||
history.failure();
|
||||
assertEquals(average(0), history.getMean(), 0.01);
|
||||
history.failure();
|
||||
@@ -110,7 +115,7 @@ public class ExponentialMovingAverageRatioTests {
|
||||
assertEquals(0, history.getStandardDeviation(), 0.01);
|
||||
history.success();
|
||||
history.failure();
|
||||
assertFalse(0==history.getStandardDeviation());
|
||||
assertThat(history.getStandardDeviation(), not(equalTo(0)));
|
||||
history.reset();
|
||||
assertEquals(0, history.getStandardDeviation(), 0.01);
|
||||
assertEquals(0, history.getCount());
|
||||
@@ -118,6 +123,8 @@ public class ExponentialMovingAverageRatioTests {
|
||||
assertEquals(1, history.getMean(), 0.01);
|
||||
assertEquals(0, history.getMin(), 0.01);
|
||||
assertEquals(0, history.getMax(), 0.01);
|
||||
history.success();
|
||||
assertEquals(1, history.getMin(), 0.01);
|
||||
}
|
||||
|
||||
private double average(double... values) {
|
||||
@@ -132,8 +139,23 @@ public class ExponentialMovingAverageRatioTests {
|
||||
|
||||
@Test
|
||||
public void testRatio() {
|
||||
ExponentialMovingAverageRatio ratio = new ExponentialMovingAverageRatio(60, 10, true);
|
||||
for (int i = 0; i < 100; i++) {
|
||||
if (i % 10 == 1) {
|
||||
ratio.failure();
|
||||
}
|
||||
else {
|
||||
ratio.success();
|
||||
}
|
||||
}
|
||||
assertEquals(0.9, ratio.getMax(), 0.02);
|
||||
assertEquals(0.9, ratio.getMean(), 0.03);
|
||||
}
|
||||
|
||||
@Test @Ignore
|
||||
public void testPerf() {
|
||||
ExponentialMovingAverageRatio ratio = new ExponentialMovingAverageRatio(60, 10);
|
||||
for (int i = 0; i < 10000; i++) {
|
||||
for (int i = 0; i < 100000; i++) {
|
||||
if (i % 10 == 0) {
|
||||
ratio.failure();
|
||||
}
|
||||
@@ -141,8 +163,6 @@ public class ExponentialMovingAverageRatioTests {
|
||||
ratio.success();
|
||||
}
|
||||
}
|
||||
assertEquals(0.9, ratio.getMax(), 0.01);
|
||||
assertEquals(0.9, ratio.getMean(), 0.01);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ package org.springframework.integration.support.management;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
@@ -63,6 +64,8 @@ public class ExponentialMovingAverageTests {
|
||||
assertEquals(0, history.getStandardDeviation(), 0.01);
|
||||
// INT-2165
|
||||
assertEquals(String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", 0, 0d, 0d, 0d, 0d), history.toString());
|
||||
history.append(1);
|
||||
assertEquals(1, history.getMin(), 0.01);
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -86,5 +89,26 @@ public class ExponentialMovingAverageTests {
|
||||
assertEquals(30, av.getMean(), 1.0);
|
||||
}
|
||||
|
||||
@Test @Ignore
|
||||
public void testPerf() throws Exception {
|
||||
ExponentialMovingAverage av = new ExponentialMovingAverage(10);
|
||||
for (int i = 0; i < 10000000; i++) {
|
||||
switch (i % 4) {
|
||||
case 0:
|
||||
av.append(20);
|
||||
break;
|
||||
case 1:
|
||||
av.append(30);
|
||||
break;
|
||||
case 2:
|
||||
av.append(40);
|
||||
break;
|
||||
|
||||
case 3:
|
||||
av.append(50);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -27,6 +27,7 @@ import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.integration.channel.NullChannel;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
@@ -66,7 +67,7 @@ public class DynamicRouterTests {
|
||||
private PollableChannel processCChannel;
|
||||
|
||||
@Autowired
|
||||
private MessageChannel nullChannel;
|
||||
private NullChannel nullChannel;
|
||||
|
||||
|
||||
@Test @DirtiesContext
|
||||
@@ -116,10 +117,11 @@ public class DynamicRouterTests {
|
||||
|
||||
@Test @DirtiesContext @Ignore
|
||||
public void testPerf() throws Exception {
|
||||
for (int i = 0; i < 10000000; i++) {
|
||||
// this.nullChannel.enableStats(false);
|
||||
for (int i = 0; i < 1000000000; i++) {
|
||||
this.nullChannel.send(null);
|
||||
}
|
||||
System.out.println("done");
|
||||
System.out.println(this.nullChannel.getSendRate());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user