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