package org.infinispan.distribution;

import java.util.Arrays;

/**
 * <a href = "http://weblogs.java.net/blog/tomwhite/archive/2007/11/consistent_hash.html">Consistent
 * hashing algorithm</a>.  Each target is entered into the pool <i>weight[i]</i>*<i>weightFactor</i>
 * times. Where <i>weight[i]</i> and <i>weightFactor</i> are integers greater than zero.
 *
 * @author akluge
 * @version $Rev: $, $Date: $
 */
public class GenericConsistentHash<T>
{

    private static final int FNV_OFFSET_BASIS = (int)2166136261L;
    private static final int FNV_PRIME        = 16777619;

    private final Hasher     hasher;
    private final T[]        nodes;
    private final Entry<T>[] pool;
    private final int        poolSize;
    private final int        replication;
    private final int        weightFactor;
    private final int[]      weights;

    public static class Builder<T>
    {
        private static final Hasher DEFAULT_HASH         = new ReHash();
        private static final int    DEFAULT_REPLICATION  = 2;
        private static final int    DEFAULT_WEIGHT       = 1;
        private static final int    DEFAULT_WEIGHTFACTOR = 30;

        private Hasher      hasher;
        private final T[]   nodes;
        private       int   replication;
        private       int   weightFactor;
        private       int[] weights;

        /**
         * Create a consistent hash Builder with the given set of nodes.
         * Defaults are used for replication and weights.
         *
         * @param nodes An array of objects to me entered into a consistent hash.
         */
        public Builder(T[] nodes)
        {
            hasher       = DEFAULT_HASH;
            this.nodes   = nodes;
            replication  = DEFAULT_REPLICATION;
            weightFactor = DEFAULT_WEIGHTFACTOR;
            weights      = new int[this.nodes.length];
            Arrays.fill(weights, DEFAULT_WEIGHT);
        }

        /**
         * @param replication  An int giving the number of nodes to be returned
         *                     by the get method.
         *
         * @return The Builder with the replication value set.
         *
         * @throws IllegalArgumentException If replication is less than 1. You must return
         *         at least one item with each get call.
         */
        public Builder<T> replication(int replication)
        {
            if (replication < 1) {
                throw new IllegalArgumentException("The minimum replication is 1.");
            }
            this.replication = replication;
            return this;
        }

        /**
         * @param weights      An array of integers controling the number of times
         *                     each node is entered into the hash pool. The ith node
         *                     occures <i>weights</i>[i]*<i>weightfactor</i> times
         *                     in the pool.
         *
         * @return The Builder with the weights set.
         *
         * @throws IllegalArgumentException If the weights and the nodes arrays are not
         *         the same size.
         *
         * @throws IllegalArgumentException If any of the weights are less than zero.
         */
        public Builder<T> weights(int[] weights)
        {
            if (weights.length != nodes.length)
            {
                throw new IllegalArgumentException("The weights and nodes arrays are of different sizes."
                                                   + "They must be of the same size.");
            }

            int nweights = weights.length;
            for (int i=0; i<nweights; i++)
            {
                // Does a zero weight actually make sense? Allows an item to be skipped in the map.
                if(weights[i] < 0) {
                    throw new IllegalArgumentException("Negative weights are not allowed.");
                }
            }

            this.weights = weights;
            return this;
        }

        /**
         * @param weightFactor An int containing a constant multiplier for the weights
         *                     array.
         *
         * @return The Builder with the weight factor set.
         *
         * @throws IllegalArgumentException if the weight factor is less than one.
         */
        public Builder<T> weightFactor(int weightFactor)
        {
            if (weightFactor < 1)
            {
                throw new IllegalArgumentException("The minimum weight factor is one.");
            }
            this.weightFactor = weightFactor;
            return this;
        }

        /**
         * Provide a strategy for computing the hash of the keys. This must provide
         * a hash method which produces a uniform distribution of values given a random
         * object.
         *
         * @param hasher Implements the {@link Hasher} interface.
         * 
         * @return The Builder with the hasher set.
         */
        public Builder<T> hasher(Hasher hasher)
        {
            if (hasher == null)
            {
                throw new IllegalArgumentException("A null hash is not allowed.");
            }
            this.hasher = hasher;
            return this;
        }

        /**
         * Construct a ConsistentHash from the settings provided to the Builder.
         *
         * @return A GenericConsistentHash<T>.
         */
        public GenericConsistentHash<T> build()
        {
            return new GenericConsistentHash<T>(this);
        }


    }

    /**
     * Build a consistent hash as described by the builder.
     *
     * @param builder A Bilder<T> instance.
     */
    private GenericConsistentHash(Builder<T> builder)
    {
        hasher       = builder.hasher;
        nodes        = builder.nodes;
        replication  = builder.replication;
        weightFactor = builder.weightFactor;
        weights      = builder.weights;

       int nnodes   = nodes.length;

        int poolSize = 0;
        for (int i=0; i<nnodes; i++)
        {
            poolSize += weights[i] * weightFactor;
        }
        this.poolSize = poolSize;
        // Each entry will wrap an object of type T, this odd construct is
        // because you can not create an array with a generic type.
        pool = (Entry<T>[])new Entry[this.poolSize];

        int nentries = 0;
        for (int i=0; i<nnodes; i++)
        {
            nentries = add(nodes[i], weights[i] * weightFactor,
                           pool,     nentries);
        }
        Arrays.sort(pool);
    }

    /**
     * Adds a node of type T count times to the pool of nodes.
     *
     * @param node     A generic (T class) node to be added to the hash pool.
     * @param count    An int giving the number of times the node is added to the pool.
     *                 The count is greater than zero, and is likely greater than 10.
     *                 The count for the ith node is usually the
     *                 <i>weights</i>[i]*<i>weightfactor</i>
     * @param pool     The expanded set of all nodes mapped multiple times into the
     *                 range of the hash function.
     * @param position The position in the pool to begin adding node entries.
     *
     * @return position;
     */
    private int add(T node, int count, Entry<T>[] pool, int position)
    {
        Entry<T> entry;
        int      hash;
        String   nodeName = node.toString();
        for (int i = 0; i < count; i++)
        {
            hash = hash((Integer.toString(i) + nodeName).getBytes());
            //System.out.print(" Hash: " + inode);
            entry = new Entry<T>(node, nodeName, i, hash);
            pool[position++] = entry;
        }
        return position;
    }

    /**
     * Removes a node of type T from the pool of nodes. This requires the location
     * and removal of <i>weights</i>[i]*<i>weightfactor</i> copies of the node from
     * the pool. The entries corresponding to the given node are set to null.
     *
     * @param node A generic node of type T to be removed from the pool.
     */
    public void remove(T node)
    {
        int inode = find(nodes, node);
        // The number of expected instances of this node in the pool.
        int count = weights[inode]*weightFactor;
        int hash;
        for (int i = 0; i < count; i++)
        {
            // Find the ith occurance of the node.
            hash  = hash((node.toString() + i).getBytes());
            inode = binarySearch(pool, 0, poolSize-1, hash);
            if (inode >= 0)
            {
                pool[inode] = null;
            } // Eventually, throw exception here?? Something has gone very wrong.
        }
    }

    /**
     * Get the number of entries we return for get request.
     *
     * @return An int giving the number of nodes we return with each get.
     */
    public int getReplication()
    {
        return replication;
    }

    /**
     * Returns and array containing <i>replication</i> unique nodes for a key.
     *
     * @param key   An object used as a key in the hash. The object's toString method
     *              is called, and a fast hash is used to compute the hash. The toString
     *              method on the key class must present the variables that contain the
     *              relevant state of the object.
     *
     * @param nodes An array of the paramaterized type T, which must match the original
     *              type used in creating the consistent hash, and whose size must match
     *              the replication count.
     * @see #getReplication()
     *
     * @return An array containing <i>replication</i> objects of the paramaterized
     *         type <i>T</i>.
     */
    public T[] locate(Object key, T[] nodes)
    {
        if (key == null)
        {
            throw new NullPointerException("Attempt to get with null key");
        }

        if (nodes.length < replication)
        {
            throw new IllegalArgumentException("Nodes must contain at least "
                                               + replication + " entries.");
        }

        int hashValue = hasher.hash(key);

        return locate(hashValue, nodes, replication);
    }

    /**
     * Returns and array containing the requested <i>count</i> of unique nodes for a key.
     *
     * @param key   An object used as a key in the hash. By default, the key's
     *              hashCode is used, so it is important that this method be
     *              implemented.
     *
     * @param nodes An array of the paramaterized type <i>T</i>, which must match the original
     *              type used in creating the consistent hash, and whose size must match
     *              the replication count.
     *
     * @oaram count An int giving the number of node to be returned.
     *
     * @return An array containing <i>count</i> objects of the paramaterized
     *         type <i>T</i>.
     */
    public T[] locate(Object key, T[] nodes, int count)
    {
        if (key == null)
        {
            throw new NullPointerException("Attempt to get with null key");
        }

        if (nodes.length < count)
        {
            throw new IllegalArgumentException("Nodes must contain at least "
                                               + count + " entries.");
        }
        
        if (count > this.nodes.length)
        {
            throw new IllegalArgumentException();
        }

        int hashValue = hasher.hash(key);

        return locate(hashValue, nodes, count);
    }

    /**
     * Returns and array containing <i>replication</i> unique nodes for a value.
     *
     * @param value An int, usually a hash, to be mapped to a bin via the CH.
     *
     * @param nodes An array of the paramaterized type T, which must match the original
     *              type used in creating the consistent hash, and whose size must match
     *              the replication count.
     * @see #getReplication()
     *
     * @return An array containing <i>replication</i> objects of the paramaterized
     *         type <i>T</i>.
     */
    public T[] locate(int value, T[] nodes, int count)
    {
        // Stop looking if we have checked the entire pool.
        int checked = 0;
        // Or if we have found as many instances as we need.
        int found   = 0;
        // Start looking at the first (primary) node for entries for this value.
        int inode   = findNearest(pool, value);
        //System.out.println(key + " : " + hash(key.toString().getBytes()) + " : " + inode);
        while (found < count && checked < poolSize)
        {
            if (pool[inode] != null && find(nodes, pool[inode].object)<0)
            {
                nodes[found++] = pool[inode].object;
            }
            inode = (++inode)%poolSize;
            checked++;
        }
        return nodes;
    }

    /**
     * Return the single node that owns the key. Used to validate the destination of
     * the primary request.
     *
     * @param value An int, usually representing a hash. Locate the bin, or node,
     *              that owns the int.
     *
     * @return A node that owns the given int.
     */
    public T locate(int value)
    {
        return pool[findNearest(pool, value)].object;
    }

    /**
     * The number of bins in the consistent hash.
     * @return An int giving the number of items in the consistent hash.
     */
    public int getPoolSize()
    {
        return poolSize;
    }

    /**
     * Get the objects that define the consistent hash, in the order they are sorted into for the hash.
     *
     * @param values An array ot type T to hold the original values, now duplicated (by weight) and sorted.
     * @return An array of type T.
     */
    public T[] getPoolValues(T[] values)
    {
        if (values.length != poolSize)
        {
            throw new IllegalArgumentException("Size mismatch in values array. Expected " + poolSize + " got " + values.length);
        }

        for (int i=0; i<poolSize; i++)
        {
            values[i] = pool[i].object;
        }
        return values;
    }


    /**
     * An implimentation of the <a href = "http://www.isthe.com/chongo/tech/comp/fnv/">FNV 1
     * hash</a>. If needed, consider a <a href = "http://www.azillionmonkeys.com/qed/hash.html">faster
     * hash</a>. This is a fast hash with a good distribution, which is all that matters in this
     * case.
     *
     * @param bytes The array of bytes to be hashed.
     * @return An int containing the FNV hash of the byte array.
     */
    public int hash(byte[] bytes)
    {
        int nbytes = bytes.length;
        int hash   = FNV_OFFSET_BASIS;
        for(int i=0; i<nbytes; i++)
        {
            hash ^= bytes[i];
            hash *= FNV_PRIME;
        }
        return hash;
    }

    /**
     * Find a target for a hash key within a list of targets. We search within
     * a slice of the array bounded by lowerBound and upperBound. Further
     * we assume that lowerBound and upperBound are small enough that their
     * sum will not overflow an int.
     *
     * TODO: Replace wity Arrays binary search call, with comparitor.
     *
     * @param targets    The sorted array of Entry objects.
     * @param lowerBound The initial lower bound for the search.
     * @param upperBound The initial upper bound for the search.
     * @param hash       The desired hash key.
     *
     * @return An int giving the index of the desired entry in the list of
     *         targets. If the target is not found, then -(lowerBound +1)
     *         will be returned, where lowerBound is the lower bound
     *         of the search after possibly several iterations.
     */
    public int binarySearch(Entry<T>[] targets,    int lowerBound,
                            int        upperBound, int hash)
    {
	    while (lowerBound <= upperBound)
        {
            // Fast div by 2. We assume that the number of targets is small enough
            // that the sum will not overflow an int.
	        int mid         = (lowerBound + upperBound) >>> 1;
	        int currentHash = targets[mid].hash;

	        if (currentHash < hash)
            {
		        lowerBound = mid + 1;
            }
	        else if (currentHash > hash)
            {
		        upperBound = mid - 1;
            }
	        else
            {
		        return mid;
            }
	    }
        // The +1 ensures that the return value is negative, even when the hash
        // is off the left edge of the array.
	    return -(lowerBound +1);
    }

    /**
     * Finds the lowest index into the targets array such that T[i] >= hash.
     *
     * @param targets The list of targets over which we maintain the consistent hash.
     * @param hash    The hash being maped to a bin via the consistent hash.
     *
     * @return An int, the lowest index into the target array such that T[i] >= hash.
     */
    public int findNearest(Entry<T>[] targets, int hash)
    {
        // Find the index of the node - or at least a near one.
        // We only search up to targets.length-1. If the element
        // is greater than the last entry in the list, then map
        // it to the first one.
        int inode = binarySearch(targets, 0, targets.length -1, hash);

        // If the returned value is less than zero, then no exact match was found.
        if (inode < 0)
        {
            // The value returned is -(lowerBound +1), we want the lower bound back.
            inode = -(inode + 1);

            // If hash is greater than the last entry, wrap around to the first.
            if (inode >= targets.length)
            {
                inode = 0;
            }
        }

        return inode;
    }

    /**
     * Linear search for a target in a list of targets.
     *
     * @param nodes  An array of the paramaterized type T.
     * @param target An entry for T objects.
     *
     * @return An int giving the location of the target in the array
     *         of nodes, or -1 if not found.
     */
    private int find(T[] nodes, T target)
    {
        int nnodes = nodes.length;
        for (int i=0; i<nnodes; i++)
        {
            if (nodes[i] != null && nodes[i].equals(target))
            {
                return i;
            }
        }
        return -1;
    }

    /**
     * @return A String representing the object pool.
     */
    @Override
    public String toString()
    {
        StringBuilder stringBuilder = new StringBuilder(1000);
        stringBuilder.append(" pool: ");
        return " pool: " + Arrays.deepToString(pool) + "\n"
               + " replication: " + this.replication + "\n";
    }

    @Override
    public boolean equals(Object other)
    {
        if (other == null
           || !(other instanceof GenericConsistentHash))
        {
            return false;
        }

        GenericConsistentHash otherHash = (GenericConsistentHash)other;
        return Arrays.equals(pool, otherHash.pool)
               && replication  == otherHash.replication
               && weightFactor == otherHash.weightFactor
               && Arrays.equals(weights, otherHash.weights);
    }

    @Override
    public int hashCode()
    {
        int hashCode = 1;
        for (int i=0; i<poolSize; i++)
        {
            hashCode = 31*hashCode + pool[i].hash;
        }

        for (int i=0; i<weights.length; i++)
        {
            hashCode = 31*hashCode + weights[i];
        }

        hashCode = 31*hashCode + replication;

        hashCode = 31*hashCode + weightFactor;

        return hashCode;
    }

    /**
     * An entry into a consistent hash. It wrapps the original object,
     * the object's hash as used to generate the consistent hash, the
     * value extracted from the object used to generate the hash, and
     * the modifier used to differentiate the hash.
     */
    public class Entry<T extends Object> implements Comparable<Entry>
    {
        public final int    differentiator;
        public final int    hash;
        public final T      object;
        public final String string;

        public Entry(T object, String string, int differentiator, int hash)
        {
            this.differentiator = differentiator;
            this.hash           = hash;
            this.object         = object;
            this.string         = string;
        }

        /**
         * Compare this Entry with another Entry. First the hash values
         * are compaired, then the differentiator is compaired. if the
         * hash values are equal.
         *
         * @param other An Entry object to be compared with this object. Returns
         *              <ul>
         *                <li>-1 if this Entry is less than the other Entry.</li>
         *                <li>0  if they are equal.</li>
         *                <li>+1 if this Entry is greater than the other Entry.</li>
         *              </ul>
         *
         * @return
         */
        //@Override
        public int compareTo(Entry other)
        {
            if (this.hash < other.hash)
            {
                return -1;
            }

            if (this.hash > other.hash)
            {
                return 1;
            }

            if (this.differentiator < other.differentiator)
            {
                return -1;
            }

            if (this.differentiator > other.differentiator)
            {
                return +1;
            }

            return 0;
        }

        @Override
        public boolean equals(Object other)
        {
            if (other instanceof Entry)
            {
                Entry otherEntry = (Entry)other;

                return hash == otherEntry.hash
                       && differentiator == otherEntry.differentiator
                       && object.equals(otherEntry.object);
            }
            return false;
        }

        @Override
        public int hashCode()
        {
            return hash;
        }


        @Override
        public String toString()
        {
            return string + ":" + Integer.toHexString(hash);
        }
    }
}