You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2016/05/27 15:55:16 UTC
[01/10] incubator-tinkerpop git commit: TraversalVertexProgram can
use DetachFactory or ReferenceFactory when sending halted traversers to the
master traversal. The default is to use ReferenceFactory (more efficient),
but if someone wants to use Deatched
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 86d8b5cdf -> 2c089280a
TraversalVertexProgram can use DetachFactory or ReferenceFactory when sending halted traversers to the master traversal. The default is to use ReferenceFactory (more efficient), but if someone wants to use DeatchedFactory, then they do g.withComputer().withStrategies(HaltedTraverserFactoryStrategy.detached()). Easy peasy lemon squeezy. Still need to write specific test cases, but manual testing shows things workings.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/b72ea666
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/b72ea666
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/b72ea666
Branch: refs/heads/master
Commit: b72ea666ce18920a8dd7d1d4c30bbe029fe4ae80
Parents: afd4048
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue May 24 16:50:19 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue May 24 16:50:19 2016 -0600
----------------------------------------------------------------------
.../computer/traversal/MasterExecutor.java | 32 +++++++-----
.../traversal/TraversalVertexProgram.java | 30 +++++++----
.../computer/traversal/WorkerExecutor.java | 44 +++++++++-------
.../step/map/TraversalVertexProgramStep.java | 10 +++-
.../HaltedTraverserFactoryStrategy.java | 53 ++++++++++++++++++++
5 files changed, 126 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b72ea666/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 88570fe..b5ec12b 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
@@ -42,6 +42,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.HashSet;
@@ -57,6 +58,16 @@ final class MasterExecutor {
}
+ protected static <R> Traverser.Admin<R> detach(final Traverser.Admin<R> traverser, final Class haltedTraverserFactory) {
+ if (haltedTraverserFactory.equals(DetachedFactory.class))
+ traverser.set(DetachedFactory.detach(traverser.get(), true));
+ else if (haltedTraverserFactory.equals(ReferenceFactory.class))
+ traverser.set(ReferenceFactory.detach(traverser.get()));
+ else
+ throw new IllegalArgumentException("The following detaching factory is unknown: " + haltedTraverserFactory);
+ return traverser;
+ }
+
protected static void processMemory(final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final TraverserSet<Object> traverserSet, final Set<String> completedBarriers) {
if (memory.exists(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
for (final String key : memory.<Set<String>>get(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
@@ -82,7 +93,8 @@ final class MasterExecutor {
final TraversalMatrix<?, ?> traversalMatrix,
TraverserSet<Object> toProcessTraversers,
final TraverserSet<Object> remoteActiveTraversers,
- final TraverserSet<Object> haltedTraversers) {
+ final TraverserSet<Object> haltedTraversers,
+ final Class haltedTraverserFactory) {
while (!toProcessTraversers.isEmpty()) {
@@ -96,23 +108,19 @@ final class MasterExecutor {
traverser.set(DetachedFactory.detach(traverser.get(), true));
traverser.setSideEffects(traversal.get().getSideEffects());
if (traverser.isHalted()) {
- traverser.detach();
- haltedTraversers.add(traverser);
+ haltedTraversers.add(MasterExecutor.detach(traverser, haltedTraverserFactory));
} else if (isRemoteTraverser(traverser, traversalMatrix)) { // this is so that patterns like order().name work as expected.
- traverser.detach();
- remoteActiveTraversers.add(traverser);
+ remoteActiveTraversers.add(traverser.detach());
} else {
currentStep = traversalMatrix.getStepById(traverser.getStepId());
if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep)) {
while (previousStep.hasNext()) {
final Traverser.Admin<Object> result = previousStep.next();
if (result.isHalted()) {
- result.detach();
- haltedTraversers.add(result);
+ haltedTraversers.add(MasterExecutor.detach(result, haltedTraverserFactory));
} else {
if (isRemoteTraverser(result, traversalMatrix)) {
- result.detach();
- remoteActiveTraversers.add(result);
+ remoteActiveTraversers.add(result.detach());
} else
localActiveTraversers.add(result);
}
@@ -126,12 +134,10 @@ final class MasterExecutor {
while (currentStep.hasNext()) {
final Traverser.Admin<Object> traverser = currentStep.next();
if (traverser.isHalted()) {
- traverser.detach();
- haltedTraversers.add(traverser);
+ haltedTraversers.add(MasterExecutor.detach(traverser, haltedTraverserFactory));
} else {
if (isRemoteTraverser(traverser, traversalMatrix)) {
- traverser.detach();
- remoteActiveTraversers.add(traverser);
+ remoteActiveTraversers.add(traverser.detach());
} else
localActiveTraversers.add(traverser);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b72ea666/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 266426f..d4daaac 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
@@ -64,6 +64,7 @@ import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.function.MutableMetricsSupplier;
import org.apache.tinkerpop.gremlin.util.iterator.EmptyIterator;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
@@ -93,6 +94,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
public static final String TRAVERSAL = "gremlin.traversalVertexProgram.traversal";
public static final String HALTED_TRAVERSERS = "gremlin.traversalVertexProgram.haltedTraversers";
public static final String ACTIVE_TRAVERSERS = "gremlin.traversalVertexProgram.activeTraversers";
+ public static final String HALTED_TRAVERSER_FACTORY = "gremlin.traversalVertexProgram.haltedTraverserFactory";
protected static final String MUTATED_MEMORY_KEYS = "gremlin.traversalVertexProgram.mutatedMemoryKeys";
private static final String VOTE_TO_HALT = "gremlin.traversalVertexProgram.voteToHalt";
private static final String COMPLETED_BARRIERS = "gremlin.traversalVertexProgram.completedBarriers";
@@ -108,6 +110,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
private final Set<MapReduce> mapReducers = new HashSet<>();
private TraverserSet<Object> haltedTraversers;
private boolean returnHaltedTraversers = false;
+ private Class haltedTraverserDetachFactory;
private TraversalVertexProgram() {
}
@@ -164,6 +167,8 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
this.traversal.get().getParent().asStep().getNextStep() instanceof EmptyStep || // same as above, but if using TraversalVertexProgramStep directly
(this.traversal.get().getParent().asStep().getNextStep() instanceof ProfileStep && // same as above, but needed for profiling
this.traversal.get().getParent().asStep().getNextStep().getNextStep() instanceof ComputerResultStep));
+ // determine how to store halted traversers
+ this.haltedTraverserDetachFactory = configuration.containsKey(HALTED_TRAVERSER_FACTORY) ? (Class) configuration.getProperty(HALTED_TRAVERSER_FACTORY) : ReferenceFactory.class;
// register traversal side-effects in memory
this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
// register MapReducer memory compute keys
@@ -191,6 +196,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
public void storeState(final Configuration configuration) {
VertexProgram.super.storeState(configuration);
this.traversal.storeState(configuration, TRAVERSAL);
+ configuration.setProperty(HALTED_TRAVERSER_FACTORY, this.haltedTraverserDetachFactory);
TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
}
@@ -211,7 +217,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
});
assert haltedTraversers.isEmpty();
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
- MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers);
+ MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers, this.haltedTraverserDetachFactory);
memory.set(HALTED_TRAVERSERS, this.haltedTraversers);
memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
} else {
@@ -265,18 +271,17 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
graphStep.setIteratorSupplier(() -> (Iterator) IteratorUtils.filter(vertex.edges(Direction.OUT), edge -> ElementHelper.idExists(edge.id(), graphStep.getIds())));
graphStep.forEachRemaining(traverser -> {
if (traverser.isHalted()) {
- traverser.detach();
if (this.returnHaltedTraversers)
- memory.add(HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ memory.add(HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, this.haltedTraverserDetachFactory)));
else
- haltedTraversers.add((Traverser.Admin) traverser);
+ haltedTraversers.add((Traverser.Admin) traverser.detach());
} else
activeTraversers.add((Traverser.Admin) traverser);
});
}
- memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers));
+ memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserDetachFactory));
} else { // ITERATION 1+
- memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers));
+ memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserDetachFactory));
}
if (this.returnHaltedTraversers || haltedTraversers.isEmpty())
vertex.<TraverserSet>property(HALTED_TRAVERSERS).remove();
@@ -298,7 +303,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
final Set<String> completedBarriers = new HashSet<>();
MasterExecutor.processMemory(this.traversalMatrix, memory, toProcessTraversers, completedBarriers);
// process all results from barriers locally and when elements are touched, put them in remoteActiveTraversers
- MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers);
+ MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers, this.haltedTraverserDetachFactory);
// tell parallel barriers that might not have been active in the last round that they are no longer active
memory.set(COMPLETED_BARRIERS, completedBarriers);
if (!remoteActiveTraversers.isEmpty() ||
@@ -308,11 +313,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
return false;
} else {
// finalize locally with any last traversers dangling in the local traversal
- final Step<?, ?> endStep = this.traversal.get().getEndStep();
+ final Step<?, Object> endStep = (Step<?, Object>) this.traversal.get().getEndStep();
while (endStep.hasNext()) {
- final Traverser.Admin traverser = endStep.next();
- traverser.detach();
- haltedTraversers.add(traverser);
+ haltedTraversers.add(MasterExecutor.detach(endStep.next(), this.haltedTraverserDetachFactory));
}
// the result of a TraversalVertexProgram are the halted traversers
memory.set(HALTED_TRAVERSERS, haltedTraversers);
@@ -409,6 +412,11 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
return this;
}
+ public Builder haltedTraverserFactory(final Class detachFactory) {
+ this.configuration.setProperty(HALTED_TRAVERSER_FACTORY, detachFactory);
+ return this;
+ }
+
public Builder traversal(final TraversalSource traversalSource, final String scriptEngine, final String traversalScript, final Object... bindings) {
return this.traversal(new ScriptTraversal<>(traversalSource, scriptEngine, traversalScript, bindings));
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b72ea666/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
index c4bd659..5bc3da9 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
@@ -35,6 +35,7 @@ import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.Collections;
@@ -51,7 +52,12 @@ final class WorkerExecutor {
}
- protected static boolean execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final boolean returnHaltedTraversers) {
+ protected static boolean execute(final Vertex vertex,
+ final Messenger<TraverserSet<Object>> messenger,
+ final TraversalMatrix<?, ?> traversalMatrix,
+ final Memory memory,
+ final boolean returnHaltedTraversers,
+ final Class haltedTraverserFactory) {
final TraversalSideEffects traversalSideEffects = traversalMatrix.getTraversal().getSideEffects();
final AtomicBoolean voteToHalt = new AtomicBoolean(true);
@@ -93,7 +99,7 @@ final class WorkerExecutor {
traversers.remove();
if (traverser.isHalted()) {
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
else
haltedTraversers.add(traverser);
} else {
@@ -118,11 +124,11 @@ final class WorkerExecutor {
traversers.remove();
final Step<Object, Object> currentStep = traversalMatrix.getStepById(traverser.getStepId());
if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep))
- WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers);
+ WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserFactory);
currentStep.addStart(traverser);
previousStep = currentStep;
}
- WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers);
+ WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserFactory);
assert toProcessTraversers.isEmpty();
// process all the local objects and send messages or store locally again
if (!activeTraversers.isEmpty()) {
@@ -135,8 +141,7 @@ final class WorkerExecutor {
final Vertex hostingVertex = WorkerExecutor.getHostingVertex(traverser.get());
if (!vertex.equals(hostingVertex)) { // necessary for path access
voteToHalt.set(false);
- traverser.detach();
- messenger.sendMessage(MessageScope.Global.of(hostingVertex), new TraverserSet<>(traverser));
+ messenger.sendMessage(MessageScope.Global.of(hostingVertex), new TraverserSet<>(traverser.detach()));
} else {
if (traverser.get() instanceof Attachable) // necessary for path access to local object
traverser.attach(Attachable.Method.get(vertex));
@@ -151,7 +156,13 @@ final class WorkerExecutor {
return voteToHalt.get();
}
- private static void drainStep(final Vertex vertex, final Step<Object, Object> step, final TraverserSet<Object> activeTraversers, final TraverserSet<Object> haltedTraversers, final Memory memory, final boolean returnHaltedTraversers) {
+ private static void drainStep(final Vertex vertex,
+ final Step<Object, Object> step,
+ final TraverserSet<Object> activeTraversers,
+ final TraverserSet<Object> haltedTraversers,
+ final Memory memory,
+ final boolean returnHaltedTraversers,
+ final Class haltedTraverserFactory) {
if (step instanceof Barrier) {
if (step instanceof Bypassing)
((Bypassing) step).setBypass(true);
@@ -167,15 +178,12 @@ final class WorkerExecutor {
(returnHaltedTraversers ||
(!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
getHostingVertex(traverser.get()).equals(vertex))) {
- traverser.detach();
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
else
- haltedTraversers.add(traverser);
- } else {
- traverser.detach();
- traverserSet.add(traverser);
- }
+ haltedTraversers.add(traverser.detach());
+ } else
+ traverserSet.add(traverser.detach());
});
}
memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
@@ -189,14 +197,14 @@ final class WorkerExecutor {
} else { // LOCAL PROCESSING
step.forEachRemaining(traverser -> {
if (traverser.isHalted() &&
- (returnHaltedTraversers ||
+ // if its a ReferenceFactory (one less iteration)
+ ((returnHaltedTraversers || haltedTraverserFactory == ReferenceFactory.class) &&
(!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
getHostingVertex(traverser.get()).equals(vertex))) {
- traverser.detach();
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(traverser));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
else
- haltedTraversers.add(traverser);
+ haltedTraversers.add(traverser.detach());
} else {
activeTraversers.add(traverser);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b72ea666/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
index 58e44a2..2b2498b 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
@@ -30,6 +30,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequire
import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import java.util.Collections;
import java.util.List;
@@ -41,6 +42,7 @@ import java.util.Set;
public final class TraversalVertexProgramStep extends VertexProgramStep implements TraversalParent {
public PureTraversal<?, ?> computerTraversal;
+ private Class haltedTraverserFactory = ReferenceFactory.class;
public TraversalVertexProgramStep(final Traversal.Admin traversal, final Traversal.Admin<?, ?> computerTraversal) {
super(traversal);
@@ -69,7 +71,9 @@ public final class TraversalVertexProgramStep extends VertexProgramStep implemen
this.getTraversal().getStrategies().toList().forEach(computerSpecificTraversal.getStrategies()::addStrategies);
computerSpecificTraversal.setSideEffects(new MemoryTraversalSideEffects(this.getTraversal().getSideEffects()));
computerSpecificTraversal.setParent(this);
- final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build().traversal(computerSpecificTraversal);
+ final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build()
+ .traversal(computerSpecificTraversal)
+ .haltedTraverserFactory(this.haltedTraverserFactory);
if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
builder.haltedTraversers(memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
return builder.create(graph);
@@ -96,6 +100,10 @@ public final class TraversalVertexProgramStep extends VertexProgramStep implemen
this.integrateChild(this.computerTraversal.get());
}
+ public void setHaltedTraverserFactory(final Class haltedTraverserDetachFactory) {
+ this.haltedTraverserFactory = haltedTraverserDetachFactory;
+ }
+
/*@Override
public int hashCode() {
return super.hashCode() ^ this.computerTraversal.hashCode();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b72ea666/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
new file mode 100644
index 0000000..44542c7
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
@@ -0,0 +1,53 @@
+/*
+ * 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.computer.traversal.strategy.decoration;
+
+import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class HaltedTraverserFactoryStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy> implements TraversalStrategy.DecorationStrategy {
+
+ private final Class haltedTraverserFactory;
+
+ private HaltedTraverserFactoryStrategy(final Class haltedTraverserFactory) {
+ this.haltedTraverserFactory = haltedTraverserFactory;
+ }
+
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ TraversalHelper.getStepsOfAssignableClass(TraversalVertexProgramStep.class, traversal)
+ .forEach(step -> step.setHaltedTraverserFactory(this.haltedTraverserFactory));
+ }
+
+ public static HaltedTraverserFactoryStrategy detach() {
+ return new HaltedTraverserFactoryStrategy(DetachedFactory.class);
+ }
+
+ public static HaltedTraverserFactoryStrategy reference() {
+ return new HaltedTraverserFactoryStrategy(ReferenceFactory.class);
+ }
+}
[07/10] incubator-tinkerpop git commit: Moved MasterExecutor.detach()
to HaltedTraverserStrategy.halt() as this is now generally useful for OLTP
and not just OLAP. Lots more nick nack cleanups and comments.
Posted by ok...@apache.org.
Moved MasterExecutor.detach() to HaltedTraverserStrategy.halt() as this is now generally useful for OLTP and not just OLAP. Lots more nick nack cleanups and comments.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/56c43ea7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/56c43ea7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/56c43ea7
Branch: refs/heads/master
Commit: 56c43ea7bd89e0ef72b958400d7bd83a10ad6a15
Parents: f78d1cb
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu May 26 07:38:12 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu May 26 07:38:12 2016 -0600
----------------------------------------------------------------------
.../computer/traversal/MasterExecutor.java | 21 ++-------
.../traversal/TraversalVertexProgram.java | 21 +++++----
.../computer/traversal/WorkerExecutor.java | 49 ++++++++++++--------
.../decoration/HaltedTraverserStrategy.java | 17 ++++++-
4 files changed, 61 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/56c43ea7/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 1c1e9d2..0462950 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
@@ -20,6 +20,7 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.Path;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
@@ -57,18 +58,6 @@ final class MasterExecutor {
}
- // the halted traversers can either be reference or detached elements -- this is good for determining how much data around the element the user wants to get
- // see HaltedTraverserFactoryStrategy for how this is all connected
- protected static <R> Traverser.Admin<R> detach(final Traverser.Admin<R> traverser, final Class haltedTraverserFactory) {
- if (haltedTraverserFactory.equals(DetachedFactory.class))
- traverser.set(DetachedFactory.detach(traverser.get(), true));
- else if (haltedTraverserFactory.equals(ReferenceFactory.class))
- traverser.set(ReferenceFactory.detach(traverser.get()));
- else
- throw new IllegalArgumentException("The following detaching factory is unknown: " + haltedTraverserFactory);
- return traverser;
- }
-
protected static void processMemory(final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final TraverserSet<Object> toProcessTraversers, final Set<String> completedBarriers) {
// handle traversers and data that were sent from the workers to the master traversal via memory
if (memory.exists(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
@@ -94,7 +83,7 @@ final class MasterExecutor {
TraverserSet<Object> toProcessTraversers,
final TraverserSet<Object> remoteActiveTraversers,
final TraverserSet<Object> haltedTraversers,
- final Class haltedTraverserFactory) {
+ final HaltedTraverserStrategy haltedTraverserStrategy) {
while (!toProcessTraversers.isEmpty()) {
final TraverserSet<Object> localActiveTraversers = new TraverserSet<>();
@@ -109,7 +98,7 @@ final class MasterExecutor {
traverser.set(DetachedFactory.detach(traverser.get(), true)); // why?
traverser.setSideEffects(traversal.get().getSideEffects());
if (traverser.isHalted())
- haltedTraversers.add(MasterExecutor.detach(traverser, haltedTraverserFactory));
+ haltedTraversers.add(haltedTraverserStrategy.halt(traverser));
else if (isRemoteTraverser(traverser, traversalMatrix)) // this is so that patterns like order().name work as expected. try and stay local as long as possible
remoteActiveTraversers.add(traverser.detach());
else {
@@ -118,7 +107,7 @@ final class MasterExecutor {
while (previousStep.hasNext()) {
final Traverser.Admin<Object> result = previousStep.next();
if (result.isHalted())
- haltedTraversers.add(MasterExecutor.detach(result, haltedTraverserFactory));
+ haltedTraversers.add(haltedTraverserStrategy.halt(traverser));
else if (isRemoteTraverser(result, traversalMatrix))
remoteActiveTraversers.add(result.detach());
else
@@ -133,7 +122,7 @@ final class MasterExecutor {
while (currentStep.hasNext()) {
final Traverser.Admin<Object> traverser = currentStep.next();
if (traverser.isHalted())
- haltedTraversers.add(MasterExecutor.detach(traverser, haltedTraverserFactory));
+ haltedTraversers.add(haltedTraverserStrategy.halt(traverser));
else if (isRemoteTraverser(traverser, traversalMatrix))
remoteActiveTraversers.add(traverser.detach());
else
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/56c43ea7/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 07ab98a..c9dc053 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
@@ -109,7 +109,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
private final Set<MapReduce> mapReducers = new HashSet<>();
private TraverserSet<Object> haltedTraversers;
private boolean returnHaltedTraversers = false;
- private Class haltedTraverserDetachFactory;
+ private HaltedTraverserStrategy haltedTraverserStrategy;
private TraversalVertexProgram() {
}
@@ -167,11 +167,11 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
(this.traversal.get().getParent().asStep().getNextStep() instanceof ProfileStep && // same as above, but needed for profiling
this.traversal.get().getParent().asStep().getNextStep().getNextStep() instanceof ComputerResultStep));
// determine how to store halted traversers
- this.haltedTraverserDetachFactory = ((HaltedTraverserStrategy) this.traversal.get().getStrategies().toList()
+ this.haltedTraverserStrategy = ((HaltedTraverserStrategy) this.traversal.get().getStrategies().toList()
.stream()
.filter(strategy -> strategy instanceof HaltedTraverserStrategy)
.findAny()
- .orElse(HaltedTraverserStrategy.reference())).getHaltedTraverserFactory();
+ .orElse(HaltedTraverserStrategy.reference()));
// register traversal side-effects in memory
this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
// register MapReducer memory compute keys
@@ -217,9 +217,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
traverser.setStepId(this.traversal.get().getStartStep().getId());
toProcessTraversers.add(traverser);
});
- assert haltedTraversers.isEmpty();
+ assert this.haltedTraversers.isEmpty();
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
- MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers, this.haltedTraverserDetachFactory);
+ MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers, this.haltedTraverserStrategy);
memory.set(HALTED_TRAVERSERS, this.haltedTraversers);
memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
} else {
@@ -238,6 +238,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public void execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final Memory memory) {
// if any global halted traversers, simply don't use them as they were handled by master setup()
+ // these halted traversers are typically from a previous OLAP job that yielded traversers at the master traversal
if (null != this.haltedTraversers)
this.haltedTraversers = null;
// memory is distributed
@@ -274,16 +275,16 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
graphStep.forEachRemaining(traverser -> {
if (traverser.isHalted()) {
if (this.returnHaltedTraversers)
- memory.add(HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, this.haltedTraverserDetachFactory)));
+ memory.add(HALTED_TRAVERSERS, new TraverserSet<>(this.haltedTraverserStrategy.halt(traverser)));
else
haltedTraversers.add((Traverser.Admin) traverser.detach());
} else
activeTraversers.add((Traverser.Admin) traverser);
});
}
- memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserDetachFactory));
+ memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserStrategy));
} else // ITERATION 1+
- memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserDetachFactory));
+ memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserStrategy));
// save space by not having an empty halted traversers property
if (this.returnHaltedTraversers || haltedTraversers.isEmpty())
vertex.<TraverserSet>property(HALTED_TRAVERSERS).remove();
@@ -307,7 +308,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
final Set<String> completedBarriers = new HashSet<>();
MasterExecutor.processMemory(this.traversalMatrix, memory, toProcessTraversers, completedBarriers);
// process all results from barriers locally and when elements are touched, put them in remoteActiveTraversers
- MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers, this.haltedTraverserDetachFactory);
+ MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers, this.haltedTraverserStrategy);
// tell parallel barriers that might not have been active in the last round that they are no longer active
memory.set(COMPLETED_BARRIERS, completedBarriers);
if (!remoteActiveTraversers.isEmpty() ||
@@ -319,7 +320,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
// finalize locally with any last traversers dangling in the local traversal
final Step<?, Object> endStep = (Step<?, Object>) this.traversal.get().getEndStep();
while (endStep.hasNext()) {
- haltedTraversers.add(MasterExecutor.detach(endStep.next(), this.haltedTraverserDetachFactory));
+ haltedTraversers.add(this.haltedTraverserStrategy.halt(endStep.next()));
}
// the result of a TraversalVertexProgram are the halted traversers
memory.set(HALTED_TRAVERSERS, haltedTraversers);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/56c43ea7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
index 5798af0..7181992 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
@@ -57,7 +58,7 @@ final class WorkerExecutor {
final TraversalMatrix<?, ?> traversalMatrix,
final Memory memory,
final boolean returnHaltedTraversers,
- final Class haltedTraverserFactory) {
+ final HaltedTraverserStrategy haltedTraverserStrategy) {
final TraversalSideEffects traversalSideEffects = traversalMatrix.getTraversal().getSideEffects();
final AtomicBoolean voteToHalt = new AtomicBoolean(true);
final TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
@@ -68,6 +69,7 @@ final class WorkerExecutor {
// GENERATE LOCAL TRAVERSERS //
///////////////////////////////
+ // MASTER ACTIVE
// these are traversers that are going from OLTP (master) to OLAP (workers)
// these traversers were broadcasted from the master traversal to the workers for attachment
final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
@@ -75,17 +77,21 @@ final class WorkerExecutor {
// its better to reduce the memory footprint and shorten the active traverser list so synchronization is worth it.
// most distributed OLAP systems have the memory partitioned and thus, this synchronization does nothing.
synchronized (maybeActiveTraversers) {
- final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
- while (iterator.hasNext()) {
- final Traverser.Admin<Object> traverser = iterator.next();
- if (vertex.equals(WorkerExecutor.getHostingVertex(traverser.get()))) {
- iterator.remove();
- traverser.attach(Attachable.Method.get(vertex));
- traverser.setSideEffects(traversalSideEffects);
- toProcessTraversers.add(traverser);
+ if (!maybeActiveTraversers.isEmpty()) {
+ final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
+ while (iterator.hasNext()) {
+ final Traverser.Admin<Object> traverser = iterator.next();
+ if (vertex.equals(WorkerExecutor.getHostingVertex(traverser.get()))) {
+ iterator.remove();
+ traverser.attach(Attachable.Method.get(vertex));
+ traverser.setSideEffects(traversalSideEffects);
+ toProcessTraversers.add(traverser);
+ }
}
}
}
+
+ // WORKER ACTIVE
// these are traversers that exist from from a local barrier
// these traversers will simply saved at the local vertex while the master traversal synchronized the barrier
vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).ifPresent(previousActiveTraversers -> {
@@ -98,17 +104,20 @@ final class WorkerExecutor {
// remove the property to save space
vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS).remove();
});
+
+ // TRAVERSER MESSAGES (WORKER -> WORKER)
// these are traversers that have been messaged to the vertex from another vertex
final Iterator<TraverserSet<Object>> messages = messenger.receiveMessages();
while (messages.hasNext()) {
IteratorUtils.removeOnNext(messages.next().iterator()).forEachRemaining(traverser -> {
if (traverser.isHalted()) {
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(haltedTraverserStrategy.halt(traverser)));
else
- haltedTraversers.add(traverser.detach());
+ haltedTraversers.add(traverser); // the traverser has already been detached so no need to detach it again
} else {
// traverser is not halted and thus, should be processed locally
+ // attach it and process
traverser.attach(Attachable.Method.get(vertex));
traverser.setSideEffects(traversalSideEffects);
toProcessTraversers.add(traverser);
@@ -120,7 +129,7 @@ final class WorkerExecutor {
// PROCESS LOCAL TRAVERSERS //
//////////////////////////////
- // while there are still local traversers, process them until they leave the vertex or halt (i.e. isHalted()).
+ // while there are still local traversers, process them until they leave the vertex (message pass) or halt (store).
while (!toProcessTraversers.isEmpty()) {
Step<Object, Object> previousStep = EmptyStep.instance();
Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
@@ -130,11 +139,11 @@ final class WorkerExecutor {
final Step<Object, Object> currentStep = traversalMatrix.getStepById(traverser.getStepId());
// try and fill up the current step as much as possible with traversers to get a bulking optimization
if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep))
- WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserFactory);
+ WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserStrategy);
currentStep.addStart(traverser);
previousStep = currentStep;
}
- WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserFactory);
+ WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserStrategy);
// all processed traversers should be either halted or active
assert toProcessTraversers.isEmpty();
// process all the local objects and send messages or store locally again
@@ -151,7 +160,7 @@ final class WorkerExecutor {
voteToHalt.set(false); // if message is passed, then don't vote to halt
messenger.sendMessage(MessageScope.Global.of(hostingVertex), new TraverserSet<>(traverser.detach()));
} else {
- traverser.attach(Attachable.Method.get(vertex));
+ traverser.attach(Attachable.Method.get(vertex)); // necessary for select() steps that reference the current object
toProcessTraversers.add(traverser);
}
} else // STANDARD OBJECT
@@ -169,7 +178,7 @@ final class WorkerExecutor {
final TraverserSet<Object> haltedTraversers,
final Memory memory,
final boolean returnHaltedTraversers,
- final Class haltedTraverserFactory) {
+ final HaltedTraverserStrategy haltedTraverserStrategy) {
if (step instanceof Barrier) {
if (step instanceof Bypassing)
((Bypassing) step).setBypass(true);
@@ -187,7 +196,7 @@ final class WorkerExecutor {
(!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
getHostingVertex(traverser.get()).equals(vertex))) {
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(haltedTraverserStrategy.halt(traverser)));
else
haltedTraversers.add(traverser.detach());
} else
@@ -205,12 +214,12 @@ final class WorkerExecutor {
} else { // LOCAL PROCESSING
step.forEachRemaining(traverser -> {
if (traverser.isHalted() &&
- // if its a ReferenceFactory (one less iteration)
- ((returnHaltedTraversers || ReferenceFactory.class == haltedTraverserFactory) &&
+ // if its a ReferenceFactory (one less iteration required)
+ ((returnHaltedTraversers || ReferenceFactory.class == haltedTraverserStrategy.getHaltedTraverserFactory()) &&
(!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
getHostingVertex(traverser.get()).equals(vertex))) {
if (returnHaltedTraversers)
- memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
+ memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(haltedTraverserStrategy.halt(traverser)));
else
haltedTraversers.add(traverser.detach());
} else {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/56c43ea7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
index fd07e23..968e167 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decorat
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
@@ -33,7 +34,10 @@ public final class HaltedTraverserStrategy extends AbstractTraversalStrategy<Tra
private final Class haltedTraverserFactory;
private HaltedTraverserStrategy(final Class haltedTraverserFactory) {
- this.haltedTraverserFactory = haltedTraverserFactory;
+ if (haltedTraverserFactory.equals(DetachedFactory.class) || haltedTraverserFactory.equals(ReferenceFactory.class))
+ this.haltedTraverserFactory = haltedTraverserFactory;
+ else
+ throw new IllegalArgumentException("The provided traverser detachment factory is unknown: " + haltedTraverserFactory);
}
public void apply(final Traversal.Admin<?, ?> traversal) {
@@ -44,6 +48,16 @@ public final class HaltedTraverserStrategy extends AbstractTraversalStrategy<Tra
return this.haltedTraverserFactory;
}
+ public <R> Traverser.Admin<R> halt(final Traverser.Admin<R> traverser) {
+ if (ReferenceFactory.class.equals(this.haltedTraverserFactory))
+ traverser.set(ReferenceFactory.detach(traverser.get()));
+ else
+ traverser.set(DetachedFactory.detach(traverser.get(), true));
+ return traverser;
+ }
+
+ ////////////
+
public static HaltedTraverserStrategy detached() {
return new HaltedTraverserStrategy(DetachedFactory.class);
}
@@ -51,4 +65,5 @@ public final class HaltedTraverserStrategy extends AbstractTraversalStrategy<Tra
public static HaltedTraverserStrategy reference() {
return new HaltedTraverserStrategy(ReferenceFactory.class);
}
+
}
[10/10] incubator-tinkerpop git commit: updated CHANGELOG.
Posted by ok...@apache.org.
updated CHANGELOG.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/2c089280
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/2c089280
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/2c089280
Branch: refs/heads/master
Commit: 2c089280a20c48464a76eaa083c5aadcf5951d6e
Parents: aa34888
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri May 27 09:55:10 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri May 27 09:55:10 2016 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/2c089280/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 7705701..8f00e2f 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,7 @@ image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/
TinkerPop 3.2.1 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Added `HaltedTraverserStrategy` to allow users to get back different element detachments in OLAP.
* Fixed a `NullPointerException` bug around nested `group()`-steps in OLAP.
* Fixed a severe bug around halted traversers in a multi-job OLAP traversal chain.
* Ensure a separation of `GraphComputer` and `VertexProgram` configurations in `SparkGraphComputer` and `GiraphGraphComputer`.
[04/10] incubator-tinkerpop git commit: Merge branch 'master' into
TINKERPOP-1310
Posted by ok...@apache.org.
Merge branch 'master' into TINKERPOP-1310
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/7255844c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/7255844c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/7255844c
Branch: refs/heads/master
Commit: 7255844c03fc1b6d447a2631250100758c4f84cb
Parents: 3978e7b 6113c92
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 25 12:25:51 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 25 12:25:51 2016 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../process/traversal/step/map/GroupStep.java | 307 ++++++++++++-------
.../step/sideEffect/GroupSideEffectStep.java | 61 ++--
.../step/sideEffect/GroovyGroupTest.groovy | 10 +
.../traversal/step/sideEffect/GroupTest.java | 77 +++++
.../gremlin/hadoop/structure/HadoopGraph.java | 124 ++++++--
.../SparkStarBarrierInterceptor.java | 1 -
.../structure/TinkerGraphPlayTest.java | 4 +-
8 files changed, 390 insertions(+), 195 deletions(-)
----------------------------------------------------------------------
[06/10] incubator-tinkerpop git commit: Giraph integration tests are
passing now. @spmallette and I chatted tonight and came up with some good
ideas regarding how to leverage HaltedTraverserStrategy for both OLTP and
OLAP. In short, 'metadata strategies'
Posted by ok...@apache.org.
Giraph integration tests are passing now. @spmallette and I chatted tonight and came up with some good ideas regarding how to leverage HaltedTraverserStrategy for both OLTP and OLAP. In short, 'metadata strategies' might be the new thing.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/f78d1cb6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/f78d1cb6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/f78d1cb6
Branch: refs/heads/master
Commit: f78d1cb631c91fefb46829840e6f4247955a1a5b
Parents: e6f2caa
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 25 19:58:28 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 25 19:58:28 2016 -0600
----------------------------------------------------------------------
.../traversal/TraversalVertexProgram.java | 19 ++--
.../step/map/TraversalVertexProgramStep.java | 10 +-
.../HaltedTraverserFactoryStrategy.java | 67 ------------
.../decoration/HaltedTraverserStrategy.java | 54 ++++++++++
.../HaltedTraverserFactoryStrategyTest.java | 96 -----------------
.../decoration/HaltedTraverserStrategyTest.java | 102 +++++++++++++++++++
6 files changed, 165 insertions(+), 183 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f78d1cb6/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 4479306..07ab98a 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
@@ -32,6 +32,7 @@ import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.ComputerResultStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
import org.apache.tinkerpop.gremlin.process.computer.util.SingleMessenger;
@@ -64,7 +65,6 @@ import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.function.MutableMetricsSupplier;
import org.apache.tinkerpop.gremlin.util.iterator.EmptyIterator;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
@@ -94,7 +94,6 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
public static final String TRAVERSAL = "gremlin.traversalVertexProgram.traversal";
public static final String HALTED_TRAVERSERS = "gremlin.traversalVertexProgram.haltedTraversers";
public static final String ACTIVE_TRAVERSERS = "gremlin.traversalVertexProgram.activeTraversers";
- public static final String HALTED_TRAVERSER_FACTORY = "gremlin.traversalVertexProgram.haltedTraverserFactory";
protected static final String MUTATED_MEMORY_KEYS = "gremlin.traversalVertexProgram.mutatedMemoryKeys";
private static final String VOTE_TO_HALT = "gremlin.traversalVertexProgram.voteToHalt";
private static final String COMPLETED_BARRIERS = "gremlin.traversalVertexProgram.completedBarriers";
@@ -168,7 +167,11 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
(this.traversal.get().getParent().asStep().getNextStep() instanceof ProfileStep && // same as above, but needed for profiling
this.traversal.get().getParent().asStep().getNextStep().getNextStep() instanceof ComputerResultStep));
// determine how to store halted traversers
- this.haltedTraverserDetachFactory = configuration.containsKey(HALTED_TRAVERSER_FACTORY) ? (Class) configuration.getProperty(HALTED_TRAVERSER_FACTORY) : ReferenceFactory.class;
+ this.haltedTraverserDetachFactory = ((HaltedTraverserStrategy) this.traversal.get().getStrategies().toList()
+ .stream()
+ .filter(strategy -> strategy instanceof HaltedTraverserStrategy)
+ .findAny()
+ .orElse(HaltedTraverserStrategy.reference())).getHaltedTraverserFactory();
// register traversal side-effects in memory
this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
// register MapReducer memory compute keys
@@ -196,7 +199,6 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
public void storeState(final Configuration configuration) {
VertexProgram.super.storeState(configuration);
this.traversal.storeState(configuration, TRAVERSAL);
- configuration.setProperty(HALTED_TRAVERSER_FACTORY, this.haltedTraverserDetachFactory);
TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
}
@@ -280,9 +282,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
});
}
memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserDetachFactory));
- } else { // ITERATION 1+
+ } else // ITERATION 1+
memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers, this.haltedTraverserDetachFactory));
- }
+ // save space by not having an empty halted traversers property
if (this.returnHaltedTraversers || haltedTraversers.isEmpty())
vertex.<TraverserSet>property(HALTED_TRAVERSERS).remove();
}
@@ -414,11 +416,6 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
return this;
}
- public Builder haltedTraverserFactory(final Class detachFactory) {
- this.configuration.setProperty(HALTED_TRAVERSER_FACTORY, detachFactory);
- return this;
- }
-
public Builder traversal(final TraversalSource traversalSource, final String scriptEngine, final String traversalScript, final Object... bindings) {
return this.traversal(new ScriptTraversal<>(traversalSource, scriptEngine, traversalScript, bindings));
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f78d1cb6/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
index 2b2498b..58e44a2 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/TraversalVertexProgramStep.java
@@ -30,7 +30,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequire
import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import java.util.Collections;
import java.util.List;
@@ -42,7 +41,6 @@ import java.util.Set;
public final class TraversalVertexProgramStep extends VertexProgramStep implements TraversalParent {
public PureTraversal<?, ?> computerTraversal;
- private Class haltedTraverserFactory = ReferenceFactory.class;
public TraversalVertexProgramStep(final Traversal.Admin traversal, final Traversal.Admin<?, ?> computerTraversal) {
super(traversal);
@@ -71,9 +69,7 @@ public final class TraversalVertexProgramStep extends VertexProgramStep implemen
this.getTraversal().getStrategies().toList().forEach(computerSpecificTraversal.getStrategies()::addStrategies);
computerSpecificTraversal.setSideEffects(new MemoryTraversalSideEffects(this.getTraversal().getSideEffects()));
computerSpecificTraversal.setParent(this);
- final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build()
- .traversal(computerSpecificTraversal)
- .haltedTraverserFactory(this.haltedTraverserFactory);
+ final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build().traversal(computerSpecificTraversal);
if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
builder.haltedTraversers(memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
return builder.create(graph);
@@ -100,10 +96,6 @@ public final class TraversalVertexProgramStep extends VertexProgramStep implemen
this.integrateChild(this.computerTraversal.get());
}
- public void setHaltedTraverserFactory(final Class haltedTraverserDetachFactory) {
- this.haltedTraverserFactory = haltedTraverserDetachFactory;
- }
-
/*@Override
public int hashCode() {
return super.hashCode() ^ this.computerTraversal.hashCode();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f78d1cb6/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
deleted file mode 100644
index c2f3855..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.computer.traversal.strategy.decoration;
-
-import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class HaltedTraverserFactoryStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy> implements TraversalStrategy.DecorationStrategy {
-
- private final Class haltedTraverserFactory;
-
- private HaltedTraverserFactoryStrategy(final Class haltedTraverserFactory) {
- this.haltedTraverserFactory = haltedTraverserFactory;
- }
-
- public void apply(final Traversal.Admin<?, ?> traversal) {
- // only the root traversal should be processed
- if (traversal.getParent() instanceof EmptyStep) {
- final List<TraversalVertexProgramStep> steps = TraversalHelper.getStepsOfAssignableClass(TraversalVertexProgramStep.class, traversal);
- // only the last step (the one returning data) needs to have a non-reference traverser factory
- if (!steps.isEmpty())
- steps.get(steps.size() - 1).setHaltedTraverserFactory(this.haltedTraverserFactory);
- }
- }
-
- public static HaltedTraverserFactoryStrategy detached() {
- return new HaltedTraverserFactoryStrategy(DetachedFactory.class);
- }
-
- public static HaltedTraverserFactoryStrategy reference() {
- return new HaltedTraverserFactoryStrategy(ReferenceFactory.class);
- }
-
- public Set<Class<? extends DecorationStrategy>> applyPrior() {
- return Collections.singleton(VertexProgramStrategy.class);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f78d1cb6/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
new file mode 100644
index 0000000..fd07e23
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
@@ -0,0 +1,54 @@
+/*
+ * 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.computer.traversal.strategy.decoration;
+
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class HaltedTraverserStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy> implements TraversalStrategy.DecorationStrategy {
+
+ private final Class haltedTraverserFactory;
+
+ private HaltedTraverserStrategy(final Class haltedTraverserFactory) {
+ this.haltedTraverserFactory = haltedTraverserFactory;
+ }
+
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ // do nothing as this is simply a metadata strategy
+ }
+
+ public Class getHaltedTraverserFactory() {
+ return this.haltedTraverserFactory;
+ }
+
+ public static HaltedTraverserStrategy detached() {
+ return new HaltedTraverserStrategy(DetachedFactory.class);
+ }
+
+ public static HaltedTraverserStrategy reference() {
+ return new HaltedTraverserStrategy(ReferenceFactory.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f78d1cb6/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
deleted file mode 100644
index 43bc94e..0000000
--- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.tinkergraph.process.computer.traversal.strategy.decoration;
-
-import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserFactoryStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedPath;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedProperty;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferencePath;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class HaltedTraverserFactoryStrategyTest {
-
- @Before
- public void setup() {
- // necessary as ComputerResult step for testing purposes attaches Attachables
- System.setProperty("is.testing", "false");
- }
-
- @After
- public void shutdown() {
- System.setProperty("is.testing", "true");
- }
-
- @Test
- public void shouldReturnDetachedElements() {
- Graph graph = TinkerFactory.createModern();
- GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserFactoryStrategy.detached());
- g.V().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
- g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(DetachedVertexProperty.class, vertexProperty.getClass()));
- g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
- g.V().out().outE().forEachRemaining(edge -> assertEquals(DetachedEdge.class, edge.getClass()));
- g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(DetachedProperty.class, property.getClass()));
- g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
- g.V().out().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
- g.V().out().out().path().forEachRemaining(path -> assertEquals(DetachedPath.class, path.getClass()));
- }
-
- @Test
- public void shouldReturnReferenceElements() {
- Graph graph = TinkerFactory.createModern();
- GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserFactoryStrategy.reference());
- g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
- g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
- g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
- g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
- g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
- g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
- // the default should be reference elements
- g = graph.traversal().withComputer();
- g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
- g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
- g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
- g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
- g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
- g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f78d1cb6/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
new file mode 100644
index 0000000..9c92efe
--- /dev/null
+++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.tinkergraph.process.computer.traversal.strategy.decoration;
+
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedPath;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedProperty;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferencePath;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class HaltedTraverserStrategyTest {
+
+ @Before
+ public void setup() {
+ // necessary as ComputerResult step for testing purposes attaches Attachables
+ System.setProperty("is.testing", "false");
+ }
+
+ @After
+ public void shutdown() {
+ System.setProperty("is.testing", "true");
+ }
+
+ @Test
+ public void shouldReturnDetachedElements() {
+ final Graph graph = TinkerFactory.createModern();
+ final GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserStrategy.detached());
+ g.V().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(DetachedVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(DetachedEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(DetachedProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(DetachedPath.class, path.getClass()));
+ g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ }
+
+ @Test
+ public void shouldReturnReferenceElements() {
+ final Graph graph = TinkerFactory.createModern();
+ GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserStrategy.reference());
+ g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
+ g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ // the default should be reference elements
+ g = graph.traversal().withComputer();
+ g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
+ g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ }
+
+}
[08/10] incubator-tinkerpop git commit: moved HaltedTraverserStrategy
to the general strategy package as it can now be used with OLTP.
Posted by ok...@apache.org.
moved HaltedTraverserStrategy to the general strategy package as it can now be used with OLTP.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/8be75a92
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/8be75a92
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/8be75a92
Branch: refs/heads/master
Commit: 8be75a92aab1f647db966458fd77fa33eedb7c60
Parents: 56c43ea
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu May 26 12:52:04 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu May 26 12:52:04 2016 -0600
----------------------------------------------------------------------
.../computer/traversal/MasterExecutor.java | 5 +-
.../traversal/TraversalVertexProgram.java | 2 +-
.../computer/traversal/WorkerExecutor.java | 2 +-
.../decoration/HaltedTraverserStrategy.java | 69 -------------
.../decoration/HaltedTraverserStrategy.java | 69 +++++++++++++
.../decoration/HaltedTraverserStrategyTest.java | 102 -------------------
.../decoration/HaltedTraverserStrategyTest.java | 102 +++++++++++++++++++
7 files changed, 175 insertions(+), 176 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/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 0462950..b83b6d6 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
@@ -20,7 +20,7 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.Path;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
@@ -43,7 +43,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import java.util.HashSet;
import java.util.Iterator;
@@ -107,7 +106,7 @@ final class MasterExecutor {
while (previousStep.hasNext()) {
final Traverser.Admin<Object> result = previousStep.next();
if (result.isHalted())
- haltedTraversers.add(haltedTraverserStrategy.halt(traverser));
+ haltedTraversers.add(haltedTraverserStrategy.halt(result));
else if (isRemoteTraverser(result, traversalMatrix))
remoteActiveTraversers.add(result.detach());
else
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/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 c9dc053..1a54721 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
@@ -32,7 +32,7 @@ import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.ComputerResultStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
import org.apache.tinkerpop.gremlin.process.computer.util.SingleMessenger;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
index 7181992..0a3aad2 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
@@ -21,7 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
-import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
deleted file mode 100644
index 968e167..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserStrategy.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.computer.traversal.strategy.decoration;
-
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class HaltedTraverserStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy> implements TraversalStrategy.DecorationStrategy {
-
- private final Class haltedTraverserFactory;
-
- private HaltedTraverserStrategy(final Class haltedTraverserFactory) {
- if (haltedTraverserFactory.equals(DetachedFactory.class) || haltedTraverserFactory.equals(ReferenceFactory.class))
- this.haltedTraverserFactory = haltedTraverserFactory;
- else
- throw new IllegalArgumentException("The provided traverser detachment factory is unknown: " + haltedTraverserFactory);
- }
-
- public void apply(final Traversal.Admin<?, ?> traversal) {
- // do nothing as this is simply a metadata strategy
- }
-
- public Class getHaltedTraverserFactory() {
- return this.haltedTraverserFactory;
- }
-
- public <R> Traverser.Admin<R> halt(final Traverser.Admin<R> traverser) {
- if (ReferenceFactory.class.equals(this.haltedTraverserFactory))
- traverser.set(ReferenceFactory.detach(traverser.get()));
- else
- traverser.set(DetachedFactory.detach(traverser.get(), true));
- return traverser;
- }
-
- ////////////
-
- public static HaltedTraverserStrategy detached() {
- return new HaltedTraverserStrategy(DetachedFactory.class);
- }
-
- public static HaltedTraverserStrategy reference() {
- return new HaltedTraverserStrategy(ReferenceFactory.class);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/HaltedTraverserStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/HaltedTraverserStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/HaltedTraverserStrategy.java
new file mode 100644
index 0000000..45966f0
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/decoration/HaltedTraverserStrategy.java
@@ -0,0 +1,69 @@
+/*
+ * 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.strategy.decoration;
+
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class HaltedTraverserStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy> implements TraversalStrategy.DecorationStrategy {
+
+ private final Class haltedTraverserFactory;
+
+ private HaltedTraverserStrategy(final Class haltedTraverserFactory) {
+ if (haltedTraverserFactory.equals(DetachedFactory.class) || haltedTraverserFactory.equals(ReferenceFactory.class))
+ this.haltedTraverserFactory = haltedTraverserFactory;
+ else
+ throw new IllegalArgumentException("The provided traverser detachment factory is unknown: " + haltedTraverserFactory);
+ }
+
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ // do nothing as this is simply a metadata strategy
+ }
+
+ public Class getHaltedTraverserFactory() {
+ return this.haltedTraverserFactory;
+ }
+
+ public <R> Traverser.Admin<R> halt(final Traverser.Admin<R> traverser) {
+ if (ReferenceFactory.class.equals(this.haltedTraverserFactory))
+ traverser.set(ReferenceFactory.detach(traverser.get()));
+ else
+ traverser.set(DetachedFactory.detach(traverser.get(), true));
+ return traverser;
+ }
+
+ ////////////
+
+ public static HaltedTraverserStrategy detached() {
+ return new HaltedTraverserStrategy(DetachedFactory.class);
+ }
+
+ public static HaltedTraverserStrategy reference() {
+ return new HaltedTraverserStrategy(ReferenceFactory.class);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
deleted file mode 100644
index 9c92efe..0000000
--- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.tinkergraph.process.computer.traversal.strategy.decoration;
-
-import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedPath;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedProperty;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferencePath;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class HaltedTraverserStrategyTest {
-
- @Before
- public void setup() {
- // necessary as ComputerResult step for testing purposes attaches Attachables
- System.setProperty("is.testing", "false");
- }
-
- @After
- public void shutdown() {
- System.setProperty("is.testing", "true");
- }
-
- @Test
- public void shouldReturnDetachedElements() {
- final Graph graph = TinkerFactory.createModern();
- final GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserStrategy.detached());
- g.V().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
- g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(DetachedVertexProperty.class, vertexProperty.getClass()));
- g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
- g.V().out().outE().forEachRemaining(edge -> assertEquals(DetachedEdge.class, edge.getClass()));
- g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(DetachedProperty.class, property.getClass()));
- g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
- g.V().out().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
- g.V().out().out().path().forEachRemaining(path -> assertEquals(DetachedPath.class, path.getClass()));
- g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
- g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
- }
-
- @Test
- public void shouldReturnReferenceElements() {
- final Graph graph = TinkerFactory.createModern();
- GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserStrategy.reference());
- g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
- g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
- g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
- g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
- g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
- g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
- g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- // the default should be reference elements
- g = graph.traversal().withComputer();
- g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
- g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
- g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
- g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
- g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
- g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
- g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8be75a92/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/traversal/strategy/decoration/HaltedTraverserStrategyTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
new file mode 100644
index 0000000..0943391
--- /dev/null
+++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/traversal/strategy/decoration/HaltedTraverserStrategyTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.tinkergraph.process.traversal.strategy.decoration;
+
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.HaltedTraverserStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedPath;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedProperty;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferencePath;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class HaltedTraverserStrategyTest {
+
+ @Before
+ public void setup() {
+ // necessary as ComputerResult step for testing purposes attaches Attachables
+ System.setProperty("is.testing", "false");
+ }
+
+ @After
+ public void shutdown() {
+ System.setProperty("is.testing", "true");
+ }
+
+ @Test
+ public void shouldReturnDetachedElements() {
+ final Graph graph = TinkerFactory.createModern();
+ final GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserStrategy.detached());
+ g.V().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(DetachedVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(DetachedEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(DetachedProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(DetachedPath.class, path.getClass()));
+ g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ }
+
+ @Test
+ public void shouldReturnReferenceElements() {
+ final Graph graph = TinkerFactory.createModern();
+ GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserStrategy.reference());
+ g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
+ g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ // the default should be reference elements
+ g = graph.traversal().withComputer();
+ g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
+ g.V().out().pageRank().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().pageRank().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ }
+
+}
[09/10] incubator-tinkerpop git commit: Merge branch 'TINKERPOP-1310'
Posted by ok...@apache.org.
Merge branch 'TINKERPOP-1310'
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/aa348886
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/aa348886
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/aa348886
Branch: refs/heads/master
Commit: aa34888677e748f7316a0870a3a7cc71c4f52df5
Parents: 86d8b5c 8be75a9
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri May 27 09:54:19 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri May 27 09:54:19 2016 -0600
----------------------------------------------------------------------
.../computer/traversal/MasterExecutor.java | 65 +++++------
.../traversal/TraversalVertexProgram.java | 36 +++---
.../computer/traversal/WorkerExecutor.java | 111 ++++++++++++-------
.../decoration/HaltedTraverserStrategy.java | 69 ++++++++++++
.../gremlin/hadoop/structure/HadoopGraph.java | 2 +-
.../decoration/HaltedTraverserStrategyTest.java | 102 +++++++++++++++++
6 files changed, 290 insertions(+), 95 deletions(-)
----------------------------------------------------------------------
[03/10] incubator-tinkerpop git commit: lots of documentation on
TraversalVertexProgram and got HaltedTraveserFactoryStrategy tested and
optimized.
Posted by ok...@apache.org.
lots of documentation on TraversalVertexProgram and got HaltedTraveserFactoryStrategy tested and optimized.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/3978e7bd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/3978e7bd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/3978e7bd
Branch: refs/heads/master
Commit: 3978e7bda5f4896ea1c2815c889e206afcbaccaa
Parents: 19f16f1
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 25 12:25:31 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 25 12:25:31 2016 -0600
----------------------------------------------------------------------
.../computer/traversal/MasterExecutor.java | 6 +-
.../computer/traversal/WorkerExecutor.java | 56 +++++++-----
.../HaltedTraverserFactoryStrategy.java | 13 ++-
.../HaltedTraverserFactoryStrategyTest.java | 91 ++++++++++++++++++++
4 files changed, 140 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3978e7bd/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 b5ec12b..b994f1e 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
@@ -58,6 +58,8 @@ final class MasterExecutor {
}
+ // the halted traversers can either be reference or detached elements -- this is good for determining how much data around the element the user wants to get
+ // see HaltedTraverserFactoryStrategy for how this is all connected
protected static <R> Traverser.Admin<R> detach(final Traverser.Admin<R> traverser, final Class haltedTraverserFactory) {
if (haltedTraverserFactory.equals(DetachedFactory.class))
traverser.set(DetachedFactory.detach(traverser.get(), true));
@@ -68,11 +70,12 @@ final class MasterExecutor {
return traverser;
}
+ // handle traversers and data that were sent from the workers to the master traversal via memory
protected static void processMemory(final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final TraverserSet<Object> traverserSet, final Set<String> completedBarriers) {
if (memory.exists(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
for (final String key : memory.<Set<String>>get(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
final Step<Object, Object> step = traversalMatrix.getStepById(key);
- if (null == step) continue;
+ if (null == step) continue; // why? how can this happen?
assert step instanceof Barrier;
completedBarriers.add(step.getId());
if (!(step instanceof LocalBarrier)) { // local barriers don't do any processing on the master traversal (they just lock on the workers)
@@ -81,6 +84,7 @@ final class MasterExecutor {
while (step.hasNext()) {
traverserSet.add(step.next());
}
+ // if it was a reducing barrier step, reset the barrier to its seed value
if (step instanceof ReducingBarrierStep)
memory.set(step.getId(), ((ReducingBarrierStep) step).getSeedSupplier().get());
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3978e7bd/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
index 5bc3da9..f833b6f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
+import org.apache.tinkerpop.gremlin.process.computer.util.SingleMessenger;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
@@ -35,6 +36,7 @@ import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceElement;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
@@ -58,30 +60,37 @@ final class WorkerExecutor {
final Memory memory,
final boolean returnHaltedTraversers,
final Class haltedTraverserFactory) {
-
final TraversalSideEffects traversalSideEffects = traversalMatrix.getTraversal().getSideEffects();
final AtomicBoolean voteToHalt = new AtomicBoolean(true);
final TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
final TraverserSet<Object> activeTraversers = new TraverserSet<>();
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
+ final boolean isTesting = Boolean.valueOf(System.getProperty("is.testing", "false"));
////////////////////////////////
// GENERATE LOCAL TRAVERSERS //
///////////////////////////////
- // these are traversers that are going from OLTP to OLAP
- final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
- final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
- while (iterator.hasNext()) {
- final Traverser.Admin<Object> traverser = iterator.next();
- if (vertex.equals(WorkerExecutor.getHostingVertex(traverser.get()))) {
- // iterator.remove(); ConcurrentModificationException
- traverser.attach(Attachable.Method.get(vertex));
- traverser.setSideEffects(traversalSideEffects);
- toProcessTraversers.add(traverser);
+ // some memory systems are interacted by multiple threads and thus, concurrent modification can happen at iterator.remove()
+ // its better to reduce the memory footprint and shorten the active traverser list so synchronization is worth it.
+ // most distributed OLAP systems have the memory partitioned and thus, this synchronization does nothing
+ synchronized (memory) {
+ // these are traversers that are going from OLTP to OLAP
+ // these traversers were broadcasted from the master traversal to the workers for attachment
+ final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
+ final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
+ while (iterator.hasNext()) {
+ final Traverser.Admin<Object> traverser = iterator.next();
+ if (vertex.equals(WorkerExecutor.getHostingVertex(traverser.get()))) {
+ iterator.remove();
+ traverser.attach(Attachable.Method.get(vertex));
+ traverser.setSideEffects(traversalSideEffects);
+ toProcessTraversers.add(traverser);
+ }
}
}
// these are traversers that exist from from a local barrier
+ // these traversers will simply saved at the local vertex while the master traversal synchronized the barrier
vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).ifPresent(previousActiveTraversers -> {
IteratorUtils.removeOnNext(previousActiveTraversers.iterator()).forEachRemaining(traverser -> {
traverser.attach(Attachable.Method.get(vertex));
@@ -90,24 +99,25 @@ final class WorkerExecutor {
});
vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS).remove();
});
- // these are traversers that have been messaged to the vertex
+ // these are traversers that have been messaged to the vertex from another vertex
final Iterator<TraverserSet<Object>> messages = messenger.receiveMessages();
while (messages.hasNext()) {
- final Iterator<Traverser.Admin<Object>> traversers = messages.next().iterator();
- while (traversers.hasNext()) {
- final Traverser.Admin<Object> traverser = traversers.next();
- traversers.remove();
+ IteratorUtils.removeOnNext(messages.next().iterator()).forEachRemaining(traverser -> {
+ // this is internal testing to ensure that messaged elements are always ReferenceXXX and not DetachedXXX (related to HaltedTraverserFactoryStrategy)
+ if (isTesting && !(messenger instanceof SingleMessenger) && traverser.get() instanceof Element)
+ assert traverser.get() instanceof ReferenceElement;
if (traverser.isHalted()) {
if (returnHaltedTraversers)
memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
else
haltedTraversers.add(traverser);
} else {
+ // traverser is not halted and thus, should be processed locally
traverser.attach(Attachable.Method.get(vertex));
traverser.setSideEffects(traversalSideEffects);
toProcessTraversers.add(traverser);
}
- }
+ });
}
///////////////////////////////
@@ -116,7 +126,6 @@ final class WorkerExecutor {
// while there are still local traversers, process them until they leave the vertex or halt (i.e. isHalted()).
while (!toProcessTraversers.isEmpty()) {
- // process local traversers and if alive, repeat, else halt.
Step<Object, Object> previousStep = EmptyStep.instance();
Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
while (traversers.hasNext()) {
@@ -129,6 +138,7 @@ final class WorkerExecutor {
previousStep = currentStep;
}
WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserFactory);
+ // all processed traversers should be either halted or active
assert toProcessTraversers.isEmpty();
// process all the local objects and send messages or store locally again
if (!activeTraversers.isEmpty()) {
@@ -136,6 +146,7 @@ final class WorkerExecutor {
while (traversers.hasNext()) {
final Traverser.Admin<Object> traverser = traversers.next();
traversers.remove();
+ // decide whether to message the traverser or to process it locally
if (traverser.get() instanceof Element || traverser.get() instanceof Property) { // GRAPH OBJECT
// if the element is remote, then message, else store it locally for re-processing
final Vertex hostingVertex = WorkerExecutor.getHostingVertex(traverser.get());
@@ -167,9 +178,10 @@ final class WorkerExecutor {
if (step instanceof Bypassing)
((Bypassing) step).setBypass(true);
if (step instanceof LocalBarrier) {
+ // local barrier traversers are stored on the vertex until the master traversal synchronizes the system
final LocalBarrier<Object> barrier = (LocalBarrier<Object>) step;
- final TraverserSet<Object> traverserSet = vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).orElse(new TraverserSet<>());
- vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS, traverserSet);
+ final TraverserSet<Object> localBarrierTraversers = vertex.<TraverserSet<Object>>property(TraversalVertexProgram.ACTIVE_TRAVERSERS).orElse(new TraverserSet<>());
+ vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS, localBarrierTraversers);
while (barrier.hasNextBarrier()) {
final TraverserSet<Object> barrierSet = barrier.nextBarrier();
IteratorUtils.removeOnNext(barrierSet.iterator()).forEachRemaining(traverser -> {
@@ -183,7 +195,7 @@ final class WorkerExecutor {
else
haltedTraversers.add(traverser.detach());
} else
- traverserSet.add(traverser.detach());
+ localBarrierTraversers.add(traverser);
});
}
memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
@@ -198,7 +210,7 @@ final class WorkerExecutor {
step.forEachRemaining(traverser -> {
if (traverser.isHalted() &&
// if its a ReferenceFactory (one less iteration)
- ((returnHaltedTraversers || haltedTraverserFactory == ReferenceFactory.class) &&
+ ((returnHaltedTraversers || ReferenceFactory.class == haltedTraverserFactory) &&
(!(traverser.get() instanceof Element) && !(traverser.get() instanceof Property)) ||
getHostingVertex(traverser.get()).equals(vertex))) {
if (returnHaltedTraversers)
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3978e7bd/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
index b046986..c2f3855 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
@@ -22,12 +22,14 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decorat
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import java.util.Collections;
+import java.util.List;
import java.util.Set;
/**
@@ -42,11 +44,16 @@ public final class HaltedTraverserFactoryStrategy extends AbstractTraversalStrat
}
public void apply(final Traversal.Admin<?, ?> traversal) {
- TraversalHelper.getStepsOfAssignableClass(TraversalVertexProgramStep.class, traversal)
- .forEach(step -> step.setHaltedTraverserFactory(this.haltedTraverserFactory));
+ // only the root traversal should be processed
+ if (traversal.getParent() instanceof EmptyStep) {
+ final List<TraversalVertexProgramStep> steps = TraversalHelper.getStepsOfAssignableClass(TraversalVertexProgramStep.class, traversal);
+ // only the last step (the one returning data) needs to have a non-reference traverser factory
+ if (!steps.isEmpty())
+ steps.get(steps.size() - 1).setHaltedTraverserFactory(this.haltedTraverserFactory);
+ }
}
- public static HaltedTraverserFactoryStrategy detach() {
+ public static HaltedTraverserFactoryStrategy detached() {
return new HaltedTraverserFactoryStrategy(DetachedFactory.class);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3978e7bd/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
new file mode 100644
index 0000000..af0b3b7
--- /dev/null
+++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.tinkergraph.process.computer.traversal.strategy.decoration;
+
+import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.HaltedTraverserFactoryStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedProperty;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class HaltedTraverserFactoryStrategyTest {
+
+ @Before
+ public void setup() {
+ // necessary as ComputerResult step for testing purposes attaches Attachables
+ System.setProperty("is.testing", "false");
+ }
+
+ @After
+ public void shutdown() {
+ System.setProperty("is.testing", "true");
+ }
+
+ @Test
+ public void shouldReturnDetachedElements() {
+ Graph graph = TinkerFactory.createModern();
+ GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserFactoryStrategy.detached());
+ g.V().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(DetachedVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(DetachedEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(DetachedProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ }
+
+ @Test
+ public void shouldReturnReferenceElements() {
+ Graph graph = TinkerFactory.createModern();
+ GraphTraversalSource g = graph.traversal().withComputer().withStrategies(HaltedTraverserFactoryStrategy.reference());
+ g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ //
+ g = graph.traversal().withComputer();
+ g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
+ g.V().out().values("name").forEachRemaining(value -> assertEquals(String.class, value.getClass()));
+ g.V().out().outE().forEachRemaining(edge -> assertEquals(ReferenceEdge.class, edge.getClass()));
+ g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
+ g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
+ g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ }
+
+}
[02/10] incubator-tinkerpop git commit: minor tweak. need to work in
another branch.
Posted by ok...@apache.org.
minor tweak. need to work in another branch.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/19f16f1c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/19f16f1c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/19f16f1c
Branch: refs/heads/master
Commit: 19f16f1c6a9954a30fc2031df34724e56b2124fc
Parents: b72ea66
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 25 08:42:51 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 25 08:42:51 2016 -0600
----------------------------------------------------------------------
.../strategy/decoration/HaltedTraverserFactoryStrategy.java | 7 +++++++
1 file changed, 7 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/19f16f1c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
index 44542c7..b046986 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategy.java
@@ -27,6 +27,9 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
+import java.util.Collections;
+import java.util.Set;
+
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
@@ -50,4 +53,8 @@ public final class HaltedTraverserFactoryStrategy extends AbstractTraversalStrat
public static HaltedTraverserFactoryStrategy reference() {
return new HaltedTraverserFactoryStrategy(ReferenceFactory.class);
}
+
+ public Set<Class<? extends DecorationStrategy>> applyPrior() {
+ return Collections.singleton(VertexProgramStrategy.class);
+ }
}
[05/10] incubator-tinkerpop git commit: lots of more documentation on
TraversalVertexProgram and I really combed through the code and was able to
find numerous minor optimizations here and there.
Posted by ok...@apache.org.
lots of more documentation on TraversalVertexProgram and I really combed through the code and was able to find numerous minor optimizations here and there.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/e6f2caa8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/e6f2caa8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/e6f2caa8
Branch: refs/heads/master
Commit: e6f2caa89adc8e6630dadc613ca9b6b92416c223
Parents: 7255844
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 25 14:54:15 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 25 14:54:15 2016 -0600
----------------------------------------------------------------------
.../computer/traversal/MasterExecutor.java | 47 +++++++++-----------
.../traversal/TraversalVertexProgram.java | 2 +
.../computer/traversal/WorkerExecutor.java | 32 ++++++-------
.../gremlin/hadoop/structure/HadoopGraph.java | 2 +-
.../HaltedTraverserFactoryStrategyTest.java | 7 ++-
5 files changed, 43 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e6f2caa8/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 b994f1e..1c1e9d2 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
@@ -43,7 +43,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.HashSet;
import java.util.Iterator;
@@ -70,20 +69,17 @@ final class MasterExecutor {
return traverser;
}
- // handle traversers and data that were sent from the workers to the master traversal via memory
- protected static void processMemory(final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final TraverserSet<Object> traverserSet, final Set<String> completedBarriers) {
+ protected static void processMemory(final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final TraverserSet<Object> toProcessTraversers, final Set<String> completedBarriers) {
+ // handle traversers and data that were sent from the workers to the master traversal via memory
if (memory.exists(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
for (final String key : memory.<Set<String>>get(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
final Step<Object, Object> step = traversalMatrix.getStepById(key);
- if (null == step) continue; // why? how can this happen?
assert step instanceof Barrier;
completedBarriers.add(step.getId());
if (!(step instanceof LocalBarrier)) { // local barriers don't do any processing on the master traversal (they just lock on the workers)
final Barrier<Object> barrier = (Barrier<Object>) step;
barrier.addBarrier(memory.get(key));
- while (step.hasNext()) {
- traverserSet.add(step.next());
- }
+ step.forEachRemaining(toProcessTraversers::add);
// if it was a reducing barrier step, reset the barrier to its seed value
if (step instanceof ReducingBarrierStep)
memory.set(step.getId(), ((ReducingBarrierStep) step).getSeedSupplier().get());
@@ -100,34 +96,33 @@ final class MasterExecutor {
final TraverserSet<Object> haltedTraversers,
final Class haltedTraverserFactory) {
-
while (!toProcessTraversers.isEmpty()) {
final TraverserSet<Object> localActiveTraversers = new TraverserSet<>();
Step<Object, Object> previousStep = EmptyStep.instance();
Step<Object, Object> currentStep = EmptyStep.instance();
- final Iterator<Traverser.Admin<Object>> traversers = IteratorUtils.removeOnNext(toProcessTraversers.iterator());
+ // these are traversers that are at the master traversal and will either halt here or be distributed back to the workers as needed
+ final Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
while (traversers.hasNext()) {
final Traverser.Admin<Object> traverser = traversers.next();
- traverser.set(DetachedFactory.detach(traverser.get(), true));
+ traversers.remove();
+ traverser.set(DetachedFactory.detach(traverser.get(), true)); // why?
traverser.setSideEffects(traversal.get().getSideEffects());
- if (traverser.isHalted()) {
+ if (traverser.isHalted())
haltedTraversers.add(MasterExecutor.detach(traverser, haltedTraverserFactory));
- } else if (isRemoteTraverser(traverser, traversalMatrix)) { // this is so that patterns like order().name work as expected.
+ else if (isRemoteTraverser(traverser, traversalMatrix)) // this is so that patterns like order().name work as expected. try and stay local as long as possible
remoteActiveTraversers.add(traverser.detach());
- } else {
+ else {
currentStep = traversalMatrix.getStepById(traverser.getStepId());
if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep)) {
while (previousStep.hasNext()) {
final Traverser.Admin<Object> result = previousStep.next();
- if (result.isHalted()) {
+ if (result.isHalted())
haltedTraversers.add(MasterExecutor.detach(result, haltedTraverserFactory));
- } else {
- if (isRemoteTraverser(result, traversalMatrix)) {
- remoteActiveTraversers.add(result.detach());
- } else
- localActiveTraversers.add(result);
- }
+ else if (isRemoteTraverser(result, traversalMatrix))
+ remoteActiveTraversers.add(result.detach());
+ else
+ localActiveTraversers.add(result);
}
}
currentStep.addStart(traverser);
@@ -137,14 +132,12 @@ final class MasterExecutor {
if (!(currentStep instanceof EmptyStep)) {
while (currentStep.hasNext()) {
final Traverser.Admin<Object> traverser = currentStep.next();
- if (traverser.isHalted()) {
+ if (traverser.isHalted())
haltedTraversers.add(MasterExecutor.detach(traverser, haltedTraverserFactory));
- } else {
- if (isRemoteTraverser(traverser, traversalMatrix)) {
- remoteActiveTraversers.add(traverser.detach());
- } else
- localActiveTraversers.add(traverser);
- }
+ else if (isRemoteTraverser(traverser, traversalMatrix))
+ remoteActiveTraversers.add(traverser.detach());
+ else
+ localActiveTraversers.add(traverser);
}
}
assert toProcessTraversers.isEmpty();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e6f2caa8/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 d4daaac..4479306 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
@@ -297,7 +297,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
if (voteToHalt) {
// local traverser sets to process
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
+ // traversers that need to be sent back to the workers (no longer can be processed locally by the master traversal)
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
+ // halted traversers that have completed their journey
final TraverserSet<Object> haltedTraversers = memory.get(HALTED_TRAVERSERS);
// get all barrier traversers
final Set<String> completedBarriers = new HashSet<>();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e6f2caa8/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
index f833b6f..5798af0 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/WorkerExecutor.java
@@ -21,7 +21,6 @@ package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
-import org.apache.tinkerpop.gremlin.process.computer.util.SingleMessenger;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
@@ -36,7 +35,6 @@ import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceElement;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
@@ -65,19 +63,18 @@ final class WorkerExecutor {
final TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
final TraverserSet<Object> activeTraversers = new TraverserSet<>();
final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
- final boolean isTesting = Boolean.valueOf(System.getProperty("is.testing", "false"));
////////////////////////////////
// GENERATE LOCAL TRAVERSERS //
///////////////////////////////
- // some memory systems are interacted by multiple threads and thus, concurrent modification can happen at iterator.remove()
+ // these are traversers that are going from OLTP (master) to OLAP (workers)
+ // these traversers were broadcasted from the master traversal to the workers for attachment
+ final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
+ // some memory systems are interacted with by multiple threads and thus, concurrent modification can happen at iterator.remove().
// its better to reduce the memory footprint and shorten the active traverser list so synchronization is worth it.
- // most distributed OLAP systems have the memory partitioned and thus, this synchronization does nothing
- synchronized (memory) {
- // these are traversers that are going from OLTP to OLAP
- // these traversers were broadcasted from the master traversal to the workers for attachment
- final TraverserSet<Object> maybeActiveTraversers = memory.get(TraversalVertexProgram.ACTIVE_TRAVERSERS);
+ // most distributed OLAP systems have the memory partitioned and thus, this synchronization does nothing.
+ synchronized (maybeActiveTraversers) {
final Iterator<Traverser.Admin<Object>> iterator = maybeActiveTraversers.iterator();
while (iterator.hasNext()) {
final Traverser.Admin<Object> traverser = iterator.next();
@@ -97,20 +94,19 @@ final class WorkerExecutor {
traverser.setSideEffects(traversalSideEffects);
toProcessTraversers.add(traverser);
});
+ assert previousActiveTraversers.isEmpty();
+ // remove the property to save space
vertex.property(TraversalVertexProgram.ACTIVE_TRAVERSERS).remove();
});
// these are traversers that have been messaged to the vertex from another vertex
final Iterator<TraverserSet<Object>> messages = messenger.receiveMessages();
while (messages.hasNext()) {
IteratorUtils.removeOnNext(messages.next().iterator()).forEachRemaining(traverser -> {
- // this is internal testing to ensure that messaged elements are always ReferenceXXX and not DetachedXXX (related to HaltedTraverserFactoryStrategy)
- if (isTesting && !(messenger instanceof SingleMessenger) && traverser.get() instanceof Element)
- assert traverser.get() instanceof ReferenceElement;
if (traverser.isHalted()) {
if (returnHaltedTraversers)
memory.add(TraversalVertexProgram.HALTED_TRAVERSERS, new TraverserSet<>(MasterExecutor.detach(traverser, haltedTraverserFactory)));
else
- haltedTraversers.add(traverser);
+ haltedTraversers.add(traverser.detach());
} else {
// traverser is not halted and thus, should be processed locally
traverser.attach(Attachable.Method.get(vertex));
@@ -132,6 +128,7 @@ final class WorkerExecutor {
final Traverser.Admin<Object> traverser = traversers.next();
traversers.remove();
final Step<Object, Object> currentStep = traversalMatrix.getStepById(traverser.getStepId());
+ // try and fill up the current step as much as possible with traversers to get a bulking optimization
if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep))
WorkerExecutor.drainStep(vertex, previousStep, activeTraversers, haltedTraversers, memory, returnHaltedTraversers, haltedTraverserFactory);
currentStep.addStart(traverser);
@@ -150,12 +147,11 @@ final class WorkerExecutor {
if (traverser.get() instanceof Element || traverser.get() instanceof Property) { // GRAPH OBJECT
// if the element is remote, then message, else store it locally for re-processing
final Vertex hostingVertex = WorkerExecutor.getHostingVertex(traverser.get());
- if (!vertex.equals(hostingVertex)) { // necessary for path access
- voteToHalt.set(false);
+ if (!vertex.equals(hostingVertex)) { // if its host is not the current vertex, then send the traverser to the hosting vertex
+ voteToHalt.set(false); // if message is passed, then don't vote to halt
messenger.sendMessage(MessageScope.Global.of(hostingVertex), new TraverserSet<>(traverser.detach()));
} else {
- if (traverser.get() instanceof Attachable) // necessary for path access to local object
- traverser.attach(Attachable.Method.get(vertex));
+ traverser.attach(Attachable.Method.get(vertex));
toProcessTraversers.add(traverser);
}
} else // STANDARD OBJECT
@@ -195,7 +191,7 @@ final class WorkerExecutor {
else
haltedTraversers.add(traverser.detach());
} else
- localBarrierTraversers.add(traverser);
+ localBarrierTraversers.add(traverser.detach());
});
}
memory.add(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>(Collections.singleton(step.getId())));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e6f2caa8/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
index d643cd4..d0f50d0 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
@@ -109,7 +109,7 @@ import java.util.stream.Stream;
test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.GroovyMatchTest$Traversals",
method = "g_V_matchXa_0sungBy_b__a_0sungBy_c__b_writtenBy_d__c_writtenBy_e__d_hasXname_George_HarisonX__e_hasXname_Bob_MarleyXX",
reason = "Hadoop-Gremlin is OLAP-oriented and for OLTP operations, linear-scan joins are required. This particular tests takes many minutes to execute.",
- computers = {"ALL"})
+ computers = {"org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer"}) // this is a nasty long test, just do it once in Java MatchTest
@Graph.OptOut(
test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.GroovyMatchTest$Traversals",
method = "g_V_matchXa_0sungBy_b__a_0writtenBy_c__b_writtenBy_d__c_sungBy_d__d_hasXname_GarciaXX",
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e6f2caa8/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
index af0b3b7..43bc94e 100644
--- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
+++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/traversal/strategy/decoration/HaltedTraverserFactoryStrategyTest.java
@@ -23,10 +23,12 @@ import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decorati
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedPath;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedProperty;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferencePath;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
@@ -64,6 +66,7 @@ public class HaltedTraverserFactoryStrategyTest {
g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(DetachedProperty.class, property.getClass()));
g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
g.V().out().out().forEachRemaining(vertex -> assertEquals(DetachedVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(DetachedPath.class, path.getClass()));
}
@Test
@@ -77,7 +80,8 @@ public class HaltedTraverserFactoryStrategyTest {
g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
- //
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
+ // the default should be reference elements
g = graph.traversal().withComputer();
g.V().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
g.V().out().properties("name").forEachRemaining(vertexProperty -> assertEquals(ReferenceVertexProperty.class, vertexProperty.getClass()));
@@ -86,6 +90,7 @@ public class HaltedTraverserFactoryStrategyTest {
g.V().out().outE().properties("weight").forEachRemaining(property -> assertEquals(ReferenceProperty.class, property.getClass()));
g.V().out().outE().values("weight").forEachRemaining(value -> assertEquals(Double.class, value.getClass()));
g.V().out().out().forEachRemaining(vertex -> assertEquals(ReferenceVertex.class, vertex.getClass()));
+ g.V().out().out().path().forEachRemaining(path -> assertEquals(ReferencePath.class, path.getClass()));
}
}