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 2015/02/20 20:00:47 UTC
[6/7] incubator-tinkerpop git commit: Merge master.
Merge master.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/995c5dcc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/995c5dcc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/995c5dcc
Branch: refs/heads/master
Commit: 995c5dcc3c4d66eada485770c169480fd5fa489b
Parents: 91ad775 f37abf3
Author: rjbriody <bo...@datastax.com>
Authored: Fri Feb 20 13:31:20 2015 -0500
Committer: rjbriody <bo...@datastax.com>
Committed: Fri Feb 20 13:31:20 2015 -0500
----------------------------------------------------------------------
docs/src/implementations.asciidoc | 34 ++++----
.../traversal/step/map/CountGlobalStep.java | 10 +--
.../traversal/step/map/GroupCountStep.java | 18 ++--
.../graph/traversal/step/map/GroupStep.java | 66 ++++++++-------
.../graph/traversal/step/map/MaxGlobalStep.java | 20 ++---
.../traversal/step/map/MeanGlobalStep.java | 20 ++---
.../graph/traversal/step/map/MinGlobalStep.java | 20 ++---
.../graph/traversal/step/map/SumGlobalStep.java | 20 ++---
.../step/sideEffect/AggregateStep.java | 77 +++++++++++++++--
.../sideEffect/GroupCountSideEffectStep.java | 14 ++--
.../step/sideEffect/GroupSideEffectStep.java | 55 ++++++------
.../traversal/step/sideEffect/ProfileStep.java | 57 ++++++++++++-
.../traversal/step/sideEffect/StoreStep.java | 74 ++++++++++++++--
.../traversal/step/sideEffect/TreeStep.java | 59 ++++++++++++-
.../mapreduce/AggregateMapReduce.java | 88 --------------------
.../sideEffect/mapreduce/ProfileMapReduce.java | 66 ---------------
.../sideEffect/mapreduce/StoreMapReduce.java | 88 --------------------
.../sideEffect/mapreduce/TreeMapReduce.java | 80 ------------------
.../step/util/ReducingBarrierStep.java | 1 -
.../process/traversal/util/TraversalUtil.java | 8 ++
.../TraversalVerificationStrategyTest.java | 3 -
hadoop-gremlin/conf/hadoop-kryo.properties | 2 +-
22 files changed, 406 insertions(+), 474 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/995c5dcc/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/sideEffect/ProfileStep.java
----------------------------------------------------------------------
diff --cc gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/sideEffect/ProfileStep.java
index 3c14fe2,830a969..4d00473
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/sideEffect/ProfileStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/sideEffect/ProfileStep.java
@@@ -21,12 -21,18 +21,19 @@@ package org.apache.tinkerpop.gremlin.pr
import org.apache.tinkerpop.gremlin.process.Step;
import org.apache.tinkerpop.gremlin.process.Traversal;
import org.apache.tinkerpop.gremlin.process.Traverser;
+ import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
- import org.apache.tinkerpop.gremlin.process.graph.traversal.step.sideEffect.mapreduce.ProfileMapReduce;
+ import org.apache.tinkerpop.gremlin.process.computer.traversal.VertexTraversalSideEffects;
+ import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
-import org.apache.tinkerpop.gremlin.process.traversal.step.AbstractStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Reversible;
+import org.apache.tinkerpop.gremlin.process.traversal.step.*;
- import org.apache.tinkerpop.gremlin.process.util.metric.*;
++import org.apache.tinkerpop.gremlin.process.util.metric.DependantMutableMetrics;
++import org.apache.tinkerpop.gremlin.process.util.metric.MutableMetrics;
+ import org.apache.tinkerpop.gremlin.process.util.metric.StandardTraversalMetrics;
+ import org.apache.tinkerpop.gremlin.process.util.metric.TraversalMetrics;
+ import org.apache.tinkerpop.gremlin.structure.Vertex;
+ import java.util.Iterator;
+import java.util.List;
import java.util.NoSuchElementException;
/**
@@@ -78,85 -90,55 +85,129 @@@ public final class ProfileStep<S> exten
return this.starts.next();
}
- private StandardTraversalMetrics getTraversalMetricsUtil() {
- StandardTraversalMetrics traversalMetrics = this.getTraversal().getSideEffects().getOrCreate(TraversalMetrics.METRICS_KEY, StandardTraversalMetrics::new);
- final boolean isComputer = this.getTraversal().getEngine().isComputer();
- traversalMetrics.initializeIfNecessary(this.getId(), this.getTraversal().getSteps().indexOf(this), name, isComputer);
- return traversalMetrics;
+ private void initializeIfNeeded() {
+ if (traversalMetrics != null) {
+ return;
+ }
+
+ createTraversalMetricsSideEffectIfNecessary();
+
+ // How can traversalMetrics still be null? When running on computer it may need to be re-initialized from
+ // sideEffects after serialization.
+ if (traversalMetrics == null) {
+ // look up the TraversalMetrics in the root traversal's sideEffects
+ Traversal t = this.getTraversal();
+ while (!(t.asAdmin().getParent() instanceof EmptyStep)) {
+ t = t.asAdmin().getParent().asStep().getTraversal();
+ }
+ traversalMetrics = t.asAdmin().getSideEffects().get(TraversalMetrics.METRICS_KEY);
+ }
+ }
+
+ private void createTraversalMetricsSideEffectIfNecessary() {
+ if (this.getTraversal().getSideEffects().exists(TraversalMetrics.METRICS_KEY)) {
+ // Already initialized
+ return;
+ }
+
+ if (!(this.getTraversal().getParent() instanceof EmptyStep)) {
+ // Initialization is handled at the top-level of the traversal only.
+ return;
+ }
+
+ // The following code is executed once per top-level (non-nested) Traversal for all Profile steps. (Technically,
+ // once per thread if using Computer.)
+
+ traversalMetrics = this.getTraversal().getSideEffects().getOrCreate(TraversalMetrics.METRICS_KEY, StandardTraversalMetrics::new);
+ prepTraversalForProfiling(this.getTraversal().asAdmin(), null);
+ }
+
+ // Walk the traversal steps and initialize the Metrics timers.
+ private void prepTraversalForProfiling(Traversal.Admin<?, ?> traversal, MutableMetrics parentMetrics) {
+
+ DependantMutableMetrics prevMetrics = null;
+ final List<Step> steps = traversal.getSteps();
+ for (int ii = 0; ii + 1 < steps.size(); ii = ii + 2) {
+ Step step = steps.get(ii);
+ ProfileStep profileStep = (ProfileStep) steps.get(ii + 1);
+
+ // Create metrics
+ MutableMetrics metrics;
+
+ // Computer metrics are "stand-alone" but Standard metrics handle double-counted upstream time.
+ if (traversal.getEngine().isComputer()) {
+ metrics = new MutableMetrics(step.getId(), step.toString());
+ } else {
+ metrics = new DependantMutableMetrics(step.getId(), step.toString(), prevMetrics);
+ prevMetrics = (DependantMutableMetrics) metrics;
+ }
+
+ // Initialize counters (necessary because some steps might end up being 0)
+ metrics.incrementCount(TraversalMetrics.ELEMENT_COUNT_ID, 0);
+ metrics.incrementCount(TraversalMetrics.TRAVERSER_COUNT_ID, 0);
+
+ // Add metrics to parent, if necessary
+ if (parentMetrics != null) {
+ parentMetrics.addNested(metrics);
+ }
+
+ // The TraversalMetrics sideEffect is shared across all the steps.
+ profileStep.traversalMetrics = this.traversalMetrics;
+
+ // Add root metrics to traversalMetrics
+ this.traversalMetrics.addMetrics(metrics, step.getId(), ii / 2, parentMetrics == null, profileStep.getId());
+
+ // Handle nested traversal
+ if (step instanceof TraversalParent) {
+ for (Traversal.Admin<?, ?> t : ((TraversalParent) step).getLocalChildren()) {
+ prepTraversalForProfiling(t, metrics);
+ }
+ }
+ }
}
+ //////////////////
+
+ public static final class ProfileMapReduce extends StaticMapReduce<MapReduce.NullObject, StandardTraversalMetrics, MapReduce.NullObject, StandardTraversalMetrics, StandardTraversalMetrics> {
+
+ private static ProfileMapReduce INSTANCE = new ProfileMapReduce();
+
+ private ProfileMapReduce() {
+
+ }
+
+ @Override
+ public boolean doStage(final Stage stage) {
+ return true;
+ }
+
+ @Override
+ public String getMemoryKey() {
+ return TraversalMetrics.METRICS_KEY;
+ }
+
+ @Override
+ public void map(final Vertex vertex, final MapEmitter<NullObject, StandardTraversalMetrics> emitter) {
+ VertexTraversalSideEffects.of(vertex).<StandardTraversalMetrics>ifPresent(TraversalMetrics.METRICS_KEY, emitter::emit);
+ }
+
+ @Override
+ public void combine(final NullObject key, final Iterator<StandardTraversalMetrics> values, final ReduceEmitter<NullObject, StandardTraversalMetrics> emitter) {
+ reduce(key, values, emitter);
+ }
+
+ @Override
+ public void reduce(final NullObject key, final Iterator<StandardTraversalMetrics> values, final ReduceEmitter<NullObject, StandardTraversalMetrics> emitter) {
+ emitter.emit(StandardTraversalMetrics.merge(values));
+ }
+
+ @Override
+ public StandardTraversalMetrics generateFinalResult(final Iterator<KeyValue<NullObject, StandardTraversalMetrics>> keyValues) {
+ return keyValues.next().getValue();
+ }
+
+ public static ProfileMapReduce instance() {
+ return INSTANCE;
+ }
+ }
}