You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ni...@apache.org on 2013/01/04 21:52:39 UTC
[2/23] GIRAPH-409: Refactor / cleanups (nitay)
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/graph/partition/TestPartitionStores.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/graph/partition/TestPartitionStores.java b/giraph-core/src/test/java/org/apache/giraph/graph/partition/TestPartitionStores.java
deleted file mode 100644
index 098899b..0000000
--- a/giraph-core/src/test/java/org/apache/giraph/graph/partition/TestPartitionStores.java
+++ /dev/null
@@ -1,211 +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.giraph.graph.partition;
-
-import org.apache.giraph.conf.GiraphConfiguration;
-import org.apache.giraph.conf.GiraphConstants;
-import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
-import org.apache.giraph.graph.IntIntNullIntVertex;
-import org.apache.giraph.graph.Vertex;
-import org.apache.giraph.utils.UnsafeByteArrayInputStream;
-import org.apache.giraph.utils.UnsafeByteArrayOutputStream;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.junit.Before;
-import org.junit.Test;
-
-import com.google.common.collect.Iterables;
-
-import java.io.IOException;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-
-/**
- * Test case for partition stores.
- */
-public class TestPartitionStores {
- private ImmutableClassesGiraphConfiguration conf;
- private Mapper<?, ?, ?, ?>.Context context;
-
- public static class MyVertex extends IntIntNullIntVertex {
- @Override
- public void compute(Iterable<IntWritable> messages) throws IOException {}
- }
-
- private Partition<IntWritable, IntWritable, NullWritable,
- IntWritable> createPartition(ImmutableClassesGiraphConfiguration conf,
- Integer id,
- Vertex<IntWritable, IntWritable,
- NullWritable, IntWritable>... vertices) {
- Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition =
- conf.createPartition(id, context);
- for (Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v :
- vertices) {
- partition.putVertex(v);
- }
- return partition;
- }
-
- @Before
- public void setUp() {
- GiraphConfiguration configuration = new GiraphConfiguration();
- configuration.setVertexClass(MyVertex.class);
- conf = new ImmutableClassesGiraphConfiguration(configuration);
- context = mock(Mapper.Context.class);
- }
-
- @Test
- public void testSimplePartitionStore() {
- PartitionStore<IntWritable, IntWritable, NullWritable, IntWritable>
- partitionStore = new SimplePartitionStore<IntWritable, IntWritable,
- NullWritable, IntWritable>(conf, context);
- testReadWrite(partitionStore, conf);
- }
-
- @Test
- public void testUnsafePartitionSerializationClass() throws IOException {
- conf.setPartitionClass(ByteArrayPartition.class);
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v1 =
- new MyVertex();
- v1.initialize(new IntWritable(1), new IntWritable(1));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v2 =
- new MyVertex();
- v2.initialize(new IntWritable(2), new IntWritable(2));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v3 =
- new MyVertex();
- v3.initialize(new IntWritable(3), new IntWritable(3));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v4 =
- new MyVertex();
- v4.initialize(new IntWritable(4), new IntWritable(4));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v5 =
- new MyVertex();
- v5.initialize(new IntWritable(5), new IntWritable(5));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v6 =
- new MyVertex();
- v6.initialize(new IntWritable(6), new IntWritable(6));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v7 =
- new MyVertex();
- v7.initialize(new IntWritable(7), new IntWritable(7));
-
- Partition<IntWritable, IntWritable, NullWritable,
- IntWritable> partition =
- createPartition(conf, 3, v1, v2, v3, v4, v5, v6, v7);
- assertEquals(3, partition.getId());
- assertEquals(0, partition.getEdgeCount());
- assertEquals(7, partition.getVertexCount());
- UnsafeByteArrayOutputStream outputStream = new
- UnsafeByteArrayOutputStream();
- partition.write(outputStream);
- UnsafeByteArrayInputStream inputStream = new UnsafeByteArrayInputStream(
- outputStream.getByteArray(), 0, outputStream.getPos());
- Partition<IntWritable, IntWritable, NullWritable,
- IntWritable> deserializatedPartition = conf.createPartition(-1,
- context);
- deserializatedPartition.readFields(inputStream);
-
- assertEquals(3, deserializatedPartition.getId());
- assertEquals(0, deserializatedPartition.getEdgeCount());
- assertEquals(7, deserializatedPartition.getVertexCount());
- }
-
- @Test
- public void testDiskBackedPartitionStore() {
- conf.setBoolean(GiraphConstants.USE_OUT_OF_CORE_GRAPH, true);
- conf.setInt(GiraphConstants.MAX_PARTITIONS_IN_MEMORY, 1);
-
- PartitionStore<IntWritable, IntWritable, NullWritable, IntWritable>
- partitionStore = new DiskBackedPartitionStore<IntWritable,
- IntWritable, NullWritable, IntWritable>(conf, context);
- testReadWrite(partitionStore, conf);
-
- conf.setInt(GiraphConstants.MAX_PARTITIONS_IN_MEMORY, 2);
- partitionStore = new DiskBackedPartitionStore<IntWritable,
- IntWritable, NullWritable, IntWritable>(conf, context);
- testReadWrite(partitionStore, conf);
- }
-
- /**
- * Test reading/writing to/from a partition store
- *
- * @param partitionStore Partition store to test
- * @param conf Configuration to use
- */
- public void testReadWrite(
- PartitionStore<IntWritable, IntWritable,
- NullWritable, IntWritable> partitionStore,
- ImmutableClassesGiraphConfiguration conf) {
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v1 =
- new MyVertex();
- v1.initialize(new IntWritable(1), new IntWritable(1));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v2 =
- new MyVertex();
- v2.initialize(new IntWritable(2), new IntWritable(2));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v3 =
- new MyVertex();
- v3.initialize(new IntWritable(3), new IntWritable(3));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v4 =
- new MyVertex();
- v4.initialize(new IntWritable(4), new IntWritable(4));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v5 =
- new MyVertex();
- v5.initialize(new IntWritable(5), new IntWritable(5));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v6 =
- new MyVertex();
- v6.initialize(new IntWritable(7), new IntWritable(7));
- Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v7 =
- new MyVertex();
- v7.initialize(new IntWritable(7), new IntWritable(7));
-
- partitionStore.addPartition(createPartition(conf, 1, v1, v2));
- partitionStore.addPartition(createPartition(conf, 2, v3));
- partitionStore.addPartition(createPartition(conf, 2, v4));
- partitionStore.addPartition(createPartition(conf, 3, v5));
- partitionStore.addPartition(createPartition(conf, 1, v6));
- partitionStore.addPartition(createPartition(conf, 4, v7));
-
- Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition1 =
- partitionStore.getPartition(1);
- Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition2 =
- partitionStore.getPartition(2);
- Partition<IntWritable, IntWritable, NullWritable,
- IntWritable> partition3 = partitionStore.removePartition(3);
- Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition4 =
- partitionStore.getPartition(4);
-
- assertEquals(3, partitionStore.getNumPartitions());
- assertEquals(3, Iterables.size(partitionStore.getPartitionIds()));
- assertEquals(3, Iterables.size(partitionStore.getPartitions()));
- assertTrue(partitionStore.hasPartition(1));
- assertTrue(partitionStore.hasPartition(2));
- assertFalse(partitionStore.hasPartition(3));
- assertTrue(partitionStore.hasPartition(4));
- assertEquals(3, partition1.getVertexCount());
- assertEquals(2, partition2.getVertexCount());
- assertEquals(1, partition3.getVertexCount());
- assertEquals(1, partition4.getVertexCount());
-
- partitionStore.deletePartition(2);
-
- assertEquals(2, partitionStore.getNumPartitions());
- }
-}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/io/TestAdjacencyListTextVertexOutputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/io/TestAdjacencyListTextVertexOutputFormat.java b/giraph-core/src/test/java/org/apache/giraph/io/TestAdjacencyListTextVertexOutputFormat.java
index 36f73ab..76686a0 100644
--- a/giraph-core/src/test/java/org/apache/giraph/io/TestAdjacencyListTextVertexOutputFormat.java
+++ b/giraph-core/src/test/java/org/apache/giraph/io/TestAdjacencyListTextVertexOutputFormat.java
@@ -18,7 +18,8 @@
package org.apache.giraph.io;
import org.apache.giraph.graph.Edge;
-import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.vertex.Vertex;
+import org.apache.giraph.io.formats.AdjacencyListTextVertexOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
@@ -37,8 +38,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
-public class TestAdjacencyListTextVertexOutputFormat extends
- AdjacencyListTextVertexOutputFormat<Text, DoubleWritable, DoubleWritable> {
+public class TestAdjacencyListTextVertexOutputFormat extends AdjacencyListTextVertexOutputFormat<Text, DoubleWritable, DoubleWritable> {
protected AdjacencyListTextVertexWriter createVertexWriter(
final RecordWriter<Text, Text> tw) {
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/io/TestEdgeInput.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/io/TestEdgeInput.java b/giraph-core/src/test/java/org/apache/giraph/io/TestEdgeInput.java
new file mode 100644
index 0000000..112a267
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/io/TestEdgeInput.java
@@ -0,0 +1,166 @@
+/*
+ * 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.giraph.io;
+
+import org.apache.giraph.BspCase;
+import org.apache.giraph.conf.GiraphClasses;
+import org.apache.giraph.vertex.EdgeListVertex;
+import org.apache.giraph.io.formats.IdWithValueTextOutputFormat;
+import org.apache.giraph.io.formats.IntIntTextVertexValueInputFormat;
+import org.apache.giraph.io.formats.IntNullTextEdgeInputFormat;
+import org.apache.giraph.utils.InternalVertexRunner;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * A test case to ensure that loading a graph from a list of edges works as
+ * expected.
+ */
+public class TestEdgeInput extends BspCase {
+ public TestEdgeInput() {
+ super(TestEdgeInput.class.getName());
+ }
+
+ // It should be able to build a graph starting from the edges only.
+ // Vertices should be implicitly created with default values.
+ @Test
+ public void testEdgesOnly() throws Exception {
+ String[] edges = new String[] {
+ "1 2",
+ "2 3",
+ "2 4",
+ "4 1"
+ };
+
+ GiraphClasses classes = new GiraphClasses();
+ classes.setVertexClass(TestVertexWithNumEdges.class);
+ classes.setEdgeInputFormatClass(IntNullTextEdgeInputFormat.class);
+ classes.setVertexOutputFormatClass(IdWithValueTextOutputFormat.class);
+ Map<String, String> params = ImmutableMap.of();
+ Iterable<String> results = InternalVertexRunner.run(classes, params,
+ null, edges);
+
+ Map<Integer, Integer> values = parseResults(results);
+
+ // Check that all vertices with outgoing edges have been created
+ assertEquals(3, values.size());
+ // Check the number of edges for each vertex
+ assertEquals(1, (int) values.get(1));
+ assertEquals(2, (int) values.get(2));
+ assertEquals(1, (int) values.get(4));
+ }
+
+ // It should be able to build a graph by specifying vertex data and edges
+ // as separate input formats.
+ @Test
+ public void testMixedFormat() throws Exception {
+ String[] vertices = new String[] {
+ "1 75",
+ "2 34",
+ "3 13",
+ "4 32"
+ };
+ String[] edges = new String[] {
+ "1 2",
+ "2 3",
+ "2 4",
+ "4 1",
+ "5 3"
+ };
+
+ GiraphClasses classes = new GiraphClasses();
+ classes.setVertexClass(TestVertexDoNothing.class);
+ classes.setVertexInputFormatClass(IntIntTextVertexValueInputFormat.class);
+ classes.setEdgeInputFormatClass(IntNullTextEdgeInputFormat.class);
+ classes.setVertexOutputFormatClass(IdWithValueTextOutputFormat.class);
+ Map<String, String> emptyParams = ImmutableMap.of();
+
+ // Run a job with a vertex that does nothing
+ Iterable<String> results = InternalVertexRunner.run(classes, emptyParams,
+ vertices, edges);
+
+ Map<Integer, Integer> values = parseResults(results);
+
+ // Check that all vertices with either initial values or outgoing edges
+ // have been created
+ assertEquals(5, values.size());
+ // Check that the vertices have been created with correct values
+ assertEquals(75, (int) values.get(1));
+ assertEquals(34, (int) values.get(2));
+ assertEquals(13, (int) values.get(3));
+ assertEquals(32, (int) values.get(4));
+ // A vertex with edges but no initial value should have the default value
+ assertEquals(0, (int) values.get(5));
+
+ classes = new GiraphClasses();
+ classes.setVertexClass(TestVertexWithNumEdges.class);
+ classes.setVertexInputFormatClass(IntIntTextVertexValueInputFormat.class);
+ classes.setEdgeInputFormatClass(IntNullTextEdgeInputFormat.class);
+ classes.setVertexOutputFormatClass(IdWithValueTextOutputFormat.class);
+
+ // Run a job with a vertex that counts outgoing edges
+ results = InternalVertexRunner.run(classes, emptyParams, vertices, edges);
+
+ values = parseResults(results);
+
+ // Check the number of edges for each vertex
+ assertEquals(1, (int) values.get(1));
+ assertEquals(2, (int) values.get(2));
+ assertEquals(0, (int) values.get(3));
+ assertEquals(1, (int) values.get(4));
+ assertEquals(1, (int) values.get(5));
+ }
+
+ public static class TestVertexWithNumEdges extends EdgeListVertex<IntWritable,
+ IntWritable, NullWritable, NullWritable> {
+ @Override
+ public void compute(Iterable<NullWritable> messages) throws IOException {
+ setValue(new IntWritable(getNumEdges()));
+ voteToHalt();
+ }
+ }
+
+ public static class TestVertexDoNothing extends EdgeListVertex<IntWritable,
+ IntWritable, NullWritable, NullWritable> {
+ @Override
+ public void compute(Iterable<NullWritable> messages) throws IOException {
+ voteToHalt();
+ }
+ }
+
+ private static Map<Integer, Integer> parseResults(Iterable<String> results) {
+ Map<Integer, Integer> values = Maps.newHashMap();
+ for (String line : results) {
+ String[] tokens = line.split("\\s+");
+ int id = Integer.valueOf(tokens[0]);
+ int value = Integer.valueOf(tokens[1]);
+ values.put(id, value);
+ }
+ return values;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/io/TestIdWithValueTextOutputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/io/TestIdWithValueTextOutputFormat.java b/giraph-core/src/test/java/org/apache/giraph/io/TestIdWithValueTextOutputFormat.java
index 17511c4..e770338 100644
--- a/giraph-core/src/test/java/org/apache/giraph/io/TestIdWithValueTextOutputFormat.java
+++ b/giraph-core/src/test/java/org/apache/giraph/io/TestIdWithValueTextOutputFormat.java
@@ -18,7 +18,8 @@
package org.apache.giraph.io;
-import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.vertex.Vertex;
+import org.apache.giraph.io.formats.IdWithValueTextOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
@@ -37,8 +38,7 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.ArrayList;
-public class TestIdWithValueTextOutputFormat extends
- IdWithValueTextOutputFormat<Text, DoubleWritable, Writable> {
+public class TestIdWithValueTextOutputFormat extends IdWithValueTextOutputFormat<Text, DoubleWritable, Writable> {
@Test
public void testHappyPath() throws IOException, InterruptedException {
Configuration conf = new Configuration();
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/io/TestJsonBase64Format.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/io/TestJsonBase64Format.java b/giraph-core/src/test/java/org/apache/giraph/io/TestJsonBase64Format.java
new file mode 100644
index 0000000..13f4fe1
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/io/TestJsonBase64Format.java
@@ -0,0 +1,105 @@
+/*
+ * 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.giraph.io;
+import org.apache.giraph.BspCase;
+import org.apache.giraph.benchmark.EdgeListVertexPageRankBenchmark;
+import org.apache.giraph.benchmark.PageRankComputation;
+import org.apache.giraph.conf.GiraphClasses;
+import org.apache.giraph.graph.GiraphJob;
+import org.apache.giraph.io.formats.GiraphFileInputFormat;
+import org.apache.giraph.io.formats.JsonBase64VertexInputFormat;
+import org.apache.giraph.io.formats.JsonBase64VertexOutputFormat;
+import org.apache.giraph.io.formats.PseudoRandomVertexInputFormat;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test out the JsonBase64 format.
+ */
+public class TestJsonBase64Format extends BspCase {
+ /**
+ * Constructor.
+ */
+ public TestJsonBase64Format() {
+ super(TestJsonBase64Format.class.getName());
+ }
+
+ /**
+ * Start a job and finish after i supersteps, then begin a new job and
+ * continue on more j supersteps. Check the results against a single job
+ * with i + j supersteps.
+ *
+ * @throws IOException
+ * @throws ClassNotFoundException
+ * @throws InterruptedException
+ */
+ @Test
+ public void testContinue()
+ throws IOException, InterruptedException, ClassNotFoundException {
+
+ Path outputPath = getTempPath(getCallingMethodName());
+ GiraphClasses classes = new GiraphClasses();
+ classes.setVertexClass(EdgeListVertexPageRankBenchmark.class);
+ classes.setVertexInputFormatClass(PseudoRandomVertexInputFormat.class);
+ classes.setVertexOutputFormatClass(JsonBase64VertexOutputFormat.class);
+ GiraphJob job = prepareJob(getCallingMethodName(), classes, outputPath);
+ job.getConfiguration().setLong(
+ PseudoRandomVertexInputFormat.AGGREGATE_VERTICES, 101);
+ job.getConfiguration().setLong(
+ PseudoRandomVertexInputFormat.EDGES_PER_VERTEX, 2);
+ job.getConfiguration().setInt(PageRankComputation.SUPERSTEP_COUNT, 2);
+
+ assertTrue(job.run(true));
+
+ Path outputPath2 = getTempPath(getCallingMethodName() + "2");
+ classes = new GiraphClasses();
+ classes.setVertexClass(EdgeListVertexPageRankBenchmark.class);
+ classes.setVertexInputFormatClass(JsonBase64VertexInputFormat.class);
+ classes.setVertexOutputFormatClass(JsonBase64VertexOutputFormat.class);
+ job = prepareJob(getCallingMethodName(), classes, outputPath2);
+ job.getConfiguration().setInt(PageRankComputation.SUPERSTEP_COUNT, 3);
+ GiraphFileInputFormat.addVertexInputPath(job.getInternalJob(), outputPath);
+ assertTrue(job.run(true));
+
+ Path outputPath3 = getTempPath(getCallingMethodName() + "3");
+ classes = new GiraphClasses();
+ classes.setVertexClass(EdgeListVertexPageRankBenchmark.class);
+ classes.setVertexInputFormatClass(PseudoRandomVertexInputFormat.class);
+ classes.setVertexOutputFormatClass(JsonBase64VertexOutputFormat.class);
+ job = prepareJob(getCallingMethodName(), classes, outputPath3);
+ job.getConfiguration().setLong(
+ PseudoRandomVertexInputFormat.AGGREGATE_VERTICES, 101);
+ job.getConfiguration().setLong(
+ PseudoRandomVertexInputFormat.EDGES_PER_VERTEX, 2);
+ job.getConfiguration().setInt(PageRankComputation.SUPERSTEP_COUNT, 5);
+ assertTrue(job.run(true));
+
+ Configuration conf = job.getConfiguration();
+
+ assertEquals(101, getNumResults(conf, outputPath));
+ assertEquals(101, getNumResults(conf, outputPath2));
+ assertEquals(101, getNumResults(conf, outputPath3));
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/io/TestLongDoubleDoubleAdjacencyListVertexInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/io/TestLongDoubleDoubleAdjacencyListVertexInputFormat.java b/giraph-core/src/test/java/org/apache/giraph/io/TestLongDoubleDoubleAdjacencyListVertexInputFormat.java
index b03c8e4..1b1ebb0 100644
--- a/giraph-core/src/test/java/org/apache/giraph/io/TestLongDoubleDoubleAdjacencyListVertexInputFormat.java
+++ b/giraph-core/src/test/java/org/apache/giraph/io/TestLongDoubleDoubleAdjacencyListVertexInputFormat.java
@@ -20,9 +20,11 @@ package org.apache.giraph.io;
import org.apache.giraph.conf.GiraphConstants;
import org.apache.giraph.graph.Edge;
-import org.apache.giraph.graph.EdgeListVertex;
+import org.apache.giraph.io.formats.AdjacencyListTextVertexInputFormat;
+import org.apache.giraph.io.formats.LongDoubleDoubleAdjacencyListVertexInputFormat;
+import org.apache.giraph.vertex.EdgeListVertex;
import org.apache.giraph.graph.GraphState;
-import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.vertex.Vertex;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.DoubleWritable;
@@ -45,8 +47,7 @@ import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-public class TestLongDoubleDoubleAdjacencyListVertexInputFormat extends
- LongDoubleDoubleAdjacencyListVertexInputFormat<BooleanWritable> {
+public class TestLongDoubleDoubleAdjacencyListVertexInputFormat extends LongDoubleDoubleAdjacencyListVertexInputFormat<BooleanWritable> {
private RecordReader<LongWritable, Text> rr;
private Configuration conf;
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/io/TestTextDoubleDoubleAdjacencyListVertexInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/io/TestTextDoubleDoubleAdjacencyListVertexInputFormat.java b/giraph-core/src/test/java/org/apache/giraph/io/TestTextDoubleDoubleAdjacencyListVertexInputFormat.java
index c447598..8e6c92b 100644
--- a/giraph-core/src/test/java/org/apache/giraph/io/TestTextDoubleDoubleAdjacencyListVertexInputFormat.java
+++ b/giraph-core/src/test/java/org/apache/giraph/io/TestTextDoubleDoubleAdjacencyListVertexInputFormat.java
@@ -19,8 +19,14 @@ package org.apache.giraph.io;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
+
+import org.apache.giraph.bsp.BspUtils;
import org.apache.giraph.conf.GiraphConstants;
import org.apache.giraph.graph.*;
+import org.apache.giraph.io.formats.AdjacencyListTextVertexInputFormat;
+import org.apache.giraph.io.formats.TextDoubleDoubleAdjacencyListVertexInputFormat;
+import org.apache.giraph.vertex.EdgeListVertex;
+import org.apache.giraph.vertex.Vertex;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.InputSplit;
@@ -39,8 +45,7 @@ import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-public class TestTextDoubleDoubleAdjacencyListVertexInputFormat extends
- TextDoubleDoubleAdjacencyListVertexInputFormat<BooleanWritable> {
+public class TestTextDoubleDoubleAdjacencyListVertexInputFormat extends TextDoubleDoubleAdjacencyListVertexInputFormat<BooleanWritable> {
private RecordReader<LongWritable, Text> rr;
private Configuration conf;
@@ -214,7 +219,7 @@ public class TestTextDoubleDoubleAdjacencyListVertexInputFormat extends
public static class DummyVertex
extends EdgeListVertex<Text, DoubleWritable,
- DoubleWritable, BooleanWritable> {
+ DoubleWritable, BooleanWritable> {
@Override
public void compute(Iterable<BooleanWritable> messages) throws IOException {
// ignore
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/master/TestMasterObserver.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/master/TestMasterObserver.java b/giraph-core/src/test/java/org/apache/giraph/master/TestMasterObserver.java
new file mode 100644
index 0000000..f67a6e0
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/master/TestMasterObserver.java
@@ -0,0 +1,102 @@
+/*
+ * 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.giraph.master;
+
+import org.apache.giraph.conf.GiraphClasses;
+import org.apache.giraph.conf.GiraphConstants;
+import org.apache.giraph.vertex.IntNullNullNullVertex;
+import org.apache.giraph.io.formats.IntNullNullNullTextInputFormat;
+import org.apache.giraph.utils.InternalVertexRunner;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.util.StringUtils;
+import org.junit.Test;
+
+import com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestMasterObserver {
+ public static class NoOpVertex extends IntNullNullNullVertex {
+ private int count = 0;
+
+ @Override
+ public void compute(Iterable<NullWritable> messages) throws IOException {
+ if (count == 2) {
+ voteToHalt();
+ }
+ ++count;
+ }
+ }
+
+ public static class Obs extends DefaultMasterObserver {
+ public static int preApp = 0;
+ public static int preSuperstep = 0;
+ public static int postSuperstep = 0;
+ public static int postApp = 0;
+
+ @Override
+ public void preApplication() {
+ ++preApp;
+ }
+
+ @Override
+ public void postApplication() {
+ ++postApp;
+ }
+
+ @Override
+ public void preSuperstep() {
+ ++preSuperstep;
+ }
+
+ @Override
+ public void postSuperstep() {
+ ++postSuperstep;
+ }
+ }
+
+ @Test
+ public void testGetsCalled() throws Exception {
+ assertEquals(0, Obs.postApp);
+
+ String[] graph = new String[] { "1", "2", "3" };
+
+ Map<String, String> params = Maps.newHashMap();
+ String klasses[] = new String[] {
+ Obs.class.getName(),
+ Obs.class.getName()
+ };
+ params.put(GiraphConstants.MASTER_OBSERVER_CLASSES,
+ StringUtils.arrayToString(klasses));
+
+ GiraphClasses classes = new GiraphClasses();
+ classes.setVertexClass(NoOpVertex.class);
+ classes.setVertexInputFormatClass(IntNullNullNullTextInputFormat.class);
+ InternalVertexRunner.run(classes, params, graph);
+
+ assertEquals(2, Obs.preApp);
+ // 3 supersteps + 1 input superstep * 2 observers = 8 callbacks
+ assertEquals(8, Obs.preSuperstep);
+ assertEquals(8, Obs.postSuperstep);
+ assertEquals(2, Obs.postApp);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/partition/TestGiraphTransferRegulator.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/partition/TestGiraphTransferRegulator.java b/giraph-core/src/test/java/org/apache/giraph/partition/TestGiraphTransferRegulator.java
new file mode 100644
index 0000000..61e8863
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/partition/TestGiraphTransferRegulator.java
@@ -0,0 +1,94 @@
+/*
+ * 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.giraph.partition;
+
+import org.apache.giraph.graph.Edge;
+import org.apache.giraph.partition.PartitionOwner;
+import org.apache.giraph.vertex.EdgeListVertex;
+import org.apache.giraph.graph.GiraphJob;
+import org.apache.giraph.graph.GiraphTransferRegulator;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.FloatWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.Lists;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Test the GiraphTransferRegulator.
+ */
+public class TestGiraphTransferRegulator {
+ /** Job filled in by setup() */
+ private GiraphJob job;
+ /** Instantiated vertex filled in from setup() */
+ private IFDLEdgeListVertex vertex = new IFDLEdgeListVertex();
+
+ /**
+ * Simple instantiable class that extends
+ * {@link org.apache.giraph.vertex.EdgeListVertex}.
+ */
+ public static class IFDLEdgeListVertex extends
+ EdgeListVertex<IntWritable, FloatWritable, DoubleWritable, LongWritable> {
+ @Override
+ public void compute(Iterable<LongWritable> messages) throws IOException { }
+ }
+
+ @Before
+ public void setUp() {
+ try {
+ job = new GiraphJob("TestGiraphTransferRegulator");
+ } catch (IOException e) {
+ throw new RuntimeException("setUp: Failed", e);
+ }
+ job.getConfiguration().setVertexClass(IFDLEdgeListVertex.class);
+ }
+
+ @Test
+ public void testGiraphTransferRegulator() {
+ job.getConfiguration()
+ .setInt(GiraphTransferRegulator.MAX_VERTICES_PER_TRANSFER, 1);
+ job.getConfiguration()
+ .setInt(GiraphTransferRegulator.MAX_EDGES_PER_TRANSFER, 3);
+ List<Edge<IntWritable, DoubleWritable>> edges = Lists.newLinkedList();
+ edges.add(new Edge<IntWritable, DoubleWritable>(new IntWritable(2),
+ new DoubleWritable(22)));
+ edges.add(new Edge<IntWritable, DoubleWritable>(new IntWritable(3),
+ new DoubleWritable(33)));
+ edges.add(new Edge<IntWritable, DoubleWritable>(new IntWritable(4),
+ new DoubleWritable(44)));
+ vertex.initialize(null, null, edges);
+ GiraphTransferRegulator gtr =
+ new GiraphTransferRegulator(job.getConfiguration());
+ PartitionOwner owner = mock(PartitionOwner.class);
+ when(owner.getPartitionId()).thenReturn(57);
+ assertFalse(gtr.transferThisPartition(owner));
+ gtr.incrementCounters(owner, vertex);
+ assertTrue(gtr.transferThisPartition(owner));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/partition/TestPartitionStores.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/partition/TestPartitionStores.java b/giraph-core/src/test/java/org/apache/giraph/partition/TestPartitionStores.java
new file mode 100644
index 0000000..b02ed3a
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/partition/TestPartitionStores.java
@@ -0,0 +1,211 @@
+/*
+ * 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.giraph.partition;
+
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.GiraphConstants;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.vertex.IntIntNullIntVertex;
+import org.apache.giraph.vertex.Vertex;
+import org.apache.giraph.utils.UnsafeByteArrayInputStream;
+import org.apache.giraph.utils.UnsafeByteArrayOutputStream;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.collect.Iterables;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Test case for partition stores.
+ */
+public class TestPartitionStores {
+ private ImmutableClassesGiraphConfiguration conf;
+ private Mapper<?, ?, ?, ?>.Context context;
+
+ public static class MyVertex extends IntIntNullIntVertex {
+ @Override
+ public void compute(Iterable<IntWritable> messages) throws IOException {}
+ }
+
+ private Partition<IntWritable, IntWritable, NullWritable,
+ IntWritable> createPartition(ImmutableClassesGiraphConfiguration conf,
+ Integer id,
+ Vertex<IntWritable, IntWritable,
+ NullWritable, IntWritable>... vertices) {
+ Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition =
+ conf.createPartition(id, context);
+ for (Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v :
+ vertices) {
+ partition.putVertex(v);
+ }
+ return partition;
+ }
+
+ @Before
+ public void setUp() {
+ GiraphConfiguration configuration = new GiraphConfiguration();
+ configuration.setVertexClass(MyVertex.class);
+ conf = new ImmutableClassesGiraphConfiguration(configuration);
+ context = mock(Mapper.Context.class);
+ }
+
+ @Test
+ public void testSimplePartitionStore() {
+ PartitionStore<IntWritable, IntWritable, NullWritable, IntWritable>
+ partitionStore = new SimplePartitionStore<IntWritable, IntWritable,
+ NullWritable, IntWritable>(conf, context);
+ testReadWrite(partitionStore, conf);
+ }
+
+ @Test
+ public void testUnsafePartitionSerializationClass() throws IOException {
+ conf.setPartitionClass(ByteArrayPartition.class);
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v1 =
+ new MyVertex();
+ v1.initialize(new IntWritable(1), new IntWritable(1));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v2 =
+ new MyVertex();
+ v2.initialize(new IntWritable(2), new IntWritable(2));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v3 =
+ new MyVertex();
+ v3.initialize(new IntWritable(3), new IntWritable(3));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v4 =
+ new MyVertex();
+ v4.initialize(new IntWritable(4), new IntWritable(4));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v5 =
+ new MyVertex();
+ v5.initialize(new IntWritable(5), new IntWritable(5));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v6 =
+ new MyVertex();
+ v6.initialize(new IntWritable(6), new IntWritable(6));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v7 =
+ new MyVertex();
+ v7.initialize(new IntWritable(7), new IntWritable(7));
+
+ Partition<IntWritable, IntWritable, NullWritable,
+ IntWritable> partition =
+ createPartition(conf, 3, v1, v2, v3, v4, v5, v6, v7);
+ assertEquals(3, partition.getId());
+ assertEquals(0, partition.getEdgeCount());
+ assertEquals(7, partition.getVertexCount());
+ UnsafeByteArrayOutputStream outputStream = new
+ UnsafeByteArrayOutputStream();
+ partition.write(outputStream);
+ UnsafeByteArrayInputStream inputStream = new UnsafeByteArrayInputStream(
+ outputStream.getByteArray(), 0, outputStream.getPos());
+ Partition<IntWritable, IntWritable, NullWritable,
+ IntWritable> deserializatedPartition = conf.createPartition(-1,
+ context);
+ deserializatedPartition.readFields(inputStream);
+
+ assertEquals(3, deserializatedPartition.getId());
+ assertEquals(0, deserializatedPartition.getEdgeCount());
+ assertEquals(7, deserializatedPartition.getVertexCount());
+ }
+
+ @Test
+ public void testDiskBackedPartitionStore() {
+ conf.setBoolean(GiraphConstants.USE_OUT_OF_CORE_GRAPH, true);
+ conf.setInt(GiraphConstants.MAX_PARTITIONS_IN_MEMORY, 1);
+
+ PartitionStore<IntWritable, IntWritable, NullWritable, IntWritable>
+ partitionStore = new DiskBackedPartitionStore<IntWritable,
+ IntWritable, NullWritable, IntWritable>(conf, context);
+ testReadWrite(partitionStore, conf);
+
+ conf.setInt(GiraphConstants.MAX_PARTITIONS_IN_MEMORY, 2);
+ partitionStore = new DiskBackedPartitionStore<IntWritable,
+ IntWritable, NullWritable, IntWritable>(conf, context);
+ testReadWrite(partitionStore, conf);
+ }
+
+ /**
+ * Test reading/writing to/from a partition store
+ *
+ * @param partitionStore Partition store to test
+ * @param conf Configuration to use
+ */
+ public void testReadWrite(
+ PartitionStore<IntWritable, IntWritable,
+ NullWritable, IntWritable> partitionStore,
+ ImmutableClassesGiraphConfiguration conf) {
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v1 =
+ new MyVertex();
+ v1.initialize(new IntWritable(1), new IntWritable(1));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v2 =
+ new MyVertex();
+ v2.initialize(new IntWritable(2), new IntWritable(2));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v3 =
+ new MyVertex();
+ v3.initialize(new IntWritable(3), new IntWritable(3));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v4 =
+ new MyVertex();
+ v4.initialize(new IntWritable(4), new IntWritable(4));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v5 =
+ new MyVertex();
+ v5.initialize(new IntWritable(5), new IntWritable(5));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v6 =
+ new MyVertex();
+ v6.initialize(new IntWritable(7), new IntWritable(7));
+ Vertex<IntWritable, IntWritable, NullWritable, IntWritable> v7 =
+ new MyVertex();
+ v7.initialize(new IntWritable(7), new IntWritable(7));
+
+ partitionStore.addPartition(createPartition(conf, 1, v1, v2));
+ partitionStore.addPartition(createPartition(conf, 2, v3));
+ partitionStore.addPartition(createPartition(conf, 2, v4));
+ partitionStore.addPartition(createPartition(conf, 3, v5));
+ partitionStore.addPartition(createPartition(conf, 1, v6));
+ partitionStore.addPartition(createPartition(conf, 4, v7));
+
+ Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition1 =
+ partitionStore.getPartition(1);
+ Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition2 =
+ partitionStore.getPartition(2);
+ Partition<IntWritable, IntWritable, NullWritable,
+ IntWritable> partition3 = partitionStore.removePartition(3);
+ Partition<IntWritable, IntWritable, NullWritable, IntWritable> partition4 =
+ partitionStore.getPartition(4);
+
+ assertEquals(3, partitionStore.getNumPartitions());
+ assertEquals(3, Iterables.size(partitionStore.getPartitionIds()));
+ assertEquals(3, Iterables.size(partitionStore.getPartitions()));
+ assertTrue(partitionStore.hasPartition(1));
+ assertTrue(partitionStore.hasPartition(2));
+ assertFalse(partitionStore.hasPartition(3));
+ assertTrue(partitionStore.hasPartition(4));
+ assertEquals(3, partition1.getVertexCount());
+ assertEquals(2, partition2.getVertexCount());
+ assertEquals(1, partition3.getVertexCount());
+ assertEquals(1, partition4.getVertexCount());
+
+ partitionStore.deletePartition(2);
+
+ assertEquals(2, partitionStore.getNumPartitions());
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/utils/BspUtilsTest.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/utils/BspUtilsTest.java b/giraph-core/src/test/java/org/apache/giraph/utils/BspUtilsTest.java
deleted file mode 100644
index 68f6553..0000000
--- a/giraph-core/src/test/java/org/apache/giraph/utils/BspUtilsTest.java
+++ /dev/null
@@ -1,193 +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.giraph.utils;
-
-import org.apache.giraph.conf.GiraphConfiguration;
-import org.apache.giraph.conf.GiraphConstants;
-import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
-import org.apache.giraph.graph.BspUtils;
-import org.apache.giraph.graph.EdgeListVertex;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.DoubleWritable;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableComparable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestName;
-
-import java.io.IOException;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Benchmark tests to insure that object creation is fast
- */
-public class BspUtilsTest {
- @Rule
- public TestName name = new TestName();
- private static final Time TIME = SystemTime.get();
- private static final long COUNT = 200000;
- private Configuration conf = new Configuration();
- private long startNanos = -1;
- private long totalNanos = -1;
- private long total = 0;
- private long expected = COUNT * (COUNT - 1) / 2L;
- private ImmutableClassesGiraphConfiguration<LongWritable, LongWritable,
- LongWritable, LongWritable> configuration;
-
- @Before
- public void setUp() {
- conf.setClass(GiraphConstants.VERTEX_ID_CLASS, IntWritable.class,
- WritableComparable.class);
- conf.setClass(GiraphConstants.VERTEX_VALUE_CLASS, LongWritable.class,
- Writable.class);
- conf.setClass(GiraphConstants.EDGE_VALUE_CLASS, DoubleWritable.class,
- Writable.class);
- conf.setClass(GiraphConstants.MESSAGE_VALUE_CLASS, LongWritable.class,
- Writable.class);
- GiraphConfiguration conf = new GiraphConfiguration();
- conf.setVertexClass(ImmutableVertex.class);
- configuration =
- new ImmutableClassesGiraphConfiguration<LongWritable, LongWritable,
- LongWritable, LongWritable>(conf);
- total = 0;
- System.gc();
- }
-
- @After
- public void cleanUp() {
- totalNanos = Times.getNanosSince(TIME, startNanos);
- System.out.println(name.getMethodName() + ": took "
- + totalNanos +
- " ns for " + COUNT + " elements " + (totalNanos * 1f / COUNT) +
- " ns / element");
- assertEquals(expected, total);
- System.gc();
- }
-
- @Test
- public void testCreateClass() {
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = BspUtils.createVertexValue(conf);
- value.set(i);
- total += value.get();
- }
- }
-
- @Test
- public void testNativeCreateClass() {
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = new LongWritable();
- value.set(i);
- total += value.get();
- }
- }
-
- private Class<?> getLongWritableClass() {
- return LongWritable.class;
- }
-
- @Test
- public void testNewInstance()
- throws IllegalAccessException, InstantiationException {
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = (LongWritable)
- getLongWritableClass().newInstance();
- value.set(i);
- total += value.get();
- }
- }
-
- private synchronized Class<?> getSyncLongWritableClass() {
- return LongWritable.class;
- }
-
- @Test
- public void testSyncNewInstance()
- throws IllegalAccessException, InstantiationException {
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = (LongWritable)
- getSyncLongWritableClass().newInstance();
- value.set(i);
- total += value.get();
- }
- }
-
- @Test
- public void testReflectionUtilsNewInstance()
- throws IllegalAccessException, InstantiationException {
- // Throwaway to put into cache
- org.apache.hadoop.util.ReflectionUtils.newInstance(LongWritable.class,
- null);
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = (LongWritable)
- org.apache.hadoop.util.ReflectionUtils.newInstance(
- getLongWritableClass(), null);
- value.set(i);
- total += value.get();
- }
- }
-
- @Test
- public void testConstructorNewInstance()
- throws IllegalAccessException, InstantiationException,
- NoSuchMethodException, InvocationTargetException {
- Constructor<?> constructor = LongWritable.class.getDeclaredConstructor
- (new Class[]{});
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = (LongWritable) constructor.newInstance();
- value.set(i);
- total += value.get();
- }
- }
-
- private static class ImmutableVertex extends EdgeListVertex<LongWritable,
- LongWritable, LongWritable, LongWritable> {
- @Override
- public void compute(Iterable<LongWritable> messages) throws IOException {
- }
- }
-
- private ImmutableClassesGiraphConfiguration<LongWritable, LongWritable,
- LongWritable, LongWritable> getConfiguration() {
- return configuration;
- }
-
- @Test
- public void testImmutableClassesGiraphConfigurationNewInstance() {
- startNanos = TIME.getNanoseconds();
- for (int i = 0; i < COUNT; ++i) {
- LongWritable value = getConfiguration().createVertexValue();
- value.set(i);
- total += value.get();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/utils/MockUtils.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/utils/MockUtils.java b/giraph-core/src/test/java/org/apache/giraph/utils/MockUtils.java
index 04d2282..82dc283 100644
--- a/giraph-core/src/test/java/org/apache/giraph/utils/MockUtils.java
+++ b/giraph-core/src/test/java/org/apache/giraph/utils/MockUtils.java
@@ -23,10 +23,10 @@ import org.apache.giraph.bsp.CentralizedServiceWorker;
import org.apache.giraph.comm.ServerData;
import org.apache.giraph.comm.WorkerClientRequestProcessor;
import org.apache.giraph.comm.messages.ByteArrayMessagesPerVertexStore;
-import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.vertex.Vertex;
import org.apache.giraph.graph.GraphState;
-import org.apache.giraph.graph.partition.BasicPartitionOwner;
-import org.apache.giraph.graph.partition.PartitionOwner;
+import org.apache.giraph.partition.BasicPartitionOwner;
+import org.apache.giraph.partition.PartitionOwner;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/vertex/TestIntIntNullIntVertex.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/vertex/TestIntIntNullIntVertex.java b/giraph-core/src/test/java/org/apache/giraph/vertex/TestIntIntNullIntVertex.java
new file mode 100644
index 0000000..73bd135
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/vertex/TestIntIntNullIntVertex.java
@@ -0,0 +1,79 @@
+/*
+ * 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.giraph.vertex;
+
+import org.apache.giraph.graph.Edge;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.common.collect.Lists;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Tests {@link org.apache.giraph.vertex.IntIntNullIntVertex}.
+ */
+public class TestIntIntNullIntVertex {
+ /**
+ * Simple instantiable class that extends {@link org.apache.giraph.vertex.IntIntNullIntVertex}.
+ */
+ private static class MyIntIntNullVertex extends IntIntNullIntVertex {
+ @Override
+ public void compute(Iterable<IntWritable> messages) throws IOException {
+ }
+ }
+
+ @Test
+ public void testSerialize() throws IOException {
+ IntIntNullIntVertex vertex = new MyIntIntNullVertex();
+
+ List<Edge<IntWritable, NullWritable>> edges = Lists.newLinkedList();
+ edges.add(new Edge<IntWritable, NullWritable>(new IntWritable(3),
+ NullWritable.get()));
+ edges.add(new Edge<IntWritable, NullWritable>(new IntWritable(47),
+ NullWritable.get()));
+
+ vertex.initialize(new IntWritable(23), new IntWritable(7), edges);
+ vertex.voteToHalt();
+
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ DataOutput out = new DataOutputStream(outStream);
+ vertex.write(out);
+
+ IntIntNullIntVertex vertex1 = new MyIntIntNullVertex();
+
+ ByteArrayInputStream inStream = new ByteArrayInputStream(
+ outStream.toByteArray());
+ DataInput in = new DataInputStream(inStream);
+ vertex1.readFields(in);
+
+ assertEquals(2, vertex1.getNumEdges());
+ assertEquals(true, vertex1.isHalted());
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/vertex/TestMultiGraphVertex.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/vertex/TestMultiGraphVertex.java b/giraph-core/src/test/java/org/apache/giraph/vertex/TestMultiGraphVertex.java
new file mode 100644
index 0000000..8822d2e
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/vertex/TestMultiGraphVertex.java
@@ -0,0 +1,130 @@
+/*
+ * 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.giraph.vertex;
+
+import com.google.common.collect.Lists;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.graph.Edge;
+import org.apache.hadoop.io.IntWritable;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test all multigraph mutable vertices.
+ */
+public class TestMultiGraphVertex {
+ private Collection<Class<? extends Vertex<IntWritable, IntWritable,
+ IntWritable, IntWritable>>> vertexClasses = Lists.newArrayList();
+
+ public static class MyMultiGraphEdgeListVertex
+ extends MultiGraphEdgeListVertex<IntWritable, IntWritable, IntWritable,
+ IntWritable> {
+ @Override
+ public void compute(Iterable<IntWritable> messages) throws IOException { }
+ }
+
+ public static class MyMultiGraphRepresentativeVertex
+ extends MultiGraphRepresentativeVertex<IntWritable, IntWritable,
+ IntWritable, IntWritable> {
+ @Override
+ public void compute(Iterable<IntWritable> messages) throws IOException { }
+ }
+
+ @Before
+ public void setUp() {
+ vertexClasses.add(MyMultiGraphEdgeListVertex.class);
+ vertexClasses.add(MyMultiGraphRepresentativeVertex.class);
+ }
+
+ @Test
+ public void testAddRemoveEdges() {
+ for (Class<? extends Vertex<IntWritable, IntWritable,
+ IntWritable, IntWritable>> vertexClass : vertexClasses) {
+ testAddRemoveEdgesVertexClass(vertexClass);
+ }
+ }
+
+ private void testAddRemoveEdgesVertexClass(Class<? extends Vertex<IntWritable,
+ IntWritable, IntWritable, IntWritable>> vertexClass) {
+ MutableVertex<IntWritable, IntWritable, IntWritable,
+ IntWritable> vertex = instantiateVertex(vertexClass);
+
+ assertEquals(vertex.removeEdges(new IntWritable(1)), 0);
+
+ // We test a few different patterns for duplicate edges,
+ // in order to catch corner cases:
+
+ // Edge list of form: [A, B, A]
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(1),
+ new IntWritable(1)));
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(2),
+ new IntWritable(2)));
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(1),
+ new IntWritable(10)));
+ assertEquals(vertex.getNumEdges(), 3);
+ assertEquals(vertex.removeEdges(new IntWritable(1)), 2);
+ assertEquals(vertex.getNumEdges(), 1);
+
+ // Edge list of form: [A, B, B]
+ vertex = instantiateVertex(vertexClass);
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(2),
+ new IntWritable(2)));
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(1),
+ new IntWritable(1)));
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(1),
+ new IntWritable(10)));
+ assertEquals(vertex.getNumEdges(), 3);
+ assertEquals(vertex.removeEdges(new IntWritable(1)), 2);
+ assertEquals(vertex.getNumEdges(), 1);
+
+ // Edge list of form: [A, A, B]
+ vertex = instantiateVertex(vertexClass);
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(1),
+ new IntWritable(1)));
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(1),
+ new IntWritable(10)));
+ vertex.addEdge(new Edge<IntWritable, IntWritable>(new IntWritable(2),
+ new IntWritable(2)));
+ assertEquals(vertex.getNumEdges(), 3);
+ assertEquals(vertex.removeEdges(new IntWritable(1)), 2);
+ assertEquals(vertex.getNumEdges(), 1);
+ }
+
+ private MutableVertex<IntWritable, IntWritable, IntWritable, IntWritable>
+ instantiateVertex(Class<? extends Vertex<IntWritable, IntWritable,
+ IntWritable, IntWritable>> vertexClass) {
+ GiraphConfiguration giraphConfiguration = new GiraphConfiguration();
+ giraphConfiguration.setVertexClass(vertexClass);
+ ImmutableClassesGiraphConfiguration immutableClassesGiraphConfiguration =
+ new ImmutableClassesGiraphConfiguration(giraphConfiguration);
+ MutableVertex<IntWritable, IntWritable, IntWritable,
+ IntWritable> vertex =
+ (MutableVertex<IntWritable, IntWritable,
+ IntWritable, IntWritable>)
+ immutableClassesGiraphConfiguration.createVertex();
+ vertex.initialize(new IntWritable(), new IntWritable());
+ return vertex;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/1684891e/giraph-core/src/test/java/org/apache/giraph/vertex/TestMutableVertex.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/test/java/org/apache/giraph/vertex/TestMutableVertex.java b/giraph-core/src/test/java/org/apache/giraph/vertex/TestMutableVertex.java
new file mode 100644
index 0000000..9ed2d81
--- /dev/null
+++ b/giraph-core/src/test/java/org/apache/giraph/vertex/TestMutableVertex.java
@@ -0,0 +1,467 @@
+/*
+ * 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.giraph.vertex;
+
+import com.google.common.collect.Lists;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.graph.Edge;
+import org.apache.giraph.time.SystemTime;
+import org.apache.giraph.time.Time;
+import org.apache.giraph.time.Times;
+import org.apache.giraph.utils.*;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.FloatWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+/**
+ * Test all the mutable vertices (except multigraph versions)
+ */
+public class TestMutableVertex {
+ /** Number of repetitions */
+ public static final int REPS = 100;
+ /** Vertex classes to be tested filled in from setup() */
+ private Collection<
+ Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>>> vertexClasses = Lists.newArrayList();
+
+ /**
+ * Simple instantiable class that extends
+ * {@link org.apache.giraph.vertex.HashMapVertex}.
+ */
+ public static class IFDLHashMapVertex extends HashMapVertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable> {
+ @Override
+ public void compute(Iterable<LongWritable> messages) throws IOException { }
+ }
+
+ /**
+ * Simple instantiable class that extends
+ * {@link org.apache.giraph.vertex.EdgeListVertex}.
+ */
+ public static class IFDLEdgeListVertex extends EdgeListVertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable> {
+ @Override
+ public void compute(Iterable<LongWritable> messages) throws IOException { }
+ }
+
+ /**
+ * Simple instantiable class that extends
+ * {@link org.apache.giraph.vertex.RepresentativeVertex}.
+ */
+ public static class IFDLRepresentativeVertex extends RepresentativeVertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable> {
+ @Override
+ public void compute(Iterable<LongWritable> messages) throws IOException { }
+ }
+
+ @Before
+ public void setUp() {
+ vertexClasses.add(IFDLHashMapVertex.class);
+ vertexClasses.add(IFDLEdgeListVertex.class);
+ vertexClasses.add(IFDLRepresentativeVertex.class);
+ }
+
+ @Test
+ public void testInstantiate() throws IOException {
+ for (Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass : vertexClasses) {
+ testInstantiateVertexClass(vertexClass);
+ }
+ }
+
+ /**
+ * Test a vertex class for instantiation
+ *
+ * @param vertexClass Vertex class to check
+ * @return Instantiated mutable vertex
+ */
+ private MutableVertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable> testInstantiateVertexClass(
+ Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass) {
+ GiraphConfiguration giraphConfiguration = new GiraphConfiguration();
+ giraphConfiguration.setVertexClass(vertexClass);
+ ImmutableClassesGiraphConfiguration immutableClassesGiraphConfiguration =
+ new ImmutableClassesGiraphConfiguration(giraphConfiguration);
+ MutableVertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable> vertex =
+ (MutableVertex<IntWritable, FloatWritable,
+ DoubleWritable, LongWritable>)
+ immutableClassesGiraphConfiguration.createVertex();
+ assertNotNull(vertex);
+ return vertex;
+ }
+
+ @Test
+ public void testEdges() {
+ for (Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass : vertexClasses) {
+ testEdgesVertexClass(vertexClass);
+ }
+ }
+
+ /**
+ * Test a vertex class for edges
+ *
+ * @param vertexClass Vertex class to check
+ */
+ private void testEdgesVertexClass(Class<? extends Vertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable>> vertexClass) {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ testInstantiateVertexClass(vertexClass);
+
+ List<Edge<IntWritable, DoubleWritable>> edges = Lists.newLinkedList();
+ for (int i = 1000; i > 0; --i) {
+ edges.add(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(i), new DoubleWritable(i * 2.0)));
+ }
+
+ vertex.initialize(null, null, edges);
+ assertEquals(vertex.getNumEdges(), 1000);
+ for (Edge<IntWritable, DoubleWritable> edge : vertex.getEdges()) {
+ assertEquals(edge.getValue().get(),
+ edge.getTargetVertexId().get() * 2.0d, 0d);
+ }
+ assertEquals(vertex.removeEdges(new IntWritable(500)), 1);
+ assertEquals(vertex.getNumEdges(), 999);
+ }
+
+ @Test
+ public void testGetEdges() {
+ for (Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass : vertexClasses) {
+ testGetEdgesVertexClass(vertexClass);
+ }
+ }
+
+ /**
+ * Test a vertex class for getting edges
+ *
+ * @param vertexClass Vertex class to check
+ */
+ private void testGetEdgesVertexClass(Class<? extends Vertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable>> vertexClass) {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ testInstantiateVertexClass(vertexClass);
+
+ List<Edge<IntWritable, DoubleWritable>> edges = Lists.newLinkedList();
+ for (int i = 1000; i > 0; --i) {
+ edges.add(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(i), new DoubleWritable(i * 3.0)));
+ }
+
+ vertex.initialize(null, null, edges);
+ assertEquals(vertex.getNumEdges(), 1000);
+ assertEquals(vertex.getEdgeValue(new IntWritable(600)),
+ new DoubleWritable(600 * 3.0));
+ assertEquals(vertex.removeEdges(new IntWritable(600)), 1);
+ assertEquals(vertex.getNumEdges(), 999);
+ assertEquals(vertex.getEdgeValue(new IntWritable(500)),
+ new DoubleWritable(500 * 3.0));
+ assertEquals(vertex.getEdgeValue(new IntWritable(700)),
+ new DoubleWritable(700 * 3.0));
+ }
+
+ @Test
+ public void testAddRemoveEdges() {
+ for (Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass : vertexClasses) {
+ testAddRemoveEdgesVertexClass(vertexClass);
+ }
+ }
+
+ /**
+ * Test a vertex class for adding/removing edges
+ *
+ * @param vertexClass Vertex class to check
+ */
+ private void testAddRemoveEdgesVertexClass(Class<? extends
+ Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass) {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ testInstantiateVertexClass(vertexClass);
+
+ vertex.initialize(new IntWritable(0), new FloatWritable(0.0f));
+ assertEquals(vertex.getNumEdges(), 0);
+ assertTrue(vertex.addEdge(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(2),
+ new DoubleWritable(2.0))));
+ assertEquals(vertex.getNumEdges(), 1);
+ assertEquals(vertex.getEdgeValue(new IntWritable(2)),
+ new DoubleWritable(2.0));
+ assertTrue(vertex.addEdge(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(4),
+ new DoubleWritable(4.0))));
+ assertTrue(vertex.addEdge(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(3),
+ new DoubleWritable(3.0))));
+ assertTrue(vertex.addEdge(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(1),
+ new DoubleWritable(1.0))));
+ assertEquals(vertex.getNumEdges(), 4);
+ assertNull(vertex.getEdgeValue(new IntWritable(5)));
+ assertNull(vertex.getEdgeValue(new IntWritable(0)));
+ for (Edge<IntWritable, DoubleWritable> edge : vertex.getEdges()) {
+ assertEquals(edge.getTargetVertexId().get() * 1.0d,
+ edge.getValue().get(), 0d);
+ }
+ assertEquals(vertex.removeEdges(new IntWritable(1)), 1);
+ assertEquals(vertex.getNumEdges(), 3);
+ assertEquals(vertex.removeEdges(new IntWritable(3)), 1);
+ assertEquals(vertex.getNumEdges(), 2);
+ assertEquals(vertex.removeEdges(new IntWritable(2)), 1);
+ assertEquals(vertex.getNumEdges(), 1);
+ assertEquals(vertex.removeEdges(new IntWritable(4)), 1);
+ assertEquals(vertex.getNumEdges(), 0);
+ }
+
+ @Test
+ public void testSerialized() throws IOException {
+ for (Class<? extends Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass : vertexClasses) {
+ testSerializeVertexClass(vertexClass);
+ testDynamicChannelBufferSerializeVertexClass(vertexClass);
+ testUnsafeSerializeVertexClass(vertexClass);
+ }
+ }
+
+ /**
+ * Build a vertex for testing
+ *
+ * @param vertexClass Vertex class to use for testing
+ * @return Vertex that has some initial data
+ */
+ private MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> buildVertex(Class<? extends
+ Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass) {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ testInstantiateVertexClass(vertexClass);
+
+ final int edgesCount = 200;
+ List<Edge<IntWritable, DoubleWritable>> edges =
+ Lists.newArrayListWithCapacity(edgesCount);
+ for (int i = edgesCount; i > 0; --i) {
+ edges.add(new Edge<IntWritable, DoubleWritable>(
+ new IntWritable(i), new DoubleWritable(i * 2.0)));
+ }
+ vertex.initialize(new IntWritable(2), new FloatWritable(3.0f), edges);
+ return vertex;
+ }
+
+ /**
+ * Test a vertex class for serializing
+ *
+ * @param vertexClass Vertex class to check
+ */
+ private void testSerializeVertexClass(Class<? extends
+ Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass) {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ buildVertex(vertexClass);
+
+ long serializeNanosStart = 0;
+ long serializeNanos = 0;
+ byte[] byteArray = null;
+ for (int i = 0; i < REPS; ++i) {
+ serializeNanosStart = SystemTime.get().getNanoseconds();
+ byteArray = WritableUtils.writeToByteArray(vertex);
+ serializeNanos += Times.getNanosecondsSince(SystemTime.get(), serializeNanosStart);
+ }
+ serializeNanos /= REPS;
+ System.out.println("testSerialize: Serializing took " +
+ serializeNanos +
+ " ns for " + byteArray.length + " bytes " +
+ (byteArray.length * 1f * Time.NS_PER_SECOND / serializeNanos) +
+ " bytes / sec for " + vertexClass.getName());
+
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> readVertex =
+ testInstantiateVertexClass(vertexClass);
+
+ long deserializeNanosStart = 0;
+ long deserializeNanos = 0;
+ for (int i = 0; i < REPS; ++i) {
+ deserializeNanosStart = SystemTime.get().getNanoseconds();
+ WritableUtils.readFieldsFromByteArray(byteArray, readVertex);
+ deserializeNanos += Times.getNanosecondsSince(SystemTime.get(),
+ deserializeNanosStart);
+ }
+ deserializeNanos /= REPS;
+ System.out.println("testSerialize: " +
+ "Deserializing " +
+ "took " +
+ deserializeNanos +
+ " ns for " + byteArray.length + " bytes " +
+ (byteArray.length * 1f * Time.NS_PER_SECOND / deserializeNanos) +
+ " bytes / sec for " + vertexClass.getName());
+
+ assertEquals(vertex.getId(), readVertex.getId());
+ assertEquals(vertex.getValue(), readVertex.getValue());
+ assertEquals(Lists.newArrayList(vertex.getEdges()),
+ Lists.newArrayList(readVertex.getEdges()));
+ }
+
+ /**
+ * Test a vertex class for serializing with DynamicChannelBuffers
+ *
+ * @param vertexClass Vertex class to check
+ */
+ private void testDynamicChannelBufferSerializeVertexClass(Class<? extends
+ Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass) throws IOException {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ buildVertex(vertexClass);
+
+ long serializeNanosStart = 0;
+ long serializeNanos = 0;
+ DynamicChannelBufferOutputStream outputStream = null;
+ for (int i = 0; i <
+ REPS; ++i) {
+ serializeNanosStart = SystemTime.get().getNanoseconds();
+ outputStream =
+ new DynamicChannelBufferOutputStream(32);
+ vertex.write(outputStream);
+ serializeNanos += Times.getNanosecondsSince(SystemTime.get(),
+ serializeNanosStart);
+ }
+ serializeNanos /= REPS;
+ System.out.println("testDynamicChannelBufferSerializeVertexClass: " +
+ "Serializing took " +
+ serializeNanos +
+ " ns for " + outputStream.getDynamicChannelBuffer().writerIndex()
+ + " bytes " +
+ (outputStream.getDynamicChannelBuffer().writerIndex() * 1f *
+ Time.NS_PER_SECOND / serializeNanos) +
+ " bytes / sec for " + vertexClass.getName());
+
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> readVertex =
+ testInstantiateVertexClass(vertexClass);
+
+ long deserializeNanosStart = 0;
+ long deserializeNanos = 0;
+ for (int i = 0; i < REPS; ++i) {
+ deserializeNanosStart = SystemTime.get().getNanoseconds();
+ DynamicChannelBufferInputStream inputStream = new
+ DynamicChannelBufferInputStream(
+ outputStream.getDynamicChannelBuffer());
+ readVertex.readFields(inputStream);
+ deserializeNanos += Times.getNanosecondsSince(SystemTime.get(),
+ deserializeNanosStart);
+ outputStream.getDynamicChannelBuffer().readerIndex(0);
+ }
+ deserializeNanos /= REPS;
+ System.out.println("testDynamicChannelBufferSerializeVertexClass: " +
+ "Deserializing took " +
+ deserializeNanos +
+ " ns for " + outputStream.getDynamicChannelBuffer().writerIndex() +
+ " bytes " +
+ (outputStream.getDynamicChannelBuffer().writerIndex() * 1f *
+ Time.NS_PER_SECOND / deserializeNanos) +
+ " bytes / sec for " + vertexClass.getName());
+
+ assertEquals(vertex.getId(), readVertex.getId());
+ assertEquals(vertex.getValue(), readVertex.getValue());
+ assertEquals(Lists.newArrayList(vertex.getEdges()),
+ Lists.newArrayList(readVertex.getEdges()));
+ }
+
+
+ /**
+ * Test a vertex class for serializing with UnsafeByteArray(Input/Output)
+ * Stream
+ *
+ * @param vertexClass Vertex class to check
+ */
+ private void testUnsafeSerializeVertexClass(Class<? extends
+ Vertex<IntWritable, FloatWritable, DoubleWritable,
+ LongWritable>> vertexClass) throws IOException {
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> vertex =
+ buildVertex(vertexClass);
+
+ long serializeNanosStart = 0;
+ long serializeNanos = 0;
+ UnsafeByteArrayOutputStream outputStream = null;
+ for (int i = 0; i <
+ REPS; ++i) {
+ serializeNanosStart = SystemTime.get().getNanoseconds();
+ outputStream =
+ new UnsafeByteArrayOutputStream(32);
+ vertex.write(outputStream);
+ serializeNanos += Times.getNanosecondsSince(SystemTime.get(),
+ serializeNanosStart);
+ }
+ serializeNanos /= REPS;
+ System.out.println("testUnsafeSerializeVertexClass: " +
+ "Serializing took " +
+ serializeNanos +
+ " ns for " + outputStream.getPos()
+ + " bytes " +
+ (outputStream.getPos() * 1f *
+ Time.NS_PER_SECOND / serializeNanos) +
+ " bytes / sec for " + vertexClass.getName());
+
+ MutableVertex<IntWritable,
+ FloatWritable, DoubleWritable, LongWritable> readVertex =
+ testInstantiateVertexClass(vertexClass);
+
+ long deserializeNanosStart = 0;
+ long deserializeNanos = 0;
+ for (int i = 0; i < REPS; ++i) {
+ deserializeNanosStart = SystemTime.get().getNanoseconds();
+ UnsafeByteArrayInputStream inputStream = new
+ UnsafeByteArrayInputStream(
+ outputStream.getByteArray(), 0, outputStream.getPos());
+ readVertex.readFields(inputStream);
+ deserializeNanos += Times.getNanosecondsSince(SystemTime.get(),
+ deserializeNanosStart);
+ }
+ deserializeNanos /= REPS;
+ System.out.println("testUnsafeSerializeVertexClass: " +
+ "Deserializing took " +
+ deserializeNanos +
+ " ns for " + outputStream.getPos() +
+ " bytes " +
+ (outputStream.getPos() * 1f *
+ Time.NS_PER_SECOND / deserializeNanos) +
+ " bytes / sec for " + vertexClass.getName());
+
+ assertEquals(vertex.getId(), readVertex.getId());
+ assertEquals(vertex.getValue(), readVertex.getValue());
+ assertEquals(Lists.newArrayList(vertex.getEdges()),
+ Lists.newArrayList(readVertex.getEdges()));
+ }
+}