You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2025/01/06 15:21:29 UTC
(tinkerpop) branch TINKERPOP-3124 updated (72f04ab46c -> a19f3653d5)
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a change to branch TINKERPOP-3124
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
discard 72f04ab46c TINKERPOP-3124 Added custom MessageSizeEstimator to cover Frame
new a19f3653d5 TINKERPOP-3124 Added custom MessageSizeEstimator to cover Frame
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (72f04ab46c)
\
N -- N -- N refs/heads/TINKERPOP-3124 (a19f3653d5)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 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.
Summary of changes:
docs/src/reference/gremlin-applications.asciidoc | 35 ++++++++++++++++++++++--
1 file changed, 33 insertions(+), 2 deletions(-)
(tinkerpop) 01/01: TINKERPOP-3124 Added custom MessageSizeEstimator to cover Frame
Posted by sp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-3124
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit a19f3653d5d1488d9972b01ed3fda721d275e776
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Fri Jan 3 15:54:58 2025 -0500
TINKERPOP-3124 Added custom MessageSizeEstimator to cover Frame
Improved logging around channel pauses. Added new metrics for Gremlin Server around channels and pauses. Added some extra capabilities to the java driver profiling tool.
---
CHANGELOG.asciidoc | 5 +-
docs/src/reference/gremlin-applications.asciidoc | 58 +++++++++++---
docs/src/upgrade/release-3.7.x.asciidoc | 22 +++++
.../gremlin/driver/util/ProfilingApplication.java | 44 ++++++----
.../gremlin/server/AbstractChannelizer.java | 12 +++
.../tinkerpop/gremlin/server/GremlinServer.java | 45 ++++++-----
.../gremlin/server/handler/AbstractSession.java | 33 ++++++--
.../server/handler/FrameMessageSizeEstimator.java | 93 ++++++++++++++++++++++
.../gremlin/server/op/AbstractOpProcessor.java | 45 +++++++++--
.../server/op/session/SessionOpProcessor.java | 32 ++++++--
.../server/op/traversal/TraversalOpProcessor.java | 30 +++++--
.../gremlin/server/util/MetricManager.java | 4 +-
.../gremlin/server/util/ServerGremlinExecutor.java | 43 +++++++++-
.../gremlin/server/GremlinServerIntegrateTest.java | 16 ++--
14 files changed, 397 insertions(+), 85 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 557aec0372..161b0dd594 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -23,7 +23,10 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
[[release-3-7-4]]
=== TinkerPop 3.7.4 (NOT OFFICIALLY RELEASED YET)
-* Add log entry in `WsAndHttpChannelizerHandler`.
+* Added log entry in `WsAndHttpChannelizerHandler` to catch general errors that escape the handlers.
+* Added a `MessageSizeEstimator` implementation to cover `Frame` allowing Gremlin Server to better estimate message sizes for the direct buffer.
+* Improved logging around triggers of the `writeBufferHighWaterMark` so that they occur more than once but do not excessively fill the logs.
+* Added server metrics to help better detect and diagnose write pauses due to the `writeBufferHighWaterMark`: `channels.paused`, `channels.total`, and `channels.write-pauses`.
* Changed `IdentityRemovalStrategy` to omit `IdentityStep` if only with `RepeatEndStep` under `RepeatStep`.
* Changed Gremlin grammar to make use of `g` to spawn child traversals a syntax error.
diff --git a/docs/src/reference/gremlin-applications.asciidoc b/docs/src/reference/gremlin-applications.asciidoc
index 196e59fd4d..9df004cfc6 100644
--- a/docs/src/reference/gremlin-applications.asciidoc
+++ b/docs/src/reference/gremlin-applications.asciidoc
@@ -930,6 +930,7 @@ that iterates thousands of results will serialize each of those in memory into a
quite possible that such a script will generate `OutOfMemoryError` exceptions on the server. Consider the default
WebSocket configuration, which supports streaming, if that type of use case is required.
+[[server-configuring]]
=== Configuring
The `gremlin-server.sh` file serves multiple purposes. It can be used to "install" dependencies to the Gremlin
@@ -1062,7 +1063,7 @@ The following table describes the various YAML configuration options that Gremli
|useEpollEventLoop |Try to use epoll event loops (works only on Linux os) instead of netty NIO. |false
|useGlobalFunctionCacheForSessions |Enable the global function cache for sessions when using the `UnifiedChannelizer`. When `true` it means that functions created in one request to a session remain available on the next request to that session. This setting is only relevant when `useGlobalFunctionCacheForSessions` is `false`. |true
|writeBufferHighWaterMark | If the number of bytes in the network send buffer exceeds this value then the channel is no longer writeable, accepting no additional writes until buffer is drained and the `writeBufferLowWaterMark` is met. |65536
-|writeBufferLowWaterMark | Once the number of bytes queued in the network send buffer exceeds the `writeBufferHighWaterMark`, the channel will not become writeable again until the buffer is drained and it drops below this value. |65536
+|writeBufferLowWaterMark | Once the number of bytes queued in the network send buffer exceeds the `writeBufferHighWaterMark`, the channel will not become writeable again until the buffer is drained and it drops below this value. |32768
|=========================================================
See the <<metrics,Metrics>> section for more information on how to configure Ganglia and Graphite.
@@ -1273,9 +1274,17 @@ NOTE: Installing Ganglia will include `org.acplt:oncrpc`, which is an LGPL licen
Regardless of the output, the metrics gathered are the same. Each metric is prefixed with
`org.apache.tinkerpop.gremlin.server.GremlinServer` and the following metrics are reported:
-* `sessions` - The number of sessions open at the time the metric was last measured. For the `UnifiedChannelizer`, each
-request creates a "session", even a so-called "sessionless request", which is basically a session that will only
-execute within the context of that single request.
+* `channels.paused` - The current number of open channels (HTTP and Websocket) that are have their writes to buffer
+paused when the `writeBufferHighWaterMark` configuration is exceeded.
+* `channels.total` - The current number of open channels (HTTP and Websocket).
+* `channels.write-pauses` - The total number of pauses across all channels (HTTP and Websocket) to buffer writes where
+the `writeBufferHighWaterMark` configuration is exceeded, with mean rate, as well as the 1, 5, and 15-minute rates.
+* `engine-name.session.session-id.*` - Metrics related to different `GremlinScriptEngine` instances configured for
+session-based requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy" and
+"session-id" will be the identifier for the session itself. This metric is not measured under the `UnifiedChannelizer`.
+* `engine-name.sessionless.*` - Metrics related to different `GremlinScriptEngine` instances configured for sessionless
+requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy". This metric is not
+measured under the `UnifiedChannelizer`.
* `errors` - The number of total errors, mean rate, as well as the 1, 5, and 15-minute error rates.
* `op.eval` - The number of script evaluations, mean rate, 1, 5, and 15 minute rates, minimum, maximum, median, mean,
and standard deviation evaluation times, as well as the 75th, 95th, 98th, 99th and 99.9th percentile evaluation times
@@ -1283,12 +1292,9 @@ and standard deviation evaluation times, as well as the 75th, 95th, 98th, 99th a
* `op.traversal` - The number of `Traversal` bytecode-based executions, mean rate, 1, 5, and 15 minute rates, minimum,
maximum, median, mean, and standard deviation evaluation times, as well as the 75th, 95th, 98th, 99th and 99.9th
percentile evaluation times.
-* `engine-name.session.session-id.*` - Metrics related to different `GremlinScriptEngine` instances configured for
-session-based requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy" and
-"session-id" will be the identifier for the session itself. This metric is not measured under the `UnifiedChannelizer`.
-* `engine-name.sessionless.*` - Metrics related to different `GremlinScriptEngine` instances configured for sessionless
-requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy". This metric is not
-measured under the `UnifiedChannelizer`.
+* `sessions` - The number of sessions open at the time the metric was last measured. For the `UnifiedChannelizer`, each
+request creates a "session", even a so-called "sessionless request", which is basically a session that will only
+execute within the context of that single request.
* `user-agent.*` - Counts the number of connection requests from clients providing a given user agent.
NOTE: Gremlin Server has a limit of 10000 unique user agents to be tracked by metrics. If this cap is exceeded
@@ -2103,7 +2109,8 @@ The following sections define best practices for working with Gremlin Server.
==== Tuning
-image:gremlin-handdrawn.png[width=120,float=right] Tuning Gremlin Server for a particular environment may require some simple trial-and-error, but the following represent some basic guidelines that might be useful:
+image:gremlin-handdrawn.png[width=120,float=right] Tuning Gremlin Server for a particular environment may require some
+simple trial-and-error, but the following represent some basic guidelines that might be useful:
* Gremlin Server defaults to a very modest maximum heap size. Consider increasing this value for non-trivial uses.
Maximum heap size (`-Xmx`) is defined with the `JAVA_OPTIONS` setting in `gremlin-server.conf`.
@@ -2153,6 +2160,35 @@ data that is required. For example, if only two properties of a `Vertex` are nee
than returning the entire `Vertex` object itself. Even with an entire `Vertex`, it is typically much faster to issue
the query as `g.V(1).elementMap()` than `g.V(1)`, as the former returns a `Map` of the same data as a `Vertex`, but
without all the associated structure which can slow the response.
+* Gremlin Server writes responses to a buffer held in direct memory prior to flushing them to the TCP socket. If the
+logs show `OutOfDirectMemoryError`, particularly when the `channels.write-pauses` <<metrics,metric>> is high, it is
+likely caused by this buffer being filled. The buffer can fill when clients are slow to consume results being sent to
+them (e.g. network problems, underpowered client instances, etc.). Gremlin Server will attempt to throttle the speed at
+which the buffer gets filled by pausing writes for any channel that exceeds its allowed buffer space allotment as
+determined by the `writeBufferHighWaterMark` and `writeBufferLowWaterMark` described in the
+<<server-configuring,Server Configuration Section>>. Pauses obviously increase latency, but do so for benefit of
+server stability in continuing to serve channels that have clients without issue consuming the results.
+** Write pauses are generally considered a natural part of server operations, though a continuous amount of pausing
+means that threads used for query execution are tied up and are therefore preventing the processing of other requests.
+As a result, requests may begin to queue which further adds to server load and potential latency. Increasing the
+`writeBufferHighWaterMark` and `writeBufferLowWaterMark` settings could allow the server to delay pauses at the expense
+of direct memory and therefore allow more requests to be handled by freeing those query execution threads.
+** Client applications should be selective in their retries. Quickly resending a query that triggered an
+`OutOfDirectMemoryError` without giving the server time to recover will just further burden a taxed system. Even retry
+systems that use exponential back-off may not be suitable for these cases as early retries may land too quickly and
+therefore just queue another heavy request.
+** Consider the shape of query results as they can have an impact on server performance. The "shape" refers to the form
+of the result given the query. For example, `g.V()` and `g.V().fold()` both return the same results (i.e. all the
+vertices in the graph) but the former returns them one at a time in a stream and the latter collects them all in
+memory in a `List` and then returns the one `List` result. Writing queries in ways that allow results that can stream
+(only applies for websockets) is preferable and will allow the server to perform better. Another aspect of "shape"
+can come into play when returning data of individual graph elements. For example, the `g.V()` form of query will stream,
+but if each `Vertex` returned has lots of properties (e.g. properties with large strings or heavy blobs), this could
+trigger scenarios where each streamed batch immediately exceeds `writeBufferHighWaterMark`. Simply exceeding the
+`writeBufferHighWaterMark` may not trigger a pause as the server may quickly flush the buffer before the next batch, but
+one could see how easily a write pause could be triggered in that state. It could make sense to configure a smaller
+`batchSize` for queries results that have heavy individual objects in them as that would reduce the byte size of the
+batch and allow buffer flushes to happen more often (though that may be a cost in and of itself).
[[parameterized-scripts]]
==== Parameterized Scripts
diff --git a/docs/src/upgrade/release-3.7.x.asciidoc b/docs/src/upgrade/release-3.7.x.asciidoc
index 342d5c8602..8649da72fe 100644
--- a/docs/src/upgrade/release-3.7.x.asciidoc
+++ b/docs/src/upgrade/release-3.7.x.asciidoc
@@ -30,6 +30,28 @@ complete list of all the modifications that are part of this release.
=== Upgrading for Users
+==== Improved Server Memory Management
+
+A TinkerPop-specific `MessageSizeEstimator` was added to more accurately measure the size of responses being written
+back to the client. With a more accurate measurement, the server is able to better prevent exhaustion of direct memory
+by overly eager channels writing large result sets to slower clients. Overall, this change should help reduce the
+likelihood of the server hitting `OutOfMemoryExceptions` and other performance problems that may appear under certain
+workloads and network conditions.
+
+It is worth nothing that logging around the `writeBufferHighWaterMark` has been modified to include a bit more
+information about the pause. This warning formerly was only issued on the first pause per request. Additional pauses
+for that request would not be noted in the logs. The warnings now appear periodically for a request, immediately for
+the first pause and then warnings will continue for subsequent pauses using an exponential backoff.
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-3124[TINKERPOP-3124]
+
+==== Channel Metrics
+
+Gremlin Server has three new metrics in the `org.apache.tinkerpop.gremlin.server.GremlinServer` space:
+`channels.paused`, `channels.total`, and `channels.write-pauses`. These metrics are designed to provide more insight
+into channel (websocket and http) operations to use as a tool in understanding server memory issues and latency.
+
+See: link:https://tinkerpop.apache.org/docs/3.7.4/reference/#metrics[Reference Documentation - Metrics]
=== Upgrading for Providers
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 898f3b1977..c962ec4cd6 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
@@ -76,10 +76,12 @@ public class ProfilingApplication {
private final int tooSlowThreshold;
private final boolean exercise;
+ private final boolean suppressStackTraces;
+
private final ExecutorService executor;
public ProfilingApplication(final String executionName, final Cluster cluster, final int requests, final ExecutorService executor,
- final String script, final int tooSlowThreshold, final boolean exercise) {
+ final String script, final int tooSlowThreshold, final boolean exercise, final boolean suppressStackTraces) {
this.executionName = executionName;
this.cluster = cluster;
this.requests = requests;
@@ -87,10 +89,12 @@ public class ProfilingApplication {
this.script = script;
this.tooSlowThreshold = tooSlowThreshold;
this.exercise = exercise;
+ this.suppressStackTraces = suppressStackTraces;
}
public long executeThroughput() throws Exception {
final AtomicInteger tooSlow = new AtomicInteger(0);
+ final AtomicInteger errors = new AtomicInteger(0);
final Client client = cluster.connect();
final String executionId = "[" + executionName + "]";
@@ -107,7 +111,8 @@ public class ProfilingApplication {
} catch (TimeoutException ex) {
tooSlow.incrementAndGet();
} catch (Exception ex) {
- ex.printStackTrace();
+ errors.incrementAndGet();
+ if (!suppressStackTraces) ex.printStackTrace();
} finally {
latch.countDown();
}
@@ -122,11 +127,17 @@ public class ProfilingApplication {
final double totalSeconds = total / 1000000000d;
final long requestCount = requests;
final long reqSec = Math.round(requestCount / totalSeconds);
- System.out.println(String.format(StringUtils.rightPad(executionId, 10) + " requests: %s | time(s): %s | req/sec: %s | too slow: %s", requestCount, StringUtils.rightPad(String.valueOf(totalSeconds), 14), StringUtils.rightPad(String.valueOf(reqSec), 7), exercise ? "N/A" : tooSlow.get()));
+ System.out.println(String.format(StringUtils.rightPad(executionId, 10) + " requests: %s | time(s): %s | req/sec: %s | too slow: %s | errors: %s",
+ requestCount,
+ StringUtils.rightPad(String.valueOf(totalSeconds), 14),
+ StringUtils.rightPad(String.valueOf(reqSec), 7), exercise ? "N/A" : tooSlow.get(),
+ errors.get()));
return reqSec;
} catch (Exception ex) {
- ex.printStackTrace();
- throw new RuntimeException(ex);
+ // catch all so that it doesn't tank the whole execution. returning 0 will hose up calculations on
+ // averages, but presumably you'd see the failed executions and be suspicious.
+ System.out.println("Failed Execution: " + executionName + " - " + ex.getMessage());
+ return 0;
} finally {
client.close();
}
@@ -152,12 +163,11 @@ public class ProfilingApplication {
final long end = System.nanoTime();
final long 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();
+ if (!suppressStackTraces) ex.printStackTrace();
throw new RuntimeException(ex);
} finally {
client.close();
@@ -196,6 +206,8 @@ public class ProfilingApplication {
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();
+ final int pauseBetweenRuns = Integer.parseInt(options.getOrDefault("pauseBetweenRuns", "1000").toString());
+ final boolean suppressStackTraces = Boolean.parseBoolean(options.getOrDefault("suppressStackTraces", "false").toString());
final boolean exercise = Boolean.parseBoolean(options.getOrDefault("exercise", "false").toString());
final String script = options.getOrDefault("script", "1+1").toString();
@@ -245,9 +257,9 @@ public class ProfilingApplication {
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 long averageRequestsPerSecond = new ProfilingApplication("warmup-" + (ix + 1), cluster, 1000, executor, script, tooSlowThreshold, exercise, suppressStackTraces).executeThroughput();
+ meetsRpsExpectation.set(averageRequestsPerSecond >= minExpectedRps);
+ TimeUnit.MILLISECONDS.sleep(pauseBetweenRuns); // pause between executions
}
final AtomicBoolean exceededTimeout = new AtomicBoolean(false);
@@ -258,9 +270,9 @@ public class ProfilingApplication {
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();
+ totalRequestsPerSecond += new ProfilingApplication("test-" + (ix + 1), cluster, requests, executor, script, tooSlowThreshold, exercise, suppressStackTraces).executeThroughput();
exceededTimeout.set((System.nanoTime() - start) > TimeUnit.NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS));
- TimeUnit.SECONDS.sleep(1); // pause between executions
+ TimeUnit.MILLISECONDS.sleep(pauseBetweenRuns); // pause between executions
}
}
@@ -275,9 +287,9 @@ public class ProfilingApplication {
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();
+ final double latency = new ProfilingApplication("warmup-" + (ix + 1), cluster, 1000, executor, script, tooSlowThreshold, exercise, suppressStackTraces).executeLatency();
meetsTimeoutExpectation.set(latency < timeout);
- TimeUnit.SECONDS.sleep(1); // pause between executions
+ TimeUnit.MILLISECONDS.sleep(pauseBetweenRuns); // pause between executions
}
final AtomicBoolean exceededTimeout = new AtomicBoolean(false);
@@ -288,9 +300,9 @@ public class ProfilingApplication {
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();
+ totalTime += new ProfilingApplication("test-" + (ix + 1), cluster, requests, executor, script, tooSlowThreshold, exercise, suppressStackTraces).executeLatency();
exceededTimeout.set((System.nanoTime() - start) > TimeUnit.NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS));
- TimeUnit.SECONDS.sleep(1); // pause between executions
+ TimeUnit.MILLISECONDS.sleep(pauseBetweenRuns); // pause between executions
}
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java
index c086ffae24..c133f68652 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.server;
+import io.netty.channel.group.ChannelGroup;
import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
@@ -111,6 +112,8 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann
protected final Map<String, MessageSerializer<?>> serializers = new HashMap<>();
+ protected ChannelGroup channels;
+
private OpSelectorHandler opSelectorHandler;
private OpExecutorHandler opExecutorHandler;
@@ -138,6 +141,7 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann
graphManager = serverGremlinExecutor.getGraphManager();
gremlinExecutorService = serverGremlinExecutor.getGremlinExecutorService();
scheduledExecutorService = serverGremlinExecutor.getScheduledExecutorService();
+ channels = serverGremlinExecutor.getChannels();
// instantiate and configure the serializers that gremlin server will use - could error out here
// and fail the server startup
@@ -156,6 +160,10 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann
opExecutorHandler = new OpExecutorHandler(settings, graphManager, gremlinExecutor, scheduledExecutorService);
}
+ /**
+ * It is best not to override this method as it sets up some core parts to the server. Prefer implementing the
+ * {@link #configure(ChannelPipeline)} and {@link #finalize(ChannelPipeline)} methods to alter the pipeline.
+ */
@Override
public void initChannel(final SocketChannel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
@@ -179,6 +187,10 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann
pipeline.addLast(PIPELINE_OP_EXECUTOR, opExecutorHandler);
finalize(pipeline);
+
+ // track the newly created channel in the channel group
+ channels.add(ch);
+
}
protected AbstractAuthenticationHandler createAuthenticationHandler(final Settings settings) {
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/GremlinServer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/GremlinServer.java
index f2568c8187..565a471e40 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/GremlinServer.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/GremlinServer.java
@@ -21,7 +21,6 @@ package org.apache.tinkerpop.gremlin.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
@@ -34,13 +33,13 @@ import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Slf4JLoggerFactory;
import org.apache.commons.lang3.SystemUtils;
+import org.apache.tinkerpop.gremlin.server.handler.FrameMessageSizeEstimator;
import org.apache.tinkerpop.gremlin.server.op.OpLoader;
import org.apache.tinkerpop.gremlin.server.util.LifeCycleHook;
import org.apache.tinkerpop.gremlin.server.util.MetricManager;
import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor;
import org.apache.tinkerpop.gremlin.server.util.ThreadFactoryUtil;
import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
import org.apache.tinkerpop.gremlin.util.Gremlin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,7 +70,11 @@ public class GremlinServer {
private static final Logger logger = LoggerFactory.getLogger(GremlinServer.class);
private final Settings settings;
- private Channel ch;
+
+ /**
+ * The {@code ServerSocketChannel} established when binding the server to the host/port.
+ */
+ private Channel serverSocketChannel;
private CompletableFuture<Void> serverStopped = null;
private CompletableFuture<ServerGremlinExecutor> serverStarted = null;
@@ -145,6 +148,9 @@ public class GremlinServer {
try {
final ServerBootstrap b = new ServerBootstrap();
+ // need a custom MessageSizeEstimator to cover Frame
+ b.childOption(ChannelOption.MESSAGE_SIZE_ESTIMATOR, FrameMessageSizeEstimator.instance());
+
// when high value is reached then the channel becomes non-writable and stays like that until the
// low value is so that there is time to recover
b.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
@@ -173,22 +179,19 @@ public class GremlinServer {
b.channel(NioServerSocketChannel.class);
}
- // bind to host/port and wait for channel to be ready
- b.bind(settings.host, settings.port).addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(final ChannelFuture channelFuture) throws Exception {
- if (channelFuture.isSuccess()) {
- ch = channelFuture.channel();
-
- logger.info("Gremlin Server configured with worker thread pool of {}, gremlin pool of {} and boss thread pool of {}.",
- settings.threadPoolWorker, settings.gremlinPool, settings.threadPoolBoss);
- logger.info("Channel started at port {}.", settings.port);
-
- serverReadyFuture.complete(serverGremlinExecutor);
- } else {
- serverReadyFuture.completeExceptionally(new IOException(
- String.format("Could not bind to %s and %s - perhaps something else is bound to that address.", settings.host, settings.port)));
- }
+ // bind to host/port and wait for ServerSocketChannel to be ready
+ b.bind(settings.host, settings.port).addListener((ChannelFutureListener) channelFuture -> {
+ if (channelFuture.isSuccess()) {
+ serverSocketChannel = channelFuture.channel();
+
+ logger.info("Gremlin Server configured with worker thread pool of {}, gremlin pool of {} and boss thread pool of {}.",
+ settings.threadPoolWorker, settings.gremlinPool, settings.threadPoolBoss);
+ logger.info("Channel started at port {}.", settings.port);
+
+ serverReadyFuture.complete(serverGremlinExecutor);
+ } else {
+ serverReadyFuture.completeExceptionally(new IOException(
+ String.format("Could not bind to %s and %s - perhaps something else is bound to that address.", settings.host, settings.port)));
}
});
} catch (Exception ex) {
@@ -249,10 +252,10 @@ public class GremlinServer {
// it's possible that a channel might not be initialized in the first place if bind() fails because
// of port conflict. in that case, there's no need to wait for the channel to close.
- if (null == ch)
+ if (null == serverSocketChannel)
servicesLeftToShutdown.countDown();
else
- ch.close().addListener(f -> servicesLeftToShutdown.countDown());
+ serverSocketChannel.close().addListener(f -> servicesLeftToShutdown.countDown());
logger.info("Shutting down thread pools.");
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java
index 7c04e9eb02..6680127d51 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractSession.java
@@ -80,6 +80,8 @@ import java.util.stream.Stream;
import static org.apache.tinkerpop.gremlin.process.traversal.GraphOp.TX_COMMIT;
import static org.apache.tinkerpop.gremlin.process.traversal.GraphOp.TX_ROLLBACK;
+import static org.apache.tinkerpop.gremlin.server.op.AbstractOpProcessor.WRITE_PAUSE_TIME_MS;
+import static org.apache.tinkerpop.gremlin.server.op.AbstractOpProcessor.writePausesMeter;
/**
* A base implementation of {@link Session} which offers some common functionality that matches typical Gremlin Server
@@ -129,7 +131,7 @@ public abstract class AbstractSession implements Session, AutoCloseable {
/**
* The session was interrupted by the channel closing, which can be something initiated by closing the
- * {@link Client} or might be triggered by the server. This may not be considered an error situation and
+ * {@code Client} or might be triggered by the server. This may not be considered an error situation and
* depending on context, might be similar to a {@link #EXIT_PROCESSING} termination.
*/
CHANNEL_CLOSED,
@@ -530,7 +532,11 @@ public abstract class AbstractSession implements Session, AutoCloseable {
final ChannelHandlerContext nettyContext = sessionTask.getChannelHandlerContext();
final RequestMessage msg = sessionTask.getRequestMessage();
final Settings settings = sessionTask.getSettings();
- boolean warnOnce = false;
+
+ // used to limit warnings for when netty fills the buffer and hits the high watermark - prevents
+ // over-logging of the same message.
+ long lastWarningTime = 0;
+ int warnCounter = 0;
// sessionless requests are always transaction managed, but in-session requests are configurable.
final boolean managedTransactionsForRequest = transactionManaged ?
@@ -650,15 +656,28 @@ public abstract class AbstractSession implements Session, AutoCloseable {
sessionTask.writeAndFlush(code, frame);
}
} else {
- // don't keep triggering this warning over and over again for the same request
- if (!warnOnce) {
- logger.warn("Pausing response writing as writeBufferHighWaterMark exceeded on {} - writing will continue once client has caught up", msg);
- warnOnce = true;
+ final long currentTime = System.currentTimeMillis();
+
+ // exponential delay between warnings. don't keep triggering this warning over and over again for the
+ // same request. totalPendingWriteBytes is volatile so it is possible that by the time this warning
+ // hits the log the low watermark may have been hit
+ long interval = (long) Math.pow(2, warnCounter) * 1000;
+ if (currentTime - lastWarningTime >= interval) {
+ final Channel ch = nettyContext.channel();
+ logger.warn("Warning {}: Outbound buffer size={}, pausing response writing as writeBufferHighWaterMark exceeded on request {} for channel {} - writing will continue once client has caught up",
+ warnCounter,
+ ch.unsafe().outboundBuffer().totalPendingWriteBytes(),
+ msg.getRequestId(),
+ ch.id());
+
+ lastWarningTime = currentTime;
+ warnCounter++;
}
// since the client is lagging we can hold here for a period of time for the client to catch up.
// this isn't blocking the IO thread - just a worker.
- TimeUnit.MILLISECONDS.sleep(10);
+ TimeUnit.MILLISECONDS.sleep(WRITE_PAUSE_TIME_MS);
+ writePausesMeter.mark();
}
}
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/FrameMessageSizeEstimator.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/FrameMessageSizeEstimator.java
new file mode 100644
index 0000000000..d60be6e578
--- /dev/null
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/FrameMessageSizeEstimator.java
@@ -0,0 +1,93 @@
+/*
+ * 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.ByteBufHolder;
+import io.netty.channel.FileRegion;
+import io.netty.channel.MessageSizeEstimator;
+
+import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
+
+/**
+ * A {@code MessageSizeEstimator} that sizes the {@link Frame} class. This is basically an implementation of netty's
+ * default but with that additional feature.
+ */
+public final class FrameMessageSizeEstimator implements MessageSizeEstimator {
+
+ private static final class HandleImpl implements Handle {
+ private final int unknownSize;
+
+ private HandleImpl(int unknownSize) {
+ this.unknownSize = unknownSize;
+ }
+
+ @Override
+ public int size(Object msg) {
+ if (msg instanceof Frame) {
+ return size0(((Frame) msg).getMsg());
+ } else {
+ return size0(msg);
+ }
+ }
+
+ /**
+ * Standard netty implementation.
+ */
+ private int size0(final Object msg) {
+ if (msg instanceof ByteBuf) {
+ return ((ByteBuf) msg).readableBytes();
+ }
+ if (msg instanceof ByteBufHolder) {
+ return ((ByteBufHolder) msg).content().readableBytes();
+ }
+ if (msg instanceof FileRegion) {
+ return 0;
+ }
+ return unknownSize;
+ }
+ }
+
+ /**
+ * Return the default implementation which returns {@code 8} for unknown messages which matches the
+ * implementation of netty at {@code DefaultMessageSizeEstimator}
+ */
+ private static final FrameMessageSizeEstimator instance = new FrameMessageSizeEstimator(8);
+
+ private final Handle handle;
+
+ /**
+ * Create a new instance
+ *
+ * @param unknownSize The size which is returned for unknown messages.
+ */
+ public FrameMessageSizeEstimator(int unknownSize) {
+ checkPositiveOrZero(unknownSize, "unknownSize");
+ handle = new HandleImpl(unknownSize);
+ }
+
+ public static FrameMessageSizeEstimator instance() {
+ return instance;
+ }
+
+ @Override
+ public Handle newHandle() {
+ return handle;
+ }
+}
\ No newline at end of file
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java
index 49b9f6f1f4..c071780284 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractOpProcessor.java
@@ -18,8 +18,12 @@
*/
package org.apache.tinkerpop.gremlin.server.op;
+import com.codahale.metrics.Meter;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.tinkerpop.gremlin.server.GremlinServer;
+import org.apache.tinkerpop.gremlin.server.util.MetricManager;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
import org.apache.tinkerpop.gremlin.util.Tokens;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
@@ -49,6 +53,8 @@ import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
+import static com.codahale.metrics.MetricRegistry.name;
+
/**
* A base {@link OpProcessor} implementation that processes an {@code Iterator} of results in a generalized way while
* ensuring that graph transactions are properly managed.
@@ -58,6 +64,16 @@ import java.util.stream.Stream;
public abstract class AbstractOpProcessor implements OpProcessor {
private static final Logger logger = LoggerFactory.getLogger(AbstractEvalOpProcessor.class);
+ /**
+ * Length of time to pause writes in milliseconds when the high watermark is exceeded.
+ */
+ public static final long WRITE_PAUSE_TIME_MS = 10;
+
+ /**
+ * Tracks the rate of pause to writes when the high watermark is exceeded.
+ */
+ public static final Meter writePausesMeter = MetricManager.INSTANCE.getMeter(name(GremlinServer.class, "channels", "write-pauses"));
+
/**
* When set to {@code true}, transactions are always managed otherwise they can be overridden by the request.
*/
@@ -88,7 +104,11 @@ public abstract class AbstractOpProcessor implements OpProcessor {
final Settings settings = context.getSettings();
final MessageSerializer<?> serializer = nettyContext.channel().attr(StateKey.SERIALIZER).get();
final boolean useBinary = nettyContext.channel().attr(StateKey.USE_BINARY).get();
- boolean warnOnce = false;
+
+ // used to limit warnings for when netty fills the buffer and hits the high watermark - prevents
+ // over-logging of the same message.
+ long lastWarningTime = 0;
+ int warnCounter = 0;
// sessionless requests are always transaction managed, but in-session requests are configurable.
final boolean managedTransactionsForRequest = manageTransactions ?
@@ -213,15 +233,28 @@ public abstract class AbstractOpProcessor implements OpProcessor {
context.writeAndFlush(code, frame);
}
} else {
- // don't keep triggering this warning over and over again for the same request
- if (!warnOnce) {
- logger.warn("Pausing response writing as writeBufferHighWaterMark exceeded on {} - writing will continue once client has caught up", msg);
- warnOnce = true;
+ final long currentTime = System.currentTimeMillis();
+
+ // exponential delay between warnings. don't keep triggering this warning over and over again for the
+ // same request. totalPendingWriteBytes is volatile so it is possible that by the time this warning
+ // hits the log the low watermark may have been hit
+ long interval = (long) Math.pow(2, warnCounter) * 1000;
+ if (currentTime - lastWarningTime >= interval) {
+ final Channel ch = context.getChannelHandlerContext().channel();
+ logger.warn("Warning {}: Outbound buffer size={}, pausing response writing as writeBufferHighWaterMark exceeded on request {} for channel {} - writing will continue once client has caught up",
+ warnCounter,
+ ch.unsafe().outboundBuffer().totalPendingWriteBytes(),
+ msg.getRequestId(),
+ ch.id());
+
+ lastWarningTime = currentTime;
+ warnCounter++;
}
// since the client is lagging we can hold here for a period of time for the client to catch up.
// this isn't blocking the IO thread - just a worker.
- TimeUnit.MILLISECONDS.sleep(10);
+ TimeUnit.MILLISECONDS.sleep(WRITE_PAUSE_TIME_MS);
+ writePausesMeter.mark();
}
}
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
index 2c02c65761..17a76e740e 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.server.op.session;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
import org.apache.tinkerpop.gremlin.util.Tokens;
@@ -96,7 +97,7 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
protected static ConcurrentHashMap<String, Session> sessions = new ConcurrentHashMap<>();
static {
- MetricManager.INSTANCE.getGuage(sessions::size, name(GremlinServer.class, "sessions"));
+ MetricManager.INSTANCE.getGauge(sessions::size, name(GremlinServer.class, "sessions"));
}
/**
@@ -608,7 +609,11 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
final Settings settings = context.getSettings();
final MessageSerializer serializer = nettyContext.channel().attr(StateKey.SERIALIZER).get();
final boolean useBinary = nettyContext.channel().attr(StateKey.USE_BINARY).get();
- boolean warnOnce = false;
+
+ // used to limit warnings for when netty fills the buffer and hits the high watermark - prevents
+ // over-logging of the same message.
+ long lastWarningTime = 0;
+ int warnCounter = 0;
// we have an empty iterator - happens on stuff like: g.V().iterate()
if (!itty.hasNext()) {
@@ -728,15 +733,28 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
context.writeAndFlush(code, frame);
}
} else {
- // don't keep triggering this warning over and over again for the same request
- if (!warnOnce) {
- logger.warn("Pausing response writing as writeBufferHighWaterMark exceeded on {} - writing will continue once client has caught up", msg);
- warnOnce = true;
+ final long currentTime = System.currentTimeMillis();
+
+ // exponential delay between warnings. don't keep triggering this warning over and over again for the
+ // same request. totalPendingWriteBytes is volatile so it is possible that by the time this warning
+ // hits the log the low watermark may have been hit
+ long interval = (long) Math.pow(2, warnCounter) * 1000;
+ if (currentTime - lastWarningTime >= interval) {
+ final Channel ch = context.getChannelHandlerContext().channel();
+ logger.warn("Warning {}: Outbound buffer size={}, pausing response writing as writeBufferHighWaterMark exceeded on request {} for channel {} - writing will continue once client has caught up",
+ warnCounter,
+ ch.unsafe().outboundBuffer().totalPendingWriteBytes(),
+ msg.getRequestId(),
+ ch.id());
+
+ lastWarningTime = currentTime;
+ warnCounter++;
}
// since the client is lagging we can hold here for a period of time for the client to catch up.
// this isn't blocking the IO thread - just a worker.
- TimeUnit.MILLISECONDS.sleep(10);
+ TimeUnit.MILLISECONDS.sleep(WRITE_PAUSE_TIME_MS);
+ writePausesMeter.mark();
}
}
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
index b8038528d3..4698c81212 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
@@ -19,6 +19,7 @@
package org.apache.tinkerpop.gremlin.server.op.traversal;
import com.codahale.metrics.Timer;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelHandlerContext;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
@@ -344,7 +345,11 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
final Settings settings = context.getSettings();
final MessageSerializer<?> serializer = nettyContext.channel().attr(StateKey.SERIALIZER).get();
final boolean useBinary = nettyContext.channel().attr(StateKey.USE_BINARY).get();
- boolean warnOnce = false;
+
+ // used to limit warnings for when netty fills the buffer and hits the high watermark - prevents
+ // over-logging of the same message.
+ long lastWarningTime = 0;
+ int warnCounter = 0;
// we have an empty iterator - happens on stuff like: g.V().iterate()
if (!itty.hasNext()) {
@@ -464,15 +469,28 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
context.writeAndFlush(code, frame);
}
} else {
- // don't keep triggering this warning over and over again for the same request
- if (!warnOnce) {
- logger.warn("Pausing response writing as writeBufferHighWaterMark exceeded on {} - writing will continue once client has caught up", msg);
- warnOnce = true;
+ final long currentTime = System.currentTimeMillis();
+
+ // exponential delay between warnings. don't keep triggering this warning over and over again for the
+ // same request. totalPendingWriteBytes is volatile so it is possible that by the time this warning
+ // hits the log the low watermark may have been hit
+ long interval = (long) Math.pow(2, warnCounter) * 1000;
+ if (currentTime - lastWarningTime >= interval) {
+ final Channel ch = context.getChannelHandlerContext().channel();
+ logger.warn("Warning {}: Outbound buffer size={}, pausing response writing as writeBufferHighWaterMark exceeded on request {} for channel {} - writing will continue once client has caught up",
+ warnCounter,
+ ch.unsafe().outboundBuffer().totalPendingWriteBytes(),
+ msg.getRequestId(),
+ ch.id());
+
+ lastWarningTime = currentTime;
+ warnCounter++;
}
// since the client is lagging we can hold here for a period of time for the client to catch up.
// this isn't blocking the IO thread - just a worker.
- TimeUnit.MILLISECONDS.sleep(10);
+ TimeUnit.MILLISECONDS.sleep(WRITE_PAUSE_TIME_MS);
+ writePausesMeter.mark();
}
}
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/MetricManager.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/MetricManager.java
index f07572d51a..69bde41eca 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/MetricManager.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/MetricManager.java
@@ -408,11 +408,11 @@ public enum MetricManager {
return getRegistry().counter(MetricRegistry.name(prefix, names));
}
- public <T> Gauge<T> getGuage(final Gauge<T> gauge, final String name) {
+ public <T> Gauge<T> getGauge(final Gauge<T> gauge, final String name) {
return getRegistry().register(name, gauge);
}
- public <T> Gauge<T> getGuage(final Gauge<T> gauge, final String prefix, final String... names) {
+ public <T> Gauge<T> getGauge(final Gauge<T> gauge, final String prefix, final String... names) {
return getRegistry().register(MetricRegistry.name(prefix, names), gauge);
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java
index 956bb0371e..570af9466b 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/util/ServerGremlinExecutor.java
@@ -18,9 +18,11 @@
*/
package org.apache.tinkerpop.gremlin.server.util;
+import com.codahale.metrics.Gauge;
+import io.netty.channel.Channel;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinLangScriptEngine;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinLangScriptEngineFactory;
import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngine;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
import org.apache.tinkerpop.gremlin.server.Channelizer;
@@ -47,6 +49,8 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import static com.codahale.metrics.MetricRegistry.name;
+
/**
* The core of script execution in Gremlin Server. Given {@link Settings} and optionally other arguments, this
* class will construct a {@link GremlinExecutor} to be used by Gremlin Server. A typical usage would be to
@@ -67,9 +71,22 @@ public class ServerGremlinExecutor {
private final ScheduledExecutorService scheduledExecutorService;
private final ExecutorService gremlinExecutorService;
private final GremlinExecutor gremlinExecutor;
+ private final DefaultChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
private final Map<String,Object> hostOptions = new ConcurrentHashMap<>();
+ /**
+ * The total number of channels.
+ */
+ public final Gauge channelsTotalGauge = MetricManager.INSTANCE.getGauge(
+ this::getChannelCount, name(GremlinServer.class, "channels", "total"));
+
+ /**
+ * The total number of paused channels.
+ */
+ public final Gauge channelsPausedGauge = MetricManager.INSTANCE.getGauge(
+ this::getPausedChannelCount, name(GremlinServer.class, "channels", "paused"));
+
/**
* Create a new object from {@link Settings} where thread pools are externally assigned. Note that if the
* {@code scheduleExecutorServiceClass} is set to {@code null} it will be created via
@@ -226,4 +243,26 @@ public class ServerGremlinExecutor {
public List<LifeCycleHook> getHooks() {
return hooks;
}
+
+ /**
+ * Gets all open channels (excluding the {@code ServerSocketChannel}).
+ */
+ public DefaultChannelGroup getChannels() {
+ return channels;
+ }
+
+ /**
+ * Gets the number of open channels currently in a paused state that are not writeable.
+ */
+ public long getPausedChannelCount() {
+ return channels.stream().filter(ch -> !ch.isWritable()).count();
+ }
+
+ /**
+ * Gets the number of open channels.
+ */
+ public long getChannelCount() {
+ return channels.size();
+ }
+
}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index 0458c86cc8..0078b43d41 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
@@ -25,6 +25,8 @@ import io.netty.util.AttributeKey;
import nl.altindag.log.LogCaptor;
import org.apache.commons.configuration2.BaseConfiguration;
import org.apache.commons.configuration2.Configuration;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.tinkerpop.gremlin.server.channel.HttpTestChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.TestChannelizer;
import org.apache.tinkerpop.gremlin.server.channel.UnifiedChannelizer;
@@ -691,23 +693,25 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
// the highwatermark should get exceeded on the server and thus pause the writes, but have no problem catching
// itself up - this is a tricky tests to get passing on all environments so this assumption will deny the
// test for most cases
- TestHelper.assumeNonDeterministic();
+ //TestHelper.assumeNonDeterministic();
final Cluster cluster = TestClientFactory.open();
final Client client = cluster.connect();
try {
- final int resultCountToGenerate = 1000;
- final int batchSize = 3;
- final String fatty = IntStream.range(0, 175).mapToObj(String::valueOf).collect(Collectors.joining());
- final String fattyX = "['" + fatty + "'] * " + resultCountToGenerate;
+ final int resultCountToGenerate = 300;
+ final int batchSize = 1;
+ final String fatty = RandomStringUtils.random(128);
+ final String fattyX = "[x] * " + resultCountToGenerate;
// don't allow the thread to proceed until all results are accounted for
final CountDownLatch latch = new CountDownLatch(resultCountToGenerate);
final AtomicBoolean expected = new AtomicBoolean(false);
final AtomicBoolean faulty = new AtomicBoolean(false);
final RequestMessage request = RequestMessage.build(Tokens.OPS_EVAL)
+ .addArg(ARGS_EVAL_TIMEOUT, 10 * 60 * 1000)
.addArg(Tokens.ARGS_BATCH_SIZE, batchSize)
+ .addArg(Tokens.ARGS_BINDINGS, new HashMap<String,Object>() {{ put("x", fatty); }})
.addArg(Tokens.ARGS_GREMLIN, fattyX).create();
client.submitAsync(request).thenAcceptAsync(r -> {
@@ -730,7 +734,7 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
assertThat(expected.get(), is(true));
assertThat(logCaptor.getLogs().stream().anyMatch(m -> m.contains(
- "Pausing response writing as writeBufferHighWaterMark exceeded on")), is(true));
+ "pausing response writing as writeBufferHighWaterMark exceeded on")), is(true));
} catch (Exception ex) {
fail("Shouldn't have tossed an exception");
} finally {