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/03/01 18:32:46 UTC
incubator-tinkerpop git commit:
SparkGraphComputerProcessXXXIntegrateTest is now
SparkGraphComputerProcessXXXTest -- per dev@ email VOTE. Fixed a traverser
timing issue in TraversalVertexProgram (minor) and added comments throughout
so others know why ea
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 19f24d9be -> 980c950c6
SparkGraphComputerProcessXXXIntegrateTest is now SparkGraphComputerProcessXXXTest -- per dev@ email VOTE. Fixed a traverser timing issue in TraversalVertexProgram (minor) and added comments throughout so others know why each line does what it does. CTR.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/980c950c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/980c950c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/980c950c
Branch: refs/heads/master
Commit: 980c950c649a9a2094f50fa04d8429cd00ba1699
Parents: 19f24d9
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Mar 1 10:32:40 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Mar 1 10:32:40 2016 -0700
----------------------------------------------------------------------
.../traversal/TraversalVertexProgram.java | 29 ++++++++++-------
.../SparkGraphComputerProcessIntegrateTest.java | 32 -------------------
.../computer/SparkGraphComputerProcessTest.java | 32 +++++++++++++++++++
...GraphComputerGroovyProcessIntegrateTest.java | 33 --------------------
.../SparkGraphComputerGroovyProcessTest.java | 33 ++++++++++++++++++++
.../structure/TinkerGraphPlayTest.java | 3 +-
6 files changed, 85 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/980c950c/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 0636bbd..7da6984 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
@@ -222,45 +222,51 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
@Override
public boolean terminate(final Memory memory) {
- final Set<String> mutatedMemoryKeys = memory.get(MUTATED_MEMORY_KEYS); // TODO: if not touched we still have to initialize the seeds?
final boolean voteToHalt = memory.<Boolean>get(VOTE_TO_HALT);
memory.set(VOTE_TO_HALT, true);
- memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
memory.set(ACTIVE_TRAVERSERS, new TraverserSet<>());
-
// put all side-effect memory into traversal side-effects
if (voteToHalt) {
+ final Set<String> mutatedMemoryKeys = memory.get(MUTATED_MEMORY_KEYS);
+ memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
// put all memory side-effects into the traversal side-effects
memory.keys().stream().
filter(key -> !PROGRAM_KEYS.contains(key)).
filter(key -> !StepPosition.isStepId(key)).
forEach(key -> this.traversal.get().getSideEffects().set(key, memory.get(key)));
+ // local traverser sets to process
TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
TraverserSet<Object> localActiveTraversers = new TraverserSet<>();
final TraverserSet<Object> remoteActiveTraversers = new TraverserSet<>();
final TraverserSet<Object> haltedTraversers = memory.get(HALTED_TRAVERSERS);
+ // get all barrier traversers
final Set<String> completedBarriers = new HashSet<>();
this.processMemory(memory, mutatedMemoryKeys, toProcessTraversers, completedBarriers);
+ // process all results from barriers locally and when elements are touched, put them in remoteActiveTraversers
while (!toProcessTraversers.isEmpty()) {
this.processTraversers(toProcessTraversers, localActiveTraversers, remoteActiveTraversers, haltedTraversers);
toProcessTraversers = localActiveTraversers;
localActiveTraversers = new TraverserSet<>();
}
- // pull all traversal side-effects into memory
+ // put all traversal side-effects into memory
this.traversal.get().getSideEffects().keys().forEach(key -> {
if (memory.exists(key))
memory.set(key, this.traversal.get().getSideEffects().get(key).get());
});
+ // 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()) {
+ // send active traversers back to workers
memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
return false;
} else {
+ // finalize locally with any last traversers dangling in the local traversal
while (this.traversal.get().getEndStep().hasNext()) {
final Traverser.Admin<Object> traverser = (Traverser.Admin) this.traversal.get().getEndStep().next().asAdmin();
traverser.detach();
haltedTraversers.add(traverser);
}
+ // the result of a TraversalVertexProgram are the halted traversers
memory.set(HALTED_TRAVERSERS, haltedTraversers);
return true;
}
@@ -287,10 +293,11 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
private void processTraversers(final TraverserSet<Object> toProcessTraversers, final TraverserSet<Object> localActiveTraversers, final TraverserSet<Object> remoteActiveTraversers, final TraverserSet<Object> haltedTraversers) {
Step<Object, Object> previousStep = EmptyStep.instance();
- final Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
+ Step<Object, Object> currentStep = EmptyStep.instance();
+
+ final Iterator<Traverser.Admin<Object>> traversers = IteratorUtils.removeOnNext(toProcessTraversers.iterator());
while (traversers.hasNext()) {
final Traverser.Admin<Object> traverser = traversers.next();
- traversers.remove();
traverser.set(DetachedFactory.detach(traverser.get(), true));
traverser.setSideEffects(this.traversal.get().getSideEffects());
if (traverser.isHalted()) {
@@ -301,9 +308,9 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
!TraversalHelper.isLocalElement(this.traversalMatrix.getStepById(traverser.getStepId()))) { // this is so that patterns like order().name work as expected.
remoteActiveTraversers.add(traverser);
} else {
- final Step<Object, Object> currentStep = this.traversalMatrix.getStepById(traverser.getStepId());
+ currentStep = this.traversalMatrix.getStepById(traverser.getStepId());
if (!currentStep.getId().equals(previousStep.getId()) && !(previousStep instanceof EmptyStep)) {
- currentStep.forEachRemaining(result -> {
+ previousStep.forEachRemaining(result -> {
if (result.asAdmin().isHalted()) {
result.asAdmin().detach();
haltedTraversers.add(result.asAdmin());
@@ -319,8 +326,8 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
previousStep = currentStep;
}
}
- if (!(previousStep instanceof EmptyStep)) {
- previousStep.forEachRemaining(traverser -> {
+ if (!(currentStep instanceof EmptyStep)) {
+ currentStep.forEachRemaining(traverser -> {
if (traverser.asAdmin().isHalted()) {
traverser.asAdmin().detach();
haltedTraversers.add(traverser.asAdmin());
@@ -332,7 +339,7 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
}
});
}
- assert (toProcessTraversers.isEmpty());
+ assert toProcessTraversers.isEmpty();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/980c950c/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessIntegrateTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessIntegrateTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessIntegrateTest.java
deleted file mode 100644
index f38dcf5..0000000
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessIntegrateTest.java
+++ /dev/null
@@ -1,32 +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.spark.process.computer;
-
-import org.apache.tinkerpop.gremlin.GraphProviderClass;
-import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
-import org.apache.tinkerpop.gremlin.process.ProcessComputerSuite;
-import org.junit.runner.RunWith;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-@RunWith(ProcessComputerSuite.class)
-@GraphProviderClass(provider = SparkHadoopGraphProvider.class, graph = HadoopGraph.class)
-public class SparkGraphComputerProcessIntegrateTest {
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/980c950c/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessTest.java
new file mode 100644
index 0000000..055c9bd
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputerProcessTest.java
@@ -0,0 +1,32 @@
+/*
+ * 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.spark.process.computer;
+
+import org.apache.tinkerpop.gremlin.GraphProviderClass;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.process.ProcessComputerSuite;
+import org.junit.runner.RunWith;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+@RunWith(ProcessComputerSuite.class)
+@GraphProviderClass(provider = SparkHadoopGraphProvider.class, graph = HadoopGraph.class)
+public class SparkGraphComputerProcessTest {
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/980c950c/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessIntegrateTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessIntegrateTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessIntegrateTest.java
deleted file mode 100644
index 6f259dd..0000000
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessIntegrateTest.java
+++ /dev/null
@@ -1,33 +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.spark.process.computer.groovy;
-
-import org.apache.tinkerpop.gremlin.GraphProviderClass;
-import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
-import org.apache.tinkerpop.gremlin.process.GroovyProcessComputerSuite;
-import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
-import org.junit.runner.RunWith;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-@RunWith(GroovyProcessComputerSuite.class)
-@GraphProviderClass(provider = SparkHadoopGraphProvider.class, graph = HadoopGraph.class)
-public class SparkGraphComputerGroovyProcessIntegrateTest {
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/980c950c/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessTest.java
new file mode 100644
index 0000000..65ebd75
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkGraphComputerGroovyProcessTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spark.process.computer.groovy;
+
+import org.apache.tinkerpop.gremlin.GraphProviderClass;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.process.GroovyProcessComputerSuite;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
+import org.junit.runner.RunWith;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+@RunWith(GroovyProcessComputerSuite.class)
+@GraphProviderClass(provider = SparkHadoopGraphProvider.class, graph = HadoopGraph.class)
+public class SparkGraphComputerGroovyProcessTest {
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/980c950c/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraphPlayTest.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraphPlayTest.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraphPlayTest.java
index 0c9f24c..1d682b5 100644
--- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraphPlayTest.java
+++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/structure/TinkerGraphPlayTest.java
@@ -52,6 +52,7 @@ import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.count;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.fold;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.has;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.in;
+import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.inE;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.out;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.outE;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.select;
@@ -72,7 +73,7 @@ public class TinkerGraphPlayTest {
GraphTraversalSource g = graph.traversal().withComputer();//GraphTraversalSource.computer());
//System.out.println(g.V().outE("knows").identity().inV().count().is(P.eq(5)).explain());
//System.out.println(g.V().hasLabel("person").fold().order(Scope.local).by("age").toList());
- final Traversal<?,?> traversal = g.V().both().groupCount("a").out().cap("a").select(Column.keys).unfold().both().groupCount("a").cap("a"); // unfold.select(values) [{v[1]=6, v[2]=2, v[3]=6, v[4]=6, v[5]=2, v[6]=2}]
+ final Traversal<?,?> traversal = g.V(1, 2).union(outE().count(), inE().count(), (Traversal) outE().values("weight").sum());
System.out.println(traversal.asAdmin().clone().toString());
final Traversal<?,?> clone = traversal.asAdmin().clone();