You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ke...@apache.org on 2024/06/17 18:29:29 UTC
(tinkerpop) 01/01: Update profiling with latency test
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch 370-profiling
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 5e3d8eca62383a9402ec25f24cd8b896053964a0
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Sat Jun 15 13:04:27 2024 -0700
Update profiling with latency test
---
.../gremlin/driver/util/ProfilingApplication.java | 132 ++++++++++++++++-----
1 file changed, 100 insertions(+), 32 deletions(-)
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
index 4259e2545f..c1fbc7f69f 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
@@ -30,7 +30,9 @@ import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.PrintWriter;
+import java.util.Iterator;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -48,7 +50,7 @@ import java.util.stream.IntStream;
*/
public class ProfilingApplication {
- private static final Random random = new Random();
+ private static final Random random = new Random(0);
private static final String[] scripts = new String[]{
"g.V()",
"g.V(1).out('knows')",
@@ -87,7 +89,7 @@ public class ProfilingApplication {
this.exercise = exercise;
}
- public long execute() throws Exception {
+ public long executeThroughput() throws Exception {
final AtomicInteger tooSlow = new AtomicInteger(0);
final Client client = cluster.connect();
@@ -130,16 +132,53 @@ public class ProfilingApplication {
}
}
+ public double executeLatency() throws Exception {
+ final Client client = cluster.connect();
+ final String executionId = "[" + executionName + "]";
+ try {
+ client.init();
+
+ long total = 0;
+
+ final long start = System.nanoTime();
+ int size = 0;
+ final Iterator itr = client.submitAsync(script).get().iterator();
+ try {
+ while (true) {
+ itr.next();
+ size++;
+ }
+ } catch (NoSuchElementException nsee) {
+ ; // Expected as hasNext() not called to increase performance.
+ }
+ final long end = System.nanoTime();
+ total += (end - start);
+
+
+ final double totalSeconds = total / 1000000000d;
+ System.out.println(String.format(StringUtils.rightPad(executionId, 10) + "time: %s, result count: %s", StringUtils.rightPad(String.valueOf(totalSeconds), 7), StringUtils.rightPad(String.valueOf(size), 10)));
+ return totalSeconds;
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ throw new RuntimeException(ex);
+ } finally {
+ client.close();
+ }
+ }
+
private String chooseScript() {
return scripts[random.nextInt(scripts.length - 1)];
}
+ public enum TestType { UNKNOWN, LATENCY, THROUGHPUT };
+
public static void main(final String[] args) {
final Map<String,Object> options = ElementHelper.asMap(args);
final boolean noExit = Boolean.parseBoolean(options.getOrDefault("noExit", "false").toString());
final int parallelism = Integer.parseInt(options.getOrDefault("parallelism", "16").toString());
final BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("profiler-%d").build();
final ExecutorService executor = Executors.newFixedThreadPool(parallelism, threadFactory);
+ final TestType testType = TestType.values()[(Integer.parseInt(options.getOrDefault("testType", "2").toString()) % TestType.values().length)];
final String host = options.getOrDefault("host", "localhost").toString();
final int minExpectedRps = Integer.parseInt(options.getOrDefault("minExpectedRps", "1000").toString());
@@ -155,7 +194,7 @@ public class ProfilingApplication {
final int maxInProcessPerConnection = Integer.parseInt(options.getOrDefault("maxInProcessPerConnection", "64").toString());
final int minInProcessPerConnection = Integer.parseInt(options.getOrDefault("minInProcessPerConnection", "16").toString());
final int maxWaitForConnection = Integer.parseInt(options.getOrDefault("maxWaitForConnection", "3000").toString());
- final int workerPoolSize = Integer.parseInt(options.getOrDefault("workerPoolSize", "2").toString());
+ final int workerPoolSize = Integer.parseInt(options.getOrDefault("workerPoolSize", Runtime.getRuntime().availableProcessors() * 2).toString());
final int tooSlowThreshold = Integer.parseInt(options.getOrDefault("tooSlowThreshold", "125").toString());
final String channelizer = options.getOrDefault("channelizer", Channelizer.WebSocketChannelizer.class.getName()).toString();
final String serializer = options.getOrDefault("serializer", Serializers.GRAPHBINARY_V1.name()).toString();
@@ -189,43 +228,72 @@ public class ProfilingApplication {
System.out.println("Modern graph loaded");
}
- final Object fileName = options.get("store");
- final File f = null == fileName ? null : new File(fileName.toString());
- if (f != null && f.length() == 0) {
- try (final PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter(f, true)))) {
- writer.println("parallelism\tnioPoolSize\tminConnectionPoolSize\tmaxConnectionPoolSize\tminSimultaneousUsagePerConnection\tmaxSimultaneousUsagePerConnection\tminInProcessPerConnection\tmaxInProcessPerConnection\tworkerPoolSize\trequestPerSecond");
+ if (TestType.THROUGHPUT == testType) {
+ final Object fileName = options.get("store");
+ final File f = null == fileName ? null : new File(fileName.toString());
+ if (f != null && f.length() == 0) {
+ try (final PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter(f, true)))) {
+ writer.println("parallelism\tnioPoolSize\tminConnectionPoolSize\tmaxConnectionPoolSize\tminSimultaneousUsagePerConnection\tmaxSimultaneousUsagePerConnection\tminInProcessPerConnection\tmaxInProcessPerConnection\tworkerPoolSize\trequestPerSecond");
+ }
}
- }
- // not much point to continuing with a line of tests if we can't get at least minExpectedRps.
- final AtomicBoolean meetsRpsExpectation = new AtomicBoolean(true);
- System.out.println("---------------------------WARMUP CYCLE---------------------------");
- for (int ix = 0; ix < warmups && meetsRpsExpectation.get(); ix++) {
- final long averageRequestsPerSecond = new ProfilingApplication("warmup-" + (ix + 1), cluster, 1000, executor, script, tooSlowThreshold, exercise).execute();
- meetsRpsExpectation.set(averageRequestsPerSecond > minExpectedRps);
- TimeUnit.SECONDS.sleep(1); // pause between executions
- }
+ // not much point to continuing with a line of tests if we can't get at least minExpectedRps.
+ final AtomicBoolean meetsRpsExpectation = new AtomicBoolean(true);
+ System.out.println("---------------------------WARMUP CYCLE---------------------------");
+ for (int ix = 0; ix < warmups && meetsRpsExpectation.get(); ix++) {
+ final long averageRequestsPerSecond = new ProfilingApplication("warmup-" + (ix + 1), cluster, 1000, executor, script, tooSlowThreshold, exercise).executeThroughput();
+ meetsRpsExpectation.set(averageRequestsPerSecond > minExpectedRps);
+ TimeUnit.SECONDS.sleep(1); // pause between executions
+ }
- final AtomicBoolean exceededTimeout = new AtomicBoolean(false);
- long totalRequestsPerSecond = 0;
+ final AtomicBoolean exceededTimeout = new AtomicBoolean(false);
+ long totalRequestsPerSecond = 0;
- // no need to execute this if we didn't pass the basic expectation in the warmups
- if (exercise || meetsRpsExpectation.get()) {
- final long start = System.nanoTime();
- System.out.println("----------------------------TEST CYCLE----------------------------");
- for (int ix = 0; ix < executions && !exceededTimeout.get(); ix++) {
- totalRequestsPerSecond += new ProfilingApplication("test-" + (ix + 1), cluster, requests, executor, script, tooSlowThreshold, exercise).execute();
- exceededTimeout.set((System.nanoTime() - start) > TimeUnit.NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS));
+ // no need to execute this if we didn't pass the basic expectation in the warmups
+ if (exercise || meetsRpsExpectation.get()) {
+ final long start = System.nanoTime();
+ System.out.println("----------------------------TEST CYCLE----------------------------");
+ for (int ix = 0; ix < executions && !exceededTimeout.get(); ix++) {
+ totalRequestsPerSecond += new ProfilingApplication("test-" + (ix + 1), cluster, requests, executor, script, tooSlowThreshold, exercise).executeThroughput();
+ exceededTimeout.set((System.nanoTime() - start) > TimeUnit.NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS));
+ TimeUnit.SECONDS.sleep(1); // pause between executions
+ }
+ }
+
+ final int averageRequestPerSecond = !meetsRpsExpectation.get() || exceededTimeout.get() ? 0 : Math.round(totalRequestsPerSecond / executions);
+ System.out.println(String.format("avg req/sec: %s", averageRequestPerSecond));
+ if (f != null) {
+ try (final PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter(f, true)))) {
+ writer.println(String.join("\t", String.valueOf(parallelism), String.valueOf(nioPoolSize), String.valueOf(minConnectionPoolSize), String.valueOf(maxConnectionPoolSize), String.valueOf(minSimultaneousUsagePerConnection), String.valueOf(maxSimultaneousUsagePerConnection), String.valueOf(minInProcessPerConnection), String.valueOf(maxInProcessPerConnection), String.valueOf(workerPoolSize), String.valueOf(averageRequestPerSecond)));
+ }
+ }
+ } else if (TestType.LATENCY == testType) {
+ final AtomicBoolean meetsTimeoutExpectation = new AtomicBoolean(true);
+ System.out.println("---------------------------WARMUP CYCLE---------------------------");
+ for (int ix = 0; ix < warmups && meetsTimeoutExpectation.get(); ix++) {
+ final double latency = new ProfilingApplication("warmup-" + (ix + 1), cluster, 1000, executor, script, tooSlowThreshold, exercise).executeLatency();
+ meetsTimeoutExpectation.set(latency < timeout);
TimeUnit.SECONDS.sleep(1); // pause between executions
}
- }
- final int averageRequestPerSecond = !meetsRpsExpectation.get() || exceededTimeout.get() ? 0 : Math.round(totalRequestsPerSecond / executions);
- System.out.println(String.format("avg req/sec: %s", averageRequestPerSecond));
- if (f != null) {
- try (final PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter(f, true)))) {
- writer.println(String.join("\t", String.valueOf(parallelism), String.valueOf(nioPoolSize), String.valueOf(minConnectionPoolSize), String.valueOf(maxConnectionPoolSize), String.valueOf(minSimultaneousUsagePerConnection), String.valueOf(maxSimultaneousUsagePerConnection), String.valueOf(minInProcessPerConnection), String.valueOf(maxInProcessPerConnection), String.valueOf(workerPoolSize), String.valueOf(averageRequestPerSecond)));
+ final AtomicBoolean exceededTimeout = new AtomicBoolean(false);
+ double totalTime = 0;
+
+ // no need to execute this if we didn't pass the basic expectation in the warmups
+ if (exercise || meetsTimeoutExpectation.get()) {
+ final long start = System.nanoTime();
+ System.out.println("----------------------------TEST CYCLE----------------------------");
+ for (int ix = 0; ix < executions && !exceededTimeout.get(); ix++) {
+ totalTime += new ProfilingApplication("test-" + (ix + 1), cluster, requests, executor, script, tooSlowThreshold, exercise).executeLatency();
+ exceededTimeout.set((System.nanoTime() - start) > TimeUnit.NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS));
+ TimeUnit.SECONDS.sleep(1); // pause between executions
+ }
}
+
+ final double averageLatency = !meetsTimeoutExpectation.get() || exceededTimeout.get() ? 0 : (totalTime / executions);
+ System.out.println(String.format("avg latency (sec/req): %s", averageLatency));
+ } else {
+ System.out.println("Encountered unknown testType. Please enter a valid value and try again.");
}
if (!noExit) System.exit(0);