You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ik...@apache.org on 2015/07/03 01:33:13 UTC
[1/2] git commit: updated refs/heads/trunk to 3b7c68e
Repository: giraph
Updated Branches:
refs/heads/trunk d7e4bde11 -> 3b7c68e54
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/TestMessageChain.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/TestMessageChain.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/TestMessageChain.java
new file mode 100644
index 0000000..0f5f1ac
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/TestMessageChain.java
@@ -0,0 +1,224 @@
+/*
+ * 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.block_app.library;
+
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.api.local.LocalBlockRunner;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.test_setup.NumericTestGraph;
+import org.apache.giraph.combiner.MaxMessageCombiner;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.GiraphConstants;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.function.primitive.PrimitiveRefs.LongRef;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.google.common.collect.Iterators;
+
+/**
+ * Tests and examples of using SendMessageChain
+ */
+public class TestMessageChain {
+
+ private static GiraphConfiguration createConf() {
+ GiraphConfiguration conf = new GiraphConfiguration();
+ GiraphConstants.VERTEX_ID_CLASS.set(conf, LongWritable.class);
+ GiraphConstants.VERTEX_VALUE_CLASS.set(conf, LongWritable.class);
+ GiraphConstants.EDGE_VALUE_CLASS.set(conf, NullWritable.class);
+ return conf;
+ }
+
+ private static NumericTestGraph<LongWritable, LongWritable, NullWritable> createTestGraph() {
+ NumericTestGraph<LongWritable, LongWritable, NullWritable> graph =
+ new NumericTestGraph<LongWritable, LongWritable, NullWritable>(createConf());
+ graph.addVertex(1);
+ graph.addVertex(2);
+ graph.addVertex(3);
+ graph.addVertex(4);
+
+ graph.addSymmetricEdge(1, 2);
+ graph.addSymmetricEdge(2, 3);
+ return graph;
+ }
+
+ @Test
+ public void testReply() {
+ NumericTestGraph<LongWritable, LongWritable, NullWritable> graph = createTestGraph();
+
+ // calculates max ID of FOFs
+ Block reply = SendMessageChain.<LongWritable, LongWritable, NullWritable, LongWritable>
+ startSendToNeighbors(
+ "SendMyIdToAllNeighbors",
+ LongWritable.class,
+ VertexSuppliers.vertexIdSupplier()
+ ).thenSendToNeighbors(
+ "SendMaxIReceivedToAllNeighbors",
+ LongWritable.class,
+ (vertex, messages) -> new LongWritable(max(messages))
+ ).endConsume(
+ (vertex, messages) -> vertex.getValue().set(max(messages))
+ );
+
+ LocalBlockRunner.runBlock(graph.getTestGraph(), reply, new Object());
+
+ Assert.assertEquals(3, graph.getVertex(1).getValue().get());
+ Assert.assertEquals(2, graph.getVertex(2).getValue().get());
+ Assert.assertEquals(3, graph.getVertex(3).getValue().get());
+ Assert.assertEquals(0, graph.getVertex(4).getValue().get());
+ }
+
+ @Test
+ public void testReplyCombiner() {
+ NumericTestGraph<LongWritable, LongWritable, NullWritable> graph = createTestGraph();
+
+ // calculates max ID of FOFs
+ Block reply = SendMessageChain.<LongWritable, LongWritable, NullWritable, LongWritable>
+ startSendToNeighbors(
+ "SendMyIdToAllNeighbors",
+ MaxMessageCombiner.LONG,
+ VertexSuppliers.vertexIdSupplier()
+ ).thenSendToNeighbors(
+ "SendMaxIReceivedToAllNeighbors",
+ MaxMessageCombiner.LONG,
+ (vertex, message) -> message
+ ).endConsume(
+ (vertex, message) -> vertex.getValue().set(message != null ? message.get() : 0)
+ );
+
+ LocalBlockRunner.runBlock(graph.getTestGraph(), reply, new Object());
+
+ Assert.assertEquals(3, graph.getVertex(1).getValue().get());
+ Assert.assertEquals(2, graph.getVertex(2).getValue().get());
+ Assert.assertEquals(3, graph.getVertex(3).getValue().get());
+ Assert.assertEquals(0, graph.getVertex(4).getValue().get());
+ }
+
+ @Test
+ public void testReplyCombinerEndReduce() {
+ NumericTestGraph<LongWritable, LongWritable, NullWritable> graph = createTestGraph();
+
+ LongRef sumOfAll = new LongRef(0);
+
+ // calculates max ID of FOFs
+ Block reply = SendMessageChain.<LongWritable, LongWritable, NullWritable, LongWritable>
+ startSendToNeighbors(
+ "SendMyIdToAllNeighbors",
+ MaxMessageCombiner.LONG,
+ VertexSuppliers.vertexIdSupplier()
+ ).thenSendToNeighbors(
+ "SendMaxIReceivedToAllNeighbors",
+ MaxMessageCombiner.LONG,
+ (vertex, message) -> message
+ ).endReduce(
+ "SumAllReceivedValues",
+ SumReduce.LONG,
+ (vertex, message) -> message != null ? message : new LongWritable(0),
+ (value) -> sumOfAll.value = value.get()
+ );
+
+ LocalBlockRunner.runBlock(
+ graph.getTestGraph(),
+ new SequenceBlock(
+ reply,
+ Pieces.forAllVertices(
+ "SetAllValuesToReduced",
+ (vertex) -> ((LongWritable) vertex.getValue()).set(sumOfAll.value))),
+ new Object());
+
+ Assert.assertEquals(8, graph.getVertex(1).getValue().get());
+ Assert.assertEquals(8, graph.getVertex(2).getValue().get());
+ Assert.assertEquals(8, graph.getVertex(3).getValue().get());
+ Assert.assertEquals(8, graph.getVertex(4).getValue().get());
+
+ // Block execution is happening in the separate environment if SERIALIZE_MASTER is used,
+ // so our instance of sumOfAll will be unchanged
+ Assert.assertEquals(LocalBlockRunner.SERIALIZE_MASTER.getDefaultValue() ? 0 : 8, sumOfAll.value);
+ }
+
+
+ @Test
+ public void testStartCustom() {
+ NumericTestGraph<LongWritable, LongWritable, NullWritable> graph = createTestGraph();
+
+ Block reply = SendMessageChain.<LongWritable, LongWritable, NullWritable, LongWritable>
+ startCustom(
+ // Sends ID to it's first neighbor, passing max of received messages to next part of the chain
+ (consumer) -> new Piece<LongWritable, LongWritable, NullWritable, LongWritable, Object>() {
+ @Override
+ public VertexSender<LongWritable, LongWritable, NullWritable> getVertexSender(
+ BlockWorkerSendApi<LongWritable, LongWritable, NullWritable, LongWritable> workerApi,
+ Object executionStage) {
+ return (vertex) -> {
+ Edge<LongWritable, NullWritable> edge =
+ Iterators.getNext(vertex.getEdges().iterator(), null);
+ if (edge != null) {
+ workerApi.sendMessage(edge.getTargetVertexId(), vertex.getId());
+ }
+ };
+ }
+
+ @Override
+ public VertexReceiver<LongWritable, LongWritable, NullWritable, LongWritable>
+ getVertexReceiver(BlockWorkerReceiveApi<LongWritable> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ consumer.apply(vertex, new LongWritable(max(messages)));
+ };
+ }
+
+ @Override
+ protected Class<LongWritable> getMessageClass() {
+ return LongWritable.class;
+ }
+ }
+ ).thenSendToNeighbors(
+ "SendMaxIReceivedToAllNeighbors",
+ SumMessageCombiner.LONG,
+ (vertex, message) -> message
+ ).endConsume(
+ (vertex, message) -> vertex.getValue().set(message != null ? message.get() : 0)
+ );
+
+ LocalBlockRunner.runBlock(graph.getTestGraph(), reply, new Object());
+
+ Assert.assertEquals(3, graph.getVertex(1).getValue().get());
+ Assert.assertEquals(2, graph.getVertex(2).getValue().get());
+ Assert.assertEquals(3, graph.getVertex(3).getValue().get());
+ Assert.assertEquals(0, graph.getVertex(4).getValue().get());
+ }
+
+
+
+
+ private static long max(Iterable<LongWritable> messages) {
+ long result = 0;
+ for (LongWritable message : messages) {
+ result = Math.max(result, message.get());
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchBlockFactory.java
new file mode 100644
index 0000000..e16313a
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchBlockFactory.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.block_app.library.algo;
+
+import org.apache.giraph.block_app.framework.AbstractBlockFactory;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Example Application of BFS calculation
+ */
+public class BreadthFirstSearchBlockFactory extends AbstractBlockFactory<Object> {
+ @Override
+ public Block createBlock(GiraphConfiguration conf) {
+ SupplierFromVertex<LongWritable, BreadthFirstSearchVertexValue, Writable, Boolean>
+ isVertexInSeedSet =
+ (vertex) -> {
+ return vertex.getValue().isSeedVertex();
+ };
+
+ SupplierFromVertex<LongWritable, BreadthFirstSearchVertexValue, Writable, IntWritable>
+ getDistance =
+ (vertex) -> {
+ return new IntWritable(vertex.getValue().getDistance());
+ };
+
+ ConsumerWithVertex<LongWritable, BreadthFirstSearchVertexValue, Writable, IntWritable>
+ setDistance =
+ (vertex, value) -> {
+ vertex.getValue().setDistance(value.get());
+ };
+
+ return BreadthFirstSearch.bfs(
+ isVertexInSeedSet,
+ getDistance,
+ setDistance);
+ }
+
+ @Override
+ public Object createExecutionStage(GiraphConfiguration conf) {
+ return new Object();
+ }
+
+ @Override
+ protected Class<LongWritable> getVertexIDClass(GiraphConfiguration conf) {
+ return LongWritable.class;
+ }
+
+ @Override
+ protected Class<BreadthFirstSearchVertexValue> getVertexValueClass(GiraphConfiguration conf) {
+ return BreadthFirstSearchVertexValue.class;
+ }
+
+ @Override
+ protected Class<NullWritable> getEdgeValueClass(GiraphConfiguration conf) {
+ return NullWritable.class;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchVertexValue.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchVertexValue.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchVertexValue.java
new file mode 100644
index 0000000..88b78e5
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearchVertexValue.java
@@ -0,0 +1,61 @@
+/*
+ * 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.block_app.library.algo;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Vertex-value class for BreadthFirstSearchBlockFactory
+ */
+public class BreadthFirstSearchVertexValue implements Writable {
+ private boolean seedVertex;
+ private int distance;
+
+ public boolean isSeedVertex() {
+ return seedVertex;
+ }
+
+ public void setSeedVertex(boolean seedVertex) {
+ this.seedVertex = seedVertex;
+ }
+
+ public int getDistance() {
+ return distance;
+ }
+
+ public void setDistance(int distance) {
+ this.distance = distance;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeBoolean(seedVertex);
+ out.writeInt(distance);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ seedVertex = in.readBoolean();
+ distance = in.readInt();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetBlockFactory.java
new file mode 100644
index 0000000..76c9f89
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetBlockFactory.java
@@ -0,0 +1,60 @@
+/*
+ * 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.block_app.library.algo;
+
+import org.apache.giraph.block_app.framework.AbstractBlockFactory;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+
+/**
+ * Example Application of distributed independent set calculation
+ */
+public class DistributedIndependentSetBlockFactory extends AbstractBlockFactory<Object> {
+ @Override
+ public Block createBlock(GiraphConfiguration conf) {
+ return DistributedIndependentSet.
+ <LongWritable, DistributedIndependentSetVertexValue>independentSets(
+ getVertexIDClass(conf),
+ (vertex) -> vertex.getValue().getIndependentSetID(),
+ (vertex, value) -> vertex.getValue().setIndependentSetID(value)
+ );
+ }
+
+ @Override
+ public Object createExecutionStage(GiraphConfiguration conf) {
+ return new Object();
+ }
+
+ @Override
+ protected Class<LongWritable> getVertexIDClass(GiraphConfiguration conf) {
+ return LongWritable.class;
+ }
+
+ @Override
+ protected Class<DistributedIndependentSetVertexValue> getVertexValueClass(
+ GiraphConfiguration conf) {
+ return DistributedIndependentSetVertexValue.class;
+ }
+
+ @Override
+ protected Class<NullWritable> getEdgeValueClass(GiraphConfiguration conf) {
+ return NullWritable.class;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetVertexValue.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetVertexValue.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetVertexValue.java
new file mode 100644
index 0000000..d29ee00
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSetVertexValue.java
@@ -0,0 +1,56 @@
+/*
+ * 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.block_app.library.algo;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Vertex-value class for DistributedIndependentSetBlockFactory
+ */
+public class DistributedIndependentSetVertexValue implements Writable {
+ private IntWritable independentSetID;
+
+ public IntWritable getIndependentSetID() {
+ return independentSetID;
+ }
+
+ public void setIndependentSetID(IntWritable independentSetID) {
+ this.independentSetID = independentSetID;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ if (independentSetID == null) {
+ independentSetID = new IntWritable();
+ }
+ out.writeInt(independentSetID.get());
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ if (independentSetID == null) {
+ independentSetID = new IntWritable();
+ }
+ independentSetID.set(in.readInt());
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestBreadthFirstSearch.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestBreadthFirstSearch.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestBreadthFirstSearch.java
new file mode 100644
index 0000000..c19fe27
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestBreadthFirstSearch.java
@@ -0,0 +1,165 @@
+/*
+ * 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.block_app.library.algo;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.giraph.block_app.framework.BlockUtils;
+import org.apache.giraph.block_app.test_setup.NumericTestGraph;
+import org.apache.giraph.block_app.test_setup.TestGraphModifier;
+import org.apache.giraph.block_app.test_setup.TestGraphUtils;
+import org.apache.giraph.block_app.test_setup.graphs.Small1GraphInit;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.junit.Test;
+
+public class TestBreadthFirstSearch {
+ private void run(
+ TestGraphModifier<LongWritable, BreadthFirstSearchVertexValue, NullWritable> graphLoader,
+ int[] expectedDistances,
+ int[] seedVertices
+ ) throws Exception {
+ TestGraphModifier<LongWritable, BreadthFirstSearchVertexValue, NullWritable> valueLoader =
+ (graph) -> {
+ List<Integer> seeds = Arrays.asList(ArrayUtils.toObject(seedVertices));
+ for (int i = 0; i < graph.getVertexCount(); i++)
+ graph.getVertex(i).getValue().setSeedVertex(seeds.contains(i));
+ };
+
+ TestGraphUtils.runTest(
+ TestGraphUtils.chainModifiers(graphLoader, valueLoader),
+ (graph) -> {
+ for (int i = 0; i < expectedDistances.length; i++) {
+ assertEquals(expectedDistances[i], graph.getValue(i).getDistance());
+ }
+ },
+ (conf) -> {
+ BlockUtils.setBlockFactoryClass(conf, BreadthFirstSearchBlockFactory.class);
+ }
+ );
+ }
+
+ @Test
+ public void testSmall1SingleSeed() throws Exception {
+ int[] expected = {0, 1, 1, 2, 3, 3, -1};
+ int[] seeds = {0};
+ run(new Small1GraphInit<>(), expected, seeds);
+ }
+
+ @Test
+ public void testSmall1TwoSeeds() throws Exception {
+ int[] expected = {0, 1, 1, 1, 0, 1, -1};
+ int[] seeds = {0, 4};
+ run(new Small1GraphInit<>(), expected, seeds);
+ }
+
+ @Test
+ public void testSmall1IsolatedSeed() throws Exception {
+ int[] expected = {-1, -1, -1, -1, -1, -1, 0};
+ int[] seeds = {6};
+ run(new Small1GraphInit<>(), expected, seeds);
+ }
+
+ @Test
+ public void testSmallGraphTwoSeeds() throws Exception {
+ int[] expected = {0, 1, 2, 2, 2, 2, 3, 4, 5, 5, 5, 1, 2, 2, 2, 0};
+ int[] seeds = {0, 15};
+ run(new Graph1Init<>(), expected, seeds);
+ }
+
+ @Test
+ public void testSmallGraphTwoCloseSeeds() throws Exception {
+ int[] expected = {1, 0, 1, 0, 1, 1, 1, 2, 3, 3, 3, 2, 3, 3, 3, 3};
+ int[] seeds = {1, 3};
+ run(new Graph1Init<>(), expected, seeds);
+ }
+
+ @Test
+ public void testMultipleComponentGraphCloseSeeds() throws Exception {
+ int[] expected = {2, 1, 0, 1, 2, 3, 3, 3, 2, 2, 2, 2, 1, 0, 2, -1, -1, -1, -1, -1, -1};
+ int[] seeds = {13, 2};
+ run(new Graph2Init(), expected, seeds);
+ }
+
+ @Test
+ public void testMultipleComponentGraphFarSeeds() throws Exception {
+ int[] expected = {3, 2, 3, 2, 1, 0, 1, 2, 1, 2, 3, 3, 2, 3, 3, 3, 2, 1, 0, 1, -1};
+ int[] seeds = {5, 18};
+ run(new Graph2Init(), expected, seeds);
+ }
+
+
+ public class Graph1Init<I extends WritableComparable, V extends Writable, E extends Writable>
+ implements TestGraphModifier<I, V, E> {
+
+ @Override
+ public void modifyGraph(NumericTestGraph<I, V, E> graph) {
+ graph.addVertex(0, (Number) null, null, 1);
+ graph.addVertex(1, (Number) null, null, 0,2,3,4,5);
+ graph.addVertex(2, (Number) null, null, 1,3,4,5);
+ graph.addVertex(3, (Number) null, null, 1,2,4,5,6);
+ graph.addVertex(4, (Number) null, null, 1,2,3,5);
+ graph.addVertex(5, (Number) null, null, 1,2,3,4,11);
+ graph.addVertex(6, (Number) null, null, 3,7);
+ graph.addVertex(7, (Number) null, null, 6,8,9,10);
+ graph.addVertex(8, (Number) null, null, 7,9,10);
+ graph.addVertex(9, (Number) null, null, 7,8,10);
+ graph.addVertex(10, (Number) null, null, 7,8,9);
+ graph.addVertex(11, (Number) null, null, 5,12,13,14,15);
+ graph.addVertex(12, (Number) null, null, 11);
+ graph.addVertex(13, (Number) null, null, 11);
+ graph.addVertex(14, (Number) null, null, 11);
+ graph.addVertex(15, (Number) null, null, 11);
+ }
+ }
+
+ public class Graph2Init<I extends WritableComparable, V extends Writable, E extends Writable>
+ implements TestGraphModifier<I, V, E> {
+
+ @Override
+ public void modifyGraph(NumericTestGraph<I, V, E> graph) {
+ graph.addVertex(0, (Number) null, null, 1);
+ graph.addVertex(1, (Number) null, null, 0,2,3,4);
+ graph.addVertex(2, (Number) null, null, 1,3);
+ graph.addVertex(3, (Number) null, null, 2,4,9,10,11);
+ graph.addVertex(4, (Number) null, null, 1,3,5,6,7);
+ graph.addVertex(5, (Number) null, null, 4,6,5,8);
+ graph.addVertex(6, (Number) null, null, 4,5,7);
+ graph.addVertex(7, (Number) null, null, 4,5,6);
+ graph.addVertex(8, (Number) null, null, 5,9,12);
+ graph.addVertex(9, (Number) null, null, 3,8,10,11,12);
+ graph.addVertex(10, (Number) null, null, 3,9,11);
+ graph.addVertex(11, (Number) null, null, 3,9,10);
+ graph.addVertex(12, (Number) null, null, 8,9,13,14);
+ graph.addVertex(13, (Number) null, null, 12);
+ graph.addVertex(14, (Number) null, null, 12);
+ graph.addVertex(15, (Number) null, null, 16);
+ graph.addVertex(16, (Number) null, null, 15,17,19);
+ graph.addVertex(17, (Number) null, null, 16,18);
+ graph.addVertex(18, (Number) null, null, 17,19);
+ graph.addVertex(19, (Number) null, null, 16,18);
+ graph.addVertex(20);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestDistributedIndependentSet.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestDistributedIndependentSet.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestDistributedIndependentSet.java
new file mode 100644
index 0000000..fbd380b
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/algo/TestDistributedIndependentSet.java
@@ -0,0 +1,220 @@
+/*
+ * 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.block_app.library.algo;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+
+import org.apache.giraph.block_app.framework.BlockUtils;
+import org.apache.giraph.block_app.test_setup.NumericTestGraph;
+import org.apache.giraph.block_app.test_setup.TestGraphModifier;
+import org.apache.giraph.block_app.test_setup.TestGraphUtils;
+import org.apache.giraph.edge.Edge;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestDistributedIndependentSet {
+
+ private void runTest(
+ TestGraphModifier<LongWritable, DistributedIndependentSetVertexValue, NullWritable>
+ graphLoader) throws Exception {
+ TestGraphUtils.runTest(graphLoader, (graph) -> {
+ checkDecomposition(graph);
+ }, (conf) -> {
+ BlockUtils.setBlockFactoryClass(conf, DistributedIndependentSetBlockFactory.class);
+ });
+ }
+
+ private static void checkDecomposition(
+ NumericTestGraph<LongWritable, DistributedIndependentSetVertexValue, NullWritable> graph) {
+ final int UNASSIGNED = -1;
+ final int HAS_EDGE_TO_IND_SET = -2;
+ // Hold (id -> List of vertices) for each independent set id
+ HashMap<Integer, ArrayList<Integer>> indSets = new HashMap<>();
+ int numVertices = graph.getVertexCount();
+
+ int numIndSets = 0;
+ for (int i = 0; i < numVertices; ++i) {
+ int indSetID = graph.getVertex(i).getValue().getIndependentSetID().get();
+ // Number of independent sets are less than or equal to the number of vertices in the input
+ // graph.
+ Assert.assertTrue(indSetID >= 0 && indSetID < numVertices);
+ // All tests assign (0, 1, ...) as independent set ids. The vertex assigned to the max id
+ // is a specifier for the total number of independent sets the input graph is decomposed
+ // into.
+ if (indSetID > numIndSets)
+ numIndSets = indSetID;
+ ArrayList<Integer> mapValue = indSets.get(indSetID);
+ if (mapValue == null) {
+ mapValue = new ArrayList<>();
+ indSets.put(indSetID, mapValue);
+ }
+ mapValue.add(i);
+ }
+ numIndSets++;
+
+ int[] label = new int[numVertices];
+ for (int i = 0; i < numVertices; ++i)
+ label[i] = UNASSIGNED;
+
+ for (int i = 0; i < numIndSets; ++i) {
+ ArrayList<Integer> indSet = indSets.get(i);
+
+ // All independent set ids are assigned consecutively starting from 0. There should be at
+ // least one vertex per assigned independent set.
+ Assert.assertTrue(indSet != null && indSet.size() > 0);
+ for (int v : indSet) {
+ // Check if vertices in this independent set is not assigned to any of the previous
+ // independent sets.
+ Assert.assertTrue(label[v] == UNASSIGNED);
+ label[v] = i;
+ }
+
+ for (int v : indSet) {
+ for (Edge<LongWritable, NullWritable> edge : graph.getVertex(v).getEdges()) {
+ int u = (int) edge.getTargetVertexId().get();
+ // Check all vertices in the current independent set do not have edge to each other.
+ Assert.assertTrue(label[u] != label[v]);
+ // Mark unassigned vertices neighboring this independent set. This is necessary to check
+ // if this independent set is 'maximal'.
+ if (label[u] == UNASSIGNED)
+ label[u] = HAS_EDGE_TO_IND_SET;
+ }
+ }
+
+ // Check if the independent set is maximal.
+ for (int j = 0; j < numVertices; ++j) {
+ Assert.assertTrue(label[j] != UNASSIGNED);
+ // Reset marked vertices neighboring to this independent set.
+ if (label[j] == HAS_EDGE_TO_IND_SET)
+ label[j] = UNASSIGNED;
+ }
+ }
+ }
+
+ private static void createVertices(
+ NumericTestGraph<LongWritable, DistributedIndependentSetVertexValue, NullWritable> graph,
+ int numVertices) {
+ for (int i = 0; i < numVertices; ++i)
+ graph.addVertex(i);
+ }
+
+ @Test
+ public void testSmallGraph() throws Exception {
+ /*
+ * 1 5
+ * / \ / \ 6
+ * 0---2--3---4
+ */
+ final int NUM_VERTICES = 7;
+ runTest((graph) -> {
+ createVertices(graph, NUM_VERTICES);
+ graph.addSymmetricEdge(0, 1);
+ graph.addSymmetricEdge(0, 2);
+ graph.addSymmetricEdge(1, 2);
+ graph.addSymmetricEdge(2, 3);
+ graph.addSymmetricEdge(3, 4);
+ graph.addSymmetricEdge(3, 5);
+ graph.addSymmetricEdge(4, 5);
+ });
+ }
+
+ @Test
+ public void testSmallGraphOrderingEffect() throws Exception {
+ /*
+ * 4 5
+ * / \ / \ 6
+ * 0---2--1---3
+ */
+ final int NUM_VERTICES = 7;
+ runTest((graph) -> {
+ createVertices(graph, NUM_VERTICES);
+ graph.addSymmetricEdge(0, 4);
+ graph.addSymmetricEdge(0, 2);
+ graph.addSymmetricEdge(2, 4);
+ graph.addSymmetricEdge(1, 2);
+ graph.addSymmetricEdge(1, 5);
+ graph.addSymmetricEdge(1, 3);
+ graph.addSymmetricEdge(3, 5);
+ });
+ }
+
+ @Test
+ public void testRingOdd() throws Exception {
+ final int NUM_VERTICES = 13;
+ runTest((graph) -> {
+ createVertices(graph, NUM_VERTICES);
+ for (int i = 1; i < NUM_VERTICES; ++i)
+ graph.addSymmetricEdge(i - 1, i);
+ graph.addSymmetricEdge(0, NUM_VERTICES - 1);
+ });
+ }
+
+ @Test
+ public void testStarGraph() throws Exception {
+ final int NUM_VERTICES = 15;
+ runTest((graph) -> {
+ createVertices(graph, NUM_VERTICES);
+ for (int i = 1; i < NUM_VERTICES; ++i)
+ graph.addSymmetricEdge(0, i);
+ });
+ }
+
+ @Test
+ public void testMultipleStarGraphs() throws Exception {
+ final int NUM_VERTICES1 = 15;
+ final int NUM_VERTICES2 = 21;
+ runTest((graph) -> {
+ createVertices(graph, NUM_VERTICES1 + NUM_VERTICES2);
+ for (int i = 1; i < NUM_VERTICES1; ++i)
+ graph.addSymmetricEdge(0, i);
+
+ for (int i = 1 + NUM_VERTICES1; i < NUM_VERTICES1 + NUM_VERTICES2; ++i)
+ graph.addSymmetricEdge(NUM_VERTICES1, i);
+ });
+ }
+
+ @Test
+ public void testMeshGraph() throws Exception {
+ final int M = 11;
+ final int N = 13;
+ runTest((graph) -> {
+ createVertices(graph, M * N);
+ for (int i = 0; i < M; ++i)
+ for (int j = 0; j < N; ++j) {
+ if (i != M - 1)
+ graph.addSymmetricEdge(i * N + j, (i + 1) * N + j);
+ if (j != N - 1)
+ graph.addSymmetricEdge(i * N + j, i * N + (j + 1));
+ }
+ });
+ }
+
+ @Test
+ public void testCompleteGraph() throws Exception {
+ final int NUM_VERTICES = 17;
+ runTest((graph) -> {
+ createVertices(graph, NUM_VERTICES);
+ for (int i = 0; i < NUM_VERTICES; ++i)
+ for (int j = i + 1; j < NUM_VERTICES; ++j)
+ graph.addSymmetricEdge(i, j);
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/coarsening/TestCoarseningUtils.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/coarsening/TestCoarseningUtils.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/coarsening/TestCoarseningUtils.java
new file mode 100644
index 0000000..1f5ece6
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/library/coarsening/TestCoarseningUtils.java
@@ -0,0 +1,133 @@
+/*
+ * 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.block_app.library.coarsening;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.giraph.block_app.framework.AbstractBlockFactory;
+import org.apache.giraph.block_app.framework.BlockUtils;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.library.ReusableSuppliers;
+import org.apache.giraph.block_app.test_setup.TestGraphUtils;
+import org.apache.giraph.block_app.test_setup.graphs.EachVertexInit;
+import org.apache.giraph.block_app.test_setup.graphs.Small1GraphInit;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.edge.Edge;
+import org.apache.hadoop.io.IntWritable;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestCoarseningUtils {
+
+ public static class TestCoarseningUtilsBlockFactory extends AbstractBlockFactory<Object> {
+ @Override
+ public Block createBlock(GiraphConfiguration conf) {
+ return CoarseningUtils.<IntWritable, IntWritable, IntWritable>createCoarseningBlock(
+ (ImmutableClassesGiraphConfiguration<IntWritable, IntWritable, IntWritable>) conf,
+ ReusableSuppliers.fromInt((vertex) -> {
+ int id = vertex.getId().get();
+ if (id == 0 || id == 1) {
+ return -1;
+ } else if (id == 2 || id == 3) {
+ return -2;
+ } else if (id == 4 || id == 5) {
+ return -4;
+ } else return -id;
+ }),
+ id -> id.get() >= 0,
+ id -> id.get() < 0);
+ }
+
+ @Override
+ public Object createExecutionStage(GiraphConfiguration conf) {
+ return new Object();
+ }
+
+ @Override
+ protected Class<IntWritable> getVertexIDClass(GiraphConfiguration conf) {
+ return IntWritable.class;
+ }
+
+ @Override
+ protected Class<IntWritable> getVertexValueClass(GiraphConfiguration conf) {
+ return IntWritable.class;
+ }
+
+ @Override
+ protected Class<IntWritable> getEdgeValueClass(GiraphConfiguration conf) {
+ return IntWritable.class;
+ }
+ }
+
+ @Test
+ public void testSmallGraph() throws Exception {
+ /* We take small graph:
+ *
+ * 1 5
+ * / \ / \ 6
+ * 0---2--3---4
+ *
+ * And coarsen it into 4 groups: (0,1), (2,3), (4,5), (6)
+ */
+
+ TestGraphUtils.<IntWritable, IntWritable, IntWritable>runTest(
+ TestGraphUtils.chainModifiers(
+ new Small1GraphInit<>(() -> new IntWritable(1)),
+ new EachVertexInit<>((vertex) -> vertex.getValue().set(1))),
+ (graph) -> {
+ for (int i : new int[] {-1, -4}) {
+ Assert.assertEquals(2, graph.getValue(i).get());
+ Assert.assertEquals(2, graph.getVertex(i).getNumEdges());
+
+ Map<Integer, Integer> edges = edgesToMap(graph.getVertex(i).getEdges());
+ // self loop
+ Assert.assertEquals(2, edges.get(i).intValue());
+
+ Assert.assertEquals(2, edges.get(-2).intValue());
+ }
+ for (int i : new int[] {-2}) {
+ Assert.assertEquals(2, graph.getValue(i).get());
+ Assert.assertEquals(3, graph.getVertex(i).getNumEdges());
+
+ Map<Integer, Integer> edges = edgesToMap(graph.getVertex(i).getEdges());
+ // self loop
+ Assert.assertEquals(2, edges.get(i).intValue());
+
+ Assert.assertEquals(2, edges.get(-1).intValue());
+ Assert.assertEquals(2, edges.get(-4).intValue());
+ }
+ for (int i : new int[] {-6}) {
+ Assert.assertEquals(1, graph.getValue(i).get());
+ Assert.assertEquals(0, graph.getVertex(i).getNumEdges());
+ }
+ },
+ (conf) -> {
+ BlockUtils.setBlockFactoryClass(conf, TestCoarseningUtilsBlockFactory.class);
+ });
+ }
+
+ private static Map<Integer, Integer> edgesToMap(Iterable<Edge<IntWritable, IntWritable>> edges) {
+ Map<Integer, Integer> map = new HashMap<>();
+ for(Edge<IntWritable, IntWritable> edge : edges) {
+ Assert.assertNull(map.put(edge.getTargetVertexId().get(), edge.getValue().get()));
+ }
+ return map;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/writable/kryo/KryoWritableWrapperJava8Test.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/writable/kryo/KryoWritableWrapperJava8Test.java b/giraph-block-app-8/src/test/java/org/apache/giraph/writable/kryo/KryoWritableWrapperJava8Test.java
new file mode 100644
index 0000000..d8b4cc1
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/writable/kryo/KryoWritableWrapperJava8Test.java
@@ -0,0 +1,174 @@
+package org.apache.giraph.writable.kryo;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Iterator;
+import java.util.Random;
+
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.piece.AbstractPiece;
+import org.apache.giraph.block_app.library.striping.StripingUtils;
+import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.Predicate;
+import org.apache.giraph.function.Supplier;
+import org.apache.giraph.function.primitive.Int2ObjFunction;
+import org.apache.giraph.function.primitive.Obj2IntFunction;
+import org.apache.giraph.utils.WritableUtils;
+import org.apache.hadoop.io.LongWritable;
+import org.junit.Assert;
+import org.junit.Test;
+
+
+public class KryoWritableWrapperJava8Test {
+ // Copy from KryoWritableWrapperTest, since we cannot extend it, since tests are not in jars
+ public static <T> T kryoSerDeser(T t) throws IOException {
+ KryoWritableWrapper<T> wrapped = new KryoWritableWrapper<>(t);
+ KryoWritableWrapper<T> deser = new KryoWritableWrapper<>();
+ WritableUtils.copyInto(wrapped, deser, true);
+ return deser.get();
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void testNonSerializableLambda() throws IOException {
+ Runnable nonCapturing = () -> System.out.println("works");
+ kryoSerDeser(nonCapturing).run();
+ }
+
+ @Test
+ public void testLambda() throws IOException {
+ Runnable nonCapturing = (Runnable & Serializable) () -> System.out.println("works");
+ kryoSerDeser(nonCapturing).run();
+
+ String works = "works";
+ Runnable capturing = (Runnable & Serializable) () -> System.out.println(works);
+ kryoSerDeser(capturing).run();
+ }
+
+ @Test
+ public void testLambdaCapturingThisRef() throws IOException {
+ KryoWritableWrapperJava8Test o = this;
+ Runnable capturingThisRef = (Runnable & Serializable) () ->
+ System.out.println(o);
+ kryoSerDeser(capturingThisRef).run();
+ }
+
+ @Test
+ public void testLambdaCapturingThis() throws IOException {
+ Runnable capturingThis = (Runnable & Serializable) () ->
+ System.out.println(this);
+ kryoSerDeser(capturingThis).run();
+ }
+
+ @Test
+ public void testLambdaCapturingLambda() throws IOException {
+ Supplier<Boolean> nonCapturing = () -> true;
+ Runnable capturingLambda = (Runnable & Serializable) () ->
+ System.out.println(nonCapturing);
+ kryoSerDeser(capturingLambda).run();
+ }
+
+ @Test
+ public void testLambdaCapturingLambdaWithCapture() throws IOException {
+ boolean trueVar = new Random().nextDouble() < 1;
+ Supplier<Boolean> capturing = () -> trueVar;
+ Runnable capturingLambda = (Runnable & Serializable) () ->
+ System.out.println(capturing + " " + trueVar);
+ kryoSerDeser(capturingLambda).run();
+ }
+
+
+ @Test
+ public void testLambdaFunctions() throws IOException {
+ Supplier<Boolean> nonCapturing = () -> true;
+ Assert.assertTrue(kryoSerDeser(nonCapturing).get());
+
+ boolean trueVar = new Random().nextDouble() < 1;
+ Supplier<Boolean> capturing = () -> trueVar;
+ Assert.assertTrue(kryoSerDeser(capturing).get());
+ }
+
+ @Test
+ public void testLambdasFromCode() throws IOException {
+ Assert.assertNotNull(kryoSerDeser(StripingUtils.fastHashStriping(3)));
+ Assert.assertNotNull(kryoSerDeser(StripingUtils.fastHashStripingPredicate(3)));
+
+ Assert.assertNotNull(kryoSerDeser(
+ (Int2ObjFunction<Obj2IntFunction<LongWritable>>) StripingUtils::fastHashStriping));
+
+ Int2ObjFunction<Int2ObjFunction<Predicate<LongWritable>>> stripingPredicate = StripingUtils::fastHashStripingPredicate;
+ Assert.assertNotNull(kryoSerDeser(stripingPredicate));
+
+ Assert.assertNotNull(kryoSerDeser(stripingPredicate.apply(3)));
+
+ Assert.assertNotNull(kryoSerDeser(StripingUtils.fastHashStripingPredicate(3).apply(2)));
+
+ Assert.assertNotNull(kryoSerDeser(stripingPredicate.apply(3).apply(2)));
+
+ Predicate<LongWritable> predicate = stripingPredicate.apply(3).apply(2);
+ Assert.assertNotNull(kryoSerDeser(predicate));
+
+ Runnable capturingLambda = (Runnable & Serializable) () ->
+ System.out.println(predicate.apply(new LongWritable()));
+ kryoSerDeser(capturingLambda).run();
+
+ Assert.assertNotNull(kryoSerDeser(
+ StripingUtils.generateStripedBlock(
+ 5,
+ (filter) -> new Block() {
+ private final Predicate<LongWritable> test = filter;
+ {
+ Assert.assertTrue(filter instanceof Serializable);
+ }
+
+ @Override
+ public Iterator<AbstractPiece> iterator() {
+ return null;
+ }
+
+ @Override
+ public void forAllPossiblePieces(Consumer<AbstractPiece> consumer) { }
+ })));
+ }
+
+ @Test
+ public void testLambdaCapturingSameReference() throws IOException {
+ ObjectTransfer<Integer> transfer = new ObjectTransfer<>();
+
+ Consumer<Integer> consumer = (t) -> transfer.apply(t);
+ Supplier<Integer> supplier = () -> transfer.get();
+
+ class TwoLambdaObject {
+ Consumer<Integer> consumer;
+ Supplier<Integer> supplier;
+
+ public TwoLambdaObject(Consumer<Integer> consumer, Supplier<Integer> supplier) {
+ this.consumer = consumer;
+ this.supplier = supplier;
+ }
+ }
+
+ TwoLambdaObject object = new TwoLambdaObject(consumer, supplier);
+ // test transfer before serialization
+ object.consumer.apply(5);
+ Assert.assertEquals(5, object.supplier.get().intValue());
+
+ // test transfer through serialization
+ object.consumer.apply(6);
+ TwoLambdaObject deser = kryoSerDeser(object);
+ Assert.assertEquals(6, deser.supplier.get().intValue());
+
+ // test that after serialization, both lambdas point to the same object
+ deser.consumer.apply(4);
+ Assert.assertEquals(4, deser.supplier.get().intValue());
+ }
+
+ // Bug in Java, have test to know when it becomes fixed
+ @Test //(expected=RuntimeException.class)
+ public void testNestedLambda() throws IOException {
+ Int2ObjFunction<Int2ObjFunction<Integer>> f = (x) -> (y) -> x+y;
+ Assert.assertNotNull(kryoSerDeser(f));
+ Assert.assertNotNull(kryoSerDeser(f.apply(0)));
+ Assert.assertNotNull(kryoSerDeser(f.apply(0).apply(1)));
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/block_app/framework/api/local/LocalBlockRunner.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/block_app/framework/api/local/LocalBlockRunner.java b/giraph-block-app/src/main/java/org/apache/giraph/block_app/framework/api/local/LocalBlockRunner.java
index ad6ca15..d582cb2 100644
--- a/giraph-block-app/src/main/java/org/apache/giraph/block_app/framework/api/local/LocalBlockRunner.java
+++ b/giraph-block-app/src/main/java/org/apache/giraph/block_app/framework/api/local/LocalBlockRunner.java
@@ -144,6 +144,8 @@ public class LocalBlockRunner {
Block block, Object executionStage, TestGraph<I, V, E> graph,
final VertexSaver<I, V, E> vertexSaver
) {
+ Preconditions.checkNotNull(block);
+ Preconditions.checkNotNull(graph);
ImmutableClassesGiraphConfiguration<I, V, E> conf = graph.getConf();
int numWorkers = NUM_THREADS.get(conf);
boolean runAllChecks = RUN_ALL_CHECKS.get(conf);
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/Pieces.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/Pieces.java b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/Pieces.java
index 88b78a3..d926cdd 100644
--- a/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/Pieces.java
+++ b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/Pieces.java
@@ -32,6 +32,7 @@ import org.apache.giraph.block_app.library.internal.SendMessagePiece;
import org.apache.giraph.block_app.library.internal.SendMessageWithCombinerPiece;
import org.apache.giraph.combiner.MessageCombiner;
import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.PairConsumer;
import org.apache.giraph.function.vertex.ConsumerWithVertex;
import org.apache.giraph.function.vertex.SupplierFromVertex;
import org.apache.giraph.graph.Vertex;
@@ -170,10 +171,38 @@ public class Pieces {
<S, R extends Writable, I extends WritableComparable, V extends Writable,
E extends Writable>
Piece<I, V, E, NoMessage, Object> reduce(
+ String name,
+ ReduceOperation<S, R> reduceOp,
+ SupplierFromVertex<I, V, E, S> valueSupplier,
+ final Consumer<R> reducedValueConsumer) {
+ return reduceWithMaster(
+ name, reduceOp, valueSupplier,
+ new PairConsumer<R, BlockMasterApi>() {
+ @Override
+ public void apply(R input, BlockMasterApi master) {
+ reducedValueConsumer.apply(input);
+ }
+ });
+ }
+
+ /**
+ * Creates single reducer piece - given reduce class, supplier of values on
+ * worker, reduces and passes the result to given consumer on master.
+ *
+ * @param <S> Single value type, objects passed on workers
+ * @param <R> Reduced value type
+ * @param <I> Vertex id type
+ * @param <V> Vertex value type
+ * @param <E> Edge value type
+ */
+ public static
+ <S, R extends Writable, I extends WritableComparable, V extends Writable,
+ E extends Writable>
+ Piece<I, V, E, NoMessage, Object> reduceWithMaster(
final String name,
final ReduceOperation<S, R> reduceOp,
final SupplierFromVertex<I, V, E, S> valueSupplier,
- final Consumer<R> reducedValueConsumer) {
+ final PairConsumer<R, BlockMasterApi> reducedValueConsumer) {
return new Piece<I, V, E, NoMessage, Object>() {
private ReducerHandle<S, R> handle;
@@ -197,7 +226,7 @@ public class Pieces {
@Override
public void masterCompute(BlockMasterApi master, Object executionStage) {
- reducedValueConsumer.apply(handle.getReducedValue(master));
+ reducedValueConsumer.apply(handle.getReducedValue(master), master);
}
@Override
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/SendMessageChain.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/SendMessageChain.java b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/SendMessageChain.java
index b606a34..d1efd5b 100644
--- a/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/SendMessageChain.java
+++ b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/SendMessageChain.java
@@ -19,12 +19,14 @@ package org.apache.giraph.block_app.library;
import java.util.Iterator;
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
import org.apache.giraph.block_app.framework.block.Block;
import org.apache.giraph.block_app.framework.block.SequenceBlock;
import org.apache.giraph.combiner.MessageCombiner;
import org.apache.giraph.function.Consumer;
import org.apache.giraph.function.Function;
import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.PairConsumer;
import org.apache.giraph.function.vertex.ConsumerWithVertex;
import org.apache.giraph.function.vertex.FunctionWithVertex;
import org.apache.giraph.function.vertex.SupplierFromVertex;
@@ -145,6 +147,18 @@ public class SendMessageChain<I extends WritableComparable, V extends Writable,
}
/**
+ * Start chain by providing a function that will produce Block representing
+ * beginning of the chain, given a consumer of messages send
+ * by the last link in the created block.
+ */
+ public static <I extends WritableComparable, V extends Writable,
+ E extends Writable, P extends Writable>
+ SendMessageChain<I, V, E, P> startCustom(
+ Function<ConsumerWithVertex<I, V, E, P>, Block> createStartingBlock) {
+ return new SendMessageChain<>(createStartingBlock);
+ }
+
+ /**
* Give previously received message(s) to messageSupplier, and send message
* it returns to all targets provided by targetsSupplier.
*/
@@ -245,23 +259,51 @@ public class SendMessageChain<I extends WritableComparable, V extends Writable,
* by reducedValueConsumer.
*/
public <S, R extends Writable>
- Block endReduce(String name, ReduceOperation<S, R> reduceOp,
+ Block endReduce(final String name, final ReduceOperation<S, R> reduceOp,
final FunctionWithVertex<I, V, E, P, S> valueSupplier,
- Consumer<R> reducedValueConsumer) {
- final ObjectTransfer<P> prevMessagesTransfer = new ObjectTransfer<>();
+ final Consumer<R> reducedValueConsumer) {
+ return endCustom(new Function<SupplierFromVertex<I, V, E, P>, Block>() {
+ @Override
+ public Block apply(final SupplierFromVertex<I, V, E, P> prevMessages) {
+ return Pieces.reduce(
+ name,
+ reduceOp,
+ new SupplierFromVertex<I, V, E, S>() {
+ @Override
+ public S get(Vertex<I, V, E> vertex) {
+ return valueSupplier.apply(vertex, prevMessages.get(vertex));
+ }
+ },
+ reducedValueConsumer);
+ }
+ });
+ }
- return new SequenceBlock(
- blockCreator.apply(prevMessagesTransfer.<I, V, E>castToConsumer()),
- Pieces.reduce(
- name,
- reduceOp,
- new SupplierFromVertex<I, V, E, S>() {
- @Override
- public S get(Vertex<I, V, E> vertex) {
- return valueSupplier.apply(vertex, prevMessagesTransfer.get());
- }
- },
- reducedValueConsumer));
+ /**
+ * End chain by giving received messages to valueSupplier,
+ * to produce value that should be reduced, and consumed on master
+ * by reducedValueConsumer.
+ */
+ public <S, R extends Writable>
+ Block endReduceWithMaster(
+ final String name, final ReduceOperation<S, R> reduceOp,
+ final FunctionWithVertex<I, V, E, P, S> valueSupplier,
+ final PairConsumer<R, BlockMasterApi> reducedValueConsumer) {
+ return endCustom(new Function<SupplierFromVertex<I, V, E, P>, Block>() {
+ @Override
+ public Block apply(final SupplierFromVertex<I, V, E, P> prevMessages) {
+ return Pieces.reduceWithMaster(
+ name,
+ reduceOp,
+ new SupplierFromVertex<I, V, E, S>() {
+ @Override
+ public S get(Vertex<I, V, E> vertex) {
+ return valueSupplier.apply(vertex, prevMessages.get(vertex));
+ }
+ },
+ reducedValueConsumer);
+ }
+ });
}
/**
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/WorkerGCPiece.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/WorkerGCPiece.java b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/WorkerGCPiece.java
new file mode 100644
index 0000000..83ef3fd
--- /dev/null
+++ b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/WorkerGCPiece.java
@@ -0,0 +1,42 @@
+/*
+ * 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.block_app.library.gc;
+
+import org.apache.giraph.block_app.framework.api.BlockWorkerContextSendApi;
+import org.apache.giraph.block_app.framework.piece.PieceWithWorkerContext;
+import org.apache.giraph.types.NoMessage;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+/**
+ * Dummy piece to hint System.gc()
+ */
+public class WorkerGCPiece extends PieceWithWorkerContext<WritableComparable,
+ Writable, Writable, NoMessage, Object, NoMessage, Object> {
+
+ @Override
+ @SuppressFBWarnings(value = "DM_GC")
+ public void workerContextSend(
+ BlockWorkerContextSendApi<NoMessage> workerContextApi,
+ Object executionStage,
+ Object workerValue) {
+ System.gc();
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/package-info.java b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/package-info.java
new file mode 100644
index 0000000..bdf5ded
--- /dev/null
+++ b/giraph-block-app/src/main/java/org/apache/giraph/block_app/library/gc/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * GC utility pieces.
+ */
+package org.apache.giraph.block_app.library.gc;
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/Double2ObjFunction.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/Double2ObjFunction.java b/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/Double2ObjFunction.java
new file mode 100644
index 0000000..8b3c0a3
--- /dev/null
+++ b/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/Double2ObjFunction.java
@@ -0,0 +1,36 @@
+/*
+ * 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.function.primitive;
+
+import java.io.Serializable;
+
+/**
+ * Primitive specialization of Function:
+ * (double) -> T
+ *
+ * @param <T> Result type
+ */
+public interface Double2ObjFunction<T> extends Serializable {
+ /**
+ * Returns the result of applying this function to given {@code input}.
+ *
+ * The returned object may or may not be a new instance,
+ * depending on the implementation.
+ */
+ T apply(double input);
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/DoubleConsumer.java
----------------------------------------------------------------------
diff --git a/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/DoubleConsumer.java b/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/DoubleConsumer.java
new file mode 100644
index 0000000..4725d4b
--- /dev/null
+++ b/giraph-block-app/src/main/java/org/apache/giraph/function/primitive/DoubleConsumer.java
@@ -0,0 +1,32 @@
+/*
+ * 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.function.primitive;
+
+import java.io.Serializable;
+
+
+/**
+ * Primitive specialization of Function:
+ * (double) -> void
+ */
+public interface DoubleConsumer extends Serializable {
+ /**
+ * Applies this function to {@code input}
+ */
+ void apply(double input);
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-core/src/main/java/org/apache/giraph/combiner/MaxMessageCombiner.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/combiner/MaxMessageCombiner.java b/giraph-core/src/main/java/org/apache/giraph/combiner/MaxMessageCombiner.java
new file mode 100644
index 0000000..5a7c101
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/combiner/MaxMessageCombiner.java
@@ -0,0 +1,74 @@
+/*
+ * 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.combiner;
+
+import org.apache.giraph.types.ops.DoubleTypeOps;
+import org.apache.giraph.types.ops.FloatTypeOps;
+import org.apache.giraph.types.ops.IntTypeOps;
+import org.apache.giraph.types.ops.LongTypeOps;
+import org.apache.giraph.types.ops.NumericTypeOps;
+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.apache.hadoop.io.WritableComparable;
+
+/**
+ * Message combiner which calculates max of all messages.
+ *
+ * @param <M> Message type
+ */
+public class MaxMessageCombiner<M extends WritableComparable>
+ implements MessageCombiner<WritableComparable, M> {
+ /** DoubleWritable specialization */
+ public static final MaxMessageCombiner<DoubleWritable> DOUBLE =
+ new MaxMessageCombiner<>(DoubleTypeOps.INSTANCE);
+ /** DoubleWritable specialization */
+ public static final MaxMessageCombiner<FloatWritable> FLOAT =
+ new MaxMessageCombiner<>(FloatTypeOps.INSTANCE);
+ /** LongWritable specialization */
+ public static final MaxMessageCombiner<LongWritable> LONG =
+ new MaxMessageCombiner<>(LongTypeOps.INSTANCE);
+ /** IntWritable specialization */
+ public static final MaxMessageCombiner<IntWritable> INT =
+ new MaxMessageCombiner<>(IntTypeOps.INSTANCE);
+
+ /** Value type operations */
+ private final NumericTypeOps<M> typeOps;
+
+ /**
+ * Constructor
+ * @param typeOps Value type operations
+ */
+ public MaxMessageCombiner(NumericTypeOps<M> typeOps) {
+ this.typeOps = typeOps;
+ }
+
+ @Override
+ public void combine(
+ WritableComparable vertexIndex, M originalMessage, M messageToCombine) {
+ if (originalMessage.compareTo(messageToCombine) < 0) {
+ typeOps.set(originalMessage, messageToCombine);
+ }
+ }
+
+ @Override
+ public M createInitialMessage() {
+ return typeOps.createZero();
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-core/src/main/java/org/apache/giraph/reducers/impl/PairReduce.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/reducers/impl/PairReduce.java b/giraph-core/src/main/java/org/apache/giraph/reducers/impl/PairReduce.java
new file mode 100644
index 0000000..1aea293
--- /dev/null
+++ b/giraph-core/src/main/java/org/apache/giraph/reducers/impl/PairReduce.java
@@ -0,0 +1,104 @@
+/*
+ * 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.reducers.impl;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.giraph.reducers.ReduceOperation;
+import org.apache.giraph.utils.WritableUtils;
+import org.apache.giraph.writable.tuple.PairWritable;
+import org.apache.hadoop.io.Writable;
+import org.python.google.common.base.Preconditions;
+
+/**
+ * Combines two individual reducers, to create a single reducer of pairs that
+ * reduces each of them individually.
+ *
+ * @param <S1> First single value type
+ * @param <R1> First reduced value type
+ * @param <S2> Second single value type
+ * @param <R2> Second reduced value type
+ */
+public class PairReduce<S1, R1 extends Writable, S2, R2 extends Writable>
+ implements ReduceOperation<Pair<S1, S2>, PairWritable<R1, R2>> {
+ /** First reduceOp */
+ private ReduceOperation<S1, R1> reduce1;
+ /** Second reduceOp */
+ private ReduceOperation<S2, R2> reduce2;
+
+ /** Constructor */
+ public PairReduce() {
+ }
+
+ /**
+ * Constructor
+ * @param reduce1 First reduceOp
+ * @param reduce2 Second reduceOp
+ */
+ public PairReduce(
+ ReduceOperation<S1, R1> reduce1, ReduceOperation<S2, R2> reduce2) {
+ this.reduce1 = reduce1;
+ this.reduce2 = reduce2;
+ }
+
+
+ @Override
+ public PairWritable<R1, R2> createInitialValue() {
+ return new PairWritable<>(
+ reduce1.createInitialValue(), reduce2.createInitialValue());
+ }
+
+ @Override
+ public PairWritable<R1, R2> reduce(
+ PairWritable<R1, R2> curValue, Pair<S1, S2> valueToReduce) {
+ Preconditions.checkState(
+ curValue.getLeft() ==
+ reduce1.reduce(curValue.getLeft(), valueToReduce.getLeft()));
+ Preconditions.checkState(
+ curValue.getRight() ==
+ reduce2.reduce(curValue.getRight(), valueToReduce.getRight()));
+ return curValue;
+ }
+
+ @Override
+ public PairWritable<R1, R2> reduceMerge(
+ PairWritable<R1, R2> curValue, PairWritable<R1, R2> valueToReduce) {
+ Preconditions.checkState(
+ curValue.getLeft() ==
+ reduce1.reduceMerge(curValue.getLeft(), valueToReduce.getLeft()));
+ Preconditions.checkState(
+ curValue.getRight() ==
+ reduce2.reduceMerge(curValue.getRight(), valueToReduce.getRight()));
+ return curValue;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ WritableUtils.writeWritableObject(reduce1, out);
+ WritableUtils.writeWritableObject(reduce2, out);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ reduce1 = WritableUtils.readWritableObject(in, null);
+ reduce2 = WritableUtils.readWritableObject(in, null);
+ }
+}
[2/2] git commit: updated refs/heads/trunk to 3b7c68e
Posted by ik...@apache.org.
[GIRAPH-1013] Adding more libraries, algos and examples
Summary:
Adding more libraries, algos and examples
Only changes from our internal state:
New classes:
PairReduce
MaxMessageCombiner
PartitioningStats
TestMessageChain
Change to:
Pieces
SendMessageChain
Test Plan: mvn clean install -Phadoop_facebook
Reviewers: maja.kabiljo, dionysis.logothetis, sergey.edunov
Reviewed By: sergey.edunov
Differential Revision: https://reviews.facebook.net/D40935
Project: http://git-wip-us.apache.org/repos/asf/giraph/repo
Commit: http://git-wip-us.apache.org/repos/asf/giraph/commit/3b7c68e5
Tree: http://git-wip-us.apache.org/repos/asf/giraph/tree/3b7c68e5
Diff: http://git-wip-us.apache.org/repos/asf/giraph/diff/3b7c68e5
Branch: refs/heads/trunk
Commit: 3b7c68e54a76d02ae721161b1b81c15c2e22e44a
Parents: d7e4bde
Author: Igor Kabiljo <ik...@fb.com>
Authored: Mon Jun 29 23:20:58 2015 -0700
Committer: Igor Kabiljo <ik...@fb.com>
Committed: Thu Jul 2 16:32:41 2015 -0700
----------------------------------------------------------------------
.../library/algo/BreadthFirstSearch.java | 202 +++++++++++
.../library/algo/DistributedIndependentSet.java | 337 ++++++++++++++++++
.../block_app/library/algo/package-info.java | 21 ++
.../library/coarsening/CoarseningUtils.java | 347 +++++++++++++++++++
.../library/coarsening/package-info.java | 21 ++
.../library/stats/DirectedGraphStats.java | 187 ++++++++++
.../library/stats/PartitioningStats.java | 125 +++++++
.../block_app/library/stats/package-info.java | 21 ++
.../AbstractPageRankExampleBlockFactory.java | 54 +++
.../pagerank/PageRankExampleBlockFactory.java | 47 +++
...eRankWithConvergenceExampleBlockFactory.java | 70 ++++
...PiecesAndConvergenceExampleBlockFactory.java | 146 ++++++++
.../PageRankWithPiecesExampleBlockFactory.java | 83 +++++
...ansferAndConvergenceExampleBlockFactory.java | 74 ++++
.../examples/pagerank/TestPageRankExample.java | 127 +++++++
.../block_app/library/TestMessageChain.java | 224 ++++++++++++
.../algo/BreadthFirstSearchBlockFactory.java | 79 +++++
.../algo/BreadthFirstSearchVertexValue.java | 61 ++++
.../DistributedIndependentSetBlockFactory.java | 60 ++++
.../DistributedIndependentSetVertexValue.java | 56 +++
.../library/algo/TestBreadthFirstSearch.java | 165 +++++++++
.../algo/TestDistributedIndependentSet.java | 220 ++++++++++++
.../library/coarsening/TestCoarseningUtils.java | 133 +++++++
.../kryo/KryoWritableWrapperJava8Test.java | 174 ++++++++++
.../framework/api/local/LocalBlockRunner.java | 2 +
.../apache/giraph/block_app/library/Pieces.java | 33 +-
.../block_app/library/SendMessageChain.java | 72 +++-
.../block_app/library/gc/WorkerGCPiece.java | 42 +++
.../block_app/library/gc/package-info.java | 21 ++
.../function/primitive/Double2ObjFunction.java | 36 ++
.../function/primitive/DoubleConsumer.java | 32 ++
.../giraph/combiner/MaxMessageCombiner.java | 74 ++++
.../apache/giraph/reducers/impl/PairReduce.java | 104 ++++++
33 files changed, 3433 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearch.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearch.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearch.java
new file mode 100644
index 0000000..fe290fb
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/BreadthFirstSearch.java
@@ -0,0 +1,202 @@
+/*
+ * 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.block_app.library.algo;
+
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.api.CreateReducersApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatUntilBlock;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.Supplier;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.giraph.types.NoMessage;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.log4j.Logger;
+
+import com.google.common.collect.Iterators;
+
+/**
+ * Class for running breadth-first search on the graph.
+ *
+ * Graph is expected to be symmetric before calling any of the methods here.
+ */
+public class BreadthFirstSearch {
+ private static final Logger LOG = Logger.getLogger(BreadthFirstSearch.class);
+ private static final IntWritable NOT_REACHABLE_VERTEX_VALUE =
+ new IntWritable(-1);
+
+ private BreadthFirstSearch() {
+ }
+
+ /**
+ * Default block, which calculates connected components using the vertex's
+ * default edges.
+ */
+ public static <I extends WritableComparable, V extends Writable>
+ Block bfs(
+ SupplierFromVertex<I, V, Writable, Boolean> isVertexInSeedSet,
+ SupplierFromVertex<I, V, Writable, IntWritable> getDistance,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setDistance
+ ) {
+ ObjectTransfer<Boolean> converged = new ObjectTransfer<>();
+ ObjectTransfer<Boolean> vertexUpdatedDistance = new ObjectTransfer<>();
+
+ return new SequenceBlock(
+ createInitializePiece(
+ vertexUpdatedDistance,
+ isVertexInSeedSet,
+ getDistance,
+ setDistance,
+ VertexSuppliers.vertexEdgesSupplier()
+ ),
+ RepeatUntilBlock.unlimited(
+ createPropagateConnectedComponentsPiece(
+ vertexUpdatedDistance,
+ vertexUpdatedDistance,
+ converged,
+ getDistance,
+ setDistance,
+ VertexSuppliers.vertexEdgesSupplier()
+ ),
+ converged
+ )
+ );
+ }
+
+ /**
+ * Initialize vertex values for connected components calculation
+ */
+ private static <I extends WritableComparable, V extends Writable>
+ Piece<I, V, Writable, NoMessage, Object> createInitializePiece(
+ Consumer<Boolean> vertexUpdatedDistance,
+ SupplierFromVertex<I, V, Writable, Boolean> isVertexInSeedSet,
+ SupplierFromVertex<I, V, Writable, IntWritable> getDistance,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setDistance,
+ SupplierFromVertex<I, V, Writable, ? extends Iterable<? extends Edge<I, ?>>>
+ edgeSupplier
+ ) {
+ IntWritable zero = new IntWritable(0);
+ return Pieces.forAllVerticesOnReceive(
+ "InitializeBFS",
+ (vertex) -> {
+ if (isVertexInSeedSet.get(vertex)) {
+ setDistance.apply(vertex, zero);
+ vertexUpdatedDistance.apply(true);
+ } else {
+ setDistance.apply(vertex, NOT_REACHABLE_VERTEX_VALUE);
+ vertexUpdatedDistance.apply(false);
+ }
+ }
+ );
+ }
+
+ /**
+ * Propagate connected components to neighbor pieces
+ */
+ private static <I extends WritableComparable, V extends Writable>
+ Block createPropagateConnectedComponentsPiece(
+ Supplier<Boolean> vertexToPropagate,
+ Consumer<Boolean> vertexUpdatedDistance,
+ Consumer<Boolean> converged,
+ SupplierFromVertex<I, V, Writable, IntWritable> getDistance,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setDistance,
+ SupplierFromVertex<I, V, Writable, ? extends Iterable<?
+ extends Edge<I, ?>>> edgeSupplier) {
+ return new Piece<I, V, Writable, IntWritable, Object>() {
+ private ReducerHandle<IntWritable, IntWritable> propagatedAggregator;
+
+ @Override
+ public void registerReducers(
+ CreateReducersApi reduceApi, Object executionStage) {
+ propagatedAggregator = reduceApi.createLocalReducer(SumReduce.INT);
+ }
+
+ @Override
+ public VertexSender<I, V, Writable> getVertexSender(
+ BlockWorkerSendApi<I, V, Writable, IntWritable> workerApi,
+ Object executionStage
+ ) {
+ return (vertex) -> {
+ if (vertexToPropagate.get()) {
+ workerApi.sendMessageToMultipleEdges(
+ Iterators.transform(
+ edgeSupplier.get(vertex).iterator(),
+ Edge::getTargetVertexId
+ ),
+ getDistance.get(vertex)
+ );
+ reduceInt(propagatedAggregator, 1);
+ }
+ };
+ }
+
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+ converged.apply(
+ propagatedAggregator.getReducedValue(master).get() == 0);
+ LOG.info("BFS: " + propagatedAggregator.getReducedValue(master).get() +
+ " many vertices sent in this iteration");
+ }
+
+ @Override
+ public VertexReceiver<I, V, Writable, IntWritable> getVertexReceiver(
+ BlockWorkerReceiveApi<I> workerApi,
+ Object executionStage
+ ) {
+ IntWritable next = new IntWritable();
+ return (vertex, messages) -> {
+ vertexUpdatedDistance.apply(false);
+ for (IntWritable receivedValue : messages) {
+ IntWritable currentValue = getDistance.get(vertex);
+ next.set(receivedValue.get() + 1);
+ if (currentValue.compareTo(NOT_REACHABLE_VERTEX_VALUE) == 0 ||
+ currentValue.compareTo(next) > 0) {
+ setDistance.apply(vertex, next);
+ vertexUpdatedDistance.apply(true);
+ }
+ }
+ };
+ }
+
+ @Override
+ public Class<IntWritable> getMessageClass() {
+ return IntWritable.class;
+ }
+
+ @Override
+ public String toString() {
+ return "PropagateConnectedComponentsPiece";
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSet.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSet.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSet.java
new file mode 100644
index 0000000..8beef59
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/DistributedIndependentSet.java
@@ -0,0 +1,337 @@
+/*
+ * 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.block_app.library.algo;
+
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.api.CreateReducersApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatUntilBlock;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.primitive.PrimitiveRefs.IntRef;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.giraph.types.NoMessage;
+import org.apache.hadoop.io.BooleanWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.log4j.Logger;
+
+import com.google.common.collect.Iterables;
+
+/**
+ * Class for computing maximal independent sets of a graph.
+ *
+ * Graph is expected to be symmetric before calling methods here.
+ */
+public class DistributedIndependentSet {
+ private static final Logger LOG =
+ Logger.getLogger(DistributedIndependentSet.class);
+ private static final IntWritable UNKNOWN = new IntWritable(-1);
+ private static final IntWritable NOT_IN_SET = new IntWritable(-2);
+ private static final IntWritable IN_SET = new IntWritable(-3);
+
+ private DistributedIndependentSet() {
+ }
+
+ /**
+ * Default block which decomposes the input graph into independent sets of
+ * vertices. An independent set of vertices is defined as a set of vertices
+ * that do not have edge to each other, i.e. the sub-graph induced by an
+ * independent set of vertices is an empty graph.
+ *
+ * The algorithm finds independent sets as follows:
+ * 1) Find the maximal independent set of vertices amongst unassigned
+ * vertices.
+ * 2) Assign found vertices to a new independent set.
+ * 3) If all vertices of the graph are assigned, go to step 1. Otherwise
+ * terminate.
+ *
+ * @param numberClass Independent set id type
+ * @param chooseNumber Process that returns a deterministic id of type
+ * <code>numberClass</code> for each vertex
+ * @param getIndependentSet Getter for independent set id of each vertex
+ * @param setIndependentSet Setter for independent set id of each vertex
+ */
+ public static
+ <I extends WritableComparable, V extends Writable,
+ N extends WritableComparable>
+ Block independentSets(
+ Class<N> numberClass,
+ SupplierFromVertex<I, V, Writable, N> chooseNumber,
+ SupplierFromVertex<I, V, Writable, IntWritable> getIndependentSet,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setIndependentSet
+ ) {
+ ObjectTransfer<Boolean> done = new ObjectTransfer<>();
+ IntRef iteration = new IntRef(-1);
+
+ return new SequenceBlock(
+ Pieces.<I, V, Writable>forAllVerticesOnReceive(
+ "InitializeIndependentSet",
+ (vertex) -> setIndependentSet.apply(vertex, UNKNOWN)),
+ RepeatUntilBlock.unlimited(
+ // find maximal independent sets amongst remaining un-assigned
+ // vertices, one after another, until all vertices of the graph
+ // are assigned to independent sets.
+ findMaximalIndependentSet(
+ numberClass,
+ chooseNumber,
+ getIndependentSet,
+ setIndependentSet,
+ iteration, done
+ ),
+ done
+ )
+ );
+ }
+
+ /**
+ * Independent set calculation with vertex ids as messages to choose the
+ * vertices in a set.
+ */
+ public static <I extends WritableComparable, V extends Writable>
+ Block independentSets(
+ Class<I> vertexIdClass,
+ SupplierFromVertex<I, V, Writable, IntWritable> getIndependentSet,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setIndependentSet
+ ) {
+ return independentSets(
+ vertexIdClass,
+ VertexSuppliers.vertexIdSupplier(),
+ getIndependentSet,
+ setIndependentSet
+ );
+ }
+
+ /**
+ * Finds a maximal independent set, amongst un-assigned vertices of the
+ * graph. The algorithm is as follows:
+ * 1) Normalize the state of all un-assigned vertices of the graph to
+ * UNKNOWN.
+ * 2) Each UNKNOWN vertex sends a number (output of <code>chooseNumber
+ * </code>) to all its neighbors.
+ * 3) Each UNKNOWN vertex finds the maximum value of all incoming
+ * messages. If the max value is less than the chosen number for the
+ * vertex (output of <code>chooseNumber</code> in step 2), the vertex
+ * assigns itself to the independent set (changes its state to IN_SET),
+ * and sends 'ack' messages to all its neighbors.
+ * 4) Each UNKNOWN vertex that receives an 'ack' message, changes its
+ * state to NOT_IN_SET.
+ * 5) If there are any UNKNOWN vertex, go to step 2. Otherwise, we found
+ * the maximal independent set, and terminate.
+ */
+ private static
+ <I extends WritableComparable, V extends Writable,
+ N extends WritableComparable>
+ Block findMaximalIndependentSet(
+ Class<N> numberClass,
+ SupplierFromVertex<I, V, Writable, N> chooseNumber,
+ SupplierFromVertex<I, V, Writable, IntWritable> getIndependentSet,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setIndependentSet,
+ IntRef iteration, ObjectTransfer<Boolean> done
+ ) {
+ ObjectTransfer<Boolean> foundMIS = new ObjectTransfer<>();
+ return new SequenceBlock(
+ createInitializePiece(getIndependentSet, setIndependentSet, iteration),
+ RepeatUntilBlock.unlimited(
+ new SequenceBlock(
+ // Announce the number to all the neighbors.
+ createAnnounceBlock(
+ numberClass,
+ chooseNumber,
+ getIndependentSet,
+ setIndependentSet
+ ),
+ // Select the vertices in the independent set, and refine the
+ // state of neighboring vertices so not to consider them for
+ // this independent set.
+ createSelectAndRefinePiece(
+ getIndependentSet,
+ setIndependentSet,
+ iteration,
+ foundMIS,
+ done
+ )
+ ),
+ foundMIS
+ )
+ );
+ }
+
+ /**
+ * Piece to initialize vertex values so they are either assigned to a
+ * previously found independent set, or UNKNOWN to be considered for the
+ * discovery of the current independent set.
+ */
+ private static <I extends WritableComparable, V extends Writable>
+ Block createInitializePiece(
+ SupplierFromVertex<I, V, Writable, IntWritable> getIndependentSet,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setIndependentSet,
+ IntRef iteration) {
+ return new Piece<I, V, Writable, NoMessage, Object>() {
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+ iteration.value++;
+ LOG.info("Start finding independent set with ID " + iteration.value);
+ }
+
+ @Override
+ public VertexReceiver<I, V, Writable, NoMessage> getVertexReceiver(
+ final BlockWorkerReceiveApi<I> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ if (getIndependentSet.get(vertex).equals(NOT_IN_SET)) {
+ setIndependentSet.apply(vertex, UNKNOWN);
+ }
+ };
+ }
+
+ @Override
+ public String toString() {
+ return "InitializePiece";
+ }
+ };
+ }
+
+ /**
+ * A Block that assigns some of the UNKNOWN vertices to the new independent
+ * set based on numbers each vertex broadcast to all its neighbors.
+ */
+ private static
+ <I extends WritableComparable, V extends Writable,
+ N extends WritableComparable>
+ Block createAnnounceBlock(
+ Class<N> numberClass,
+ SupplierFromVertex<I, V, Writable, N> chooseNumber,
+ SupplierFromVertex<I, V, Writable, IntWritable> getIndependentSet,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setIndependentSet) {
+ return Pieces.<I, V, Writable, N>sendMessageToNeighbors(
+ "AnnounceNumber",
+ numberClass,
+ (vertex) -> {
+ if (getIndependentSet.get(vertex).equals(UNKNOWN)) {
+ return chooseNumber.get(vertex);
+ } else {
+ return null;
+ }
+ },
+ (vertex, messages) -> {
+ if (getIndependentSet.get(vertex).equals(UNKNOWN)) {
+ N curVal = chooseNumber.get(vertex);
+ boolean myValIsMax = true;
+ for (N receivedValue : messages) {
+ if (receivedValue.compareTo(curVal) > 0) {
+ myValIsMax = false;
+ break;
+ }
+ }
+ if (myValIsMax) {
+ setIndependentSet.apply(vertex, IN_SET);
+ }
+ }
+ }
+ );
+ }
+
+ /**
+ * Piece to confirm selection of some vertices for the independent set. Also,
+ * changes the state of neighboring vertices of newly assigned vertices to
+ * NOT_IN_SET, so not to consider them for the discovery of the current
+ * independent set.
+ *
+ * @param foundMIS Specifies the end of discovery for current independent set.
+ * @param done Specifies the end of whole computation of decomposing to
+ * independent sets.
+ */
+ private static <I extends WritableComparable, V extends Writable>
+ Block createSelectAndRefinePiece(
+ SupplierFromVertex<I, V, Writable, IntWritable> getIndependentSet,
+ ConsumerWithVertex<I, V, Writable, IntWritable> setIndependentSet,
+ IntRef iteration,
+ Consumer<Boolean> foundMIS,
+ Consumer<Boolean> done) {
+ return new Piece<I, V, Writable, BooleanWritable, Object>() {
+ private ReducerHandle<IntWritable, IntWritable> numVerticesUnknown;
+ private ReducerHandle<IntWritable, IntWritable> numVerticesNotAssigned;
+
+ @Override
+ public void registerReducers(
+ CreateReducersApi reduceApi, Object executionStage) {
+ numVerticesUnknown = reduceApi.createLocalReducer(SumReduce.INT);
+ numVerticesNotAssigned = reduceApi.createLocalReducer(SumReduce.INT);
+ }
+
+ @Override
+ public VertexSender<I, V, Writable> getVertexSender(
+ final BlockWorkerSendApi<I, V, Writable, BooleanWritable> workerApi,
+ Object executionStage) {
+ BooleanWritable ack = new BooleanWritable(true);
+ IntWritable one = new IntWritable(1);
+ return (vertex) -> {
+ IntWritable vertexState = getIndependentSet.get(vertex);
+ if (vertexState.equals(IN_SET)) {
+ setIndependentSet.apply(vertex, new IntWritable(iteration.value));
+ workerApi.sendMessageToAllEdges(vertex, ack);
+ } else if (vertexState.equals(UNKNOWN)) {
+ numVerticesUnknown.reduce(one);
+ numVerticesNotAssigned.reduce(one);
+ } else if (vertexState.equals(NOT_IN_SET)) {
+ numVerticesNotAssigned.reduce(one);
+ }
+ };
+ }
+
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+ done.apply(numVerticesNotAssigned.getReducedValue(master).get() == 0);
+ foundMIS.apply(numVerticesUnknown.getReducedValue(master).get() == 0);
+ }
+
+ @Override
+ public VertexReceiver<I, V, Writable, BooleanWritable> getVertexReceiver(
+ final BlockWorkerReceiveApi<I> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ if (getIndependentSet.get(vertex).equals(UNKNOWN) &&
+ Iterables.size(messages) > 0) {
+ setIndependentSet.apply(vertex, NOT_IN_SET);
+ }
+ };
+ }
+
+ @Override
+ public Class<BooleanWritable> getMessageClass() {
+ return BooleanWritable.class;
+ }
+
+ @Override
+ public String toString() {
+ return "SelectAndRefinePiece";
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/package-info.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/package-info.java
new file mode 100644
index 0000000..03a550a
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/algo/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * Utility algorithms
+ */
+package org.apache.giraph.block_app.library.algo;
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/CoarseningUtils.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/CoarseningUtils.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/CoarseningUtils.java
new file mode 100644
index 0000000..c74a056
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/CoarseningUtils.java
@@ -0,0 +1,347 @@
+/*
+ * 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.block_app.library.coarsening;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.delegate.FilteringPiece;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.combiner.MessageCombiner;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.edge.EdgeFactory;
+import org.apache.giraph.factories.MessageValueFactory;
+import org.apache.giraph.function.Function;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.vertex.ConsumerWithVertex;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.types.ops.NumericTypeOps;
+import org.apache.giraph.types.ops.PrimitiveIdTypeOps;
+import org.apache.giraph.types.ops.TypeOps;
+import org.apache.giraph.types.ops.TypeOpsUtils;
+import org.apache.giraph.types.ops.collections.Basic2ObjectMap;
+import org.apache.giraph.types.ops.collections.WritableWriter;
+import org.apache.giraph.utils.ReflectionUtils;
+import org.apache.giraph.writable.tuple.PairWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * Utilities class that creates a new coarsened graph, while keeping the
+ * original graph, allowing us to execute things on either representation
+ * of the graph.
+ */
+public class CoarseningUtils {
+
+ private CoarseningUtils() { }
+
+ //CHECKSTYLE: stop ParameterNumberCheck
+
+ /**
+ * Create block that creates coarsened graph from given set of vertices
+ * (original vertices, that return true for originalNodesChecker).
+ *
+ * Old graph is not deleted, if needed, you can trivially do that by calling
+ * Pieces.removeVertices(originalNodesChecker) after this block.
+ *
+ *
+ * @param idTypeOps Type ops for vertex ids
+ * @param edgeTypeOps Type ops for edge values
+ * @param coarsenedVertexIdSupplier Supplier called on original nodes,
+ * that should return it's coarsened vertex
+ * id
+ * @param vertexInfoClass Part of the vertex value that is needed to create
+ * coarsened vertices
+ * @param vertexInfoSupplier Supplier providing part of the vertex value that
+ * is needed to create coarsened vertices
+ * @param vertexInfoCombiner Combiner that aggregates individual vertex info
+ * values into coarsened value
+ * @param coarsenedVertexValueInitializer Function that initializes coarsened
+ * vertex value, from an aggregated
+ * vertex info of original vertex
+ * values that are coarsened into this
+ * vertex
+ * @param edgeCoarseningCombiner Combiner that aggregates individual edge
+ * values into coarsened edge value
+ * @param originalNodesChecker Vertices with id that this gives true for will
+ * be coarsened
+ * @param clusterNodesChecker Should return true only for coarsened ids
+ * (ones returned by coarsenedVertexIdSupplier)
+ * @return Block that does coarsening
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable,
+ VI extends Writable>
+ Block createCustomCoarseningBlock(
+ PrimitiveIdTypeOps<I> idTypeOps,
+ TypeOps<E> edgeTypeOps,
+ SupplierFromVertex<I, V, E, I> coarsenedVertexIdSupplier,
+ Class<VI> vertexInfoClass,
+ SupplierFromVertex<I, V, E, VI> vertexInfoSupplier,
+ MessageCombiner<? super I, VI> vertexInfoCombiner,
+ ConsumerWithVertex<I, V, E, VI> coarsenedVertexValueInitializer,
+ MessageCombiner<? super I, E> edgeCoarseningCombiner,
+ Function<I, Boolean> originalNodesChecker,
+ Function<I, Boolean> clusterNodesChecker) {
+
+ //CHECKSTYLE: resume ParameterNumberCheck
+
+ ObjectTransfer<Iterable<PairWritable<I, E>>> edgesHolder =
+ new ObjectTransfer<>();
+
+ MessageValueFactory<PairWritable<I, E>> pairMessageFactory =
+ () -> new PairWritable<I, E>(
+ idTypeOps.create(), edgeTypeOps.create());
+ Piece<I, V, E, PairWritable<I, E>, Object> calcWeightsOnEdgesPiece =
+ new Piece<I, V, E, PairWritable<I, E>, Object>() {
+ @Override
+ public VertexSender<I, V, E> getVertexSender(
+ BlockWorkerSendApi<I, V, E, PairWritable<I, E>> workerApi,
+ Object executionStage) {
+ PairWritable<I, E> message = pairMessageFactory.newInstance();
+ return (vertex) -> {
+ idTypeOps.set(
+ message.getLeft(), coarsenedVertexIdSupplier.get(vertex));
+ for (Edge<I, E> edge : vertex.getEdges()) {
+ edgeTypeOps.set(message.getRight(), edge.getValue());
+ workerApi.sendMessage(edge.getTargetVertexId(), message);
+ }
+ };
+ }
+
+ @Override
+ public VertexReceiver<I, V, E, PairWritable<I, E>> getVertexReceiver(
+ BlockWorkerReceiveApi<I> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ edgesHolder.apply(messages);
+ };
+ }
+
+ @Override
+ protected MessageValueFactory<PairWritable<I, E>> getMessageFactory(
+ ImmutableClassesGiraphConfiguration conf) {
+ return pairMessageFactory;
+ }
+
+ @Override
+ public String toString() {
+ return "CoarseningCalcWeightsOnEdges";
+ }
+ };
+
+ WritableWriter<E> edgeValueWriter = new WritableWriter<E>() {
+ @Override
+ public void write(DataOutput out, E value) throws IOException {
+ value.write(out);
+ }
+
+ @Override
+ public E readFields(DataInput in) throws IOException {
+ E edge = edgeTypeOps.create();
+ edge.readFields(in);
+ return edge;
+ }
+ };
+
+ Piece<I, V, E, PairWritable<VI, Basic2ObjectMap<I, E>>, Object>
+ createNewVerticesPiece =
+ new Piece<I, V, E, PairWritable<VI, Basic2ObjectMap<I, E>>, Object>() {
+ @Override
+ public VertexSender<I, V, E> getVertexSender(
+ BlockWorkerSendApi<I, V, E, PairWritable<VI, Basic2ObjectMap<I, E>>>
+ workerApi,
+ Object executionStage) {
+ return (vertex) -> {
+ Basic2ObjectMap<I, E> map =
+ idTypeOps.create2ObjectOpenHashMap(edgeValueWriter);
+ for (PairWritable<I, E> message : edgesHolder.get()) {
+
+ E value = map.get(message.getLeft());
+ if (value == null) {
+ value = edgeCoarseningCombiner.createInitialMessage();
+ map.put(message.getLeft(), value);
+ }
+
+ edgeCoarseningCombiner.combine(
+ message.getLeft(), value, message.getRight());
+ }
+
+ workerApi.sendMessage(
+ coarsenedVertexIdSupplier.get(vertex),
+ new PairWritable<>(vertexInfoSupplier.get(vertex), map));
+ };
+ }
+
+ @Override
+ public
+ VertexReceiver<I, V, E, PairWritable<VI, Basic2ObjectMap<I, E>>>
+ getVertexReceiver(
+ BlockWorkerReceiveApi<I> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ VI vertexInfo = vertexInfoCombiner.createInitialMessage();
+ Basic2ObjectMap<I, E> map = idTypeOps.create2ObjectOpenHashMap(null);
+ for (PairWritable<VI, Basic2ObjectMap<I, E>> message : messages) {
+ vertexInfoCombiner.combine(
+ vertex.getId(), vertexInfo, message.getLeft());
+
+
+ for (Iterator<I> iter = message.getRight().fastKeyIterator();
+ iter.hasNext();) {
+ I key = iter.next();
+ E value = map.get(key);
+ if (value == null) {
+ value = edgeCoarseningCombiner.createInitialMessage();
+ map.put(key, value);
+ }
+ edgeCoarseningCombiner.combine(
+ vertex.getId(), value, message.getRight().get(key));
+ }
+ }
+
+ for (Iterator<I> iter = map.fastKeyIterator(); iter.hasNext();) {
+ I key = iter.next();
+ Edge<I, E> edge = EdgeFactory.create(
+ idTypeOps.createCopy(key), map.get(key));
+ vertex.addEdge(edge);
+ }
+
+ coarsenedVertexValueInitializer.apply(vertex, vertexInfo);
+ };
+ }
+
+ @Override
+ protected MessageValueFactory<PairWritable<VI, Basic2ObjectMap<I, E>>>
+ getMessageFactory(
+ ImmutableClassesGiraphConfiguration conf) {
+ return () -> new PairWritable<>(
+ ReflectionUtils.newInstance(vertexInfoClass),
+ idTypeOps.create2ObjectOpenHashMap(edgeValueWriter));
+ }
+
+ @Override
+ public String toString() {
+ return "CoarseningCreateNewVertices";
+ }
+ };
+
+ return new SequenceBlock(
+ new FilteringPiece<>(
+ (vertex) -> originalNodesChecker.apply(vertex.getId()),
+ calcWeightsOnEdgesPiece),
+ new FilteringPiece<>(
+ (vertex) -> originalNodesChecker.apply(vertex.getId()),
+ (vertex) -> clusterNodesChecker.apply(vertex.getId()),
+ createNewVerticesPiece));
+ }
+
+ /**
+ * Create block that creates coarsened graph from given set of vertices
+ * (original vertices, that return true for originalNodesChecker), when
+ * vertex and edge values are primitives (or TypeOps exist for them).
+ * Coarsening vertex values and edge values are computed as sum of their
+ * individual values.
+ *
+ * Old graph is not deleted, if needed, you can trivially do that by calling
+ * Pieces.removeVertices(originalNodesChecker) after this block.
+ *
+ * @param idTypeOps Vertex id TypeOps
+ * @param valueTypeOps Vertex value TypeOps
+ * @param edgeTypeOps Edge value TypeOps
+ * @param coarsenedVertexIdSupplier Supplier called on original nodes,
+ * that should return it's coarsened vertex
+ * id
+ * @param originalNodesChecker Vertices with id that this gives true for will
+ * be coarsened
+ * @param clusterNodesChecker Should return true only for coarsened ids
+ * (ones returned by coarsenedVertexIdSupplier)
+ * @return Block that does coarsening
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable>
+ Block createCoarseningBlock(
+ PrimitiveIdTypeOps<I> idTypeOps,
+ NumericTypeOps<V> valueTypeOps,
+ NumericTypeOps<E> edgeTypeOps,
+ SupplierFromVertex<I, V, E, I> coarsenedVertexIdSupplier,
+ Function<I, Boolean> originalNodesChecker,
+ Function<I, Boolean> clusterNodesChecker) {
+ return CoarseningUtils.<I, V, E, V>createCustomCoarseningBlock(
+ idTypeOps,
+ edgeTypeOps,
+ coarsenedVertexIdSupplier,
+ valueTypeOps.getTypeClass(),
+ VertexSuppliers.vertexValueSupplier(),
+ new MessageCombiner<I, V>() {
+ @Override
+ public void combine(
+ I vertexIndex, V originalMessage, V messageToCombine) {
+ valueTypeOps.plusInto(originalMessage, messageToCombine);
+ }
+
+ @Override
+ public V createInitialMessage() {
+ return valueTypeOps.createZero();
+ }
+ },
+ (vertex, value) -> valueTypeOps.set(vertex.getValue(), value),
+ new MessageCombiner<I, E>() {
+ @Override
+ public void combine(
+ I vertexIndex, E originalMessage, E messageToCombine) {
+ edgeTypeOps.plusInto(originalMessage, messageToCombine);
+ }
+
+ @Override
+ public E createInitialMessage() {
+ return edgeTypeOps.createZero();
+ }
+ },
+ originalNodesChecker,
+ clusterNodesChecker);
+ }
+
+ /**
+ * Uses configuration to figure out vertex id, value and edge value types,
+ * and calls above createCoarseningBlock with it, look at it's
+ * documentation for more details.
+ */
+ public static
+ <I extends WritableComparable, V extends Writable, E extends Writable>
+ Block createCoarseningBlock(
+ ImmutableClassesGiraphConfiguration<I, V, E> conf,
+ SupplierFromVertex<I, V, E, I> coarsenedVertexIdSupplier,
+ Function<I, Boolean> originalNodesChecker,
+ Function<I, Boolean> clusterNodesChecker) {
+ return createCoarseningBlock(
+ TypeOpsUtils.getPrimitiveIdTypeOps(conf.getVertexIdClass()),
+ (NumericTypeOps<V>)
+ TypeOpsUtils.getTypeOps(conf.getVertexValueClass()),
+ (NumericTypeOps<E>) TypeOpsUtils.getTypeOps(conf.getEdgeValueClass()),
+ coarsenedVertexIdSupplier, originalNodesChecker, clusterNodesChecker);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/package-info.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/package-info.java
new file mode 100644
index 0000000..16693fa
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/coarsening/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * Coarsening utils
+ */
+package org.apache.giraph.block_app.library.coarsening;
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/DirectedGraphStats.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/DirectedGraphStats.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/DirectedGraphStats.java
new file mode 100644
index 0000000..1739718
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/DirectedGraphStats.java
@@ -0,0 +1,187 @@
+/*
+ * 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.block_app.library.stats;
+
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.api.CreateReducersApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
+import org.apache.giraph.block_app.framework.piece.global_comm.map.ReducerMapHandle;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.block_app.library.VertexSuppliers;
+import org.apache.giraph.block_app.reducers.map.BasicMapReduce;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.reducers.impl.MaxReduce;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.giraph.types.ops.IntTypeOps;
+import org.apache.giraph.types.ops.LongTypeOps;
+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.apache.log4j.Logger;
+
+import com.google.common.collect.Iterables;
+
+/** Utility class for calculating stats of a directed graph */
+public class DirectedGraphStats {
+ private static final Logger LOG = Logger.getLogger(DirectedGraphStats.class);
+ private static final double LOG_2 = Math.log(2);
+ private static final int MAX_LOG_DEGREE = 20;
+
+ private DirectedGraphStats() { }
+
+ /**
+ * Calculate and print on master statistics about in and out degrees
+ * of all vertices.
+ */
+ public static <I extends WritableComparable>
+ Block createInAndOutDegreeStatsBlock(Class<I> idClass) {
+ ObjectTransfer<Iterable<I>> inEdges = new ObjectTransfer<>();
+
+ Block announceToNeighbors = Pieces.sendMessageToNeighbors(
+ "AnnounceToNeighbors",
+ idClass,
+ VertexSuppliers.<I, Writable, Writable>vertexIdSupplier(),
+ inEdges.<I, Writable, Writable>castToConsumer());
+
+ return new SequenceBlock(
+ announceToNeighbors,
+ new AggregateInAndOutDegreeStatsPiece<>(
+ inEdges.<I, Writable, Writable>castToSupplier()));
+ }
+
+ /** Aggregating in and out degree statistics */
+ private static class AggregateInAndOutDegreeStatsPiece
+ <I extends WritableComparable>
+ extends Piece<I, Writable, Writable, Writable, Object> {
+ private final
+ SupplierFromVertex<I, Writable, Writable, Iterable<I>> inEdges;
+
+ private ReducerHandle<IntWritable, IntWritable> maxDegreeAgg;
+
+ private ReducerMapHandle<IntWritable, LongWritable, LongWritable>
+ inHistograms;
+ private ReducerMapHandle<IntWritable, LongWritable, LongWritable>
+ outHistograms;
+
+ private ReducerMapHandle<IntWritable, LongWritable, LongWritable>
+ inVsOutHistograms;
+
+
+ public AggregateInAndOutDegreeStatsPiece(
+ SupplierFromVertex<I, Writable, Writable, Iterable<I>> inEdges) {
+ this.inEdges = inEdges;
+ }
+
+ @Override
+ public void registerReducers(
+ CreateReducersApi reduceApi, Object executionStage) {
+ inHistograms = BasicMapReduce.createLocalMapHandles(
+ IntTypeOps.INSTANCE, LongTypeOps.INSTANCE, SumReduce.LONG, reduceApi);
+ outHistograms = BasicMapReduce.createLocalMapHandles(
+ IntTypeOps.INSTANCE, LongTypeOps.INSTANCE, SumReduce.LONG, reduceApi);
+
+ inVsOutHistograms = BasicMapReduce.createLocalMapHandles(
+ IntTypeOps.INSTANCE, LongTypeOps.INSTANCE, SumReduce.LONG, reduceApi);
+
+ maxDegreeAgg =
+ reduceApi.createLocalReducer(new MaxReduce<>(IntTypeOps.INSTANCE));
+ }
+
+ @Override
+ public VertexSender<I, Writable, Writable> getVertexSender(
+ BlockWorkerSendApi<I, Writable, Writable, Writable> workerApi,
+ Object executionStage) {
+ final IntWritable indexWrap = new IntWritable();
+
+ return new InnerVertexSender() {
+ @Override
+ public void vertexSend(Vertex<I, Writable, Writable> vertex) {
+ Iterable<I> in = inEdges.get(vertex);
+
+ int inCount = Iterables.size(in);
+ int outCount = vertex.getNumEdges();
+
+ reduceInt(maxDegreeAgg, Math.max(inCount, outCount));
+ increment(inHistograms, inCount);
+ increment(outHistograms, outCount);
+ increment(inVsOutHistograms,
+ log2(inCount + 1) * MAX_LOG_DEGREE + log2(outCount + 1));
+
+ // TODO add count for common edges.
+ }
+
+ private int log2(int value) {
+ return (int) (Math.log(value) / LOG_2);
+ }
+
+ private void increment(
+ ReducerMapHandle<IntWritable, LongWritable, LongWritable>
+ reduceHandle,
+ int index) {
+ indexWrap.set(index);
+ reduceLong(inHistograms.get(indexWrap), 1);
+ }
+ };
+ }
+
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+
+ int maxDegree = maxDegreeAgg.getReducedValue(master).get();
+ LOG.info("Max degree : " + maxDegree);
+
+ StringBuilder sb = new StringBuilder("In and out degree histogram:\n");
+ sb.append("degree\tnumIn\tnumOut\n");
+
+ final IntWritable index = new IntWritable();
+
+ for (int i = 0; i <= maxDegree; i++) {
+ index.set(i);
+ long numIn = inHistograms.get(index).getReducedValue(master).get();
+ long numOut = outHistograms.get(index).getReducedValue(master).get();
+ if (numIn > 0 || numOut > 0) {
+ sb.append(i + "\t" + numIn + "\t" + numOut + "\n");
+ }
+ }
+ LOG.info(sb);
+
+ sb = new StringBuilder("In vs out degree log/log histogram:\n");
+ sb.append("<inDeg\t<outDeg\tnum\n");
+
+ for (int in = 0; in < MAX_LOG_DEGREE; in++) {
+ for (int out = 0; out < MAX_LOG_DEGREE; out++) {
+ index.set(in * MAX_LOG_DEGREE + out);
+ long num = inVsOutHistograms.get(index).getReducedValue(master).get();
+ if (num > 0) {
+ sb.append(Math.pow(2, in) + "\t" + Math.pow(2, out) +
+ "\t" + num + "\n");
+ }
+ }
+ }
+ LOG.info(sb);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/PartitioningStats.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/PartitioningStats.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/PartitioningStats.java
new file mode 100644
index 0000000..950c144
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/PartitioningStats.java
@@ -0,0 +1,125 @@
+/*
+ * 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.block_app.library.stats;
+
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongSet;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.library.SendMessageChain;
+import org.apache.giraph.function.primitive.DoubleConsumer;
+import org.apache.giraph.function.vertex.SupplierFromVertex;
+import org.apache.giraph.reducers.impl.PairReduce;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.log4j.Logger;
+
+/**
+ * Utility blocks for calculating stats for a given partitioning - an
+ * assignment of vertices to buckets.
+ */
+public class PartitioningStats {
+ private static final Logger LOG = Logger.getLogger(PartitioningStats.class);
+
+ private PartitioningStats() { }
+
+ /**
+ * Calculate edge locality - ratio of edges that are within a same bucket.
+ */
+ public static <V extends Writable> Block calculateEdgeLocality(
+ SupplierFromVertex<WritableComparable, V, Writable, LongWritable>
+ bucketSupplier,
+ DoubleConsumer edgeLocalityConsumer) {
+ final Pair<LongWritable, LongWritable> pair =
+ Pair.of(new LongWritable(), new LongWritable());
+ return SendMessageChain.<WritableComparable, V, Writable, LongWritable>
+ startSendToNeighbors(
+ "CalcLocalEdgesPiece",
+ LongWritable.class,
+ bucketSupplier
+ ).endReduceWithMaster(
+ "AggregateEdgeLocalityPiece",
+ new PairReduce<>(SumReduce.LONG, SumReduce.LONG),
+ (vertex, messages) -> {
+ long bucket = bucketSupplier.get(vertex).get();
+ int local = 0;
+ int total = 0;
+ for (LongWritable otherCluster : messages) {
+ total++;
+ if (bucket == otherCluster.get()) {
+ local++;
+ }
+ }
+ pair.getLeft().set(local);
+ pair.getRight().set(total);
+ return pair;
+ },
+ (reducedPair, master) -> {
+ long localEdges = reducedPair.getLeft().get();
+ long totalEdges = reducedPair.getRight().get();
+ double edgeLocality = (double) localEdges / totalEdges;
+ LOG.info("locality ratio = " + edgeLocality);
+ master.getCounter(
+ "Edge locality stats", "edge locality (in percent * 1000)")
+ .setValue((long) (edgeLocality * 100000));
+ edgeLocalityConsumer.apply(edgeLocality);
+ }
+ );
+ }
+
+ /**
+ * Calculates average fanout - average number of distinct buckets that vertex
+ * has neighbors in.
+ */
+ public static <V extends Writable> Block calculateFanout(
+ SupplierFromVertex<WritableComparable, V, Writable, LongWritable>
+ bucketSupplier,
+ DoubleConsumer averageFanoutConsumer) {
+ final Pair<LongWritable, LongWritable> pair =
+ Pair.of(new LongWritable(), new LongWritable(1));
+ return SendMessageChain.<WritableComparable, V, Writable, LongWritable>
+ startSendToNeighbors(
+ "CalcFanoutPiece",
+ LongWritable.class,
+ bucketSupplier
+ ).endReduceWithMaster(
+ "AggregateFanoutPiece",
+ new PairReduce<>(SumReduce.LONG, SumReduce.LONG),
+ (vertex, messages) -> {
+ LongSet setOfNeighborBuckets = new LongOpenHashSet();
+ for (LongWritable neighborBucket : messages) {
+ setOfNeighborBuckets.add(neighborBucket.get());
+ }
+ pair.getLeft().set(setOfNeighborBuckets.size());
+ return pair;
+ },
+ (reducedPair, master) -> {
+ long fanout = reducedPair.getLeft().get();
+ long numVertices = reducedPair.getRight().get();
+ double avgFanout = (double) fanout / numVertices;
+ LOG.info("fanout ratio = " + avgFanout);
+ master.getCounter("Fanout stats", "fanout (in percent * 1000)")
+ .setValue((long) (avgFanout * 100000));
+ averageFanoutConsumer.apply(avgFanout);
+ }
+ );
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/package-info.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/package-info.java b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/package-info.java
new file mode 100644
index 0000000..4240dd7
--- /dev/null
+++ b/giraph-block-app-8/src/main/java/org/apache/giraph/block_app/library/stats/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * Utilities for calculating graph statistics.
+ */
+package org.apache.giraph.block_app.library.stats;
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/AbstractPageRankExampleBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/AbstractPageRankExampleBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/AbstractPageRankExampleBlockFactory.java
new file mode 100644
index 0000000..13dfbdb
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/AbstractPageRankExampleBlockFactory.java
@@ -0,0 +1,54 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.AbstractBlockFactory;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.IntConfOption;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+
+/**
+ * Parent class for PageRank example of using BlockFactory. Contains same typing information needed
+ * for all examples.
+ */
+public abstract class AbstractPageRankExampleBlockFactory extends AbstractBlockFactory<Object> {
+ public static final IntConfOption NUM_ITERATIONS = new IntConfOption(
+ "page_rank_example.num_iterations", 10, "num iterations");
+
+ @Override
+ public Object createExecutionStage(GiraphConfiguration conf) {
+ return new Object();
+ }
+
+ @Override
+ protected Class<LongWritable> getVertexIDClass(GiraphConfiguration conf) {
+ return LongWritable.class;
+ }
+
+ @Override
+ protected Class<DoubleWritable> getVertexValueClass(GiraphConfiguration conf) {
+ return DoubleWritable.class;
+ }
+
+ @Override
+ protected Class<NullWritable> getEdgeValueClass(GiraphConfiguration conf) {
+ return NullWritable.class;
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankExampleBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankExampleBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankExampleBlockFactory.java
new file mode 100644
index 0000000..dd70c0e
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankExampleBlockFactory.java
@@ -0,0 +1,47 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatBlock;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * PageRank example of using BlockFactory - in it's simplest form, using functional primitives.
+ */
+public class PageRankExampleBlockFactory extends AbstractPageRankExampleBlockFactory {
+ @Override
+ @SuppressWarnings("rawtypes")
+ public Block createBlock(GiraphConfiguration conf) {
+ Block iter = Pieces.<WritableComparable, DoubleWritable, Writable, DoubleWritable>
+ sendMessageToNeighbors(
+ "IterationPiece",
+ SumMessageCombiner.DOUBLE,
+ (vertex) -> new DoubleWritable(vertex.getValue().get() / vertex.getNumEdges()),
+ (vertex, value) -> {
+ double sum = value != null ? value.get() : 0;
+ vertex.getValue().set(0.15f + 0.85f * sum);
+ });
+ return new RepeatBlock(NUM_ITERATIONS.get(conf), iter);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithConvergenceExampleBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithConvergenceExampleBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithConvergenceExampleBlockFactory.java
new file mode 100644
index 0000000..1585557
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithConvergenceExampleBlockFactory.java
@@ -0,0 +1,70 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatUntilBlock;
+import org.apache.giraph.block_app.library.SendMessageChain;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * PageRank example with convergence check, using functional primitives and SendMessageChain
+ * for send/receive/reply logic.
+ */
+public class PageRankWithConvergenceExampleBlockFactory
+ extends AbstractPageRankExampleBlockFactory {
+ private static final double EPS = 1e-3;
+ private static final LongWritable ONE = new LongWritable(1);
+ private static final LongWritable ZERO = new LongWritable(0);
+
+ @Override
+ @SuppressWarnings("rawtypes")
+ public Block createBlock(GiraphConfiguration conf) {
+ ObjectTransfer<Boolean> converged = new ObjectTransfer<>();
+
+ Block iter = SendMessageChain.<WritableComparable, DoubleWritable, Writable, DoubleWritable>
+ startSendToNeighbors(
+ "PageRankUpdate",
+ SumMessageCombiner.DOUBLE,
+ (vertex) -> new DoubleWritable(vertex.getValue().get() / vertex.getNumEdges())
+ ).endReduce(
+ "PageRankCheckConvergence",
+ SumReduce.LONG,
+ (vertex, value) -> {
+ double sum = value != null ? value.get() : 0;
+ double newValue = 0.15f + 0.85f * sum;
+ double change = Math.abs(newValue - vertex.getValue().get());
+ vertex.getValue().set(newValue);
+ return (change > EPS) ? ONE : ZERO;
+ },
+ (changingCount) -> converged.apply(changingCount.get() == 0)
+ );
+
+ return new RepeatUntilBlock(
+ NUM_ITERATIONS.get(conf),
+ iter,
+ converged);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesAndConvergenceExampleBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesAndConvergenceExampleBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesAndConvergenceExampleBlockFactory.java
new file mode 100644
index 0000000..878bf1f
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesAndConvergenceExampleBlockFactory.java
@@ -0,0 +1,146 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.api.BlockMasterApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.api.CreateReducersApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatUntilBlock;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.combiner.MessageCombiner;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.giraph.function.Consumer;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.function.Supplier;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.giraph.types.NoMessage;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * PageRank example with convergence check.
+ */
+@SuppressWarnings("rawtypes")
+public class PageRankWithPiecesAndConvergenceExampleBlockFactory extends AbstractPageRankExampleBlockFactory {
+ private static final double EPS = 1e-3;
+
+ @Override
+ public Block createBlock(GiraphConfiguration conf) {
+ ObjectTransfer<Boolean> converged = new ObjectTransfer<>();
+ ObjectTransfer<Double> vertexValueChange = new ObjectTransfer<>();
+ return new RepeatUntilBlock(
+ NUM_ITERATIONS.get(conf),
+ new SequenceBlock(
+ new PageRankUpdatePiece(vertexValueChange),
+ new PageRankConvergencePiece(vertexValueChange, converged)
+ ),
+ converged);
+ }
+
+ /** One PageRank iteration */
+ public static class PageRankUpdatePiece
+ extends Piece<WritableComparable, DoubleWritable, Writable, DoubleWritable, Object> {
+ private final Consumer<Double> changeConsumer;
+
+ public PageRankUpdatePiece(Consumer<Double> changeConsumer) {
+ this.changeConsumer = changeConsumer;
+ }
+
+ @Override
+ public VertexSender<WritableComparable, DoubleWritable, Writable> getVertexSender(
+ final BlockWorkerSendApi<WritableComparable, DoubleWritable, Writable,
+ DoubleWritable> workerApi,
+ Object executionStage) {
+ DoubleWritable message = new DoubleWritable();
+ return (vertex) -> {
+ message.set(vertex.getValue().get() / vertex.getNumEdges());
+ workerApi.sendMessageToAllEdges(vertex, message);
+ };
+ }
+
+ @Override
+ public VertexReceiver<WritableComparable, DoubleWritable, Writable, DoubleWritable>
+ getVertexReceiver(BlockWorkerReceiveApi<WritableComparable> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ double sum = 0;
+ for (DoubleWritable value : messages) {
+ sum += value.get();
+ }
+ double newValue = 0.15f + 0.85f * sum;
+ changeConsumer.apply(Math.abs(newValue - vertex.getValue().get()));
+ vertex.getValue().set(newValue);
+ };
+ }
+
+ @Override
+ public MessageCombiner<? super WritableComparable, DoubleWritable> getMessageCombiner(
+ ImmutableClassesGiraphConfiguration conf) {
+ return SumMessageCombiner.DOUBLE;
+ }
+ }
+
+ /** PageRank convergence check */
+ public static class PageRankConvergencePiece
+ extends Piece<WritableComparable, DoubleWritable, Writable, NoMessage, Object> {
+ private ReducerHandle<LongWritable, LongWritable> countModified;
+
+ private final Supplier<Double> changeSupplier;
+ private final Consumer<Boolean> converged;
+
+ public PageRankConvergencePiece(
+ Supplier<Double> changeSupplier,
+ Consumer<Boolean> converged) {
+ this.changeSupplier = changeSupplier;
+ this.converged = converged;
+ }
+
+ @Override
+ public void registerReducers(CreateReducersApi reduceApi, Object executionStage) {
+ countModified = reduceApi.createLocalReducer(SumReduce.LONG);
+ }
+
+ @Override
+ public VertexSender<WritableComparable, DoubleWritable, Writable>
+ getVertexSender(
+ BlockWorkerSendApi<WritableComparable, DoubleWritable, Writable, NoMessage> workerApi,
+ Object executionStage) {
+ return (vertex) -> {
+ double change = changeSupplier.get();
+ if (change > EPS) {
+ reduceLong(countModified, 1);
+ }
+ };
+ }
+
+ @Override
+ public void masterCompute(BlockMasterApi master, Object executionStage) {
+ LongWritable count = countModified.getReducedValue(master);
+ converged.apply(count.get() == 0);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesExampleBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesExampleBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesExampleBlockFactory.java
new file mode 100644
index 0000000..8881752
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithPiecesExampleBlockFactory.java
@@ -0,0 +1,83 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
+import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatBlock;
+import org.apache.giraph.block_app.framework.piece.Piece;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
+import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
+import org.apache.giraph.combiner.MessageCombiner;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * PageRank example of using BlockFactory - in it's simplest form.
+ * This single class represents everything needed for the application.
+ * To use it - set BlockUtils.BLOCK_FACTORY_CLASS property to this class.
+ *
+ * Note, as a general practice, for iteration Piece to be reusable, it should not
+ * assume fixed vertex value class, but specify interface it needs, and use it instead.
+ */
+public class PageRankWithPiecesExampleBlockFactory extends AbstractPageRankExampleBlockFactory {
+ @Override
+ public Block createBlock(GiraphConfiguration conf) {
+ return new RepeatBlock(NUM_ITERATIONS.get(conf), new PageRankUpdatePiece());
+ }
+
+ /** One PageRank iteration */
+ @SuppressWarnings("rawtypes")
+ public static class PageRankUpdatePiece
+ extends Piece<WritableComparable, DoubleWritable, Writable, DoubleWritable, Object> {
+ @Override
+ public VertexSender<WritableComparable, DoubleWritable, Writable>
+ getVertexSender(BlockWorkerSendApi<WritableComparable, DoubleWritable, Writable,
+ DoubleWritable> workerApi,
+ Object executionStage) {
+ DoubleWritable message = new DoubleWritable();
+ return (vertex) -> {
+ message.set(vertex.getValue().get() / vertex.getNumEdges());
+ workerApi.sendMessageToAllEdges(vertex, message);
+ };
+ }
+
+ @Override
+ public VertexReceiver<WritableComparable, DoubleWritable, Writable, DoubleWritable>
+ getVertexReceiver(BlockWorkerReceiveApi<WritableComparable> workerApi, Object executionStage) {
+ return (vertex, messages) -> {
+ double sum = 0;
+ for (DoubleWritable value : messages) {
+ sum += value.get();
+ }
+ vertex.getValue().set(0.15f + 0.85f * sum);
+ };
+ }
+
+ @Override
+ public MessageCombiner<? super WritableComparable, DoubleWritable> getMessageCombiner(
+ ImmutableClassesGiraphConfiguration conf) {
+ return SumMessageCombiner.DOUBLE;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithTransferAndConvergenceExampleBlockFactory.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithTransferAndConvergenceExampleBlockFactory.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithTransferAndConvergenceExampleBlockFactory.java
new file mode 100644
index 0000000..c939701
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/PageRankWithTransferAndConvergenceExampleBlockFactory.java
@@ -0,0 +1,74 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.block.Block;
+import org.apache.giraph.block_app.framework.block.RepeatUntilBlock;
+import org.apache.giraph.block_app.framework.block.SequenceBlock;
+import org.apache.giraph.block_app.library.Pieces;
+import org.apache.giraph.combiner.SumMessageCombiner;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.function.ObjectTransfer;
+import org.apache.giraph.reducers.impl.SumReduce;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+/**
+ * PageRank example with convergence check, using functional primitives
+ */
+public class PageRankWithTransferAndConvergenceExampleBlockFactory
+ extends AbstractPageRankExampleBlockFactory {
+ private static final double EPS = 1e-3;
+ private static final LongWritable ONE = new LongWritable(1);
+ private static final LongWritable ZERO = new LongWritable(0);
+
+ @Override
+ @SuppressWarnings("rawtypes")
+ public Block createBlock(GiraphConfiguration conf) {
+ ObjectTransfer<Boolean> converged = new ObjectTransfer<>();
+ ObjectTransfer<Double> vertexValueChange = new ObjectTransfer<>();
+
+ Block iter = Pieces.<WritableComparable, DoubleWritable, Writable, DoubleWritable>
+ sendMessageToNeighbors(
+ "IterationPiece",
+ SumMessageCombiner.DOUBLE,
+ (vertex) -> new DoubleWritable(vertex.getValue().get() / vertex.getNumEdges()),
+ (vertex, value) -> {
+ double sum = value != null ? value.get() : 0;
+ double newValue = 0.15f + 0.85f * sum;
+ vertexValueChange.apply(Math.abs(newValue - vertex.getValue().get()));
+ vertex.getValue().set(newValue);
+ });
+
+ Block checkConverged = Pieces.reduce(
+ "CheckConvergedPiece",
+ SumReduce.LONG,
+ (vertex) -> {
+ double change = vertexValueChange.get();
+ return (change > EPS) ? ONE : ZERO;
+ },
+ (changingCount) -> converged.apply(changingCount.get() == 0));
+
+ return new RepeatUntilBlock(
+ NUM_ITERATIONS.get(conf),
+ new SequenceBlock(iter, checkConverged),
+ converged);
+ }
+}
http://git-wip-us.apache.org/repos/asf/giraph/blob/3b7c68e5/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/TestPageRankExample.java
----------------------------------------------------------------------
diff --git a/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/TestPageRankExample.java b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/TestPageRankExample.java
new file mode 100644
index 0000000..b6b65ba
--- /dev/null
+++ b/giraph-block-app-8/src/test/java/org/apache/giraph/block_app/examples/pagerank/TestPageRankExample.java
@@ -0,0 +1,127 @@
+/*
+ * 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.block_app.examples.pagerank;
+
+import org.apache.giraph.block_app.framework.BlockFactory;
+import org.apache.giraph.block_app.framework.BlockUtils;
+import org.apache.giraph.block_app.test_setup.TestGraphUtils;
+import org.apache.giraph.block_app.test_setup.graphs.EachVertexInit;
+import org.apache.giraph.block_app.test_setup.graphs.Small1GraphInit;
+import org.apache.giraph.conf.GiraphConfiguration;
+import org.apache.giraph.conf.GiraphConstants;
+import org.apache.hadoop.io.DoubleWritable;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.WritableComparable;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestPageRankExample {
+ private <I extends WritableComparable>
+ void testTenIterations(final Class<I> type,
+ final Class<? extends BlockFactory<?>> factory) throws Exception {
+ TestGraphUtils.runTest(
+ TestGraphUtils.chainModifiers(
+ new Small1GraphInit<LongWritable, DoubleWritable, NullWritable>(),
+ new EachVertexInit<>((vertex) -> vertex.getValue().set(1.0))),
+ (graph) -> {
+ float outside = 0.8759f;
+ float inside = 1.2481f;
+ float isolated = 0.15f;
+
+ for (int i : new int[] {0, 1, 4, 5}) {
+ Assert.assertEquals(outside, graph.getValue(i).get(), 0.001);
+ }
+ for (int i : new int[] {2, 3}) {
+ Assert.assertEquals(inside, graph.getValue(i).get(), 0.001);
+ }
+ for (int i : new int[] {6}) {
+ Assert.assertEquals(isolated, graph.getValue(i).get(), 0.001);
+ }
+ },
+ (GiraphConfiguration conf) -> {
+ GiraphConstants.VERTEX_ID_CLASS.set(conf, type);
+ BlockUtils.setBlockFactoryClass(conf, factory);
+ AbstractPageRankExampleBlockFactory.NUM_ITERATIONS.set(conf, 10);
+ });
+ }
+
+ private void testTenIterationsOnAllTypes(Class<? extends BlockFactory<?>> factory)
+ throws Exception {
+ testTenIterations(
+ IntWritable.class,
+ factory);
+ testTenIterations(
+ LongWritable.class,
+ factory);
+ }
+
+ private <I extends WritableComparable>
+ void testConverging(
+ final Class<I> type, Class<? extends BlockFactory<?>> factory) throws Exception {
+ TestGraphUtils.runTest(
+ new Small1GraphInit<LongWritable, DoubleWritable, NullWritable>(),
+ (graph) -> {
+ },
+ (GiraphConfiguration conf) -> {
+ GiraphConstants.VERTEX_ID_CLASS.set(conf, type);
+ BlockUtils.setBlockFactoryClass(conf, factory);
+ AbstractPageRankExampleBlockFactory.NUM_ITERATIONS.set(conf, Integer.MAX_VALUE);
+ });
+ // Test is that this doesn't loop forever, even though num_iterations=Integer.MAX_VALUE
+ }
+
+ private <I extends WritableComparable>
+ void testConvergingOnAllTypes(Class<? extends BlockFactory<?>> factory) throws Exception {
+ testConverging(
+ IntWritable.class,
+ factory);
+ testConverging(
+ LongWritable.class,
+ factory);
+ }
+
+ @Test
+ public void testExample() throws Exception {
+ testTenIterationsOnAllTypes(PageRankWithPiecesExampleBlockFactory.class);
+ }
+
+ @Test
+ public void testConvergenceExample() throws Exception {
+ testTenIterationsOnAllTypes(PageRankWithPiecesAndConvergenceExampleBlockFactory.class);
+ testConvergingOnAllTypes(PageRankWithPiecesAndConvergenceExampleBlockFactory.class);
+ }
+
+ @Test
+ public void testFunctionalExample() throws Exception {
+ testTenIterationsOnAllTypes(PageRankExampleBlockFactory.class);
+ }
+
+ @Test
+ public void testFunctionalConvergenceExample() throws Exception {
+ testTenIterationsOnAllTypes(PageRankWithTransferAndConvergenceExampleBlockFactory.class);
+ testConvergingOnAllTypes(PageRankWithTransferAndConvergenceExampleBlockFactory.class);
+ }
+
+ @Test
+ public void testFunctionalChainConvergenceExample() throws Exception {
+ testTenIterationsOnAllTypes(PageRankWithConvergenceExampleBlockFactory.class);
+ testConvergingOnAllTypes(PageRankWithConvergenceExampleBlockFactory.class);
+ }
+}