/*
 * 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]);
    }
  }
}