/*
    * 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.io.formats;
  
  import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
  import it.unimi.dsi.fastutil.longs.LongSet;
  
  import java.io.IOException;
  import java.util.List;
  import java.util.Random;
  
  import org.apache.giraph.bsp.BspInputSplit;
  import org.apache.giraph.edge.Edge;
  import org.apache.giraph.edge.OutEdges;
  import org.apache.giraph.edge.ReusableEdge;
  import org.apache.giraph.graph.Vertex;
  import org.apache.giraph.io.VertexInputFormat;
  import org.apache.giraph.io.VertexReader;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.io.DoubleWritable;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.mapreduce.InputSplit;
  import org.apache.hadoop.mapreduce.JobContext;
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  
  /**
   * Generates a random Watts-Strogatz graph by re-wiring a ring lattice.
   * The resulting graph is a random graph with high clustering coefficient
   * and low average path length. The graph has these two characteristics that
   * are typical of small-world scale-free graphs, however the degree
   * distribution is more similar to a random graph.
   * It supports a seed for pseudo-random generation.
   */
  public class WattsStrogatzVertexInputFormat extends
    VertexInputFormat<LongWritable, DoubleWritable, DoubleWritable> {
    /** The number of vertices in the graph */
    private static final String AGGREGATE_VERTICES =
        "wattsStrogatz.aggregateVertices";
    /** The number of outgoing edges per vertex */
    private static final String EDGES_PER_VERTEX =
        "wattsStrogatz.edgesPerVertex";
    /** The probability to re-wire an outgoing edge from the ring lattice */
    private static final String BETA =
        "wattsStrogatz.beta";
    /** The seed to generate random values for pseudo-randomness */
    private static final String SEED =
        "wattsStrogatz.seed";
  
    @Override
    public void checkInputSpecs(Configuration conf) { }
  
    @Override
    public final List<InputSplit> getSplits(final JobContext context,
        final int minSplitCountHint) throws IOException, InterruptedException {
      return PseudoRandomUtils.getSplits(minSplitCountHint);
    }
  
    @Override
    public VertexReader<LongWritable, DoubleWritable, DoubleWritable>
    createVertexReader(InputSplit split,
        TaskAttemptContext context) throws IOException {
      return new WattsStrogatzVertexReader();
    }
  
    /**
     * Vertex reader used to generate the graph
     */
    private static class WattsStrogatzVertexReader extends
      VertexReader<LongWritable, DoubleWritable, DoubleWritable> {
      /** the re-wiring probability */
      private float beta = 0;
      /** The total number of vertices */
      private long aggregateVertices = 0;
      /** The starting vertex id for this split */
      private long startingVertexId = -1;
      /** The number of vertices read so far */
      private long verticesRead = 0;
      /** The total number of vertices in the split */
      private long totalSplitVertices = -1;
      /** the total number of outgoing edges per vertex */
      private int edgesPerVertex = -1;
      /** The target ids of the outgoing edges */
      private final LongSet destVertices = new LongOpenHashSet();
      /** The random values generator */
     private Random rnd;
     /** The reusable edge */
     private ReusableEdge<LongWritable, DoubleWritable> reusableEdge = null;
 
     /**
      * Default constructor
      */
     public WattsStrogatzVertexReader() { }
 
     @Override
     public void initialize(InputSplit inputSplit,
         TaskAttemptContext context) throws IOException {
       beta = getConf().getFloat(
           BETA, 0.0f);
       aggregateVertices = getConf().getLong(
           AGGREGATE_VERTICES, 0);
       BspInputSplit bspInputSplit = (BspInputSplit) inputSplit;
       long extraVertices = aggregateVertices % bspInputSplit.getNumSplits();
       totalSplitVertices = aggregateVertices / bspInputSplit.getNumSplits();
       if (bspInputSplit.getSplitIndex() < extraVertices) {
         ++totalSplitVertices;
       }
       startingVertexId = bspInputSplit.getSplitIndex() *
           (aggregateVertices / bspInputSplit.getNumSplits()) +
           Math.min(bspInputSplit.getSplitIndex(), extraVertices);
       edgesPerVertex = getConf().getInt(
           EDGES_PER_VERTEX, 0);
       if (getConf().reuseEdgeObjects()) {
         reusableEdge = getConf().createReusableEdge();
       }
       int seed = getConf().getInt(SEED, -1);
       if (seed != -1) {
         rnd = new Random(seed);
       } else {
         rnd = new Random();
       }
     }
 
     @Override
     public boolean nextVertex() throws IOException, InterruptedException {
       return totalSplitVertices > verticesRead;
     }
 
     /**
      * Return a long value uniformly distributed between 0 (inclusive) and n.
      *
      * @param n the upper bound for the random long value
      * @return the random value
      */
     private long nextLong(long n) {
       long bits;
       long val;
       do {
         bits = (rnd.nextLong() << 1) >>> 1;
         val = bits % n;
       } while (bits - val + (n - 1) < 0L);
       return val;
     }
 
     /**
      * Get a destination id that is not already in the neighborhood and
      * that is not the vertex itself (no self-loops). For the second condition
      * it expects destVertices to contain the own id already.
      *
      * @return the destination vertex id
      */
     private long getRandomDestination() {
       long randomId;
       do {
         randomId = nextLong(aggregateVertices);
       } while (!destVertices.add(randomId));
       return randomId;
     }
 
     @Override
     public Vertex<LongWritable, DoubleWritable, DoubleWritable>
     getCurrentVertex() throws IOException, InterruptedException {
       Vertex<LongWritable, DoubleWritable, DoubleWritable> vertex =
           getConf().createVertex();
       long vertexId = startingVertexId + verticesRead;
       OutEdges<LongWritable, DoubleWritable> edges =
           getConf().createOutEdges();
       edges.initialize(edgesPerVertex);
       destVertices.clear();
       destVertices.add(vertexId);
       long destVertexId = vertexId - edgesPerVertex / 2;
       if (destVertexId < 0) {
         destVertexId = aggregateVertices + destVertexId;
       }
       for (int i = 0; i < edgesPerVertex + 1; ++i) {
         if (destVertexId != vertexId) {
           Edge<LongWritable, DoubleWritable> edge =
               (reusableEdge == null) ? getConf().createEdge() : reusableEdge;
           edge.getTargetVertexId().set(
               rnd.nextFloat() < beta ? getRandomDestination() : destVertexId);
           edge.getValue().set(rnd.nextDouble());
           edges.add(edge);
         }
         destVertexId = (destVertexId + 1) % aggregateVertices;
       }
       vertex.initialize(new LongWritable(vertexId),
           new DoubleWritable(rnd.nextDouble()), edges);
       ++verticesRead;
       return vertex;
     }
 
     @Override
     public void close() throws IOException { }
 
     @Override
     public float getProgress() throws IOException {
       return verticesRead * 100.0f / totalSplitVertices;
     }
   }
 }