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/03/05 23:47:46 UTC
[1/3] incubator-tinkerpop git commit: fixed a obnoxious bug around
GraphMemory in SparkGraphComputer --- had to use a combination of
BroadcastVariables and Accumulators. What a time pit that was.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master dd14d4fe6 -> a7f941ed5
fixed a obnoxious bug around GraphMemory in SparkGraphComputer --- had to use a combination of BroadcastVariables and Accumulators. What a time pit that was.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/b4246a0e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/b4246a0e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/b4246a0e
Branch: refs/heads/master
Commit: b4246a0e0508878004250c1eb4af63d24269bbe3
Parents: dd14d4f
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu Mar 5 13:33:53 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu Mar 5 15:46:16 2015 -0700
----------------------------------------------------------------------
.../process/computer/spark/RuleAccumulator.java | 14 +++-
.../computer/spark/SparkGraphComputer.java | 16 +++--
.../process/computer/spark/SparkMemory.java | 70 +++++++++++++++-----
.../hadoop/process/computer/util/Rule.java | 2 +-
4 files changed, 75 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b4246a0e/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/RuleAccumulator.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/RuleAccumulator.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/RuleAccumulator.java
index 446dbdb..6bd9921 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/RuleAccumulator.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/RuleAccumulator.java
@@ -28,12 +28,22 @@ public final class RuleAccumulator implements AccumulatorParam<Rule> {
@Override
public Rule addAccumulator(final Rule a, final Rule b) {
- return new Rule(b.operation, b.operation.compute(a.object, b.object));
+ if (a.operation.equals(Rule.Operation.NO_OP))
+ return b;
+ if (b.operation.equals(Rule.Operation.NO_OP))
+ return a;
+ else
+ return new Rule(b.operation, b.operation.compute(a.object, b.object));
}
@Override
public Rule addInPlace(final Rule a, final Rule b) {
- return new Rule(b.operation, b.operation.compute(a.object, b.object));
+ if (a.operation.equals(Rule.Operation.NO_OP))
+ return b;
+ if (b.operation.equals(Rule.Operation.NO_OP))
+ return a;
+ else
+ return new Rule(b.operation, b.operation.compute(a.object, b.object));
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b4246a0e/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
index 92b60ab..b061304 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
@@ -28,7 +28,6 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.storage.StorageLevel;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.util.SApacheConfiguration;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.util.SparkHelper;
@@ -124,7 +123,7 @@ public final class SparkGraphComputer implements GraphComputer {
// wire up a spark context
final SparkConf sparkConfiguration = new SparkConf();
- sparkConfiguration.setAppName(Constants.GREMLIN_HADOOP_SPARK_JOB_PREFIX + this.vertexProgram + "[" + this.mapReducers + "]");
+ sparkConfiguration.setAppName(Constants.GREMLIN_HADOOP_SPARK_JOB_PREFIX + (null == this.vertexProgram ? "No VertexProgram" : this.vertexProgram) + "[" + this.mapReducers + "]");
hadoopConfiguration.forEach(entry -> sparkConfiguration.set(entry.getKey(), entry.getValue()));
if (FileInputFormat.class.isAssignableFrom(hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputFormat.class)))
hadoopConfiguration.set(Constants.MAPRED_INPUT_DIR, hadoopConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); // necessary for Spark and newAPIHadoopRDD
@@ -146,7 +145,8 @@ public final class SparkGraphComputer implements GraphComputer {
if (null != this.vertexProgram) {
// set up the vertex program and wire up configurations
memory = new SparkMemory(this.vertexProgram, this.mapReducers, sparkContext);
- this.vertexProgram.setup(memory); // TODO: setup variables are not being broadcasted on first call (force using broadcast or dummy reduce that has a memory reference?)
+ this.vertexProgram.setup(memory);
+ memory.broadcastMemory(sparkContext);
final SApacheConfiguration vertexProgramConfiguration = new SApacheConfiguration();
this.vertexProgram.storeState(vertexProgramConfiguration);
ConfigurationUtils.copy(vertexProgramConfiguration, apacheConfiguration);
@@ -156,18 +156,20 @@ public final class SparkGraphComputer implements GraphComputer {
while (true) {
memory.setInTask(true);
graphRDD = SparkHelper.executeStep(graphRDD, memory, vertexProgramConfiguration);
- graphRDD.foreachPartition(iterator -> doNothing()); // TODO: i think this is a fast way to execute the rdd (wish there was a "execute()" method.
+ graphRDD.foreachPartition(iterator -> doNothing()); // TODO: i think this is a fast way to execute the rdd (wish there was a "execute()" method).
memory.setInTask(false);
- if (this.vertexProgram.terminate(memory)) {
- memory.incrIteration();
+ if (this.vertexProgram.terminate(memory))
break;
- } else
+ else {
memory.incrIteration();
+ memory.broadcastMemory(sparkContext);
+ }
}
// write the output graph back to disk
SparkHelper.saveGraphRDD(graphRDD, hadoopConfiguration);
}
+ // reuse the graphRDD for all map reduce jobs
graphRDD = graphRDD.cache();
//////////////////////////////
// process the map reducers //
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b4246a0e/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
index b21f752..3ef69f4 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
@@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.hadoop.process.computer.spark;
import org.apache.spark.Accumulator;
import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.util.Rule;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
@@ -43,9 +44,10 @@ import java.util.concurrent.atomic.AtomicLong;
public final class SparkMemory implements Memory.Admin, Serializable {
public final Set<String> memoryKeys = new HashSet<>();
- private final AtomicInteger iteration = new AtomicInteger(0);
+ private final AtomicInteger iteration = new AtomicInteger(0); // do these need to be atomics?
private final AtomicLong runtime = new AtomicLong(0l);
private final Map<String, Accumulator<Rule>> memory = new HashMap<>();
+ private Broadcast<Map<String, Object>> broadcast;
private boolean inTask = false;
public SparkMemory(final VertexProgram<?> vertexProgram, final Set<MapReduce> mapReducers, final JavaSparkContext sparkContext) {
@@ -59,18 +61,23 @@ public final class SparkMemory implements Memory.Admin, Serializable {
this.memoryKeys.add(mapReduce.getMemoryKey());
}
for (final String key : this.memoryKeys) {
- this.memory.put(key, sparkContext.accumulator(new Rule(Rule.Operation.NO_OP, null), new RuleAccumulator()));
+ this.memory.put(key, sparkContext.accumulator(new Rule(Rule.Operation.NO_OP, null), key, new RuleAccumulator()));
}
+ this.broadcast = sparkContext.broadcast(new HashMap<>());
}
@Override
public Set<String> keys() {
- final Set<String> trueKeys = new HashSet<>();
- this.memory.forEach((key, value) -> {
- if (value.value().object != null)
- trueKeys.add(key);
- });
- return Collections.unmodifiableSet(trueKeys);
+ if (this.inTask)
+ return this.broadcast.getValue().keySet();
+ else {
+ final Set<String> trueKeys = new HashSet<>();
+ this.memory.forEach((key, value) -> {
+ if (value.value().object != null)
+ trueKeys.add(key);
+ });
+ return Collections.unmodifiableSet(trueKeys);
+ }
}
@Override
@@ -105,7 +112,7 @@ public final class SparkMemory implements Memory.Admin, Serializable {
@Override
public <R> R get(final String key) throws IllegalArgumentException {
- final R r = (R) (this.inTask ? this.memory.get(key).localValue() : this.memory.get(key).value()).object;
+ final R r = this.getValue(key);
if (null == r)
throw Memory.Exceptions.memoryDoesNotExist(key);
else
@@ -115,28 +122,43 @@ public final class SparkMemory implements Memory.Admin, Serializable {
@Override
public long incr(final String key, final long delta) {
checkKeyValue(key, delta);
- this.memory.get(key).add(new Rule(Rule.Operation.INCR, delta));
- return (Long) (this.inTask ? this.memory.get(key).localValue() : this.memory.get(key).value()).object + delta;
+ final Long returnValue = this.<Long>getValue(key) + delta;
+ if (this.inTask)
+ this.memory.get(key).add(new Rule(Rule.Operation.INCR, delta));
+ else
+ this.memory.get(key).setValue(new Rule(Rule.Operation.INCR, returnValue));
+ return returnValue;
}
@Override
public boolean and(final String key, final boolean bool) {
checkKeyValue(key, bool);
- this.memory.get(key).add(new Rule(Rule.Operation.AND, bool));
- return (Boolean) (this.inTask ? this.memory.get(key).localValue() : this.memory.get(key).value()).object && bool;
+ final Boolean returnValue = this.<Boolean>getValue(key) && bool;
+ if (this.inTask)
+ this.memory.get(key).add(new Rule(Rule.Operation.AND, bool));
+ else
+ this.memory.get(key).setValue(new Rule(Rule.Operation.AND, returnValue));
+ return returnValue;
}
@Override
public boolean or(final String key, final boolean bool) {
checkKeyValue(key, bool);
- this.memory.get(key).add(new Rule(Rule.Operation.OR, bool));
- return (Boolean) (this.inTask ? this.memory.get(key).localValue() : this.memory.get(key).value()).object || bool;
+ final Boolean returnValue = this.<Boolean>getValue(key) || bool;
+ if (this.inTask)
+ this.memory.get(key).add(new Rule(Rule.Operation.OR, bool));
+ else
+ this.memory.get(key).setValue(new Rule(Rule.Operation.OR, returnValue));
+ return returnValue;
}
@Override
public void set(final String key, final Object value) {
checkKeyValue(key, value);
- this.memory.get(key).setValue(new Rule(Rule.Operation.SET, value));
+ if (this.inTask)
+ this.memory.get(key).add(new Rule(Rule.Operation.SET, value));
+ else
+ this.memory.get(key).setValue(new Rule(Rule.Operation.SET, value));
}
@Override
@@ -144,13 +166,27 @@ public final class SparkMemory implements Memory.Admin, Serializable {
return StringFactory.memoryString(this);
}
- public void setInTask(final boolean inTask) {
+ protected void setInTask(final boolean inTask) {
this.inTask = inTask;
}
+ protected void broadcastMemory(final JavaSparkContext sparkContext) {
+ this.broadcast.destroy(true); // do we need to block?
+ final Map<String, Object> toBroadcast = new HashMap<>();
+ this.memory.forEach((key, rule) -> {
+ if (null != rule.value().object)
+ toBroadcast.put(key, rule.value().object);
+ });
+ this.broadcast = sparkContext.broadcast(toBroadcast);
+ }
+
private void checkKeyValue(final String key, final Object value) {
if (!this.memoryKeys.contains(key))
throw GraphComputer.Exceptions.providedKeyIsNotAMemoryComputeKey(key);
MemoryHelper.validateValue(value);
}
+
+ private <R> R getValue(final String key) {
+ return this.inTask ? (R) this.broadcast.getValue().get(key) : (R) this.memory.get(key).value().object;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/b4246a0e/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/Rule.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/Rule.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/Rule.java
index 1e9198b..d86f34e 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/Rule.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/Rule.java
@@ -56,7 +56,7 @@ public final class Rule implements Serializable {
}
}, SET {
public Object compute(final Object first, final Object second) {
- return second;
+ return null == first ? second : first;
}
}, NO_OP {
public Object compute(final Object first, final Object second) {
[3/3] incubator-tinkerpop git commit: Minor tweak to CHANGELOG.
Posted by ok...@apache.org.
Minor tweak to 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/a7f941ed
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/a7f941ed
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/a7f941ed
Branch: refs/heads/master
Commit: a7f941ed5fe8052771fe7ceb4e9015677d4e6e7c
Parents: 30326b5
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu Mar 5 15:46:10 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu Mar 5 15:47:35 2015 -0700
----------------------------------------------------------------------
CHANGELOG.asciidoc | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a7f941ed/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 2f0fe0e..f99718b 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,8 +26,8 @@ TinkerPop 3.0.0.M8 (Release Date: NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
* Added `compile` method to `GremlinExecutor` and related classes.
-* Changed the `Memory` API to not return the mutated value on `or`, `and`, `incr` as it is too difficult to implement faithfully in distributed systems.
-* Added `SparkGraphComputer` to Hadoop-Gremlin.
+* Changed the `Memory` API to not return the mutated value on `or`, `and`, `incr` as it is too difficult to implement faithfully in a distributed system.
+* Added `SparkGraphComputer` to Hadoop-Gremlin which uses Apache Spark as the underlying computing engine.
* 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.
[2/3] incubator-tinkerpop git commit: GraphComputer Memory no longer
returns the mutated value on worker set -- its too difficult to get right in
distributed systems.
Posted by ok...@apache.org.
GraphComputer Memory no longer returns the mutated value on worker set -- its too difficult to get right in distributed systems.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/30326b5b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/30326b5b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/30326b5b
Branch: refs/heads/master
Commit: 30326b5b4ba4c357383eb144283e6262fdc1ab4d
Parents: b4246a0
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu Mar 5 14:07:23 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu Mar 5 15:46:59 2015 -0700
----------------------------------------------------------------------
CHANGELOG.asciidoc | 2 +
.../gremlin/process/computer/Memory.java | 11 +-
.../process/computer/util/DefaultMemory.java | 109 -------------------
.../process/computer/util/ImmutableMemory.java | 6 +-
.../process/computer/util/MapMemory.java | 23 ++--
.../computer/GroovyGraphComputerTest.groovy | 10 +-
.../process/computer/GraphComputerTest.java | 12 +-
.../process/computer/giraph/GiraphMemory.java | 17 +--
.../computer/spark/SparkGraphComputer.java | 4 +-
.../process/computer/spark/SparkMemory.java | 18 +--
.../process/computer/TinkerMemory.java | 9 +-
11 files changed, 49 insertions(+), 172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 02a221f..2f0fe0e 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,8 @@ TinkerPop 3.0.0.M8 (Release Date: NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
* Added `compile` method to `GremlinExecutor` and related classes.
+* Changed the `Memory` API to not return the mutated value on `or`, `and`, `incr` as it is too difficult to implement faithfully in distributed systems.
+* Added `SparkGraphComputer` to Hadoop-Gremlin.
* 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.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/Memory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/Memory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/Memory.java
index fb0c977..72cba8c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/Memory.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/Memory.java
@@ -69,7 +69,7 @@ public interface Memory {
* @param key they key to set a value for
* @param value the value to set for the key
*/
- public void set(final String key, Object value);
+ public void set(final String key, final Object value);
/**
* A helper method that generates a {@link Map} of the memory key/values.
@@ -103,27 +103,24 @@ public interface Memory {
*
* @param key the key of the long value
* @param delta the adjusting amount (can be negative for decrement)
- * @return the adjusted value (according to the previous iterations current value + delta)
*/
- public long incr(final String key, final long delta);
+ public void incr(final String key, final long delta);
/**
* Logically AND the provided boolean value with the boolean value currently stored at the key.
*
* @param key the key of the boolean value
* @param bool the boolean to AND
- * @return the adjusted value (according to the previous iterations current value && bool)
*/
- public boolean and(final String key, final boolean bool);
+ public void and(final String key, final boolean bool);
/**
* Logically OR the provided boolean value with the boolean value currently stored at the key.
*
* @param key the key of the boolean value
* @param bool the boolean to OR
- * @return the adjusted value (according to the previous iterations current value || bool)
*/
- public boolean or(final String key, final boolean bool);
+ public void or(final String key, final boolean bool);
/**
* A helper method that states whether the current iteration is 0.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/DefaultMemory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/DefaultMemory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/DefaultMemory.java
deleted file mode 100644
index 9fee439..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/DefaultMemory.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.process.computer.util;
-
-import org.apache.tinkerpop.gremlin.process.computer.Memory;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class DefaultMemory implements Memory.Admin {
-
- private Map<String, Object> memory = new HashMap<>();
- private int iteration = 0;
- private long runtime = 0l;
-
- public DefaultMemory() {
-
- }
-
- public DefaultMemory(final Memory.Admin copyMemory) {
- this.iteration = copyMemory.getIteration();
- this.runtime = copyMemory.getRuntime();
- copyMemory.keys().forEach(key -> this.memory.put(key, copyMemory.get(key)));
- }
-
- @Override
- public void setIteration(final int iteration) {
- this.iteration = iteration;
- }
-
- @Override
- public void setRuntime(final long runtime) {
- this.runtime = runtime;
- }
-
- @Override
- public Set<String> keys() {
- return Collections.unmodifiableSet(this.memory.keySet());
- }
-
- @Override
- public <R> R get(final String key) throws IllegalArgumentException {
- final R r = (R) this.memory.get(key);
- if (null == r)
- throw Memory.Exceptions.memoryDoesNotExist(key);
- else
- return r;
- }
-
- @Override
- public void set(final String key, final Object value) {
- this.memory.put(key, value);
- }
-
- @Override
- public int getIteration() {
- return this.iteration;
- }
-
- @Override
- public long getRuntime() {
- return this.runtime;
- }
-
- @Override
- public long incr(final String key, final long delta) {
- final Long value = (Long) this.memory.get(key);
- final Long newValue = (null == value) ? delta : delta + value;
- this.memory.put(key, newValue);
- return newValue;
- }
-
- @Override
- public boolean and(final String key, final boolean bool) {
- final Boolean value = (Boolean) this.memory.get(key);
- final Boolean newValue = (null == value) ? bool : bool && value;
- this.memory.put(key, newValue);
- return newValue;
- }
-
- @Override
- public boolean or(final String key, final boolean bool) {
- final Boolean value = (Boolean) this.memory.get(key);
- final Boolean newValue = (null == value) ? bool : bool || value;
- this.memory.put(key, newValue);
- return newValue;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ImmutableMemory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ImmutableMemory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ImmutableMemory.java
index c04d231..02fa051 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ImmutableMemory.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/ImmutableMemory.java
@@ -60,17 +60,17 @@ public final class ImmutableMemory implements Memory.Admin {
}
@Override
- public long incr(final String key, final long delta) {
+ public void incr(final String key, final long delta) {
throw Memory.Exceptions.memoryIsCurrentlyImmutable();
}
@Override
- public boolean and(final String key, final boolean bool) {
+ public void and(final String key, final boolean bool) {
throw Memory.Exceptions.memoryIsCurrentlyImmutable();
}
@Override
- public boolean or(final String key, final boolean bool) {
+ public void or(final String key, final boolean bool) {
throw Memory.Exceptions.memoryIsCurrentlyImmutable();
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/MapMemory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/MapMemory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/MapMemory.java
index d9a0c06..71bd185 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/MapMemory.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/MapMemory.java
@@ -40,6 +40,17 @@ public final class MapMemory implements Memory.Admin, Serializable {
private final Map<String, Object> memoryMap = new HashMap<>();
private final Set<String> memoryComputeKeys = new HashSet<>();
+ public MapMemory() {
+
+ }
+
+ public MapMemory(final Memory otherMemory) {
+ otherMemory.keys().forEach(key -> {
+ this.memoryMap.put(key, otherMemory.get(key));
+ });
+ this.iteration = otherMemory.getIteration();
+ }
+
public void addVertexProgramMemoryComputeKeys(final VertexProgram<?> vertexProgram) {
this.memoryComputeKeys.addAll(vertexProgram.getMemoryComputeKeys());
}
@@ -78,41 +89,35 @@ public final class MapMemory implements Memory.Admin, Serializable {
}
@Override
- public long incr(final String key, final long delta) {
+ public void incr(final String key, final long delta) {
this.checkKeyValue(key, delta);
if (this.memoryMap.containsKey(key)) {
final long newValue = (long) this.memoryMap.get(key) + delta;
this.memoryMap.put(key, newValue);
- return newValue;
} else {
this.memoryMap.put(key, delta);
- return delta;
}
}
@Override
- public boolean and(final String key, final boolean bool) {
+ public void and(final String key, final boolean bool) {
this.checkKeyValue(key, bool);
if (this.memoryMap.containsKey(key)) {
final boolean newValue = (boolean) this.memoryMap.get(key) && bool;
this.memoryMap.put(key, newValue);
- return newValue;
} else {
this.memoryMap.put(key, bool);
- return bool;
}
}
@Override
- public boolean or(final String key, final boolean bool) {
+ public void or(final String key, final boolean bool) {
this.checkKeyValue(key, bool);
if (this.memoryMap.containsKey(key)) {
final boolean newValue = (boolean) this.memoryMap.get(key) || bool;
this.memoryMap.put(key, newValue);
- return newValue;
} else {
this.memoryMap.put(key, bool);
- return bool;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/computer/GroovyGraphComputerTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/computer/GroovyGraphComputerTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/computer/GroovyGraphComputerTest.groovy
index daa04b6..eaf8bd9 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/computer/GroovyGraphComputerTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/computer/GroovyGraphComputerTest.groovy
@@ -108,11 +108,11 @@ public abstract class GroovyGraphComputerTest {
assertTrue(c.get("e"));
// update current step values and make sure returns are correct
- assertEquals(Long.valueOf(6 * c.getIteration()) + 1l, c.incr("a", 1l));
- assertEquals(Long.valueOf(0) + 1l, c.incr("b", 1l));
- assertFalse(c.and("c", false));
- assertTrue(c.or("d", true));
- assertFalse(c.and("e", false));
+ c.incr("a", 1l);
+ c.incr("b", 1l);
+ c.and("c", false);
+ c.or("d", true);
+ c.and("e", false);
// test current step values, should be the same as previous prior to update
assertEquals(Long.valueOf(6 * c.getIteration()), c.get("a"));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
index 1316ceb..ea5e6c1 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
@@ -403,12 +403,12 @@ public abstract class GraphComputerTest extends AbstractGremlinProcessTest {
}
assertTrue(memory.get("e"));
- // update current step values and make sure returns are correct
- assertEquals(Long.valueOf(6 * memory.getIteration()) + 1l, memory.incr("a", 1l));
- assertEquals(Long.valueOf(0) + 1l, memory.incr("b", 1l));
- assertFalse(memory.and("c", false));
- assertTrue(memory.or("d", true));
- assertFalse(memory.and("e", false));
+ // update current step values
+ memory.incr("a", 1l);
+ memory.incr("b", 1l);
+ memory.and("c", false);
+ memory.or("d", true);
+ memory.and("e", false);
// test current step values, should be the same as previous prior to update
assertEquals(Long.valueOf(6 * memory.getIteration()), memory.get("a"));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
index a2b1709..9af0cd0 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
+import org.apache.giraph.master.MasterCompute;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
@@ -25,7 +26,6 @@ import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.MemoryHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.giraph.master.MasterCompute;
import java.io.DataInput;
import java.io.DataOutput;
@@ -137,47 +137,38 @@ public class GiraphMemory extends MasterCompute implements Memory {
}
@Override
- public boolean and(final String key, final boolean bool) {
+ public void and(final String key, final boolean bool) {
this.checkKeyValue(key, bool);
if (this.isMasterCompute) { // only called on setup() and terminate()
Boolean value = this.<RuleWritable>getAggregatedValue(key).<Boolean>getObject();
value = null == value ? bool : bool && value;
this.setAggregatedValue(key, new RuleWritable(RuleWritable.Rule.AND, value));
- return value;
} else {
- final Boolean result = ((RuleWritable) this.worker.getAggregatedValue(key)).getObject();
this.worker.aggregate(key, new RuleWritable(RuleWritable.Rule.AND, bool));
- return null == result ? bool : result && bool;
}
}
@Override
- public boolean or(final String key, final boolean bool) {
+ public void or(final String key, final boolean bool) {
this.checkKeyValue(key, bool);
if (this.isMasterCompute) { // only called on setup() and terminate()
Boolean value = this.<RuleWritable>getAggregatedValue(key).<Boolean>getObject();
value = null == value ? bool : bool || value;
this.setAggregatedValue(key, new RuleWritable(RuleWritable.Rule.OR, value));
- return value;
} else {
- final Boolean result = ((RuleWritable) this.worker.getAggregatedValue(key)).getObject();
this.worker.aggregate(key, new RuleWritable(RuleWritable.Rule.OR, bool));
- return null == result ? bool : result || bool;
}
}
@Override
- public long incr(final String key, final long delta) {
+ public void incr(final String key, final long delta) {
this.checkKeyValue(key, delta);
if (this.isMasterCompute) { // only called on setup() and terminate()
Number value = this.<RuleWritable>getAggregatedValue(key).<Number>getObject();
value = null == value ? delta : value.longValue() + delta;
this.setAggregatedValue(key, new RuleWritable(RuleWritable.Rule.INCR, value));
- return value.longValue();
} else {
- final Long result = ((RuleWritable) this.worker.getAggregatedValue(key)).getObject();
this.worker.aggregate(key, new RuleWritable(RuleWritable.Rule.INCR, delta));
- return null == result ? delta : result + delta;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
index b061304..a8ad2f4 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
@@ -41,8 +41,8 @@ import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
-import org.apache.tinkerpop.gremlin.process.computer.util.DefaultMemory;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
+import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
import org.slf4j.Logger;
@@ -174,7 +174,7 @@ public final class SparkGraphComputer implements GraphComputer {
//////////////////////////////
// process the map reducers //
//////////////////////////////
- final Memory.Admin finalMemory = null == memory ? new DefaultMemory() : new DefaultMemory(memory);
+ final Memory.Admin finalMemory = null == memory ? new MapMemory() : new MapMemory(memory);
for (final MapReduce mapReduce : this.mapReducers) {
// execute the map reduce job
final SApacheConfiguration newApacheConfiguration = new SApacheConfiguration(apacheConfiguration);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
index 3ef69f4..c25d25e 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkMemory.java
@@ -120,36 +120,30 @@ public final class SparkMemory implements Memory.Admin, Serializable {
}
@Override
- public long incr(final String key, final long delta) {
+ public void incr(final String key, final long delta) {
checkKeyValue(key, delta);
- final Long returnValue = this.<Long>getValue(key) + delta;
if (this.inTask)
this.memory.get(key).add(new Rule(Rule.Operation.INCR, delta));
else
- this.memory.get(key).setValue(new Rule(Rule.Operation.INCR, returnValue));
- return returnValue;
+ this.memory.get(key).setValue(new Rule(Rule.Operation.INCR, this.<Long>getValue(key) + delta));
}
@Override
- public boolean and(final String key, final boolean bool) {
+ public void and(final String key, final boolean bool) {
checkKeyValue(key, bool);
- final Boolean returnValue = this.<Boolean>getValue(key) && bool;
if (this.inTask)
this.memory.get(key).add(new Rule(Rule.Operation.AND, bool));
else
- this.memory.get(key).setValue(new Rule(Rule.Operation.AND, returnValue));
- return returnValue;
+ this.memory.get(key).setValue(new Rule(Rule.Operation.AND, this.<Boolean>getValue(key) && bool));
}
@Override
- public boolean or(final String key, final boolean bool) {
+ public void or(final String key, final boolean bool) {
checkKeyValue(key, bool);
- final Boolean returnValue = this.<Boolean>getValue(key) || bool;
if (this.inTask)
this.memory.get(key).add(new Rule(Rule.Operation.OR, bool));
else
- this.memory.get(key).setValue(new Rule(Rule.Operation.OR, returnValue));
- return returnValue;
+ this.memory.get(key).setValue(new Rule(Rule.Operation.OR, this.<Boolean>getValue(key) || bool));
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/30326b5b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMemory.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMemory.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMemory.java
index dfe2d41..6e90b75 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMemory.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerMemory.java
@@ -113,24 +113,21 @@ public class TinkerMemory implements Memory.Admin {
}
@Override
- public long incr(final String key, final long delta) {
+ public void incr(final String key, final long delta) {
checkKeyValue(key, delta);
this.currentMap.compute(key, (k, v) -> null == v ? delta : delta + (Long) v);
- return (Long) this.previousMap.getOrDefault(key, 0l) + delta;
}
@Override
- public boolean and(final String key, final boolean bool) {
+ public void and(final String key, final boolean bool) {
checkKeyValue(key, bool);
this.currentMap.compute(key, (k, v) -> null == v ? bool : bool && (Boolean) v);
- return (Boolean) this.previousMap.getOrDefault(key, true) && bool;
}
@Override
- public boolean or(final String key, final boolean bool) {
+ public void or(final String key, final boolean bool) {
checkKeyValue(key, bool);
this.currentMap.compute(key, (k, v) -> null == v ? bool : bool || (Boolean) v);
- return (Boolean) this.previousMap.getOrDefault(key, true) || bool;
}
@Override