TINKERPOP-1490 Implemented promise API for Traversal Added two promise() methods that return CompletableFuture on Traversal. Provided an override on DefaultTraversal for those methods because the function that transforms the Traversal is executed in a different thread and therefore requires Graph transaction management (or else we would orphan transactions). Did not update gremlin-python with the promise API because it seemed to beg discussion on the "right" way to do that (i.e. what library to use to support promises?, just use futures from the core lib?, etc).
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/03e37e81 Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/03e37e81 Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/03e37e81 Branch: refs/heads/TINKERPOP-1490 Commit: 03e37e81fde3291f155d4ac7cf5d6fb42f76d6d0 Parents: d7fb966 Author: Stephen Mallette <sp...@genoprime.com> Authored: Tue Nov 1 09:30:28 2016 -0400 Committer: Stephen Mallette <sp...@genoprime.com> Committed: Wed Nov 2 11:10:49 2016 -0400 ---------------------------------------------------------------------- CHANGELOG.asciidoc | 1 + .../upgrade/release-3.2.x-incubating.asciidoc | 20 + gremlin-core/pom.xml | 5 + .../gremlin/process/traversal/Traversal.java | 50 ++ .../traversal/util/DefaultTraversal.java | 38 ++ .../process/traversal/TraversalTest.java | 473 +++++++++++++++++++ gremlin-groovy/pom.xml | 5 - .../gremlin/python/jsr223/PythonProvider.java | 1 + .../process/traversal/CoreTraversalTest.java | 42 ++ .../TinkerGraphGroovyTranslatorProvider.java | 1 + .../TinkerGraphJavaTranslatorProvider.java | 1 + 11 files changed, 632 insertions(+), 5 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/CHANGELOG.asciidoc ---------------------------------------------------------------------- diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc index 328bc9d..23c49ca 100644 --- a/CHANGELOG.asciidoc +++ b/CHANGELOG.asciidoc @@ -26,6 +26,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET) ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +* Added `Traversal.promise()` methods to allow for asynchronous traversal processing. * Added `choose(predicate,traversal)` and `choose(traversal,traversal)` to effect if/then-semantics (no else). Equivalent to `choose(x,y,identity())`. * `SparkGraphComputer` no longer starts a worker iteration if the worker's partition is empty. * Added `ProjectStep.getProjectKeys()` for strategies that rely on such information. http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/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 f121aa4..acaa83b 100644 --- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc +++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc @@ -32,6 +32,26 @@ Please see the link:https://github.com/apache/tinkerpop/blob/3.2.4/CHANGELOG.asc Upgrading for Users ~~~~~~~~~~~~~~~~~~~ +Traversal Promises +^^^^^^^^^^^^^^^^^^ + +The `Traversal` API now has two `promise()` method overloads. These methods return a promise in the form of a +`CompleteableFuture`. Usage is as follows: + +[source,groovy] +---- +gremlin> promise = g.V().out().promise{it.next()} +==>java.util.concurrent.CompletableFuture@4aa3d36[Completed normally] +gremlin> promise.join() +==>v[3] +gremlin> promise.isDone() +==>true +gremlin> g.V().out().promise{it.toList()}.thenApply{it.size()}.get() +==>6 +---- + +See: link:https://issues.apache.org/jira/browse/TINKERPOP-1490[TINKERPOP-1490] + If/Then-Semantics with Choose Step ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-core/pom.xml ---------------------------------------------------------------------- diff --git a/gremlin-core/pom.xml b/gremlin-core/pom.xml index e8f3a34..0594448 100644 --- a/gremlin-core/pom.xml +++ b/gremlin-core/pom.xml @@ -61,6 +61,11 @@ limitations under the License. </exclusion> </exclusions> </dependency> + <dependency> + <groupId>org.apache.commons</groupId> + <artifactId>commons-lang3</artifactId> + <version>3.3.1</version> + </dependency> <!-- LOGGING --> <dependency> <groupId>org.slf4j</groupId> http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java index edc18e0..e4ba5a6 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java @@ -18,6 +18,7 @@ */ package org.apache.tinkerpop.gremlin.process.traversal; +import org.apache.commons.lang3.concurrent.BasicThreadFactory; import org.apache.tinkerpop.gremlin.process.computer.GraphComputer; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal; import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent; @@ -42,7 +43,13 @@ import java.util.Optional; import java.util.Set; import java.util.Spliterator; import java.util.Spliterators; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.function.Consumer; +import java.util.function.Function; import java.util.stream.Stream; import java.util.stream.StreamSupport; @@ -140,6 +147,43 @@ public interface Traversal<S, E> extends Iterator<E>, Serializable, Cloneable, A } /** + * Starts a promise to execute a function on the current {@code Traversal} that will be completed in the future. + * This implementation uses {@link Admin#traversalExecutorService} to execute the supplied + * {@code traversalFunction}. + */ + public default <T> CompletableFuture<T> promise(final Function<Traversal, T> traversalFunction) { + return promise(traversalFunction, Admin.traversalExecutorService); + } + + /** + * Starts a promise to execute a function on the current {@code Traversal} that will be completed in the future. + * This implementation uses the caller supplied {@code ExecutorService} to execute the {@code traversalFunction}. + */ + public default <T> CompletableFuture<T> promise(final Function<Traversal, T> traversalFunction, final ExecutorService service) { + final CompletableFuture<T> promise = new CompletableFuture<>(); + final Future iterationFuture = service.submit(() -> { + try { + promise.complete(traversalFunction.apply(this)); + } catch (Exception ex) { + // the promise may have been cancelled by the caller, in which case, there is no need to attempt + // another write on completion + if (!promise.isDone()) promise.completeExceptionally(ex); + } + }); + + // if the user cancels the promise then attempt to kill the iteration. + promise.exceptionally(t -> { + if (t instanceof CancellationException) { + iterationFuture.cancel(true); + } + + return null; + }); + + return promise; + } + + /** * Add all the results of the traversal to the provided collection. * * @param collection the collection to fill @@ -253,6 +297,12 @@ public interface Traversal<S, E> extends Iterator<E>, Serializable, Cloneable, A public interface Admin<S, E> extends Traversal<S, E> { /** + * Service that handles promises. + */ + static final ExecutorService traversalExecutorService = Executors.newCachedThreadPool( + new BasicThreadFactory.Builder().namingPattern("traversal-executor-%d").build()); + + /** * Get the {@link Bytecode} associated with the construction of this traversal. * * @return the byte code representation of the traversal http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java index 5bd01da..a81f34e 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java @@ -42,6 +42,10 @@ import java.util.Iterator; import java.util.List; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ForkJoinPool; +import java.util.function.Function; /** * @author Marko A. Rodriguez (http://markorodriguez.com) @@ -316,6 +320,40 @@ public class DefaultTraversal<S, E> implements Traversal.Admin<S, E> { this.graph = graph; } + /** + * Override of {@link Traversal#promise(Function)} that is aware of graph transactions. + */ + @Override + public <T2> CompletableFuture<T2> promise(final Function<Traversal, T2> traversalFunction) { + return this.promise(traversalFunction, Traversal.Admin.traversalExecutorService); + } + + /** + * Override of {@link Traversal#promise(Function)} that is aware of graph transactions. In a transactional graph + * a promise represents the full scope of a transaction, even if the graph is only partially iterated. + */ + @Override + public <T2> CompletableFuture<T2> promise(final Function<Traversal, T2> traversalFunction, final ExecutorService service) { + if (graph != null && graph.features().graph().supportsTransactions()) { + final Function<Traversal, T2> transactionAware = traversal -> { + + try { + if (graph.tx().isOpen()) graph.tx().rollback(); + final T2 obj = traversalFunction.apply(traversal); + if (graph.tx().isOpen()) graph.tx().commit(); + return obj; + } catch (Exception ex) { + if (graph.tx().isOpen()) graph.tx().rollback(); + throw ex; + } + }; + + return Traversal.Admin.super.promise(transactionAware, service); + } else { + return Traversal.Admin.super.promise(traversalFunction, service); + } + } + @Override public boolean equals(final Object other) { return other != null && other.getClass().equals(this.getClass()) && this.equals(((Traversal.Admin) other)); http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java new file mode 100644 index 0000000..aa1b99b --- /dev/null +++ b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java @@ -0,0 +1,473 @@ +/* + * 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.process.traversal; + +import org.apache.tinkerpop.gremlin.process.remote.traversal.DefaultRemoteTraverser; +import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent; +import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement; +import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Optional; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.hamcrest.core.Is.is; +import static org.hamcrest.core.IsCollectionContaining.hasItems; +import static org.hamcrest.core.IsInstanceOf.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; + +/** + * @author Stephen Mallette (http://stephen.genoprime.com) + */ +public class TraversalTest { + + private final ExecutorService service = Executors.newFixedThreadPool(2); + + @Test + public void shouldTryNext() { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3); + final Optional<Integer> optFirst = t.tryNext(); + assertEquals(1, optFirst.get().intValue()); + final Optional<Integer> optSecond = t.tryNext(); + assertEquals(2, optSecond.get().intValue()); + final Optional<Integer> optThird = t.tryNext(); + assertEquals(3, optThird.get().intValue()); + + IntStream.range(0, 100).forEach(i -> { + assertThat(t.tryNext().isPresent(), is(false)); + }); + } + + @Test + public void shouldGetTwoAtATime() { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7); + final List<Integer> batchOne = t.next(2); + assertEquals(2, batchOne.size()); + assertThat(batchOne, hasItems(1 ,2)); + + final List<Integer> batchTwo = t.next(2); + assertEquals(2, batchTwo.size()); + assertThat(batchTwo, hasItems(3 ,4)); + + final List<Integer> batchThree = t.next(2); + assertEquals(2, batchThree.size()); + assertThat(batchThree, hasItems(5, 6)); + + final List<Integer> batchFour = t.next(2); + assertEquals(1, batchFour.size()); + assertThat(batchFour, hasItems(7)); + + final List<Integer> batchFive = t.next(2); + assertEquals(0, batchFive.size()); + } + + @Test + public void shouldFillList() { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7); + final List<Integer> listToFill = new ArrayList<>(); + final List<Integer> batch = t.fill(listToFill); + assertEquals(7, batch.size()); + assertThat(batch, hasItems(1 ,2, 3, 4, 5, 6, 7)); + assertThat(t.hasNext(), is(false)); + assertSame(listToFill, batch); + } + + @Test + public void shouldStream() { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7); + final List<Integer> batch = t.toStream().collect(Collectors.toList()); + assertEquals(7, batch.size()); + assertThat(batch, hasItems(1 ,2, 3, 4, 5, 6, 7)); + assertThat(t.hasNext(), is(false)); + } + + @Test + public void shouldPromiseNextThreeUsingForkJoin() throws Exception { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7); + final CompletableFuture<List<Integer>> promiseFirst = t.promise(traversal -> traversal.next(3)); + final List<Integer> listFirst = promiseFirst.get(); + assertEquals(3, listFirst.size()); + assertThat(listFirst, hasItems(1 ,2, 3)); + assertThat(t.hasNext(), is(true)); + assertThat(promiseFirst.isDone(), is(true)); + + final CompletableFuture<List<Integer>> promiseSecond = t.promise(traversal -> traversal.next(3)); + final List<Integer> listSecond = promiseSecond.get(); + assertEquals(3, listSecond.size()); + assertThat(listSecond, hasItems(4, 5, 6)); + assertThat(t.hasNext(), is(true)); + assertThat(promiseSecond.isDone(), is(true)); + + final CompletableFuture<List<Integer>> promiseThird = t.promise(traversal -> traversal.next(3)); + final List<Integer> listThird = promiseThird.get(); + assertEquals(1, listThird.size()); + assertThat(listThird, hasItems(7)); + assertThat(t.hasNext(), is(false)); + assertThat(promiseThird.isDone(), is(true)); + + final CompletableFuture<Integer> promiseDead = t.promise(traversal -> (Integer) traversal.next()); + final AtomicBoolean dead = new AtomicBoolean(false); + promiseDead.exceptionally(tossed -> { + dead.set(tossed instanceof NoSuchElementException); + return null; + }); + + try { + promiseDead.get(10000, TimeUnit.MILLISECONDS); + fail("Should have gotten an exception"); + } catch (Exception ex) { + if (ex instanceof TimeoutException) { + fail("This should not have timed out but should have gotten an exception caught above in the exceptionally() clause"); + } + + assertThat(ex.getCause(), instanceOf(NoSuchElementException.class)); + } + + assertThat(dead.get(), is(true)); + assertThat(t.hasNext(), is(false)); + assertThat(promiseDead.isDone(), is(true)); + } + + @Test + public void shouldPromiseNextThreeUsingSpecificExecutor() throws Exception { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7); + final CompletableFuture<List<Integer>> promiseFirst = t.promise(traversal -> traversal.next(3), service); + final List<Integer> listFirst = promiseFirst.get(); + assertEquals(3, listFirst.size()); + assertThat(listFirst, hasItems(1 ,2, 3)); + assertThat(t.hasNext(), is(true)); + assertThat(promiseFirst.isDone(), is(true)); + + final CompletableFuture<List<Integer>> promiseSecond = t.promise(traversal -> traversal.next(3), service); + final List<Integer> listSecond = promiseSecond.get(); + assertEquals(3, listSecond.size()); + assertThat(listSecond, hasItems(4, 5, 6)); + assertThat(t.hasNext(), is(true)); + assertThat(promiseSecond.isDone(), is(true)); + + final CompletableFuture<List<Integer>> promiseThird = t.promise(traversal -> traversal.next(3), service); + final List<Integer> listThird = promiseThird.get(); + assertEquals(1, listThird.size()); + assertThat(listThird, hasItems(7)); + assertThat(t.hasNext(), is(false)); + assertThat(promiseThird.isDone(), is(true)); + + final CompletableFuture<Integer> promiseDead = t.promise(traversal -> (Integer) traversal.next(), service); + final AtomicBoolean dead = new AtomicBoolean(false); + promiseDead.exceptionally(tossed -> { + dead.set(tossed instanceof NoSuchElementException); + return null; + }); + + try { + promiseDead.get(10000, TimeUnit.MILLISECONDS); + fail("Should have gotten an exception"); + } catch (Exception ex) { + if (ex instanceof TimeoutException) { + fail("This should not have timed out but should have gotten an exception caught above in the exceptionally() clause"); + } + + assertThat(ex.getCause(), instanceOf(NoSuchElementException.class)); + } + + assertThat(dead.get(), is(true)); + assertThat(t.hasNext(), is(false)); + assertThat(promiseDead.isDone(), is(true)); + } + + @Test + public void shouldInterruptTraversalFunction() throws Exception { + final Random rand = new Random(1234567890); + + // infinite traversal + final MockTraversal<Integer> t = new MockTraversal<>(IntStream.generate(rand::nextInt).iterator()); + + // iterate a bunch of it + final CompletableFuture<List<Integer>> promise10 = t.promise(traversal -> traversal.next(10), service); + assertEquals(10, promise10.get(10000, TimeUnit.MILLISECONDS).size()); + final CompletableFuture<List<Integer>> promise100 = t.promise(traversal -> traversal.next(100), service); + assertEquals(100, promise100.get(10000, TimeUnit.MILLISECONDS).size()); + final CompletableFuture<List<Integer>> promise1000 = t.promise(traversal -> traversal.next(1000), service); + assertEquals(1000, promise1000.get(10000, TimeUnit.MILLISECONDS).size()); + + // this is endless, so let's cancel + final CompletableFuture<List<Integer>> promiseForevers = t.promise(traversal -> traversal.next(Integer.MAX_VALUE), service); + + // specify what to do on exception + final AtomicBoolean failed = new AtomicBoolean(false); + promiseForevers.exceptionally(ex -> { + failed.set(true); + return null; + }); + + try { + // let it actually iterate a moment + promiseForevers.get(500, TimeUnit.MILLISECONDS); + fail("This should have timed out because the traversal has infinite items in it"); + } catch (TimeoutException tex) { + + } + + assertThat(promiseForevers.isDone(), is(false)); + promiseForevers.cancel(true); + assertThat(failed.get(), is(true)); + assertThat(promiseForevers.isDone(), is(true)); + } + + @Test + public void shouldIterate() { + final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7); + assertThat(t.hasNext(), is(true)); + t.iterate(); + assertThat(t.hasNext(), is(false)); + } + + private static class MockStep<E> implements Step<E,E> { + + private final Iterator<E> itty; + + MockStep(final Iterator<E> itty) { + this.itty = itty; + } + + @Override + public void addStarts(final Iterator starts) { + + } + + @Override + public void addStart(final Traverser.Admin start) { + + } + + @Override + public void setPreviousStep(final Step step) { + + } + + @Override + public Step getPreviousStep() { + return null; + } + + @Override + public void setNextStep(final Step step) { + + } + + @Override + public Step getNextStep() { + return null; + } + + @Override + public Traversal.Admin getTraversal() { + return null; + } + + @Override + public void setTraversal(final Traversal.Admin traversal) { + + } + + @Override + public void reset() { + + } + + @Override + public Step clone() { + return null; + } + + @Override + public Set<String> getLabels() { + return null; + } + + @Override + public void addLabel(final String label) { + + } + + @Override + public void removeLabel(final String label) { + + } + + @Override + public void setId(final String id) { + + } + + @Override + public String getId() { + return null; + } + + @Override + public boolean hasNext() { + return itty.hasNext(); + } + + @Override + public Traverser.Admin<E> next() { + return new DefaultRemoteTraverser<>(itty.next(), 1L); + } + } + + + private static class MockTraversal<T> implements Traversal.Admin<T,T> { + + private Iterator<T> itty; + + private Step mockEndStep; + + private List<Step> steps; + + MockTraversal(final T... objects) { + this(Arrays.asList(objects)); + } + + MockTraversal(final List<T> list) { + this(list.iterator()); + } + + MockTraversal(final Iterator<T> itty) { + this.itty = itty; + mockEndStep = new MockStep<>(itty); + steps = Collections.singletonList(mockEndStep); + } + + @Override + public Bytecode getBytecode() { + return null; + } + + @Override + public List<Step> getSteps() { + return steps; + } + + @Override + public <S2, E2> Admin<S2, E2> addStep(final int index, final Step<?, ?> step) throws IllegalStateException { + return null; + } + + @Override + public <S2, E2> Admin<S2, E2> removeStep(final int index) throws IllegalStateException { + return null; + } + + @Override + public void applyStrategies() throws IllegalStateException { + + } + + @Override + public TraverserGenerator getTraverserGenerator() { + return null; + } + + @Override + public Set<TraverserRequirement> getTraverserRequirements() { + return null; + } + + @Override + public void setSideEffects(final TraversalSideEffects sideEffects) { + + } + + @Override + public TraversalSideEffects getSideEffects() { + return null; + } + + @Override + public void setStrategies(final TraversalStrategies strategies) { + + } + + @Override + public TraversalStrategies getStrategies() { + return null; + } + + @Override + public void setParent(final TraversalParent step) { + + } + + @Override + public TraversalParent getParent() { + return null; + } + + @Override + public Admin<T, T> clone() { + return null; + } + + @Override + public boolean isLocked() { + return false; + } + + @Override + public Optional<Graph> getGraph() { + return null; + } + + @Override + public void setGraph(final Graph graph) { + + } + + @Override + public boolean hasNext() { + return itty.hasNext(); + } + + @Override + public T next() { + if (Thread.interrupted()) throw new TraversalInterruptedException(); + return itty.next(); + } + } +} http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-groovy/pom.xml ---------------------------------------------------------------------- diff --git a/gremlin-groovy/pom.xml b/gremlin-groovy/pom.xml index acc51b7..dae5e8a 100644 --- a/gremlin-groovy/pom.xml +++ b/gremlin-groovy/pom.xml @@ -61,11 +61,6 @@ limitations under the License. <classifier>indy</classifier> </dependency> <dependency> - <groupId>org.apache.commons</groupId> - <artifactId>commons-lang3</artifactId> - <version>3.3.1</version> - </dependency> - <dependency> <groupId>com.github.jeremyh</groupId> <artifactId>jBCrypt</artifactId> <version>jbcrypt-0.4</version> http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java ---------------------------------------------------------------------- diff --git a/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java b/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java index fe04156..3bf4239 100644 --- a/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java +++ b/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java @@ -72,6 +72,7 @@ public class PythonProvider extends AbstractGraphProvider { "shouldNeverPropagateANullValuedTraverser", "shouldHidePartitionKeyForValues", "g_withSackXBigInteger_TEN_powX1000X_assignX_V_localXoutXknowsX_barrierXnormSackXX_inXknowsX_barrier_sack", + "shouldUsePromiseAndControlTransactionsIfAvailable", // ProgramTest.Traversals.class.getCanonicalName(), TraversalInterruptionTest.class.getCanonicalName(), http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java ---------------------------------------------------------------------- diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java index bf3acf3..e68d872 100644 --- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java +++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java @@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.process.traversal; import org.apache.tinkerpop.gremlin.ExceptionCoverage; import org.apache.tinkerpop.gremlin.FeatureRequirement; +import org.apache.tinkerpop.gremlin.FeatureRequirementSet; import org.apache.tinkerpop.gremlin.LoadGraphWith; import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal; @@ -37,6 +38,9 @@ import java.util.List; import java.util.Map; import java.util.Random; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN; @@ -272,4 +276,42 @@ public class CoreTraversalTest extends AbstractGremlinProcessTest { assertEquals(1, IteratorUtils.count(t)); g.tx().rollback(); } + + @Test + @FeatureRequirementSet(FeatureRequirementSet.Package.SIMPLE) + public void shouldUsePromiseAndControlTransactionsIfAvailable() throws Exception { + // this test will validate that transactional graphs can properly open/close transactions within a promise. + // as there is a feature check, non-transactional graphs can use this to simply exercise the promise API + final Vertex vAdded = g.addV("person").property("name", "stephen").promise(t -> (Vertex) t.next()).get(10000, TimeUnit.MILLISECONDS); + final Vertex vRead = g.V().has("name", "stephen").next(); + assertEquals(vAdded.id(), vRead.id()); + + // transaction should have been committed at this point so test the count in this thread to validate persistence + assertVertexEdgeCounts(graph, 1, 0); + + // cancel a promise and ensure the transaction ended in failure. hold the traversal in park until it can be + // interrupted, then the promise will have to rollback the transaction. + final CompletableFuture promiseToCancel = g.addV("person").property("name", "marko").sideEffect(traverser -> { + try { + Thread.sleep(100000); + } catch (Exception ignored) { + + } + }).promise(t -> (Vertex) t.next()); + + try { + promiseToCancel.get(500, TimeUnit.MILLISECONDS); + fail("Should have timed out"); + } catch (TimeoutException te) { + + } + + promiseToCancel.cancel(true); + + // graphs that support transactions will rollback the transaction + if (graph.features().graph().supportsTransactions()) + assertVertexEdgeCounts(graph, 1, 0); + else + assertVertexEdgeCounts(graph, 2, 0); + } } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java ---------------------------------------------------------------------- diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java index dd118d7..a595c34 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java @@ -51,6 +51,7 @@ public class TinkerGraphGroovyTranslatorProvider extends TinkerGraphProvider { "g_VX1X_out_injectXv2X_name", "shouldNeverPropagateANoBulkTraverser", "shouldNeverPropagateANullValuedTraverser", + "shouldUsePromiseAndControlTransactionsIfAvailable", GraphComputerTest.class.getCanonicalName(), ProgramTest.Traversals.class.getCanonicalName(), TraversalInterruptionTest.class.getCanonicalName(), http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/03e37e81/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java ---------------------------------------------------------------------- diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java index f40a8c7..15a6c0b 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java @@ -49,6 +49,7 @@ public class TinkerGraphJavaTranslatorProvider extends TinkerGraphProvider { TraversalInterruptionComputerTest.class.getCanonicalName(), "shouldNeverPropagateANoBulkTraverser", "shouldNeverPropagateANullValuedTraverser", + "shouldUsePromiseAndControlTransactionsIfAvailable", ElementIdStrategyProcessTest.class.getCanonicalName(), EventStrategyProcessTest.class.getCanonicalName(), ProgramTest.Traversals.class.getCanonicalName()));
