/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 *
 * Modified for https://jira.jboss.org/jira/browse/ISPN-299
 * Includes ideas described in http://portal.acm.org/citation.cfm?id=1547428
 *
 */
package io.debezium.util;

import static java.util.Collections.singletonMap;
import static java.util.Collections.unmodifiableMap;

import java.io.IOException;
import java.io.Serializable;
import java.util.AbstractCollection;
import java.util.AbstractMap;
import java.util.AbstractSet;
import java.util.Collection;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * A hash table supporting full concurrency of retrievals and
 * adjustable expected concurrency for updates. This class obeys the
 * same functional specification as {@link java.util.Hashtable}, and
 * includes versions of methods corresponding to each method of
 * <tt>Hashtable</tt>. However, even though all operations are
 * thread-safe, retrieval operations do <em>not</em> entail locking,
 * and there is <em>not</em> any support for locking the entire table
 * in a way that prevents all access.  This class is fully
 * interoperable with <tt>Hashtable</tt> in programs that rely on its
 * thread safety but not on its synchronization details.
 * <p/>
 * <p> Retrieval operations (including <tt>get</tt>) generally do not
 * block, so may overlap with update operations (including
 * <tt>put</tt> and <tt>remove</tt>). Retrievals reflect the results
 * of the most recently <em>completed</em> update operations holding
 * upon their onset.  For aggregate operations such as <tt>putAll</tt>
 * and <tt>clear</tt>, concurrent retrievals may reflect insertion or
 * removal of only some entries.  Similarly, Iterators and
 * Enumerations return elements reflecting the state of the hash table
 * at some point at or since the creation of the iterator/enumeration.
 * They do <em>not</em> throw {@link java.util.ConcurrentModificationException}.
 * However, iterators are designed to be used by only one thread at a time.
 * <p/>
 * <p> The allowed concurrency among update operations is guided by
 * the optional <tt>concurrencyLevel</tt> constructor argument
 * (default <tt>16</tt>), which is used as a hint for internal sizing.  The
 * table is internally partitioned to try to permit the indicated
 * number of concurrent updates without contention. Because placement
 * in hash tables is essentially random, the actual concurrency will
 * vary.  Ideally, you should choose a value to accommodate as many
 * threads as will ever concurrently modify the table. Using a
 * significantly higher value than you need can waste space and time,
 * and a significantly lower value can lead to thread contention. But
 * overestimates and underestimates within an order of magnitude do
 * not usually have much noticeable impact. A value of one is
 * appropriate when it is known that only one thread will modify and
 * all others will only read. Also, resizing this or any other kind of
 * hash table is a relatively slow operation, so, when possible, it is
 * a good idea to provide estimates of expected table sizes in
 * constructors.
 * <p/>
 * <p>This class and its views and iterators implement all of the
 * <em>optional</em> methods of the {@link Map} and {@link Iterator}
 * interfaces.
 * <p/>
 * <p>This class is copied from Hibernate (which took it from Infinispan),
 * and was originally written
 * by Doug Lea with assistance from members of JCP JSR-166 Expert Group and
 * released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain</p>
 * <p/>
 * <p/>
 * <p> Like {@link java.util.Hashtable} but unlike {@link HashMap}, this class
 * does <em>not</em> allow <tt>null</tt> to be used as a key or value.
 *
 * @param <K> the type of keys maintained by this map
 * @param <V> the type of mapped values
 *
 * @author Doug Lea
 */
public class BoundedConcurrentHashMap<K, V> extends AbstractMap<K, V>
        implements ConcurrentMap<K, V>, Serializable {
    private static final long serialVersionUID = 7249069246763182397L;

    /*
     * The basic strategy is to subdivide the table among Segments,
     * each of which itself is a concurrently readable hash table.
     */

    /* ---------------- Constants -------------- */

    /**
     * The default initial capacity for this table,
     * used when not otherwise specified in a constructor.
     */
    static final int DEFAULT_MAXIMUM_CAPACITY = 512;

    /**
     * The default load factor for this table, used when not
     * otherwise specified in a constructor.
     */
    static final float DEFAULT_LOAD_FACTOR = 0.75f;

    /**
     * The default concurrency level for this table, used when not
     * otherwise specified in a constructor.
     */
    static final int DEFAULT_CONCURRENCY_LEVEL = 16;

    /**
     * The maximum capacity, used if a higher value is implicitly
     * specified by either of the constructors with arguments.  MUST
     * be a power of two <= 1<<30 to ensure that entries are indexable
     * using ints.
     */
    static final int MAXIMUM_CAPACITY = 1 << 30;

    /**
     * The maximum number of segments to allow; used to bound
     * constructor arguments.
     */
    static final int MAX_SEGMENTS = 1 << 16; // slightly conservative

    /**
     * Number of unsynchronized retries in size and containsValue
     * methods before resorting to locking. This is used to avoid
     * unbounded retries if tables undergo continuous modification
     * which would make it impossible to obtain an accurate result.
     */
    static final int RETRIES_BEFORE_LOCK = 2;

    /* ---------------- Fields -------------- */

    /**
     * Mask value for indexing into segments. The upper bits of a
     * key's hash code are used to choose the segment.
     */
    final int segmentMask;

    /**
     * Shift value for indexing within segments.
     */
    final int segmentShift;

    /**
     * The segments, each of which is a specialized hash table
     */
    final Segment<K, V>[] segments;

    transient Set<K> keySet;
    transient Set<Map.Entry<K, V>> entrySet;
    transient Collection<V> values;

    /* ---------------- Small Utilities -------------- */

    /**
     * Applies a supplemental hash function to a given hashCode, which
     * defends against poor quality hash functions.  This is critical
     * because ConcurrentHashMap uses power-of-two length hash tables,
     * that otherwise encounter collisions for hashCodes that do not
     * differ in lower or upper bits.
     */
    private static int hash(int h) {
        // Spread bits to regularize both segment and index locations,
        // using variant of single-word Wang/Jenkins hash.
        h += h << 15 ^ 0xffffcd7d;
        h ^= h >>> 10;
        h += h << 3;
        h ^= h >>> 6;
        h += (h << 2) + (h << 14);
        return h ^ h >>> 16;
    }

    /**
     * Returns the segment that should be used for key with given hash
     *
     * @param hash the hash code for the key
     *
     * @return the segment
     */
    final Segment<K, V> segmentFor(int hash) {
        return segments[hash >>> segmentShift & segmentMask];
    }

    /* ---------------- Inner Classes -------------- */

    /**
     * ConcurrentHashMap list entry. Note that this is never exported
     * out as a user-visible Map.Entry.
     * <p/>
     * Because the value field is volatile, not final, it is legal wrt
     * the Java Memory Model for an unsynchronized reader to see null
     * instead of initial value when read via a data race.  Although a
     * reordering leading to this is not likely to ever actually
     * occur, the Segment.readValueUnderLock method is used as a
     * backup in case a null (pre-initialized) value is ever seen in
     * an unsynchronized access method.
     */
    private static class HashEntry<K, V> {
        final K key;
        final int hash;
        volatile V value;
        final HashEntry<K, V> next;

        HashEntry(K key, int hash, HashEntry<K, V> next, V value) {
            this.key = key;
            this.hash = hash;
            this.next = next;
            this.value = value;
        }

        @Override
        public int hashCode() {
            int result = 17;
            result = result * 31 + hash;
            result = result * 31 + key.hashCode();
            return result;
        }

        @Override
        public boolean equals(Object o) {
            // HashEntry is internal class, never leaks out of CHM, hence slight optimization
            if (this == o) {
                return true;
            }
            if (o == null) {
                return false;
            }
            HashEntry<?, ?> other = (HashEntry<?, ?>) o;
            return hash == other.hash && key.equals(other.key);
        }

        @SuppressWarnings("unchecked")
        static <K, V> HashEntry<K, V>[] newArray(int i) {
            return new HashEntry[i];
        }
    }

    private enum Recency {
        HIR_RESIDENT,
        LIR_RESIDENT,
        HIR_NONRESIDENT
    }

    public enum Eviction {
        NONE {
            @Override
            public <K, V> EvictionPolicy<K, V> make(Segment<K, V> s, int capacity, float lf) {
                return new NullEvictionPolicy<K, V>();
            }
        },
        LRU {
            @Override
            public <K, V> EvictionPolicy<K, V> make(Segment<K, V> s, int capacity, float lf) {
                return new LRU<K, V>(s, capacity, lf, capacity * 10, lf);
            }
        },
        LIRS {
            @Override
            public <K, V> EvictionPolicy<K, V> make(Segment<K, V> s, int capacity, float lf) {
                return new LIRS<K, V>(s, capacity, capacity * 10, lf);
            }
        };

        abstract <K, V> EvictionPolicy<K, V> make(Segment<K, V> s, int capacity, float lf);
    }

    public interface EvictionListener<K, V> {
        void onEntryEviction(Map<K, V> evicted);

        void onEntryChosenForEviction(V internalCacheEntry);
    }

    static final class NullEvictionListener<K, V> implements EvictionListener<K, V> {
        @Override
        public void onEntryEviction(Map<K, V> evicted) {
            // Do nothing.
        }

        @Override
        public void onEntryChosenForEviction(V internalCacheEntry) {
            // Do nothing.
        }
    }

    public interface EvictionPolicy<K, V> {

        public final static int MAX_BATCH_SIZE = 64;

        HashEntry<K, V> createNewEntry(K key, int hash, HashEntry<K, V> next, V value);

        /**
         * Invokes eviction policy algorithm and returns set of evicted entries.
         * <p/>
         * <p/>
         * Set cannot be null but could possibly be an empty set.
         *
         * @return set of evicted entries.
         */
        Set<HashEntry<K, V>> execute();

        /**
         * Invoked to notify EvictionPolicy implementation that there has been an attempt to access
         * an entry in Segment, however that entry was not present in Segment.
         *
         * @param e accessed entry in Segment
         *
         * @return non null set of evicted entries.
         */
        Set<HashEntry<K, V>> onEntryMiss(HashEntry<K, V> e);

        /**
         * Invoked to notify EvictionPolicy implementation that an entry in Segment has been
         * accessed. Returns true if batching threshold has been reached, false otherwise.
         * <p/>
         * Note that this method is potentially invoked without holding a lock on Segment.
         *
         * @param e accessed entry in Segment
         *
         * @return true if batching threshold has been reached, false otherwise.
         */
        boolean onEntryHit(HashEntry<K, V> e);

        /**
         * Invoked to notify EvictionPolicy implementation that an entry e has been removed from
         * Segment.
         *
         * @param e removed entry in Segment
         */
        void onEntryRemove(HashEntry<K, V> e);

        /**
         * Invoked to notify EvictionPolicy implementation that all Segment entries have been
         * cleared.
         */
        void clear();

        /**
         * Returns type of eviction algorithm (strategy).
         *
         * @return type of eviction algorithm
         */
        Eviction strategy();

        /**
         * Returns true if batching threshold has expired, false otherwise.
         * <p/>
         * Note that this method is potentially invoked without holding a lock on Segment.
         *
         * @return true if batching threshold has expired, false otherwise.
         */
        boolean thresholdExpired();
    }

    static class NullEvictionPolicy<K, V> implements EvictionPolicy<K, V> {

        @Override
        public void clear() {
            // Do nothing.
        }

        @Override
        public Set<HashEntry<K, V>> execute() {
            return Collections.emptySet();
        }

        @Override
        public boolean onEntryHit(HashEntry<K, V> e) {
            return false;
        }

        @Override
        public Set<HashEntry<K, V>> onEntryMiss(HashEntry<K, V> e) {
            return Collections.emptySet();
        }

        @Override
        public void onEntryRemove(HashEntry<K, V> e) {
            // Do nothing.
        }

        @Override
        public boolean thresholdExpired() {
            return false;
        }

        @Override
        public Eviction strategy() {
            return Eviction.NONE;
        }

        @Override
        public HashEntry<K, V> createNewEntry(K key, int hash, HashEntry<K, V> next, V value) {
            return new HashEntry<K, V>(key, hash, next, value);
        }
    }

    static final class LRU<K, V> extends LinkedHashMap<HashEntry<K, V>, V> implements EvictionPolicy<K, V> {

        /**
         * The serialVersionUID
         */
        private static final long serialVersionUID = -7645068174197717838L;

        private final ConcurrentLinkedQueue<HashEntry<K, V>> accessQueue;
        private final Segment<K, V> segment;
        private final int maxBatchQueueSize;
        private final int trimDownSize;
        private final float batchThresholdFactor;
        private final Set<HashEntry<K, V>> evicted;

        public LRU(Segment<K, V> s, int capacity, float lf, int maxBatchSize, float batchThresholdFactor) {
            super(capacity, lf, true);
            this.segment = s;
            this.trimDownSize = capacity;
            this.maxBatchQueueSize = maxBatchSize > MAX_BATCH_SIZE ? MAX_BATCH_SIZE : maxBatchSize;
            this.batchThresholdFactor = batchThresholdFactor;
            this.accessQueue = new ConcurrentLinkedQueue<HashEntry<K, V>>();
            this.evicted = new HashSet<HashEntry<K, V>>();
        }

        @Override
        public Set<HashEntry<K, V>> execute() {
            Set<HashEntry<K, V>> evictedCopy = new HashSet<HashEntry<K, V>>();
            for (HashEntry<K, V> e : accessQueue) {
                put(e, e.value);
            }
            evictedCopy.addAll(evicted);
            accessQueue.clear();
            evicted.clear();
            return evictedCopy;
        }

        @Override
        public Set<HashEntry<K, V>> onEntryMiss(HashEntry<K, V> e) {
            put(e, e.value);
            if (!evicted.isEmpty()) {
                Set<HashEntry<K, V>> evictedCopy = new HashSet<HashEntry<K, V>>();
                evictedCopy.addAll(evicted);
                evicted.clear();
                return evictedCopy;
            }
            else {
                return Collections.emptySet();
            }
        }

        /*
         * Invoked without holding a lock on Segment
         */
        @Override
        public boolean onEntryHit(HashEntry<K, V> e) {
            accessQueue.add(e);
            return accessQueue.size() >= maxBatchQueueSize * batchThresholdFactor;
        }

        /*
         * Invoked without holding a lock on Segment
         */
        @Override
        public boolean thresholdExpired() {
            return accessQueue.size() >= maxBatchQueueSize;
        }

        @Override
        public void onEntryRemove(HashEntry<K, V> e) {
            remove(e);
            // we could have multiple instances of e in accessQueue; remove them all
            while (accessQueue.remove(e)) {
                continue;
            }
        }

        @Override
        public void clear() {
            super.clear();
            accessQueue.clear();
        }

        @Override
        public Eviction strategy() {
            return Eviction.LRU;
        }

        protected boolean isAboveThreshold() {
            return size() > trimDownSize;
        }

        @Override
        protected boolean removeEldestEntry(Map.Entry<HashEntry<K, V>, V> eldest) {
            boolean aboveThreshold = isAboveThreshold();
            if (aboveThreshold) {
                HashEntry<K, V> evictedEntry = eldest.getKey();
                segment.evictionListener.onEntryChosenForEviction(evictedEntry.value);
                segment.remove(evictedEntry.key, evictedEntry.hash, null);
                evicted.add(evictedEntry);
            }
            return aboveThreshold;
        }

        @Override
        public HashEntry<K, V> createNewEntry(K key, int hash, HashEntry<K, V> next, V value) {
            return new HashEntry<K, V>(key, hash, next, value);
        }
    }

    /**
     * Adapted to Infinispan BoundedConcurrentHashMap using LIRS implementation ideas from Charles Fry (fry@google.com)
     * See http://code.google.com/p/concurrentlinkedhashmap/source/browse/trunk/src/test/java/com/googlecode/concurrentlinkedhashmap/caches/LirsMap.java
     * for original sources
     */
    private static final class LIRSHashEntry<K, V> extends HashEntry<K, V> {

        // LIRS stack S
        private LIRSHashEntry<K, V> previousInStack;
        private LIRSHashEntry<K, V> nextInStack;

        // LIRS queue Q
        private LIRSHashEntry<K, V> previousInQueue;
        private LIRSHashEntry<K, V> nextInQueue;
        volatile Recency state;

        LIRS<K, V> owner;

        LIRSHashEntry(LIRS<K, V> owner, K key, int hash, HashEntry<K, V> next, V value) {
            super(key, hash, next, value);
            this.owner = owner;
            this.state = Recency.HIR_RESIDENT;

            // initially point everything back to self
            this.previousInStack = this;
            this.nextInStack = this;
            this.previousInQueue = this;
            this.nextInQueue = this;
        }

        @Override
        public int hashCode() {
            int result = 17;
            result = result * 31 + hash;
            result = result * 31 + key.hashCode();
            return result;
        }

        @Override
        public boolean equals(Object o) {
            // HashEntry is internal class, never leaks out of CHM, hence slight optimization
            if (this == o) {
                return true;
            }
            if (o == null) {
                return false;
            }
            HashEntry<?, ?> other = (HashEntry<?, ?>) o;
            return hash == other.hash && key.equals(other.key);
        }

        /**
         * Returns true if this entry is in the stack, false otherwise.
         */
        public boolean inStack() {
            return (nextInStack != null);
        }

        /**
         * Returns true if this entry is in the queue, false otherwise.
         */
        public boolean inQueue() {
            return (nextInQueue != null);
        }

        /**
         * Records a cache hit.
         */
        public void hit(Set<HashEntry<K, V>> evicted) {
            switch (state) {
                case LIR_RESIDENT:
                    hotHit(evicted);
                    break;
                case HIR_RESIDENT:
                    coldHit(evicted);
                    break;
                case HIR_NONRESIDENT:
                    throw new IllegalStateException("Can't hit a non-resident entry!");
                default:
                    throw new AssertionError("Hit with unknown status: " + state);
            }
        }

        /**
         * Records a cache hit on a hot block.
         */
        private void hotHit(Set<HashEntry<K, V>> evicted) {
            // See section 3.3 case 1:
            // "Upon accessing an LIR block X:
            // This access is guaranteed to be a hit in the cache."

            // "We move it to the top of stack S."
            boolean onBottom = (owner.stackBottom() == this);
            moveToStackTop();

            // "If the LIR block is originally located in the bottom of the stack,
            // we conduct a stack pruning."
            if (onBottom) {
                owner.pruneStack(evicted);
            }
        }

        /**
         * Records a cache hit on a cold block.
         */
        private void coldHit(Set<HashEntry<K, V>> evicted) {
            // See section 3.3 case 2:
            // "Upon accessing an HIR resident block X:
            // This is a hit in the cache."

            // "We move it to the top of stack S."
            boolean inStack = inStack();
            moveToStackTop();

            // "There are two cases for block X:"
            if (inStack) {
                // "(1) If X is in the stack S, we change its status to LIR."
                hot();

                // "This block is also removed from list Q."
                removeFromQueue();

                // "The LIR block in the bottom of S is moved to the end of list Q
                // with its status changed to HIR."
                owner.stackBottom().migrateToQueue();

                // "A stack pruning is then conducted."
                owner.pruneStack(evicted);
            }
            else {
                // "(2) If X is not in stack S, we leave its status in HIR and move
                // it to the end of list Q."
                moveToQueueEnd();
            }
        }

        /**
         * Records a cache miss. This is how new entries join the LIRS stack and
         * queue. This is called both when a new entry is first created, and when a
         * non-resident entry is re-computed.
         */
        private Set<HashEntry<K, V>> miss() {
            Set<HashEntry<K, V>> evicted = Collections.emptySet();
            if (owner.hotSize < owner.maximumHotSize) {
                warmupMiss();
            }
            else {
                evicted = new HashSet<HashEntry<K, V>>();
                fullMiss(evicted);
            }

            // now the missed item is in the cache
            owner.size++;
            return evicted;
        }

        /**
         * Records a miss when the hot entry set is not full.
         */
        private void warmupMiss() {
            // See section 3.3:
            // "When LIR block set is not full, all the referenced blocks are
            // given an LIR status until its size reaches L_lirs."
            hot();
            moveToStackTop();
        }

        /**
         * Records a miss when the hot entry set is full.
         */
        private void fullMiss(Set<HashEntry<K, V>> evicted) {
            // See section 3.3 case 3:
            // "Upon accessing an HIR non-resident block X:
            // This is a miss."

            // This condition is unspecified in the paper, but appears to be
            // necessary.
            if (owner.size >= owner.maximumSize) {
                // "We remove the HIR resident block at the front of list Q (it then
                // becomes a non-resident block), and replace it out of the cache."
                LIRSHashEntry<K, V> evictedNode = owner.queueFront();
                evicted.add(evictedNode);
            }

            // "Then we load the requested block X into the freed buffer and place
            // it on the top of stack S."
            boolean inStack = inStack();
            moveToStackTop();

            // "There are two cases for block X:"
            if (inStack) {
                // "(1) If X is in stack S, we change its status to LIR and move the
                // LIR block in the bottom of stack S to the end of list Q with its
                // status changed to HIR. A stack pruning is then conducted.
                hot();
                owner.stackBottom().migrateToQueue();
                owner.pruneStack(evicted);
            }
            else {
                // "(2) If X is not in stack S, we leave its status in HIR and place
                // it in the end of list Q."
                cold();
            }
        }

        /**
         * Marks this entry as hot.
         */
        private void hot() {
            if (state != Recency.LIR_RESIDENT) {
                owner.hotSize++;
            }
            state = Recency.LIR_RESIDENT;
        }

        /**
         * Marks this entry as cold.
         */
        private void cold() {
            if (state == Recency.LIR_RESIDENT) {
                owner.hotSize--;
            }
            state = Recency.HIR_RESIDENT;
            moveToQueueEnd();
        }

        /**
         * Marks this entry as non-resident.
         */
        @SuppressWarnings("fallthrough")
        private void nonResident() {
            switch (state) {
                case LIR_RESIDENT:
                    owner.hotSize--;
                    // fallthrough
                case HIR_RESIDENT:
                    owner.size--;
                    break;
            }
            state = Recency.HIR_NONRESIDENT;
        }

        /**
         * Returns true if this entry is resident in the cache, false otherwise.
         */
        public boolean isResident() {
            return (state != Recency.HIR_NONRESIDENT);
        }

        /**
         * Temporarily removes this entry from the stack, fixing up neighbor links.
         * This entry's links remain unchanged, meaning that {@link #inStack()} will
         * continue to return true. This should only be called if this node's links
         * will be subsequently changed.
         */
        private void tempRemoveFromStack() {
            if (inStack()) {
                previousInStack.nextInStack = nextInStack;
                nextInStack.previousInStack = previousInStack;
            }
        }

        /**
         * Removes this entry from the stack.
         */
        private void removeFromStack() {
            tempRemoveFromStack();
            previousInStack = null;
            nextInStack = null;
        }

        /**
         * Inserts this entry before the specified existing entry in the stack.
         */
        private void addToStackBefore(LIRSHashEntry<K, V> existingEntry) {
            previousInStack = existingEntry.previousInStack;
            nextInStack = existingEntry;
            previousInStack.nextInStack = this;
            nextInStack.previousInStack = this;
        }

        /**
         * Moves this entry to the top of the stack.
         */
        private void moveToStackTop() {
            tempRemoveFromStack();
            addToStackBefore(owner.header.nextInStack);
        }

        /**
         * Moves this entry to the bottom of the stack.
         */
        private void moveToStackBottom() {
            tempRemoveFromStack();
            addToStackBefore(owner.header);
        }

        /**
         * Temporarily removes this entry from the queue, fixing up neighbor links.
         * This entry's links remain unchanged. This should only be called if this
         * node's links will be subsequently changed.
         */
        private void tempRemoveFromQueue() {
            if (inQueue()) {
                previousInQueue.nextInQueue = nextInQueue;
                nextInQueue.previousInQueue = previousInQueue;
            }
        }

        /**
         * Removes this entry from the queue.
         */
        private void removeFromQueue() {
            tempRemoveFromQueue();
            previousInQueue = null;
            nextInQueue = null;
        }

        /**
         * Inserts this entry before the specified existing entry in the queue.
         */
        private void addToQueueBefore(LIRSHashEntry<K, V> existingEntry) {
            previousInQueue = existingEntry.previousInQueue;
            nextInQueue = existingEntry;
            previousInQueue.nextInQueue = this;
            nextInQueue.previousInQueue = this;
        }

        /**
         * Moves this entry to the end of the queue.
         */
        private void moveToQueueEnd() {
            tempRemoveFromQueue();
            addToQueueBefore(owner.header);
        }

        /**
         * Moves this entry from the stack to the queue, marking it cold
         * (as hot entries must remain in the stack). This should only be called
         * on resident entries, as non-resident entries should not be made resident.
         * The bottom entry on the queue is always hot due to stack pruning.
         */
        private void migrateToQueue() {
            removeFromStack();
            cold();
        }

        /**
         * Moves this entry from the queue to the stack, marking it hot (as cold
         * resident entries must remain in the queue).
         */
        private void migrateToStack() {
            removeFromQueue();
            if (!inStack()) {
                moveToStackBottom();
            }
            hot();
        }

        /**
         * Evicts this entry, removing it from the queue and setting its status to
         * cold non-resident. If the entry is already absent from the stack, it is
         * removed from the backing map; otherwise it remains in order for its
         * recency to be maintained.
         */
        private void evict() {
            removeFromQueue();
            removeFromStack();
            nonResident();
            owner = null;
        }

        /**
         * Removes this entry from the cache. This operation is not specified in
         * the paper, which does not account for forced eviction.
         */
        private V remove() {
            boolean wasHot = (state == Recency.LIR_RESIDENT);
            V result = value;
            LIRSHashEntry<K, V> end = owner != null ? owner.queueEnd() : null;
            evict();

            // attempt to maintain a constant number of hot entries
            if (wasHot) {
                if (end != null) {
                    end.migrateToStack();
                }
            }

            return result;
        }
    }

    static final class LIRS<K, V> implements EvictionPolicy<K, V> {

        /**
         * The percentage of the cache which is dedicated to hot blocks.
         * See section 5.1
         */
        private static final float L_LIRS = 0.95f;

        /**
         * The owning segment
         */
        private final Segment<K, V> segment;

        /**
         * The accessQueue for reducing lock contention
         * See "BP-Wrapper: a system framework making any replacement algorithms
         * (almost) lock contention free"
         * <p/>
         * http://www.cse.ohio-state.edu/hpcs/WWW/HTML/publications/abs09-1.html
         */
        private final ConcurrentLinkedQueue<LIRSHashEntry<K, V>> accessQueue;

        /**
         * The maxBatchQueueSize
         * <p/>
         * See "BP-Wrapper: a system framework making any replacement algorithms (almost) lock
         * contention free"
         */
        private final int maxBatchQueueSize;

        /**
         * The number of LIRS entries in a segment
         */
        private int size;

        private final float batchThresholdFactor;

        /**
         * This header encompasses two data structures:
         * <p/>
         * <ul>
         * <li>The LIRS stack, S, which is maintains recency information. All hot
         * entries are on the stack. All cold and non-resident entries which are more
         * recent than the least recent hot entry are also stored in the stack (the
         * stack is always pruned such that the last entry is hot, and all entries
         * accessed more recently than the last hot entry are present in the stack).
         * The stack is ordered by recency, with its most recently accessed entry
         * at the top, and its least recently accessed entry at the bottom.</li>
         * <p/>
         * <li>The LIRS queue, Q, which enqueues all cold entries for eviction. Cold
         * entries (by definition in the queue) may be absent from the stack (due to
         * pruning of the stack). Cold entries are added to the end of the queue
         * and entries are evicted from the front of the queue.</li>
         * </ul>
         */
        private final LIRSHashEntry<K, V> header = new LIRSHashEntry<K, V>(null, null, 0, null, null);

        /**
         * The maximum number of hot entries (L_lirs in the paper).
         */
        private final int maximumHotSize;

        /**
         * The maximum number of resident entries (L in the paper).
         */
        private final int maximumSize;

        /**
         * The actual number of hot entries.
         */
        private int hotSize;

        public LIRS(Segment<K, V> s, int capacity, int maxBatchSize, float batchThresholdFactor) {
            this.segment = s;
            this.maximumSize = capacity;
            this.maximumHotSize = calculateLIRSize(capacity);
            this.maxBatchQueueSize = maxBatchSize > MAX_BATCH_SIZE ? MAX_BATCH_SIZE : maxBatchSize;
            this.batchThresholdFactor = batchThresholdFactor;
            this.accessQueue = new ConcurrentLinkedQueue<LIRSHashEntry<K, V>>();
        }

        private static int calculateLIRSize(int maximumSize) {
            int result = (int) (L_LIRS * maximumSize);
            return (result == maximumSize) ? maximumSize - 1 : result;
        }

        @Override
        public Set<HashEntry<K, V>> execute() {
            Set<HashEntry<K, V>> evicted = new HashSet<HashEntry<K, V>>();
            try {
                for (LIRSHashEntry<K, V> e : accessQueue) {
                    if (e.isResident()) {
                        e.hit(evicted);
                    }
                }
                removeFromSegment(evicted);
            }
            finally {
                accessQueue.clear();
            }
            return evicted;
        }

        /**
         * Prunes HIR blocks in the bottom of the stack until an HOT block sits in
         * the stack bottom. If pruned blocks were resident, then they
         * remain in the queue; otherwise they are no longer referenced, and are thus
         * removed from the backing map.
         */
        private void pruneStack(Set<HashEntry<K, V>> evicted) {
            // See section 3.3:
            // "We define an operation called "stack pruning" on the LIRS
            // stack S, which removes the HIR blocks in the bottom of
            // the stack until an LIR block sits in the stack bottom. This
            // operation serves for two purposes: (1) We ensure the block in
            // the bottom of the stack always belongs to the LIR block set.
            // (2) After the LIR block in the bottom is removed, those HIR
            // blocks contiguously located above it will not have chances to
            // change their status from HIR to LIR, because their recencies
            // are larger than the new maximum recency of LIR blocks."
            LIRSHashEntry<K, V> bottom = stackBottom();
            while (bottom != null && bottom.state != Recency.LIR_RESIDENT) {
                bottom.removeFromStack();
                if (bottom.state == Recency.HIR_NONRESIDENT) {
                    evicted.add(bottom);
                }
                bottom = stackBottom();
            }
        }

        @Override
        public Set<HashEntry<K, V>> onEntryMiss(HashEntry<K, V> en) {
            LIRSHashEntry<K, V> e = (LIRSHashEntry<K, V>) en;
            Set<HashEntry<K, V>> evicted = e.miss();
            removeFromSegment(evicted);
            return evicted;
        }

        private void removeFromSegment(Set<HashEntry<K, V>> evicted) {
            for (HashEntry<K, V> e : evicted) {
                ((LIRSHashEntry<K, V>) e).evict();
                segment.evictionListener.onEntryChosenForEviction(e.value);
                segment.remove(e.key, e.hash, null);
            }
        }

        /*
         * Invoked without holding a lock on Segment
         */
        @Override
        public boolean onEntryHit(HashEntry<K, V> e) {
            accessQueue.add((LIRSHashEntry<K, V>) e);
            return accessQueue.size() >= maxBatchQueueSize * batchThresholdFactor;
        }

        /*
         * Invoked without holding a lock on Segment
         */
        @Override
        public boolean thresholdExpired() {
            return accessQueue.size() >= maxBatchQueueSize;
        }

        @Override
        public void onEntryRemove(HashEntry<K, V> e) {

            ((LIRSHashEntry<K, V>) e).remove();
            // we could have multiple instances of e in accessQueue; remove them all
            while (accessQueue.remove(e)) {
            }
        }

        @Override
        public void clear() {
            accessQueue.clear();
        }

        @Override
        public Eviction strategy() {
            return Eviction.LIRS;
        }

        /**
         * Returns the entry at the bottom of the stack.
         */
        private LIRSHashEntry<K, V> stackBottom() {
            LIRSHashEntry<K, V> bottom = header.previousInStack;
            return (bottom == header) ? null : bottom;
        }

        /**
         * Returns the entry at the front of the queue.
         */
        private LIRSHashEntry<K, V> queueFront() {
            LIRSHashEntry<K, V> front = header.nextInQueue;
            return (front == header) ? null : front;
        }

        /**
         * Returns the entry at the end of the queue.
         */
        private LIRSHashEntry<K, V> queueEnd() {
            LIRSHashEntry<K, V> end = header.previousInQueue;
            return (end == header) ? null : end;
        }

        @Override
        public HashEntry<K, V> createNewEntry(K key, int hash, HashEntry<K, V> next, V value) {
            return new LIRSHashEntry<K, V>(this, key, hash, next, value);
        }
    }

    /**
     * Segments are specialized versions of hash tables.  This
     * subclasses from ReentrantLock opportunistically, just to
     * simplify some locking and avoid separate construction.
     */
    static final class Segment<K, V> extends ReentrantLock {
        /*
         * Segments maintain a table of entry lists that are ALWAYS
         * kept in a consistent state, so can be read without locking.
         * Next fields of nodes are immutable (final). All list
         * additions are performed at the front of each bin. This
         * makes it easy to check changes, and also fast to traverse.
         * When nodes would otherwise be changed, new nodes are
         * created to replace them. This works well for hash tables
         * since the bin lists tend to be short. (The average length
         * is less than two for the default load factor threshold.)
         *
         * Read operations can thus proceed without locking, but rely
         * on selected uses of volatiles to ensure that completed
         * write operations performed by other threads are
         * noticed. For most purposes, the "count" field, tracking the
         * number of elements, serves as that volatile variable
         * ensuring visibility. This is convenient because this field
         * needs to be read in many read operations anyway:
         *
         * - All (unsynchronized) read operations must first read the
         * "count" field, and should not look at table entries if
         * it is 0.
         *
         * - All (synchronized) write operations should write to
         * the "count" field after structurally changing any bin.
         * The operations must not take any action that could even
         * momentarily cause a concurrent read operation to see
         * inconsistent data. This is made easier by the nature of
         * the read operations in Map. For example, no operation
         * can reveal that the table has grown but the threshold
         * has not yet been updated, so there are no atomicity
         * requirements for this with respect to reads.
         *
         * As a guide, all critical volatile reads and writes to the
         * count field are marked in code comments.
         */

        private static final long serialVersionUID = 2249069246763182397L;

        /**
         * The number of elements in this segment's region.
         */
        transient volatile int count;

        /**
         * Number of updates that alter the size of the table. This is
         * used during bulk-read methods to make sure they see a
         * consistent snapshot: If modCounts change during a traversal
         * of segments computing size or checking containsValue, then
         * we might have an inconsistent view of state so (usually)
         * must retry.
         */
        transient int modCount;

        /**
         * The table is rehashed when its size exceeds this threshold.
         * (The value of this field is always <tt>(int)(capacity *
         * loadFactor)</tt>.)
         */
        transient int threshold;

        /**
         * The per-segment table.
         */
        transient volatile HashEntry<K, V>[] table;

        /**
         * The load factor for the hash table.  Even though this value
         * is same for all segments, it is replicated to avoid needing
         * links to outer object.
         *
         * @serial
         */
        final float loadFactor;

        final int evictCap;

        transient final EvictionPolicy<K, V> eviction;

        transient final EvictionListener<K, V> evictionListener;

        Segment(int cap, int evictCap, float lf, Eviction es, EvictionListener<K, V> listener) {
            loadFactor = lf;
            this.evictCap = evictCap;
            eviction = es.make(this, evictCap, lf);
            evictionListener = listener;
            setTable(HashEntry.<K, V> newArray(cap));
        }

        @SuppressWarnings("unchecked")
        static <K, V> Segment<K, V>[] newArray(int i) {
            return new Segment[i];
        }

        EvictionListener<K, V> getEvictionListener() {
            return evictionListener;
        }

        /**
         * Sets table to new HashEntry array.
         * Call only while holding lock or in constructor.
         */
        void setTable(HashEntry<K, V>[] newTable) {
            threshold = (int) (newTable.length * loadFactor);
            table = newTable;
        }

        /**
         * Returns properly casted first entry of bin for given hash.
         */
        HashEntry<K, V> getFirst(int hash) {
            HashEntry<K, V>[] tab = table;
            return tab[hash & tab.length - 1];
        }

        /**
         * Reads value field of an entry under lock. Called if value
         * field ever appears to be null. This is possible only if a
         * compiler happens to reorder a HashEntry initialization with
         * its table assignment, which is legal under memory model
         * but is not known to ever occur.
         */
        V readValueUnderLock(HashEntry<K, V> e) {
            lock();
            try {
                return e.value;
            }
            finally {
                unlock();
            }
        }

        /* Specialized implementations of map methods */

        V get(Object key, int hash) {
            int c = count;
            if (c != 0) { // read-volatile
                V result = null;
                HashEntry<K, V> e = getFirst(hash);
                while (e != null) {
                    if (e.hash == hash && key.equals(e.key)) {
                        V v = e.value;
                        if (v != null) {
                            result = v;
                            break;
                        }
                        else {
                            result = readValueUnderLock(e); // recheck
                            break;
                        }
                    }
                    e = e.next;
                }
                // a hit
                if (result != null) {
                    if (eviction.onEntryHit(e)) {
                        Set<HashEntry<K, V>> evicted = attemptEviction(false);
                        notifyEvictionListener(evicted);
                    }
                }
                return result;
            }
            return null;
        }

        boolean containsKey(Object key, int hash) {
            if (count != 0) { // read-volatile
                HashEntry<K, V> e = getFirst(hash);
                while (e != null) {
                    if (e.hash == hash && key.equals(e.key)) {
                        return true;
                    }
                    e = e.next;
                }
            }
            return false;
        }

        boolean containsValue(Object value) {
            if (count != 0) { // read-volatile
                HashEntry<K, V>[] tab = table;
                int len = tab.length;
                for (int i = 0; i < len; i++) {
                    for (HashEntry<K, V> e = tab[i]; e != null; e = e.next) {
                        V v = e.value;
                        if (v == null) {
                            v = readValueUnderLock(e);
                        }
                        if (value.equals(v)) {
                            return true;
                        }
                    }
                }
            }
            return false;
        }

        boolean replace(K key, int hash, V oldValue, V newValue) {
            lock();
            Set<HashEntry<K, V>> evicted = null;
            try {
                HashEntry<K, V> e = getFirst(hash);
                while (e != null && (e.hash != hash || !key.equals(e.key))) {
                    e = e.next;
                }

                boolean replaced = false;
                if (e != null && oldValue.equals(e.value)) {
                    replaced = true;
                    e.value = newValue;
                    if (eviction.onEntryHit(e)) {
                        evicted = attemptEviction(true);
                    }
                }
                return replaced;
            }
            finally {
                unlock();
                notifyEvictionListener(evicted);
            }
        }

        V replace(K key, int hash, V newValue) {
            lock();
            Set<HashEntry<K, V>> evicted = null;
            try {
                HashEntry<K, V> e = getFirst(hash);
                while (e != null && (e.hash != hash || !key.equals(e.key))) {
                    e = e.next;
                }

                V oldValue = null;
                if (e != null) {
                    oldValue = e.value;
                    e.value = newValue;
                    if (eviction.onEntryHit(e)) {
                        evicted = attemptEviction(true);
                    }
                }
                return oldValue;
            }
            finally {
                unlock();
                notifyEvictionListener(evicted);
            }
        }

        V put(K key, int hash, V value, boolean onlyIfAbsent) {
            lock();
            Set<HashEntry<K, V>> evicted = null;
            try {
                int c = count;
                if (c++ > threshold && eviction.strategy() == Eviction.NONE) {
                    rehash();
                }
                HashEntry<K, V>[] tab = table;
                int index = hash & tab.length - 1;
                HashEntry<K, V> first = tab[index];
                HashEntry<K, V> e = first;
                while (e != null && (e.hash != hash || !key.equals(e.key))) {
                    e = e.next;
                }

                V oldValue;
                if (e != null) {
                    oldValue = e.value;
                    if (!onlyIfAbsent) {
                        e.value = value;
                        eviction.onEntryHit(e);
                    }
                }
                else {
                    oldValue = null;
                    ++modCount;
                    count = c; // write-volatile
                    if (eviction.strategy() != Eviction.NONE) {
                        if (c > evictCap) {
                            // remove entries;lower count
                            evicted = eviction.execute();
                            // re-read first
                            first = tab[index];
                        }
                        // add a new entry
                        tab[index] = eviction.createNewEntry(key, hash, first, value);
                        // notify a miss
                        Set<HashEntry<K, V>> newlyEvicted = eviction.onEntryMiss(tab[index]);
                        if (!newlyEvicted.isEmpty()) {
                            if (evicted != null) {
                                evicted.addAll(newlyEvicted);
                            }
                            else {
                                evicted = newlyEvicted;
                            }
                        }
                    }
                    else {
                        tab[index] = eviction.createNewEntry(key, hash, first, value);
                    }
                }
                return oldValue;
            }
            finally {
                unlock();
                notifyEvictionListener(evicted);
            }
        }

        void rehash() {
            HashEntry<K, V>[] oldTable = table;
            int oldCapacity = oldTable.length;
            if (oldCapacity >= MAXIMUM_CAPACITY) {
                return;
            }

            /*
             * Reclassify nodes in each list to new Map. Because we are
             * using power-of-two expansion, the elements from each bin
             * must either stay at same index, or move with a power of two
             * offset. We eliminate unnecessary node creation by catching
             * cases where old nodes can be reused because their next
             * fields won't change. Statistically, at the default
             * threshold, only about one-sixth of them need cloning when
             * a table doubles. The nodes they replace will be garbage
             * collectable as soon as they are no longer referenced by any
             * reader thread that may be in the midst of traversing table
             * right now.
             */

            HashEntry<K, V>[] newTable = HashEntry.newArray(oldCapacity << 1);
            threshold = (int) (newTable.length * loadFactor);
            int sizeMask = newTable.length - 1;
            for (int i = 0; i < oldCapacity; i++) {
                // We need to guarantee that any existing reads of old Map can
                // proceed. So we cannot yet null out each bin.
                HashEntry<K, V> e = oldTable[i];

                if (e != null) {
                    HashEntry<K, V> next = e.next;
                    int idx = e.hash & sizeMask;

                    // Single node on list
                    if (next == null) {
                        newTable[idx] = e;
                    }
                    else {
                        // Reuse trailing consecutive sequence at same slot
                        HashEntry<K, V> lastRun = e;
                        int lastIdx = idx;
                        for (HashEntry<K, V> last = next; last != null; last = last.next) {
                            int k = last.hash & sizeMask;
                            if (k != lastIdx) {
                                lastIdx = k;
                                lastRun = last;
                            }
                        }
                        newTable[lastIdx] = lastRun;

                        // Clone all remaining nodes
                        for (HashEntry<K, V> p = e; p != lastRun; p = p.next) {
                            int k = p.hash & sizeMask;
                            HashEntry<K, V> n = newTable[k];
                            newTable[k] = eviction.createNewEntry(p.key, p.hash, n, p.value);
                        }
                    }
                }
            }
            table = newTable;
        }

        /**
         * Remove; match on key only if value null, else match both.
         */
        V remove(Object key, int hash, Object value) {
            lock();
            try {
                int c = count - 1;
                HashEntry<K, V>[] tab = table;
                int index = hash & tab.length - 1;
                HashEntry<K, V> first = tab[index];
                HashEntry<K, V> e = first;
                while (e != null && (e.hash != hash || !key.equals(e.key))) {
                    e = e.next;
                }

                V oldValue = null;
                if (e != null) {
                    V v = e.value;
                    if (value == null || value.equals(v)) {
                        oldValue = v;
                        // All entries following removed node can stay
                        // in list, but all preceding ones need to be
                        // cloned.
                        ++modCount;

                        // e was removed
                        eviction.onEntryRemove(e);

                        HashEntry<K, V> newFirst = e.next;
                        for (HashEntry<K, V> p = first; p != e; p = p.next) {
                            // TODO A remove operation makes the map behave like all the other keys in the bucket were just added???
                            // allow p to be GC-ed
                            eviction.onEntryRemove(p);
                            newFirst = eviction.createNewEntry(p.key, p.hash, newFirst, p.value);
                            // and notify eviction algorithm about new hash entries
                            eviction.onEntryMiss(newFirst);
                        }

                        tab[index] = newFirst;
                        count = c; // write-volatile
                    }
                }
                return oldValue;
            }
            finally {
                unlock();
            }
        }

        void clear() {
            if (count != 0) {
                lock();
                try {
                    HashEntry<K, V>[] tab = table;
                    for (int i = 0; i < tab.length; i++) {
                        tab[i] = null;
                    }
                    ++modCount;
                    eviction.clear();
                    count = 0; // write-volatile
                }
                finally {
                    unlock();
                }
            }
        }

        private Set<HashEntry<K, V>> attemptEviction(boolean lockedAlready) {
            Set<HashEntry<K, V>> evicted = null;
            boolean obtainedLock = lockedAlready || tryLock();
            if (!obtainedLock && eviction.thresholdExpired()) {
                lock();
                obtainedLock = true;
            }
            if (obtainedLock) {
                try {
                    if (eviction.thresholdExpired()) {
                        evicted = eviction.execute();
                    }
                }
                finally {
                    if (!lockedAlready) {
                        unlock();
                    }
                }
            }
            return evicted;
        }

        private void notifyEvictionListener(Set<HashEntry<K, V>> evicted) {
            // piggyback listener invocation on callers thread outside lock
            if (evicted != null) {
                Map<K, V> evictedCopy;
                if (evicted.size() == 1) {
                    HashEntry<K, V> evictedEntry = evicted.iterator().next();
                    evictedCopy = singletonMap(evictedEntry.key, evictedEntry.value);
                }
                else {
                    evictedCopy = new HashMap<K, V>(evicted.size());
                    for (HashEntry<K, V> he : evicted) {
                        evictedCopy.put(he.key, he.value);
                    }
                    evictedCopy = unmodifiableMap(evictedCopy);
                }
                evictionListener.onEntryEviction(evictedCopy);
            }
        }
    }

    /* ---------------- Public operations -------------- */

    /**
     * Creates a new, empty map with the specified maximum capacity, load factor and concurrency
     * level.
     *
     * @param capacity is the upper bound capacity for the number of elements in this map
     * @param concurrencyLevel the estimated number of concurrently updating threads. The implementation performs
     * internal sizing to try to accommodate this many threads.
     * @param evictionStrategy the algorithm used to evict elements from this map
     * @param evictionListener the evicton listener callback to be notified about evicted elements
     *
     * @throws IllegalArgumentException if the initial capacity is negative or the load factor or concurrencyLevel are
     * nonpositive.
     */
    public BoundedConcurrentHashMap(
                                    int capacity, int concurrencyLevel,
                                    Eviction evictionStrategy, EvictionListener<K, V> evictionListener) {
        if (capacity < 0 || concurrencyLevel <= 0) {
            throw new IllegalArgumentException();
        }

        concurrencyLevel = Math.min(capacity / 2, concurrencyLevel); // concurrencyLevel cannot be > capacity/2
        concurrencyLevel = Math.max(concurrencyLevel, 1); // concurrencyLevel cannot be less than 1

        // minimum two elements per segment
        if (capacity < concurrencyLevel * 2 && capacity != 1) {
            throw new IllegalArgumentException("Maximum capacity has to be at least twice the concurrencyLevel");
        }

        if (evictionStrategy == null || evictionListener == null) {
            throw new IllegalArgumentException();
        }

        if (concurrencyLevel > MAX_SEGMENTS) {
            concurrencyLevel = MAX_SEGMENTS;
        }

        // Find power-of-two sizes best matching arguments
        int sshift = 0;
        int ssize = 1;
        while (ssize < concurrencyLevel) {
            ++sshift;
            ssize <<= 1;
        }
        segmentShift = 32 - sshift;
        segmentMask = ssize - 1;
        this.segments = Segment.newArray(ssize);

        if (capacity > MAXIMUM_CAPACITY) {
            capacity = MAXIMUM_CAPACITY;
        }
        int c = capacity / ssize;
        int cap = 1;
        while (cap < c) {
            cap <<= 1;
        }

        for (int i = 0; i < this.segments.length; ++i) {
            this.segments[i] = new Segment<K, V>(cap, c, DEFAULT_LOAD_FACTOR, evictionStrategy, evictionListener);
        }
    }

    /**
     * Creates a new, empty map with the specified maximum capacity, load factor, concurrency
     * level and LRU eviction policy.
     *
     * @param capacity is the upper bound capacity for the number of elements in this map
     * @param concurrencyLevel the estimated number of concurrently updating threads. The implementation performs
     * internal sizing to try to accommodate this many threads.
     *
     * @throws IllegalArgumentException if the initial capacity is negative or the load factor or concurrencyLevel are
     * nonpositive.
     */
    public BoundedConcurrentHashMap(int capacity, int concurrencyLevel) {
        this(capacity, concurrencyLevel, Eviction.LRU);
    }

    /**
     * Creates a new, empty map with the specified maximum capacity, load factor, concurrency
     * level and eviction strategy.
     *
     * @param capacity is the upper bound capacity for the number of elements in this map
     * @param concurrencyLevel the estimated number of concurrently updating threads. The implementation performs
     * internal sizing to try to accommodate this many threads.
     * @param evictionStrategy the algorithm used to evict elements from this map
     *
     * @throws IllegalArgumentException if the initial capacity is negative or the load factor or concurrencyLevel are
     * nonpositive.
     */
    public BoundedConcurrentHashMap(int capacity, int concurrencyLevel, Eviction evictionStrategy) {
        this(capacity, concurrencyLevel, evictionStrategy, new NullEvictionListener<K, V>());
    }

    /**
     * Creates a new, empty map with the specified maximum capacity, default concurrency
     * level and LRU eviction policy.
     *
     * @param capacity is the upper bound capacity for the number of elements in this map
     *
     * @throws IllegalArgumentException if the initial capacity of
     * elements is negative or the load factor is nonpositive
     * @since 1.6
     */
    public BoundedConcurrentHashMap(int capacity) {
        this(capacity, DEFAULT_CONCURRENCY_LEVEL);
    }

    /**
     * Creates a new, empty map with the default maximum capacity
     */
    public BoundedConcurrentHashMap() {
        this(DEFAULT_MAXIMUM_CAPACITY, DEFAULT_CONCURRENCY_LEVEL);
    }

    /**
     * Returns <tt>true</tt> if this map contains no key-value mappings.
     *
     * @return <tt>true</tt> if this map contains no key-value mappings
     */
    @Override
    public boolean isEmpty() {
        final Segment<K, V>[] segments = this.segments;
        /*
         * We keep track of per-segment modCounts to avoid ABA
         * problems in which an element in one segment was added and
         * in another removed during traversal, in which case the
         * table was never actually empty at any point. Note the
         * similar use of modCounts in the size() and containsValue()
         * methods, which are the only other methods also susceptible
         * to ABA problems.
         */
        int[] mc = new int[segments.length];
        int mcsum = 0;
        for (int i = 0; i < segments.length; ++i) {
            if (segments[i].count != 0) {
                return false;
            }
            else {
                mcsum += mc[i] = segments[i].modCount;
            }
        }
        // If mcsum happens to be zero, then we know we got a snapshot
        // before any modifications at all were made. This is
        // probably common enough to bother tracking.
        if (mcsum != 0) {
            for (int i = 0; i < segments.length; ++i) {
                if (segments[i].count != 0 || mc[i] != segments[i].modCount) {
                    return false;
                }
            }
        }
        return true;
    }

    /**
     * Returns the number of key-value mappings in this map.  If the
     * map contains more than <tt>Integer.MAX_VALUE</tt> elements, returns
     * <tt>Integer.MAX_VALUE</tt>.
     *
     * @return the number of key-value mappings in this map
     */
    @Override
    public int size() {
        final Segment<K, V>[] segments = this.segments;
        long sum = 0;
        long check = 0;
        int[] mc = new int[segments.length];
        // Try a few times to get accurate count. On failure due to
        // continuous async changes in table, resort to locking.
        for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
            check = 0;
            sum = 0;
            int mcsum = 0;
            for (int i = 0; i < segments.length; ++i) {
                sum += segments[i].count;
                mcsum += mc[i] = segments[i].modCount;
            }
            if (mcsum != 0) {
                for (int i = 0; i < segments.length; ++i) {
                    check += segments[i].count;
                    if (mc[i] != segments[i].modCount) {
                        check = -1; // force retry
                        break;
                    }
                }
            }
            if (check == sum) {
                break;
            }
        }
        if (check != sum) { // Resort to locking all segments
            sum = 0;
            for (int i = 0; i < segments.length; ++i) {
                segments[i].lock();
            }
            try {
                for (int i = 0; i < segments.length; ++i) {
                    sum += segments[i].count;
                }
            }
            finally {
                for (int i = 0; i < segments.length; ++i) {
                    segments[i].unlock();
                }
            }
        }
        if (sum > Integer.MAX_VALUE) {
            return Integer.MAX_VALUE;
        }
        else {
            return (int) sum;
        }
    }

    /**
     * Returns the value to which the specified key is mapped,
     * or {@code null} if this map contains no mapping for the key.
     * <p/>
     * <p>More formally, if this map contains a mapping from a key
     * {@code k} to a value {@code v} such that {@code key.equals(k)},
     * then this method returns {@code v}; otherwise it returns
     * {@code null}.  (There can be at most one such mapping.)
     *
     * @throws NullPointerException if the specified key is null
     */
    @Override
    public V get(Object key) {
        int hash = hash(key.hashCode());
        return segmentFor(hash).get(key, hash);
    }

    /**
     * Tests if the specified object is a key in this table.
     *
     * @param key possible key
     *
     * @return <tt>true</tt> if and only if the specified object
     * is a key in this table, as determined by the
     * <tt>equals</tt> method; <tt>false</tt> otherwise.
     *
     * @throws NullPointerException if the specified key is null
     */
    @Override
    public boolean containsKey(Object key) {
        int hash = hash(key.hashCode());
        return segmentFor(hash).containsKey(key, hash);
    }

    /**
     * Returns <tt>true</tt> if this map maps one or more keys to the
     * specified value. Note: This method requires a full internal
     * traversal of the hash table, and so is much slower than
     * method <tt>containsKey</tt>.
     *
     * @param value value whose presence in this map is to be tested
     *
     * @return <tt>true</tt> if this map maps one or more keys to the
     * specified value
     *
     * @throws NullPointerException if the specified value is null
     */
    @Override
    public boolean containsValue(Object value) {
        if (value == null) {
            throw new NullPointerException();
        }

        // See explanation of modCount use above

        final Segment<K, V>[] segments = this.segments;
        int[] mc = new int[segments.length];

        // Try a few times without locking
        for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
            int mcsum = 0;
            for (int i = 0; i < segments.length; ++i) {
                @SuppressWarnings("unused")
                int c = segments[i].count; // read-volatile
                mcsum += mc[i] = segments[i].modCount;
                if (segments[i].containsValue(value)) {
                    return true;
                }
            }
            boolean cleanSweep = true;
            if (mcsum != 0) {
                for (int i = 0; i < segments.length; ++i) {
                    @SuppressWarnings("unused")
                    int c = segments[i].count; // read-volatile
                    if (mc[i] != segments[i].modCount) {
                        cleanSweep = false;
                        break;
                    }
                }
            }
            if (cleanSweep) {
                return false;
            }
        }
        // Resort to locking all segments
        for (int i = 0; i < segments.length; ++i) {
            segments[i].lock();
        }
        boolean found = false;
        try {
            for (int i = 0; i < segments.length; ++i) {
                if (segments[i].containsValue(value)) {
                    found = true;
                    break;
                }
            }
        }
        finally {
            for (int i = 0; i < segments.length; ++i) {
                segments[i].unlock();
            }
        }
        return found;
    }

    /**
     * Legacy method testing if some key maps into the specified value
     * in this table.  This method is identical in functionality to
     * {@link #containsValue}, and exists solely to ensure
     * full compatibility with class {@link java.util.Hashtable},
     * which supported this method prior to introduction of the
     * Java Collections framework.
     *
     * @param value a value to search for
     *
     * @return <tt>true</tt> if and only if some key maps to the
     * <tt>value</tt> argument in this table as
     * determined by the <tt>equals</tt> method;
     * <tt>false</tt> otherwise
     *
     * @throws NullPointerException if the specified value is null
     */
    public boolean contains(Object value) {
        return containsValue(value);
    }

    /**
     * Maps the specified key to the specified value in this table.
     * Neither the key nor the value can be null.
     * <p/>
     * <p> The value can be retrieved by calling the <tt>get</tt> method
     * with a key that is equal to the original key.
     *
     * @param key key with which the specified value is to be associated
     * @param value value to be associated with the specified key
     *
     * @return the previous value associated with <tt>key</tt>, or
     * <tt>null</tt> if there was no mapping for <tt>key</tt>
     *
     * @throws NullPointerException if the specified key or value is null
     */
    @Override
    public V put(K key, V value) {
        if (value == null) {
            throw new NullPointerException();
        }
        int hash = hash(key.hashCode());
        return segmentFor(hash).put(key, hash, value, false);
    }

    /**
     * {@inheritDoc}
     *
     * @return the previous value associated with the specified key,
     * or <tt>null</tt> if there was no mapping for the key
     *
     * @throws NullPointerException if the specified key or value is null
     */
    @Override
    public V putIfAbsent(K key, V value) {
        if (value == null) {
            throw new NullPointerException();
        }
        int hash = hash(key.hashCode());
        return segmentFor(hash).put(key, hash, value, true);
    }

    /**
     * Copies all of the mappings from the specified map to this one.
     * These mappings replace any mappings that this map had for any of the
     * keys currently in the specified map.
     *
     * @param m mappings to be stored in this map
     */
    @Override
    public void putAll(Map<? extends K, ? extends V> m) {
        for (Map.Entry<? extends K, ? extends V> e : m.entrySet()) {
            put(e.getKey(), e.getValue());
        }
    }

    /**
     * Removes the key (and its corresponding value) from this map.
     * This method does nothing if the key is not in the map.
     *
     * @param key the key that needs to be removed
     *
     * @return the previous value associated with <tt>key</tt>, or
     * <tt>null</tt> if there was no mapping for <tt>key</tt>
     *
     * @throws NullPointerException if the specified key is null
     */
    @Override
    public V remove(Object key) {
        int hash = hash(key.hashCode());
        return segmentFor(hash).remove(key, hash, null);
    }

    /**
     * {@inheritDoc}
     *
     * @throws NullPointerException if the specified key is null
     */
    @Override
    public boolean remove(Object key, Object value) {
        int hash = hash(key.hashCode());
        if (value == null) {
            return false;
        }
        return segmentFor(hash).remove(key, hash, value) != null;
    }

    /**
     * {@inheritDoc}
     *
     * @throws NullPointerException if any of the arguments are null
     */
    @Override
    public boolean replace(K key, V oldValue, V newValue) {
        if (oldValue == null || newValue == null) {
            throw new NullPointerException();
        }
        int hash = hash(key.hashCode());
        return segmentFor(hash).replace(key, hash, oldValue, newValue);
    }

    /**
     * {@inheritDoc}
     *
     * @return the previous value associated with the specified key,
     * or <tt>null</tt> if there was no mapping for the key
     *
     * @throws NullPointerException if the specified key or value is null
     */
    @Override
    public V replace(K key, V value) {
        if (value == null) {
            throw new NullPointerException();
        }
        int hash = hash(key.hashCode());
        return segmentFor(hash).replace(key, hash, value);
    }

    /**
     * Removes all of the mappings from this map.
     */
    @Override
    public void clear() {
        for (int i = 0; i < segments.length; ++i) {
            segments[i].clear();
        }
    }

    /**
     * Returns a {@link Set} view of the keys contained in this map.
     * The set is backed by the map, so changes to the map are
     * reflected in the set, and vice-versa.  The set supports element
     * removal, which removes the corresponding mapping from this map,
     * via the <tt>Iterator.remove</tt>, <tt>Set.remove</tt>,
     * <tt>removeAll</tt>, <tt>retainAll</tt>, and <tt>clear</tt>
     * operations.  It does not support the <tt>add</tt> or
     * <tt>addAll</tt> operations.
     * <p/>
     * <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator
     * that will never throw {@link java.util.ConcurrentModificationException},
     * and guarantees to traverse elements as they existed upon
     * construction of the iterator, and may (but is not guaranteed to)
     * reflect any modifications subsequent to construction.
     */
    @Override
    public Set<K> keySet() {
        Set<K> ks = keySet;
        return ks != null ? ks : (keySet = new KeySet());
    }

    /**
     * Returns a {@link Collection} view of the values contained in this map.
     * The collection is backed by the map, so changes to the map are
     * reflected in the collection, and vice-versa.  The collection
     * supports element removal, which removes the corresponding
     * mapping from this map, via the <tt>Iterator.remove</tt>,
     * <tt>Collection.remove</tt>, <tt>removeAll</tt>,
     * <tt>retainAll</tt>, and <tt>clear</tt> operations.  It does not
     * support the <tt>add</tt> or <tt>addAll</tt> operations.
     * <p/>
     * <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator
     * that will never throw {@link java.util.ConcurrentModificationException},
     * and guarantees to traverse elements as they existed upon
     * construction of the iterator, and may (but is not guaranteed to)
     * reflect any modifications subsequent to construction.
     */
    @Override
    public Collection<V> values() {
        Collection<V> vs = values;
        return vs != null ? vs : (values = new Values());
    }

    /**
     * Returns a {@link Set} view of the mappings contained in this map.
     * The set is backed by the map, so changes to the map are
     * reflected in the set, and vice-versa.  The set supports element
     * removal, which removes the corresponding mapping from the map,
     * via the <tt>Iterator.remove</tt>, <tt>Set.remove</tt>,
     * <tt>removeAll</tt>, <tt>retainAll</tt>, and <tt>clear</tt>
     * operations.  It does not support the <tt>add</tt> or
     * <tt>addAll</tt> operations.
     * <p/>
     * <p>The view's <tt>iterator</tt> is a "weakly consistent" iterator
     * that will never throw {@link java.util.ConcurrentModificationException},
     * and guarantees to traverse elements as they existed upon
     * construction of the iterator, and may (but is not guaranteed to)
     * reflect any modifications subsequent to construction.
     */
    @Override
    public Set<Map.Entry<K, V>> entrySet() {
        Set<Map.Entry<K, V>> es = entrySet;
        return es != null ? es : (entrySet = new EntrySet());
    }

    /**
     * Returns an enumeration of the keys in this table.
     *
     * @return an enumeration of the keys in this table
     *
     * @see #keySet()
     */
    public Enumeration<K> keys() {
        return new KeyIterator();
    }

    /**
     * Returns an enumeration of the values in this table.
     *
     * @return an enumeration of the values in this table
     *
     * @see #values()
     */
    public Enumeration<V> elements() {
        return new ValueIterator();
    }

    /* ---------------- Iterator Support -------------- */

    abstract class HashIterator {
        int nextSegmentIndex;

        int nextTableIndex;

        HashEntry<K, V>[] currentTable;

        HashEntry<K, V> nextEntry;

        HashEntry<K, V> lastReturned;

        HashIterator() {
            nextSegmentIndex = segments.length - 1;
            nextTableIndex = -1;
            advance();
        }

        public boolean hasMoreElements() {
            return hasNext();
        }

        final void advance() {
            if (nextEntry != null && (nextEntry = nextEntry.next) != null) {
                return;
            }

            while (nextTableIndex >= 0) {
                if ((nextEntry = currentTable[nextTableIndex--]) != null) {
                    return;
                }
            }

            while (nextSegmentIndex >= 0) {
                Segment<K, V> seg = segments[nextSegmentIndex--];
                if (seg.count != 0) {
                    currentTable = seg.table;
                    for (int j = currentTable.length - 1; j >= 0; --j) {
                        if ((nextEntry = currentTable[j]) != null) {
                            nextTableIndex = j - 1;
                            return;
                        }
                    }
                }
            }
        }

        public boolean hasNext() {
            return nextEntry != null;
        }

        HashEntry<K, V> nextEntry() {
            if (nextEntry == null) {
                throw new NoSuchElementException();
            }
            lastReturned = nextEntry;
            advance();
            return lastReturned;
        }

        public void remove() {
            if (lastReturned == null) {
                throw new IllegalStateException();
            }
            BoundedConcurrentHashMap.this.remove(lastReturned.key);
            lastReturned = null;
        }
    }

    final class KeyIterator extends HashIterator implements Iterator<K>, Enumeration<K> {
        @Override
        public K next() {
            return super.nextEntry().key;
        }

        @Override
        public K nextElement() {
            return super.nextEntry().key;
        }
    }

    final class ValueIterator extends HashIterator implements Iterator<V>, Enumeration<V> {
        @Override
        public V next() {
            return super.nextEntry().value;
        }

        @Override
        public V nextElement() {
            return super.nextEntry().value;
        }
    }

    /**
     * Custom Entry class used by EntryIterator.next(), that relays
     * setValue changes to the underlying map.
     */
    final class WriteThroughEntry extends AbstractMap.SimpleEntry<K, V> {

        private static final long serialVersionUID = -7041346694785573824L;

        WriteThroughEntry(K k, V v) {
            super(k, v);
        }

        /**
         * Set our entry's value and write through to the map. The
         * value to return is somewhat arbitrary here. Since a
         * WriteThroughEntry does not necessarily track asynchronous
         * changes, the most recent "previous" value could be
         * different from what we return (or could even have been
         * removed in which case the put will re-establish). We do not
         * and cannot guarantee more.
         */
        @Override
        public V setValue(V value) {
            if (value == null) {
                throw new NullPointerException();
            }
            V v = super.setValue(value);
            BoundedConcurrentHashMap.this.put(getKey(), value);
            return v;
        }
    }

    final class EntryIterator extends HashIterator implements Iterator<Entry<K, V>> {
        @Override
        public Map.Entry<K, V> next() {
            HashEntry<K, V> e = super.nextEntry();
            return new WriteThroughEntry(e.key, e.value);
        }
    }

    final class KeySet extends AbstractSet<K> {
        @Override
        public Iterator<K> iterator() {
            return new KeyIterator();
        }

        @Override
        public int size() {
            return BoundedConcurrentHashMap.this.size();
        }

        @Override
        public boolean isEmpty() {
            return BoundedConcurrentHashMap.this.isEmpty();
        }

        @Override
        public boolean contains(Object o) {
            return BoundedConcurrentHashMap.this.containsKey(o);
        }

        @Override
        public boolean remove(Object o) {
            return BoundedConcurrentHashMap.this.remove(o) != null;
        }

        @Override
        public void clear() {
            BoundedConcurrentHashMap.this.clear();
        }
    }

    final class Values extends AbstractCollection<V> {
        @Override
        public Iterator<V> iterator() {
            return new ValueIterator();
        }

        @Override
        public int size() {
            return BoundedConcurrentHashMap.this.size();
        }

        @Override
        public boolean isEmpty() {
            return BoundedConcurrentHashMap.this.isEmpty();
        }

        @Override
        public boolean contains(Object o) {
            return BoundedConcurrentHashMap.this.containsValue(o);
        }

        @Override
        public void clear() {
            BoundedConcurrentHashMap.this.clear();
        }
    }

    final class EntrySet extends AbstractSet<Map.Entry<K, V>> {
        @Override
        public Iterator<Map.Entry<K, V>> iterator() {
            return new EntryIterator();
        }

        @Override
        public boolean contains(Object o) {
            if (!(o instanceof Map.Entry)) {
                return false;
            }
            Map.Entry<?, ?> e = (Map.Entry<?, ?>) o;
            V v = BoundedConcurrentHashMap.this.get(e.getKey());
            return v != null && v.equals(e.getValue());
        }

        @Override
        public boolean remove(Object o) {
            if (!(o instanceof Map.Entry)) {
                return false;
            }
            Map.Entry<?, ?> e = (Map.Entry<?, ?>) o;
            return BoundedConcurrentHashMap.this.remove(e.getKey(), e.getValue());
        }

        @Override
        public int size() {
            return BoundedConcurrentHashMap.this.size();
        }

        @Override
        public boolean isEmpty() {
            return BoundedConcurrentHashMap.this.isEmpty();
        }

        @Override
        public void clear() {
            BoundedConcurrentHashMap.this.clear();
        }
    }

    /* ---------------- Serialization Support -------------- */

    /**
     * Save the state of the <tt>ConcurrentHashMap</tt> instance to a
     * stream (i.e., serialize it).
     *
     * @param s the stream
     *
     * @serialData the key (Object) and value (Object)
     * for each key-value mapping, followed by a null pair.
     * The key-value mappings are emitted in no particular order.
     */
    private void writeObject(java.io.ObjectOutputStream s) throws IOException {
        s.defaultWriteObject();

        for (int k = 0; k < segments.length; ++k) {
            Segment<K, V> seg = segments[k];
            seg.lock();
            try {
                HashEntry<K, V>[] tab = seg.table;
                for (int i = 0; i < tab.length; ++i) {
                    for (HashEntry<K, V> e = tab[i]; e != null; e = e.next) {
                        s.writeObject(e.key);
                        s.writeObject(e.value);
                    }
                }
            }
            finally {
                seg.unlock();
            }
        }
        s.writeObject(null);
        s.writeObject(null);
    }

    /**
     * Reconstitute the <tt>ConcurrentHashMap</tt> instance from a
     * stream (i.e., deserialize it).
     *
     * @param s the stream
     */
    @SuppressWarnings("unchecked")
    private void readObject(java.io.ObjectInputStream s) throws IOException,
            ClassNotFoundException {
        s.defaultReadObject();

        // Initialize each segment to be minimally sized, and let grow.
        for (int i = 0; i < segments.length; ++i) {
            segments[i].setTable(new HashEntry[1]);
        }

        // Read the keys and values, and put the mappings in the table
        for (;;) {
            K key = (K) s.readObject();
            V value = (V) s.readObject();
            if (key == null) {
                break;
            }
            put(key, value);
        }
    }
}
