/*
    * 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.framework;
  
  import static org.apache.giraph.utils.ReflectionUtils.getTypeArguments;
  
  import java.lang.reflect.Field;
  import java.lang.reflect.Modifier;
  
  import org.apache.giraph.block_app.framework.api.giraph.BlockComputation;
  import org.apache.giraph.block_app.framework.api.giraph.BlockMasterCompute;
  import org.apache.giraph.block_app.framework.api.giraph.BlockWorkerContext;
  import org.apache.giraph.block_app.framework.block.Block;
  import org.apache.giraph.block_app.framework.block.PieceCount;
  import org.apache.giraph.block_app.framework.piece.AbstractPiece;
  import org.apache.giraph.block_app.framework.piece.Piece;
  import org.apache.giraph.conf.BooleanConfOption;
  import org.apache.giraph.conf.ClassConfOption;
  import org.apache.giraph.conf.GiraphConfiguration;
  import org.apache.giraph.conf.GiraphConstants;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.conf.MessageClasses;
  import org.apache.giraph.function.Consumer;
  import org.apache.giraph.types.NoMessage;
  import org.apache.giraph.utils.ReflectionUtils;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.log4j.Logger;
  
  import com.google.common.base.Preconditions;
  
  /**
   * Utility functions for block applications
   */
  @SuppressWarnings({ "rawtypes", "unchecked" })
  public class BlockUtils {
    /** Property describing BlockFactory to use for current application run */
    public static final ClassConfOption<BlockFactory> BLOCK_FACTORY_CLASS =
        ClassConfOption.create("digraph.block_factory", null, BlockFactory.class,
            "block factory describing giraph job");
  
    /** Property describing block worker context value class to use */
    public static final ClassConfOption<Object> BLOCK_WORKER_CONTEXT_VALUE_CLASS =
        ClassConfOption.create(
            "digraph.block_worker_context_value_class",
            Object.class, Object.class,
            "block worker context value class");
  
    /** Property describing whether to log execution status as application runs */
    public static final
    BooleanConfOption LOG_EXECUTION_STATUS = new BooleanConfOption(
        "giraph.block_utils.log_execution_status", true,
        "Log execution status (of which pieces are being executed, etc)");
  
    private static final Logger LOG = Logger.getLogger(BlockUtils.class);
  
    /** Dissallow constructor */
    private BlockUtils() { }
  
    /**
     * Create new BlockFactory that is specified in the configuration.
     */
    public static <S> BlockFactory<S> createBlockFactory(Configuration conf) {
      return ReflectionUtils.newInstance(BLOCK_FACTORY_CLASS.get(conf));
    }
  
    /**
     * Set which BlockFactory class to be used for the application.
     * (generally useful within tests only)
     */
    public static void setBlockFactoryClass(Configuration conf,
        Class<? extends BlockFactory<?>> clazz) {
      BLOCK_FACTORY_CLASS.set(conf, clazz);
    }
  
    /**
     * Set block factory, and initialize configs with it.
     * Should be used only if there are no configuration options set after
     * this method call.
     */
    public static void setAndInitBlockFactoryClass(GiraphConfiguration conf,
        Class<? extends BlockFactory<?>> clazz) {
      BLOCK_FACTORY_CLASS.set(conf, clazz);
      initAndCheckConfig(conf);
    }
 
   /**
    * Initializes configuration, such that running it executes block application.
    *
    * Additionally, checks types of all pieces with a block application.
    */
   public static void initAndCheckConfig(GiraphConfiguration conf) {
     conf.setMasterComputeClass(BlockMasterCompute.class);
     conf.setComputationClass(BlockComputation.class);
     conf.setWorkerContextClass(BlockWorkerContext.class);
 
     Preconditions.checkState(
         GiraphConstants.OUTGOING_MESSAGE_VALUE_CLASS.get(conf) == null,
         "Message types should only be specified in Pieces, " +
         "but outgoing was specified globally");
     Preconditions.checkState(
         GiraphConstants.OUTGOING_MESSAGE_VALUE_FACTORY_CLASS
           .isDefaultValue(conf),
         "Message types should only be specified in Pieces, " +
         "but factory was specified globally");
     Preconditions.checkState(
         GiraphConstants.MESSAGE_COMBINER_CLASS.get(conf) == null,
         "Message combiner should only be specified in Pieces, " +
         "but was specified globally");
 
     BlockFactory<?> blockFactory = createBlockFactory(conf);
     blockFactory.initConfig(conf);
 
     Preconditions.checkState(
         GiraphConstants.OUTGOING_MESSAGE_VALUE_CLASS.get(conf) == null,
         "Outgoing message type was specified in blockFactory.initConfig");
     Preconditions.checkState(
         GiraphConstants.OUTGOING_MESSAGE_VALUE_FACTORY_CLASS
           .isDefaultValue(conf),
         "Outgoing message factory type was specified in " +
         "blockFactory.initConfig");
     Preconditions.checkState(
         GiraphConstants.MESSAGE_COMBINER_CLASS.get(conf) == null,
         "Message combiner type was specified in blockFactory.initConfig");
 
     GiraphConstants.OUTGOING_MESSAGE_VALUE_CLASS.set(conf, NoMessage.class);
 
     final ImmutableClassesGiraphConfiguration immConf =
         new ImmutableClassesGiraphConfiguration<>(conf);
 
     // Create blocks to detect issues before creating a Giraph job
     // They will not be used here
     Block executionBlock = blockFactory.createBlock(immConf);
     checkBlockTypes(
         executionBlock, blockFactory.createExecutionStage(immConf), immConf);
 
     PieceCount pieceCount = executionBlock.getPieceCount();
     if (pieceCount.isKnown()) {
       GiraphConstants.SUPERSTEP_COUNT.set(conf, pieceCount.getCount() + 1);
     }
 
     // check for non 'static final' fields in BlockFactories
     Class<?> bfClass = blockFactory.getClass();
     while (!bfClass.equals(Object.class)) {
       for (Field field : bfClass.getDeclaredFields()) {
         if (!Modifier.isStatic(field.getModifiers()) ||
             !Modifier.isFinal(field.getModifiers())) {
           throw new IllegalStateException("BlockFactory (" + bfClass +
               ") cannot have any mutable (non 'static final') fields as a " +
               "safety measure, as createBlock function is called from a " +
               "different context then all other functions, use conf argument " +
               "instead, or make it 'static final'. Field present: " + field);
         }
       }
       bfClass = bfClass.getSuperclass();
     }
 
     // Register outputs
     blockFactory.registerOutputs(conf);
   }
 
   public static void checkBlockTypes(
       Block executionBlock, Object executionStage,
       final ImmutableClassesGiraphConfiguration conf) {
     LOG.info("Executing application - " + executionBlock);
 
     final Class<?> vertexIdClass = conf.getVertexIdClass();
     final Class<?> vertexValueClass = conf.getVertexValueClass();
     final Class<?> edgeValueClass = conf.getEdgeValueClass();
     final Class<?> workerContextValueClass =
         BLOCK_WORKER_CONTEXT_VALUE_CLASS.get(conf);
     final Class<?> executionStageClass = executionStage.getClass();
 
     // Check for type inconsistencies
     executionBlock.forAllPossiblePieces(new Consumer<AbstractPiece>() {
       @Override
       public void apply(AbstractPiece piece) {
         if (!piece.getClass().equals(Piece.class)) {
           Class<?>[] classList = getTypeArguments(
               AbstractPiece.class, piece.getClass());
           Preconditions.checkArgument(classList.length == 7);
 
           ReflectionUtils.verifyTypes(
               vertexIdClass, classList[0], "vertexId", piece.getClass());
           ReflectionUtils.verifyTypes(
               vertexValueClass, classList[1], "vertexValue", piece.getClass());
           ReflectionUtils.verifyTypes(
               edgeValueClass, classList[2], "edgeValue", piece.getClass());
 
           MessageClasses classes = piece.getMessageClasses(conf);
           Class<?> messageType = classes.getMessageClass();
           if (messageType == null) {
             messageType = NoMessage.class;
           }
           ReflectionUtils.verifyTypes(
               messageType, classList[3], "message", piece.getClass());
 
           ReflectionUtils.verifyTypes(
               workerContextValueClass, classList[4],
               "workerContextValue", piece.getClass());
           // No need to check worker context message class at all
 
           ReflectionUtils.verifyTypes(
               executionStageClass, classList[6],
               "executionStage", piece.getClass());
         }
       }
     });
   }
 }