/*
    * 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.worker;
  
  import java.io.IOException;
  
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.edge.Edge;
  import org.apache.giraph.edge.OutEdges;
  import org.apache.giraph.graph.Vertex;
  import org.apache.giraph.graph.VertexEdgeCount;
  import org.apache.giraph.io.GiraphInputFormat;
  import org.apache.giraph.io.VertexInputFormat;
  import org.apache.giraph.io.VertexReader;
  import org.apache.giraph.io.filters.VertexInputFilter;
  import org.apache.giraph.mapping.translate.TranslateEdge;
  import org.apache.giraph.io.InputType;
  import org.apache.giraph.ooc.OutOfCoreEngine;
  import org.apache.giraph.partition.PartitionOwner;
  import org.apache.giraph.utils.LoggerUtils;
  import org.apache.giraph.utils.MemoryUtils;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.hadoop.mapreduce.InputSplit;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.log4j.Level;
  import org.apache.log4j.Logger;
  
  import com.yammer.metrics.core.Counter;
  import com.yammer.metrics.core.Meter;
  
  /**
   * Load as many vertex input splits as possible.
   * Every thread will has its own instance of WorkerClientRequestProcessor
   * to send requests.
   *
   * @param <I> Vertex index value
   * @param <V> Vertex value
   * @param <E> Edge value
   */
  @SuppressWarnings("unchecked")
  public class VertexInputSplitsCallable<I extends WritableComparable,
      V extends Writable, E extends Writable>
      extends InputSplitsCallable<I, V, E> {
    /** How often to update metrics and print info */
    public static final int VERTICES_UPDATE_PERIOD = 250000;
    /** How often to update filtered out metrics */
    public static final int VERTICES_FILTERED_UPDATE_PERIOD = 2500;
  
    /** Class logger */
    private static final Logger LOG =
        Logger.getLogger(VertexInputSplitsCallable.class);
    /** Vertex input format */
    private final VertexInputFormat<I, V, E> vertexInputFormat;
    /** Input split max vertices (-1 denotes all) */
    private final long inputSplitMaxVertices;
    /** Bsp service worker (only use thread-safe methods) */
    private final BspServiceWorker<I, V, E> bspServiceWorker;
    /** Filter to select which vertices to keep */
    private final VertexInputFilter<I, V, E> vertexInputFilter;
    /** Can embedInfo in vertexIds */
    private final boolean canEmbedInIds;
    /**
     * Whether the chosen {@link OutEdges} implementation allows for Edge
     * reuse.
     */
    private final boolean reuseEdgeObjects;
    /** Used to translate Edges during vertex input phase based on localData */
    private final TranslateEdge<I, E> translateEdge;
  
    // Metrics
    /** number of vertices loaded meter across all readers */
    private final Meter totalVerticesMeter;
    /** number of vertices filtered out */
    private final Counter totalVerticesFilteredCounter;
    /** number of edges loaded meter across all readers */
    private final Meter totalEdgesMeter;
  
    /**
     * Constructor.
     *
     * @param vertexInputFormat Vertex input format
     * @param context Context
    * @param configuration Configuration
    * @param bspServiceWorker service worker
    * @param splitsHandler Handler for input splits
    */
   public VertexInputSplitsCallable(
       VertexInputFormat<I, V, E> vertexInputFormat,
       Mapper<?, ?, ?, ?>.Context context,
       ImmutableClassesGiraphConfiguration<I, V, E> configuration,
       BspServiceWorker<I, V, E> bspServiceWorker,
       WorkerInputSplitsHandler splitsHandler)  {
     super(context, configuration, bspServiceWorker, splitsHandler);
     this.vertexInputFormat = vertexInputFormat;
 
     inputSplitMaxVertices = configuration.getInputSplitMaxVertices();
     this.bspServiceWorker = bspServiceWorker;
     vertexInputFilter = configuration.getVertexInputFilter();
     reuseEdgeObjects = configuration.reuseEdgeObjects();
     canEmbedInIds = bspServiceWorker
         .getLocalData()
         .getMappingStoreOps() != null &&
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .hasEmbedding();
     translateEdge = bspServiceWorker.getTranslateEdge();
 
     // Initialize Metrics
     totalVerticesMeter = getTotalVerticesLoadedMeter();
     totalVerticesFilteredCounter = getTotalVerticesFilteredCounter();
     totalEdgesMeter = getTotalEdgesLoadedMeter();
   }
 
   @Override
   public GiraphInputFormat getInputFormat() {
     return vertexInputFormat;
   }
 
   @Override
   public InputType getInputType() {
     return InputType.VERTEX;
   }
 
   /**
    * Read vertices from input split.  If testing, the user may request a
    * maximum number of vertices to be read from an input split.
    *
    * @param inputSplit Input split to process with vertex reader
    * @return Vertices and edges loaded from this input split
    * @throws IOException
    * @throws InterruptedException
    */
   @Override
   protected VertexEdgeCount readInputSplit(
       InputSplit inputSplit) throws IOException, InterruptedException {
     VertexReader<I, V, E> vertexReader =
         vertexInputFormat.createVertexReader(inputSplit, context);
     vertexReader.setConf(configuration);
 
     WorkerThreadGlobalCommUsage globalCommUsage =
       this.bspServiceWorker
         .getAggregatorHandler().newThreadAggregatorUsage();
 
     vertexReader.initialize(inputSplit, context);
     // Set aggregator usage to vertex reader
     vertexReader.setWorkerGlobalCommUsage(globalCommUsage);
 
     long inputSplitVerticesLoaded = 0;
     long inputSplitVerticesFiltered = 0;
 
     long edgesSinceLastUpdate = 0;
     long inputSplitEdgesLoaded = 0;
 
     int count = 0;
     OutOfCoreEngine oocEngine = bspServiceWorker.getServerData().getOocEngine();
     while (vertexReader.nextVertex()) {
       // If out-of-core mechanism is used, check whether this thread
       // can stay active or it should temporarily suspend and stop
       // processing and generating more data for the moment.
       if (oocEngine != null &&
           (++count & OutOfCoreEngine.CHECK_IN_INTERVAL) == 0) {
         oocEngine.activeThreadCheckIn();
       }
       Vertex<I, V, E> readerVertex = vertexReader.getCurrentVertex();
       if (readerVertex.getId() == null) {
         throw new IllegalArgumentException(
             "readInputSplit: Vertex reader returned a vertex " +
                 "without an id!  - " + readerVertex);
       }
       if (canEmbedInIds) {
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .embedTargetInfo(readerVertex.getId());
       }
       if (readerVertex.getValue() == null) {
         readerVertex.setValue(configuration.createVertexValue());
       }
       readerVertex.setConf(configuration);
 
       ++inputSplitVerticesLoaded;
 
       if (vertexInputFilter.dropVertex(readerVertex)) {
         ++inputSplitVerticesFiltered;
         if (inputSplitVerticesFiltered % VERTICES_FILTERED_UPDATE_PERIOD == 0) {
           totalVerticesFilteredCounter.inc(inputSplitVerticesFiltered);
           inputSplitVerticesFiltered = 0;
         }
         continue;
       }
 
       // Before saving to partition-store translate all edges (if present)
       if (translateEdge != null) {
         // only iff vertexInput reads edges also
         if (readerVertex.getEdges() != null && readerVertex.getNumEdges() > 0) {
           OutEdges<I, E> vertexOutEdges = configuration
               .createAndInitializeOutEdges(readerVertex.getNumEdges());
           // TODO : this works for generic OutEdges, can create a better api
           // to support more efficient translation for specific types
 
           // NOTE : for implementations where edge is reusable, space is
           // consumed by the OutEdges data structure itself, but if not reusable
           // space is consumed by the newly created edge -> and the new OutEdges
           // data structure just holds a reference to the newly created edge
           // so in any way we virtually hold edges twice - similar to
           // OutEdges.trim() -> this has the same complexity as OutEdges.trim()
           for (Edge<I, E> edge : readerVertex.getEdges()) {
             if (reuseEdgeObjects) {
               bspServiceWorker
                   .getLocalData()
                   .getMappingStoreOps()
                   .embedTargetInfo(edge.getTargetVertexId());
               vertexOutEdges.add(edge); // edge can be re-used
             } else { // edge objects cannot be reused - so create new edges
               vertexOutEdges.add(configuration.createEdge(translateEdge, edge));
             }
           }
           // set out edges to translated instance -> old instance is released
           readerVertex.setEdges(vertexOutEdges);
         }
       }
 
       PartitionOwner partitionOwner =
           bspServiceWorker.getVertexPartitionOwner(readerVertex.getId());
       workerClientRequestProcessor.sendVertexRequest(
           partitionOwner, readerVertex);
       edgesSinceLastUpdate += readerVertex.getNumEdges();
 
       // Update status every VERTICES_UPDATE_PERIOD vertices
       if (inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD == 0) {
         totalVerticesMeter.mark(VERTICES_UPDATE_PERIOD);
         WorkerProgress.get().addVerticesLoaded(VERTICES_UPDATE_PERIOD);
         totalEdgesMeter.mark(edgesSinceLastUpdate);
         inputSplitEdgesLoaded += edgesSinceLastUpdate;
         edgesSinceLastUpdate = 0;
 
         LoggerUtils.setStatusAndLog(
             context, LOG, Level.INFO,
             "readVertexInputSplit: Loaded " +
                 totalVerticesMeter.count() + " vertices at " +
                 totalVerticesMeter.meanRate() + " vertices/sec " +
                 totalEdgesMeter.count() + " edges at " +
                 totalEdgesMeter.meanRate() + " edges/sec " +
                 MemoryUtils.getRuntimeMemoryStats());
       }
 
       // For sampling, or to limit outlier input splits, the number of
       // records per input split can be limited
       if (inputSplitMaxVertices > 0 &&
           inputSplitVerticesLoaded >= inputSplitMaxVertices) {
         if (LOG.isInfoEnabled()) {
           LOG.info("readInputSplit: Leaving the input " +
               "split early, reached maximum vertices " +
               inputSplitVerticesLoaded);
         }
         break;
       }
     }
 
     totalVerticesMeter.mark(inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD);
     totalEdgesMeter.mark(edgesSinceLastUpdate);
     totalVerticesFilteredCounter.inc(inputSplitVerticesFiltered);
 
     vertexReader.close();
 
     WorkerProgress.get().addVerticesLoaded(
         inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD);
     WorkerProgress.get().incrementVertexInputSplitsLoaded();
 
     return new VertexEdgeCount(inputSplitVerticesLoaded,
         inputSplitEdgesLoaded + edgesSinceLastUpdate, 0);
   }
 }