migrates to RequestInterceptor to track Stats

Signed-off-by: Oleh Dokuka <shadowgun@i.ua>

Author: OlegDokuka

benchmarks/README.md

@@ -17,7 +17,7 @@ Specify extra profilers:
 Prominent profilers (for full list call `jmhProfilers` task):
 - comp - JitCompilations, tune your iterations
 - stack - which methods used most time
-- gc - print garbage collection stats
+- gc - print garbage collection defaultWeightedStats
 - hs_thr - thread usage
 
 Change report format from JSON to one of [CSV, JSON, NONE, SCSV, TEXT]:

rsocket-core/src/main/java/io/rsocket/loadbalance/BaseWeightedStats.java

@@ -0,0 +1,221 @@
+package io.rsocket.loadbalance;
+
+import io.rsocket.util.Clock;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+/**
+ * The base implementation of the {@link WeightedStats} interface
+ *
+ * @since 1.1
+ */
+public class BaseWeightedStats implements WeightedStats {
+
+  private static final double DEFAULT_LOWER_QUANTILE = 0.5;
+  private static final double DEFAULT_HIGHER_QUANTILE = 0.8;
+  private static final int INACTIVITY_FACTOR = 500;
+  private static final long DEFAULT_INITIAL_INTER_ARRIVAL_TIME =
+      Clock.unit().convert(1L, TimeUnit.SECONDS);
+
+  private static final double STARTUP_PENALTY = Long.MAX_VALUE >> 12;
+
+  private final Quantile lowerQuantile;
+  private final Quantile higherQuantile;
+  private final Ewma availabilityPercentage;
+  private final Median median;
+  private final Ewma interArrivalTime;
+
+  private final long tau;
+  private final long inactivityFactor;
+
+  private long errorStamp; // last we got an error
+  private long stamp; // last timestamp we sent a request
+  private long stamp0; // last timestamp we sent a request or receive a response
+  private long duration; // instantaneous cumulative duration
+
+  private double availability = 1.0;
+
+  private volatile int pendingRequests; // instantaneous rate
+  private static final AtomicIntegerFieldUpdater<BaseWeightedStats> PENDING_REQUESTS =
+      AtomicIntegerFieldUpdater.newUpdater(BaseWeightedStats.class, "pendingRequests");
+  private volatile int pendingStreams; // number of active streams
+  private static final AtomicIntegerFieldUpdater<BaseWeightedStats> PENDING_STREAMS =
+      AtomicIntegerFieldUpdater.newUpdater(BaseWeightedStats.class, "pendingStreams");
+
+  protected BaseWeightedStats() {
+    this(
+        new FrugalQuantile(DEFAULT_LOWER_QUANTILE),
+        new FrugalQuantile(DEFAULT_HIGHER_QUANTILE),
+        INACTIVITY_FACTOR);
+  }
+
+  private BaseWeightedStats(
+      Quantile lowerQuantile, Quantile higherQuantile, long inactivityFactor) {
+    this.lowerQuantile = lowerQuantile;
+    this.higherQuantile = higherQuantile;
+    this.inactivityFactor = inactivityFactor;
+
+    long now = Clock.now();
+    this.stamp = now;
+    this.errorStamp = now;
+    this.stamp0 = now;
+    this.duration = 0L;
+    this.pendingRequests = 0;
+    this.median = new Median();
+    this.interArrivalTime = new Ewma(1, TimeUnit.MINUTES, DEFAULT_INITIAL_INTER_ARRIVAL_TIME);
+    this.availabilityPercentage = new Ewma(5, TimeUnit.SECONDS, 1.0);
+    this.tau = Clock.unit().convert((long) (5 / Math.log(2)), TimeUnit.SECONDS);
+  }
+
+  @Override
+  public double lowerQuantileLatency() {
+    return lowerQuantile.estimation();
+  }
+
+  @Override
+  public double higherQuantileLatency() {
+    return higherQuantile.estimation();
+  }
+
+  @Override
+  public int pending() {
+    return pendingRequests + pendingStreams;
+  }
+
+  @Override
+  public double availability() {
+    if (Clock.now() - stamp > tau) {
+      updateAvailability(1.0);
+    }
+    return availability * availabilityPercentage.value();
+  }
+
+  @Override
+  public double predictedLatency() {
+    final long now = Clock.now();
+    final long elapsed;
+
+    synchronized (this) {
+      elapsed = Math.max(now - stamp, 1L);
+    }
+
+    final double latency;
+    final double prediction = median.estimation();
+
+    final int pending = this.pending();
+    if (prediction == 0.0) {
+      if (pending == 0) {
+        latency = 0.0; // first request
+      } else {
+        // subsequent requests while we don't have any history
+        latency = STARTUP_PENALTY + pending;
+      }
+    } else if (pending == 0 && elapsed > inactivityFactor * interArrivalTime.value()) {
+      // if we did't see any data for a while, we decay the prediction by inserting
+      // artificial 0.0 into the median
+      median.insert(0.0);
+      latency = median.estimation();
+    } else {
+      final double predicted = prediction * pending;
+      final double instant = instantaneous(now, pending);
+
+      if (predicted < instant) { // NB: (0.0 < 0.0) == false
+        latency = instant / pending; // NB: pending never equal 0 here
+      } else {
+        // we are under the predictions
+        latency = prediction;
+      }
+    }
+
+    return latency;
+  }
+
+  long instantaneous(long now, int pending) {
+    return duration + (now - stamp0) * pending;
+  }
+
+  void startStream() {
+    PENDING_STREAMS.incrementAndGet(this);
+  }
+
+  void stopStream() {
+    PENDING_STREAMS.decrementAndGet(this);
+  }
+
+  synchronized long startRequest() {
+    final long now = Clock.now();
+    final int pendingRequests = this.pendingRequests;
+
+    interArrivalTime.insert(now - stamp);
+    duration += Math.max(0, now - stamp0) * pendingRequests;
+    PENDING_REQUESTS.lazySet(this, pendingRequests + 1);
+    stamp = now;
+    stamp0 = now;
+
+    return now;
+  }
+
+  synchronized long stopRequest(long timestamp) {
+    final long now = Clock.now();
+    final int pendingRequests = this.pendingRequests;
+
+    duration += Math.max(0, now - stamp0) * pendingRequests - (now - timestamp);
+    PENDING_REQUESTS.lazySet(this, pendingRequests - 1);
+    stamp0 = now;
+
+    return now;
+  }
+
+  synchronized void record(double roundTripTime) {
+    median.insert(roundTripTime);
+    lowerQuantile.insert(roundTripTime);
+    higherQuantile.insert(roundTripTime);
+  }
+
+  void updateAvailability(double value) {
+    availabilityPercentage.insert(value);
+    if (value == 0.0d) {
+      synchronized (this) {
+        errorStamp = Clock.now();
+      }
+    }
+  }
+
+  void setAvailability(double availability) {
+    this.availability = availability;
+  }
+
+  @Override
+  public String toString() {
+    return "Stats{"
+        + "lowerQuantile="
+        + lowerQuantile.estimation()
+        + ", higherQuantile="
+        + higherQuantile.estimation()
+        + ", inactivityFactor="
+        + inactivityFactor
+        + ", tau="
+        + tau
+        + ", errorPercentage="
+        + availabilityPercentage.value()
+        + ", pending="
+        + pendingRequests
+        + ", errorStamp="
+        + errorStamp
+        + ", stamp="
+        + stamp
+        + ", stamp0="
+        + stamp0
+        + ", duration="
+        + duration
+        + ", median="
+        + median.estimation()
+        + ", interArrivalTime="
+        + interArrivalTime.value()
+        + ", pendingStreams="
+        + pendingStreams
+        + ", availability="
+        + availability
+        + '}';
+  }
+}

rsocket-core/src/main/java/io/rsocket/loadbalance/ClientLoadbalanceStrategy.java

@@ -0,0 +1,14 @@
+package io.rsocket.loadbalance;
+
+import io.rsocket.core.RSocketConnector;
+
+/**
+ * Extension for {@link LoadbalanceStrategy} which allows pre-setup {@link RSocketConnector} for
+ * {@link LoadbalanceStrategy} needs
+ *
+ * @since 1.1
+ */
+public interface ClientLoadbalanceStrategy extends LoadbalanceStrategy {
+
+  void initialize(RSocketConnector connector);
+}

rsocket-core/src/main/java/io/rsocket/loadbalance/Ewma.java

@@ -18,6 +18,7 @@
 
 import io.rsocket.util.Clock;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 /**
  * Compute the exponential weighted moving average of a series of values. The time at which you
@@ -28,20 +29,27 @@
  * equal to (200 - 100)/2 = 150 (half of the distance between the new and the old value)
  */
 class Ewma {
-  private final long tau;
-  private volatile long stamp;
-  private volatile double ewma;
+
+  final long tau;
+
+  volatile long stamp;
+  static final AtomicLongFieldUpdater<Ewma> STAMP =
+      AtomicLongFieldUpdater.newUpdater(Ewma.class, "stamp");
+  volatile double ewma;
 
   public Ewma(long halfLife, TimeUnit unit, double initialValue) {
     this.tau = Clock.unit().convert((long) (halfLife / Math.log(2)), unit);
-    stamp = 0L;
-    ewma = initialValue;
+
+    this.ewma = initialValue;
+
+    STAMP.lazySet(this, 0L);
   }
 
   public synchronized void insert(double x) {
-    long now = Clock.now();
-    double elapsed = Math.max(0, now - stamp);
-    stamp = now;
+    final long now = Clock.now();
+    final double elapsed = Math.max(0, now - stamp);
+
+    STAMP.lazySet(this, now);
 
     double w = Math.exp(-elapsed / tau);
     ewma = w * ewma + (1.0 - w) * x;

rsocket-core/src/main/java/io/rsocket/loadbalance/FluxDeferredResolution.java

@@ -83,7 +83,7 @@ public final Context currentContext() {
 
   @Nullable
   @Override
-  public Object scanUnsafe(Attr key) {
+  public final Object scanUnsafe(Attr key) {
     long state = this.requested;
 
     if (key == Attr.PARENT) {
@@ -145,7 +145,7 @@ public final void onNext(Payload payload) {
   }
 
   @Override
-  public void onError(Throwable t) {
+  public final void onError(Throwable t) {
     if (this.done) {
       Operators.onErrorDropped(t, this.actual.currentContext());
       return;
@@ -156,7 +156,7 @@ public void onError(Throwable t) {
   }
 
   @Override
-  public void onComplete() {
+  public final void onComplete() {
     if (this.done) {
       return;
     }
@@ -206,7 +206,7 @@ public final void request(long n) {
     }
   }
 
-  public void cancel() {
+  public final void cancel() {
     long state = REQUESTED.getAndSet(this, STATE_TERMINATED);
     if (state == STATE_TERMINATED) {
       return;

rsocket-core/src/main/java/io/rsocket/loadbalance/FrugalQuantile.java

@@ -26,12 +26,14 @@
  * <p>More info: http://blog.aggregateknowledge.com/2013/09/16/sketch-of-the-day-frugal-streaming/
  */
 class FrugalQuantile implements Quantile {
-  private final double increment;
-  volatile double estimate;
+  final double increment;
+  final SplittableRandom rnd;
+
   int step;
   int sign;
-  private double quantile;
-  private SplittableRandom rnd;
+  double quantile;
+
+  volatile double estimate;
 
   public FrugalQuantile(double quantile, double increment) {
     this.increment = increment;