Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *   http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.apache.reef.wake.metrics;
020
021import java.util.concurrent.TimeUnit;
022import java.util.concurrent.atomic.AtomicLong;
023
024/**
025 * Exponentially weighted moving average.
026 */
027public class EWMA {
028
029  private final double alpha;
030  private final double interval;
031  private final AtomicLong uncounted = new AtomicLong();
032  private volatile boolean initialized = false;
033  private volatile double rate = 0.0;
034
035  /**
036   * Constructs an EWMA object.
037   *
038   * @param alpha
039   * @param interval
040   * @param intervalUnit
041   */
042  public EWMA(final double alpha, final long interval, final TimeUnit intervalUnit) {
043    this.interval = intervalUnit.toNanos(interval);
044    this.alpha = alpha;
045  }
046
047  /**
048   * Updates the EWMA with a new value.
049   *
050   * @param n the new value
051   */
052  public void update(final long n) {
053    uncounted.addAndGet(n);
054  }
055
056  /**
057   * Updates the rate.
058   */
059  public void tick() {
060    final long count = uncounted.getAndSet(0);
061    final double instantRate = count / interval;
062    if (initialized) {
063      rate += alpha * (instantRate - rate);
064    } else {
065      rate = instantRate;
066      initialized = true;
067    }
068  }
069
070  /**
071   * Gets the rate.
072   *
073   * @param rateUnit the unit of the rate
074   * @return the rate
075   */
076  public double getRate(final TimeUnit rateUnit) {
077    return rate * (double) rateUnit.toNanos(1);
078  }
079}