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 2016/02/16 18:17:47 UTC
[10/10] incubator-tinkerpop git commit: More fixes for in-session
binding concurrency.
More fixes for in-session binding concurrency.
Tried to again ensure that multiple threads don't act on bindings at the same time. Fixed a logging message for session close that was logging a rollback that may not have actually happened.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/a5a00bf1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/a5a00bf1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/a5a00bf1
Branch: refs/heads/TINKERPOP-1148-wip
Commit: a5a00bf1c42c286b5c1d042003c2521f7dc317d3
Parents: 888ac6a
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon Feb 15 16:23:38 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Feb 16 12:17:14 2016 -0500
----------------------------------------------------------------------
.../server/op/AbstractEvalOpProcessor.java | 49 ++++++++++++--------
.../gremlin/server/op/session/Session.java | 6 ++-
.../server/op/session/SessionOpProcessor.java | 4 +-
.../server/GremlinServerIntegrateTest.java | 27 +++++++++++
4 files changed, 64 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a5a00bf1/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
index 1e37fa0..b8f48eb 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
@@ -46,6 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.script.Bindings;
+import javax.script.SimpleBindings;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
@@ -188,30 +189,40 @@ public abstract class AbstractEvalOpProcessor implements OpProcessor {
final String script = (String) args.get(Tokens.ARGS_GREMLIN);
final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ? (String) args.get(Tokens.ARGS_LANGUAGE) : null;
- final Bindings bindings = bindingsSupplier.get();
+ final Bindings bindings = new SimpleBindings();
// sessionless requests are always transaction managed, but in-session requests are configurable.
final boolean managedTransactionsForRequest = manageTransactions ?
true : (Boolean) args.getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false);
- final CompletableFuture<Object> evalFuture = gremlinExecutor.eval(script, language, bindings, null, o -> {
- final Iterator itty = IteratorUtils.asIterator(o);
-
- logger.debug("Preparing to iterate results from - {} - in thread [{}]", msg, Thread.currentThread().getName());
-
- try {
- handleIterator(context, itty);
- } catch (TimeoutException ex) {
- final String errorMessage = String.format("Response iteration exceeded the configured threshold for request [%s] - %s", msg, ex.getMessage());
- logger.warn(errorMessage);
- ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT).statusMessage(errorMessage).create());
- if (managedTransactionsForRequest) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
- } catch (Exception ex) {
- logger.warn(String.format("Exception processing a script on request [%s].", msg), ex);
- ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR).statusMessage(ex.getMessage()).create());
- if (managedTransactionsForRequest) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
- }
- });
+ final GremlinExecutor.LifeCycle lifeCycle = GremlinExecutor.LifeCycle.build()
+ .beforeEval(b -> {
+ try {
+ b.putAll(bindingsSupplier.get());
+ } catch (OpProcessorException ope) {
+ ope.printStackTrace();
+ }
+ })
+ .withResult(o -> {
+ final Iterator itty = IteratorUtils.asIterator(o);
+
+ logger.debug("Preparing to iterate results from - {} - in thread [{}]", msg, Thread.currentThread().getName());
+
+ try {
+ handleIterator(context, itty);
+ } catch (TimeoutException ex) {
+ final String errorMessage = String.format("Response iteration exceeded the configured threshold for request [%s] - %s", msg, ex.getMessage());
+ logger.warn(errorMessage);
+ ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT).statusMessage(errorMessage).create());
+ if (managedTransactionsForRequest) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
+ } catch (Exception ex) {
+ logger.warn(String.format("Exception processing a script on request [%s].", msg), ex);
+ ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR).statusMessage(ex.getMessage()).create());
+ if (managedTransactionsForRequest) attemptRollback(msg, context.getGraphManager(), settings.strictTransactionManagement);
+ }
+ }).create();
+
+ final CompletableFuture<Object> evalFuture = gremlinExecutor.eval(script, language, bindings, lifeCycle);
evalFuture.handle((v, t) -> {
timerContext.stop();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a5a00bf1/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
index 70fa593..c29b45a 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/Session.java
@@ -122,8 +122,10 @@ public class Session {
// that thread of execution from this session
try {
executor.submit(() -> {
- logger.info("Rolling back open transactions on {} before killing session: {}", kv.getKey(), session);
- if (g.tx().isOpen()) g.tx().rollback();
+ if (g.tx().isOpen()) {
+ logger.info("Rolling back open transactions on {} before killing session: {}", kv.getKey(), session);
+ g.tx().rollback();
+ }
}).get(30000, TimeUnit.MILLISECONDS);
} catch (Exception ex) {
logger.warn("An error occurred while attempting rollback when closing session: " + session, ex);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a5a00bf1/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
----------------------------------------------------------------------
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 d5e5e7c..b7e0d06 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
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.script.Bindings;
+import javax.script.SimpleBindings;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -149,7 +150,8 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
context.getChannelHandlerContext().channel().attr(StateKey.SESSION).set(session);
evalOpInternal(context, session::getGremlinExecutor, () -> {
- final Bindings bindings = session.getBindings();
+ final Bindings bindings = new SimpleBindings();
+ bindings.putAll(session.getBindings());
// parameter bindings override session bindings if present
Optional.ofNullable((Map<String, Object>) msg.getArgs().get(Tokens.ARGS_BINDINGS)).ifPresent(bindings::putAll);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a5a00bf1/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 6ca7dff..7714a93 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
@@ -780,4 +780,31 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
assertEquals("jason", v.value("name"));
}
}
+
+ @Test
+ public void shouldEnsureSessionBindingsAreThreadSafe() throws Exception {
+ final Cluster cluster = Cluster.open();
+ final Client client = cluster.connect(name.getMethodName());
+
+ client.submitAsync("a=100;b=1000;c=10000;null");
+ final int requests = 1000;
+ final List<CompletableFuture<ResultSet>> futures = new ArrayList<>(requests);
+ IntStream.range(0, requests).forEach(i -> {
+ try {
+ futures.add(client.submitAsync("a+b+c"));
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ });
+
+ assertEquals(requests, futures.size());
+
+ for(CompletableFuture<ResultSet> f : futures) {
+ final Result r = f.get().one();
+ assertEquals(11100, r.getInt());
+ }
+
+ client.close();
+ cluster.close();
+ }
}