package org.apache.helix.tools;

/*
 * 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.
 */

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.helix.ZNRecord;
import org.apache.helix.model.IdealState.IdealStateProperty;

public class IdealCalculatorByConsistentHashing {
  /**
   * Interface to calculate the hash function value of a string
   */
  public interface HashFunction {
    public int getHashValue(String key);
  }

  /**
   * The default string hash function. Same as the default function used by
   * Voldmort
   */
  public static class FnvHash implements HashFunction {
    private static final long FNV_BASIS = 0x811c9dc5;
    private static final long FNV_PRIME = (1 << 24) + 0x193;
    public static final long FNV_BASIS_64 = 0xCBF29CE484222325L;
    public static final long FNV_PRIME_64 = 1099511628211L;

    public int hash(byte[] key) {
      long hash = FNV_BASIS;
      for (int i = 0; i < key.length; i++) {
        hash ^= 0xFF & key[i];
        hash *= FNV_PRIME;
      }
      return (int) hash;
    }

    public long hash64(long val) {
      long hashval = FNV_BASIS_64;
      for (int i = 0; i < 8; i++) {
        long octet = val & 0x00ff;
        val = val >> 8;
        hashval = hashval ^ octet;
        hashval = hashval * FNV_PRIME_64;
      }
      return Math.abs(hashval);
    }

    @Override
    public int getHashValue(String key) {
      return hash(key.getBytes());
    }

  }

  /**
   * Calculate the ideal state for list of instances clusters using consistent
   * hashing.
   * @param instanceNames
   *          List of instance names.
   * @param partitions
   *          the partition number of the database
   * @param replicas
   *          the replication degree
   * @param resourceName
   *          the name of the database
   * @return The ZNRecord that contains the ideal state
   */
  public static ZNRecord calculateIdealState(List<String> instanceNames, int partitions,
      int replicas, String resourceName, HashFunction hashFunc) {
    return calculateIdealState(instanceNames, partitions, replicas, resourceName, hashFunc, 65536);
  }

  /**
   * Calculate the ideal state for list of instances clusters using consistent
   * hashing.
   * @param instanceNames
   *          List of instance names.
   * @param partitions
   *          the partition number of the database
   * @param replicas
   *          the replication degree
   * @param resourceName
   *          the name of the database
   * @param hashRingSize
   *          the size of the hash ring used by consistent hashing
   * @return The ZNRecord that contains the ideal state
   */
  public static ZNRecord calculateIdealState(List<String> instanceNames, int partitions,
      int replicas, String resourceName, HashFunction hashFunc, int hashRingSize) {
    ZNRecord result = new ZNRecord(resourceName);

    int[] hashRing = generateEvenHashRing(instanceNames, hashRingSize);
    result.setSimpleField(IdealStateProperty.NUM_PARTITIONS.toString(), String.valueOf(partitions));
    Random rand = new Random(0xc0ffee);
    for (int i = 0; i < partitions; i++) {
      String partitionName = resourceName + ".partition-" + i;
      int hashPos = rand.nextInt() % hashRingSize;
      // (int)(hashFunc.getHashValue(partitionName) % hashRingSize);
      hashPos = hashPos < 0 ? (hashPos + hashRingSize) : hashPos;
      // System.out.print(hashPos+ " ");
      // if(i % 120 == 0) System.out.println();
      Map<String, String> partitionAssignment = new TreeMap<String, String>();
      // the first in the list is the node that contains the master
      int masterPos = hashRing[hashPos];
      partitionAssignment.put(instanceNames.get(masterPos), "MASTER");

      // partitionAssignment.put("hash", "" + hashPos + " " + masterPos);

      // Put slaves in next has ring positions. We need to make sure that no
      // more than 2 slaves
      // are mapped to one node.
      for (int j = 1; j <= replicas; j++) {
        String next = instanceNames.get(hashRing[(hashPos + j) % hashRingSize]);
        while (partitionAssignment.containsKey(next)) {
          hashPos++;
          next = instanceNames.get(hashRing[(hashPos + j) % hashRingSize]);
        }
        partitionAssignment.put(next, "SLAVE");
      }
      result.setMapField(partitionName, partitionAssignment);
    }
    return result;
  }

  /**
   * Generate the has ring for consistent hashing.
   * @param instanceNames
   *          List of instance names.
   * @param hashRingSize
   *          the size of the hash ring used by consistent hashing
   * @return The int array as the hashing. it contains random values ranges from
   *         0..size of instanceNames-1
   */
  public static int[] generateHashRing(List<String> instanceNames, int hashRingSize) {
    int[] result = new int[hashRingSize];
    for (int i = 0; i < result.length; i++) {
      result[i] = 0;
    }
    int instances = instanceNames.size();
    // The following code generates the random distribution
    for (int i = 1; i < instances; i++) {
      putNodeOnHashring(result, i, hashRingSize / (i + 1), i);
    }
    return result;
  }

  public static int[] generateEvenHashRing(List<String> instanceNames, int hashRingSize) {
    int[] result = new int[hashRingSize];
    for (int i = 0; i < result.length; i++) {
      result[i] = 0;
    }
    int instances = instanceNames.size();
    // The following code generates the random distribution
    for (int i = 1; i < instances; i++) {
      putNodeEvenOnHashRing(result, i, i + 1);
    }
    return result;
  }

  private static void putNodeEvenOnHashRing(int[] hashRing, int nodeVal, int totalValues) {
    int newValNum = hashRing.length / totalValues;
    assert (newValNum > 0);
    Map<Integer, List<Integer>> valueIndex = buildValueIndex(hashRing);
    int nSources = valueIndex.size();
    int remainder = newValNum % nSources;

    List<List<Integer>> positionLists = new ArrayList<List<Integer>>();
    for (List<Integer> list : valueIndex.values()) {
      positionLists.add(list);
    }
    class ListComparator implements Comparator<List<Integer>> {
      @Override
      public int compare(List<Integer> o1, List<Integer> o2) {
        return (o1.size() > o2.size() ? -1 : (o1.size() == o2.size() ? 0 : 1));
      }
    }
    Collections.sort(positionLists, new ListComparator());

    for (List<Integer> oldValPositions : positionLists) {
      // List<Integer> oldValPositions = valueIndex.get(oldVal);
      int nValsToReplace = newValNum / nSources;
      assert (nValsToReplace > 0);
      if (remainder > 0) {
        nValsToReplace++;
        remainder--;
      }
      // System.out.print(oldValPositions.size()+" "+nValsToReplace+"  ");
      putNodeValueOnHashRing(hashRing, nodeVal, nValsToReplace, oldValPositions);
      // randomly take nValsToReplace positions in oldValPositions and make them
    }
    // System.out.println();
  }

  private static void putNodeValueOnHashRing(int[] hashRing, int nodeVal, int numberOfValues,
      List<Integer> positions) {
    Random rand = new Random(nodeVal);
    // initialize the index array
    int[] index = new int[positions.size()];
    for (int i = 0; i < index.length; i++) {
      index[i] = i;
    }

    int nodesLeft = index.length;

    for (int i = 0; i < numberOfValues; i++) {
      // Calculate a random index
      int randIndex = rand.nextInt() % nodesLeft;
      if (randIndex < 0) {
        randIndex += nodesLeft;
      }
      hashRing[positions.get(index[randIndex])] = nodeVal;

      // swap the random index and the last available index, and decrease the
      // nodes left
      int temp = index[randIndex];
      index[randIndex] = index[nodesLeft - 1];
      index[nodesLeft - 1] = temp;
      nodesLeft--;
    }
  }

  private static Map<Integer, List<Integer>> buildValueIndex(int[] hashRing) {
    Map<Integer, List<Integer>> result = new TreeMap<Integer, List<Integer>>();
    for (int i = 0; i < hashRing.length; i++) {
      if (!result.containsKey(hashRing[i])) {
        List<Integer> list = new ArrayList<Integer>();
        result.put(hashRing[i], list);
      }
      result.get(hashRing[i]).add(i);
    }
    return result;
  }

  /**
   * Uniformly put node values on the hash ring. Derived from the shuffling
   * algorithm
   * @param result
   *          the hash ring array.
   * @param nodeValue
   *          the int value to be added to the hash ring this time
   * @param numberOfNodes
   *          number of node values to put on the hash ring array
   * @param randomSeed
   *          the random seed
   */
  public static void putNodeOnHashring(int[] result, int nodeValue, int numberOfNodes,
      int randomSeed) {
    Random rand = new Random(randomSeed);
    // initialize the index array
    int[] index = new int[result.length];
    for (int i = 0; i < index.length; i++) {
      index[i] = i;
    }

    int nodesLeft = index.length;

    for (int i = 0; i < numberOfNodes; i++) {
      // Calculate a random index
      int randIndex = rand.nextInt() % nodesLeft;
      if (randIndex < 0) {
        randIndex += nodesLeft;
      }
      if (result[index[randIndex]] == nodeValue) {
        assert (false);
      }
      result[index[randIndex]] = nodeValue;

      // swap the random index and the last available index, and decrease the
      // nodes left
      int temp = index[randIndex];
      index[randIndex] = index[nodesLeft - 1];
      index[nodesLeft - 1] = temp;

      nodesLeft--;
    }
  }

  /**
   * Helper function to see how many partitions are mapped to different
   * instances in two ideal states
   */
  public static void printDiff(ZNRecord record1, ZNRecord record2) {
    int diffCount = 0;
    for (String key : record1.getMapFields().keySet()) {
      Map<String, String> map1 = record1.getMapField(key);
      Map<String, String> map2 = record2.getMapField(key);

      for (String k : map1.keySet()) {
        if (!map2.containsKey(k)) {
          diffCount++;
        } else if (!map1.get(k).equalsIgnoreCase(map2.get(k))) {
          diffCount++;
        }
      }
    }
    System.out.println("diff count = " + diffCount);
  }

  /**
   * Helper function to compare the difference between two hashing buffers
   */
  public static void compareHashrings(int[] ring1, int[] ring2) {
    int diff = 0;
    for (int i = 0; i < ring1.length; i++) {
      if (ring1[i] != ring2[i]) {
        diff++;
      }
    }
    System.out.println("ring diff: " + diff);
  }

  public static void printNodeOfflineOverhead(ZNRecord record) {
    // build node -> partition map
    Map<String, Set<String>> nodeNextMap = new TreeMap<String, Set<String>>();
    for (String partitionName : record.getMapFields().keySet()) {
      Map<String, String> map1 = record.getMapField(partitionName);
      String master = "", slave = "";
      for (String nodeName : map1.keySet()) {
        if (!nodeNextMap.containsKey(nodeName)) {
          nodeNextMap.put(nodeName, new TreeSet<String>());
        }

        // String master = "", slave = "";
        if (map1.get(nodeName).equalsIgnoreCase("MASTER")) {
          master = nodeName;
        } else {
          if (slave.equalsIgnoreCase("")) {
            slave = nodeName;
          }
        }

      }
      nodeNextMap.get(master).add(slave);
    }
    System.out.println("next count: ");
    for (String key : nodeNextMap.keySet()) {
      System.out.println(nodeNextMap.get(key).size() + " ");
    }
    System.out.println();
  }

  /**
   * Helper function to calculate and print the standard deviation of the
   * partition assignment ideal state, also the min/max of master partitions
   * that is hosted on each node
   */
  public static void printIdealStateStats(ZNRecord record, String value) {
    Map<String, Integer> countsMap = new TreeMap<String, Integer>();
    for (String key : record.getMapFields().keySet()) {
      Map<String, String> map1 = record.getMapField(key);
      for (String k : map1.keySet()) {
        if (!countsMap.containsKey(k)) {
          countsMap.put(k, new Integer(0));//
        }
        if (value.equals("") || map1.get(k).equalsIgnoreCase(value)) {
          countsMap.put(k, countsMap.get(k).intValue() + 1);
        }
      }
    }
    double sum = 0;
    int maxCount = 0;
    int minCount = Integer.MAX_VALUE;

    System.out.println("Partition distributions: ");
    for (String k : countsMap.keySet()) {
      int count = countsMap.get(k);
      sum += count;
      if (maxCount < count) {
        maxCount = count;
      }
      if (minCount > count) {
        minCount = count;
      }
      System.out.print(count + " ");
    }
    System.out.println();
    double mean = sum / (countsMap.size());
    // calculate the deviation of the node distribution
    double deviation = 0;
    for (String k : countsMap.keySet()) {
      double count = countsMap.get(k);
      deviation += (count - mean) * (count - mean);
    }
    System.out.println("Mean: " + mean + " normal deviation:"
        + Math.sqrt(deviation / countsMap.size()));

    System.out.println("Max count: " + maxCount + " min count:" + minCount);
    /*
     * int steps = 10; int stepLen = (maxCount - minCount)/steps; List<Integer>
     * histogram = new ArrayList<Integer>((maxCount - minCount)/stepLen + 1);
     * for(int i = 0; i< (maxCount - minCount)/stepLen + 1; i++) {
     * histogram.add(0); } for(String k :countsMap.keySet()) { int count =
     * countsMap.get(k); int stepNo = (count - minCount)/stepLen;
     * histogram.set(stepNo, histogram.get(stepNo) +1); }
     * System.out.println("histogram:"); for(Integer x : histogram) {
     * System.out.print(x+" "); }
     */
  }

  public static void printHashRingStat(int[] hashRing) {
    double sum = 0, mean = 0, deviation = 0;
    Map<Integer, Integer> countsMap = new TreeMap<Integer, Integer>();
    for (int i = 0; i < hashRing.length; i++) {
      if (!countsMap.containsKey(hashRing[i])) {
        countsMap.put(hashRing[i], new Integer(0));//
      }
      countsMap.put(hashRing[i], countsMap.get(hashRing[i]).intValue() + 1);
    }
    int maxCount = Integer.MIN_VALUE;
    int minCount = Integer.MAX_VALUE;
    for (Integer k : countsMap.keySet()) {
      int count = countsMap.get(k);
      sum += count;
      if (maxCount < count) {
        maxCount = count;
      }
      if (minCount > count) {
        minCount = count;
      }
    }
    mean = sum / countsMap.size();
    for (Integer k : countsMap.keySet()) {
      int count = countsMap.get(k);
      deviation += (count - mean) * (count - mean);
    }
    System.out.println("hashring Mean: " + mean + " normal deviation:"
        + Math.sqrt(deviation / countsMap.size()));

  }

  static int[] getFnvHashArray(List<String> strings) {
    int[] result = new int[strings.size()];
    int i = 0;
    IdealCalculatorByConsistentHashing.FnvHash hashfunc =
        new IdealCalculatorByConsistentHashing.FnvHash();
    for (String s : strings) {
      int val = hashfunc.getHashValue(s) % 65536;
      if (val < 0)
        val += 65536;
      result[i++] = val;
    }
    return result;
  }

  static void printArrayStat(int[] vals) {
    double sum = 0, mean = 0, deviation = 0;

    for (int i = 0; i < vals.length; i++) {
      sum += vals[i];
    }
    mean = sum / vals.length;
    for (int i = 0; i < vals.length; i++) {
      deviation += (mean - vals[i]) * (mean - vals[i]);
    }
    System.out.println("normalized deviation: " + Math.sqrt(deviation / vals.length) / mean);
  }

  public static void main(String args[]) throws Exception {
    // Test the hash ring generation
    List<String> instanceNames = new ArrayList<String>();
    for (int i = 0; i < 10; i++) {
      instanceNames.add("localhost_123" + i);
    }

    // int[] ring1 =
    // IdealCalculatorByConsistentHashing.generateEvenHashRing(instanceNames,
    // 65535);
    // printHashRingStat(ring1);
    // int[] ring1 = getFnvHashArray(instanceNames);
    // printArrayStat(ring1);

    int partitions = 200, replicas = 2;
    String dbName = "espressoDB1";

    ZNRecord result =
        IdealCalculatorByConsistentHashing.calculateIdealState(instanceNames, partitions, replicas,
            dbName, new IdealCalculatorByConsistentHashing.FnvHash());
    System.out.println("\nMaster :");
    printIdealStateStats(result, "MASTER");

    System.out.println("\nSlave :");
    printIdealStateStats(result, "SLAVE");

    System.out.println("\nTotal :");
    printIdealStateStats(result, "");

    printNodeOfflineOverhead(result);
    /*
     * ZNRecordSerializer serializer = new ZNRecordSerializer(); byte[] bytes;
     * bytes = serializer.serialize(result); // System.out.println(new
     * String(bytes));
     * List<String> instanceNames2 = new ArrayList<String>(); for(int i = 0;i <
     * 40; i++) { instanceNames2.add("localhost_123"+i); }
     * ZNRecord result2 =
     * IdealCalculatorByConsistentHashing.calculateIdealState( instanceNames2,
     * partitions, replicas, dbName, new
     * IdealCalculatorByConsistentHashing.FnvHash());
     * printDiff(result, result2);
     * //IdealCalculatorByConsistentHashing.printIdealStateStats(result2);
     * int[] ring2 =
     * IdealCalculatorByConsistentHashing.generateHashRing(instanceNames2,
     * 30000);
     * IdealCalculatorByConsistentHashing.compareHashrings(ring1, ring2);
     * //printNodeStats(result); //printNodeStats(result2); bytes =
     * serializer.serialize(result2); printHashRingStat(ring2); //
     * System.out.println(new String(bytes));
     */

  }
}