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/02/09 18:09:06 UTC
[1/8] incubator-tinkerpop git commit: Bump to 3.2.0-SNAPSHOT.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/tp31 1286b4f19 -> b07400ccb
Bump to 3.2.0-SNAPSHOT.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/ffbecdb8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/ffbecdb8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/ffbecdb8
Branch: refs/heads/tp31
Commit: ffbecdb870e5d4a840b0445447d96c27c3d84e3a
Parents: b50bb3f
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon Feb 1 08:09:12 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Mon Feb 1 08:09:12 2016 -0500
----------------------------------------------------------------------
giraph-gremlin/pom.xml | 2 +-
gremlin-console/pom.xml | 2 +-
gremlin-core/pom.xml | 2 +-
gremlin-driver/pom.xml | 2 +-
gremlin-groovy-test/pom.xml | 2 +-
gremlin-groovy/pom.xml | 2 +-
gremlin-server/pom.xml | 2 +-
gremlin-shaded/pom.xml | 2 +-
gremlin-test/pom.xml | 2 +-
hadoop-gremlin/pom.xml | 2 +-
neo4j-gremlin/pom.xml | 2 +-
pom.xml | 2 +-
spark-gremlin/pom.xml | 2 +-
tinkergraph-gremlin/pom.xml | 2 +-
14 files changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/giraph-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/giraph-gremlin/pom.xml b/giraph-gremlin/pom.xml
index ba037cf..c3eadc0 100644
--- a/giraph-gremlin/pom.xml
+++ b/giraph-gremlin/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>giraph-gremlin</artifactId>
<name>Apache TinkerPop :: Giraph Gremlin</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-console/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-console/pom.xml b/gremlin-console/pom.xml
index 419fd7e..f5afe32 100644
--- a/gremlin-console/pom.xml
+++ b/gremlin-console/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<artifactId>tinkerpop</artifactId>
<groupId>org.apache.tinkerpop</groupId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-console</artifactId>
<name>Apache TinkerPop :: Gremlin Console</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-core/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-core/pom.xml b/gremlin-core/pom.xml
index 668830a..fabf0cb 100644
--- a/gremlin-core/pom.xml
+++ b/gremlin-core/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-core</artifactId>
<name>Apache TinkerPop :: Gremlin Core</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-driver/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-driver/pom.xml b/gremlin-driver/pom.xml
index 30248b8..910068a 100644
--- a/gremlin-driver/pom.xml
+++ b/gremlin-driver/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-driver</artifactId>
<name>Apache TinkerPop :: Gremlin Driver</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-groovy-test/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/pom.xml b/gremlin-groovy-test/pom.xml
index b731d40..23cfa06 100644
--- a/gremlin-groovy-test/pom.xml
+++ b/gremlin-groovy-test/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-groovy-test</artifactId>
<name>Apache TinkerPop :: Gremlin Groovy Test</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-groovy/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-groovy/pom.xml b/gremlin-groovy/pom.xml
index 0a682e5..aa459ab 100644
--- a/gremlin-groovy/pom.xml
+++ b/gremlin-groovy/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-groovy</artifactId>
<name>Apache TinkerPop :: Gremlin Groovy</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-server/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-server/pom.xml b/gremlin-server/pom.xml
index 583eb23..1e67584 100644
--- a/gremlin-server/pom.xml
+++ b/gremlin-server/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-server</artifactId>
<name>Apache TinkerPop :: Gremlin Server</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-shaded/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-shaded/pom.xml b/gremlin-shaded/pom.xml
index 5417b73..614a33c 100644
--- a/gremlin-shaded/pom.xml
+++ b/gremlin-shaded/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-shaded</artifactId>
<name>Apache TinkerPop :: Gremlin Shaded</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/gremlin-test/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-test/pom.xml b/gremlin-test/pom.xml
index e9aacad..590e5dd 100644
--- a/gremlin-test/pom.xml
+++ b/gremlin-test/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>gremlin-test</artifactId>
<name>Apache TinkerPop :: Gremlin Test</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/hadoop-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/pom.xml b/hadoop-gremlin/pom.xml
index 3542275..c0b733f 100644
--- a/hadoop-gremlin/pom.xml
+++ b/hadoop-gremlin/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>hadoop-gremlin</artifactId>
<name>Apache TinkerPop :: Hadoop Gremlin</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/neo4j-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/neo4j-gremlin/pom.xml b/neo4j-gremlin/pom.xml
index 6bfd5ee..8732566 100644
--- a/neo4j-gremlin/pom.xml
+++ b/neo4j-gremlin/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>neo4j-gremlin</artifactId>
<name>Apache TinkerPop :: Neo4j Gremlin</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 4d3328c..91f515a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -25,7 +25,7 @@ limitations under the License.
</parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache TinkerPop</name>
<description>A Graph Computing Framework</description>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/spark-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/spark-gremlin/pom.xml b/spark-gremlin/pom.xml
index 76b616f..e6731c6 100644
--- a/spark-gremlin/pom.xml
+++ b/spark-gremlin/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>spark-gremlin</artifactId>
<name>Apache TinkerPop :: Spark Gremlin</name>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ffbecdb8/tinkergraph-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/pom.xml b/tinkergraph-gremlin/pom.xml
index 2c6fbf6..bde9783 100644
--- a/tinkergraph-gremlin/pom.xml
+++ b/tinkergraph-gremlin/pom.xml
@@ -21,7 +21,7 @@ limitations under the License.
<parent>
<groupId>org.apache.tinkerpop</groupId>
<artifactId>tinkerpop</artifactId>
- <version>3.1.1-SNAPSHOT</version>
+ <version>3.2.0-SNAPSHOT</version>
</parent>
<artifactId>tinkergraph-gremlin</artifactId>
<name>Apache TinkerPop :: TinkerGraph Gremlin</name>
[4/8] incubator-tinkerpop git commit: Made significant memory
improvements to TraverserExecutor. Realized some massive heaps on some jobs
on Friendster using SparkGraphComputer and tracked it down to how I'm dealing
with traversers in TraverserVertexProg
Posted by ok...@apache.org.
Made significant memory improvements to TraverserExecutor. Realized some massive heaps on some jobs on Friendster using SparkGraphComputer and tracked it down to how I'm dealing with traversers in TraverserVertexProgram. I was not 'draining' sets of traversers and thus, was using an excessive amount of memory. This really shows itself when touching edges where you can easily generate million of traversers and to have multiple copies of that data is bad. To make draining work, I had to update all the Iterators to support .remove() which simply call .remove() of the child iterator. Found a simple optimization for CountGlobalStep that will make OLAP counting much faster.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/b54ddf28
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/b54ddf28
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/b54ddf28
Branch: refs/heads/tp31
Commit: b54ddf2830483705c4c4a865b9f2586ed457223a
Parents: 12b9b21
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri Feb 5 14:56:10 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri Feb 5 14:56:10 2016 -0700
----------------------------------------------------------------------
.../computer/traversal/TraverserExecutor.java | 101 +++++++++++--------
.../traversal/step/map/CountGlobalStep.java | 10 ++
.../gremlin/util/iterator/DoubleIterator.java | 12 ++-
.../gremlin/util/iterator/IteratorUtils.java | 25 +++++
.../gremlin/util/iterator/MultiIterator.java | 7 +-
.../gremlin/util/iterator/SingleIterator.java | 9 +-
6 files changed, 118 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b54ddf28/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
index 038e309..2df682c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
@@ -23,6 +23,7 @@ import org.apache.tinkerpop.gremlin.process.computer.Messenger;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -31,6 +32,7 @@ import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.Attachable;
+import java.util.Iterator;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -42,58 +44,75 @@ public final class TraverserExecutor {
final TraverserSet<Object> haltedTraversers = vertex.value(TraversalVertexProgram.HALTED_TRAVERSERS);
final AtomicBoolean voteToHalt = new AtomicBoolean(true);
-
final TraverserSet<Object> aliveTraversers = new TraverserSet<>();
- // gather incoming traversers into a traverser set and gain the 'weighted-set' optimization
+ final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
+
+ final Iterator<TraverserSet<?>> messages = messenger.receiveMessages();
final TraversalSideEffects traversalSideEffects = traversalMatrix.getTraversal().getSideEffects();
- messenger.receiveMessages().forEachRemaining(traverserSet -> {
- traverserSet.forEach(traverser -> {
+ while (messages.hasNext()) {
+ final Iterator<Traverser.Admin<Object>> traversers = (Iterator) messages.next().iterator();
+ messages.remove();
+ while (traversers.hasNext()) {
+ final Traverser.Admin<Object> traverser = traversers.next();
+ traversers.remove();
traverser.setSideEffects(traversalSideEffects);
traverser.attach(Attachable.Method.get(vertex));
- aliveTraversers.add((Traverser.Admin) traverser);
- });
- });
-
+ toProcessTraversers.add((Traverser.Admin) traverser);
+ }
+ }
// while there are still local traversers, process them until they leave the vertex or halt (i.e. isHalted()).
- final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
- while (!aliveTraversers.isEmpty()) {
+ Step<?, ?> previousStep = EmptyStep.instance();
+ while (!toProcessTraversers.isEmpty()) {
+ // process local traversers and if alive, repeat, else halt.
+ Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
+ while (traversers.hasNext()) {
+ final Traverser.Admin<Object> traverser = traversers.next();
+ traversers.remove();
+ final Step<?, ?> currentStep = traversalMatrix.getStepById(traverser.getStepId());
+ if (!currentStep.getId().equals(previousStep.getId()))
+ TraverserExecutor.drainStep(previousStep, aliveTraversers, haltedTraversers);
+ currentStep.addStart((Traverser.Admin) traverser);
+ previousStep = currentStep;
+ }
+ TraverserExecutor.drainStep(previousStep, aliveTraversers, haltedTraversers);
+ assert toProcessTraversers.isEmpty();
// process all the local objects and send messages or store locally again
- aliveTraversers.forEach(traverser -> {
- 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 = TraverserExecutor.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));
- } else {
- if (traverser.get() instanceof Attachable) // necessary for path access to local object
- traverser.attach(Attachable.Method.get(vertex));
+ if (!aliveTraversers.isEmpty()) {
+ traversers = aliveTraversers.iterator();
+ while (traversers.hasNext()) {
+ final Traverser.Admin<Object> traverser = traversers.next();
+ traversers.remove();
+ 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 = TraverserExecutor.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));
+ } else {
+ if (traverser.get() instanceof Attachable) // necessary for path access to local object
+ traverser.attach(Attachable.Method.get(vertex));
+ toProcessTraversers.add(traverser);
+ }
+ } else // STANDARD OBJECT
toProcessTraversers.add(traverser);
- }
- } else // STANDARD OBJECT
- toProcessTraversers.add(traverser);
- });
-
- // process local traversers and if alive, repeat, else halt.
- aliveTraversers.clear();
- toProcessTraversers.forEach(start -> {
- final Step<?, ?> step = traversalMatrix.getStepById(start.getStepId());
- step.addStart((Traverser.Admin) start);
- step.forEachRemaining(end -> {
- if (end.asAdmin().isHalted()) {
- end.asAdmin().detach();
- haltedTraversers.add((Traverser.Admin) end);
- } else
- aliveTraversers.add((Traverser.Admin) end);
- });
- });
-
- toProcessTraversers.clear();
+ }
+ assert aliveTraversers.isEmpty();
+ }
}
return voteToHalt.get();
}
+ private static void drainStep(final Step<?, ?> step, final TraverserSet<?> aliveTraversers, final TraverserSet<?> haltedTraversers) {
+ step.forEachRemaining(traverser -> {
+ if (traverser.asAdmin().isHalted()) {
+ traverser.asAdmin().detach();
+ haltedTraversers.add((Traverser.Admin) traverser);
+ } else
+ aliveTraversers.add((Traverser.Admin) traverser);
+ });
+ }
+
private static Vertex getHostingVertex(final Object object) {
Object obj = object;
while (true) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b54ddf28/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
index 409c445..d090832 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
@@ -61,6 +61,16 @@ public final class CountGlobalStep<S> extends ReducingBarrierStep<S, Long> imple
return CountGlobalMapReduce.instance();
}
+ @Override
+ public Traverser<Long> processNextStart() {
+ if (this.byPass) {
+ final Traverser.Admin<S> traverser = this.starts.next();
+ return traverser.asAdmin().split(1l, this); // if bypassing, just key all the traversers to 1 long (the count is going to be the bulk of course)
+ } else {
+ return super.processNextStart();
+ }
+ }
+
///////////
private static class CountBiFunction<S> implements BiFunction<Long, Traverser<S>, Long>, Serializable {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b54ddf28/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/DoubleIterator.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/DoubleIterator.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/DoubleIterator.java
index e07bcef..2d7982c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/DoubleIterator.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/DoubleIterator.java
@@ -28,8 +28,8 @@ import java.util.Iterator;
*/
final class DoubleIterator<T> implements Iterator<T>, Serializable {
- private final T a;
- private final T b;
+ private T a;
+ private T b;
private char current = 'a';
protected DoubleIterator(final T a, final T b) {
@@ -43,6 +43,14 @@ final class DoubleIterator<T> implements Iterator<T>, Serializable {
}
@Override
+ public void remove() {
+ if (this.current == 'b')
+ this.a = null;
+ else if (this.current == 'x')
+ this.b = null;
+ }
+
+ @Override
public T next() {
if (this.current == 'x')
throw FastNoSuchElementException.instance();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b54ddf28/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
index dc38a07..808317f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
@@ -105,6 +105,11 @@ public final class IteratorUtils {
}
@Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
public S next() {
if (this.count++ >= limit)
throw FastNoSuchElementException.instance();
@@ -205,6 +210,11 @@ public final class IteratorUtils {
}
@Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
public S next() {
final S s = iterator.next();
consumer.accept(s);
@@ -228,6 +238,11 @@ public final class IteratorUtils {
}
@Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
public E next() {
return function.apply(iterator.next());
}
@@ -257,6 +272,11 @@ public final class IteratorUtils {
}
@Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
public S next() {
try {
if (null != this.nextResult) {
@@ -312,6 +332,11 @@ public final class IteratorUtils {
}
@Override
+ public void remove() {
+ iterator.remove();
+ }
+
+ @Override
public E next() {
if (this.hasNext())
return this.currentIterator.next();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b54ddf28/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
index 02c15f4..e272a08 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
@@ -42,7 +42,7 @@ public final class MultiIterator<T> implements Iterator<T>, Serializable {
if (this.current >= this.iterators.size())
return false;
- Iterator<T> currentIterator = iterators.get(this.current);
+ Iterator<T> currentIterator = this.iterators.get(this.current);
while (true) {
if (currentIterator.hasNext()) {
@@ -58,6 +58,11 @@ public final class MultiIterator<T> implements Iterator<T>, Serializable {
}
@Override
+ public void remove() {
+ // this.iterators.get(this.current).remove();
+ }
+
+ @Override
public T next() {
if (this.iterators.isEmpty()) throw FastNoSuchElementException.instance();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b54ddf28/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/SingleIterator.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/SingleIterator.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/SingleIterator.java
index b680819..bb2da72 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/SingleIterator.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/SingleIterator.java
@@ -26,9 +26,9 @@ import java.util.Iterator;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-final class SingleIterator<T> implements Iterator<T>,Serializable {
+final class SingleIterator<T> implements Iterator<T>, Serializable {
- private final T t;
+ private T t;
private boolean alive = true;
protected SingleIterator(final T t) {
@@ -41,6 +41,11 @@ final class SingleIterator<T> implements Iterator<T>,Serializable {
}
@Override
+ public void remove() {
+ this.t = null;
+ }
+
+ @Override
public T next() {
if (!this.alive)
throw FastNoSuchElementException.instance();
[6/8] incubator-tinkerpop git commit: Greatly optimized all
ReducingBarrierSteps in OLAP execution. I realized that there are actually
two 'combine' stages. Given that the HALTED_TRAVERSERS are all group'd at a
Vertex prior to MAP, you can COMBINE in Map
Posted by ok...@apache.org.
Greatly optimized all ReducingBarrierSteps in OLAP execution. I realized that there are actually two 'combine' stages. Given that the HALTED_TRAVERSERS are all group'd at a Vertex prior to MAP, you can COMBINE in Map. Thus, you map/combine, combine, and then reduce. For steps like outE().count(), this means that we don't Map traversers for each edge --- instead we create a single count locally before pushing off to combine/reduce. This will make GC and memory usage minimal for these operations.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/388d1f25
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/388d1f25
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/388d1f25
Branch: refs/heads/tp31
Commit: 388d1f25fec61aebc175508452d74b6375bacabe
Parents: 047ee20
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Sun Feb 7 14:19:28 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Sun Feb 7 14:19:28 2016 -0700
----------------------------------------------------------------------
.../traversal/step/map/CountGlobalStep.java | 31 ++++++------
.../traversal/step/map/GroupCountStep.java | 16 ++-----
.../process/traversal/step/map/GroupStep.java | 10 +++-
.../traversal/step/map/GroupStepV3d0.java | 10 +++-
.../traversal/step/map/MaxGlobalStep.java | 23 +++++----
.../traversal/step/map/MinGlobalStep.java | 22 +++++----
.../traversal/step/map/SumGlobalStep.java | 23 +++++----
.../process/traversal/step/map/TreeStep.java | 10 ----
.../step/util/ReducingBarrierStep.java | 50 ++++++++++----------
9 files changed, 106 insertions(+), 89 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
index d090832..096580d 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/CountGlobalStep.java
@@ -27,11 +27,12 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.ConstantSupplier;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.io.Serializable;
+import java.util.Collections;
import java.util.EnumSet;
import java.util.Iterator;
import java.util.Set;
@@ -61,16 +62,6 @@ public final class CountGlobalStep<S> extends ReducingBarrierStep<S, Long> imple
return CountGlobalMapReduce.instance();
}
- @Override
- public Traverser<Long> processNextStart() {
- if (this.byPass) {
- final Traverser.Admin<S> traverser = this.starts.next();
- return traverser.asAdmin().split(1l, this); // if bypassing, just key all the traversers to 1 long (the count is going to be the bulk of course)
- } else {
- return super.processNextStart();
- }
- }
-
///////////
private static class CountBiFunction<S> implements BiFunction<Long, Traverser<S>, Long>, Serializable {
@@ -108,7 +99,11 @@ public final class CountGlobalStep<S> extends ReducingBarrierStep<S, Long> imple
@Override
public void map(final Vertex vertex, final MapEmitter<NullObject, Long> emitter) {
- vertex.<TraverserSet<?>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet -> traverserSet.forEach(traverser -> emitter.emit(traverser.bulk())));
+ final Iterator<Long> values = IteratorUtils.map(vertex.<Set<Traverser.Admin<Long>>>property(TraversalVertexProgram.HALTED_TRAVERSERS).orElse(Collections.emptySet()).iterator(),
+ traverser -> traverser.get() * traverser.bulk());
+ long count = getCount(values);
+ if (count > 0)
+ emitter.emit(count);
}
@Override
@@ -118,11 +113,17 @@ public final class CountGlobalStep<S> extends ReducingBarrierStep<S, Long> imple
@Override
public void reduce(final NullObject key, final Iterator<Long> values, final ReduceEmitter<NullObject, Long> emitter) {
+ long count = getCount(values);
+ if (count > 0)
+ emitter.emit(count);
+ }
+
+ private Long getCount(final Iterator<Long> longs) {
long count = 0l;
- while (values.hasNext()) {
- count = count + values.next();
+ while (longs.hasNext()) {
+ count = count + longs.next();
}
- emitter.emit(count);
+ return count;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupCountStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupCountStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupCountStep.java
index 7f3904e..95b954e 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupCountStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupCountStep.java
@@ -97,16 +97,6 @@ public final class GroupCountStep<S, E> extends ReducingBarrierStep<S, Map<E, Lo
}
@Override
- public Traverser<Map<E, Long>> processNextStart() {
- if (this.byPass) {
- final Traverser.Admin<S> traverser = this.starts.next();
- return traverser.asAdmin().split(TraversalUtil.applyNullable(traverser, (Traversal.Admin<S, Map<E, Long>>) this.groupTraversal), this);
- } else {
- return super.processNextStart();
- }
- }
-
- @Override
public String toString() {
return StringFactory.stepString(this, this.groupTraversal);
}
@@ -146,7 +136,11 @@ public final class GroupCountStep<S, E> extends ReducingBarrierStep<S, Map<E, Lo
@Override
public void map(final Vertex vertex, final MapEmitter<E, Long> emitter) {
- vertex.<TraverserSet<E>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet -> traverserSet.forEach(traverser -> emitter.emit(traverser.get(), traverser.bulk())));
+ final Map<E, Long> groupCount = new HashMap<>();
+ vertex.<TraverserSet<Map<E, Long>>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet ->
+ traverserSet.forEach(traverser ->
+ traverser.get().forEach((k, v) -> MapHelper.incr(groupCount, k, (v * traverser.bulk())))));
+ groupCount.forEach(emitter::emit);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
index 32a16a7..069d57e 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
@@ -24,9 +24,11 @@ import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
+import org.apache.tinkerpop.gremlin.process.traversal.step.EngineDependent;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
@@ -55,7 +57,7 @@ import java.util.function.Supplier;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>> implements MapReducer, TraversalParent {
+public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>> implements MapReducer, EngineDependent, TraversalParent {
private char state = 'k';
@@ -63,6 +65,7 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
private Traversal.Admin<S, ?> valueTraversal = this.integrateChild(__.identity().asAdmin()); // used in OLAP
private Traversal.Admin<?, V> reduceTraversal = this.integrateChild(__.fold().asAdmin()); // used in OLAP
private Traversal.Admin<S, V> valueReduceTraversal = this.integrateChild(__.fold().asAdmin()); // used in OLTP
+ private boolean byPass = false;
public GroupStep(final Traversal.Admin traversal) {
super(traversal);
@@ -71,6 +74,11 @@ public final class GroupStep<S, K, V> extends ReducingBarrierStep<S, Map<K, V>>
}
@Override
+ public void onEngine(final TraversalEngine traversalEngine) {
+ this.byPass = traversalEngine.isComputer();
+ }
+
+ @Override
public List<Traversal.Admin<?, ?>> getLocalChildren() {
final List<Traversal.Admin<?, ?>> children = new ArrayList<>(4);
if (null != this.keyTraversal)
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStepV3d0.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStepV3d0.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStepV3d0.java
index 04acf62..a988d4e 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStepV3d0.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStepV3d0.java
@@ -25,7 +25,9 @@ import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.EngineDependent;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
@@ -55,13 +57,14 @@ import java.util.function.Supplier;
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
@Deprecated
-public final class GroupStepV3d0<S, K, V, R> extends ReducingBarrierStep<S, Map<K, R>> implements MapReducer, TraversalParent {
+public final class GroupStepV3d0<S, K, V, R> extends ReducingBarrierStep<S, Map<K, R>> implements MapReducer, EngineDependent, TraversalParent {
private char state = 'k';
private Traversal.Admin<S, K> keyTraversal = null;
private Traversal.Admin<S, V> valueTraversal = null;
private Traversal.Admin<Collection<V>, R> reduceTraversal = null;
+ private boolean byPass = false;
public GroupStepV3d0(final Traversal.Admin traversal) {
super(traversal);
@@ -70,6 +73,11 @@ public final class GroupStepV3d0<S, K, V, R> extends ReducingBarrierStep<S, Map<
}
@Override
+ public void onEngine(final TraversalEngine traversalEngine) {
+ this.byPass = traversalEngine.isComputer();
+ }
+
+ @Override
public <A, B> List<Traversal.Admin<A, B>> getLocalChildren() {
final List<Traversal.Admin<A, B>> children = new ArrayList<>(3);
if (null != this.keyTraversal)
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MaxGlobalStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MaxGlobalStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MaxGlobalStep.java
index d876d2d..c552bab 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MaxGlobalStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MaxGlobalStep.java
@@ -27,9 +27,9 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.ConstantSupplier;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.io.Serializable;
import java.util.Collections;
@@ -98,7 +98,9 @@ public final class MaxGlobalStep<S extends Number> extends ReducingBarrierStep<S
@Override
public void map(final Vertex vertex, final MapEmitter<NullObject, Number> emitter) {
- vertex.<TraverserSet<Number>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet -> traverserSet.forEach(traverser -> emitter.emit(traverser.get())));
+ final Iterator<Number> values = IteratorUtils.map(vertex.<Set<Traverser.Admin<Number>>>property(TraversalVertexProgram.HALTED_TRAVERSERS).orElse(Collections.emptySet()).iterator(), Traverser.Admin::get);
+ if (values.hasNext())
+ emitter.emit(getMax(values));
}
@Override
@@ -108,14 +110,17 @@ public final class MaxGlobalStep<S extends Number> extends ReducingBarrierStep<S
@Override
public void reduce(final NullObject key, final Iterator<Number> values, final ReduceEmitter<NullObject, Number> emitter) {
- if (values.hasNext()) {
- Number max = null;
- while (values.hasNext()) {
- final Number value = values.next();
- max = max != null ? max(value, max) : value;
- }
- emitter.emit(max);
+ if (values.hasNext())
+ emitter.emit(getMax(values));
+ }
+
+ private Number getMax(final Iterator<Number> numbers) {
+ Number max = null;
+ while (numbers.hasNext()) {
+ final Number value = numbers.next();
+ max = max != null ? max(value, max) : value;
}
+ return max;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MinGlobalStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MinGlobalStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MinGlobalStep.java
index 750e36c..88dd46f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MinGlobalStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/MinGlobalStep.java
@@ -27,9 +27,9 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.ConstantSupplier;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.io.Serializable;
import java.util.Collections;
@@ -98,7 +98,9 @@ public final class MinGlobalStep<S extends Number> extends ReducingBarrierStep<S
@Override
public void map(final Vertex vertex, final MapEmitter<NullObject, Number> emitter) {
- vertex.<TraverserSet<Number>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet -> traverserSet.forEach(traverser -> emitter.emit(traverser.get())));
+ final Iterator<Number> values = IteratorUtils.map(vertex.<Set<Traverser.Admin<Number>>>property(TraversalVertexProgram.HALTED_TRAVERSERS).orElse(Collections.emptySet()).iterator(), Traverser.Admin::get);
+ if (values.hasNext())
+ emitter.emit(getMin(values));
}
@Override
@@ -108,14 +110,16 @@ public final class MinGlobalStep<S extends Number> extends ReducingBarrierStep<S
@Override
public void reduce(final NullObject key, final Iterator<Number> values, final ReduceEmitter<NullObject, Number> emitter) {
- if (values.hasNext()) {
- Number min = null;
- while (values.hasNext()) {
- final Number value = values.next();
- min = min != null ? min(value, min) : value;
- }
- emitter.emit(min);
+ if (values.hasNext()) emitter.emit(getMin(values));
+ }
+
+ private Number getMin(final Iterator<Number> numbers) {
+ Number min = null;
+ while (numbers.hasNext()) {
+ final Number value = numbers.next();
+ min = min != null ? min(value, min) : value;
}
+ return min;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/SumGlobalStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/SumGlobalStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/SumGlobalStep.java
index 14573d1..8c13192 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/SumGlobalStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/SumGlobalStep.java
@@ -27,11 +27,12 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.ConstantSupplier;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.io.Serializable;
+import java.util.Collections;
import java.util.EnumSet;
import java.util.Iterator;
import java.util.Set;
@@ -104,7 +105,10 @@ public final class SumGlobalStep<S extends Number> extends ReducingBarrierStep<S
@Override
public void map(final Vertex vertex, final MapEmitter<NullObject, Number> emitter) {
- vertex.<TraverserSet<Number>>property(TraversalVertexProgram.HALTED_TRAVERSERS).ifPresent(traverserSet -> traverserSet.forEach(traverser -> emitter.emit(mul(traverser.get(), traverser.bulk()))));
+ final Iterator<Number> values = IteratorUtils.map(vertex.<Set<Traverser.Admin<Number>>>property(TraversalVertexProgram.HALTED_TRAVERSERS).orElse(Collections.emptySet()).iterator(),
+ traverser -> mul(traverser.get(), traverser.bulk()));
+ if (values.hasNext())
+ emitter.emit(getSum(values));
}
@Override
@@ -114,13 +118,16 @@ public final class SumGlobalStep<S extends Number> extends ReducingBarrierStep<S
@Override
public void reduce(final NullObject key, final Iterator<Number> values, final ReduceEmitter<NullObject, Number> emitter) {
- if (values.hasNext()) {
- Number sum = values.next();
- while (values.hasNext()) {
- sum = add(sum, values.next());
- }
- emitter.emit(sum);
+ if (values.hasNext())
+ emitter.emit(getSum(values));
+ }
+
+ private Number getSum(final Iterator<Number> numbers) {
+ Number sum = numbers.next();
+ while (numbers.hasNext()) {
+ sum = add(sum, numbers.next());
}
+ return sum;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/TreeStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/TreeStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/TreeStep.java
index 3de0a9d..92ffb76 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/TreeStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/TreeStep.java
@@ -94,16 +94,6 @@ public final class TreeStep<S> extends ReducingBarrierStep<S, Tree> implements M
}
@Override
- public Traverser<Tree> processNextStart() {
- if (this.byPass) {
- final Traverser.Admin<S> traverser = this.starts.next();
- return traverser.split(this.reducingBiFunction.apply(new Tree(), traverser), this);
- } else {
- return super.processNextStart();
- }
- }
-
- @Override
public String toString() {
return StringFactory.stepString(this, this.traversalRing);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/388d1f25/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
index b459071..fb7ecb7 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/util/ReducingBarrierStep.java
@@ -26,10 +26,8 @@ import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
-import org.apache.tinkerpop.gremlin.process.traversal.step.EngineDependent;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.process.traversal.util.FastNoSuchElementException;
@@ -44,14 +42,13 @@ import java.util.function.Supplier;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> implements MapReducer, EngineDependent, Barrier {
+public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> implements MapReducer, Barrier {
public static final String REDUCING = Graph.Hidden.hide("reducing");
protected Supplier<E> seedSupplier;
protected BiFunction<E, Traverser<S>, E> reducingBiFunction;
private boolean done = false;
- protected boolean byPass = false;
private E seed = null;
@@ -68,11 +65,6 @@ public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> imple
}
@Override
- public void onEngine(final TraversalEngine traversalEngine) {
- this.byPass = traversalEngine.isComputer();
- }
-
- @Override
public void reset() {
super.reset();
this.done = false;
@@ -80,27 +72,35 @@ public abstract class ReducingBarrierStep<S, E> extends AbstractStep<S, E> imple
}
@Override
- public void processAllStarts() {
- if (!this.byPass) {
- if (this.seed == null) this.seed = this.seedSupplier.get();
- while (this.starts.hasNext())
- this.seed = this.reducingBiFunction.apply(this.seed, this.starts.next());
+ public void addStarts(final Iterator<Traverser<S>> starts) {
+ if (starts.hasNext()) {
+ this.done = false;
+ super.addStarts(starts);
}
}
@Override
+ public void addStart(final Traverser<S> start) {
+ this.done = false;
+ super.addStart(start);
+ }
+
+ @Override
+ public void processAllStarts() {
+ if (this.seed == null) this.seed = this.seedSupplier.get();
+ while (this.starts.hasNext())
+ this.seed = this.reducingBiFunction.apply(this.seed, this.starts.next());
+ }
+
+ @Override
public Traverser<E> processNextStart() {
- if (this.byPass) {
- return (Traverser<E>) this.starts.next();
- } else {
- if (this.done)
- throw FastNoSuchElementException.instance();
- this.processAllStarts();
- this.done = true;
- final Traverser<E> traverser = TraversalHelper.getRootTraversal(this.getTraversal()).getTraverserGenerator().generate(FinalGet.tryFinalGet(this.seed), (Step) this, 1l);
- this.seed = null;
- return traverser;
- }
+ if (this.done)
+ throw FastNoSuchElementException.instance();
+ this.processAllStarts();
+ this.done = true;
+ final Traverser<E> traverser = TraversalHelper.getRootTraversal(this.getTraversal()).getTraverserGenerator().generate(FinalGet.tryFinalGet(this.seed), (Step) this, 1l);
+ this.seed = null;
+ return traverser;
}
@Override
[7/8] incubator-tinkerpop git commit: minor nothing.
Posted by ok...@apache.org.
minor nothing.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/ba22c7aa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/ba22c7aa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/ba22c7aa
Branch: refs/heads/tp31
Commit: ba22c7aadab066e1f685368cc51faca0b7597831
Parents: 388d1f2
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Sun Feb 7 19:38:13 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Sun Feb 7 19:38:13 2016 -0700
----------------------------------------------------------------------
.../gremlin/process/computer/traversal/TraverserExecutor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ba22c7aa/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
index 2df682c..04f33f2 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraverserExecutor.java
@@ -61,9 +61,9 @@ public final class TraverserExecutor {
}
}
// while there are still local traversers, process them until they leave the vertex or halt (i.e. isHalted()).
- Step<?, ?> previousStep = EmptyStep.instance();
while (!toProcessTraversers.isEmpty()) {
// process local traversers and if alive, repeat, else halt.
+ Step<?, ?> previousStep = EmptyStep.instance();
Iterator<Traverser.Admin<Object>> traversers = toProcessTraversers.iterator();
while (traversers.hasNext()) {
final Traverser.Admin<Object> traverser = traversers.next();
[3/8] incubator-tinkerpop git commit: Merge branch 'tp31'
Posted by ok...@apache.org.
Merge branch 'tp31'
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/12b9b21f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/12b9b21f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/12b9b21f
Branch: refs/heads/tp31
Commit: 12b9b21fd8dac5a0ff6c645ec38e742ee400e72e
Parents: d6e22ff 93f430e
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu Feb 4 07:33:51 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu Feb 4 07:33:51 2016 -0700
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../clustering/peerpressure/PeerPressureVertexProgramTest.java | 4 ++--
.../gremlin/hadoop/structure/io/gryo/GryoRecordReader.java | 3 ++-
3 files changed, 5 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
[8/8] incubator-tinkerpop git commit: merge tweaks. updated CHANGELOG.
Posted by ok...@apache.org.
merge tweaks. 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/b07400cc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/b07400cc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/b07400cc
Branch: refs/heads/tp31
Commit: b07400ccb8c1154b75440ae1ad96c12c78e74680
Parents: 1286b4f ba22c7a
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Feb 9 10:09:01 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Feb 9 10:09:01 2016 -0700
----------------------------------------------------------------------
CHANGELOG.asciidoc | 3 +-
.../process/computer/GiraphComputation.java | 3 +-
.../computer/traversal/TraverserExecutor.java | 101 +++++++++++--------
.../traversal/step/map/CountGlobalStep.java | 21 +++-
.../traversal/step/map/GroupCountStep.java | 16 +--
.../process/traversal/step/map/GroupStep.java | 10 +-
.../traversal/step/map/GroupStepV3d0.java | 10 +-
.../traversal/step/map/MaxGlobalStep.java | 23 +++--
.../traversal/step/map/MinGlobalStep.java | 22 ++--
.../traversal/step/map/SumGlobalStep.java | 23 +++--
.../process/traversal/step/map/TreeStep.java | 10 --
.../step/util/ReducingBarrierStep.java | 50 ++++-----
.../gremlin/util/iterator/DoubleIterator.java | 12 ++-
.../gremlin/util/iterator/IteratorUtils.java | 44 ++++++++
.../gremlin/util/iterator/MultiIterator.java | 7 +-
.../gremlin/util/iterator/SingleIterator.java | 9 +-
.../process/computer/TinkerMessenger.java | 13 +--
17 files changed, 244 insertions(+), 133 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b07400cc/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --cc CHANGELOG.asciidoc
index 6b51ba8,4881c52..0cd221f
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@@ -22,16 -22,10 +22,17 @@@ TinkerPop 3.1.0 (A 187 On The Undercove
image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/static/images/gremlin-gangster.png[width=185]
-[[release-3.1.1-incubating]]
-TinkerPop 3.1.1 (NOT OFFICIALLY RELEASED YET)
+[[release-3.1.2-incubating]]
+TinkerPop 3.1.2 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
++* Optimized memory-usage in `TraversalVertexProgram`.
++* Greatly reduced the amount of objects required in OLAP for the `ReducingBarrierStep` steps.
+
+[[release-3.1.1-incubating]]
+TinkerPop 3.1.1 (Release Date: February 8, 2016)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
* Made `GryoRecordReader` more robust to 0 byte record splits.
* Fixed a constructor/serialization bug in `LP_O_OB_S_SE_SL_Traverser`.
* Added a lazy iterator, memory safe implementation of MapReduce to `SparkGraphComputer`.
[5/8] incubator-tinkerpop git commit: Figured out a neat trick.
IteratorUtils.noRemove(Iterator) so if a providers Messenger doesn't support
Iterator.remove(),
they can simply wrap the iterator they provide in Iterator.noRemove().
Chillin'.
Posted by ok...@apache.org.
Figured out a neat trick. IteratorUtils.noRemove(Iterator) so if a providers Messenger doesn't support Iterator.remove(), they can simply wrap the iterator they provide in Iterator.noRemove(). Chillin'.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/047ee206
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/047ee206
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/047ee206
Branch: refs/heads/tp31
Commit: 047ee2065023227381e1d2e594354b05331f06fd
Parents: b54ddf2
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri Feb 5 15:30:53 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri Feb 5 15:30:53 2016 -0700
----------------------------------------------------------------------
.../process/computer/GiraphComputation.java | 3 ++-
.../gremlin/util/iterator/IteratorUtils.java | 19 +++++++++++++++++++
.../gremlin/util/iterator/MultiIterator.java | 2 +-
.../process/computer/TinkerMessenger.java | 13 +++++++------
4 files changed, 29 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/047ee206/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
index 1d52566..f094f2d 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
@@ -26,6 +26,7 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.io.IOException;
@@ -41,7 +42,7 @@ public final class GiraphComputation extends BasicComputation<ObjectWritable, Ve
public void compute(final Vertex<ObjectWritable, VertexWritable, NullWritable> vertex, final Iterable<ObjectWritable> messages) throws IOException {
final GiraphWorkerContext workerContext = this.getWorkerContext();
final VertexProgram<?> vertexProgram = workerContext.getVertexProgramPool().take();
- vertexProgram.execute(ComputerGraph.vertexProgram(vertex.getValue().get(), vertexProgram), workerContext.getMessenger((GiraphVertex) vertex, this, messages.iterator()), workerContext.getMemory());
+ vertexProgram.execute(ComputerGraph.vertexProgram(vertex.getValue().get(), vertexProgram), workerContext.getMessenger((GiraphVertex) vertex, this, IteratorUtils.noRemove(messages.iterator())), workerContext.getMemory());
workerContext.getVertexProgramPool().offer(vertexProgram);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/047ee206/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
index 808317f..78ffe4a 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/IteratorUtils.java
@@ -392,4 +392,23 @@ public final class IteratorUtils {
public static <T> Stream<T> stream(final Iterable<T> iterable) {
return IteratorUtils.stream(iterable.iterator());
}
+
+ public static <T> Iterator<T> noRemove(final Iterator<T> iterator) {
+ return new Iterator<T>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public void remove() {
+ // do nothing
+ }
+
+ @Override
+ public T next() {
+ return iterator.next();
+ }
+ };
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/047ee206/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
index e272a08..20c0946 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/util/iterator/MultiIterator.java
@@ -59,7 +59,7 @@ public final class MultiIterator<T> implements Iterator<T>, Serializable {
@Override
public void remove() {
- // this.iterators.get(this.current).remove();
+ this.iterators.get(this.current).remove();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/047ee206/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMessenger.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMessenger.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMessenger.java
index d0c302a..71d7030 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMessenger.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMessenger.java
@@ -29,6 +29,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.apache.tinkerpop.gremlin.util.iterator.MultiIterator;
import java.util.Iterator;
@@ -64,19 +65,19 @@ public final class TinkerMessenger<M> implements Messenger<M> {
final Traversal.Admin<Vertex, Edge> incidentTraversal = TinkerMessenger.setVertexStart(localMessageScope.getIncidentTraversal().get().asAdmin(), this.vertex);
final Direction direction = TinkerMessenger.getDirection(incidentTraversal);
final Edge[] edge = new Edge[1]; // simulates storage side-effects available in Gremlin, but not Java8 streams
- multiIterator.addIterator(StreamSupport.stream(Spliterators.spliteratorUnknownSize(VertexProgramHelper.reverse(incidentTraversal.asAdmin()), Spliterator.IMMUTABLE | Spliterator.SIZED), false)
+ multiIterator.addIterator(IteratorUtils.noRemove(StreamSupport.stream(Spliterators.spliteratorUnknownSize(VertexProgramHelper.reverse(incidentTraversal.asAdmin()), Spliterator.IMMUTABLE | Spliterator.SIZED), false)
.map(e -> this.messageBoard.receiveMessages.get((edge[0] = e).vertices(direction).next()))
.filter(q -> null != q)
.flatMap(Queue::stream)
.map(message -> localMessageScope.getEdgeFunction().apply(message, edge[0]))
- .iterator());
+ .iterator()));
} else {
- multiIterator.addIterator(Stream.of(this.vertex)
+ multiIterator.addIterator(IteratorUtils.noRemove(Stream.of(this.vertex)
.map(this.messageBoard.receiveMessages::get)
.filter(q -> null != q)
.flatMap(Queue::stream)
- .iterator());
+ .iterator()));
}
}
return multiIterator;
@@ -93,8 +94,8 @@ public final class TinkerMessenger<M> implements Messenger<M> {
}
private void addMessage(final Vertex vertex, final M message) {
- this.messageBoard.sendMessages.compute(vertex, (v,queue) -> {
- if(null == queue) queue = new ConcurrentLinkedQueue<>();
+ this.messageBoard.sendMessages.compute(vertex, (v, queue) -> {
+ if (null == queue) queue = new ConcurrentLinkedQueue<>();
queue.add(null != this.combiner && !queue.isEmpty() ? this.combiner.combine(queue.remove(), message) : message);
return queue;
});
[2/8] incubator-tinkerpop git commit: Merge branch 'tp31'
Posted by ok...@apache.org.
Merge branch 'tp31'
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/d6e22ff9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/d6e22ff9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/d6e22ff9
Branch: refs/heads/tp31
Commit: d6e22ff901f2941ecb9e28a018a4686dc3588433
Parents: ffbecdb 580c683
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Feb 3 14:28:19 2016 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed Feb 3 14:28:19 2016 -0700
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
docs/src/dev/developer/contributing.asciidoc | 11 ++++++-----
gremlin-console/src/main/NOTICE | 12 ++++++++++--
.../traversal/traverser/LP_O_OB_S_SE_SL_Traverser.java | 3 +++
gremlin-server/src/main/NOTICE | 12 ++++++++++--
.../gremlin/spark/structure/io/InputFormatRDD.java | 6 +-----
6 files changed, 31 insertions(+), 14 deletions(-)
----------------------------------------------------------------------