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 2017/09/13 10:54:26 UTC
[4/4] tinkerpop git commit: TINKERPOP-1770 Enable timeouts for remote
traversals
TINKERPOP-1770 Enable timeouts for remote traversals
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/63191aef
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/63191aef
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/63191aef
Branch: refs/heads/TINKERPOP-1770
Commit: 63191aef22e6cce2f052cb1c8fa898d18a07615b
Parents: c59393f
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Thu Sep 7 09:15:37 2017 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Wed Sep 13 06:54:10 2017 -0400
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../upgrade/release-3.2.x-incubating.asciidoc | 20 ++++-
.../op/traversal/TraversalOpProcessor.java | 84 ++++++++++----------
.../server/GremlinServerIntegrateTest.java | 34 ++++++++
4 files changed, 93 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/63191aef/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index c15835c..70d6134 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -28,6 +28,7 @@ TinkerPop 3.2.7 (Release Date: NOT OFFICIALLY RELEASED YET)
* Bump to Jackson 2.8.10.
* Added an `EmbeddedRemoteConnection` so that it's possible to mimic a remote connection within the same JVM.
+* Supported interruption for remote traversals.
* The Console's `plugin.txt` file is only updated if there were manually uninstalled plugins.
* Fixed a bug in `MatchStep` where mid-traversal `where()` variables were not being considered in start-scope.
* Generalized `MatchStep` to locally compute all clauses with barriers (not just reducing barriers).
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/63191aef/docs/src/upgrade/release-3.2.x-incubating.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/upgrade/release-3.2.x-incubating.asciidoc b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
index 407cc89..ae504d4 100644
--- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
@@ -41,10 +41,22 @@ simple way to provide a "remote" that is actually local to the same JVM.
See: link:https://issues.apache.org/jira/browse/TINKERPOP-1756[TINKERPOP-1756]
-Changes to match()
-~~~~~~~~~~~~~~~~~~
-The `match()`-step has been generalized to suppor the local scoping of all barrier steps, not just reducing barrier steps.
+Remote Traversal Timeout
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+There was limited support for "timeouts" with remote traversals (i.e. those traversals executed using the `withRemote()`
+option) prior to 3.2.7. Remote traversals will now interrupt on the server using the `scriptEvaluationTimeout`
+setting in the same way that normal script evaluations would. As a reminder, interruptions for traversals are always
+considered "attempts to interrupt" and may not always succeed (a graph database implementation might not respect the
+interruption, for example).
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-1770[TINKERPOP-1770]
+
+Modifications to match()
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+The `match()`-step has been generalized to support the local scoping of all barrier steps, not just reducing barrier steps.
Previously, the `order().limit()` clause would have worked globally yielding:
[source,groovy]
@@ -74,6 +86,8 @@ This includes steps like `count()`, `min()`, `max()`, `sum()`, `group()`, `group
generalized this behavior to all barriers and thus, adds `aggregate()`, `dedup()`, `range()`, `limit()`, `tail()`, and `order()`
to the list of locally computed clauses.
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-1764[TINKERPOP-1764]
+
TinkerPop 3.2.6
---------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/63191aef/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java
----------------------------------------------------------------------
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 4e35a85..fc3066a 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
@@ -35,6 +35,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.util.BytecodeHelper;
+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;
@@ -56,6 +57,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.script.SimpleBindings;
+import java.lang.ref.WeakReference;
+import java.lang.reflect.UndeclaredThrowableException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -64,6 +67,9 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -347,6 +353,10 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
// earlier validation in selection of this op method should free us to cast this without worry
final Map<String, String> aliases = (Map<String, String>) msg.optionalArgs(Tokens.ARGS_ALIASES).get();
+ // timeout override
+ final long seto = msg.getArgs().containsKey(Tokens.ARGS_SCRIPT_EVAL_TIMEOUT) ?
+ Long.parseLong(msg.getArgs().get(Tokens.ARGS_SCRIPT_EVAL_TIMEOUT).toString()) : context.getSettings().scriptEvaluationTimeout;
+
final GraphManager graphManager = context.getGraphManager();
final String traversalSourceName = aliases.entrySet().iterator().next().getValue();
final TraversalSource g = graphManager.getTraversalSource(traversalSourceName);
@@ -370,51 +380,54 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
}
final Timer.Context timerContext = traversalOpTimer.time();
- try {
+ final FutureTask<Void> evalFuture = new FutureTask<>(() -> {
final ChannelHandlerContext ctx = context.getChannelHandlerContext();
final Graph graph = g.getGraph();
- context.getGremlinExecutor().getExecutorService().submit(() -> {
+ try {
+ beforeProcessing(graph, context);
+
try {
- beforeProcessing(graph, context);
+ // compile the traversal - without it getEndStep() has nothing in it
+ traversal.applyStrategies();
+ handleIterator(context, new TraverserIterator(traversal), graph);
+ } catch (Exception ex) {
+ Throwable t = ex;
+ if (ex instanceof UndeclaredThrowableException)
+ t = t.getCause();
- try {
- // compile the traversal - without it getEndStep() has nothing in it
- traversal.applyStrategies();
- handleIterator(context, new TraverserIterator(traversal), graph);
- } catch (TimeoutException ex) {
- final String errorMessage = String.format("Response iteration exceeded the configured threshold for request [%s] - %s", msg.getRequestId(), ex.getMessage());
+ if (t instanceof InterruptedException || t instanceof TraversalInterruptedException) {
+ final String errorMessage = String.format("A timeout occurred during traversal evaluation of [%s] - consider increasing the limit given to scriptEvaluationTimeout", msg);
logger.warn(errorMessage);
ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage(errorMessage)
.statusAttributeException(ex).create());
onError(graph, context);
- return;
- } catch (Exception ex) {
+ } else {
logger.warn(String.format("Exception processing a Traversal on iteration for request [%s].", msg.getRequestId()), ex);
ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(ex.getMessage())
.statusAttributeException(ex).create());
onError(graph, context);
- return;
}
- } catch (Exception ex) {
- logger.warn(String.format("Exception processing a Traversal on request [%s].", msg.getRequestId()), ex);
- ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
- .statusMessage(ex.getMessage())
- .statusAttributeException(ex).create());
- onError(graph, context);
- } finally {
- timerContext.stop();
}
- });
+ } catch (Exception ex) {
+ logger.warn(String.format("Exception processing a Traversal on request [%s].", msg.getRequestId()), ex);
+ ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
+ .statusMessage(ex.getMessage())
+ .statusAttributeException(ex).create());
+ onError(graph, context);
+ } finally {
+ timerContext.stop();
+ }
- } catch (Exception ex) {
- timerContext.stop();
- throw new OpProcessorException("Could not iterate the Traversal instance",
- ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
- .statusMessage(ex.getMessage())
- .statusAttributeException(ex).create());
+ return null;
+ });
+
+ final Future<?> executionFuture = context.getGremlinExecutor().getExecutorService().submit(evalFuture);
+ if (seto > 0) {
+ // Schedule a timeout in the thread pool for future execution
+ context.getScheduledExecutorService().schedule(() -> executionFuture.cancel(true), seto, TimeUnit.MILLISECONDS);
}
}
@@ -463,7 +476,7 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
return metaData;
}
- protected void handleIterator(final Context context, final Iterator itty, final Graph graph) throws TimeoutException, InterruptedException {
+ protected void handleIterator(final Context context, final Iterator itty, final Graph graph) throws InterruptedException {
final ChannelHandlerContext ctx = context.getChannelHandlerContext();
final RequestMessage msg = context.getRequestMessage();
final Settings settings = context.getSettings();
@@ -483,10 +496,6 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
return;
}
- // timer for the total serialization time
- final StopWatch stopWatch = new StopWatch();
- stopWatch.start();
-
// the batch size can be overridden by the request
final int resultIterationBatchSize = (Integer) msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
.orElse(settings.resultIterationBatchSize);
@@ -585,17 +594,6 @@ public class TraversalOpProcessor extends AbstractOpProcessor {
// this isn't blocking the IO thread - just a worker.
TimeUnit.MILLISECONDS.sleep(10);
}
-
- stopWatch.split();
- if (settings.serializedResponseTimeout > 0 && stopWatch.getSplitTime() > settings.serializedResponseTimeout) {
- final String timeoutMsg = String.format("Serialization of the entire response exceeded the 'serializeResponseTimeout' setting %s",
- warnOnce ? "[Gremlin Server paused writes to client as messages were not being consumed quickly enough]" : "");
- throw new TimeoutException(timeoutMsg.trim());
- }
-
- stopWatch.unsplit();
}
-
- stopWatch.stop();
}
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/63191aef/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
----------------------------------------------------------------------
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 8b0c280..c401de6 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
@@ -34,6 +34,7 @@ import org.apache.tinkerpop.gremlin.driver.Cluster;
import org.apache.tinkerpop.gremlin.driver.Result;
import org.apache.tinkerpop.gremlin.driver.ResultSet;
import org.apache.tinkerpop.gremlin.driver.Tokens;
+import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseStatusCode;
@@ -51,6 +52,7 @@ import org.apache.tinkerpop.gremlin.process.remote.RemoteGraph;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
import org.apache.tinkerpop.gremlin.server.op.AbstractEvalOpProcessor;
import org.apache.tinkerpop.gremlin.server.op.standard.StandardOpProcessor;
@@ -229,6 +231,9 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
settings.processors.clear();
settings.processors.add(processorSettingsSmall);
break;
+ case "shouldTimeOutRemoteTraversal":
+ settings.scriptEvaluationTimeout = 500;
+ break;
}
return settings;
@@ -289,6 +294,35 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
}
@Test
+ public void shouldTimeOutRemoteTraversal() throws Exception {
+ final Graph graph = EmptyGraph.instance();
+ final GraphTraversalSource g = graph.traversal().withRemote(conf);
+
+ try {
+ // tests sleeping thread
+ g.inject(1).sideEffect(Lambda.consumer("Thread.sleep(10000)")).iterate();
+ fail("This traversal should have timed out");
+ } catch (Exception ex) {
+ final Throwable t = ex.getCause();
+ assertThat(t, instanceOf(ResponseException.class));
+ assertEquals(ResponseStatusCode.SERVER_ERROR_TIMEOUT, ((ResponseException) t).getResponseStatusCode());
+ }
+
+ // make a graph with a cycle in it to force a long run traversal
+ graphGetter.get().traversal().addV("person").as("p").addE("self").to("p").iterate();
+
+ try {
+ // tests an "unending" traversal
+ g.V().repeat(__.out()).until(__.outE().count().is(0)).iterate();
+ fail("This traversal should have timed out");
+ } catch (Exception ex) {
+ final Throwable t = ex.getCause();
+ assertThat(t, instanceOf(ResponseException.class));
+ assertEquals(ResponseStatusCode.SERVER_ERROR_TIMEOUT, ((ResponseException) t).getResponseStatusCode());
+ }
+ }
+
+ @Test
public void shouldUseBaseScript() throws Exception {
final Cluster cluster = TestClientFactory.open();
final Client client = cluster.connect(name.getMethodName());