/*

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

 import org.apache.giraph.aggregators.BooleanOrAggregator;

 import org.apache.giraph.aggregators.IntOverwriteAggregator;

 import org.apache.giraph.master.DefaultMasterCompute;

 import org.apache.hadoop.io.BooleanWritable;

 import org.apache.hadoop.io.IntWritable;

 /**

  * This master compute keeps track of what phase is being currently executed by

  * the Strongly Connected Components computation. The phases comprehend the

  * following: 1 - Transpose (comprehends 2 supersteps, one to propagate parent

  * vertices ids and another one to store them by their respective children) 2 -

  * Trimming (this phase can happen multiple times) 3 - Forward Traversal 4 -

  * Backward Traversal

  */

 public class SccPhaseMasterCompute extends DefaultMasterCompute {

   /**

    * Aggregator that stores the current phase

    */

   public static final String PHASE = "scccompute.phase";

   /**

    * Flags whether a new maximum was found in the Forward Traversal phase

    */

   public static final String NEW_MAXIMUM = "scccompute.max";

   /**

    * Flags whether a vertex converged in the Backward Traversal phase

    */

   public static final String CONVERGED = "scccompute.converged";

   /**

    * Enumerates the possible phases of the algorithm.

    */

   public enum Phases {

     /** Tranpose and Trimming phases **/

     TRANSPOSE, TRIMMING,

     /** Maximum id propagation **/

     FORWARD_TRAVERSAL,

     /** Vertex convergence in SCC **/

     BACKWARD_TRAVERSAL_START, BACKWARD_TRAVERSAL_REST

   };

   @Override

   public void initialize() throws InstantiationException,

       IllegalAccessException {

     registerPersistentAggregator(PHASE, IntOverwriteAggregator.class);

     registerAggregator(NEW_MAXIMUM, BooleanOrAggregator.class);

     registerAggregator(CONVERGED, BooleanOrAggregator.class);

   }

   @Override

   public void compute() {

     if (getSuperstep() == 0) {

       setPhase(Phases.TRANSPOSE);

     } else {

       Phases currPhase = getPhase();

       switch (currPhase) {

       case TRANSPOSE:

         setPhase(Phases.TRIMMING);

         break;

       case TRIMMING :

         setPhase(Phases.FORWARD_TRAVERSAL);

         break;

       case FORWARD_TRAVERSAL :

         BooleanWritable newMaxFound = getAggregatedValue(NEW_MAXIMUM);

         // If no new maximum value was found it means the propagation

         // converged, so we can move to the next phase

         if (!newMaxFound.get()) {

           setPhase(Phases.BACKWARD_TRAVERSAL_START);

         }

         break;

       case BACKWARD_TRAVERSAL_START :

         setPhase(Phases.BACKWARD_TRAVERSAL_REST);

         break;

       case BACKWARD_TRAVERSAL_REST :

         BooleanWritable converged = getAggregatedValue(CONVERGED);

         if (!converged.get()) {

           setPhase(Phases.TRANSPOSE);

         }

         break;

       default :

         break;

       }

     }

   }

   /**

    * Sets the next phase of the algorithm.

    * @param phase

    *          Next phase.

    */

   private void setPhase(Phases phase) {

     setAggregatedValue(PHASE, new IntWritable(phase.ordinal()));

   }

   /**

    * Get current phase.

    * @return Current phase as enumerator.

    */

   private Phases getPhase() {

     IntWritable phaseInt = getAggregatedValue(PHASE);

     return getPhase(phaseInt);

   }

   /**

    * Helper function to convert from internal aggregated value to a Phases

    * enumerator.

    * @param phaseInt

    *          An integer that matches a position in the Phases enumerator.

    * @return A Phases' item for the given position.

    */

   public static Phases getPhase(IntWritable phaseInt) {

     return Phases.values()[phaseInt.get()];

   }

 }