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/05/18 13:11:11 UTC
[04/16] incubator-tinkerpop git commit: wow. was in a massive hole.
Anywho, lots of good stuff here. PR will have all the details.
wow. was in a massive hole. Anywho, lots of good stuff here. PR will have all the details.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/6acbf96a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/6acbf96a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/6acbf96a
Branch: refs/heads/TINKERPOP-1274
Commit: 6acbf96a79e64cd5a69635d4de3c959e2c0bf425
Parents: c929fd4
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu May 12 10:45:55 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu May 12 10:45:55 2016 -0600
----------------------------------------------------------------------
docs/src/reference/the-traversal.asciidoc | 80 +++++++++++++++++
.../computer/traversal/MasterExecutor.java | 2 +-
.../traversal/MemoryTraversalSideEffects.java | 29 ++++--
.../traversal/TraversalVertexProgram.java | 13 +--
.../traversal/step/VertexComputing.java | 22 +++++
.../step/map/PageRankVertexProgramStep.java | 5 --
.../step/map/PeerPressureVertexProgramStep.java | 5 --
.../step/map/ProgramVertexProgramStep.java | 4 +-
.../traversal/step/map/VertexProgramStep.java | 21 +++--
.../gremlin/process/ProcessComputerSuite.java | 4 +-
.../process/traversal/step/map/ProgramTest.java | 95 ++++++++++++++------
.../SparkStarBarrierInterceptor.java | 2 +-
12 files changed, 221 insertions(+), 61 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/docs/src/reference/the-traversal.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/reference/the-traversal.asciidoc b/docs/src/reference/the-traversal.asciidoc
index 94cb6a2..71708e2 100644
--- a/docs/src/reference/the-traversal.asciidoc
+++ b/docs/src/reference/the-traversal.asciidoc
@@ -1419,6 +1419,86 @@ g.V().out('created').
select('a')
----
+[[program-step]]
+Program Step
+~~~~~~~~~~~~
+
+The `program()`-step (*map*/*sideEffect*) is the "lambda" step for `GraphComputer` jobs. The step takes a
+<<vertexprogram,`VertexProgram`>> as an argument and will process the incoming graph accordingly. Thus, the user
+can create their own `VertexProgram` and have it execute within a traversal. The configuration provided to the
+vertex program includes:
+
+* `gremlin.vertexProgramStep.rootTraversal` is a serialization of a `PureTraversal` form of the root traversal.
+* `gremlin.vertexProgramStep.stepId` is the step string id of the `program()`-step being executed.
+
+The user supplied `VertexProgram` can leverage that information accordingly within their vertex program. Example uses
+are provided below.
+
+[source,java]
+----
+public void loadState(final Graph graph, final Configuration configuration) {
+ VertexProgram.super.loadState(graph, configuration);
+ this.traversal = PureTraversal.loadState(configuration, VertexProgramStep.ROOT_TRAVERSAL, graph);
+ this.programStep = new TraversalMatrix<>(this.traversal.get()).getStepById(configuration.getString(ProgramVertexProgramStep.STEP_ID));
+ // if the traversal sideEffects will be used in the computation, add them as memory compute keys
+ this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
+ // if master-traversal traversers may be propagated, create a memory compute key
+ this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.HALTED_TRAVERSERS, Operator.addAll, false, false));
+}
+
+public void setup(final Memory memory) {
+ if(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
+ // haltedTraversers in setup() represent master-traversal traversers
+ // for example, from a traversal of the form g.V().groupCount().program(...)
+ TraverserSet<Object> haltedTraversers = sideEffects.get(TraversalVertexProgram.HALTED_TRAVERSERS);
+ // do as you please with this information
+ }
+}
+
+public void execute(final Vertex vertex, final Messenger messenger, final Memory memory) {
+ if(vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS).isPresent()) {
+ // haltedTraversers in execute() represent worker-traversal traversers
+ // for example, from a traversal of the form g.V().out().program(...)
+ TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS) :
+ // create a new halted traverser set that can be used by the next OLAP job in the chain
+ // these are worker-traversers that are distributed throughout the graph
+ TraverserSet<Object> newHaltedTraversers = new TraverserSet<>();
+ haltedTraversers.forEach(traverser -> {
+ newHaltedTraversers.add(traverser.split(traverser.get().toString(), this.programStep));
+ });
+ vertex.property(VertexProperty.Cardinality.single, TraversalVertexProgram.HALTED_TRAVERSERS, newHaltedTraversers);
+ // it is possible to create master-traversers that are localized to the master traversal (thread)
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS,
+ new TraverserSet<>(this.traversal().get().getTraverserGenerator().generate("an example", this.programStep, 1l)));
+ }
+
+public boolean terminate(final Memory memory) {
+ // the master-traversal will have halted traversers
+ assert memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS);
+ final TraverserSet<String> haltedTraversers = memory.get(TraversalVertexProgram.HALTED_TRAVERSERS);
+ // it will only have the traversers sent to the master traversal via memory
+ assert haltedTraversers.stream().map(Traverser::get).filter(s -> s.equals("an example")).findAny().isPresent();
+ // it will not contain the worker traversers distributed throughout the vertices
+ assert !haltedTraversers.stream().map(Traverser::get).filter(s -> !s.equals("an example")).findAny().isPresent();
+ return true;
+}
+----
+
+NOTE: The test case `ProgramTest` in `gremlin-test` has an example vertex program called `TestProgram` that demonstrates
+all the various ways in which traversal and traverser information is propagated within a vertex program and ultimately
+usable by other vertex programs (including `TraversalVertexProgram`) down the line in an OLAP compute chain.
+
+Finally, an example is provided using `PageRankVertexProgram` which doesn't use <<pagerank-step,`pageRank()`>>-step.
+
+[gremlin-groovy,modern]
+----
+g = graph.traversal().withComputer()
+g.V().hasLabel('person').
+ program(PageRankVertexProgram.build().property('rank').create(graph)).
+ order().by('rank', incr).
+ valueMap('name', 'rank')
+----
+
[[range-step]]
Range Step
~~~~~~~~~~
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MasterExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MasterExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MasterExecutor.java
index ac0fc29..88570fe 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MasterExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MasterExecutor.java
@@ -51,7 +51,7 @@ import java.util.Set;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class MasterExecutor {
+final class MasterExecutor {
private MasterExecutor() {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
index 89f4f5c..1327a7f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/MemoryTraversalSideEffects.java
@@ -38,7 +38,23 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
private TraversalSideEffects sideEffects;
private Memory memory;
- private boolean worker;
+ private State state;
+
+ public enum State {
+ SETUP,
+ WORKER_ITERATION_START,
+ EXECUTE,
+ WORKER_ITERATION_END,
+ TERMINATE;
+
+ public boolean masterState() {
+ return this == SETUP || this == TERMINATE;
+ }
+
+ public boolean workerState() {
+ return !this.masterState();
+ }
+ }
private MemoryTraversalSideEffects() {
// for serialization
@@ -77,7 +93,7 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
@Override
public void add(final String key, final Object value) {
- if (this.worker)
+ if (this.state.workerState())
this.memory.add(key, value);
else
this.memory.set(key, this.sideEffects.getReducer(key).apply(this.memory.get(key), value));
@@ -152,19 +168,20 @@ public final class MemoryTraversalSideEffects implements TraversalSideEffects {
}
public void storeSideEffectsInMemory() {
- if (this.worker)
+ if (this.state.workerState())
this.sideEffects.forEach(this.memory::add);
else
this.sideEffects.forEach(this.memory::set);
}
- public static void setMemorySideEffects(final Traversal.Admin<?, ?> traversal, final Memory memory, final boolean worker) {
+ public static void setMemorySideEffects(final Traversal.Admin<?, ?> traversal, final Memory memory, final State state) {
final TraversalSideEffects sideEffects = traversal.getSideEffects();
if (!(sideEffects instanceof MemoryTraversalSideEffects)) {
traversal.setSideEffects(new MemoryTraversalSideEffects(sideEffects));
}
- ((MemoryTraversalSideEffects) traversal.getSideEffects()).memory = memory;
- ((MemoryTraversalSideEffects) traversal.getSideEffects()).worker = worker;
+ final MemoryTraversalSideEffects memoryTraversalSideEffects = ((MemoryTraversalSideEffects) traversal.getSideEffects());
+ memoryTraversalSideEffects.memory = memory;
+ memoryTraversalSideEffects.state = state;
}
public static Set<MemoryComputeKey> getMemoryComputeKeys(final Traversal.Admin<?, ?> traversal) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index 21d7f06..4912dbe 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -163,15 +163,15 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public void setup(final Memory memory) {
// memory is local
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, false);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.SETUP);
final MemoryTraversalSideEffects sideEffects = ((MemoryTraversalSideEffects) this.traversal.get().getSideEffects());
sideEffects.storeSideEffectsInMemory();
memory.set(VOTE_TO_HALT, true);
memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
memory.set(COMPLETED_BARRIERS, new HashSet<>());
// if halted traversers are being sent from a previous VertexProgram in an OLAP chain (non-distributed traversers), get them into the stream
- if (sideEffects.exists(HALTED_TRAVERSERS)) {
- final TraverserSet<Object> haltedTraversers = sideEffects.get(HALTED_TRAVERSERS);
+ if (sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
+ final TraverserSet<Object> haltedTraversers = sideEffects.get(TraversalVertexProgram.HALTED_TRAVERSERS);
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
IteratorUtils.removeOnNext(haltedTraversers.iterator()).forEachRemaining(traverser -> {
traverser.setStepId(this.traversal.get().getStartStep().getId());
@@ -180,6 +180,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
assert haltedTraversers.isEmpty(); // TODO: this should be empty
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers);
+ memory.set(HALTED_TRAVERSERS, haltedTraversers);
memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
} else {
memory.set(HALTED_TRAVERSERS, new TraverserSet<>());
@@ -195,8 +196,8 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public void execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final Memory memory) {
// memory is distributed
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, true);
- // if a barrier was completed in another worker, it is also completed here (ensure distributed barries are synchronized)
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.EXECUTE);
+ // if a barrier was completed in another worker, it is also completed here (ensure distributed barriers are synchronized)
final Set<String> completedBarriers = memory.get(COMPLETED_BARRIERS);
for (final String stepId : completedBarriers) {
final Step<?, ?> step = this.traversalMatrix.getStepById(stepId);
@@ -245,7 +246,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public boolean terminate(final Memory memory) {
// memory is local
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, false);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.TERMINATE);
final boolean voteToHalt = memory.<Boolean>get(VOTE_TO_HALT);
memory.set(VOTE_TO_HALT, true);
memory.set(ACTIVE_TRAVERSERS, new TraverserSet<>());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
index 48bed14..57e1a3e 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/VertexComputing.java
@@ -29,12 +29,34 @@ import org.apache.tinkerpop.gremlin.structure.Graph;
*/
public interface VertexComputing {
+ /**
+ * Set the {@link Computer} to be used to generate the {@link GraphComputer}.
+ *
+ * @param computer the computer specification.
+ */
public void setComputer(final Computer computer);
+ /**
+ * Get the {@link Computer} for generating the {@link GraphComputer}.
+ * Inferences on the state of the {@link org.apache.tinkerpop.gremlin.process.traversal.Step}
+ * within the {@link org.apache.tinkerpop.gremlin.process.traversal.Traversal} can be use applied here.
+ *
+ * @return the computer specification for generating the graph computer.
+ */
public Computer getComputer();
+ /**
+ * Generate the {@link VertexProgram}.
+ *
+ * @param graph the {@link Graph} that the program will be executed over.
+ * @return the generated vertex program instance.
+ */
public VertexProgram generateProgram(final Graph graph);
+ /**
+ * @deprecated As of release 3.2.1. Please use {@link VertexComputing#getComputer()}.
+ */
+ @Deprecated
public default GraphComputer generateComputer(final Graph graph) {
return this.getComputer().apply(graph);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
index af435a3..5d10e67 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PageRankVertexProgramStep.java
@@ -97,11 +97,6 @@ public final class PageRankVertexProgramStep extends VertexProgramStep implement
}
@Override
- public GraphComputer generateComputer(final Graph graph) {
- return this.computer.apply(graph).persist(GraphComputer.Persist.EDGES).result(GraphComputer.ResultGraph.NEW);
- }
-
- @Override
public Set<TraverserRequirement> getRequirements() {
return TraversalParent.super.getSelfAndChildRequirements();
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
index b5353e9..f42ce93 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/PeerPressureVertexProgramStep.java
@@ -100,11 +100,6 @@ public final class PeerPressureVertexProgramStep extends VertexProgramStep imple
}
@Override
- public GraphComputer generateComputer(final Graph graph) {
- return this.computer.apply(graph).persist(GraphComputer.Persist.EDGES).result(GraphComputer.ResultGraph.NEW);
- }
-
- @Override
public PeerPressureVertexProgramStep clone() {
final PeerPressureVertexProgramStep clone = (PeerPressureVertexProgramStep) super.clone();
clone.edgeTraversal = this.edgeTraversal.clone();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
index 764df2b..a1342b6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ProgramVertexProgramStep.java
@@ -21,9 +21,9 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
import org.apache.commons.configuration.MapConfiguration;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -51,7 +51,7 @@ public final class ProgramVertexProgramStep extends VertexProgramStep {
public VertexProgram generateProgram(final Graph graph) {
final MapConfiguration base = new MapConfiguration(this.configuration);
base.setDelimiterParsingDisabled(true);
- PureTraversal.storeState(base, TraversalVertexProgram.TRAVERSAL, this.getTraversal().clone());
+ PureTraversal.storeState(base, ROOT_TRAVERSAL, TraversalHelper.getRootTraversal(this.getTraversal()).clone());
base.setProperty(STEP_ID, this.getId());
return VertexProgram.createVertexProgram(graph, base);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
index 9b9d5d7..eaf06c4 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/VertexProgramStep.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.step.map;
import org.apache.tinkerpop.gremlin.process.computer.Computer;
import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.VertexComputing;
@@ -47,7 +48,9 @@ import java.util.concurrent.Future;
*/
public abstract class VertexProgramStep extends AbstractStep<ComputerResult, ComputerResult> implements VertexComputing {
+ public static final String ROOT_TRAVERSAL = "gremlin.vertexProgramStep.rootTraversal";
public static final String STEP_ID = "gremlin.vertexProgramStep.stepId";
+
protected Computer computer = Computer.compute();
protected boolean first = true;
@@ -75,7 +78,7 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
this.processMemorySideEffects(result.memory());
return traverser.split(result, this);
}
- } catch (InterruptedException ie) {
+ } catch (final InterruptedException ie) {
// the thread running the traversal took an interruption while waiting on the call the future.get().
// the future should then be cancelled with interruption so that the the GraphComputer that created
// the future knows we don't care about it anymore. The GraphComputer should attempt to respect this
@@ -89,7 +92,14 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
@Override
public Computer getComputer() {
- return this.computer;
+ Computer tempComputer = this.computer;
+ if (!this.isEndStep()) {
+ if (null == tempComputer.getPersist())
+ tempComputer = tempComputer.persist(GraphComputer.Persist.EDGES);
+ if (null == tempComputer.getResultGraph())
+ tempComputer = tempComputer.result(GraphComputer.ResultGraph.NEW);
+ }
+ return tempComputer;
}
@Override
@@ -100,7 +110,6 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
protected boolean previousTraversalVertexProgram() {
Step<?, ?> currentStep = this;
while (!(currentStep instanceof EmptyStep)) {
- if (Thread.interrupted()) throw new TraversalInterruptedException();
if (currentStep instanceof TraversalVertexProgramStep)
return true;
currentStep = currentStep.getPreviousStep();
@@ -109,15 +118,14 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
}
private void processMemorySideEffects(final Memory memory) {
- // unfortunately there is no easy way to test this in a test case
- // assert this.isEndStep() == memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS);
+ // update the traversal side-effects with the state of the memory after the OLAP job execution
final TraversalSideEffects sideEffects = this.getTraversal().getSideEffects();
for (final String key : memory.keys()) {
if (sideEffects.exists(key)) {
sideEffects.set(key, memory.get(key));
}
}
- if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS) && !this.isEndStep()) {
+ if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS)) {
final TraverserSet<Object> haltedTraversers = memory.get(TraversalVertexProgram.HALTED_TRAVERSERS);
if (!haltedTraversers.isEmpty()) {
if (sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
@@ -128,6 +136,7 @@ public abstract class VertexProgramStep extends AbstractStep<ComputerResult, Com
}
}
+
protected boolean isEndStep() {
return this.getNextStep() instanceof ComputerResultStep || (this.getNextStep() instanceof ProfileStep && this.getNextStep().getNextStep() instanceof ComputerResultStep);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
index d8021eb..02f1b27 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/ProcessComputerSuite.java
@@ -112,7 +112,7 @@ public class ProcessComputerSuite extends AbstractGremlinSuite {
private static final Class<?>[] allTests = new Class<?>[]{
// computer, vertex program, and map/reduce semantics
- GraphComputerTest.class,
+ /*GraphComputerTest.class,
// branch
BranchTest.Traversals.class,
@@ -159,7 +159,7 @@ public class ProcessComputerSuite extends AbstractGremlinSuite {
PathTest.Traversals.class,
PeerPressureTest.Traversals.class,
ProfileTest.Traversals.class,
- ProjectTest.Traversals.class,
+ ProjectTest.Traversals.class, */
ProgramTest.Traversals.class,
PropertiesTest.Traversals.class,
SelectTest.Traversals.class,
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
index dcef79c..2c505e7 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/ProgramTest.java
@@ -28,16 +28,18 @@ import org.apache.tinkerpop.gremlin.process.computer.MemoryComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
+import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.MemoryTraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.ProgramVertexProgramStep;
-import org.apache.tinkerpop.gremlin.process.computer.util.StaticVertexProgram;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.VertexProgramStep;
import org.apache.tinkerpop.gremlin.process.traversal.Operator;
import org.apache.tinkerpop.gremlin.process.traversal.Order;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
+import org.apache.tinkerpop.gremlin.process.traversal.TraverserGenerator;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
@@ -112,7 +114,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
final Traversal<Vertex, Map<String, Object>> traversal = get_g_V_outXcreatedX_aggregateXxX_byXlangX_groupCount_programXTestProgramX_asXaX_selectXa_xX();
final List<Map<String, Object>> results = traversal.toList();
assertFalse(traversal.hasNext());
- assertEquals(4, results.size());
+ assertEquals(6, results.size());
final BulkSet<String> bulkSet = new BulkSet<>();
bulkSet.add("java", 4);
for (int i = 0; i < 4; i++) {
@@ -123,11 +125,15 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
strings.add((String) results.get(1).get("a"));
strings.add((String) results.get(2).get("a"));
strings.add((String) results.get(3).get("a"));
- assertEquals(4, strings.size());
+ strings.add((String) results.get(4).get("a"));
+ strings.add((String) results.get(5).get("a"));
+ assertEquals(6, strings.size());
assertTrue(strings.contains("hello"));
assertTrue(strings.contains("gremlin"));
assertTrue(strings.contains("lop"));
assertTrue(strings.contains("ripple"));
+ assertTrue(strings.contains("marko-is-my-name"));
+ assertTrue(strings.contains("the-v-o-double-g"));
}
@@ -151,7 +157,7 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
/////////////////////
- public static class TestProgram extends StaticVertexProgram {
+ public static class TestProgram implements VertexProgram {
private PureTraversal<?, ?> traversal = new PureTraversal<>(EmptyTraversal.instance());
private Step programStep = EmptyStep.instance();
@@ -160,8 +166,8 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
@Override
public void loadState(final Graph graph, final Configuration configuration) {
- super.loadState(graph, configuration);
- this.traversal = PureTraversal.loadState(configuration, TraversalVertexProgram.TRAVERSAL, graph);
+ VertexProgram.super.loadState(graph, configuration);
+ this.traversal = PureTraversal.loadState(configuration, VertexProgramStep.ROOT_TRAVERSAL, graph);
this.programStep = new TraversalMatrix<>(this.traversal.get()).getStepById(configuration.getString(ProgramVertexProgramStep.STEP_ID));
this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.HALTED_TRAVERSERS, Operator.addAll, false, false));
@@ -170,18 +176,18 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
@Override
public void storeState(final Configuration configuration) {
- super.storeState(configuration);
- this.traversal.storeState(configuration, TraversalVertexProgram.TRAVERSAL);
+ VertexProgram.super.storeState(configuration);
+ this.traversal.storeState(configuration, VertexProgramStep.ROOT_TRAVERSAL);
configuration.setProperty(ProgramVertexProgramStep.STEP_ID, this.programStep.getId());
}
@Override
public void setup(final Memory memory) {
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, false);
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.SETUP);
final TraversalSideEffects sideEffects = this.traversal.get().getSideEffects();
final TraverserSet<Object> haltedTraversers = sideEffects.get(TraversalVertexProgram.HALTED_TRAVERSERS);
sideEffects.remove(TraversalVertexProgram.HALTED_TRAVERSERS);
- this.checkSideEffects();
+ this.checkSideEffects(MemoryTraversalSideEffects.State.SETUP);
final Map<Vertex, Long> map = (Map<Vertex, Long>) haltedTraversers.iterator().next().get();
assertEquals(2, map.size());
assertTrue(map.values().contains(3l));
@@ -194,8 +200,10 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
@Override
public void execute(final Vertex vertex, final Messenger messenger, final Memory memory) {
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, true);
- this.checkSideEffects();
+ assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.EXECUTE);
+ this.checkSideEffects(MemoryTraversalSideEffects.State.EXECUTE);
final TraverserSet<Vertex> activeTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
if (vertex.label().equals("software")) {
assertEquals(1, activeTraversers.stream().filter(v -> v.get().equals(vertex)).count());
@@ -203,37 +211,53 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
assertFalse(vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS).isPresent());
vertex.property(
TraversalVertexProgram.HALTED_TRAVERSERS,
- new TraverserSet<>(this.traversal.get().getTraverserGenerator().generate(vertex.value("name"), this.programStep, 1l)));
+ new TraverserSet<>(generator.generate(vertex.value("name"), this.programStep, 1l)));
} else {
assertTrue(vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS).isPresent());
}
} else {
assertFalse(vertex.property(TraversalVertexProgram.HALTED_TRAVERSERS).isPresent());
assertEquals(0, activeTraversers.stream().filter(v -> v.get().equals(vertex)).count());
+ if (!memory.isInitialIteration()) {
+ if (vertex.value("name").equals("marko"))
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(generator.generate("marko-is-my-name", this.programStep, 1l)));
+ else if (vertex.value("name").equals("vadas"))
+ this.traversal.get().getSideEffects().add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(generator.generate("the-v-o-double-g", this.programStep, 1l)));
+ }
}
}
@Override
public boolean terminate(final Memory memory) {
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, false);
- checkSideEffects();
- final TraverserSet<String> haltedTraversers = new TraverserSet<>();
- haltedTraversers.add(this.traversal.get().getTraverserGenerator().generate("hello", this.programStep, 1l));
- haltedTraversers.add(this.traversal.get().getTraverserGenerator().generate("gremlin", this.programStep, 1l));
- memory.set(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversers);
- return !memory.isInitialIteration();
+ final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.TERMINATE);
+ checkSideEffects(MemoryTraversalSideEffects.State.TERMINATE);
+ if (memory.isInitialIteration()) {
+ assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ return false;
+ } else {
+ ///
+ assertTrue(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ final TraverserSet<String> haltedTraversers = memory.get(TraversalVertexProgram.HALTED_TRAVERSERS);
+ haltedTraversers.add(generator.generate("hello", this.programStep, 1l));
+ haltedTraversers.add(generator.generate("gremlin", this.programStep, 1l));
+ memory.set(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversers);
+ return true;
+ }
}
@Override
public void workerIterationStart(final Memory memory) {
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, true);
- checkSideEffects();
+ assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.WORKER_ITERATION_START);
+ checkSideEffects(MemoryTraversalSideEffects.State.WORKER_ITERATION_START);
}
@Override
public void workerIterationEnd(final Memory memory) {
- MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, true);
- checkSideEffects();
+ assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, MemoryTraversalSideEffects.State.WORKER_ITERATION_END);
+ checkSideEffects(MemoryTraversalSideEffects.State.WORKER_ITERATION_END);
}
@Override
@@ -252,6 +276,18 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
}
@Override
+ public TestProgram clone() {
+ try {
+ final TestProgram clone = (TestProgram) super.clone();
+ clone.traversal = this.traversal.clone();
+ clone.programStep = new TraversalMatrix<>(clone.traversal.get()).getStepById(this.programStep.getId());
+ return clone;
+ } catch (final CloneNotSupportedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ @Override
public GraphComputer.ResultGraph getPreferredResultGraph() {
return GraphComputer.ResultGraph.NEW;
}
@@ -263,12 +299,17 @@ public abstract class ProgramTest extends AbstractGremlinProcessTest {
////////
- private void checkSideEffects() {
+ private void checkSideEffects(final MemoryTraversalSideEffects.State state) {
final TraversalSideEffects sideEffects = this.traversal.get().getSideEffects();
assertTrue(sideEffects instanceof MemoryTraversalSideEffects);
-// assertEquals(1, sideEffects.keys().size());
+ if (state.masterState()) {
+ assertEquals(1, sideEffects.keys().size());
+ assertFalse(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ } else {
+ assertEquals(2, sideEffects.keys().size());
+ assertTrue(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
+ }
assertTrue(sideEffects.exists("x"));
-// assertFalse(sideEffects.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
final BulkSet<String> bulkSet = sideEffects.get("x");
assertEquals(4, bulkSet.size());
assertEquals(4, bulkSet.get("java"));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6acbf96a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
index 4980832..6e35cf8 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/traversal/strategy/optimization/interceptor/SparkStarBarrierInterceptor.java
@@ -75,7 +75,7 @@ public final class SparkStarBarrierInterceptor implements SparkVertexProgramInte
traversal.applyStrategies(); // compile
boolean identityTraversal = traversal.getSteps().isEmpty(); // if the traversal is empty, just return the vertex (fast)
///////////////////////////////
- MemoryTraversalSideEffects.setMemorySideEffects(traversal, memory, true); // any intermediate sideEffect steps are backed by SparkMemory
+ MemoryTraversalSideEffects.setMemorySideEffects(traversal, memory, MemoryTraversalSideEffects.State.EXECUTE); // any intermediate sideEffect steps are backed by SparkMemory
memory.setInExecute(true);
final JavaRDD<Traverser.Admin<Object>> nextRDD = inputRDD.values()
.filter(vertexWritable -> ElementHelper.idExists(vertexWritable.get().id(), graphStepIds)) // ensure vertex ids are in V(x)