/*

  * 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.examples;

 import org.apache.giraph.graph.BasicComputation;

 import org.apache.giraph.edge.Edge;

 import org.apache.giraph.graph.Vertex;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.NullWritable;

 import java.io.IOException;

 /**

  * Implementation of the HCC algorithm that identifies connected components and

  * assigns each vertex its "component identifier" (the smallest vertex id

  * in the component)

  *

  * The idea behind the algorithm is very simple: propagate the smallest

  * vertex id along the edges to all vertices of a connected component. The

  * number of supersteps necessary is equal to the length of the maximum

  * diameter of all components + 1

  *

  * The original Hadoop-based variant of this algorithm was proposed by Kang,

  * Charalampos, Tsourakakis and Faloutsos in

  * "PEGASUS: Mining Peta-Scale Graphs", 2010

  *

  * http://www.cs.cmu.edu/~ukang/papers/PegasusKAIS.pdf

  */

 @Algorithm(

     name = "Connected components",

     description = "Finds connected components of the graph"

 )

 public class ConnectedComponentsComputation extends

     BasicComputation<IntWritable, IntWritable, NullWritable, IntWritable> {

   /**

    * Propagates the smallest vertex id to all neighbors. Will always choose to

    * halt and only reactivate if a smaller id has been sent to it.

    *

    * @param vertex Vertex

    * @param messages Iterator of messages from the previous superstep.

    * @throws IOException

    */

   @Override

   public void compute(

       Vertex<IntWritable, IntWritable, NullWritable> vertex,

       Iterable<IntWritable> messages) throws IOException {

     int currentComponent = vertex.getValue().get();

     // First superstep is special, because we can simply look at the neighbors

     if (getSuperstep() == 0) {

       for (Edge<IntWritable, NullWritable> edge : vertex.getEdges()) {

         int neighbor = edge.getTargetVertexId().get();

         if (neighbor < currentComponent) {

           currentComponent = neighbor;

         }

       }

       // Only need to send value if it is not the own id

       if (currentComponent != vertex.getValue().get()) {

         vertex.setValue(new IntWritable(currentComponent));

         for (Edge<IntWritable, NullWritable> edge : vertex.getEdges()) {

           IntWritable neighbor = edge.getTargetVertexId();

           if (neighbor.get() > currentComponent) {

             sendMessage(neighbor, vertex.getValue());

           }

         }

       }

       vertex.voteToHalt();

       return;

     }

     boolean changed = false;

     // did we get a smaller id ?

     for (IntWritable message : messages) {

       int candidateComponent = message.get();

       if (candidateComponent < currentComponent) {

         currentComponent = candidateComponent;

         changed = true;

       }

     }

     // propagate new component id to the neighbors

     if (changed) {

       vertex.setValue(new IntWritable(currentComponent));

       sendMessageToAllEdges(vertex, vertex.getValue());

     }

     vertex.voteToHalt();

   }

 }