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/11/03 11:58:01 UTC
[12/12] tinkerpop git commit: TINKERPOP-1434 Block calls on "remote"
traversal to get side-effects
TINKERPOP-1434 Block calls on "remote" traversal to get side-effects
This prevents calls for side-effects that are not yet ready on the server. This was generally a problem in tests, but usrs could have also run afoul of the issue if they decided to get low-level with the API. It is more likely that they would grab side-effects with a call to cap() so this would be a non-issue in that case.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/1921e031
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/1921e031
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/1921e031
Branch: refs/heads/TINKERPOP-1434
Commit: 1921e031a5d584ef52426a83d5a81aca6d621508
Parents: 53e1c21
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Wed Nov 2 17:12:36 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Thu Nov 3 07:57:32 2016 -0400
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../tinkerpop/gremlin/driver/ResultSet.java | 13 +++
.../driver/remote/DriverRemoteTraversal.java | 18 +++-
.../tinkerpop/gremlin/driver/ResultSetTest.java | 26 +++++-
.../server/GremlinServerIntegrateTest.java | 86 ++++++++++++++++++++
5 files changed, 141 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1921e031/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 57d7119..c6809a8 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -36,6 +36,7 @@ TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
* Added `ProjectStep.getProjectKeys()` for strategies that rely on such information.
* Added `VertexFeatures.supportsDuplicateMultiProperties()` for graphs that only support unique values in multi-properties.
* Deprecated the "performance" tests in `OptIn`.
+* Block calls to "remote" traversal side-effects until the traversal read is complete which signifies an end to iteration.
* Added `Pick.none` and `Pick.any` to the serializers and importers.
* Added a class loader to `TraversalStrategies.GlobalCache` which guarantees strategies are registered prior to `GlobalCache.getStrategies()`.
* Fixed a severe bug where `GraphComputer` strategies are not being loaded until the second use of the traversal source.
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1921e031/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java
index b9c8636..f82876c 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/ResultSet.java
@@ -81,6 +81,19 @@ public final class ResultSet implements Iterable<Result> {
}
/**
+ * Returns a future that will complete when all items have been returned from the server.
+ */
+ public CompletableFuture<Void> allItemsAvailableAsync() {
+ final CompletableFuture<Void> allAvailable = new CompletableFuture<>();
+ readCompleted.thenRun(() -> allAvailable.complete(null));
+ readCompleted.exceptionally(t -> {
+ allAvailable.completeExceptionally(t);
+ return null;
+ });
+ return allAvailable;
+ }
+
+ /**
* Gets the number of items available on the client.
*/
public int getAvailableItemCount() {
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1921e031/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
index e6e1c9b..88ee794 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
@@ -51,7 +51,7 @@ public class DriverRemoteTraversal<S, E> extends AbstractRemoteTraversal<S, E> {
private final Iterator<Traverser.Admin<E>> traversers;
private Traverser.Admin<E> lastTraverser = EmptyTraverser.instance();
private final RemoteTraversalSideEffects sideEffects;
- private final Client client;
+ private final ResultSet rs;
public DriverRemoteTraversal(final ResultSet rs, final Client client, final boolean attach, final Optional<Configuration> conf) {
// attaching is really just for testing purposes. it doesn't make sense in any real-world scenario as it would
@@ -65,15 +65,29 @@ public class DriverRemoteTraversal<S, E> extends AbstractRemoteTraversal<S, E> {
this.traversers = new TraverserIterator<>(rs.iterator());
}
- this.client = client;
+ this.rs = rs;
this.sideEffects = new DriverRemoteTraversalSideEffects(
client,
rs.getOriginalRequestMessage().getRequestId(),
rs.getHost());
}
+ /**
+ * Gets a side-effect from the server. Do not call this method prior to completing the iteration of the
+ * {@link DriverRemoteTraversal} that spawned this as the side-effect will not be ready. If this method is called
+ * prior to iteration being complete, then it will block until the traversal notifies it of completion. Generally
+ * speaking, the common user would not get side-effects this way - they would use a call to {@code cap()}.
+ */
@Override
public RemoteTraversalSideEffects getSideEffects() {
+ // wait for the read to complete (i.e. iteration on the server) before allowing the caller to get the
+ // side-effect. calling prior to this will result in the side-effect not being found. of course, the
+ // bad part here is that the method blocks indefinitely waiting for the result, but it prevents the
+ // test failure problems that happen on slower systems. in practice, it's unlikely that a user would
+ // try to get a side-effect prior to iteration, but since the API allows it, this at least prevents
+ // the error.
+ rs.allItemsAvailableAsync().join();
+
return this.sideEffects;
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1921e031/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java
index f33168d..4a57ebb 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/ResultSetTest.java
@@ -24,7 +24,6 @@ import org.junit.Test;
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -49,6 +48,14 @@ public class ResultSetTest extends AbstractResultQueueTest {
}
@Test
+ public void shouldHaveAllItemsAvailableAsynchronouslyOnReadComplete() {
+ final CompletableFuture<Void> all = resultSet.allItemsAvailableAsync();
+ assertThat(all.isDone(), is(false));
+ readCompleted.complete(null);
+ assertThat(all.isDone(), is(true));
+ }
+
+ @Test
public void shouldHaveAllItemsAvailableOnReadComplete() {
assertThat(resultSet.allItemsAvailable(), is(false));
readCompleted.complete(null);
@@ -56,6 +63,23 @@ public class ResultSetTest extends AbstractResultQueueTest {
}
@Test
+ public void shouldHaveAllItemsAvailableAsynchronouslyOnReadCompleteWhileLoading() throws Exception {
+ final CompletableFuture<Void> all = resultSet.allItemsAvailableAsync();
+ assertThat(all.isDone(), is(false));
+
+ final AtomicBoolean atLeastOnce = new AtomicBoolean(false);
+ addToQueue(1000, 1, true, true);
+ while (!readCompleted.isDone()) {
+ atLeastOnce.set(true);
+ if (!atLeastOnce.get())
+ assertThat(all.isDone(), is(false));
+ }
+
+ assertThat(atLeastOnce.get(), is(true));
+ assertThat(all.isDone(), is(true));
+ }
+
+ @Test
public void shouldHaveAllItemsAvailableOnReadCompleteWhileLoading() throws Exception {
assertThat(resultSet.allItemsAvailable(), is(false));
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1921e031/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 9ab1f0d..26b532b 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
@@ -895,4 +895,90 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
}
assertThat(error, is(true));
}
+
+ @Test
+ public void shouldBlockWhenGettingSideEffectKeysUsingWithRemote() throws Exception {
+ final Graph graph = EmptyGraph.instance();
+ final GraphTraversalSource g = graph.traversal().withRemote(conf);
+ g.addV("person").property("age", 20).iterate();
+ g.addV("person").property("age", 10).iterate();
+ final GraphTraversal traversal = g.V().aggregate("a")
+ .sideEffect(Lambda.consumer("{Thread.sleep(3000)}"))
+ .aggregate("b");
+
+ // force strategy application - if this doesn't happen then getSideEffects() returns DefaultTraversalSideEffects
+ traversal.hasNext();
+
+ // start a separate thread to iterate
+ final Thread t = new Thread(traversal::iterate);
+ t.start();
+
+ // blocks here until traversal iteration is complete
+ final DriverRemoteTraversalSideEffects se = (DriverRemoteTraversalSideEffects) traversal.asAdmin().getSideEffects();
+
+ // Get keys
+ final Set<String> sideEffectKeys = se.keys();
+ assertEquals(2, sideEffectKeys.size());
+
+ // Get side effects
+ final BulkSet aSideEffects = se.get("a");
+ assertThat(aSideEffects.isEmpty(), is(false));
+ final BulkSet bSideEffects = se.get("b");
+ assertThat(bSideEffects.isEmpty(), is(false));
+
+ // Should get local keys/side effects after close
+ se.close();
+
+ final Set<String> localSideEffectKeys = se.keys();
+ assertEquals(2, localSideEffectKeys.size());
+
+ final BulkSet localASideEffects = se.get("a");
+ assertThat(localASideEffects.isEmpty(), is(false));
+
+ final BulkSet localBSideEffects = se.get("b");
+ assertThat(localBSideEffects.isEmpty(), is(false));
+ }
+
+ @Test
+ public void shouldBlockWhenGettingSideEffectValuesUsingWithRemote() throws Exception {
+ final Graph graph = EmptyGraph.instance();
+ final GraphTraversalSource g = graph.traversal().withRemote(conf);
+ g.addV("person").property("age", 20).iterate();
+ g.addV("person").property("age", 10).iterate();
+ final GraphTraversal traversal = g.V().aggregate("a")
+ .sideEffect(Lambda.consumer("{Thread.sleep(3000)}"))
+ .aggregate("b");
+
+ // force strategy application - if this doesn't happen then getSideEffects() returns DefaultTraversalSideEffects
+ traversal.hasNext();
+
+ // start a separate thread to iterate
+ final Thread t = new Thread(traversal::iterate);
+ t.start();
+
+ // blocks here until traversal iteration is complete
+ final DriverRemoteTraversalSideEffects se = (DriverRemoteTraversalSideEffects) traversal.asAdmin().getSideEffects();
+
+ // Get side effects
+ final BulkSet aSideEffects = se.get("a");
+ assertThat(aSideEffects.isEmpty(), is(false));
+ final BulkSet bSideEffects = se.get("b");
+ assertThat(bSideEffects.isEmpty(), is(false));
+
+ // Get keys
+ final Set<String> sideEffectKeys = se.keys();
+ assertEquals(2, sideEffectKeys.size());
+
+ // Should get local keys/side effects after close
+ se.close();
+
+ final Set<String> localSideEffectKeys = se.keys();
+ assertEquals(2, localSideEffectKeys.size());
+
+ final BulkSet localASideEffects = se.get("a");
+ assertThat(localASideEffects.isEmpty(), is(false));
+
+ final BulkSet localBSideEffects = se.get("b");
+ assertThat(localBSideEffects.isEmpty(), is(false));
+ }
}