Async map improvements (#2212)

Async map improvements

java/drivers/driver-hazelcast4plus/src/main/java/com/hazelcast/simulator/hz/map/AsyncLongStringMapTest.java

@@ -26,15 +26,33 @@
 import com.hazelcast.simulator.test.annotations.Teardown;
 import com.hazelcast.simulator.worker.loadsupport.Streamer;
 import com.hazelcast.simulator.worker.loadsupport.StreamerFactory;
+import com.hazelcast.simulator.worker.testcontainer.Probability;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CountDownLatch;
 
 import static com.hazelcast.simulator.utils.GeneratorUtils.generateAsciiStrings;
 
+/**
+ * An IMap test that uses async calls to generate load which is much more efficient
+ * than using sync calls. So instead of having a set of threads calling blocking methods,
+ * just a few threads are needed that keep a level of concurrent request in flight.
+ * <p/>
+ * The big advantage is that a client can generate much higher levels of load without
+ * becoming the bottleneck itself.
+ * <p/>
+ * This is a POC implementation. In the future code generation should be used to do the
+ * actual work.
+ */
 public class AsyncLongStringMapTest extends HazelcastTest {
 
+    private final static byte METHOD_GET = 0;
+    private final static byte METHOD_SET = 1;
+    private final static byte METHOD_PUT = 2;
+
     // properties
     public int concurrency = 100;
     public long keyDomain = 10000;
@@ -43,13 +61,25 @@ public class AsyncLongStringMapTest extends HazelcastTest {
     public int maxValueLength = 10;
     public boolean fillOnPrepare = true;
     public boolean destroyOnExit = true;
+    public double getProb = 1;
+    public double putProb = 0;
+    public double setProb = 0;
+
     private IMap<Long, String> map;
     private String[] values;
+    private byte[] probabilityArray;
 
     @Setup
     public void setUp() {
         map = targetInstance.getMap(name);
         values = generateAsciiStrings(valueCount, minValueLength, maxValueLength);
+
+        List<Probability> probabilityList = new ArrayList<>();
+        probabilityList.add(new Probability(getProb));
+        probabilityList.add(new Probability(putProb));
+        probabilityList.add(new Probability(setProb));
+
+        probabilityArray = Probability.loadTimeStepProbabilityArray(probabilityList);
     }
 
     @Prepare(global = true)
@@ -80,9 +110,13 @@ public void run() throws InterruptedException {
     private class Task implements Runnable {
         private final ThreadState state = new ThreadState();
         private final LatencyProbe getLatencyProbe = testContext.getLatencyProbe("get");
+        private final LatencyProbe putLatencyProbe = testContext.getLatencyProbe("put");
+        private final LatencyProbe setLatencyProbe = testContext.getLatencyProbe("set");
         private final CountDownLatch completionLatch;
         private long startNs;
         private boolean initializing = true;
+        private final ThreadState threadState = new ThreadState();
+        private int method;
 
         private Task(CountDownLatch completionLatch) {
             this.completionLatch = completionLatch;
@@ -95,15 +129,50 @@ public void run() {
                 startNs = System.nanoTime();
             } else {
                 long nowNs = System.nanoTime();
-                getLatencyProbe.recordValue(nowNs - startNs);
+                long durationNs = nowNs - startNs;
+                switch (method) {
+                    case METHOD_GET:
+                        getLatencyProbe.recordValue(durationNs);
+                        break;
+                    case METHOD_SET:
+                        setLatencyProbe.recordValue(durationNs);
+                        break;
+                    case METHOD_PUT:
+                        putLatencyProbe.recordValue(durationNs);
+                        break;
+                    default:
+                        throw new RuntimeException("unexpected method index: " + method);
+                }
+
                 startNs = nowNs;
             }
 
             if (testContext.isStopped()) {
                 completionLatch.countDown();
-            } else {
-                CompletionStage<String> f = map.getAsync(state.randomKey());
-                f.thenRunAsync(this);
+                return;
+            }
+
+            int probabilityPosition = threadState.randomInt(probabilityArray.length);
+            this.method = probabilityArray[probabilityPosition];
+
+            switch (method) {
+                case METHOD_GET: {
+                    CompletionStage<String> f = map.getAsync(state.randomKey());
+                    f.thenRunAsync(this);
+                }
+                break;
+                case METHOD_SET: {
+                    CompletionStage<Void> f = map.setAsync(state.randomKey(), threadState.randomValue());
+                    f.thenRunAsync(this);
+                }
+                break;
+                case METHOD_PUT: {
+                    CompletionStage<String> f = map.putAsync(state.randomKey(), threadState.randomValue());
+                    f.thenRunAsync(this);
+                }
+                break;
+                default:
+                    throw new RuntimeException("unexpected method index: " + method);
             }
         }
     }

java/simulator/src/main/java/com/hazelcast/simulator/worker/testcontainer/Probability.java

@@ -88,6 +88,18 @@ public static byte[] loadTimeStepProbabilityArray(Map<Method, Probability> metho
         return ratiosToMethodProbabilityArray(methodRatios);
     }
 
+
+    public static byte[] loadTimeStepProbabilityArray(List<Probability> methods) {
+        double[] methodProbabilities = new double[methods.size()];
+        for (int methodIndex = 0; methodIndex < methods.size(); methodIndex++) {
+            Probability probability = methods.get(methodIndex);
+            methodProbabilities[methodIndex] = probability.getValue();
+        }
+
+        int[] methodRatios = methodProbabilitiesToMethodRatios(methodProbabilities);
+        return ratiosToMethodProbabilityArray(methodRatios);
+    }
+
     public static int[] methodProbabilitiesToMethodRatios(double... methodProbabilities) {
         int[] roundedMethodProbabilities = new int[methodProbabilities.length];
 

templates/hazelcast5-ec2/async-tests.yaml

@@ -0,0 +1,50 @@
+# Using asynchronous load generation.
+#
+# So instead of having thread per request, a concurrency level is specified. And the load
+# generator will keep that number of concurrent request using the IMap async methods.
+# With synchronous load generation, the clients can quickly become a bottleneck, especially
+# when testing high throughput due to context switching. With async load generation this is much less
+# if a problem since there are fewer threads involved. So they are a much better alternative to do
+# throughput testing.
+
+- name: read_only
+  duration: 300s
+  repetitions: 1
+  clients: 1
+  members: 1
+  driver: hazelcast5
+  version: maven=5.3.0
+  client_args: >
+    -Xms3g
+    -Xmx3g
+    --add-modules java.se 
+    --add-exports java.base/jdk.internal.ref=ALL-UNNAMED 
+    --add-opens java.base/java.lang=ALL-UNNAMED 
+    --add-opens java.base/sun.nio.ch=ALL-UNNAMED 
+    --add-opens java.management/sun.management=ALL-UNNAMED 
+    --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED
+  member_args: >
+    -Xms3g
+    -Xmx3g
+    --add-modules java.se 
+    --add-exports java.base/jdk.internal.ref=ALL-UNNAMED 
+    --add-opens java.base/java.lang=ALL-UNNAMED 
+    --add-opens java.base/sun.nio.ch=ALL-UNNAMED 
+    --add-opens java.management/sun.management=ALL-UNNAMED 
+    --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED
+  loadgenerator_hosts: loadgenerators
+  node_hosts: nodes
+  verify_enabled: False
+  performance_monitor_interval_seconds: 1
+  warmup_seconds: 0
+  cooldown_seconds: 0
+    - class: com.hazelcast.simulator.hz.map.AsyncLongStringMapTest
+      name: map
+      getProb: 1
+      putProb: 0
+      setProb: 0
+      concurrency: 100
+      keyDomain: 1_000_000
+      valueCount: 100
+      minValueLength: 100
+      maxValueLength: 100