/*
    * 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.job;
  
  import static org.apache.giraph.conf.GiraphConstants.VERTEX_EDGES_CLASS;
  import static org.apache.giraph.conf.GiraphConstants.VERTEX_RESOLVER_CLASS;
  import static org.apache.giraph.utils.ReflectionUtils.getTypeArguments;
  
  import org.apache.giraph.combiner.MessageCombiner;
  import org.apache.giraph.conf.GiraphConstants;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.edge.OutEdges;
  import org.apache.giraph.factories.DefaultVertexValueFactory;
  import org.apache.giraph.factories.VertexValueFactory;
  import org.apache.giraph.graph.DefaultVertexResolver;
  import org.apache.giraph.graph.VertexResolver;
  import org.apache.giraph.graph.VertexValueCombiner;
  import org.apache.giraph.io.EdgeInputFormat;
  import org.apache.giraph.io.EdgeOutputFormat;
  import org.apache.giraph.io.VertexInputFormat;
  import org.apache.giraph.io.VertexOutputFormat;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.log4j.Logger;
  
  /**
   * GiraphConfigurationValidator attempts to verify the consistency of
   * user-chosen InputFormat, OutputFormat, and Vertex generic type
   * parameters as well as the general Configuration settings
   * before the job run actually begins.
   *
   * @param <I> the Vertex ID type
   * @param <V> the Vertex Value type
   * @param <E> the Edge Value type
   * @param <M1> the incoming Message type
   * @param <M2> the outgoing Message type
   */
  public class GiraphConfigurationValidator<I extends WritableComparable,
      V extends Writable, E extends Writable, M1 extends Writable,
      M2 extends Writable> {
    /**
     * Class logger object.
     */
    private static Logger LOG =
      Logger.getLogger(GiraphConfigurationValidator.class);
  
    /** I param vertex index in classList */
    private static final int ID_PARAM_INDEX = 0;
    /** V param vertex index in classList */
    private static final int VALUE_PARAM_INDEX = 1;
    /** E param vertex index in classList */
    private static final int EDGE_PARAM_INDEX = 2;
    /** M param vertex combiner index in classList */
    private static final int MSG_COMBINER_PARAM_INDEX = 1;
    /** E param edge input format index in classList */
    private static final int EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX = 1;
    /** E param vertex edges index in classList */
    private static final int EDGE_PARAM_OUT_EDGES_INDEX = 1;
    /** V param vertex value factory index in classList */
    private static final int VALUE_PARAM_VERTEX_VALUE_FACTORY_INDEX = 0;
    /** V param vertex value combiner index in classList */
    private static final int VALUE_PARAM_VERTEX_VALUE_COMBINER_INDEX = 0;
  
    /**
     * The Configuration object for use in the validation test.
     */
    private final ImmutableClassesGiraphConfiguration conf;
  
    /**
     * Constructor to execute the validation test, throws
     * unchecked exception to end job run on failure.
     *
     * @param conf the Configuration for this run.
     */
    public GiraphConfigurationValidator(Configuration conf) {
      this.conf = new ImmutableClassesGiraphConfiguration(conf);
    }
  
    /**
     * Get vertex id type
     *
     * @return vertex id type
    */
   private Class<? extends WritableComparable> vertexIndexType() {
     return conf.getGiraphTypes().getVertexIdClass();
   }
 
   /**
    * Get vertex value type
    *
    * @return vertex value type
    */
   private Class<? extends Writable> vertexValueType() {
     return conf.getGiraphTypes().getVertexValueClass();
   }
 
   /**
    * Get edge value type
    *
    * @return edge value type
    */
   private Class<? extends Writable> edgeValueType() {
     return conf.getGiraphTypes().getEdgeValueClass();
   }
 
   /**
    * Get outgoing message value type
    *
    * @return outgoing message value type
    */
   private Class<? extends Writable> outgoingMessageValueType() {
     return conf.getOutgoingMessageValueClass();
   }
 
   /**
    * Make sure that all registered classes have matching types.  This
    * is a little tricky due to type erasure, cannot simply get them from
    * the class type arguments.  Also, set the vertex index, vertex value,
    * edge value and message value classes.
    */
   public void validateConfiguration() {
     checkConfiguration();
     verifyOutEdgesGenericTypes();
     verifyVertexInputFormatGenericTypes();
     verifyEdgeInputFormatGenericTypes();
     verifyVertexOutputFormatGenericTypes();
     verifyEdgeOutputFormatGenericTypes();
     verifyVertexResolverGenericTypes();
     verifyVertexValueCombinerGenericTypes();
     verifyMessageCombinerGenericTypes();
     verifyVertexValueFactoryGenericTypes();
   }
 
   /**
    * Make sure the configuration is set properly by the user prior to
    * submitting the job.
    */
   private void checkConfiguration() {
     if (conf.getMaxWorkers() < 0) {
       throw new RuntimeException("checkConfiguration: No valid " +
           GiraphConstants.MAX_WORKERS);
     }
     if (conf.getMinPercentResponded() <= 0.0f ||
         conf.getMinPercentResponded() > 100.0f) {
       throw new IllegalArgumentException(
           "checkConfiguration: Invalid " + conf.getMinPercentResponded() +
               " for " + GiraphConstants.MIN_PERCENT_RESPONDED.getKey());
     }
     if (conf.getMinWorkers() < 0) {
       throw new IllegalArgumentException("checkConfiguration: No valid " +
           GiraphConstants.MIN_WORKERS);
     }
     conf.createComputationFactory().checkConfiguration(conf);
     if (conf.getVertexInputFormatClass() == null &&
         conf.getEdgeInputFormatClass() == null) {
       throw new IllegalArgumentException("checkConfiguration: One of " +
           GiraphConstants.VERTEX_INPUT_FORMAT_CLASS.getKey() + " and " +
           GiraphConstants.EDGE_INPUT_FORMAT_CLASS.getKey() +
           " must be non-null");
     }
     if (conf.getVertexResolverClass() == null) {
       if (LOG.isInfoEnabled()) {
         LOG.info("checkConfiguration: No class found for " +
             VERTEX_RESOLVER_CLASS.getKey() +
             ", defaulting to " +
             VERTEX_RESOLVER_CLASS.getDefaultClass().getCanonicalName());
       }
     }
     if (conf.getOutEdgesClass() == null) {
       if (LOG.isInfoEnabled()) {
         LOG.info("checkConfiguration: No class found for " +
             VERTEX_EDGES_CLASS.getKey() + ", defaulting to " +
             VERTEX_EDGES_CLASS.getDefaultClass().getCanonicalName());
       }
     }
   }
 
   /**
    * Verify matching generic types for a specific OutEdges class.
    *
    * @param outEdgesClass {@link org.apache.giraph.edge.OutEdges} class to check
    */
   private void verifyOutEdgesGenericTypesClass(
       Class<? extends OutEdges<I, E>> outEdgesClass) {
     Class<?>[] classList = getTypeArguments(OutEdges.class, outEdgesClass);
     checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
         OutEdges.class, "vertex index");
     checkAssignable(classList, EDGE_PARAM_OUT_EDGES_INDEX, edgeValueType(),
         OutEdges.class, "edge value");
   }
 
   /** Verify matching generic types in OutEdges. */
   private void verifyOutEdgesGenericTypes() {
     Class<? extends OutEdges<I, E>> outEdgesClass =
         conf.getOutEdgesClass();
     Class<? extends OutEdges<I, E>> inputOutEdgesClass =
         conf.getInputOutEdgesClass();
     verifyOutEdgesGenericTypesClass(outEdgesClass);
     verifyOutEdgesGenericTypesClass(inputOutEdgesClass);
   }
 
   /** Verify matching generic types in VertexInputFormat. */
   private void verifyVertexInputFormatGenericTypes() {
     Class<? extends VertexInputFormat<I, V, E>> vertexInputFormatClass =
       conf.getVertexInputFormatClass();
     if (vertexInputFormatClass != null) {
       Class<?>[] classList =
           getTypeArguments(VertexInputFormat.class, vertexInputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           VertexInputFormat.class, "vertex index");
       checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
           VertexInputFormat.class, "vertex value");
       checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
           VertexInputFormat.class, "edge value");
     }
   }
 
   /** Verify matching generic types in EdgeInputFormat. */
   private void verifyEdgeInputFormatGenericTypes() {
     Class<? extends EdgeInputFormat<I, E>> edgeInputFormatClass =
         conf.getEdgeInputFormatClass();
     if (edgeInputFormatClass != null) {
       Class<?>[] classList =
           getTypeArguments(EdgeInputFormat.class, edgeInputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           EdgeInputFormat.class, "vertex index");
       checkAssignable(classList, EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX,
           edgeValueType(), EdgeInputFormat.class, "edge value");
     }
   }
 
   /**
    * If there is a vertex value combiner type, verify its
    * generic params match the job.
    */
   private void verifyVertexValueCombinerGenericTypes() {
     Class<? extends VertexValueCombiner<V>> vertexValueCombiner =
         conf.getVertexValueCombinerClass();
     if (vertexValueCombiner != null) {
       Class<?>[] classList =
           getTypeArguments(VertexValueCombiner.class, vertexValueCombiner);
       checkAssignable(classList, VALUE_PARAM_VERTEX_VALUE_COMBINER_INDEX,
           vertexValueType(), VertexValueCombiner.class, "vertex value");
     }
   }
 
   /**
    * If there is a message combiner type, verify its
    * generic params match the job.
    */
   private void verifyMessageCombinerGenericTypes() {
     MessageCombiner<I, M2> messageCombiner =
       conf.createOutgoingMessageCombiner();
     if (messageCombiner != null) {
       Class<?>[] classList =
           getTypeArguments(MessageCombiner.class, messageCombiner.getClass());
       checkEquals(classList, ID_PARAM_INDEX, vertexIndexType(),
           MessageCombiner.class, "vertex index");
       checkEquals(classList, MSG_COMBINER_PARAM_INDEX,
           outgoingMessageValueType(), MessageCombiner.class, "message value");
     }
   }
 
   /** Verify that the vertex output format's generic params match the job. */
   private void verifyVertexOutputFormatGenericTypes() {
     Class<? extends EdgeOutputFormat<I, V, E>>
       edgeOutputFormatClass = conf.getEdgeOutputFormatClass();
     if (conf.hasEdgeOutputFormat()) {
       Class<?>[] classList =
         getTypeArguments(EdgeOutputFormat.class, edgeOutputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           VertexOutputFormat.class, "vertex index");
       checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
           VertexOutputFormat.class, "vertex value");
       checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
           VertexOutputFormat.class, "edge value");
     }
   }
 
   /** Verify that the edge output format's generic params match the job. */
   private void verifyEdgeOutputFormatGenericTypes() {
     Class<? extends VertexOutputFormat<I, V, E>>
       vertexOutputFormatClass = conf.getVertexOutputFormatClass();
     if (conf.hasVertexOutputFormat()) {
       Class<?>[] classList =
         getTypeArguments(VertexOutputFormat.class, vertexOutputFormatClass);
       checkAssignable(classList, ID_PARAM_INDEX, vertexIndexType(),
           VertexOutputFormat.class, "vertex index");
       checkAssignable(classList, VALUE_PARAM_INDEX, vertexValueType(),
           VertexOutputFormat.class, "vertex value");
       checkAssignable(classList, EDGE_PARAM_INDEX, edgeValueType(),
           VertexOutputFormat.class, "edge value");
     }
   }
 
   /** Verify that the vertex value factory's type matches the job */
   private void verifyVertexValueFactoryGenericTypes() {
     Class<? extends VertexValueFactory<V>>
         vvfClass = conf.getVertexValueFactoryClass();
     if (DefaultVertexValueFactory.class.equals(vvfClass)) {
       return;
     }
     Class<?>[] classList = getTypeArguments(VertexValueFactory.class, vvfClass);
     checkEquals(classList, VALUE_PARAM_VERTEX_VALUE_FACTORY_INDEX,
         vertexValueType(), VertexValueFactory.class, "vertex value");
   }
 
   /**
    * If there is a vertex resolver,
    * validate the generic parameter types.
    * */
   private void verifyVertexResolverGenericTypes() {
     Class<? extends VertexResolver<I, V, E>>
         vrClass = conf.getVertexResolverClass();
     if (DefaultVertexResolver.class.equals(vrClass)) {
       return;
     }
     Class<?>[] classList =
         getTypeArguments(VertexResolver.class, vrClass);
     checkEquals(classList, ID_PARAM_INDEX, vertexIndexType(),
         VertexResolver.class, "vertex index");
     checkEquals(classList, VALUE_PARAM_INDEX, vertexValueType(),
         VertexResolver.class, "vertex value");
     checkEquals(classList, EDGE_PARAM_INDEX, edgeValueType(),
         VertexResolver.class, "edge value");
   }
 
   /**
    * Check that the type from computation equals the type from the class.
    *
    * @param classList classes from type
    * @param index array index of class to check
    * @param classFromComputation class from computation
    * @param klass Class type we're checking, only used for printing name
    * @param typeName Name of type we're checking
    */
   private static void checkEquals(Class<?>[] classList, int index,
       Class<?> classFromComputation, Class klass, String typeName) {
     if (classList[index] == null) {
       LOG.warn(klass.getSimpleName() + " " + typeName + " type is not known");
     } else if (!classList[index].equals(classFromComputation)) {
       throw new IllegalStateException(
           "checkClassTypes: " + typeName + " types not equal, " +
               "computation - " + classFromComputation +
               ", " + klass.getSimpleName() + " - " +
               classList[index]);
     }
   }
 
   /**
    * Check that the type from computation is assignable to type from the class.
    *
    * @param classList classes from type
    * @param index array index of class to check
    * @param classFromComputation class from computation
    * @param klass Class type we're checking, only used for printing name
    * @param typeName Name of type we're checking
    */
   private static void checkAssignable(Class<?>[] classList, int index,
       Class<?> classFromComputation, Class klass, String typeName) {
     if (classList[index] == null) {
       LOG.warn(klass.getSimpleName() + " " + typeName + " type is not known");
     } else if (!classList[index].isAssignableFrom(classFromComputation)) {
       throw new IllegalStateException(
           "checkClassTypes: " + typeName + " types not assignable, " +
               "computation - " + classFromComputation +
               ", " + klass.getSimpleName() + " - " +
               classList[EDGE_PARAM_EDGE_INPUT_FORMAT_INDEX]);
     }
   }
 }