Rearrangement and refinement of statistic code in core

Signed-off-by: Eric Zhao <sczyh16@gmail.com>

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/MetricEvent.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.csp.sentinel.slots.statistic;
+
+/**
+ * @author Eric Zhao
+ */
+public enum MetricEvent {
+
+    /**
+     * Normal pass.
+     */
+    PASS,
+    /**
+     * Normal block.
+     */
+    BLOCK,
+    EXCEPTION,
+    SUCCESS,
+    RT,
+    OCCUPIED_PASS,
+    OCCUPIED_BLOCK,
+    WAITING
+}

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/base/LeapArray.java

@@ -25,12 +25,12 @@ import com.alibaba.csp.sentinel.util.TimeUtil;
 
 /**
  * <p>
- * Basic data structure for statistic metrics.
+ * Basic data structure for statistic metrics in Sentinel.
  * </p>
  * <p>
- * Using sliding window algorithm to count data. Each bucket cover {@link #windowLengthInMs} time span,
+ * Leap array use sliding window algorithm to count data. Each bucket cover {code windowLengthInMs} time span,
- * and the total time span is {@link #intervalInMs}, so the total bucket count is:
+ * and the total time span is {@link #intervalInMs}, so the total bucket amount is:
- * {@link #sampleCount} = intervalInMs / windowLengthInMs.
+ * {@code sampleCount = intervalInMs / windowLengthInMs}.
  * </p>
  *
  * @param <T> type of statistic data
@@ -47,7 +47,7 @@ public abstract class LeapArray<T> {
     protected final AtomicReferenceArray<WindowWrap<T>> array;
 
     /**
-     * The fine-grained update lock is used only when current bucket is deprecated.
+     * The conditional (predicate) update lock is used only when current bucket is deprecated.
      */
     private final ReentrantLock updateLock = new ReentrantLock();
 
@@ -58,9 +58,10 @@ public abstract class LeapArray<T> {
      * @param intervalInSec    the total time span of this {@link LeapArray} in seconds.
      */
     public LeapArray(int windowLengthInMs, int intervalInSec) {
+        // TODO: change `intervalInSec` to `intervalInMs`
         AssertUtil.isTrue(windowLengthInMs > 0, "bucket length is invalid: " + windowLengthInMs);
         int intervalInMs = intervalInSec * 1000;
-        AssertUtil.isTrue(intervalInSec * 1000 > windowLengthInMs,
+        AssertUtil.isTrue(intervalInMs > windowLengthInMs,
             "total time span of the window should be greater than bucket length");
         AssertUtil.isTrue(intervalInMs % windowLengthInMs == 0, "time span needs to be evenly divided");
 
@@ -90,28 +91,36 @@ public abstract class LeapArray<T> {
     /**
      * Reset given bucket to provided start time and reset the value.
      *
-     * @param startTime  the start time of the bucket
+     * @param startTime  the start time of the bucket in milliseconds
      * @param windowWrap current bucket
      * @return new clean bucket at given start time
      */
     protected abstract WindowWrap<T> resetWindowTo(WindowWrap<T> windowWrap, long startTime);
 
+    protected int calculateTimeIdx(/*@Valid*/ long timeMillis) {
+        long timeId = timeMillis / windowLengthInMs;
+        // Calculate current index so we can map the timestamp to the leap array.
+        return (int)(timeId % array.length());
+    }
+
+    protected long calculateWindowStart(/*@Valid*/ long timeMillis) {
+        return timeMillis - timeMillis % windowLengthInMs;
+    }
+
     /**
      * Get bucket item at provided timestamp.
      *
-     * @param time a valid timestamp
+     * @param timeMillis a valid timestamp in milliseconds
      * @return current bucket item at provided timestamp if the time is valid; null if time is invalid
      */
-    public WindowWrap<T> currentWindow(long time) {
+    public WindowWrap<T> currentWindow(long timeMillis) {
-        if (time < 0) {
+        if (timeMillis < 0) {
             return null;
         }
-        long timeId = time / windowLengthInMs;
-        // Calculate current index so we can map the timestamp to the leap array.
-        int idx = (int)(timeId % array.length());
 
+        int idx = calculateTimeIdx(timeMillis);
         // Calculate current bucket start time.
-        long windowStart = time - time % windowLengthInMs;
+        long windowStart = calculateWindowStart(timeMillis);
 
         /*
          * Get bucket item at given time from the array.
@@ -171,7 +180,7 @@ public abstract class LeapArray<T> {
                  * Note that the reset and clean-up operations are hard to be atomic,
                  * so we need a update lock to guarantee the correctness of bucket update.
                  *
-                 * The update lock is fine-grained and will take effect only when
+                 * The update lock is conditional (tiny scope) and will take effect only when
                  * bucket is deprecated, so in most cases it won't lead to performance loss.
                  */
                 if (updateLock.tryLock()) {
@@ -195,23 +204,23 @@ public abstract class LeapArray<T> {
     /**
      * Get the previous bucket item before provided timestamp.
      *
-     * @param time a valid timestamp
+     * @param timeMillis a valid timestamp in milliseconds
      * @return the previous bucket item before provided timestamp
      */
-    public WindowWrap<T> getPreviousWindow(long time) {
+    public WindowWrap<T> getPreviousWindow(long timeMillis) {
-        if (time < 0) {
+        if (timeMillis < 0) {
             return null;
         }
-        long timeId = (time - windowLengthInMs) / windowLengthInMs;
+        int idx = calculateTimeIdx(timeMillis);
-        int idx = (int)(timeId % array.length());
+
-        time = time - windowLengthInMs;
+        long previousTime = timeMillis - windowLengthInMs;
         WindowWrap<T> wrap = array.get(idx);
 
         if (wrap == null || isWindowDeprecated(wrap)) {
             return null;
         }
 
-        if (wrap.windowStart() + windowLengthInMs < (time)) {
+        if (wrap.windowStart() + windowLengthInMs < previousTime) {
             return null;
         }
 
@@ -230,15 +239,14 @@ public abstract class LeapArray<T> {
     /**
      * Get statistic value from bucket for provided timestamp.
      *
-     * @param time a valid timestamp
+     * @param time a valid timestamp in milliseconds
      * @return the statistic value if bucket for provided timestamp is up-to-date; otherwise null
      */
     public T getWindowValue(long time) {
         if (time < 0) {
             return null;
         }
-        long timeId = time / windowLengthInMs;
+        int idx = calculateTimeIdx(time);
-        int idx = (int)(timeId % array.length());
 
         WindowWrap<T> old = array.get(idx);
         if (old == null || isWindowDeprecated(old)) {
@@ -255,7 +263,7 @@ public abstract class LeapArray<T> {
      * @param windowWrap a non-null bucket
      * @return true if the bucket is deprecated; otherwise false
      */
-    private boolean isWindowDeprecated(/*@NonNull*/ WindowWrap<T> windowWrap) {
+    protected boolean isWindowDeprecated(/*@NonNull*/ WindowWrap<T> windowWrap) {
         return TimeUtil.currentTimeMillis() - windowWrap.windowStart() >= intervalInMs;
     }
 
@@ -266,9 +274,10 @@ public abstract class LeapArray<T> {
      * @return valid bucket list for entire sliding window.
      */
     public List<WindowWrap<T>> list() {
-        List<WindowWrap<T>> result = new ArrayList<WindowWrap<T>>();
+        int size = array.length();
+        List<WindowWrap<T>> result = new ArrayList<WindowWrap<T>>(size);
 
-        for (int i = 0; i < array.length(); i++) {
+        for (int i = 0; i < size; i++) {
             WindowWrap<T> windowWrap = array.get(i);
             if (windowWrap == null || isWindowDeprecated(windowWrap)) {
                 continue;
@@ -286,9 +295,10 @@ public abstract class LeapArray<T> {
      * @return aggregated value list for entire sliding window
      */
     public List<T> values() {
-        List<T> result = new ArrayList<T>();
+        int size = array.length();
+        List<T> result = new ArrayList<T>(size);
 
-        for (int i = 0; i < array.length(); i++) {
+        for (int i = 0; i < size; i++) {
             WindowWrap<T> windowWrap = array.get(i);
             if (windowWrap == null || isWindowDeprecated(windowWrap)) {
                 continue;
@@ -298,6 +308,34 @@ public abstract class LeapArray<T> {
         return result;
     }
 
+    /**
+     * Get the valid "head" bucket of the sliding window for provided timestamp.
+     * Package-private for test.
+     *
+     * @param timeMillis a valid timestamp in milliseconds
+     * @return the "head" bucket if it exists and is valid; otherwise null
+     */
+    WindowWrap<T> getValidHead(long timeMillis) {
+        // Calculate index for expected head time.
+        int idx = calculateTimeIdx(timeMillis + windowLengthInMs);
+
+        WindowWrap<T> wrap = array.get(idx);
+        if (wrap == null || isWindowDeprecated(wrap)) {
+            return null;
+        }
+
+        return wrap;
+    }
+
+    /**
+     * Get the valid "head" bucket of the sliding window at current timestamp.
+     *
+     * @return the "head" bucket if it exists and is valid; otherwise null
+     */
+    public WindowWrap<T> getValidHead() {
+        return getValidHead(TimeUtil.currentTimeMillis());
+    }
+
     /**
      * Get sample count (total amount of buckets).
      *

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/data/MetricBucket.java

@@ -13,9 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.alibaba.csp.sentinel.slots.statistic.base;
+package com.alibaba.csp.sentinel.slots.statistic.data;
 
 import com.alibaba.csp.sentinel.Constants;
+import com.alibaba.csp.sentinel.slots.statistic.base.LongAdder;
 
 /**
  * Represents metrics data in a period of time span.

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/metric/ArrayMetric.java

@@ -20,7 +20,7 @@ import java.util.List;
 
 import com.alibaba.csp.sentinel.Constants;
 import com.alibaba.csp.sentinel.node.metric.MetricNode;
-import com.alibaba.csp.sentinel.slots.statistic.base.MetricBucket;
+import com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket;
 import com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap;
 
 /**

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/metric/Metric.java

@@ -18,7 +18,7 @@ package com.alibaba.csp.sentinel.slots.statistic.metric;
 import java.util.List;
 
 import com.alibaba.csp.sentinel.node.metric.MetricNode;
-import com.alibaba.csp.sentinel.slots.statistic.base.MetricBucket;
+import com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket;
 
 /**
  * Represents a basic structure recording invocation metrics of protected resources.

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/metric/MetricsLeapArray.java

@@ -16,7 +16,7 @@
 package com.alibaba.csp.sentinel.slots.statistic.metric;
 
 import com.alibaba.csp.sentinel.slots.statistic.base.LeapArray;
-import com.alibaba.csp.sentinel.slots.statistic.base.MetricBucket;
+import com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket;
 import com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap;
 
 /**

sentinel-core/src/test/java/com/alibaba/csp/sentinel/base/metric/ArrayMetricTest.java

@@ -19,7 +19,7 @@ import java.util.ArrayList;
 
 import org.junit.Test;
 
-import com.alibaba.csp.sentinel.slots.statistic.base.MetricBucket;
+import com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket;
 import com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap;
 import com.alibaba.csp.sentinel.slots.statistic.metric.ArrayMetric;
 import com.alibaba.csp.sentinel.slots.statistic.metric.MetricsLeapArray;

sentinel-core/src/test/java/com/alibaba/csp/sentinel/base/metric/MetricsLeapArrayTest.java

@@ -21,7 +21,7 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 
-import com.alibaba.csp.sentinel.slots.statistic.base.MetricBucket;
+import com.alibaba.csp.sentinel.slots.statistic.data.MetricBucket;
 import com.alibaba.csp.sentinel.util.TimeUtil;
 import com.alibaba.csp.sentinel.slots.statistic.base.WindowWrap;
 import com.alibaba.csp.sentinel.slots.statistic.metric.MetricsLeapArray;

sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/statistic/base/LeapArrayTest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 1999-2018 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alibaba.csp.sentinel.slots.statistic.base;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author Eric Zhao
+ */
+public class LeapArrayTest {
+
+    @Test
+    public void testGetValidHead() {
+        int windowLengthInMs = 100;
+        int intervalInSec = 1;
+        int sampleCount = intervalInSec * 1000 / windowLengthInMs;
+        LeapArray<AtomicInteger> leapArray = new LeapArray<AtomicInteger>(windowLengthInMs, intervalInSec) {
+            @Override
+            public AtomicInteger newEmptyBucket() {
+                return new AtomicInteger(0);
+            }
+
+            @Override
+            protected WindowWrap<AtomicInteger> resetWindowTo(WindowWrap<AtomicInteger> windowWrap, long startTime) {
+                windowWrap.resetTo(startTime);
+                windowWrap.value().set(0);
+                return windowWrap;
+            }
+        };
+        WindowWrap<AtomicInteger> expected1 = leapArray.currentWindow();
+        expected1.value().addAndGet(1);
+        sleep(windowLengthInMs);
+        WindowWrap<AtomicInteger> expected2 = leapArray.currentWindow();
+        expected2.value().addAndGet(2);
+        for (int i = 0; i < sampleCount - 2; i++) {
+            sleep(windowLengthInMs);
+            leapArray.currentWindow().value().addAndGet(i + 3);
+        }
+
+        assertSame(expected1, leapArray.getValidHead());
+        sleep(windowLengthInMs);
+        assertSame(expected2, leapArray.getValidHead());
+    }
+
+    private void sleep(int t) {
+        try {
+            Thread.sleep(t);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}