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/24 22:38:09 UTC
(tinkerpop) branch http-compression created (now 1385ee9921)
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a change to branch http-compression
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
at 1385ee9921 Initial implementation for transfer encoded compression
This branch includes the following new commits:
new 157ab777be Update profiling with latency test
new 9a0517c456 Update profiling app
new 1385ee9921 Initial implementation for transfer encoded compression
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tinkerpop) 03/03: Initial implementation for transfer encoded compression
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch http-compression
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 1385ee992198c4d48879d7fa7a909c1bf1ba697c
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Mon Jun 24 15:34:54 2024 -0700
Initial implementation for transfer encoded compression
---
.../tinkerpop/gremlin/driver/Channelizer.java | 5 ++
.../handler/HttpContentCompressionDecoder.java | 97 ++++++++++++++++++++++
.../gremlin/server/channel/HttpChannelizer.java | 4 +
.../handler/HttpContentCompressionEncoder.java | 86 +++++++++++++++++++
.../server/handler/HttpGremlinEndpointHandler.java | 4 +-
5 files changed, 195 insertions(+), 1 deletion(-)
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
index 5d8af18a3f..f2984e3e13 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
@@ -30,6 +30,7 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import org.apache.tinkerpop.gremlin.driver.handler.GremlinResponseHandler;
+import org.apache.tinkerpop.gremlin.driver.handler.HttpContentCompressionDecoder;
import org.apache.tinkerpop.gremlin.driver.handler.HttpGremlinRequestEncoder;
import org.apache.tinkerpop.gremlin.driver.handler.HttpGremlinResponseStreamDecoder;
import org.apache.tinkerpop.gremlin.driver.handler.SslCheckHandler;
@@ -167,10 +168,13 @@ public interface Channelizer extends ChannelHandler {
private HttpGremlinRequestEncoder gremlinRequestEncoder;
private HttpGremlinResponseStreamDecoder gremlinResponseDecoder;
+ private HttpContentCompressionDecoder httpCompressionDecoder;
+
@Override
public void init(final Connection connection) {
super.init(connection);
+ httpCompressionDecoder = new HttpContentCompressionDecoder();
gremlinRequestEncoder = new HttpGremlinRequestEncoder(cluster.getSerializer(), cluster.getRequestInterceptor(), cluster.isUserAgentOnConnectEnabled());
gremlinResponseDecoder = new HttpGremlinResponseStreamDecoder(cluster.getSerializer(), cluster.getMaxContentLength());
}
@@ -202,6 +206,7 @@ public interface Channelizer extends ChannelHandler {
pipeline.addLast(PIPELINE_HTTP_CODEC, handler);
pipeline.addLast(PIPELINE_HTTP_ENCODER, gremlinRequestEncoder);
+ pipeline.addLast("compression-decoder", httpCompressionDecoder);
pipeline.addLast(PIPELINE_HTTP_DECODER, gremlinResponseDecoder);
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpContentCompressionDecoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpContentCompressionDecoder.java
new file mode 100644
index 0000000000..3351631237
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpContentCompressionDecoder.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.MessageToMessageCodec;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.HttpContent;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpObject;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.util.AttributeKey;
+import io.netty.util.ReferenceCountUtil;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.zip.Inflater;
+
+@ChannelHandler.Sharable
+public class HttpContentCompressionDecoder extends ChannelInboundHandlerAdapter {
+ private static final AttributeKey<Inflater> INFLATER = AttributeKey.valueOf("inflater");
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ Object out = msg;
+
+ if (msg instanceof HttpResponse && (hasDeflateTransferEncoding(((HttpResponse) msg).headers()))) {
+ ctx.attr(INFLATER).set(new Inflater());
+ } else if (msg instanceof HttpContent) {
+ final Inflater decompressor = ctx.attr(INFLATER).get();
+ if (decompressor != null) {
+ final ByteBuf chunk = ((HttpContent) msg).content();
+ decompressor.setInput(ByteBufUtil.getBytes(chunk));
+ final int decompressedSizeEstimate = chunk.readableBytes()*8;
+ byte[] outBuf = new byte[decompressedSizeEstimate > 10 ? decompressedSizeEstimate : 10];
+
+ int writeIdx = 0;
+ int bytesWritten = decompressor.inflate(outBuf, writeIdx, outBuf.length);
+
+ while (bytesWritten == (outBuf.length - writeIdx)) {
+ writeIdx += bytesWritten;
+ outBuf = Arrays.copyOf(outBuf, outBuf.length * 2);
+ bytesWritten = decompressor.inflate(outBuf, writeIdx, outBuf.length - writeIdx);
+ }
+
+ writeIdx += bytesWritten;
+ out = ((HttpContent) msg).replace(Unpooled.wrappedBuffer(outBuf).writerIndex(writeIdx));
+ ReferenceCountUtil.release(msg);
+ }
+ }
+
+ if (out instanceof LastHttpContent) {
+ ctx.attr(INFLATER).getAndSet(null).end();
+ }
+
+ super.channelRead(ctx, out);
+ }
+
+ private boolean hasDeflateTransferEncoding(HttpHeaders headers) {
+ for (String value : headers.getAll(HttpHeaderNames.TRANSFER_ENCODING)) {
+ if ("deflate".equals(value)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java
index 93bc5aa826..2935fc95f4 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/HttpChannelizer.java
@@ -29,6 +29,7 @@ import org.apache.tinkerpop.gremlin.server.auth.AllowAllAuthenticator;
import org.apache.tinkerpop.gremlin.server.handler.AbstractAuthenticationHandler;
import org.apache.tinkerpop.gremlin.server.handler.HttpBasicAuthenticationHandler;
import org.apache.tinkerpop.gremlin.server.handler.HttpBasicAuthorizationHandler;
+import org.apache.tinkerpop.gremlin.server.handler.HttpContentCompressionEncoder;
import org.apache.tinkerpop.gremlin.server.handler.HttpRequestCheckingHandler;
import org.apache.tinkerpop.gremlin.server.handler.HttpRequestIdHandler;
import org.apache.tinkerpop.gremlin.server.handler.HttpRequestMessageDecoder;
@@ -56,6 +57,8 @@ public class HttpChannelizer extends AbstractChannelizer {
private HttpRequestMessageDecoder httpRequestMessageDecoder = new HttpRequestMessageDecoder(serializers);
private HttpRequestIdHandler httpRequestIdHandler = new HttpRequestIdHandler();
+ private HttpContentCompressionEncoder httpContentEncoder = new HttpContentCompressionEncoder();
+
@Override
public void init(final ServerGremlinExecutor serverGremlinExecutor) {
super.init(serverGremlinExecutor);
@@ -81,6 +84,7 @@ public class HttpChannelizer extends AbstractChannelizer {
pipeline.addLast(PIPELINE_HTTP_AGGREGATOR, aggregator);
pipeline.addLast("http-request-checker", httpRequestCheckingHandler);
pipeline.addLast("http-user-agent-handler", new HttpUserAgentHandler());
+ pipeline.addLast("http-compression-encoder", httpContentEncoder);
if (authenticator != null) {
// Cannot add the same handler instance multiple times unless
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpContentCompressionEncoder.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpContentCompressionEncoder.java
new file mode 100644
index 0000000000..c7f96fffa8
--- /dev/null
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpContentCompressionEncoder.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.server.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.MessageToMessageCodec;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.HttpContent;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpObject;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.util.AttributeKey;
+import io.netty.util.ReferenceCountUtil;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.zip.Deflater;
+
+@ChannelHandler.Sharable
+public class HttpContentCompressionEncoder extends ChannelOutboundHandlerAdapter {
+ private static final AttributeKey<Deflater> DEFLATER = AttributeKey.valueOf("deflater");
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+ Object out = msg;
+ int writeIdx = 0;
+ int bytesWritten = 0;
+
+ if (msg instanceof HttpResponse && (((HttpResponse) msg).headers().contains(HttpHeaderNames.TRANSFER_ENCODING))) {
+ ctx.attr(DEFLATER).set(new Deflater());
+ } else if (msg instanceof HttpContent) {
+ final ByteBuf chunk = ((HttpContent) msg).content();
+ final Deflater compressor = ctx.attr(DEFLATER).get();
+
+ if (compressor != null) {
+ compressor.setInput(ByteBufUtil.getBytes(chunk));
+
+ if (msg instanceof LastHttpContent) compressor.finish();
+
+ final int compressedSizeEstimate = chunk.readableBytes() / 8;
+ byte[] outBuf = new byte[compressedSizeEstimate > 10 ? compressedSizeEstimate : 10];
+ bytesWritten = compressor.deflate(outBuf, writeIdx, outBuf.length, Deflater.SYNC_FLUSH);
+ while (bytesWritten == (outBuf.length - writeIdx)) {
+ writeIdx += bytesWritten;
+ outBuf = Arrays.copyOf(outBuf, outBuf.length * 2);
+ bytesWritten = compressor.deflate(outBuf, writeIdx, outBuf.length - writeIdx, Deflater.SYNC_FLUSH);
+ }
+ writeIdx += bytesWritten;
+ out = ((HttpContent) msg).replace(Unpooled.wrappedBuffer(outBuf).writerIndex(writeIdx));
+ ReferenceCountUtil.release(msg);
+ }
+ }
+
+ if (out instanceof LastHttpContent) {
+ ctx.attr(DEFLATER).getAndSet(null).end();
+ }
+
+ super.write(ctx, out, promise);
+ }
+}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
index 711a0d87ae..cf4fa913a4 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
@@ -93,6 +93,7 @@ import java.util.stream.Stream;
import static com.codahale.metrics.MetricRegistry.name;
import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING;
import static io.netty.handler.codec.http.HttpHeaderValues.CHUNKED;
+import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
@@ -198,7 +199,8 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
// Send back the 200 OK response header here since the response is always chunk transfer encoded. Any
// failures that follow this will show up in the response body instead.
final HttpResponse responseHeader = new DefaultHttpResponse(HTTP_1_1, OK);
- responseHeader.headers().set(TRANSFER_ENCODING, CHUNKED);
+ responseHeader.headers().add(TRANSFER_ENCODING, DEFLATE);
+ responseHeader.headers().add(TRANSFER_ENCODING, CHUNKED);
responseHeader.headers().set(HttpHeaderNames.CONTENT_TYPE, serializer.getValue0());
ctx.writeAndFlush(responseHeader);
ctx.channel().attr(StateKey.HTTP_RESPONSE_SENT).set(true);
(tinkerpop) 01/03: Update profiling with latency test
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch http-compression
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 157ab777be67eac7a9b558f2f5ebd020ea208d04
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 e505fbad70..1df5896d5f 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());
@@ -151,7 +190,7 @@ public class ProfilingApplication {
final int minConnectionPoolSize = Integer.parseInt(options.getOrDefault("minConnectionPoolSize", "256").toString());
final int maxConnectionPoolSize = Integer.parseInt(options.getOrDefault("maxConnectionPoolSize", "256").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 serializer = options.getOrDefault("serializer", SerializersV4.GRAPHBINARY_V4.name()).toString();
@@ -179,43 +218,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(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);
(tinkerpop) 02/03: Update profiling app
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch http-compression
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 9a0517c4567cacffdf3d20d457cb29ad768e112f
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Mon Jun 17 11:28:27 2024 -0700
Update profiling app
---
.../tinkerpop/gremlin/driver/util/ProfilingApplication.java | 11 ++++++-----
1 file changed, 6 insertions(+), 5 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 1df5896d5f..d4a4fb109f 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
@@ -187,9 +187,9 @@ public class ProfilingApplication {
final int executions = Integer.parseInt(options.getOrDefault("executions", "10").toString());
final int nioPoolSize = Integer.parseInt(options.getOrDefault("nioPoolSize", "1").toString());
final int requests = Integer.parseInt(options.getOrDefault("requests", "10000").toString());
- final int minConnectionPoolSize = Integer.parseInt(options.getOrDefault("minConnectionPoolSize", "256").toString());
- final int maxConnectionPoolSize = Integer.parseInt(options.getOrDefault("maxConnectionPoolSize", "256").toString());
- final int maxWaitForConnection = Integer.parseInt(options.getOrDefault("maxWaitForConnection", "3000").toString());
+ final int minConnectionPoolSize = Integer.parseInt(options.getOrDefault("minConnectionPoolSize", "128").toString());
+ final int maxConnectionPoolSize = Integer.parseInt(options.getOrDefault("maxConnectionPoolSize", "1024").toString());
+ final int maxWaitForConnection = Integer.parseInt(options.getOrDefault("maxWaitForConnection", "300000").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 serializer = options.getOrDefault("serializer", SerializersV4.GRAPHBINARY_V4.name()).toString();
@@ -198,6 +198,7 @@ public class ProfilingApplication {
final String script = options.getOrDefault("script", "1+1").toString();
final Cluster cluster = Cluster.build(host)
+ .maxContentLength(Integer.MAX_VALUE)
.minConnectionPoolSize(minConnectionPoolSize)
.maxConnectionPoolSize(maxConnectionPoolSize)
.nioPoolSize(nioPoolSize)
@@ -223,7 +224,7 @@ public class ProfilingApplication {
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");
+ writer.println("parallelism\tnioPoolSize\tminConnectionPoolSize\tmaxConnectionPoolSize\tworkerPoolSize\trequestPerSecond");
}
}
@@ -254,7 +255,7 @@ public class ProfilingApplication {
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)));
+ writer.println(String.join("\t", String.valueOf(parallelism), String.valueOf(nioPoolSize), String.valueOf(minConnectionPoolSize), String.valueOf(maxConnectionPoolSize), String.valueOf(workerPoolSize), String.valueOf(averageRequestPerSecond)));
}
}
} else if (TestType.LATENCY == testType) {