You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2015/03/05 14:24:22 UTC
incubator-tinkerpop git commit: Refactored TinkerWorkerPool to allow
reuse of it for mapreduce or vertexprograms.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 6d40a9b0f -> b34785f06
Refactored TinkerWorkerPool to allow reuse of it for mapreduce or vertexprograms.
Prior to this change, the threadpool was being created and destroyed each between use in a vertex program or for each step of a mapreduce. That was an unecessary expense.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/b34785f0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/b34785f0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/b34785f0
Branch: refs/heads/master
Commit: b34785f06dd38ead3aa8211ea464ada7f36cd564
Parents: 6d40a9b
Author: Stephen Mallette <sp...@apache.org>
Authored: Thu Mar 5 08:21:23 2015 -0500
Committer: Stephen Mallette <sp...@apache.org>
Committed: Thu Mar 5 08:21:23 2015 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../process/computer/TinkerGraphComputer.java | 48 ++++++------
.../process/computer/TinkerWorkerPool.java | 81 +++++++++-----------
3 files changed, 61 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b34785f0/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 87fab47..01ad2d9 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -27,6 +27,7 @@ TinkerPop 3.0.0.M8 (Release Date: NOT OFFICIALLY RELEASED YET)
* Renamed "Gremlin Kryo" to "Gryo".
* Implement profiling for nested traversals.
+* Refactored `TinkerWorkerPool` to use `ExecutorService` so as to reuse threads when executing graph computer functions.
* Removed `Reducing.Reducer` and `ReducingStrategy`. Previous `Reducing` classes are now `MapReducer` classes.
* Refactored the "process" test suite to allow for better test configuration with respect to different `TraversalEngine` implementations.
* Added `hasNot(traversal)` which is a faster way of doing `has(traversal.count().is(0L))`.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b34785f0/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
index 013bd36..5b6306a 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
@@ -96,15 +96,15 @@ public class TinkerGraphComputer implements GraphComputer {
this.memory = new TinkerMemory(this.vertexProgram, this.mapReducers);
return CompletableFuture.<ComputerResult>supplyAsync(() -> {
final long time = System.currentTimeMillis();
- if (null != this.vertexProgram) {
- TinkerHelper.createGraphView(this.graph, this.isolation, this.vertexProgram.getElementComputeKeys());
+ try (final TinkerWorkerPool workers = new TinkerWorkerPool(Runtime.getRuntime().availableProcessors())) {
+ if (null != this.vertexProgram) {
+ TinkerHelper.createGraphView(this.graph, this.isolation, this.vertexProgram.getElementComputeKeys());
- // execute the vertex program
- this.vertexProgram.setup(this.memory);
- this.memory.completeSubRound();
- try (final TinkerWorkerPool workers = new TinkerWorkerPool(Runtime.getRuntime().availableProcessors(), this.vertexProgram)) {
+ // execute the vertex program
+ this.vertexProgram.setup(this.memory);
+ this.memory.completeSubRound();
while (true) {
- workers.executeVertexProgram(vertexProgram -> vertexProgram.workerIterationStart(this.memory.asImmutable()));
+ workers.executeVertexProgram(vertexProgram -> vertexProgram.workerIterationStart(this.memory.asImmutable()), vertexProgram);
final SynchronizedIterator<Vertex> vertices = new SynchronizedIterator<>(sg.iterators().vertexIterator());
workers.executeVertexProgram(vertexProgram -> {
while (true) {
@@ -112,8 +112,8 @@ public class TinkerGraphComputer implements GraphComputer {
if (null == vertex) return;
vertexProgram.execute(vertex, new TinkerMessenger(vertex, this.messageBoard, vertexProgram.getMessageCombiner()), this.memory);
}
- });
- workers.executeVertexProgram(vertexProgram -> vertexProgram.workerIterationEnd(this.memory.asImmutable()));
+ }, vertexProgram);
+ workers.executeVertexProgram(vertexProgram -> vertexProgram.workerIterationEnd(this.memory.asImmutable()), vertexProgram);
this.messageBoard.completeIteration();
this.memory.completeSubRound();
if (this.vertexProgram.terminate(this.memory)) {
@@ -125,14 +125,10 @@ public class TinkerGraphComputer implements GraphComputer {
this.memory.completeSubRound();
}
}
- } catch (final Exception ex) {
- throw new IllegalStateException(ex.getMessage(), ex);
}
- }
- // execute mapreduce jobs
- for (final MapReduce mapReduce : this.mapReducers) {
- try (final TinkerWorkerPool workers = new TinkerWorkerPool(Runtime.getRuntime().availableProcessors(), mapReduce)) {
+ // execute mapreduce jobs
+ for (final MapReduce mapReduce : mapReducers) {
if (mapReduce.doStage(MapReduce.Stage.MAP)) {
final TinkerMapEmitter<?, ?> mapEmitter = new TinkerMapEmitter<>(mapReduce.doStage(MapReduce.Stage.REDUCE));
final SynchronizedIterator<Vertex> vertices = new SynchronizedIterator<>(sg.iterators().vertexIterator());
@@ -142,8 +138,11 @@ public class TinkerGraphComputer implements GraphComputer {
if (null == vertex) return;
workerMapReduce.map(vertex, mapEmitter);
}
- });
- mapEmitter.complete(mapReduce); // sort results if a map output sort is defined
+ }, mapReduce);
+
+ // sort results if a map output sort is defined
+ mapEmitter.complete(mapReduce);
+
// no need to run combiners as this is single machine
if (mapReduce.doStage(MapReduce.Stage.REDUCE)) {
final TinkerReduceEmitter<?, ?> reduceEmitter = new TinkerReduceEmitter<>();
@@ -154,21 +153,22 @@ public class TinkerGraphComputer implements GraphComputer {
if (null == entry) return;
workerMapReduce.reduce(entry.getKey(), entry.getValue().iterator(), reduceEmitter);
}
- });
+ }, mapReduce);
reduceEmitter.complete(mapReduce); // sort results if a reduce output sort is defined
mapReduce.addResultToMemory(this.memory, reduceEmitter.reduceQueue.iterator());
} else {
mapReduce.addResultToMemory(this.memory, mapEmitter.mapQueue.iterator());
}
}
- } catch (Exception ex) {
- throw new RuntimeException(ex);
}
+ // update runtime and return the newly computed graph
+ this.memory.setRuntime(System.currentTimeMillis() - time);
+ this.memory.complete();
+ return new TinkerComputerResult(sg, this.memory.asImmutable());
+
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
}
- // update runtime and return the newly computed graph
- this.memory.setRuntime(System.currentTimeMillis() - time);
- this.memory.complete();
- return new TinkerComputerResult(sg, this.memory.asImmutable());
});
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b34785f0/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerWorkerPool.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerWorkerPool.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerWorkerPool.java
index d418d4a..d4e3e7d 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerWorkerPool.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerWorkerPool.java
@@ -39,61 +39,26 @@ import java.util.stream.Collectors;
public class TinkerWorkerPool implements AutoCloseable {
private static final BasicThreadFactory threadFactoryWorker = new BasicThreadFactory.Builder().namingPattern("tinker-worker-%d").build();
- public static enum State {VERTEX_PROGRAM, MAP_REDUCE}
-
- private final List<MapReduce> mapReducers;
- private final List<VertexProgram> vertexPrograms;
- private final State state;
+ private final int numberOfWorkers;
private final ExecutorService workerPool;
- public TinkerWorkerPool(final int numberOfWorkers, final VertexProgram vertexProgram) {
- try {
- this.state = State.VERTEX_PROGRAM;
- this.vertexPrograms = new ArrayList<>(numberOfWorkers);
- this.mapReducers = Collections.emptyList();
- for (int i = 0; i < numberOfWorkers; i++) {
- this.vertexPrograms.add(vertexProgram.clone());
- }
-
- workerPool = Executors.newFixedThreadPool(numberOfWorkers, threadFactoryWorker);
-
- } catch (final CloneNotSupportedException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- }
-
- public TinkerWorkerPool(final int numberOfWorkers, final MapReduce mapReduce) {
- try {
- this.state = State.MAP_REDUCE;
- this.vertexPrograms = Collections.emptyList();
- this.mapReducers = new ArrayList<>(numberOfWorkers);
- for (int i = 0; i < numberOfWorkers; i++) {
- this.mapReducers.add(mapReduce.clone());
- }
-
- workerPool = Executors.newFixedThreadPool(numberOfWorkers, threadFactoryWorker);
-
- } catch (final CloneNotSupportedException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
+ public TinkerWorkerPool(final int numberOfWorkers) {
+ this.numberOfWorkers = numberOfWorkers;
+ workerPool = Executors.newFixedThreadPool(numberOfWorkers, threadFactoryWorker);
}
- public void executeVertexProgram(final Consumer<VertexProgram> worker) {
- if (!this.state.equals(State.VERTEX_PROGRAM))
- throw new IllegalStateException("The provided TinkerWorkerPool is not setup for VertexProgram: " + this.state);
-
- this.vertexPrograms.stream()
+ public void executeVertexProgram(final Consumer<VertexProgram> worker, final VertexProgram vertexProgram) {
+ final List<VertexProgram> vertexPrograms = cloneVertexProgram(vertexProgram);
+ vertexPrograms.stream()
.map(vp -> (Future<Void>) workerPool.submit(() -> worker.accept(vp)))
.collect(Collectors.toList())
.forEach(FunctionUtils.wrapConsumer(Future::get));
}
- public void executeMapReduce(final Consumer<MapReduce> worker) {
- if (!this.state.equals(State.MAP_REDUCE))
- throw new IllegalStateException("The provided TinkerWorkerPool is not setup for MapReduce: " + this.state);
-
- this.mapReducers.stream()
+ public void executeMapReduce(final Consumer<MapReduce> worker, final MapReduce mapReduce) {
+ final List<MapReduce> mapReducers = cloneMapReducer(mapReduce);
+ mapReducers.stream()
.map(mr -> (Future<Void>) workerPool.submit(() -> worker.accept(mr)))
.collect(Collectors.toList())
.forEach(FunctionUtils.wrapConsumer(Future::get));
@@ -103,4 +68,30 @@ public class TinkerWorkerPool implements AutoCloseable {
public void close() throws Exception {
workerPool.shutdown();
}
+
+ private List<VertexProgram> cloneVertexProgram(final VertexProgram vertexProgram) {
+ final List<VertexProgram> vertexPrograms;
+ try {
+ vertexPrograms = new ArrayList<>(numberOfWorkers);
+ for (int i = 0; i < numberOfWorkers; i++) {
+ vertexPrograms.add(vertexProgram.clone());
+ }
+ } catch (final CloneNotSupportedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ return vertexPrograms;
+ }
+
+ private List<MapReduce> cloneMapReducer(final MapReduce mapReduce) {
+ final List<MapReduce> mapReducers;
+ try {
+ mapReducers = new ArrayList<>(numberOfWorkers);
+ for (int i = 0; i < numberOfWorkers; i++) {
+ mapReducers.add(mapReduce.clone());
+ }
+ } catch (final CloneNotSupportedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ return mapReducers;
+ }
}
\ No newline at end of file