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/28 17:26:37 UTC
incubator-tinkerpop git commit: more play work....
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1214 1a8851f8e -> 6b43a5da4
more play work....
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/6b43a5da
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/6b43a5da
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/6b43a5da
Branch: refs/heads/TINKERPOP-1214
Commit: 6b43a5da4842785611f8fdadcf544394330f8f2b
Parents: 1a8851f
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Mon Mar 28 09:26:28 2016 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Mon Mar 28 09:26:28 2016 -0600
----------------------------------------------------------------------
.../traversal/step/branch/ParallelStep.java | 29 ++++++++++++--------
.../structure/TinkerGraphPlayTest.java | 12 +++++---
2 files changed, 26 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6b43a5da/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/branch/ParallelStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/branch/ParallelStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/branch/ParallelStep.java
index 3cd4741..5ed2d69 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/branch/ParallelStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/branch/ParallelStep.java
@@ -19,6 +19,7 @@
package org.apache.tinkerpop.gremlin.process.traversal.step.branch;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
@@ -27,8 +28,6 @@ import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequire
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
@@ -42,46 +41,54 @@ import java.util.concurrent.Future;
*/
public final class ParallelStep<S, E> extends AbstractStep<S, E> implements TraversalParent {
+ //private final Queue<Traverser.Admin<E>> barrier = new ConcurrentLinkedQueue<>();
private final TraverserSet<E> barrier = new TraverserSet<>();
private boolean first = true;
- private final Traversal.Admin<S, E> threadedTraversal;
private final ExecutorService executor;
private List<Callable<Boolean>> callables;
private List<Future<Boolean>> futures;
+ private List<Traversal.Admin<S, E>> threadedTraversals;
+
+ private final Object barrierMutex = new Object();
//private final Set<String> activeWorkers = new HashSet<>();
public ParallelStep(final Traversal.Admin traversal, final int threads, final Traversal<S, E> threadedTraversal) {
super(traversal);
- this.threadedTraversal = threadedTraversal.asAdmin();
+ this.threadedTraversals = new ArrayList<>();
this.executor = Executors.newFixedThreadPool(threads);
this.callables = new ArrayList<>();
for (int i = 0; i < threads; i++) {
+ final Traversal.Admin<S, E> parallelTraversal = threadedTraversal.asAdmin().clone();
+ this.threadedTraversals.add(parallelTraversal);
this.callables.add(() -> {
try {
- final Traversal.Admin<S, E> parallelTraversal = this.threadedTraversal.clone();
parallelTraversal.addStarts((Iterator) this.starts);
- while (parallelTraversal.hasNext()) {
+ final Step<?, E> endStep = parallelTraversal.getEndStep();
+ while (true) {
//activeWorkers.add(Thread.currentThread().getName());
- this.barrier.add(parallelTraversal.getEndStep().next());
+ final Traverser.Admin<E> end = endStep.next();
+ synchronized (this.barrierMutex) {
+ this.barrier.add(end);
+ }
}
- return true;
} catch (final Exception e) {
+ // System.out.println(e);
// can get null pointer cause of threading that is pulling on previous step (need to create a thread safe barrier)
- return true;
}
+ return true;
});
}
}
@Override
public List<Traversal.Admin<S, E>> getGlobalChildren() {
- return Collections.singletonList(this.threadedTraversal);
+ return this.threadedTraversals;
}
@Override
public Set<TraverserRequirement> getRequirements() {
- return this.threadedTraversal.getTraverserRequirements();
+ return this.threadedTraversals.get(0).getTraverserRequirements();
}
public Traverser.Admin<E> processNextStart() {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6b43a5da/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 0b25bc0..3bd254e 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
@@ -71,18 +71,22 @@ public class TinkerGraphPlayTest {
graph.io(GraphMLIo.build()).readGraph("../data/grateful-dead.xml");
//final Traversal traversal = g.V().parallel(3, __.<Vertex>out().out().out()).count();
- System.out.println(TimeUtil.clockWithResult(2, () -> g.V().both().parallel(8, __.repeat(both()).times(2)).count().next()));
- System.out.println(TimeUtil.clockWithResult(2, () -> g.V().repeat(both()).times(3).count().next()));
+ //System.out.println(TimeUtil.clockWithResult(1, () -> g.V().parallel(8, __.repeat(both()).times(2)).count().next()));
+ //System.out.println(TimeUtil.clockWithResult(1, () -> g.V().repeat(both()).times(2).count().next()));
+ //////
+ System.out.println(TimeUtil.clockWithResult(1, () -> g.V().both().parallel(8, __.repeat(out()).times(2)).count().next()));
+ System.out.println(TimeUtil.clockWithResult(1, () -> g.V().both().barrier().repeat(out()).times(2).count().next()));
+ //////
- System.out.println(TimeUtil.clockWithResult(2, () -> g.V().parallel(8,__.<Vertex,Vertex>match(
+ System.out.println(TimeUtil.clockWithResult(10, () -> g.V().parallel(8, __.<Vertex, Vertex>match(
as("a").in("sungBy").count().as("b"),
as("a").in("sungBy").as("c"),
as("c").out("followedBy").as("d"),
as("d").out("sungBy").as("e"),
as("e").in("sungBy").count().as("b"),
where("a", P.neq("e"))).select("a", "e").by("name")).count().next()));
- System.out.println(TimeUtil.clockWithResult(2, () -> g.V().match(
+ System.out.println(TimeUtil.clockWithResult(10, () -> g.V().match(
as("a").in("sungBy").count().as("b"),
as("a").in("sungBy").as("c"),
as("c").out("followedBy").as("d"),