From 3a114ec83a4d295cc01dfa4749f1b6a7a58e8598 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 25 Feb 2015 14:28:02 +0200 Subject: [PATCH] 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 --- .../DefaultMessageChannelMetrics.java | 25 +-- .../DefaultMessageHandlerMetrics.java | 13 +- .../management/ExponentialMovingAverage.java | 114 ++++++++---- .../ExponentialMovingAverageRate.java | 161 +++++++++++++---- .../ExponentialMovingAverageRatio.java | 167 ++++++++++++++---- .../IntegrationManagedResource.java | 2 +- .../ExponentialMovingAverageRateTests.java | 26 +-- .../ExponentialMovingAverageRatioTests.java | 36 +++- .../ExponentialMovingAverageTests.java | 24 +++ .../jmx/config/DynamicRouterTests.java | 8 +- 10 files changed, 431 insertions(+), 145 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java index 29c84c657b..6ace7ec15f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java index 6a0b2c9f33..208f42484e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java index 6b424e2863..da0d4879de 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java @@ -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 samples = new LinkedList(); + + 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 1/e. - * * @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 1/e. + * @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 copy; + long count; + synchronized (this) { + copy = new ArrayList(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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java index 93dabbbf90..86e08c3c2a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java @@ -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; *
  • per measurement according to the lapse window supplied: weight = exp(-i/L) where L is * the lapse window and i is the sequence number of the measurement.
    • * - * + * 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 times = new LinkedList(); + + 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 copy; + long count; + synchronized (this) { + copy = new ArrayList(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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java index 352494c63f..cb6b0d9216 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java @@ -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; *
  • per measurement according to the lapse window supplied: weight = exp(-i/L) where L is * the lapse window and i is the sequence number of the measurement.
    • * - * + * 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 times = new LinkedList(); + private final List values = new LinkedList(); + + 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 copyTimes; + List copyValues; + long count; + synchronized (this) { + copyTimes = new ArrayList(this.times); + copyValues = new ArrayList(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 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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagedResource.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagedResource.java index 9a584a15e5..a2b641ca4c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagedResource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/IntegrationManagedResource.java @@ -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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java index de5be40888..d49d420d16 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java @@ -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(); + } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java index 2591ba4f39..b7fcabfb76 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java @@ -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); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java index d6b9319f1d..02a648da5a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java @@ -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; + } + } + } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java index e2a6038f3b..c79d223eee 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java @@ -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()); } }