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/09/30 17:03:41 UTC
incubator-tinkerpop git commit: GiraphGraphComputer supports load
balancing TinkerPop workers across a Hadoop cluster and across threads on the
machines of that cluster. Thus,
mappers * threads = TinkerPop workers. Extended GraphTest.shouldSupportWorkerC
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 53b712b8c -> 6cbff01b7
GiraphGraphComputer supports load balancing TinkerPop workers across a Hadoop cluster and across threads on the machines of that cluster. Thus, mappers * threads = TinkerPop workers. Extended GraphTest.shouldSupportWorkerCount() to ensure that the number of workers spawned is consistent and that there is no thread leaks between VertexPrograms in workers. Every worker has its own VertexProgram (this was a nasty bug in TinkerGraph that was fixed last night and further verified in the updated test suite). OPT_OUT computers() returns List<String> now as opposed to List<Class> given that we can gurantee that the computer classes will be with the engine (i.e. spark is split from hadoop now).
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/6cbff01b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/6cbff01b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/6cbff01b
Branch: refs/heads/master
Commit: 6cbff01b7d9a93afef279f2069e2e58dac0a07c5
Parents: 53b712b
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Sep 30 09:03:13 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed Sep 30 09:03:34 2015 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 3 +
.../process/computer/GiraphGraphComputer.java | 47 ++++++++--
.../computer/HadoopGiraphGraphProvider.java | 2 +
.../gremlin/process/computer/GraphComputer.java | 8 +-
.../tinkerpop/gremlin/structure/Graph.java | 4 +-
.../tinkerpop/gremlin/AbstractGremlinSuite.java | 10 ++-
.../process/computer/GraphComputerTest.java | 29 ++++---
.../computer/AbstractHadoopGraphComputer.java | 90 +++++++++++---------
.../gremlin/hadoop/structure/HadoopGraph.java | 3 +-
.../process/computer/TinkerGraphComputer.java | 16 ++++
10 files changed, 150 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index a1ff11a..0510c74 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,9 @@ image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/
TinkerPop 3.1.0 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Fixed a severe threading issue in `TinkerGraphComputer`.
+* Fixed a `clone()` bug in `RepeatStep` and `TraversalRing`.
+* `GiraphGraphComputer.workers()` is smart about using threads and machines to load balance TinkerPop workers across cluster.
* `GraphComputer.workers(int)` allows the user to programmatically set the number of workers to spawn.
* Added `GryoSerializer` as the new recommended Spark `Serializer`. Handles `Graph` and `GryoMapper` registries.
* `GryoPool` now makes use of `GryoPool.Builder` for its construction.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
index 7c11b69..ec200b7d 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
@@ -52,6 +52,7 @@ import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import java.io.File;
+import java.io.IOException;
import java.io.NotSerializableException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
@@ -79,6 +80,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
this.giraphConfiguration.setBoolean(GiraphConstants.STATIC_GRAPH.getKey(), true);
this.giraphConfiguration.setVertexInputFormatClass(GiraphVertexInputFormat.class);
this.giraphConfiguration.setVertexOutputFormatClass(GiraphVertexOutputFormat.class);
+ this.workers(this.giraphConfiguration.getNumComputeThreads() * (this.giraphConfiguration.getMaxWorkers() < 1 ? 1 : this.giraphConfiguration.getMaxWorkers()));
}
@Override
@@ -93,12 +95,6 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
}
@Override
- public GraphComputer workers(final int workers) {
- this.giraphConfiguration.setWorkerConfiguration(this.workers, this.workers, 100.0F);
- return super.workers(workers);
- }
-
- @Override
public Future<ComputerResult> submit() {
final long startTime = System.currentTimeMillis();
super.validateStatePriorToExecution();
@@ -109,6 +105,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
fs.delete(new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)), true);
ToolRunner.run(this, new String[]{});
} catch (final Exception e) {
+ System.out.println(this.giraphConfiguration.getMaxWorkers() + "$%$%$");
//e.printStackTrace();
throw new IllegalStateException(e.getMessage(), e);
}
@@ -133,6 +130,21 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
}
// prepare the giraph vertex-centric computing job
final GiraphJob job = new GiraphJob(this.giraphConfiguration, Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
+ // split required workers across system (open map slots + max threads per machine = total amount of TinkerPop workers)
+ if (this.giraphConfiguration.getLocalTestMode()) {
+ this.giraphConfiguration.setWorkerConfiguration(1, 1, 100.0F);
+ this.giraphConfiguration.setNumComputeThreads(this.workers);
+ } else {
+ int totalMappers = job.getInternalJob().getCluster().getClusterStatus().getMapSlotCapacity() - 1; // 1 is needed for master
+ if (this.workers <= totalMappers) {
+ this.giraphConfiguration.setWorkerConfiguration(this.workers, this.workers, 100.0F);
+ this.giraphConfiguration.setNumComputeThreads(1);
+ } else {
+ int threadsPerMapper = Long.valueOf(Math.round((double) this.workers / (double) totalMappers)).intValue(); // TODO: need to find least common denominator
+ this.giraphConfiguration.setWorkerConfiguration(totalMappers, totalMappers, 100.0F);
+ this.giraphConfiguration.setNumComputeThreads(threadsPerMapper);
+ }
+ }
// handle input paths (if any)
if (FileInputFormat.class.isAssignableFrom(this.giraphConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputFormat.class))) {
final Path inputPath = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
@@ -226,4 +238,27 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
final FileConfiguration configuration = new PropertiesConfiguration(args[0]);
new GiraphGraphComputer(HadoopGraph.open(configuration)).program(VertexProgram.createVertexProgram(HadoopGraph.open(configuration), configuration)).submit().get();
}
+
+ public Features features() {
+ return new Features();
+ }
+
+ public class Features extends AbstractHadoopGraphComputer.Features {
+
+ @Override
+ public int getMaxWorkers() {
+ if (GiraphGraphComputer.this.giraphConfiguration.getLocalTestMode())
+ return Runtime.getRuntime().availableProcessors();
+ else {
+ try {
+ final GiraphJob job = new GiraphJob(GiraphGraphComputer.this.giraphConfiguration, "GiraphGraphComputer.Features.getMaxWorkers()");
+ int maxWorkers = job.getInternalJob().getCluster().getClusterStatus().getMapSlotCapacity() * 32; // max 32 threads per machine hardcoded :|
+ job.getInternalJob().killJob();
+ return maxWorkers;
+ } catch (final IOException | InterruptedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
index 86007f8..ce1edf7 100644
--- a/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
+++ b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
@@ -115,12 +115,14 @@ public final class HadoopGiraphGraphProvider extends AbstractGraphProvider {
put(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
put("mapreduce.job.reduces", 4);
/// giraph configuration
+ put(GiraphConstants.LOCAL_TEST_MODE.getKey(), true); // local testing can only spawn one worker
put(GiraphConstants.MIN_WORKERS, 1);
put(GiraphConstants.MAX_WORKERS, 1);
put(GiraphConstants.SPLIT_MASTER_WORKER.getKey(), false);
put(GiraphConstants.ZOOKEEPER_SERVER_PORT.getKey(), 2181); // you must have a local zookeeper running on this port
put(GiraphConstants.NETTY_SERVER_USE_EXECUTION_HANDLER.getKey(), false); // this prevents so many integration tests running out of threads
put(GiraphConstants.NETTY_CLIENT_USE_EXECUTION_HANDLER.getKey(), false); // this prevents so many integration tests running out of threads
+ put(GiraphConstants.NETTY_USE_DIRECT_MEMORY.getKey(),true);
put(GiraphConstants.NUM_INPUT_THREADS.getKey(), 3);
put(GiraphConstants.NUM_COMPUTE_THREADS.getKey(), 3);
put(GiraphConstants.MAX_MASTER_SUPERSTEP_WAIT_MSECS.getKey(), TimeUnit.MINUTES.toMillis(60L));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
index 18d4f74..0818ed8 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
@@ -117,8 +117,8 @@ public interface GraphComputer {
public interface Features {
- public default boolean supportsWorkerCount(int workers) {
- return workers != 0;
+ public default int getMaxWorkers() {
+ return Integer.MAX_VALUE;
}
public default boolean supportsGlobalMessageScopes() {
@@ -218,6 +218,10 @@ public interface GraphComputer {
public static UnsupportedOperationException vertexPropertiesCanNotBeUpdatedInMapReduce() {
return new UnsupportedOperationException("The computer is in MapReduce mode and a vertex's properties can not be updated");
}
+
+ public static IllegalArgumentException computerRequiresMoreWorkersThanSupported(final int workers, final int maxWorkers) {
+ return new IllegalArgumentException("The computer requires more workers than supported: " + workers + " [max:" + maxWorkers + "]");
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
index ddd5d2e..e901141 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
@@ -1190,13 +1190,13 @@ public interface Graph extends AutoCloseable, Host {
public String specific() default "";
/**
- * The list of {@link GraphComputer} implementations that a test should opt-out from using (i.e. other
+ * The list of {@link GraphComputer} implementations by class name that a test should opt-out from using (i.e. other
* graph computers not in this list will execute the test). This setting should only be included when
* the test is one that uses the {@link ComputerTraversalEngine} - it will otherwise be ignored. By
* default, an empty array is assigned and it is thus assumed that all computers are excluded when an
* {@code OptOut} annotation is used, therefore this value must be overridden to be more specific.
*/
- public Class<? extends GraphComputer>[] computers() default { };
+ public String[] computers() default { };
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/AbstractGremlinSuite.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/AbstractGremlinSuite.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/AbstractGremlinSuite.java
index 186d4c8..5f8fd4f 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/AbstractGremlinSuite.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/AbstractGremlinSuite.java
@@ -333,11 +333,15 @@ public abstract class AbstractGremlinSuite extends Suite {
|| optOut.computers().length == 0) {
return true;
}
-
// can assume that that GraphProvider.Descriptor is not null at this point. a test should
// only opt out if it matches the expected computer
- final boolean x = Stream.of(optOut.computers()).anyMatch(c -> c == graphProviderDescriptor.get().computer());
- return Stream.of(optOut.computers()).anyMatch(c -> c == graphProviderDescriptor.get().computer());
+ return Stream.of(optOut.computers()).map(c -> {
+ try {
+ return Class.forName(c);
+ } catch (ClassNotFoundException e) {
+ return Object.class;
+ }
+ }).filter(c -> c.equals(Object.class)).anyMatch(c -> c == graphProviderDescriptor.get().computer());
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/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 b859f4b..b546717 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
@@ -60,7 +60,8 @@ import static org.junit.Assert.*;
"adjacentVertexPropertiesCanNotBeReadOrUpdated",
"adjacentVertexEdgesAndVerticesCanNotBeReadOrUpdated",
"resultGraphPersistCombinationNotSupported",
- "vertexPropertiesCanNotBeUpdatedInMapReduce"
+ "vertexPropertiesCanNotBeUpdatedInMapReduce",
+ "computerRequiresMoreWorkersThanSupported"
})
@ExceptionCoverage(exceptionClass = Graph.Exceptions.class, methods = {
"graphDoesNotSupportProvidedGraphComputer"
@@ -1371,19 +1372,27 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(GRATEFUL)
public void shouldSupportWorkerCount() throws Exception {
- assertFalse(graph.compute(graphComputerClass.get()).features().supportsWorkerCount(0));
- for (int i = 0; i < 10; i++) { // the GraphComputer should not support 0 workers
- final GraphComputer computer = graph.compute(graphComputerClass.get());
- if (computer.features().supportsWorkerCount(i)) {
- ComputerResult result = computer.program(new VertexProgramL()).workers(i).submit().get();
- assertEquals(Integer.valueOf(i).longValue(), (long) result.memory().get("workerCount"));
+ int maxWorkers = graph.compute(graphComputerClass.get()).features().getMaxWorkers();
+ if (maxWorkers != Integer.MAX_VALUE) {
+ for (int i = maxWorkers + 1; i < maxWorkers + 10; i++) {
+ try {
+ graph.compute(graphComputerClass.get()).program(new VertexProgramL()).workers(i).submit().get();
+ fail("Should throw a GraphComputer.Exceptions.computerRequiresMoreWorkersThanSupported() exception");
+ } catch (final IllegalArgumentException e) {
+ assertTrue(e.getMessage().contains("computer requires more workers"));
+ }
}
}
+ if (maxWorkers > 25) maxWorkers = 25;
+ for (int i = 1; i <= maxWorkers; i++) {
+ ComputerResult result = graph.compute(graphComputerClass.get()).program(new VertexProgramL()).workers(i).submit().get();
+ assertEquals(Integer.valueOf(i).longValue(), (long) result.memory().get("workerCount"));
+ }
}
public static class VertexProgramL implements VertexProgram {
- final Set<String> threadIds = new HashSet<>();
+ boolean announced = false;
@Override
public void setup(final Memory memory) {
@@ -1397,9 +1406,9 @@ public class GraphComputerTest extends AbstractGremlinProcessTest {
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
- if (!this.threadIds.contains(Thread.currentThread().getName())) {
+ if (!this.announced) {
memory.incr("workerCount", 1l);
- this.threadIds.add(Thread.currentThread().getName());
+ this.announced = true;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
index ca2a931..1553240 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/AbstractHadoopGraphComputer.java
@@ -47,7 +47,7 @@ public abstract class AbstractHadoopGraphComputer implements GraphComputer {
protected boolean executed = false;
protected final Set<MapReduce> mapReducers = new HashSet<>();
protected VertexProgram<Object> vertexProgram;
- protected int workers = -1;
+ protected int workers = 1;
protected ResultGraph resultGraph = null;
protected Persist persist = null;
@@ -112,58 +112,72 @@ public abstract class AbstractHadoopGraphComputer implements GraphComputer {
// determine persistence and result graph options
if (!this.features().supportsResultGraphPersistCombination(this.resultGraph, this.persist))
throw GraphComputer.Exceptions.resultGraphPersistCombinationNotSupported(this.resultGraph, this.persist);
+ // if too many workers are requested, throw appropriate exception
+ if (this.workers > this.features().getMaxWorkers())
+ throw GraphComputer.Exceptions.computerRequiresMoreWorkersThanSupported(this.workers, this.features().getMaxWorkers());
}
@Override
public Features features() {
- return new Features() {
+ return new Features();
+ }
- public boolean supportsVertexAddition() {
- return false;
- }
+ public class Features implements GraphComputer.Features {
- public boolean supportsVertexRemoval() {
- return false;
- }
+ @Override
+ public boolean supportsVertexAddition() {
+ return false;
+ }
- public boolean supportsVertexPropertyRemoval() {
- return false;
- }
+ @Override
+ public boolean supportsVertexRemoval() {
+ return false;
+ }
- public boolean supportsEdgeAddition() {
- return false;
- }
+ @Override
+ public boolean supportsVertexPropertyRemoval() {
+ return false;
+ }
- public boolean supportsEdgeRemoval() {
- return false;
- }
+ @Override
+ public boolean supportsEdgeAddition() {
+ return false;
+ }
- public boolean supportsEdgePropertyAddition() {
- return false;
- }
+ @Override
+ public boolean supportsEdgeRemoval() {
+ return false;
+ }
- public boolean supportsEdgePropertyRemoval() {
- return false;
- }
+ @Override
+ public boolean supportsEdgePropertyAddition() {
+ return false;
+ }
- public boolean supportsResultGraphPersistCombination(final ResultGraph resultGraph, final Persist persist) {
- if (hadoopGraph.configuration().containsKey(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT)) {
- final OutputFormat<NullWritable, VertexWritable> outputFormat = ReflectionUtils.newInstance(hadoopGraph.configuration().getGraphOutputFormat(), ConfUtil.makeHadoopConfiguration(hadoopGraph.configuration()));
- if (outputFormat instanceof PersistResultGraphAware)
- return ((PersistResultGraphAware) outputFormat).supportsResultGraphPersistCombination(resultGraph, persist);
- else {
- logger.warn(outputFormat.getClass() + " does not implement " + PersistResultGraphAware.class.getSimpleName() + " and thus, persistence options are unknown -- assuming all options are possible");
- return true;
- }
- } else {
- logger.warn("Unknown OutputFormat class and thus, persistence options are unknown -- assuming all options are possible");
+ @Override
+ public boolean supportsEdgePropertyRemoval() {
+ return false;
+ }
+
+ @Override
+ public boolean supportsResultGraphPersistCombination(final ResultGraph resultGraph, final Persist persist) {
+ if (hadoopGraph.configuration().containsKey(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT)) {
+ final OutputFormat<NullWritable, VertexWritable> outputFormat = ReflectionUtils.newInstance(hadoopGraph.configuration().getGraphOutputFormat(), ConfUtil.makeHadoopConfiguration(hadoopGraph.configuration()));
+ if (outputFormat instanceof PersistResultGraphAware)
+ return ((PersistResultGraphAware) outputFormat).supportsResultGraphPersistCombination(resultGraph, persist);
+ else {
+ logger.warn(outputFormat.getClass() + " does not implement " + PersistResultGraphAware.class.getSimpleName() + " and thus, persistence options are unknown -- assuming all options are possible");
return true;
}
+ } else {
+ logger.warn("Unknown OutputFormat class and thus, persistence options are unknown -- assuming all options are possible");
+ return true;
}
+ }
- public boolean supportsDirectObjects() {
- return false;
- }
- };
+ @Override
+ public boolean supportsDirectObjects() {
+ return false;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
index 7927a15..641dc0f 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopGraph.java
@@ -156,7 +156,8 @@ import java.util.stream.Stream;
@Graph.OptOut(
test = "org.apache.tinkerpop.gremlin.process.computer.GraphComputerTest",
method = "shouldSupportWorkerCount",
- reason = "It is not possible to control Spark and Giraph worker counts in integration testing.")
+ reason = "It is not possible to control Spark worker counts in integration testing.",
+ computers = {"org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer"})
public final class HadoopGraph implements Graph {
public static final Logger LOGGER = LoggerFactory.getLogger(HadoopGraph.class);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/6cbff01b/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 c88fb4c..07ad0c8 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
@@ -111,6 +111,10 @@ public final class TinkerGraphComputer implements GraphComputer {
this.persist = GraphComputerHelper.getPersistState(Optional.ofNullable(this.vertexProgram), Optional.ofNullable(this.persist));
if (!this.features().supportsResultGraphPersistCombination(this.resultGraph, this.persist))
throw GraphComputer.Exceptions.resultGraphPersistCombinationNotSupported(this.resultGraph, this.persist);
+ // ensure requested workers are not larger than supported workers
+ if (this.workers > this.features().getMaxWorkers())
+ throw GraphComputer.Exceptions.computerRequiresMoreWorkersThanSupported(this.workers, this.features().getMaxWorkers());
+
// initialize the memory
this.memory = new TinkerMemory(this.vertexProgram, this.mapReducers);
@@ -226,30 +230,42 @@ public final class TinkerGraphComputer implements GraphComputer {
public Features features() {
return new Features() {
+ @Override
+ public int getMaxWorkers() {
+ return Runtime.getRuntime().availableProcessors();
+ }
+
+ @Override
public boolean supportsVertexAddition() {
return false;
}
+ @Override
public boolean supportsVertexRemoval() {
return false;
}
+ @Override
public boolean supportsVertexPropertyRemoval() {
return false;
}
+ @Override
public boolean supportsEdgeAddition() {
return false;
}
+ @Override
public boolean supportsEdgeRemoval() {
return false;
}
+ @Override
public boolean supportsEdgePropertyAddition() {
return false;
}
+ @Override
public boolean supportsEdgePropertyRemoval() {
return false;
}