/*
    * 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.piece;
  
  import org.apache.giraph.block_app.framework.api.BlockMasterApi;
  import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
  import org.apache.giraph.block_app.framework.api.CreateReducersApi;
  import org.apache.giraph.block_app.framework.piece.global_comm.ReduceUtilsObject;
  import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
  import org.apache.giraph.block_app.framework.piece.global_comm.internal.CreateReducersApiWrapper;
  import org.apache.giraph.block_app.framework.piece.global_comm.internal.ReducersForPieceHandler;
  import org.apache.giraph.block_app.framework.piece.interfaces.VertexPostprocessor;
  import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
  import org.apache.giraph.block_app.framework.piece.messages.ObjectMessageClasses;
  import org.apache.giraph.block_app.framework.piece.messages.SupplierFromConf;
  import org.apache.giraph.block_app.framework.piece.messages.SupplierFromConf.DefaultMessageFactorySupplierFromConf;
  import org.apache.giraph.block_app.framework.piece.messages.SupplierFromConf.SupplierFromConfByCopy;
  import org.apache.giraph.combiner.MessageCombiner;
  import org.apache.giraph.comm.messages.MessageEncodeAndStoreType;
  import org.apache.giraph.conf.EnumConfOption;
  import org.apache.giraph.conf.GiraphConfigurationSettable;
  import org.apache.giraph.conf.GiraphConstants;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.conf.MessageClasses;
  import org.apache.giraph.factories.MessageValueFactory;
  import org.apache.giraph.graph.Vertex;
  import org.apache.giraph.types.NoMessage;
  import org.apache.hadoop.io.DoubleWritable;
  import org.apache.hadoop.io.FloatWritable;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  
  import com.google.common.base.Preconditions;
  
  /**
   * Additional abstract implementations for all pieces to be used.
   * Code here is not in AbstractPiece only to allow for non-standard
   * non-user-defined pieces. <br>
   * Only logic used by the underlying framework directly is in AbstractPiece
   * itself.
   *
   * @param <I> Vertex id type
   * @param <V> Vertex value type
   * @param <E> Edge value type
   * @param <M> Message type
   * @param <WV> Worker value type
   * @param <WM> Worker message type
   * @param <S> Execution stage type
   */
  @SuppressWarnings({ "rawtypes", "unchecked" })
  public abstract class DefaultParentPiece<I extends WritableComparable,
      V extends Writable, E extends Writable, M extends Writable, WV,
      WM extends Writable, S> extends AbstractPiece<I, V, E, M, WV, WM, S> {
    // TODO move to GiraphConstants
    /**
     * This option will tell which message encode &amp; store enum to force,
     * when combining is not enabled.
     *
     * MESSAGE_ENCODE_AND_STORE_TYPE and this property are basically upper
     * and lower bound on message store type, when looking them in order from
     * not doing anything special, to most advanced type:
     * BYTEARRAY_PER_PARTITION,
     * EXTRACT_BYTEARRAY_PER_PARTITION,
     * POINTER_LIST_PER_VERTEX
     * resulting encode type is going to be:
     * pieceEncodingType = piece.allowOneMessageToManyIdsEncoding() ?
     *    POINTER_LIST_PER_VERTEX : BYTEARRAY_PER_PARTITION)
     * Math.max(index(minForce), Math.min(index(maxAllowed), index(pieceType);
     *
     * This is useful to force all pieces onto particular message store, even
     * if they do not overrideallowOneMessageToManyIdsEncoding, though that might
     * be rarely needed.
     * This option might be more useful for fully local computation,
     * where overall job behavior is quite different.
     */
    public static final EnumConfOption<MessageEncodeAndStoreType>
    MESSAGE_ENCODE_AND_STORE_TYPE_MIN_FORCE =
        EnumConfOption.create("giraph.messageEncodeAndStoreTypeMinForce",
            MessageEncodeAndStoreType.class,
            MessageEncodeAndStoreType.BYTEARRAY_PER_PARTITION,
            "Select the message_encode_and_store_type min force to use");
  
   private final ReduceUtilsObject reduceUtils = new ReduceUtilsObject();
   private ReducersForPieceHandler reducersHandler;
 
   // Overridable functions
 
   /**
    * Override to register any potential reducers used by this piece,
    * through calls to {@code reduceApi}, which will return reducer handles
    * for simple.
    * Tip: Without defining a field, first write here name of the field and what
    * you want to reduce, like:
    * {@code totalSum = reduceApi.createLocalReducer(SumReduce.DOUBLE); }
    * and then use tools your IDE provides to generate field signature itself,
    * which might be slightly complex:
    * {@code ReducerHandle<DoubleWritable, DoubleWritable> totalSum; }
    */
   public void registerReducers(CreateReducersApi reduceApi, S executionStage) {
   }
 
   /**
    * Override to do vertex send processing.
    *
    * Creates handler that defines what should be executed on worker
    * during send phase.
    *
    * This logic gets executed first.
    * This function is called once on each worker on each thread, in parallel,
    * on their copy of Piece object to create functions handler.
    *
    * If returned object implements Postprocessor interface, then corresponding
    * postprocess() function is going to be called once, after all vertices
    * corresponding thread needed to process are done.
    */
   public VertexSender<I, V, E> getVertexSender(
       BlockWorkerSendApi<I, V, E, M> workerApi, S executionStage) {
     return null;
   }
 
   /**
    * Override to specify type of the message this Piece sends, if it does
    * send messages.
    *
    * If not overwritten, no messages can be sent.
    */
   protected Class<M> getMessageClass() {
     return null;
   }
 
   /**
    * Override to specify message value factory to be used,
    * which creates objects into which messages will be deserialized.
    *
    * If not overwritten, or null is returned, DefaultMessageValueFactory
    * will be used.
    */
   protected MessageValueFactory<M> getMessageFactory(
       ImmutableClassesGiraphConfiguration conf) {
     return null;
   }
 
   /**
    * Override to specify message combiner to be used, if any.
    *
    * Message combiner itself should be immutable
    * (i.e. it will be call simultanously from multiple threads)
    */
   protected MessageCombiner<? super I, M> getMessageCombiner(
       ImmutableClassesGiraphConfiguration conf) {
     return null;
   }
 
   /**
    * Override to specify that this Piece allows one to many ids encoding to be
    * used for messages.
    * You should override this function, if you are sending identical message to
    * all targets, and message itself is not extremely small.
    */
   protected boolean allowOneMessageToManyIdsEncoding() {
     return false;
   }
 
   /**
    * Override to specify that receive of this Piece (and send of next Piece)
    * ignore existing vertices, and just process received messages.
    *
    * Useful when distributed processing on groups that are not vertices is
    * needed. This flag allows you not to worry whether a destination vertex
    * exist, and removes need to clean it up when finished.
    * One example is if each vertex is in a cluster, and we need to process
    * something per cluster.
    *
    * Alternative are reducers, which have distributed reduction, but mostly
    * master still does the processing afterwards, and amount of data needs to
    * fit single machine (master).
    */
   protected boolean receiveIgnoreExistingVertices() {
     return false;
   }
 
   @Override
   public MessageClasses<I, M> getMessageClasses(
       ImmutableClassesGiraphConfiguration conf) {
     Class<M> messageClass = null;
     MessageValueFactory<M> messageFactory = getMessageFactory(conf);
     MessageCombiner<? super I, M> messageCombiner = getMessageCombiner(conf);
 
     if (messageFactory != null) {
       messageClass = (Class) messageFactory.newInstance().getClass();
     } else if (messageCombiner != null) {
       messageClass = (Class) messageCombiner.createInitialMessage().getClass();
     }
 
     if (messageClass != null) {
       Preconditions.checkState(getMessageClass() == null,
           "Piece %s defines getMessageFactory or getMessageCombiner, " +
           "so it doesn't need to define getMessageClass.",
           toString());
     } else {
       messageClass = getMessageClass();
       if (messageClass == null) {
         messageClass = (Class) NoMessage.class;
       }
     }
 
     SupplierFromConf<MessageValueFactory<M>> messageFactorySupplier;
     if (messageFactory != null) {
       messageFactorySupplier =
           new SupplierFromConfByCopy<MessageValueFactory<M>>(messageFactory);
     } else {
       messageFactorySupplier =
           new DefaultMessageFactorySupplierFromConf<>(messageClass);
     }
 
     SupplierFromConf<? extends MessageCombiner<? super I, M>>
     messageCombinerSupplier;
     if (messageCombiner != null) {
       messageCombinerSupplier = new SupplierFromConfByCopy<>(messageCombiner);
     } else {
       messageCombinerSupplier = null;
     }
 
     int maxAllowed =
         GiraphConstants.MESSAGE_ENCODE_AND_STORE_TYPE.get(conf).ordinal();
     int minForce = MESSAGE_ENCODE_AND_STORE_TYPE_MIN_FORCE.get(conf).ordinal();
     Preconditions.checkState(maxAllowed >= minForce);
 
     int pieceEncodeType = (allowOneMessageToManyIdsEncoding() ?
         MessageEncodeAndStoreType.POINTER_LIST_PER_VERTEX :
         MessageEncodeAndStoreType.BYTEARRAY_PER_PARTITION).ordinal();
     // bound piece type with boundaries:
     pieceEncodeType = Math.max(minForce, Math.min(maxAllowed, pieceEncodeType));
 
     MessageEncodeAndStoreType messageEncodeAndStoreType =
         MessageEncodeAndStoreType.values()[pieceEncodeType];
 
     if (messageFactory instanceof GiraphConfigurationSettable) {
       throw new IllegalStateException(
           messageFactory.getClass() + " MessageFactory in " + this +
           " Piece implements GiraphConfigurationSettable");
     }
     if (messageCombiner instanceof GiraphConfigurationSettable) {
       throw new IllegalStateException(
           messageCombiner.getClass() + " MessageCombiner in " + this +
           " Piece implements GiraphConfigurationSettable");
     }
 
     return new ObjectMessageClasses<>(
         messageClass, messageFactorySupplier,
         messageCombinerSupplier, messageEncodeAndStoreType,
         receiveIgnoreExistingVertices());
   }
 
   // Internal implementation
 
   @Override
   public final InnerVertexSender getWrappedVertexSender(
       final BlockWorkerSendApi<I, V, E, M> workerApi, S executionStage) {
     reducersHandler.vertexSenderWorkerPreprocess(workerApi);
     final VertexSender<I, V, E> functions =
         getVertexSender(workerApi, executionStage);
     return new InnerVertexSender() {
       @Override
       public void vertexSend(Vertex<I, V, E> vertex) {
         if (functions != null) {
           functions.vertexSend(vertex);
         }
       }
       @Override
       public void postprocess() {
         if (functions instanceof VertexPostprocessor) {
           ((VertexPostprocessor) functions).postprocess();
         }
         reducersHandler.vertexSenderWorkerPostprocess(workerApi);
       }
     };
   }
 
   @Override
   public final void wrappedRegisterReducers(
       BlockMasterApi masterApi, S executionStage) {
     reducersHandler = new ReducersForPieceHandler();
     registerReducers(new CreateReducersApiWrapper(
         masterApi, reducersHandler), executionStage);
   }
 
   // utility functions:
   // TODO Java8 - move these as default functions to VertexSender interface
   protected final void reduceDouble(
       ReducerHandle<DoubleWritable, ?> reduceHandle, double value) {
     reduceUtils.reduceDouble(reduceHandle, value);
   }
 
   protected final void reduceFloat(
       ReducerHandle<FloatWritable, ?> reduceHandle, float value) {
     reduceUtils.reduceFloat(reduceHandle, value);
   }
 
   protected final void reduceLong(
       ReducerHandle<LongWritable, ?> reduceHandle, long value) {
     reduceUtils.reduceLong(reduceHandle, value);
   }
 
   protected final void reduceInt(
       ReducerHandle<IntWritable, ?> reduceHandle, int value) {
     reduceUtils.reduceInt(reduceHandle, value);
   }
 }