Issue ReactiveX#12 speed optimisations and cleanup

storozhukBM · storozhukBM · commit 8b07f08712b3 · 2016-12-02T19:15:15.000+02:00
diff --git a/src/jmh/java/io/github/robwin/circuitbreaker/CircuitBreakerBenchmark.java b/src/jmh/java/io/github/robwin/circuitbreaker/CircuitBreakerBenchmark.java
@@ -18,40 +18,55 @@
  */
 package io.github.robwin.circuitbreaker;
 
-//@State(Scope.Benchmark)
-//@OutputTimeUnit(TimeUnit.MILLISECONDS)
-//@BenchmarkMode(Mode.Throughput)
-//public class CircuitBreakerBenchmark {
-//
-//    private CircuitBreaker circuitBreaker;
-//    private Supplier<String> supplier;
-//    private static final int ITERATION_COUNT = 10;
-//    private static final int WARMUP_COUNT = 10;
-//    private static final int THREAD_COUNT = 10;
-//
-//    @Setup
-//    public void setUp() {
-//        CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.of(CircuitBreakerConfig.custom()
-//                .failureRateThreshold(1)
-//                .waitDurationInOpenState(Duration.ofSeconds(1))
-//                .build());
-//        circuitBreaker = circuitBreakerRegistry.circuitBreaker("testCircuitBreaker");
-//
-//        supplier = CircuitBreaker.decorateSupplier(() -> {
-//            try {
-//                Thread.sleep(100);
-//            } catch (InterruptedException e) {
-//                e.printStackTrace();
-//            }
-//            return "Hello Benchmark";
-//        }, circuitBreaker);
-//    }
-//
-//    @Benchmark
-//    @Threads(value = THREAD_COUNT)
-//    @Warmup(iterations = WARMUP_COUNT)
-//    @Measurement(iterations = ITERATION_COUNT)
-//    public String invokeSupplier(){
-//        return supplier.get();
-//    }
-//}
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@BenchmarkMode(Mode.Throughput)
+public class CircuitBreakerBenchmark {
+
+    private static final int ITERATION_COUNT = 10;
+    private static final int WARMUP_COUNT = 10;
+    private static final int THREAD_COUNT = 10;
+    private CircuitBreaker circuitBreaker;
+    private Supplier<String> supplier;
+
+    @Setup
+    public void setUp() {
+        CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.of(CircuitBreakerConfig.custom()
+            .failureRateThreshold(1)
+            .waitDurationInOpenState(Duration.ofSeconds(1))
+            .build());
+        circuitBreaker = circuitBreakerRegistry.circuitBreaker("testCircuitBreaker");
+
+        supplier = CircuitBreaker.decorateSupplier(() -> {
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+            return "Hello Benchmark";
+        }, circuitBreaker);
+    }
+
+    @Benchmark
+    @Threads(value = THREAD_COUNT)
+    @Warmup(iterations = WARMUP_COUNT)
+    @Measurement(iterations = ITERATION_COUNT)
+    public String invokeSupplier() {
+        return supplier.get();
+    }
+}
diff --git a/src/jmh/java/javaslang/circuitbreaker/RateLimiterBenchmark.java b/src/jmh/java/javaslang/circuitbreaker/RateLimiterBenchmark.java
@@ -23,18 +23,16 @@
 
 @State(Scope.Benchmark)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
-@BenchmarkMode(Mode.AverageTime)
+@BenchmarkMode(Mode.All)
 public class RateLimiterBenchmark {
 
     public static final int FORK_COUNT = 2;
     private static final int WARMUP_COUNT = 10;
-    private static final int ITERATION_COUNT = 5;
+    private static final int ITERATION_COUNT = 10;
     private static final int THREAD_COUNT = 2;
 
     private RateLimiter semaphoreBasedRateLimiter;
     private AtomicRateLimiter atomicRateLimiter;
-    private AtomicRateLimiter.State state;
-    private static final Object mutex = new Object();
 
     private Supplier<String> semaphoreGuardedSupplier;
     private Supplier<String> atomicGuardedSupplier;
@@ -48,7 +46,6 @@ public void setUp() {
             .build();
         semaphoreBasedRateLimiter = new SemaphoreBasedRateLimiter("semaphoreBased", rateLimiterConfig);
         atomicRateLimiter = new AtomicRateLimiter("atomicBased", rateLimiterConfig);
-        state = atomicRateLimiter.state.get();
 
         Supplier<String> stringSupplier = () -> {
             Blackhole.consumeCPU(1);
@@ -63,104 +60,16 @@ public void setUp() {
     @Warmup(iterations = WARMUP_COUNT)
     @Fork(value = FORK_COUNT)
     @Measurement(iterations = ITERATION_COUNT)
-    public void calculateNextState(Blackhole bh) {
-        AtomicRateLimiter.State next = atomicRateLimiter.calculateNextState(Duration.ZERO.toNanos(), this.state);
-        bh.consume(next);
+    public String semaphoreBasedPermission() {
+        return semaphoreGuardedSupplier.get();
     }
 
     @Benchmark
     @Threads(value = THREAD_COUNT)
     @Warmup(iterations = WARMUP_COUNT)
     @Fork(value = FORK_COUNT)
     @Measurement(iterations = ITERATION_COUNT)
-    public void nanosToWaitForPermission(Blackhole bh) {
-        long next = atomicRateLimiter.nanosToWaitForPermission(1, 315L, 31L);
-        bh.consume(next);
+    public String atomicPermission() {
+        return atomicGuardedSupplier.get();
     }
-
-    @Benchmark
-    @Threads(value = THREAD_COUNT)
-    @Warmup(iterations = WARMUP_COUNT)
-    @Fork(value = FORK_COUNT)
-    @Measurement(iterations = ITERATION_COUNT)
-    public void reservePermissions(Blackhole bh) {
-        AtomicRateLimiter.State next = atomicRateLimiter.reservePermissions(0L, 31L, 1, 0L);
-        bh.consume(next);
-    }
-
-    @Benchmark
-    @Threads(value = THREAD_COUNT)
-    @Warmup(iterations = WARMUP_COUNT)
-    @Fork(value = FORK_COUNT)
-    @Measurement(iterations = ITERATION_COUNT)
-    public void currentNanoTime(Blackhole bh) {
-        long next = atomicRateLimiter.currentNanoTime();
-        bh.consume(next);
-    }
-
-//    @Benchmark
-//    @Threads(value = THREAD_COUNT)
-//    @Warmup(iterations = WARMUP_COUNT)
-//    @Fork(value = FORK_COUNT)
-//    @Measurement(iterations = ITERATION_COUNT)
-//    public void mutex(Blackhole bh) {
-//        synchronized (mutex) {
-//            state = atomicRateLimiter.calculateNextState(Duration.ZERO.toNanos(), state);
-//        }
-//    }
-//
-//    @Benchmark
-//    @Threads(value = THREAD_COUNT)
-//    @Warmup(iterations = WARMUP_COUNT)
-//    @Fork(value = FORK_COUNT)
-//    @Measurement(iterations = ITERATION_COUNT)
-//    public void atomic(Blackhole bh) {
-//        atomicRateLimiter.state.updateAndGet(state -> {
-//            return atomicRateLimiter.calculateNextState(Duration.ZERO.toNanos(), state);
-//        });
-//    }
-//
-//    @Benchmark
-//    @Threads(value = THREAD_COUNT)
-//    @Warmup(iterations = WARMUP_COUNT)
-//    @Fork(value = FORK_COUNT)
-//    @Measurement(iterations = ITERATION_COUNT)
-//    public void atomicBackOf(Blackhole bh) {
-//        AtomicRateLimiter.State prev;
-//        AtomicRateLimiter.State next;
-//        do {
-//            prev = atomicRateLimiter.state.get();
-//            next = atomicRateLimiter.calculateNextState(Duration.ZERO.toNanos(), prev);
-//        } while (!compareAndSet(prev, next));
-//    }
-//
-//    /*
-//    https://arxiv.org/abs/1305.5800  https://dzone.com/articles/wanna-get-faster-wait-bit
-//     */
-//    public boolean compareAndSet(final AtomicRateLimiter.State current, final AtomicRateLimiter.State next) {
-//        if (atomicRateLimiter.state.compareAndSet(current, next)) {
-//            return true;
-//        } else {
-//            LockSupport.parkNanos(1);
-//            return false;
-//        }
-//    }
-//
-//    @Benchmark
-//    @Threads(value = THREAD_COUNT)
-//    @Warmup(iterations = WARMUP_COUNT)
-//    @Fork(value = FORK_COUNT)
-//    @Measurement(iterations = ITERATION_COUNT)
-//    public String semaphoreBasedPermission() {
-//        return semaphoreGuardedSupplier.get();
-//    }
-//
-//    @Benchmark
-//    @Threads(value = THREAD_COUNT)
-//    @Warmup(iterations = WARMUP_COUNT)
-//    @Fork(value = FORK_COUNT)
-//    @Measurement(iterations = ITERATION_COUNT)
-//    public String atomicPermission() {
-//        return atomicGuardedSupplier.get();
-//    }
 }
diff --git a/src/jmh/java/javaslang/circuitbreaker/stateCalculationBenchmark.txt b/src/jmh/java/javaslang/circuitbreaker/stateCalculationBenchmark.txt
diff --git a/src/main/java/javaslang/ratelimiter/internal/AtomicRateLimiter.java b/src/main/java/javaslang/ratelimiter/internal/AtomicRateLimiter.java
@@ -11,6 +11,7 @@
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.UnaryOperator;
 
 /**
  * {@link AtomicRateLimiter} splits all nanoseconds from the start of epoch into cycles.
@@ -31,7 +32,7 @@ public class AtomicRateLimiter implements RateLimiter {
     private final long cyclePeriodInNanos;
     private final int permissionsPerCycle;
     private final AtomicInteger waitingThreads;
-    public final AtomicReference<State> state;
+    private final AtomicReference<State> state;
 
 
     public AtomicRateLimiter(String name, RateLimiterConfig rateLimiterConfig) {
@@ -51,12 +52,57 @@ public AtomicRateLimiter(String name, RateLimiterConfig rateLimiterConfig) {
     @Override
     public boolean getPermission(final Duration timeoutDuration) {
         long timeoutInNanos = timeoutDuration.toNanos();
-        State modifiedState = state.updateAndGet(
-            activeState -> calculateNextState(timeoutInNanos, activeState)
-        );
+        State modifiedState = updateStateWithBackOff(timeoutInNanos);
         return waitForPermissionIfNecessary(timeoutInNanos, modifiedState.nanosToWait);
     }
 
+    /**
+     * Atomically updates the current {@link State} with the results of
+     * applying the {@link AtomicRateLimiter#calculateNextState}, returning the updated {@link State}.
+     * It differs from {@link AtomicReference#updateAndGet(UnaryOperator)} by constant back off.
+     * It means that after one try to {@link AtomicReference#compareAndSet(Object, Object)}
+     * this method will wait for a while before try one more time.
+     * This technique was originally described in this
+     * <a href="https://arxiv.org/abs/1305.5800"> paper</a>
+     * and showed great results with {@link AtomicRateLimiter} in benchmark tests.
+     *
+     * @param timeoutInNanos a side-effect-free function
+     * @return the updated value
+     */
+    private State updateStateWithBackOff(final long timeoutInNanos) {
+        AtomicRateLimiter.State prev;
+        AtomicRateLimiter.State next;
+        do {
+            prev = state.get();
+            next = calculateNextState(timeoutInNanos, prev);
+        } while (!compareAndSet(prev, next));
+        return next;
+    }
+
+    /**
+     * Atomically sets the value to the given updated value
+     * if the current value {@code ==} the expected value.
+     * It differs from {@link AtomicReference#updateAndGet(UnaryOperator)} by constant back off.
+     * It means that after one try to {@link AtomicReference#compareAndSet(Object, Object)}
+     * this method will wait for a while before try one more time.
+     * This technique was originally described in this
+     * <a href="https://arxiv.org/abs/1305.5800"> paper</a>
+     * and showed great results with {@link AtomicRateLimiter} in benchmark tests.
+     *
+     * @param current the expected value
+     * @param next    the new value
+     * @return {@code true} if successful. False return indicates that
+     * the actual value was not equal to the expected value.
+     */
+    private boolean compareAndSet(final State current, final State next) {
+        if (state.compareAndSet(current, next)) {
+            return true;
+        } else {
+            parkNanos(1); // back-off
+            return false;
+        }
+    }
+
     /**
      * A side-effect-free function that can calculate next {@link State} from current.
      * It determines time duration that you should wait for permission and reserves it for you,
@@ -66,7 +112,7 @@ public boolean getPermission(final Duration timeoutDuration) {
      * @param activeState    current state of {@link AtomicRateLimiter}
      * @return next {@link State}
      */
-    public State calculateNextState(final long timeoutInNanos, final State activeState) {
+    private State calculateNextState(final long timeoutInNanos, final State activeState) {
         long currentNanos = currentNanoTime();
         long currentCycle = currentNanos / cyclePeriodInNanos;
 
@@ -93,7 +139,7 @@ public State calculateNextState(final long timeoutInNanos, final State activeSta
      * @param currentCycle         current {@link AtomicRateLimiter} cycle
      * @return nanoseconds to wait for the next permission
      */
-    public long nanosToWaitForPermission(final int availablePermissions, final long currentNanos, final long currentCycle) {
+    private long nanosToWaitForPermission(final int availablePermissions, final long currentNanos, final long currentCycle) {
         if (availablePermissions > 0) {
             return 0L;
         } else {
@@ -114,7 +160,7 @@ public long nanosToWaitForPermission(final int availablePermissions, final long
      * @param nanosToWait    nanoseconds to wait for the next permission
      * @return new {@link State} with possibly reserved permissions and time to wait
      */
-    public State reservePermissions(final long timeoutInNanos, final long cycle, final int permissions, final long nanosToWait) {
+    private State reservePermissions(final long timeoutInNanos, final long cycle, final int permissions, final long nanosToWait) {
         boolean canAcquireInTime = timeoutInNanos >= nanosToWait;
         int permissionsWithReservation = permissions;
         if (canAcquireInTime) {
@@ -130,7 +176,7 @@ public State reservePermissions(final long timeoutInNanos, final long cycle, fin
      * @param nanosToWait    nanoseconds caller need to wait
      * @return true if caller was able to wait for nanosToWait without {@link Thread#interrupt} and not exceed timeout
      */
-    public boolean waitForPermissionIfNecessary(final long timeoutInNanos, final long nanosToWait) {
+    private boolean waitForPermissionIfNecessary(final long timeoutInNanos, final long nanosToWait) {
         boolean canAcquireImmediately = nanosToWait <= 0;
         boolean canAcquireInTime = timeoutInNanos >= nanosToWait;
 
@@ -153,7 +199,7 @@ public boolean waitForPermissionIfNecessary(final long timeoutInNanos, final lon
      * @param nanosToWait nanoseconds caller need to wait
      * @return true if caller was not {@link Thread#interrupted} while waiting
      */
-    public boolean waitForPermission(final long nanosToWait) {
+    private boolean waitForPermission(final long nanosToWait) {
         waitingThreads.incrementAndGet();
         long deadline = currentNanoTime() + nanosToWait;
         boolean wasInterrupted = false;
@@ -207,13 +253,14 @@ public AtomicRateLimiterMetrics getMetrics() {
      * the last {@link AtomicRateLimiter#getPermission(Duration)} call.</li>
      * </ul>
      */
-    public static class State {
+    private static class State {
 
         private final long activeCycle;
+
         private final int activePermissions;
         private final long nanosToWait;
 
-        public State(final long activeCycle, final int activePermissions, final long nanosToWait) {
+        private State(final long activeCycle, final int activePermissions, final long nanosToWait) {
             this.activeCycle = activeCycle;
             this.activePermissions = activePermissions;
             this.nanosToWait = nanosToWait;
@@ -259,12 +306,13 @@ public long getAvailablePermissions() {
             State estimatedState = calculateNextState(-1, currentState);
             return estimatedState.activePermissions;
         }
+
     }
 
     /**
      * Created only for test purposes. Simply calls {@link System#nanoTime()}
      */
-    public long currentNanoTime() {
+    private long currentNanoTime() {
         return nanoTime();
     }
 }
