FFM-6392 - Java SDK - Investigate thread contention MetricsProcessor.pushToQueue (#133)

What
- Remove synchronized keyword from pushToQueue (also renamed to registerEvaluation)
- Remove synchronized keyword from runOneIteration, instead a new method was added to drain frequencyMap atomically which means we no longer need synchronized here
- Remove debug logs causing excessive string concat operations in pushToQueue as reported by profiler
- New stress test added to place load on this method with multiple threads, for easier profiling

Why
Platform team are reporting contention around MetricsProcessor.pushToQueue with JFR when using a large number of targets and flags.

Testing
Tested manually with new junit stress test added for MatricsProcessor with list of targets/flags used by platform.

Author: andybharness

src/main/java/io/harness/cf/client/api/InnerClient.java

@@ -374,7 +374,7 @@ public JsonObject jsonVariation(
   public void processEvaluation(
       @NonNull FeatureConfig featureConfig, Target target, @NonNull Variation variation) {
     if (this.options.isAnalyticsEnabled()) {
-      metricsProcessor.pushToQueue(target, featureConfig.getFeature(), variation);
+      metricsProcessor.registerEvaluation(target, featureConfig.getFeature(), variation);
     }
   }
 

src/main/java/io/harness/cf/client/api/MetricsProcessor.java

@@ -16,6 +16,46 @@
 @Slf4j
 class MetricsProcessor extends AbstractScheduledService {
 
+  static class FrequencyMap<K> {
+
+    private final AtomicLongMap<K> map;
+
+    FrequencyMap() {
+      map = AtomicLongMap.create();
+    }
+
+    void increment(K key) {
+      map.incrementAndGet(key);
+    }
+
+    long get(K key) {
+      return map.get(key);
+    }
+
+    int size() {
+      return map.size();
+    }
+
+    long sum() {
+      return map.sum();
+    }
+
+    Map<K, Long> drainToMap() {
+      // Guava doesn't have a function to atomically drain an AtomicLongMap.
+      // Here we need to atomically set each key to zero as we transfer it to the new map else we
+      // see missed evaluations
+      final HashMap<K, Long> result = new HashMap<>();
+      map.asMap()
+          .forEach(
+              (k, v) -> {
+                final long oldVal = map.getAndUpdate(k, old -> 0);
+                result.put(k, oldVal);
+              });
+      map.removeAllZeros();
+      return result;
+    }
+  }
+
   private static final String FEATURE_NAME_ATTRIBUTE = "featureName";
   private static final String VARIATION_IDENTIFIER_ATTRIBUTE = "variationIdentifier";
   private static final String TARGET_ATTRIBUTE = "target";
@@ -36,7 +76,7 @@ class MetricsProcessor extends AbstractScheduledService {
 
   private final Connector connector;
   private final BaseConfig config;
-  private final AtomicLongMap<MetricEvent> frequencyMap;
+  private final FrequencyMap<MetricEvent> frequencyMap;
   private final Set<Target> uniqueTargetSet;
   private final ScheduledExecutorService executor;
 
@@ -50,26 +90,32 @@ public MetricsProcessor(
     this.config = config;
     this.executor = executor();
 
-    this.frequencyMap = AtomicLongMap.create();
+    this.frequencyMap = new FrequencyMap<>();
     this.uniqueTargetSet = ConcurrentHashMap.newKeySet();
 
     callback.onMetricsReady();
   }
 
-  // push the incoming data to the queue
-  public synchronized void pushToQueue(Target target, String featureName, Variation variation) {
+  @Deprecated /* The name of this method no longer makes sense since we moved to a map, kept for source compatibility */
+  public void pushToQueue(Target target, String featureName, Variation variation) {
+    registerEvaluation(target, featureName, variation);
+  }
 
-    if (frequencyMap.size() > config.getBufferSize()) {
-      log.warn(
-          "Metric frequency map exceeded buffer size ({} > {}), force flushing",
-          frequencyMap.size(),
-          config.getBufferSize());
+  void registerEvaluation(Target target, String featureName, Variation variation) {
+
+    final int freqMapSize = frequencyMap.size();
+
+    if (freqMapSize > config.getBufferSize()) {
+      if (log.isWarnEnabled()) {
+        log.warn(
+            "Metric frequency map exceeded buffer size ({} > {}), force flushing",
+            freqMapSize,
+            config.getBufferSize());
+      }
       // If the map is starting to grow too much then push the events now and reset the counters
       executor.submit(this::runOneIteration);
     }
 
-    log.debug("Flag: " + featureName + " Target: " + target + " Variation: " + variation);
-
     Target metricTarget = globalTarget;
 
     if (target != null) {
@@ -79,7 +125,7 @@ public synchronized void pushToQueue(Target target, String featureName, Variatio
       }
     }
 
-    frequencyMap.incrementAndGet(new MetricEvent(featureName, metricTarget, variation));
+    frequencyMap.increment(new MetricEvent(featureName, metricTarget, variation));
   }
 
   /** This method sends the metrics data to the analytics server and resets the cache */
@@ -202,17 +248,15 @@ private String getVersion() {
   }
 
   @Override
-  protected synchronized void runOneIteration() {
+  protected void runOneIteration() {
     if (log.isDebugEnabled()) {
       log.debug(
-          "Drain metrics queue : frequencyMap size={} uniqueTargetSet size={} metric events pending={}",
+          "Drain metrics queue : frequencyMap size={} uniqueTargetSet size={}",
           frequencyMap.size(),
-          uniqueTargetSet.size(),
-          frequencyMap.sum());
+          uniqueTargetSet.size());
     }
-    sendDataAndResetCache(new HashMap<>(frequencyMap.asMap()), new HashSet<>(uniqueTargetSet));
+    sendDataAndResetCache(frequencyMap.drainToMap(), new HashSet<>(uniqueTargetSet));
 
-    frequencyMap.clear();
     uniqueTargetSet.clear();
   }
 
@@ -238,6 +282,8 @@ public void close() {
     log.info("Closing MetricsProcessor");
   }
 
+  /* package private */
+
   synchronized void flushQueue() {
     executor.submit(this::runOneIteration);
   }

src/test/java/io/harness/cf/client/api/CfClientTest.java

@@ -10,19 +10,12 @@
 import com.google.gson.JsonObject;
 import io.harness.cf.client.api.dispatchers.TestWebServerDispatcher;
 import io.harness.cf.client.api.dispatchers.UnimplementedStreamDispatcher;
+import io.harness.cf.client.api.testutils.DummyConnector;
 import io.harness.cf.client.common.Cache;
-import io.harness.cf.client.connector.Connector;
-import io.harness.cf.client.connector.ConnectorException;
-import io.harness.cf.client.connector.Service;
-import io.harness.cf.client.connector.Updater;
 import io.harness.cf.client.dto.Target;
-import io.harness.cf.model.FeatureConfig;
-import io.harness.cf.model.Metrics;
-import io.harness.cf.model.Segment;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.time.Duration;
-import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -367,46 +360,4 @@ public List<String> keys() {
       return null;
     }
   }
-
-  static class DummyConnector implements Connector {
-
-    @Override
-    public String authenticate() throws ConnectorException {
-      return "dummy";
-    }
-
-    @Override
-    public void setOnUnauthorized(Runnable runnable) {}
-
-    @Override
-    public List<FeatureConfig> getFlags() throws ConnectorException {
-      return Collections.emptyList();
-    }
-
-    @Override
-    public FeatureConfig getFlag(@NonNull String identifier) throws ConnectorException {
-      return null;
-    }
-
-    @Override
-    public List<Segment> getSegments() throws ConnectorException {
-      return Collections.emptyList();
-    }
-
-    @Override
-    public Segment getSegment(@NonNull String identifier) throws ConnectorException {
-      return null;
-    }
-
-    @Override
-    public void postMetrics(Metrics metrics) throws ConnectorException {}
-
-    @Override
-    public Service stream(Updater updater) throws ConnectorException {
-      return null;
-    }
-
-    @Override
-    public void close() {}
-  }
 }

src/test/java/io/harness/cf/client/api/MetricsProcessorStressTest.java

@@ -0,0 +1,166 @@
+package io.harness.cf.client.api;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import io.harness.cf.client.api.testutils.DummyConnector;
+import io.harness.cf.client.dto.Target;
+import io.harness.cf.model.FeatureConfig;
+import io.harness.cf.model.Variation;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.stream.Stream;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.NonNull;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+/*
+ * This stress test is disabled by default and needs to be run manually. Make sure to set TARGETS_FILE and
+ * FLAGS_FILE from the production files in ff-sdk-testgrid or create files with at least 18K unique targets
+ * and 400 unique flags if you want to run this test.
+ *
+ * If running in IntelliJ you can attach JFR from a terminal after starting the test using something like:
+ *
+ *  jcmd  $(jcmd | grep -i junit | cut -f 1 -d " ")  JFR.start duration=10m filename=flight.jfr
+ */
+@Disabled
+class MetricsProcessorStressTest {
+
+  boolean RUN_PERPETUALLY = false; // useful for longer tests, note if true, test will never exit
+  boolean DUMP_POSTED_METRICS = false;
+  int NUM_THREADS = 32;
+  int VARIATION_COUNT = 4;
+  String TARGETS_FILE = "/tmp/prod2_targets.txt";
+  String FLAGS_FILE = "/tmp/flags.txt";
+
+  @Test
+  void testRegisterEvaluationContention() throws Exception {
+
+    final DummyConnector dummyConnector = new DummyConnector(DUMP_POSTED_METRICS);
+
+    final MetricsProcessor metricsProcessor =
+        new MetricsProcessor(
+            dummyConnector,
+            BaseConfig.builder()
+                // .globalTargetEnabled(false)
+                .build(),
+            new DummyMetricsCallback());
+
+    metricsProcessor.start();
+
+    System.out.println("Loading...");
+
+    final List<String> targets = loadFile(TARGETS_FILE);
+    final List<String> flags = loadFile(FLAGS_FILE);
+
+    System.out.printf("Loaded %d targets\n", targets.size());
+    System.out.printf("Loaded %d flags\n", flags.size());
+
+    final ConcurrentLinkedQueue<TargetAndFlag> targetAndFlags =
+        createFlagTargetVariationPermutations(flags, targets);
+
+    System.out.printf("Starting...processing %d flags/targets\n", targetAndFlags.size());
+
+    final ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
+    final LongAdder totalProcessed = new LongAdder();
+
+    for (int i = 0; i < NUM_THREADS; i++) {
+      final int threadNum = i;
+      executor.submit(
+          () -> {
+            Thread.currentThread().setName("THREAD" + threadNum);
+            System.out.println("start thread " + Thread.currentThread().getName());
+
+            TargetAndFlag next;
+
+            while ((next = targetAndFlags.poll()) != null) {
+              final Target target = Target.builder().identifier(next.target).build();
+              final FeatureConfig feature = FeatureConfig.builder().feature(next.flag).build();
+              final Variation variation =
+                  Variation.builder()
+                      .identifier(next.variation)
+                      .value(next.variation + "Value")
+                      .build();
+
+              metricsProcessor.registerEvaluation(target, feature.getFeature(), variation);
+
+              if (RUN_PERPETUALLY) {
+                targetAndFlags.add(next);
+              }
+
+              totalProcessed.increment();
+            }
+
+            System.out.printf("thread %s finished\n", Thread.currentThread().getName());
+          });
+    }
+
+    while (targetAndFlags.size() > 0) {
+      System.out.printf(
+          "target/flags/variations processed %d, map size %d, pending evaluations=%d \n",
+          totalProcessed.sum(),
+          metricsProcessor.getQueueSize(),
+          metricsProcessor.getPendingMetricsToBeSent());
+
+      Thread.sleep(1000);
+    }
+
+    metricsProcessor.runOneIteration();
+
+    assertEquals(flags.size() * targets.size() * VARIATION_COUNT, (int) totalProcessed.sum());
+    assertEquals(
+        flags.size() * targets.size() * VARIATION_COUNT,
+        dummyConnector.getTotalMetricEvaluations());
+  }
+
+  private List<String> loadFile(String filename) throws IOException {
+    final List<String> map = new ArrayList<>();
+    try (Stream<String> stream = Files.lines(Paths.get(filename))) {
+      stream.forEach(map::add);
+    }
+    return map;
+  }
+
+  private ConcurrentLinkedQueue<TargetAndFlag> createFlagTargetVariationPermutations(
+      List<String> flags, List<String> targets) {
+    final ConcurrentLinkedQueue<TargetAndFlag> targetAndFlags = new ConcurrentLinkedQueue<>();
+
+    for (String flag : flags) {
+      for (String target : targets) {
+        for (int v = 0; v < VARIATION_COUNT; v++) { // variations per flag/target combination
+          targetAndFlags.add(new TargetAndFlag(target, flag, "variation" + v));
+        }
+      }
+    }
+    return targetAndFlags;
+  }
+
+  @EqualsAndHashCode
+  @AllArgsConstructor
+  static class TargetAndFlag {
+    String target, flag, variation;
+  }
+
+  static class DummyMetricsCallback implements MetricsCallback {
+    @Override
+    public void onMetricsReady() {
+      System.out.println("onMetricsReady");
+    }
+
+    @Override
+    public void onMetricsError(@NonNull String error) {
+      System.out.println("onMetricsError " + error);
+    }
+
+    @Override
+    public void onMetricsFailure() {
+      System.out.println("onMetricsFailure");
+    }
+  }
+}