/*

  * 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.aggregators.DoubleMaxAggregator;

 import org.apache.giraph.aggregators.DoubleMinAggregator;

 import org.apache.giraph.aggregators.LongSumAggregator;

 import org.apache.giraph.graph.BspUtils;

 import org.apache.giraph.graph.DefaultMasterCompute;

 import org.apache.giraph.graph.LongDoubleFloatDoubleVertex;

 import org.apache.giraph.graph.Vertex;

 import org.apache.giraph.graph.VertexReader;

 import org.apache.giraph.graph.VertexWriter;

 import org.apache.giraph.graph.WorkerContext;

 import org.apache.giraph.io.GeneratedVertexInputFormat;

 import org.apache.giraph.io.TextVertexOutputFormat;

 import org.apache.giraph.io.TextVertexOutputFormat.TextVertexWriter;

 import org.apache.hadoop.io.DoubleWritable;

 import org.apache.hadoop.io.FloatWritable;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.InputSplit;

 import org.apache.hadoop.mapreduce.RecordWriter;

 import org.apache.hadoop.mapreduce.TaskAttemptContext;

 import org.apache.log4j.Logger;

 import com.google.common.collect.Maps;

 import java.io.IOException;

 import java.util.Map;

 /**

  * Demonstrates the basic Pregel PageRank implementation.

  */

 @Algorithm(

     name = "Page rank"

 )

 public class SimplePageRankVertex extends LongDoubleFloatDoubleVertex {

   /** Number of supersteps for this test */

   public static final int MAX_SUPERSTEPS = 30;

   /** Logger */

   private static final Logger LOG =

       Logger.getLogger(SimplePageRankVertex.class);

   /** Sum aggregator name */

   private static String SUM_AGG = "sum";

   /** Min aggregator name */

   private static String MIN_AGG = "min";

   /** Max aggregator name */

   private static String MAX_AGG = "max";

   @Override

   public void compute(Iterable<DoubleWritable> messages) {

     if (getSuperstep() >= 1) {

       double sum = 0;

       for (DoubleWritable message : messages) {

         sum += message.get();

       }

       DoubleWritable vertexValue =

           new DoubleWritable((0.15f / getTotalNumVertices()) + 0.85f * sum);

       setValue(vertexValue);

       aggregate(MAX_AGG, vertexValue);

       aggregate(MIN_AGG, vertexValue);

       aggregate(SUM_AGG, new LongWritable(1));

       LOG.info(getId() + ": PageRank=" + vertexValue +

           " max=" + getAggregatedValue(MAX_AGG) +

           " min=" + getAggregatedValue(MIN_AGG));

     }

     if (getSuperstep() < MAX_SUPERSTEPS) {

       long edges = getNumEdges();

       sendMessageToAllEdges(

           new DoubleWritable(getValue().get() / edges));

     } else {

       voteToHalt();

     }

   }

   /**

    * Worker context used with {@link SimplePageRankVertex}.

    */

   public static class SimplePageRankVertexWorkerContext extends

       WorkerContext {

     /** Final max value for verification for local jobs */

     private static double FINAL_MAX;

     /** Final min value for verification for local jobs */

     private static double FINAL_MIN;

     /** Final sum value for verification for local jobs */

     private static long FINAL_SUM;

     public static double getFinalMax() {

       return FINAL_MAX;

     }

     public static double getFinalMin() {

       return FINAL_MIN;

     }

     public static long getFinalSum() {

       return FINAL_SUM;

     }

     @Override

     public void preApplication()

       throws InstantiationException, IllegalAccessException {

     }

     @Override

     public void postApplication() {

       FINAL_SUM = this.<LongWritable>getAggregatedValue(SUM_AGG).get();

       FINAL_MAX = this.<DoubleWritable>getAggregatedValue(MAX_AGG).get();

       FINAL_MIN = this.<DoubleWritable>getAggregatedValue(MIN_AGG).get();

       LOG.info("aggregatedNumVertices=" + FINAL_SUM);

       LOG.info("aggregatedMaxPageRank=" + FINAL_MAX);

       LOG.info("aggregatedMinPageRank=" + FINAL_MIN);

     }

     @Override

     public void preSuperstep() {

       if (getSuperstep() >= 3) {

         LOG.info("aggregatedNumVertices=" +

             getAggregatedValue(SUM_AGG) +

             " NumVertices=" + getTotalNumVertices());

         if (this.<LongWritable>getAggregatedValue(SUM_AGG).get() !=

             getTotalNumVertices()) {

           throw new RuntimeException("wrong value of SumAggreg: " +

               getAggregatedValue(SUM_AGG) + ", should be: " +

               getTotalNumVertices());

         }

         DoubleWritable maxPagerank = getAggregatedValue(MAX_AGG);

         LOG.info("aggregatedMaxPageRank=" + maxPagerank.get());

         DoubleWritable minPagerank = getAggregatedValue(MIN_AGG);

         LOG.info("aggregatedMinPageRank=" + minPagerank.get());

       }

     }

     @Override

     public void postSuperstep() { }

   }

   /**

    * Master compute associated with {@link SimplePageRankVertex}.

    * It registers required aggregators.

    */

   public static class SimplePageRankVertexMasterCompute extends

       DefaultMasterCompute {

     @Override

     public void initialize() throws InstantiationException,

         IllegalAccessException {

       registerAggregator(SUM_AGG, LongSumAggregator.class);

       registerPersistentAggregator(MIN_AGG, DoubleMinAggregator.class);

       registerPersistentAggregator(MAX_AGG, DoubleMaxAggregator.class);

     }

   }

   /**

    * Simple VertexReader that supports {@link SimplePageRankVertex}

    */

   public static class SimplePageRankVertexReader extends

       GeneratedVertexReader<LongWritable, DoubleWritable, FloatWritable,

       DoubleWritable> {

     /** Class logger */

     private static final Logger LOG =

         Logger.getLogger(SimplePageRankVertexReader.class);

     /**

      * Constructor.

      */

     public SimplePageRankVertexReader() {

       super();

     }

     @Override

     public boolean nextVertex() {

       return totalRecords > recordsRead;

     }

     @Override

     public Vertex<LongWritable, DoubleWritable,

         FloatWritable, DoubleWritable>

     getCurrentVertex() throws IOException {

       Vertex<LongWritable, DoubleWritable, FloatWritable, DoubleWritable>

       vertex = BspUtils.createVertex(configuration);

       LongWritable vertexId = new LongWritable(

           (inputSplit.getSplitIndex() * totalRecords) + recordsRead);

       DoubleWritable vertexValue = new DoubleWritable(vertexId.get() * 10d);

       long targetVertexId =

           (vertexId.get() + 1) %

           (inputSplit.getNumSplits() * totalRecords);

       float edgeValue = vertexId.get() * 100f;

       Map<LongWritable, FloatWritable> edges = Maps.newHashMap();

       edges.put(new LongWritable(targetVertexId), new FloatWritable(edgeValue));

       vertex.initialize(vertexId, vertexValue, edges, null);

       ++recordsRead;

       if (LOG.isInfoEnabled()) {

         LOG.info("next: Return vertexId=" + vertex.getId().get() +

             ", vertexValue=" + vertex.getValue() +

             ", targetVertexId=" + targetVertexId + ", edgeValue=" + edgeValue);

       }

       return vertex;

     }

   }

   /**

    * Simple VertexInputFormat that supports {@link SimplePageRankVertex}

    */

   public static class SimplePageRankVertexInputFormat extends

     GeneratedVertexInputFormat<LongWritable,

             DoubleWritable, FloatWritable, DoubleWritable> {

     @Override

     public VertexReader<LongWritable, DoubleWritable,

     FloatWritable, DoubleWritable> createVertexReader(InputSplit split,

       TaskAttemptContext context)

       throws IOException {

       return new SimplePageRankVertexReader();

     }

   }

   /**

    * Simple VertexWriter that supports {@link SimplePageRankVertex}

    */

   public static class SimplePageRankVertexWriter extends

       TextVertexWriter<LongWritable, DoubleWritable, FloatWritable> {

     /**

      * Constructor with line writer.

      *

      * @param lineRecordWriter Line writer that will do the writing.

      */

     public SimplePageRankVertexWriter(

         RecordWriter<Text, Text> lineRecordWriter) {

       super(lineRecordWriter);

     }

     @Override

     public void writeVertex(

       Vertex<LongWritable, DoubleWritable, FloatWritable, ?> vertex)

       throws IOException, InterruptedException {

       getRecordWriter().write(

           new Text(vertex.getId().toString()),

           new Text(vertex.getValue().toString()));

     }

   }

   /**

    * Simple VertexOutputFormat that supports {@link SimplePageRankVertex}

    */

   public static class SimplePageRankVertexOutputFormat extends

       TextVertexOutputFormat<LongWritable, DoubleWritable, FloatWritable> {

     @Override

     public VertexWriter<LongWritable, DoubleWritable, FloatWritable>

     createVertexWriter(TaskAttemptContext context)

       throws IOException, InterruptedException {

       RecordWriter<Text, Text> recordWriter =

           textOutputFormat.getRecordWriter(context);

       return new SimplePageRankVertexWriter(recordWriter);

     }

   }

 }