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/10 15:27:41 UTC
(tinkerpop) 02/02: Merge branch '3.7-dev'
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit cf4218620729f764ec5498bb214c0c31d97d4450
Merge: e093151527 9627b78bcf
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Fri Jan 10 10:27:12 2025 -0500
Merge branch '3.7-dev'
CHANGELOG.asciidoc | 5 ++-
docs/src/reference/gremlin-applications.asciidoc | 52 ++++++++++++++++++----
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 | 41 +++++++++--------
.../server/handler/HttpGremlinEndpointHandler.java | 41 ++++++++++++++---
.../gremlin/server/util/MetricManager.java | 12 +++--
.../gremlin/server/util/ServerGremlinExecutor.java | 43 +++++++++++++++++-
.../gremlin/server/GremlinServerIntegrateTest.java | 6 ++-
10 files changed, 218 insertions(+), 60 deletions(-)
diff --cc docs/src/reference/gremlin-applications.asciidoc
index 75e3b8c893,876942f1cf..c7262d7447
--- a/docs/src/reference/gremlin-applications.asciidoc
+++ b/docs/src/reference/gremlin-applications.asciidoc
@@@ -717,6 -925,12 +717,7 @@@ curl -X POST -d "{\"gremlin\":\"100-x\"
By default this value is set to `gremlin-groovy`. If using a `GET` operation, this value can be set as a query
string argument with by setting the `language` key.
-WARNING: Consider the size of the result of a submitted script being returned from the HTTP endpoint. A script
-that iterates thousands of results will serialize each of those in memory into a single JSON result set. It is
-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
@@@ -840,9 -1059,11 +841,9 @@@ The following table describes the vario
|strictTransactionManagement |Set to `true` to require `aliases` to be submitted on every requests, where the `aliases` become the scope of transaction management. |false
|threadPoolBoss |The number of threads available to Gremlin Server for accepting connections. Should always be set to `1`. |1
|threadPoolWorker |The number of threads available to Gremlin Server for processing non-blocking reads and writes. |1
-|useCommonEngineForSessions |Ensures that the same `ScriptEngine` is used to support sessions and sessionless requests which will lead to better performance. Do not change this setting from the default without a specific use case in mind. This configuration only applies to the `UnifiedChannelizer`. |true
|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.
@@@ -985,7 -1274,17 +986,16 @@@ NOTE: Installing Ganglia will include `
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.
+ * `channels.paused` - The current number of open channels (HTTP and Websocket) that 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`.
++"session-id" will be the identifier for the session itself.
+ * `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`.
++requests where "engine-name" will be the actual name of the engine, such as "gremlin-groovy".
* `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
@@@ -993,11 -1292,9 +1003,7 @@@
* `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.
- * `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".
-* `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.
++* `sessions` - The number of sessions open at the time the metric was last measured.
* `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
diff --cc gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
index 8518d7b64f,c962ec4cd6..7e285f21d2
--- 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
@@@ -188,7 -200,14 +198,9 @@@ public class ProfilingApplication
final int maxWaitForConnection = Integer.parseInt(options.getOrDefault("maxWaitForConnection", "3000").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();
+ final String serializer = options.getOrDefault("serializer", Serializers.GRAPHBINARY_V4.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();
diff --cc gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java
index 8a9fd95e2c,c133f68652..a8969fc964
--- 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 +18,8 @@@
*/
package org.apache.tinkerpop.gremlin.server;
++import io.netty.channel.group.ChannelGroup;
+ import io.netty.channel.group.ChannelGroup;
import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
@@@ -103,6 -112,11 +105,8 @@@ public abstract class AbstractChanneliz
protected final Map<String, MessageSerializer<?>> serializers = new HashMap<>();
+ protected ChannelGroup channels;
+
- private OpSelectorHandler opSelectorHandler;
- private OpExecutorHandler opExecutorHandler;
-
protected Authenticator authenticator;
protected Authorizer authorizer;
@@@ -131,8 -154,16 +136,12 @@@
authenticator = createAuthenticator(settings.authentication);
authorizer = createAuthorizer(settings.authorization);
-
- // these handlers don't share any state and can thus be initialized once per pipeline
- opSelectorHandler = new OpSelectorHandler(settings, graphManager, gremlinExecutor, scheduledExecutorService, this);
- 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.
++ * {@link #configure(ChannelPipeline)} methods to alter the pipeline.
+ */
@Override
public void initChannel(final SocketChannel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
@@@ -151,6 -182,15 +160,9 @@@
// pipeline must decode to an incoming RequestMessage instances and encode to a outgoing ResponseMessage
// instance
configure(pipeline);
+
- pipeline.addLast(PIPELINE_OP_SELECTOR, opSelectorHandler);
- 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 --cc gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/GremlinServer.java
index 80e1a816ca,565a471e40..383142753e
--- 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
@@@ -238,12 -240,22 +237,12 @@@ public class GremlinServer
serverStopped = new CompletableFuture<>();
final CountDownLatch servicesLeftToShutdown = new CountDownLatch(3);
- // release resources in the OpProcessors (e.g. kill sessions)
- OpLoader.getProcessors().entrySet().forEach(kv -> {
- logger.info("Shutting down OpProcessor[{}]", kv.getKey());
- try {
- kv.getValue().close();
- } catch (Exception ex) {
- logger.warn("Shutdown will continue but, there was an error encountered while closing " + kv.getKey(), ex);
- }
- });
-
// 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 --cc gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
index ce906977e4,665cf1ef3c..de314d12ed
--- 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
@@@ -18,43 -18,35 +18,45 @@@
*/
package org.apache.tinkerpop.gremlin.server.handler;
++import com.codahale.metrics.Meter;
import com.codahale.metrics.Timer;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
++import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.channel.ChannelPromise;
+import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.TooLongFrameException;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.HttpUtil;
-import io.netty.util.ReferenceCountUtil;
+import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptChecker;
-import org.apache.tinkerpop.gremlin.process.remote.traversal.DefaultRemoteTraverser;
+import org.apache.tinkerpop.gremlin.groovy.jsr223.TimedInterruptTimeoutException;
+import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngine;
+import org.apache.tinkerpop.gremlin.process.traversal.Failure;
+import org.apache.tinkerpop.gremlin.process.traversal.Operator;
+import org.apache.tinkerpop.gremlin.process.traversal.Order;
+import org.apache.tinkerpop.gremlin.process.traversal.Pop;
+import org.apache.tinkerpop.gremlin.process.traversal.Scope;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.AbstractTraverser;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
+import org.apache.tinkerpop.gremlin.server.Context;
import org.apache.tinkerpop.gremlin.server.GraphManager;
import org.apache.tinkerpop.gremlin.server.GremlinServer;
+import org.apache.tinkerpop.gremlin.server.ProcessingException;
import org.apache.tinkerpop.gremlin.server.Settings;
import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
+import org.apache.tinkerpop.gremlin.server.util.GremlinError;
import org.apache.tinkerpop.gremlin.server.util.MetricManager;
-import org.apache.tinkerpop.gremlin.server.util.TextPlainMessageSerializer;
-import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.server.util.TraverserIterator;
+import org.apache.tinkerpop.gremlin.structure.Column;
import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
+import org.apache.tinkerpop.gremlin.structure.T;
+import org.apache.tinkerpop.gremlin.structure.util.TemporaryException;
import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
import org.apache.tinkerpop.gremlin.util.Tokens;
@@@ -119,37 -101,15 +121,47 @@@ public class HttpGremlinEndpointHandle
private static final Timer evalOpTimer = MetricManager.INSTANCE.getTimer(name(GremlinServer.class, "op", "eval"));
+ /**
- * Serializers for the response.
++ * Length of time to pause writes in milliseconds when the high watermark is exceeded.
+ */
- private final Map<String, MessageSerializer<?>> serializers;
++ public static final long WRITE_PAUSE_TIME_MS = 10;
+
+ /**
- * Serializer for {@code text/plain} which is a serializer exclusive to HTTP.
++ * Tracks the rate of pause to writes when the high watermark is exceeded.
+ */
- private static final TextPlainMessageSerializer textPlainSerializer = new TextPlainMessageSerializer();
++ public static final Meter writePausesMeter = MetricManager.INSTANCE.getMeter(name(GremlinServer.class, "channels", "write-pauses"));
++
+ protected static final Set<String> INVALID_BINDINGS_KEYS = new HashSet<>();
+
+ static {
+ INVALID_BINDINGS_KEYS.addAll(Arrays.asList(
+ T.id.name(), T.key.name(),
+ T.label.name(), T.value.name(),
+ T.id.getAccessor(), T.key.getAccessor(),
+ T.label.getAccessor(), T.value.getAccessor(),
+ T.id.getAccessor().toUpperCase(), T.key.getAccessor().toUpperCase(),
+ T.label.getAccessor().toUpperCase(), T.value.getAccessor().toUpperCase()));
+
+ for (Column enumItem : Column.values()) {
+ INVALID_BINDINGS_KEYS.add(enumItem.name());
+ }
+
+ for (Order enumItem : Order.values()) {
+ INVALID_BINDINGS_KEYS.add(enumItem.name());
+ }
+
+ for (Operator enumItem : Operator.values()) {
+ INVALID_BINDINGS_KEYS.add(enumItem.name());
+ }
+
+ for (Scope enumItem : Scope.values()) {
+ INVALID_BINDINGS_KEYS.add(enumItem.name());
+ }
+
+ for (Pop enumItem : Pop.values()) {
+ INVALID_BINDINGS_KEYS.add(enumItem.name());
+ }
+ }
private final GremlinExecutor gremlinExecutor;
private final GraphManager graphManager;
@@@ -341,180 -330,57 +353,197 @@@
}
}
- private Bindings createBindings(final Map<String, Object> bindingMap, final Map<String, String> rebindingMap) {
- final Bindings bindings = new SimpleBindings();
+ private Bindings mergeBindingsFromRequest(final Context ctx, final Bindings bindings) throws ProcessingException {
+ // alias any global bindings to a different variable.
+ final RequestMessage msg = ctx.getRequestMessage();
- // rebind any global bindings to a different variable.
- if (!rebindingMap.isEmpty()) {
- for (Map.Entry<String, String> kv : rebindingMap.entrySet()) {
- boolean found = false;
- final Graph g = this.graphManager.getGraph(kv.getValue());
- if (null != g) {
- bindings.put(kv.getKey(), g);
+ // add any bindings to override any other supplied
+ Optional.ofNullable((Map<String, Object>) msg.getFields().get(Tokens.ARGS_BINDINGS)).ifPresent(bindings::putAll);
+
+ if (msg.getFields().containsKey(Tokens.ARGS_G)) {
+ final String aliased = msg.getField(Tokens.ARGS_G);
+ boolean found = false;
+
+ // first check if the alias refers to a Graph instance
+ final Graph graph = ctx.getGraphManager().getGraph(aliased);
+ if (null != graph) {
+ bindings.put(Tokens.ARGS_G, graph);
+ found = true;
+ }
+
+ // if the alias wasn't found as a Graph then perhaps it is a TraversalSource - it needs to be
+ // something
+ if (!found) {
+ final TraversalSource ts = ctx.getGraphManager().getTraversalSource(aliased);
+ if (null != ts) {
+ bindings.put(Tokens.ARGS_G, ts);
found = true;
}
+ }
- if (!found) {
- final TraversalSource ts = this.graphManager.getTraversalSource(kv.getValue());
- if (null != ts) {
- bindings.put(kv.getKey(), ts);
- found = true;
- }
+ // this validation is important to calls to GraphManager.commit() and rollback() as they both
+ // expect that the aliases supplied are valid
+ if (!found) {
+ throw new ProcessingException(GremlinError.binding(aliased));
+ }
+ }
+
+ return bindings;
+ }
+
+ private void handleIterator(final Context context, final Iterator itty, final MessageSerializer<?> serializer, final boolean bulking) throws InterruptedException {
+ final ChannelHandlerContext nettyContext = context.getChannelHandlerContext();
+ final RequestMessage msg = context.getRequestMessage();
+ final Settings settings = context.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;
+
+ // we have an empty iterator - happens on stuff like: g.V().iterate()
+ if (!itty.hasNext()) {
+ ByteBuf chunk = null;
+ try {
+ chunk = makeChunk(context, serializer, new ArrayList<>(), false, bulking);
+ nettyContext.writeAndFlush(new DefaultHttpContent(chunk));
+ } catch (Exception ex) {
+ // Bytebuf is a countable release - if it does not get written downstream
+ // it needs to be released here
+ if (chunk != null) chunk.release();
+ }
+ sendTrailingHeaders(nettyContext, HttpResponseStatus.OK, "");
+ return;
+ }
+
+ // the batch size can be overridden by the request
+ final int resultIterationBatchSize = (Integer) msg.optionalField(Tokens.ARGS_BATCH_SIZE)
+ .orElse(settings.resultIterationBatchSize);
+ List<Object> aggregate = new ArrayList<>(resultIterationBatchSize);
+
+ // use an external control to manage the loop as opposed to just checking hasNext() in the while. this
+ // prevent situations where auto transactions create a new transaction after calls to commit() withing
+ // the loop on calls to hasNext().
+ boolean hasMore = itty.hasNext();
+
+ while (hasMore) {
+ if (Thread.interrupted()) throw new InterruptedException();
+
+ // have to check the aggregate size because it is possible that the channel is not writeable (below)
+ // so iterating next() if the message is not written and flushed would bump the aggregate size beyond
+ // the expected resultIterationBatchSize. Total serialization time for the response remains in
+ // effect so if the client is "slow" it may simply timeout.
+ //
+ // there is a need to check hasNext() on the iterator because if the channel is not writeable the
+ // previous pass through the while loop will have next()'d the iterator and if it is "done" then a
+ // NoSuchElementException will raise its head. also need a check to ensure that this iteration doesn't
+ // require a forced flush which can be forced by sub-classes.
+ //
+ // this could be placed inside the isWriteable() portion of the if-then below but it seems better to
+ // allow iteration to continue into a batch if that is possible rather than just doing nothing at all
+ // while waiting for the client to catch up
+ if (aggregate.size() < resultIterationBatchSize && itty.hasNext()) {
+ if (bulking) {
+ Traverser traverser = (Traverser) itty.next();
+ aggregate.add(traverser.get());
+ aggregate.add(traverser.bulk());
+ } else {
+ aggregate.add(itty.next());
}
+ }
+
+ // Don't keep executor busy if client has already given up; there is no way to catch up if the channel is
+ // not active, and hence we should break the loop.
+ if (!nettyContext.channel().isActive()) {
+ break;
+ }
+
+ // send back a page of results if batch size is met or if it's the end of the results being iterated.
+ // also check writeability of the channel to prevent OOME for slow clients.
+ //
+ // clients might decide to close the Netty channel to the server with a CloseWebsocketFrame after errors
+ // like CorruptedFrameException. On the server, although the channel gets closed, there might be some
+ // executor threads waiting for watermark to clear which will not clear in these cases since client has
+ // already given up on these requests. This leads to these executors waiting for the client to consume
+ // results till the timeout. checking for isActive() should help prevent that.
+ if (nettyContext.channel().isActive() && nettyContext.channel().isWritable()) {
+ if (aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
+ ByteBuf chunk = null;
+ try {
+ chunk = makeChunk(context, serializer, aggregate, itty.hasNext(), bulking);
+ } catch (Exception ex) {
+ // Bytebuf is a countable release - if it does not get written downstream
+ // it needs to be released here
+ if (chunk != null) chunk.release();
+
+ // exception is handled in makeFrame() - serialization error gets written back to driver
+ // at that point
+ break;
+ }
+
+ // track whether there is anything left in the iterator because it needs to be accessed after
+ // the transaction could be closed - in that case a call to hasNext() could open a new transaction
+ // unintentionally
+ hasMore = itty.hasNext();
+
+ try {
+ // only need to reset the aggregation list if there's more stuff to write
+ if (hasMore) {
+ aggregate = new ArrayList<>(resultIterationBatchSize);
+ }
+ } catch (Exception ex) {
+ // Bytebuf is a countable release - if it does not get written downstream
+ // it needs to be released here
+ if (chunk != null) chunk.release();
+ throw ex;
+ }
- if (!found) {
- final String error = String.format("Could not rebind [%s] to [%s] as [%s] not in the Graph or TraversalSource global bindings",
- kv.getKey(), kv.getValue(), kv.getValue());
- throw new IllegalStateException(error);
+ nettyContext.writeAndFlush(new DefaultHttpContent(chunk));
+
+ if (!hasMore) {
+ sendTrailingHeaders(nettyContext, HttpResponseStatus.OK, "");
+ }
}
+ } 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(),
++ ch.attr(StateKey.REQUEST_ID),
++ 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();
}
}
+ }
- bindings.putAll(bindingMap);
-
- return bindings;
+ /**
+ * Check if any exception in the chain is {@link TemporaryException} or {@link Failure} then respond with the
+ * right error code so that the client knows to retry.
+ */
+ private Optional<Throwable> determineIfSpecialException(final Throwable ex) {
+ return Stream.of(ExceptionUtils.getThrowables(ex)).
+ filter(i -> i instanceof TemporaryException || i instanceof Failure).findFirst();
}
- private Pair<String, MessageTextSerializer<?>> chooseSerializer(final String mimeType) {
- final List<Pair<String, Double>> ordered = Stream.of(mimeType.split(",")).map(mediaType -> {
- // parse out each mediaType with its params - keeping it simple and just looking for "quality". if
- // that value isn't there, default it to 1.0. not really validating here so users better get their
- // accept headers straight
- final Matcher matcher = pattern.matcher(mediaType);
- return (matcher.matches()) ? Pair.with(matcher.group(1), Double.parseDouble(matcher.group(2))) : Pair.with(mediaType, 1.0);
- }).sorted((o1, o2) -> o2.getValue0().compareTo(o1.getValue0())).collect(Collectors.toList());
-
- for (Pair<String, Double> p : ordered) {
- // this isn't perfect as it doesn't really account for wildcards. that level of complexity doesn't seem
- // super useful for gremlin server really.
- final String accept = p.getValue0().equals("*/*") ? "application/json" : p.getValue0();
- if (serializers.containsKey(accept))
- return Pair.with(accept, (MessageTextSerializer<?>) serializers.get(accept));
- else if (accept.equals("text/plain")) {
- return Pair.with(accept, textPlainSerializer);
+ private boolean acceptsDeflateEncoding(List<String> encodings) {
+ for (String encoding : encodings) {
+ if (encoding.contains(DEFLATE.toString())) {
+ return true;
}
}
diff --cc gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index df392480aa,d644a9534d..ab366e5a8d
--- 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
@@@ -24,8 -25,17 +24,10 @@@ 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;
-import org.apache.tinkerpop.gremlin.server.channel.UnifiedTestChannelizer;
-import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer;
-import org.apache.tinkerpop.gremlin.server.channel.WebSocketTestChannelizer;
-import org.apache.tinkerpop.gremlin.server.channel.WsAndHttpTestChannelizer;
+import org.apache.tinkerpop.gremlin.driver.RequestOptions;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.TestHelper;
import org.apache.tinkerpop.gremlin.driver.Client;
import org.apache.tinkerpop.gremlin.driver.Cluster;
@@@ -1008,52 -1402,4 +1010,52 @@@ public class GremlinServerIntegrateTes
final Edge r2_tokens = g.with("materializeProperties", "tokens").E().next();
assertFalse(r2_tokens.properties().hasNext());
}
-
++
+ @Test
+ public void shouldBulkResultsWithClusterOptions() {
+ final Cluster cluster = TestClientFactory.build().bulkResults(true).create();
+ try {
+ final GraphTraversalSource g = traversal().with(DriverRemoteConnection.using(cluster));
+ final List resultBulked = g.with("language", "gremlin-lang").inject(1, 2, 3, 2, 1).toList();
+ assertTrue(cluster.isBulkResultsEnabled());
+ assertEquals(5, resultBulked.size());
+
+ // the result itself should not be affected by bulking through the wire
+ final List result = g.with("language", "gremlin-lang").with("bulkResults", false).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result.size());
+
+ assertTrue(cluster.isBulkResultsEnabled());
+
+ // the result itself should not be affected by bulking through the wire
+ final List resultBulked2 = g.with("language", "gremlin-lang").with("bulkResults", true).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, resultBulked2.size());
+
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldBulkResultsWithRequestOptions() {
+ final Cluster cluster = TestClientFactory.build().create();
+ try {
+ final GraphTraversalSource g = traversal().with(DriverRemoteConnection.using(cluster));
+ final List result = g.inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result.size());
+ assertFalse(cluster.isBulkResultsEnabled());
+
+ // the result itself should not be affected by bulking through the wire
+ final List resultBulked = g.with("language", "gremlin-lang").with("bulkResults", true).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, resultBulked.size());
+
+ assertFalse(cluster.isBulkResultsEnabled());
+
+ // the result itself should not be affected by bulking through the wire
+ final List result2 = g.with("language", "gremlin-lang").with("bulkResults", false).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result2.size());
+
+ } finally {
+ cluster.close();
+ }
+ }
}