/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
    * with the License.  You may obtain a copy of the License at
    *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.giraph.master;
  
  import java.util.List;
  
  import org.apache.giraph.aggregators.Aggregator;
  import org.apache.giraph.bsp.CentralizedServiceMaster;
  import org.apache.giraph.combiner.MessageCombiner;
  import org.apache.giraph.conf.DefaultImmutableClassesGiraphConfigurable;
  import org.apache.giraph.conf.MessageClasses;
  import org.apache.giraph.graph.Computation;
  import org.apache.giraph.graph.GraphState;
  import org.apache.giraph.reducers.ReduceOperation;
  import org.apache.giraph.worker.WorkerInfo;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.hadoop.mapreduce.Mapper;
  
  /**
   * Interface for defining a master vertex that can perform centralized
   * computation between supersteps. This class will be instantiated on the
   * master node and will run every superstep before the workers do.
   *
   * Communication with the workers should be performed via aggregators. The
   * values of the aggregators are broadcast to the workers before
   * vertex.compute() is called and collected by the master before
   * master.compute() is called. This means aggregator values used by the workers
   * are consistent with aggregator values from the master from the same
   * superstep and aggregator used by the master are consistent with aggregator
   * values from the workers from the previous superstep.
   */
  public abstract class MasterCompute
      extends DefaultImmutableClassesGiraphConfigurable
      implements MasterAggregatorUsage, MasterGlobalCommUsage, Writable {
    /** If true, do not do anymore computation on this vertex. */
    private boolean halt = false;
    /** Master aggregator usage */
    private CentralizedServiceMaster serviceMaster;
    /** Graph state */
    private GraphState graphState;
    /**
     * Computation and MessageCombiner classes used, which can be
     * switched by master
     */
    private SuperstepClasses superstepClasses;
  
    /**
     * Must be defined by user to specify what the master has to do.
     */
    public abstract void compute();
  
    /**
     * Initialize the MasterCompute class, this is the place to register
     * aggregators.
     *
     * @throws InstantiationException
     * @throws IllegalAccessException
     */
    public abstract void initialize() throws InstantiationException,
      IllegalAccessException;
  
    /**
     * Retrieves the current superstep.
     *
     * @return Current superstep
     */
    public final long getSuperstep() {
      return graphState.getSuperstep();
    }
  
    /**
     * Get the total (all workers) number of vertices that
     * existed in the previous superstep.
     *
     * @return Total number of vertices (-1 if first superstep)
     */
    public final long getTotalNumVertices() {
      return graphState.getTotalNumVertices();
    }
  
    /**
     * Get the total (all workers) number of edges that
    * existed in the previous superstep.
    *
    * @return Total number of edges (-1 if first superstep)
    */
   public final long getTotalNumEdges() {
     return graphState.getTotalNumEdges();
   }
 
   /**
    * After this is called, the computation will stop, even if there are
    * still messages in the system or vertices that have not voted to halt.
    */
   public final void haltComputation() {
     halt = true;
   }
 
   /**
    * Has the master halted?
    *
    * @return True if halted, false otherwise.
    */
   public final boolean isHalted() {
     return halt;
   }
 
   /**
    * Get the mapper context
    *
    * @return Mapper context
    */
   public final Mapper.Context getContext() {
     return graphState.getContext();
   }
 
   /**
    * Get list of workers
    *
    * @return List of workers
    */
   public final List<WorkerInfo> getWorkerInfoList() {
     return serviceMaster.getWorkerInfoList();
   }
 
   /**
    * Set Computation class to be used
    *
    * @param computationClass Computation class
    */
   public final void setComputation(
       Class<? extends Computation> computationClass) {
     superstepClasses.setComputationClass(computationClass);
   }
 
   /**
    * Get Computation class to be used
    *
    * @return Computation class
    */
   public final Class<? extends Computation> getComputation() {
     // Might be called prior to classes being set, do not return NPE
     if (superstepClasses == null) {
       return null;
     }
 
     return superstepClasses.getComputationClass();
   }
 
   /**
    * Set MessageCombiner class to be used
    *
    * @param combinerClass MessageCombiner class
    */
   public final void setMessageCombiner(
       Class<? extends MessageCombiner> combinerClass) {
     superstepClasses.setMessageCombinerClass(combinerClass);
   }
 
   /**
    * Get MessageCombiner class to be used
    *
    * @return MessageCombiner class
    */
   public final Class<? extends MessageCombiner> getMessageCombiner() {
     // Might be called prior to classes being set, do not return NPE
     if (superstepClasses == null) {
       return null;
     }
 
     return superstepClasses.getMessageCombinerClass();
   }
 
   /**
    * Set incoming message class to be used
    *
    * @param incomingMessageClass incoming message class
    */
   @Deprecated
   public final void setIncomingMessage(
       Class<? extends Writable> incomingMessageClass) {
     superstepClasses.setIncomingMessageClass(incomingMessageClass);
   }
 
   /**
    * Set outgoing message class to be used
    *
    * @param outgoingMessageClass outgoing message class
    */
   public final void setOutgoingMessage(
       Class<? extends Writable> outgoingMessageClass) {
     superstepClasses.setOutgoingMessageClass(outgoingMessageClass);
   }
 
   /**
    * Set outgoing message classes to be used
    *
    * @param outgoingMessageClasses outgoing message classes
    */
   public void setOutgoingMessageClasses(
       MessageClasses<? extends WritableComparable, ? extends Writable>
         outgoingMessageClasses) {
     superstepClasses.setOutgoingMessageClasses(outgoingMessageClasses);
   }
 
   @Override
   public final <S, R extends Writable> void registerReducer(
       String name, ReduceOperation<S, R> reduceOp) {
     serviceMaster.getGlobalCommHandler().registerReducer(name, reduceOp);
   }
 
   @Override
   public final <S, R extends Writable> void registerReducer(
       String name, ReduceOperation<S, R> reduceOp, R globalInitialValue) {
     serviceMaster.getGlobalCommHandler().registerReducer(
         name, reduceOp, globalInitialValue);
   }
 
   @Override
   public final <T extends Writable> T getReduced(String name) {
     return serviceMaster.getGlobalCommHandler().getReduced(name);
   }
 
   @Override
   public final void broadcast(String name, Writable object) {
     serviceMaster.getGlobalCommHandler().broadcast(name, object);
   }
 
   @Override
   public final <A extends Writable> boolean registerAggregator(
     String name, Class<? extends Aggregator<A>> aggregatorClass)
     throws InstantiationException, IllegalAccessException {
     return serviceMaster.getAggregatorTranslationHandler().registerAggregator(
         name, aggregatorClass);
   }
 
   @Override
   public final <A extends Writable> boolean registerPersistentAggregator(
       String name,
       Class<? extends Aggregator<A>> aggregatorClass) throws
       InstantiationException, IllegalAccessException {
     return serviceMaster.getAggregatorTranslationHandler()
         .registerPersistentAggregator(name, aggregatorClass);
   }
 
   @Override
   public final <A extends Writable> A getAggregatedValue(String name) {
     return serviceMaster.getAggregatorTranslationHandler()
         .<A>getAggregatedValue(name);
   }
 
   @Override
   public final <A extends Writable> void setAggregatedValue(
       String name, A value) {
     serviceMaster.getAggregatorTranslationHandler()
         .setAggregatedValue(name, value);
   }
 
   /**
    * Call this to log a line to command line of the job. Use in moderation -
    * it's a synchronous call to Job client
    *
    * @param line Line to print
    */
   public void logToCommandLine(String line) {
     serviceMaster.getJobProgressTracker().logInfo(line);
   }
 
   public final void setGraphState(GraphState graphState) {
     this.graphState = graphState;
   }
 
   public final void setMasterService(CentralizedServiceMaster serviceMaster) {
     this.serviceMaster = serviceMaster;
   }
 
   public final void setSuperstepClasses(SuperstepClasses superstepClasses) {
     this.superstepClasses = superstepClasses;
   }
 }